tests: SAE and opportunistic key caching and PMK lifetime

Signed-off-by: Jouni Malinen <jouni@codeaurora.org>

tests/hwsim/test_ap_ft.py

@@ -1906,7 +1906,7 @@ def test_ap_ft_oom(dev, apdev):
     """WPA2-PSK-FT and OOM"""
     dst = setup_ap_ft_oom(dev, apdev)
     with alloc_fail(dev[0], 1, "wpa_ft_gen_req_ies"):
-        dev[0].roam(dst)
+        dev[0].roam(dst, check_bssid=False)
 
 def test_ap_ft_oom2(dev, apdev):
     """WPA2-PSK-FT and OOM (2)"""
@@ -1949,7 +1949,7 @@ def test_ap_ft_ap_oom(dev, apdev):
     bssid1 = hapd1.own_addr()
     dev[0].scan_for_bss(bssid1, freq="2412")
     # This roam will fail due to missing PMK-R0 (OOM prevented storing it)
-    dev[0].roam(bssid1)
+    dev[0].roam(bssid1, check_bssid=False)
 
 def test_ap_ft_ap_oom2(dev, apdev):
     """WPA2-PSK-FT and AP OOM 2"""
@@ -1994,15 +1994,15 @@ def test_ap_ft_ap_oom3(dev, apdev):
     dev[0].scan_for_bss(bssid1, freq="2412")
     with alloc_fail(hapd1, 1, "wpa_ft_pull_pmk_r1"):
         # This will fail due to not being able to send out PMK-R1 pull request
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
 
     with fail_test(hapd1, 2, "os_get_random;wpa_ft_pull_pmk_r1"):
         # This will fail due to not being able to send out PMK-R1 pull request
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
 
     with fail_test(hapd1, 2, "aes_siv_encrypt;wpa_ft_pull_pmk_r1"):
         # This will fail due to not being able to send out PMK-R1 pull request
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
 
 def test_ap_ft_ap_oom3b(dev, apdev):
     """WPA2-PSK-FT and AP OOM 3b"""
@@ -2076,23 +2076,23 @@ def test_ap_ft_ap_oom5(dev, apdev):
     dev[0].scan_for_bss(bssid1, freq="2412")
     with alloc_fail(hapd1, 1, "=wpa_ft_process_auth_req"):
         # This will fail to roam
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
 
     with fail_test(hapd1, 1, "os_get_random;wpa_ft_process_auth_req"):
         # This will fail to roam
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
 
     with fail_test(hapd1, 1, "sha256_prf_bits;wpa_pmk_r1_to_ptk;wpa_ft_process_auth_req"):
         # This will fail to roam
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
 
     with fail_test(hapd1, 3, "wpa_pmk_r1_to_ptk;wpa_ft_process_auth_req"):
         # This will fail to roam
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
 
     with fail_test(hapd1, 1, "wpa_derive_pmk_r1_name;wpa_ft_process_auth_req"):
         # This will fail to roam
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
 
 def test_ap_ft_ap_oom6(dev, apdev):
     """WPA2-PSK-FT and AP OOM 6"""
@@ -2231,10 +2231,10 @@ def test_ap_ft_ap_oom8(dev, apdev):
     dev[0].scan_for_bss(bssid1, freq="2412")
     with fail_test(hapd1, 1, "wpa_derive_pmk_r0;wpa_ft_psk_pmk_r1"):
         # This will fail to roam
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
     with fail_test(hapd1, 1, "wpa_derive_pmk_r1;wpa_ft_psk_pmk_r1"):
         # This will fail to roam
-        dev[0].roam(bssid1)
+        dev[0].roam(bssid1, check_bssid=False)
 
 def test_ap_ft_ap_oom9(dev, apdev):
     """WPA2-PSK-FT and AP OOM 9"""

tests/hwsim/test_sae.py

@@ -2318,3 +2318,151 @@ def test_sae_forced_anti_clogging_h2e_loop(dev, apdev):
     finally:
         for i in range(2):
             dev[i].set("sae_pwe", "0")
