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
|
from ..constants import BusType
from ..errors import InvalidAddressError
from urllib.parse import unquote
import re
import os
invalid_address_chars_re = re.compile(r'[^-0-9A-Za-z_/.%]')
def parse_address(address_str):
addresses = []
for address in filter(lambda a: a, address_str.split(';')):
if address.find(':') == -1:
raise InvalidAddressError('address did not contain a transport')
transport, opt_string = address.split(':', 1)
options = {}
for kv in filter(lambda s: s, opt_string.split(',')):
if kv.find('=') == -1:
raise InvalidAddressError('address option did not contain a value')
k, v = kv.split('=', 1)
if invalid_address_chars_re.search(v):
raise InvalidAddressError('address contains invalid characters')
# XXX the actual unquote rules are simpler than this
v = unquote(v)
options[k] = v
addresses.append((transport, options))
if not addresses:
raise InvalidAddressError(f'address string contained no addresses: "{address_str}"')
return addresses
def get_system_bus_address():
if 'DBUS_SYSTEM_BUS_ADDRESS' in os.environ:
return os.environ['DBUS_SYSTEM_BUS_ADDRESS']
else:
return 'unix:path=/var/run/dbus/system_bus_socket'
display_re = re.compile(r'.*:([0-9]+)\.?.*')
remove_quotes_re = re.compile(r'''^['"]?(.*?)['"]?$''')
def get_session_bus_address():
if 'DBUS_SESSION_BUS_ADDRESS' in os.environ:
return os.environ['DBUS_SESSION_BUS_ADDRESS']
home = os.environ['HOME']
if 'DISPLAY' not in os.environ:
raise InvalidAddressError(
'DBUS_SESSION_BUS_ADDRESS not set and could not get DISPLAY environment variable to get bus address'
)
display = os.environ['DISPLAY']
try:
display = display_re.search(display).group(1)
except Exception:
raise InvalidAddressError(
f'DBUS_SESSION_BUS_ADDRESS not set and could not parse DISPLAY environment variable to get bus address: {display}'
)
# XXX: this will block but they're very small files and fs operations
# should be fairly reliable. fix this by passing in an async func to read
# the file for each io backend.
machine_id = None
with open('/var/lib/dbus/machine-id') as f:
machine_id = f.read().rstrip()
dbus_info_file_name = f'{home}/.dbus/session-bus/{machine_id}-{display}'
dbus_info = None
try:
with open(dbus_info_file_name) as f:
dbus_info = f.read().rstrip()
except Exception:
raise InvalidAddressError(f'could not open dbus info file: {dbus_info_file_name}')
for line in dbus_info.split('\n'):
if line.strip().startswith('DBUS_SESSION_BUS_ADDRESS='):
_, addr = line.split('=', 1)
if not addr:
raise InvalidAddressError(
f'DBUS_SESSION_BUS_ADDRESS variable not set correctly in dbus info file: {dbus_info_file_name}'
)
addr = remove_quotes_re.search(addr).group(1)
return addr
raise InvalidAddressError('could not find dbus session bus address')
def get_bus_address(bus_type):
if bus_type == BusType.SESSION:
return get_session_bus_address()
elif bus_type == BusType.SYSTEM:
return get_system_bus_address()
else:
raise Exception('got unknown bus type: {bus_type}')
|
