tests: Secure mesh network setup failing due to wpa_init() OOM
[mech_eap.git] / tests / hwsim / test_wpas_mesh.py
index e247aa5..b6188d1 100644 (file)
@@ -8,11 +8,19 @@
 
 import logging
 logger = logging.getLogger()
+import subprocess
 
 import hwsim_utils
 from wpasupplicant import WpaSupplicant
+from utils import HwsimSkip, alloc_fail
 
-def check_mesh_scan(dev, params, other_started=False):
+def check_mesh_support(dev, secure=False):
+    if "MESH" not in dev.get_capability("modes"):
+        raise HwsimSkip("Driver does not support mesh")
+    if secure and "SAE" not in dev.get_capability("auth_alg"):
+        raise HwsimSkip("SAE not supported")
+
+def check_mesh_scan(dev, params, other_started=False, beacon_int=0):
     if not other_started:
         dev.dump_monitor()
     id = dev.request("SCAN " + params)
@@ -57,6 +65,11 @@ def check_mesh_scan(dev, params, other_started=False):
         raise Exception("Could not get BSS entry for mesh")
     if 'mesh_capability' not in bss:
         raise Exception("mesh_capability missing from BSS entry")
+    if beacon_int:
+        if 'beacon_int' not in bss:
+            raise Exception("beacon_int missing from BSS entry")
+        if str(beacon_int) != bss['beacon_int']:
+            raise Exception("Unexpected beacon_int in BSS entry: " + bss['beacon_int'])
 
 def check_mesh_group_added(dev):
     ev = dev.wait_event(["MESH-GROUP-STARTED"])
@@ -70,8 +83,8 @@ def check_mesh_group_removed(dev):
         raise Exception("Test exception: Couldn't leave mesh")
 
 
-def check_mesh_peer_connected(dev):
-    ev = dev.wait_event(["MESH-PEER-CONNECTED"])
+def check_mesh_peer_connected(dev, timeout=10):
+    ev = dev.wait_event(["MESH-PEER-CONNECTED"], timeout=timeout)
     if ev is None:
         raise Exception("Test exception: Remote peer did not connect.")
 
@@ -84,24 +97,26 @@ def check_mesh_peer_disconnected(dev):
 
 def test_wpas_add_set_remove_support(dev):
     """wpa_supplicant MESH add/set/remove network support"""
+    check_mesh_support(dev[0])
     id = dev[0].add_network()
     dev[0].set_network(id, "mode", "5")
     dev[0].remove_network(id)
 
-def add_open_mesh_network(dev, ht_mode=False, freq="2412", start=True):
+def add_open_mesh_network(dev, freq="2412", start=True, beacon_int=0):
     id = dev.add_network()
     dev.set_network(id, "mode", "5")
     dev.set_network_quoted(id, "ssid", "wpas-mesh-open")
     dev.set_network(id, "key_mgmt", "NONE")
     dev.set_network(id, "frequency", freq)
-    if ht_mode:
-        dev.set_network(id, "mesh_ht_mode", ht_mode)
+    if beacon_int:
+        dev.set_network(id, "beacon_int", str(beacon_int))
     if start:
         dev.mesh_group_add(id)
     return id
 
 def test_wpas_mesh_group_added(dev):
     """wpa_supplicant MESH group add"""
+    check_mesh_support(dev[0])
     add_open_mesh_network(dev[0])
 
     # Check for MESH-GROUP-STARTED event
@@ -110,7 +125,8 @@ def test_wpas_mesh_group_added(dev):
 
 def test_wpas_mesh_group_remove(dev):
     """wpa_supplicant MESH group remove"""
-    add_open_mesh_network(dev[0], ht_mode="NOHT")
+    check_mesh_support(dev[0])
+    add_open_mesh_network(dev[0])
     # Check for MESH-GROUP-STARTED event
     check_mesh_group_added(dev[0])
     dev[0].mesh_group_remove()
@@ -120,8 +136,9 @@ def test_wpas_mesh_group_remove(dev):
 
 def test_wpas_mesh_peer_connected(dev):
     """wpa_supplicant MESH peer connected"""
