[PATCH v2 6/6] tests: hwsim: add hwsim relay code
Johannes Berg
johannes at sipsolutions.net
Fri Sep 11 15:59:54 PDT 2026
From: Johannes Berg <johannes.berg at intel.com>
For a lot of tests that we don't have now, it'd be useful to be
able to modify or drop arbitrary frames going across the medium,
such as beacons to test various CSA/critical update scenarios,
missed beacons overlapping with CSA to force mac80211 to detect
it from another link, etc. Even where it's not error conditions
hostapd doesn't necessarily support all features we want to test.
Add a hwsim relay module that registers with the wmediumd APIs
and can be used by tests to intercept frames. By default it'll
relay all frames unmodified, but frame handling layers can be
stacked on top of it as in the example test. Right now the code
has one to wait for frames (relay.WaitFrame), recalculate BIP
for protected group management frames and beacons after they've
been modified by further layers (bip.GroupMgmtSigner) and one to
decrypt management frames before passing to the next layer and
re-encrypt them if they were changed (ccmp.ProtectedMgmtCrypto).
The choice for CCMP is because that's simpler with python.
All of this will likely only work right with UML's time-travel,
but that's not really that much of an issue, and it allows us
to write a large number of tests that'd otherwise be impossible.
Signed-off-by: Johannes Berg <johannes.berg at intel.com>
---
v2: async
---
tests/hwsim/hostapd.py | 7 +
tests/hwsim/relay/__init__.py | 305 ++++++++++++++++++++++++++++++++
tests/hwsim/relay/bip.py | 86 +++++++++
tests/hwsim/relay/ccmp.py | 97 ++++++++++
tests/hwsim/relay/frame.py | 119 +++++++++++++
tests/hwsim/test_hwsim_relay.py | 146 +++++++++++++++
6 files changed, 760 insertions(+)
create mode 100644 tests/hwsim/relay/__init__.py
create mode 100644 tests/hwsim/relay/bip.py
create mode 100644 tests/hwsim/relay/ccmp.py
create mode 100644 tests/hwsim/relay/frame.py
create mode 100644 tests/hwsim/test_hwsim_relay.py
diff --git a/tests/hwsim/hostapd.py b/tests/hwsim/hostapd.py
index e855390c7fd6..00b18f9357b4 100644
--- a/tests/hwsim/hostapd.py
+++ b/tests/hwsim/hostapd.py
@@ -432,6 +432,13 @@ class Hostapd:
res = self.request(cmd + addr + " " + info)
else:
res = self.request(cmd + addr)
+ return self._parse_sta_mib(res)
+
+ def get_link_sta(self, addr):
+ return self._parse_sta_mib(self.request("LINK_STA " + addr))
+
+ @staticmethod
+ def _parse_sta_mib(res):
lines = res.splitlines()
vals = dict()
first = True
diff --git a/tests/hwsim/relay/__init__.py b/tests/hwsim/relay/__init__.py
new file mode 100644
index 000000000000..972d9a5c78c5
--- /dev/null
+++ b/tests/hwsim/relay/__init__.py
@@ -0,0 +1,305 @@
+#
+# hwsim relay - interceptor for hwsim using wmediumd API
+#
+# Copyright (C) 2026 Intel Corporation
+#
+# This software may be distributed under the terms of the BSD license.
+# See README for more details.
+#
+#
+# This code allows intercepting and manipulating/dropping frames that
+# are being transmitted over the hwsim medium simulation, for testing
+# purposes.
+#
+# Note that it will only really work in UML's time-travel.
+#
+
+import asyncio
+import os
+import errno
+import logging
+
+import netlink
+
+logger = logging.getLogger()
+
+HWSIM_CMD_REGISTER = 1
+HWSIM_CMD_FRAME = 2
+HWSIM_CMD_TX_INFO_FRAME = 3
+
+HWSIM_ATTR_ADDR_RECEIVER = 1
+HWSIM_ATTR_ADDR_TRANSMITTER = 2
+HWSIM_ATTR_FRAME = 3
+HWSIM_ATTR_FLAGS = 4
+HWSIM_ATTR_RX_RATE = 5
+HWSIM_ATTR_SIGNAL = 6
+HWSIM_ATTR_TX_INFO = 7
+HWSIM_ATTR_COOKIE = 8
+HWSIM_ATTR_FREQ = 19
+HWSIM_ATTR_NO_MONITOR = 32
+
+HWSIM_TX_CTL_REQ_TX_STATUS = 1 << 0
+HWSIM_TX_CTL_NO_ACK = 1 << 1
+HWSIM_TX_STAT_ACK = 1 << 2
+
+IEEE80211_TX_MAX_RATES = 4
+
+
+def _tx_info_all_invalid():
+ # struct hwsim_tx_rate { idx = -1, count = 0 }
+ return b'\xff\x00' * IEEE80211_TX_MAX_RATES
+
+
+def radio_addr_for_iface(ifname):
+ phy = os.path.basename(os.readlink(
+ '/sys/class/net/%s/phy80211' % ifname))
+ with open('/sys/class/ieee80211/%s/addresses' % phy) as f:
+ addrs = f.read().split()
+ return bytes.fromhex(addrs[1].replace(':', ''))
+
+
+class RelayLayer:
+ """
+ Base layer for frame interception - each layer has its own
+ .intercept() like the HwsimRelay() itself.
+
+ Used as "async with XyzLayer(...) as layer:" just like
+ the relay itself.
+ """
+
+ def __init__(self, upstream):
+ self._upstream = upstream
+ self._downstream = self._default
+
+ def __enter__(self):
+ self._prev = self._upstream.intercept
+ self._upstream.intercept = self._run
+ return self
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ self._upstream.intercept = self._prev
+
+ @property
+ def intercept(self):
+ return self._downstream
+
+ @intercept.setter
+ def intercept(self, fn):
+ self._downstream = fn
+
+ def _default(self, src, freq, frame):
+ return frame
+
+ def _run(self, src, freq, frame):
+ raise NotImplementedError
+
+
+class WaitFrame(RelayLayer):
+ def __init__(self, upstream, matchers):
+ super().__init__(upstream)
+ self._matchers = list(matchers)
+ self._results = asyncio.Queue()
+
+ def _run(self, src, freq, frame):
+ for matcher in self._matchers:
+ result = matcher.match(src, freq, frame)
+ if result is not None:
+ self._results.put_nowait(result)
+ break
+ return self.intercept(src, freq, frame)
+
+ async def wait_for_frame(self, timeout=5):
+ try:
+ return await asyncio.wait_for(self._results.get(), timeout=timeout)
+ except asyncio.TimeoutError:
+ return None
+
+
+_MGMT_SUBTYPES = {
+ 0: 'AssocReq',
+ 1: 'AssocResp',
+ 2: 'ReassocReq',
+ 3: 'ReassocResp',
+ 4: 'ProbeReq',
+ 5: 'ProbeResp',
+ 6: 'TimingAdv',
+ 8: 'Beacon',
+ 9: 'ATIM',
+ 10: 'Disassoc',
+ 11: 'Auth',
+ 12: 'Deauth',
+ 13: 'Action',
+ 14: 'ActionNoAck',
+}
+
+
+def _frame_type_str(frame):
+ if len(frame) < 2:
+ return '?'
+
+ fc = frame[0]
+ ftype = (fc >> 2) & 0x3
+ subtype = (fc >> 4) & 0xf
+ if ftype == 0:
+ return _MGMT_SUBTYPES.get(subtype, 'Mgmt(%d)' % subtype)
+
+ return {
+ 1: 'Ctrl',
+ 2: 'Data',
+ 3: 'Ext'
+ }[ftype] + '(%d)' % subtype
+
+
+class HwsimRelay:
+ def __init__(self, intercept=None, rx_rate=1, signal=-30, radios=None):
+ """
+ By default every frame is passed through unmodified to every
+ radio other than the sender, but the intercept call can override
+ that.
+
+ intercept(src_addr: bytes, freq: int|None, frame: bytes) -> frame: bytes | None
+
+ The returned frame (or None to drop it) is sent to every radio
+ other than the sender - all receivers always see the same frame,
+ as they would over the air.
+
+ Used as "async with HwsimRelay() as relay:".
+ """
+ self._conn = netlink.Connection(netlink.NETLINK_GENERIC)
+ self._fid = netlink.genl_controller.get_family_id(b'MAC80211_HWSIM')
+ self._radios = set()
+ for addr in (radios or []):
+ self.add_radio(addr)
+ self._intercept_fn = intercept or self._default_intercept
+ self._rx_rate = rx_rate
+ self._signal = signal
+ self._running = False
+ self._exc = None
+ self._task = None
+
+ @property
+ def intercept(self):
+ return self._intercept_fn
+
+ @intercept.setter
+ def intercept(self, fn):
+ self._intercept_fn = fn or self._default_intercept
+
+ def add_radio(self, addr):
+ if isinstance(addr, str):
+ addr = bytes.fromhex(addr.replace(':', ''))
+ self._radios.add(addr)
+
+ def add_radio_for_iface(self, ifname):
+ self.add_radio(radio_addr_for_iface(ifname))
+
+ def _default_intercept(self, src, freq, frame):
+ return frame
+
+ def register(self):
+ msg = netlink.GenlMessage(self._fid, HWSIM_CMD_REGISTER,
+ flags=netlink.NLM_F_REQUEST |
+ netlink.NLM_F_ACK)
+ msg.send_and_recv(self._conn)
+
+ async def __aenter__(self):
+ self.register()
+ self._running = True
+ self._task = asyncio.current_task()
+ asyncio.get_running_loop().add_reader(self._conn.descriptor.fileno(),
+ self._on_readable)
+ return self
+
+ async def __aexit__(self, exc_type, exc_value, traceback):
+ asyncio.get_running_loop().remove_reader(self._conn.descriptor.fileno())
+ self._running = False
+ self._conn.descriptor.close()
+ if self._exc is not None:
+ exc, self._exc = self._exc, None
+ # replaces the CancelledError we injected below to unblock
+ # whatever the body was awaiting when the reader died
+ raise exc
+
+ def _recv_one(self):
+ msg = self._conn.recv()
+ if msg.type != self._fid or len(msg.payload) < 4:
+ return
+ cmd = msg.payload[0]
+ if cmd != HWSIM_CMD_FRAME:
+ logger.debug("hwsim_relay: non-FRAME cmd=%d" % (cmd,))
+ return
+ attrs = netlink.parse_attributes(msg.payload[4:])
+ self._handle_frame(attrs)
+
+ def _on_readable(self):
+ try:
+ self._recv_one()
+ except OSError as e:
+ # hwsim rejects frames for radios not started with EINVAL
+ if e.errno != errno.EINVAL:
+ logger.warning("hwsim_relay: dropping failed op: %r" % (e,))
+ except Exception as e:
+ self._exc = e
+ # wake up whatever the body is currently awaiting instead of
+ # waiting for it to time out or hang forever
+ if self._task is not None:
+ self._task.cancel()
+
+ def _handle_frame(self, attrs):
+ if HWSIM_ATTR_ADDR_TRANSMITTER not in attrs or \
+ HWSIM_ATTR_FRAME not in attrs or \
+ HWSIM_ATTR_COOKIE not in attrs:
+ return
+
+ src = attrs[HWSIM_ATTR_ADDR_TRANSMITTER].str()
+ frame = attrs[HWSIM_ATTR_FRAME].str()
+ freq = attrs[HWSIM_ATTR_FREQ].u32() if HWSIM_ATTR_FREQ in attrs else None
+ flags = attrs[HWSIM_ATTR_FLAGS].u32() if HWSIM_ATTR_FLAGS in attrs else 0
+ cookie = attrs[HWSIM_ATTR_COOKIE].str()
+
+ self._radios.add(src)
+ delivered = False
+ logger.debug("hwsim_relay: %s src=%s len=%d freq=%s flags=0x%x" %
+ (_frame_type_str(frame), src.hex(), len(frame), freq, flags))
+
+ try:
+ out_frame = self._intercept_fn(src, freq, frame)
+
+ if out_frame is not None:
+ for dst in tuple(self._radios):
+ if dst != src:
+ self._deliver(src, dst, out_frame, freq, cookie)
+ delivered = True
+ finally:
+ self._ack(src, flags, cookie, delivered)
+
+ def _deliver(self, src, dst, frame, freq, cookie):
+ attrs = [
+ netlink.StrAttr(HWSIM_ATTR_ADDR_TRANSMITTER, src),
+ netlink.StrAttr(HWSIM_ATTR_ADDR_RECEIVER, dst),
+ netlink.StrAttr(HWSIM_ATTR_FRAME, frame),
+ netlink.U32Attr(HWSIM_ATTR_RX_RATE, self._rx_rate),
+ netlink.U32Attr(HWSIM_ATTR_SIGNAL, self._signal & 0xffffffff),
+ netlink.Attr(HWSIM_ATTR_COOKIE, cookie),
+ ]
+ if freq is not None:
+ attrs.append(netlink.U32Attr(HWSIM_ATTR_FREQ, freq))
+ msg = netlink.GenlMessage(self._fid, HWSIM_CMD_FRAME, attrs=attrs,
+ flags=netlink.NLM_F_REQUEST)
+ msg.send(self._conn)
+
+ def _ack(self, src, flags, cookie, delivered):
+ tx_flags = HWSIM_TX_STAT_ACK if delivered and \
+ not (flags & HWSIM_TX_CTL_NO_ACK) else 0
+ attrs = [
+ netlink.StrAttr(HWSIM_ATTR_ADDR_TRANSMITTER, src),
+ netlink.U32Attr(HWSIM_ATTR_FLAGS, tx_flags),
+ netlink.Attr(HWSIM_ATTR_COOKIE, cookie),
+ netlink.U32Attr(HWSIM_ATTR_SIGNAL, self._signal & 0xffffffff),
+ netlink.StrAttr(HWSIM_ATTR_TX_INFO, _tx_info_all_invalid()),
+ ]
+ if not delivered:
+ attrs.append(netlink.FlagAttr(HWSIM_ATTR_NO_MONITOR))
+ msg = netlink.GenlMessage(self._fid, HWSIM_CMD_TX_INFO_FRAME,
+ attrs=attrs, flags=netlink.NLM_F_REQUEST)
+ msg.send(self._conn)
diff --git a/tests/hwsim/relay/bip.py b/tests/hwsim/relay/bip.py
new file mode 100644
index 000000000000..0e1395c01506
--- /dev/null
+++ b/tests/hwsim/relay/bip.py
@@ -0,0 +1,86 @@
+#
+# Copyright (C) 2026 Intel Corporation
+#
+# This software may be distributed under the terms of the BSD license.
+# See README for more details.
+
+try:
+ from Cryptodome.Cipher import AES
+ from Cryptodome.Hash import CMAC
+except ImportError:
+ from Crypto.Cipher import AES
+ from Crypto.Hash import CMAC
+
+from . import RelayLayer
+from .frame import MGMT_HDR_LEN, is_beacon, masked_fc_aad
+
+WLAN_EID_MMIE = 76
+
+
+def find_mmie(frame, mic_len=8):
+ mmie_len = 2 + 2 + 6 + mic_len
+ if len(frame) < MGMT_HDR_LEN + mmie_len:
+ return None
+ mmie = frame[-mmie_len:]
+ if mmie[0] != WLAN_EID_MMIE or mmie[1] != mmie_len - 2:
+ return None
+ return mmie
+
+def bip_aad(frame):
+ return masked_fc_aad(frame)
+
+def bip_cmac_mic(key, frame, mic_len=8):
+ aad = bip_aad(frame)
+ body = bytearray(frame[MGMT_HDR_LEN:])
+ if is_beacon(frame):
+ body[0:8] = b'\x00' * 8
+ body[-mic_len:] = b'\x00' * mic_len
+
+ c = CMAC.new(key, ciphermod=AES)
+ c.update(aad + bytes(body))
+ return c.digest()[:mic_len]
+
+
+def resign_frame(frame, key):
+ mmie = find_mmie(frame)
+ if mmie is None:
+ return frame
+ mic_len = len(mmie) - (2 + 2 + 6)
+ mic = bip_cmac_mic(key, frame, mic_len)
+ return frame[:-mic_len] + mic
+
+
+def group_key_getter(hapd):
+ def get_key(frame):
+ cmd = "GET_BIGTK" if is_beacon(frame) else "GET_IGTK"
+ res = hapd.request(cmd)
+ if res is None or "FAIL" in res:
+ return None
+ return bytes.fromhex(res.strip())
+ return get_key
+
+
+class GroupMgmtSigner(RelayLayer):
+ """
+ Recalculate the MIC for a group-addressed frame after the inner
+ layers have changed it.
+
+ get_key: callable get_key(frame) or hostapd instance
+ """
+
+ def __init__(self, upstream, get_key):
+ super().__init__(upstream)
+ self._get_key = get_key if callable(get_key) else \
+ group_key_getter(get_key)
+
+ def _run(self, src, freq, frame):
+ out_frame = self.intercept(src, freq, frame)
+ if out_frame is None or find_mmie(out_frame) is None:
+ return out_frame
+ return self._resign(frame, out_frame)
+
+ def _resign(self, old, frame):
+ if frame == old:
+ return frame
+ key = self._get_key(frame)
+ return resign_frame(frame, key) if key is not None else frame
diff --git a/tests/hwsim/relay/ccmp.py b/tests/hwsim/relay/ccmp.py
new file mode 100644
index 000000000000..3f666178291d
--- /dev/null
+++ b/tests/hwsim/relay/ccmp.py
@@ -0,0 +1,97 @@
+#
+# Copyright (C) 2026 Intel Corporation
+#
+# This software may be distributed under the terms of the BSD license.
+# See README for more details.
+#
+
+try:
+ from Cryptodome.Cipher import AES
+except ImportError:
+ from Crypto.Cipher import AES
+
+from . import RelayLayer
+from .frame import (IEEE80211_FCTL_PROTECTED, MGMT_HDR_LEN, is_protected_mgmt,
+ masked_fc_aad, client_addr)
+
+CCMP_HDR_LEN = 8
+
+
+def _mgmt_aad_nonce(frame):
+ aad = masked_fc_aad(frame, extra_mask=IEEE80211_FCTL_PROTECTED)
+ aad += bytes([frame[22] & 0x0f, 0])
+
+ ccmp_hdr = frame[MGMT_HDR_LEN:MGMT_HDR_LEN + CCMP_HDR_LEN]
+ pn = bytes([ccmp_hdr[7], ccmp_hdr[6], ccmp_hdr[5], ccmp_hdr[4],
+ ccmp_hdr[1], ccmp_hdr[0]])
+ nonce = bytes([0x10]) + frame[10:16] + pn
+ return aad, nonce
+
+
+def ccmp_decrypt_mgmt(frame, tk, mic_len=8):
+ aad, nonce = _mgmt_aad_nonce(frame)
+
+ body_start = MGMT_HDR_LEN + CCMP_HDR_LEN
+ ciphertext = frame[body_start:len(frame) - mic_len]
+ mic = frame[len(frame) - mic_len:]
+
+ cipher = AES.new(tk, AES.MODE_CCM, nonce=nonce, mac_len=mic_len)
+ cipher.update(aad)
+ return cipher.decrypt_and_verify(ciphertext, mic)
+
+
+def ccmp_encrypt_mgmt(frame, tk, body, mic_len=8):
+ aad, nonce = _mgmt_aad_nonce(frame)
+
+ cipher = AES.new(tk, AES.MODE_CCM, nonce=nonce, mac_len=mic_len)
+ cipher.update(aad)
+ ciphertext, mic = cipher.encrypt_and_digest(body)
+
+ body_start = MGMT_HDR_LEN + CCMP_HDR_LEN
+ return frame[:body_start] + ciphertext + mic
+
+
+def sta_tk_getter(hapd, cipher="CCMP"):
+ def get_tk(addr):
+ addr_str = addr.hex(':')
+
+ # might be a link address, try to resolve it
+ sta = hapd.get_link_sta(addr_str)
+ resolved = sta.get('addr')
+ if resolved and resolved != 'FAIL':
+ addr_str = resolved
+
+ ptksa = hapd.get_ptksa(addr_str, cipher)
+ if ptksa is None:
+ return None
+ return bytes.fromhex(ptksa['tk'])
+ return get_tk
+
+
+class ProtectedMgmtCrypto(RelayLayer):
+ """
+ Decrypt CCMP-protected management frames and pass them to
+ the next layer unencrypted (but protected bit still set)
+
+ get_tk: callable get_tk(client_addr) or hostapd instance
+ """
+ def __init__(self, upstream, get_tk):
+ super().__init__(upstream)
+ self._get_tk = get_tk if callable(get_tk) else sta_tk_getter(get_tk)
+
+ def _run(self, src, freq, frame):
+ if is_protected_mgmt(frame):
+ tk = self._get_tk(client_addr(frame))
+ if tk is not None:
+ try:
+ body = ccmp_decrypt_mgmt(frame, tk)
+ except ValueError:
+ body = None
+ if body is not None:
+ clear_frame = frame[:MGMT_HDR_LEN] + body
+ out_frame = self.intercept(src, freq, clear_frame)
+ if out_frame is not None and is_protected_mgmt(out_frame):
+ return ccmp_encrypt_mgmt(frame, tk,
+ out_frame[MGMT_HDR_LEN:])
+ return out_frame
+ return self.intercept(src, freq, frame)
diff --git a/tests/hwsim/relay/frame.py b/tests/hwsim/relay/frame.py
new file mode 100644
index 000000000000..36c406121f68
--- /dev/null
+++ b/tests/hwsim/relay/frame.py
@@ -0,0 +1,119 @@
+#
+# 802.11 frame helpers for frame relay handling
+#
+# Copyright (C) 2026 Intel Corporation
+#
+# This software may be distributed under the terms of the BSD license.
+# See README for more details.
+#
+
+import struct
+
+IEEE80211_FCTL_RETRY = 0x0800
+IEEE80211_FCTL_PM = 0x1000
+IEEE80211_FCTL_MOREDATA = 0x2000
+IEEE80211_FCTL_PROTECTED = 0x4000
+IEEE80211_FCTL_FTYPE = 0x000c
+IEEE80211_FTYPE_MGMT = 0x0000
+
+STYPE_ASSOC_REQ = 0
+STYPE_ASSOC_RESP = 1
+STYPE_REASSOC_RESP = 3
+STYPE_PROBE_RESP = 5
+STYPE_BEACON = 8
+STYPE_DEAUTH = 12
+
+MGMT_HDR_LEN = 24
+
+WLAN_EID_EXTENSION = 255
+
+
+def frame_fc(frame):
+ return struct.unpack_from('<H', frame, 0)[0]
+
+
+def is_ftype_mgmt(frame):
+ return (frame_fc(frame) & IEEE80211_FCTL_FTYPE) == IEEE80211_FTYPE_MGMT
+
+
+def mgmt_subtype(frame):
+ return (frame_fc(frame) >> 4) & 0xf
+
+
+def is_mgmt_subtype(frame, stype):
+ return is_ftype_mgmt(frame) and mgmt_subtype(frame) == stype
+
+
+def is_beacon(frame):
+ return is_mgmt_subtype(frame, STYPE_BEACON)
+
+
+def is_protected_mgmt(frame):
+ if len(frame) < MGMT_HDR_LEN:
+ return False
+ return is_ftype_mgmt(frame) and bool(frame_fc(frame) & IEEE80211_FCTL_PROTECTED)
+
+
+def is_protected_deauth(frame):
+ return is_protected_mgmt(frame) and is_mgmt_subtype(frame, STYPE_DEAUTH)
+
+
+def ies_start_offset(frame):
+ """Byte offset of the IE section for beacon/probe-resp/assoc-resp."""
+ stype = mgmt_subtype(frame)
+ if stype in (STYPE_BEACON, STYPE_PROBE_RESP):
+ return 24 + 12 # timestamp(8) + beacon_interval(2) + capab(2)
+ if stype in (STYPE_ASSOC_RESP, STYPE_REASSOC_RESP):
+ return 24 + 6 # capab(2) + status(2) + aid(2)
+ raise ValueError("unsupported mgmt subtype %d for IE parsing" % stype)
+
+
+def iter_elements(buf, start=0):
+ """Yield (eid, ext_id_or_None, elem_start, payload_off, payload_len)."""
+ i = start
+ while i + 2 <= len(buf):
+ eid = buf[i]
+ elen = buf[i + 1]
+ if i + 2 + elen > len(buf):
+ break
+ payload_off = i + 2
+ payload_len = elen
+ ext_id = None
+ if eid == WLAN_EID_EXTENSION and elen >= 1:
+ ext_id = buf[payload_off]
+ payload_off += 1
+ payload_len -= 1
+ yield eid, ext_id, i, payload_off, payload_len
+ i += 2 + elen
+
+
+def find_element(buf, start, eid, ext_id=None):
+ for e, x, elem_start, payload_off, payload_len in iter_elements(buf, start):
+ if e == eid and x == ext_id:
+ elem_total_len = (payload_off + payload_len) - elem_start
+ return elem_start, elem_total_len, payload_off, payload_len
+ return None
+
+
+def build_element(payload, eid=None, ext_id=None):
+ if ext_id is None:
+ return struct.pack('BB', eid, len(payload)) + payload
+ return struct.pack('BBB', WLAN_EID_EXTENSION, 1 + len(payload), ext_id) + payload
+
+
+def client_addr(frame):
+ addr1, addr2, bssid = frame[4:10], frame[10:16], frame[16:22]
+ return addr1 if addr2 == bssid else addr2
+
+
+def masked_fc_aad(frame, extra_mask=0):
+ fc = frame_fc(frame)
+ masked_fc = (fc & ~(IEEE80211_FCTL_RETRY | IEEE80211_FCTL_PM |
+ IEEE80211_FCTL_MOREDATA)) | extra_mask
+ return struct.pack('<H', masked_fc) + frame[4:22]
+
+
+class ProtectedFrame:
+ @classmethod
+ def match(cls, src, freq, frame):
+ return frame if is_protected_mgmt(frame) else None
diff --git a/tests/hwsim/test_hwsim_relay.py b/tests/hwsim/test_hwsim_relay.py
new file mode 100644
index 000000000000..f9a4fd73d5e5
--- /dev/null
+++ b/tests/hwsim/test_hwsim_relay.py
@@ -0,0 +1,146 @@
+#
+# Trivial test for the hwsim relay
+#
+# Copyright (C) 2026 Intel Corporation
+#
+# This software may be distributed under the terms of the BSD license.
+# See README for more details.
+
+import hostapd
+import hwsim_utils
+from relay import HwsimRelay, WaitFrame, radio_addr_for_iface
+from relay import bip, ccmp
+from relay import frame as frame_mod
+
+async def test_hwsim_relay_passthrough(dev, apdev):
+ """
+ Simple test for HwsimRelay() without interceptor function.
+ """
+ async with HwsimRelay() as relay:
+ relay.add_radio_for_iface(apdev[0]['ifname'])
+ relay.add_radio_for_iface(dev[0].ifname)
+
+ hapd = await hostapd.add_ap_async(apdev[0], {"ssid": "open"})
+ await dev[0].connect_async("open", key_mgmt="NONE", scan_freq="2412",
+ bg_scan_period="0")
+ ev = await hapd.wait_event_async(["AP-STA-CONNECTED"], timeout=5)
+ if ev is None:
+ raise Exception("STA didn't connect")
+ await hwsim_utils.test_connectivity_async(dev[0], hapd)
+
+async def test_hwsim_relay_bip_verify(dev, apdev):
+ """
+ Check python CMAC/CCMP calculations against mac80211
+ """
+ hapd = None
+
+ async with HwsimRelay() as relay:
+ relay.add_radio_for_iface(apdev[0]['ifname'])
+ relay.add_radio_for_iface(dev[0].ifname)
+
+ def get_tk(addr):
+ if hapd is None:
+ return None
+ ptksa = hapd.get_ptksa(addr.hex(':'), "CCMP")
+ if ptksa is None:
+ return None
+ return bytes.fromhex(ptksa['tk'])
+
+ class CheckBip:
+ @staticmethod
+ def match(src, freq, frame):
+ mmie = bip.find_mmie(frame)
+ if mmie is None or not frame_mod.is_beacon(frame) or hapd is None:
+ return None
+ key = bytes.fromhex(hapd.request("GET_BIGTK"))
+ mic = bip.bip_cmac_mic(key, frame)
+ assert mic == mmie[-8:], \
+ f"MIC mismatch: {mic.hex()} != {bytes(mmie[-8:]).hex()}"
+ return frame
+
+ with ccmp.ProtectedMgmtCrypto(relay, get_tk) as pmc, \
+ WaitFrame(pmc, [CheckBip]) as bip_waiter, \
+ WaitFrame(bip_waiter, [frame_mod.ProtectedFrame]) as prot_waiter:
+
+ params = hostapd.wpa2_params(ssid="test-beacon-prot",
+ passphrase="12345678")
+ params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
+ params["ieee80211w"] = "2"
+ params["beacon_prot"] = "1"
+ params["group_mgmt_cipher"] = "AES-128-CMAC"
+ hapd = await hostapd.add_ap_async(apdev[0], params)
+ await dev[0].connect_async("test-beacon-prot", psk="12345678",
+ ieee80211w="2", beacon_prot="1",
+ key_mgmt="WPA-PSK-SHA256", proto="WPA2",
+ scan_freq="2412")
+
+ assert await bip_waiter.wait_for_frame(), "No protected beacon found"
+
+ # disconnect to check CCMP with the encrypted deauth
+ dev[0].request("DISCONNECT")
+ assert await prot_waiter.wait_for_frame(), \
+ "No protected management frame found"
+ assert await dev[0].wait_event_async(["CTRL-EVENT-DISCONNECTED"],
+ timeout=5), \
+ "STA did not disconnect"
+
+async def test_hwsim_relay_drop_frame(dev, apdev):
+ """
+ validate relay frame dropping
+ """
+ dropped = []
+ sta_radio = radio_addr_for_iface(dev[0].ifname)
+
+ def intercept(src, freq, frame):
+ if src == sta_radio and \
+ frame_mod.is_mgmt_subtype(frame, frame_mod.STYPE_DEAUTH):
+ dropped.append(frame)
+ return None
+ return frame
+
+ async with HwsimRelay(intercept=intercept) as relay:
+ relay.add_radio_for_iface(apdev[0]['ifname'])
+ relay.add_radio_for_iface(dev[0].ifname)
+
+ hapd = await hostapd.add_ap_async(apdev[0], {"ssid": "open"})
+ await dev[0].connect_async("open", key_mgmt="NONE", scan_freq="2412",
+ bg_scan_period="0")
+ ev = await hapd.wait_event_async(["AP-STA-CONNECTED"], timeout=5)
+ if ev is None:
+ raise Exception("STA didn't connect")
+ await hwsim_utils.test_connectivity_async(dev[0], hapd)
+
+ dev[0].request("DISCONNECT")
+ ev = await dev[0].wait_event_async(["CTRL-EVENT-DISCONNECTED"],
+ timeout=5)
+ assert ev, "STA did not disconnect"
+
+ ev = await hapd.wait_event_async(["AP-STA-DISCONNECTED"], timeout=1)
+ assert ev is None, "AP received the dropped deauth"
+
+ assert dropped, "no deauth frame dropped"
+
+async def test_hwsim_relay_intercept_exception(dev, apdev):
+ """
+ intercept callbacks can raise exceptions and that bubbles out
+ """
+ class InjectedError(Exception):
+ pass
+
+ def intercept(src, freq, frame):
+ if frame_mod.is_mgmt_subtype(frame, frame_mod.STYPE_ASSOC_REQ):
+ raise InjectedError("boom")
+ return frame
+
+ try:
+ async with HwsimRelay(intercept=intercept) as relay:
+ relay.add_radio_for_iface(apdev[0]['ifname'])
+ relay.add_radio_for_iface(dev[0].ifname)
+
+ hapd = await hostapd.add_ap_async(apdev[0], {"ssid": "open"})
+ await dev[0].connect_async("open", key_mgmt="NONE",
+ scan_freq="2412", bg_scan_period="0")
+ except InjectedError as e:
+ pass
+ else:
+ assert False, "intercept exception was not propagated"
--
2.55.0
More information about the Hostap
mailing list