1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
#!/usr/bin/env python3
import os
import sys
sys.path.append(os.path.abspath(os.path.dirname(__file__) + "/.."))
import asyncio
import json
from argparse import OPTIONAL, ArgumentParser
from dbus_fast import BusType, Message, MessageType, Variant
from dbus_fast.aio import MessageBus
from dbus_fast.validators import (
is_bus_name_valid,
is_interface_name_valid,
is_member_name_valid,
is_object_path_valid,
)
parser = ArgumentParser()
parser.add_argument("--system", help="Use the system bus", action="store_true")
parser.add_argument("--session", help="Use the session bus", action="store_true")
parser.add_argument(
"--dest", help="The destination address for the message", required=True
)
parser.add_argument("--signature", help="The signature for the message body")
parser.add_argument(
"--type",
help="The type of message to send",
choices=[e.name for e in MessageType],
default=MessageType.METHOD_CALL.name,
nargs=OPTIONAL,
)
parser.add_argument("object_path", help="The object path for the message")
parser.add_argument("interface.member", help="The interface and member for the message")
parser.add_argument(
"body",
help="The JSON encoded body of the message. Must match the signature",
nargs=OPTIONAL,
)
args = parser.parse_args()
def exit_error(message):
parser.print_usage()
print()
print(message)
sys.exit(1)
interface_member = vars(args)["interface.member"].split(".")
if len(interface_member) < 2:
exit_error(
f"Expecting an interface and member separated by a dot: {vars(args)['interface.member']}"
)
destination = args.dest
member = interface_member[-1]
interface = ".".join(interface_member[: len(interface_member) - 1])
object_path = args.object_path
signature = args.signature
body = args.body
message_type = MessageType[args.type]
signature = args.signature
bus_type = BusType.SESSION
if args.system:
bus_type = BusType.SYSTEM
if message_type is not MessageType.METHOD_CALL:
exit_error("only message type METHOD_CALL is supported right now")
if not is_bus_name_valid(destination):
exit_error(f"got invalid bus name: {destination}")
if not is_object_path_valid(object_path):
exit_error(f"got invalid object path: {object_path}")
if not is_interface_name_valid(interface):
exit_error(f"got invalid interface name: {interface}")
if not is_member_name_valid(member):
exit_error(f"got invalid member name: {member}")
if body is None:
body = []
signature = ""
else:
try:
body = json.loads(body)
except json.JSONDecodeError as e:
exit_error(f"could not parse body as JSON: ({e})")
if type(body) is not list:
exit_error("body must be an array of arguments")
if not signature:
exit_error("--signature is a required argument when passing a message body")
async def main():
bus = await MessageBus(bus_type=bus_type).connect()
message = Message(
destination=destination,
member=member,
interface=interface,
path=object_path,
signature=signature,
body=body,
)
result = await bus.call(message)
ret = 0
if result.message_type is MessageType.ERROR:
print(f"Error: {result.error_name}", file=sys.stderr)
ret = 1
def default(o):
if type(o) is Variant:
return [o.signature, o.value]
raise json.JSONDecodeError()
print(json.dumps(result.body, indent=2, default=default))
sys.exit(ret)
asyncio.run(main())
|