-    add_open_mesh_network(dev[0], ht_mode="HT20")
-    add_open_mesh_network(dev[1], ht_mode="HT20")
+    check_mesh_support(dev[0])
+    add_open_mesh_network(dev[0], beacon_int=160)
+    add_open_mesh_network(dev[1], beacon_int=160)
 
     # Check for mesh joined
     check_mesh_group_added(dev[0])
@@ -134,6 +151,7 @@ def test_wpas_mesh_peer_connected(dev):
 
 def test_wpas_mesh_peer_disconnected(dev):
     """wpa_supplicant MESH peer disconnected"""
+    check_mesh_support(dev[0])
     add_open_mesh_network(dev[0])
     add_open_mesh_network(dev[1])
 
@@ -153,27 +171,22 @@ def test_wpas_mesh_peer_disconnected(dev):
 
 def test_wpas_mesh_mode_scan(dev):
     """wpa_supplicant MESH scan support"""
-    add_open_mesh_network(dev[0], ht_mode="HT40+")
-    add_open_mesh_network(dev[1], ht_mode="HT40+")
+    check_mesh_support(dev[0])
+    add_open_mesh_network(dev[0])
+    add_open_mesh_network(dev[1], beacon_int=175)
 
     # Check for mesh joined
     check_mesh_group_added(dev[0])
     check_mesh_group_added(dev[1])
 
     # Check for Mesh scan
-    check_mesh_scan(dev[0], "use_id=1")
-
-
-def wrap_wpas_mesh_test(test, dev, apdev):
-    def _test_connectivity(dev1, dev2):
-        return hwsim_utils.test_connectivity(dev1, dev2)
-
-    return test(dev, apdev, _test_connectivity)
+    check_mesh_scan(dev[0], "use_id=1", beacon_int=175)
 
-
-def _test_wpas_mesh_open(dev, apdev, test_connectivity):
-    add_open_mesh_network(dev[0], ht_mode="HT40-", freq="2462")
-    add_open_mesh_network(dev[1], ht_mode="HT40-", freq="2462")
+def test_wpas_mesh_open(dev, apdev):
+    """wpa_supplicant open MESH network connectivity"""
+    check_mesh_support(dev[0])
+    add_open_mesh_network(dev[0], freq="2462")
+    add_open_mesh_network(dev[1], freq="2462")
 
     # Check for mesh joined
     check_mesh_group_added(dev[0])
@@ -184,15 +197,11 @@ def _test_wpas_mesh_open(dev, apdev, test_connectivity):
     check_mesh_peer_connected(dev[1])
 
     # Test connectivity 0->1 and 1->0
-    test_connectivity(dev[0], dev[1])
+    hwsim_utils.test_connectivity(dev[0], dev[1])
 
-
-def test_wpas_mesh_open(dev, apdev):
+def test_wpas_mesh_open_no_auto(dev, apdev):
     """wpa_supplicant open MESH network connectivity"""
-    return wrap_wpas_mesh_test(_test_wpas_mesh_open, dev, apdev)
-
-
-def _test_wpas_mesh_open_no_auto(dev, apdev, test_connectivity):
+    check_mesh_support(dev[0])
     id = add_open_mesh_network(dev[0], start=False)
     dev[0].set_network(id, "dot11MeshMaxRetries", "16")
     dev[0].set_network(id, "dot11MeshRetryTimeout", "255")
@@ -207,16 +216,11 @@ def _test_wpas_mesh_open_no_auto(dev, apdev, test_connectivity):
     check_mesh_group_added(dev[1])
 
     # Check for peer connected
-    check_mesh_peer_connected(dev[0])
+    check_mesh_peer_connected(dev[0], timeout=30)
     check_mesh_peer_connected(dev[1])
 
     # Test connectivity 0->1 and 1->0
-    test_connectivity(dev[0], dev[1])
-
-
-def test_wpas_mesh_open_no_auto(dev, apdev):
-    """wpa_supplicant open MESH network connectivity"""
-    return wrap_wpas_mesh_test(_test_wpas_mesh_open_no_auto, dev, apdev)
+    hwsim_utils.test_connectivity(dev[0], dev[1])
 
 def add_mesh_secure_net(dev, psk=True):
     id = dev.add_network()
