[PATCH 3/3] tests: hwsim: add hwsim relay code
Johannes Berg
johannes at sipsolutions.net
Fri Sep 11 07:21:58 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.
Note that the involvement of a background thread which is doing
control interface commands now requires a lock in wpaspy.
Signed-off-by: Johannes Berg <johannes.berg at intel.com>
---
tests/hwsim/hostapd.py | 7 +
tests/hwsim/relay/__init__.py | 309 ++++++++++++++++++++++++++++++++
tests/hwsim/relay/bip.py | 86 +++++++++
tests/hwsim/relay/ccmp.py | 97 ++++++++++
tests/hwsim/relay/frame.py | 118 ++++++++++++
tests/hwsim/test_hwsim_relay.py | 110 ++++++++++++
wpaspy/wpaspy.py | 29 +--
7 files changed, 743 insertions(+), 13 deletions(-)
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 9cfe5a2b61d0..41f7c7db96f6 100644
--- a/tests/hwsim/hostapd.py
+++ b/tests/hwsim/hostapd.py
@@ -443,6 +443,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..3d970d369829
--- /dev/null
+++ b/tests/hwsim/relay/__init__.py
@@ -0,0 +1,309 @@
+#
+# 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 threading
+import queue
+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.
+
+ Note: the intercept is run on a background thread
+ """
+
+ 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 = queue.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(result)
+ break
+ return self.intercept(src, freq, frame)
+
+ def wait_for_frame(self, timeout=5):
+ try:
+ return self._results.get(timeout=timeout)
+ except queue.Empty:
+ 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.
+
+ Note: intercept calls are on a background thread
+ """
+ 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._thread = None
+ self._exc = 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)
+
+ def start(self):
+ self.register()
+ self._running = True
+ self._thread = threading.Thread(target=self._loop, daemon=True)
+ self._thread.start()
+
+ def stop(self):
+ self._running = False
+ self._conn.descriptor.close()
+ if self._thread:
+ self._thread.join(timeout=5)
+ self._thread = None
+
+ def __enter__(self):
+ self.start()
+ return self
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ self.stop()
+ # raise thread exceptions into the test
+ if self._exc is not None:
+ exc, self._exc = self._exc, None
+ raise exc
+
+ def _loop(self):
+ while self._running:
+ try:
+ msg = self._conn.recv()
+ if msg.type != self._fid or len(msg.payload) < 4:
+ continue
+ cmd = msg.payload[0]
+ if cmd != HWSIM_CMD_FRAME:
+ logger.debug("hwsim_relay: non-FRAME cmd=%d" % (cmd,))
+ continue
+ attrs = netlink.parse_attributes(msg.payload[4:])
+ self._handle_frame(attrs)
+ except OSError as e:
+ # stop() closes the fd to stop the thread, ignore errors
+ if not self._running:
+ break
+ # 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:
+ logger.error("hwsim_relay: exception in thread: %s" % (e,))
+ # record exception for __exit__
+ self._exc = e
+ break
+
+ 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)
+ out_frame = self._intercept_fn(src, freq, frame)
+ logger.debug("hwsim_relay: %s src=%s len=%d freq=%s flags=0x%x" %
+ (_frame_type_str(frame), src.hex(), len(frame), freq, flags))
+
+ delivered = False
+ 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
+
+ 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..27d9a81ff254
--- /dev/null
+++ b/tests/hwsim/relay/frame.py
@@ -0,0 +1,118 @@
+#
+# 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_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..53937b06105b
--- /dev/null
+++ b/tests/hwsim/test_hwsim_relay.py
@@ -0,0 +1,110 @@
+#
+# 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
+from test_ap_pmf import run_ap_pmf_beacon_protection
+
+def test_hwsim_relay_passthrough(dev, apdev):
+ """
+ Simple test for HwsimRelay() without interceptor function.
+ """
+ with HwsimRelay() as relay:
+ relay.add_radio_for_iface(apdev[0]['ifname'])
+ relay.add_radio_for_iface(dev[0].ifname)
+
+ hapd = hostapd.add_ap(apdev[0], {"ssid": "open"})
+ dev[0].connect("open", key_mgmt="NONE", scan_freq="2412",
+ bg_scan_period="0")
+ ev = hapd.wait_event(["AP-STA-CONNECTED"], timeout=5)
+ if ev is None:
+ raise Exception("STA didn't connect")
+ hwsim_utils.test_connectivity(dev[0], hapd)
+
+def test_hwsim_relay_bip_verify(dev, apdev):
+ """
+ Check python CMAC/CCMP calculations against mac80211
+ """
+ hapd = None
+
+ with HwsimRelay() as relay:
+ relay.add_radio_for_iface(apdev[0]['ifname'])
+ relay.add_radio_for_iface(dev[0].ifname)
+ relay.add_radio_for_iface(dev[1].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:
+
+ hapd = run_ap_pmf_beacon_protection(dev, apdev, "AES-128-CMAC")
+
+ assert bip_waiter.wait_for_frame(), "No protected beacon found"
+
+ # disconnect to check CCMP with the encrypted deauth
+ dev[1].request("DISCONNECT")
+ assert prot_waiter.wait_for_frame(), "No protected management frame found"
+ assert dev[1].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5), \
+ "STA did not disconnect"
+
+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
+
+ with HwsimRelay(intercept=intercept) as relay:
+ relay.add_radio_for_iface(apdev[0]['ifname'])
+ relay.add_radio_for_iface(dev[0].ifname)
+
+ hapd = hostapd.add_ap(apdev[0], {"ssid": "open"})
+ dev[0].connect("open", key_mgmt="NONE", scan_freq="2412",
+ bg_scan_period="0")
+ ev = hapd.wait_event(["AP-STA-CONNECTED"], timeout=5)
+ if ev is None:
+ raise Exception("STA didn't connect")
+ hwsim_utils.test_connectivity(dev[0], hapd)
+
+ dev[0].request("DISCONNECT")
+ assert dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5), \
+ "STA did not disconnect"
+
+ assert hapd.wait_event(["AP-STA-DISCONNECTED"], timeout=1) is None, \
+ "AP received the dropped deauth"
+
+ assert dropped, "no deauth frame dropped"
diff --git a/wpaspy/wpaspy.py b/wpaspy/wpaspy.py
index 5b8140b7c99f..d193d8d86a97 100644
--- a/wpaspy/wpaspy.py
+++ b/wpaspy/wpaspy.py
@@ -10,6 +10,7 @@ import os
import stat
import socket
import select
+import threading
counter = 0
@@ -20,6 +21,7 @@ class Ctrl:
self.attached = False
self.path = path
self.port = port
+ self._lock = threading.Lock()
self.udp = False
if not path.startswith('/'):
@@ -87,19 +89,20 @@ class Ctrl:
cmd = cmd2
except UnicodeDecodeError as e:
pass
- if self.udp:
- self.s.sendto(self.cookie + cmd, self.sockaddr)
- else:
- self.s.send(cmd)
- [r, w, e] = select.select([self.s], [], [], timeout)
- if r:
- res = self.s.recv(4096).decode()
- try:
- r = str(res)
- except UnicodeDecodeError as e:
- r = res
- return r
- raise Exception("Timeout on waiting response")
+ with self._lock:
+ if self.udp:
+ self.s.sendto(self.cookie + cmd, self.sockaddr)
+ else:
+ self.s.send(cmd)
+ [r, w, e] = select.select([self.s], [], [], timeout)
+ if r:
+ res = self.s.recv(4096).decode()
+ try:
+ r = str(res)
+ except UnicodeDecodeError as e:
+ r = res
+ return r
+ raise Exception("Timeout on waiting response")
def attach(self):
if self.attached:
--
2.55.0
More information about the Hostap
mailing list