# /usr/bin/env python """varlink cli tool Call with: $ python -m varlink.cli --help """ import argparse import json import shlex import socket import sys import varlink def varlink_call(args): deli = args.METHOD.rfind(".") if deli == -1: print("No method found", file=sys.stderr) sys.exit(1) method = args.METHOD[deli + 1 :] interface = args.METHOD[:deli] deli = interface.rfind("/") if deli != -1: address = interface[:deli] interface = interface[deli + 1 :] else: address = None def new_client(address): if address: client = varlink.Client.new_with_address(address) else: if args.activate: client = varlink.Client.new_with_activate(shlex.split(args.activate)) elif args.bridge: client = varlink.Client.new_with_bridge(shlex.split(args.bridge)) else: client = varlink.Client.new_with_resolved_interface(interface, args.resolver) return client with new_client(address) as client: got = False try: with client.open(interface) as con: out = { "method": interface + "." + method, "more": args.more, "parameters": json.loads(args.ARGUMENTS or "{}"), } con._send_message(json.dumps(out, cls=varlink.VarlinkEncoder).encode("utf-8")) more = True while more: (message, more) = con._next_varlink_message() if message: print(json.dumps(message, cls=varlink.VarlinkEncoder, indent=2, sort_keys=True)) got = True except varlink.VarlinkError as e: print(e, file=sys.stderr) except BrokenPipeError: if not got or args.more: print("Connection closed") sys.exit(1) def varlink_bridge(args): message = b"" last_interface = None con = None client = None if args.connect: client = varlink.Client.new_with_address(args.connect) con = client.open_connection() elif args.bridge: client = varlink.Client.new_with_bridge(shlex.split(args.bridge)) con = client.open_connection() elif args.activate: client = varlink.Client.new_with_activate(shlex.split(args.activate)) con = client.open_connection() else: print("No --connect or --bridge or --activate") if hasattr(sys.stdout, "buffer"): stdout = sys.stdout.buffer else: stdout = sys.stdout if hasattr(sys.stdin, "buffer"): stdin = sys.stdin.buffer else: stdin = sys.stdin while not sys.stdin.closed: c = stdin.read(1) if c == b"": break if c != b"\0": message += c continue req = json.loads(message.decode("utf-8")) if not args.connect and not args.activate and not args.bridge: if req["method"] == "org.varlink.service.GetInfo": req["method"] = "org.varlink.resolver.GetInfo" interface_name, _, method_name = req.get("method", "").rpartition(".") if req["method"] == "org.varlink.service.GetInterfaceDescription": resolving_interface = req["parameters"]["interface"] else: resolving_interface = interface_name if not interface_name or not method_name: stdout.write( json.dumps(varlink.InterfaceNotFound(interface_name), cls=varlink.VarlinkEncoder).encode( "utf-8" ) + b"\0" ) sys.stdout.flush() continue if last_interface != resolving_interface: if con: if hasattr(con, "shutdown"): con.shutdown(socket.SHUT_RDWR) else: con.close() if client: client.cleanup() try: client = varlink.Client.new_with_resolved_interface( resolving_interface, resolver_address=args.resolver ) except varlink.VarlinkError as e: stdout.write(json.dumps(e, cls=varlink.VarlinkEncoder).encode("utf-8") + b"\0") sys.stdout.flush() continue con = client.open_connection() last_interface = resolving_interface if hasattr(con, "send"): con.send(message + b"\0") else: con.write(message + b"\0") if req.get("oneway", False): continue message = b"" ret_message = b"" while True: c = con.recv(1) if c == b"": break if c != b"\0": ret_message += c continue stdout.write(ret_message + b"\0") sys.stdout.flush() ret = json.loads(ret_message.decode("utf-8")) ret_message = b"" if not ret.get("continues", False): break if ret.get("upgraded", False): raise NotImplementedError("Bridging upgraded connection not yet supported") def varlink_help(args): deli = args.INTERFACE.rfind("/") if deli != -1: address = args.INTERFACE[:deli] interface_name = args.INTERFACE[deli + 1 :] client = varlink.Client.new_with_address(address) else: interface_name = args.INTERFACE if args.activate: client = varlink.Client.new_with_activate(shlex.split(args.activate)) elif args.bridge: client = varlink.Client.new_with_bridge(shlex.split(args.bridge)) else: client = varlink.Client.new_with_resolved_interface(interface_name, args.resolver) interface = client.get_interface(interface_name) del client print(interface.description) def varlink_info(args): if args.ADDRESS: client = varlink.Client.new_with_address(args.ADDRESS) elif args.bridge: client = varlink.Client.new_with_bridge(shlex.split(args.bridge)) elif args.activate: client = varlink.Client.new_with_activate(shlex.split(args.activate)) else: print("No ADDRESS or --bridge or --activate") client.get_interfaces() info = client.info del client print("Vendor:", info["vendor"]) print("Product:", info["product"]) print("Version:", info["version"]) print("URL:", info["url"]) print("Interfaces:") for i in info["interfaces"]: print(" ", i) if __name__ == "__main__": parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(title="commands") parser.add_argument("-r", "--resolver", default=None, help="address of the resolver") parser.add_argument( "-A", "--activate", default=None, help="Service to socket-activate and connect to. " "The temporary UNIX socket address is exported as $VARLINK_ADDRESS.", ) parser.add_argument("-b", "--bridge", default=None, help="Command to execute and connect to") parser_info = subparsers.add_parser("info", help="Print information about a service") parser_info.add_argument("ADDRESS", nargs="?") parser_info.set_defaults(func=varlink_info) parser_help = subparsers.add_parser("help", help="Print interface description or service information") parser_help.add_argument("INTERFACE") parser_help.set_defaults(func=varlink_help) parser_bridge = subparsers.add_parser( "bridge", help="Bridge varlink messages to services on this machine" ) parser_bridge.add_argument( "-c", "--connect", default=None, help="Optional varlink address to connect to, without using the resolver.", ) parser_bridge.set_defaults(func=varlink_bridge) parser_call = subparsers.add_parser("call", help="Call a method") parser_call.add_argument( "-m", "--more", action="store_true", help="wait for multiple method returns if supported" ) parser_call.add_argument("METHOD") parser_call.add_argument("ARGUMENTS", nargs="?", default="") parser_call.set_defaults(func=varlink_call) args = parser.parse_args() if not hasattr(args, "func"): parser.print_usage(sys.stderr) sys.exit(1) args.func(args)