@@ -228,7 +232,9 @@ def add_mesh_secure_net(dev, psk=True):
         dev.set_network_quoted(id, "psk", "thisismypassphrase!")
     return id
 
-def _test_wpas_mesh_secure(dev, apdev, test_connectivity):
+def test_wpas_mesh_secure(dev, apdev):
+    """wpa_supplicant secure MESH network connectivity"""
+    check_mesh_support(dev[0], secure=True)
     dev[0].request("SET sae_groups ")
     id = add_mesh_secure_net(dev[0])
     dev[0].mesh_group_add(id)
@@ -246,15 +252,11 @@ def _test_wpas_mesh_secure(dev, apdev, test_connectivity):
     check_mesh_peer_connected(dev[1])
 
     # Test connectivity 0->1 and 1->0
-    test_connectivity(dev[0], dev[1])
-
-
-def test_wpas_mesh_secure(dev, apdev):
-    """wpa_supplicant secure MESH network connectivity"""
-    return wrap_wpas_mesh_test(_test_wpas_mesh_secure, dev, apdev)
+    hwsim_utils.test_connectivity(dev[0], dev[1])
 
 def test_wpas_mesh_secure_sae_group_mismatch(dev, apdev):
     """wpa_supplicant secure MESH and SAE group mismatch"""
+    check_mesh_support(dev[0], secure=True)
     addr0 = dev[0].p2p_interface_addr()
     addr1 = dev[1].p2p_interface_addr()
     addr2 = dev[2].p2p_interface_addr()
@@ -305,6 +307,7 @@ def test_wpas_mesh_secure_sae_group_mismatch(dev, apdev):
 
 def test_wpas_mesh_secure_sae_missing_password(dev, apdev):
     """wpa_supplicant secure MESH and missing SAE password"""
+    check_mesh_support(dev[0], secure=True)
     id = add_mesh_secure_net(dev[0], psk=False)
     dev[0].set_network(id, "psk", "8f20b381f9b84371d61b5080ad85cac3c61ab3ca9525be5b2d0f4da3d979187a")
     dev[0].mesh_group_add(id)
@@ -318,7 +321,9 @@ def test_wpas_mesh_secure_sae_missing_password(dev, apdev):
     if ev is not None:
         raise Exception("Unexpected mesh group start")
 
-def _test_wpas_mesh_secure_no_auto(dev, apdev, test_connectivity):
+def test_wpas_mesh_secure_no_auto(dev, apdev):
+    """wpa_supplicant secure MESH network connectivity"""
+    check_mesh_support(dev[0], secure=True)
     dev[0].request("SET sae_groups 19")
     id = add_mesh_secure_net(dev[0])
     dev[0].mesh_group_add(id)
@@ -333,21 +338,18 @@ def _test_wpas_mesh_secure_no_auto(dev, apdev, test_connectivity):
     check_mesh_group_added(dev[1])
 
     # Check for peer connected
-    check_mesh_peer_connected(dev[0])
+    check_mesh_peer_connected(dev[0], timeout=30)
     check_mesh_peer_connected(dev[1])
 
     # Test connectivity 0->1 and 1->0
-    test_connectivity(dev[0], dev[1])
+    hwsim_utils.test_connectivity(dev[0], dev[1])
 
     dev[0].request("SET sae_groups ")
     dev[1].request("SET sae_groups ")
 
-def test_wpas_mesh_secure_no_auto(dev, apdev):
-    """wpa_supplicant secure MESH network connectivity"""
-    return wrap_wpas_mesh_test(_test_wpas_mesh_secure_no_auto, dev, apdev)
-
 def test_wpas_mesh_ctrl(dev):
     """wpa_supplicant ctrl_iface mesh command error cases"""
+    check_mesh_support(dev[0])
     if "FAIL" not in dev[0].request("MESH_GROUP_ADD 123"):
         raise Exception("Unexpected MESH_GROUP_ADD success")
     id = dev[0].add_network()