+
+def test_sae_okc(dev, apdev):
+    """SAE and opportunistic key caching"""
+    check_sae_capab(dev[0])
+    params = hostapd.wpa2_params(ssid="test-sae", passphrase="12345678")
+    params['wpa_key_mgmt'] = 'SAE'
+    params['okc'] = '1'
+    hapd = hostapd.add_ap(apdev[0], params)
+    bssid = hapd.own_addr()
+
+    dev[0].set("sae_groups", "")
+    id = dev[0].connect("test-sae", psk="12345678", key_mgmt="SAE",
+                        okc=True, scan_freq="2412")
+    dev[0].dump_monitor()
+    hapd.wait_sta()
+    if "sae_group" not in dev[0].get_status():
+        raise Exception("SAE authentication not used")
+
+    hapd2 = hostapd.add_ap(apdev[1], params)
+    bssid2 = hapd2.own_addr()
+
+    dev[0].scan_for_bss(bssid2, freq=2412)
+    dev[0].roam(bssid2)
+    dev[0].dump_monitor()
+    hapd2.wait_sta()
+    if "sae_group" in dev[0].get_status():
+        raise Exception("SAE authentication used during roam to AP2")
+
+    dev[0].roam(bssid)
+    dev[0].dump_monitor()
+    hapd.wait_sta()
+    if "sae_group" in dev[0].get_status():
+        raise Exception("SAE authentication used during roam to AP1")
+
+def test_sae_okc_sta_only(dev, apdev):
+    """SAE and opportunistic key caching only on STA"""
+    check_sae_capab(dev[0])
+    params = hostapd.wpa2_params(ssid="test-sae", passphrase="12345678")
+    params['wpa_key_mgmt'] = 'SAE'
+    hapd = hostapd.add_ap(apdev[0], params)
+    bssid = hapd.own_addr()
+
+    dev[0].set("sae_groups", "")
+    id = dev[0].connect("test-sae", psk="12345678", key_mgmt="SAE",
+                        okc=True, scan_freq="2412")
+    dev[0].dump_monitor()
+    hapd.wait_sta()
+    if "sae_group" not in dev[0].get_status():
+        raise Exception("SAE authentication not used")
+
+    hapd2 = hostapd.add_ap(apdev[1], params)
+    bssid2 = hapd2.own_addr()
+
+    dev[0].scan_for_bss(bssid2, freq=2412)
+    dev[0].roam(bssid2, assoc_reject_ok=True)
+    dev[0].dump_monitor()
+    hapd2.wait_sta()
+    if "sae_group" not in dev[0].get_status():
+        raise Exception("SAE authentication not used during roam to AP2")
+
+def test_sae_okc_pmk_lifetime(dev, apdev):
+    """SAE and opportunistic key caching and PMK lifetime"""
+    check_sae_capab(dev[0])
+    params = hostapd.wpa2_params(ssid="test-sae", passphrase="12345678")
+    params['wpa_key_mgmt'] = 'SAE'
+    params['okc'] = '1'
+    hapd = hostapd.add_ap(apdev[0], params)
+    bssid = hapd.own_addr()
+
+    dev[0].set("sae_groups", "")
+    dev[0].set("dot11RSNAConfigPMKLifetime", "10")
+    dev[0].set("dot11RSNAConfigPMKReauthThreshold", "30")
+    id = dev[0].connect("test-sae", psk="12345678", key_mgmt="SAE",
+                        okc=True, scan_freq="2412")
+    dev[0].dump_monitor()
+    hapd.wait_sta()
+    if "sae_group" not in dev[0].get_status():
+        raise Exception("SAE authentication not used")
+
+    hapd2 = hostapd.add_ap(apdev[1], params)
+    bssid2 = hapd2.own_addr()
+
+    time.sleep(5)
+    dev[0].scan_for_bss(bssid2, freq=2412)
+    dev[0].roam(bssid2)
+    dev[0].dump_monitor()
+    hapd2.wait_sta()
+    if "sae_group" not in dev[0].get_status():
+        raise Exception("SAE authentication not used during roam to AP2 after reauth threshold")
+
+def test_sae_pmk_lifetime(dev, apdev):
+    """SAE and opportunistic key caching and PMK lifetime"""
+    check_sae_capab(dev[0])
+    params = hostapd.wpa2_params(ssid="test-sae", passphrase="12345678")
+    params['wpa_key_mgmt'] = 'SAE'
+    hapd = hostapd.add_ap(apdev[0], params)
+    bssid = hapd.own_addr()
+
+    dev[0].set("sae_groups", "")
+    dev[0].set("dot11RSNAConfigPMKLifetime", "10")
+    dev[0].set("dot11RSNAConfigPMKReauthThreshold", "50")
+    id = dev[0].connect("test-sae", psk="12345678", key_mgmt="SAE",
+                        scan_freq="2412")
+    dev[0].dump_monitor()
+    hapd.wait_sta()
+    if "sae_group" not in dev[0].get_status():
+        raise Exception("SAE authentication not used")
+
+    hapd2 = hostapd.add_ap(apdev[1], params)
+    bssid2 = hapd2.own_addr()
+
+    dev[0].scan_for_bss(bssid2, freq=2412)
+    dev[0].roam(bssid2)
+    dev[0].dump_monitor()
+    hapd2.wait_sta()
+    if "sae_group" not in dev[0].get_status():
+        raise Exception("SAE authentication not used during roam to AP2")
+
+    dev[0].roam(bssid)
+    dev[0].dump_monitor()
+    hapd.wait_sta()
+    if "sae_group" in dev[0].get_status():
+        raise Exception("SAE authentication used during roam to AP1")
+
+    time.sleep(6)
+    dev[0].scan_for_bss(bssid2, freq=2412)
+    dev[0].roam(bssid2)
+    dev[0].dump_monitor()
+    hapd2.wait_sta()
+    if "sae_group" not in dev[0].get_status():
+        raise Exception("SAE authentication not used during roam to AP2 after reauth threshold")
+
+    ev = dev[0].wait_event(["PMKSA-CACHE-REMOVED"], 11)
+    if ev is None:
+        raise Exception("PMKSA cache entry did not expire")
+    if bssid2 not in ev:
+        ev = dev[0].wait_event(["PMKSA-CACHE-REMOVED"], 11)
+        if ev is None:
+            raise Exception("PMKSA cache entry did not expire")
+        if bssid2 not in ev:
+            raise Exception("PMKSA cache entry for the current AP did not expire")
+    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], 1)
+    if ev is None:
+        raise Exception("Disconnection not reported after PMKSA cache entry expiration")
+
+    dev[0].wait_connected()
+    if "sae_group" not in dev[0].get_status():
+        raise Exception("SAE authentication not used after PMKSA cache entry expiration")

