From: Benjamin Berg <benjamin.berg@xxxxxxxxx> This adds a test that neighbor entries with ML element work and will apply to the the entire MLD AP instead of just the BSS that was listed explicitly. Signed-off-by: Benjamin Berg <benjamin.berg@xxxxxxxxx> Reviewed-by: Andrei Otcheretianski <andrei.otcheretianski@xxxxxxxxx> --- tests/hwsim/hostapd.py | 18 ++++++++ tests/hwsim/test_eht.py | 95 +++++++++++++++++++++++++++++++++++++++++ wpaspy/wpaspy.py | 4 ++ 3 files changed, 117 insertions(+) diff --git a/tests/hwsim/hostapd.py b/tests/hwsim/hostapd.py index 9cfe5a2b61..163de0b2a3 100644 --- a/tests/hwsim/hostapd.py +++ b/tests/hwsim/hostapd.py @@ -15,6 +15,7 @@ import wpaspy import remotehost import utils import subprocess +import select from remotectrl import RemoteCtrl logger = logging.getLogger() @@ -658,6 +659,23 @@ class Hostapd: return vals return None +def mld_wait_event(hapds, events, timeout): + start = os.times()[4] + while True: + for hapd in hapds: + ev = hapd.wait_event(events, 0) + if ev: + return ev + + now = os.times()[4] + remaining = start + timeout - now + if remaining <= 0: + break + socks = [hapd.mon.socket for hapd in hapds] + r, w, e = select.select(socks, [], [], remaining) + if not r: + break + def add_ap(apdev, params, wait_enabled=True, no_enable=False, timeout=30, global_ctrl_override=None, driver=False, set_channel=True): if isinstance(apdev, dict): diff --git a/tests/hwsim/test_eht.py b/tests/hwsim/test_eht.py index aa11765a6a..e3ce7da149 100644 --- a/tests/hwsim/test_eht.py +++ b/tests/hwsim/test_eht.py @@ -1249,6 +1249,101 @@ def test_eht_mld_bss_trans_mgmt_link_removal_imminent(dev, apdev): if ev is not None: raise Exception("Unexpected action on STA: " + ev) +def test_eht_mld_bss_trans_mgmt_ml_neighbor(dev, apdev): + """EHT MLD with two links. BSS transition management with ML neighbor""" + + with HWSimRadio(use_mlo=True) as (hapd0_radio, hapd0_iface), \ + HWSimRadio(use_mlo=True) as (hapd1_radio, hapd1_iface), \ + HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): + + # This test does a HT scan on an MLD. Check kernel scanning support + buf = subprocess.Popen(['iw', 'phy', f'phy{hapd1_radio}', 'info'], + stdout=subprocess.PIPE).communicate()[0] + if b'Device supports AP scan.' not in buf: + raise HwsimSkip('Kernel does not support AP scanning') + + try: + wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') + wpas.interface_add(wpas_iface) + + ssid = "mld_ap_owe" + params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") + + # Enable BSS transition management support + params['mbo'] = '1' + + # Start first MLD AP + hapd0_0 = eht_mld_enable_ap(hapd0_iface, 0, params) + params['channel'] = '6' + hapd0_1 = eht_mld_enable_ap(hapd0_iface, 1, params) + + # Start second MLD AP, with 40MHz bw to make it more interesting + params_5ghz = eht_5ghz_params(36, 0, 38, 0) + params.update(params_5ghz) + hapd1_0 = eht_mld_enable_ap(hapd1_iface, 0, params) + + params_5ghz = eht_5ghz_params(149, 1, 155, 0) + params.update(params_5ghz) + hapd1_1 = eht_mld_enable_ap(hapd1_iface, 1, params) + + # Give APs some time to settle + time.sleep(1) + + # Connect to and make sure we are on hapd0 + wpas.connect(ssid, key_mgmt="OWE", ieee80211w="2") + wpas.roam(hapd0_0.own_addr(), check_bssid=False) + + eht_verify_status(wpas, hapd0_0, 2412, 20, is_ht=True, mld=True, + valid_links=3, active_links=3) + + # Neighbor for hapd_1_0 with preference 0x00 and for the entire MLD + neighbor = hapd1_0.own_addr() + ',0x0000,' + '116,36,9,030100' + # ML neighbor report, control, common info with MLD addr + neighbor += 'c909000007' + hapd1_0.own_mld_addr().replace(':', '') + + hapd0_0.request("BSS_TM_REQ " + wpas.own_addr() + ' pref=1 abridged=0 disassoc_imminent=1 valid_int=255 neighbor=' + neighbor) + + # The station may respond on either link + ev = hostapd.mld_wait_event((hapd0_0, hapd0_1), ['BSS-TM-RESP'], timeout=20) + if ev is None: + raise Exception("No BSS Transition Management Response") + # Should be accepted, but the client has no where to roam + if 'status_code=0' not in ev or 'target_bssid=00:00:00:00:00:00' not in ev: + raise Exception("Unexpected BSS TM response status: " + ev) + + # Give it a bit of time, we shouldn't roam as the other AP is forbidden + time.sleep(5) + + # Didn't connect to the other AP + assert wpas.get_status()['ap_mld_addr'] == hapd0_0.own_mld_addr(), \ + 'STA should still be connected to same AP' + + eht_verify_status(wpas, hapd0_0, 2412, 20, is_ht=True, mld=True, + valid_links=3, active_links=3) + + # Similar, but now ask to roam to the better AP + neighbor = hapd1_0.own_addr() + ',0x0000,' + '116,36,9,0301ff' + neighbor += 'c909000007' + hapd1_0.own_mld_addr().replace(':', '') + + hapd0_0.request("BSS_TM_REQ " + wpas.own_addr() + ' pref=1 abridged=1 disassoc_imminent=1 valid_int=255 neighbor=' + neighbor) + + # The station may respond on either link + ev = hostapd.mld_wait_event((hapd0_0, hapd0_1), ['BSS-TM-RESP'], timeout=20) + if ev is None: + raise Exception("No BSS Transition Management Response") + # Client accepts and roams to one the second AP + if 'status_code=0' not in ev: + raise Exception("Unexpected BSS TM response status: " + ev) + + wpas.wait_connected(timeout=2) + + eht_verify_status(wpas, hapd1_0, 5, 40, is_ht=True, is_vht=True, + mld=True, valid_links=3, active_links=3) + + finally: + wpas.cmd_execute(['iw', 'reg', 'set', '00']) + wait_regdom_changes(wpas) + def send_check(hapd, frame, no_tx_status=False): cmd = "MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=" hapd.request(cmd + frame) diff --git a/wpaspy/wpaspy.py b/wpaspy/wpaspy.py index 5b8140b7c9..46862edc95 100644 --- a/wpaspy/wpaspy.py +++ b/wpaspy/wpaspy.py @@ -63,6 +63,10 @@ class Ctrl: raise self.started = True + @property + def socket(self): + return self.s + def __del__(self): self.close() -- 2.50.0 _______________________________________________ Hostap mailing list Hostap@xxxxxxxxxxxxxxxxxxx http://lists.infradead.org/mailman/listinfo/hostap