@@ -363,6 +365,7 @@ def test_wpas_mesh_ctrl(dev):
 
 def test_wpas_mesh_dynamic_interface(dev):
     """wpa_supplicant mesh with dynamic interface"""
+    check_mesh_support(dev[0])
     mesh0 = None
     mesh1 = None
     try:
@@ -436,3 +439,174 @@ def test_wpas_mesh_dynamic_interface(dev):
             dev[0].request("MESH_GROUP_REMOVE " + mesh0)
         if mesh1:
             dev[1].request("MESH_GROUP_REMOVE " + mesh1)
+
+def test_wpas_mesh_max_peering(dev, apdev):
+    """Mesh max peering limit"""
+    check_mesh_support(dev[0])
+    try:
+        dev[0].request("SET max_peer_links 1")
+
+        # first, connect dev[0] and dev[1]
+        add_open_mesh_network(dev[0])
+        add_open_mesh_network(dev[1])
+        for i in range(2):
+            ev = dev[i].wait_event(["MESH-PEER-CONNECTED"])
+            if ev is None:
+                raise Exception("dev%d did not connect with any peer" % i)
+
+        # add dev[2] which will try to connect with both dev[0] and dev[1],
+        # but can complete connection only with dev[1]
+        add_open_mesh_network(dev[2])
+        for i in range(1, 3):
+            ev = dev[i].wait_event(["MESH-PEER-CONNECTED"])
+            if ev is None:
+                raise Exception("dev%d did not connect the second peer" % i)
+
+        ev = dev[0].wait_event(["MESH-PEER-CONNECTED"], timeout=1)
+        if ev is not None:
+            raise Exception("dev0 connection beyond max peering limit")
+
+        ev = dev[2].wait_event(["MESH-PEER-CONNECTED"], timeout=0.1)
+        if ev is not None:
+            raise Exception("dev2 reported unexpected peering: " + ev)
+
+        for i in range(3):
+            dev[i].mesh_group_remove()
+            check_mesh_group_removed(dev[i])
+    finally:
+        dev[0].request("SET max_peer_links 99")
+
+def test_wpas_mesh_open_5ghz(dev, apdev):
+    """wpa_supplicant open MESH network on 5 GHz band"""
+    try:
+        _test_wpas_mesh_open_5ghz(dev, apdev)
+    finally:
+        subprocess.call(['iw', 'reg', 'set', '00'])
+        dev[0].flush_scan_cache()
+        dev[1].flush_scan_cache()
+
+def _test_wpas_mesh_open_5ghz(dev, apdev):
+    check_mesh_support(dev[0])
+    subprocess.call(['iw', 'reg', 'set', 'US'])
+    for i in range(2):
+        for j in range(5):
+            ev = dev[i].wait_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=5)
+            if ev is None:
+                raise Exception("No regdom change event")
+            if "alpha2=US" in ev:
+                break
+        add_open_mesh_network(dev[i], freq="5180")
+
+    # Check for mesh joined
+    check_mesh_group_added(dev[0])
+    check_mesh_group_added(dev[1])
+
+    # Check for peer connected
+    check_mesh_peer_connected(dev[0])
+    check_mesh_peer_connected(dev[1])
+
+    # Test connectivity 0->1 and 1->0
+    hwsim_utils.test_connectivity(dev[0], dev[1])
+
+def test_wpas_mesh_password_mismatch(dev, apdev):
+    """Mesh network and one device with mismatching password"""
+    check_mesh_support(dev[0], secure=True)
+    dev[0].request("SET sae_groups ")
+    id = add_mesh_secure_net(dev[0])
+    dev[0].mesh_group_add(id)
+
+    dev[1].request("SET sae_groups ")
+    id = add_mesh_secure_net(dev[1])
+    dev[1].mesh_group_add(id)
+
+    dev[2].request("SET sae_groups ")
+    id = add_mesh_secure_net(dev[2])
+    dev[2].set_network_quoted(id, "psk", "wrong password")
+    dev[2].mesh_group_add(id)
+
+    # The two peers with matching password need to be able to connect
+    check_mesh_group_added(dev[0])
+    check_mesh_group_added(dev[1])
+    check_mesh_peer_connected(dev[0])
+    check_mesh_peer_connected(dev[1])
+
+    ev = dev[2].wait_event(["MESH-SAE-AUTH-FAILURE"], timeout=20)
+    if ev is None:
+        raise Exception("dev2 did not report auth failure (1)")
+    ev = dev[2].wait_event(["MESH-SAE-AUTH-FAILURE"], timeout=20)
+    if ev is None:
+        raise Exception("dev2 did not report auth failure (2)")
+
+    count = 0
+    ev = dev[0].wait_event(["MESH-SAE-AUTH-FAILURE"], timeout=1)
+    if ev is None:
+        logger.info("dev0 did not report auth failure")
+    else:
+        if "addr=" + dev[2].own_addr() not in ev:
+            raise Exception("Unexpected peer address in dev0 event: " + ev)
+        count += 1
+
+    ev = dev[1].wait_event(["MESH-SAE-AUTH-FAILURE"], timeout=1)
+    if ev is None:
+        logger.info("dev1 did not report auth failure")
+    else:
+        if "addr=" + dev[2].own_addr() not in ev:
+            raise Exception("Unexpected peer address in dev1 event: " + ev)
+        count += 1
+
+    hwsim_utils.test_connectivity(dev[0], dev[1])
+
+    for i in range(2):
+        try:
+            hwsim_utils.test_connectivity(dev[i], dev[2], timeout=1)
+            raise Exception("Data connectivity test passed unexpectedly")
+        except Exception, e:
+            if "data delivery failed" not in str(e):
+                raise
+
+    if count == 0:
+        raise Exception("Neither dev0 nor dev1 reported auth failure")
+
+def test_wpas_mesh_password_mismatch_retry(dev, apdev, params):
+    """Mesh password mismatch and retry [long]"""
+    if not params['long']:
+        raise HwsimSkip("Skip test case with long duration due to --long not specified")
+    check_mesh_support(dev[0], secure=True)
+    dev[0].request("SET sae_groups ")
+    id = add_mesh_secure_net(dev[0])
+    dev[0].mesh_group_add(id)
+
+    dev[1].request("SET sae_groups ")
+    id = add_mesh_secure_net(dev[1])
+    dev[1].set_network_quoted(id, "psk", "wrong password")
+    dev[1].mesh_group_add(id)
+
+    # Check for mesh joined
+    check_mesh_group_added(dev[0])
+    check_mesh_group_added(dev[1])
+
+    for i in range(4):
+        ev = dev[0].wait_event(["MESH-SAE-AUTH-FAILURE"], timeout=20)
+        if ev is None:
+            raise Exception("dev0 did not report auth failure (%d)" % i)
+        ev = dev[1].wait_event(["MESH-SAE-AUTH-FAILURE"], timeout=20)
+        if ev is None:
+            raise Exception("dev1 did not report auth failure (%d)" % i)
+
+    ev = dev[0].wait_event(["MESH-SAE-AUTH-BLOCKED"], timeout=10)
+    if ev is None:
+        raise Exception("dev0 did not report auth blocked")
+    ev = dev[1].wait_event(["MESH-SAE-AUTH-BLOCKED"], timeout=10)
+    if ev is None:
+        raise Exception("dev1 did not report auth blocked")
+
+def test_mesh_wpa_auth_init_oom(dev, apdev):
+    """Secure mesh network setup failing due to wpa_init() OOM"""
+    check_mesh_support(dev[0], secure=True)
+    dev[0].request("SET sae_groups ")
+    with alloc_fail(dev[0], 1, "wpa_init"):
+        id = add_mesh_secure_net(dev[0])
+        dev[0].mesh_group_add(id)
+        ev = dev[0].wait_event(["MESH-GROUP-STARTED"], timeout=0.2)
+        if ev is not None:
+            raise Exception("Unexpected mesh group start during OOM")