diff --git a/.github/workflows/gh-pages.yml b/.github/workflows/gh-pages.yml new file mode 100644 index 0000000..dbefb5b --- /dev/null +++ b/.github/workflows/gh-pages.yml @@ -0,0 +1,25 @@ +name: github pages + +on: + push: + branches: + - main + +jobs: + deploy: + runs-on: ubuntu-18.04 + steps: + - uses: actions/checkout@v2 + + - name: Build + uses: ammaraskar/sphinx-action@master + with: + pre-build-command: python -mpip install sphinx-rtd-theme + docs-folder: "docs/" + + - name: Deploy + uses: peaceiris/actions-gh-pages@v3 + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + publish_dir: ./docs/_build/html + force_orphan: true diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish.yml new file mode 100644 index 0000000..4e1ef42 --- /dev/null +++ b/.github/workflows/python-publish.yml @@ -0,0 +1,31 @@ +# This workflows will upload a Python Package using Twine when a release is created +# For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries + +name: Upload Python Package + +on: + release: + types: [created] + +jobs: + deploy: + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v2 + - name: Set up Python + uses: actions/setup-python@v2 + with: + python-version: '3.x' + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install setuptools wheel twine + - name: Build and publish + env: + TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }} + TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }} + run: | + python setup.py sdist bdist_wheel + twine upload dist/* diff --git a/COPYING b/COPYING index 58643d2..07971ab 100644 --- a/COPYING +++ b/COPYING @@ -1,5 +1,5 @@ python-networkmanager - Easy communication with NetworkManager -Copyright (C) 2011-2016 Dennis Kaarsemaker +Copyright (C) 2011-2021 Dennis Kaarsemaker This software is provided 'as-is', without any express or implied warranty. In no event will the authors be held liable for any damages diff --git a/NetworkManager.py b/NetworkManager.py index ddcc664..e66beba 100644 --- a/NetworkManager.py +++ b/NetworkManager.py @@ -1,155 +1,525 @@ # NetworkManager - a library to make interacting with the NetworkManager daemon # easier. # -# (C)2011-2016 Dennis Kaarsemaker +# (C)2011-2021 Dennis Kaarsemaker # License: zlib +import copy import dbus +import dbus.service import os +import six import socket import struct import sys +import time +import warnings +import weakref +import xml.etree.ElementTree as etree + +class ObjectVanished(Exception): + def __init__(self, obj): + self.obj = obj + super(ObjectVanished, self).__init__(obj.object_path) + +class SignalDispatcher(object): + def __init__(self): + self.handlers = {} + self.args = {} + self.interfaces = set() + self.setup = False + + def setup_signals(self): + if not self.setup: + bus = dbus.SystemBus() + for interface in self.interfaces: + bus.add_signal_receiver(self.handle_signal, dbus_interface=interface, interface_keyword='interface', member_keyword='signal', path_keyword='path') + self.setup = True + self.listen_for_restarts() + + def listen_for_restarts(self): + # If we have a mainloop, listen for disconnections + if not NMDbusInterface.last_disconnect and dbus.get_default_main_loop(): + dbus.SystemBus().add_signal_receiver(self.handle_restart, 'NameOwnerChanged', 'org.freedesktop.DBus') + NMDbusInterface.last_disconnect = 1 + + def add_signal_receiver(self, interface, signal, obj, func, args, kwargs): + self.setup_signals() + key = (interface, signal) + if key not in self.handlers: + self.handlers[key] = [] + self.handlers[key].append((obj, func, args, kwargs)) + + def handle_signal(self, *args, **kwargs): + key = (kwargs['interface'], kwargs['signal']) + skwargs = {} + sargs = [] + if key not in self.handlers: + return + try: + sender = fixups.base_to_python(kwargs['path']) + for arg, (name, signature) in zip(args, self.args[key]): + if name: + skwargs[name] = fixups.to_python(type(sender).__name__, kwargs['signal'], name, arg, signature) + else: + # Older NetworkManager versions don't supply attribute names. Hope for the best. + sargs.append(fixups.to_python(type(sender).__name__, kwargs['signal'], None, arg, signature)) + except dbus.exceptions.DBusException: + # This happens if the sender went away. Tough luck, no signal for you. + return + to_delete = [] + for pos, (match, receiver, rargs, rkwargs) in enumerate(self.handlers[key]): + try: + match == sender + except ObjectVanished: + to_delete.append(pos) + continue + if match == sender: + rkwargs['interface'] = kwargs['interface'] + rkwargs['signal'] = kwargs['signal'] + rkwargs.update(skwargs) + receiver(sender, *(sargs + rargs), **rkwargs) + for pos in reversed(to_delete): + self.handlers[key].pop(pos) + + def handle_restart(self, name, old, new): + if str(new) == "" or str(name) != 'org.freedesktop.NetworkManager': + return + NMDbusInterface.last_disconnect = time.time() + time.sleep(1) # Give NetworkManager a bit of time to start and rediscover itself. + for key in self.handlers: + val, self.handlers[key] = self.handlers[key], [] + for obj, func, args, kwargs in val: + try: + # This resets the object path if needed + obj.proxy + self.add_signal_receiver(key[0], key[1], obj, func, args, kwargs) + except ObjectVanished: + pass +SignalDispatcher = SignalDispatcher() + +# We completely dynamically generate all classes using introspection data. As +# this is done at import time, use a special dbus connection that does not get +# in the way of setting a mainloop and doing async stuff later. +init_bus = dbus.SystemBus(private=True) +xml_cache = {} + +class NMDbusInterfaceType(type): + """Metaclass that generates our classes based on introspection data""" + dbus_service = 'org.freedesktop.NetworkManager' + + def __new__(type_, name, bases, attrs): + attrs['dbus_service'] = type_.dbus_service + attrs['properties'] = [] + attrs['introspection_data'] = None + attrs['signals'] = [] + + # Derive the interface name from the name of the class, but let classes + # override it if needed + if 'interface_names' not in attrs and 'NMDbusInterface' not in name: + attrs['interface_names'] = ['org.freedesktop.NetworkManager.%s' % name] + for base in bases: + if hasattr(base, 'interface_names'): + attrs['interface_names'] = ['%s.%s' % (base.interface_names[0], name)] + base.interface_names + break + else: + for base in bases: + if hasattr(base, 'interface_names'): + attrs['interface_names'] += base.interface_names + break + + if 'interface_names' in attrs: + SignalDispatcher.interfaces.update(attrs['interface_names']) + + # If we know where to find this object, let's introspect it and + # generate properties and methods + if 'object_path' in attrs and attrs['object_path']: + proxy = init_bus.get_object(type_.dbus_service, attrs['object_path']) + attrs['introspection_data'] = proxy.Introspect(dbus_interface='org.freedesktop.DBus.Introspectable') + root = etree.fromstring(attrs['introspection_data']) + for element in root: + if element.tag == 'interface' and element.attrib['name'] in attrs['interface_names']: + for item in element: + if item.tag == 'property': + attrs[item.attrib['name']] = type_.make_property(name, element.attrib['name'], item.attrib) + attrs['properties'].append(item.attrib['name']) + elif item.tag == 'method': + aname = item.attrib['name'] + if aname in attrs: + aname = '_' + aname + attrs[aname] = type_.make_method(name, element.attrib['name'], item.attrib, list(item)) + elif item.tag == 'signal': + SignalDispatcher.args[(element.attrib['name'], item.attrib['name'])] = [(arg.attrib.get('name',None), arg.attrib['type']) for arg in item] + attrs['On' + item.attrib['name']] = type_.make_signal(name, element.attrib['name'], item.attrib) + attrs['signals'].append(item.attrib['name']) + + klass = super(NMDbusInterfaceType, type_).__new__(type_, name, bases, attrs) + return klass + + @staticmethod + def make_property(klass, interface, attrib): + name = attrib['name'] + def get_func(self): + try: + data = self.proxy.Get(interface, name, dbus_interface='org.freedesktop.DBus.Properties') + except dbus.exceptions.DBusException as e: + if e.get_dbus_name() == 'org.freedesktop.DBus.Error.UnknownMethod': + raise ObjectVanished(self) + raise + return fixups.to_python(klass, 'Get', name, data, attrib['type']) + if attrib['access'] == 'read': + return property(get_func) + def set_func(self, value): + value = fixups.to_dbus(klass, 'Set', name, value, attrib['type']) + try: + return self.proxy.Set(interface, name, value, dbus_interface='org.freedesktop.DBus.Properties') + except dbus.exceptions.DBusException as e: + if e.get_dbus_name() == 'org.freedesktop.DBus.Error.UnknownMethod': + raise ObjectVanished(self) + raise + return property(get_func, set_func) -PY3 = sys.version_info >= (3,0) -if PY3: - basestring = str - unicode = str -elif not hasattr(__builtins__, 'bytes'): - bytes = lambda x, y=None: chr(x[0]) if x else x - -try: - debuglevel = int(os.environ['NM_DEBUG']) - def debug(msg, data): - sys.stderr.write(msg + "\n") - sys.stderr.write(repr(data)+"\n") -except: - debug = lambda *args: None + @staticmethod + def make_method(klass, interface, attrib, args): + name = attrib['name'] + outargs = [x for x in args if x.tag == 'arg' and x.attrib['direction'] == 'out'] + outargstr = ', '.join([x.attrib['name'] for x in outargs]) or 'ret' + args = [x for x in args if x.tag == 'arg' and x.attrib['direction'] == 'in'] + argstr = ', '.join([x.attrib['name'] for x in args]) + ret = {} + code = "def %s(self%s):\n" % (name, ', ' + argstr if argstr else '') + for arg in args: + argname = arg.attrib['name'] + signature = arg.attrib['type'] + code += " %s = fixups.to_dbus('%s', '%s', '%s', %s, '%s')\n" % (argname, klass, name, argname, argname, signature) + code += " try:\n" + code += " %s = dbus.Interface(self.proxy, '%s').%s(%s)\n" % (outargstr, interface, name, argstr) + code += " except dbus.exceptions.DBusException as e:\n" + code += " if e.get_dbus_name() == 'org.freedesktop.DBus.Error.UnknownMethod':\n" + code += " raise ObjectVanished(self)\n" + code += " raise\n" + for arg in outargs: + argname = arg.attrib['name'] + signature = arg.attrib['type'] + code += " %s = fixups.to_python('%s', '%s', '%s', %s, '%s')\n" % (argname, klass, name, argname, argname, signature) + code += " return (%s)" % outargstr + exec(code, globals(), ret) + return ret[name] + @staticmethod + def make_signal(klass, interface, attrib): + name = attrib['name'] + ret = {} + code = "def On%s(self, func, *args, **kwargs):" % name + code += " SignalDispatcher.add_signal_receiver('%s', '%s', self, func, list(args), kwargs)" % (interface, name) + exec(code, globals(), ret) + return ret['On' + name] + +@six.add_metaclass(NMDbusInterfaceType) class NMDbusInterface(object): - bus = dbus.SystemBus() - dbus_service = 'org.freedesktop.NetworkManager' object_path = None + last_disconnect = 0 + is_transient = False + + def __new__(klass, object_path=None): + # If we didn't introspect this one at definition time, let's do it now. + if object_path and not klass.introspection_data: + proxy = dbus.SystemBus().get_object(klass.dbus_service, object_path) + klass.introspection_data = proxy.Introspect(dbus_interface='org.freedesktop.DBus.Introspectable') + root = etree.fromstring(klass.introspection_data) + for element in root: + if element.tag == 'interface' and element.attrib['name'] in klass.interface_names: + for item in element: + if item.tag == 'property': + setattr(klass, item.attrib['name'], type(klass).make_property(klass.__name__, element.attrib['name'], item.attrib)) + klass.properties.append(item.attrib['name']) + elif item.tag == 'method': + aname = item.attrib['name'] + if hasattr(klass, aname): + aname = '_' + aname + setattr(klass, aname, type(klass).make_method(klass.__name__, element.attrib['name'], item.attrib, list(item))) + elif item.tag == 'signal': + SignalDispatcher.args[(element.attrib['name'], item.attrib['name'])] = [(arg.attrib.get('name',None), arg.attrib['type']) for arg in item] + setattr(klass, 'On' + item.attrib['name'], type(klass).make_signal(klass.__name__, element.attrib['name'], item.attrib)) + klass.signals.append(item.attrib['name']) + + SignalDispatcher.listen_for_restarts() + return super(NMDbusInterface, klass).__new__(klass) def __init__(self, object_path=None): if isinstance(object_path, NMDbusInterface): object_path = object_path.object_path self.object_path = self.object_path or object_path - self.proxy = self.bus.get_object(self.dbus_service, self.object_path) - self.interface = dbus.Interface(self.proxy, self.interface_name) + self._proxy = None + + def __eq__(self, other): + return isinstance(other, NMDbusInterface) and self.object_path and other.object_path == self.object_path + + @property + def proxy(self): + if not self._proxy: + self._proxy = dbus.SystemBus().get_object(self.dbus_service, self.object_path, follow_name_owner_changes=True) + self._proxy.created = time.time() + elif self._proxy.created < self.last_disconnect: + if self.is_transient: + raise ObjectVanished(self) + obj = type(self)(self.object_path) + if obj.object_path != self.object_path: + self.object_path = obj.object_path + self._proxy = dbus.SystemBus().get_object(self.dbus_service, self.object_path) + self._proxy.created = time.time() + return self._proxy + + # Backwards compatibility interface + def connect_to_signal(self, signal, handler, *args, **kwargs): + return getattr(self, 'On' + signal)(handler, *args, **kwargs) + +class TransientNMDbusInterface(NMDbusInterface): + is_transient = True + +class NetworkManager(NMDbusInterface): + interface_names = ['org.freedesktop.NetworkManager'] + object_path = '/org/freedesktop/NetworkManager' + + # noop method for backward compatibility. It is no longer necessary to call + # this but let's not break code that does so. + def auto_reconnect(self): + pass + +class Statistics(NMDbusInterface): + object_path = '/org/freedesktop/NetworkManager/Statistics' - properties = [] +class Settings(NMDbusInterface): + object_path = '/org/freedesktop/NetworkManager/Settings' + +class AgentManager(NMDbusInterface): + object_path = '/org/freedesktop/NetworkManager/AgentManager' + +class Connection(NMDbusInterface): + interface_names = ['org.freedesktop.NetworkManager.Settings.Connection'] + has_secrets = ['802-1x', '802-11-wireless-security', 'cdma', 'gsm', 'pppoe', 'vpn'] + + def __init__(self, object_path): + super(Connection, self).__init__(object_path) + self.uuid = self.GetSettings()['connection']['uuid'] + + def GetSecrets(self, name=None): + settings = self.GetSettings() + if name is None: + name = settings['connection']['type'] + name = settings[name].get('security', name) try: - properties = self.proxy.GetAll(self.interface_name, - dbus_interface='org.freedesktop.DBus.Properties') + return self._GetSecrets(name) except dbus.exceptions.DBusException as e: - if e.get_dbus_name() != 'org.freedesktop.DBus.Error.UnknownMethod': + if e.get_dbus_name() != 'org.freedesktop.NetworkManager.AgentManager.NoSecrets': raise - for p in properties: - p = str(p) - if not hasattr(self.__class__, p): - setattr(self.__class__, p, self._make_property(p)) + return {key: {} for key in settings} - def _make_property(self, name): - def get_func(self): - data = self.proxy.Get(self.interface_name, name, dbus_interface='org.freedesktop.DBus.Properties') - debug("Received property %s.%s" % (self.interface_name, name), data) - return self.postprocess(name, self.unwrap(data)) - def set_func(self, value): - value = self.wrap(self.preprocess(name, (value,), {})[0][0]) - debug("Setting property %s.%s" % (self.interface_name, name), value) - return self.proxy.Set(self.interface_name, name, value, dbus_interface='org.freedesktop.DBus.Properties') - return property(get_func, set_func) + @staticmethod + def all(): + return Settings.ListConnections() - def unwrap(self, val): - if isinstance(val, dbus.ByteArray): - return "".join([str(x) for x in val]) - if isinstance(val, (dbus.Array, list, tuple)): - return [self.unwrap(x) for x in val] - if isinstance(val, (dbus.Dictionary, dict)): - return dict([(self.unwrap(x), self.unwrap(y)) for x,y in val.items()]) - if isinstance(val, dbus.ObjectPath): - if val.startswith('/org/freedesktop/NetworkManager/'): - classname = val.split('/')[4] - classname = { - 'Settings': 'Connection', - 'Devices': 'Device', - }.get(classname, classname) - return globals()[classname](val) - if isinstance(val, (dbus.Signature, dbus.String)): - return unicode(val) - if isinstance(val, dbus.Boolean): - return bool(val) - if isinstance(val, (dbus.Int16, dbus.UInt16, dbus.Int32, dbus.UInt32, dbus.Int64, dbus.UInt64)): - return int(val) - if isinstance(val, dbus.Byte): - return bytes([int(val)]) - return val + def __eq__(self, other): + return isinstance(other, type(self)) and self.uuid == other.uuid - def wrap(self, val): - if isinstance(val, NMDbusInterface): - return val.object_path - if hasattr(val, 'mro'): - for klass in val.mro(): - if klass.__module__ == '_dbus_bindings': - return val - if hasattr(val, '__iter__') and not isinstance(val, basestring): - if hasattr(val, 'items'): - return dict([(x, self.wrap(y)) for x, y in val.items()]) - else: - if (isinstance(val, dbus.Struct) or isinstance(val, dbus.ByteArray)): - return val - return [self.wrap(x) for x in val] - return val +class ActiveConnection(TransientNMDbusInterface): + interface_names = ['org.freedesktop.NetworkManager.Connection.Active'] + def __new__(klass, object_path): + if klass == ActiveConnection: + # Automatically turn this into a VPNConnection if needed + obj = dbus.SystemBus().get_object(klass.dbus_service, object_path) + if obj.Get('org.freedesktop.NetworkManager.Connection.Active', 'Vpn', dbus_interface='org.freedesktop.DBus.Properties'): + return VPNConnection.__new__(VPNConnection, object_path) + return super(ActiveConnection, klass).__new__(klass, object_path) - def __getattr__(self, name): - try: - return super(NMDbusInterface, self).__getattribute__(name) - except AttributeError: - return self.make_proxy_call(name) - - def make_proxy_call(self, name): - def proxy_call(*args, **kwargs): - func = getattr(self.interface, name) - args, kwargs = self.preprocess(name, args, kwargs) - args = self.wrap(args) - kwargs = self.wrap(kwargs) - debug("Calling function %s.%s" % (self.interface_name, name), (args, kwargs)) - ret = func(*args, **kwargs) - debug("Received return value for %s.%s" % (self.interface_name, name), ret) - return self.postprocess(name, self.unwrap(ret)) - return proxy_call + def __eq__(self, other): + return isinstance(other, type(self)) and self.Uuid == other.Uuid - def connect_to_signal(self, signal, handler, *args, **kwargs): - def helper(*args, **kwargs): - args = [self.unwrap(x) for x in args] - handler(*args, **kwargs) - args = self.wrap(args) - kwargs = self.wrap(kwargs) - return self.proxy.connect_to_signal(signal, helper, *args, **kwargs) - - def postprocess(self, name, val): - return val +class VPNConnection(ActiveConnection): + interface_names = ['org.freedesktop.NetworkManager.VPN.Connection'] - def preprocess(self, name, args, kwargs): - return args, kwargs +class Device(NMDbusInterface): + interface_names = ['org.freedesktop.NetworkManager.Device', 'org.freedesktop.NetworkManager.Device.Statistics'] + def __new__(klass, object_path): + if klass == Device: + # Automatically specialize the device + try: + obj = dbus.SystemBus().get_object(klass.dbus_service, object_path) + klass = device_class(obj.Get('org.freedesktop.NetworkManager.Device', 'DeviceType', dbus_interface='org.freedesktop.DBus.Properties')) + return klass.__new__(klass, object_path) + except ObjectVanished: + pass + return super(Device, klass).__new__(klass, object_path) -class NetworkManager(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager' - object_path = '/org/freedesktop/NetworkManager' + @staticmethod + def all(): + return NetworkManager.Devices + + def __eq__(self, other): + return isinstance(other, type(self)) and self.IpInterface == other.IpInterface + + # Backwards compatibility method. Devices now auto-specialize, so this is + # no longer needed. But code may use it. + def SpecificDevice(self): + return self + + +def device_class(typ): + return { + NM_DEVICE_TYPE_ADSL: Adsl, + NM_DEVICE_TYPE_BOND: Bond, + NM_DEVICE_TYPE_BRIDGE: Bridge, + NM_DEVICE_TYPE_BT: Bluetooth, + NM_DEVICE_TYPE_ETHERNET: Wired, + NM_DEVICE_TYPE_GENERIC: Generic, + NM_DEVICE_TYPE_INFINIBAND: Infiniband, + NM_DEVICE_TYPE_IP_TUNNEL: IPTunnel, + NM_DEVICE_TYPE_MACVLAN: Macvlan, + NM_DEVICE_TYPE_MODEM: Modem, + NM_DEVICE_TYPE_OLPC_MESH: OlpcMesh, + NM_DEVICE_TYPE_TEAM: Team, + NM_DEVICE_TYPE_TUN: Tun, + NM_DEVICE_TYPE_VETH: Veth, + NM_DEVICE_TYPE_VLAN: Vlan, + NM_DEVICE_TYPE_VXLAN: Vxlan, + NM_DEVICE_TYPE_WIFI: Wireless, + NM_DEVICE_TYPE_WIMAX: Wimax, + NM_DEVICE_TYPE_MACSEC: MacSec, + NM_DEVICE_TYPE_DUMMY: Dummy, + NM_DEVICE_TYPE_PPP: PPP, + NM_DEVICE_TYPE_OVS_INTERFACE: OvsIf, + NM_DEVICE_TYPE_OVS_PORT: OvsPort, + NM_DEVICE_TYPE_OVS_BRIDGE: OvsBridge, + NM_DEVICE_TYPE_WPAN: Wpan, + NM_DEVICE_TYPE_6LOWPAN: SixLoWpan, + NM_DEVICE_TYPE_WIREGUARD: WireGuard, + NM_DEVICE_TYPE_VRF: Vrf, + NM_DEVICE_TYPE_WIFI_P2P: WifiP2p, + }[typ] + +class Adsl(Device): pass +class Bluetooth(Device): pass +class Bond(Device): pass +class Bridge(Device): pass +class Generic(Device): pass +class Infiniband(Device): pass +class IPTunnel(Device): pass +class Macvlan(Device): pass +class Modem(Device): pass +class OlpcMesh(Device): pass +class Team(Device): pass +class Tun(Device): pass +class Veth(Device): pass +class Vlan(Device): pass +class Vxlan(Device): pass +class Wimax(Device): pass +class Wired(Device): pass +class Wireless(Device): pass +class MacSec(Device): pass +class Dummy(Device): pass +class PPP(Device): pass +class OvsIf(Device): pass +class OvsPort(Device): pass +class OvsBridge(Device): pass +class Wpan(Device): pass +class SixLoWpan(Device): pass +class WireGuard(Device): pass +class WifiP2p(Device): pass +class Vrf(Device): pass + +class NSP(TransientNMDbusInterface): + interface_names = ['org.freedesktop.NetworkManager.Wimax.NSP'] + +class AccessPoint(NMDbusInterface): + @staticmethod + def all(): + for device in Device.all(): + if isinstance(device, Wireless): + for ap in device.AccessPoints: + yield ap + def __eq__(self, other): + return isinstance(other, type(self)) and self.HwAddress == other.HwAddress + +class IP4Config(TransientNMDbusInterface): pass +class IP6Config(TransientNMDbusInterface): pass +class DHCP4Config(TransientNMDbusInterface): pass +class DHCP6Config(TransientNMDbusInterface): pass + +# Evil hack to work around not being able to specify a method name in the +# dbus.service.method decorator. +class SecretAgentType(type(dbus.service.Object)): + def __new__(type_, name, bases, attrs): + if bases != (dbus.service.Object,): + attrs['GetSecretsImpl'] = attrs.pop('GetSecrets') + return super(SecretAgentType, type_).__new__(type_, name, bases, attrs) + +@six.add_metaclass(SecretAgentType) +class SecretAgent(dbus.service.Object): + object_path = '/org/freedesktop/NetworkManager/SecretAgent' + interface_name = 'org.freedesktop.NetworkManager.SecretAgent' + + def __init__(self, identifier): + self.identifier = identifier + dbus.service.Object.__init__(self, dbus.SystemBus(), self.object_path) + AgentManager.Register(self.identifier) + + @dbus.service.method(dbus_interface=interface_name, in_signature='a{sa{sv}}osasu', out_signature='a{sa{sv}}') + def GetSecrets(self, connection, connection_path, setting_name, hints, flags): + settings = fixups.to_python('SecretAgent', 'GetSecrets', 'connection', connection, 'a{sa{sv}}') + connection = fixups.to_python('SecretAgent', 'GetSecrets', 'connection_path', connection_path, 'o') + setting_name = fixups.to_python('SecretAgent', 'GetSecrets', 'setting_name', setting_name, 's') + hints = fixups.to_python('SecretAgent', 'GetSecrets', 'hints', hints, 'as') + return self.GetSecretsImpl(settings, connection, setting_name, hints, flags) + +# These two are interfaces that must be provided to NetworkManager. Keep them +# as comments for documentation purposes. +# +# class PPP(NMDbusInterface): pass +# class VPNPlugin(NMDbusInterface): +# interface_names = ['org.freedesktop.NetworkManager.VPN.Plugin'] - def preprocess(self, name, args, kwargs): - if name in ('AddConnection', 'Update', 'AddAndActivateConnection'): - settings = args[0] +def const(prefix, val): + prefix = 'NM_' + prefix.upper() + '_' + for key, vval in globals().items(): + if 'REASON' in key and 'REASON' not in prefix: + continue + if key.startswith(prefix) and val == vval: + return key.replace(prefix,'').lower() + raise ValueError("No constant found for %s* with value %d", (prefix, val)) + +# Several fixer methods to make the data easier to handle in python +# - SSID sent/returned as bytes (only encoding tried is utf-8) +# - IP, Mac address and route metric encoding/decoding +class fixups(object): + @staticmethod + def to_dbus(klass, method, arg, val, signature): + if arg in ('connection' 'properties') and signature == 'a{sa{sv}}': + settings = copy.deepcopy(val) for key in settings: if 'mac-address' in settings[key]: settings[key]['mac-address'] = fixups.mac_to_dbus(settings[key]['mac-address']) if 'cloned-mac-address' in settings[key]: settings[key]['cloned-mac-address'] = fixups.mac_to_dbus(settings[key]['cloned-mac-address']) if 'bssid' in settings[key]: - settings[key]['bssid'] = fixups.mac_to_dbus(settings[key]['mac-address']) + settings[key]['bssid'] = fixups.mac_to_dbus(settings[key]['bssid']) + for cert in ['ca-cert', 'client-cert', 'phase2-ca-cert', 'phase2-client-cert', 'private-key']: + if cert in settings[key]: + settings[key][cert] = fixups.cert_to_dbus(settings[key][cert]) if 'ssid' in settings.get('802-11-wireless', {}): settings['802-11-wireless']['ssid'] = fixups.ssid_to_dbus(settings['802-11-wireless']['ssid']) if 'ipv4' in settings: + if 'address-data' in settings['ipv4']: + for item in settings['ipv4']['address-data']: + item['prefix'] = dbus.UInt32(item['prefix']) + settings['ipv4']['address-data'] = dbus.Array( + settings['ipv4']['address-data'], + signature=dbus.Signature('a{sv}')) + if 'route-data' in settings['ipv4']: + for item in settings['ipv4']['route-data']: + item['prefix'] = dbus.UInt32(item['prefix']) + settings['ipv4']['route-data'] = dbus.Array( + settings['ipv4']['route-data'], + signature=dbus.Signature('a{sv}')) if 'addresses' in settings['ipv4']: settings['ipv4']['addresses'] = [fixups.addrconf_to_dbus(addr,socket.AF_INET) for addr in settings['ipv4']['addresses']] if 'routes' in settings['ipv4']: @@ -157,41 +527,82 @@ def preprocess(self, name, args, kwargs): if 'dns' in settings['ipv4']: settings['ipv4']['dns'] = [fixups.addr_to_dbus(addr,socket.AF_INET) for addr in settings['ipv4']['dns']] if 'ipv6' in settings: + if 'address-data' in settings['ipv6']: + for item in settings['ipv6']['address-data']: + item['prefix'] = dbus.UInt32(item['prefix']) + settings['ipv6']['address-data'] = dbus.Array( + settings['ipv6']['address-data'], + signature=dbus.Signature('a{sv}')) + if 'route-data' in settings['ipv6']: + for item in settings['ipv6']['route-data']: + item['prefix'] = dbus.UInt32(item['prefix']) + settings['ipv6']['route-data'] = dbus.Array( + settings['ipv6']['route-data'], + signature=dbus.Signature('a{sv}')) if 'addresses' in settings['ipv6']: settings['ipv6']['addresses'] = [fixups.addrconf_to_dbus(addr,socket.AF_INET6) for addr in settings['ipv6']['addresses']] if 'routes' in settings['ipv6']: settings['ipv6']['routes'] = [fixups.route_to_dbus(route,socket.AF_INET6) for route in settings['ipv6']['routes']] if 'dns' in settings['ipv6']: settings['ipv6']['dns'] = [fixups.addr_to_dbus(addr,socket.AF_INET6) for addr in settings['ipv6']['dns']] - return args, kwargs -NetworkManager = NetworkManager() - -class Settings(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Settings' - object_path = '/org/freedesktop/NetworkManager/Settings' - preprocess = NetworkManager.preprocess -Settings = Settings() + # Get rid of empty arrays/dicts. dbus barfs on them (can't guess + # signatures), and if they were to get through, NetworkManager + # ignores them anyway. + for key in list(settings.keys()): + if isinstance(settings[key], dict): + for key2 in list(settings[key].keys()): + if settings[key][key2] in ({}, []): + del settings[key][key2] + if settings[key] in ({}, []): + del settings[key] + val = settings + return fixups.base_to_dbus(val) -class Connection(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Settings.Connection' - has_secrets = ['802-1x', '802-11-wireless-security', 'cdma', 'gsm', 'pppoe', 'vpn'] - - def GetSecrets(self, name=None): - if name == None: - settings = self.GetSettings() - for key in self.has_secrets: - if key in settings: - name = key - break + @staticmethod + def base_to_dbus(val): + if isinstance(val, NMDbusInterface): + return val.object_path + if hasattr(val.__class__, 'mro'): + for klass in val.__class__.mro(): + if klass.__module__ in ('dbus', '_dbus_bindings'): + return val + if hasattr(val, '__iter__') and not isinstance(val, six.string_types): + if hasattr(val, 'items'): + return dict([(x, fixups.base_to_dbus(y)) for x, y in val.items()]) else: - return {} - try: - return self.make_proxy_call('GetSecrets')(name) - except: - return {} + return [fixups.base_to_dbus(x) for x in val] + return val - def postprocess(self, name, val): - if name == 'GetSettings': + @staticmethod + def to_python(klass, method, arg, val, signature): + val = fixups.base_to_python(val) + klass_af = {'IP4Config': socket.AF_INET, 'IP6Config': socket.AF_INET6}.get(klass, socket.AF_INET) + if method == 'Get': + if arg == 'Ip4Address': + return fixups.addr_to_python(val, socket.AF_INET) + if arg == 'Ip6Address': + return fixups.addr_to_python(val, socket.AF_INET6) + if arg == 'Ssid': + return fixups.ssid_to_python(val) + if arg == 'Strength': + return fixups.strength_to_python(val) + if arg == 'Addresses': + return [fixups.addrconf_to_python(addr, klass_af) for addr in val] + if arg == 'Routes': + return [fixups.route_to_python(route, klass_af) for route in val] + if arg in ('Nameservers', 'WinsServers'): + return [fixups.addr_to_python(addr, klass_af) for addr in val] + if arg == 'Options': + for key in val: + if key.startswith('requested_'): + val[key] = bool(int(val[key])) + elif val[key].isdigit(): + val[key] = int(val[key]) + elif key in ('domain_name_servers', 'ntp_servers', 'routers'): + val[key] = val[key].split() + + return val + if method == 'GetSettings': if 'ssid' in val.get('802-11-wireless', {}): val['802-11-wireless']['ssid'] = fixups.ssid_to_python(val['802-11-wireless']['ssid']) for key in val: @@ -210,156 +621,55 @@ def postprocess(self, name, val): val['ipv6']['addresses'] = [fixups.addrconf_to_python(addr,socket.AF_INET6) for addr in val['ipv6']['addresses']] val['ipv6']['routes'] = [fixups.route_to_python(route,socket.AF_INET6) for route in val['ipv6']['routes']] val['ipv6']['dns'] = [fixups.addr_to_python(addr,socket.AF_INET6) for addr in val['ipv6']['dns']] + return val + if method == 'PropertiesChanged': + for prop in val: + val[prop] = fixups.to_python(klass, 'Get', prop, val[prop], None) return val - preprocess = NetworkManager.preprocess - -class ActiveConnection(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Connection.Active' - -class Device(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device' - - def SpecificDevice(self): - return { - NM_DEVICE_TYPE_ETHERNET: Wired, - NM_DEVICE_TYPE_WIFI: Wireless, - NM_DEVICE_TYPE_MODEM: Modem, - NM_DEVICE_TYPE_BT: Bluetooth, - NM_DEVICE_TYPE_OLPC_MESH: OlpcMesh, - NM_DEVICE_TYPE_WIMAX: Wimax, - NM_DEVICE_TYPE_INFINIBAND: Infiniband, - NM_DEVICE_TYPE_BOND: Bond, - NM_DEVICE_TYPE_VLAN: Vlan, - NM_DEVICE_TYPE_ADSL: Adsl, - NM_DEVICE_TYPE_BRIDGE: Bridge, - NM_DEVICE_TYPE_GENERIC: Generic - }[self.DeviceType](self.object_path) - - def postprocess(self, name, val): - if name == 'Ip4Address': - return fixups.addr_to_python(val,socket.AF_INET) - if name == 'Ip6Address': - return fixups.addr_to_python(val,socket.AF_INET6) - return val - -class AccessPoint(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.AccessPoint' - - def postprocess(self, name, val): - if name == 'Ssid': - return fixups.ssid_to_python(val) - elif name == 'Strength': - return fixups.strength_to_python(val) - return val - -class Wired(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Wired' - -class Wireless(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Wireless' - -class Modem(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Modem' - -class Bluetooth(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Bluetooth' - -class OlpcMesh(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.OlpcMesh' - -class Wimax(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Wimax' - -class Infiniband(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Infiniband' - -class Bond(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Bond' - -class Bridge(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Bridge' - -class Vlan(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Vlan' - -class Adsl(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.adsl' - -class Generic(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Device.Generic' - -class NSP(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.Wimax.NSP' - -class IP4Config(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.IP4Config' - - def postprocess(self, name, val): - if name == 'Addresses': - return [fixups.addrconf_to_python(addr,socket.AF_INET) for addr in val] - if name == 'Routes': - return [fixups.route_to_python(route,socket.AF_INET) for route in val] - if name in ('Nameservers', 'WinsServers'): - return [fixups.addr_to_python(addr,socket.AF_INET) for addr in val] - return val - -class IP6Config(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.IP6Config' - def postprocess(self, name, val): - if name == 'Addresses': - return [fixups.addrconf_to_python(addr,socket.AF_INET6) for addr in val] - if name == 'Routes': - return [fixups.route_to_python(route,socket.AF_INET6) for route in val] - if name in ('Nameservers', 'WinsServers'): - return [fixups.addr_to_python(addr,socket.AF_INET6) for addr in val] + @staticmethod + def base_to_python(val): + if isinstance(val, dbus.ByteArray): + return "".join([str(x) for x in val]) + if isinstance(val, (dbus.Array, list, tuple)): + return [fixups.base_to_python(x) for x in val] + if isinstance(val, (dbus.Dictionary, dict)): + return dict([(fixups.base_to_python(x), fixups.base_to_python(y)) for x,y in val.items()]) + if isinstance(val, dbus.ObjectPath): + for obj in (NetworkManager, Settings, AgentManager): + if val == obj.object_path: + return obj + if val.startswith('/org/freedesktop/NetworkManager/'): + classname = val.split('/')[4] + classname = { + 'Settings': 'Connection', + 'Devices': 'Device', + }.get(classname, classname) + return globals()[classname](val) + if val == '/': + return None + if isinstance(val, (dbus.Signature, dbus.String)): + return six.text_type(val) + if isinstance(val, dbus.Boolean): + return bool(val) + if isinstance(val, (dbus.Int16, dbus.UInt16, dbus.Int32, dbus.UInt32, dbus.Int64, dbus.UInt64)): + return int(val) + if isinstance(val, dbus.Byte): + return six.int2byte(int(val)) return val -class DHCP4Config(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.DHCP4Config' - -class DHCP6Config(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.DHCP6Config' - -class AgentManager(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.AgentManager' - -class SecretAgent(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.SecretAgent' - -class VPNConnection(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.VPN.Connection' - - def preprocess(self, name, args, kwargs): - conf = args[0] - conf['addresses'] = [fixups.addrconf_to_python(addr,socket.AF_INET) for addr in conf['addresses']] - conf['routes'] = [fixups.route_to_python(route,socket.AF_INET) for route in conf['routes']] - conf['dns'] = [fixups.addr_to_python(addr,socket.AF_INET) for addr in conf['dns']] - return args, kwargs - -class VPNPlugin(NMDbusInterface): - interface_name = 'org.freedesktop.NetworkManager.VPN.Plugin' - -def const(prefix, val): - prefix = 'NM_' + prefix.upper() + '_' - for key, vval in globals().items(): - if 'REASON' in key and 'REASON' not in prefix: - continue - if key.startswith(prefix) and val == vval: - return key.replace(prefix,'').lower() - raise ValueError("No constant found for %s* with value %d", (prefix, val)) - -# Several fixer methods to make the data easier to handle in python -# - SSID sent/returned as bytes (only encoding tried is utf-8) -# - IP, Mac address and route metric encoding/decoding -class fixups(object): @staticmethod def ssid_to_python(ssid): - return bytes("",'ascii').join(ssid).decode('utf-8') + try: + return bytes().join(ssid).decode('utf-8') + except UnicodeDecodeError: + ssid = bytes().join(ssid).decode('utf-8', 'replace') + warnings.warn("Unable to decode ssid %s properly" % ssid, UnicodeWarning) + return ssid @staticmethod def ssid_to_dbus(ssid): - if isinstance(ssid, unicode): + if isinstance(ssid, six.text_type): ssid = ssid.encode('utf-8') return [dbus.Byte(x) for x in ssid] @@ -427,7 +737,7 @@ def route_to_python(route,family): fixups.addr_to_python(addr,family), netmask, fixups.addr_to_python(gateway,family), - socket.ntohl(metric) + metric ] @staticmethod @@ -437,10 +747,28 @@ def route_to_dbus(route,family): fixups.addr_to_dbus(addr,family), fixups.mask_to_dbus(netmask), fixups.addr_to_dbus(gateway,family), - socket.htonl(metric) + metric ] + @staticmethod + def cert_to_dbus(cert): + if not isinstance(cert, bytes): + if not cert.startswith('file://'): + cert = 'file://' + cert + cert = cert.encode('utf-8') + b'\0' + return [dbus.Byte(x) for x in cert] + +# Turn NetworkManager and Settings into singleton objects +NetworkManager = NetworkManager() +Settings = Settings() +AgentManager = AgentManager() +init_bus.close() +del init_bus +del xml_cache + # Constants below are generated with makeconstants.py. Do not edit manually. +NM_CAPABILITY_TEAM = 1 +NM_CAPABILITY_OVS = 2 NM_STATE_UNKNOWN = 0 NM_STATE_ASLEEP = 10 NM_STATE_DISCONNECTED = 20 @@ -470,10 +798,27 @@ def route_to_dbus(route,family): NM_DEVICE_TYPE_BRIDGE = 13 NM_DEVICE_TYPE_GENERIC = 14 NM_DEVICE_TYPE_TEAM = 15 +NM_DEVICE_TYPE_TUN = 16 +NM_DEVICE_TYPE_IP_TUNNEL = 17 +NM_DEVICE_TYPE_MACVLAN = 18 +NM_DEVICE_TYPE_VXLAN = 19 +NM_DEVICE_TYPE_VETH = 20 +NM_DEVICE_TYPE_MACSEC = 21 +NM_DEVICE_TYPE_DUMMY = 22 +NM_DEVICE_TYPE_PPP = 23 +NM_DEVICE_TYPE_OVS_INTERFACE = 24 +NM_DEVICE_TYPE_OVS_PORT = 25 +NM_DEVICE_TYPE_OVS_BRIDGE = 26 +NM_DEVICE_TYPE_WPAN = 27 +NM_DEVICE_TYPE_6LOWPAN = 28 +NM_DEVICE_TYPE_WIREGUARD = 29 +NM_DEVICE_TYPE_WIFI_P2P = 30 +NM_DEVICE_TYPE_VRF = 31 NM_DEVICE_CAP_NONE = 0 NM_DEVICE_CAP_NM_SUPPORTED = 1 NM_DEVICE_CAP_CARRIER_DETECT = 2 NM_DEVICE_CAP_IS_SOFTWARE = 4 +NM_DEVICE_CAP_SRIOV = 8 NM_WIFI_DEVICE_CAP_NONE = 0 NM_WIFI_DEVICE_CAP_CIPHER_WEP40 = 1 NM_WIFI_DEVICE_CAP_CIPHER_WEP104 = 2 @@ -486,8 +831,13 @@ def route_to_dbus(route,family): NM_WIFI_DEVICE_CAP_FREQ_VALID = 256 NM_WIFI_DEVICE_CAP_FREQ_2GHZ = 512 NM_WIFI_DEVICE_CAP_FREQ_5GHZ = 1024 +NM_WIFI_DEVICE_CAP_MESH = 4096 +NM_WIFI_DEVICE_CAP_IBSS_RSN = 8192 NM_802_11_AP_FLAGS_NONE = 0 NM_802_11_AP_FLAGS_PRIVACY = 1 +NM_802_11_AP_FLAGS_WPS = 2 +NM_802_11_AP_FLAGS_WPS_PBC = 4 +NM_802_11_AP_FLAGS_WPS_PIN = 8 NM_802_11_AP_SEC_NONE = 0 NM_802_11_AP_SEC_PAIR_WEP40 = 1 NM_802_11_AP_SEC_PAIR_WEP104 = 2 @@ -499,10 +849,14 @@ def route_to_dbus(route,family): NM_802_11_AP_SEC_GROUP_CCMP = 128 NM_802_11_AP_SEC_KEY_MGMT_PSK = 256 NM_802_11_AP_SEC_KEY_MGMT_802_1X = 512 +NM_802_11_AP_SEC_KEY_MGMT_SAE = 1024 +NM_802_11_AP_SEC_KEY_MGMT_OWE = 2048 +NM_802_11_AP_SEC_KEY_MGMT_OWE_TM = 4096 NM_802_11_MODE_UNKNOWN = 0 NM_802_11_MODE_ADHOC = 1 NM_802_11_MODE_INFRA = 2 NM_802_11_MODE_AP = 3 +NM_802_11_MODE_MESH = 4 NM_BT_CAPABILITY_NONE = 0 NM_BT_CAPABILITY_DUN = 1 NM_BT_CAPABILITY_NAP = 2 @@ -511,6 +865,10 @@ def route_to_dbus(route,family): NM_DEVICE_MODEM_CAPABILITY_CDMA_EVDO = 2 NM_DEVICE_MODEM_CAPABILITY_GSM_UMTS = 4 NM_DEVICE_MODEM_CAPABILITY_LTE = 8 +NM_WIMAX_NSP_NETWORK_TYPE_UNKNOWN = 0 +NM_WIMAX_NSP_NETWORK_TYPE_HOME = 1 +NM_WIMAX_NSP_NETWORK_TYPE_PARTNER = 2 +NM_WIMAX_NSP_NETWORK_TYPE_ROAMING_PARTNER = 3 NM_DEVICE_STATE_UNKNOWN = 0 NM_DEVICE_STATE_UNMANAGED = 10 NM_DEVICE_STATE_UNAVAILABLE = 20 @@ -587,12 +945,129 @@ def route_to_dbus(route,family): NM_DEVICE_STATE_REASON_NEW_ACTIVATION = 60 NM_DEVICE_STATE_REASON_PARENT_CHANGED = 61 NM_DEVICE_STATE_REASON_PARENT_MANAGED_CHANGED = 62 -NM_DEVICE_STATE_REASON_LAST = 65535 +NM_DEVICE_STATE_REASON_OVSDB_FAILED = 63 +NM_DEVICE_STATE_REASON_IP_ADDRESS_DUPLICATE = 64 +NM_DEVICE_STATE_REASON_IP_METHOD_UNSUPPORTED = 65 +NM_DEVICE_STATE_REASON_SRIOV_CONFIGURATION_FAILED = 66 +NM_DEVICE_STATE_REASON_PEER_NOT_FOUND = 67 +NM_METERED_UNKNOWN = 0 +NM_METERED_YES = 1 +NM_METERED_NO = 2 +NM_METERED_GUESS_YES = 3 +NM_METERED_GUESS_NO = 4 +NM_CONNECTION_MULTI_CONNECT_DEFAULT = 0 +NM_CONNECTION_MULTI_CONNECT_SINGLE = 1 +NM_CONNECTION_MULTI_CONNECT_MANUAL_MULTIPLE = 2 +NM_CONNECTION_MULTI_CONNECT_MULTIPLE = 3 NM_ACTIVE_CONNECTION_STATE_UNKNOWN = 0 NM_ACTIVE_CONNECTION_STATE_ACTIVATING = 1 NM_ACTIVE_CONNECTION_STATE_ACTIVATED = 2 NM_ACTIVE_CONNECTION_STATE_DEACTIVATING = 3 NM_ACTIVE_CONNECTION_STATE_DEACTIVATED = 4 +NM_ACTIVE_CONNECTION_STATE_REASON_UNKNOWN = 0 +NM_ACTIVE_CONNECTION_STATE_REASON_NONE = 1 +NM_ACTIVE_CONNECTION_STATE_REASON_USER_DISCONNECTED = 2 +NM_ACTIVE_CONNECTION_STATE_REASON_DEVICE_DISCONNECTED = 3 +NM_ACTIVE_CONNECTION_STATE_REASON_SERVICE_STOPPED = 4 +NM_ACTIVE_CONNECTION_STATE_REASON_IP_CONFIG_INVALID = 5 +NM_ACTIVE_CONNECTION_STATE_REASON_CONNECT_TIMEOUT = 6 +NM_ACTIVE_CONNECTION_STATE_REASON_SERVICE_START_TIMEOUT = 7 +NM_ACTIVE_CONNECTION_STATE_REASON_SERVICE_START_FAILED = 8 +NM_ACTIVE_CONNECTION_STATE_REASON_NO_SECRETS = 9 +NM_ACTIVE_CONNECTION_STATE_REASON_LOGIN_FAILED = 10 +NM_ACTIVE_CONNECTION_STATE_REASON_CONNECTION_REMOVED = 11 +NM_ACTIVE_CONNECTION_STATE_REASON_DEPENDENCY_FAILED = 12 +NM_ACTIVE_CONNECTION_STATE_REASON_DEVICE_REALIZE_FAILED = 13 +NM_ACTIVE_CONNECTION_STATE_REASON_DEVICE_REMOVED = 14 +NM_SECRET_AGENT_GET_SECRETS_FLAG_NONE = 0 +NM_SECRET_AGENT_GET_SECRETS_FLAG_ALLOW_INTERACTION = 1 +NM_SECRET_AGENT_GET_SECRETS_FLAG_REQUEST_NEW = 2 +NM_SECRET_AGENT_GET_SECRETS_FLAG_USER_REQUESTED = 4 +NM_SECRET_AGENT_GET_SECRETS_FLAG_WPS_PBC_ACTIVE = 8 +NM_SECRET_AGENT_GET_SECRETS_FLAG_ONLY_SYSTEM = 2147483648 +NM_SECRET_AGENT_GET_SECRETS_FLAG_NO_ERRORS = 1073741824 +NM_IP_TUNNEL_MODE_UNKNOWN = 0 +NM_IP_TUNNEL_MODE_IPIP = 1 +NM_IP_TUNNEL_MODE_GRE = 2 +NM_IP_TUNNEL_MODE_SIT = 3 +NM_IP_TUNNEL_MODE_ISATAP = 4 +NM_IP_TUNNEL_MODE_VTI = 5 +NM_IP_TUNNEL_MODE_IP6IP6 = 6 +NM_IP_TUNNEL_MODE_IPIP6 = 7 +NM_IP_TUNNEL_MODE_IP6GRE = 8 +NM_IP_TUNNEL_MODE_VTI6 = 9 +NM_IP_TUNNEL_MODE_GRETAP = 10 +NM_IP_TUNNEL_MODE_IP6GRETAP = 11 +NM_CHECKPOINT_CREATE_FLAG_NONE = 0 +NM_CHECKPOINT_CREATE_FLAG_DESTROY_ALL = 1 +NM_CHECKPOINT_CREATE_FLAG_DELETE_NEW_CONNECTIONS = 2 +NM_CHECKPOINT_CREATE_FLAG_DISCONNECT_NEW_DEVICES = 4 +NM_CHECKPOINT_CREATE_FLAG_ALLOW_OVERLAPPING = 8 +NM_ROLLBACK_RESULT_OK = 0 +NM_ROLLBACK_RESULT_ERR_NO_DEVICE = 1 +NM_ROLLBACK_RESULT_ERR_DEVICE_UNMANAGED = 2 +NM_ROLLBACK_RESULT_ERR_FAILED = 3 +NM_SETTINGS_CONNECTION_FLAG_NONE = 0 +NM_SETTINGS_CONNECTION_FLAG_UNSAVED = 1 +NM_SETTINGS_CONNECTION_FLAG_NM_GENERATED = 2 +NM_SETTINGS_CONNECTION_FLAG_VOLATILE = 4 +NM_SETTINGS_CONNECTION_FLAG_EXTERNAL = 8 +NM_ACTIVATION_STATE_FLAG_NONE = 0 +NM_ACTIVATION_STATE_FLAG_IS_MASTER = 1 +NM_ACTIVATION_STATE_FLAG_IS_SLAVE = 2 +NM_ACTIVATION_STATE_FLAG_LAYER2_READY = 4 +NM_ACTIVATION_STATE_FLAG_IP4_READY = 8 +NM_ACTIVATION_STATE_FLAG_IP6_READY = 16 +NM_ACTIVATION_STATE_FLAG_MASTER_HAS_SLAVES = 32 +NM_ACTIVATION_STATE_FLAG_LIFETIME_BOUND_TO_PROFILE_VISIBILITY = 64 +NM_ACTIVATION_STATE_FLAG_EXTERNAL = 128 +NM_SETTINGS_ADD_CONNECTION2_FLAG_NONE = 0 +NM_SETTINGS_ADD_CONNECTION2_FLAG_TO_DISK = 1 +NM_SETTINGS_ADD_CONNECTION2_FLAG_IN_MEMORY = 2 +NM_SETTINGS_ADD_CONNECTION2_FLAG_BLOCK_AUTOCONNECT = 32 +NM_SETTINGS_UPDATE2_FLAG_NONE = 0 +NM_SETTINGS_UPDATE2_FLAG_TO_DISK = 1 +NM_SETTINGS_UPDATE2_FLAG_IN_MEMORY = 2 +NM_SETTINGS_UPDATE2_FLAG_IN_MEMORY_DETACHED = 4 +NM_SETTINGS_UPDATE2_FLAG_IN_MEMORY_ONLY = 8 +NM_SETTINGS_UPDATE2_FLAG_VOLATILE = 16 +NM_SETTINGS_UPDATE2_FLAG_BLOCK_AUTOCONNECT = 32 +NM_SETTINGS_UPDATE2_FLAG_NO_REAPPLY = 64 +NM_TERNARY_DEFAULT = -1 +NM_TERNARY_FALSE = 0 +NM_TERNARY_TRUE = 1 +NM_MANAGER_RELOAD_FLAG_NONE = 0 +NM_MANAGER_RELOAD_FLAG_CONF = 1 +NM_MANAGER_RELOAD_FLAG_DNS_RC = 2 +NM_MANAGER_RELOAD_FLAG_DNS_FULL = 4 +NM_MANAGER_RELOAD_FLAG_ALL = 7 +NM_DEVICE_INTERFACE_FLAG_NONE = 0 +NM_DEVICE_INTERFACE_FLAG_UP = 1 +NM_DEVICE_INTERFACE_FLAG_LOWER_UP = 2 +NM_DEVICE_INTERFACE_FLAG_CARRIER = 65536 +NM_CLIENT_PERMISSION_NONE = 0 +NM_CLIENT_PERMISSION_ENABLE_DISABLE_NETWORK = 1 +NM_CLIENT_PERMISSION_ENABLE_DISABLE_WIFI = 2 +NM_CLIENT_PERMISSION_ENABLE_DISABLE_WWAN = 3 +NM_CLIENT_PERMISSION_ENABLE_DISABLE_WIMAX = 4 +NM_CLIENT_PERMISSION_SLEEP_WAKE = 5 +NM_CLIENT_PERMISSION_NETWORK_CONTROL = 6 +NM_CLIENT_PERMISSION_WIFI_SHARE_PROTECTED = 7 +NM_CLIENT_PERMISSION_WIFI_SHARE_OPEN = 8 +NM_CLIENT_PERMISSION_SETTINGS_MODIFY_SYSTEM = 9 +NM_CLIENT_PERMISSION_SETTINGS_MODIFY_OWN = 10 +NM_CLIENT_PERMISSION_SETTINGS_MODIFY_HOSTNAME = 11 +NM_CLIENT_PERMISSION_SETTINGS_MODIFY_GLOBAL_DNS = 12 +NM_CLIENT_PERMISSION_RELOAD = 13 +NM_CLIENT_PERMISSION_CHECKPOINT_ROLLBACK = 14 +NM_CLIENT_PERMISSION_ENABLE_DISABLE_STATISTICS = 15 +NM_CLIENT_PERMISSION_ENABLE_DISABLE_CONNECTIVITY_CHECK = 16 +NM_CLIENT_PERMISSION_WIFI_SCAN = 17 +NM_CLIENT_PERMISSION_LAST = 17 +NM_CLIENT_PERMISSION_RESULT_UNKNOWN = 0 +NM_CLIENT_PERMISSION_RESULT_YES = 1 +NM_CLIENT_PERMISSION_RESULT_AUTH = 2 +NM_CLIENT_PERMISSION_RESULT_NO = 3 NM_VPN_SERVICE_STATE_UNKNOWN = 0 NM_VPN_SERVICE_STATE_INIT = 1 NM_VPN_SERVICE_STATE_SHUTDOWN = 2 diff --git a/README b/README index 55ebddf..be21729 100644 --- a/README +++ b/README @@ -1,28 +1,10 @@ -python-networkmanager Easy communication with NetworkManager -============================================================ +python-networkmanager +===================== -python-networkmanager wraps NetworkManagers D-Bus interface so you can be less -verbose when talking to NetworkManager from python. All interfaces have been -wrapped in classes, properties are exposed as python properties and function -calls are forwarded to the correct interface. +⚠️ This project is no longer maintained. ⚠️ -See docs/index.rst for the documentation. An HTML version can be found on -http://packages.python.org/python-networkmanager/ +If you are looking for an alternative, please try +https://github.com/python-sdbus/python-sdbus-networkmanager -Requirements -============ -Python 2.5 or newer (Python 3 is supported as well) and the python D-Bus -bindings. - -Quick install instructions -========================== - -Stable version: - -$ pip install python-networkmanager - -Latest code: - -$ git clone http://github.com/seveas/python-networkmanager -$ cd python-networkmanager -$ python setup.py install +If python-sdbus-networkmanager does not fit your needs, and you wish to revive this project, please +contact me. diff --git a/ci/python-networkmanager.yml b/ci/python-networkmanager.yml new file mode 100644 index 0000000..f806955 --- /dev/null +++ b/ci/python-networkmanager.yml @@ -0,0 +1,252 @@ +--- +resource_types: + - name: pypi + type: docker-image + source: + repository: cfplatformeng/concourse-pypi-resource + + - name: dput + type: docker-image + source: + repository: seveas/concourse-dput-resource + + - name: ppa + type: docker-image + source: + repository: seveas/concourse-ppa-resource + +# - name: copr +# type: docker-image +# source: +# repository: seveas/concourse-copr-resource + +resources: + - name: git-master + type: git + source: + uri: https://github.com/seveas/python-networkmanager.git + branch: master + + - name: git-gh-pages + type: git + source: + uri: https://github.com/seveas/python-networkmanager.git + branch: gh-pages + username: ((github-username)) + password: ((github-password)) + + - name: git-releases + type: git + source: + uri: https://github.com/seveas/python-networkmanager.git + branch: master + tag_filter: '*' + + - name: git-debian + type: git + source: + uri: https://github.com/seveas/python-networkmanager.git + branch: debian + +# - name: git-rpm +# type: git +# source: +# uri: https://github.com/seveas/python-networkmanager.git +# branch: rpm + + - name: pypi + type: pypi + source: + name: python-networkmanager + username: ((pypi-username)) + password: ((pypi-password)) + + - name: dput + type: dput + source: + archive: ppa:dennis/python + + - name: ppa + type: ppa + source: + ppa: dennis/python + package: python-networkmanager + api_token: ((launchpad-token)) + +# - name: copr +# type: copr +# source: +# project: seveas/python-networkmanager +# package: python-networkmanager +# api_token: ((copr-token)) + +jobs: + - name: docs + plan: + - get: git-master + trigger: true + - get: git-gh-pages + - task: build + config: + platform: linux + image_resource: + type: docker-image + source: + repository: seveas/build-python + tag: latest + inputs: + - name: git-master + - name: git-gh-pages + outputs: + - name: git-gh-pages-out + run: + path: sh + args: + - -exc + - | + make -C git-master/docs html + git -C git-gh-pages checkout gh-pages + rsync -av --delete --filter 'protect .git' git-master/docs/_build/html/ git-gh-pages/ + touch git-gh-pages/.nojekyll + git -C git-gh-pages add -A + if ! git -C git-gh-pages diff --quiet --cached; then + git -C git-gh-pages commit -m "Automated update from $(git -C git-master rev-parse --short HEAD)" + fi + # We must make a copy because the put doesn't see our changes to the input + rsync -av --delete git-gh-pages/ git-gh-pages-out/ + params: + GIT_COMMITTER_NAME: Concourse CI + GIT_AUTHOR_NAME: Concourse CI + GIT_COMMITTER_EMAIL: concourse@localhost.invalid + GIT_AUTHOR_EMAIL: concourse@localhost.invalid + - put: git-gh-pages + params: + repository: git-gh-pages-out + + - name: tarball + plan: + - get: git-releases + trigger: true + - task: tarball + config: + platform: linux + image_resource: + type: docker-image + source: + repository: python + tag: latest + inputs: + - name: git-releases + outputs: + - name: tarball + run: + path: sh + args: + - -exc + - | + cd git-releases + python setup.py sdist -d ../tarball + - put: pypi + params: + glob: tarball/python-networkmanager*.tar.gz + + - name: ppa + plan: + - get: pypi + trigger: true + - get: git-debian + - task: build + config: + platform: linux + image_resource: + type: docker-image + source: + repository: seveas/build-python + tag: latest + inputs: + - name: git-debian + - name: pypi + outputs: + - name: sources + run: + path: sh + args: + - -exc + - | + perl -E 'say $ENV{GPG_KEY}' | gpg --import + cd pypi + version=$(cat version) + mv python-networkmanager-$version.tar.gz python-networkmanager_$version.orig.tar.gz + tar zxvf python-networkmanager_$version.orig.tar.gz + cd python-networkmanager-$version + mv ../../git-debian/debian . + debuild -d -S -si + cd .. + mv python-networkmanager_$version* ../sources + params: + GPG_KEY: ((gpg-key)) + - put: dput + params: + glob: sources/*.changes + + - name: ppa-porter + plan: + - get: ppa + trigger: true + - task: split + config: + platform: linux + image_resource: + type: docker-image + source: + repository: seveas/concourse-ppa-resource + tag: latest + inputs: + - name: ppa + outputs: + - name: sources + run: + path: sh + args: + - -exc + - | + perl -E 'say $ENV{GPG_KEY}' | gpg --import + ppa-split ppa ../sources + params: + GPG_KEY: ((gpg-key)) + API_TOKEN: ((launchpad-token)) + DEBFULLNAME: Dennis Kaarsemaker + DEBEMAIL: dennis@kaarsemaker.net + - put: dput + params: + glob: sources/*.changes + allow_noop: true + +# - name: copr +# plan: +# - get: pypi +# trigger: true +# - get: git-rpm +# - task: build +# config: +# platform: linux +# image_resource: +# type: docker-image +# source: +# repository: seveas/build-rpm-minimal +# tag: latest +# inputs: +# - name: git-rpm +# - name: pypi +# outputs: +# - name: srpm +# run: +# path: sh +# args: +# - -exc +# - | +# rpmbuild --define '_sourcedir pypi' --define '_srcrpmdir srpm' \ +# --define '_topdir /tmp' -bs git-rpm/python-networkmanager.spec +# - put: copr +# params: +# glob: srpm/python-networkmanager*.src.rpm diff --git a/docs/conf.py b/docs/conf.py index c096292..d68ba1d 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -12,6 +12,7 @@ # serve to show the default. import sys, os +import sphinx_rtd_theme as theme # If extensions (or modules to document with autodoc) are in another directory, # add these directories to sys.path here. If the directory is relative to the @@ -41,16 +42,16 @@ # General information about the project. project = u'python-networkmanager' -copyright = u'2011-2016, Dennis Kaarsemaker' +copyright = u'2011-2021, Dennis Kaarsemaker' # The version info for the project you're documenting, acts as replacement for # |version| and |release|, also used in various other places throughout the # built documents. # # The short X.Y version. -version = '1.1' +version = '2.2' # The full version, including alpha/beta/rc tags. -release = '1.1' +release = '2.2' # The language for content autogenerated by Sphinx. Refer to documentation # for a list of supported languages. @@ -91,7 +92,7 @@ # The theme to use for HTML and HTML Help pages. See the documentation for # a list of builtin themes. -html_theme = 'default' +html_theme = 'sphinx_rtd_theme' # Theme options are theme-specific and customize the look and feel of a theme # further. For a list of options available for each theme, see the @@ -99,7 +100,7 @@ #html_theme_options = {} # Add any paths that contain custom themes here, relative to this directory. -#html_theme_path = [] +html_theme_path = [theme.get_html_theme_path()] # The name for this set of Sphinx documents. If None, it defaults to # " v documentation". diff --git a/docs/index.rst b/docs/index.rst index e43f497..765c6aa 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -4,17 +4,15 @@ Python API to talk to NetworkManager NetworkManager provides a detailed and capable D-Bus interface on the system bus. You can use this interface to query NetworkManager about the overall state of the network and details of network devices like current IP addresses or DHCP -options, and to activate and deactivate network connections. +options, and to configure, activate and deactivate network connections. python-networkmanager takes this D-Bus interface and wraps D-Bus interfaces in -classes and D-Bus properties in python properties. It also provides a -command-line utility to inspect the configuration and (de-)activate -connections. +classes and D-Bus methods and properties in their python equivalents. The :mod:`NetworkManager` module -------------------------------- .. module:: NetworkManager - :platform: Linux systems with NetworkManager 0.9 and 1.0 + :platform: Linux systems with NetworkManager 0.9 and newer :synopsis: Talk to NetworkManager from python All the code is contained in one module: :mod:`NetworkManager`. Using it is as @@ -24,37 +22,12 @@ simple as you think it is: >>> import NetworkManager >>> NetworkManager.NetworkManager.Version - '1.0.4' + '1.2.0' NetworkManager exposes a lot of information via D-Bus and also allows full -control of network settings. The full D-Bus API can be found on `NetworkManager -project website`_. All interfaces listed there have been wrapped in classes as -listed below. With a few exceptions, they behave exactly like the D-Bus -methods. These exceptions are for convenience and limited to this list: - -* IP addresses are returned as strings of the form :data:`1.2.3.4` instead of - network byte ordered integers. -* Route metrics are returned in host byte order, so you can use them as - integers. -* Mac addresses and BSSIDs are always returned as strings of the form - :data:`00:11:22:33:44:55` instead of byte sequences. -* Wireless SSID's are returned as strings instead of byte sequences. They will - be decoded as UTF-8 data, so using any other encoding for your SSID will - result in errors. - -.. class:: NMDbusInterface - -This is the base class for all interface wrappers. It adds a few useful -features to standard D-Bus interfaces: - -* All D-Bus properties are exposed as python properties -* Return values are automatically converted to python basic types (so no more - :data:`dbus.String`, but simple :data:`str` (python 3) or :data:`unicode` - (python 2)) -* Object paths in return values are automatically replaced with proxy objects, - so you don't need to do that manually -* ...and vice versa when sending -* And also when receiving signals +control of network settings. The full D-Bus interface can be found on +`NetworkManager project website`_. All interfaces listed there have been +wrapped in classes as listed below. .. function:: const(prefix, value) @@ -71,101 +44,171 @@ translate them to text. For example: >>> NetworkManager.const('device_type', 2) 'wifi' -.. _`NetworkManager project website`: https://developer.gnome.org/NetworkManager/0.9/spec.html +.. _`NetworkManager project website`: https://developer.gnome.org/NetworkManager/1.2/spec.html List of classes --------------- +.. class:: ObjectVanished -.. class:: NetworkManager +This Exception will be raised when you try to call a method or access a +property on a dbus object that no longer exists. Objects can go missing if +devices are removed, connections are disabled or NetworkManager is restarted. -This class represents the :data:`org.freedesktop.NetworkManager` interface. -Note that :data:`NetworkManager.NetworkManager` actually is the singleton -instance of this class. - -.. class:: Settings - -This class represents the :data:`org.freedesktop.NetworkManager.Settings` -interface. Note that :data:`NetworkManager.Settings` actually is the singleton -instance of this class. - -.. class:: Connection - -This class represents the -:data:`org.freedesktop.NetworkManager.Settings.Connection` interface. +.. class:: NMDbusInterface -.. class:: ActiveConnection +This is the base class of all classes below. It handles the marshalling of data +and the automatic creation of properties and methods. -This class represents the -:data:`org.freedesktop.NetworkManager.Connection.Active` interface. +Each property, method and signal exposed via the D-Bus interface is +automatically mirrored as an attribute of the actual classes. Moreover, the +data is made slightly more usable by performing the following transformations +on received and sent data. -.. class:: AccessPoint - -This class represents the :data:`org.freedesktop.NetworkManager.AccessPoint` -interface. +* IP addresses are returned as strings of the form :data:`1.2.3.4` instead of + network byte ordered integers. +* Route metrics are returned in host byte order, so you can use them as + integers. +* Mac addresses and BSSIDs are always returned as strings of the form + :data:`00:11:22:33:44:55` instead of byte sequences. +* Wireless SSID's are returned as strings instead of byte sequences. They will + be decoded as UTF-8 data, so using any other encoding for your SSID will + result in errors. +* DHCP options are turned into integers or booleans as appropriate +* Signals can be connected to using calls to On\ *SignalName* functions. -.. class:: Device +Here's a short example to illustrate: -.. class:: Wired + >>> import NetworkManager + >>> NetworkManager.NetworkManager.Version + '1.4.4' + >>> NetworkManager.NetworkManager.GetPermissions() + {'org.freedesktop.NetworkManager.checkpoint-rollback': 'auth', + 'org.freedesktop.NetworkManager.enable-disable-network': 'yes', + ...} + # Must have a mainloop to use signals + >>> import dbus.mainloop.glib + >>> dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + >>> NetworkManager.NetworkManager.OnStateChanged(handle_state_change) -.. class:: Wireless +.. class:: TransientNMDbusInterface -.. class:: Modem +Subclasses of this class, which are ActiveConnection, NSP, IP[46]Config and +DHCP[46]Config never survive a NetworkManager restart. Other objects may +survive a restart, but get a different object path. -.. class:: Bluetooth +.. class:: NetworkManager -.. class:: OlpcMesh +The main `NetworkManager +`_ +object; the `NetworkManager.Networkmanager` object is actually the singleton +instance of this class. -.. class:: Wimax +.. class:: Settings -.. class:: Infiniband +The `Settings +`_ +object, which can be used to add connections, list connections or update your +hostname; the `NetworkManager.Settings` object is actually the singleton +instance of this class. -.. class:: Bond +.. class:: AgentManager -.. class:: Bridge +The `AgentManager +`_ +object, whose methods you'll need to call when implementing a secrets agent; +the `NetworkManager.AgentManager` object is actually the singleton instance of +this class. -.. class:: Vlan +.. class:: Connection -.. class:: Adsl +`Connection +`_ +objects represent network configurations configured by the user. -.. class:: Generic +.. class:: ActiveConnection +.. class:: VPNConnection -These classes represent D-Bus interfaces for various types of hardware. Note -that methods such as :data:`NetworkManager.GetDevices()` will only return -:class:`Device` instances. To get the hardware-specific class, you can call the -:func:`Device.SpecificDevice` method. +Active connections are represented by `ActiveConnection +`_ +objects. `VPNConnection +`_ +is a subclass for active VPN connection that implements both the +Connection.Active and VPN.Connection interfaces. -.. code-block:: py +.. class:: IP4Config +.. class:: IP6Config +.. class:: DHCP4Config +.. class:: DHCP6Config - >>> [(dev.Interface, dev.SpecificDevice().__class__.__name__) - ... for dev in NetworkManager.NetworkManager.GetDevices()] - [('eth0', 'Wired'), ('wlan0', 'Wireless'), ('wwan0', 'Modem')] +Active network connections and devices can all have `IPv4 +`_, +`IPv6 +`_, +`IPv4 DHCP +`_ +and `IPv6 DHCP +`_ +information attached to them, which is represented by instances of these +classes. -.. class:: IP4Config +.. class:: AccessPoint -.. class:: IP6Config +Wifi `Accesspoints +`_, +as visibly by any 802.11 wifi interface. -.. class:: DHCP4Config +.. class:: NSP -.. class:: DHCP6Config +Wimax `Network Service Providers `_. -These classes represent the various IP configuration interfaces. +.. class:: Device -.. class:: AgentManager +All device classes implement the `Device +`_ +interface, which gives you access to basic device properties. Note that you will never see instances of this class, only of its devicetype-specific subclasses which impletemnt not only the Device interface but also their own specific interface. Supported device types are +`Adsl `_, +`Bluetooth `_, +`Bond `_, +`Bridge `_, +`Generic `_, +`Infiniband `_, +`IPTunnel `_, +`Macvlan `_, +`Modem `_, +`OlpcMesh `_, +`Team `_, +`Tun `_, +`Veth `_, +`Vlan `_, +`Vxlan `_, +`Wimax `_, +`Wired `_ and +`Wireless `_ .. class:: SecretAgent -Classes that can be used to handle and store secrets. Note that these are not -for querying NetworkManager's exisiting secret stores. For that the -:func:`GetSecrets` method of the :class:`Connection` class can be used. +The NetworkManager daemon can ask separate programs, called agents, for secrets +if it needs them. The NetworkManager applet and the `nmcli` command-line tool +are examples of such agents. You can also write such agents by subclassing the +`SecretAgent` class and providing a `GetSecrets` method as in the following +example, which returns a static password for each secret:: -.. class:: VPNConnection + import dbus.mainloop.glib + import GObject + import NetworkManager -This class represents the :data:`org.freedesktop.NetworkManager.VPN.Connection` -interface. + class MyAgent(NetworkManager.SecretAgent): + def GetSecrets(self, settings, connection, setting_name, hints, flags): + return {setting_name: {'secrets': {'password': 'TopSecret!'}}} -.. class:: VPNPlugin + agent = MyAgent() + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + Gobject.MainLoop().run() -A class that can be used to query VPN plugins. +Beware that NetworkManager will ask each agent in turn in what is in essence +random order. Except it will prioritize the program that activated the +connection. So if you want to make sure your agent is called first, activate +the connection from the same application. .. toctree:: :maxdepth: 2 diff --git a/examples/connection_detail.py b/examples/connection_detail.py index 43b0491..30ea5e7 100644 --- a/examples/connection_detail.py +++ b/examples/connection_detail.py @@ -20,7 +20,7 @@ if conn.Devices: devices = " (on %s)" % ", ".join([x.Interface for x in conn.Devices]) print("Active connection: %s%s" % (settings['connection']['id'], devices)) - size = max([max([len(y) for y in x.keys() + ['']]) for x in settings.values()]) + size = max([max([len(y) for y in list(x.keys()) + ['']]) for x in settings.values()]) format = " %%-%ds %%s" % (size + 5) for key, val in sorted(settings.items()): print(" %s" % key) @@ -30,9 +30,8 @@ print("Device: %s" % dev.Interface) print(" Type %s" % c('device_type', dev.DeviceType)) # print(" IPv4 address %s" % socket.inet_ntoa(struct.pack('L', dev.Ip4Address))) - devicedetail = dev.SpecificDevice() - if not callable(devicedetail.HwAddress): - print(" MAC address %s" % devicedetail.HwAddress) + if hasattr(dev, 'HwAddress'): + print(" MAC address %s" % dev.HwAddress) print(" IPv4 config") print(" Addresses") for addr in dev.Ip4Config.Addresses: diff --git a/examples/listener.py b/examples/listener.py index 3e866ea..1cc787d 100644 --- a/examples/listener.py +++ b/examples/listener.py @@ -2,31 +2,32 @@ Listen to some available signals from NetworkManager """ -import dbus.mainloop.glib; dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) +import dbus.mainloop.glib from gi.repository import GObject import NetworkManager +import time -d_args = ('sender', 'destination', 'interface', 'member', 'path') -d_args = dict([(x + '_keyword', 'd_' + x) for x in d_args]) +def out(msg): + print("%s %s" % (time.strftime('%H:%M:%S'), msg)) -def main(): - NetworkManager.NetworkManager.connect_to_signal('CheckPermissions', display_sig, **d_args) - NetworkManager.NetworkManager.connect_to_signal('StateChanged', display_sig, **d_args) - NetworkManager.NetworkManager.connect_to_signal('PropertiesChanged', display_sig, **d_args) - NetworkManager.NetworkManager.connect_to_signal('DeviceAdded', display_sig, **d_args) - NetworkManager.NetworkManager.connect_to_signal('DeviceRemoved', display_sig, **d_args) +def statechange(nm, interface, signal, state): + out("State changed to %s" % NetworkManager.const('STATE', state)) + +def adddevice(nm, interface, signal, device_path): + try: + out("Device %s added" % device_path.IpInterface) + except NetworkManager.ObjectVanished: + # Sometimes this signal is sent for *removed* devices. Ignore. + pass - print("Waiting for signals") - print("-------------------") +def main(): + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + NetworkManager.NetworkManager.OnStateChanged(statechange) + NetworkManager.NetworkManager.OnDeviceAdded(adddevice) + out("Waiting for signals") loop = GObject.MainLoop() loop.run() -def display_sig(*args, **kwargs): - print("Received signal: %s.%s" % (kwargs['d_interface'], kwargs['d_member'])) - print("Sender: (%s)%s" % (kwargs['d_sender'], kwargs['d_path'])) - print("Arguments: (%s)" % ", ".join([str(x) for x in args])) - print("-------------------") - if __name__ == '__main__': main() diff --git a/examples/n-m b/examples/n-m index 9bde772..46f02e1 100755 --- a/examples/n-m +++ b/examples/n-m @@ -3,7 +3,7 @@ # Command-line tool to interact with NetworkManager. With this tool, you can # inspect various configuration items and (de-)activate connections. # -# (C) 2011-2016 Dennis Kaarsemaker +# (C) 2011-2021 Dennis Kaarsemaker # License: zlib from __future__ import print_function diff --git a/examples/otp-agent b/examples/otp-agent new file mode 100755 index 0000000..7e7cd80 --- /dev/null +++ b/examples/otp-agent @@ -0,0 +1,71 @@ +#!/usr/bin/python3 +# +# Automate vpn connections that require a one-time password. +# Requirements: +# - secretstorage (find on pypi) +# - pyotp (likewise) +# +# usage: ./otp-agent name-of-connection +# +# The connection will be activated and when networkmanager asks for a secret, +# it will be provided. If the secret isn't known yet, it will be asked and +# stored with the secretstorage api (so in e.g. your gnome keyring) + +import dbus.mainloop.glib +from gi.repository import GObject +import NetworkManager +import pyotp +import traceback +import secretstorage +import sys + +def main(): + print("Connecting to %s" % sys.argv[1]) + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + loop = GObject.MainLoop() + agent = SecretAgent(loop) + for connection in NetworkManager.Settings.ListConnections(): + settings = connection.GetSettings()['connection'] + if settings['id'] == sys.argv[1]: + NetworkManager.NetworkManager.ActivateConnection(connection, "/", "/") + loop.run() + break + else: + print("Connection %s not found" % sys.argv[1]) + +class SecretAgent(NetworkManager.SecretAgent): + def __init__(self, loop): + self.loop = loop + self.collection = secretstorage.get_default_collection(secretstorage.dbus_init()) + super(SecretAgent, self).__init__('net.seveas.otp-agent') + + def GetSecrets(self, settings, connection, setting_name, hints, flags): + try: + print("NetworkManager is asking us for a secret") + if setting_name != 'vpn': + return {} + attrs = { + 'xdg:schema': 'net.seveas.otp-agent', + 'hostname': settings['vpn']['data']['remote'], + } + items = list(self.collection.search_items(attrs)) + if not items: + print("No secrets found yet, asking user") + secret = input("Enter secret code for %s: " % settings['vpn']['data']['remote']) + self.collection.create_item(settings['vpn']['data']['remote'], attrs, secret) + items = list(self.collection.search_items(attrs)) + else: + print("Found secret key, generating otp code") + secret = items[0].get_secret().decode('ascii') + otp = pyotp.TOTP(secret).now() + print("otp code: %s" % otp) + return {setting_name: {'secrets': {'password': otp}}} + except: + import traceback + traceback.print_exc() + return {} + finally: + self.loop.quit() + +if __name__ == '__main__': + main() diff --git a/examples/ssids.py b/examples/ssids.py index 1140508..a6bf694 100644 --- a/examples/ssids.py +++ b/examples/ssids.py @@ -7,5 +7,5 @@ for dev in NetworkManager.NetworkManager.GetDevices(): if dev.DeviceType != NetworkManager.NM_DEVICE_TYPE_WIFI: continue - for ap in dev.SpecificDevice().GetAccessPoints(): - print '%-30s %dMHz %d%%' % (ap.Ssid, ap.Frequency, ap.Strength) + for ap in dev.GetAccessPoints(): + print('%-30s %dMHz %d%%' % (ap.Ssid, ap.Frequency, ap.Strength)) diff --git a/examples/wifi_monitor.py b/examples/wifi_monitor.py new file mode 100644 index 0000000..bfa5dc0 --- /dev/null +++ b/examples/wifi_monitor.py @@ -0,0 +1,42 @@ +""" +Show and monitor available access points +""" +from gi.repository import GObject +import dbus.mainloop.glib +import NetworkManager + +# Cache the ssids, as the SSid property will be unavailable when an AP +# disappears +ssids = {} + +def main(): + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + # Listen for added and removed access points + for dev in NetworkManager.Device.all(): + if dev.DeviceType == NetworkManager.NM_DEVICE_TYPE_WIFI: + dev.OnAccessPointAdded(ap_added) + dev.OnAccessPointRemoved(ap_removed) + for ap in NetworkManager.AccessPoint.all(): + try: + ssids[ap.object_path] = ap.Ssid + print("* %-30s %s %sMHz %s%%" % (ap.Ssid, ap.HwAddress, ap.Frequency, ap.Strength)) + ap.OnPropertiesChanged(ap_propchange) + except NetworkManager.ObjectVanished: + pass + GObject.MainLoop().run() + +def ap_added(dev, interface, signal, access_point): + ssids[access_point.object_path] = access_point.Ssid + print("+ %-30s %s %sMHz %s%%" % (access_point.Ssid, access_point.HwAddress, access_point.Frequency, access_point.Strength)) + access_point.OnPropertiesChanged(ap_propchange) + +def ap_removed(dev, interface, signal, access_point): + print("- %-30s" % ssids.pop(access_point.object_path)) + +def ap_propchange(ap, interface, signal, properties): + if 'Strength' in properties: + print(" %-30s %s %sMHz %s%%" % (ap.Ssid, ap.HwAddress, ap.Frequency, properties['Strength'])) + + +if __name__ == '__main__': + main() diff --git a/makeconstants.py b/makeconstants.py index ca8f654..9408e5d 100644 --- a/makeconstants.py +++ b/makeconstants.py @@ -5,9 +5,10 @@ enum_regex = re.compile(r'typedef enum(?:\s+[a-zA-Z]+)?\s*\{(.*?)\}', re.DOTALL) comment_regex = re.compile(r'/\*.*?\*/', re.DOTALL) -headers = ['/usr/include/NetworkManager/NetworkManager.h', +headers = [ '/usr/include/libnm/nm-dbus-interface.h', '/usr/include/NetworkManager/NetworkManagerVPN.h', '/usr/include/libnm-glib/nm-secret-agent.h'] + for h in headers: for enum in enum_regex.findall(open(h).read()): enum = comment_regex.sub('', enum) @@ -17,7 +18,7 @@ continue if '=' in key: key, val = key.split('=') - val = eval(val) + val = eval(val.replace('LL','')) else: val = last + 1 key = key.strip() diff --git a/setup.py b/setup.py index 4101936..b1d897a 100755 --- a/setup.py +++ b/setup.py @@ -1,14 +1,15 @@ #!/usr/bin/python -from distutils.core import setup +from setuptools import setup setup(name = "python-networkmanager", - version = "1.1", + version = "2.2", author = "Dennis Kaarsemaker", author_email = "dennis@kaarsemaker.net", url = "http://github.com/seveas/python-networkmanager", description = "Easy communication with NetworkManager", py_modules = ["NetworkManager"], + install_requires = ["dbus-python", "six"], classifiers = [ 'Development Status :: 5 - Production/Stable', 'Intended Audience :: Developers', diff --git a/test/README b/test/README new file mode 100644 index 0000000..9218f82 --- /dev/null +++ b/test/README @@ -0,0 +1,17 @@ +WARNING: THE TESTS MESS WITH YOUR SETTINGS + +Some of the things the testsuite does: + +- Repeatedly kill all network connections +- Delete, modify and add network connections +- Change your hostname + +Think twice before you want to run this testsuire. And read the tests to see +what they do. + +The following tests are safe to run, as they only read data: + +test_accesspoint.py +test_activeconnection.py +test_devices.py +test_ipconfig.py diff --git a/test/__init__.py b/test/__init__.py new file mode 100644 index 0000000..70b8a44 --- /dev/null +++ b/test/__init__.py @@ -0,0 +1,59 @@ +# Set this variable in your environment if you understand that +# +# THE TESTS WILL MESS WITH YOUR COMPUTER! +# +# You will repeatedly go offline and it will attempt to add and delete +# connections, change the hostname and more nasty things. + +import os +if 'NM_TESTS' not in os.environ: + print("Cowardly refusing to run tests") + os._exit(1) + +import dbus +import ipaddress +import six +import time +import unittest + +import NetworkManager + +class TestCase(unittest.TestCase): + def assertRaisesDBus(self, exception, func, *args, **kwargs): + if '.' not in exception: + exception = 'org.freedesktop.NetworkManager.' + exception + with self.assertRaises(dbus.exceptions.DBusException) as cm: + func(*args, **kwargs) + self.assertEqual(cm.exception.get_dbus_name(), exception) + + def assertIsStrictSubclass(self, klass1, klass2): + self.assertTrue(issubclass(klass1, klass2) and klass1 != klass2) + + def assertIsIpAddress(self, ip): + try: + ipaddress.ip_address(six.text_type(ip)) + except ValueError as e: + raise self.failureException(str(e)) + + def assertIsIpNetwork(self, ip, prefix): + try: + ipaddress.ip_network((ip, prefix), strict=False) + except ValueError as e: + raise self.failureException(str(e)) + + def assertIsMacAddress(self, address): + self.assertRegex(address, '^[0-9a-fA-F]{2}(?::[0-9a-fA-F]{2}){5}$', '%s is not a mac address' % address) + + def waitForConnection(self): + while NetworkManager.NetworkManager.State < NetworkManager.NM_STATE_CONNECTED_LOCAL: + time.sleep(0.5) + + def waitForDisconnection(self): + while NetworkManager.NetworkManager.State >= NetworkManager.NM_STATE_CONNECTED_LOCAL: + time.sleep(0.5) + +permissions = NetworkManager.NetworkManager.GetPermissions() +def have_permission(permission): + permission = 'org.freedesktop.NetworkManager.' + permission + return permissions.get(permission, None) == 'yes' + diff --git a/test/test_accesspoint.py b/test/test_accesspoint.py new file mode 100644 index 0000000..1bc538b --- /dev/null +++ b/test/test_accesspoint.py @@ -0,0 +1,26 @@ +from test import * + +class AccessPointTest(TestCase): + def test_accesspoints(self): + for dev in NetworkManager.NetworkManager.Devices: + if isinstance(dev, NetworkManager.Wireless): + for ap in dev.AccessPoints: + self.assertIsInstance(ap.Flags, int) + # Frequencies from https://en.wikipedia.org/wiki/List_of_WLAN_channels + f = ap.Frequency + if not ( + (f > 2400 and f < 2500) or + (f > 3650 and f < 3700) or + (f > 4900 and f < 6000)): + self.fail("Frequency is not a valid wifi frequency") + self.assertIsMacAddress(ap.HwAddress) + self.assertIsInstance(ap.LastSeen, int) + self.assertIsInstance(ap.MaxBitrate, int) + self.assertIsInstance(ap.WpaFlags, int) + self.assertIsInstance(ap.RsnFlags, int) + self.assertLess(ap.Strength, 100) + self.assertIsInstance(ap.Ssid, six.text_type) + self.assertIn(ap.Mode, (NetworkManager.NM_802_11_MODE_ADHOC, NetworkManager.NM_802_11_MODE_INFRA, NetworkManager.NM_802_11_MODE_AP)) + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_activeconnection.py b/test/test_activeconnection.py new file mode 100644 index 0000000..1847023 --- /dev/null +++ b/test/test_activeconnection.py @@ -0,0 +1,25 @@ +from test import * + +class ActiveConnectionTest(TestCase): + def test_properties(self): + for conn in NetworkManager.NetworkManager.ActiveConnections: + self.assertIsInstance(conn.Connection, NetworkManager.Connection) + for device in conn.Devices: + self.assertIsInstance(device, NetworkManager.Device) + if conn.Connection.GetSettings()['connection']['type'] == '802-11-wireless': + self.assertIsInstance(conn.SpecificObject, NetworkManager.AccessPoint) + if conn.Vpn: + self.assertIsInstance(conn, NetworkManager.VPNConnection) + self.assertIsInstance(conn.Banner, six.text_type) + self.assertIsInstance(conn.SpecificObject, NetworkManager.ActiveConnection) + self.assertTrue(conn.State == NetworkManager.NM_ACTIVE_CONNECTION_STATE_ACTIVATED) + self.assertIsInstance(conn.Ip4Config, NetworkManager.IP4Config) + self.assertIsInstance(conn.Ip6Config, (NetworkManager.IP6Config, type(None))) + self.assertIsInstance(conn.Dhcp4Config, (NetworkManager.DHCP4Config, type(None))) + self.assertIsInstance(conn.Dhcp6Config, (NetworkManager.DHCP6Config, type(None))) + if conn.Master != None: + self.assertIsInstance(conn.Master, NetworkManager.Device) + self.assertEqual(conn.Master, conn.SpecificObject.Devices[0]) + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_agentmanager.py b/test/test_agentmanager.py new file mode 100644 index 0000000..7cf0a9b --- /dev/null +++ b/test/test_agentmanager.py @@ -0,0 +1,9 @@ +from test import * + +class AgentManagerTest(TestCase): + def test_registration(self): + NetworkManager.AgentManager.Register('python-network-manager-test') + NetworkManager.AgentManager.Unregister() + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_connection.py b/test/test_connection.py new file mode 100644 index 0000000..c2b95fd --- /dev/null +++ b/test/test_connection.py @@ -0,0 +1,63 @@ +from test import * + +class ConnectionTest(TestCase): + def test_settings(self): + for connection in NetworkManager.Settings.ListConnections(): + settings = connection.GetSettings() + self.assertIn(settings['connection']['type'], settings) + + secrets = connection.GetSecrets() + for key in settings: + self.assertIn(key, secrets) + + if 'ipv4' in settings: + for address, prefix, gateway in settings['ipv4']['addresses']: + self.assertIsIpAddress(address) + self.assertIsIpAddress(gateway) + if 'ipv6' in settings: + for address, prefix, gateway in settings['ipv6']['addresses']: + self.assertIsIpAddress(address) + self.assertIsIpAddress(gateway) + + def test_update(self): + active = [x.Connection for x in NetworkManager.NetworkManager.ActiveConnections] + for connection in NetworkManager.Settings.Connections: + if connection in active: + continue + settings = connection.GetSettings() + connection.Update(settings) + # FIXME: this causes assertion failures in n-m, which cause the dbus call to hang + #settings['connection']['timestamp'] -= 1 + #connection.UpdateUnsaved(settings) + #self.assertTrue(connection.Unsaved) + #print("Saving") + #connection.Save() + #print("Saved") + #self.assertFalse(connection.Unsaved) + break + + def test_secrets(self): + active = [x.Connection for x in NetworkManager.NetworkManager.ActiveConnections] + key = '802-11-wireless-security' + for connection in NetworkManager.Settings.Connections: + if connection in active: + continue + settings = connection.GetSettings() + if key not in settings: + continue + secrets = connection.GetSecrets() + if not secrets[key]: + continue + settings[key].update(secrets[key]) + + connection.ClearSecrets() + secrets = connection.GetSecrets() + self.assertEqual(secrets[key], {}) + + connection.Update(settings) + secrets = connection.GetSecrets() + self.assertNotEqual(secrets[key], {}) + break + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_connection_addremove.py b/test/test_connection_addremove.py new file mode 100644 index 0000000..54e2bd1 --- /dev/null +++ b/test/test_connection_addremove.py @@ -0,0 +1,53 @@ +from test import * + +class ConnectionAddRemoveTest(TestCase): + + def test_activate(self): + active = NetworkManager.NetworkManager.ActiveConnections[0] + ap = active.SpecificObject + conn = active.Connection + dev = active.Devices[0] + + NetworkManager.NetworkManager.DeactivateConnection(active) + self.waitForDisconnection() + NetworkManager.NetworkManager.ActivateConnection(conn, dev, ap) + self.waitForConnection() + + def test_delete_addactivate(self): + active = NetworkManager.NetworkManager.ActiveConnections[0] + ap = active.SpecificObject + conn = active.Connection + dev = active.Devices[0] + settings = conn.GetSettings() + typ = settings['connection']['type'] + if 'security' in settings[typ]: + key2 = settings[typ]['security'] + settings[key2].update(conn.GetSecrets(key2)[key2]) + + conn.Delete() + self.waitForDisconnection() + conn, active = NetworkManager.NetworkManager.AddAndActivateConnection(settings, dev, ap) + self.assertIsInstance(conn, NetworkManager.Connection) + self.assertIsInstance(active, NetworkManager.ActiveConnection) + self.waitForConnection() + + def test_delete_add_activate(self): + active = NetworkManager.NetworkManager.ActiveConnections[0] + ap = active.SpecificObject + conn = active.Connection + dev = active.Devices[0] + settings = conn.GetSettings() + typ = settings['connection']['type'] + if 'security' in settings[typ]: + key2 = settings[typ]['security'] + settings[key2].update(conn.GetSecrets(key2)[key2]) + + conn.Delete() + self.waitForDisconnection() + conn = NetworkManager.Settings.AddConnection(settings) + self.assertIsInstance(conn, NetworkManager.Connection) + NetworkManager.NetworkManager.ActivateConnection(conn, dev, ap) + self.waitForConnection() + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_devices.py b/test/test_devices.py new file mode 100644 index 0000000..77d2e8e --- /dev/null +++ b/test/test_devices.py @@ -0,0 +1,37 @@ +from test import * + +class DeviceTest(TestCase): + def test_devices(self): + for device in NetworkManager.NetworkManager.Devices: + self.assertIsStrictSubclass(type(device), NetworkManager.Device) + if device.Dhcp4Config: + self.assertIsInstance(device.Dhcp4Config, NetworkManager.DHCP4Config) + if device.Dhcp6Config: + self.assertIsInstance(device.Dhcp6Config, NetworkManager.DHCP6Config) + if device.Ip4Config: + self.assertIsInstance(device.Ip4Config, NetworkManager.IP4Config) + if device.Ip6Config: + self.assertIsInstance(device.Ip6Config, NetworkManager.IP6Config) + if device.Ip4Address: + self.assertIsIpAddress(device.Ip4Address) + if hasattr(device, 'HwAddress') and device.HwAddress: + self.assertIsMacAddress(device.HwAddress) + if hasattr(device, 'PermHwAddress') and device.PermHwAddress: + self.assertIsMacAddress(device.PermHwAddress) + if device.DeviceType == NetworkManager.NM_DEVICE_TYPE_WIFI: + for ap in device.AccessPoints: + self.assertIsInstance(ap, NetworkManager.AccessPoint) + device.RequestScan({}) + elif device.DeviceType == NetworkManager.NM_DEVICE_TYPE_ETHERNET: + self.assertIn(device.Carrier, (True, False)) + elif device.DeviceType == NetworkManager.NM_DEVICE_TYPE_GENERIC: + self.assertIsInstance(device.TypeDescription, six.text_type) + elif device.DeviceType == NetworkManager.NM_DEVICE_TYPE_TUN: + if device.Owner != -1: + import pwd + pwd.getpwuid(device.Owner) + else: + self.fail("I don't know how to test %s devices" % type(device).__name__) + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_ipconfig.py b/test/test_ipconfig.py new file mode 100644 index 0000000..889a0a6 --- /dev/null +++ b/test/test_ipconfig.py @@ -0,0 +1,55 @@ +from test import * + +class IpConfigTest(TestCase): + def test_configs(self): + for device in NetworkManager.NetworkManager.Devices: + if device.State != NetworkManager.NM_DEVICE_STATE_ACTIVATED: + continue + self.do_check(device) + for connection in NetworkManager.NetworkManager.ActiveConnections: + self.do_check(connection) + + def do_check(self, thing): + if thing.Dhcp4Config: + self.assertIsInstance(thing.Dhcp4Config, NetworkManager.DHCP4Config) + self.assertIsInstance(thing.Dhcp4Config.Options, dict) + o = thing.Dhcp4Config.Options + self.assertIsInstance(o['domain_name_servers'], list) + self.assertIsInstance(o['ntp_servers'], list) + self.assertIsIpAddress(o['ip_address']) + for key in o: + if key.endswith('_requested'): + self.assertTrue(o[key]) + if thing.Dhcp6Config: + self.assertIsInstance(thing.Dhcp6Config, NetworkManager.DHCP6Config) + self.assertIsInstance(thing.Dhcp6Config.Options, dict) + for c in (thing.Ip4Config, thing.Ip6Config): + if not c: + continue + for addr, prefix, gateway in c.Addresses: + self.assertIsIpAddress(addr) + self.assertIsIpAddress(gateway) + self.assertIsIpNetwork(addr, prefix) + for data in c.AddressData: + self.assertIsIpAddress(data['address']) + self.assertIsIpNetwork(data['address'], data['prefix']) + if 'peer' in data: + self.assertIsIpAddress(data['peer']) + if c.Gateway: + self.assertIsIpAddress(c.Gateway) + for addr in c.Nameservers: + self.assertIsIpAddress(addr) + for addr in getattr(c, 'WinsServers', []): + self.assertIsIpAddress(addr) + for dest, prefix, next_hop, metric in c.Routes: + self.assertIsIpNetwork(dest, prefix) + self.assertIsIpAddress(next_hop) + self.assertLessEqual(metric, 1000) + for data in c.RouteData: + self.assertIsIpNetwork(data['dest'], data['prefix']) + if 'next-hop' in data: + self.assertIsIpAddress(data['next-hop']) + self.assertLessEqual(data['metric'], 1000) + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_networkmanager.py b/test/test_networkmanager.py new file mode 100644 index 0000000..1760d8d --- /dev/null +++ b/test/test_networkmanager.py @@ -0,0 +1,64 @@ +from test import * + +class NetworkManagerTest(TestCase): + def test_properties(self): + self.assertIsInstance(NetworkManager.NetworkManager.NetworkingEnabled, bool) + self.assertIsInstance(NetworkManager.NetworkManager.Metered, int) + self.assertIsInstance(NetworkManager.NetworkManager.Version, six.string_types) + self.assertIsInstance(NetworkManager.NetworkManager.ActiveConnections, list) + for conn in NetworkManager.NetworkManager.ActiveConnections: + self.assertIsInstance(conn, NetworkManager.ActiveConnection) + self.assertIsInstance(NetworkManager.NetworkManager.Devices, list) + for dev in NetworkManager.NetworkManager.Devices: + self.assertIsInstance(dev, NetworkManager.Device) + self.assertIsInstance(NetworkManager.NetworkManager.PrimaryConnection, NetworkManager.ActiveConnection) + + @unittest.skipUnless(have_permission('sleep-wake'), "Not allowed to make networkmanager sleep") + def test_sleep(self): + NetworkManager.NetworkManager.Sleep(True) + self.assertRaisesDBus('AlreadyAsleepOrAwake', NetworkManager.NetworkManager.Sleep, True) + NetworkManager.NetworkManager.Sleep(False) + self.assertRaisesDBus('AlreadyAsleepOrAwake', NetworkManager.NetworkManager.Sleep, False) + self.waitForConnection() + + def test_enable(self): + NetworkManager.NetworkManager.Enable(False) + self.assertRaisesDBus('AlreadyEnabledOrDisabled', NetworkManager.NetworkManager.Enable, False) + NetworkManager.NetworkManager.Enable(True) + self.assertRaisesDBus('AlreadyEnabledOrDisabled', NetworkManager.NetworkManager.Enable, True) + self.waitForConnection() + + @unittest.skipUnless(os.getuid() == 0, "Must be root to modify logging") + def test_logging(self): + level1, domains = NetworkManager.NetworkManager.GetLogging() + self.assertIn(level1, ['ERR', 'WARN', 'INFO', 'DEBUG', 'TRACE']) + self.assertIn('PLATFORM', domains) + NetworkManager.NetworkManager.SetLogging("KEEP", "PLATFORM:DEBUG") + level2, domains = NetworkManager.NetworkManager.GetLogging() + self.assertEqual(level1, level2) + self.assertIn('PLATFORM:DEBUG', domains) + self.assertIn('CORE', domains) + NetworkManager.NetworkManager.SetLogging("KEEP", "PLATFORM:" + level1) + level2, domains = NetworkManager.NetworkManager.GetLogging() + self.assertIn('PLATFORM', domains) + self.assertNotIn('PLATFORM:DEBUG', domains) + + def test_permissions(self): + permissions = NetworkManager.NetworkManager.GetPermissions() + self.assertIsInstance(permissions, dict) + for key in permissions: + self.assertTrue(key.startswith('org.freedesktop.NetworkManager.')) + self.assertIn(permissions[key], ('yes', 'no', 'auth')) + + def test_devices(self): + dev1 = NetworkManager.NetworkManager.GetDevices() + dev2 = NetworkManager.NetworkManager.GetAllDevices() + for dev in dev1: + self.assertIsInstance(dev, NetworkManager.Device) + self.assertIsStrictSubclass(dev.__class__, NetworkManager.Device) + self.assertIn(dev, dev2) + dev = NetworkManager.NetworkManager.GetDeviceByIpIface(dev1[0].IpInterface) + self.assertEqual(dev, dev1[0]) + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_settings.py b/test/test_settings.py new file mode 100644 index 0000000..5d5ed3f --- /dev/null +++ b/test/test_settings.py @@ -0,0 +1,29 @@ +from test import * +import socket + +class SettingsTest(TestCase): + def test_connections(self): + conn1 = NetworkManager.Settings.Connections + conn2 = NetworkManager.Settings.ListConnections() + self.assertIsInstance(conn1, list) + for conn in conn1: + self.assertIn(conn, conn2) + for conn in conn2: + self.assertIn(conn, conn1) + conn = NetworkManager.Settings.GetConnectionByUuid(conn1[0].GetSettings()['connection']['uuid']) + + @unittest.skipUnless(os.getuid() == 0, "Must be root to reload connections") + def test_reload(self): + self.assertTrue(NetworkManager.Settings.ReloadConnections()) + + @unittest.skipUnless(have_permission('settings.modify.hostname'), "don't have permission to modify the hostname") + def test_hostname(self): + hn = NetworkManager.Settings.Hostname + self.assertEqual(hn, socket.gethostname()) + NetworkManager.Settings.SaveHostname(hn + '-test') + self.assertEqual(NetworkManager.Settings.Hostname, hn + '-test') + NetworkManager.Settings.SaveHostname(hn) + self.assertEqual(NetworkManager.Settings.Hostname, hn) + +if __name__ == '__main__': + unittest.main()