[PATCH 14/15] auto-t: add PMKSA tests

[James Prestwood <prestwoj@gmail.com>]

Adds a test for just PMKSA and testing expiration as well as includes
some PMKSA tests in the SAE roaming test to ensure FT/reassociation
works.
---
 autotests/testPMKSA-SAE/connection_test.py | 114 +++++++++++++++++++++
 autotests/testPMKSA-SAE/hw.conf            |   7 ++
 autotests/testPMKSA-SAE/ssidSAE.conf       |  12 +++
 autotests/testSAE-roam/connection_test.py  |  60 ++++++++++-
 4 files changed, 192 insertions(+), 1 deletion(-)
 create mode 100644 autotests/testPMKSA-SAE/connection_test.py
 create mode 100644 autotests/testPMKSA-SAE/hw.conf
 create mode 100644 autotests/testPMKSA-SAE/ssidSAE.conf

diff --git a/autotests/testPMKSA-SAE/connection_test.py b/autotests/testPMKSA-SAE/connection_test.py
new file mode 100644
index 00000000..5bab3ff8
--- /dev/null
+++ b/autotests/testPMKSA-SAE/connection_test.py
@@ -0,0 +1,114 @@
+#!/usr/bin/python3
+
+import unittest
+import sys
+
+sys.path.append('../util')
+from iwd import IWD
+from iwd import PSKAgent
+from iwd import NetworkType
+from hostapd import HostapdCLI
+import testutil
+
+class Test(unittest.TestCase):
+
+    def validate_connection(self, wd, ssid, hostapd, expected_group):
+        psk_agent = PSKAgent("secret123")
+        wd.register_psk_agent(psk_agent)
+
+        devices = wd.list_devices(1)
+        self.assertIsNotNone(devices)
+        device = devices[0]
+
+        device.disconnect()
+
+        network = device.get_ordered_network(ssid, full_scan=True)
+
+        self.assertEqual(network.type, NetworkType.psk)
+
+        network.network_object.connect()
+
+        condition = 'obj.state == DeviceState.connected'
+        wd.wait_for_object_condition(device, condition)
+
+        wd.wait(2)
+
+        testutil.test_iface_operstate(intf=device.name)
+        testutil.test_ifaces_connected(if0=device.name, if1=hostapd.ifname)
+
+        # Initial connection PMKSA should not be used. So we should see the
+        # SAE group set.
+        sta_status = hostapd.sta_status(device.address)
+        self.assertEqual(int(sta_status["sae_group"]), expected_group)
+
+        device.disconnect()
+
+        condition = 'not obj.connected'
+        wd.wait_for_object_condition(network.network_object, condition)
+
+        wd.unregister_psk_agent(psk_agent)
+
+        network.network_object.connect(wait=False)
+
+        condition = 'obj.state == DeviceState.connected'
+        wd.wait_for_object_condition(device, condition)
+
+        wd.wait(2)
+
+        testutil.test_iface_operstate(intf=device.name)
+        testutil.test_ifaces_connected(if0=device.name, if1=hostapd.ifname)
+
+        # Having connected once prior we should have a PMKSA and SAE should not
+        # have been used.
+        sta_status = hostapd.sta_status(device.address)
+        self.assertNotIn("sae_group", sta_status.keys())
+
+        device.disconnect()
+
+        condition = 'not obj.connected'
+        wd.wait_for_object_condition(network.network_object, condition)
+
+        hostapd.pmksa_flush()
+
+        wd.wait(5)
+
+        network.network_object.connect()
+
+        device.wait_for_event("pmksa-invalid-pmkid")
+
+        condition = 'obj.state == DeviceState.connected'
+        wd.wait_for_object_condition(device, condition)
+
+        wd.wait(2)
+
+        testutil.test_iface_operstate(intf=device.name)
+        testutil.test_ifaces_connected(if0=device.name, if1=hostapd.ifname)
+
+        # Manually flushing the PMKSA from the AP then reconnecting we should
+        # have failed (INVALID_PMKID) then retried the same BSS with SAE, not
+        # PMKSA.
+        sta_status = hostapd.sta_status(device.address)
+        self.assertEqual(int(sta_status["sae_group"]), expected_group)
+
+    def test_pmksa_sae(self):
+        self.hostapd.wait_for_event("AP-ENABLED")
+        self.validate_connection(self.wd, "ssidSAE", self.hostapd, 19)
+
+    def setUp(self):
+        self.hostapd.default()
+        self.wd = IWD(True)
+
+    def tearDown(self):
+        self.wd.clear_storage()
+        self.wd = None
+
+    @classmethod
+    def setUpClass(cls):
+        cls.hostapd = HostapdCLI(config='ssidSAE.conf')
+
+    @classmethod
+    def tearDownClass(cls):
+        pass
+
+if __name__ == '__main__':
+    unittest.main(exit=True)
diff --git a/autotests/testPMKSA-SAE/hw.conf b/autotests/testPMKSA-SAE/hw.conf
new file mode 100644
index 00000000..72b161b8
--- /dev/null
+++ b/autotests/testPMKSA-SAE/hw.conf
@@ -0,0 +1,7 @@
+[SETUP]
+num_radios=2
+start_iwd=0
+hwsim_medium=yes
+
+[HOSTAPD]
+rad0=ssidSAE.conf
diff --git a/autotests/testPMKSA-SAE/ssidSAE.conf b/autotests/testPMKSA-SAE/ssidSAE.conf
new file mode 100644
index 00000000..377646b2
--- /dev/null
+++ b/autotests/testPMKSA-SAE/ssidSAE.conf
@@ -0,0 +1,12 @@
+hw_mode=g
+channel=1
+ssid=ssidSAE
+
+wpa=2
+wpa_key_mgmt=SAE
+wpa_pairwise=CCMP
+sae_password=secret123
+sae_groups=19
+ieee80211w=2
+sae_pwe=0
+rsn_preauth=1
diff --git a/autotests/testSAE-roam/connection_test.py b/autotests/testSAE-roam/connection_test.py
index ca7234a6..718bfd77 100644
--- a/autotests/testSAE-roam/connection_test.py
+++ b/autotests/testSAE-roam/connection_test.py
@@ -13,7 +13,7 @@ import testutil
 from config import ctx
 
 class Test(unittest.TestCase):