tests/hwsim/wpasupplicant.py

@@ -1191,7 +1191,8 @@ class WpaSupplicant:
                                  "CTRL-EVENT-SCAN-RESULTS"], timeout=0.5)
         self.dump_monitor()
 
-    def roam(self, bssid, fail_test=False, assoc_reject_ok=False):
+    def roam(self, bssid, fail_test=False, assoc_reject_ok=False,
+             check_bssid=True):
         self.dump_monitor()
         if "OK" not in self.request("ROAM " + bssid):
             raise Exception("ROAM failed")
@@ -1205,6 +1206,9 @@ class WpaSupplicant:
                 raise Exception("Unexpected connection")
             self.dump_monitor()
             return
+        if assoc_reject_ok:
+            ev = self.wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
+        else:
             ev = self.wait_event(["CTRL-EVENT-CONNECTED",
                                   "CTRL-EVENT-ASSOC-REJECT"], timeout=10)
         if ev is None:
@@ -1212,6 +1216,8 @@ class WpaSupplicant:
         if "CTRL-EVENT-ASSOC-REJECT" in ev:
             raise Exception("Roaming association rejected")
         self.dump_monitor()
+        if check_bssid and self.get_status_field('bssid') != bssid:
+            raise Exception("Did not roam to correct BSSID")
 
     def roam_over_ds(self, bssid, fail_test=False):
         self.dump_monitor()