| # wpa_supplicant D-Bus interface tests |
| # Copyright (c) 2014-2015, Jouni Malinen <j@w1.fi> |
| # |
| # This software may be distributed under the terms of the BSD license. |
| # See README for more details. |
| |
| import binascii |
| import logging |
| logger = logging.getLogger() |
| import subprocess |
| import time |
| |
| try: |
| import gobject |
| import dbus |
| dbus_imported = True |
| except ImportError: |
| dbus_imported = False |
| |
| import hostapd |
| from wpasupplicant import WpaSupplicant |
| from utils import HwsimSkip, alloc_fail |
| from test_ap_tdls import connect_2sta_open |
| |
| WPAS_DBUS_SERVICE = "fi.w1.wpa_supplicant1" |
| WPAS_DBUS_PATH = "/fi/w1/wpa_supplicant1" |
| WPAS_DBUS_IFACE = "fi.w1.wpa_supplicant1.Interface" |
| WPAS_DBUS_IFACE_WPS = WPAS_DBUS_IFACE + ".WPS" |
| WPAS_DBUS_NETWORK = "fi.w1.wpa_supplicant1.Network" |
| WPAS_DBUS_BSS = "fi.w1.wpa_supplicant1.BSS" |
| WPAS_DBUS_IFACE_P2PDEVICE = WPAS_DBUS_IFACE + ".P2PDevice" |
| WPAS_DBUS_P2P_PEER = "fi.w1.wpa_supplicant1.Peer" |
| WPAS_DBUS_GROUP = "fi.w1.wpa_supplicant1.Group" |
| WPAS_DBUS_PERSISTENT_GROUP = "fi.w1.wpa_supplicant1.PersistentGroup" |
| |
| def prepare_dbus(dev): |
| if not dbus_imported: |
| logger.info("No dbus module available") |
| raise HwsimSkip("No dbus module available") |
| try: |
| from dbus.mainloop.glib import DBusGMainLoop |
| dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) |
| bus = dbus.SystemBus() |
| wpas_obj = bus.get_object(WPAS_DBUS_SERVICE, WPAS_DBUS_PATH) |
| wpas = dbus.Interface(wpas_obj, WPAS_DBUS_SERVICE) |
| path = wpas.GetInterface(dev.ifname) |
| if_obj = bus.get_object(WPAS_DBUS_SERVICE, path) |
| return (bus,wpas_obj,path,if_obj) |
| except Exception, e: |
| raise HwsimSkip("Could not connect to D-Bus: %s" % e) |
| |
| class TestDbus(object): |
| def __init__(self, bus): |
| self.loop = gobject.MainLoop() |
| self.signals = [] |
| self.bus = bus |
| |
| def __exit__(self, type, value, traceback): |
| for s in self.signals: |
| s.remove() |
| |
| def add_signal(self, handler, interface, name, byte_arrays=False): |
| s = self.bus.add_signal_receiver(handler, dbus_interface=interface, |
| signal_name=name, |
| byte_arrays=byte_arrays) |
| self.signals.append(s) |
| |
| def timeout(self, *args): |
| logger.debug("timeout") |
| self.loop.quit() |
| return False |
| |
| class alloc_fail_dbus(object): |
| def __init__(self, dev, count, funcs, operation="Operation", |
| expected="NoMemory"): |
| self._dev = dev |
| self._count = count |
| self._funcs = funcs |
| self._operation = operation |
| self._expected = expected |
| def __enter__(self): |
| cmd = "TEST_ALLOC_FAIL %d:%s" % (self._count, self._funcs) |
| if "OK" not in self._dev.request(cmd): |
| raise HwsimSkip("TEST_ALLOC_FAIL not supported") |
| def __exit__(self, type, value, traceback): |
| if type is None: |
| raise Exception("%s succeeded during out-of-memory" % self._operation) |
| if type == dbus.exceptions.DBusException and self._expected in str(value): |
| return True |
| if self._dev.request("GET_ALLOC_FAIL") != "0:%s" % self._funcs: |
| raise Exception("%s did not trigger allocation failure" % self._operation) |
| return False |
| |
| def start_ap(ap): |
| ssid = "test-wps" |
| params = { "ssid": ssid, "eap_server": "1", "wps_state": "2", |
| "wpa_passphrase": "12345678", "wpa": "2", |
| "wpa_key_mgmt": "WPA-PSK", "rsn_pairwise": "CCMP", |
| "ap_pin": "12345670"} |
| return hostapd.add_ap(ap['ifname'], params) |
| |
| def test_dbus_getall(dev, apdev): |
| """D-Bus GetAll""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| |
| props = wpas_obj.GetAll(WPAS_DBUS_SERVICE, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| logger.debug("GetAll(fi.w1.wpa.supplicant1, /fi/w1/wpa_supplicant1) ==> " + str(props)) |
| |
| props = if_obj.GetAll(WPAS_DBUS_IFACE, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| logger.debug("GetAll(%s, %s): %s" % (WPAS_DBUS_IFACE, path, str(props))) |
| |
| props = if_obj.GetAll(WPAS_DBUS_IFACE_WPS, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| logger.debug("GetAll(%s, %s): %s" % (WPAS_DBUS_IFACE_WPS, path, str(props))) |
| |
| res = if_obj.Get(WPAS_DBUS_IFACE, 'BSSs', |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if len(res) != 0: |
| raise Exception("Unexpected BSSs entry: " + str(res)) |
| |
| res = if_obj.Get(WPAS_DBUS_IFACE, 'Networks', |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if len(res) != 0: |
| raise Exception("Unexpected Networks entry: " + str(res)) |
| |
| hapd = hostapd.add_ap(apdev[0]['ifname'], { "ssid": "open" }) |
| bssid = apdev[0]['bssid'] |
| dev[0].scan_for_bss(bssid, freq=2412) |
| id = dev[0].add_network() |
| dev[0].set_network(id, "disabled", "0") |
| dev[0].set_network_quoted(id, "ssid", "test") |
| |
| res = if_obj.Get(WPAS_DBUS_IFACE, 'BSSs', |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if len(res) != 1: |
| raise Exception("Missing BSSs entry: " + str(res)) |
| bss_obj = bus.get_object(WPAS_DBUS_SERVICE, res[0]) |
| props = bss_obj.GetAll(WPAS_DBUS_BSS, dbus_interface=dbus.PROPERTIES_IFACE) |
| logger.debug("GetAll(%s, %s): %s" % (WPAS_DBUS_BSS, res[0], str(props))) |
| bssid_str = '' |
| for item in props['BSSID']: |
| if len(bssid_str) > 0: |
| bssid_str += ':' |
| bssid_str += '%02x' % item |
| if bssid_str != bssid: |
| raise Exception("Unexpected BSSID in BSSs entry") |
| |
| res = if_obj.Get(WPAS_DBUS_IFACE, 'Networks', |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if len(res) != 1: |
| raise Exception("Missing Networks entry: " + str(res)) |
| net_obj = bus.get_object(WPAS_DBUS_SERVICE, res[0]) |
| props = net_obj.GetAll(WPAS_DBUS_NETWORK, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| logger.debug("GetAll(%s, %s): %s" % (WPAS_DBUS_NETWORK, res[0], str(props))) |
| ssid = props['Properties']['ssid'] |
| if ssid != '"test"': |
| raise Exception("Unexpected SSID in network entry") |
| |
| def dbus_get(dbus, wpas_obj, prop, expect=None, byte_arrays=False): |
| val = wpas_obj.Get(WPAS_DBUS_SERVICE, prop, |
| dbus_interface=dbus.PROPERTIES_IFACE, |
| byte_arrays=byte_arrays) |
| if expect is not None and val != expect: |
| raise Exception("Unexpected %s: %s (expected: %s)" % |
| (prop, str(val), str(expect))) |
| return val |
| |
| def dbus_set(dbus, wpas_obj, prop, val): |
| wpas_obj.Set(WPAS_DBUS_SERVICE, prop, val, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| def test_dbus_properties(dev, apdev): |
| """D-Bus Get/Set fi.w1.wpa_supplicant1 properties""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| |
| dbus_get(dbus, wpas_obj, "DebugLevel", expect="msgdump") |
| dbus_set(dbus, wpas_obj, "DebugLevel", "debug") |
| dbus_get(dbus, wpas_obj, "DebugLevel", expect="debug") |
| for (val,err) in [ (3, "Error.Failed: wrong property type"), |
| ("foo", "Error.Failed: wrong debug level value") ]: |
| try: |
| dbus_set(dbus, wpas_obj, "DebugLevel", val) |
| raise Exception("Invalid DebugLevel value accepted: " + str(val)) |
| except dbus.exceptions.DBusException, e: |
| if err not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| dbus_set(dbus, wpas_obj, "DebugLevel", "msgdump") |
| dbus_get(dbus, wpas_obj, "DebugLevel", expect="msgdump") |
| |
| dbus_get(dbus, wpas_obj, "DebugTimestamp", expect=True) |
| dbus_set(dbus, wpas_obj, "DebugTimestamp", False) |
| dbus_get(dbus, wpas_obj, "DebugTimestamp", expect=False) |
| try: |
| dbus_set(dbus, wpas_obj, "DebugTimestamp", "foo") |
| raise Exception("Invalid DebugTimestamp value accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: wrong property type" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| dbus_set(dbus, wpas_obj, "DebugTimestamp", True) |
| dbus_get(dbus, wpas_obj, "DebugTimestamp", expect=True) |
| |
| dbus_get(dbus, wpas_obj, "DebugShowKeys", expect=True) |
| dbus_set(dbus, wpas_obj, "DebugShowKeys", False) |
| dbus_get(dbus, wpas_obj, "DebugShowKeys", expect=False) |
| try: |
| dbus_set(dbus, wpas_obj, "DebugShowKeys", "foo") |
| raise Exception("Invalid DebugShowKeys value accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: wrong property type" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| dbus_set(dbus, wpas_obj, "DebugShowKeys", True) |
| dbus_get(dbus, wpas_obj, "DebugShowKeys", expect=True) |
| |
| res = dbus_get(dbus, wpas_obj, "Interfaces") |
| if len(res) != 1: |
| raise Exception("Unexpected Interfaces value: " + str(res)) |
| |
| res = dbus_get(dbus, wpas_obj, "EapMethods") |
| if len(res) < 5 or "TTLS" not in res: |
| raise Exception("Unexpected EapMethods value: " + str(res)) |
| |
| res = dbus_get(dbus, wpas_obj, "Capabilities") |
| if len(res) < 2 or "p2p" not in res: |
| raise Exception("Unexpected Capabilities value: " + str(res)) |
| |
| dbus_get(dbus, wpas_obj, "WFDIEs", byte_arrays=True) |
| val = binascii.unhexlify("010006020304050608") |
| dbus_set(dbus, wpas_obj, "WFDIEs", dbus.ByteArray(val)) |
| res = dbus_get(dbus, wpas_obj, "WFDIEs", byte_arrays=True) |
| if val != res: |
| raise Exception("WFDIEs value changed") |
| try: |
| dbus_set(dbus, wpas_obj, "WFDIEs", dbus.ByteArray('\x00')) |
| raise Exception("Invalid WFDIEs value accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| dbus_set(dbus, wpas_obj, "WFDIEs", dbus.ByteArray('')) |
| dbus_set(dbus, wpas_obj, "WFDIEs", dbus.ByteArray(val)) |
| dbus_set(dbus, wpas_obj, "WFDIEs", dbus.ByteArray('')) |
| res = dbus_get(dbus, wpas_obj, "WFDIEs", byte_arrays=True) |
| if len(res) != 0: |
| raise Exception("WFDIEs not cleared properly") |
| |
| res = dbus_get(dbus, wpas_obj, "EapMethods") |
| try: |
| dbus_set(dbus, wpas_obj, "EapMethods", res) |
| raise Exception("Invalid Set accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs: Property is read-only" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| try: |
| wpas_obj.SetFoo(WPAS_DBUS_SERVICE, "DebugShowKeys", True, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Unknown method accepted") |
| except dbus.exceptions.DBusException, e: |
| if "UnknownMethod" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| try: |
| wpas_obj.Get("foo", "DebugShowKeys", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Get accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs: No such property" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| test_obj = bus.get_object(WPAS_DBUS_SERVICE, WPAS_DBUS_PATH, |
| introspect=False) |
| try: |
| test_obj.Get(123, "DebugShowKeys", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Get accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs: Invalid arguments" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| try: |
| test_obj.Get(WPAS_DBUS_SERVICE, 123, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Get accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs: Invalid arguments" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| try: |
| wpas_obj.Set(WPAS_DBUS_SERVICE, "WFDIEs", |
| dbus.ByteArray('', variant_level=2), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs: invalid message format" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| def test_dbus_invalid_method(dev, apdev): |
| """D-Bus invalid method""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wps = dbus.Interface(if_obj, WPAS_DBUS_IFACE_WPS) |
| |
| try: |
| wps.Foo() |
| raise Exception("Unknown method accepted") |
| except dbus.exceptions.DBusException, e: |
| if "UnknownMethod" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| test_obj = bus.get_object(WPAS_DBUS_SERVICE, path, introspect=False) |
| test_wps = dbus.Interface(test_obj, WPAS_DBUS_IFACE_WPS) |
| try: |
| test_wps.Start(123) |
| raise Exception("WPS.Start with incorrect signature accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs: Invalid arg" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| def test_dbus_get_set_wps(dev, apdev): |
| """D-Bus Get/Set for WPS properties""" |
| try: |
| _test_dbus_get_set_wps(dev, apdev) |
| finally: |
| dev[0].request("SET wps_cred_processing 0") |
| dev[0].request("SET config_methods display keypad virtual_display nfc_interface p2ps") |
| |
| def _test_dbus_get_set_wps(dev, apdev): |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| |
| if_obj.Get(WPAS_DBUS_IFACE_WPS, "ConfigMethods", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| val = "display keypad virtual_display nfc_interface" |
| dev[0].request("SET config_methods " + val) |
| |
| config = if_obj.Get(WPAS_DBUS_IFACE_WPS, "ConfigMethods", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if config != val: |
| raise Exception("Unexpected Get(ConfigMethods) result: " + config) |
| |
| val2 = "push_button display" |
| if_obj.Set(WPAS_DBUS_IFACE_WPS, "ConfigMethods", val2, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| config = if_obj.Get(WPAS_DBUS_IFACE_WPS, "ConfigMethods", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if config != val2: |
| raise Exception("Unexpected Get(ConfigMethods) result after Set: " + config) |
| |
| dev[0].request("SET config_methods " + val) |
| |
| for i in range(3): |
| dev[0].request("SET wps_cred_processing " + str(i)) |
| val = if_obj.Get(WPAS_DBUS_IFACE_WPS, "ProcessCredentials", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| expected_val = False if i == 1 else True |
| if val != expected_val: |
| raise Exception("Unexpected Get(ProcessCredentials) result({}): {}".format(i, val)) |
| |
| class TestDbusGetSet(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.signal_received = False |
| self.signal_received_deprecated = False |
| self.sets_done = False |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.run_sets) |
| gobject.timeout_add(1000, self.timeout) |
| self.add_signal(self.propertiesChanged, WPAS_DBUS_IFACE_WPS, |
| "PropertiesChanged") |
| self.add_signal(self.propertiesChanged2, dbus.PROPERTIES_IFACE, |
| "PropertiesChanged") |
| self.loop.run() |
| return self |
| |
| def propertiesChanged(self, properties): |
| logger.debug("PropertiesChanged: " + str(properties)) |
| if properties.has_key("ProcessCredentials"): |
| self.signal_received_deprecated = True |
| if self.sets_done and self.signal_received: |
| self.loop.quit() |
| |
| def propertiesChanged2(self, interface_name, changed_properties, |
| invalidated_properties): |
| logger.debug("propertiesChanged2: interface_name=%s changed_properties=%s invalidated_properties=%s" % (interface_name, str(changed_properties), str(invalidated_properties))) |
| if interface_name != WPAS_DBUS_IFACE_WPS: |
| return |
| if changed_properties.has_key("ProcessCredentials"): |
| self.signal_received = True |
| if self.sets_done and self.signal_received_deprecated: |
| self.loop.quit() |
| |
| def run_sets(self, *args): |
| logger.debug("run_sets") |
| if_obj.Set(WPAS_DBUS_IFACE_WPS, "ProcessCredentials", |
| dbus.Boolean(1), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if if_obj.Get(WPAS_DBUS_IFACE_WPS, "ProcessCredentials", |
| dbus_interface=dbus.PROPERTIES_IFACE) != True: |
| raise Exception("Unexpected Get(ProcessCredentials) result after Set"); |
| if_obj.Set(WPAS_DBUS_IFACE_WPS, "ProcessCredentials", |
| dbus.Boolean(0), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if if_obj.Get(WPAS_DBUS_IFACE_WPS, "ProcessCredentials", |
| dbus_interface=dbus.PROPERTIES_IFACE) != False: |
| raise Exception("Unexpected Get(ProcessCredentials) result after Set"); |
| |
| self.dbus_sets_done = True |
| return False |
| |
| def success(self): |
| return self.signal_received and self.signal_received_deprecated |
| |
| with TestDbusGetSet(bus) as t: |
| if not t.success(): |
| raise Exception("No signal received for ProcessCredentials change") |
| |
| def test_dbus_wps_invalid(dev, apdev): |
| """D-Bus invaldi WPS operation""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wps = dbus.Interface(if_obj, WPAS_DBUS_IFACE_WPS) |
| |
| failures = [ {'Role': 'foo', 'Type': 'pbc'}, |
| {'Role': 123, 'Type': 'pbc'}, |
| {'Type': 'pbc'}, |
| {'Role': 'enrollee'}, |
| {'Role': 'registrar'}, |
| {'Role': 'enrollee', 'Type': 123}, |
| {'Role': 'enrollee', 'Type': 'foo'}, |
| {'Role': 'enrollee', 'Type': 'pbc', |
| 'Bssid': '02:33:44:55:66:77'}, |
| {'Role': 'enrollee', 'Type': 'pin', 'Pin': 123}, |
| {'Role': 'enrollee', 'Type': 'pbc', |
| 'Bssid': dbus.ByteArray('12345')}, |
| {'Role': 'enrollee', 'Type': 'pbc', |
| 'P2PDeviceAddress': 12345}, |
| {'Role': 'enrollee', 'Type': 'pbc', |
| 'P2PDeviceAddress': dbus.ByteArray('12345')}, |
| {'Role': 'enrollee', 'Type': 'pbc', 'Foo': 'bar'} ] |
| for args in failures: |
| try: |
| wps.Start(args) |
| raise Exception("Invalid WPS.Start() arguments accepted: " + str(args)) |
| except dbus.exceptions.DBusException, e: |
| if not str(e).startswith("fi.w1.wpa_supplicant1.InvalidArgs"): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| def test_dbus_wps_oom(dev, apdev): |
| """D-Bus WPS operation (OOM)""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wps = dbus.Interface(if_obj, WPAS_DBUS_IFACE_WPS) |
| |
| with alloc_fail_dbus(dev[0], 1, "=wpas_dbus_getter_state", "Get"): |
| if_obj.Get(WPAS_DBUS_IFACE, "State", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| hapd = hostapd.add_ap(apdev[0]['ifname'], { "ssid": "open" }) |
| bssid = apdev[0]['bssid'] |
| dev[0].scan_for_bss(bssid, freq=2412) |
| |
| for i in range(1, 3): |
| with alloc_fail_dbus(dev[0], i, "=wpas_dbus_getter_bsss", "Get"): |
| if_obj.Get(WPAS_DBUS_IFACE, "BSSs", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| res = if_obj.Get(WPAS_DBUS_IFACE, 'BSSs', |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| bss_obj = bus.get_object(WPAS_DBUS_SERVICE, res[0]) |
| with alloc_fail_dbus(dev[0], 1, "=wpas_dbus_getter_bss_rates", "Get"): |
| bss_obj.Get(WPAS_DBUS_BSS, "Rates", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| id = dev[0].add_network() |
| dev[0].set_network(id, "disabled", "0") |
| dev[0].set_network_quoted(id, "ssid", "test") |
| |
| for i in range(1, 3): |
| with alloc_fail_dbus(dev[0], i, "=wpas_dbus_getter_networks", "Get"): |
| if_obj.Get(WPAS_DBUS_IFACE, "Networks", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| with alloc_fail_dbus(dev[0], 1, "wpas_dbus_getter_interfaces", "Get"): |
| dbus_get(dbus, wpas_obj, "Interfaces") |
| |
| for i in range(1, 6): |
| with alloc_fail_dbus(dev[0], i, "=eap_get_names_as_string_array;wpas_dbus_getter_eap_methods", "Get"): |
| dbus_get(dbus, wpas_obj, "EapMethods") |
| |
| with alloc_fail_dbus(dev[0], 1, "wpas_dbus_setter_config_methods", "Set", |
| expected="Error.Failed: Failed to set property"): |
| val2 = "push_button display" |
| if_obj.Set(WPAS_DBUS_IFACE_WPS, "ConfigMethods", val2, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| with alloc_fail_dbus(dev[0], 1, "=wpa_config_add_network;wpas_dbus_handler_wps_start", |
| "WPS.Start", |
| expected="UnknownError: WPS start failed"): |
| wps.Start({'Role': 'enrollee', 'Type': 'pin', 'Pin': '12345670'}) |
| |
| def test_dbus_wps_pbc(dev, apdev): |
| """D-Bus WPS/PBC operation and signals""" |
| try: |
| _test_dbus_wps_pbc(dev, apdev) |
| finally: |
| dev[0].request("SET wps_cred_processing 0") |
| |
| def _test_dbus_wps_pbc(dev, apdev): |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wps = dbus.Interface(if_obj, WPAS_DBUS_IFACE_WPS) |
| |
| hapd = start_ap(apdev[0]) |
| hapd.request("WPS_PBC") |
| bssid = apdev[0]['bssid'] |
| dev[0].scan_for_bss(bssid, freq="2412") |
| dev[0].request("SET wps_cred_processing 2") |
| |
| res = if_obj.Get(WPAS_DBUS_IFACE, 'BSSs', |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if len(res) != 1: |
| raise Exception("Missing BSSs entry: " + str(res)) |
| bss_obj = bus.get_object(WPAS_DBUS_SERVICE, res[0]) |
| props = bss_obj.GetAll(WPAS_DBUS_BSS, dbus_interface=dbus.PROPERTIES_IFACE) |
| logger.debug("GetAll(%s, %s): %s" % (WPAS_DBUS_BSS, res[0], str(props))) |
| if 'WPS' not in props: |
| raise Exception("No WPS information in the BSS entry") |
| if 'Type' not in props['WPS']: |
| raise Exception("No Type field in the WPS dictionary") |
| if props['WPS']['Type'] != 'pbc': |
| raise Exception("Unexpected WPS Type: " + props['WPS']['Type']) |
| |
| class TestDbusWps(TestDbus): |
| def __init__(self, bus, wps): |
| TestDbus.__init__(self, bus) |
| self.success_seen = False |
| self.credentials_received = False |
| self.wps = wps |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.start_pbc) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.wpsEvent, WPAS_DBUS_IFACE_WPS, "Event") |
| self.add_signal(self.credentials, WPAS_DBUS_IFACE_WPS, |
| "Credentials") |
| self.loop.run() |
| return self |
| |
| def wpsEvent(self, name, args): |
| logger.debug("wpsEvent: %s args='%s'" % (name, str(args))) |
| if name == "success": |
| self.success_seen = True |
| if self.credentials_received: |
| self.loop.quit() |
| |
| def credentials(self, args): |
| logger.debug("credentials: " + str(args)) |
| self.credentials_received = True |
| if self.success_seen: |
| self.loop.quit() |
| |
| def start_pbc(self, *args): |
| logger.debug("start_pbc") |
| self.wps.Start({'Role': 'enrollee', 'Type': 'pbc'}) |
| return False |
| |
| def success(self): |
| return self.success_seen and self.credentials_received |
| |
| with TestDbusWps(bus, wps) as t: |
| if not t.success(): |
| raise Exception("Failure in D-Bus operations") |
| |
| dev[0].wait_connected(timeout=10) |
| dev[0].request("DISCONNECT") |
| hapd.disable() |
| dev[0].flush_scan_cache() |
| |
| def test_dbus_wps_pin(dev, apdev): |
| """D-Bus WPS/PIN operation and signals""" |
| try: |
| _test_dbus_wps_pin(dev, apdev) |
| finally: |
| dev[0].request("SET wps_cred_processing 0") |
| |
| def _test_dbus_wps_pin(dev, apdev): |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wps = dbus.Interface(if_obj, WPAS_DBUS_IFACE_WPS) |
| |
| hapd = start_ap(apdev[0]) |
| hapd.request("WPS_PIN any 12345670") |
| bssid = apdev[0]['bssid'] |
| dev[0].scan_for_bss(bssid, freq="2412") |
| dev[0].request("SET wps_cred_processing 2") |
| |
| class TestDbusWps(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.success_seen = False |
| self.credentials_received = False |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.start_pin) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.wpsEvent, WPAS_DBUS_IFACE_WPS, "Event") |
| self.add_signal(self.credentials, WPAS_DBUS_IFACE_WPS, |
| "Credentials") |
| self.loop.run() |
| return self |
| |
| def wpsEvent(self, name, args): |
| logger.debug("wpsEvent: %s args='%s'" % (name, str(args))) |
| if name == "success": |
| self.success_seen = True |
| if self.credentials_received: |
| self.loop.quit() |
| |
| def credentials(self, args): |
| logger.debug("credentials: " + str(args)) |
| self.credentials_received = True |
| if self.success_seen: |
| self.loop.quit() |
| |
| def start_pin(self, *args): |
| logger.debug("start_pin") |
| bssid_ay = dbus.ByteArray(bssid.replace(':','').decode('hex')) |
| wps.Start({'Role': 'enrollee', 'Type': 'pin', 'Pin': '12345670', |
| 'Bssid': bssid_ay}) |
| return False |
| |
| def success(self): |
| return self.success_seen and self.credentials_received |
| |
| with TestDbusWps(bus) as t: |
| if not t.success(): |
| raise Exception("Failure in D-Bus operations") |
| |
| dev[0].wait_connected(timeout=10) |
| |
| def test_dbus_wps_pin2(dev, apdev): |
| """D-Bus WPS/PIN operation and signals (PIN from wpa_supplicant)""" |
| try: |
| _test_dbus_wps_pin2(dev, apdev) |
| finally: |
| dev[0].request("SET wps_cred_processing 0") |
| |
| def _test_dbus_wps_pin2(dev, apdev): |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wps = dbus.Interface(if_obj, WPAS_DBUS_IFACE_WPS) |
| |
| hapd = start_ap(apdev[0]) |
| bssid = apdev[0]['bssid'] |
| dev[0].scan_for_bss(bssid, freq="2412") |
| dev[0].request("SET wps_cred_processing 2") |
| |
| class TestDbusWps(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.success_seen = False |
| self.failed = False |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.start_pin) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.wpsEvent, WPAS_DBUS_IFACE_WPS, "Event") |
| self.add_signal(self.credentials, WPAS_DBUS_IFACE_WPS, |
| "Credentials") |
| self.loop.run() |
| return self |
| |
| def wpsEvent(self, name, args): |
| logger.debug("wpsEvent: %s args='%s'" % (name, str(args))) |
| if name == "success": |
| self.success_seen = True |
| if self.credentials_received: |
| self.loop.quit() |
| |
| def credentials(self, args): |
| logger.debug("credentials: " + str(args)) |
| self.credentials_received = True |
| if self.success_seen: |
| self.loop.quit() |
| |
| def start_pin(self, *args): |
| logger.debug("start_pin") |
| bssid_ay = dbus.ByteArray(bssid.replace(':','').decode('hex')) |
| res = wps.Start({'Role': 'enrollee', 'Type': 'pin', |
| 'Bssid': bssid_ay}) |
| pin = res['Pin'] |
| h = hostapd.Hostapd(apdev[0]['ifname']) |
| h.request("WPS_PIN any " + pin) |
| return False |
| |
| def success(self): |
| return self.success_seen and self.credentials_received |
| |
| with TestDbusWps(bus) as t: |
| if not t.success(): |
| raise Exception("Failure in D-Bus operations") |
| |
| dev[0].wait_connected(timeout=10) |
| |
| def test_dbus_wps_pin_m2d(dev, apdev): |
| """D-Bus WPS/PIN operation and signals with M2D""" |
| try: |
| _test_dbus_wps_pin_m2d(dev, apdev) |
| finally: |
| dev[0].request("SET wps_cred_processing 0") |
| |
| def _test_dbus_wps_pin_m2d(dev, apdev): |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wps = dbus.Interface(if_obj, WPAS_DBUS_IFACE_WPS) |
| |
| hapd = start_ap(apdev[0]) |
| bssid = apdev[0]['bssid'] |
| dev[0].scan_for_bss(bssid, freq="2412") |
| dev[0].request("SET wps_cred_processing 2") |
| |
| class TestDbusWps(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.success_seen = False |
| self.credentials_received = False |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.start_pin) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.wpsEvent, WPAS_DBUS_IFACE_WPS, "Event") |
| self.add_signal(self.credentials, WPAS_DBUS_IFACE_WPS, |
| "Credentials") |
| self.loop.run() |
| return self |
| |
| def wpsEvent(self, name, args): |
| logger.debug("wpsEvent: %s args='%s'" % (name, str(args))) |
| if name == "success": |
| self.success_seen = True |
| if self.credentials_received: |
| self.loop.quit() |
| elif name == "m2d": |
| h = hostapd.Hostapd(apdev[0]['ifname']) |
| h.request("WPS_PIN any 12345670") |
| |
| def credentials(self, args): |
| logger.debug("credentials: " + str(args)) |
| self.credentials_received = True |
| if self.success_seen: |
| self.loop.quit() |
| |
| def start_pin(self, *args): |
| logger.debug("start_pin") |
| bssid_ay = dbus.ByteArray(bssid.replace(':','').decode('hex')) |
| wps.Start({'Role': 'enrollee', 'Type': 'pin', 'Pin': '12345670', |
| 'Bssid': bssid_ay}) |
| return False |
| |
| def success(self): |
| return self.success_seen and self.credentials_received |
| |
| with TestDbusWps(bus) as t: |
| if not t.success(): |
| raise Exception("Failure in D-Bus operations") |
| |
| dev[0].wait_connected(timeout=10) |
| |
| def test_dbus_wps_reg(dev, apdev): |
| """D-Bus WPS/Registrar operation and signals""" |
| try: |
| _test_dbus_wps_reg(dev, apdev) |
| finally: |
| dev[0].request("SET wps_cred_processing 0") |
| |
| def _test_dbus_wps_reg(dev, apdev): |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wps = dbus.Interface(if_obj, WPAS_DBUS_IFACE_WPS) |
| |
| hapd = start_ap(apdev[0]) |
| hapd.request("WPS_PIN any 12345670") |
| bssid = apdev[0]['bssid'] |
| dev[0].scan_for_bss(bssid, freq="2412") |
| dev[0].request("SET wps_cred_processing 2") |
| |
| class TestDbusWps(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.credentials_received = False |
| |
| def __enter__(self): |
| gobject.timeout_add(100, self.start_reg) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.wpsEvent, WPAS_DBUS_IFACE_WPS, "Event") |
| self.add_signal(self.credentials, WPAS_DBUS_IFACE_WPS, |
| "Credentials") |
| self.loop.run() |
| return self |
| |
| def wpsEvent(self, name, args): |
| logger.debug("wpsEvent: %s args='%s'" % (name, str(args))) |
| |
| def credentials(self, args): |
| logger.debug("credentials: " + str(args)) |
| self.credentials_received = True |
| self.loop.quit() |
| |
| def start_reg(self, *args): |
| logger.debug("start_reg") |
| bssid_ay = dbus.ByteArray(bssid.replace(':','').decode('hex')) |
| wps.Start({'Role': 'registrar', 'Type': 'pin', |
| 'Pin': '12345670', 'Bssid': bssid_ay}) |
| return False |
| |
| def success(self): |
| return self.credentials_received |
| |
| with TestDbusWps(bus) as t: |
| if not t.success(): |
| raise Exception("Failure in D-Bus operations") |
| |
| dev[0].wait_connected(timeout=10) |
| |
| def test_dbus_scan_invalid(dev, apdev): |
| """D-Bus invalid scan method""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| tests = [ ({}, "InvalidArgs"), |
| ({'Type': 123}, "InvalidArgs"), |
| ({'Type': 'foo'}, "InvalidArgs"), |
| ({'Type': 'active', 'Foo': 'bar'}, "InvalidArgs"), |
| ({'Type': 'active', 'SSIDs': 'foo'}, "InvalidArgs"), |
| ({'Type': 'active', 'SSIDs': ['foo']}, "InvalidArgs"), |
| ({'Type': 'active', |
| 'SSIDs': [ dbus.ByteArray("1"), dbus.ByteArray("2"), |
| dbus.ByteArray("3"), dbus.ByteArray("4"), |
| dbus.ByteArray("5"), dbus.ByteArray("6"), |
| dbus.ByteArray("7"), dbus.ByteArray("8"), |
| dbus.ByteArray("9"), dbus.ByteArray("10"), |
| dbus.ByteArray("11"), dbus.ByteArray("12"), |
| dbus.ByteArray("13"), dbus.ByteArray("14"), |
| dbus.ByteArray("15"), dbus.ByteArray("16"), |
| dbus.ByteArray("17") ]}, |
| "InvalidArgs"), |
| ({'Type': 'active', |
| 'SSIDs': [ dbus.ByteArray("1234567890abcdef1234567890abcdef1") ]}, |
| "InvalidArgs"), |
| ({'Type': 'active', 'IEs': 'foo'}, "InvalidArgs"), |
| ({'Type': 'active', 'IEs': ['foo']}, "InvalidArgs"), |
| ({'Type': 'active', 'Channels': 2412 }, "InvalidArgs"), |
| ({'Type': 'active', 'Channels': [ 2412 ] }, "InvalidArgs"), |
| ({'Type': 'active', |
| 'Channels': [ (dbus.Int32(2412), dbus.UInt32(20)) ] }, |
| "InvalidArgs"), |
| ({'Type': 'active', |
| 'Channels': [ (dbus.UInt32(2412), dbus.Int32(20)) ] }, |
| "InvalidArgs"), |
| ({'Type': 'active', 'AllowRoam': "yes" }, "InvalidArgs"), |
| ({'Type': 'passive', 'IEs': [ dbus.ByteArray("\xdd\x00") ]}, |
| "InvalidArgs"), |
| ({'Type': 'passive', 'SSIDs': [ dbus.ByteArray("foo") ]}, |
| "InvalidArgs")] |
| for (t,err) in tests: |
| try: |
| iface.Scan(t) |
| raise Exception("Invalid Scan() arguments accepted: " + str(t)) |
| except dbus.exceptions.DBusException, e: |
| if err not in str(e): |
| raise Exception("Unexpected error message for invalid Scan(%s): %s" % (str(t), str(e))) |
| |
| def test_dbus_scan_oom(dev, apdev): |
| """D-Bus scan method and OOM""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| with alloc_fail_dbus(dev[0], 1, |
| "wpa_scan_clone_params;wpas_dbus_handler_scan", |
| "Scan", expected="ScanError: Scan request rejected"): |
| iface.Scan({ 'Type': 'passive', |
| 'Channels': [ (dbus.UInt32(2412), dbus.UInt32(20)) ] }) |
| |
| with alloc_fail_dbus(dev[0], 1, |
| "=wpas_dbus_get_scan_channels;wpas_dbus_handler_scan", |
| "Scan"): |
| iface.Scan({ 'Type': 'passive', |
| 'Channels': [ (dbus.UInt32(2412), dbus.UInt32(20)) ] }) |
| |
| with alloc_fail_dbus(dev[0], 1, |
| "=wpas_dbus_get_scan_ies;wpas_dbus_handler_scan", |
| "Scan"): |
| iface.Scan({ 'Type': 'active', |
| 'IEs': [ dbus.ByteArray("\xdd\x00") ], |
| 'Channels': [ (dbus.UInt32(2412), dbus.UInt32(20)) ] }) |
| |
| with alloc_fail_dbus(dev[0], 1, |
| "=wpas_dbus_get_scan_ssids;wpas_dbus_handler_scan", |
| "Scan"): |
| iface.Scan({ 'Type': 'active', |
| 'SSIDs': [ dbus.ByteArray("open"), |
| dbus.ByteArray() ], |
| 'Channels': [ (dbus.UInt32(2412), dbus.UInt32(20)) ] }) |
| |
| def test_dbus_scan(dev, apdev): |
| """D-Bus scan and related signals""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| hapd = hostapd.add_ap(apdev[0]['ifname'], { "ssid": "open" }) |
| |
| class TestDbusScan(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.scan_completed = 0 |
| self.bss_added = False |
| self.fail_reason = None |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.run_scan) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.scanDone, WPAS_DBUS_IFACE, "ScanDone") |
| self.add_signal(self.bssAdded, WPAS_DBUS_IFACE, "BSSAdded") |
| self.add_signal(self.bssRemoved, WPAS_DBUS_IFACE, "BSSRemoved") |
| self.loop.run() |
| return self |
| |
| def scanDone(self, success): |
| logger.debug("scanDone: success=%s" % success) |
| self.scan_completed += 1 |
| if self.scan_completed == 1: |
| iface.Scan({'Type': 'passive', |
| 'AllowRoam': True, |
| 'Channels': [(dbus.UInt32(2412), dbus.UInt32(20))]}) |
| elif self.scan_completed == 2: |
| iface.Scan({'Type': 'passive', |
| 'AllowRoam': False}) |
| elif self.bss_added and self.scan_completed == 3: |
| self.loop.quit() |
| |
| def bssAdded(self, bss, properties): |
| logger.debug("bssAdded: %s" % bss) |
| logger.debug(str(properties)) |
| if 'WPS' in properties: |
| if 'Type' in properties['WPS']: |
| self.fail_reason = "Unexpected WPS dictionary entry in non-WPS BSS" |
| self.loop.quit() |
| self.bss_added = True |
| if self.scan_completed == 3: |
| self.loop.quit() |
| |
| def bssRemoved(self, bss): |
| logger.debug("bssRemoved: %s" % bss) |
| |
| def run_scan(self, *args): |
| logger.debug("run_scan") |
| iface.Scan({'Type': 'active', |
| 'SSIDs': [ dbus.ByteArray("open"), |
| dbus.ByteArray() ], |
| 'IEs': [ dbus.ByteArray("\xdd\x00"), |
| dbus.ByteArray() ], |
| 'AllowRoam': False, |
| 'Channels': [(dbus.UInt32(2412), dbus.UInt32(20))]}) |
| return False |
| |
| def success(self): |
| return self.scan_completed == 3 and self.bss_added |
| |
| with TestDbusScan(bus) as t: |
| if t.fail_reason: |
| raise Exception(t.fail_reason) |
| if not t.success(): |
| raise Exception("Expected signals not seen") |
| |
| res = if_obj.Get(WPAS_DBUS_IFACE, "BSSs", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if len(res) < 1: |
| raise Exception("Scan result not in BSSs property") |
| iface.FlushBSS(0) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "BSSs", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if len(res) != 0: |
| raise Exception("FlushBSS() did not remove scan results from BSSs property") |
| iface.FlushBSS(1) |
| |
| def test_dbus_scan_busy(dev, apdev): |
| """D-Bus scan trigger rejection when busy with previous scan""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| if "OK" not in dev[0].request("SCAN freq=2412-2462"): |
| raise Exception("Failed to start scan") |
| ev = dev[0].wait_event(["CTRL-EVENT-SCAN-STARTED"], 15) |
| if ev is None: |
| raise Exception("Scan start timed out") |
| |
| try: |
| iface.Scan({'Type': 'active', 'AllowRoam': False}) |
| raise Exception("Scan() accepted when busy") |
| except dbus.exceptions.DBusException, e: |
| if "ScanError: Scan request reject" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"], 15) |
| if ev is None: |
| raise Exception("Scan timed out") |
| |
| def test_dbus_connect(dev, apdev): |
| """D-Bus AddNetwork and connect""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| ssid = "test-wpa2-psk" |
| passphrase = 'qwertyuiop' |
| params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase) |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| |
| class TestDbusConnect(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.network_added = False |
| self.network_selected = False |
| self.network_removed = False |
| self.state = 0 |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.run_connect) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.networkAdded, WPAS_DBUS_IFACE, "NetworkAdded") |
| self.add_signal(self.networkRemoved, WPAS_DBUS_IFACE, |
| "NetworkRemoved") |
| self.add_signal(self.networkSelected, WPAS_DBUS_IFACE, |
| "NetworkSelected") |
| self.add_signal(self.propertiesChanged, WPAS_DBUS_IFACE, |
| "PropertiesChanged") |
| self.loop.run() |
| return self |
| |
| def networkAdded(self, network, properties): |
| logger.debug("networkAdded: %s" % str(network)) |
| logger.debug(str(properties)) |
| self.network_added = True |
| |
| def networkRemoved(self, network): |
| logger.debug("networkRemoved: %s" % str(network)) |
| self.network_removed = True |
| |
| def networkSelected(self, network): |
| logger.debug("networkSelected: %s" % str(network)) |
| self.network_selected = True |
| |
| def propertiesChanged(self, properties): |
| logger.debug("propertiesChanged: %s" % str(properties)) |
| if 'State' in properties and properties['State'] == "completed": |
| if self.state == 0: |
| self.state = 1 |
| iface.Disconnect() |
| elif self.state == 2: |
| self.state = 3 |
| iface.Disconnect() |
| elif self.state == 4: |
| self.state = 5 |
| iface.Reattach() |
| elif self.state == 5: |
| self.state = 6 |
| res = iface.SignalPoll() |
| logger.debug("SignalPoll: " + str(res)) |
| if 'frequency' not in res or res['frequency'] != 2412: |
| self.state = -1 |
| logger.info("Unexpected SignalPoll result") |
| iface.RemoveNetwork(self.netw) |
| if 'State' in properties and properties['State'] == "disconnected": |
| if self.state == 1: |
| self.state = 2 |
| iface.SelectNetwork(self.netw) |
| elif self.state == 3: |
| self.state = 4 |
| iface.Reassociate() |
| elif self.state == 6: |
| self.state = 7 |
| self.loop.quit() |
| |
| def run_connect(self, *args): |
| logger.debug("run_connect") |
| args = dbus.Dictionary({ 'ssid': ssid, |
| 'key_mgmt': 'WPA-PSK', |
| 'psk': passphrase, |
| 'scan_freq': 2412 }, |
| signature='sv') |
| self.netw = iface.AddNetwork(args) |
| iface.SelectNetwork(self.netw) |
| return False |
| |
| def success(self): |
| if not self.network_added or \ |
| not self.network_removed or \ |
| not self.network_selected: |
| return False |
| return self.state == 7 |
| |
| with TestDbusConnect(bus) as t: |
| if not t.success(): |
| raise Exception("Expected signals not seen") |
| |
| def test_dbus_connect_psk_mem(dev, apdev): |
| """D-Bus AddNetwork and connect with memory-only PSK""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| ssid = "test-wpa2-psk" |
| passphrase = 'qwertyuiop' |
| params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase) |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| |
| class TestDbusConnect(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.connected = False |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.run_connect) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.propertiesChanged, WPAS_DBUS_IFACE, |
| "PropertiesChanged") |
| self.add_signal(self.networkRequest, WPAS_DBUS_IFACE, |
| "NetworkRequest") |
| self.loop.run() |
| return self |
| |
| def propertiesChanged(self, properties): |
| logger.debug("propertiesChanged: %s" % str(properties)) |
| if 'State' in properties and properties['State'] == "completed": |
| self.connected = True |
| self.loop.quit() |
| |
| def networkRequest(self, path, field, txt): |
| logger.debug("networkRequest: %s %s %s" % (path, field, txt)) |
| if field == "PSK_PASSPHRASE": |
| iface.NetworkReply(path, field, '"' + passphrase + '"') |
| |
| def run_connect(self, *args): |
| logger.debug("run_connect") |
| args = dbus.Dictionary({ 'ssid': ssid, |
| 'key_mgmt': 'WPA-PSK', |
| 'mem_only_psk': 1, |
| 'scan_freq': 2412 }, |
| signature='sv') |
| self.netw = iface.AddNetwork(args) |
| iface.SelectNetwork(self.netw) |
| return False |
| |
| def success(self): |
| return self.connected |
| |
| with TestDbusConnect(bus) as t: |
| if not t.success(): |
| raise Exception("Expected signals not seen") |
| |
| def test_dbus_connect_oom(dev, apdev): |
| """D-Bus AddNetwork and connect when out-of-memory""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| if "OK" not in dev[0].request("TEST_ALLOC_FAIL 0:"): |
| raise HwsimSkip("TEST_ALLOC_FAIL not supported in the build") |
| |
| ssid = "test-wpa2-psk" |
| passphrase = 'qwertyuiop' |
| params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase) |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| |
| class TestDbusConnect(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.network_added = False |
| self.network_selected = False |
| self.network_removed = False |
| self.state = 0 |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.run_connect) |
| gobject.timeout_add(1500, self.timeout) |
| self.add_signal(self.networkAdded, WPAS_DBUS_IFACE, "NetworkAdded") |
| self.add_signal(self.networkRemoved, WPAS_DBUS_IFACE, |
| "NetworkRemoved") |
| self.add_signal(self.networkSelected, WPAS_DBUS_IFACE, |
| "NetworkSelected") |
| self.add_signal(self.propertiesChanged, WPAS_DBUS_IFACE, |
| "PropertiesChanged") |
| self.loop.run() |
| return self |
| |
| def networkAdded(self, network, properties): |
| logger.debug("networkAdded: %s" % str(network)) |
| logger.debug(str(properties)) |
| self.network_added = True |
| |
| def networkRemoved(self, network): |
| logger.debug("networkRemoved: %s" % str(network)) |
| self.network_removed = True |
| |
| def networkSelected(self, network): |
| logger.debug("networkSelected: %s" % str(network)) |
| self.network_selected = True |
| |
| def propertiesChanged(self, properties): |
| logger.debug("propertiesChanged: %s" % str(properties)) |
| if 'State' in properties and properties['State'] == "completed": |
| if self.state == 0: |
| self.state = 1 |
| iface.Disconnect() |
| elif self.state == 2: |
| self.state = 3 |
| iface.Disconnect() |
| elif self.state == 4: |
| self.state = 5 |
| iface.Reattach() |
| elif self.state == 5: |
| self.state = 6 |
| res = iface.SignalPoll() |
| logger.debug("SignalPoll: " + str(res)) |
| if 'frequency' not in res or res['frequency'] != 2412: |
| self.state = -1 |
| logger.info("Unexpected SignalPoll result") |
| iface.RemoveNetwork(self.netw) |
| if 'State' in properties and properties['State'] == "disconnected": |
| if self.state == 1: |
| self.state = 2 |
| iface.SelectNetwork(self.netw) |
| elif self.state == 3: |
| self.state = 4 |
| iface.Reassociate() |
| elif self.state == 6: |
| self.state = 7 |
| self.loop.quit() |
| |
| def run_connect(self, *args): |
| logger.debug("run_connect") |
| args = dbus.Dictionary({ 'ssid': ssid, |
| 'key_mgmt': 'WPA-PSK', |
| 'psk': passphrase, |
| 'scan_freq': 2412 }, |
| signature='sv') |
| try: |
| self.netw = iface.AddNetwork(args) |
| except Exception, e: |
| logger.info("Exception on AddNetwork: " + str(e)) |
| self.loop.quit() |
| return False |
| try: |
| iface.SelectNetwork(self.netw) |
| except Exception, e: |
| logger.info("Exception on SelectNetwork: " + str(e)) |
| self.loop.quit() |
| |
| return False |
| |
| def success(self): |
| if not self.network_added or \ |
| not self.network_removed or \ |
| not self.network_selected: |
| return False |
| return self.state == 7 |
| |
| count = 0 |
| for i in range(1, 1000): |
| for j in range(3): |
| dev[j].dump_monitor() |
| dev[0].request("TEST_ALLOC_FAIL %d:main" % i) |
| try: |
| with TestDbusConnect(bus) as t: |
| if not t.success(): |
| logger.info("Iteration %d - Expected signals not seen" % i) |
| else: |
| logger.info("Iteration %d - success" % i) |
| |
| state = dev[0].request('GET_ALLOC_FAIL') |
| logger.info("GET_ALLOC_FAIL: " + state) |
| dev[0].dump_monitor() |
| dev[0].request("TEST_ALLOC_FAIL 0:") |
| if i < 3: |
| raise Exception("Connection succeeded during out-of-memory") |
| if not state.startswith('0:'): |
| count += 1 |
| if count == 5: |
| break |
| except: |
| pass |
| |
| def test_dbus_while_not_connected(dev, apdev): |
| """D-Bus invalid operations while not connected""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| try: |
| iface.Disconnect() |
| raise Exception("Disconnect() accepted when not connected") |
| except dbus.exceptions.DBusException, e: |
| if "NotConnected" not in str(e): |
| raise Exception("Unexpected error message for invalid Disconnect: " + str(e)) |
| |
| try: |
| iface.Reattach() |
| raise Exception("Reattach() accepted when not connected") |
| except dbus.exceptions.DBusException, e: |
| if "NotConnected" not in str(e): |
| raise Exception("Unexpected error message for invalid Reattach: " + str(e)) |
| |
| def test_dbus_connect_eap(dev, apdev): |
| """D-Bus AddNetwork and connect to EAP network""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| ssid = "ieee8021x-open" |
| params = hostapd.radius_params() |
| params["ssid"] = ssid |
| params["ieee8021x"] = "1" |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| |
| class TestDbusConnect(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.certification_received = False |
| self.eap_status = False |
| self.state = 0 |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.run_connect) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.propertiesChanged, WPAS_DBUS_IFACE, |
| "PropertiesChanged") |
| self.add_signal(self.certification, WPAS_DBUS_IFACE, |
| "Certification", byte_arrays=True) |
| self.add_signal(self.networkRequest, WPAS_DBUS_IFACE, |
| "NetworkRequest") |
| self.add_signal(self.eap, WPAS_DBUS_IFACE, "EAP") |
| self.loop.run() |
| return self |
| |
| def propertiesChanged(self, properties): |
| logger.debug("propertiesChanged: %s" % str(properties)) |
| if 'State' in properties and properties['State'] == "completed": |
| if self.state == 0: |
| self.state = 1 |
| iface.EAPLogoff() |
| logger.info("Set dNSName constraint") |
| net_obj = bus.get_object(WPAS_DBUS_SERVICE, self.netw) |
| args = dbus.Dictionary({ 'altsubject_match': |
| self.server_dnsname }, |
| signature='sv') |
| net_obj.Set(WPAS_DBUS_NETWORK, "Properties", args, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| elif self.state == 2: |
| self.state = 3 |
| iface.Disconnect() |
| logger.info("Set non-matching dNSName constraint") |
| net_obj = bus.get_object(WPAS_DBUS_SERVICE, self.netw) |
| args = dbus.Dictionary({ 'altsubject_match': |
| self.server_dnsname + "FOO" }, |
| signature='sv') |
| net_obj.Set(WPAS_DBUS_NETWORK, "Properties", args, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if 'State' in properties and properties['State'] == "disconnected": |
| if self.state == 1: |
| self.state = 2 |
| iface.EAPLogon() |
| iface.SelectNetwork(self.netw) |
| if self.state == 3: |
| self.state = 4 |
| iface.SelectNetwork(self.netw) |
| |
| def certification(self, args): |
| logger.debug("certification: %s" % str(args)) |
| self.certification_received = True |
| if args['depth'] == 0: |
| # The test server certificate is supposed to have dNSName |
| if len(args['altsubject']) < 1: |
| raise Exception("Missing dNSName") |
| dnsname = args['altsubject'][0] |
| if not dnsname.startswith("DNS:"): |
| raise Exception("Expected dNSName not found: " + dnsname) |
| logger.info("altsubject: " + dnsname) |
| self.server_dnsname = dnsname |
| |
| def eap(self, status, parameter): |
| logger.debug("EAP: status=%s parameter=%s" % (status, parameter)) |
| if status == 'completion' and parameter == 'success': |
| self.eap_status = True |
| if self.state == 4 and status == 'remote certificate verification' and parameter == 'AltSubject mismatch': |
| self.state = 5 |
| self.loop.quit() |
| |
| def networkRequest(self, path, field, txt): |
| logger.debug("networkRequest: %s %s %s" % (path, field, txt)) |
| if field == "PASSWORD": |
| iface.NetworkReply(path, field, "password") |
| |
| def run_connect(self, *args): |
| logger.debug("run_connect") |
| args = dbus.Dictionary({ 'ssid': ssid, |
| 'key_mgmt': 'IEEE8021X', |
| 'eapol_flags': 0, |
| 'eap': 'TTLS', |
| 'anonymous_identity': 'ttls', |
| 'identity': 'pap user', |
| 'ca_cert': 'auth_serv/ca.pem', |
| 'phase2': 'auth=PAP', |
| 'scan_freq': 2412 }, |
| signature='sv') |
| self.netw = iface.AddNetwork(args) |
| iface.SelectNetwork(self.netw) |
| return False |
| |
| def success(self): |
| if not self.eap_status or not self.certification_received: |
| return False |
| return self.state == 5 |
| |
| with TestDbusConnect(bus) as t: |
| if not t.success(): |
| raise Exception("Expected signals not seen") |
| |
| def test_dbus_network(dev, apdev): |
| """D-Bus AddNetwork/RemoveNetwork parameters and error cases""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| args = dbus.Dictionary({ 'ssid': "foo", |
| 'key_mgmt': 'WPA-PSK', |
| 'psk': "12345678", |
| 'identity': dbus.ByteArray([ 1, 2 ]), |
| 'priority': dbus.Int32(0), |
| 'scan_freq': dbus.UInt32(2412) }, |
| signature='sv') |
| netw = iface.AddNetwork(args) |
| iface.RemoveNetwork(netw) |
| try: |
| iface.RemoveNetwork(netw) |
| raise Exception("Invalid RemoveNetwork() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "NetworkUnknown" not in str(e): |
| raise Exception("Unexpected error message for invalid RemoveNetwork: " + str(e)) |
| try: |
| iface.SelectNetwork(netw) |
| raise Exception("Invalid SelectNetwork() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "NetworkUnknown" not in str(e): |
| raise Exception("Unexpected error message for invalid RemoveNetwork: " + str(e)) |
| |
| args = dbus.Dictionary({ 'ssid': "foo1", 'key_mgmt': 'NONE', |
| 'identity': "testuser", 'scan_freq': '2412' }, |
| signature='sv') |
| netw1 = iface.AddNetwork(args) |
| args = dbus.Dictionary({ 'ssid': "foo2", 'key_mgmt': 'NONE' }, |
| signature='sv') |
| netw2 = iface.AddNetwork(args) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "Networks", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if len(res) != 2: |
| raise Exception("Unexpected number of networks") |
| |
| net_obj = bus.get_object(WPAS_DBUS_SERVICE, netw1) |
| res = net_obj.Get(WPAS_DBUS_NETWORK, "Enabled", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != False: |
| raise Exception("Added network was unexpectedly enabled by default") |
| net_obj.Set(WPAS_DBUS_NETWORK, "Enabled", dbus.Boolean(True), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| res = net_obj.Get(WPAS_DBUS_NETWORK, "Enabled", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != True: |
| raise Exception("Set(Enabled,True) did not seem to change property value") |
| net_obj.Set(WPAS_DBUS_NETWORK, "Enabled", dbus.Boolean(False), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| res = net_obj.Get(WPAS_DBUS_NETWORK, "Enabled", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != False: |
| raise Exception("Set(Enabled,False) did not seem to change property value") |
| try: |
| net_obj.Set(WPAS_DBUS_NETWORK, "Enabled", dbus.UInt32(1), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(Enabled,1) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: wrong property type" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(Enabled,1): " + str(e)) |
| |
| args = dbus.Dictionary({ 'ssid': "foo1new" }, signature='sv') |
| net_obj.Set(WPAS_DBUS_NETWORK, "Properties", args, |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| res = net_obj.Get(WPAS_DBUS_NETWORK, "Properties", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res['ssid'] != '"foo1new"': |
| raise Exception("Set(Properties) failed to update ssid") |
| if res['identity'] != '"testuser"': |
| raise Exception("Set(Properties) unexpectedly changed unrelated parameter") |
| |
| iface.RemoveAllNetworks() |
| res = if_obj.Get(WPAS_DBUS_IFACE, "Networks", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if len(res) != 0: |
| raise Exception("Unexpected number of networks") |
| iface.RemoveAllNetworks() |
| |
| tests = [ dbus.Dictionary({ 'psk': "1234567" }, signature='sv'), |
| dbus.Dictionary({ 'identity': dbus.ByteArray() }, |
| signature='sv'), |
| dbus.Dictionary({ 'identity': dbus.Byte(1) }, signature='sv'), |
| dbus.Dictionary({ 'identity': "" }, signature='sv') ] |
| for args in tests: |
| try: |
| iface.AddNetwork(args) |
| raise Exception("Invalid AddNetwork args accepted: " + str(args)) |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid AddNetwork: " + str(e)) |
| |
| def test_dbus_network_oom(dev, apdev): |
| """D-Bus AddNetwork/RemoveNetwork parameters and OOM error cases""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| args = dbus.Dictionary({ 'ssid': "foo1", 'key_mgmt': 'NONE', |
| 'identity': "testuser", 'scan_freq': '2412' }, |
| signature='sv') |
| netw1 = iface.AddNetwork(args) |
| net_obj = bus.get_object(WPAS_DBUS_SERVICE, netw1) |
| |
| with alloc_fail_dbus(dev[0], 1, |
| "wpa_config_get_all;wpas_dbus_getter_network_properties", |
| "Get"): |
| net_obj.Get(WPAS_DBUS_NETWORK, "Properties", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| iface.RemoveAllNetworks() |
| |
| with alloc_fail_dbus(dev[0], 1, |
| "wpas_dbus_new_decompose_object_path;wpas_dbus_handler_remove_network", |
| "RemoveNetwork", "InvalidArgs"): |
| iface.RemoveNetwork(dbus.ObjectPath("/fi/w1/wpa_supplicant1/Interfaces/1234/Networks/1234")) |
| |
| with alloc_fail(dev[0], 1, "wpa_dbus_register_object_per_iface;wpas_dbus_register_network"): |
| args = dbus.Dictionary({ 'ssid': "foo2", 'key_mgmt': 'NONE' }, |
| signature='sv') |
| try: |
| netw = iface.AddNetwork(args) |
| # Currently, AddNetwork() succeeds even if os_strdup() for path |
| # fails, so remove the network if that occurs. |
| iface.RemoveNetwork(netw) |
| except dbus.exceptions.DBusException, e: |
| pass |
| |
| for i in range(1, 3): |
| with alloc_fail(dev[0], i, "=wpas_dbus_register_network"): |
| try: |
| netw = iface.AddNetwork(args) |
| # Currently, AddNetwork() succeeds even if network registration |
| # fails, so remove the network if that occurs. |
| iface.RemoveNetwork(netw) |
| except dbus.exceptions.DBusException, e: |
| pass |
| |
| with alloc_fail_dbus(dev[0], 1, |
| "=wpa_config_add_network;wpas_dbus_handler_add_network", |
| "AddNetwork", |
| "UnknownError: wpa_supplicant could not add a network"): |
| args = dbus.Dictionary({ 'ssid': "foo2", 'key_mgmt': 'NONE' }, |
| signature='sv') |
| netw = iface.AddNetwork(args) |
| |
| tests = [ (1, |
| 'wpa_dbus_dict_get_entry;set_network_properties;wpas_dbus_handler_add_network', |
| dbus.Dictionary({ 'ssid': dbus.ByteArray(' ') }, |
| signature='sv')), |
| (1, '=set_network_properties;wpas_dbus_handler_add_network', |
| dbus.Dictionary({ 'ssid': 'foo' }, signature='sv')), |
| (1, '=set_network_properties;wpas_dbus_handler_add_network', |
| dbus.Dictionary({ 'eap': 'foo' }, signature='sv')), |
| (1, '=set_network_properties;wpas_dbus_handler_add_network', |
| dbus.Dictionary({ 'priority': dbus.UInt32(1) }, |
| signature='sv')), |
| (1, '=set_network_properties;wpas_dbus_handler_add_network', |
| dbus.Dictionary({ 'priority': dbus.Int32(1) }, |
| signature='sv')), |
| (1, '=set_network_properties;wpas_dbus_handler_add_network', |
| dbus.Dictionary({ 'ssid': dbus.ByteArray(' ') }, |
| signature='sv')) ] |
| for (count,funcs,args) in tests: |
| with alloc_fail_dbus(dev[0], count, funcs, "AddNetwork", "InvalidArgs"): |
| netw = iface.AddNetwork(args) |
| |
| if len(if_obj.Get(WPAS_DBUS_IFACE, 'Networks', |
| dbus_interface=dbus.PROPERTIES_IFACE)) > 0: |
| raise Exception("Unexpected network block added") |
| if len(dev[0].list_networks()) > 0: |
| raise Exception("Unexpected network block visible") |
| |
| def test_dbus_interface(dev, apdev): |
| """D-Bus CreateInterface/GetInterface/RemoveInterface parameters and error cases""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wpas = dbus.Interface(wpas_obj, WPAS_DBUS_SERVICE) |
| |
| params = dbus.Dictionary({ 'Ifname': 'lo', 'Driver': 'none' }, |
| signature='sv') |
| path = wpas.CreateInterface(params) |
| logger.debug("New interface path: " + str(path)) |
| path2 = wpas.GetInterface("lo") |
| if path != path2: |
| raise Exception("Interface object mismatch") |
| |
| params = dbus.Dictionary({ 'Ifname': 'lo', |
| 'Driver': 'none', |
| 'ConfigFile': 'foo', |
| 'BridgeIfname': 'foo', }, |
| signature='sv') |
| try: |
| wpas.CreateInterface(params) |
| raise Exception("Invalid CreateInterface() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InterfaceExists" not in str(e): |
| raise Exception("Unexpected error message for invalid CreateInterface: " + str(e)) |
| |
| wpas.RemoveInterface(path) |
| try: |
| wpas.RemoveInterface(path) |
| raise Exception("Invalid RemoveInterface() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InterfaceUnknown" not in str(e): |
| raise Exception("Unexpected error message for invalid RemoveInterface: " + str(e)) |
| |
| params = dbus.Dictionary({ 'Ifname': 'lo', 'Driver': 'none', |
| 'Foo': 123 }, |
| signature='sv') |
| try: |
| wpas.CreateInterface(params) |
| raise Exception("Invalid CreateInterface() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid CreateInterface: " + str(e)) |
| |
| params = dbus.Dictionary({ 'Driver': 'none' }, signature='sv') |
| try: |
| wpas.CreateInterface(params) |
| raise Exception("Invalid CreateInterface() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid CreateInterface: " + str(e)) |
| |
| try: |
| wpas.GetInterface("lo") |
| raise Exception("Invalid GetInterface() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InterfaceUnknown" not in str(e): |
| raise Exception("Unexpected error message for invalid RemoveInterface: " + str(e)) |
| |
| def test_dbus_interface_oom(dev, apdev): |
| """D-Bus CreateInterface/GetInterface/RemoveInterface OOM error cases""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| wpas = dbus.Interface(wpas_obj, WPAS_DBUS_SERVICE) |
| |
| with alloc_fail_dbus(dev[0], 1, "wpa_dbus_dict_get_entry;wpas_dbus_handler_create_interface", "CreateInterface", "InvalidArgs"): |
| params = dbus.Dictionary({ 'Ifname': 'lo', 'Driver': 'none' }, |
| signature='sv') |
| wpas.CreateInterface(params) |
| |
| for i in range(1, 1000): |
| dev[0].request("TEST_ALLOC_FAIL %d:wpa_supplicant_add_iface;wpas_dbus_handler_create_interface" % i) |
| params = dbus.Dictionary({ 'Ifname': 'lo', 'Driver': 'none' }, |
| signature='sv') |
| try: |
| npath = wpas.CreateInterface(params) |
| wpas.RemoveInterface(npath) |
| logger.info("CreateInterface succeeds after %d allocation failures" % i) |
| state = dev[0].request('GET_ALLOC_FAIL') |
| logger.info("GET_ALLOC_FAIL: " + state) |
| dev[0].dump_monitor() |
| dev[0].request("TEST_ALLOC_FAIL 0:") |
| if i < 5: |
| raise Exception("CreateInterface succeeded during out-of-memory") |
| if not state.startswith('0:'): |
| break |
| except dbus.exceptions.DBusException, e: |
| pass |
| |
| for arg in [ 'Driver', 'Ifname', 'ConfigFile', 'BridgeIfname' ]: |
| with alloc_fail_dbus(dev[0], 1, "=wpas_dbus_handler_create_interface", |
| "CreateInterface"): |
| params = dbus.Dictionary({ arg: 'foo' }, signature='sv') |
| wpas.CreateInterface(params) |
| |
| def test_dbus_blob(dev, apdev): |
| """D-Bus AddNetwork/RemoveNetwork parameters and error cases""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| blob = dbus.ByteArray("\x01\x02\x03") |
| iface.AddBlob('blob1', blob) |
| try: |
| iface.AddBlob('blob1', dbus.ByteArray("\x01\x02\x04")) |
| raise Exception("Invalid AddBlob() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "BlobExists" not in str(e): |
| raise Exception("Unexpected error message for invalid AddBlob: " + str(e)) |
| res = iface.GetBlob('blob1') |
| if len(res) != len(blob): |
| raise Exception("Unexpected blob data length") |
| for i in range(len(res)): |
| if res[i] != dbus.Byte(blob[i]): |
| raise Exception("Unexpected blob data") |
| res = if_obj.Get(WPAS_DBUS_IFACE, "Blobs", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if 'blob1' not in res: |
| raise Exception("Added blob missing from Blobs property") |
| iface.RemoveBlob('blob1') |
| try: |
| iface.RemoveBlob('blob1') |
| raise Exception("Invalid RemoveBlob() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "BlobUnknown" not in str(e): |
| raise Exception("Unexpected error message for invalid RemoveBlob: " + str(e)) |
| try: |
| iface.GetBlob('blob1') |
| raise Exception("Invalid GetBlob() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "BlobUnknown" not in str(e): |
| raise Exception("Unexpected error message for invalid GetBlob: " + str(e)) |
| |
| class TestDbusBlob(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.blob_added = False |
| self.blob_removed = False |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.run_blob) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.blobAdded, WPAS_DBUS_IFACE, "BlobAdded") |
| self.add_signal(self.blobRemoved, WPAS_DBUS_IFACE, "BlobRemoved") |
| self.loop.run() |
| return self |
| |
| def blobAdded(self, blobName): |
| logger.debug("blobAdded: %s" % blobName) |
| if blobName == 'blob2': |
| self.blob_added = True |
| |
| def blobRemoved(self, blobName): |
| logger.debug("blobRemoved: %s" % blobName) |
| if blobName == 'blob2': |
| self.blob_removed = True |
| self.loop.quit() |
| |
| def run_blob(self, *args): |
| logger.debug("run_blob") |
| iface.AddBlob('blob2', dbus.ByteArray("\x01\x02\x04")) |
| iface.RemoveBlob('blob2') |
| return False |
| |
| def success(self): |
| return self.blob_added and self.blob_removed |
| |
| with TestDbusBlob(bus) as t: |
| if not t.success(): |
| raise Exception("Expected signals not seen") |
| |
| def test_dbus_blob_oom(dev, apdev): |
| """D-Bus AddNetwork/RemoveNetwork OOM error cases""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| for i in range(1, 4): |
| with alloc_fail_dbus(dev[0], i, "wpas_dbus_handler_add_blob", |
| "AddBlob"): |
| iface.AddBlob('blob_no_mem', dbus.ByteArray("\x01\x02\x03\x04")) |
| |
| def test_dbus_autoscan(dev, apdev): |
| """D-Bus Autoscan()""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| iface.AutoScan("foo") |
| iface.AutoScan("periodic:1") |
| iface.AutoScan("") |
| dev[0].request("AUTOSCAN ") |
| |
| def test_dbus_autoscan_oom(dev, apdev): |
| """D-Bus Autoscan() OOM""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| with alloc_fail_dbus(dev[0], 1, "wpas_dbus_handler_autoscan", "AutoScan"): |
| iface.AutoScan("foo") |
| dev[0].request("AUTOSCAN ") |
| |
| def test_dbus_tdls_invalid(dev, apdev): |
| """D-Bus invalid TDLS operations""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| hapd = hostapd.add_ap(apdev[0]['ifname'], { "ssid": "test-open" }) |
| connect_2sta_open(dev, hapd) |
| addr1 = dev[1].p2p_interface_addr() |
| |
| try: |
| iface.TDLSDiscover("foo") |
| raise Exception("Invalid TDLSDiscover() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid TDLSDiscover: " + str(e)) |
| |
| try: |
| iface.TDLSStatus("foo") |
| raise Exception("Invalid TDLSStatus() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid TDLSStatus: " + str(e)) |
| |
| res = iface.TDLSStatus(addr1) |
| if res != "peer does not exist": |
| raise Exception("Unexpected TDLSStatus response") |
| |
| try: |
| iface.TDLSSetup("foo") |
| raise Exception("Invalid TDLSSetup() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid TDLSSetup: " + str(e)) |
| |
| try: |
| iface.TDLSTeardown("foo") |
| raise Exception("Invalid TDLSTeardown() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid TDLSTeardown: " + str(e)) |
| |
| try: |
| iface.TDLSTeardown("00:11:22:33:44:55") |
| raise Exception("TDLSTeardown accepted for unknown peer") |
| except dbus.exceptions.DBusException, e: |
| if "UnknownError: error performing TDLS teardown" not in str(e): |
| raise Exception("Unexpected error message: " + str(e)) |
| |
| def test_dbus_tdls_oom(dev, apdev): |
| """D-Bus TDLS operations during OOM""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| with alloc_fail_dbus(dev[0], 1, "wpa_tdls_add_peer", "TDLSSetup", |
| "UnknownError: error performing TDLS setup"): |
| iface.TDLSSetup("00:11:22:33:44:55") |
| |
| def test_dbus_tdls(dev, apdev): |
| """D-Bus TDLS""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| hapd = hostapd.add_ap(apdev[0]['ifname'], { "ssid": "test-open" }) |
| connect_2sta_open(dev, hapd) |
| |
| addr1 = dev[1].p2p_interface_addr() |
| |
| class TestDbusTdls(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.tdls_setup = False |
| self.tdls_teardown = False |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.run_tdls) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.propertiesChanged, WPAS_DBUS_IFACE, |
| "PropertiesChanged") |
| self.loop.run() |
| return self |
| |
| def propertiesChanged(self, properties): |
| logger.debug("propertiesChanged: %s" % str(properties)) |
| |
| def run_tdls(self, *args): |
| logger.debug("run_tdls") |
| iface.TDLSDiscover(addr1) |
| gobject.timeout_add(100, self.run_tdls2) |
| return False |
| |
| def run_tdls2(self, *args): |
| logger.debug("run_tdls2") |
| iface.TDLSSetup(addr1) |
| gobject.timeout_add(500, self.run_tdls3) |
| return False |
| |
| def run_tdls3(self, *args): |
| logger.debug("run_tdls3") |
| res = iface.TDLSStatus(addr1) |
| if res == "connected": |
| self.tdls_setup = True |
| else: |
| logger.info("Unexpected TDLSStatus: " + res) |
| iface.TDLSTeardown(addr1) |
| gobject.timeout_add(200, self.run_tdls4) |
| return False |
| |
| def run_tdls4(self, *args): |
| logger.debug("run_tdls4") |
| res = iface.TDLSStatus(addr1) |
| if res == "peer does not exist": |
| self.tdls_teardown = True |
| else: |
| logger.info("Unexpected TDLSStatus: " + res) |
| self.loop.quit() |
| return False |
| |
| def success(self): |
| return self.tdls_setup and self.tdls_teardown |
| |
| with TestDbusTdls(bus) as t: |
| if not t.success(): |
| raise Exception("Expected signals not seen") |
| |
| def test_dbus_pkcs11(dev, apdev): |
| """D-Bus SetPKCS11EngineAndModulePath()""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| try: |
| iface.SetPKCS11EngineAndModulePath("foo", "bar") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: Reinit of the EAPOL" not in str(e): |
| raise Exception("Unexpected error message for invalid SetPKCS11EngineAndModulePath: " + str(e)) |
| |
| try: |
| iface.SetPKCS11EngineAndModulePath("foo", "") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: Reinit of the EAPOL" not in str(e): |
| raise Exception("Unexpected error message for invalid SetPKCS11EngineAndModulePath: " + str(e)) |
| |
| iface.SetPKCS11EngineAndModulePath("", "bar") |
| res = if_obj.Get(WPAS_DBUS_IFACE, "PKCS11EnginePath", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != "": |
| raise Exception("Unexpected PKCS11EnginePath value: " + res) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "PKCS11ModulePath", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != "bar": |
| raise Exception("Unexpected PKCS11ModulePath value: " + res) |
| |
| iface.SetPKCS11EngineAndModulePath("", "") |
| res = if_obj.Get(WPAS_DBUS_IFACE, "PKCS11EnginePath", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != "": |
| raise Exception("Unexpected PKCS11EnginePath value: " + res) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "PKCS11ModulePath", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != "": |
| raise Exception("Unexpected PKCS11ModulePath value: " + res) |
| |
| def test_dbus_apscan(dev, apdev): |
| """D-Bus Get/Set ApScan""" |
| try: |
| _test_dbus_apscan(dev, apdev) |
| finally: |
| dev[0].request("AP_SCAN 1") |
| |
| def _test_dbus_apscan(dev, apdev): |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| |
| res = if_obj.Get(WPAS_DBUS_IFACE, "ApScan", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != 1: |
| raise Exception("Unexpected initial ApScan value: %d" % res) |
| |
| for i in range(3): |
| if_obj.Set(WPAS_DBUS_IFACE, "ApScan", dbus.UInt32(i), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "ApScan", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != i: |
| raise Exception("Unexpected ApScan value %d (expected %d)" % (res, i)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "ApScan", dbus.Int16(-1), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(ApScan,-1) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: wrong property type" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(ApScan,-1): " + str(e)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "ApScan", dbus.UInt32(123), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(ApScan,123) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: ap_scan must be 0, 1, or 2" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(ApScan,123): " + str(e)) |
| |
| if_obj.Set(WPAS_DBUS_IFACE, "ApScan", dbus.UInt32(1), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| def test_dbus_fastreauth(dev, apdev): |
| """D-Bus Get/Set FastReauth""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| |
| res = if_obj.Get(WPAS_DBUS_IFACE, "FastReauth", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != True: |
| raise Exception("Unexpected initial FastReauth value: " + str(res)) |
| |
| for i in [ False, True ]: |
| if_obj.Set(WPAS_DBUS_IFACE, "FastReauth", dbus.Boolean(i), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "FastReauth", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != i: |
| raise Exception("Unexpected FastReauth value %d (expected %d)" % (res, i)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "FastReauth", dbus.Int16(-1), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(FastReauth,-1) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: wrong property type" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(ApScan,-1): " + str(e)) |
| |
| if_obj.Set(WPAS_DBUS_IFACE, "FastReauth", dbus.Boolean(True), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| def test_dbus_bss_expire(dev, apdev): |
| """D-Bus Get/Set BSSExpireAge and BSSExpireCount""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| |
| if_obj.Set(WPAS_DBUS_IFACE, "BSSExpireAge", dbus.UInt32(179), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "BSSExpireAge", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != 179: |
| raise Exception("Unexpected BSSExpireAge value %d (expected %d)" % (res, i)) |
| |
| if_obj.Set(WPAS_DBUS_IFACE, "BSSExpireCount", dbus.UInt32(3), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "BSSExpireCount", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != 3: |
| raise Exception("Unexpected BSSExpireCount value %d (expected %d)" % (res, i)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "BSSExpireAge", dbus.Int16(-1), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(BSSExpireAge,-1) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: wrong property type" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(BSSExpireAge,-1): " + str(e)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "BSSExpireAge", dbus.UInt32(9), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(BSSExpireAge,9) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: BSSExpireAge must be >= 10" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(BSSExpireAge,9): " + str(e)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "BSSExpireCount", dbus.Int16(-1), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(BSSExpireCount,-1) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: wrong property type" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(BSSExpireCount,-1): " + str(e)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "BSSExpireCount", dbus.UInt32(0), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(BSSExpireCount,0) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: BSSExpireCount must be > 0" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(BSSExpireCount,0): " + str(e)) |
| |
| if_obj.Set(WPAS_DBUS_IFACE, "BSSExpireAge", dbus.UInt32(180), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if_obj.Set(WPAS_DBUS_IFACE, "BSSExpireCount", dbus.UInt32(2), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| def test_dbus_country(dev, apdev): |
| """D-Bus Get/Set Country""" |
| try: |
| _test_dbus_country(dev, apdev) |
| finally: |
| dev[0].request("SET country 00") |
| subprocess.call(['iw', 'reg', 'set', '00']) |
| |
| def _test_dbus_country(dev, apdev): |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| |
| # work around issues with possible pending regdom event from the end of |
| # the previous test case |
| time.sleep(0.2) |
| dev[0].dump_monitor() |
| |
| if_obj.Set(WPAS_DBUS_IFACE, "Country", "FI", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "Country", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != "FI": |
| raise Exception("Unexpected Country value %s (expected FI)" % res) |
| |
| ev = dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"]) |
| if ev is None: |
| raise Exception("regdom change event not seen") |
| if "init=USER type=COUNTRY alpha2=FI" not in ev: |
| raise Exception("Unexpected event contents: " + ev) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "Country", dbus.Int16(-1), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(Country,-1) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: wrong property type" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(Country,-1): " + str(e)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "Country", "F", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(Country,F) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: invalid country code" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(Country,F): " + str(e)) |
| |
| if_obj.Set(WPAS_DBUS_IFACE, "Country", "00", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| ev = dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"]) |
| if ev is None: |
| raise Exception("regdom change event not seen") |
| if "init=CORE type=WORLD" not in ev: |
| raise Exception("Unexpected event contents: " + ev) |
| |
| def test_dbus_scan_interval(dev, apdev): |
| """D-Bus Get/Set ScanInterval""" |
| try: |
| _test_dbus_scan_interval(dev, apdev) |
| finally: |
| dev[0].request("SCAN_INTERVAL 5") |
| |
| def _test_dbus_scan_interval(dev, apdev): |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| |
| if_obj.Set(WPAS_DBUS_IFACE, "ScanInterval", dbus.Int32(3), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| res = if_obj.Get(WPAS_DBUS_IFACE, "ScanInterval", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| if res != 3: |
| raise Exception("Unexpected ScanInterval value %d (expected %d)" % (res, i)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "ScanInterval", dbus.UInt16(100), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(ScanInterval,100) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: wrong property type" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(ScanInterval,100): " + str(e)) |
| |
| try: |
| if_obj.Set(WPAS_DBUS_IFACE, "ScanInterval", dbus.Int32(-1), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Set(ScanInterval,-1) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: scan_interval must be >= 0" not in str(e): |
| raise Exception("Unexpected error message for invalid Set(ScanInterval,-1): " + str(e)) |
| |
| if_obj.Set(WPAS_DBUS_IFACE, "ScanInterval", dbus.Int32(5), |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| |
| def test_dbus_probe_req_reporting(dev, apdev): |
| """D-Bus Probe Request reporting""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| dev[0].p2p_start_go(freq=2412) |
| dev[1].p2p_find(social=True) |
| |
| class TestDbusProbe(TestDbus): |
| def __init__(self, bus): |
| TestDbus.__init__(self, bus) |
| self.reported = False |
| |
| def __enter__(self): |
| gobject.timeout_add(1, self.run_test) |
| gobject.timeout_add(15000, self.timeout) |
| self.add_signal(self.probeRequest, WPAS_DBUS_IFACE, "ProbeRequest", |
| byte_arrays=True) |
| iface.SubscribeProbeReq() |
| self.loop.run() |
| return self |
| |
| def probeRequest(self, args): |
| logger.debug("probeRequest: args=%s" % str(args)) |
| self.reported = True |
| self.loop.quit() |
| |
| def run_test(self, *args): |
| logger.debug("run_test") |
| return False |
| |
| def success(self): |
| return self.reported |
| |
| with TestDbusProbe(bus) as t: |
| if not t.success(): |
| raise Exception("Expected signals not seen") |
| iface.UnsubscribeProbeReq() |
| |
| try: |
| iface.UnsubscribeProbeReq() |
| raise Exception("Invalid UnsubscribeProbeReq() accepted") |
| except dbus.exceptions.DBusException, e: |
| if "NoSubscription" not in str(e): |
| raise Exception("Unexpected error message for invalid UnsubscribeProbeReq(): " + str(e)) |
| |
| with TestDbusProbe(bus) as t: |
| if not t.success(): |
| raise Exception("Expected signals not seen") |
| # On purpose, leave ProbeReq subscription in place to test automatic |
| # cleanup. |
| |
| dev[1].p2p_stop_find() |
| dev[0].remove_group() |
| |
| def test_dbus_probe_req_reporting_oom(dev, apdev): |
| """D-Bus Probe Request reporting (OOM)""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| iface = dbus.Interface(if_obj, WPAS_DBUS_IFACE) |
| |
| # Need to make sure this process has not already subscribed to avoid false |
| # failures due to the operation succeeding due to os_strdup() not even |
| # getting called. |
| try: |
| iface.UnsubscribeProbeReq() |
| was_subscribed = True |
| except dbus.exceptions.DBusException, e: |
| was_subscribed = False |
| pass |
| |
| with alloc_fail_dbus(dev[0], 1, "wpas_dbus_handler_subscribe_preq", |
| "SubscribeProbeReq"): |
| iface.SubscribeProbeReq() |
| |
| if was_subscribed: |
| # On purpose, leave ProbeReq subscription in place to test automatic |
| # cleanup. |
| iface.SubscribeProbeReq() |
| |
| def test_dbus_p2p_invalid(dev, apdev): |
| """D-Bus invalid P2P operations""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| p2p = dbus.Interface(if_obj, WPAS_DBUS_IFACE_P2PDEVICE) |
| |
| try: |
| p2p.RejectPeer(path + "/Peers/00112233445566") |
| raise Exception("Invalid RejectPeer accepted") |
| except dbus.exceptions.DBusException, e: |
| if "UnknownError: Failed to call wpas_p2p_reject" not in str(e): |
| raise Exception("Unexpected error message for invalid RejectPeer(): " + str(e)) |
| |
| try: |
| p2p.RejectPeer("/foo") |
| raise Exception("Invalid RejectPeer accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid RejectPeer(): " + str(e)) |
| |
| tests = [ {'DiscoveryType': 'foo'}, |
| {'RequestedDeviceTypes': 'foo'}, |
| {'RequestedDeviceTypes': ['foo']}, |
| {'RequestedDeviceTypes': ['1','2','3','4','5','6','7','8','9', |
| '10','11','12','13','14','15','16', |
| '17']}, |
| {'RequestedDeviceTypes': dbus.Array([], signature="s")}, |
| {'RequestedDeviceTypes': dbus.Array([['foo']], signature="as")}, |
| {'RequestedDeviceTypes': dbus.Array([], signature="i")}, |
| {'RequestedDeviceTypes': [dbus.ByteArray('12345678'), |
| dbus.ByteArray('1234567')]}, |
| {'Foo': dbus.Int16(1)}, |
| {'Foo': dbus.UInt16(1)}, |
| {'Foo': dbus.Int64(1)}, |
| {'Foo': dbus.UInt64(1)}, |
| {'Foo': dbus.Double(1.23)}, |
| {'Foo': dbus.Signature('s')}, |
| {'Foo': 'bar'}] |
| for t in tests: |
| try: |
| p2p.Find(dbus.Dictionary(t)) |
| raise Exception("Invalid Find accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid Find(): " + str(e)) |
| |
| for p in [ "/foo", |
| "/fi/w1/wpa_supplicant1/Interfaces/1234", |
| "/fi/w1/wpa_supplicant1/Interfaces/1234/Networks/1234" ]: |
| try: |
| p2p.RemovePersistentGroup(dbus.ObjectPath(p)) |
| raise Exception("Invalid RemovePersistentGroup accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid RemovePersistentGroup: " + str(e)) |
| |
| try: |
| dev[0].request("P2P_SET disabled 1") |
| p2p.Listen(5) |
| raise Exception("Invalid Listen accepted") |
| except dbus.exceptions.DBusException, e: |
| if "UnknownError: Could not start P2P listen" not in str(e): |
| raise Exception("Unexpected error message for invalid Listen: " + str(e)) |
| finally: |
| dev[0].request("P2P_SET disabled 0") |
| |
| test_obj = bus.get_object(WPAS_DBUS_SERVICE, path, introspect=False) |
| test_p2p = dbus.Interface(test_obj, WPAS_DBUS_IFACE_P2PDEVICE) |
| try: |
| test_p2p.Listen("foo") |
| raise Exception("Invalid Listen accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid Listen: " + str(e)) |
| |
| try: |
| dev[0].request("P2P_SET disabled 1") |
| p2p.ExtendedListen(dbus.Dictionary({})) |
| raise Exception("Invalid ExtendedListen accepted") |
| except dbus.exceptions.DBusException, e: |
| if "UnknownError: failed to initiate a p2p_ext_listen" not in str(e): |
| raise Exception("Unexpected error message for invalid ExtendedListen: " + str(e)) |
| finally: |
| dev[0].request("P2P_SET disabled 0") |
| |
| try: |
| dev[0].request("P2P_SET disabled 1") |
| args = { 'duration1': 30000, 'interval1': 102400, |
| 'duration2': 20000, 'interval2': 102400 } |
| p2p.PresenceRequest(args) |
| raise Exception("Invalid PresenceRequest accepted") |
| except dbus.exceptions.DBusException, e: |
| if "UnknownError: Failed to invoke presence request" not in str(e): |
| raise Exception("Unexpected error message for invalid PresenceRequest: " + str(e)) |
| finally: |
| dev[0].request("P2P_SET disabled 0") |
| |
| try: |
| params = dbus.Dictionary({'frequency': dbus.Int32(-1)}) |
| p2p.GroupAdd(params) |
| raise Exception("Invalid GroupAdd accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid GroupAdd: " + str(e)) |
| |
| try: |
| params = dbus.Dictionary({'persistent_group_object': |
| dbus.ObjectPath(path), |
| 'frequency': 2412}) |
| p2p.GroupAdd(params) |
| raise Exception("Invalid GroupAdd accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid GroupAdd: " + str(e)) |
| |
| try: |
| p2p.Disconnect() |
| raise Exception("Invalid Disconnect accepted") |
| except dbus.exceptions.DBusException, e: |
| if "UnknownError: failed to disconnect" not in str(e): |
| raise Exception("Unexpected error message for invalid Disconnect: " + str(e)) |
| |
| try: |
| dev[0].request("P2P_SET disabled 1") |
| p2p.Flush() |
| raise Exception("Invalid Flush accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: P2P is not available for this interface" not in str(e): |
| raise Exception("Unexpected error message for invalid Flush: " + str(e)) |
| finally: |
| dev[0].request("P2P_SET disabled 0") |
| |
| try: |
| dev[0].request("P2P_SET disabled 1") |
| args = { 'peer': path, |
| 'join': True, |
| 'wps_method': 'pbc', |
| 'frequency': 2412 } |
| pin = p2p.Connect(args) |
| raise Exception("Invalid Connect accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: P2P is not available for this interface" not in str(e): |
| raise Exception("Unexpected error message for invalid Connect: " + str(e)) |
| finally: |
| dev[0].request("P2P_SET disabled 0") |
| |
| tests = [ { 'frequency': dbus.Int32(-1) }, |
| { 'wps_method': 'pbc' }, |
| { 'wps_method': 'foo' } ] |
| for args in tests: |
| try: |
| pin = p2p.Connect(args) |
| raise Exception("Invalid Connect accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid Connect: " + str(e)) |
| |
| try: |
| dev[0].request("P2P_SET disabled 1") |
| args = { 'peer': path } |
| pin = p2p.Invite(args) |
| raise Exception("Invalid Invite accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: P2P is not available for this interface" not in str(e): |
| raise Exception("Unexpected error message for invalid Invite: " + str(e)) |
| finally: |
| dev[0].request("P2P_SET disabled 0") |
| |
| try: |
| args = { 'foo': 'bar' } |
| pin = p2p.Invite(args) |
| raise Exception("Invalid Invite accepted") |
| except dbus.exceptions.DBusException, e: |
| if "InvalidArgs" not in str(e): |
| raise Exception("Unexpected error message for invalid Connect: " + str(e)) |
| |
| tests = [ (path, 'display', "InvalidArgs"), |
| (dbus.ObjectPath(path + "/Peers/00112233445566"), |
| 'display', |
| "UnknownError: Failed to send provision discovery request"), |
| (dbus.ObjectPath(path + "/Peers/00112233445566"), |
| 'keypad', |
| "UnknownError: Failed to send provision discovery request"), |
| (dbus.ObjectPath(path + "/Peers/00112233445566"), |
| 'pbc', |
| "UnknownError: Failed to send provision discovery request"), |
| (dbus.ObjectPath(path + "/Peers/00112233445566"), |
| 'pushbutton', |
| "UnknownError: Failed to send provision discovery request"), |
| (dbus.ObjectPath(path + "/Peers/00112233445566"), |
| 'foo', "InvalidArgs") ] |
| for (p,method,err) in tests: |
| try: |
| p2p.ProvisionDiscoveryRequest(p, method) |
| raise Exception("Invalid ProvisionDiscoveryRequest accepted") |
| except dbus.exceptions.DBusException, e: |
| if err not in str(e): |
| raise Exception("Unexpected error message for invalid ProvisionDiscoveryRequest: " + str(e)) |
| |
| try: |
| dev[0].request("P2P_SET disabled 1") |
| if_obj.Get(WPAS_DBUS_IFACE_P2PDEVICE, "Peers", |
| dbus_interface=dbus.PROPERTIES_IFACE) |
| raise Exception("Invalid Get(Peers) accepted") |
| except dbus.exceptions.DBusException, e: |
| if "Error.Failed: P2P is not available for this interface" not in str(e): |
| raise Exception("Unexpected error message for invalid Get(Peers): " + str(e)) |
| finally: |
| dev[0].request("P2P_SET disabled 0") |
| |
| def test_dbus_p2p_oom(dev, apdev): |
| """D-Bus P2P operations and OOM""" |
| (bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0]) |
| p2p = dbus.Interface(if_obj, WPAS_DBUS_IFACE_P2PDEVICE) |
| |
| with alloc_fail_dbus(dev[0], 1, "_wpa_dbus_dict_entry_get_string_array", |
| "Find", "InvalidArgs"): |
| p2p.Find(dbus.Dictionary({ 'Foo': [ 'bar' ] })) |
| |
| with alloc_fail_dbus(dev[0], 2, "_wpa_dbus_dict_entry_get_string_array", |
| "Find", "InvalidArgs"): |
| p2p.Find(dbus.Dictionary({ 'Foo': [ 'bar' ] })) |
| |
| with alloc_fail_dbus(dev[0], 10, "_wpa_dbus_dict_entry_get_string_array", |
| "Find", "InvalidArgs"): |
| p2p.Find(dbus.Dictionary({ 'Foo': [ '1','2','3','4','5','6','7','8','9' ] })) |
| |
| with alloc_fail_dbus(dev[0], 1, ":=_wpa_dbus_dict_entry_get_binarray", |
| "Find", "InvalidArgs"): |
| p2p.Find(dbus.Dictionary({ 'Foo': [ dbus.ByteArray('123') ] })) |
| |
| with alloc_fail_dbus(dev[0], 1, "_wpa_dbus_dict_entry_get_byte_array;_wpa_dbus_dict_entry_get_binarray", |
| "Find", "InvalidArgs"): |
| p2p.Find(dbus.Dictionary({ 'Foo': [ dbus.ByteArray('123') ] })) |
| |
| with alloc_fail_dbus(dev[0], 2, "=_wpa_dbus_dict_entry_get_binarray", |
| "Find", "InvalidArgs"): |
| p2p.Find(dbus.Dictionary({ 'Foo': [ dbus.ByteArray('123'), |
| dbus.ByteArray('123'), |
| dbus.ByteArray('123'), |
| dbus.ByteArray('123'), |
| dbus.ByteArray('123'), |
| dbus.ByteArray('123'), |
| dbus.ByteArray('123'), |
| dbus.ByteArray('123'), |
| dbus.ByteArray('123'), |
| dbus.ByteArray('123'), |
| dbus.ByteArray('123') ] })) |
| |
| with alloc_fail_dbus(dev[0], 1, "wpabuf_alloc_ext_data;_wpa
|