[PATCH 7/7] tests: RADIUS/TLS reverse CoA
Seb Adamski
seb at bionic.network
Sun Oct 18 23:38:14 PDT 2026
Add a small RADIUS/TLS server (radius_tls_server.py) for testing hostapd
as a RADIUS/TLS client without an external RADIUS server. hostapd's own
RADIUS server cannot send DAS requests on the RADIUS/TLS connection of a
client, so this server accepts the TLS connections on localhost with
mutual authentication (auth_serv test certificates, shared secret
"radsec"), relays Access-Request messages unchanged over UDP to a
hostapd RADIUS server instance that uses the same shared secret, and
answers Accounting-Request messages locally. A test can send a
Disconnect-Request or CoA-Request on the authentication or accounting
connection and wait for the reply, whose Response Authenticator and
Message-Authenticator are verified independently of hostapd, or send
raw data with control over TLS records and TCP segments. The received
stream is framed by the RADIUS Length field.
Add Acct-Status-Type to the test dictionary.
The test cases are skipped if pyrad is not available or if hostapd was
built without CONFIG_RADIUS_TLS=y. They use auth_server_reverse_coa=1
without radius_das_port, with radius_das_require_event_timestamp=1 and
radius_das_require_message_authenticator=1:
- Disconnect-Request for unknown sessions (NAK with Error-Cause 503,
Proxy-State echoed) and for the session by Acct-Session-Id and by
User-Name and NAS-Identifier (ACK and the station is disconnected)
- Event-Timestamp outside radius_das_time_window: dropped without a
reply, and a following request on the same connection is answered
- CoA-Request without a supported change: CoA-NAK
- auth_server_reverse_coa=0: the request is dropped and counted as an
unknown type; enabling the option with RELOAD makes it work
- with radius_das_port also set, the UDP listener keeps using its own
shared secret and the RADIUS/TLS connection the server shared secret
- the option on a RADIUS/UDP server is a configuration error
- acct_server_reverse_coa=1: request on the accounting connection, with
the Accounting-Stop sent on the same connection
- two requests in two TLS records in one TCP segment, two requests in
one TLS record, and requests split over TLS records
- malformed requests are dropped and the connection keeps working; an
invalid Length field closes the connection
- 500 requests back to back are all answered
- requests while the server does not read the responses: hostapd queues
the responses and sends them when the server reads again, survives a
TCP reset while responses are queued, and closes the connection when
too much data would be queued
- a request followed by a TCP reset, on the authentication and on the
accounting connection
Signed-off-by: Seb Adamski <seb at bionic.network>
---
.../auth_serv/radius_clients_radsec.conf | 1 +
tests/hwsim/dictionary.radius | 1 +
tests/hwsim/radius_tls_server.py | 429 +++++++++++++
tests/hwsim/test_radius.py | 606 ++++++++++++++++++
4 files changed, 1037 insertions(+)
create mode 100644 tests/hwsim/auth_serv/radius_clients_radsec.conf
create mode 100644 tests/hwsim/radius_tls_server.py
diff --git a/tests/hwsim/auth_serv/radius_clients_radsec.conf b/tests/hwsim/auth_serv/radius_clients_radsec.conf
new file mode 100644
index 000000000..faa3e16ce
--- /dev/null
+++ b/tests/hwsim/auth_serv/radius_clients_radsec.conf
@@ -0,0 +1 @@
+0.0.0.0/0 radsec
diff --git a/tests/hwsim/dictionary.radius b/tests/hwsim/dictionary.radius
index 051e49675..72b7be7a0 100644
--- a/tests/hwsim/dictionary.radius
+++ b/tests/hwsim/dictionary.radius
@@ -7,6 +7,7 @@ ATTRIBUTE Session-Timeout 27 integer
ATTRIBUTE Calling-Station-Id 31 string
ATTRIBUTE NAS-Identifier 32 string
ATTRIBUTE Proxy-State 33 octets
+ATTRIBUTE Acct-Status-Type 40 integer
ATTRIBUTE Acct-Session-Id 44 string
ATTRIBUTE Acct-Multi-Session-Id 50 string
ATTRIBUTE Event-Timestamp 55 date
diff --git a/tests/hwsim/radius_tls_server.py b/tests/hwsim/radius_tls_server.py
new file mode 100644
index 000000000..9b452f910
--- /dev/null
+++ b/tests/hwsim/radius_tls_server.py
@@ -0,0 +1,429 @@
+# RADIUS/TLS server for testing hostapd as a RADIUS/TLS client
+# Copyright (c) 2026, Seb Adamski <seb at bionic.network>
+#
+# This software may be distributed under the terms of the BSD license.
+# See README for more details.
+
+import hashlib
+import hmac
+import logging
+logger = logging.getLogger()
+import queue
+import select
+import socket
+import ssl
+import struct
+import threading
+import time
+
+import pyrad.packet
+
+RADIUS_HDR_LEN = 20
+RADIUS_MAX_LEN = 4096
+ATTR_MESSAGE_AUTHENTICATOR = 80
+SEND_TIMEOUT = 5
+
+class RadiusTlsConn:
+ def __init__(self, sock, peer):
+ self.sock = sock
+ self.peer = peer
+ self.buf = b''
+ # "auth" or "acct" based on the first request on the connection
+ self.kind = None
+
+class RadiusTlsServer:
+ """RADIUS/TLS server (RFC 6614) with mutual TLS authentication
+
+ Access-Request messages are relayed unchanged over UDP to a backend
+ RADIUS server on localhost that uses the same shared secret, and the
+ responses are sent back on the TLS connection the request came from.
+ Accounting-Request messages are answered locally. Disconnect-Request and
+ CoA-Request messages can be sent to the client on its authentication or
+ accounting connection (reverse CoA) with send_das() or, for testing the
+ framing, as raw data with send_raw().
+
+ All socket operations are done in a single server thread; the test calls
+ are passed to it as commands."""
+
+ def __init__(self, dict, port=2084, backend_port=18140,
+ secret=b"radsec", ca_cert="auth_serv/ca.pem",
+ server_cert="auth_serv/server.pem",
+ private_key="auth_serv/server.key", rcvbuf=None):
+ self.dict = dict
+ self.port = port
+ self.rcvbuf = rcvbuf
+ self.paused = False
+ self.backend = ("127.0.0.1", backend_port)
+ self.secret = secret
+ self.ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+ self.ctx.verify_mode = ssl.CERT_REQUIRED
+ self.ctx.load_verify_locations(ca_cert)
+ self.ctx.load_cert_chain(server_cert, private_key)
+ self.ctx.num_tickets = 0
+ self.conns = []
+ self.conn_event = threading.Condition()
+ self.pending = {}
+ self.acct_reqs = []
+ self.acct_event = threading.Condition()
+ self.cmdq = queue.Queue()
+ self.das_replies = queue.Queue()
+ self.stop_event = threading.Event()
+ self.thread = None
+ self.lsock = None
+ self.usock = None
+ self.wake_r = None
+ self.wake_w = None
+
+ def start(self):
+ try:
+ self.lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.lsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ if self.rcvbuf:
+ self.lsock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF,
+ self.rcvbuf)
+ self.lsock.bind(("127.0.0.1", self.port))
+ self.lsock.listen(5)
+ self.usock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ self.usock.bind(("127.0.0.1", 0))
+ self.wake_r, self.wake_w = socket.socketpair()
+ self.thread = threading.Thread(target=self.run, daemon=True)
+ self.thread.start()
+ except:
+ self.close_sockets()
+ raise
+ logger.info("RADIUS/TLS server listening on 127.0.0.1:%d" % self.port)
+
+ def close_sockets(self):
+ for c in list(self.conns):
+ self.close_conn(c)
+ for s in [self.lsock, self.usock, self.wake_r, self.wake_w]:
+ if s:
+ try:
+ s.close()
+ except OSError:
+ pass
+ self.lsock = None
+ self.usock = None
+ self.wake_r = None
+ self.wake_w = None
+
+ def stop(self):
+ self.stop_event.set()
+ alive = False
+ if self.thread:
+ try:
+ self.wake_w.send(b'x')
+ except OSError:
+ pass
+ self.thread.join(5)
+ alive = self.thread.is_alive()
+ self.thread = None
+ self.close_sockets()
+ if alive:
+ logger.info("RADIUS/TLS server: thread did not stop")
+
+ def run(self):
+ while not self.stop_event.is_set():
+ rlist = [self.lsock, self.usock, self.wake_r]
+ if not self.paused:
+ rlist += [c.sock for c in self.conns]
+ try:
+ r, w, x = select.select(rlist, [], [], 1)
+ except (OSError, ValueError) as e:
+ logger.info("RADIUS/TLS server: select failed: " + str(e))
+ break
+ for s in r:
+ try:
+ if s == self.lsock:
+ self.accept()
+ elif s == self.usock:
+ self.backend_rx()
+ elif s == self.wake_r:
+ self.wake_r.recv(100)
+ self.run_commands()
+ else:
+ for c in self.conns:
+ if c.sock == s:
+ self.conn_rx(c)
+ break
+ except Exception as e:
+ logger.info("RADIUS/TLS server: exception: " + str(e))
+
+ def run_commands(self):
+ while True:
+ try:
+ cmd = self.cmdq.get_nowait()
+ except queue.Empty:
+ return
+ cmd()
+
+ def call(self, func, timeout=SEND_TIMEOUT + 5):
+ """Run func() in the server thread and wait for it to complete"""
+ done = threading.Event()
+ def cmd():
+ try:
+ func()
+ finally:
+ done.set()
+ self.cmdq.put(cmd)
+ self.wake_w.send(b'x')
+ if not done.wait(timeout):
+ raise Exception("RADIUS/TLS server command timed out")
+
+ def accept(self):
+ s, peer = self.lsock.accept()
+ s.settimeout(5)
+ try:
+ tls = self.ctx.wrap_socket(s, server_side=True)
+ except (ssl.SSLError, OSError) as e:
+ logger.info("RADIUS/TLS server: TLS handshake with %s failed: %s"
+ % (str(peer), str(e)))
+ s.close()
+ return
+ cert = tls.getpeercert()
+ logger.info("RADIUS/TLS server: connection from %s (%s), client %s"
+ % (str(peer), tls.version(),
+ str(cert.get('subject') if cert else None)))
+ tls.setblocking(False)
+ self.conns.append(RadiusTlsConn(tls, peer))
+
+ def close_conn(self, c):
+ logger.info("RADIUS/TLS server: close %s connection from %s" %
+ (c.kind, str(c.peer)))
+ with self.conn_event:
+ if c in self.conns:
+ self.conns.remove(c)
+ self.conn_event.notify_all()
+ try:
+ c.sock.close()
+ except OSError:
+ pass
+
+ def get_conn(self, kind):
+ for c in reversed(list(self.conns)):
+ if c.kind == kind:
+ return c
+ return None
+
+ def wait_conn(self, kind="auth", timeout=5):
+ """Wait for a connection that has carried a request of the given
+ kind ("auth" or "acct")"""
+ with self.conn_event:
+ if not self.conn_event.wait_for(
+ lambda: self.get_conn(kind) is not None, timeout):
+ raise Exception("No RADIUS/TLS %s connection" % kind)
+
+ def wait_conn_closed(self, kind="auth", timeout=5):
+ with self.conn_event:
+ if not self.conn_event.wait_for(
+ lambda: self.get_conn(kind) is None, timeout):
+ raise Exception("RADIUS/TLS %s connection not closed" % kind)
+
+ def set_kind(self, c, kind):
+ if c.kind != kind:
+ with self.conn_event:
+ c.kind = kind
+ self.conn_event.notify_all()
+
+ def conn_rx(self, c):
+ while True:
+ try:
+ data = c.sock.recv(RADIUS_MAX_LEN)
+ except ssl.SSLWantReadError:
+ break
+ except (ssl.SSLError, OSError) as e:
+ logger.info("RADIUS/TLS server: recv failed: " + str(e))
+ self.close_conn(c)
+ return
+ if not data:
+ self.close_conn(c)
+ return
+ c.buf += data
+
+ # RADIUS/TLS is a stream: use the RADIUS Length field for framing
+ while len(c.buf) >= 4:
+ length = struct.unpack('>H', c.buf[2:4])[0]
+ if length < RADIUS_HDR_LEN or length > RADIUS_MAX_LEN:
+ logger.info("RADIUS/TLS server: invalid length %d" % length)
+ self.close_conn(c)
+ return
+ if len(c.buf) < length:
+ break
+ pkt = c.buf[0:length]
+ c.buf = c.buf[length:]
+ self.handle(c, pkt)
+
+ def handle(self, c, pkt):
+ code = pkt[0]
+ logger.info("RADIUS/TLS server: received code=%d id=%d len=%d from %s"
+ % (code, pkt[1], len(pkt), str(c.peer)))
+ if code == pyrad.packet.AccessRequest:
+ self.set_kind(c, "auth")
+ self.pending[pkt[1]] = c
+ self.usock.sendto(pkt, self.backend)
+ elif code == pyrad.packet.AccountingRequest:
+ self.set_kind(c, "acct")
+ req = pyrad.packet.AcctPacket(packet=pkt, dict=self.dict,
+ secret=self.secret)
+ if not req.VerifyAcctRequest():
+ logger.info("RADIUS/TLS server: invalid Accounting-Request authenticator - drop")
+ return
+ attrs = {}
+ for a in list(req.keys()):
+ attrs[a] = req[a]
+ with self.acct_event:
+ self.acct_reqs.append(attrs)
+ self.acct_event.notify_all()
+ reply = req.CreateReply()
+ self.send(c, reply.ReplyPacket())
+ elif code in [pyrad.packet.DisconnectACK, pyrad.packet.DisconnectNAK,
+ pyrad.packet.CoAACK, pyrad.packet.CoANAK]:
+ self.das_replies.put(pkt)
+ else:
+ logger.info("RADIUS/TLS server: unexpected code %d - drop" % code)
+
+ def backend_rx(self):
+ pkt, addr = self.usock.recvfrom(RADIUS_MAX_LEN)
+ if len(pkt) < RADIUS_HDR_LEN:
+ return
+ c = self.pending.pop(pkt[1], None)
+ if c is None or c not in self.conns:
+ logger.info("RADIUS/TLS server: no connection for backend response id=%d" % pkt[1])
+ return
+ self.send(c, pkt)
+
+ def send(self, c, data):
+ """Send data in one TLS write; returns False if the connection was
+ closed"""
+ deadline = time.time() + SEND_TIMEOUT
+ while data:
+ try:
+ sent = c.sock.send(data)
+ data = data[sent:]
+ except ssl.SSLWantWriteError:
+ if time.time() > deadline:
+ logger.info("RADIUS/TLS server: send timed out")
+ self.close_conn(c)
+ return False
+ select.select([], [c.sock], [], 0.1)
+ except (ssl.SSLError, OSError) as e:
+ logger.info("RADIUS/TLS server: send failed: " + str(e))
+ self.close_conn(c)
+ return False
+ return True
+
+ def send_raw(self, chunks, kind="auth", cork=False, delay=0,
+ reset=False):
+ """Send each chunk with a separate TLS write on the connection of
+ the given kind. cork=True sends all of them in one TCP segment,
+ delay waits between the chunks, and reset=True closes the connection
+ with a TCP RST after the last chunk."""
+ def send_chunks():
+ c = self.get_conn(kind)
+ if c is None:
+ logger.info("RADIUS/TLS server: no %s connection" % kind)
+ return
+ if cork:
+ c.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_CORK, 1)
+ for i, data in enumerate(chunks):
+ if i and delay:
+ time.sleep(delay)
+ if not self.send(c, data):
+ return
+ if cork:
+ c.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_CORK, 0)
+ if reset:
+ c.sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
+ struct.pack('ii', 1, 0))
+ self.close_conn(c)
+ self.call(send_chunks)
+
+ def pause(self, paused):
+ """Stop (paused=True) or resume reading from the connections"""
+ def set_paused():
+ self.paused = paused
+ self.call(set_paused)
+
+ def clear_das_replies(self):
+ while not self.das_replies.empty():
+ self.das_replies.get()
+
+ def get_das_replies(self, count, timeout=3):
+ """Return up to count raw DAS replies received within timeout"""
+ replies = []
+ end = time.time() + timeout
+ while len(replies) < count:
+ left = end - time.time()
+ if left <= 0:
+ break
+ try:
+ replies.append(self.das_replies.get(timeout=left))
+ except queue.Empty:
+ break
+ return replies
+
+ def wait_acct(self, status_type, timeout=5):
+ """Wait for an Accounting-Request with the given Acct-Status-Type
+ (1 = Start, 2 = Stop) and return its attributes"""
+ def find():
+ for attrs in self.acct_reqs:
+ if attrs.get('Acct-Status-Type', [None])[0] == status_type:
+ return attrs
+ return None
+ with self.acct_event:
+ self.acct_event.wait_for(lambda: find() is not None, timeout)
+ attrs = find()
+ if attrs is None:
+ raise Exception("No Accounting-Request with Acct-Status-Type %d"
+ % status_type)
+ return attrs
+
+ def send_das(self, req, kind="auth", timeout=2):
+ """Send a Disconnect-Request or CoA-Request on the connection of the
+ given kind and wait for the reply on the same connection. Returns the
+ verified reply packet or None on timeout."""
+ data = req.RequestPacket()
+ self.clear_das_replies()
+ self.send_raw([data], kind=kind)
+ end = time.time() + timeout
+ while True:
+ left = end - time.time()
+ if left <= 0:
+ return None
+ try:
+ pkt = self.das_replies.get(timeout=left)
+ except queue.Empty:
+ return None
+ if pkt[1] != req.id:
+ logger.info("RADIUS/TLS server: ignore reply with id=%d" %
+ pkt[1])
+ continue
+ self.check_reply_auth(pkt, req.authenticator)
+ reply = pyrad.packet.Packet(packet=pkt, dict=self.dict,
+ secret=self.secret)
+ logger.debug("DAS reply code=%d" % reply.code)
+ for a in list(reply.keys()):
+ logger.debug("%s: %s" % (a, reply[a]))
+ return reply
+
+ def check_reply_auth(self, pkt, req_auth):
+ hdr = pkt[0:4]
+ attrs = pkt[RADIUS_HDR_LEN:]
+ auth = hashlib.md5(hdr + req_auth + attrs + self.secret).digest()
+ if auth != pkt[4:20]:
+ raise Exception("Invalid Response Authenticator in DAS reply")
+
+ pos = 0
+ while pos + 2 <= len(attrs):
+ t, l = attrs[pos], attrs[pos + 1]
+ if l < 2:
+ break
+ if t == ATTR_MESSAGE_AUTHENTICATOR and l == 18:
+ zeroed = attrs[0:pos + 2] + 16 * b'\x00' + attrs[pos + 18:]
+ mac = hmac.new(self.secret, hdr + req_auth + zeroed,
+ hashlib.md5).digest()
+ if mac != attrs[pos + 2:pos + 18]:
+ raise Exception("Invalid Message-Authenticator in DAS reply")
+ return
+ pos += l
+ raise Exception("No Message-Authenticator in DAS reply")
diff --git a/tests/hwsim/test_radius.py b/tests/hwsim/test_radius.py
index c4a6f05e8..00d344e66 100644
--- a/tests/hwsim/test_radius.py
+++ b/tests/hwsim/test_radius.py
@@ -1952,6 +1952,612 @@ def test_radius_tls_freeradius(dev, apdev, test_params):
if pid > 0:
os.kill(pid, signal.SIGTERM)
+RADIUS_TLS_PORT = 2084
+RADIUS_TLS_BACKEND_PORT = 18140
+
+def check_radius_tls_capa(apdev):
+ hapd = hostapd.add_ap(apdev, {"ssid": "radius-tls-capa",
+ "auth_server_addr": "127.0.0.1"},
+ no_enable=True)
+ if "OK" not in hapd.request("SET auth_server_type TLS"):
+ raise HwsimSkip("RADIUS/TLS not supported in the hostapd build")
+
+def start_radius_tls_reverse_coa(dev, apdev, ssid, extra_params=None,
+ backend_params=None, server_params=None):
+ try:
+ import pyrad.packet
+ import pyrad.dictionary
+ import radius_das
+ import radius_tls_server
+ except ImportError:
+ raise HwsimSkip("No pyrad modules available")
+ check_radius_tls_capa(apdev[0])
+
+ # hostapd's RADIUS server cannot send DAS requests on the RADIUS/TLS
+ # connection of a client, so a small RADIUS/TLS server is used. It relays
+ # Access-Request messages unchanged to this RADIUS server for the EAP
+ # exchange, so this server uses the RADIUS/TLS shared secret.
+ params = {'ssid': 'as-radsec',
+ 'beacon_int': '2000',
+ 'radius_server_clients': 'auth_serv/radius_clients_radsec.conf',
+ 'radius_server_auth_port': str(RADIUS_TLS_BACKEND_PORT),
+ 'eap_server': '1',
+ 'eap_user_file': 'auth_serv/eap_user.conf',
+ 'ca_cert': 'auth_serv/ca.pem',
+ 'server_cert': 'auth_serv/server.pem',
+ 'private_key': 'auth_serv/server.key'}
+ if backend_params:
+ params.update(backend_params)
+ hostapd.add_ap(apdev[1], params)
+
+ dict = pyrad.dictionary.Dictionary("dictionary.radius")
+ srv = radius_tls_server.RadiusTlsServer(
+ dict, port=RADIUS_TLS_PORT, backend_port=RADIUS_TLS_BACKEND_PORT,
+ **(server_params or {}))
+ try:
+ srv.start()
+ params = hostapd.wpa2_eap_params(ssid=ssid)
+ for s in ["auth", "acct"]:
+ params[s + '_server_addr'] = "127.0.0.1"
+ params[s + '_server_port'] = str(RADIUS_TLS_PORT)
+ params[s + '_server_type'] = "TLS"
+ params[s + '_server_shared_secret'] = "radsec"
+ params[s + '_server_ca_cert'] = "auth_serv/ca.pem"
+ params[s + '_server_client_cert'] = "auth_serv/user.pem"
+ params[s + '_server_private_key'] = "auth_serv/user.key"
+ params['auth_server_reverse_coa'] = "1"
+ params['radius_das_require_event_timestamp'] = "1"
+ params['radius_das_require_message_authenticator'] = "1"
+ params['own_ip_addr'] = "127.0.0.1"
+ params['nas_identifier'] = "nas.example.com"
+ if extra_params:
+ params.update(extra_params)
+ hapd = hostapd.add_ap(apdev[0], params)
+ except:
+ srv.stop()
+ raise
+
+ return srv, hapd, dict
+
+def rcoa_req(dict, code=None, rid=None, proxy_state=None, **attrs):
+ import pyrad.packet
+ import radius_das
+ if code is None:
+ code = pyrad.packet.DisconnectRequest
+ if 'Event_Timestamp' not in attrs:
+ attrs['Event_Timestamp'] = int(time.time())
+ req = radius_das.DisconnectPacket(code=code, id=rid, dict=dict,
+ secret=b"radsec", **attrs)
+ for val in proxy_state or []:
+ req.AddAttribute("Proxy-State", val)
+ add_message_auth_req(req)
+ return req
+
+def rcoa_connect(dev, hapd, srv, ssid="radius-tls-rcoa"):
+ connect(dev, ssid)
+ addr = dev.own_addr()
+ hapd.wait_sta(addr=addr)
+ srv.wait_conn("auth")
+ return addr
+
+def test_radius_tls_reverse_coa_disconnect(dev, apdev):
+ """RADIUS/TLS reverse CoA - Disconnect-Request on the TLS connection"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(dev, apdev,
+ "radius-tls-rcoa")
+ try:
+ import pyrad.packet
+
+ addr = rcoa_connect(dev[0], hapd, srv)
+ sta = hapd.get_sta(addr)
+ start = srv.wait_acct(1)
+ acct_id = start['Acct-Session-Id'][0]
+ if acct_id != sta['dot1xAuthSessionId']:
+ raise Exception("Unexpected Acct-Session-Id: " + acct_id)
+
+ logger.info("Disconnect-Request with unknown Acct-Session-Id and Proxy-State")
+ proxy_state = [b"proxy-state-1", b"\x00\x01\x02\x03"]
+ req = rcoa_req(dict, Acct_Session_Id="0123456789abcdef",
+ proxy_state=proxy_state)
+ check_das_reply(srv.send_das(req), pyrad.packet.DisconnectNAK,
+ error_cause=503, proxy_state=proxy_state)
+
+ logger.info("Disconnect-Request with unknown User-Name")
+ req = rcoa_req(dict, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com")
+ check_das_reply(srv.send_das(req), pyrad.packet.DisconnectNAK,
+ error_cause=503)
+ ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.5)
+ if ev is not None:
+ raise Exception("Unexpected disconnection")
+
+ logger.info("Disconnect-Request with matching Acct-Session-Id")
+ req = rcoa_req(dict, Acct_Session_Id=acct_id)
+ check_das_reply(srv.send_das(req), pyrad.packet.DisconnectACK)
+ dev[0].wait_disconnected(timeout=10)
+ hapd.wait_sta_disconnect(addr=addr)
+ dev[0].wait_connected(timeout=15, error="Re-connection timed out")
+ hapd.wait_sta(addr=addr)
+
+ logger.info("Disconnect-Request with matching User-Name and NAS-Identifier")
+ req = rcoa_req(dict, User_Name="psk.user at example.com",
+ NAS_Identifier="nas.example.com")
+ check_das_reply(srv.send_das(req), pyrad.packet.DisconnectACK)
+ dev[0].wait_disconnected(timeout=10)
+ hapd.wait_sta_disconnect(addr=addr)
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_time_window(dev, apdev):
+ """RADIUS/TLS reverse CoA - Event-Timestamp outside the time window"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(
+ dev, apdev, "radius-tls-rcoa",
+ extra_params={'radius_das_time_window': "10"})
+ try:
+ import pyrad.packet
+
+ addr = rcoa_connect(dev[0], hapd, srv)
+
+ logger.info("Disconnect-Request with Event-Timestamp outside the time window")
+ req = rcoa_req(dict, NAS_IP_Address="127.0.0.1",
+ NAS_Identifier="nas.example.com",
+ Calling_Station_Id=addr,
+ Event_Timestamp=int(time.time()) - 50)
+ reply = srv.send_das(req, timeout=1)
+ if reply is not None:
+ raise Exception("Unexpected response to Disconnect-Request: code %d" % reply.code)
+ ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.5)
+ if ev is not None:
+ raise Exception("Unexpected disconnection")
+ if hapd.get_sta(addr).get('addr') != addr:
+ raise Exception("Station not associated after dropped request")
+
+ logger.info("Disconnect-Request with Event-Timestamp within the time window")
+ req = rcoa_req(dict, NAS_IP_Address="127.0.0.1",
+ NAS_Identifier="nas.example.com",
+ Calling_Station_Id=addr)
+ check_das_reply(srv.send_das(req), pyrad.packet.DisconnectACK)
+ dev[0].wait_disconnected(timeout=10)
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_coa(dev, apdev):
+ """RADIUS/TLS reverse CoA - CoA-Request on the TLS connection"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(dev, apdev,
+ "radius-tls-rcoa")
+ try:
+ import pyrad.packet
+
+ addr = rcoa_connect(dev[0], hapd, srv)
+ sta = hapd.get_sta(addr)
+
+ # Only the Hotspot 2.0 Terms and Conditions filtering change is
+ # supported for CoA and only with CONFIG_HS20=y: without it, the
+ # NAK is for an unsupported request type (406), otherwise for a
+ # request with no supported authorization change (402).
+ hs20 = "OK" in hapd.request("SET hs20 0")
+ logger.info("CoA-Request without a supported change (HS20 build: %s)" % hs20)
+ req = rcoa_req(dict, code=pyrad.packet.CoARequest,
+ Acct_Session_Id=sta['dot1xAuthSessionId'])
+ check_das_reply(srv.send_das(req), pyrad.packet.CoANAK,
+ error_cause=402 if hs20 else 406)
+ ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.5)
+ if ev is not None:
+ raise Exception("Unexpected disconnection")
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_disabled(dev, apdev):
+ """RADIUS/TLS reverse CoA - option disabled, then enabled with RELOAD"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(
+ dev, apdev, "radius-tls-rcoa",
+ extra_params={'auth_server_reverse_coa': "0"})
+ try:
+ import pyrad.packet
+
+ addr = rcoa_connect(dev[0], hapd, srv)
+ unknown = int(hapd.get_mib().get('radiusAuthClientUnknownTypes', '0'))
+
+ logger.info("Disconnect-Request without auth_server_reverse_coa")
+ req = rcoa_req(dict, Calling_Station_Id=addr,
+ NAS_Identifier="nas.example.com")
+ reply = srv.send_das(req, timeout=1)
+ if reply is not None:
+ raise Exception("Unexpected response to Disconnect-Request: code %d" % reply.code)
+ ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.5)
+ if ev is not None:
+ raise Exception("Unexpected disconnection")
+ val = int(hapd.get_mib().get('radiusAuthClientUnknownTypes', '0'))
+ if val != unknown + 1:
+ raise Exception("Dropped request not counted (%d -> %d)" %
+ (unknown, val))
+
+ logger.info("Enable auth_server_reverse_coa with RELOAD")
+ hapd.set("auth_server_reverse_coa", "1")
+ if "OK" not in hapd.request("RELOAD"):
+ raise Exception("RELOAD failed")
+ ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=2)
+ if ev is not None:
+ dev[0].wait_connected(timeout=15)
+ hapd.wait_sta(addr=addr, timeout=5)
+ srv.wait_conn("auth")
+
+ req = rcoa_req(dict, Calling_Station_Id=addr,
+ NAS_Identifier="nas.example.com")
+ check_das_reply(srv.send_das(req), pyrad.packet.DisconnectACK)
+ dev[0].wait_disconnected(timeout=10)
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_udp_das(dev, apdev):
+ """RADIUS/TLS reverse CoA together with the UDP DAS listener"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(
+ dev, apdev, "radius-tls-rcoa",
+ extra_params={'radius_das_port': "3799",
+ 'radius_das_client': "127.0.0.1 secret"})
+ try:
+ import pyrad.client
+ import pyrad.packet
+ import radius_das
+
+ addr = rcoa_connect(dev[0], hapd, srv)
+ udp = pyrad.client.Client(server="127.0.0.1", acctport=3799,
+ secret=b"secret", dict=dict)
+ udp.retries = 1
+ udp.timeout = 1
+
+ logger.info("UDP DAS uses radius_das_client shared secret")
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
+ User_Name="unknown at example.com",
+ Event_Timestamp=int(time.time()))
+ add_message_auth_req(req)
+ check_das_reply(udp.SendPacket(req), pyrad.packet.DisconnectNAK,
+ error_cause=503)
+
+ logger.info("RADIUS/TLS uses the server shared secret")
+ req = radius_das.DisconnectPacket(dict=dict, secret=b"secret",
+ User_Name="unknown at example.com",
+ Event_Timestamp=int(time.time()))
+ add_message_auth_req(req)
+ if srv.send_das(req, timeout=1) is not None:
+ raise Exception("Unexpected response to a request with the UDP DAS shared secret")
+ req = rcoa_req(dict, Calling_Station_Id=addr,
+ NAS_Identifier="nas.example.com")
+ check_das_reply(srv.send_das(req), pyrad.packet.DisconnectACK)
+ dev[0].wait_disconnected(timeout=10)
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_udp_server(dev, apdev):
+ """RADIUS/TLS reverse CoA - option on a RADIUS/UDP server"""
+ check_radius_tls_capa(apdev[0])
+ for s in ["auth", "acct"]:
+ params = hostapd.wpa2_eap_params(ssid="radius-tls-rcoa")
+ params['acct_server_addr'] = "127.0.0.1"
+ params['acct_server_port'] = "1813"
+ params['acct_server_shared_secret'] = "radius"
+ params[s + '_server_reverse_coa'] = "1"
+ hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
+ if "FAIL" not in hapd.request("ENABLE"):
+ raise Exception("Unexpected ENABLE success with %s_server_reverse_coa=1 on a RADIUS/UDP server" % s)
+
+def test_radius_tls_reverse_coa_acct(dev, apdev):
+ """RADIUS/TLS reverse CoA - Disconnect-Request on the accounting connection"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(
+ dev, apdev, "radius-tls-rcoa",
+ extra_params={'acct_server_reverse_coa': "1"})
+ try:
+ import pyrad.packet
+
+ addr = rcoa_connect(dev[0], hapd, srv)
+ start = srv.wait_acct(1)
+ srv.wait_conn("acct")
+
+ req = rcoa_req(dict, Acct_Session_Id=start['Acct-Session-Id'][0])
+ check_das_reply(srv.send_das(req, kind="acct"),
+ pyrad.packet.DisconnectACK)
+ dev[0].wait_disconnected(timeout=10)
+ hapd.wait_sta_disconnect(addr=addr)
+ stop = srv.wait_acct(2)
+ if stop['Acct-Session-Id'][0] != start['Acct-Session-Id'][0]:
+ raise Exception("Unexpected Acct-Session-Id in Accounting-Stop")
+ finally:
+ srv.stop()
+
+def check_rcoa_replies(srv, ids, timeout=3):
+ import pyrad.packet
+ replies = srv.get_das_replies(len(ids), timeout=timeout)
+ got = [pkt[1] for pkt in replies]
+ logger.info("Replies received for ids " + str(got))
+ if sorted(got) != sorted(ids):
+ raise Exception("Replies for ids %s, expected %s" % (got, ids))
+ for pkt in replies:
+ if pkt[0] != pyrad.packet.DisconnectNAK:
+ raise Exception("Unexpected reply code %d" % pkt[0])
+
+def test_radius_tls_reverse_coa_two_records(dev, apdev):
+ """RADIUS/TLS reverse CoA - two requests in two TLS records in one TCP segment"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(dev, apdev,
+ "radius-tls-rcoa")
+ try:
+ rcoa_connect(dev[0], hapd, srv)
+ reqs = [rcoa_req(dict, rid=i, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com").RequestPacket()
+ for i in [101, 102]]
+ srv.clear_das_replies()
+ srv.send_raw(reqs, cork=True)
+ check_rcoa_replies(srv, [101, 102])
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_one_record(dev, apdev):
+ """RADIUS/TLS reverse CoA - two requests in one TLS record"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(dev, apdev,
+ "radius-tls-rcoa")
+ try:
+ rcoa_connect(dev[0], hapd, srv)
+ reqs = [rcoa_req(dict, rid=i, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com").RequestPacket()
+ for i in [111, 112]]
+ srv.clear_das_replies()
+ srv.send_raw([reqs[0] + reqs[1]])
+ check_rcoa_replies(srv, [111, 112])
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_split(dev, apdev):
+ """RADIUS/TLS reverse CoA - request split over several TLS records"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(dev, apdev,
+ "radius-tls-rcoa")
+ try:
+ rcoa_connect(dev[0], hapd, srv)
+ req = rcoa_req(dict, rid=121, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com").RequestPacket()
+ req2 = rcoa_req(dict, rid=122, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com").RequestPacket()
+ srv.clear_das_replies()
+ # Partial header, rest of the first request together with the start
+ # of the second one, and the rest of the second one
+ srv.send_raw([req[0:2], req[2:] + req2[0:30], req2[30:]], delay=0.2)
+ check_rcoa_replies(srv, [121, 122])
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_malformed(dev, apdev):
+ """RADIUS/TLS reverse CoA - malformed requests"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(dev, apdev,
+ "radius-tls-rcoa")
+ try:
+ rcoa_connect(dev[0], hapd, srv)
+ good = rcoa_req(dict, rid=131, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com").RequestPacket()
+ bad = []
+ # Invalid Message-Authenticator
+ b = bytearray(good)
+ b[-1] ^= 0xff
+ bad.append(bytes(b))
+ # Attribute with length 0
+ b = bytearray(good[0:20]) + bytes([1, 0])
+ b[2:4] = struct.pack('>H', len(b))
+ bad.append(bytes(b))
+ # Attribute extending beyond the end of the message
+ b = bytearray(good[0:20]) + bytes([1, 200, 0x41])
+ b[2:4] = struct.pack('>H', len(b))
+ bad.append(bytes(b))
+ srv.clear_das_replies()
+ for data in bad:
+ srv.send_raw([data])
+ srv.send_raw([good])
+ check_rcoa_replies(srv, [131])
+
+ logger.info("Invalid RADIUS Length field closes the connection")
+ b = bytearray(good)
+ b[2:4] = struct.pack('>H', 10)
+ srv.send_raw([bytes(b)])
+ srv.wait_conn_closed("auth")
+ if "PONG" not in hapd.request("PING"):
+ raise Exception("hostapd not responding")
+
+ logger.info("A new connection is used for the next authentication")
+ dev[0].request("REAUTHENTICATE")
+ ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
+ if ev is None:
+ raise Exception("Reauthentication timed out")
+ srv.wait_conn("auth")
+ req = rcoa_req(dict, rid=132, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com")
+ srv.clear_das_replies()
+ srv.send_raw([req.RequestPacket()])
+ check_rcoa_replies(srv, [132])
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_flood(dev, apdev):
+ """RADIUS/TLS reverse CoA - many requests back to back"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(dev, apdev,
+ "radius-tls-rcoa")
+ try:
+ addr = rcoa_connect(dev[0], hapd, srv)
+ num = 500
+ reqs = [rcoa_req(dict, rid=i % 256, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com").RequestPacket()
+ for i in range(num)]
+ srv.clear_das_replies()
+ start = time.time()
+ srv.send_raw(reqs)
+ replies = srv.get_das_replies(num, timeout=20)
+ logger.info("%d/%d replies in %.2f s" % (len(replies), num,
+ time.time() - start))
+ if len(replies) != num:
+ raise Exception("Only %d of %d requests answered" %
+ (len(replies), num))
+ if hapd.get_sta(addr).get('addr') != addr:
+ raise Exception("Station not associated after the requests")
+
+ req = rcoa_req(dict, rid=200, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com")
+ srv.send_raw([req.RequestPacket()])
+ check_rcoa_replies(srv, [200])
+ finally:
+ srv.stop()
+
+def rcoa_fill_until(srv, dict, log, pattern, batch=10, max_reqs=2000):
+ """Send requests with large Proxy-State attributes, which hostapd copies
+ into its responses, while the server does not read the responses, until
+ pattern shows up in the hostapd debug log. Each batch is processed by
+ hostapd before the log is checked and the next batch is sent, so at
+ most one batch of responses is queued when the pattern is found, and
+ with batch=1 nothing is sent after hostapd closed the connection.
+ Returns the number of requests sent."""
+ proxy_state = 15 * [250 * b'p']
+ state = {'buf': '', 'processed': 0, 'found': False}
+
+ def scan():
+ state['buf'] += log.read()
+ lines = state['buf'].split('\n')
+ state['buf'] = lines.pop()
+ for line in lines:
+ if "RADIUS: Received DAS request" in line:
+ state['processed'] += 1
+ if pattern in line:
+ state['found'] = True
+
+ sent = 0
+ while sent < max_reqs:
+ reqs = [rcoa_req(dict, rid=(sent + i) % 256,
+ User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com",
+ proxy_state=proxy_state).RequestPacket()
+ for i in range(batch)]
+ srv.send_raw(reqs)
+ sent += len(reqs)
+ for i in range(500):
+ scan()
+ if state['processed'] >= sent:
+ break
+ time.sleep(0.01)
+ time.sleep(0.05)
+ scan()
+ if state['found']:
+ return sent
+ raise Exception("'%s' not seen in hostapd log after %d requests" %
+ (pattern, sent))
+
+def test_radius_tls_reverse_coa_backpressure(dev, apdev, params):
+ """RADIUS/TLS reverse CoA - responses while the server does not read"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(
+ dev, apdev, "radius-tls-rcoa", server_params={'rcvbuf': 4096})
+ try:
+ addr = rcoa_connect(dev[0], hapd, srv)
+ log = open(os.path.join(params['logdir'], 'hostapd'), 'r',
+ errors='ignore')
+ log.seek(0, os.SEEK_END)
+
+ logger.info("Responses are queued and sent when the server reads again")
+ srv.clear_das_replies()
+ srv.pause(True)
+ num = rcoa_fill_until(srv, dict, log, "RADIUS: Queue ")
+ logger.info("hostapd queued TLS data after %d requests" % num)
+ srv.pause(False)
+ replies = srv.get_das_replies(num, timeout=20)
+ if len(replies) != num:
+ raise Exception("Only %d of %d requests answered" %
+ (len(replies), num))
+ if "RADIUS: Sent " not in log.read():
+ raise Exception("Queued TLS data not reported as sent")
+ req = rcoa_req(dict, rid=201, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com")
+ srv.send_raw([req.RequestPacket()])
+ check_rcoa_replies(srv, [201])
+
+ logger.info("TCP RST while responses are queued")
+ srv.pause(True)
+ rcoa_fill_until(srv, dict, log, "RADIUS: Queue ")
+ srv.send_raw([], reset=True)
+ srv.pause(False)
+ srv.wait_conn_closed("auth")
+ time.sleep(0.5)
+ if "PONG" not in hapd.request("PING"):
+ raise Exception("hostapd not responding after TCP RST")
+ dev[0].request("REAUTHENTICATE")
+ ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
+ if ev is None:
+ raise Exception("Reauthentication timed out")
+ srv.wait_conn("auth")
+
+ logger.info("Connection is closed when too much data is queued")
+ srv.pause(True)
+ rcoa_fill_until(srv, dict, log, "RADIUS: Queue ")
+ rcoa_fill_until(srv, dict, log, "No buffer space available", batch=1)
+ srv.pause(False)
+ srv.wait_conn_closed("auth")
+ if "PONG" not in hapd.request("PING"):
+ raise Exception("hostapd not responding")
+ if hapd.get_sta(addr).get('addr') != addr:
+ raise Exception("Station not associated")
+ dev[0].request("REAUTHENTICATE")
+ ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
+ if ev is None:
+ raise Exception("Reauthentication timed out")
+ srv.wait_conn("auth")
+ req = rcoa_req(dict, rid=202, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com")
+ srv.clear_das_replies()
+ srv.send_raw([req.RequestPacket()])
+ check_rcoa_replies(srv, [202])
+ log.close()
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_rst(dev, apdev):
+ """RADIUS/TLS reverse CoA - request followed by a TCP RST"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(dev, apdev,
+ "radius-tls-rcoa")
+ try:
+ rcoa_connect(dev[0], hapd, srv)
+ req = rcoa_req(dict, rid=141, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com")
+ srv.send_raw([req.RequestPacket()], reset=True)
+ time.sleep(1)
+ if "PONG" not in hapd.request("PING"):
+ raise Exception("hostapd not responding after TCP RST")
+
+ dev[0].request("REAUTHENTICATE")
+ ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
+ if ev is None:
+ raise Exception("Reauthentication timed out")
+ srv.wait_conn("auth")
+ req = rcoa_req(dict, rid=142, User_Name="unknown at example.com",
+ NAS_Identifier="nas.example.com")
+ srv.clear_das_replies()
+ srv.send_raw([req.RequestPacket()])
+ check_rcoa_replies(srv, [142])
+ finally:
+ srv.stop()
+
+def test_radius_tls_reverse_coa_acct_rst(dev, apdev):
+ """RADIUS/TLS reverse CoA - request on the accounting connection followed by a TCP RST"""
+ srv, hapd, dict = start_radius_tls_reverse_coa(
+ dev, apdev, "radius-tls-rcoa",
+ extra_params={'acct_server_reverse_coa': "1"})
+ try:
+ addr = rcoa_connect(dev[0], hapd, srv)
+ srv.wait_acct(1)
+ srv.wait_conn("acct")
+
+ # The request matches the station, so hostapd sends Accounting-Stop
+ # and the response on the connection that was reset.
+ req = rcoa_req(dict, rid=151, User_Name="psk.user at example.com",
+ NAS_Identifier="nas.example.com")
+ srv.send_raw([req.RequestPacket()], kind="acct", reset=True)
+ ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=2)
+ if "PONG" not in hapd.request("PING"):
+ raise Exception("hostapd not responding after TCP RST")
+ if ev is not None:
+ dev[0].wait_connected(timeout=15)
+ hapd.wait_sta(addr=addr, timeout=5)
+ finally:
+ srv.stop()
+
def test_radius_eapol_test(dev, apdev, test_params):
"""RADIUS testing with eapol_test"""
et_path = "../../wpa_supplicant/eapol_test"
--
2.54.0 (Apple Git-157)
More information about the Hostap
mailing list