"""Finite NetworkManager D-Bus adapter; no commands or caller-selected bus paths. The production constructor connects only to the system bus. Tests inject an already-connected disposable bus. All effects pass the write-ahead fence first. """ import ipaddress import time import uuid SERVICE = 'org.freedesktop.NetworkManager' ROOT = '/org/freedesktop/NetworkManager' DEVICE = SERVICE + '.Device' CONNECTION = SERVICE + '.Settings.Connection' def require(ok, message='network settings refused'): if not ok: raise ValueError(message) def validate(mutation): require(type(mutation) is dict and len(mutation) == 1) kind, value = next(iter(mutation.items())) require(type(value) is dict) if kind == 'Wifi': require(set(value) == {'ssid', 'password'}) ssid, secret = value['ssid'], value['password'] require(type(ssid) is str and 0 < len(ssid.encode()) <= 32 and '\0' not in ssid) require(type(secret) is str and (secret == '' or 8 <= len(secret) <= 63 and all(32 <= ord(c) <= 126 for c in secret) or len(secret) == 64 and all(c in '0123456789abcdefABCDEF' for c in secret))) elif kind == 'Ethernet': require(set(value) == {'interface', 'ipv4'}) name = value['interface'] require(type(name) is str and 0 < len(name) < 16 and name != 'lo' and name[0].isascii() and name[0].isalnum() and all(c.isascii() and (c.isalnum() or c in '_-.') for c in name)) mode = value['ipv4'] if mode != 'Dhcp': require(type(mode) is dict and set(mode) == {'Static'}) s = mode['Static'] require(type(s) is dict and set(s) == {'address', 'prefix', 'gateway', 'dns'}) require(type(s['address']) is str and (s['gateway'] is None or type(s['gateway']) is str)) require(type(s['prefix']) is int and 0 < s['prefix'] <= 32) ip = ipaddress.IPv4Address(s['address']) net = ipaddress.IPv4Network((ip, s['prefix']), strict=False) require(not (ip.is_unspecified or ip.is_multicast or ip.is_loopback or int(ip) == 0xffffffff)) if s['prefix'] < 31: require(ip not in (net.network_address, net.broadcast_address)) if s['gateway'] is not None: gw = ipaddress.IPv4Address(s['gateway']) require(gw in net and gw != ip and not (gw.is_unspecified or gw.is_multicast or gw.is_loopback)) if s['prefix'] < 31: require(gw not in (net.network_address, net.broadcast_address)) dns_policy(s['dns']) elif kind == 'Dns': require(set(value) == {'policy'} and value['policy'] != 'Preserve') dns_policy(value['policy']) else: raise ValueError('unknown network mutation') def dns_policy(policy): if policy in ('Automatic', 'Preserve'): return require(type(policy) is dict and set(policy) == {'Servers'}) servers = policy['Servers'] require(type(servers) is list and 0 < len(servers) <= 6) parsed = [] for value in servers: require(type(value) is str and '%' not in value) ip = ipaddress.ip_address(value) require(not ip.is_unspecified and not ip.is_multicast and value != '255.255.255.255' and not (ip.version == 6 and ip.is_link_local) and ip not in parsed) parsed.append(ip) def variant_settings(value): # Keep the original D-Bus variants, including unknown/operator properties. result = {} for index in range(value.n_children()): entry = value.get_child_value(index) group = {} values = entry.get_child_value(1) for j in range(values.n_children()): item = values.get_child_value(j) group[item.get_child_value(0).get_string()] = item.get_child_value(1).get_variant() result[entry.get_child_value(0).get_string()] = group return result class NetworkManager: def __init__(self, bus=None): from gi.repository import Gio, GLib self.Gio, self.GLib = Gio, GLib self.bus = bus if bus is not None else Gio.bus_get_sync(Gio.BusType.SYSTEM, None) self.destination = 'org.freedesktop.DBus' self.end = time.clock_gettime(time.CLOCK_BOOTTIME) + 2 self.owner = self.call('/org/freedesktop/DBus', 'org.freedesktop.DBus', 'GetNameOwner', '(s)', (SERVICE,)).unpack()[0] require(self.call('/org/freedesktop/DBus', 'org.freedesktop.DBus', 'GetConnectionUnixUser', '(s)', (self.owner,)).unpack()[0] == 0, 'NetworkManager must be root owned') self.destination = self.owner # Pin unique owner; no restart/rebinding. def call(self, path, interface, method, signature=None, args=()): remaining = self.end - time.clock_gettime(time.CLOCK_BOOTTIME) require(remaining > 0, 'network operation expired') value = self.GLib.Variant(signature, args) if signature else None return self.bus.call_sync(self.destination, path, interface, method, value, None, self.Gio.DBusCallFlags.NO_AUTO_START, max(1, min(1000, int(remaining * 1000))), None) def prop(self, path, interface, name): return self.call(path, 'org.freedesktop.DBus.Properties', 'Get', '(ss)', (interface, name)).unpack()[0] def settings(self, path): return variant_settings(self.call(path, CONNECTION, 'GetSettings').get_child_value(0)) def device_profile(self, path): active = self.prop(path, DEVICE, 'ActiveConnection') require(active != '/', 'no active profile') return self.prop(active, SERVICE + '.Connection.Active', 'Connection') def devices(self, device_type): paths = self.call(ROOT, SERVICE, 'GetDevices').unpack()[0] require(len(paths) <= 128) return [p for p in paths if self.prop(p, DEVICE, 'DeviceType') == device_type and self.prop(p, DEVICE, 'Managed') and self.prop(p, DEVICE, 'Real') and not self.prop(p, DEVICE, 'Software')] def patch_dns(self, settings, policy): if policy == 'Preserve': return values = [] if policy == 'Automatic' else [ipaddress.ip_address(s) for s in policy['Servers']] v = self.GLib.Variant for family in (4, 6): group = settings.setdefault('ipv' + str(family), {}) ips = [ip for ip in values if ip.version == family] group['dns'] = v('au', [int.from_bytes(ip.packed, 'little') for ip in ips]) if family == 4 else v('aay', [list(ip.packed) for ip in ips]) group['ignore-auto-dns'] = v('b', policy != 'Automatic') # Do not leave conflicting modern/deprecated representations behind. group.pop('dns-data', None) def execute(self, mutation, end, before_effect, verifying): validate(mutation) self.end = end kind, value = next(iter(mutation.items())) if kind == 'Wifi': return self.wifi(value, before_effect, verifying) if kind == 'Ethernet': devices = [p for p in self.devices(1) if self.prop(p, DEVICE, 'Interface') == value['interface']] require(len(devices) == 1, 'Ethernet target is not unique') else: devices = self.devices(1) + self.devices(2) devices = [p for p in devices if self.prop(p, DEVICE, 'ActiveConnection') != '/'] require(0 < len(devices) <= 8, 'no bounded active uplink set') plans = [] v = self.GLib.Variant for device in devices: profile = self.device_profile(device) settings = self.settings(profile) version = self.prop(profile, CONNECTION, 'VersionId') applied = self.call(device, DEVICE, 'GetAppliedConnection', '(u)', (0,)) applied_version = applied.get_child_value(1).unpack() require(settings['connection']['type'].unpack() in ('802-3-ethernet', '802-11-wireless')) if kind == 'Dns': self.patch_dns(settings, value['policy']) else: group = settings.setdefault('ipv4', {}) mode = value['ipv4'] group.pop('addresses', None) if mode == 'Dhcp': group.update({'method': v('s', 'auto'), 'address-data': v('aa{sv}', []), 'gateway': v('s', '')}) else: s = mode['Static'] group.update({'method': v('s', 'manual'), 'address-data': v('aa{sv}', [{'address': v('s', s['address']), 'prefix': v('u', s['prefix'])}]), 'gateway': v('s', s['gateway'] or '')}) self.patch_dns(settings, s['dns']) plans.append((device, profile, version, applied_version, settings)) require(len({p[1] for p in plans}) == len(plans), 'shared active profile refused') # Freeze all targets before first effect. A later failure is Unknown. before_effect([p[1] for p in plans]) for device, profile, version, applied_version, settings in plans: require(self.device_profile(device) == profile) require(self.prop(profile, CONNECTION, 'VersionId') == version, 'profile changed concurrently') self.call(profile, CONNECTION, 'Update2', '(a{sa{sv}}ua{sv})', (settings, 1, {'version-id': v('t', version)})) self.call(device, DEVICE, 'Reapply', '(a{sa{sv}}tu)', (settings, applied_version, 0)) if kind == 'Ethernet': self.wait_ipv4(device, value['ipv4']) verifying() for device, profile, _, _, intended in plans: require(self.device_profile(device) == profile) self.verify_ip(intended, self.settings(profile)) applied = variant_settings(self.call(device, DEVICE, 'GetAppliedConnection', '(u)', (0,)).get_child_value(0)) self.verify_ip(intended, applied) def wait_ipv4(self, device, mode): # Reapply acceptance is not a DHCP lease or completed static activation. # All property calls share the immutable operation deadline. while True: state = self.prop(device, DEVICE, 'State') require(state not in (10, 20, 30, 120), 'Ethernet activation failed') config = self.prop(device, DEVICE, 'Ip4Config') if state == 100 and config != '/': addresses = self.prop(config, SERVICE + '.IP4Config', 'AddressData') if mode == 'Dhcp': if addresses: return else: desired = mode['Static'] if any(item.get('address') == desired['address'] and item.get('prefix') == desired['prefix'] for item in addresses): if not desired['gateway'] or self.prop(config, SERVICE + '.IP4Config', 'Gateway') == desired['gateway']: return time.sleep(.1) def verify_ip(self, intended, actual): require(actual.get('connection', {}).get('uuid') is not None and actual['connection']['uuid'].unpack() == intended['connection']['uuid'].unpack(), 'profile identity changed') for group in ('ipv4', 'ipv6'): for key in ('method', 'address-data', 'gateway', 'dns', 'ignore-auto-dns'): if key in intended.get(group, {}): expected = intended[group][key].unpack() observed = actual.get(group, {}).get(key) # NM omits default false/empty fields when normalizing. require((observed.unpack() if observed is not None else None) == expected or observed is None and expected in (False, '', []), 'network readback mismatch') def wifi(self, value, before_effect, verifying): v = self.GLib.Variant devices = self.devices(2) require(len(devices) == 1, 'Wi-Fi radio is not unique') device = devices[0] profiles = self.call(ROOT + '/Settings', SERVICE + '.Settings', 'ListConnections').unpack()[0] require(len(profiles) <= 256) matches = [] ssid = value['ssid'].encode() for path in profiles: settings = self.settings(path) if bytes(settings.get('802-11-wireless', {}).get('ssid', v('ay', [])).unpack()) == ssid: matches.append((path, settings)) require(len(matches) <= 1, 'ambiguous Wi-Fi profiles') if matches: profile, settings = matches[0] require(settings.get('user', {}).get('data', v('a{ss}', {})).unpack().get('archipelago.network-settings') == '1', 'operator profile preserved') old = settings.get('802-11-wireless-security', {}).get('key-mgmt', v('s', '')).unpack() require(old in ('', 'wpa-psk') and bool(old) == bool(value['password']), 'Wi-Fi security downgrade/change refused') version = self.prop(profile, CONNECTION, 'VersionId') else: profile, version = None, None settings = {'connection': {'id': v('s', 'Archipelago Wi-Fi'), 'uuid': v('s', str(uuid.uuid4())), 'type': v('s', '802-11-wireless'), 'autoconnect': v('b', True)}, '802-11-wireless': {'ssid': v('ay', list(ssid)), 'mode': v('s', 'infrastructure')}, 'ipv4': {'method': v('s', 'auto')}, 'ipv6': {'method': v('s', 'auto')}, 'user': {'data': v('a{ss}', {'archipelago.network-settings': '1'})}} # Carry explicitly selected DNS from this radio's existing uplink. if self.prop(device, DEVICE, 'ActiveConnection') != '/': previous = self.settings(self.device_profile(device)) for group in ('ipv4', 'ipv6'): for key in ('dns', 'dns-data', 'ignore-auto-dns', 'dns-search', 'dns-priority'): if key in previous.get(group, {}): settings[group][key] = previous[group][key] if value['password']: settings.setdefault('802-11-wireless-security', {}).update( {'key-mgmt': v('s', 'wpa-psk'), 'psk': v('s', value['password']), 'psk-flags': v('u', 0)}) before_effect([settings['connection']['uuid'].unpack()]) try: if profile: self.call(profile, CONNECTION, 'Update2', '(a{sa{sv}}ua{sv})', (settings, 1, {'version-id': v('t', version)})) else: profile = self.call(ROOT + '/Settings', SERVICE + '.Settings', 'AddConnection2', '(a{sa{sv}}ua{sv})', (settings, 1, {})).unpack()[0] active = self.call(ROOT, SERVICE, 'ActivateConnection', '(ooo)', (profile, device, '/')).unpack()[0] finally: settings.get('802-11-wireless-security', {}).pop('psk', None) value['password'] = '' while self.prop(active, SERVICE + '.Connection.Active', 'State') != 2: require(self.prop(active, SERVICE + '.Connection.Active', 'State') < 3, 'Wi-Fi activation failed') time.sleep(.1) verifying() require(self.device_profile(device) == profile) actual = self.settings(profile) require(actual['802-11-wireless']['ssid'].unpack() == list(ssid)) expected = settings.get('802-11-wireless-security', {}).get('key-mgmt') observed = actual.get('802-11-wireless-security', {}).get('key-mgmt') require((expected.unpack() if expected else '') == (observed.unpack() if observed else '')) self.verify_ip(settings, actual) applied = variant_settings(self.call(device, DEVICE, 'GetAppliedConnection', '(u)', (0,)).get_child_value(0)) self.verify_ip(settings, applied)