diff options
Diffstat (limited to 'tools')
| -rwxr-xr-x | tools/debug-helper.py | 79 | ||||
| -rwxr-xr-x | tools/generate-docs-nm-property-infos.pl | 159 | ||||
| -rwxr-xr-x | tools/generate-docs-nm-settings-docs-gir.py | 288 | ||||
| -rwxr-xr-x | tools/generate-docs-nm-settings-docs-merge.py | 176 | ||||
| -rwxr-xr-x | tools/meson-post-install.sh | 1 | ||||
| -rwxr-xr-x | tools/run-nm-test.sh | 1 | ||||
| -rwxr-xr-x | tools/test-networkmanager-service.py | 1883 |
7 files changed, 1873 insertions, 714 deletions
diff --git a/tools/debug-helper.py b/tools/debug-helper.py index b841c39c..708cf73c 100755 --- a/tools/debug-helper.py +++ b/tools/debug-helper.py @@ -6,56 +6,85 @@ import argparse bus = dbus.SystemBus() -parser = argparse.ArgumentParser(description='Interface to easily control logging levels for NetworkManager, ModemManager, and wpasupplicant.') +parser = argparse.ArgumentParser( + description="Interface to easily control logging levels for NetworkManager, ModemManager, and wpasupplicant." +) # NM options -parser.add_argument('--nm', dest='do_set_nm_logging', action='store', - help='modify log level for NetworkManager (debug, info, etc.)') -parser.add_argument('--domains', dest='log_domains', action='store', - default=[], nargs='+', - help='log "domains" to use with NetworkManager (HW, CORE, etc.)') +parser.add_argument( + "--nm", + dest="do_set_nm_logging", + action="store", + help="modify log level for NetworkManager (debug, info, etc.)", +) +parser.add_argument( + "--domains", + dest="log_domains", + action="store", + default=[], + nargs="+", + help='log "domains" to use with NetworkManager (HW, CORE, etc.)', +) # MM options -parser.add_argument('--mm', dest='do_set_mm_logging', action='store', - help='modify log level for ModemManager (debug, info, etc.)') +parser.add_argument( + "--mm", + dest="do_set_mm_logging", + action="store", + help="modify log level for ModemManager (debug, info, etc.)", +) # wpasupplicant options -parser.add_argument('--wpa', dest='do_set_wpa_logging', action='store', - help='modify log level for wpasupplicant (debug, msgdump, info, etc.)') +parser.add_argument( + "--wpa", + dest="do_set_wpa_logging", + action="store", + help="modify log level for wpasupplicant (debug, msgdump, info, etc.)", +) args = parser.parse_args() if args.do_set_nm_logging: - #print args.log_domains + # print args.log_domains dom_msg = "" if args.log_domains: - dom_msg = " for domains: " + ','.join(args.log_domains) - print("Setting NetworkManager log level to '" + args.do_set_nm_logging + "'" + dom_msg) + dom_msg = " for domains: " + ",".join(args.log_domains) + print( + "Setting NetworkManager log level to '" + args.do_set_nm_logging + "'" + dom_msg + ) - nm_bus = bus.get_object('org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager') - nm = dbus.Interface(nm_bus, dbus_interface='org.freedesktop.NetworkManager') - nm.SetLogging(args.do_set_nm_logging, ','.join(args.log_domains)) + nm_bus = bus.get_object( + "org.freedesktop.NetworkManager", "/org/freedesktop/NetworkManager" + ) + nm = dbus.Interface(nm_bus, dbus_interface="org.freedesktop.NetworkManager") + nm.SetLogging(args.do_set_nm_logging, ",".join(args.log_domains)) if args.do_set_mm_logging: print("Setting ModemManager log level to '" + args.do_set_mm_logging + "'") - mm_bus = bus.get_object('org.freedesktop.ModemManager', '/org/freedesktop/ModemManager') - mm = dbus.Interface(mm_bus, dbus_interface='org.freedesktop.ModemManager') + mm_bus = bus.get_object( + "org.freedesktop.ModemManager", "/org/freedesktop/ModemManager" + ) + mm = dbus.Interface(mm_bus, dbus_interface="org.freedesktop.ModemManager") mm.SetLogging(args.do_set_mm_logging) if args.do_set_wpa_logging: print("Setting wpa_supplicant log level to '" + args.do_set_wpa_logging + "'") - if 'debug' in args.do_set_wpa_logging or 'msgdump' in args.do_set_wpa_logging: + if "debug" in args.do_set_wpa_logging or "msgdump" in args.do_set_wpa_logging: print("Enabling timestamps for wpasupplicant debugging logs") use_timestamps = dbus.Boolean(True, variant_level=1) else: print("Disabling timestamps for wpasupplicant debugging logs") use_timestamps = dbus.Boolean(False, variant_level=1) - wpa_bus = bus.get_object('fi.w1.wpa_supplicant1', '/fi/w1/wpa_supplicant1') - wpa_properties = dbus.Interface(wpa_bus, dbus_interface='org.freedesktop.DBus.Properties') - wpa_properties.Set('fi.w1.wpa_supplicant1', 'DebugTimestamp', use_timestamps) - wpa_properties.Set('fi.w1.wpa_supplicant1', 'DebugLevel', - dbus.String(args.do_set_wpa_logging, variant_level=1)) - + wpa_bus = bus.get_object("fi.w1.wpa_supplicant1", "/fi/w1/wpa_supplicant1") + wpa_properties = dbus.Interface( + wpa_bus, dbus_interface="org.freedesktop.DBus.Properties" + ) + wpa_properties.Set("fi.w1.wpa_supplicant1", "DebugTimestamp", use_timestamps) + wpa_properties.Set( + "fi.w1.wpa_supplicant1", + "DebugLevel", + dbus.String(args.do_set_wpa_logging, variant_level=1), + ) diff --git a/tools/generate-docs-nm-property-infos.pl b/tools/generate-docs-nm-property-infos.pl new file mode 100755 index 00000000..bbdec3b6 --- /dev/null +++ b/tools/generate-docs-nm-property-infos.pl @@ -0,0 +1,159 @@ +#!/usr/bin/env perl +# SPDX-License-Identifier: GPL-2.0+ +# +# Copyright (C) 2014 Red Hat, Inc. +# + +# +# The script parses nm-setting-*.c files and extracts documentation related +# to setting plugins. The documentation is in a simple format of lines +# "keyword: value". The documentation is enclosed between tags +# ---<plugin-name>--- and ---end--- +# Recognized keywords are: +# "property: " - property name +# "variable: " - name of the variable used by the plugin +# "format: " - format of the value in 'keyfile' plugin +# "default: " - default value when variable is not used +# "values: " - allowed values (e.g. for enumerations) +# "example: " - example(s) +# "description: " - description text +# Value is an arbitrary string that can span over multiple lines. +# +# ifcfg-rh specifics: +# - mark NM extension variables with (+), e.g. variable: UUID(+) +# + +use strict; +use warnings; +use v5.10; + +# global variables +my @keywords = ("property", "variable", "format", "values", "default", "example", "description"); +my @data; +my $fo; + +(scalar @ARGV >= 3) or die "Usage: $0 <plugin> <output-xml-file> <srcfiles>\n"; +my ($plugin, $output, (@source_files)) = @ARGV; +my $start_tag = "---$plugin---\\s*\$"; +my $end_tag = '---end---'; + +# open output file +open $fo, '>', $output or die "Can't open $output: $!"; + +# write XML header +write_header(); + +# write generated documentation for each setting +foreach my $c_file (@source_files) { + my $setting_name = get_setting_name($c_file); + if ($setting_name) { + write_item("<setting name=\"$setting_name\">"); + scan_doc_comments($c_file, $start_tag, $end_tag); + write_item("</setting>"); + } +} + +# write XML footer +write_footer(); + +# close output file +close $fo; + + +### --- subroutines --- ### + +# get setting name from NM_SETTING_*_SETTING_NAME constant in C header file +sub get_setting_name { + my $path = $_[0]; + $path =~ s/c$/h/; # use header file to find out setting name + open my $fh, '<', $path or die "Can't open $path: $!"; + while (my $line = <$fh>) { + if ($line =~ /NM_SETTING_.+SETTING_NAME\s+\"(\S+)\"/) { + return $1; + } + } +} + +# scan source setting file for documentation tags and write them to XML +sub scan_doc_comments { + my($setting_file, $start, $end) = @_; + open my $fi, '<', $setting_file or die "Can't open $setting_file: $!"; + while (<$fi>) { + if (/$start/ .. /$end/) { + next if /$start/; + if (/$end/) { + process_data(); + } else { + push @data, $_; + } + next; + } + # ignore text not inside marks + } + close $fi; +} + +# process plugin property documentation comments +sub process_data { + return if not @data; + my $kwd_pat = join("|", @keywords); + my %parsed_data; + my $this_key; + + foreach (@data) { + if (/^\s*\**\s+($kwd_pat):\s+(.*?)\s*$/) { + $this_key = $1; + $parsed_data{$this_key} = "$2\n"; + } elsif (/^\s*\**\s+(.*?)\s*$/) { + die "Extra mess in a comment: $_" unless $this_key; + $parsed_data{$this_key} .= "$1\n"; + } + } + + # now write a line into the XML + my $name = $parsed_data{property} // ""; + my $var = $parsed_data{variable} // $name; # fallback to "property: " + my $format = $parsed_data{format} // ""; + my $values = $parsed_data{values} // ""; + my $def = $parsed_data{default} // ""; + my $exam = $parsed_data{example} // ""; + my $desc = $parsed_data{description} // ""; + + chomp($name, $var, $format, $values, $def, $exam, $desc); + escape_xml_chars($name, $var, $format, $values, $def, $exam, $desc); + my $foo = sprintf("<property name=\"%s\" variable=\"%s\" format=\"%s\" values=\"%s\" ". + "default=\"%s\" example=\"%s\" description=\"%s\"/>", + $name, $var, $format, $values, $def, $exam, $desc); + write_item($foo); + @data = (); +} + +# - XML handling - +sub write_header { + (my $header = + qq{<nm-setting-docs> + }) =~ s/^ {7}//mg; + print {$fo} $header; +} + +sub write_footer { + my $footer = "</nm-setting-docs>"; + print {$fo} $footer; +} + +sub write_item { + my $str = join("", @_); + print {$fo} $str, "\n"; +} + +sub escape_xml_chars { + # http://en.wikipedia.org/wiki/List_of_XML_and_HTML_character_entity_references#Predefined%5Fentities%5Fin%5FXML + foreach my $val (@_) { + $val =~ s/&/&/sg; + $val =~ s/</</sg; + $val =~ s/>/>/sg; + $val =~ s/"/"/sg; + $val =~ s/'/'/sg; + } +} + diff --git a/tools/generate-docs-nm-settings-docs-gir.py b/tools/generate-docs-nm-settings-docs-gir.py new file mode 100755 index 00000000..0eb0da97 --- /dev/null +++ b/tools/generate-docs-nm-settings-docs-gir.py @@ -0,0 +1,288 @@ +#!/usr/bin/env python +# SPDX-License-Identifier: LGPL-2.1+ +# +# Copyright (C) 2009 - 2017 Red Hat, Inc. +# + +from __future__ import print_function + +import os +import gi + +gi.require_version("GIRepository", "2.0") +from gi.repository import GIRepository +import argparse, re, sys +import xml.etree.ElementTree as ET + +try: + libs = os.environ["LD_LIBRARY_PATH"].split(":") + libs.reverse() + for lib in libs: + GIRepository.Repository.prepend_library_path(lib) +except AttributeError: + # An old GI version, that has no prepend_library_path + # It's alright, it probably interprets LD_LIBRARY_PATH + # correctly. + pass +except KeyError: + pass + +gi.require_version("NM", "1.0") +from gi.repository import NM, GObject + +dbus_type_name_map = { + "b": "boolean", + "s": "string", + "i": "int32", + "u": "uint32", + "t": "uint64", + "x": "int64", + "y": "byte", + "as": "array of string", + "au": "array of uint32", + "ay": "byte array", + "a{ss}": "dict of string to string", + "a{sv}": "vardict", + "aa{sv}": "array of vardict", + "aau": "array of array of uint32", + "aay": "array of byte array", + "a(ayuay)": "array of legacy IPv6 address struct", + "a(ayuayu)": "array of legacy IPv6 route struct", +} + +ns_map = { + "c": "http://www.gtk.org/introspection/c/1.0", + "gi": "http://www.gtk.org/introspection/core/1.0", + "glib": "http://www.gtk.org/introspection/glib/1.0", +} +identifier_key = "{%s}identifier" % ns_map["c"] +nick_key = "{%s}nick" % ns_map["glib"] +symbol_prefix_key = "{%s}symbol-prefix" % ns_map["c"] + +constants = { + "TRUE": "TRUE", + "FALSE": "FALSE", + "G_MAXUINT32": "G_MAXUINT32", + "NULL": "NULL", +} +setting_names = {} + + +def get_setting_name_define(setting): + n = setting.attrib[symbol_prefix_key] + if n and n.startswith("setting_"): + return n[8:].upper() + raise Exception('Unexpected symbol_prefix_key "%s"' % (n)) + + +def init_constants(girxml, settings): + for const in girxml.findall("./gi:namespace/gi:constant", ns_map): + cname = const.attrib["{%s}type" % ns_map["c"]] + cvalue = const.attrib["value"] + if const.find('./gi:type[@name="utf8"]', ns_map) is not None: + cvalue = '"%s"' % cvalue + constants[cname] = cvalue + + for enum in girxml.findall("./gi:namespace/gi:enumeration", ns_map): + for enumval in enum.findall("./gi:member", ns_map): + cname = enumval.attrib[identifier_key] + cvalue = "%s (%s)" % (cname, enumval.attrib["value"]) + constants[cname] = cvalue + + for enum in girxml.findall("./gi:namespace/gi:bitfield", ns_map): + for enumval in enum.findall("./gi:member", ns_map): + cname = enumval.attrib[identifier_key] + cvalue = "%s (0x%x)" % (cname, int(enumval.attrib["value"])) + constants[cname] = cvalue + + for setting in settings: + setting_type_name = "NM" + setting.attrib["name"] + setting_name_symbol = ( + "NM_SETTING_" + get_setting_name_define(setting) + "_SETTING_NAME" + ) + if setting_name_symbol in constants: + setting_name = constants[setting_name_symbol] + setting_names[setting_type_name] = setting_name + + +def get_prop_type(setting, pspec): + dbus_type = setting.get_dbus_property_type(pspec.name).dup_string() + prop_type = dbus_type_name_map[dbus_type] + + if GObject.type_is_a(pspec.value_type, GObject.TYPE_ENUM) or GObject.type_is_a( + pspec.value_type, GObject.TYPE_FLAGS + ): + prop_type = "%s (%s)" % (pspec.value_type.name, prop_type) + + return prop_type + + +def get_docs(propxml): + doc_xml = propxml.find("gi:doc", ns_map) + if doc_xml is None: + return None + + doc = doc_xml.text + if "deprecated" in propxml.attrib: + doc = doc + " Deprecated: " + propxml.attrib["deprecated"] + + doc = re.sub(r"\n\s*", r" ", doc) + + # Expand constants + doc = re.sub(r"%([^%]\w*)", lambda match: constants[match.group(1)], doc) + + # #NMSettingWired:mac-address -> "mac-address" + doc = re.sub(r"#[A-Za-z0-9_]*:([A-Za-z0-9_-]*)", r'"\1"', doc) + + # #NMSettingWired setting -> "802-3-ethernet" setting + doc = re.sub( + r"#([A-Z]\w*) setting", + lambda match: setting_names[match.group(1)] + " setting", + doc, + ) + + # remaining gtk-doc cleanup + doc = doc.replace("%%", "%") + doc = doc.replace("<!-- -->", "") + doc = re.sub(r" Element-.ype:.*", "", doc) + doc = re.sub(r"#([A-Z]\w*)", r"\1", doc) + + # Remove sentences that refer to functions + doc = re.sub(r"\.\s+[^.]*\w\(\)[^.]*\.", r".", doc) + + return doc + + +def get_default_value(setting, pspec, propxml): + default_value = setting.get_property(pspec.name.replace("-", "_")) + if default_value is None: + return default_value + + value_type = get_prop_type(setting, pspec) + if value_type == "string" and default_value != "" and pspec.name != "name": + default_value = '"%s"' % default_value + elif value_type == "boolean": + default_value = str(default_value).upper() + elif value_type == "byte array": + default_value = "[]" + elif str(default_value).startswith("<"): + default_value = None + elif str(default_value).startswith("["): + default_value = None + + return default_value + + +def settings_sort_key(x): + x_prefix = x.attrib["{%s}symbol-prefix" % ns_map["c"]] + # always sort NMSettingConnection first + return (x_prefix != "setting_connection", x_prefix) + + +def escape(val): + return str(val).replace('"', """) + + +def usage(): + print("Usage: %s --gir FILE --output FILE" % sys.argv[0]) + exit() + + +parser = argparse.ArgumentParser() +parser.add_argument( + "-l", + "--lib-path", + metavar="PATH", + action="append", + help="path to scan for shared libraries", +) +parser.add_argument("-g", "--gir", metavar="FILE", help="NM-1.0.gir file") +parser.add_argument("-o", "--output", metavar="FILE", help="output file") + +args = parser.parse_args() +if args.gir is None or args.output is None: + usage() + +if args.lib_path: + for lib in args.lib_path: + GIRepository.Repository.prepend_library_path(lib) + +girxml = ET.parse(args.gir).getroot() +outfile = open(args.output, mode="w") + +basexml = girxml.find('./gi:namespace/gi:class[@name="Setting"]', ns_map) +settings = girxml.findall('./gi:namespace/gi:class[@parent="Setting"]', ns_map) +# Hack. Need a better way to do this +ipxml = girxml.find('./gi:namespace/gi:class[@name="SettingIPConfig"]', ns_map) +settings.extend( + girxml.findall('./gi:namespace/gi:class[@parent="SettingIPConfig"]', ns_map) +) +settings = sorted(settings, key=settings_sort_key) + +init_constants(girxml, settings) + +outfile.write( + """<?xml version=\"1.0\"?> +<!DOCTYPE nm-setting-docs [ +<!ENTITY quot """> +]> +<nm-setting-docs> +""" +) + +for settingxml in settings: + if "abstract" in settingxml.attrib: + continue + + new_func = NM.__getattr__(settingxml.attrib["name"]) + setting = new_func() + + class_desc = get_docs(settingxml) + if class_desc is None: + raise Exception( + "%s needs a gtk-doc block with one-line description" % setting.props.name + ) + outfile.write( + ' <setting name="%s" description="%s" name_upper="%s" >\n' + % (setting.props.name, class_desc, get_setting_name_define(settingxml)) + ) + + setting_properties = { + prop.name: prop + for prop in GObject.list_properties(setting) + if prop.name != "name" + } + + for prop in sorted(setting_properties): + pspec = setting_properties[prop] + + propxml = settingxml.find('./gi:property[@name="%s"]' % pspec.name, ns_map) + if propxml is None: + propxml = basexml.find('./gi:property[@name="%s"]' % pspec.name, ns_map) + if propxml is None: + propxml = ipxml.find('./gi:property[@name="%s"]' % pspec.name, ns_map) + + value_type = get_prop_type(setting, pspec) + value_desc = get_docs(propxml) + default_value = get_default_value(setting, pspec, propxml) + + prop_upper = prop.upper().replace("-", "_") + + if value_desc is None: + raise Exception( + "%s.%s needs a documentation description" % (setting.props.name, prop) + ) + + default_value_as_xml = "" + if default_value is not None: + default_value_as_xml = ' default="%s"' % (escape(default_value)) + + outfile.write( + ' <property name="%s" name_upper="%s" type="%s"%s description="%s" />\n' + % (prop, prop_upper, value_type, default_value_as_xml, escape(value_desc)) + ) + + outfile.write(" </setting>\n") + +outfile.write("</nm-setting-docs>\n") +outfile.close() diff --git a/tools/generate-docs-nm-settings-docs-merge.py b/tools/generate-docs-nm-settings-docs-merge.py new file mode 100755 index 00000000..0da3a0bb --- /dev/null +++ b/tools/generate-docs-nm-settings-docs-merge.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python +# SPDX-License-Identifier: LGPL-2.1+ + +from __future__ import print_function + +import os +import sys +import collections +import xml.etree.ElementTree as ET + +############################################################################### + +_setting_name_order = [ + "connection", + "6lowpan", + "802-1x", + "adsl", + "bluetooth", + "bond", + "bridge", + "bridge-port", + "cdma", + "dcb", + "dummy", + "ethtool", + "generic", + "gsm", + "infiniband", + "ipv4", + "ipv6", + "ip-tunnel", + "macsec", + "macvlan", + "match", + "802-11-olpc-mesh", + "ovs-bridge", + "ovs-dpdk", + "ovs-interface", + "ovs-patch", + "ovs-port", + "ppp", + "pppoe", + "proxy", + "serial", + "sriov", + "tc", + "team", + "team-port", + "tun", + "user", + "vlan", + "vpn", + "vrf", + "vxlan", + "wifi-p2p", + "wimax", + "802-3-ethernet", + "wireguard", + "802-11-wireless", + "802-11-wireless-security", + "wpan", +] + + +def _setting_name_order_idx(name): + try: + return _setting_name_order.index(name) + except ValueError: + return len(_setting_name_order) + + +def key_fcn_setting_name(n1): + return (_setting_name_order_idx(n1), n1) + + +def iter_keys_of_dicts(dicts, key=None): + keys = set([k for d in dicts for k in d.keys()]) + return sorted(keys, key=key) + + +def node_to_dict(node, tag, key_attr): + dictionary = collections.OrderedDict() + if node is not None: + for n in node.iter(tag): + k = n.get(key_attr) + assert k is not None + dictionary[k] = n + return dictionary + + +def node_get_attr(nodes, name): + for n in nodes: + if n is None: + continue + x = n.get(name, None) + if x: + return x + return None + + +def node_set_attr(dst_node, name, nodes): + x = node_get_attr(nodes, name) + if x: + dst_node.set(name, x) + + +############################################################################### + +gl_only_from_first = False + +argv = list(sys.argv[1:]) +while True: + if argv[0] == "--only-from-first": + gl_only_from_first = True + del argv[0] + continue + break +if len(argv) < 2: + print("%s [--only-from-first] [OUT_FILE] [SETTING_XML [...]]" % (sys.argv[0])) + exit(1) + +gl_output_xml_file = argv[0] +gl_input_files = list(argv[1:]) + +############################################################################### + +xml_roots = list([ET.parse(f).getroot() for f in gl_input_files]) + +assert all([root.tag == "nm-setting-docs" for root in xml_roots]) + +settings_roots = list([node_to_dict(root, "setting", "name") for root in xml_roots]) + +root_node = ET.Element("nm-setting-docs") + +for setting_name in iter_keys_of_dicts(settings_roots, key_fcn_setting_name): + + settings = list([d.get(setting_name) for d in settings_roots]) + + if gl_only_from_first and settings[0] is None: + continue + + properties = list([node_to_dict(s, "property", "name") for s in settings]) + + if gl_only_from_first and not properties[0]: + continue + + setting_node = ET.SubElement(root_node, "setting") + + setting_node.set("name", setting_name) + + node_set_attr(setting_node, "description", settings) + node_set_attr(setting_node, "name_upper", settings) + node_set_attr(setting_node, "alias", settings) + + for property_name in iter_keys_of_dicts(properties): + + properties_attrs = list([p.get(property_name) for p in properties]) + + if gl_only_from_first and properties_attrs[0] is None: + continue + + property_node = ET.SubElement(setting_node, "property") + property_node.set("name", property_name) + property_node.set("name_upper", property_name.upper().replace("-", "_")) + + x = node_get_attr(properties_attrs, "format") + if x: + property_node.set("type", x) + else: + node_set_attr(property_node, "type", properties_attrs) + + node_set_attr(property_node, "default", properties_attrs) + node_set_attr(property_node, "description", properties_attrs) + node_set_attr(property_node, "alias", properties_attrs) + +ET.ElementTree(root_node).write(gl_output_xml_file) diff --git a/tools/meson-post-install.sh b/tools/meson-post-install.sh index a6fd0961..897a8c5c 100755 --- a/tools/meson-post-install.sh +++ b/tools/meson-post-install.sh @@ -52,6 +52,7 @@ if [ "$enable_docs" = 1 ]; then done ln -f "${DESTDIR}${nm_mandir}/man5/NetworkManager.conf.5" "${DESTDIR}${nm_mandir}/man5/nm-system-settings.conf.5" + ln -f "${DESTDIR}${nm_mandir}/man5/nm-settings-nmcli.5" "${DESTDIR}${nm_mandir}/man5/nm-settings.5" fi if [ "$enable_ifcfg_rh" = 1 ]; then diff --git a/tools/run-nm-test.sh b/tools/run-nm-test.sh index 02fced01..11c9e17a 100755 --- a/tools/run-nm-test.sh +++ b/tools/run-nm-test.sh @@ -310,6 +310,7 @@ if [ $HAS_ERRORS -eq 0 ]; then # valgrind doesn't support setns syscall and spams the logfile. # hack around it... case "$TEST_NAME" in + 'test-config' | \ 'test-link-linux' | \ 'test-acd' | \ 'test-service-providers' | \ diff --git a/tools/test-networkmanager-service.py b/tools/test-networkmanager-service.py index 56a53a45..359c5e5d 100755 --- a/tools/test-networkmanager-service.py +++ b/tools/test-networkmanager-service.py @@ -8,7 +8,7 @@ import gi from gi.repository import GLib try: - gi.require_version('NM', '1.0') + gi.require_version("NM", "1.0") from gi.repository import NM except Exception as e: print("Cannot load gi.NM: %s" % (str(e))) @@ -30,23 +30,28 @@ _DEFAULT_ARG = object() ############################################################################### + class Global: pass + gl = None ############################################################################### + class TestError(AssertionError): - def __init__(self, message = 'Unspecified error', errors = None): + def __init__(self, message="Unspecified error", errors=None): AssertionError.__init__(self, message) self.errors = errors + ############################################################################### + class Util: - PY3 = (sys.version_info[0] == 3) + PY3 = sys.version_info[0] == 3 @staticmethod def g_source_remove(source_id): @@ -54,14 +59,14 @@ class Util: GLib.source_remove(source_id) @staticmethod - def addr_family_check(family, allow_af_unspec = False): + def addr_family_check(family, allow_af_unspec=False): if family == socket.AF_INET: return if family == socket.AF_INET6: return if allow_af_unspec and family == socket.AF_UNSPEC: return - raise TestError('invalid address family %s' % (family)) + raise TestError("invalid address family %s" % (family)) @staticmethod def ip_addr_pton(addr, family=None): @@ -86,11 +91,11 @@ class Util: return (a, family) @staticmethod - def ip_addr_ntop(addr, family = None): + def ip_addr_ntop(addr, family=None): if Util.PY3: a = bytes(addr) else: - a = ''.join([chr(c) for c in addr]) + a = "".join([chr(c) for c in addr]) if len(a) == 4: f = socket.AF_INET elif len(a) == 16: @@ -98,11 +103,14 @@ class Util: else: raise TestError("Invalid binary IP address '%s'" % (repr(addr))) if family is not None and f != family: - raise TestError("Unexpected address family. Expected %s but ip address was %s" % (family, repr(addr))) + raise TestError( + "Unexpected address family. Expected %s but ip address was %s" + % (family, repr(addr)) + ) return socket.inet_ntop(f, a) @staticmethod - def ip_addr_norm(addr, family = None): + def ip_addr_norm(addr, family=None): a, family = Util.ip_addr_pton(addr, family) return (Util.ip_addr_ntop(a, family), family) @@ -121,10 +129,12 @@ class Util: return Util.ip_addr_pton(addr, socket.AF_INET6)[0] @staticmethod - def ip_net_parse(net, family = None): - parts = net.split('/') + def ip_net_parse(net, family=None): + parts = net.split("/") if len(parts) != 2: - raise TestError("Invalid IP network '%s' has not '/' for the prefix length" % (net)) + raise TestError( + "Invalid IP network '%s' has not '/' for the prefix length" % (net) + ) prefix = int(parts[1]) addr, family = Util.ip_addr_norm(parts[0], family) if family == socket.AF_INET: @@ -135,14 +145,16 @@ class Util: raise TestError("Invalid prefix length for IPv4 address '%s'" % (net)) return (addr, prefix, family) - class RandomSeed(): + class RandomSeed: def __init__(self, seed): self.cnt = 0 self.seed = str(seed) + def _next(self): c = self.cnt self.cnt += 1 - return self.seed + '-' + str(c) + return self.seed + "-" + str(c) + @staticmethod def wrap(seed): if seed is None: @@ -150,8 +162,9 @@ class Util: if isinstance(seed, Util.RandomSeed): return seed return Util.RandomSeed(seed) + @staticmethod - def get(seed, extra_seed = None): + def get(seed, extra_seed=None): if seed is None: return None if isinstance(seed, Util.RandomSeed): @@ -162,12 +175,14 @@ class Util: try: extra_seed = Util.RandomSeed._extra_seed except: - extra_seed = os.environ.get('NM_TEST_NETWORKMANAGER_SERVICE_SEED', '') + extra_seed = os.environ.get( + "NM_TEST_NETWORKMANAGER_SERVICE_SEED", "" + ) Util.RandomSeed._extra_seed = extra_seed return extra_seed + seed @staticmethod - def random_stream(seed, length = None): + def random_stream(seed, length=None): seed = Util.RandomSeed.wrap(seed) # generates a stream of integers, in the range [0..255] if seed is None: @@ -181,7 +196,7 @@ class Util: while length is None or length > 0: if not v: s = Util.RandomSeed.get(seed) - s = s.encode('utf8') + s = s.encode("utf8") v = hashlib.sha256(s).hexdigest() yield int(v[0:2], 16) v = v[2:] @@ -189,7 +204,7 @@ class Util: length -= 1 @staticmethod - def random_int(seed, v_start = _DEFAULT_ARG, v_end = _DEFAULT_ARG): + def random_int(seed, v_start=_DEFAULT_ARG, v_end=_DEFAULT_ARG): # - if neither start not end is give, return a number in the range # u32 range [0, 0xFFFFFFFF] # - if only start is given (the first argument), interpret it as @@ -231,7 +246,9 @@ class Util: all_set = list(all_set) result = [] seed = Util.RandomSeed.wrap(seed) - for i in list(range(Util.random_int(Util.RandomSeed.get(seed), len(all_set) + 1))): + for i in list( + range(Util.random_int(Util.RandomSeed.get(seed), len(all_set) + 1)) + ): idx = Util.random_int(Util.RandomSeed.get(seed), len(all_set)) result.append(all_set[idx]) del all_set[idx] @@ -239,10 +256,10 @@ class Util: @staticmethod def random_mac(seed): - return '%02X:%02X:%02X:%02X:%02X:%02X' % tuple(Util.random_stream(seed, 6)) + return "%02X:%02X:%02X:%02X:%02X:%02X" % tuple(Util.random_stream(seed, 6)) @staticmethod - def random_ip(seed, net = None, family = None): + def random_ip(seed, net=None, family=None): if net is not None: mask, prefix, family = Util.ip_net_parse(net, family) a_mask, unused = Util.ip_addr_pton(mask, family) @@ -277,120 +294,156 @@ class Util: @staticmethod def variant_from_dbus(val): if isinstance(val, (dbus.String, str)): - return GLib.Variant('s', str(val)) + return GLib.Variant("s", str(val)) if isinstance(val, dbus.UInt32): - return GLib.Variant('u', int(val)) + return GLib.Variant("u", int(val)) if isinstance(val, dbus.UInt64): - return GLib.Variant('t', int(val)) + return GLib.Variant("t", int(val)) if isinstance(val, dbus.Boolean): - return GLib.Variant('b', bool(val)) + return GLib.Variant("b", bool(val)) if isinstance(val, dbus.Byte): - return GLib.Variant('y', int(val)) + return GLib.Variant("y", int(val)) if isinstance(val, dbus.Array): try: - if val.signature == 's': - return GLib.Variant('as', [Util.variant_from_dbus(x) for x in val]) - if val.signature == 'b': - return GLib.Variant('ab', [Util.variant_from_dbus(x) for x in val]) - if val.signature == 'y': - return GLib.Variant('ay', [int(x) for x in val]) - if val.signature == 'u': - return GLib.Variant('au', [int(x) for x in val]) - if val.signature == 'ay': - return GLib.Variant('aay', [Util.variant_from_dbus(x) for x in val]) - if val.signature == 'au': - return GLib.Variant('aau', [Util.variant_from_dbus(x) for x in val]) - if val.signature == 'a{sv}': - return GLib.Variant('aa{sv}', [collections.OrderedDict([(str(k), Util.variant_from_dbus(v)) for k, v in addr.items()]) for addr in val]) - if val.signature == '(ayuay)': - return GLib.Variant('a(ayuay)', [Util.variant_from_dbus(x) for x in val]) - if val.signature == '(ayuayu)': - return GLib.Variant('a(ayuayu)', [Util.variant_from_dbus(x) for x in val]) + if val.signature == "s": + return GLib.Variant("as", [Util.variant_from_dbus(x) for x in val]) + if val.signature == "b": + return GLib.Variant("ab", [Util.variant_from_dbus(x) for x in val]) + if val.signature == "y": + return GLib.Variant("ay", [int(x) for x in val]) + if val.signature == "u": + return GLib.Variant("au", [int(x) for x in val]) + if val.signature == "ay": + return GLib.Variant("aay", [Util.variant_from_dbus(x) for x in val]) + if val.signature == "au": + return GLib.Variant("aau", [Util.variant_from_dbus(x) for x in val]) + if val.signature == "a{sv}": + return GLib.Variant( + "aa{sv}", + [ + collections.OrderedDict( + [ + (str(k), Util.variant_from_dbus(v)) + for k, v in addr.items() + ] + ) + for addr in val + ], + ) + if val.signature == "(ayuay)": + return GLib.Variant( + "a(ayuay)", [Util.variant_from_dbus(x) for x in val] + ) + if val.signature == "(ayuayu)": + return GLib.Variant( + "a(ayuayu)", [Util.variant_from_dbus(x) for x in val] + ) except Exception as e: - raise Exception("Cannot convert array element to type '%s': %s" % (val.signature, e.message)) + raise Exception( + "Cannot convert array element to type '%s': %s" + % (val.signature, e.message) + ) if isinstance(val, dbus.Dictionary): - if val.signature == 'ss': - return GLib.Variant('a{ss}', collections.OrderedDict([(str(k), str(v)) for k, v in val.items()])) - if val.signature == 'sv': - return GLib.Variant('a{sv}', collections.OrderedDict([(str(k), Util.variant_from_dbus(v)) for k, v in val.items()])) - if val.signature == 'sa{sv}': - c = collections.OrderedDict([ - (str(key1), - collections.OrderedDict([(str(key2), Util.variant_from_dbus(arr2)) for key2, arr2 in arr1.items()]) - ) for key1, arr1 in val.items() - ]) - return GLib.Variant('a{sa{sv}}', c) + if val.signature == "ss": + return GLib.Variant( + "a{ss}", + collections.OrderedDict([(str(k), str(v)) for k, v in val.items()]), + ) + if val.signature == "sv": + return GLib.Variant( + "a{sv}", + collections.OrderedDict( + [(str(k), Util.variant_from_dbus(v)) for k, v in val.items()] + ), + ) + if val.signature == "sa{sv}": + c = collections.OrderedDict( + [ + ( + str(key1), + collections.OrderedDict( + [ + (str(key2), Util.variant_from_dbus(arr2)) + for key2, arr2 in arr1.items() + ] + ), + ) + for key1, arr1 in val.items() + ] + ) + return GLib.Variant("a{sa{sv}}", c) raise Exception("Unsupported type for value '%s'" % (repr(val))) + ############################################################################### -IFACE_DBUS = 'org.freedesktop.DBus' -IFACE_OBJECT_MANAGER = 'org.freedesktop.DBus.ObjectManager' -IFACE_CONNECTION = 'org.freedesktop.NetworkManager.Settings.Connection' -IFACE_DEVICE = 'org.freedesktop.NetworkManager.Device' -IFACE_WIFI = 'org.freedesktop.NetworkManager.Device.Wireless' -IFACE_TEST = 'org.freedesktop.NetworkManager.LibnmGlibTest' -IFACE_NM = 'org.freedesktop.NetworkManager' -IFACE_SETTINGS = 'org.freedesktop.NetworkManager.Settings' -IFACE_AGENT_MANAGER = 'org.freedesktop.NetworkManager.AgentManager' -IFACE_AGENT = 'org.freedesktop.NetworkManager.SecretAgent' -IFACE_WIRED = 'org.freedesktop.NetworkManager.Device.Wired' -IFACE_VLAN = 'org.freedesktop.NetworkManager.Device.Vlan' -IFACE_WIFI_AP = 'org.freedesktop.NetworkManager.AccessPoint' -IFACE_ACTIVE_CONNECTION = 'org.freedesktop.NetworkManager.Connection.Active' -IFACE_VPN_CONNECTION = 'org.freedesktop.NetworkManager.VPN.Connection' -IFACE_DNS_MANAGER = 'org.freedesktop.NetworkManager.DnsManager' -IFACE_IP4_CONFIG = 'org.freedesktop.NetworkManager.IP4Config' -IFACE_IP6_CONFIG = 'org.freedesktop.NetworkManager.IP6Config' -IFACE_DHCP4_CONFIG = 'org.freedesktop.NetworkManager.DHCP4Config' -IFACE_DHCP6_CONFIG = 'org.freedesktop.NetworkManager.DHCP6Config' +IFACE_DBUS = "org.freedesktop.DBus" +IFACE_OBJECT_MANAGER = "org.freedesktop.DBus.ObjectManager" +IFACE_CONNECTION = "org.freedesktop.NetworkManager.Settings.Connection" +IFACE_DEVICE = "org.freedesktop.NetworkManager.Device" +IFACE_WIFI = "org.freedesktop.NetworkManager.Device.Wireless" +IFACE_TEST = "org.freedesktop.NetworkManager.LibnmGlibTest" +IFACE_NM = "org.freedesktop.NetworkManager" +IFACE_SETTINGS = "org.freedesktop.NetworkManager.Settings" +IFACE_AGENT_MANAGER = "org.freedesktop.NetworkManager.AgentManager" +IFACE_AGENT = "org.freedesktop.NetworkManager.SecretAgent" +IFACE_WIRED = "org.freedesktop.NetworkManager.Device.Wired" +IFACE_VLAN = "org.freedesktop.NetworkManager.Device.Vlan" +IFACE_WIFI_AP = "org.freedesktop.NetworkManager.AccessPoint" +IFACE_ACTIVE_CONNECTION = "org.freedesktop.NetworkManager.Connection.Active" +IFACE_VPN_CONNECTION = "org.freedesktop.NetworkManager.VPN.Connection" +IFACE_DNS_MANAGER = "org.freedesktop.NetworkManager.DnsManager" +IFACE_IP4_CONFIG = "org.freedesktop.NetworkManager.IP4Config" +IFACE_IP6_CONFIG = "org.freedesktop.NetworkManager.IP6Config" +IFACE_DHCP4_CONFIG = "org.freedesktop.NetworkManager.DHCP4Config" +IFACE_DHCP6_CONFIG = "org.freedesktop.NetworkManager.DHCP6Config" ############################################################################### -class BusErr: +class BusErr: class UnknownInterfaceException(dbus.DBusException): - _dbus_error_name = IFACE_DBUS + '.UnknownInterface' + _dbus_error_name = IFACE_DBUS + ".UnknownInterface" class UnknownPropertyException(dbus.DBusException): - _dbus_error_name = IFACE_DBUS + '.UnknownProperty' + _dbus_error_name = IFACE_DBUS + ".UnknownProperty" class InvalidPropertyException(dbus.DBusException): - _dbus_error_name = IFACE_CONNECTION + '.InvalidProperty' + _dbus_error_name = IFACE_CONNECTION + ".InvalidProperty" class MissingPropertyException(dbus.DBusException): - _dbus_error_name = IFACE_CONNECTION + '.MissingProperty' + _dbus_error_name = IFACE_CONNECTION + ".MissingProperty" class InvalidSettingException(dbus.DBusException): - _dbus_error_name = IFACE_CONNECTION + '.InvalidSetting' + _dbus_error_name = IFACE_CONNECTION + ".InvalidSetting" class MissingSettingException(dbus.DBusException): - _dbus_error_name = IFACE_CONNECTION + '.MissingSetting' + _dbus_error_name = IFACE_CONNECTION + ".MissingSetting" class NotSoftwareException(dbus.DBusException): - _dbus_error_name = IFACE_DEVICE + '.NotSoftware' + _dbus_error_name = IFACE_DEVICE + ".NotSoftware" class ApNotFoundException(dbus.DBusException): - _dbus_error_name = IFACE_WIFI + '.AccessPointNotFound' + _dbus_error_name = IFACE_WIFI + ".AccessPointNotFound" class PermissionDeniedException(dbus.DBusException): - _dbus_error_name = IFACE_NM + '.PermissionDenied' + _dbus_error_name = IFACE_NM + ".PermissionDenied" class UnknownDeviceException(dbus.DBusException): - _dbus_error_name = IFACE_NM + '.UnknownDevice' + _dbus_error_name = IFACE_NM + ".UnknownDevice" class UnknownConnectionException(dbus.DBusException): - _dbus_error_name = IFACE_NM + '.UnknownConnection' + _dbus_error_name = IFACE_NM + ".UnknownConnection" class InvalidHostnameException(dbus.DBusException): - _dbus_error_name = IFACE_SETTINGS + '.InvalidHostname' + _dbus_error_name = IFACE_SETTINGS + ".InvalidHostname" class NoSecretsException(dbus.DBusException): - _dbus_error_name = IFACE_AGENT_MANAGER + '.NoSecrets' + _dbus_error_name = IFACE_AGENT_MANAGER + ".NoSecrets" class UserCanceledException(dbus.DBusException): - _dbus_error_name = IFACE_AGENT_MANAGER + '.UserCanceled' + _dbus_error_name = IFACE_AGENT_MANAGER + ".UserCanceled" @staticmethod def from_nmerror(e): @@ -412,34 +465,43 @@ class BusErr: raise e2 raise e + ############################################################################### -class NmUtil: +class NmUtil: @staticmethod - def con_hash_to_connection(con_hash, do_verify = False, do_normalize = False): + def con_hash_to_connection(con_hash, do_verify=False, do_normalize=False): x_con = [] for v_setting_name, v_setting in list(con_hash.items()): if isinstance(v_setting_name, (dbus.String, str)): v_setting_name = str(v_setting_name) else: - raise Exception("Expected string dict, but got '%s' key" % (v_setting_name)) + raise Exception( + "Expected string dict, but got '%s' key" % (v_setting_name) + ) x_setting = [] for v_property_name, v_value in list(v_setting.items()): if isinstance(v_property_name, (dbus.String, str)): v_property_name = str(v_property_name) else: - raise Exception("Expected string dict, but got '%s' subkey under %s (%s)" % (v_property_name, v_setting_name, repr(con_hash))) + raise Exception( + "Expected string dict, but got '%s' subkey under %s (%s)" + % (v_property_name, v_setting_name, repr(con_hash)) + ) try: v = Util.variant_from_dbus(v_value) except Exception as e: - raise Exception("Unsupported value %s.%s = %s (%s)" % (v_setting_name, v_property_name, v_value, str(e))) + raise Exception( + "Unsupported value %s.%s = %s (%s)" + % (v_setting_name, v_property_name, v_value, str(e)) + ) x_setting.append((v_property_name, v)) x_con.append((v_setting_name, collections.OrderedDict(x_setting))) - x_con = GLib.Variant('a{sa{sv}}', collections.OrderedDict(x_con)) + x_con = GLib.Variant("a{sa{sv}}", collections.OrderedDict(x_con)) assert GLib.Variant.equal(x_con, Util.variant_from_dbus(con_hash)) @@ -463,30 +525,40 @@ class NmUtil: return con @staticmethod - def con_hash_verify(con_hash, do_verify_strict = True): + def con_hash_verify(con_hash, do_verify_strict=True): if NM.SETTING_CONNECTION_SETTING_NAME not in con_hash: - raise BusErr.MissingSettingException('connection: setting is required') + raise BusErr.MissingSettingException("connection: setting is required") s_con = con_hash[NM.SETTING_CONNECTION_SETTING_NAME] if NM.SETTING_CONNECTION_TYPE not in s_con: - raise BusErr.MissingPropertyException('connection.type: property is required') + raise BusErr.MissingPropertyException( + "connection.type: property is required" + ) if NM.SETTING_CONNECTION_UUID not in s_con: - raise BusErr.MissingPropertyException('connection.uuid: property is required') + raise BusErr.MissingPropertyException( + "connection.uuid: property is required" + ) if NM.SETTING_CONNECTION_ID not in s_con: - raise BusErr.MissingPropertyException('connection.id: property is required') + raise BusErr.MissingPropertyException("connection.id: property is required") if not do_verify_strict: - return; + return t = s_con[NM.SETTING_CONNECTION_TYPE] - if t not in [ NM.SETTING_GSM_SETTING_NAME, - NM.SETTING_VLAN_SETTING_NAME, - NM.SETTING_VPN_SETTING_NAME, - NM.SETTING_WIMAX_SETTING_NAME, - NM.SETTING_WIRED_SETTING_NAME, - NM.SETTING_WIRELESS_SETTING_NAME ]: - raise BusErr.InvalidPropertyException('connection.type: unsupported connection type "%s"' % (t)) + if t not in [ + NM.SETTING_GSM_SETTING_NAME, + NM.SETTING_VLAN_SETTING_NAME, + NM.SETTING_VPN_SETTING_NAME, + NM.SETTING_WIMAX_SETTING_NAME, + NM.SETTING_WIRED_SETTING_NAME, + NM.SETTING_WIRELESS_SETTING_NAME, + ]: + raise BusErr.InvalidPropertyException( + 'connection.type: unsupported connection type "%s"' % (t) + ) try: - con_nm = NmUtil.con_hash_to_connection(con_hash, do_verify = True, do_normalize = True) + con_nm = NmUtil.con_hash_to_connection( + con_hash, do_verify=True, do_normalize=True + ) except Exception as e: BusErr.raise_nmerror(e) @@ -514,14 +586,18 @@ class NmUtil: return s_con[NM.SETTING_CONNECTION_TYPE] return None + ############################################################################### + class ExportedObj(dbus.service.Object): - DBusInterface = collections.namedtuple('DBusInterface', ['dbus_iface', 'props', 'legacy_prop_changed_func']) + DBusInterface = collections.namedtuple( + "DBusInterface", ["dbus_iface", "props", "legacy_prop_changed_func"] + ) @staticmethod - def create_path(klass, path_prefix = None): + def create_path(klass, path_prefix=None): if path_prefix is None: path_prefix = klass.path_prefix path = path_prefix + str(klass.path_counter_next) @@ -530,8 +606,10 @@ class ExportedObj(dbus.service.Object): @staticmethod def to_path_array(src): - return dbus.Array([ExportedObj.to_path(o) for o in src] if src else [], - signature=dbus.Signature('o')) + return dbus.Array( + [ExportedObj.to_path(o) for o in src] if src else [], + signature=dbus.Signature("o"), + ) @staticmethod def to_path(src): @@ -539,7 +617,7 @@ class ExportedObj(dbus.service.Object): return dbus.ObjectPath(src.path) return dbus.ObjectPath("/") - def __init__(self, object_path, ident = None): + def __init__(self, object_path, ident=None): dbus.service.Object.__init__(self) self._dbus_ifaces = {} @@ -564,15 +642,17 @@ class ExportedObj(dbus.service.Object): gl.object_manager.remove_object(self) self.remove_from_connection() - def dbus_interface_add(self, dbus_iface, props, legacy_prop_changed_func = None): - self._dbus_ifaces[dbus_iface] = ExportedObj.DBusInterface(dbus_iface, props, legacy_prop_changed_func) + def dbus_interface_add(self, dbus_iface, props, legacy_prop_changed_func=None): + self._dbus_ifaces[dbus_iface] = ExportedObj.DBusInterface( + dbus_iface, props, legacy_prop_changed_func + ) def _dbus_interface_get(self, dbus_iface): if dbus_iface not in self._dbus_ifaces: raise BusErr.UnknownInterfaceException() return self._dbus_ifaces[dbus_iface] - def _dbus_interface_get_property(self, dbus_interface, propname = None): + def _dbus_interface_get_property(self, dbus_interface, propname=None): props = dbus_interface.props if propname is None: return props @@ -580,33 +660,53 @@ class ExportedObj(dbus.service.Object): raise BusErr.UnknownPropertyException() return props[propname] - def _dbus_property_get(self, dbus_iface, propname = None): - return self._dbus_interface_get_property(self._dbus_interface_get(dbus_iface), - propname) - - def _dbus_property_set(self, dbus_iface, propname, value, allow_detect_dbus_iface = False, dry_run = False, force_update = False): + def _dbus_property_get(self, dbus_iface, propname=None): + return self._dbus_interface_get_property( + self._dbus_interface_get(dbus_iface), propname + ) + + def _dbus_property_set( + self, + dbus_iface, + propname, + value, + allow_detect_dbus_iface=False, + dry_run=False, + force_update=False, + ): if allow_detect_dbus_iface and not dbus_iface: props = None for p, dbus_interface in self._dbus_ifaces.items(): if propname in dbus_interface.props: if props is not None: - raise TestError("Cannot uniquely find the property '%s' on object '%s'" % (propname, self.path)) + raise TestError( + "Cannot uniquely find the property '%s' on object '%s'" + % (propname, self.path) + ) props = dbus_interface.props dbus_iface = p if props is None: - raise TestError("Cannot find the property '%s' on object '%s'" % (propname, self.path)) + raise TestError( + "Cannot find the property '%s' on object '%s'" + % (propname, self.path) + ) else: try: dbus_interface = self._dbus_interface_get(dbus_iface) props = self._dbus_interface_get_property(dbus_interface) except: if dry_run: - raise TestError("No interface '%s' on '%s'" % (dbus_iface, self.path)) + raise TestError( + "No interface '%s' on '%s'" % (dbus_iface, self.path) + ) raise if dry_run: if propname not in props: - raise TestError("No property '%s' on '%s' on '%s'" % (propname, dbus_iface, self.path)) + raise TestError( + "No property '%s' on '%s' on '%s'" + % (propname, dbus_iface, self.path) + ) permission_granted = False @@ -619,7 +719,10 @@ class ExportedObj(dbus.service.Object): permission_granted = True if not permission_granted: - raise TestError("Cannot set property '%s' on '%s' on '%s' via D-Bus" % (propname, dbus_iface, self.path)) + raise TestError( + "Cannot set property '%s' on '%s' on '%s' via D-Bus" + % (propname, dbus_iface, self.path) + ) return @@ -636,7 +739,7 @@ class ExportedObj(dbus.service.Object): dbus_interface = self._dbus_interface_get(dbus_iface) prop = self._dbus_interface_get_property(dbus_interface, propname) if propname is not None: - prop = { propname: prop } + prop = {propname: prop} ExportedObj.PropertiesChanged(self, dbus_iface, prop, []) # the legacy_prop_changed_func signal is a legacy signal that got obsoleted by the standard @@ -648,15 +751,19 @@ class ExportedObj(dbus.service.Object): if dbus_interface.legacy_prop_changed_func is not None: dbus_interface.legacy_prop_changed_func(self, prop) - @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') + @dbus.service.signal(dbus.PROPERTIES_IFACE, signature="sa{sv}as") def PropertiesChanged(self, iface, changed, invalidated): pass - @dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') + @dbus.service.method( + dbus_interface=dbus.PROPERTIES_IFACE, in_signature="s", out_signature="a{sv}" + ) def GetAll(self, dbus_iface): return self._dbus_property_get(dbus_iface) - @dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') + @dbus.service.method( + dbus_interface=dbus.PROPERTIES_IFACE, in_signature="ss", out_signature="v" + ) def Get(self, dbus_iface, name): return self._dbus_property_get(dbus_iface, name) @@ -666,31 +773,33 @@ class ExportedObj(dbus.service.Object): my_ifaces[iface] = self._dbus_ifaces[iface].props return my_ifaces + ############################################################################### -PRP_DEVICE_UDI = "Udi" -PRP_DEVICE_IFACE = "Interface" -PRP_DEVICE_DRIVER = "Driver" -PRP_DEVICE_STATE = "State" -PRP_DEVICE_STATE_REASON = "StateReason" -PRP_DEVICE_ACTIVE_CONNECTION = "ActiveConnection" -PRP_DEVICE_IP4_CONFIG = "Ip4Config" -PRP_DEVICE_IP6_CONFIG = "Ip6Config" -PRP_DEVICE_DHCP4_CONFIG = "Dhcp4Config" -PRP_DEVICE_DHCP6_CONFIG = "Dhcp6Config" -PRP_DEVICE_MANAGED = "Managed" -PRP_DEVICE_AUTOCONNECT = "Autoconnect" -PRP_DEVICE_DEVICE_TYPE = "DeviceType" +PRP_DEVICE_UDI = "Udi" +PRP_DEVICE_IFACE = "Interface" +PRP_DEVICE_DRIVER = "Driver" +PRP_DEVICE_STATE = "State" +PRP_DEVICE_STATE_REASON = "StateReason" +PRP_DEVICE_ACTIVE_CONNECTION = "ActiveConnection" +PRP_DEVICE_IP4_CONFIG = "Ip4Config" +PRP_DEVICE_IP6_CONFIG = "Ip6Config" +PRP_DEVICE_DHCP4_CONFIG = "Dhcp4Config" +PRP_DEVICE_DHCP6_CONFIG = "Dhcp6Config" +PRP_DEVICE_MANAGED = "Managed" +PRP_DEVICE_AUTOCONNECT = "Autoconnect" +PRP_DEVICE_DEVICE_TYPE = "DeviceType" PRP_DEVICE_AVAILABLE_CONNECTIONS = "AvailableConnections" -PRP_DEVICE_LLDP_NEIGHBORS = "LldpNeighbors" -PRP_DEVICE_INTERFACE_FLAGS = "InterfaceFlags" +PRP_DEVICE_LLDP_NEIGHBORS = "LldpNeighbors" +PRP_DEVICE_INTERFACE_FLAGS = "InterfaceFlags" + class Device(ExportedObj): path_counter_next = 1 path_prefix = "/org/freedesktop/NetworkManager/Devices/" - def __init__(self, iface, devtype, ident = None): + def __init__(self, iface, devtype, ident=None): if ident is None: ident = iface @@ -705,145 +814,211 @@ class Device(ExportedObj): self.prp_state = NM.DeviceState.UNAVAILABLE props = { - PRP_DEVICE_UDI: "/sys/devices/virtual/%s" % (iface), - PRP_DEVICE_IFACE: iface, - PRP_DEVICE_DRIVER: "virtual", - PRP_DEVICE_STATE: dbus.UInt32(self.prp_state), - PRP_DEVICE_STATE_REASON: dbus.Struct((dbus.UInt32(self.prp_state), dbus.UInt32(NM.DeviceStateReason.NONE))), - PRP_DEVICE_ACTIVE_CONNECTION: ExportedObj.to_path(None), - PRP_DEVICE_IP4_CONFIG: ExportedObj.to_path(self.ip4_config), - PRP_DEVICE_IP6_CONFIG: ExportedObj.to_path(self.ip6_config), - PRP_DEVICE_DHCP4_CONFIG: ExportedObj.to_path(self.dhcp4_config), - PRP_DEVICE_DHCP6_CONFIG: ExportedObj.to_path(self.dhcp6_config), - PRP_DEVICE_MANAGED: True, - PRP_DEVICE_AUTOCONNECT: True, - PRP_DEVICE_DEVICE_TYPE: dbus.UInt32(devtype), + PRP_DEVICE_UDI: "/sys/devices/virtual/%s" % (iface), + PRP_DEVICE_IFACE: iface, + PRP_DEVICE_DRIVER: "virtual", + PRP_DEVICE_STATE: dbus.UInt32(self.prp_state), + PRP_DEVICE_STATE_REASON: dbus.Struct( + (dbus.UInt32(self.prp_state), dbus.UInt32(NM.DeviceStateReason.NONE)) + ), + PRP_DEVICE_ACTIVE_CONNECTION: ExportedObj.to_path(None), + PRP_DEVICE_IP4_CONFIG: ExportedObj.to_path(self.ip4_config), + PRP_DEVICE_IP6_CONFIG: ExportedObj.to_path(self.ip6_config), + PRP_DEVICE_DHCP4_CONFIG: ExportedObj.to_path(self.dhcp4_config), + PRP_DEVICE_DHCP6_CONFIG: ExportedObj.to_path(self.dhcp6_config), + PRP_DEVICE_MANAGED: True, + PRP_DEVICE_AUTOCONNECT: True, + PRP_DEVICE_DEVICE_TYPE: dbus.UInt32(devtype), PRP_DEVICE_AVAILABLE_CONNECTIONS: ExportedObj.to_path_array([]), - PRP_DEVICE_INTERFACE_FLAGS: dbus.UInt32(3), # up,lower-up - PRP_DEVICE_LLDP_NEIGHBORS: dbus.Array([ - dbus.Dictionary({ - 'chassis-id-type': dbus.UInt32(6), - 'chassis-id': dbus.String('00:11:22:33:44:00'), - 'port-id-type': dbus.UInt32(7), - 'port-id': dbus.String('Uplink port'), - 'port-description': dbus.String('GigabitEthernet #1'), - 'system-name': dbus.String('test1.example.com'), - 'system-description': dbus.String('Test system #1'), - 'system-capabilities': dbus.UInt32(20), - 'destination': dbus.String('nearest-bridge'), - }), - dbus.Dictionary({ - 'chassis-id-type': dbus.UInt32(2), - 'chassis-id': dbus.String('chassis1'), - 'port-id-type': dbus.UInt32(3), - 'port-id': dbus.String('44:44:44:44:44:44'), - 'port-description': dbus.String('GigabitEthernet #2'), - 'system-name': dbus.String('test2.example.com'), - 'system-description': dbus.String('Test system #2'), - 'system-capabilities': dbus.UInt32(2047), - 'destination': dbus.String('nearest-non-tpmr-bridge'), - 'ieee-802-1-vlans': dbus.Array([ - dbus.Dictionary({ - 'vid': dbus.UInt32(80), - 'name': dbus.String('vlan80'), - }, signature = 'sv'), - dbus.Dictionary({ - 'vid': dbus.UInt32(4000), - 'name': dbus.String('My VLAN'), - }, signature = 'sv'), - ]), - 'ieee-802-1-ppvids': dbus.Array([ - dbus.Dictionary({ - 'ppvid': dbus.UInt32(4), - 'flags': dbus.UInt32(0x12), - }, signature = 'sv'), - dbus.Dictionary({ - 'ppvid': dbus.UInt32(10), - 'flags': dbus.UInt32(0x31), - }, signature = 'sv'), - ]), - - }), - dbus.Dictionary({ - 'chassis-id-type': dbus.UInt32(6), - 'chassis-id': dbus.String('00:11:22:33:44:22'), - 'port-id-type': dbus.UInt32(1), - 'port-id': dbus.String('port1'), - 'port-description': dbus.String('GigabitEthernet #3'), - 'system-name': dbus.String('test3.example.com'), - 'system-description': dbus.String('Test system #3'), - 'system-capabilities': dbus.UInt32(40), - 'destination': dbus.String('nearest-customer-bridge'), - 'management-addresses': dbus.Array([ - dbus.Dictionary({ - 'address-subtype': dbus.UInt32(1), - 'address': dbus.ByteArray(b'\xc0\xa8\x01\x01'), - 'interface-number': dbus.UInt32(4), - 'interface-number-subtype': dbus.UInt32(3), - 'object-id': dbus.ByteArray(b'\x01\x02\x03\x04') - }, signature = 'sv'), - dbus.Dictionary({ - 'address-subtype': dbus.UInt32(2), - 'address': dbus.ByteArray(b'\xfd\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x12\x34\x56\x78'), - 'interface-number': dbus.UInt32(1), - 'interface-number-subtype': dbus.UInt32(2), - }, signature = 'sv'), - ]), - 'ieee-802-3-mac-phy-conf': dbus.Dictionary({ - 'autoneg': dbus.UInt32(3), - 'pmd-autoneg-cap': dbus.UInt32(0xfe), - 'operational-mau-type': dbus.UInt32(5), - }, signature = 'sv'), - 'ieee-802-3-power-via-mdi': dbus.Dictionary({ - 'mdi-power-support': dbus.UInt32(7), - 'pse-power-pair': dbus.UInt32(6), - 'power-class': dbus.UInt32(1), - }, signature = 'sv'), - }) - ], 'a{sv}') + PRP_DEVICE_INTERFACE_FLAGS: dbus.UInt32(3), # up,lower-up + PRP_DEVICE_LLDP_NEIGHBORS: dbus.Array( + [ + dbus.Dictionary( + { + "chassis-id-type": dbus.UInt32(6), + "chassis-id": dbus.String("00:11:22:33:44:00"), + "port-id-type": dbus.UInt32(7), + "port-id": dbus.String("Uplink port"), + "port-description": dbus.String("GigabitEthernet #1"), + "system-name": dbus.String("test1.example.com"), + "system-description": dbus.String("Test system #1"), + "system-capabilities": dbus.UInt32(20), + "destination": dbus.String("nearest-bridge"), + } + ), + dbus.Dictionary( + { + "chassis-id-type": dbus.UInt32(2), + "chassis-id": dbus.String("chassis1"), + "port-id-type": dbus.UInt32(3), + "port-id": dbus.String("44:44:44:44:44:44"), + "port-description": dbus.String("GigabitEthernet #2"), + "system-name": dbus.String("test2.example.com"), + "system-description": dbus.String("Test system #2"), + "system-capabilities": dbus.UInt32(2047), + "destination": dbus.String("nearest-non-tpmr-bridge"), + "ieee-802-1-vlans": dbus.Array( + [ + dbus.Dictionary( + { + "vid": dbus.UInt32(80), + "name": dbus.String("vlan80"), + }, + signature="sv", + ), + dbus.Dictionary( + { + "vid": dbus.UInt32(4000), + "name": dbus.String("My VLAN"), + }, + signature="sv", + ), + ] + ), + "ieee-802-1-ppvids": dbus.Array( + [ + dbus.Dictionary( + { + "ppvid": dbus.UInt32(4), + "flags": dbus.UInt32(0x12), + }, + signature="sv", + ), + dbus.Dictionary( + { + "ppvid": dbus.UInt32(10), + "flags": dbus.UInt32(0x31), + }, + signature="sv", + ), + ] + ), + } + ), + dbus.Dictionary( + { + "chassis-id-type": dbus.UInt32(6), + "chassis-id": dbus.String("00:11:22:33:44:22"), + "port-id-type": dbus.UInt32(1), + "port-id": dbus.String("port1"), + "port-description": dbus.String("GigabitEthernet #3"), + "system-name": dbus.String("test3.example.com"), + "system-description": dbus.String("Test system #3"), + "system-capabilities": dbus.UInt32(40), + "destination": dbus.String("nearest-customer-bridge"), + "management-addresses": dbus.Array( + [ + dbus.Dictionary( + { + "address-subtype": dbus.UInt32(1), + "address": dbus.ByteArray( + b"\xc0\xa8\x01\x01" + ), + "interface-number": dbus.UInt32(4), + "interface-number-subtype": dbus.UInt32(3), + "object-id": dbus.ByteArray( + b"\x01\x02\x03\x04" + ), + }, + signature="sv", + ), + dbus.Dictionary( + { + "address-subtype": dbus.UInt32(2), + "address": dbus.ByteArray( + b"\xfd\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x12\x34\x56\x78" + ), + "interface-number": dbus.UInt32(1), + "interface-number-subtype": dbus.UInt32(2), + }, + signature="sv", + ), + ] + ), + "ieee-802-3-mac-phy-conf": dbus.Dictionary( + { + "autoneg": dbus.UInt32(3), + "pmd-autoneg-cap": dbus.UInt32(0xFE), + "operational-mau-type": dbus.UInt32(5), + }, + signature="sv", + ), + "ieee-802-3-power-via-mdi": dbus.Dictionary( + { + "mdi-power-support": dbus.UInt32(7), + "pse-power-pair": dbus.UInt32(6), + "power-class": dbus.UInt32(1), + }, + signature="sv", + ), + } + ), + ], + "a{sv}", + ), } self.dbus_interface_add(IFACE_DEVICE, props, Device.PropertiesChanged) def start(self): self.ip4_config = IP4Config() - self._dbus_property_set(IFACE_DEVICE, PRP_DEVICE_IP4_CONFIG, ExportedObj.to_path(self.ip4_config)) + self._dbus_property_set( + IFACE_DEVICE, PRP_DEVICE_IP4_CONFIG, ExportedObj.to_path(self.ip4_config) + ) self.ip6_config = IP6Config() - self._dbus_property_set(IFACE_DEVICE, PRP_DEVICE_IP6_CONFIG, ExportedObj.to_path(self.ip6_config)) + self._dbus_property_set( + IFACE_DEVICE, PRP_DEVICE_IP6_CONFIG, ExportedObj.to_path(self.ip6_config) + ) self.dhcp4_config = Dhcp4Config() - self._dbus_property_set(IFACE_DEVICE, PRP_DEVICE_DHCP4_CONFIG, ExportedObj.to_path(self.dhcp4_config)) + self._dbus_property_set( + IFACE_DEVICE, + PRP_DEVICE_DHCP4_CONFIG, + ExportedObj.to_path(self.dhcp4_config), + ) self.dhcp6_config = Dhcp6Config() - self._dbus_property_set(IFACE_DEVICE, PRP_DEVICE_DHCP6_CONFIG, ExportedObj.to_path(self.dhcp6_config)) + self._dbus_property_set( + IFACE_DEVICE, + PRP_DEVICE_DHCP6_CONFIG, + ExportedObj.to_path(self.dhcp6_config), + ) def stop(self): - self._dbus_property_set(IFACE_DEVICE, PRP_DEVICE_IP4_CONFIG, ExportedObj.to_path(None)) + self._dbus_property_set( + IFACE_DEVICE, PRP_DEVICE_IP4_CONFIG, ExportedObj.to_path(None) + ) if self.ip4_config is not None: self.ip4_config.unexport() self.ip4_config = None - self._dbus_property_set(IFACE_DEVICE, PRP_DEVICE_IP6_CONFIG, ExportedObj.to_path(None)) + self._dbus_property_set( + IFACE_DEVICE, PRP_DEVICE_IP6_CONFIG, ExportedObj.to_path(None) + ) if self.ip6_config is not None: self.ip6_config.unexport() self.ip6_config = None - self._dbus_property_set(IFACE_DEVICE, PRP_DEVICE_DHCP4_CONFIG, ExportedObj.to_path(None)) + self._dbus_property_set( + IFACE_DEVICE, PRP_DEVICE_DHCP4_CONFIG, ExportedObj.to_path(None) + ) if self.dhcp4_config is not None: self.dhcp4_config.unexport() self.dhcp4_config = None - self._dbus_property_set(IFACE_DEVICE, PRP_DEVICE_DHCP6_CONFIG, ExportedObj.to_path(None)) + self._dbus_property_set( + IFACE_DEVICE, PRP_DEVICE_DHCP6_CONFIG, ExportedObj.to_path(None) + ) if self.dhcp6_config is not None: self.dhcp6_config.unexport() self.dhcp6_config = None - @dbus.service.method(dbus_interface=IFACE_DEVICE, in_signature='', out_signature='') + @dbus.service.method(dbus_interface=IFACE_DEVICE, in_signature="", out_signature="") def Disconnect(self): pass - @dbus.service.method(dbus_interface=IFACE_DEVICE, in_signature='', out_signature='') + @dbus.service.method(dbus_interface=IFACE_DEVICE, in_signature="", out_signature="") def Delete(self): # We don't currently support any software device types, so... raise BusErr.NotSoftwareException() pass - @dbus.service.signal(IFACE_DEVICE, signature='a{sv}') + @dbus.service.signal(IFACE_DEVICE, signature="a{sv}") def PropertiesChanged(self, changed): pass @@ -862,88 +1037,109 @@ class Device(ExportedObj): return False def available_connections_get(self): - return [c for c in gl.settings.get_connections() if self.connection_is_available(c)] + return [ + c for c in gl.settings.get_connections() if self.connection_is_available(c) + ] def available_connections_update(self): - self._dbus_property_set(IFACE_DEVICE, PRP_DEVICE_AVAILABLE_CONNECTIONS, - ExportedObj.to_path_array(self.available_connections_get())) + self._dbus_property_set( + IFACE_DEVICE, + PRP_DEVICE_AVAILABLE_CONNECTIONS, + ExportedObj.to_path_array(self.available_connections_get()), + ) + ############################################################################### -PRP_WIRED_HW_ADDRESS = "HwAddress" -PRP_WIRED_PERM_HW_ADDRESS = "PermHwAddress" -PRP_WIRED_SPEED = "Speed" -PRP_WIRED_CARRIER = "Carrier" +PRP_WIRED_HW_ADDRESS = "HwAddress" +PRP_WIRED_PERM_HW_ADDRESS = "PermHwAddress" +PRP_WIRED_SPEED = "Speed" +PRP_WIRED_CARRIER = "Carrier" PRP_WIRED_S390_SUBCHANNELS = "S390Subchannels" + class WiredDevice(Device): - def __init__(self, iface, mac = None, subchannels = None, ident = None): + def __init__(self, iface, mac=None, subchannels=None, ident=None): Device.__init__(self, iface, NM.DeviceType.ETHERNET, ident) if mac is None: mac = Util.random_mac(self.ident) if subchannels is None: - subchannels = dbus.Array(signature = 's') + subchannels = dbus.Array(signature="s") props = { - PRP_WIRED_HW_ADDRESS: mac, - PRP_WIRED_PERM_HW_ADDRESS: mac, - PRP_WIRED_SPEED: dbus.UInt32(100), - PRP_WIRED_CARRIER: False, + PRP_WIRED_HW_ADDRESS: mac, + PRP_WIRED_PERM_HW_ADDRESS: mac, + PRP_WIRED_SPEED: dbus.UInt32(100), + PRP_WIRED_CARRIER: False, PRP_WIRED_S390_SUBCHANNELS: subchannels, } self.dbus_interface_add(IFACE_WIRED, props, WiredDevice.PropertiesChanged) - @dbus.service.signal(IFACE_WIRED, signature='a{sv}') + @dbus.service.signal(IFACE_WIRED, signature="a{sv}") def PropertiesChanged(self, changed): pass + ############################################################################### PRP_VLAN_HW_ADDRESS = "HwAddress" -PRP_VLAN_CARRIER = "Carrier" -PRP_VLAN_VLAN_ID = "VlanId" +PRP_VLAN_CARRIER = "Carrier" +PRP_VLAN_VLAN_ID = "VlanId" + class VlanDevice(Device): - def __init__(self, iface, ident = None): + def __init__(self, iface, ident=None): Device.__init__(self, iface, NM.DeviceType.VLAN, ident) props = { PRP_VLAN_HW_ADDRESS: Util.random_mac(self.ident), - PRP_VLAN_CARRIER: False, - PRP_VLAN_VLAN_ID: dbus.UInt32(1), + PRP_VLAN_CARRIER: False, + PRP_VLAN_VLAN_ID: dbus.UInt32(1), } self.dbus_interface_add(IFACE_VLAN, props, VlanDevice.PropertiesChanged) - @dbus.service.signal(IFACE_VLAN, signature='a{sv}') + @dbus.service.signal(IFACE_VLAN, signature="a{sv}") def PropertiesChanged(self, changed): pass + ############################################################################### -PRP_WIFI_AP_FLAGS = "Flags" -PRP_WIFI_AP_WPA_FLAGS = "WpaFlags" -PRP_WIFI_AP_RSN_FLAGS = "RsnFlags" -PRP_WIFI_AP_SSID = "Ssid" -PRP_WIFI_AP_FREQUENCY = "Frequency" -PRP_WIFI_AP_HW_ADDRESS = "HwAddress" -PRP_WIFI_AP_MODE = "Mode" +PRP_WIFI_AP_FLAGS = "Flags" +PRP_WIFI_AP_WPA_FLAGS = "WpaFlags" +PRP_WIFI_AP_RSN_FLAGS = "RsnFlags" +PRP_WIFI_AP_SSID = "Ssid" +PRP_WIFI_AP_FREQUENCY = "Frequency" +PRP_WIFI_AP_HW_ADDRESS = "HwAddress" +PRP_WIFI_AP_MODE = "Mode" PRP_WIFI_AP_MAX_BITRATE = "MaxBitrate" -PRP_WIFI_AP_STRENGTH = "Strength" -PRP_WIFI_AP_LAST_SEEN = "LastSeen" +PRP_WIFI_AP_STRENGTH = "Strength" +PRP_WIFI_AP_LAST_SEEN = "LastSeen" + class WifiAp(ExportedObj): path_counter_next = 1 path_prefix = "/org/freedesktop/NetworkManager/AccessPoint/" - def __init__(self, ssid, bssid = None, flags = None, wpaf = None, rsnf = None, freq = None, strength = None, ident = None): + def __init__( + self, + ssid, + bssid=None, + flags=None, + wpaf=None, + rsnf=None, + freq=None, + strength=None, + ident=None, + ): ExportedObj.__init__(self, ExportedObj.create_path(WifiAp), ident) - NM_AP_FLAGS = getattr(NM, '80211ApSecurityFlags') + NM_AP_FLAGS = getattr(NM, "80211ApSecurityFlags") if flags is None: flags = 0x1 if wpaf is None: @@ -970,24 +1166,25 @@ class WifiAp(ExportedObj): self.ssid = ssid props = { - PRP_WIFI_AP_FLAGS: dbus.UInt32(flags), - PRP_WIFI_AP_WPA_FLAGS: dbus.UInt32(wpaf), - PRP_WIFI_AP_RSN_FLAGS: dbus.UInt32(rsnf), - PRP_WIFI_AP_SSID: dbus.ByteArray(self.ssid.encode('utf-8')), - PRP_WIFI_AP_FREQUENCY: dbus.UInt32(freq), - PRP_WIFI_AP_HW_ADDRESS: bssid, - PRP_WIFI_AP_MODE: dbus.UInt32(getattr(NM,'80211Mode').INFRA), + PRP_WIFI_AP_FLAGS: dbus.UInt32(flags), + PRP_WIFI_AP_WPA_FLAGS: dbus.UInt32(wpaf), + PRP_WIFI_AP_RSN_FLAGS: dbus.UInt32(rsnf), + PRP_WIFI_AP_SSID: dbus.ByteArray(self.ssid.encode("utf-8")), + PRP_WIFI_AP_FREQUENCY: dbus.UInt32(freq), + PRP_WIFI_AP_HW_ADDRESS: bssid, + PRP_WIFI_AP_MODE: dbus.UInt32(getattr(NM, "80211Mode").INFRA), PRP_WIFI_AP_MAX_BITRATE: dbus.UInt32(54000), - PRP_WIFI_AP_STRENGTH: dbus.Byte(strength), - PRP_WIFI_AP_LAST_SEEN: dbus.Int32(NM.utils_get_timestamp_msec() / 1000), + PRP_WIFI_AP_STRENGTH: dbus.Byte(strength), + PRP_WIFI_AP_LAST_SEEN: dbus.Int32(NM.utils_get_timestamp_msec() / 1000), } self.dbus_interface_add(IFACE_WIFI_AP, props, WifiAp.PropertiesChanged) - @dbus.service.signal(IFACE_WIFI_AP, signature='a{sv}') + @dbus.service.signal(IFACE_WIFI_AP, signature="a{sv}") def PropertiesChanged(self, changed): pass + ############################################################################### PRP_WIFI_HW_ADDRESS = "HwAddress" @@ -999,8 +1196,9 @@ PRP_WIFI_ACTIVE_ACCESS_POINT = "ActiveAccessPoint" PRP_WIFI_WIRELESS_CAPABILITIES = "WirelessCapabilities" PRP_WIFI_LAST_SCAN = "LastScan" + class WifiDevice(Device): - def __init__(self, iface, mac = None, ident = None): + def __init__(self, iface, mac=None, ident=None): Device.__init__(self, iface, NM.DeviceType.WIFI, ident) if mac is None: @@ -1010,58 +1208,70 @@ class WifiDevice(Device): self.scan_cb_id = None # Use a randomly older timestamp to trigger RequestScan() from the client - ts = max(0, NM.utils_get_timestamp_msec() - Util.random_int(self.path, 20000, 40000)) + ts = max( + 0, NM.utils_get_timestamp_msec() - Util.random_int(self.path, 20000, 40000) + ) props = { - PRP_WIFI_HW_ADDRESS: mac, - PRP_WIFI_PERM_HW_ADDRESS: mac, - PRP_WIFI_MODE: dbus.UInt32(getattr(NM,'80211Mode').INFRA), - PRP_WIFI_BITRATE: dbus.UInt32(21000), + PRP_WIFI_HW_ADDRESS: mac, + PRP_WIFI_PERM_HW_ADDRESS: mac, + PRP_WIFI_MODE: dbus.UInt32(getattr(NM, "80211Mode").INFRA), + PRP_WIFI_BITRATE: dbus.UInt32(21000), PRP_WIFI_WIRELESS_CAPABILITIES: dbus.UInt32(0xFF), - PRP_WIFI_ACCESS_POINTS: ExportedObj.to_path_array(self.aps), - PRP_WIFI_ACTIVE_ACCESS_POINT: ExportedObj.to_path(None), - PRP_WIFI_LAST_SCAN: dbus.Int64(ts), + PRP_WIFI_ACCESS_POINTS: ExportedObj.to_path_array(self.aps), + PRP_WIFI_ACTIVE_ACCESS_POINT: ExportedObj.to_path(None), + PRP_WIFI_LAST_SCAN: dbus.Int64(ts), } self.dbus_interface_add(IFACE_WIFI, props, WifiDevice.PropertiesChanged) - @dbus.service.method(dbus_interface=IFACE_WIFI, in_signature='', out_signature='ao') + @dbus.service.method(dbus_interface=IFACE_WIFI, in_signature="", out_signature="ao") def GetAccessPoints(self): # only include non-hidden APs return ExportedObj.to_path_array([a for a in self.aps if a.ssid]) - @dbus.service.method(dbus_interface=IFACE_WIFI, in_signature='', out_signature='ao') + @dbus.service.method(dbus_interface=IFACE_WIFI, in_signature="", out_signature="ao") def GetAllAccessPoints(self): # include all APs including hidden ones return ExportedObj.to_path_array(self.aps) - @dbus.service.method(dbus_interface=IFACE_WIFI, in_signature='a{sv}', out_signature='') + @dbus.service.method( + dbus_interface=IFACE_WIFI, in_signature="a{sv}", out_signature="" + ) def RequestScan(self, props): self.scan_cb_id = Util.g_source_remove(self.scan_cb_id) + def cb(): ts = NM.utils_get_timestamp_msec() for ap in self.aps: - ap._dbus_property_set(IFACE_WIFI_AP, PRP_WIFI_AP_LAST_SEEN, dbus.Int32(ts / 1000)) + ap._dbus_property_set( + IFACE_WIFI_AP, PRP_WIFI_AP_LAST_SEEN, dbus.Int32(ts / 1000) + ) self._dbus_property_set(IFACE_WIFI, PRP_WIFI_LAST_SCAN, dbus.Int64(ts)) self.scan_cb_id = None return False + self.scan_cb_id = GLib.idle_add(cb) pass - @dbus.service.signal(IFACE_WIFI, signature='o') + @dbus.service.signal(IFACE_WIFI, signature="o") def AccessPointAdded(self, ap_path): pass def add_ap(self, ap): ap.export() self.aps.append(ap) - self._dbus_property_set(IFACE_WIFI, PRP_WIFI_ACCESS_POINTS, ExportedObj.to_path_array(self.aps)) + self._dbus_property_set( + IFACE_WIFI, PRP_WIFI_ACCESS_POINTS, ExportedObj.to_path_array(self.aps) + ) self.AccessPointAdded(ExportedObj.to_path(ap)) return ap def remove_ap(self, ap): self.aps.remove(ap) - self._dbus_property_set(IFACE_WIFI, PRP_WIFI_ACCESS_POINTS, ExportedObj.to_path_array(self.aps)) + self._dbus_property_set( + IFACE_WIFI, PRP_WIFI_ACCESS_POINTS, ExportedObj.to_path_array(self.aps) + ) self.AccessPointRemoved(ExportedObj.to_path(ap)) ap.unexport() @@ -1069,11 +1279,11 @@ class WifiDevice(Device): self.scan_cb_id = Util.g_source_remove(self.scan_cb_id) super(WifiDevice, self).stop() - @dbus.service.signal(IFACE_WIFI, signature='o') + @dbus.service.signal(IFACE_WIFI, signature="o") def AccessPointRemoved(self, ap_path): pass - @dbus.service.signal(IFACE_WIFI, signature='a{sv}') + @dbus.service.signal(IFACE_WIFI, signature="a{sv}") def PropertiesChanged(self, changed): pass @@ -1103,8 +1313,9 @@ PRP_ACTIVE_CONNECTION_DHCP6CONFIG = "Dhcp6Config" PRP_ACTIVE_CONNECTION_VPN = "Vpn" PRP_ACTIVE_CONNECTION_MASTER = "Master" -PRP_VPN_CONNECTION_VPN_STATE = 'VpnState' -PRP_VPN_CONNECTION_BANNER = 'Banner' +PRP_VPN_CONNECTION_VPN_STATE = "VpnState" +PRP_VPN_CONNECTION_BANNER = "Banner" + class ActiveConnection(ExportedObj): @@ -1124,37 +1335,45 @@ class ActiveConnection(ExportedObj): s_con = con_inst.con_hash[NM.SETTING_CONNECTION_SETTING_NAME] props = { - PRP_ACTIVE_CONNECTION_CONNECTION: ExportedObj.to_path(con_inst), + PRP_ACTIVE_CONNECTION_CONNECTION: ExportedObj.to_path(con_inst), PRP_ACTIVE_CONNECTION_SPECIFIC_OBJECT: ExportedObj.to_path(specific_object), - PRP_ACTIVE_CONNECTION_ID: s_con[NM.SETTING_CONNECTION_ID], - PRP_ACTIVE_CONNECTION_UUID: s_con[NM.SETTING_CONNECTION_UUID], - PRP_ACTIVE_CONNECTION_TYPE: s_con[NM.SETTING_CONNECTION_TYPE], - PRP_ACTIVE_CONNECTION_DEVICES: ExportedObj.to_path_array([self.device]), - PRP_ACTIVE_CONNECTION_STATE: dbus.UInt32(NM.ActiveConnectionState.UNKNOWN), - PRP_ACTIVE_CONNECTION_DEFAULT: False, - PRP_ACTIVE_CONNECTION_IP4CONFIG: ExportedObj.to_path(None), - PRP_ACTIVE_CONNECTION_DHCP4CONFIG: ExportedObj.to_path(None), - PRP_ACTIVE_CONNECTION_DEFAULT6: False, - PRP_ACTIVE_CONNECTION_IP6CONFIG: ExportedObj.to_path(None), - PRP_ACTIVE_CONNECTION_DHCP6CONFIG: ExportedObj.to_path(None), - PRP_ACTIVE_CONNECTION_VPN: self.is_vpn, - PRP_ACTIVE_CONNECTION_MASTER: ExportedObj.to_path(None), + PRP_ACTIVE_CONNECTION_ID: s_con[NM.SETTING_CONNECTION_ID], + PRP_ACTIVE_CONNECTION_UUID: s_con[NM.SETTING_CONNECTION_UUID], + PRP_ACTIVE_CONNECTION_TYPE: s_con[NM.SETTING_CONNECTION_TYPE], + PRP_ACTIVE_CONNECTION_DEVICES: ExportedObj.to_path_array([self.device]), + PRP_ACTIVE_CONNECTION_STATE: dbus.UInt32(NM.ActiveConnectionState.UNKNOWN), + PRP_ACTIVE_CONNECTION_DEFAULT: False, + PRP_ACTIVE_CONNECTION_IP4CONFIG: ExportedObj.to_path(None), + PRP_ACTIVE_CONNECTION_DHCP4CONFIG: ExportedObj.to_path(None), + PRP_ACTIVE_CONNECTION_DEFAULT6: False, + PRP_ACTIVE_CONNECTION_IP6CONFIG: ExportedObj.to_path(None), + PRP_ACTIVE_CONNECTION_DHCP6CONFIG: ExportedObj.to_path(None), + PRP_ACTIVE_CONNECTION_VPN: self.is_vpn, + PRP_ACTIVE_CONNECTION_MASTER: ExportedObj.to_path(None), } - self.dbus_interface_add(IFACE_ACTIVE_CONNECTION, props, ActiveConnection.PropertiesChanged) + self.dbus_interface_add( + IFACE_ACTIVE_CONNECTION, props, ActiveConnection.PropertiesChanged + ) if self.is_vpn: props = { - PRP_VPN_CONNECTION_VPN_STATE: dbus.UInt32(NM.VpnConnectionState.UNKNOWN), - PRP_VPN_CONNECTION_BANNER: '*** VPN connection %s ***' % (con_inst.get_id()), + PRP_VPN_CONNECTION_VPN_STATE: dbus.UInt32( + NM.VpnConnectionState.UNKNOWN + ), + PRP_VPN_CONNECTION_BANNER: "*** VPN connection %s ***" + % (con_inst.get_id()), } - self.dbus_interface_add(IFACE_VPN_CONNECTION, props, ActiveConnection.VpnPropertiesChanged) - + self.dbus_interface_add( + IFACE_VPN_CONNECTION, props, ActiveConnection.VpnPropertiesChanged + ) def _set_state(self, state, reason): state = dbus.UInt32(state) - self._dbus_property_set(IFACE_ACTIVE_CONNECTION, PRP_ACTIVE_CONNECTION_STATE, state) + self._dbus_property_set( + IFACE_ACTIVE_CONNECTION, PRP_ACTIVE_CONNECTION_STATE, state + ) self.StateChanged(state, dbus.UInt32(reason)) def activation_cancel(self): @@ -1163,55 +1382,62 @@ class ActiveConnection(ExportedObj): def _activation_step2(self): assert self._activation_id is not None self._activation_id = None - self._set_state(NM.ActiveConnectionState.ACTIVATED, NM.ActiveConnectionStateReason.UNKNOWN) + self._set_state( + NM.ActiveConnectionState.ACTIVATED, NM.ActiveConnectionStateReason.UNKNOWN + ) return False def _activation_step1(self): assert self._activation_id is not None self._activation_id = GLib.timeout_add(50, self._activation_step2) self.device.set_active_connection(self) - self._set_state(NM.ActiveConnectionState.ACTIVATING, NM.ActiveConnectionStateReason.UNKNOWN) + self._set_state( + NM.ActiveConnectionState.ACTIVATING, NM.ActiveConnectionStateReason.UNKNOWN + ) return False def start_activation(self): assert self._activation_id is None self._activation_id = GLib.timeout_add(50, self._activation_step1) - @dbus.service.signal(IFACE_VPN_CONNECTION, signature='a{sv}') + @dbus.service.signal(IFACE_VPN_CONNECTION, signature="a{sv}") def PropertiesChanged(self, changed): pass + VpnPropertiesChanged = PropertiesChanged - @dbus.service.signal(IFACE_ACTIVE_CONNECTION, signature='a{sv}') + @dbus.service.signal(IFACE_ACTIVE_CONNECTION, signature="a{sv}") def PropertiesChanged(self, changed): pass - @dbus.service.signal(IFACE_ACTIVE_CONNECTION, signature='uu') + @dbus.service.signal(IFACE_ACTIVE_CONNECTION, signature="uu") def StateChanged(self, state, reason): pass - @dbus.service.signal(IFACE_VPN_CONNECTION, signature='uu') + @dbus.service.signal(IFACE_VPN_CONNECTION, signature="uu") def VpnStateChanged(self, state, reason): pass + ############################################################################### -PRP_NM_DEVICES = 'Devices' -PRP_NM_ALL_DEVICES = 'AllDevices' -PRP_NM_NETWORKING_ENABLED = 'NetworkingEnabled' -PRP_NM_WWAN_ENABLED = 'WwanEnabled' -PRP_NM_WWAN_HARDWARE_ENABLED = 'WwanHardwareEnabled' -PRP_NM_WIRELESS_ENABLED = 'WirelessEnabled' -PRP_NM_WIRELESS_HARDWARE_ENABLED = 'WirelessHardwareEnabled' -PRP_NM_WIMAX_ENABLED = 'WimaxEnabled' -PRP_NM_WIMAX_HARDWARE_ENABLED = 'WimaxHardwareEnabled' -PRP_NM_ACTIVE_CONNECTIONS = 'ActiveConnections' -PRP_NM_PRIMARY_CONNECTION = 'PrimaryConnection' -PRP_NM_ACTIVATING_CONNECTION = 'ActivatingConnection' -PRP_NM_STARTUP = 'Startup' -PRP_NM_STATE = 'State' -PRP_NM_VERSION = 'Version' -PRP_NM_CONNECTIVITY = 'Connectivity' +PRP_NM_DEVICES = "Devices" +PRP_NM_ALL_DEVICES = "AllDevices" +PRP_NM_NETWORKING_ENABLED = "NetworkingEnabled" +PRP_NM_WWAN_ENABLED = "WwanEnabled" +PRP_NM_WWAN_HARDWARE_ENABLED = "WwanHardwareEnabled" +PRP_NM_WIRELESS_ENABLED = "WirelessEnabled" +PRP_NM_WIRELESS_HARDWARE_ENABLED = "WirelessHardwareEnabled" +PRP_NM_WIMAX_ENABLED = "WimaxEnabled" +PRP_NM_WIMAX_HARDWARE_ENABLED = "WimaxHardwareEnabled" +PRP_NM_ACTIVE_CONNECTIONS = "ActiveConnections" +PRP_NM_PRIMARY_CONNECTION = "PrimaryConnection" +PRP_NM_ACTIVATING_CONNECTION = "ActivatingConnection" +PRP_NM_STARTUP = "Startup" +PRP_NM_STATE = "State" +PRP_NM_VERSION = "Version" +PRP_NM_CONNECTIVITY = "Connectivity" + class NetworkManager(ExportedObj): def __init__(self): @@ -1220,28 +1446,30 @@ class NetworkManager(ExportedObj): self.active_connections = [] props = { - PRP_NM_DEVICES: ExportedObj.to_path_array(self.devices), - PRP_NM_ALL_DEVICES: ExportedObj.to_path_array(self.devices), - PRP_NM_NETWORKING_ENABLED: True, - PRP_NM_WWAN_ENABLED: True, - PRP_NM_WWAN_HARDWARE_ENABLED: True, - PRP_NM_WIRELESS_ENABLED: True, + PRP_NM_DEVICES: ExportedObj.to_path_array(self.devices), + PRP_NM_ALL_DEVICES: ExportedObj.to_path_array(self.devices), + PRP_NM_NETWORKING_ENABLED: True, + PRP_NM_WWAN_ENABLED: True, + PRP_NM_WWAN_HARDWARE_ENABLED: True, + PRP_NM_WIRELESS_ENABLED: True, PRP_NM_WIRELESS_HARDWARE_ENABLED: True, - PRP_NM_WIMAX_ENABLED: True, - PRP_NM_WIMAX_HARDWARE_ENABLED: True, - PRP_NM_ACTIVE_CONNECTIONS: ExportedObj.to_path_array(self.active_connections), - PRP_NM_PRIMARY_CONNECTION: ExportedObj.to_path(None), - PRP_NM_ACTIVATING_CONNECTION: ExportedObj.to_path(None), - PRP_NM_STARTUP: False, - PRP_NM_STATE: dbus.UInt32(NM.State.DISCONNECTED), - PRP_NM_VERSION: "0.9.9.0", - PRP_NM_CONNECTIVITY: dbus.UInt32(NM.ConnectivityState.NONE), + PRP_NM_WIMAX_ENABLED: True, + PRP_NM_WIMAX_HARDWARE_ENABLED: True, + PRP_NM_ACTIVE_CONNECTIONS: ExportedObj.to_path_array( + self.active_connections + ), + PRP_NM_PRIMARY_CONNECTION: ExportedObj.to_path(None), + PRP_NM_ACTIVATING_CONNECTION: ExportedObj.to_path(None), + PRP_NM_STARTUP: False, + PRP_NM_STATE: dbus.UInt32(NM.State.DISCONNECTED), + PRP_NM_VERSION: "0.9.9.0", + PRP_NM_CONNECTIVITY: dbus.UInt32(NM.ConnectivityState.NONE), } self.dbus_interface_add(IFACE_NM, props, NetworkManager.PropertiesChanged) self.export() - @dbus.service.signal(IFACE_NM, signature='u') + @dbus.service.signal(IFACE_NM, signature="u") def StateChanged(self, new_state): pass @@ -1249,20 +1477,22 @@ class NetworkManager(ExportedObj): self._dbus_property_set(IFACE_NM, PRP_NM_STATE, state) self.StateChanged(dbus.UInt32(self.state)) - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='', out_signature='ao') + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="", out_signature="ao") def GetDevices(self): return ExportedObj.to_path_array(self.devices) - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='', out_signature='ao') + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="", out_signature="ao") def GetAllDevices(self): return ExportedObj.to_path_array(self.devices) - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='s', out_signature='o') + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="s", out_signature="o") def GetDeviceByIpIface(self, ip_iface): - d = self.find_device_first(ip_iface = ip_iface, require = BusErr.UnknownDeviceException) + d = self.find_device_first( + ip_iface=ip_iface, require=BusErr.UnknownDeviceException + ) return ExportedObj.to_path(d) - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='ooo', out_signature='o') + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="ooo", out_signature="o") def ActivateConnection(self, conpath, devpath, specific_object): try: con_inst = gl.settings.get_connection(conpath) @@ -1272,14 +1502,14 @@ class NetworkManager(ExportedObj): con_hash = con_inst.con_hash con_type = NmUtil.con_hash_get_type(con_hash) - device = self.find_device_first(path = devpath) + device = self.find_device_first(path=devpath) if not device: if con_type == NM.SETTING_WIRED_SETTING_NAME: - device = self.find_device_first(dev_type = WiredDevice) + device = self.find_device_first(dev_type=WiredDevice) elif con_type == NM.SETTING_WIRELESS_SETTING_NAME: - device = self.find_device_first(dev_type = WifiDevice) + device = self.find_device_first(dev_type=WifiDevice) elif con_type == NM.SETTING_VLAN_SETTING_NAME: - ifname = con_hash[NM.SETTING_CONNECTION_SETTING_NAME]['interface-name'] + ifname = con_hash[NM.SETTING_CONNECTION_SETTING_NAME]["interface-name"] device = VlanDevice(ifname) self.add_device(device) elif con_type == NM.SETTING_VPN_SETTING_NAME: @@ -1291,19 +1521,23 @@ class NetworkManager(ExportedObj): break if not device: - raise BusErr.UnknownDeviceException("No device found for the requested iface.") + raise BusErr.UnknownDeviceException( + "No device found for the requested iface." + ) # See if we need secrets. For the moment, we only support WPA - if '802-11-wireless-security' in con_hash: - s_wsec = con_hash['802-11-wireless-security'] - if (s_wsec['key-mgmt'] == 'wpa-psk' and 'psk' not in s_wsec): - secrets = gl.agent_manager.get_secrets(con_hash, conpath, '802-11-wireless-security') + if "802-11-wireless-security" in con_hash: + s_wsec = con_hash["802-11-wireless-security"] + if s_wsec["key-mgmt"] == "wpa-psk" and "psk" not in s_wsec: + secrets = gl.agent_manager.get_secrets( + con_hash, conpath, "802-11-wireless-security" + ) if secrets is None: raise BusErr.NoSecretsException("No secret agent available") - if '802-11-wireless-security' not in secrets: + if "802-11-wireless-security" not in secrets: raise BusErr.NoSecretsException("No secrets provided") - s_wsec = secrets['802-11-wireless-security'] - if 'psk' not in s_wsec: + s_wsec = secrets["802-11-wireless-security"] + if "psk" not in s_wsec: raise BusErr.NoSecretsException("No secrets provided") ac = ActiveConnection(device, con_inst, None) @@ -1313,31 +1547,49 @@ class NetworkManager(ExportedObj): def active_connection_add(self, ac): ac.export() self.active_connections.append(ac) - self._dbus_property_set(IFACE_NM, PRP_NM_ACTIVE_CONNECTIONS, ExportedObj.to_path_array(self.active_connections)) + self._dbus_property_set( + IFACE_NM, + PRP_NM_ACTIVE_CONNECTIONS, + ExportedObj.to_path_array(self.active_connections), + ) ac.start_activation() def active_connection_remove(self, ac): ac.activation_cancel() self.active_connections.remove(ac) - self._dbus_property_set(IFACE_NM, PRP_NM_ACTIVE_CONNECTIONS, ExportedObj.to_path_array(self.active_connections)) + self._dbus_property_set( + IFACE_NM, + PRP_NM_ACTIVE_CONNECTIONS, + ExportedObj.to_path_array(self.active_connections), + ) ac.unexport() - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='a{sa{sv}}oo', out_signature='oo') + @dbus.service.method( + dbus_interface=IFACE_NM, in_signature="a{sa{sv}}oo", out_signature="oo" + ) def AddAndActivateConnection(self, con_hash, devpath, specific_object): - conpath, acpath, result = self.AddAndActivateConnection2(con_hash, devpath, specific_object, dict()) + conpath, acpath, result = self.AddAndActivateConnection2( + con_hash, devpath, specific_object, dict() + ) return (conpath, acpath) - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='a{sa{sv}}ooa{sv}', out_signature='ooa{sv}') + @dbus.service.method( + dbus_interface=IFACE_NM, + in_signature="a{sa{sv}}ooa{sv}", + out_signature="ooa{sv}", + ) def AddAndActivateConnection2(self, con_hash, devpath, specific_object, options): - device = self.find_device_first(path = devpath, require = BusErr.UnknownDeviceException) + device = self.find_device_first( + path=devpath, require=BusErr.UnknownDeviceException + ) conpath = gl.settings.AddConnection(con_hash) return (conpath, self.ActivateConnection(conpath, devpath, specific_object), []) - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='o', out_signature='') + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="o", out_signature="") def DeactivateConnection(self, active_connection): pass - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='b', out_signature='') + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="b", out_signature="") def Sleep(self, do_sleep): if do_sleep: state = NM.State.ASLEEP @@ -1345,44 +1597,54 @@ class NetworkManager(ExportedObj): state = NM.State.DISCONNECTED self.set_state(state) - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='b', out_signature='') + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="b", out_signature="") def Enable(self, do_enable): pass - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='', out_signature='a{ss}') + @dbus.service.method( + dbus_interface=IFACE_NM, in_signature="", out_signature="a{ss}" + ) def GetPermissions(self): - return { "org.freedesktop.NetworkManager.enable-disable-network": "yes", - "org.freedesktop.NetworkManager.sleep-wake": "no", - "org.freedesktop.NetworkManager.enable-disable-wifi": "yes", - "org.freedesktop.NetworkManager.enable-disable-wwan": "yes", - "org.freedesktop.NetworkManager.enable-disable-wimax": "yes", - "org.freedesktop.NetworkManager.network-control": "yes", - "org.freedesktop.NetworkManager.wifi.share.protected": "yes", - "org.freedesktop.NetworkManager.wifi.share.open": "yes", - "org.freedesktop.NetworkManager.settings.modify.own": "yes", - "org.freedesktop.NetworkManager.settings.modify.system": "yes", - "org.freedesktop.NetworkManager.settings.modify.hostname": "yes", - "org.freedesktop.NetworkManager.settings.modify.global-dns": "no", - "org.freedesktop.NetworkManager.reload": "no", - } - - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='ss', out_signature='') + return { + "org.freedesktop.NetworkManager.enable-disable-network": "yes", + "org.freedesktop.NetworkManager.sleep-wake": "no", + "org.freedesktop.NetworkManager.enable-disable-wifi": "yes", + "org.freedesktop.NetworkManager.enable-disable-wwan": "yes", + "org.freedesktop.NetworkManager.enable-disable-wimax": "yes", + "org.freedesktop.NetworkManager.network-control": "yes", + "org.freedesktop.NetworkManager.wifi.share.protected": "yes", + "org.freedesktop.NetworkManager.wifi.share.open": "yes", + "org.freedesktop.NetworkManager.settings.modify.own": "yes", + "org.freedesktop.NetworkManager.settings.modify.system": "yes", + "org.freedesktop.NetworkManager.settings.modify.hostname": "yes", + "org.freedesktop.NetworkManager.settings.modify.global-dns": "no", + "org.freedesktop.NetworkManager.reload": "no", + } + + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="ss", out_signature="") def SetLogging(self, level, domains): pass - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='', out_signature='ss') + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="", out_signature="ss") def GetLogging(self): return ("info", "HW,RFKILL,CORE,DEVICE,WIFI,ETHER") - @dbus.service.method(dbus_interface=IFACE_NM, in_signature='', out_signature='u') + @dbus.service.method(dbus_interface=IFACE_NM, in_signature="", out_signature="u") def CheckConnectivity(self): raise BusErr.PermissionDeniedException("You fail") - @dbus.service.signal(IFACE_NM, signature='o') + @dbus.service.signal(IFACE_NM, signature="o") def DeviceAdded(self, devpath): pass - def find_devices(self, ident = _DEFAULT_ARG, path = _DEFAULT_ARG, iface = _DEFAULT_ARG, ip_iface = _DEFAULT_ARG, dev_type = _DEFAULT_ARG): + def find_devices( + self, + ident=_DEFAULT_ARG, + path=_DEFAULT_ARG, + iface=_DEFAULT_ARG, + ip_iface=_DEFAULT_ARG, + dev_type=_DEFAULT_ARG, + ): r = None for d in self.devices: if ident is not _DEFAULT_ARG: @@ -1403,24 +1665,40 @@ class NetworkManager(ExportedObj): continue yield d - def find_device_first(self, ident = _DEFAULT_ARG, path = _DEFAULT_ARG, iface = _DEFAULT_ARG, ip_iface = _DEFAULT_ARG, dev_type = _DEFAULT_ARG, require = None): + def find_device_first( + self, + ident=_DEFAULT_ARG, + path=_DEFAULT_ARG, + iface=_DEFAULT_ARG, + ip_iface=_DEFAULT_ARG, + dev_type=_DEFAULT_ARG, + require=None, + ): r = None - for d in self.find_devices(ident = ident, path = path, iface = iface, ip_iface = ip_iface, dev_type = dev_type): + for d in self.find_devices( + ident=ident, path=path, iface=iface, ip_iface=ip_iface, dev_type=dev_type + ): r = d break if r is None and require: if require is TestError: - raise TestError('Device not found') - raise BusErr.UnknownDeviceException('Device not found') + raise TestError("Device not found") + raise BusErr.UnknownDeviceException("Device not found") return r def add_device(self, device): - if self.find_device_first(ident = device.ident, path = device.path) is not None: - raise TestError("Duplicate device ident=%s / path=%s" % (device.ident, device.path)) + if self.find_device_first(ident=device.ident, path=device.path) is not None: + raise TestError( + "Duplicate device ident=%s / path=%s" % (device.ident, device.path) + ) device.export() self.devices.append(device) - self._dbus_property_set(IFACE_NM, PRP_NM_DEVICES, ExportedObj.to_path_array(self.devices)) - self._dbus_property_set(IFACE_NM, PRP_NM_ALL_DEVICES, ExportedObj.to_path_array(self.devices)) + self._dbus_property_set( + IFACE_NM, PRP_NM_DEVICES, ExportedObj.to_path_array(self.devices) + ) + self._dbus_property_set( + IFACE_NM, PRP_NM_ALL_DEVICES, ExportedObj.to_path_array(self.devices) + ) self.DeviceAdded(ExportedObj.to_path(device)) device.start() return device @@ -1428,8 +1706,12 @@ class NetworkManager(ExportedObj): def remove_device(self, device): device.stop() self.devices.remove(device) - self._dbus_property_set(IFACE_NM, PRP_NM_DEVICES, ExportedObj.to_path_array(self.devices)) - self._dbus_property_set(IFACE_NM, PRP_NM_ALL_DEVICES, ExportedObj.to_path_array(self.devices)) + self._dbus_property_set( + IFACE_NM, PRP_NM_DEVICES, ExportedObj.to_path_array(self.devices) + ) + self._dbus_property_set( + IFACE_NM, PRP_NM_ALL_DEVICES, ExportedObj.to_path_array(self.devices) + ) self.DeviceRemoved(ExportedObj.to_path(device)) device.unexport() @@ -1437,23 +1719,26 @@ class NetworkManager(ExportedObj): for d in self.devices: d.available_connections_update() - @dbus.service.signal(IFACE_NM, signature='o') + @dbus.service.signal(IFACE_NM, signature="o") def DeviceRemoved(self, devpath): pass - @dbus.service.signal(IFACE_NM, signature='a{sv}') + @dbus.service.signal(IFACE_NM, signature="a{sv}") def PropertiesChanged(self, changed): pass - @dbus.service.method(IFACE_TEST, in_signature='', out_signature='') + @dbus.service.method(IFACE_TEST, in_signature="", out_signature="") def Quit(self): gl.mainloop.quit() - @dbus.service.method(IFACE_TEST, in_signature='a{ss}', out_signature='a(sss)') + @dbus.service.method(IFACE_TEST, in_signature="a{ss}", out_signature="a(sss)") def FindConnections(self, selector_args): - return [(c.path, c.get_uuid(), c.get_id()) for c in gl.settings.find_connections(**selector_args)] + return [ + (c.path, c.get_uuid(), c.get_id()) + for c in gl.settings.find_connections(**selector_args) + ] - @dbus.service.method(IFACE_TEST, in_signature='a(oa(sa(sv)))', out_signature='') + @dbus.service.method(IFACE_TEST, in_signature="a(oa(sa(sv)))", out_signature="") def SetProperties(self, all_args): for i in [0, 1]: for path, iface_args in all_args: @@ -1462,73 +1747,82 @@ class NetworkManager(ExportedObj): raise TestError("Object %s does not exist" % (path)) for iface_name, args in iface_args: for propname, value in args: - o._dbus_property_set(iface_name, propname, value, - allow_detect_dbus_iface = True, - dry_run = (i == 0)) - - - @dbus.service.method(IFACE_TEST, in_signature='sa{sv}', out_signature='o') + o._dbus_property_set( + iface_name, + propname, + value, + allow_detect_dbus_iface=True, + dry_run=(i == 0), + ) + + @dbus.service.method(IFACE_TEST, in_signature="sa{sv}", out_signature="o") def AddObj(self, class_name, args): - if class_name in ['WiredDevice', 'WifiDevice']: + if class_name in ["WiredDevice", "WifiDevice"]: py_class = globals()[class_name] d = py_class(**args) return ExportedObj.to_path(self.add_device(d)) - elif class_name in ['WifiAp']: - if 'device' not in args: + elif class_name in ["WifiAp"]: + if "device" not in args: raise TestError('missing "device" paramter') - d = self.find_device_first(ident = args['device'], require = TestError) - del args['device'] - if 'ssid' not in args: - args['ssid'] = d.ident + '-ap-' + str(WifiAp.path_counter_next) + d = self.find_device_first(ident=args["device"], require=TestError) + del args["device"] + if "ssid" not in args: + args["ssid"] = d.ident + "-ap-" + str(WifiAp.path_counter_next) ap = WifiAp(**args) return ExportedObj.to_path(d.add_ap(ap)) - raise TestError("Invalid python type \"%s\"" % (class_name)) + raise TestError('Invalid python type "%s"' % (class_name)) - @dbus.service.method(IFACE_TEST, in_signature='ssas', out_signature='o') + @dbus.service.method(IFACE_TEST, in_signature="ssas", out_signature="o") def AddWiredDevice(self, ifname, mac, subchannels): dev = WiredDevice(ifname, mac, subchannels) return ExportedObj.to_path(self.add_device(dev)) - @dbus.service.method(IFACE_TEST, in_signature='s', out_signature='o') + @dbus.service.method(IFACE_TEST, in_signature="s", out_signature="o") def AddWifiDevice(self, ifname): dev = WifiDevice(ifname) return ExportedObj.to_path(self.add_device(dev)) - @dbus.service.method(IFACE_TEST, in_signature='o', out_signature='') + @dbus.service.method(IFACE_TEST, in_signature="o", out_signature="") def RemoveDevice(self, path): - d = self.find_device_first(path = path, require = TestError) + d = self.find_device_first(path=path, require=TestError) self.remove_device(d) - @dbus.service.method(IFACE_TEST, in_signature='sss', out_signature='o') + @dbus.service.method(IFACE_TEST, in_signature="sss", out_signature="o") def AddWifiAp(self, ident, ssid, bssid): - d = self.find_device_first(ident = ident, require = TestError) + d = self.find_device_first(ident=ident, require=TestError) ap = WifiAp(ssid, bssid) return ExportedObj.to_path(d.add_ap(ap)) - @dbus.service.method(IFACE_TEST, in_signature='so', out_signature='') + @dbus.service.method(IFACE_TEST, in_signature="so", out_signature="") def RemoveWifiAp(self, ident, ap_path): - d = self.find_device_first(ident = ident, require = TestError) + d = self.find_device_first(ident=ident, require=TestError) d.remove_ap_by_path(ap_path) - @dbus.service.method(IFACE_TEST, in_signature='', out_signature='') + @dbus.service.method(IFACE_TEST, in_signature="", out_signature="") def AutoRemoveNextConnection(self): gl.settings.auto_remove_next_connection() - @dbus.service.method(dbus_interface=IFACE_TEST, in_signature='a{sa{sv}}b', out_signature='o') + @dbus.service.method( + dbus_interface=IFACE_TEST, in_signature="a{sa{sv}}b", out_signature="o" + ) def AddConnection(self, con_hash, do_verify_strict): return gl.settings.add_connection(con_hash, do_verify_strict) - @dbus.service.method(dbus_interface=IFACE_TEST, in_signature='sa{sa{sv}}b', out_signature='') + @dbus.service.method( + dbus_interface=IFACE_TEST, in_signature="sa{sa{sv}}b", out_signature="" + ) def UpdateConnection(self, path, con_hash, do_verify_strict): return gl.settings.update_connection(con_hash, path, do_verify_strict) - @dbus.service.method(dbus_interface=IFACE_TEST, in_signature='ba{ss}', out_signature='') + @dbus.service.method( + dbus_interface=IFACE_TEST, in_signature="ba{ss}", out_signature="" + ) def ConnectionSetVisible(self, vis, selector_args): cons = list(gl.settings.find_connections(**selector_args)) - assert(len(cons) == 1) + assert len(cons) == 1 cons[0].SetVisible(vis) - @dbus.service.method(dbus_interface=IFACE_TEST, in_signature='', out_signature='') + @dbus.service.method(dbus_interface=IFACE_TEST, in_signature="", out_signature="") def Restart(self): gl.bus.release_name("org.freedesktop.NetworkManager") gl.bus.request_name("org.freedesktop.NetworkManager") @@ -1536,8 +1830,9 @@ class NetworkManager(ExportedObj): ############################################################################### -PRP_CONNECTION_UNSAVED = 'Unsaved' -PRP_CONNECTION_FILENAME = 'Filename' +PRP_CONNECTION_UNSAVED = "Unsaved" +PRP_CONNECTION_FILENAME = "Filename" + class Connection(ExportedObj): def __init__(self, path_counter, con_hash, do_verify_strict=True): @@ -1551,9 +1846,11 @@ class Connection(ExportedObj): s_con = {} con_hash[NM.SETTING_CONNECTION_SETTING_NAME] = s_con if NmUtil.con_hash_get_id(con_hash) is None: - s_con[NM.SETTING_CONNECTION_ID] = 'connection-%s' % (path_counter) + s_con[NM.SETTING_CONNECTION_ID] = "connection-%s" % (path_counter) if NmUtil.con_hash_get_uuid(con_hash) is None: - s_con[NM.SETTING_CONNECTION_UUID] = str(uuid.uuid3(uuid.NAMESPACE_URL, path)) + s_con[NM.SETTING_CONNECTION_UUID] = str( + uuid.uuid3(uuid.NAMESPACE_URL, path) + ) NmUtil.con_hash_verify(con_hash, do_verify_strict=do_verify_strict) @@ -1563,7 +1860,8 @@ class Connection(ExportedObj): props = { PRP_CONNECTION_UNSAVED: False, - PRP_CONNECTION_FILENAME: "/etc/NetworkManager/system-connections/" + self.get_id(), + PRP_CONNECTION_FILENAME: "/etc/NetworkManager/system-connections/" + + self.get_id(), } self.dbus_interface_add(IFACE_CONNECTION, props) @@ -1582,56 +1880,73 @@ class Connection(ExportedObj): def update_connection(self, con_hash, do_verify_strict): - NmUtil.con_hash_verify(con_hash, do_verify_strict = do_verify_strict) + NmUtil.con_hash_verify(con_hash, do_verify_strict=do_verify_strict) old_uuid = self.get_uuid() new_uuid = NmUtil.con_hash_get_uuid(con_hash) if old_uuid != new_uuid: - raise BusErr.InvalidPropertyException('connection.uuid: cannot change the uuid from %s to %s' % (old_uuid, new_uuid)) + raise BusErr.InvalidPropertyException( + "connection.uuid: cannot change the uuid from %s to %s" + % (old_uuid, new_uuid) + ) - self.con_hash = con_hash; + self.con_hash = con_hash self.Updated() - @dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='', out_signature='a{sa{sv}}') + @dbus.service.method( + dbus_interface=IFACE_CONNECTION, in_signature="", out_signature="a{sa{sv}}" + ) def GetSettings(self): - if hasattr(self, '_remove_next_connection_cb'): + if hasattr(self, "_remove_next_connection_cb"): self._remove_next_connection_cb() raise BusErr.UnknownConnectionException("Connection not found") if not self.visible: raise BusErr.PermissionDeniedException() return self.con_hash - @dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='b', out_signature='') + @dbus.service.method( + dbus_interface=IFACE_CONNECTION, in_signature="b", out_signature="" + ) def SetVisible(self, vis): self.visible = vis self.Updated() - @dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='', out_signature='') + @dbus.service.method( + dbus_interface=IFACE_CONNECTION, in_signature="", out_signature="" + ) def Delete(self): gl.settings.delete_connection(self) - @dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='a{sa{sv}}', out_signature='') + @dbus.service.method( + dbus_interface=IFACE_CONNECTION, in_signature="a{sa{sv}}", out_signature="" + ) def Update(self, con_hash): self.update_connection(con_hash, True) - @dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='a{sa{sv}}ua{sv}', out_signature='a{sv}') + @dbus.service.method( + dbus_interface=IFACE_CONNECTION, + in_signature="a{sa{sv}}ua{sv}", + out_signature="a{sv}", + ) def Update2(self, con_hash, flags, args): self.update_connection(con_hash, True) return [] - @dbus.service.signal(IFACE_CONNECTION, signature='') + @dbus.service.signal(IFACE_CONNECTION, signature="") def Removed(self): pass - @dbus.service.signal(IFACE_CONNECTION, signature='') + @dbus.service.signal(IFACE_CONNECTION, signature="") def Updated(self): pass + ############################################################################### -PRP_SETTINGS_HOSTNAME = 'Hostname' -PRP_SETTINGS_CAN_MODIFY = 'CanModify' -PRP_SETTINGS_CONNECTIONS = 'Connections' +PRP_SETTINGS_HOSTNAME = "Hostname" +PRP_SETTINGS_CAN_MODIFY = "CanModify" +PRP_SETTINGS_CONNECTIONS = "Connections" + class Settings(ExportedObj): def __init__(self): @@ -1642,30 +1957,32 @@ class Settings(ExportedObj): self.remove_next_connection = False props = { - PRP_SETTINGS_HOSTNAME: "foobar.baz", - PRP_SETTINGS_CAN_MODIFY: True, - PRP_SETTINGS_CONNECTIONS: dbus.Array([], 'o'), + PRP_SETTINGS_HOSTNAME: "foobar.baz", + PRP_SETTINGS_CAN_MODIFY: True, + PRP_SETTINGS_CONNECTIONS: dbus.Array([], "o"), } self.dbus_interface_add(IFACE_SETTINGS, props, Settings.PropertiesChanged) self.export() def auto_remove_next_connection(self): - self.remove_next_connection = True; + self.remove_next_connection = True def get_connection(self, path): return self.connections[path] - def get_connections(self, stable_order = True): + def get_connections(self, stable_order=True): cons = list(self.connections.values()) if stable_order: - cons.sort(key = lambda c: (Util.random_int(c.get_id()), Util.random_int(c.path))) + cons.sort( + key=lambda c: (Util.random_int(c.get_id()), Util.random_int(c.path)) + ) return cons - def get_connection_paths(self, stable_order = True): - return [c.path for c in self.get_connections(stable_order = stable_order)] + def get_connection_paths(self, stable_order=True): + return [c.path for c in self.get_connections(stable_order=stable_order)] - def find_connections(self, path = None, con_id = None, con_uuid = None): + def find_connections(self, path=None, con_id=None, con_uuid=None): for c in self.get_connections(): if path is not None: if c.path != path: @@ -1678,11 +1995,15 @@ class Settings(ExportedObj): continue yield c - @dbus.service.method(dbus_interface=IFACE_SETTINGS, in_signature='', out_signature='ao') + @dbus.service.method( + dbus_interface=IFACE_SETTINGS, in_signature="", out_signature="ao" + ) def ListConnections(self): return self.get_connection_paths() - @dbus.service.method(dbus_interface=IFACE_SETTINGS, in_signature='a{sa{sv}}', out_signature='o') + @dbus.service.method( + dbus_interface=IFACE_SETTINGS, in_signature="a{sa{sv}}", out_signature="o" + ) def AddConnection(self, con_hash): return self.add_connection(con_hash) @@ -1691,23 +2012,31 @@ class Settings(ExportedObj): con_inst = Connection(self.c_counter, con_hash, do_verify_strict) uuid = con_inst.get_uuid() - if uuid in [c.get_uuid() for c in self.get_connections(stable_order = False)]: - raise BusErr.InvalidSettingException('cannot add duplicate connection with uuid %s' % (uuid)) + if uuid in [c.get_uuid() for c in self.get_connections(stable_order=False)]: + raise BusErr.InvalidSettingException( + "cannot add duplicate connection with uuid %s" % (uuid) + ) con_inst.export() self.connections[con_inst.path] = con_inst self.NewConnection(con_inst.path) - self._dbus_property_set(IFACE_SETTINGS, PRP_SETTINGS_CONNECTIONS, dbus.Array(self.get_connection_paths(), 'o')) + self._dbus_property_set( + IFACE_SETTINGS, + PRP_SETTINGS_CONNECTIONS, + dbus.Array(self.get_connection_paths(), "o"), + ) gl.manager.devices_available_connections_update() if self.remove_next_connection: self.remove_next_connection = False + def cb(): - if hasattr(con_inst, '_remove_next_connection_cb'): + if hasattr(con_inst, "_remove_next_connection_cb"): del con_inst._remove_next_connection_cb self.delete_connection(con_inst) return False + # We will delete the connection right away on an idle handler. However, # the test races with initializing the connection (calling GetSettings()). # To avoid the race, we will check in GetSettings() whether the profile @@ -1719,56 +2048,64 @@ class Settings(ExportedObj): def update_connection(self, con_hash, path=None, do_verify_strict=True): if path not in self.connections: - raise BusErr.UnknownConnectionException('Connection not found') + raise BusErr.UnknownConnectionException("Connection not found") self.connections[path].update_connection(con_hash, do_verify_strict) def delete_connection(self, con_inst): del self.connections[con_inst.path] - self._dbus_property_set(IFACE_SETTINGS, PRP_SETTINGS_CONNECTIONS, dbus.Array(self.get_connection_paths(), 'o')) + self._dbus_property_set( + IFACE_SETTINGS, + PRP_SETTINGS_CONNECTIONS, + dbus.Array(self.get_connection_paths(), "o"), + ) con_inst.Removed() con_inst.unexport() gl.manager.devices_available_connections_update() - @dbus.service.method(dbus_interface=IFACE_SETTINGS, in_signature='s', out_signature='') + @dbus.service.method( + dbus_interface=IFACE_SETTINGS, in_signature="s", out_signature="" + ) def SaveHostname(self, hostname): # Arbitrary requirement to test error handling - if hostname.find('.') == -1: + if hostname.find(".") == -1: raise BusErr.InvalidHostnameException() self._dbus_property_set(IFACE_SETTINGS, PRP_SETTINGS_HOSTNAME, hostname) - @dbus.service.signal(IFACE_SETTINGS, signature='o') + @dbus.service.signal(IFACE_SETTINGS, signature="o") def NewConnection(self, path): pass - @dbus.service.signal(IFACE_SETTINGS, signature='a{sv}') + @dbus.service.signal(IFACE_SETTINGS, signature="a{sv}") def PropertiesChanged(self, path): pass - @dbus.service.method(IFACE_SETTINGS, in_signature='', out_signature='') + @dbus.service.method(IFACE_SETTINGS, in_signature="", out_signature="") def Quit(self): gl.mainloop.quit() + ############################################################################### -PRP_IP4_CONFIG_ADDRESSES = 'Addresses' -PRP_IP4_CONFIG_ADDRESSDATA = 'AddressData' -PRP_IP4_CONFIG_GATEWAY = 'Gateway' -PRP_IP4_CONFIG_ROUTES = 'Routes' -PRP_IP4_CONFIG_ROUTEDATA = 'RouteData' -PRP_IP4_CONFIG_NAMESERVERS = 'Nameservers' -PRP_IP4_CONFIG_DOMAINS = 'Domains' -PRP_IP4_CONFIG_SEARCHES = 'Searches' -PRP_IP4_CONFIG_DNSOPTIONS = 'DnsOptions' -PRP_IP4_CONFIG_DNSPRIORITY = 'DnsPriority' -PRP_IP4_CONFIG_WINSSERVERS = 'WinsServers' +PRP_IP4_CONFIG_ADDRESSES = "Addresses" +PRP_IP4_CONFIG_ADDRESSDATA = "AddressData" +PRP_IP4_CONFIG_GATEWAY = "Gateway" +PRP_IP4_CONFIG_ROUTES = "Routes" +PRP_IP4_CONFIG_ROUTEDATA = "RouteData" +PRP_IP4_CONFIG_NAMESERVERS = "Nameservers" +PRP_IP4_CONFIG_DOMAINS = "Domains" +PRP_IP4_CONFIG_SEARCHES = "Searches" +PRP_IP4_CONFIG_DNSOPTIONS = "DnsOptions" +PRP_IP4_CONFIG_DNSPRIORITY = "DnsPriority" +PRP_IP4_CONFIG_WINSSERVERS = "WinsServers" + class IP4Config(ExportedObj): path_counter_next = 1 path_prefix = "/org/freedesktop/NetworkManager/IP4Config/" - def __init__(self, generate_seed = _DEFAULT_ARG): + def __init__(self, generate_seed=_DEFAULT_ARG): ExportedObj.__init__(self, ExportedObj.create_path(IP4Config)) if generate_seed == _DEFAULT_ARG: @@ -1784,15 +2121,15 @@ class IP4Config(ExportedObj): gateway = None if seed: if Util.random_bool(seed): - gateway = Util.random_ip(seed, net = '192.168.0.0/16')[0] + gateway = Util.random_ip(seed, net="192.168.0.0/16")[0] addrs = [] if seed: for n in range(0, Util.random_int(seed, 4)): a = { - 'addr': Util.random_ip(seed, net = '192.168.0.0/16')[0], - 'prefix': Util.random_int(seed, 17, 32), - 'gateway': gateway if n == 0 else None, + "addr": Util.random_ip(seed, net="192.168.0.0/16")[0], + "prefix": Util.random_int(seed, 17, 32), + "gateway": gateway if n == 0 else None, } addrs.append(a) @@ -1800,30 +2137,48 @@ class IP4Config(ExportedObj): if seed: for n in range(0, Util.random_int(seed, 4)): a = { - 'dest': Util.random_ip(seed, net = '192.168.0.0/16')[0], - 'prefix': Util.random_int(seed, 17, 32), - 'next-hop': None if (Util.random_int(seed) % 3 == 0) else Util.random_ip(seed, net = '192.168.0.0/16')[0], - 'metric': -1 if (Util.random_int(seed) % 3 == 0) else Util.random_int(seed, 0, 0xFFFFFFFF), + "dest": Util.random_ip(seed, net="192.168.0.0/16")[0], + "prefix": Util.random_int(seed, 17, 32), + "next-hop": None + if (Util.random_int(seed) % 3 == 0) + else Util.random_ip(seed, net="192.168.0.0/16")[0], + "metric": -1 + if (Util.random_int(seed) % 3 == 0) + else Util.random_int(seed, 0, 0xFFFFFFFF), } routes.append(a) nameservers = [] if seed: - nameservers = list([Util.random_ip(seed, net = '192.168.0.0/16')[0] for x in range(Util.random_int(seed, 4))]) - - names_selection = ['foo1.bar', 'foo2.bar', 'foo3.bar', 'foo4.bar', 'fo.o.bar', 'fo.x.y']; + nameservers = list( + [ + Util.random_ip(seed, net="192.168.0.0/16")[0] + for x in range(Util.random_int(seed, 4)) + ] + ) + + names_selection = [ + "foo1.bar", + "foo2.bar", + "foo3.bar", + "foo4.bar", + "fo.o.bar", + "fo.x.y", + ] domains = [] if seed: - domains = Util.random_subset(seed, ['dom4.' + s for s in names_selection]) + domains = Util.random_subset(seed, ["dom4." + s for s in names_selection]) searches = [] if seed: - domains = Util.random_subset(seed, ['sear4.' + s for s in names_selection]) + domains = Util.random_subset(seed, ["sear4." + s for s in names_selection]) dnsoptions = [] if seed: - dnsoptions = Util.random_subset(seed, ['dns4-opt1', 'dns4-opt2', 'dns4-opt3', 'dns4-opt4']) + dnsoptions = Util.random_subset( + seed, ["dns4-opt1", "dns4-opt2", "dns4-opt3", "dns4-opt4"] + ) dnspriority = 0 if seed: @@ -1831,78 +2186,125 @@ class IP4Config(ExportedObj): winsservers = [] if seed: - winsservers = list([Util.random_ip(seed, net = '192.168.0.0/16')[0] for x in range(Util.random_int(seed, 4))]) + winsservers = list( + [ + Util.random_ip(seed, net="192.168.0.0/16")[0] + for x in range(Util.random_int(seed, 4)) + ] + ) return { - PRP_IP4_CONFIG_ADDRESSES: dbus.Array([ - [ Util.ip4_addr_be32(a['addr']), - a['prefix'], - Util.ip4_addr_be32(a['gateway']) if a['gateway'] else 0 - ] for a in addrs - ], - 'au'), - PRP_IP4_CONFIG_ADDRESSDATA: dbus.Array([ - dbus.Dictionary(collections.OrderedDict( [ ('address', dbus.String(a['addr'])), - ('prefix', dbus.UInt32(a['prefix']))] + \ - ([ ('gateway', dbus.String(a['gateway'])) ] if a['gateway'] else [])), - 'sv') - for a in addrs - ], - 'a{sv}'), - PRP_IP4_CONFIG_GATEWAY: dbus.String(gateway) if gateway else "", - PRP_IP4_CONFIG_ROUTES: dbus.Array([ - [ Util.ip4_addr_be32(a['dest']), - a['prefix'], - Util.ip4_addr_be32(a['next-hop'] or '0.0.0.0'), - max(a['metric'], 0) - ] for a in routes - ], - 'au'), - PRP_IP4_CONFIG_ROUTEDATA: dbus.Array([ - dbus.Dictionary(collections.OrderedDict( [ ('dest', dbus.String(a['dest'])), - ('prefix', dbus.UInt32(a['prefix']))] + \ - ([ ('next-hop', dbus.String(a['next-hop'])) ] if a['next-hop'] else []) + \ - ([ ('metric', dbus.UInt32(a['metric'])) ] if a['metric'] != -1 else [])), - 'sv') - for a in routes - ], - 'a{sv}'), - PRP_IP4_CONFIG_NAMESERVERS: dbus.Array([dbus.UInt32(Util.ip4_addr_be32(n)) for n in nameservers], 'u'), - PRP_IP4_CONFIG_DOMAINS: dbus.Array(domains, 's'), - PRP_IP4_CONFIG_SEARCHES: dbus.Array(searches, 's'), - PRP_IP4_CONFIG_DNSOPTIONS: dbus.Array(dnsoptions, 's'), + PRP_IP4_CONFIG_ADDRESSES: dbus.Array( + [ + [ + Util.ip4_addr_be32(a["addr"]), + a["prefix"], + Util.ip4_addr_be32(a["gateway"]) if a["gateway"] else 0, + ] + for a in addrs + ], + "au", + ), + PRP_IP4_CONFIG_ADDRESSDATA: dbus.Array( + [ + dbus.Dictionary( + collections.OrderedDict( + [ + ("address", dbus.String(a["addr"])), + ("prefix", dbus.UInt32(a["prefix"])), + ] + + ( + [("gateway", dbus.String(a["gateway"]))] + if a["gateway"] + else [] + ) + ), + "sv", + ) + for a in addrs + ], + "a{sv}", + ), + PRP_IP4_CONFIG_GATEWAY: dbus.String(gateway) if gateway else "", + PRP_IP4_CONFIG_ROUTES: dbus.Array( + [ + [ + Util.ip4_addr_be32(a["dest"]), + a["prefix"], + Util.ip4_addr_be32(a["next-hop"] or "0.0.0.0"), + max(a["metric"], 0), + ] + for a in routes + ], + "au", + ), + PRP_IP4_CONFIG_ROUTEDATA: dbus.Array( + [ + dbus.Dictionary( + collections.OrderedDict( + [ + ("dest", dbus.String(a["dest"])), + ("prefix", dbus.UInt32(a["prefix"])), + ] + + ( + [("next-hop", dbus.String(a["next-hop"]))] + if a["next-hop"] + else [] + ) + + ( + [("metric", dbus.UInt32(a["metric"]))] + if a["metric"] != -1 + else [] + ) + ), + "sv", + ) + for a in routes + ], + "a{sv}", + ), + PRP_IP4_CONFIG_NAMESERVERS: dbus.Array( + [dbus.UInt32(Util.ip4_addr_be32(n)) for n in nameservers], "u" + ), + PRP_IP4_CONFIG_DOMAINS: dbus.Array(domains, "s"), + PRP_IP4_CONFIG_SEARCHES: dbus.Array(searches, "s"), + PRP_IP4_CONFIG_DNSOPTIONS: dbus.Array(dnsoptions, "s"), PRP_IP4_CONFIG_DNSPRIORITY: dbus.Int32(dnspriority), - PRP_IP4_CONFIG_WINSSERVERS: dbus.Array([dbus.UInt32(Util.ip4_addr_be32(n)) for n in winsservers], 'u'), + PRP_IP4_CONFIG_WINSSERVERS: dbus.Array( + [dbus.UInt32(Util.ip4_addr_be32(n)) for n in winsservers], "u" + ), } def props_regenerate(self, generate_seed): props = self.generate_props(generate_seed) - for k,v in props.items(): + for k, v in props.items(): self._dbus_property_set(IFACE_IP4_CONFIG, k, v) - @dbus.service.signal(IFACE_IP4_CONFIG, signature='a{sv}') + @dbus.service.signal(IFACE_IP4_CONFIG, signature="a{sv}") def PropertiesChanged(self, path): pass + ############################################################################### -PRP_IP6_CONFIG_ADDRESSES = "Addresses" +PRP_IP6_CONFIG_ADDRESSES = "Addresses" PRP_IP6_CONFIG_ADDRESSDATA = "AddressData" -PRP_IP6_CONFIG_GATEWAY = "Gateway" -PRP_IP6_CONFIG_ROUTES = "Routes" -PRP_IP6_CONFIG_ROUTEDATA = "RouteData" +PRP_IP6_CONFIG_GATEWAY = "Gateway" +PRP_IP6_CONFIG_ROUTES = "Routes" +PRP_IP6_CONFIG_ROUTEDATA = "RouteData" PRP_IP6_CONFIG_NAMESERVERS = "Nameservers" -PRP_IP6_CONFIG_DOMAINS = "Domains" -PRP_IP6_CONFIG_SEARCHES = "Searches" -PRP_IP6_CONFIG_DNSOPTIONS = "DnsOptions" +PRP_IP6_CONFIG_DOMAINS = "Domains" +PRP_IP6_CONFIG_SEARCHES = "Searches" +PRP_IP6_CONFIG_DNSOPTIONS = "DnsOptions" PRP_IP6_CONFIG_DNSPRIORITY = "DnsPriority" + class IP6Config(ExportedObj): path_counter_next = 1 path_prefix = "/org/freedesktop/NetworkManager/IP6Config/" - def __init__(self, generate_seed = _DEFAULT_ARG): + def __init__(self, generate_seed=_DEFAULT_ARG): ExportedObj.__init__(self, ExportedObj.create_path(IP6Config)) if generate_seed == _DEFAULT_ARG: @@ -1918,15 +2320,15 @@ class IP6Config(ExportedObj): gateway = None if seed: if Util.random_bool(seed): - gateway = Util.random_ip(seed, net = '2001:a::/64')[0] + gateway = Util.random_ip(seed, net="2001:a::/64")[0] addrs = [] if seed: for n in range(0, Util.random_int(seed, 4)): a = { - 'addr': Util.random_ip(seed, net = '2001:a::/64')[0], - 'prefix': Util.random_int(seed, 65, 128), - 'gateway': gateway if n == 0 else None, + "addr": Util.random_ip(seed, net="2001:a::/64")[0], + "prefix": Util.random_int(seed, 65, 128), + "gateway": gateway if n == 0 else None, } addrs.append(a) @@ -1934,102 +2336,162 @@ class IP6Config(ExportedObj): if seed: for n in range(0, Util.random_int(seed, 4)): a = { - 'dest': Util.random_ip(seed, net = '2001:a::/64')[0], - 'prefix': Util.random_int(seed, 65, 128), - 'next-hop': None if (Util.random_int(seed) % 3 == 0) else Util.random_ip(seed, net = '2001:a::/64')[0], - 'metric': -1 if (Util.random_int(seed) % 3 == 0) else Util.random_int(seed, 0, 0xFFFFFFFF), + "dest": Util.random_ip(seed, net="2001:a::/64")[0], + "prefix": Util.random_int(seed, 65, 128), + "next-hop": None + if (Util.random_int(seed) % 3 == 0) + else Util.random_ip(seed, net="2001:a::/64")[0], + "metric": -1 + if (Util.random_int(seed) % 3 == 0) + else Util.random_int(seed, 0, 0xFFFFFFFF), } routes.append(a) nameservers = [] if seed: - nameservers = list([Util.random_ip(seed, net = '2001:a::/64')[0] for x in range(Util.random_int(seed, 4))]) - - names_selection = ['foo1.bar', 'foo2.bar', 'foo3.bar', 'foo4.bar', 'fo.o.bar', 'fo.x.y']; + nameservers = list( + [ + Util.random_ip(seed, net="2001:a::/64")[0] + for x in range(Util.random_int(seed, 4)) + ] + ) + + names_selection = [ + "foo1.bar", + "foo2.bar", + "foo3.bar", + "foo4.bar", + "fo.o.bar", + "fo.x.y", + ] domains = [] if seed: - domains = Util.random_subset(seed, ['dom6.' + s for s in names_selection]) + domains = Util.random_subset(seed, ["dom6." + s for s in names_selection]) searches = [] if seed: - domains = Util.random_subset(seed, ['sear6.' + s for s in names_selection]) + domains = Util.random_subset(seed, ["sear6." + s for s in names_selection]) dnsoptions = [] if seed: - dnsoptions = Util.random_subset(seed, ['dns6-opt1', 'dns6-opt2', 'dns6-opt3', 'dns6-opt4']) + dnsoptions = Util.random_subset( + seed, ["dns6-opt1", "dns6-opt2", "dns6-opt3", "dns6-opt4"] + ) dnspriority = 0 if seed: dnspriority = Util.random_int(seed, -10000, 10000) return { - PRP_IP6_CONFIG_ADDRESSES: dbus.Array([ - [ Util.ip6_addr_ay(a['addr']), - a['prefix'], - Util.ip6_addr_ay(a['gateway'] or '::') - ] for a in addrs - ], - '(ayuay)'), - PRP_IP6_CONFIG_ADDRESSDATA: dbus.Array([ - dbus.Dictionary(collections.OrderedDict( [ ('address', dbus.String(a['addr'])), - ('prefix', dbus.UInt32(a['prefix']))] + \ - ([ ('gateway', dbus.String(a['gateway'])) ] if a['gateway'] else [])), - 'sv') - for a in addrs - ], - 'a{sv}'), - PRP_IP6_CONFIG_GATEWAY: dbus.String(gateway) if gateway else "", - PRP_IP6_CONFIG_ROUTES: dbus.Array([ - [ Util.ip6_addr_ay(a['dest']), - a['prefix'], - Util.ip6_addr_ay(a['next-hop'] or '::'), - max(a['metric'], 0) - ] for a in routes - ], - '(ayuayu)'), - PRP_IP6_CONFIG_ROUTEDATA: dbus.Array([ - dbus.Dictionary(collections.OrderedDict( [ ('dest', dbus.String(a['dest'])), - ('prefix', dbus.UInt32(a['prefix']))] + \ - ([ ('next-hop', dbus.String(a['next-hop'])) ] if a['next-hop'] else []) + \ - ([ ('metric', dbus.UInt32(a['metric'])) ] if a['metric'] != -1 else [])), - 'sv') - for a in routes - ], - 'a{sv}'), - PRP_IP6_CONFIG_NAMESERVERS: dbus.Array([Util.ip6_addr_ay(n) for n in nameservers], 'ay'), - PRP_IP6_CONFIG_DOMAINS: dbus.Array(domains, 's'), - PRP_IP6_CONFIG_SEARCHES: dbus.Array(searches, 's'), - PRP_IP6_CONFIG_DNSOPTIONS: dbus.Array(dnsoptions, 's'), + PRP_IP6_CONFIG_ADDRESSES: dbus.Array( + [ + [ + Util.ip6_addr_ay(a["addr"]), + a["prefix"], + Util.ip6_addr_ay(a["gateway"] or "::"), + ] + for a in addrs + ], + "(ayuay)", + ), + PRP_IP6_CONFIG_ADDRESSDATA: dbus.Array( + [ + dbus.Dictionary( + collections.OrderedDict( + [ + ("address", dbus.String(a["addr"])), + ("prefix", dbus.UInt32(a["prefix"])), + ] + + ( + [("gateway", dbus.String(a["gateway"]))] + if a["gateway"] + else [] + ) + ), + "sv", + ) + for a in addrs + ], + "a{sv}", + ), + PRP_IP6_CONFIG_GATEWAY: dbus.String(gateway) if gateway else "", + PRP_IP6_CONFIG_ROUTES: dbus.Array( + [ + [ + Util.ip6_addr_ay(a["dest"]), + a["prefix"], + Util.ip6_addr_ay(a["next-hop"] or "::"), + max(a["metric"], 0), + ] + for a in routes + ], + "(ayuayu)", + ), + PRP_IP6_CONFIG_ROUTEDATA: dbus.Array( + [ + dbus.Dictionary( + collections.OrderedDict( + [ + ("dest", dbus.String(a["dest"])), + ("prefix", dbus.UInt32(a["prefix"])), + ] + + ( + [("next-hop", dbus.String(a["next-hop"]))] + if a["next-hop"] + else [] + ) + + ( + [("metric", dbus.UInt32(a["metric"]))] + if a["metric"] != -1 + else [] + ) + ), + "sv", + ) + for a in routes + ], + "a{sv}", + ), + PRP_IP6_CONFIG_NAMESERVERS: dbus.Array( + [Util.ip6_addr_ay(n) for n in nameservers], "ay" + ), + PRP_IP6_CONFIG_DOMAINS: dbus.Array(domains, "s"), + PRP_IP6_CONFIG_SEARCHES: dbus.Array(searches, "s"), + PRP_IP6_CONFIG_DNSOPTIONS: dbus.Array(dnsoptions, "s"), PRP_IP6_CONFIG_DNSPRIORITY: dbus.Int32(dnspriority), } def props_regenerate(self, generate_seed): props = self.generate_props(generate_seed) - for k,v in props.items(): + for k, v in props.items(): self._dbus_property_set(IFACE_IP6_CONFIG, k, v) - @dbus.service.signal(IFACE_IP6_CONFIG, signature='a{sv}') + @dbus.service.signal(IFACE_IP6_CONFIG, signature="a{sv}") def PropertiesChanged(self, path): pass + ############################################################################### -PRP_DHCP4_CONFIG_OPTIONS = 'Options' +PRP_DHCP4_CONFIG_OPTIONS = "Options" + class Dhcp4Config(ExportedObj): path_counter_next = 1 path_prefix = "/org/freedesktop/NetworkManager/DHCP4Config/" - def __init__(self, generate_seed = _DEFAULT_ARG): + def __init__(self, generate_seed=_DEFAULT_ARG): ExportedObj.__init__(self, ExportedObj.create_path(Dhcp4Config)) if generate_seed == _DEFAULT_ARG: generate_seed = self.path props = self._props_generate(generate_seed) - self.dbus_interface_add(IFACE_DHCP4_CONFIG, props, Dhcp4Config.PropertiesChanged) + self.dbus_interface_add( + IFACE_DHCP4_CONFIG, props, Dhcp4Config.PropertiesChanged + ) self.export() def _props_generate(self, generate_seed): @@ -2037,39 +2499,46 @@ class Dhcp4Config(ExportedObj): options = [] if seed: - options = Util.random_subset(seed, [('dhcp-4-opt-' + str(i), 'val-' + str(i)) for i in range(10)]) + options = Util.random_subset( + seed, [("dhcp-4-opt-" + str(i), "val-" + str(i)) for i in range(10)] + ) return { - PRP_DHCP4_CONFIG_OPTIONS: dbus.Dictionary(collections.OrderedDict(options), - 'sv') + PRP_DHCP4_CONFIG_OPTIONS: dbus.Dictionary( + collections.OrderedDict(options), "sv" + ) } def props_regenerate(self, generate_seed): props = self.generate_props(generate_seed) - for k,v in props.items(): + for k, v in props.items(): self._dbus_property_set(IFACE_DHCP4_CONFIG, k, v) - @dbus.service.signal(IFACE_DHCP4_CONFIG, signature='a{sv}') + @dbus.service.signal(IFACE_DHCP4_CONFIG, signature="a{sv}") def PropertiesChanged(self, path): pass + ############################################################################### -PRP_DHCP6_CONFIG_OPTIONS = 'Options' +PRP_DHCP6_CONFIG_OPTIONS = "Options" + class Dhcp6Config(ExportedObj): path_counter_next = 1 path_prefix = "/org/freedesktop/NetworkManager/DHCP6Config/" - def __init__(self, generate_seed = _DEFAULT_ARG): + def __init__(self, generate_seed=_DEFAULT_ARG): ExportedObj.__init__(self, ExportedObj.create_path(Dhcp6Config)) if generate_seed == _DEFAULT_ARG: generate_seed = self.path props = self._props_generate(generate_seed) - self.dbus_interface_add(IFACE_DHCP6_CONFIG, props, Dhcp6Config.PropertiesChanged) + self.dbus_interface_add( + IFACE_DHCP6_CONFIG, props, Dhcp6Config.PropertiesChanged + ) self.export() def _props_generate(self, generate_seed): @@ -2077,78 +2546,98 @@ class Dhcp6Config(ExportedObj): options = [] if seed: - options = Util.random_subset(seed, [('dhcp-6-opt-' + str(i), 'val-' + str(i)) for i in range(10)]) + options = Util.random_subset( + seed, [("dhcp-6-opt-" + str(i), "val-" + str(i)) for i in range(10)] + ) return { - PRP_DHCP4_CONFIG_OPTIONS: dbus.Dictionary(collections.OrderedDict(options), - 'sv') + PRP_DHCP4_CONFIG_OPTIONS: dbus.Dictionary( + collections.OrderedDict(options), "sv" + ) } def props_regenerate(self, generate_seed): props = self.generate_props(generate_seed) - for k,v in props.items(): + for k, v in props.items(): self._dbus_property_set(IFACE_DHCP6_CONFIG, k, v) - @dbus.service.signal(IFACE_DHCP6_CONFIG, signature='a{sv}') + @dbus.service.signal(IFACE_DHCP6_CONFIG, signature="a{sv}") def PropertiesChanged(self, path): pass + ############################################################################### -PRP_DNS_MANAGER_MODE = 'Mode' -PRP_DNS_MANAGER_RC_MANAGER = 'RcManager' -PRP_DNS_MANAGER_CONFIGURATION = 'Configuration' +PRP_DNS_MANAGER_MODE = "Mode" +PRP_DNS_MANAGER_RC_MANAGER = "RcManager" +PRP_DNS_MANAGER_CONFIGURATION = "Configuration" + class DnsManager(ExportedObj): def __init__(self): ExportedObj.__init__(self, "/org/freedesktop/NetworkManager/DnsManager") props = { - PRP_DNS_MANAGER_MODE: "dnsmasq", - PRP_DNS_MANAGER_RC_MANAGER: "symlink", + PRP_DNS_MANAGER_MODE: "dnsmasq", + PRP_DNS_MANAGER_RC_MANAGER: "symlink", PRP_DNS_MANAGER_CONFIGURATION: dbus.Array( [ dbus.Dictionary( { - 'nameservers' : dbus.Array(['1.2.3.4', '5.6.7.8'], 's'), - 'priority' : dbus.Int32(100), + "nameservers": dbus.Array(["1.2.3.4", "5.6.7.8"], "s"), + "priority": dbus.Int32(100), }, - 'sv') + "sv", + ) ], - 'a{sv}'), + "a{sv}", + ), } self.dbus_interface_add(IFACE_DNS_MANAGER, props) self.export() + ############################################################################### -PATH_SECRET_AGENT = '/org/freedesktop/NetworkManager/SecretAgent' +PATH_SECRET_AGENT = "/org/freedesktop/NetworkManager/SecretAgent" FLAG_ALLOW_INTERACTION = 0x1 FLAG_REQUEST_NEW = 0x2 FLAG_USER_REQUESTED = 0x4 + class AgentManager(dbus.service.Object): def __init__(self): - dbus.service.Object.__init__(self, gl.bus, "/org/freedesktop/NetworkManager/AgentManager") + dbus.service.Object.__init__( + self, gl.bus, "/org/freedesktop/NetworkManager/AgentManager" + ) self.agents = {} - @dbus.service.method(dbus_interface=IFACE_AGENT_MANAGER, - in_signature='s', out_signature='', - sender_keyword='sender') + @dbus.service.method( + dbus_interface=IFACE_AGENT_MANAGER, + in_signature="s", + out_signature="", + sender_keyword="sender", + ) def Register(self, name, sender=None): self.RegisterWithCapabilities(name, 0, sender) - @dbus.service.method(dbus_interface=IFACE_AGENT_MANAGER, - in_signature='su', out_signature='', - sender_keyword='sender') + @dbus.service.method( + dbus_interface=IFACE_AGENT_MANAGER, + in_signature="su", + out_signature="", + sender_keyword="sender", + ) def RegisterWithCapabilities(self, name, caps, sender=None): self.agents[sender] = gl.bus.get_object(sender, PATH_SECRET_AGENT) - @dbus.service.method(dbus_interface=IFACE_AGENT_MANAGER, - in_signature='', out_signature='', - sender_keyword='sender') + @dbus.service.method( + dbus_interface=IFACE_AGENT_MANAGER, + in_signature="", + out_signature="", + sender_keyword="sender", + ) def Unregister(self, sender=None): del self.agents[sender] @@ -2160,19 +2649,25 @@ class AgentManager(dbus.service.Object): for sender in self.agents: agent = self.agents[sender] try: - secrets = agent.GetSecrets(con_hash, path, setting_name, - dbus.Array([], 's'), - FLAG_ALLOW_INTERACTION | FLAG_USER_REQUESTED, - dbus_interface=IFACE_AGENT) + secrets = agent.GetSecrets( + con_hash, + path, + setting_name, + dbus.Array([], "s"), + FLAG_ALLOW_INTERACTION | FLAG_USER_REQUESTED, + dbus_interface=IFACE_AGENT, + ) break except dbus.DBusException as e: - if e.get_dbus_name() == IFACE_AGENT + '.UserCanceled': - raise BusErr.UserCanceledException('User canceled') + if e.get_dbus_name() == IFACE_AGENT + ".UserCanceled": + raise BusErr.UserCanceledException("User canceled") continue return secrets + ############################################################################### + class ObjectManager(dbus.service.Object): def __init__(self, object_path): dbus.service.Object.__init__(self, gl.bus, object_path) @@ -2192,25 +2687,30 @@ class ObjectManager(dbus.service.Object): self.objs.remove(obj) self.InterfacesRemoved(obj.path, obj.get_managed_ifaces().keys()) - @dbus.service.signal(IFACE_OBJECT_MANAGER, signature='oa{sa{sv}}') + @dbus.service.signal(IFACE_OBJECT_MANAGER, signature="oa{sa{sv}}") def InterfacesAdded(self, name, ifaces): pass - @dbus.service.signal(IFACE_OBJECT_MANAGER, signature='oas') + @dbus.service.signal(IFACE_OBJECT_MANAGER, signature="oas") def InterfacesRemoved(self, name, ifaces): pass - @dbus.service.method(dbus_interface=IFACE_OBJECT_MANAGER, - in_signature='', out_signature='a{oa{sa{sv}}}', - sender_keyword='sender') + @dbus.service.method( + dbus_interface=IFACE_OBJECT_MANAGER, + in_signature="", + out_signature="a{oa{sa{sv}}}", + sender_keyword="sender", + ) def GetManagedObjects(self, sender=None): managed_objects = {} for obj in self.objs: managed_objects[obj.path] = obj.get_managed_ifaces() return managed_objects + ############################################################################### + def main(): dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) @@ -2222,20 +2722,24 @@ def main(): gl.mainloop = GLib.MainLoop() gl.bus = dbus.SessionBus() - gl.object_manager = ObjectManager('/org/freedesktop') + gl.object_manager = ObjectManager("/org/freedesktop") gl.manager = NetworkManager() gl.settings = Settings() gl.dns_manager = DnsManager() gl.agent_manager = AgentManager() if not gl.bus.request_name("org.freedesktop.NetworkManager"): - raise AssertionError("Failure to request D-Bus name org.freedesktop.NetworkManager") + raise AssertionError( + "Failure to request D-Bus name org.freedesktop.NetworkManager" + ) # Watch stdin; if it closes, assume our parent has crashed, and exit - id1 = GLib.io_add_watch(GLib.IOChannel.unix_new(0), - GLib.PRIORITY_DEFAULT, - GLib.IO_HUP, - lambda io, condition: gl.mainloop.quit() or True) + id1 = GLib.io_add_watch( + GLib.IOChannel.unix_new(0), + GLib.PRIORITY_DEFAULT, + GLib.IO_HUP, + lambda io, condition: gl.mainloop.quit() or True, + ) gl.mainloop.run() @@ -2249,5 +2753,6 @@ def main(): sys.exit(0) -if __name__ == '__main__': + +if __name__ == "__main__": main() |