# WPA2-Personal OCV tests # Copyright (c) 2018, Mathy Vanhoef # # This software may be distributed under the terms of the BSD license. # See README for more details from remotehost import remote_compatible import binascii, struct import logging, time logger = logging.getLogger() import hostapd from wpasupplicant import WpaSupplicant import hwsim_utils from utils import HwsimSkip from test_ap_ht import set_world_reg from test_ap_psk import parse_eapol, build_eapol, pmk_to_ptk, eapol_key_mic, recv_eapol, send_eapol, reply_eapol, hapd_connected, build_eapol_key_3_4, aes_wrap, pad_key_data #TODO: Refuse setting up AP with OCV but without MFP support #TODO: Refuse to connect to AP that advertises OCV but not MFP def make_ocikde(op_class, channel, seg1_idx): WLAN_EID_VENDOR_SPECIFIC = 221 RSN_KEY_DATA_OCI = b"\x00\x0f\xac\x0d" data = RSN_KEY_DATA_OCI + struct.pack("Q', replay_counter) msg['rsn_key_nonce'] = binascii.unhexlify('0000000000000000000000000000000000000000000000000000000000000000') msg['rsn_key_iv'] = binascii.unhexlify('00000000000000000000000000000000') msg['rsn_key_rsc'] = binascii.unhexlify('0000000000000000') msg['rsn_key_id'] = binascii.unhexlify('0000000000000000') msg['rsn_key_data_len'] = len(key_data) msg['rsn_key_data'] = key_data eapol_key_mic(kck, msg) return msg def build_eapol_key_2_2(kck, key_data, replay_counter=3, key_info=0x0302, extra_len=0, descr_type=2, key_len=16): return build_eapol_key_1_2(kck, key_data, replay_counter, key_info, extra_len, descr_type, key_len) @remote_compatible def test_wpa2_ocv(dev, apdev): """OCV on 2.4 GHz""" params = {"channel": "1", "ieee80211w": "2", "ocv": "1"} hapd, ssid, passphrase = ocv_setup_ap(apdev[0], params) for ocv in range(2): dev[0].connect(ssid, psk=passphrase, scan_freq="2412", ocv=str(ocv), ieee80211w="1") dev[0].request("REMOVE_NETWORK all") dev[0].wait_disconnected() @remote_compatible def test_wpa2_ocv_5ghz(dev, apdev): """OCV on 5 GHz""" try: run_wpa2_ocv_5ghz(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].flush_scan_cache() def run_wpa2_ocv_5ghz(dev, apdev): params = {"hw_mode": "a", "channel": "40", "ieee80211w": "2", "country_code": "US", "ocv": "1"} hapd, ssid, passphrase = ocv_setup_ap(apdev[0], params) for ocv in range(2): dev[0].connect(ssid, psk=passphrase, scan_freq="5200", ocv=str(ocv), ieee80211w="1") dev[0].wait_regdom(country_ie=True) dev[0].request("REMOVE_NETWORK all") dev[0].wait_disconnected() @remote_compatible def test_wpa2_ocv_ht20(dev, apdev): """OCV with HT20 channel""" params = {"channel": "6", "ieee80211n": "1", "ieee80211w": "1", "ocv": "1"} hapd, ssid, passphrase = ocv_setup_ap(apdev[0], params) for ocv in range(2): dev[0].connect(ssid, psk=passphrase, scan_freq="2437", ocv=str(ocv), ieee80211w="1", disable_ht="1") dev[1].connect(ssid, psk=passphrase, scan_freq="2437", ocv=str(ocv), ieee80211w="1") dev[0].request("REMOVE_NETWORK all") dev[1].request("REMOVE_NETWORK all") dev[0].wait_disconnected() dev[1].wait_disconnected() @remote_compatible def test_wpa2_ocv_ht40(dev, apdev): """OCV with HT40 channel""" try: run_wpa2_ocv_ht40(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].flush_scan_cache() dev[1].flush_scan_cache() def run_wpa2_ocv_ht40(dev, apdev): for channel, capab, freq, mode in [("6", "[HT40-]", "2437", "g"), ("6", "[HT40+]", "2437", "g"), ("40", "[HT40-]", "5200", "a"), ("36", "[HT40+]", "5180", "a")]: params = {"hw_mode": mode, "channel": channel, "country_code": "US", "ieee80211n": "1", "ht_capab": capab, "ieee80211w": "1", "ocv": "1"} hapd, ssid, passphrase = ocv_setup_ap(apdev[0], params) dev[0].flush_scan_cache() dev[1].flush_scan_cache() for ocv in range(2): dev[0].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1", disable_ht="1") dev[1].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1") dev[0].wait_regdom(country_ie=True) dev[0].request("REMOVE_NETWORK all") dev[1].request("REMOVE_NETWORK all") dev[0].wait_disconnected() dev[1].wait_disconnected() hapd.disable() @remote_compatible def test_wpa2_ocv_vht40(dev, apdev): """OCV with VHT40 channel""" try: run_wpa2_ocv_vht40(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].flush_scan_cache() dev[1].flush_scan_cache() dev[2].flush_scan_cache() def run_wpa2_ocv_vht40(dev, apdev): for channel, capab, freq in [("40", "[HT40-]", "5200"), ("36", "[HT40+]", "5180")]: params = {"hw_mode": "a", "channel": channel, "country_code": "US", "ht_capab": capab, "ieee80211n": "1", "ieee80211ac": "1", "vht_oper_chwidth": "0", "vht_oper_centr_freq_seg0_idx": "38", "ieee80211w": "1", "ocv": "1"} hapd, ssid, passphrase = ocv_setup_ap(apdev[0], params) dev[0].flush_scan_cache() dev[1].flush_scan_cache() dev[2].flush_scan_cache() for ocv in range(2): dev[0].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1", disable_ht="1") dev[1].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1", disable_vht="1") dev[2].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1") dev[0].wait_regdom(country_ie=True) dev[0].request("REMOVE_NETWORK all") dev[1].request("REMOVE_NETWORK all") dev[2].request("REMOVE_NETWORK all") dev[0].wait_disconnected() dev[1].wait_disconnected() dev[2].wait_disconnected() hapd.disable() @remote_compatible def test_wpa2_ocv_vht80(dev, apdev): """OCV with VHT80 channel""" try: run_wpa2_ocv_vht80(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].flush_scan_cache() dev[1].flush_scan_cache() dev[2].flush_scan_cache() def run_wpa2_ocv_vht80(dev, apdev): for channel, capab, freq in [("40", "[HT40-]", "5200"), ("36", "[HT40+]", "5180")]: params = {"hw_mode": "a", "channel": channel, "country_code": "US", "ht_capab": capab, "ieee80211n": "1", "ieee80211ac": "1", "vht_oper_chwidth": "1", "vht_oper_centr_freq_seg0_idx": "42", "ieee80211w": "1", "ocv": "1"} hapd, ssid, passphrase = ocv_setup_ap(apdev[0], params) for ocv in range(2): dev[0].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1", disable_ht="1") dev[1].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1", disable_vht="1") dev[2].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1") dev[0].wait_regdom(country_ie=True) dev[0].request("REMOVE_NETWORK all") dev[1].request("REMOVE_NETWORK all") dev[2].request("REMOVE_NETWORK all") dev[0].wait_disconnected() dev[1].wait_disconnected() dev[2].wait_disconnected() hapd.disable() @remote_compatible def test_wpa2_ocv_vht160(dev, apdev): """OCV with VHT160 channel""" try: run_wpa2_ocv_vht160(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].flush_scan_cache() dev[1].flush_scan_cache() dev[2].flush_scan_cache() def run_wpa2_ocv_vht160(dev, apdev): for channel, capab, freq in [("100", "[HT40+]", "5500"), ("104", "[HT40-]", "5520")]: params = {"hw_mode": "a", "channel": channel, "country_code": "ZA", "ht_capab": capab, "ieee80211n": "1", "ieee80211ac": "1", "vht_oper_chwidth": "2", "vht_oper_centr_freq_seg0_idx": "114", "ieee80211w": "1", "ocv": "1"} hapd, ssid, passphrase = ocv_setup_ap(apdev[0], params) for ocv in range(2): dev[0].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1", disable_ht="1") dev[1].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1", disable_vht="1") dev[2].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1") dev[0].wait_regdom(country_ie=True) dev[0].request("REMOVE_NETWORK all") dev[1].request("REMOVE_NETWORK all") dev[2].request("REMOVE_NETWORK all") dev[0].wait_disconnected() dev[1].wait_disconnected() dev[2].wait_disconnected() hapd.disable() @remote_compatible def test_wpa2_ocv_vht80plus80(dev, apdev): """OCV with VHT80+80 channel""" try: run_wpa2_ocv_vht80plus80(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=0.5) dev[0].flush_scan_cache() dev[1].flush_scan_cache() dev[2].flush_scan_cache() def run_wpa2_ocv_vht80plus80(dev, apdev): for channel, capab, freq in [("36", "[HT40+]", "5180"), ("40", "[HT40-]", "5200")]: params = {"hw_mode": "a", "channel": channel, "country_code": "US", "ht_capab": capab, "ieee80211n": "1", "ieee80211ac": "1", "vht_oper_chwidth": "3", "vht_oper_centr_freq_seg0_idx": "42", "vht_oper_centr_freq_seg1_idx": "155", "ieee80211w": "1", "ieee80211d": "1", "ieee80211h": "1", "ocv": "1"} hapd, ssid, passphrase = ocv_setup_ap(apdev[0], params) for ocv in range(2): dev[0].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1", disable_ht="1") dev[1].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1", disable_vht="1") dev[2].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1") dev[0].wait_regdom(country_ie=True) dev[0].request("REMOVE_NETWORK all") dev[1].request("REMOVE_NETWORK all") dev[2].request("REMOVE_NETWORK all") dev[0].wait_disconnected() dev[1].wait_disconnected() dev[2].wait_disconnected() for i in range(3): dev[i].connect(ssid, psk=passphrase, scan_freq=freq, ocv=str(ocv), ieee80211w="1") if i == 0: dev[i].wait_regdom(country_ie=True) hapd.disable() for i in range(3): dev[i].request("DISCONNECT") for i in range(3): dev[i].disconnect_and_stop_scan() class APConnection: def init_params(self): # Static parameters self.ssid = "test-wpa2-ocv" self.passphrase = "qwertyuiop" self.psk = "c2c6c255af836bed1b3f2f1ded98e052f5ad618bb554e2836757b55854a0eab7" # Dynamic parameters self.hapd = None self.addr = None self.rsne = None self.kck = None self.kek = None self.msg = None self.bssid = None self.anonce = None self.snonce = None def __init__(self, apdev, dev, params): self.init_params() # By default, OCV is enabled for both the client and AP. The following # parameters can be used to disable OCV for the client or AP. ap_ocv = params.pop("ap_ocv", "1") sta_ocv = params.pop("sta_ocv", "1") freq = params.pop("freq") params.update(hostapd.wpa2_params(ssid=self.ssid, passphrase=self.passphrase)) params["wpa_pairwise_update_count"] = "10" params["ocv"] = ap_ocv try: self.hapd = hostapd.add_ap(apdev, params) except Exception as e: if "Failed to set hostapd parameter ocv" in str(e): raise HwsimSkip("OCV not supported") raise self.hapd.request("SET ext_eapol_frame_io 1") dev.request("SET ext_eapol_frame_io 1") self.bssid = apdev['bssid'] pmk = binascii.unhexlify("c2c6c255af836bed1b3f2f1ded98e052f5ad618bb554e2836757b55854a0eab7") if sta_ocv != "0": self.rsne = binascii.unhexlify("301a0100000fac040100000fac040100000fac0280400000000fac06") else: self.rsne = binascii.unhexlify("301a0100000fac040100000fac040100000fac0280000000000fac06") self.snonce = binascii.unhexlify('1111111111111111111111111111111111111111111111111111111111111111') dev.connect(self.ssid, raw_psk=self.psk, scan_freq=freq, ocv=sta_ocv, ieee80211w="1", wait_connect=False) if "country_code" in params: dev.wait_regdom(country_ie=True) self.addr = dev.p2p_interface_addr() # Wait for EAPOL-Key msg 1/4 from hostapd to determine when associated self.msg = recv_eapol(self.hapd) self.anonce = self.msg['rsn_key_nonce'] (ptk, self.kck, self.kek) = pmk_to_ptk(pmk, self.addr, self.bssid, self.snonce, self.anonce) # hapd, addr, rsne, kck, msg, anonce, snonce def test_bad_oci(self, logmsg, op_class, channel, seg1_idx): logger.debug("Bad OCI element: " + logmsg) if op_class is None: ocikde = b'' else: ocikde = make_ocikde(op_class, channel, seg1_idx) reply_eapol("2/4", self.hapd, self.addr, self.msg, 0x010a, self.snonce, self.rsne + ocikde, self.kck) self.msg = recv_eapol(self.hapd) if self.anonce != self.msg['rsn_key_nonce'] or self.msg["rsn_key_info"] != 138: raise Exception("Didn't receive retransmitted 1/4") def confirm_valid_oci(self, op_class, channel, seg1_idx): logger.debug("Valid OCI element to complete handshake") ocikde = make_ocikde(op_class, channel, seg1_idx) reply_eapol("2/4", self.hapd, self.addr, self.msg, 0x010a, self.snonce, self.rsne + ocikde, self.kck) self.msg = recv_eapol(self.hapd) if self.anonce != self.msg['rsn_key_nonce'] or self.msg["rsn_key_info"] != 5066: raise Exception("Didn't receive 3/4 in response to valid 2/4") reply_eapol("4/4", self.hapd, self.addr, self.msg, 0x030a, None, None, self.kck) hapd_connected(self.hapd) @remote_compatible def test_wpa2_ocv_ap_mismatch(dev, apdev): """OCV AP mismatch""" params = {"channel": "1", "ieee80211w": "1", "freq": "2412"} conn = APConnection(apdev[0], dev[0], params) conn.test_bad_oci("element missing", None, 0, 0) conn.test_bad_oci("wrong channel number", 81, 6, 0) conn.test_bad_oci("invalid channel number", 81, 0, 0) conn.test_bad_oci("wrong operating class", 80, 0, 0) conn.test_bad_oci("invalid operating class", 0, 0, 0) conn.confirm_valid_oci(81, 1, 0) @remote_compatible def test_wpa2_ocv_ap_ht_mismatch(dev, apdev): """OCV AP mismatch (HT)""" params = {"channel": "6", "ht_capab": "[HT40-]", "ieee80211w": "1", "freq": "2437"} conn = APConnection(apdev[0], dev[0], params) conn.test_bad_oci("wrong primary channel", 84, 5, 0) conn.test_bad_oci("lower bandwidth than negotiated", 81, 6, 0) conn.test_bad_oci("bad upper/lower channel", 83, 6, 0) conn.confirm_valid_oci(84, 6, 0) @remote_compatible def test_wpa2_ocv_ap_vht80_mismatch(dev, apdev): """OCV AP mismatch (VHT80)""" try: run_wpa2_ocv_ap_vht80_mismatch(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].flush_scan_cache() def run_wpa2_ocv_ap_vht80_mismatch(dev, apdev): params = {"hw_mode": "a", "channel": "36", "country_code": "US", "ht_capab": "[HT40+]", "ieee80211w": "1", "ieee80211n": "1", "ieee80211ac": "1", "vht_oper_chwidth": "1", "freq": "5180", "vht_oper_centr_freq_seg0_idx": "42"} conn = APConnection(apdev[0], dev[0], params) conn.test_bad_oci("wrong primary channel", 128, 38, 0) conn.test_bad_oci("wrong primary channel", 128, 32, 0) conn.test_bad_oci("smaller bandwidth than negotiated", 116, 36, 0) conn.test_bad_oci("smaller bandwidth than negotiated", 115, 36, 0) conn.confirm_valid_oci(128, 36, 0) dev[0].dump_monitor() dev[0].request("DISCONNECT") dev[0].wait_disconnected() @remote_compatible def test_wpa2_ocv_ap_vht160_mismatch(dev, apdev): """OCV AP mismatch (VHT160)""" try: run_wpa2_ocv_ap_vht160_mismatch(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=0.5) dev[0].flush_scan_cache() def run_wpa2_ocv_ap_vht160_mismatch(dev, apdev): params = {"hw_mode": "a", "channel": "100", "country_code": "ZA", "ht_capab": "[HT40+]", "ieee80211w": "1", "ieee80211n": "1", "ieee80211ac": "1", "vht_oper_chwidth": "2", "freq": "5500", "vht_oper_centr_freq_seg0_idx": "114", "ieee80211d": "1", "ieee80211h": "1"} conn = APConnection(apdev[0], dev[0], params) conn.test_bad_oci("wrong primary channel", 129, 36, 0) conn.test_bad_oci("wrong primary channel", 129, 114, 0) conn.test_bad_oci("smaller bandwidth (20 Mhz) than negotiated", 121, 100, 0) conn.test_bad_oci("smaller bandwidth (40 Mhz) than negotiated", 122, 100, 0) conn.test_bad_oci("smaller bandwidth (80 Mhz) than negotiated", 128, 100, 0) conn.test_bad_oci("using 80+80 channel instead of 160", 130, 100, 155) conn.confirm_valid_oci(129, 100, 0) dev[0].dump_monitor() if conn.hapd: conn.hapd.request("DISABLE") dev[0].disconnect_and_stop_scan() @remote_compatible def test_wpa2_ocv_ap_vht80plus80_mismatch(dev, apdev): """OCV AP mismatch (VHT80+80)""" try: run_wpa2_ocv_ap_vht80plus80_mismatch(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=0.5) dev[0].flush_scan_cache() def run_wpa2_ocv_ap_vht80plus80_mismatch(dev, apdev): params = {"hw_mode": "a", "channel": "36", "country_code": "US", "ht_capab": "[HT40+]", "ieee80211w": "1", "ieee80211n": "1", "ieee80211ac": "1", "vht_oper_chwidth": "3", "freq": "5180", "vht_oper_centr_freq_seg0_idx": "42", "ieee80211d": "1", "vht_oper_centr_freq_seg1_idx": "155", "ieee80211h": "1"} conn = APConnection(apdev[0], dev[0], params) conn.test_bad_oci("using 80 MHz operating class", 128, 36, 155) conn.test_bad_oci("wrong frequency segment 1", 130, 36, 138) conn.confirm_valid_oci(130, 36, 155) dev[0].dump_monitor() if conn.hapd: conn.hapd.request("DISABLE") dev[0].disconnect_and_stop_scan() @remote_compatible def test_wpa2_ocv_ap_unexpected1(dev, apdev): """OCV and unexpected OCI KDE from station""" params = {"channel": "1", "ieee80211w": "1", "ap_ocv": "0", "sta_ocv": "1", "freq": "2412"} conn = APConnection(apdev[0], dev[0], params) logger.debug("Client will send OCI KDE even if it was not negotiated") conn.confirm_valid_oci(81, 1, 0) @remote_compatible def test_wpa2_ocv_ap_unexpected2(dev, apdev): """OCV and unexpected OCI KDE from station""" params = {"channel": "1", "ieee80211w": "1", "ap_ocv": "1", "sta_ocv": "0", "freq": "2412"} conn = APConnection(apdev[0], dev[0], params) logger.debug("Client will send OCI KDE even if it was not negotiated") conn.confirm_valid_oci(81, 1, 0) @remote_compatible def test_wpa2_ocv_ap_retransmit_msg3(dev, apdev): """Verify that manually retransmitted msg 3/4 contain a correct OCI""" bssid = apdev[0]['bssid'] ssid = "test-wpa2-ocv" passphrase = "qwertyuiop" psk = "c2c6c255af836bed1b3f2f1ded98e052f5ad618bb554e2836757b55854a0eab7" params = hostapd.wpa2_params(ssid=ssid) params["wpa_psk"] = psk params["ieee80211w"] = "1" params["ocv"] = "1" params['wpa_disable_eapol_key_retries'] = "1" try: hapd = hostapd.add_ap(apdev[0], params) except Exception as e: if "Failed to set hostapd parameter ocv" in str(e): raise HwsimSkip("OCV not supported") raise hapd.request("SET ext_eapol_frame_io 1") dev[0].request("SET ext_eapol_frame_io 1") dev[0].connect(ssid, psk=passphrase, scan_freq="2412", wait_connect=False, ocv="1", ieee80211w="1") addr = dev[0].own_addr() # EAPOL-Key msg 1/4 ev = hapd.wait_event(["EAPOL-TX"], timeout=15) if ev is None: raise Exception("Timeout on EAPOL-TX from hostapd") res = dev[0].request("EAPOL_RX " + bssid + " " + ev.split(' ')[2]) if "OK" not in res: raise Exception("EAPOL_RX to wpa_supplicant failed") # EAPOL-Key msg 2/4 ev = dev[0].wait_event(["EAPOL-TX"], timeout=15) if ev is None: raise Exception("Timeout on EAPOL-TX from wpa_supplicant") res = hapd.request("EAPOL_RX " + addr + " " + ev.split(' ')[2]) if "OK" not in res: raise Exception("EAPOL_RX to hostapd failed") # EAPOL-Key msg 3/4 ev = hapd.wait_event(["EAPOL-TX"], timeout=15) if ev is None: raise Exception("Timeout on EAPOL-TX from hostapd") logger.info("Drop the first EAPOL-Key msg 3/4") # Use normal EAPOL TX/RX to handle retries. hapd.request("SET ext_eapol_frame_io 0") dev[0].request("SET ext_eapol_frame_io 0") # Manually retransmit EAPOL-Key msg 3/4 if "OK" not in hapd.request("RESEND_M3 " + addr): raise Exception("RESEND_M3 failed") dev[0].wait_connected() hwsim_utils.test_connectivity(dev[0], hapd) def test_wpa2_ocv_ap_group_hs(dev, apdev): """OCV group handshake (AP)""" params = {"channel": "1", "ieee80211w": "1", "freq": "2412", "wpa_strict_rekey": "1"} conn = APConnection(apdev[0], dev[0], params) conn.confirm_valid_oci(81, 1, 0) conn.hapd.request("SET ext_eapol_frame_io 0") dev[1].connect(conn.ssid, psk=conn.passphrase, scan_freq="2412", ocv="1", ieee80211w="1") conn.hapd.wait_sta() conn.hapd.request("SET ext_eapol_frame_io 1") # Trigger a group key handshake dev[1].request("DISCONNECT") dev[0].dump_monitor() # Wait for EAPOL-Key msg 1/2 conn.msg = recv_eapol(conn.hapd) if conn.msg["rsn_key_info"] != 4994: raise Exception("Didn't receive 1/2 of group key handshake") # Send a EAPOL-Key msg 2/2 with a bad OCI logger.info("Bad OCI element") ocikde = make_ocikde(1, 1, 1) msg = build_eapol_key_2_2(conn.kck, ocikde, replay_counter=3) conn.hapd.dump_monitor() send_eapol(conn.hapd, conn.addr, build_eapol(msg)) # Wait for retransmitted EAPOL-Key msg 1/2 conn.msg = recv_eapol(conn.hapd) if conn.msg["rsn_key_info"] != 4994: raise Exception("Didn't receive 1/2 of group key handshake") # Send a EAPOL-Key msg 2/2 with a good OCI logger.info("Good OCI element") ocikde = make_ocikde(81, 1, 0) msg = build_eapol_key_2_2(conn.kck, ocikde, replay_counter=4) conn.hapd.dump_monitor() send_eapol(conn.hapd, conn.addr, build_eapol(msg)) # Verify that group key handshake has completed ev = conn.hapd.wait_event(["EAPOL-TX"], timeout=1) if ev is not None: eapol = binascii.unhexlify(ev.split(' ')[2]) msg = parse_eapol(eapol) if msg["rsn_key_info"] == 4994: raise Exception("AP didn't accept 2/2 of group key handshake") class STAConnection: def init_params(self): # Static parameters self.ssid = "test-wpa2-ocv" self.passphrase = "qwertyuiop" self.psk = "c2c6c255af836bed1b3f2f1ded98e052f5ad618bb554e2836757b55854a0eab7" # Dynamic parameters self.hapd = None self.dev = None self.addr = None self.rsne = None self.kck = None self.kek = None self.msg = None self.bssid = None self.anonce = None self.snonce = None self.gtkie = None self.counter = None def __init__(self, apdev, dev, params, sta_params=None): self.init_params() self.dev = dev self.bssid = apdev['bssid'] freq = params.pop("freq") if sta_params is None: sta_params = dict() if "ocv" not in sta_params: sta_params["ocv"] = "1" if "ieee80211w" not in sta_params: sta_params["ieee80211w"] = "1" params.update(hostapd.wpa2_params(ssid=self.ssid, passphrase=self.passphrase)) params['wpa_pairwise_update_count'] = "10" try: self.hapd = hostapd.add_ap(apdev, params) except Exception as e: if "Failed to set hostapd parameter ocv" in str(e): raise HwsimSkip("OCV not supported") raise self.hapd.request("SET ext_eapol_frame_io 1") self.dev.request("SET ext_eapol_frame_io 1") pmk = binascii.unhexlify("c2c6c255af836bed1b3f2f1ded98e052f5ad618bb554e2836757b55854a0eab7") self.gtkie = binascii.unhexlify("dd16000fac010100dc11188831bf4aa4a8678d2b41498618") if sta_params["ocv"] != "0": self.rsne = binascii.unhexlify("30140100000fac040100000fac040100000fac028c40") else: self.rsne = binascii.unhexlify("30140100000fac040100000fac040100000fac028c00") self.dev.connect(self.ssid, raw_psk=self.psk, scan_freq=freq, wait_connect=False, **sta_params) if "country_code" in params: self.dev.wait_regdom(country_ie=True) self.addr = dev.p2p_interface_addr() # Forward msg 1/4 from AP to STA self.msg = recv_eapol(self.hapd) self.anonce = self.msg['rsn_key_nonce'] send_eapol(self.dev, self.bssid, build_eapol(self.msg)) # Capture msg 2/4 from the STA so we can derive the session keys self.msg = recv_eapol(dev) self.snonce = self.msg['rsn_key_nonce'] (ptk, self.kck, self.kek) = pmk_to_ptk(pmk, self.addr, self.bssid, self.snonce, self.anonce) self.counter = struct.unpack('>Q', self.msg['rsn_replay_counter'])[0] + 1 def test_bad_oci(self, logmsg, op_class, channel, seg1_idx, errmsg): logger.info("Bad OCI element: " + logmsg) if op_class is None: ocikde = b'' else: ocikde = make_ocikde(op_class, channel, seg1_idx) plain = self.rsne + self.gtkie + ocikde wrapped = aes_wrap(self.kek, pad_key_data(plain)) msg = build_eapol_key_3_4(self.anonce, self.kck, wrapped, replay_counter=self.counter) self.dev.dump_monitor() send_eapol(self.dev, self.bssid, build_eapol(msg)) self.counter += 1 ev = self.dev.wait_event([errmsg], timeout=5) if ev is None: raise Exception("Bad OCI not reported") def confirm_valid_oci(self, op_class, channel, seg1_idx): logger.debug("Valid OCI element to complete handshake") ocikde = make_ocikde(op_class, channel, seg1_idx) plain = self.rsne + self.gtkie + ocikde wrapped = aes_wrap(self.kek, pad_key_data(plain)) msg = build_eapol_key_3_4(self.anonce, self.kck, wrapped, replay_counter=self.counter) self.dev.dump_monitor() send_eapol(self.dev, self.bssid, build_eapol(msg)) self.counter += 1 self.dev.wait_connected(timeout=1) @remote_compatible def test_wpa2_ocv_mismatch_client(dev, apdev): """OCV client mismatch""" params = {"channel": "1", "ieee80211w": "1", "ocv": "1", "freq": "2412"} conn = STAConnection(apdev[0], dev[0], params) conn.test_bad_oci("element missing", None, 0, 0, "did not receive mandatory OCI") conn.test_bad_oci("wrong channel number", 81, 6, 0, "primary channel mismatch") conn.test_bad_oci("invalid channel number", 81, 0, 0, "unable to interpret received OCI") conn.test_bad_oci("wrong operating class", 80, 0, 0, "unable to interpret received OCI") conn.test_bad_oci("invalid operating class", 0, 0, 0, "unable to interpret received OCI") conn.confirm_valid_oci(81, 1, 0) @remote_compatible def test_wpa2_ocv_vht160_mismatch_client(dev, apdev): """OCV client mismatch (VHT160)""" try: run_wpa2_ocv_vht160_mismatch_client(dev, apdev) finally: set_world_reg(apdev[0], apdev[1], dev[0]) dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=0.5) dev[0].flush_scan_cache() def run_wpa2_ocv_vht160_mismatch_client(dev, apdev): params = {"hw_mode": "a", "channel": "100", "country_code": "ZA", "ht_capab": "[HT40+]", "ieee80211w": "1", "ieee80211n": "1", "ieee80211ac": "1", "vht_oper_chwidth": "2", "ocv": "1", "vht_oper_centr_freq_seg0_idx": "114", "freq": "5500", "ieee80211d": "1", "ieee80211h": "1"} sta_params = {"disable_vht": "1"} conn = STAConnection(apdev[0], dev[0], params, sta_params) conn.test_bad_oci("smaller bandwidth (20 Mhz) than negotiated", 121, 100, 0, "channel bandwidth mismatch") conn.test_bad_oci("wrong frequency, bandwith, and secondary channel", 123, 104, 0, "primary channel mismatch") conn.test_bad_oci("wrong upper/lower behaviour", 129, 104, 0, "primary channel mismatch") conn.confirm_valid_oci(122, 100, 0) dev[0].dump_monitor() if conn.hapd: conn.hapd.request("DISABLE") dev[0].disconnect_and_stop_scan() def test_wpa2_ocv_sta_group_hs(dev, apdev): """OCV group handshake (STA)""" params = {"channel": "1", "ieee80211w": "1", "ocv": "1", "freq": "2412", "wpa_strict_rekey": "1"} conn = STAConnection(apdev[0], dev[0], params.copy()) conn.confirm_valid_oci(81, 1, 0) # Send a EAPOL-Key msg 1/2 with a bad OCI logger.info("Bad OCI element") plain = conn.gtkie + make_ocikde(1, 1, 1) wrapped = aes_wrap(conn.kek, pad_key_data(plain)) msg = build_eapol_key_1_2(conn.kck, wrapped, replay_counter=3) send_eapol(dev[0], conn.bssid, build_eapol(msg)) # We shouldn't get a EAPOL-Key message back ev = dev[0].wait_event(["EAPOL-TX"], timeout=1) if ev is not None: raise Exception("Received response to invalid EAPOL-Key 1/2") # Reset AP to try with valid OCI conn.hapd.disable() conn = STAConnection(apdev[0], dev[0], params.copy()) conn.confirm_valid_oci(81, 1, 0) # Send a EAPOL-Key msg 1/2 with a good OCI logger.info("Good OCI element") plain = conn.gtkie + make_ocikde(81, 1, 0) wrapped = aes_wrap(conn.kek, pad_key_data(plain)) msg = build_eapol_key_1_2(conn.kck, wrapped, replay_counter=4) send_eapol(dev[0], conn.bssid, build_eapol(msg)) # Wait for EAPOL-Key msg 2/2 conn.msg = recv_eapol(dev[0]) if conn.msg["rsn_key_info"] != 0x0302: raise Exception("Didn't receive 2/2 of group key handshake") def test_wpa2_ocv_auto_enable_pmf(dev, apdev): """OCV on 2.4 GHz with PMF getting enabled automatically""" params = {"channel": "1", "ocv": "1"} hapd, ssid, passphrase = ocv_setup_ap(apdev[0], params) for ocv in range(2): dev[0].connect(ssid, psk=passphrase, scan_freq="2412", ocv=str(ocv), ieee80211w="2") dev[0].request("REMOVE_NETWORK all") dev[0].wait_disconnected()