tests: More WPA2 PSK from RADIUS Tunnel-Password coverage

Signed-off-by: Jouni Malinen <jouni@qca.qualcomm.com>

tests/hwsim/dictionary.radius

@@ -2,6 +2,7 @@ ATTRIBUTE	User-Name		1	string
 ATTRIBUTE	User-Password		2	string
 ATTRIBUTE	NAS-IP-Address		4	ipaddr
 ATTRIBUTE	State			24	octets
+ATTRIBUTE	Session-Timeout		27	integer
 ATTRIBUTE	Calling-Station-Id	31	string
 ATTRIBUTE	NAS-Identifier		32	string
 ATTRIBUTE	Acct-Session-Id		44	string
@@ -13,5 +14,6 @@ ATTRIBUTE	Tunnel-Password		69	octets
 ATTRIBUTE	EAP-Message		79	string
 ATTRIBUTE	Message-Authenticator	80	octets
 ATTRIBUTE	Tunnel-Private-Group-ID	81	string
+ATTRIBUTE	Acct-Interim-Interval	85	integer
 ATTRIBUTE	Chargeable-User-Identity 89	string
 ATTRIBUTE	Error-Cause		101	integer

tests/hwsim/test_radius.py

@@ -1094,8 +1094,26 @@ def test_radius_protocol(dev, apdev):
         t_events['stop'].set()
         t.join()
 
-def test_radius_psk(dev, apdev):
+def build_tunnel_password(secret, authenticator, psk):
-    """WPA2 with PSK from RADIUS"""
+    a = "\xab\xcd"
+    padlen = 16 - (1 + len(psk)) % 16
+    if padlen == 16:
+        padlen = 0
+    p = struct.pack('B', len(psk)) + psk + padlen * b'\x00'
+    cc_all = bytes()
+    b = hashlib.md5(secret + authenticator + a).digest()
+    while len(p) > 0:
+        pp = bytearray(p[0:16])
+        p = p[16:]
+        bb = bytearray(b)
+        cc = bytearray(pp[i] ^ bb[i] for i in range(len(bb)))
+        cc_all += cc
+        b = hashlib.md5(secret + cc).digest()
+    data = '\x00' + a + bytes(cc_all)
+    return data
+
+def start_radius_psk_server(psk, invalid_code=False, acct_interim_interval=0,
+                            session_timeout=0, reject=False):
     try:
         import pyrad.server
         import pyrad.packet
@@ -1109,29 +1127,19 @@ def test_radius_psk(dev, apdev):
             logger.info("Received authentication request")
             reply = self.CreateReplyPacket(pkt)
             reply.code = pyrad.packet.AccessAccept
-            a = "\xab\xcd"
+            if self.t_events['invalid_code']:
-            secret = reply.secret
+                reply.code = pyrad.packet.AccessRequest
-            if self.t_events['long'].is_set():
+            if self.t_events['reject']:
-                p = b'\x10' + "0123456789abcdef" + 15 * b'\x00'
+                reply.code = pyrad.packet.AccessReject
-                b = hashlib.md5(secret + pkt.authenticator + a).digest()
+            data = build_tunnel_password(reply.secret, pkt.authenticator,
-                pp = bytearray(p[0:16])
+                                         self.t_events['psk'])
-                bb = bytearray(b)
-                cc = bytearray(pp[i] ^ bb[i] for i in range(len(bb)))
-
-                b = hashlib.md5(reply.secret + bytes(cc)).digest()
-                pp = bytearray(p[16:32])
-                bb = bytearray(b)
-                cc += bytearray(pp[i] ^ bb[i] for i in range(len(bb)))
-
-                data = '\x00' + a + bytes(cc)
-            else:
-                p = b'\x08' + "12345678" + 7 * b'\x00'
-                b = hashlib.md5(secret + pkt.authenticator + a).digest()
-                pp = bytearray(p)
-                bb = bytearray(b)
-                cc = bytearray(pp[i] ^ bb[i] for i in range(len(bb)))
-                data = '\x00' + a + bytes(cc)
             reply.AddAttribute("Tunnel-Password", data)
+            if self.t_events['acct_interim_interval']:
+                reply.AddAttribute("Acct-Interim-Interval",
+                                   self.t_events['acct_interim_interval'])
+            if self.t_events['session_timeout']:
+                reply.AddAttribute("Session-Timeout",
+                                   self.t_events['session_timeout'])
             self.SendReplyPacket(pkt.fd, reply)
 
         def RunWithStop(self, t_events):
@@ -1161,85 +1169,132 @@ def test_radius_psk(dev, apdev):
     srv.BindToAddress("")
     t_events = {}
     t_events['stop'] = threading.Event()
-    t_events['long'] = threading.Event()
+    t_events['psk'] = psk
+    t_events['invalid_code'] = invalid_code
+    t_events['acct_interim_interval'] = acct_interim_interval
+    t_events['session_timeout'] = session_timeout
+    t_events['reject'] = reject
     t = threading.Thread(target=run_pyrad_server, args=(srv, t_events))
     t.start()
+    return t, t_events
 
-    try:
+def hostapd_radius_psk_test_params():
-        ssid = "test-wpa2-psk"
     params = hostapd.radius_params()
-        params['ssid'] = ssid
+    params['ssid'] = "test-wpa2-psk"
     params["wpa"] = "2"
     params["wpa_key_mgmt"] = "WPA-PSK"
     params["rsn_pairwise"] = "CCMP"
     params['macaddr_acl'] = '2'
     params['wpa_psk_radius'] = '2'
     params['auth_server_port'] = "18138"
+    return params
+
+def test_radius_psk(dev, apdev):
+    """WPA2 with PSK from RADIUS"""
+    t, t_events = start_radius_psk_server("12345678")
+
+    try:
+        params = hostapd_radius_psk_test_params()
         hapd = hostapd.add_ap(apdev[0], params)
-        dev[0].connect(ssid, psk="12345678", scan_freq="2412")
+        dev[0].connect("test-wpa2-psk", psk="12345678", scan_freq="2412")
-        t_events['long'].set()
+        t_events['psk'] = "0123456789abcdef"
-        dev[1].connect(ssid, psk="0123456789abcdef", scan_freq="2412")
+        dev[1].connect("test-wpa2-psk", psk="0123456789abcdef",
+                       scan_freq="2412")
     finally:
         t_events['stop'].set()
         t.join()
 
 def test_radius_psk_invalid(dev, apdev):
     """WPA2 with invalid PSK from RADIUS"""
-    try:
+    t, t_events = start_radius_psk_server("1234567")
-        import pyrad.server
-        import pyrad.packet
-        import pyrad.dictionary
-    except ImportError:
-        raise HwsimSkip("No pyrad modules available")
-
-    class TestServer(pyrad.server.Server):
-        def _HandleAuthPacket(self, pkt):
-            pyrad.server.Server._HandleAuthPacket(self, pkt)
-            logger.info("Received authentication request")
-            reply = self.CreateReplyPacket(pkt)
-            reply.code = pyrad.packet.AccessAccept
-            a = "\xab\xcd"
-            secret = reply.secret
-            p = b'\x07' + "1234567" + 8 * b'\x00'
-            b = hashlib.md5(secret + pkt.authenticator + a).digest()
-            pp = bytearray(p)
-            bb = bytearray(b)
-            cc = bytearray(pp[i] ^ bb[i] for i in range(len(bb)))
-            data = '\x00' + a + bytes(cc)
-            reply.AddAttribute("Tunnel-Password", data)
-            self.SendReplyPacket(pkt.fd, reply)
-
-        def RunWithStop(self, t_events):
-            self._poll = select.poll()
-            self._fdmap = {}
-            self._PrepareSockets()
-            self.t_events = t_events
-
-            while not t_events['stop'].is_set():
-                for (fd, event) in self._poll.poll(1000):
-                    if event == select.POLLIN:
-                        try:
-                            fdo = self._fdmap[fd]
-                            self._ProcessInput(fdo)
-                        except pyrad.server.ServerPacketError as err:
-                            logger.info("pyrad server dropping packet: " + str(err))
-                        except pyrad.packet.PacketError as err:
-                            logger.info("pyrad server received invalid packet: " + str(err))
-                    else:
-                        logger.error("Unexpected event in pyrad server main loop")
-
-    srv = TestServer(dict=pyrad.dictionary.Dictionary("dictionary.radius"),
-                     authport=18138, acctport=18139)
-    srv.hosts["127.0.0.1"] = pyrad.server.RemoteHost("127.0.0.1",
-                                                     "radius",
-                                                     "localhost")
-    srv.BindToAddress("")
-    t_events = {}
-    t_events['stop'] = threading.Event()
-    t = threading.Thread(target=run_pyrad_server, args=(srv, t_events))
-    t.start()
 
     try:
+        params = hostapd_radius_psk_test_params()
+        hapd = hostapd.add_ap(apdev[0], params)
+        dev[0].connect("test-wpa2-psk", psk="12345678", scan_freq="2412",
+                       wait_connect=False)
+        time.sleep(1)
+    finally:
+        t_events['stop'].set()
+        t.join()
+
+def test_radius_psk_invalid2(dev, apdev):
+    """WPA2 with invalid PSK (hexstring) from RADIUS"""
+    t, t_events = start_radius_psk_server(64*'q')
+
+    try:
+        params = hostapd_radius_psk_test_params()
+        hapd = hostapd.add_ap(apdev[0], params)
+        dev[0].connect("test-wpa2-psk", psk="12345678", scan_freq="2412",
+                       wait_connect=False)
+        time.sleep(1)
+    finally:
+        t_events['stop'].set()
+        t.join()
+
+def test_radius_psk_hex_psk(dev, apdev):
+    """WPA2 with PSK hexstring from RADIUS"""
+    t, t_events = start_radius_psk_server(64*'2', acct_interim_interval=19,
+                                          session_timeout=123)
+
+    try:
+        params = hostapd_radius_psk_test_params()
+        hapd = hostapd.add_ap(apdev[0], params)
+        dev[0].connect("test-wpa2-psk", raw_psk=64*'2', scan_freq="2412")
+    finally:
+        t_events['stop'].set()
+        t.join()
+
+def test_radius_psk_unknown_code(dev, apdev):
+    """WPA2 with PSK from RADIUS and unknown code"""
+    t, t_events = start_radius_psk_server(64*'2', invalid_code=True)
+
+    try:
+        params = hostapd_radius_psk_test_params()
+        hapd = hostapd.add_ap(apdev[0], params)
+        dev[0].connect("test-wpa2-psk", psk="12345678", scan_freq="2412",
+                       wait_connect=False)
+        time.sleep(1)
+    finally:
+        t_events['stop'].set()
+        t.join()
+
+def test_radius_psk_reject(dev, apdev):
+    """WPA2 with PSK from RADIUS and reject"""
+    t, t_events = start_radius_psk_server("12345678", reject=True)
+
+    try:
+        params = hostapd_radius_psk_test_params()
+        hapd = hostapd.add_ap(apdev[0], params)
+        dev[0].connect("test-wpa2-psk", psk="12345678", scan_freq="2412",
+                       wait_connect=False)
+        ev = dev[0].wait_event(["CTRL-EVENT-AUTH-REJECT"], timeout=10)
+        if ev is None:
+            raise Exception("No CTRL-EVENT-AUTH-REJECT event")
+        dev[0].request("DISCONNECT")
+    finally:
+        t_events['stop'].set()
+        t.join()
+
+def test_radius_psk_oom(dev, apdev):
+    """WPA2 with PSK from RADIUS and OOM"""
+    t, t_events = start_radius_psk_server(64*'2')
+
+    try:
+        params = hostapd_radius_psk_test_params()
+        hapd = hostapd.add_ap(apdev[0], params)
+        bssid = hapd.own_addr()
+        dev[0].scan_for_bss(bssid, freq="2412")
+        with alloc_fail(hapd, 1, "=hostapd_acl_recv_radius"):
+            dev[0].connect("test-wpa2-psk", psk="12345678", scan_freq="2412",
+                           wait_connect=False)
+            wait_fail_trigger(hapd, "GET_ALLOC_FAIL")
+    finally:
+        t_events['stop'].set()
+        t.join()
+
+def test_radius_psk_default(dev, apdev):
+    """WPA2 with default PSK"""
     ssid = "test-wpa2-psk"
     params = hostapd.radius_params()
     params['ssid'] = ssid
@@ -1247,15 +1302,24 @@ def test_radius_psk_invalid(dev, apdev):
     params["wpa_key_mgmt"] = "WPA-PSK"
     params["rsn_pairwise"] = "CCMP"
     params['macaddr_acl'] = '2'
-        params['wpa_psk_radius'] = '2'
+    params['wpa_psk_radius'] = '1'
-        params['auth_server_port'] = "18138"
+    params['wpa_passphrase'] = 'qwertyuiop'
     hapd = hostapd.add_ap(apdev[0], params)
-        dev[0].connect(ssid, psk="12345678", scan_freq="2412",
+
-                       wait_connect=False)
+    dev[0].connect(ssid, psk="qwertyuiop", scan_freq="2412")
-        time.sleep(1)
+    dev[0].dump_monitor()
-    finally:
+    dev[0].request("REMOVE_NETWORK all")
-        t_events['stop'].set()
+    dev[0].wait_disconnected()
-        t.join()
+    dev[0].dump_monitor()
+
+    hapd.disable()
+    hapd.set("wpa_psk_radius", "2")
+    hapd.enable()
+    dev[0].connect(ssid, psk="qwertyuiop", scan_freq="2412", wait_connect=False)
+    ev = dev[0].wait_event(["CTRL-EVENT-AUTH-REJECT"], timeout=10)
+    if ev is None:
+        raise Exception("No CTRL-EVENT-AUTH-REJECT event")
+    dev[0].request("DISCONNECT")
 
 def test_radius_auth_force_client_addr(dev, apdev):
     """RADIUS client address specified"""