[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