[PATCH 05/17] Move NM dbus calls to separate module.

Radek Vykydal rvykydal at redhat.com
Fri Feb 8 14:04:21 UTC 2013


Also use GDBus instead of python-dbus / libdbus which has issues
with threads.
---
 pyanaconda/nm.py | 506 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 506 insertions(+)
 create mode 100644 pyanaconda/nm.py

diff --git a/pyanaconda/nm.py b/pyanaconda/nm.py
new file mode 100644
index 0000000..0991020
--- /dev/null
+++ b/pyanaconda/nm.py
@@ -0,0 +1,506 @@
+# Network configuration proxy to NetworkManager
+#
+# Copyright (C) 2013  Red Hat, Inc.
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# the GNU General Public License v.2, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY expressed or implied, including the implied warranties of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
+# Public License for more details.  You should have received a copy of the
+# GNU General Public License along with this program; if not, write to the
+# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
+# source code or documentation are not subject to the GNU General Public
+# License and may only be used or replicated with the express permission of
+# Red Hat, Inc.
+#
+# Red Hat Author(s): Radek Vykydal <rvykydal at redhat.com>
+#
+
+from gi.repository import Gio, GLib
+from gi.repository import NetworkManager
+import struct
+import socket
+
+device_type_interfaces = {
+    NetworkManager.DeviceType.ETHERNET: "org.freedesktop.NetworkManager.Device.Wired",
+    NetworkManager.DeviceType.WIFI: "org.freedesktop.NetworkManager.Device.Wireless",
+    NetworkManager.DeviceType.MODEM: "org.freedesktop.NetworkManager.Device.Modem",
+    NetworkManager.DeviceType.BT: "org.freedesktop.NetworkManager.Device.Bluetooth",
+    NetworkManager.DeviceType.OLPC_MESH: "org.freedesktop.NetworkManager.Device.OlpcMesh",
+    NetworkManager.DeviceType.WIMAX: "org.freedesktop.NetworkManager.Device.WiMax",
+    NetworkManager.DeviceType.INFINIBAND: "org.freedesktop.NetworkManager.Device.Infiniband",
+    NetworkManager.DeviceType.BOND: "org.freedesktop.NetworkManager.Device.Bond",
+    NetworkManager.DeviceType.VLAN: "org.freedesktop.NetworkManager.Device.Vlan",
+    NetworkManager.DeviceType.ADSL: "org.freedesktop.NetworkManager.Device.Adsl"
+}
+
+class UnknownDeviceError(ValueError):
+    """Device of specified name was not found by NM"""
+    def __str__(self):
+        return self.__repr__()
+
+class PropertyNotFoundError(ValueError):
+    """Property of NM object was not found"""
+    def __str__(self):
+        return self.__repr__()
+
+class SettingNotFoundError(ValueError):
+    """Setting of NMRemoteConnection was not found"""
+    def __str__(self):
+        return self.__repr__()
+
+class DeviceSettingsNotFoundError(ValueError):
+    """Settings (NMRemoteConnection object was not found"""
+    def __str__(self):
+        return self.__repr__()
+
+
+def _get_proxy(bus_type=Gio.BusType.SYSTEM,
+               flags=Gio.DBusProxyFlags.NONE,
+               info=None,
+               name="org.freedesktop.NetworkManager",
+               object_path="/org/freedesktop/NetworkManager",
+               interface_name="org.freedesktop.NetworkManager",
+               cancellable=None):
+    proxy = Gio.DBusProxy.new_for_bus_sync(bus_type,
+                                           flags,
+                                           info,
+                                           name,
+                                           object_path,
+                                           interface_name,
+                                           cancellable)
+    return proxy
+
+def nm_state():
+    """Return state of NetworkManager"""
+    proxy = _get_proxy()
+    return proxy.get_cached_property("State").unpack()
+
+# FIXME - use just GLOBAL? There is some connectivity checking
+# for GLOBAL in NM (nm_connectivity_get_connected), not sure if
+# and how it is implemented.
+# Also see Gio g_network_monitor_can_reach.
+def nm_is_connected():
+    """Is NetworkManager connected?"""
+    return nm_state() in (NetworkManager.State.CONNECTED_GLOBAL,
+                          NetworkManager.State.CONNECTED_SITE,
+                          NetworkManager.State.CONNECTED_LOCAL)
+
+def nm_is_connecting():
+    """Is NetworkManager connecting?"""
+    return nm_state() == NetworkManager.State.CONNECTING
+
+def nm_devices():
+    """Return list of network device names"""
+
+    interfaces = []
+
+    proxy = _get_proxy()
+    args = None
+    devices = proxy.call_sync("GetDevices",
+                              args,
+                              Gio.DBusCallFlags.NONE,
+                              -1,
+                              None)
+
+    devices = devices.unpack()[0]
+    for device in devices:
+        proxy = _get_proxy(object_path=device, interface_name="org.freedesktop.NetworkManager.Device")
+        interfaces.append(proxy.get_cached_property("Interface").unpack())
+
+    return interfaces
+
+def nm_activated_devices():
+    """Return list of names of activated network devices"""
+
+    interfaces = []
+
+    proxy = _get_proxy()
+    active_connections = proxy.get_cached_property("ActiveConnections").unpack()
+    for ac in active_connections:
+        proxy = _get_proxy(object_path=ac, interface_name="org.freedesktop.NetworkManager.Connection.Active")
+        devices = proxy.get_cached_property("Devices").unpack()
+        for device in devices:
+            proxy = _get_proxy(object_path=device, interface_name="org.freedesktop.NetworkManager.Device")
+            interfaces.append(proxy.get_cached_property("Interface").unpack())
+
+    return interfaces
+
+def nm_device_property(name, property):
+    """Return value of device NM property
+
+       Exceptions:
+       UnknownDeviceError if device is not found
+       PropertyNotFoundError if property is not found
+    """
+
+    retval = None
+
+    proxy = _get_proxy()
+    args = GLib.Variant('(s)', (name,))
+    try:
+        device = proxy.call_sync("GetDeviceByIpIface",
+                                  args,
+                                  Gio.DBusCallFlags.NONE,
+                                  -1,
+                                  None)
+    except Exception as e:
+        if "org.freedesktop.NetworkManager.UnknownDevice" in e.message:
+            raise UnknownDeviceError(name, e)
+        raise
+
+    device = device.unpack()[0]
+    proxy = _get_proxy(object_path=device, interface_name="org.freedesktop.NetworkManager.Device")
+
+    if property in proxy.get_cached_property_names():
+        retval = proxy.get_cached_property(property).unpack()
+    else:
+        # Look in device type based interface
+        device_type = proxy.get_cached_property("DeviceType").unpack()
+        proxy = _get_proxy(object_path=device, interface_name=device_type_interfaces[device_type])
+        if property in proxy.get_cached_property_names():
+            retval = proxy.get_cached_property(property).unpack()
+        else:
+            raise PropertyNotFoundError(property)
+
+    return retval
+
+def nm_device_type_is_wifi(name):
+    """Is the type of device wifi?
+
+       Exceptions:
+       UnknownDeviceError if device is not found
+       PropertyNotFoundError if type is not found
+    """
+    return nm_device_type(name) == NetworkManager.DeviceType.WIFI
+
+def nm_device_type_is_ethernet(name):
+    """Is the type of device ethernet?
+
+       Exceptions:
+       UnknownDeviceError if device is not found
+       PropertyNotFoundError if type is not found
+    """
+    return nm_device_type(name) == NetworkManager.DeviceType.ETHERNET
+
+def nm_device_hwaddress(name):
+    """Return device's 'HwAddress' property
+
+       Exceptions:
+       UnknownDeviceError if device is not found
+       PropertyNotFoundError if property is not found
+    """
+    return nm_device_property(name, "HwAddress")
+
+def nm_device_type(name):
+    """Return device's 'DeviceType' property
+
+       Exceptions:
+       UnknownDeviceError if device is not found
+       PropertyNotFoundError if property is not found
+    """
+    return nm_device_property(name, "DeviceType")
+
+def nm_device_carrier(name):
+    """Return device's 'Carrier' property
+
+       Exceptions:
+       UnknownDeviceError if device is not found
+       PropertyNotFoundError if property is not found
+    """
+    return nm_device_property(name, "Carrier")
+
+def nm_device_ip_addresses(name, version=4):
+    """Return list of devices's IP addresses
+
+       version=4 - ipv4 addresses
+       version=6 - ipv6 addresses
+
+       Exceptions:
+       UnknownDeviceError if device is not found
+       PropertyNotFoundError if ip configuration is not found
+    """
+    retval = []
+
+    state = nm_device_property(name, "State")
+    if state != NetworkManager.DeviceState.ACTIVATED:
+        return []
+
+    if version == 4:
+        dbus_iface = "org.freedesktop.NetworkManager.IP4Config"
+        property = "Ip4Config"
+    elif version == 6:
+        dbus_iface = "org.freedesktop.NetworkManager.IP6Config"
+        property = "Ip6Config"
+    else:
+        return []
+
+    config = nm_device_property(name, property)
+    if config == "/":
+        return []
+    proxy = _get_proxy(object_path=config, interface_name=dbus_iface)
+    addresses = proxy.get_cached_property("Addresses").unpack()
+    for addr, prefix, gateway in addresses:
+        # TODO - look for a library function
+        if version == 4:
+            tmp = struct.pack('I', addr)
+            addr_str = socket.inet_ntop(socket.AF_INET, tmp)
+        elif version == 6:
+            addr_str = socket.inet_ntop(socket.AF_INET6, "".join(chr(byte) for byte in addr))
+        else:
+            continue
+        retval.append(addr_str)
+
+    return retval
+
+def _device_settings(name):
+    """Return object path of device setting
+
+       Returns None if settings were not found.
+
+       Exceptions:
+       UnknownDeviceError - device was not found
+    """
+    hwaddr_str = nm_device_hwaddress(name)
+    return _settings_for_hwaddr(hwaddr_str)
+
+def _settings_for_ap(ssid):
+    """Return object path of wireless access point settings.
+
+       Returns None if settings were not found.
+`   """
+    return _find_settings(ssid, '802-11-wireless', 'ssid',
+            format_value=lambda ba: "".join(chr(b) for b in ba))
+
+def _settings_for_hwaddr(hwaddr):
+    """Return object path of settings of device specified by hw address.
+
+       Returns None if settings were not found.
+    """
+    return _find_settings(hwaddr, '802-3-ethernet', 'mac-address',
+            format_value=lambda ba: ":".join("%02X" % b for b in ba))
+
+def _find_settings(value, key1, key2, format_value=lambda x:x):
+    """Return object path of settings having given value of key1, key2 setting
+
+       Returns None if settings were not found.
+    """
+    retval = None
+
+    proxy = _get_proxy(object_path="/org/freedesktop/NetworkManager/Settings", interface_name="org.freedesktop.NetworkManager.Settings")
+
+    args = None
+    connections = proxy.call_sync("ListConnections",
+                                  args,
+                                  Gio.DBusCallFlags.NONE,
+                                  -1,
+                                  None)
+
+    for con in connections.unpack()[0]:
+        proxy = _get_proxy(object_path=con, interface_name="org.freedesktop.NetworkManager.Settings.Connection")
+        args = None
+        settings = proxy.call_sync("GetSettings",
+                                   args,
+                                   Gio.DBusCallFlags.NONE,
+                                   -1,
+                                   None)
+        settings = settings.unpack()[0]
+        try:
+            v = settings[key1][key2]
+        except KeyError:
+            continue
+        if format_value(v) == value:
+            retval = con
+            break
+
+    return retval
+
+def nm_device_setting_value(name, key1, key2):
+    """Return value of device's setting specified by key1 and key2.
+
+       Return None if the setting is not found which means NM will use
+       default value for the setting.  We can't validate that keys
+       actually correspond to a setting (see "connection" "autoconnect"
+       setting).
+
+       Caller should handle None as default value.
+
+       Exceptions:
+       UnknownDeviceError - device was not found
+       DeviceSettingsNotFoundError - settings were not found (eg for "wlan0")
+    """
+    settings_path = _device_settings(name)
+    if not settings_path:
+        raise DeviceSettingsNotFoundError(name)
+    proxy = _get_proxy(object_path=settings_path, interface_name="org.freedesktop.NetworkManager.Settings.Connection")
+    args = None
+    settings = proxy.call_sync("GetSettings",
+                               args,
+                               Gio.DBusCallFlags.NONE,
+                               -1,
+                               None)
+    settings = settings.unpack()[0]
+    try:
+        value = settings[key1][key2]
+    except KeyError:
+        #raise SettingNotFoundError(key1, key2)
+        value = None
+    return value
+
+def _update_settings_libdbus(settings_path, key1, key2, value):
+    """Update setting key1, key2 with value."""
+    import dbus
+    con = dbus.SystemBus().get_object("org.freedesktop.NetworkManager", settings_path)
+    settings = con.GetSettings()
+    try:
+        # TODO: create key1 if missing?
+        settings[key1][key2] = value
+    except KeyError:
+        return False
+    con.Update(settings)
+    return True
+
+def nm_update_settings_of_device(name, key1, key2, value):
+    """Update setting key1, key2 of device name with value.
+
+       Exceptions:
+       UnknownDeviceError - device was not found
+       DeviceSettingsNotFoundError - settings were not found (eg for "wlan0")
+    """
+    settings_path = _device_settings(name)
+    if not settings_path:
+        raise DeviceSettingsNotFoundError(name)
+    return _update_settings_libdbus(_device_settings(name), key1, key2, value)
+
+
+#def _update_settings(settings_path, key1, key2, value):
+#    """Update settings"""
+#    proxy = _get_proxy(object_path=settings_path, interface_name="org.freedesktop.NetworkManager.Settings.Connection")
+#    args = None
+#    settings = proxy.call_sync("GetSettings",
+#                               args,
+#                               Gio.DBusCallFlags.NONE,
+#                               -1,
+#                               None)
+#    # FIXME need to figure out how to create updated
+#    # GVariant and convert value to GVariant
+#    # to do this: settings[0][key1][key2] = value
+#    proxy.call_sync("Update",
+#                    settings,
+#                    Gio.DBusCallFlags.NONE,
+#                    -1,
+#                    None)
+#
+
+if __name__ == "__main__":
+    print "NM state: %s:" % nm_state()
+    print "NM is connected: %s" % nm_is_connected()
+
+    print "Devices: %s" % nm_devices()
+    print "Activated devices: %s" % nm_activated_devices()
+
+    wireless_device = ""
+
+    devs = nm_devices()
+    devs.append("nonexisting")
+    for devname in devs:
+
+        print devname
+
+        try:
+            type = nm_device_type(devname)
+        except UnknownDeviceError as e:
+            print "     %s" % e
+            type = None
+        if type == NetworkManager.DeviceType.ETHERNET:
+            print "     type %s" % "ETHERNET"
+        elif type == NetworkManager.DeviceType.WIFI:
+            print "     type %s" % "WIFI"
+            wireless_device = devname
+
+        try:
+            print "     Wifi device: %s" % nm_device_type_is_wifi(devname)
+        except UnknownDeviceError as e:
+            print "     %s" % e
+
+        try:
+            hwaddr = nm_device_hwaddress(devname)
+            print "     HwAaddress: %s" % hwaddr
+        except ValueError as e:
+            print "     %s" % e
+            hwaddr = ""
+
+        try:
+            print "     Carrier: %s" % nm_device_carrier(devname)
+        except ValueError as e:
+            print "     %s" % e
+
+        try:
+            print "     IP4 addrs: %s" % nm_device_ip_addresses(devname)
+            print "     IP6 addrs: %s" % nm_device_ip_addresses(devname, version=6)
+            print "     Udi: %s" % nm_device_property(devname, "Udi")
+        except UnknownDeviceError as e:
+            print "     %s" % e
+
+        if devname in nm_devices():
+            try:
+                print "     Nonexisting: %s" % nm_device_property(devname, "Nonexisting")
+            except PropertyNotFoundError as e:
+                print "     %s" % e
+        try:
+            print "     Nonexisting: %s" % nm_device_property(devname, "Nonexisting")
+        except ValueError as e:
+            print "     %s" % e
+
+        try:
+            print "     Settings: %s" % _device_settings(devname)
+        except UnknownDeviceError as e:
+            print "     %s" % e
+        try:
+            print "     Settings for hwaddr %s: %s" % (hwaddr, _settings_for_hwaddr(hwaddr))
+        except UnknownDeviceError as e:
+            print "     %s" % e
+        try:
+            print "     Setting value %s %s: %s" % ("ipv6", "method", nm_device_setting_value(devname, "ipv6", "method"))
+        except ValueError as e:
+            print "     %s" % e
+        try:
+            print "     Setting value %s %s: %s" % ("ipv7", "method", nm_device_setting_value(devname, "ipv7", "method"))
+        except ValueError as e:
+            print "     %s" % e
+
+    ssid = "Red Hat Guest"
+    print "Settings for AP %s: %s" % (ssid, _settings_for_ap(ssid))
+    ssid = "nonexisting"
+    print "Settings for AP %s: %s" % (ssid, _settings_for_ap(ssid))
+
+    devname = devs[0]
+    key1 = "connection"
+    key2 = "autoconnect"
+    original_value = nm_device_setting_value(devname, key1, key2)
+    print "Setting value %s %s: %s" % (key1, key2, original_value)
+    # None means default in this case, which is true
+    if original_value in (None, True):
+        new_value = False
+    else:
+        new_value = True
+
+    print "Updating to %s" % new_value
+    nm_update_settings_of_device(devname, key1, key2, new_value)
+    print "Setting value %s %s: %s" % (key1, key2, nm_device_setting_value(devname, key1, key2))
+
+    nm_update_settings_of_device(devname, key1, "nonexisting", new_value)
+    nm_update_settings_of_device(devname, "nonexisting", "nonexisting", new_value)
+    try:
+        nm_update_settings_of_device("nonexixting", key1, key2, new_value)
+    except UnknownDeviceError as e:
+        print "%s" % e
+
+    if wireless_device:
+        try:
+            nm_update_settings_of_device(wireless_device, key1, key2, new_value)
+        except DeviceSettingsNotFoundError as e:
+            print "%s" % e
-- 
1.7.11.7



More information about the anaconda-patches mailing list