-    def validate_connection(self, wd, ft=True):
+    def validate_connection(self, wd, ft=True, check_used_pmksa=False):
         device = wd.list_devices(1)[0]
 
         # This won't guarantee all BSS's are found, but at least ensures that
@@ -37,6 +37,14 @@ class Test(unittest.TestCase):
         self.assertRaises(Exception, testutil.test_ifaces_connected,
                           (self.bss_hostapd[1].ifname, device.name, True, True))
 
+        # If PMKSA was used, hostapd should not include the sae_group key in
+        # its status for the station.
+        sta_status = self.bss_hostapd[0].sta_status(device.address)
+        if check_used_pmksa:
+            self.assertNotIn("sae_group", sta_status.keys())
+        else:
+            self.assertIn("sae_group", sta_status.keys())
+
         device.roam(self.bss_hostapd[1].bssid)
 
         # Check that iwd is on BSS 1 once out of roaming state and doesn't
@@ -88,6 +96,31 @@ class Test(unittest.TestCase):
 
         self.validate_connection(wd, True)
 
+    def test_ft_roam_pmksa(self):
+        wd = IWD(True)
+
+        self.bss_hostapd[0].set_value('wpa_key_mgmt', 'FT-SAE SAE')
+        self.bss_hostapd[0].reload()
+        self.bss_hostapd[0].wait_for_event("AP-ENABLED")
+        self.bss_hostapd[1].set_value('wpa_key_mgmt', 'FT-SAE SAE')
+        self.bss_hostapd[1].reload()
+        self.bss_hostapd[1].wait_for_event("AP-ENABLED")
+        self.bss_hostapd[2].set_value('wpa_key_mgmt', 'FT-PSK')
+        self.bss_hostapd[2].reload()
+        self.bss_hostapd[2].wait_for_event("AP-ENABLED")
+
+        self.validate_connection(wd, True)
+
+        device = wd.list_devices(1)[0]
+        device.disconnect()
+
+        for hapd in self.bss_hostapd:
+            hapd.deauthenticate(device.address)
+
+        wd.wait(5)
+
+        self.validate_connection(wd, True, check_used_pmksa=True)
+
     def test_reassociate_roam_success(self):
         wd = IWD(True)
 
@@ -103,6 +136,31 @@ class Test(unittest.TestCase):
 
         self.validate_connection(wd, False)
 
+    def test_reassociate_roam_pmksa(self):
+        wd = IWD(True)
+
+        self.bss_hostapd[0].set_value('wpa_key_mgmt', 'SAE')
+        self.bss_hostapd[0].reload()
+        self.bss_hostapd[0].wait_for_event("AP-ENABLED")
+        self.bss_hostapd[1].set_value('wpa_key_mgmt', 'SAE')
+        self.bss_hostapd[1].reload()
+        self.bss_hostapd[1].wait_for_event("AP-ENABLED")
+        self.bss_hostapd[2].set_value('wpa_key_mgmt', 'WPA-PSK')
+        self.bss_hostapd[2].reload()
+        self.bss_hostapd[2].wait_for_event("AP-ENABLED")
+
+        self.validate_connection(wd, False)
+
+        device = wd.list_devices(1)[0]
+        device.disconnect()
+
+        for hapd in self.bss_hostapd:
+            hapd.deauthenticate(device.address)
+
+        wd.wait(5)
+
+        self.validate_connection(wd, False, check_used_pmksa=True)
+
     def tearDown(self):
         os.system('ip link set "' + self.bss_hostapd[0].ifname + '" down')
         os.system('ip link set "' + self.bss_hostapd[1].ifname + '" down')
-- 
2.34.1