# EHT tests # Copyright (c) 2022-2024, Qualcomm Innovation Center, Inc. # # This software may be distributed under the terms of the BSD license. # See README for more details. import binascii import tempfile import hostapd from utils import * from hwsim import HWSimRadio import hwsim_utils from wpasupplicant import WpaSupplicant import re from tshark import run_tshark from test_gas import hs20_ap_params from test_dpp import check_dpp_capab, wait_auth_success from test_rrm import build_beacon_request, run_req_beacon, BeaconReport def eht_verify_wifi_version(dev): status = dev.get_status() logger.info("station status: " + str(status)) if 'wifi_generation' not in status: raise Exception("Missing wifi_generation information") if status['wifi_generation'] != "7": raise Exception("Unexpected wifi_generation value: " + status['wifi_generation']) def _eht_get_links_bitmap(wpas, name): vfile = "/sys/kernel/debug/ieee80211/%s/netdev:%s/%s" % \ (wpas.get_driver_status_field("phyname"), wpas.ifname, name) if wpas.cmd_execute(["ls", vfile])[0] != 0: logger_info("%s not supported in mac80211: %s" % (name, vfile)) return 0 res, out = wpas.cmd_execute(["cat", vfile], shell=True) if res != 0: raise Exception("Failed to read %s" % fname) logger.info("%s=%s" % (name, out)) return int(out, 16) def _eht_valid_links(wpas): return _eht_get_links_bitmap(wpas, "valid_links") def _eht_active_links(wpas): return _eht_get_links_bitmap(wpas, "active_links") def _eht_dormant_links(wpas): return _eht_get_links_bitmap(wpas, "dormant_links") def _eht_verify_links(wpas, valid_links=0, active_links=0): vlinks = _eht_valid_links(wpas) if vlinks != valid_links: raise Exception("Unexpected valid links (0x%04x != 0x%04x)" % (vlinks, valid_links)) alinks = _eht_active_links(wpas) if alinks != active_links: raise Exception("Unexpected active links (0x%04x != 0x%04x)" % (alinks, active_links)) def eht_verify_status(wpas, hapd, freq, bw, is_ht=False, is_vht=False, mld=False, valid_links=0, active_links=0): status = hapd.get_status() logger.info("hostapd STATUS: " + str(status)) if is_ht and status["ieee80211n"] != "1": raise Exception("Unexpected STATUS ieee80211n value") if is_vht and status["ieee80211ac"] != "1": raise Exception("Unexpected STATUS ieee80211ac value") if status["ieee80211ax"] != "1": raise Exception("Unexpected STATUS ieee80211ax value") if status["ieee80211be"] != "1": raise Exception("Unexpected STATUS ieee80211be value") sta = hapd.get_sta(wpas.own_addr()) logger.info("hostapd STA: " + str(sta)) if sta['addr'] == 'FAIL': raise Exception("hostapd " + hapd.ifname + " did not have a STA entry for the STA " + wpas.own_addr()) if is_ht and "[HT]" not in sta['flags']: raise Exception("Missing STA flag: HT") if is_vht and "[VHT]" not in sta['flags']: raise Exception("Missing STA flag: VHT") if "[HE]" not in sta['flags']: raise Exception("Missing STA flag: HE") if "[EHT]" not in sta['flags']: raise Exception("Missing STA flag: EHT") sig = wpas.request("SIGNAL_POLL").splitlines() # TODO: With MLD connection, signal poll logic is still not implemented. # While mac80211 maintains the station using the MLD address, the # information is maintained in the link stations, but it is not sent to # user space yet. if not mld: if "FREQUENCY=%s" % freq not in sig: raise Exception("Unexpected SIGNAL_POLL value(1): " + str(sig)) if "WIDTH=%s MHz" % bw not in sig: raise Exception("Unexpected SIGNAL_POLL value(2): " + str(sig)) # Active links are updated in async work after the connection. # Sleep a bit to allow it to run. time.sleep(0.1) _eht_verify_links(wpas, valid_links, active_links) def traffic_test(wpas, hapd, success=True, ifname2=None): hwsim_utils.test_connectivity(wpas, hapd, success_expected=success, ifname2=ifname2) def test_eht_open(dev, apdev): """EHT AP with open mode configuration""" params = {"ssid": "eht", "ieee80211ax": "1", "ieee80211be": "1"} try: hapd = hostapd.add_ap(apdev[0], params) except Exception as e: if isinstance(e, Exception) and \ str(e) == "Failed to set hostapd parameter ieee80211be": raise HwsimSkip("EHT not supported") raise if hapd.get_status_field("ieee80211be") != "1": raise Exception("AP STATUS did not indicate ieee80211be=1") dev[0].connect("eht", key_mgmt="NONE", scan_freq="2412") sta = hapd.get_sta(dev[0].own_addr()) if "[EHT]" not in sta['flags']: raise Exception("Missing STA flag: EHT") status = dev[0].request("STATUS") if "wifi_generation=7" not in status: raise Exception("STA STATUS did not indicate wifi_generation=7") def test_prefer_eht_20(dev, apdev): """EHT AP on a 20 MHz channel""" params = {"ssid": "eht", "channel": "1", "ieee80211ax": "1", "ieee80211be" : "1", "ieee80211n": "1"} try: hapd0 = hostapd.add_ap(apdev[0], params) params["ieee80211be"] = "0" hapd1 = hostapd.add_ap(apdev[1], params) except Exception as e: if isinstance(e, Exception) and \ str(e) == "Failed to set hostapd parameter ieee80211be": raise HwsimSkip("EHT not supported") raise dev[0].connect("eht", key_mgmt="NONE") if dev[0].get_status_field('bssid') != apdev[0]['bssid']: raise Exception("dev[0] connected to unexpected AP") est = dev[0].get_bss(apdev[0]['bssid'])['est_throughput'] if est != "172103": raise Exception("Unexpected BSS1 est_throughput: " + est) def start_eht_sae_ap(apdev, ml=False, transition_mode=False): params = hostapd.wpa2_params(ssid="eht", passphrase="12345678") params["ieee80211ax"] = "1" params["ieee80211be"] = "1" params['ieee80211w'] = '1' if transition_mode else '2' params['rsn_pairwise'] = "CCMP GCMP-256" if transition_mode else "GCMP-256" params['group_cipher'] = "CCMP" if transition_mode else "GCMP-256" params["group_mgmt_cipher"] = "AES-128-CMAC" if transition_mode else "BIP-GMAC-256" params['beacon_prot'] = '1' params['wpa_key_mgmt'] = "SAE SAE-EXT-KEY WPA-PSK WPA-PSK-SHA256" if transition_mode else 'SAE-EXT-KEY' params['sae_groups'] = "19 20" if transition_mode else "20" params['sae_pwe'] = "2" if transition_mode else "1" if ml: ml_elem = "ff0d6b" + "3001" + "0a" + "021122334455" + "01" + "00" + "00" params['vendor_elements'] = ml_elem try: hapd = hostapd.add_ap(apdev, params) except Exception as e: if isinstance(e, Exception) and \ str(e) == "Failed to set hostapd parameter ieee80211be": raise HwsimSkip("EHT not supported") raise def test_eht_sae(dev, apdev): """EHT AP with SAE""" check_sae_capab(dev[0]) hapd = start_eht_sae_ap(apdev[0]) try: dev[0].set("sae_groups", "20") dev[0].set("sae_pwe", "2") dev[0].connect("eht", key_mgmt="SAE-EXT-KEY", psk="12345678", ieee80211w="2", beacon_prot="1", pairwise="GCMP-256", group="GCMP-256", group_mgmt="BIP-GMAC-256", scan_freq="2412") finally: dev[0].set("sae_groups", "") dev[0].set("sae_pwe", "0") def test_eht_sae_mlo(dev, apdev): """EHT+MLO AP with SAE""" check_sae_capab(dev[0]) hapd = start_eht_sae_ap(apdev[0], ml=True) try: dev[0].set("sae_groups", "20") dev[0].set("sae_pwe", "2") dev[0].connect("eht", key_mgmt="SAE-EXT-KEY", psk="12345678", ieee80211w="2", beacon_prot="1", pairwise="GCMP-256", group="GCMP-256", group_mgmt="BIP-GMAC-256", scan_freq="2412") finally: dev[0].set("sae_groups", "") dev[0].set("sae_pwe", "0") def test_eht_sae_mlo_tm(dev, apdev): """EHT+MLO AP with SAE and transition mode""" check_sae_capab(dev[0]) check_sae_capab(dev[1]) hapd = start_eht_sae_ap(apdev[0], ml=True, transition_mode=True) try: dev[0].set("sae_groups", "20") dev[0].set("sae_pwe", "2") dev[0].connect("eht", key_mgmt="SAE-EXT-KEY", psk="12345678", ieee80211w="2", beacon_prot="1", pairwise="GCMP-256", group="CCMP", group_mgmt="AES-128-CMAC", scan_freq="2412") dev[1].set("sae_groups", "19") dev[1].connect("eht", key_mgmt="SAE-EXT-KEY", psk="12345678", ieee80211w="2", beacon_prot="1", pairwise="CCMP", group="CCMP", group_mgmt="AES-128-CMAC", scan_freq="2412", disable_eht="1") dev[2].connect("eht", key_mgmt="WPA-PSK", psk="12345678", pairwise="CCMP", group="CCMP", group_mgmt="AES-128-CMAC", scan_freq="2412", disable_eht="1") finally: dev[0].set("sae_groups", "") dev[0].set("sae_pwe", "0") dev[1].set("sae_groups", "") def eht_mld_enable_ap(iface, params): hapd = hostapd.add_mld_link(iface, params) hapd.enable() ev = hapd.wait_event(["AP-ENABLED", "AP-DISABLED"], timeout=1) if ev is None: raise Exception("AP startup timed out") if "AP-ENABLED" not in ev: raise Exception("AP startup failed") return hapd def eht_mld_ap_wpa2_params(ssid, passphrase=None, key_mgmt="WPA-PSK-SHA256", mfp="2", pwe=None, beacon_prot="1", bridge=False): params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase, wpa_key_mgmt=key_mgmt, ieee80211w=mfp) params['ieee80211n'] = '1' params['ieee80211ax'] = '1' params['ieee80211be'] = '1' params['channel'] = '1' params['hw_mode'] = 'g' params['group_mgmt_cipher'] = "AES-128-CMAC" params['beacon_prot'] = beacon_prot if bridge: params['bridge'] = 'ap-br0' if pwe is not None: params['sae_pwe'] = pwe return params def _eht_mld_probe_req(wpas, hapd, tsf0, link_id=-1): if "OK" not in wpas.request("ML_PROBE_REQ bssid=%s mld_id=0 link_id=%d" % (hapd.own_addr(), link_id)): raise Exception("Failed to request ML probe request") ev = wpas.wait_event(["CTRL-EVENT-SCAN-STARTED"]) if ev is None: raise Exception("Scan did not start") ev = wpas.wait_event(["CTRL-EVENT-SCAN-RESULTS"]) if ev is None: raise Exception("Scan did not complete") logger.info("ML Probe request scan done") bss = wpas.get_bss(hapd.own_addr()) if not bss: raise Exception("AP did not reply to ML probe request") tsf1 = int(bss['tsf']) logger.info("tsf0=%s, tsf1=%s" % (tsf0, tsf1)) if tsf0 >= tsf1: raise Exception("AP was not found in ML probe request scan") def test_eht_mld_discovery(dev, apdev): """EHT MLD AP discovery""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap" link0_params = {"ssid": ssid, "hw_mode": "g", "channel": "1"} link1_params = {"ssid": ssid, "hw_mode": "g", "channel": "2"} hapd0 = eht_mld_enable_ap(hapd_iface, link0_params) hapd1 = eht_mld_enable_ap(hapd_iface, link1_params) # Only scan link 0 res = wpas.request("SCAN freq=2412") if "FAIL" in res: raise Exception("Failed to start scan") ev = wpas.wait_event(["CTRL-EVENT-SCAN-STARTED"]) if ev is None: raise Exception("Scan did not start") ev = wpas.wait_event(["CTRL-EVENT-SCAN-RESULTS"]) if ev is None: raise Exception("Scan did not complete") logger.info("Scan done") rnr_pattern = re.compile(".*ap_info.*, mld ID=0, link ID=", re.MULTILINE) ml_pattern = re.compile(".*multi-link:.*, MLD addr=.*", re.MULTILINE) bss = wpas.request("BSS " + hapd0.own_addr()) logger.info("BSS 0: " + str(bss)) if rnr_pattern.search(bss) is None: raise Exception("RNR element not found for first link") if ml_pattern.search(bss) is None: raise Exception("ML element not found for first link") # Save the tsf0 for checking ML Probe request scan later tsf0 = int(wpas.get_bss(hapd0.own_addr())['tsf']) if wpas.get_bss(hapd1.own_addr()) is not None: raise Exception("BSS for link 1 found without ML probe request") # Now send an ML probe request (for all links) _eht_mld_probe_req(wpas, hapd0, tsf0) tsf0 = int(wpas.get_bss(hapd0.own_addr())['tsf']) # NOTE: hostapd incorrectly reports a TSF offset of zero # This only works because the source is always the ML probe response tsf1 = int(wpas.get_bss(hapd1.own_addr())['tsf']) bss = wpas.request("BSS " + hapd1.own_addr()) logger.info("BSS 1: " + str(bss)) if rnr_pattern.search(bss) is None: raise Exception("RNR element not found for second link") if ml_pattern.search(bss) is None: raise Exception("ML element not found for second link") _eht_mld_probe_req(wpas, hapd0, tsf0, link_id=1) if int(wpas.get_bss(hapd1.own_addr())['tsf']) <= tsf1: raise Exception("Probe for link ID did not update BSS") tsf0 = int(wpas.get_bss(hapd0.own_addr())['tsf']) tsf1 = int(wpas.get_bss(hapd1.own_addr())['tsf']) # Probing the wrong link ID should not update second link _eht_mld_probe_req(wpas, hapd0, tsf0, link_id=4) if int(wpas.get_bss(hapd1.own_addr())['tsf']) != tsf1: raise Exception("Probe for other link ID not updated BSS") def test_eht_mld_owe_two_links(dev, apdev): """EHT MLD AP with MLD client OWE connection using two links""" _eht_mld_owe_two_links(dev, apdev) def _eht_mld_owe_two_links(dev, apdev, second_link_disabled=False, only_one_link=False): 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): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap_owe_two_link" params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") hapd0 = eht_mld_enable_ap(hapd0_iface, params) params['channel'] = '6' if second_link_disabled: params['mld_indicate_disabled'] = '1' hapd1 = eht_mld_enable_ap(hapd0_iface, params) # Check legacy client connection dev[0].connect(ssid, scan_freq="2437", key_mgmt="OWE", ieee80211w="2") if only_one_link: link0 = hapd0.get_status_field("link_addr") wpas.set("bssid_filter", link0) wpas.connect(ssid, scan_freq="2412 2437", key_mgmt="OWE", ieee80211w="2") active_links = 3 valid_links = 3 if second_link_disabled: dlinks = _eht_dormant_links(wpas) if dlinks != 2: raise Exception("Unexpected dormant links") active_links = 1 if only_one_link: active_links = 1 valid_links = 1 eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=valid_links, active_links=active_links) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) if not second_link_disabled: traffic_test(wpas, hapd1) if only_one_link: wpas.set("bssid_filter", "") def test_eht_mld_owe_two_links_one_disabled(dev, apdev): """AP MLD with MLD client OWE connection when one of the AP MLD links is disabled""" _eht_mld_owe_two_links(dev, apdev, second_link_disabled=True) def test_eht_mld_owe_two_links_only_one_negotiated(dev, apdev): """AP MLD with MLD client OWE connection when only one of the links is negotiated""" _eht_mld_owe_two_links(dev, apdev, only_one_link=True) def test_eht_mld_sae_single_link(dev, apdev): """EHT MLD AP with MLD client SAE H2E connection using single link""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) passphrase = 'qwertyuiop' ssid = "mld_ap_sae_single_link" params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE", mfp="2", pwe='2') hapd0 = eht_mld_enable_ap(hapd_iface, params) wpas.set("sae_pwe", "1") wpas.connect(ssid, sae_password=passphrase, scan_freq="2412", key_mgmt="SAE", ieee80211w="2") eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=1, active_links=1) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) def run_eht_mld_sae_two_links(dev, apdev, beacon_prot="1", disable_enable=False, bridge=False): with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) passphrase = 'qwertyuiop' ssid = "mld_ap_sae_two_link" params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE", mfp="2", pwe='1', beacon_prot=beacon_prot, bridge=bridge) hapd0 = eht_mld_enable_ap(hapd_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd_iface, params) if bridge: hapd0.cmd_execute(['brctl', 'setfd', 'ap-br0', '0']) hapd0.cmd_execute(['ip', 'link', 'set', 'dev', 'ap-br0', 'up']) wpas.set("sae_pwe", "1") # The first authentication attempt tries to use group 20 and the # authentication is expected to fail. The next authentication should # use group 19 and succeed. wpas.set("sae_groups", "20 19") wpas.connect(ssid, sae_password=passphrase, scan_freq="2412 2437", key_mgmt="SAE", ieee80211w="2", beacon_prot="1") eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas) if wpas.get_status_field('sae_group') != '19': raise Exception("Expected SAE group not used") traffic_test(wpas, hapd0, ifname2='ap-br0' if bridge else None) traffic_test(wpas, hapd1, ifname2='ap-br0' if bridge else None) if disable_enable: if "OK" not in hapd0.request("DISABLE_MLD"): raise Exception("DISABLE_MLD failed") ev = hapd0.wait_event(["AP-DISABLED"], timeout=1) if ev is None: raise Exception("AP-DISABLED not received (0)") ev = hapd1.wait_event(["AP-DISABLED"], timeout=1) if ev is None: raise Exception("AP-DISABLED not received (1)") # mac80211 does not seem to detect beacon loss or deauthentication # in non-AP MLD case?! For now, ignore that and just force # disconnection locally on the STA. wpas.request("DISCONNECT") wpas.wait_disconnected() if "OK" not in hapd0.request("ENABLE_MLD"): raise Exception("ENABLE_MLD failed") ev = hapd0.wait_event(["AP-ENABLED"], timeout=1) if ev is None: raise Exception("AP-ENABLED not received (0)") ev = hapd1.wait_event(["AP-ENABLED"], timeout=1) if ev is None: raise Exception("AP-ENABLED not received (1)") # TODO: Figure out why this fails without PMKSA_FLUSH. Things should # fall back to full SAE from failed PMKSA caching attempt # automatically. wpas.request("PMKSA_FLUSH") # flush the BSS table before reconnect as otherwise the old # AP MLD BSSs would be in the BSS list wpas.request("BSS_FLUSH 0") wpas.request("RECONNECT") wpas.wait_connected() hapd0.wait_sta() hapd1.wait_sta() traffic_test(wpas, hapd0, ifname2='ap-br0' if bridge else None) traffic_test(wpas, hapd1, ifname2='ap-br0' if bridge else None) def test_eht_mld_sae_two_links(dev, apdev): """EHT MLD AP with MLD client SAE H2E connection using two links""" run_eht_mld_sae_two_links(dev, apdev) def test_eht_mld_sae_two_links_no_beacon_prot(dev, apdev): """EHT MLD AP with MLD client SAE H2E connection using two links and no beacon protection""" run_eht_mld_sae_two_links(dev, apdev, beacon_prot="0") def test_eht_mld_sae_two_links_disable_enable(dev, apdev): """AP MLD with two links and disabling/enabling full AP MLD""" run_eht_mld_sae_two_links(dev, apdev, disable_enable=True) def test_eht_mld_sae_two_links_bridge(dev, apdev): """AP MLD with two links in a bridge""" run_eht_mld_sae_two_links(dev, apdev, bridge=True) def test_eht_mld_sae_ext_one_link(dev, apdev): """EHT MLD AP with MLD client SAE-EXT H2E connection using single link""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) passphrase = 'qwertyuiop' ssid = "mld_ap_sae_ext_single_link" params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE-EXT-KEY") hapd0 = eht_mld_enable_ap(hapd_iface, params) wpas.connect(ssid, sae_password=passphrase, scan_freq="2412", key_mgmt="SAE-EXT-KEY", ieee80211w="2") eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=1, active_links=1) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) def test_eht_mld_sae_ext_two_links(dev, apdev): """EHT MLD AP with MLD client SAE-EXT H2E connection using two links""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) passphrase = 'qwertyuiop' ssid = "mld_ap_sae_two_link" params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE-EXT-KEY") hapd0 = eht_mld_enable_ap(hapd_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd_iface, params) wpas.connect(ssid, sae_password=passphrase, scan_freq="2412 2437", key_mgmt="SAE-EXT-KEY", ieee80211w="2") eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) traffic_test(wpas, hapd1) def test_eht_mld_sae_legacy_client(dev, apdev): """EHT MLD AP with legacy client SAE H2E connection""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface): passphrase = 'qwertyuiop' ssid = "mld_ap_sae_two_link" params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE", mfp="2", pwe='1') hapd0 = eht_mld_enable_ap(hapd_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd_iface, params) try: dev[0].set("sae_groups", "") dev[0].set("sae_pwe", "1") dev[0].connect(ssid, sae_password=passphrase, scan_freq="2412", key_mgmt="SAE", ieee80211w="2", beacon_prot="1") logger.info("wpa_supplicant STATUS:\n" + dev[0].request("STATUS")) bssid = dev[0].get_status_field("bssid") if hapd0.own_addr() == bssid: hapd0.wait_sta(); elif hapd1.own_addr() == bssid: hapd1.wait_sta(); else: raise Exception("Unknown BSSID: " + bssid) eht_verify_status(dev[0], hapd0, 2412, 20, is_ht=True) traffic_test(dev[0], hapd0) finally: dev[0].set("sae_groups", "") dev[0].set("sae_pwe", "0") def test_eht_mld_sae_transition(dev, apdev): """EHT MLD AP in SAE/PSK transition mode with MLD client connection using two links""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) passphrase = 'qwertyuiop' ssid = "mld_ap_sae_two_link" params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE-EXT-KEY SAE WPA-PSK WPA-PSK-SHA256", mfp="1") hapd0 = eht_mld_enable_ap(hapd_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd_iface, params) wpas.connect(ssid, sae_password=passphrase, scan_freq="2412 2437", key_mgmt="SAE-EXT-KEY", ieee80211w="2") eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) traffic_test(wpas, hapd1) dev[0].set("sae_groups", "") dev[0].connect(ssid, sae_password=passphrase, scan_freq="2412", key_mgmt="SAE", ieee80211w="2", beacon_prot="1") dev[1].connect(ssid, psk=passphrase, scan_freq="2412", key_mgmt="WPA-PSK", ieee80211w="0") def test_eht_mld_ptk_rekey(dev, apdev): """EHT MLD AP and PTK rekeying with MLD client connection using two links""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) passphrase = 'qwertyuiop' ssid = "mld_ap_sae_two_link" params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE-EXT-KEY SAE WPA-PSK WPA-PSK-SHA256", mfp="1") params['wpa_ptk_rekey'] = '5' hapd0 = eht_mld_enable_ap(hapd_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd_iface, params) wpas.connect(ssid, sae_password=passphrase, scan_freq="2412 2437", key_mgmt="SAE-EXT-KEY", ieee80211w="2") ev0 = hapd0.wait_event(["AP-STA-CONNECT"], timeout=1) if ev0 is None: ev1 = hapd1.wait_event(["AP-STA-CONNECT"], timeout=1) traffic_test(wpas, hapd0) traffic_test(wpas, hapd1) ev = wpas.wait_event(["WPA: Key negotiation completed", "CTRL-EVENT-DISCONNECTED"], timeout=10) if ev is None: raise Exception("PTK rekey timed out") if "CTRL-EVENT-DISCONNECTED" in ev: raise Exception("Disconnect instead of rekey") time.sleep(0.1) traffic_test(wpas, hapd0) traffic_test(wpas, hapd1) def test_eht_mld_gtk_rekey(dev, apdev): """AP MLD and GTK rekeying with MLD client connection using two links""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) passphrase = 'qwertyuiop' ssid = "mld_ap_sae_two_link" params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE-EXT-KEY SAE WPA-PSK WPA-PSK-SHA256", mfp="1") params['wpa_group_rekey'] = '5' hapd0 = eht_mld_enable_ap(hapd_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd_iface, params) wpas.connect(ssid, sae_password=passphrase, scan_freq="2412 2437", key_mgmt="SAE-EXT-KEY", ieee80211w="2") ev0 = hapd0.wait_event(["AP-STA-CONNECT"], timeout=1) if ev0 is None: ev1 = hapd1.wait_event(["AP-STA-CONNECT"], timeout=1) traffic_test(wpas, hapd0) traffic_test(wpas, hapd1) for i in range(2): ev = wpas.wait_event(["MLO RSN: Group rekeying completed", "CTRL-EVENT-DISCONNECTED"], timeout=10) if ev is None: raise Exception("GTK rekey timed out") if "CTRL-EVENT-DISCONNECTED" in ev: raise Exception("Disconnect instead of rekey") time.sleep(0.1) traffic_test(wpas, hapd0) traffic_test(wpas, hapd1) def test_eht_ml_probe_req(dev, apdev): """AP MLD with two links and non-AP MLD sending ML Probe Request""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) passphrase = 'qwertyuiop' ssid = "mld_ap_sae_two_link" params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE-EXT-KEY") hapd0 = eht_mld_enable_ap(hapd_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd_iface, params) bssid = hapd0.own_addr() wpas.scan_for_bss(bssid, freq=2412) time.sleep(1) cmd = "ML_PROBE_REQ bssid=" + bssid + " mld_id=0" if "OK" not in wpas.request(cmd): raise Exception("Failed to run: " + cmd) ev = wpas.wait_event(["CTRL-EVENT-SCAN-RESULTS", "CTRL-EVENT-SCAN-FAILED"], timeout=10) if ev is None: raise Exception("ML_PROBE_REQ did not result in scan results") time.sleep(1) cmd = "ML_PROBE_REQ bssid=" + bssid + " mld_id=0 link_id=2" if "OK" not in wpas.request(cmd): raise Exception("Failed to run: " + cmd) ev = wpas.wait_event(["CTRL-EVENT-SCAN-RESULTS", "CTRL-EVENT-SCAN-FAILED"], timeout=10) if ev is None: raise Exception("ML_PROBE_REQ did not result in scan results") def test_eht_mld_connect_probes(dev, apdev, params): """MLD client sends ML probe to connect to not discovered links""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap" passphrase = 'qwertyuiop' link_params = eht_mld_ap_wpa2_params(ssid, passphrase, mfp="2", key_mgmt="SAE", pwe='2') link_params['channel'] = '1' link_params['bssid'] = '00:11:22:33:44:01' hapd0 = eht_mld_enable_ap(hapd_iface, link_params) link_params['channel'] = '6' link_params['bssid'] = '00:11:22:33:44:02' hapd1 = eht_mld_enable_ap(hapd_iface, link_params) wpas.set("sae_pwe", "1") wpas.connect(ssid, sae_password= passphrase, ieee80211w="2", key_mgmt="SAE", scan_freq="2412") out = run_tshark(os.path.join(params['logdir'], 'hwsim0.pcapng'), 'wlan.fc.type_subtype == 0x0004 && wlan.ext_tag.number == 107 && wlan.ext_tag.data == 11:00:02:00:00:02:11:00', display=['frame.number']) if not out.splitlines(): raise Exception('ML probe request not found') # Probe Response frame has the ML element, which will be fragmented out = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"), "wlan.fc.type_subtype == 0x0005 && wlan.ext_tag.number == 107 && wlan.ext_tag.length == 254", display=['frame.number']) if not out.splitlines(): # This requires new tshark (e.g., 4.0.6); for now, ignore the issue # to avoid forcing such upgrade. logger.info('ML probe response not found') #raise Exception('ML probe response not found') eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) traffic_test(wpas, hapd0) traffic_test(wpas, hapd1) def test_eht_tx_link_rejected_connect_other(dev, apdev, params): """EHT MLD AP with MLD client being rejected on TX link, but then connecting on second link""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap" passphrase = 'qwertyuiop' link_params = eht_mld_ap_wpa2_params(ssid, passphrase, mfp="2", key_mgmt="SAE", pwe='2') link_params['channel'] = '1' link_params['bssid'] = '00:11:22:33:44:01' hapd0 = eht_mld_enable_ap(hapd_iface, link_params) link_params['channel'] = '6' link_params['bssid'] = '00:11:22:33:44:02' hapd1 = eht_mld_enable_ap(hapd_iface, link_params) wpas.set("sae_pwe", "1") with fail_test(hapd0, 1, "hostapd_get_aid"): wpas.connect(ssid, sae_password=passphrase, ieee80211w="2", key_mgmt="SAE", scan_freq="2412") eht_verify_status(wpas, hapd1, 2437, 20, is_ht=True, mld=True, valid_links=2, active_links=2) traffic_test(wpas, hapd0) traffic_test(wpas, hapd1) def test_eht_all_links_rejected(dev, apdev, params): """EHT MLD AP with MLD client ignores all rejected links""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap" passphrase = 'qwertyuiop' link_params = eht_mld_ap_wpa2_params(ssid, passphrase, mfp="2", key_mgmt="SAE", pwe='2') link_params['channel'] = '1' link_params['bssid'] = '00:11:22:33:44:01' hapd0 = eht_mld_enable_ap(hapd_iface, link_params) link_params['channel'] = '6' link_params['bssid'] = '00:11:22:33:44:02' hapd1 = eht_mld_enable_ap(hapd_iface, link_params) wpas.set("mld_connect_bssid_pref", "00:11:22:33:44:01") wpas.set("sae_pwe", "1") with fail_test(hapd0, 1, "hostapd_get_aid", 1, "hostapd_process_assoc_ml_info"): wpas.connect(ssid, sae_password=passphrase, ieee80211w="2", key_mgmt="SAE", scan_freq="2412", wait_connect=False) ev = wpas.wait_event(['CTRL-EVENT-ASSOC-REJECT']) if not ev: raise Exception('Rejection not found') ev1 = wpas.wait_event(['Added BSSID']) ev2 = wpas.wait_event(['Added BSSID']) if (not ev1 or not ev2) or \ not ((hapd0.own_addr() in ev1 and hapd1.own_addr() in ev2) or (hapd1.own_addr() in ev1 and hapd0.own_addr() in ev2)): raise Exception('Not all BSSs were added to the ignore list') # After this message, a new scan clears the ignore and the STA # connects. wpas.wait_connected(timeout=15) def test_eht_connect_invalid_link(dev, apdev, params): """EHT MLD AP where one link is incorrectly configured and rejected by mac80211""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap" passphrase = 'qwertyuiop' ssid = "mld_ap" passphrase = 'qwertyuiop' link_params = eht_mld_ap_wpa2_params(ssid, passphrase, mfp="2", key_mgmt="SAE", pwe='2') link_params['channel'] = '1' link_params['bssid'] = '00:11:22:33:44:01' hapd0 = eht_mld_enable_ap(hapd_iface, link_params) link_params['channel'] = '6' link_params['bssid'] = '00:11:22:33:44:02' hapd1 = eht_mld_enable_ap(hapd_iface, link_params) # We scan for both APs, then try to connect to link 0, but only the # second attempt will work if mac80211 rejects the second link. wpas.set("mld_connect_bssid_pref", "00:11:22:33:44:01") wpas.set("sae_pwe", "1") with fail_test(wpas, 1, "assoc;wpa_driver_nl80211_associate", 2, "link;wpa_driver_nl80211_associate"): wpas.connect(ssid, sae_password=passphrase, ieee80211w="2", key_mgmt="SAE", scan_freq="2412") eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=1, active_links=1) out = run_tshark(os.path.join(params['logdir'], 'hwsim0.pcapng'), 'wlan.fc.type_subtype == 0x0000 && wlan.ext_tag.data == 00:01:09:%s:00:00' % wpas.own_addr(), display=['frame.number']) if not out.splitlines(): raise Exception('Association request send by mac80211 had unexpected ML element content (probably it contained a second link)') def test_eht_mld_link_removal(dev, apdev): """EHT MLD with two links. Links removed during association""" with HWSimRadio(use_mlo=True) as (hapd0_radio, hapd0_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap_owe_two_link" params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") hapd0 = eht_mld_enable_ap(hapd0_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd0_iface, params) wpas.connect(ssid, scan_freq="2412 2437", key_mgmt="OWE", ieee80211w="2") eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) logger.info("Disable the 2nd link in 4 beacon intervals") hapd1.link_remove(4) time.sleep(0.6) logger.info("Test traffic after 2nd link disabled") traffic_test(wpas, hapd0) if "OK" not in hapd0.request("REKEY_GTK"): raise Exception("REKEY_GTK failed") ev = wpas.wait_event(["MLO RSN: Group rekeying completed"], timeout=2) if ev is None: raise Exception("GTK rekey timed out") traffic_test(wpas, hapd0) logger.info("Disable the 1st link in 20 beacon intervals") hapd0.link_remove(20) time.sleep(1) logger.info("Verify that traffic is valid before the link is removed") traffic_test(wpas, hapd0) time.sleep(2) logger.info("Test traffic after 1st link disabled") traffic_test(wpas, hapd0, success=False) def test_eht_mld_bss_trans_mgmt_link_removal_imminent(dev, apdev): """EHT MLD with two links. BSS transition management with link removal imminent""" with HWSimRadio(use_mlo=True) as (hapd0_radio, hapd0_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap_owe_two_link" params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") params["bss_transition"] = "1" params["mbo"] = "1" hapd0 = eht_mld_enable_ap(hapd0_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd0_iface, params) wpas.connect(ssid, scan_freq="2412 2437", key_mgmt="OWE", ieee80211w="2") eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas) hapd0.wait_sta() hapd1.wait_sta() traffic_test(wpas, hapd0) addr = wpas.own_addr() cmd = "BSS_TM_REQ " + addr + " disassoc_timer=3 disassoc_imminent=1 link_removal_imminent=1 bss_term=0,1" if "OK" not in hapd0.request(cmd): raise Exception("BSS_TM_REQ command failed") # Only one link is terminate, so the STA is expected to remain # associated and not start a scan. ev = hapd0.wait_event(['BSS-TM-RESP'], timeout=5) # For now, allow this to pass without the BSS TM response since that # functionality with MLD needs a recent kernel change. #if ev is None: # raise Exception("No BSS TM response received") if ev and "status_code=0" not in ev: raise Exception("Unexpected BSS TM response contents: " + ev) ev = wpas.wait_event(["CTRL-EVENT-SCAN-STARTED", "CTRL-EVENT-DISCONNECTED"], timeout=10) if ev is not None: raise Exception("Unexpected action on STA: " + ev) 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) if no_tx_status: return ev = hapd.wait_event(["MGMT-TX-STATUS"], timeout=1) if ev is None: raise Exception("No TX status") def test_eht_ap_mld_proto(dev, apdev): """AP MLD protocol testing""" with HWSimRadio(use_mlo=True) as (hapd0_radio, hapd0_iface), \ HWSimRadio(use_mlo=True) as (hapd1_radio, hapd1_iface): ssid = "mld_ap_owe_two_link" params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") hapd0 = eht_mld_enable_ap(hapd0_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd0_iface, params) ap_mld_addr = hapd0.get_status_field("mld_addr[0]").replace(':', '') bssid0 = hapd0.own_addr().replace(':', '') bssid1 = hapd1.own_addr().replace(':', '') time.sleep(1) hapd0.set("ext_mgmt_frame_handling", "1") hapd1.set("ext_mgmt_frame_handling", "1") # Truncated EML missing MLD Capabilities And operations field hapd0.note("Truncated EML missing MLD Capabilities And operations field") addr0 = "021122334400" addr1 = "021122334401" mld_addr = "02112233440f" hdr = "b0003a01" + bssid0 + addr0 + bssid0 + "1000" mle = "ff0a6b000007" + mld_addr auth = hdr + "0000" + "0100" + "0000" + mle send_check(hapd0, auth) hdr = "00000000" + bssid0 + mld_addr + bssid0 + "1000" ssid = "00136d6c645f61705f6f77655f74776f5f6c696e6b" supp_rates = "010802040b160c121824" ext_supp_rates = "32043048606c" rsne = "301a0100000fac040100000fac040100000fac12cc000000000fac06" ht_capab = "2d1afe131bffff000000000000000000000100000000000000000000" ext_capab = "7f0a04004a02014000400001" he_capab = "ff16230178c81a400000bfce0000000000000000fafffaff" eht_capab = "ff126c07007c0000feffff7f0100888888880000" supp_op_classes = "3b155151525354737475767778797a7b7c7d7e7f808182" dh_param = "ff23201300ea85e693343a079500cf4d461011a0ff90ec4de1af40165adbea94a3f36eb071" wmm = "dd070050f202000100" assocreq_start = "3004" + "0500" + ssid + supp_rates + ext_supp_rates + rsne + ht_capab + ext_capab + he_capab assocreq_end = eht_capab + supp_op_classes + dh_param + wmm # --> Not enough bytes for common info mle = "ff0a6b000109" + mld_addr send_check(hapd0, hdr + assocreq_start + mle + assocreq_end) # Truncated Non-Inheritance element hapd0.note("Truncated Non-Inheritance element") addr0 = "021122334410" addr1 = "021122334411" mld_addr = "02112233441f" hdr = "b0003a01" + bssid0 + addr0 + bssid0 + "1000" mle = "ff0a6b000007" + mld_addr auth = hdr + "0000" + "0100" + "0000" + mle send_check(hapd0, auth) # --> MLD: Invalid inheritance mle = "ff7d6b000109" + mld_addr + "0000" mle += "0067" + "3100" + "07" + addr1 mle += "3004" + "010802040b160c121824" + "32043048606c" + "2d1afe131bffff000000000000000000000100000000000000000000" + "ff16230178c81a400000bfce0000000000000000fafffaff" + "ff126c07007c0000feffff7f0100888888880000" # Non-Inhericance element mle += "ff023800" # Unknown optional subelement mle += "aa00" # Vendor-Specific subelement mle += "dd0411223344" hdr = "00000000" + bssid0 + mld_addr + bssid0 + "1000" send_check(hapd0, hdr + assocreq_start + mle + assocreq_end, no_tx_status=True) # Empty Non-Inheritance element hapd0.note("Empty Non-Inheritance element") addr0 = "021122334420" addr1 = "021122334421" mld_addr = "02112233442f" hdr = "b0003a01" + bssid0 + addr0 + bssid0 + "1000" mle = "ff0a6b000007" + mld_addr auth = hdr + "0000" + "0100" + "0000" + mle send_check(hapd0, auth) mle = "ff7e6b000109" + mld_addr + "0000" mle += "0068" + "3100" + "07" + addr1 mle += "3004" + "010802040b160c121824" + "32043048606c" + "2d1afe131bffff000000000000000000000100000000000000000000" + "ff16230178c81a400000bfce0000000000000000fafffaff" + "ff126c07007c0000feffff7f0100888888880000" # Non-Inhericance element mle += "ff03380000" # Unknown optional subelement mle += "aa00" # Vendor-Specific subelement mle += "dd0411223344" hdr = "00000000" + bssid0 + mld_addr + bssid0 + "1000" send_check(hapd0, hdr + assocreq_start + mle + assocreq_end) # Non-Inheritance element hapd0.note("Non-Inheritance element") addr0 = "021122334430" addr1 = "021122334431" mld_addr = "02112233443f" hdr = "b0003a01" + bssid0 + addr0 + bssid0 + "1000" mle = "ff0a6b000007" + mld_addr auth = hdr + "0000" + "0100" + "0000" + mle send_check(hapd0, auth) mle = "ff9e6b000109" + mld_addr + "0000" mle += "0088" + "3100" + "07" + addr1 mle += "3004" + "010802040b160c121824" + "32043048606c" + "2d1afe131bffff000000000000000000000100000000000000000000" + "ff16230178c81a400000bfce0000000000000000fafffaff" + "ff126c07007c0000feffff7f0100888888880000" # Non-Inhericance element mle += "ff2338" + "1010032a362137387172756b548bedeff0" + "106b01020304050607080c0d21643b3a36" # Unknown optional subelement mle += "aa00" # Vendor-Specific subelement mle += "dd0411223344" hdr = "00000000" + bssid0 + mld_addr + bssid0 + "1000" send_check(hapd0, hdr + assocreq_start + mle + assocreq_end) # No Non-Inheritance element hapd0.note("No Non-Inheritance element") addr0 = "021122334440" addr1 = "021122334441" mld_addr = "02112233444f" hdr = "b0003a01" + bssid0 + addr0 + bssid0 + "1000" mle = "ff0a6b000007" + mld_addr auth = hdr + "0000" + "0100" + "0000" + mle send_check(hapd0, auth) mle = "ff716b000109" + mld_addr + "0000" mle += "0063" + "3100" + "07" + addr1 mle += "3004010802040b160c12182432043048606c2d1afe131bffff000000000000000000000100000000000000000000ff16230178c81a400000bfce0000000000000000fafffaffff126c07007c0000feffff7f0100888888880000" hdr = "00000000" + bssid0 + mld_addr + bssid0 + "1000" send_check(hapd0, hdr + assocreq_start + mle + assocreq_end) def _5ghz_chanwidth_to_bw(op): return { 0: "40", 1: "80", 2: "160", 3: "80+80", }.get(op, "20") def _test_eht_5ghz(dev, apdev, channel, chanwidth, ccfs1, ccfs2=0, eht_oper_puncturing_override=None, he_ccfs1=None, he_oper_chanwidth=None): if he_ccfs1 is None: he_ccfs1 = ccfs1 if he_oper_chanwidth is None: he_oper_chanwidth = chanwidth try: params = {"ssid": "eht", "country_code": "US", "hw_mode": "a", "channel": str(channel), "ieee80211n": "1", "ieee80211ac": "1", "ieee80211ax": "1", "ieee80211be": "1", "vht_oper_chwidth": str(he_oper_chanwidth), "vht_oper_centr_freq_seg0_idx": str(he_ccfs1), "vht_oper_centr_freq_seg1_idx": str(ccfs2), "he_oper_chwidth": str(he_oper_chanwidth), "he_oper_centr_freq_seg0_idx": str(he_ccfs1), "he_oper_centr_freq_seg1_idx": str(ccfs2), "eht_oper_centr_freq_seg0_idx": str(ccfs1), "eht_oper_chwidth": str(chanwidth)} if he_oper_chanwidth == 0: if channel < he_ccfs1: params["ht_capab"] = "[HT40+]" elif channel > he_ccfs1: params["ht_capab"] = "[HT40-]" else: params["ht_capab"] = "[HT40+]" if he_oper_chanwidth == 2: params["vht_capab"] = "[VHT160]" elif he_oper_chanwidth == 3: params["vht_capab"] = "[VHT160-80PLUS80]" if eht_oper_puncturing_override: params['eht_oper_puncturing_override'] = eht_oper_puncturing_override freq = 5000 + channel * 5 bw = "20" if chanwidth != 0 or channel != ccfs1: bw = _5ghz_chanwidth_to_bw(chanwidth) hapd = hostapd.add_ap(apdev[0], params) dev[0].connect("eht", key_mgmt="NONE", scan_freq=str(freq)) hapd.wait_sta() eht_verify_status(dev[0], hapd, freq, bw, is_ht=True, is_vht=True) eht_verify_wifi_version(dev[0]) hwsim_utils.test_connectivity(dev[0], hapd) if eht_oper_puncturing_override: hapd.set("eht_oper_puncturing_override", "0x0") hapd.request("UPDATE_BEACON") time.sleep(1) finally: dev[0].request("DISCONNECT") dev[0].wait_disconnected() hapd.wait_sta_disconnect() set_world_reg(apdev[0], None, dev[0]) def test_eht_5ghz_20mhz(dev, apdev): """EHT with 20 MHz channel width on 5 GHz""" _test_eht_5ghz(dev, apdev, 36, 0, 36, 0) def test_eht_5ghz_40mhz_low(dev, apdev): """EHT with 40 MHz channel width on 5 GHz - secondary channel above""" _test_eht_5ghz(dev, apdev, 36, 0, 38, 0) def test_eht_5ghz_40mhz_high(dev, apdev): """EHT with 80 MHz channel width on 5 GHz - secondary channel below""" _test_eht_5ghz(dev, apdev, 40, 0, 38, 0) def test_eht_5ghz_80mhz_1(dev, apdev): """EHT with 80 MHz channel width on 5 GHz - primary=149""" _test_eht_5ghz(dev, apdev, 36, 1, 42, 0) def test_eht_5ghz_80mhz_2(dev, apdev): """EHT with 80 MHz channel width on 5 GHz - primary=149""" _test_eht_5ghz(dev, apdev, 149, 1, 155, 0) def test_eht_5ghz_80mhz_puncturing_override_1(dev, apdev): """EHT with 80 MHz channel width on 5 GHz - primary=36 - puncturing override (2nd)""" # The 2nd 20 MHz is punctured _test_eht_5ghz(dev, apdev, 36, 1, 42, 0, eht_oper_puncturing_override="0x0002", he_ccfs1=36, he_oper_chanwidth=0) def test_eht_5ghz_80mhz_puncturing_override_2(dev, apdev): """EHT with 80 MHz channel width on 5 GHz - primary=149 - puncturing override (3rd)""" # The 3rd 20 MHz is punctured _test_eht_5ghz(dev, apdev, 149, 1, 155, 0, eht_oper_puncturing_override="0x0004", he_ccfs1=151, he_oper_chanwidth=0) def test_eht_5ghz_80mhz_puncturing_override_3(dev, apdev): """EHT with 80 MHz channel width on 5 GHz - primary=149 - puncturing override (4th)""" # The 4th 20 MHz is punctured _test_eht_5ghz(dev, apdev, 149, 1, 155, 0, eht_oper_puncturing_override="0x0008", he_ccfs1=151, he_oper_chanwidth=0) def test_eht_5ghz_80p80mhz(dev, apdev): """EHT with 80+80 MHz channel width on 5 GHz""" _test_eht_5ghz(dev, apdev, 36, 3, 42, 155) def _6ghz_op_class_to_bw(op): return { 131: "20", 132: "40", 133: "80", 134: "160", 137: "320", }.get(op, "20") def _test_eht_6ghz(dev, apdev, channel, op_class, ccfs1): check_sae_capab(dev[0]) # CA enables 320 MHz channels without NO-IR restriction dev[0].cmd_execute(['iw', 'reg', 'set', 'CA']) wait_regdom_changes(dev[0]) try: ssid = "eht_6ghz_sae" passphrase = "12345678" params = hostapd.he_wpa2_params(ssid=ssid, passphrase=passphrase) params["ieee80211be"] = "1" params["channel"] = str(channel) params["op_class"] = str(op_class) params["he_oper_centr_freq_seg0_idx"] = str(ccfs1) params["eht_oper_centr_freq_seg0_idx"] = str(ccfs1) params["country_code"] = "CA" if not he_6ghz_supported(): raise HwsimSkip("6 GHz frequency is not supported") if op_class == 137 and not eht_320mhz_supported(): raise HwsimSkip("320 MHz channels are not supported") hapd = hostapd.add_ap(apdev[0], params) status = hapd.get_status() logger.info("hostapd STATUS: " + str(status)) if hapd.get_status_field("ieee80211ax") != "1": raise Exception("STATUS did not indicate ieee80211ax=1") if hapd.get_status_field("ieee80211be") != "1": raise Exception("STATUS did not indicate ieee80211be=1") dev[0].set("sae_pwe", "1") dev[0].set("sae_groups", "") freq = 5950 + channel * 5 bw = _6ghz_op_class_to_bw(op_class) dev[0].connect(ssid, key_mgmt="SAE", psk=passphrase, ieee80211w="2", scan_freq=str(freq)) hapd.wait_sta() eht_verify_status(dev[0], hapd, freq, bw) eht_verify_wifi_version(dev[0]) sta = hapd.get_sta(dev[0].own_addr()) if 'supp_op_classes' not in sta: raise Exception("supp_op_classes not indicated") supp_op_classes = binascii.unhexlify(sta['supp_op_classes']) if op_class not in supp_op_classes: raise Exception("STA did not indicate support for opclass %d" % op_class) hwsim_utils.test_connectivity(dev[0], hapd) dev[0].request("DISCONNECT") dev[0].wait_disconnected() hapd.wait_sta_disconnect() hapd.disable() finally: dev[0].set("sae_pwe", "0") dev[0].cmd_execute(['iw', 'reg', 'set', '00']) wait_regdom_changes(dev[0]) def test_eht_6ghz_20mhz(dev, apdev): """EHT with 20 MHz channel width on 6 GHz""" _test_eht_6ghz(dev, apdev, 5, 131, 5) def test_eht_6ghz_40mhz(dev, apdev): """EHT with 40 MHz channel width on 6 GHz""" _test_eht_6ghz(dev, apdev, 5, 132, 3) def test_eht_6ghz_80mhz(dev, apdev): """EHT with 80 MHz channel width on 6 GHz""" _test_eht_6ghz(dev, apdev, 5, 133, 7) def test_eht_6ghz_160mhz(dev, apdev): """EHT with 160 MHz channel width on 6 GHz""" _test_eht_6ghz(dev, apdev, 5, 134, 15) def test_eht_6ghz_320mhz(dev, apdev): """EHT with 320 MHz channel width on 6 GHz""" _test_eht_6ghz(dev, apdev, 5, 137, 31) def test_eht_6ghz_320mhz_2(dev, apdev): """EHT with 320 MHz channel width on 6 GHz center 63""" _test_eht_6ghz(dev, apdev, 37, 137, 63) def test_eht_6ghz_320mhz_3(dev, apdev): """EHT with 320 MHz channel width on 6 GHz center 31 primary 37""" _test_eht_6ghz(dev, apdev, 37, 137, 31) def check_anqp(dev, bssid): if "OK" not in dev.request("ANQP_GET " + bssid + " 258"): raise Exception("ANQP_GET command failed") ev = dev.wait_event(["GAS-QUERY-START"], timeout=5) if ev is None: raise Exception("GAS query start timed out") ev = dev.wait_event(["GAS-QUERY-DONE"], timeout=10) if ev is None: raise Exception("GAS query timed out") ev = dev.wait_event(["RX-ANQP"], timeout=1) if ev is None or "Venue Name" not in ev: raise Exception("Did not receive Venue Name") ev = dev.wait_event(["ANQP-QUERY-DONE"], timeout=10) if ev is None: raise Exception("ANQP-QUERY-DONE event not seen") if "result=SUCCESS" not in ev: raise Exception("Unexpected result: " + ev) def test_eht_mld_gas(dev, apdev): """GAS/ANQP during MLO association""" params = hs20_ap_params() bssid = apdev[0]['bssid'] params['hessid'] = bssid params['channel'] = "11" hapd = hostapd.add_ap(apdev[0], params) 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): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) wpas.scan_for_bss(bssid, freq="2462") ssid = "owe_two_link" params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") params['interworking'] = "1" params['venue_group'] = "7" params['venue_type'] = "1" params['venue_name'] = "eng:Example venue" hapd0 = eht_mld_enable_ap(hapd0_iface, params) bssid0 = hapd0.own_addr() params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd0_iface, params) bssid1 = hapd1.own_addr() wpas.scan_for_bss(bssid0, freq="2412") wpas.scan_for_bss(bssid1, freq="2437") wpas.connect(ssid, scan_freq="2412 2437", key_mgmt="OWE", ieee80211w="2") hapd0.wait_sta() hapd1.wait_sta() eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) check_anqp(wpas, bssid) check_anqp(wpas, bssid0) check_anqp(wpas, bssid1) def test_eht_mld_dpp_responder_while_assoc(dev, apdev): """DPP responder while ML associated""" check_dpp_capab(dev[0]) 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): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) check_dpp_capab(wpas) ssid = "owe_two_link" params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") hapd0 = eht_mld_enable_ap(hapd0_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd0_iface, params) wpas.connect(ssid, scan_freq="2412 2437", key_mgmt="OWE", ieee80211w="2") hapd0.wait_sta() hapd1.wait_sta() eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) id = wpas.dpp_bootstrap_gen(chan="81/11", mac=True) uri = wpas.request("DPP_BOOTSTRAP_GET_URI %d" % id) wpas.dpp_listen(2462) dev[0].dpp_auth_init(uri=uri) wait_auth_success(dev[0], wpas) def _eht_mld_disconnect(dev, apdev, disassoc=True): with HWSimRadio(use_mlo=True) as (hapd0_radio, hapd0_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap_owe_two_link" params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") hapd0 = eht_mld_enable_ap(hapd0_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd0_iface, params) wpas.connect(ssid, scan_freq="2412 2437", key_mgmt="OWE", ieee80211w="2") hapd0.wait_sta() hapd1.wait_sta() eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) cmd = "DISASSOCIATE " if disassoc else "DEAUTHENTICATE " cmd += wpas.own_addr() for i in range(0, 3): time.sleep(1) if "OK" not in hapd0.request(cmd): raise Exception("Failed to request: " + cmd) hapd0.wait_sta_disconnect() hapd1.wait_sta_disconnect() wpas.wait_disconnected(timeout=1) wpas.wait_connected(timeout=5) hapd0.wait_sta() hapd1.wait_sta() eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) def test_eht_mld_disassociate(dev, apdev): """EHT MLD with two links. Disassociate and reconnect""" _eht_mld_disconnect(dev, apdev, disassoc=True) def test_eht_mld_deauthenticate(dev, apdev): """EHT MLD with two links. Deauthenticate and reconnect""" _eht_mld_disconnect(dev, apdev, disassoc=False) def test_eht_mld_non_pref_chan(dev, apdev): """EHT MLD with one link. MBO non preferred channels""" with HWSimRadio(use_mlo=True) as (hapd0_radio, hapd0_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) # Start the first AP ssid = "mld_ap_one_link_mbo" params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") params["bss_transition"] = "1" params["mbo"] = "1" hapd0 = eht_mld_enable_ap(hapd0_iface, params) if "OK" not in wpas.request("SET non_pref_chan 81:7:200:1 81:9:100:2"): raise Exception("Failed to set non-preferred channel list") id = wpas.connect(ssid, scan_freq="2412", key_mgmt="OWE", ieee80211w="2", owe_only="1") hapd0.wait_sta() eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=1, active_links=1) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) # Validate information received from the Association Request frame addr = wpas.own_addr() sta = hapd0.get_sta(addr) logger.debug("STA: " + str(sta)) if 'non_pref_chan[0]' not in sta: raise Exception("Missing non_pref_chan[0] value (assoc)") if sta['non_pref_chan[0]'] != '81:200:1:7': raise Exception("Unexpected non_pref_chan[0] value (assoc)") if 'non_pref_chan[1]' not in sta: raise Exception("Missing non_pref_chan[1] value (assoc)") if sta['non_pref_chan[1]'] != '81:100:2:9': raise Exception("Unexpected non_pref_chan[1] value (assoc)") if 'non_pref_chan[2]' in sta: raise Exception("Unexpected non_pref_chan[2] value (assoc)") # Verify operating class if 'supp_op_classes' not in sta: raise Exception("No supp_op_classes") supp = bytearray(binascii.unhexlify(sta['supp_op_classes'])) if supp[0] != 81: raise Exception("Unexpected current operating class %d" % supp[0]) if 115 not in supp: raise Exception("Operating class 115 missing") # Validate information from WNM action if "OK" not in wpas.request("SET non_pref_chan 81:9:100:2"): raise Exception("Failed to update non-preferred channel list") time.sleep(0.1) sta = hapd0.get_sta(addr) logger.debug("STA: " + str(sta)) if 'non_pref_chan[0]' not in sta: raise Exception("Missing non_pref_chan[0] value (update 1)") if sta['non_pref_chan[0]'] != '81:100:2:9': raise Exception("Unexpected non_pref_chan[0] value (update 1)") if 'non_pref_chan[1]' in sta: raise Exception("Unexpected non_pref_chan[1] value (update 1)") # Validate information from WNM action with multiple entries if "OK" not in wpas.request("SET non_pref_chan 81:9:100:2 81:10:100:2 81:8:100:2 81:7:100:1 81:5:100:1"): raise Exception("Failed to update non-preferred channel list") time.sleep(0.1) sta = hapd0.get_sta(addr) logger.debug("STA: " + str(sta)) if 'non_pref_chan[0]' not in sta: raise Exception("Missing non_pref_chan[0] value (update 2)") if sta['non_pref_chan[0]'] != '81:100:1:7,5': raise Exception("Unexpected non_pref_chan[0] value (update 2)") if 'non_pref_chan[1]' not in sta: raise Exception("Missing non_pref_chan[1] value (update 2)") if sta['non_pref_chan[1]'] != '81:100:2:9,10,8': raise Exception("Unexpected non_pref_chan[1] value (update 2)") if 'non_pref_chan[2]' in sta: raise Exception("Unexpected non_pref_chan[2] value (update 2)") def test_eht_mld_rrm_beacon_req(dev, apdev): """EHT MLD with one link. RRM beacon request""" 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): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) # Start the first AP and connect ssid = "mld_ap_one_link_rrm1" params = eht_mld_ap_wpa2_params(ssid, key_mgmt="OWE", mfp="2") params["bss_transition"] = "1" params["mbo"] = "1" params["rrm_beacon_report"] = "1" hapd0 = eht_mld_enable_ap(hapd0_iface, params) wpas.connect(ssid, scan_freq="2412", key_mgmt="OWE", ieee80211w="2", owe_only="1") hapd0.wait_sta() eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=1, active_links=1) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) # Start the second AP other_ssid = "other" params = eht_mld_ap_wpa2_params(other_ssid, key_mgmt="OWE", mfp="2") params["channel"] = '6' hapd1 = eht_mld_enable_ap(hapd1_iface, params) # Issue a beacon request for the second AP addr = wpas.own_addr() req = build_beacon_request(mode=1, chan=6, duration=50) # Send the request with SSID, Detail, Last Beacon Report Indication, and # Extended Request subelements. The Extended Request elements includes # the Multi-Link element ID. run_req_beacon(hapd0, addr, req + "0000" + "020101" + "a40101" + "0b02ff6b") ev = hapd0.wait_event(["BEACON-RESP-RX"], timeout=3) if ev is None: raise Exception("Beacon report response not received") fields = ev.split(' ') report = BeaconReport(binascii.unhexlify(fields[4])) logger.info("Received beacon report: " + str(report)) if report.bssid_str != hapd1.own_addr() or report.opclass != 81 or \ report.channel != 6: raise Exception("Incorrect bssid/op class/channel for hapd1") if not report.last_indication: raise Exception("Last Beacon Report Indication subelement missing") def test_eht_mld_legacy_stas(dev, apdev): """EHT AP MLD and multiple non-MLD STAs""" for i in range(3): check_sae_capab(dev[i]) with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface): password = 'qwertyuiop' ssid = "ap_mld_sae" params = eht_mld_ap_wpa2_params(ssid, password, key_mgmt="SAE SAE-EXT-KEY", mfp="2", pwe='2') params['rsn_pairwise'] = "CCMP GCMP-256" params['sae_groups'] = "19 20" hapd0 = eht_mld_enable_ap(hapd_iface, params) for i in range(3): dev[i].set("sae_groups", "") dev[i].connect(ssid, sae_password=password, scan_freq="2412", key_mgmt="SAE", ieee80211w="2", disable_eht="1") hapd0.wait_sta() hapd0.wait_sta() hapd0.wait_sta() aid = [] for i in range(3): aid.append(int(hapd0.get_sta(dev[i].own_addr())['aid'])) traffic_test(dev[i], hapd0) logger.info("Assigned AIDs: " + str(aid)) if len(set(aid)) != 3: raise Exception("AP did not assign unique AID to each STA") def test_eht_mld_and_mlds(dev, apdev): """EHT AP MLD and multiple non-AP MLDs""" check_sae_capab(dev[0]) with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio2, wpas_iface2): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) check_sae_capab(wpas) wpas2 = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas2.interface_add(wpas_iface2) check_sae_capab(wpas2) password = 'qwertyuiop' ssid = "ap_mld_sae" params = eht_mld_ap_wpa2_params(ssid, password, key_mgmt="SAE SAE-EXT-KEY", mfp="2", pwe='2') params['rsn_pairwise'] = "CCMP GCMP-256" params['sae_groups'] = "19 20" hapd0 = eht_mld_enable_ap(hapd_iface, params) wpas.set("sae_pwe", "1") wpas.connect(ssid, sae_password=password, scan_freq="2412", key_mgmt="SAE-EXT-KEY", ieee80211w="2") wpas2.set("sae_pwe", "1") wpas2.connect(ssid, sae_password=password, scan_freq="2412", key_mgmt="SAE-EXT-KEY", ieee80211w="2") hapd0.wait_sta() hapd0.wait_sta() aid = [] aid.append(int(hapd0.get_sta(wpas.own_addr())['aid'])) traffic_test(wpas, hapd0) aid.append(int(hapd0.get_sta(wpas2.own_addr())['aid'])) traffic_test(wpas2, hapd0) logger.info("Assigned AIDs: " + str(aid)) if len(set(aid)) != 2: raise Exception("AP MLD did not assign unique AID to each non-AP MLD") def mlo_perform_csa(hapd, command, freq, dev): match_str = "freq=" + str(freq) hapd.request(command) ev = hapd.wait_event(["CTRL-EVENT-STARTED-CHANNEL-SWITCH"], timeout=10) if ev is None: raise Exception("Channel switch start event not seen") if match_str not in ev: raise Exception("Unexpected channel in CS started") ev = hapd.wait_event(["CTRL-EVENT-CHANNEL-SWITCH"], timeout=10) if ev is None: raise Exception("Channel switch completion event not seen") if match_str not in ev: raise Exception("Unexpected channel in CS completed") ev = hapd.wait_event(["AP-CSA-FINISHED"], timeout=10) if ev is None: raise Exception("CSA finished event timed out") if match_str not in ev: raise Exception("Unexpected channel in CSA finished event") ev = dev.wait_event(["CTRL-EVENT-LINK-CHANNEL-SWITCH"], timeout=10) if ev is None: raise Exception("Non-AP MLD did not report CS") if match_str not in ev: raise Exception("Unexpected channel in CS event from non-AP MLD") time.sleep(0.5) def test_eht_mlo_csa(dev, apdev): """EHT MLD AP connected to non-AP MLD. Seamless channel switch""" csa_supported(dev[0]) with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) ssid = "mld_ap" passphrase = 'qwertyuiop' params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE", mfp="2", pwe='1') hapd0 = eht_mld_enable_ap(hapd_iface, params) params['channel'] = '6' hapd1 = eht_mld_enable_ap(hapd_iface, params) wpas.set("sae_pwe", "1") wpas.connect(ssid, sae_password=passphrase, scan_freq="2412 2437", key_mgmt="SAE", ieee80211w="2") eht_verify_status(wpas, hapd0, 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas) traffic_test(wpas, hapd0) logger.info("Perform CSA on 1st link") mlo_perform_csa(hapd0, "CHAN_SWITCH 5 2462 ht he eht blocktx", 2462, wpas) logger.info("Test traffic after 1st link CSA completes") traffic_test(wpas, hapd0) logger.info("Perform CSA on 2nd link") mlo_perform_csa(hapd1, "CHAN_SWITCH 5 2412 ht he eht blocktx", 2412, wpas) logger.info("Test traffic after 2nd link CSA completes") traffic_test(wpas, hapd1) logger.info("Perform CSA on 2nd link and bring it back to original channel") mlo_perform_csa(hapd1, "CHAN_SWITCH 5 2437 ht he eht blocktx", 2437, wpas) logger.info("Test traffic again after 2nd link CSA completes") traffic_test(wpas, hapd1) logger.info("Perform CSA on 1st link and bring it back to original channel") mlo_perform_csa(hapd0, "CHAN_SWITCH 5 2412 ht he eht blocktx", 2412, wpas) logger.info("Test traffic again after 1st link CSA completes") traffic_test(wpas, hapd0) def create_base_conf_file(iface, channel, prefix='hostapd-', hw_mode='g', op_class=None): # Create configuration file and add phy characteristics fd, fname = tempfile.mkstemp(dir='/tmp', prefix=prefix + iface + "-chan-" + str(channel) + "-") f = os.fdopen(fd, 'w') f.write("driver=nl80211\n") f.write("hw_mode=" + str(hw_mode) + "\n") f.write("ieee80211n=1\n") if hw_mode == 'a' and \ (op_class is None or \ op_class not in [131, 132, 133, 134, 135, 136, 137]): f.write("ieee80211ac=1\n") f.write("ieee80211ax=1\n") f.write("ieee80211be=1\n") f.write("channel=" + str(channel) + "\n") return f, fname def append_bss_conf_to_file(f, ifname, params, first=False): # Add BSS specific characteristics config = "bss" if first: config = "interface" f.write("\n" + config + "=%s\n" % ifname) for k, v in list(params.items()): f.write("{}={}\n".format(k,v)) f.write("mld_ap=1\n") def dump_config(fname): with open(fname, 'r') as f: cfg = f.read() logger.debug("hostapd config: " + str(fname) + "\n" + cfg) def get_config(iface, count, ssid, passphrase, channel, bssid_regex, rnr=False, debug=False): f, fname = create_base_conf_file(iface, channel=channel) hapds = [] for i in range(count): if i == 0: ifname = iface else: ifname = iface + "-" + str(i) set_ssid = ssid + str(i) set_passphrase = passphrase + str(i) params = hostapd.wpa2_params(ssid=set_ssid, passphrase=set_passphrase, wpa_key_mgmt="SAE", ieee80211w="2") params['sae_pwe'] = "2" params['group_mgmt_cipher'] = "AES-128-CMAC" params['beacon_prot'] = "1" params["ctrl_interface"] = "/var/run/hostapd/chan_" + str(channel) params["bssid"] = bssid_regex % (i + 1) if rnr: params["rnr"] = "1" append_bss_conf_to_file(f, ifname, params, first=(i == 0)) hapds.append([ifname, params["ctrl_interface"], i]) f.close() if debug: dump_config(fname) return fname, hapds def start_ap(prefix, configs): pid = prefix + ".hostapd.pid" configs = configs.split() cmd = ['../../hostapd/hostapd', '-ddKtB', '-P', pid, '-f', prefix + ".hostapd-log"] cmd = cmd + configs logger.info("Starting APs") res = subprocess.check_call(cmd) if res != 0: raise Exception("Could not start hostapd: %s" % str(res)) # Wait for hostapd to complete initialization and daemonize. time.sleep(2) for i in range(20): if os.path.exists(pid): break time.sleep(0.2) if not os.path.exists(pid): raise Exception("hostapd did not create PID file.") def get_mld_devs(hapd_iface, count, prefix, rnr=False): fname1, hapds1 = get_config(hapd_iface, count=count, ssid="mld-", passphrase="qwertyuiop-", channel=1, bssid_regex="02:00:00:00:07:%02x", rnr=rnr, debug=True) fname2, hapds2 = get_config(hapd_iface, count=count, ssid="mld-", passphrase="qwertyuiop-", channel=6, bssid_regex="02:00:00:00:08:%02x", rnr=rnr, debug=True) start_ap(prefix, fname1 + " " + fname2) hapd_mld1_link0 = hostapd.Hostapd(ifname=hapds1[0][0], ctrl=hapds1[0][1], bssidx=hapds1[0][2]) hapd_mld1_link1 = hostapd.Hostapd(ifname=hapds2[0][0], ctrl=hapds2[0][1], bssidx=hapds2[0][2]) hapd_mld2_link0 = hostapd.Hostapd(ifname=hapds1[1][0], ctrl=hapds1[1][1], bssidx=hapds1[1][2]) hapd_mld2_link1 = hostapd.Hostapd(ifname=hapds2[1][0], ctrl=hapds2[1][1], bssidx=hapds2[1][2]) if not hapd_mld1_link0.ping(): raise Exception("Could not ping hostapd") if not hapd_mld1_link1.ping(): raise Exception("Could not ping hostapd") if not hapd_mld2_link0.ping(): raise Exception("Could not ping hostapd") if not hapd_mld2_link1.ping(): raise Exception("Could not ping hostapd") os.remove(fname1) os.remove(fname2) return [hapd_mld1_link0, hapd_mld1_link1, hapd_mld2_link0, hapd_mld2_link1] def stop_mld_devs(hapds, pid): pid = pid + ".hostapd.pid" if "OK" not in hapds[0].request("TERMINATE"): raise Exception("Failed to terminate hostapd process") ev = hapds[0].wait_event(["CTRL-EVENT-TERMINATING"], timeout=15) if ev is None: raise Exception("CTRL-EVENT-TERMINATING not seen") time.sleep(0.5) for i in range(30): time.sleep(0.1) if not os.path.exists(pid): break if os.path.exists(pid): raise Exception("PID file exits after process termination") def eht_parse_rnr(bss, rnr=False, exp_bssid=None): partner_rnr_pattern = re.compile(".*ap_info.*, mld ID=0, link ID=", re.MULTILINE) ml_pattern = re.compile(".*multi-link:.*, MLD addr=.*", re.MULTILINE) if partner_rnr_pattern.search(bss) is None: raise Exception("RNR element not found for first link of first MLD") if ml_pattern.search(bss) is None: raise Exception("ML element not found for first link of first MLD") if not rnr: return coloc_rnr_pattern = re.compile(".*ap_info.*, mld ID=255, link ID=..", re.MULTILINE) if coloc_rnr_pattern.search(bss) is None: raise Exception("RNR element not found for co-located BSS") line = coloc_rnr_pattern.search(bss).group() if line.count('bssid') > 1: raise Exception("More than one BSS found for co-located RNR") # Get the BSSID carried in the RNR index = line.rindex('bssid') bssid = line[index+len('bssid')+1:].split(',')[0] # Get the MLD ID carried in the RNR index = line.rindex('link ID') link_id = line[index+len('link ID')+1:].split(',')[0] if link_id != "15": raise Exception("Unexpected link ID for co-located BSS which is not own partner") if bssid != exp_bssid: raise Exception("Unexpected BSSID for co-located BSS") def eht_mld_cohosted_discovery(dev, apdev, params, rnr=False): with HWSimRadio(use_mlo=True, n_channels=2) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) hapds = get_mld_devs(hapd_iface=hapd_iface, count=2, prefix=params['prefix'], rnr=rnr) # Only scan link 0 res = wpas.request("SCAN freq=2412") if "FAIL" in res: raise Exception("Failed to start scan") ev = wpas.wait_event(["CTRL-EVENT-SCAN-STARTED"]) if ev is None: raise Exception("Scan did not start") ev = wpas.wait_event(["CTRL-EVENT-SCAN-RESULTS"]) if ev is None: raise Exception("Scan did not complete") logger.info("Scan done") bss = wpas.request("BSS " + hapds[0].own_addr()) logger.info("BSS 0_0: " + str(bss)) eht_parse_rnr(bss, rnr, hapds[2].own_addr()) bss = wpas.request("BSS " + hapds[2].own_addr()) logger.info("BSS 1_0: " + str(bss)) eht_parse_rnr(bss, rnr, hapds[0].own_addr()) stop_mld_devs(hapds, params['prefix']) def test_eht_mld_cohosted_discovery(dev, apdev, params): """EHT 2 AP MLDs discovery""" eht_mld_cohosted_discovery(dev, apdev, params) def test_eht_mld_cohosted_discovery_with_rnr(dev, apdev, params): """EHT 2 AP MLDs discovery (with co-location RNR)""" eht_mld_cohosted_discovery(dev, apdev, params, rnr=True) def test_eht_mld_cohosted_connectivity(dev, apdev, params): """EHT 2 AP MLDs with 2 MLD clients connection""" check_sae_capab(dev[0]) with HWSimRadio(use_mlo=True, n_channels=2) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio1, wpas_iface1): wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas.interface_add(wpas_iface) check_sae_capab(wpas) wpas1 = WpaSupplicant(global_iface='/tmp/wpas-wlan5') wpas1.interface_add(wpas_iface1) check_sae_capab(wpas1) hapds = get_mld_devs(hapd_iface=hapd_iface, count=2, prefix=params['prefix'], rnr=False) passphrase = "qwertyuiop-" ssid = "mld-" # Connect one client to first AP MLD and verify traffic on both links wpas.set("sae_pwe", "1") wpas.connect(ssid + "0", sae_password=passphrase+"0", scan_freq="2412", key_mgmt="SAE", ieee80211w="2") eht_verify_status(wpas, hapds[0], 2412, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas) traffic_test(wpas, hapds[0]) traffic_test(wpas, hapds[1]) # Connect another client to second AP MLD and verify traffic on both # links wpas1.set("sae_pwe", "1") wpas1.connect(ssid + "1", sae_password=passphrase+"1", scan_freq="2437", key_mgmt="SAE", ieee80211w="2") eht_verify_status(wpas1, hapds[3], 2437, 20, is_ht=True, mld=True, valid_links=3, active_links=3) eht_verify_wifi_version(wpas1) traffic_test(wpas1, hapds[3]) traffic_test(wpas1, hapds[2]) stop_mld_devs(hapds, params['prefix']) def test_eht_mlo_color_change(dev, apdev): """AP MLD and Color Change Announcement""" with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \ HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface): ssid = "mld_ap" passphrase = 'qwertyuiop' params = eht_mld_ap_wpa2_params(ssid, passphrase, key_mgmt="SAE", mfp="2", pwe='1') params['he_bss_color'] = '42' hapd0 = eht_mld_enable_ap(hapd_iface, params) params['channel'] = '6' params['he_bss_color'] = '24' hapd1 = eht_mld_enable_ap(hapd_iface, params) logger.info("Perform CCA on 1st link") if "OK" not in hapd0.request("COLOR_CHANGE 10"): raise Exception("COLOR_CHANGE failed") time.sleep(1.5) color = hapd0.get_status_field("he_bss_color") if color != "10": raise Exception("Expected current he_bss_color to be 10; was " + color) logger.info("Perform CCA on 1st link again") if "OK" not in hapd0.request("COLOR_CHANGE 60"): raise Exception("COLOR_CHANGE failed") time.sleep(1.5) color = hapd0.get_status_field("he_bss_color") if color != "60": raise Exception("Expected current he_bss_color to be 60; was " + color) logger.info("Perform CCA on 2nd link") if "OK" not in hapd1.request("COLOR_CHANGE 25"): raise Exception("COLOR_CHANGE failed") time.sleep(1.5) color = hapd1.get_status_field("he_bss_color") if color != "25": raise Exception("Expected current he_bss_color to be 25; was " + color) logger.info("Perform CCA on 2nd link again") if "OK" not in hapd1.request("COLOR_CHANGE 5"): raise Exception("COLOR_CHANGE failed") time.sleep(1.5) color = hapd1.get_status_field("he_bss_color") if color != "5": raise Exception("Expected current he_bss_color to be 5; was " + color) hapd0.dump_monitor() hapd1.dump_monitor()