306 lines
16 KiB
Python
306 lines
16 KiB
Python
"""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)
|
||
# This is the IPC bound, not WPA2 policy. SAE passphrases can be longer
|
||
# than 63 characters; protocol validation happens before the write fence.
|
||
require(type(secret) is str and len(secret.encode()) <= 1024 and
|
||
all(ord(c) >= 32 and ord(c) != 127 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 valid_psk(secret):
|
||
return (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))
|
||
|
||
|
||
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')
|
||
old = None
|
||
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', 'sae') 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']:
|
||
security = 'wpa-psk'
|
||
if old == 'sae':
|
||
security = 'sae' # Never downgrade a saved WPA3-only profile.
|
||
elif not valid_psk(value['password']):
|
||
points = self.call(device, DEVICE + '.Wireless', 'GetAllAccessPoints').unpack()[0]
|
||
require(len(points) <= 512, 'too many Wi-Fi access points')
|
||
sae = any(bytes(self.prop(ap, SERVICE + '.AccessPoint', 'Ssid')) == ssid and
|
||
self.prop(ap, SERVICE + '.AccessPoint', 'RsnFlags') & 0x400 for ap in points)
|
||
require(sae, 'WPA2 needs 8–63 ASCII characters or a 64-digit hexadecimal key; this password requires a WPA3/SAE network')
|
||
security = 'sae'
|
||
settings.setdefault('802-11-wireless-security', {}).update(
|
||
{'key-mgmt': v('s', security), 'psk': v('s', value['password']), 'psk-flags': v('u', 0)})
|
||
if security == 'sae':
|
||
settings['802-11-wireless-security']['pmf'] = v('i', 3)
|
||
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)
|