blob: dba899e9fcb18a774a093aa12e7389249d09830b [file] [log] [blame]
# 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, fail_test
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",
ap_uuid="27ea801a-9e5c-4e73-bd82-f89cbcd10d7e"):
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", "uuid": ap_uuid}
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_pbc_overlap(dev, apdev):
"""D-Bus WPS/PBC operation and signal for PBC overlap"""
(bus,wpas_obj,path,if_obj) = prepare_dbus(dev[0])
wps = dbus.Interface(if_obj, WPAS_DBUS_IFACE_WPS)
hapd = start_ap(apdev[0])
hapd2 = start_ap(apdev[1], ssid="test-wps2",
ap_uuid="27ea801a-9e5c-4e73-bd82-f89cbcd10d7f")
hapd.request("WPS_PBC")
hapd2.request("WPS_PBC")
bssid = apdev[0]['bssid']
dev[0].scan_for_bss(bssid, freq="2412")
bssid2 = apdev[1]['bssid']
dev[0].scan_for_bss(bssid2, freq="2412")
class TestDbusWps(TestDbus):
def __init__(self, bus, wps):
TestDbus.__init__(self, bus)
self.overlap_seen = 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.loop.run()
return self
def wpsEvent(self, name, args):
logger.debug("wpsEvent: %s args='%s'" % (name, str(args)))
if name == "pbc-overlap":
self.overlap_seen = True
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.overlap_seen
with TestDbusWps(bus, wps) as t:
if not t.success():
raise Exception("Failure in D-Bus operations")
dev[0].request("WPS_CANCEL")
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_wps_cancel(dev, apdev):
"""D-Bus WPS Cancel operation"""
(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']
wps.Cancel()
dev[0].scan_for_bss(bssid, freq="2412")
bssid_ay = dbus.ByteArray(bssid.replace(':','').decode('hex'))
wps.Start({'Role': 'enrollee', 'Type': 'pin', 'Pin': '12345670',
'Bssid': bssid_ay})
wps.Cancel()
dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"], 1)
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
iface.Disconnect()
elif self.state == 7:
self.state = 8
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
iface.Reconnect()
elif self.state == 8:
self.state = 9
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 == 9
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)
id = int(dev[0].list_networks()[0]['id'])
val = dev[0].get_network(id, "scan_freq")
if val != "2412":
raise Exception("Invalid scan_freq value: " + str(val))
iface.RemoveNetwork(netw)
args = dbus.Dictionary({ 'ssid': "foo",
'key_mgmt': 'NONE',
'scan_freq': "2412 2432",
'freq_list': "2412 2417 2432" },
signature='sv')
netw = iface.AddNetwork(args)
id = int(dev[0].list_networks()[0]['id'])
val = dev[0].get_network(id, "scan_freq")
if val != "2412 2432":
raise Exception("Invalid scan_freq value (2): " + str(val))
val = dev[0].get_network(id, "freq_list")
if val != "2412 2417 2432":
raise Exception("Invalid freq_list value: " + str(val))
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:
# For now, work around separate P2P Device interface event delivery
ev = dev[0].wait_global_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=1)
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:
# For now, work around separate P2P Device interface event delivery
ev = dev[0].wait_global_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=1)
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])
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.groupStarted, WPAS_DBUS_IFACE_P2PDEVICE,
"GroupStarted")
self.add_signal(self.probeRequest, WPAS_DBUS_IFACE, "ProbeRequest",
byte_arrays=True)
self.loop.run()
return self
def groupStarted(self, properties):