@@ -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
new file mode 100644
@@ -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)
new file mode 100644
@@ -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
new file mode 100644
@@ -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)
new file mode 100644
@@ -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
new file mode 100644
@@ -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"
@@ -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: