# Multiple BSSID and enhanced multi-BSS advertisements (EMA) # Copyright (c) 2019, The Linux Foundation # Copyright (c) 2022, Qualcomm Innovation Center, Inc # # This software may be distributed under the terms of the BSD license. # See README for more details. import os, subprocess, time, logging, tempfile logger = logging.getLogger() import hwsim_utils, hostapd from wpasupplicant import WpaSupplicant from utils import * def mbssid_create_cfg_file(apdev, params, mbssid=1): hapd = hostapd.add_ap(apdev[0], {"ssid": "open"}) status = hapd.get_driver_status() if "capa.mbssid_max_interfaces" not in status or \ int(status["capa.mbssid_max_interfaces"]) < 8: raise HwsimSkip("MBSSID not supported") if mbssid == 2 and \ ("capa.ema_max_periodicity" not in status or \ int(status["capa.ema_max_periodicity"]) < 3): raise HwsimSkip("EMA not supported") hapd.disable() hglobal = hostapd.HostapdGlobal() hglobal.remove(apdev[0]['ifname']) # Create configuration file and add phy characteristics fd, fname = tempfile.mkstemp(dir='/tmp', prefix='hostapd-' + get_phy(apdev[0])) f = os.fdopen(fd, 'w') f.write("driver=nl80211\n") f.write("hw_mode=g\n") f.write("channel=1\n") f.write("ieee80211n=1\n") f.write("vht_oper_chwidth=0\n") f.write("ieee80211ac=1\n") f.write("he_oper_chwidth=0\n") f.write("ieee80211ax=1\n") f.write("mbssid=%d\n\n" % mbssid) if isinstance(apdev[0], dict): ifname = apdev[0]['ifname'] else: ifname = apdev return (f, fname, ifname) def mbssid_write_bss_params(f, ifname, idx, params=None): # Add BSS specific characteristics fields = ["wpa", "wpa_pairwise", "rsn_pairwise", "wpa_passphrase", "wpa_key_mgmt", "ieee80211w", "sae_pwe" ] if idx == 0: f.write("interface=%s\n" % ifname) else: f.write("\nbss=%s\n" % (ifname + '-' + str(idx))) f.write("ctrl_interface=/var/run/hostapd\n") f.write("ssid=bss-%s\n" % idx) f.write("bridge=br-lan\n") f.write("bssid=00:03:7f:12:2a:%02x\n" % idx) f.write("preamble=1\n") f.write("auth_algs=1\n") if params is None: return for field in fields: if field in params: f.write("%s=%s\n" % (field, params[field])) def mbssid_start_ap(dev, apdev, params, fname, ifname, sta_params): pid = params['prefix'] + ".hostapd.pid" cmd = ['../../hostapd/hostapd', '-dtSB', '-dt', '-P', pid, '-f', params['prefix'] + ".hostapd-log", fname] logger.info("Starting APs for " + ifname) 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. for i in range(10): if os.path.exists(pid): break time.sleep(0.2) if not os.path.exists(pid): # For now, assume MBSSID is not supported in the kernel. raise Exception("hostapd did not create PID file") hapd = hostapd.Hostapd(apdev[0]['ifname']) hapd.ping() os.remove(fname); # Allow enough time to pass for a Beacon frame to be captured. time.sleep(0.1) dev[0].connect("bss-0", **sta_params) sta = hapd.get_sta(dev[0].own_addr()) if "[HE]" not in sta['flags']: raise Exception("Missing STA flag: HE") dev[0].request("DISCONNECT") if "OK" not in hapd.request("TERMINATE"): raise Exception("Failed to terminate hostapd process") ev = hapd.wait_event(["CTRL-EVENT-TERMINATING"], timeout=15) if ev is None: raise Exception("CTRL-EVENT-TERMINATING not seen") 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 test_he_ap_mbssid_open(dev, apdev, params): """HE AP MBSSID all open""" f, fname, ifname = mbssid_create_cfg_file(apdev, params) for idx in range(0, 4): mbssid_write_bss_params(f, ifname, idx) f.close() try: sta_params = {"key_mgmt": "NONE", "scan_freq": "2412"} mbssid_start_ap(dev, apdev, params, fname, ifname, sta_params) finally: subprocess.call(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'address', apdev[0]['bssid']]) def test_he_ap_mbssid_same_security(dev, apdev, params): """HE AP MBSSID all SAE""" f, fname, ifname = mbssid_create_cfg_file(apdev, params) sae_params = {"wpa": "2", "wpa_passphrase": "12345678", "wpa_pairwise": "CCMP", "wpa_key_mgmt": "SAE", "sae_pwe": "1", "ieee80211w": "2"} for idx in range(0, 2): mbssid_write_bss_params(f, ifname, idx, sae_params) f.close() try: dev[0].set("sae_pwe", "1") dev[0].set("sae_groups", "") sta_params = {"psk": "12345678", "key_mgmt": "SAE", "ieee80211w": "2", "pairwise": "CCMP", "group": "CCMP", "scan_freq": "2412"} mbssid_start_ap(dev, apdev, params, fname, ifname, sta_params) finally: dev[0].set("sae_pwe", "0") subprocess.call(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'address', apdev[0]['bssid']]) def test_he_ap_mbssid_mixed_security1(dev, apdev, params): """HE AP MBSSID with mixed security (STA SAE)""" f, fname, ifname = mbssid_create_cfg_file(apdev, params) psk_params = {"wpa": "2", "wpa_passphrase": "12345678", "wpa_pairwise": "CCMP", "wpa_key_mgmt": "WPA-PSK"} owe_params = {"wpa": "2", "wpa_pairwise": "CCMP", "wpa_key_mgmt": "OWE"} sae_params = {"wpa": "2", "wpa_passphrase": "12345678", "wpa_pairwise": "CCMP", "wpa_key_mgmt": "SAE", "sae_pwe": "1", "ieee80211w": "2"} mbssid_write_bss_params(f, ifname, 0, sae_params) mbssid_write_bss_params(f, ifname, 1, psk_params) mbssid_write_bss_params(f, ifname, 2, owe_params) mbssid_write_bss_params(f, ifname, 3) mbssid_write_bss_params(f, ifname, 4, psk_params) mbssid_write_bss_params(f, ifname, 5, sae_params) mbssid_write_bss_params(f, ifname, 6, owe_params) mbssid_write_bss_params(f, ifname, 7) f.close() try: dev[0].set("sae_pwe", "1") dev[0].set("sae_groups", "") sta_params = {"psk": "12345678", "key_mgmt": "SAE", "ieee80211w": "2", "pairwise": "CCMP", "group": "CCMP", "scan_freq": "2412"} mbssid_start_ap(dev, apdev, params, fname, ifname, sta_params) finally: dev[0].set("sae_pwe", "0") subprocess.call(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'address', apdev[0]['bssid']]) def test_he_ap_mbssid_mixed_security2(dev, apdev, params): """HE AP MBSSID with mixed security (STA open)""" f, fname, ifname = mbssid_create_cfg_file(apdev, params) psk_params = {"wpa": "2", "wpa_passphrase": "12345678", "wpa_pairwise": "CCMP", "wpa_key_mgmt": "WPA-PSK"} owe_params = {"wpa": "2", "wpa_pairwise": "CCMP", "wpa_key_mgmt": "OWE"} sae_params = {"wpa": "2", "wpa_passphrase": "12345678", "wpa_pairwise": "CCMP", "wpa_key_mgmt": "SAE", "sae_pwe": "1", "ieee80211w": "2"} mbssid_write_bss_params(f, ifname, 0) mbssid_write_bss_params(f, ifname, 1, psk_params) mbssid_write_bss_params(f, ifname, 2, owe_params) mbssid_write_bss_params(f, ifname, 3, sae_params) mbssid_write_bss_params(f, ifname, 4) mbssid_write_bss_params(f, ifname, 5, psk_params) mbssid_write_bss_params(f, ifname, 6, sae_params) mbssid_write_bss_params(f, ifname, 7, owe_params) f.close() try: sta_params = {"key_mgmt": "NONE", "scan_freq": "2412"} mbssid_start_ap(dev, apdev, params, fname, ifname, sta_params) finally: subprocess.call(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'address', apdev[0]['bssid']]) def test_he_ap_ema(dev, apdev, params): """HE EMA AP""" f, fname, ifname = mbssid_create_cfg_file(apdev, params, 2) sae_params = {"wpa": "2", "wpa_passphrase": "12345678", "wpa_pairwise": "CCMP", "wpa_key_mgmt": "SAE", "sae_pwe": "1", "ieee80211w": "2"} for idx in range(0, 8): mbssid_write_bss_params(f, ifname, idx, sae_params) f.close() try: dev[0].set("sae_pwe", "1") dev[0].set("sae_groups", "") sta_params = {"psk": "12345678", "key_mgmt": "SAE", "ieee80211w": "2", "pairwise": "CCMP", "group": "CCMP", "scan_freq": "2412"} mbssid_start_ap(dev, apdev, params, fname, ifname, sta_params) finally: dev[0].set("sae_pwe", "0") subprocess.call(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'address', apdev[0]['bssid']])