tests/hwsim/test_ap_pmf.py

   1 # Protected management frames tests
   2 # Copyright (c) 2013, Jouni Malinen <j@w1.fi>
   3 #
   4 # This software may be distributed under the terms of the BSD license.
   5 # See README for more details.
   6
   7 from remotehost import remote_compatible
   8 import binascii
   9 import os
  10 import time
  11 import logging
  12 logger = logging.getLogger()
  13
  14 import hwsim_utils
  15 import hostapd
  16 from utils import alloc_fail, fail_test, wait_fail_trigger, HwsimSkip, \
  17     radiotap_build, start_monitor, stop_monitor
  18 from wlantest import Wlantest
  19 from wpasupplicant import WpaSupplicant
  20
  21 @remote_compatible
  22 def test_ap_pmf_required(dev, apdev):
  23     """WPA2-PSK AP with PMF required"""
  24     ssid = "test-pmf-required"
  25     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
  26     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
  27     params["ieee80211w"] = "2"
  28     hapd = hostapd.add_ap(apdev[0], params)
  29     Wlantest.setup(hapd)
  30     wt = Wlantest()
  31     wt.flush()
  32     wt.add_passphrase("12345678")
  33     key_mgmt = hapd.get_config()['key_mgmt']
  34     if key_mgmt.split(' ')[0] != "WPA-PSK-SHA256":
  35         raise Exception("Unexpected GET_CONFIG(key_mgmt): " + key_mgmt)
  36     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
  37                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
  38                    scan_freq="2412")
  39     if "[WPA2-PSK-SHA256-CCMP]" not in dev[0].request("SCAN_RESULTS"):
  40         raise Exception("Scan results missing RSN element info")
  41     hwsim_utils.test_connectivity(dev[0], hapd)
  42     dev[1].connect(ssid, psk="12345678", ieee80211w="2",
  43                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
  44                    scan_freq="2412")
  45     hwsim_utils.test_connectivity(dev[1], hapd)
  46     if "OK" not in hapd.request("SA_QUERY " + dev[0].own_addr()):
  47         raise Exception("SA_QUERY failed")
  48     if "OK" not in hapd.request("SA_QUERY " + dev[1].own_addr()):
  49         raise Exception("SA_QUERY failed")
  50     if "FAIL" not in hapd.request("SA_QUERY foo"):
  51         raise Exception("Invalid SA_QUERY accepted")
  52     wt.require_ap_pmf_mandatory(apdev[0]['bssid'])
  53     wt.require_sta_pmf(apdev[0]['bssid'], dev[0].p2p_interface_addr())
  54     wt.require_sta_pmf_mandatory(apdev[0]['bssid'], dev[1].p2p_interface_addr())
  55     time.sleep(0.1)
  56     if wt.get_sta_counter("valid_saqueryresp_tx", apdev[0]['bssid'],
  57                           dev[0].p2p_interface_addr()) < 1:
  58         raise Exception("STA did not reply to SA Query")
  59     if wt.get_sta_counter("valid_saqueryresp_tx", apdev[0]['bssid'],
  60                           dev[1].p2p_interface_addr()) < 1:
  61         raise Exception("STA did not reply to SA Query")
  62
  63 @remote_compatible
  64 def test_ocv_sa_query(dev, apdev):
  65     """Test SA Query with OCV"""
  66     ssid = "test-pmf-required"
  67     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
  68     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
  69     params["ieee80211w"] = "2"
  70     params["ocv"] = "1"
  71     try:
  72         hapd = hostapd.add_ap(apdev[0], params)
  73     except Exception as e:
  74         if "Failed to set hostapd parameter ocv" in str(e):
  75             raise HwsimSkip("OCV not supported")
  76         raise
  77     Wlantest.setup(hapd)
  78     wt = Wlantest()
  79     wt.flush()
  80     wt.add_passphrase("12345678")
  81     dev[0].connect(ssid, psk="12345678", ieee80211w="1", ocv="1",
  82                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
  83                    scan_freq="2412")
  84
  85     # Test that client can handle SA Query with OCI element
  86     if "OK" not in hapd.request("SA_QUERY " + dev[0].own_addr()):
  87         raise Exception("SA_QUERY failed")
  88     time.sleep(0.1)
  89     if wt.get_sta_counter("valid_saqueryresp_tx", apdev[0]['bssid'],
  90                           dev[0].own_addr()) < 1:
  91         raise Exception("STA did not reply to SA Query")
  92
  93     # Test that AP can handle SA Query with OCI element
  94     if "OK" not in dev[0].request("UNPROT_DEAUTH"):
  95         raise Exception("Triggering SA Query from the STA failed")
  96     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=3)
  97     if ev is not None:
  98         raise Exception("SA Query from the STA failed")
  99
 100 @remote_compatible
 101 def test_ocv_sa_query_csa(dev, apdev):
 102     """Test SA Query with OCV after channel switch"""
 103     ssid = "test-pmf-required"
 104     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 105     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 106     params["ieee80211w"] = "2"
 107     params["ocv"] = "1"
 108     try:
 109         hapd = hostapd.add_ap(apdev[0], params)
 110     except Exception as e:
 111         if "Failed to set hostapd parameter ocv" in str(e):
 112             raise HwsimSkip("OCV not supported")
 113         raise
 114     Wlantest.setup(hapd)
 115     wt = Wlantest()
 116     wt.flush()
 117     wt.add_passphrase("12345678")
 118     dev[0].connect(ssid, psk="12345678", ieee80211w="1", ocv="1",
 119                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 120                    scan_freq="2412")
 121
 122     hapd.request("CHAN_SWITCH 5 2437")
 123     time.sleep(1)
 124     if wt.get_sta_counter("valid_saqueryreq_tx", apdev[0]['bssid'],
 125                           dev[0].own_addr()) < 1:
 126         raise Exception("STA did not start SA Query after channel switch")
 127
 128 @remote_compatible
 129 def test_ap_pmf_optional(dev, apdev):
 130     """WPA2-PSK AP with PMF optional"""
 131     ssid = "test-pmf-optional"
 132     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 133     params["wpa_key_mgmt"] = "WPA-PSK"
 134     params["ieee80211w"] = "1"
 135     hapd = hostapd.add_ap(apdev[0], params)
 136     Wlantest.setup(hapd)
 137     wt = Wlantest()
 138     wt.flush()
 139     wt.add_passphrase("12345678")
 140     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 141                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 142                    scan_freq="2412")
 143     hwsim_utils.test_connectivity(dev[0], hapd)
 144     dev[1].connect(ssid, psk="12345678", ieee80211w="2",
 145                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 146                    scan_freq="2412")
 147     hwsim_utils.test_connectivity(dev[1], hapd)
 148     wt.require_ap_pmf_optional(apdev[0]['bssid'])
 149     wt.require_sta_pmf(apdev[0]['bssid'], dev[0].p2p_interface_addr())
 150     wt.require_sta_pmf_mandatory(apdev[0]['bssid'], dev[1].p2p_interface_addr())
 151
 152 @remote_compatible
 153 def test_ap_pmf_optional_2akm(dev, apdev):
 154     """WPA2-PSK AP with PMF optional (2 AKMs)"""
 155     ssid = "test-pmf-optional-2akm"
 156     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 157     params["wpa_key_mgmt"] = "WPA-PSK WPA-PSK-SHA256"
 158     params["ieee80211w"] = "1"
 159     hapd = hostapd.add_ap(apdev[0], params)
 160     Wlantest.setup(hapd)
 161     wt = Wlantest()
 162     wt.flush()
 163     wt.add_passphrase("12345678")
 164     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 165                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 166                    scan_freq="2412")
 167     hwsim_utils.test_connectivity(dev[0], hapd)
 168     dev[1].connect(ssid, psk="12345678", ieee80211w="2",
 169                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 170                    scan_freq="2412")
 171     hwsim_utils.test_connectivity(dev[1], hapd)
 172     wt.require_ap_pmf_optional(apdev[0]['bssid'])
 173     wt.require_sta_pmf(apdev[0]['bssid'], dev[0].p2p_interface_addr())
 174     wt.require_sta_key_mgmt(apdev[0]['bssid'], dev[0].p2p_interface_addr(),
 175                             "PSK-SHA256")
 176     wt.require_sta_pmf_mandatory(apdev[0]['bssid'], dev[1].p2p_interface_addr())
 177     wt.require_sta_key_mgmt(apdev[0]['bssid'], dev[1].p2p_interface_addr(),
 178                             "PSK-SHA256")
 179
 180 @remote_compatible
 181 def test_ap_pmf_negative(dev, apdev):
 182     """WPA2-PSK AP without PMF (negative test)"""
 183     ssid = "test-pmf-negative"
 184     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 185     hapd = hostapd.add_ap(apdev[0], params)
 186     Wlantest.setup(hapd)
 187     wt = Wlantest()
 188     wt.flush()
 189     wt.add_passphrase("12345678")
 190     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 191                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 192                    scan_freq="2412")
 193     hwsim_utils.test_connectivity(dev[0], hapd)
 194     try:
 195         dev[1].connect(ssid, psk="12345678", ieee80211w="2",
 196                        key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 197                        scan_freq="2412")
 198         hwsim_utils.test_connectivity(dev[1], hapd)
 199         raise Exception("PMF required STA connected to no PMF AP")
 200     except Exception as e:
 201         logger.debug("Ignore expected exception: " + str(e))
 202     wt.require_ap_no_pmf(apdev[0]['bssid'])
 203
 204 @remote_compatible
 205 def test_ap_pmf_assoc_comeback(dev, apdev):
 206     """WPA2-PSK AP with PMF association comeback"""
 207     ssid = "assoc-comeback"
 208     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 209     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 210     params["ieee80211w"] = "2"
 211     hapd = hostapd.add_ap(apdev[0], params)
 212     Wlantest.setup(hapd)
 213     wt = Wlantest()
 214     wt.flush()
 215     wt.add_passphrase("12345678")
 216     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 217                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 218                    scan_freq="2412")
 219     hapd.set("ext_mgmt_frame_handling", "1")
 220     dev[0].request("DISCONNECT")
 221     dev[0].wait_disconnected(timeout=10)
 222     hapd.set("ext_mgmt_frame_handling", "0")
 223     dev[0].request("REASSOCIATE")
 224     dev[0].wait_connected(timeout=10, error="Timeout on re-connection")
 225     if wt.get_sta_counter("assocresp_comeback", apdev[0]['bssid'],
 226                           dev[0].p2p_interface_addr()) < 1:
 227         raise Exception("AP did not use association comeback request")
 228
 229 @remote_compatible
 230 def test_ap_pmf_assoc_comeback2(dev, apdev):
 231     """WPA2-PSK AP with PMF association comeback (using DROP_SA)"""
 232     ssid = "assoc-comeback"
 233     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 234     params["wpa_key_mgmt"] = "WPA-PSK"
 235     params["ieee80211w"] = "1"
 236     hapd = hostapd.add_ap(apdev[0], params)
 237     Wlantest.setup(hapd)
 238     wt = Wlantest()
 239     wt.flush()
 240     wt.add_passphrase("12345678")
 241     dev[0].connect(ssid, psk="12345678", ieee80211w="2",
 242                    key_mgmt="WPA-PSK", proto="WPA2", scan_freq="2412")
 243     if "OK" not in dev[0].request("DROP_SA"):
 244         raise Exception("DROP_SA failed")
 245     dev[0].request("REASSOCIATE")
 246     dev[0].wait_connected(timeout=10, error="Timeout on re-connection")
 247     if wt.get_sta_counter("reassocresp_comeback", apdev[0]['bssid'],
 248                           dev[0].p2p_interface_addr()) < 1:
 249         raise Exception("AP did not use reassociation comeback request")
 250
 251 def test_ap_pmf_ap_dropping_sa(dev, apdev):
 252     """WPA2-PSK PMF AP dropping SA"""
 253     ssid = "pmf"
 254     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 255     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 256     params["ieee80211w"] = "2"
 257     hapd = hostapd.add_ap(apdev[0], params)
 258     bssid = hapd.own_addr()
 259     Wlantest.setup(hapd)
 260     wt = Wlantest()
 261     wt.flush()
 262     wt.add_passphrase("12345678")
 263     dev[0].connect(ssid, psk="12345678", ieee80211w="2",
 264                    key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
 265     addr0 = dev[0].own_addr()
 266     dev[0].dump_monitor()
 267     hapd.wait_sta()
 268     # Drop SA and association at the AP locally without notifying the STA. This
 269     # results in the STA getting unprotected Deauthentication frames when trying
 270     # to transmit the next Class 3 frame.
 271     if "OK" not in hapd.request("DEAUTHENTICATE " + addr0 + " tx=0"):
 272         raise Exception("DEAUTHENTICATE command failed")
 273     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 274     if ev is not None:
 275         raise Exception("Unexpected disconnection event after DEAUTHENTICATE tx=0: " + ev)
 276     dev[0].request("DATA_TEST_CONFIG 1")
 277     dev[0].request("DATA_TEST_TX " + bssid + " " + addr0)
 278     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
 279     dev[0].request("DATA_TEST_CONFIG 0")
 280     if ev is None or "locally_generated=1" not in ev:
 281         raise Exception("Locally generated disconnection not reported")
 282
 283 def test_ap_pmf_valid_broadcast_deauth(dev, apdev):
 284     """WPA2-PSK PMF AP sending valid broadcast deauth without dropping SA"""
 285     run_ap_pmf_valid(dev, apdev, False, True)
 286
 287 def test_ap_pmf_valid_broadcast_disassoc(dev, apdev):
 288     """WPA2-PSK PMF AP sending valid broadcast disassoc without dropping SA"""
 289     run_ap_pmf_valid(dev, apdev, True, True)
 290
 291 def test_ap_pmf_valid_unicast_deauth(dev, apdev):
 292     """WPA2-PSK PMF AP sending valid unicast deauth without dropping SA"""
 293     run_ap_pmf_valid(dev, apdev, False, False)
 294
 295 def test_ap_pmf_valid_unicast_disassoc(dev, apdev):
 296     """WPA2-PSK PMF AP sending valid unicast disassoc without dropping SA"""
 297     run_ap_pmf_valid(dev, apdev, True, False)
 298
 299 def run_ap_pmf_valid(dev, apdev, disassociate, broadcast):
 300     ssid = "pmf"
 301     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 302     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 303     params["ieee80211w"] = "2"
 304     hapd = hostapd.add_ap(apdev[0], params)
 305     bssid = hapd.own_addr()
 306     Wlantest.setup(hapd)
 307     wt = Wlantest()
 308     wt.flush()
 309     wt.add_passphrase("12345678")
 310     dev[0].connect(ssid, psk="12345678", ieee80211w="2",
 311                    key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
 312     addr0 = dev[0].own_addr()
 313     dev[0].dump_monitor()
 314     hapd.wait_sta()
 315     cmd = "DISASSOCIATE " if disassociate else "DEAUTHENTICATE "
 316     cmd += "ff:ff:ff:ff:ff:ff" if broadcast else addr0
 317     cmd += " test=1"
 318     if "OK" not in hapd.request(cmd):
 319         raise Exception("hostapd command failed")
 320     sta = hapd.get_sta(addr0)
 321     if not sta:
 322         raise Exception("STA entry lost")
 323     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
 324     if ev is None:
 325         raise Exception("Disconnection not reported")
 326     if "locally_generated=1" in ev:
 327         raise Exception("Unexpected locally generated disconnection")
 328
 329     # Wait for SA Query procedure to fail and association comeback to succeed
 330     dev[0].wait_connected()
 331
 332 def start_wpas_ap(ssid):
 333     wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
 334     wpas.interface_add("wlan5", drv_params="use_monitor=1")
 335     id = wpas.add_network()
 336     wpas.set_network(id, "mode", "2")
 337     wpas.set_network_quoted(id, "ssid", ssid)
 338     wpas.set_network(id, "proto", "WPA2")
 339     wpas.set_network(id, "key_mgmt", "WPA-PSK-SHA256")
 340     wpas.set_network(id, "ieee80211w", "2")
 341     wpas.set_network_quoted(id, "psk", "12345678")
 342     wpas.set_network(id, "pairwise", "CCMP")
 343     wpas.set_network(id, "group", "CCMP")
 344     wpas.set_network(id, "frequency", "2412")
 345     wpas.set_network(id, "scan_freq", "2412")
 346     wpas.connect_network(id)
 347     wpas.dump_monitor()
 348     return wpas
 349
 350 def test_ap_pmf_sta_sa_query(dev, apdev):
 351     """WPA2-PSK AP with station using SA Query"""
 352     ssid = "assoc-comeback"
 353     addr = dev[0].own_addr()
 354
 355     wpas = start_wpas_ap(ssid)
 356     bssid = wpas.own_addr()
 357
 358     Wlantest.setup(wpas)
 359     wt = Wlantest()
 360     wt.flush()
 361     wt.add_passphrase("12345678")
 362
 363     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 364                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 365                    scan_freq="2412")
 366     wpas.dump_monitor()
 367     wpas.request("DEAUTHENTICATE " + addr + " test=0")
 368     wpas.dump_monitor()
 369     wpas.request("DISASSOCIATE " + addr + " test=0")
 370     wpas.dump_monitor()
 371     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 372     if ev is not None:
 373         raise Exception("Unexpected disconnection")
 374
 375     wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
 376     wpas.dump_monitor()
 377     wpas.request("DISASSOCIATE " + addr + " reason=7 test=0")
 378     wpas.dump_monitor()
 379     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 380     if ev is not None:
 381         raise Exception("Unexpected disconnection")
 382     if wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr) < 1:
 383         raise Exception("STA did not send SA Query")
 384     if wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr) < 1:
 385         raise Exception("AP did not reply to SA Query")
 386     wpas.dump_monitor()
 387
 388 def test_ap_pmf_sta_sa_query_no_response(dev, apdev):
 389     """WPA2-PSK AP with station using SA Query and getting no response"""
 390     ssid = "assoc-comeback"
 391     addr = dev[0].own_addr()
 392
 393     wpas = start_wpas_ap(ssid)
 394     bssid = wpas.own_addr()
 395
 396     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 397                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 398                    scan_freq="2412")
 399     wpas.dump_monitor()
 400     wpas.request("DEAUTHENTICATE " + addr + " test=0")
 401     wpas.dump_monitor()
 402     wpas.request("DISASSOCIATE " + addr + " test=0")
 403     wpas.dump_monitor()
 404     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 405     if ev is not None:
 406         raise Exception("Unexpected disconnection")
 407
 408     wpas.request("SET ext_mgmt_frame_handling 1")
 409     wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
 410     wpas.dump_monitor()
 411     wpas.request("DISASSOCIATE " + addr + " reason=7 test=0")
 412     wpas.dump_monitor()
 413     dev[0].wait_disconnected()
 414     wpas.dump_monitor()
 415     wpas.request("SET ext_mgmt_frame_handling 0")
 416     dev[0].wait_connected()
 417     wpas.dump_monitor()
 418
 419 def test_ap_pmf_sta_unprot_deauth_burst(dev, apdev):
 420     """WPA2-PSK AP with station receiving burst of unprotected Deauthentication frames"""
 421     ssid = "deauth-attack"
 422     addr = dev[0].own_addr()
 423
 424     wpas = start_wpas_ap(ssid)
 425     bssid = wpas.own_addr()
 426
 427     Wlantest.setup(wpas)
 428     wt = Wlantest()
 429     wt.flush()
 430     wt.add_passphrase("12345678")
 431
 432     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 433                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 434                    scan_freq="2412")
 435
 436     for i in range(0, 10):
 437         wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
 438         wpas.request("DISASSOCIATE " + addr + " reason=7 test=0")
 439     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 440     if ev is not None:
 441         raise Exception("Unexpected disconnection")
 442     num_req = wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr)
 443     num_resp = wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr)
 444     if num_req < 1:
 445         raise Exception("STA did not send SA Query")
 446     if num_resp < 1:
 447         raise Exception("AP did not reply to SA Query")
 448     if num_req > 1:
 449         raise Exception("STA initiated too many SA Query procedures (%d)" % num_req)
 450
 451     time.sleep(10)
 452     for i in range(0, 5):
 453         wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
 454         wpas.request("DISASSOCIATE " + addr + " reason=7 test=0")
 455     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 456     if ev is not None:
 457         raise Exception("Unexpected disconnection")
 458     num_req = wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr)
 459     num_resp = wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr)
 460     if num_req != 2 or num_resp != 2:
 461         raise Exception("Unexpected number of SA Query procedures (req=%d resp=%d)" % (num_req, num_resp))
 462
 463 def test_ap_pmf_sta_sa_query_oom(dev, apdev):
 464     """WPA2-PSK AP with station using SA Query (OOM)"""
 465     ssid = "assoc-comeback"
 466     addr = dev[0].own_addr()
 467     wpas = start_wpas_ap(ssid)
 468     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 469                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 470                    scan_freq="2412")
 471     with alloc_fail(dev[0], 1, "=sme_sa_query_timer"):
 472         wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
 473         wait_fail_trigger(dev[0], "GET_ALLOC_FAIL")
 474     dev[0].request("DISCONNECT")
 475     wpas.request("DISCONNECT")
 476     dev[0].wait_disconnected()
 477
 478 def test_ap_pmf_sta_sa_query_local_failure(dev, apdev):
 479     """WPA2-PSK AP with station using SA Query (local failure)"""
 480     ssid = "assoc-comeback"
 481     addr = dev[0].own_addr()
 482     wpas = start_wpas_ap(ssid)
 483     dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 484                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 485                    scan_freq="2412")
 486     with fail_test(dev[0], 1, "os_get_random;sme_sa_query_timer"):
 487         wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
 488         wait_fail_trigger(dev[0], "GET_FAIL")
 489     dev[0].request("DISCONNECT")
 490     wpas.request("DISCONNECT")
 491     dev[0].wait_disconnected()
 492
 493 def test_ap_pmf_sta_sa_query_hostapd(dev, apdev):
 494     """WPA2-PSK AP with station using SA Query (hostapd)"""
 495     ssid = "assoc-comeback"
 496     passphrase = "12345678"
 497     addr = dev[0].own_addr()
 498
 499     params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase,
 500                                  wpa_key_mgmt="WPA-PSK-SHA256",
 501                                  ieee80211w="2")
 502     hapd = hostapd.add_ap(apdev[0], params)
 503     bssid = hapd.own_addr()
 504
 505     Wlantest.setup(hapd)
 506     wt = Wlantest()
 507     wt.flush()
 508     wt.add_passphrase("12345678")
 509
 510     dev[0].connect(ssid, psk=passphrase, ieee80211w="2",
 511                    key_mgmt="WPA-PSK-SHA256", proto="WPA2",
 512                    scan_freq="2412")
 513     hapd.wait_sta()
 514     if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " test=0") or \
 515        "OK" not in hapd.request("DISASSOCIATE " + addr + " test=0"):
 516         raise Exception("Failed to send unprotected disconnection messages")
 517     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 518     if ev is not None:
 519         raise Exception("Unexpected disconnection")
 520
 521     if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " reason=6 test=0") or \
 522        "OK" not in hapd.request("DISASSOCIATE " + addr + " reason=7 test=0"):
 523         raise Exception("Failed to send unprotected disconnection messages (2)")
 524     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 525     if ev is not None:
 526         raise Exception("Unexpected disconnection")
 527     if wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr) < 1:
 528         raise Exception("STA did not send SA Query")
 529     if wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr) < 1:
 530         raise Exception("AP did not reply to SA Query")
 531
 532 def test_ap_pmf_sta_sa_query_no_response_hostapd(dev, apdev):
 533     """WPA2-PSK AP with station using SA Query and getting no response (hostapd)"""
 534     ssid = "assoc-comeback"
 535     passphrase = "12345678"
 536     addr = dev[0].own_addr()
 537
 538     params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase,
 539                                  wpa_key_mgmt="WPA-PSK-SHA256",
 540                                  ieee80211w="2")
 541     hapd = hostapd.add_ap(apdev[0], params)
 542     bssid = hapd.own_addr()
 543
 544     Wlantest.setup(hapd)
 545     wt = Wlantest()
 546     wt.flush()
 547     wt.add_passphrase("12345678")
 548
 549     dev[0].connect(ssid, psk=passphrase, ieee80211w="2",
 550                    key_mgmt="WPA-PSK-SHA256", proto="WPA2",
 551                    scan_freq="2412")
 552     hapd.wait_sta()
 553     hapd.set("ext_mgmt_frame_handling", "1")
 554     if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " reason=6 test=0") or \
 555        "OK" not in hapd.request("DISASSOCIATE " + addr + " reason=7 test=0"):
 556         raise Exception("Failed to send unprotected disconnection messages")
 557     dev[0].wait_disconnected()
 558     hapd.set("ext_mgmt_frame_handling", "0")
 559     if wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr) < 1:
 560         raise Exception("STA did not send SA Query")
 561     if wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr) > 0:
 562         raise Exception("AP replied to SA Query")
 563     dev[0].wait_connected()
 564
 565 def test_ap_pmf_sta_unprot_deauth_burst_hostapd(dev, apdev):
 566     """WPA2-PSK AP with station receiving burst of unprotected Deauthentication frames (hostapd)"""
 567     ssid = "deauth-attack"
 568     passphrase = "12345678"
 569     addr = dev[0].own_addr()
 570
 571     params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase,
 572                                  wpa_key_mgmt="WPA-PSK-SHA256",
 573                                  ieee80211w="2")
 574     hapd = hostapd.add_ap(apdev[0], params)
 575     bssid = hapd.own_addr()
 576
 577     Wlantest.setup(hapd)
 578     wt = Wlantest()
 579     wt.flush()
 580     wt.add_passphrase("12345678")
 581
 582     dev[0].connect(ssid, psk=passphrase, ieee80211w="2",
 583                    key_mgmt="WPA-PSK-SHA256", proto="WPA2",
 584                    scan_freq="2412")
 585     hapd.wait_sta()
 586     for i in range(10):
 587         if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " reason=6 test=0") or \
 588            "OK" not in hapd.request("DISASSOCIATE " + addr + " reason=7 test=0"):
 589             raise Exception("Failed to send unprotected disconnection messages")
 590     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 591     if ev is not None:
 592         raise Exception("Unexpected disconnection")
 593     num_req = wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr)
 594     num_resp = wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr)
 595     if num_req < 1:
 596         raise Exception("STA did not send SA Query")
 597     if num_resp < 1:
 598         raise Exception("AP did not reply to SA Query")
 599     if num_req > 1:
 600         raise Exception("STA initiated too many SA Query procedures (%d)" % num_req)
 601
 602     time.sleep(10)
 603     for i in range(5):
 604         if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " reason=6 test=0") or \
 605            "OK" not in hapd.request("DISASSOCIATE " + addr + " reason=7 test=0"):
 606             raise Exception("Failed to send unprotected disconnection messages")
 607     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
 608     if ev is not None:
 609         raise Exception("Unexpected disconnection")
 610     num_req = wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr)
 611     num_resp = wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr)
 612     if num_req != 2 or num_resp != 2:
 613         raise Exception("Unexpected number of SA Query procedures (req=%d resp=%d)" % (num_req, num_resp))
 614
 615 def test_ap_pmf_required_eap(dev, apdev):
 616     """WPA2-EAP AP with PMF required"""
 617     ssid = "test-pmf-required-eap"
 618     params = hostapd.wpa2_eap_params(ssid=ssid)
 619     params["wpa_key_mgmt"] = "WPA-EAP-SHA256"
 620     params["ieee80211w"] = "2"
 621     hapd = hostapd.add_ap(apdev[0], params)
 622     key_mgmt = hapd.get_config()['key_mgmt']
 623     if key_mgmt.split(' ')[0] != "WPA-EAP-SHA256":
 624         raise Exception("Unexpected GET_CONFIG(key_mgmt): " + key_mgmt)
 625     dev[0].connect("test-pmf-required-eap", key_mgmt="WPA-EAP-SHA256",
 626                    ieee80211w="2", eap="PSK", identity="psk.user@example.com",
 627                    password_hex="0123456789abcdef0123456789abcdef",
 628                    scan_freq="2412")
 629     dev[1].connect("test-pmf-required-eap", key_mgmt="WPA-EAP WPA-EAP-SHA256",
 630                    ieee80211w="1", eap="PSK", identity="psk.user@example.com",
 631                    password_hex="0123456789abcdef0123456789abcdef",
 632                    scan_freq="2412")
 633
 634 def test_ap_pmf_optional_eap(dev, apdev):
 635     """WPA2EAP AP with PMF optional"""
 636     params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
 637     params["ieee80211w"] = "1"
 638     hapd = hostapd.add_ap(apdev[0], params)
 639     dev[0].connect("test-wpa2-eap", key_mgmt="WPA-EAP", eap="TTLS",
 640                    identity="pap user", anonymous_identity="ttls",
 641                    password="password",
 642                    ca_cert="auth_serv/ca.pem", phase2="auth=PAP",
 643                    ieee80211w="1", scan_freq="2412")
 644     dev[1].connect("test-wpa2-eap", key_mgmt="WPA-EAP WPA-EAP-SHA256",
 645                    eap="TTLS", identity="pap user", anonymous_identity="ttls",
 646                    password="password",
 647                    ca_cert="auth_serv/ca.pem", phase2="auth=PAP",
 648                    ieee80211w="2", scan_freq="2412")
 649
 650 @remote_compatible
 651 def test_ap_pmf_required_sha1(dev, apdev):
 652     """WPA2-PSK AP with PMF required with SHA1 AKM"""
 653     ssid = "test-pmf-required-sha1"
 654     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 655     params["wpa_key_mgmt"] = "WPA-PSK"
 656     params["ieee80211w"] = "2"
 657     hapd = hostapd.add_ap(apdev[0], params)
 658     Wlantest.setup(hapd)
 659     wt = Wlantest()
 660     wt.flush()
 661     wt.add_passphrase("12345678")
 662     key_mgmt = hapd.get_config()['key_mgmt']
 663     if key_mgmt.split(' ')[0] != "WPA-PSK":
 664         raise Exception("Unexpected GET_CONFIG(key_mgmt): " + key_mgmt)
 665     dev[0].connect(ssid, psk="12345678", ieee80211w="2",
 666                    key_mgmt="WPA-PSK", proto="WPA2", scan_freq="2412")
 667     if "[WPA2-PSK-CCMP]" not in dev[0].request("SCAN_RESULTS"):
 668         raise Exception("Scan results missing RSN element info")
 669     hwsim_utils.test_connectivity(dev[0], hapd)
 670
 671 @remote_compatible
 672 def test_ap_pmf_toggle(dev, apdev):
 673     """WPA2-PSK AP with PMF optional and changing PMF on reassociation"""
 674     try:
 675         _test_ap_pmf_toggle(dev, apdev)
 676     finally:
 677         dev[0].request("SET reassoc_same_bss_optim 0")
 678
 679 def _test_ap_pmf_toggle(dev, apdev):
 680     ssid = "test-pmf-optional"
 681     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 682     params["wpa_key_mgmt"] = "WPA-PSK"
 683     params["ieee80211w"] = "1"
 684     params["assoc_sa_query_max_timeout"] = "1"
 685     params["assoc_sa_query_retry_timeout"] = "1"
 686     hapd = hostapd.add_ap(apdev[0], params)
 687     Wlantest.setup(hapd)
 688     wt = Wlantest()
 689     wt.flush()
 690     wt.add_passphrase("12345678")
 691     bssid = apdev[0]['bssid']
 692     addr = dev[0].own_addr()
 693     dev[0].request("SET reassoc_same_bss_optim 1")
 694     id = dev[0].connect(ssid, psk="12345678", ieee80211w="1",
 695                         key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 696                         scan_freq="2412")
 697     wt.require_ap_pmf_optional(bssid)
 698     wt.require_sta_pmf(bssid, addr)
 699     sta = hapd.get_sta(addr)
 700     if '[MFP]' not in sta['flags']:
 701         raise Exception("MFP flag not present for STA")
 702
 703     dev[0].set_network(id, "ieee80211w", "0")
 704     dev[0].request("REASSOCIATE")
 705     dev[0].wait_connected()
 706     wt.require_sta_no_pmf(bssid, addr)
 707     sta = hapd.get_sta(addr)
 708     if '[MFP]' in sta['flags']:
 709         raise Exception("MFP flag unexpectedly present for STA")
 710     err, data = hapd.cmd_execute(['iw', 'dev', apdev[0]['ifname'], 'station',
 711                                   'get', addr])
 712     if "yes" in [l for l in data.splitlines() if "MFP" in l][0]:
 713         raise Exception("Kernel STA entry had MFP enabled")
 714
 715     dev[0].set_network(id, "ieee80211w", "1")
 716     dev[0].request("REASSOCIATE")
 717     dev[0].wait_connected()
 718     wt.require_sta_pmf(bssid, addr)
 719     sta = hapd.get_sta(addr)
 720     if '[MFP]' not in sta['flags']:
 721         raise Exception("MFP flag not present for STA")
 722     err, data = hapd.cmd_execute(['iw', 'dev', apdev[0]['ifname'], 'station',
 723                                   'get', addr])
 724     if "yes" not in [l for l in data.splitlines() if "MFP" in l][0]:
 725         raise Exception("Kernel STA entry did not have MFP enabled")
 726
 727 @remote_compatible
 728 def test_ap_pmf_required_sta_no_pmf(dev, apdev):
 729     """WPA2-PSK AP with PMF required and PMF disabled on STA"""
 730     ssid = "test-pmf-required"
 731     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 732     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 733     params["ieee80211w"] = "2"
 734     hapd = hostapd.add_ap(apdev[0], params)
 735
 736     # Disable PMF on the station and try to connect
 737     dev[0].connect(ssid, psk="12345678", ieee80211w="0",
 738                    key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
 739                    scan_freq="2412", wait_connect=False)
 740     ev = dev[0].wait_event(["CTRL-EVENT-NETWORK-NOT-FOUND",
 741                             "CTRL-EVENT-ASSOC-REJECT"], timeout=2)
 742     if ev is None:
 743         raise Exception("No connection result")
 744     if "CTRL-EVENT-ASSOC-REJECT" in ev:
 745         raise Exception("Tried to connect to PMF required AP without PMF enabled")
 746     dev[0].request("REMOVE_NETWORK all")
 747
 748 def test_ap_pmf_inject_auth(dev, apdev):
 749     """WPA2-PSK AP with PMF and Authentication frame injection"""
 750     ssid = "test-pmf"
 751     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 752     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 753     params["ieee80211w"] = "2"
 754     hapd = hostapd.add_ap(apdev[0], params)
 755     dev[0].connect(ssid, psk="12345678", ieee80211w="2",
 756                    key_mgmt="WPA-PSK-SHA256", proto="WPA2",
 757                    scan_freq="2412")
 758     hapd.wait_sta()
 759     hwsim_utils.test_connectivity(dev[0], hapd)
 760
 761     bssid = hapd.own_addr().replace(':', '')
 762     addr = dev[0].own_addr().replace(':', '')
 763
 764     # Inject an unprotected Authentication frame claiming to be from the
 765     # associated STA, from another STA, from the AP's own address, from all
 766     # zeros and all ones addresses, and from a multicast address.
 767     hapd.request("SET ext_mgmt_frame_handling 1")
 768     failed = False
 769     addresses = [ addr, "021122334455", bssid, 6*"00", 6*"ff", 6*"01" ]
 770     for a in addresses:
 771         auth = "b0003a01" + bssid + a + bssid + '1000000001000000'
 772         res = hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=%s" % auth)
 773         if "OK" not in res:
 774             failed = True
 775     hapd.request("SET ext_mgmt_frame_handling 0")
 776     if failed:
 777         raise Exception("MGMT_RX_PROCESS failed")
 778     time.sleep(0.1)
 779
 780     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.1)
 781     if ev:
 782         raise Exception("Unexpected disconnection reported on the STA")
 783
 784     # Verify that original association is still functional.
 785     hwsim_utils.test_connectivity(dev[0], hapd)
 786
 787     # Inject an unprotected Association Request frame (with and without RSNE)
 788     # claiming to be from the set of test addresses.
 789     hapd.request("SET ext_mgmt_frame_handling 1")
 790     for a in addresses:
 791         assoc = "00003a01" + bssid + a + bssid + '2000' + '31040500' + '0008746573742d706d66' + '010802040b160c121824' + '301a0100000fac040100000fac040100000fac06c0000000000fac06'
 792         res = hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=%s" % assoc)
 793         if "OK" not in res:
 794             failed = True
 795
 796         assoc = "00003a01" + bssid + a + bssid + '2000' + '31040500' + '0008746573742d706d66' + '010802040b160c121824' + '3000'
 797         res = hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=%s" % assoc)
 798         if "OK" not in res:
 799             failed = True
 800
 801         assoc = "00003a01" + bssid + a + bssid + '2000' + '31040500' + '0008746573742d706d66' + '010802040b160c121824'
 802         res = hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=%s" % assoc)
 803         if "OK" not in res:
 804             failed = True
 805     hapd.request("SET ext_mgmt_frame_handling 0")
 806     if failed:
 807         raise Exception("MGMT_RX_PROCESS failed")
 808     time.sleep(5)
 809
 810     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.1)
 811     if ev:
 812         raise Exception("Unexpected disconnection reported on the STA")
 813
 814     # Verify that original association is still functional.
 815     hwsim_utils.test_connectivity(dev[0], hapd)
 816
 817 def test_ap_pmf_inject_data(dev, apdev):
 818     """WPA2-PSK AP with PMF and Data frame injection"""
 819     try:
 820         run_ap_pmf_inject_data(dev, apdev)
 821     finally:
 822         stop_monitor(apdev[1]["ifname"])
 823
 824 def run_ap_pmf_inject_data(dev, apdev):
 825     ssid = "test-pmf"
 826     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 827     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 828     params["ieee80211w"] = "2"
 829     hapd = hostapd.add_ap(apdev[0], params)
 830     dev[0].connect(ssid, psk="12345678", ieee80211w="2",
 831                    key_mgmt="WPA-PSK-SHA256", proto="WPA2",
 832                    scan_freq="2412")
 833     hapd.wait_sta()
 834     hwsim_utils.test_connectivity(dev[0], hapd)
 835
 836     sock = start_monitor(apdev[1]["ifname"])
 837     radiotap = radiotap_build()
 838
 839     bssid = hapd.own_addr().replace(':', '')
 840     addr = dev[0].own_addr().replace(':', '')
 841
 842     # Inject Data frame with A2=broadcast, A2=multicast, A2=BSSID, A2=STA, and
 843     # A2=unknown unicast
 844     addresses = [ 6*"ff", 6*"01", bssid, addr, "020102030405" ]
 845     for a in addresses:
 846         frame = binascii.unhexlify("48010000" + bssid + a + bssid + "0000")
 847         sock.send(radiotap + frame)
 848
 849     time.sleep(0.1)
 850     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.1)
 851     if ev:
 852         raise Exception("Unexpected disconnection reported on the STA")
 853     hwsim_utils.test_connectivity(dev[0], hapd)
 854
 855 def test_ap_pmf_tkip_reject(dev, apdev):
 856     """Mixed mode BSS and MFP-enabled AP rejecting TKIP"""
 857     params = hostapd.wpa2_params(ssid="test-pmf", passphrase="12345678")
 858     params['wpa'] = '3'
 859     params["ieee80211w"] = "1"
 860     params["wpa_pairwise"] = "TKIP CCMP"
 861     params["rsn_pairwise"] = "TKIP CCMP"
 862     hostapd.add_ap(apdev[0], params)
 863
 864     dev[0].connect("test-pmf", psk="12345678", pairwise="CCMP", ieee80211w="2",
 865                    scan_freq="2412")
 866     dev[0].dump_monitor()
 867
 868     dev[1].connect("test-pmf", psk="12345678", proto="WPA", pairwise="TKIP",
 869                    ieee80211w="0", scan_freq="2412")
 870     dev[1].dump_monitor()
 871
 872     dev[2].connect("test-pmf", psk="12345678", pairwise="TKIP",
 873                    ieee80211w="2", scan_freq="2412", wait_connect=False)
 874     ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED",
 875                             "CTRL-EVENT-ASSOC-REJECT"], timeout=10)
 876     if ev is None:
 877         raise Exception("No connection result reported")
 878     if "CTRL-EVENT-ASSOC-REJECT" not in ev:
 879         raise Exception("MFP + TKIP connection was not rejected")
 880     if "status_code=31" not in ev:
 881         raise Exception("Unexpected status code in rejection: " + ev)
 882     dev[2].request("DISCONNECT")
 883     dev[2].dump_monitor()
 884
 885 def test_ap_pmf_sa_query_timeout(dev, apdev):
 886     """SA Query timeout"""
 887     ssid = "test-pmf-required"
 888     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 889     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 890     params["ieee80211w"] = "2"
 891     hapd = hostapd.add_ap(apdev[0], params)
 892     dev[0].connect(ssid, psk="12345678", ieee80211w="2",
 893                    key_mgmt="WPA-PSK-SHA256", proto="WPA2",
 894                    scan_freq="2412")
 895
 896     hapd.set("ext_mgmt_frame_handling", "1")
 897     if "OK" not in dev[0].request("UNPROT_DEAUTH"):
 898         raise Exception("Triggering SA Query from the STA failed")
 899     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=2)
 900     if ev is None:
 901         raise Exception("No disconnection on SA Query timeout seen")
 902     hapd.set("ext_mgmt_frame_handling", "0")
 903     dev[0].wait_connected()
 904     dev[0].dump_monitor()
 905
 906     hapd.set("ext_mgmt_frame_handling", "1")
 907     if "OK" not in dev[0].request("UNPROT_DEAUTH"):
 908         raise Exception("Triggering SA Query from the STA failed")
 909     ev = hapd.mgmt_rx()
 910     hapd.set("ext_mgmt_frame_handling", "0")
 911     dev[0].request("DISCONNECT")
 912     dev[0].wait_disconnected()
 913     dev[0].request("RECONNECT")
 914     dev[0].wait_connected()
 915     hapd.set("ext_mgmt_frame_handling", "1")
 916     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1.5)
 917     if ev is not None:
 918         raise Exception("Unexpected disconnection after reconnection seen")
 919
 920 def mac80211_read_key(keydir):
 921     vals = {}
 922     for name in os.listdir(keydir):
 923         try:
 924             with open(os.path.join(keydir, name)) as f:
 925                 vals[name] = f.read().strip()
 926         except OSError as e:
 927             pass
 928     return vals
 929
 930 def test_ap_pmf_beacon_protection_bip(dev, apdev):
 931     """WPA2-PSK Beacon protection (BIP)"""
 932     """WPA2-PSK AP with PMF required and Beacon protection enabled (BIP)"""
 933     run_ap_pmf_beacon_protection(dev, apdev, "AES-128-CMAC")
 934
 935 def test_ap_pmf_beacon_protection_bip_cmac_256(dev, apdev):
 936     """WPA2-PSK Beacon protection (BIP-CMAC-256)"""
 937     run_ap_pmf_beacon_protection(dev, apdev, "BIP-CMAC-256")
 938
 939 def test_ap_pmf_beacon_protection_bip_gmac_128(dev, apdev):
 940     """WPA2-PSK Beacon protection (BIP-GMAC-128)"""
 941     run_ap_pmf_beacon_protection(dev, apdev, "BIP-GMAC-128")
 942
 943 def test_ap_pmf_beacon_protection_bip_gmac_256(dev, apdev):
 944     """WPA2-PSK Beacon protection (BIP-GMAC-256)"""
 945     run_ap_pmf_beacon_protection(dev, apdev, "BIP-GMAC-256")
 946
 947 def run_ap_pmf_beacon_protection(dev, apdev, cipher):
 948     ssid = "test-beacon-prot"
 949     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 950     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 951     params["ieee80211w"] = "2"
 952     params["beacon_prot"] = "1"
 953     params["group_mgmt_cipher"] = cipher
 954     try:
 955         hapd = hostapd.add_ap(apdev[0], params)
 956     except Exception as e:
 957         if "Failed to enable hostapd interface" in str(e):
 958             raise HwsimSkip("Beacon protection not supported")
 959         raise
 960
 961     bssid = hapd.own_addr()
 962
 963     Wlantest.setup(hapd)
 964     wt = Wlantest()
 965     wt.flush()
 966     wt.add_passphrase("12345678")
 967
 968     # STA with Beacon protection enabled
 969     dev[0].connect(ssid, psk="12345678", ieee80211w="2", beacon_prot="1",
 970                    key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
 971
 972     # STA with Beacon protection disabled
 973     dev[1].connect(ssid, psk="12345678", ieee80211w="2",
 974                    key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
 975
 976     time.sleep(1)
 977
 978     sta_key = None
 979     ap_key = None
 980
 981     phy = dev[0].get_driver_status_field("phyname")
 982     keys = "/sys/kernel/debug/ieee80211/%s/keys" % phy
 983     try:
 984         for key in os.listdir(keys):
 985             keydir = os.path.join(keys, key)
 986             vals = mac80211_read_key(keydir)
 987             keyidx = int(vals['keyidx'])
 988             if keyidx == 6 or keyidx == 7:
 989                 sta_key = vals;
 990                 break
 991     except OSError as e:
 992         raise HwsimSkip("debugfs not supported in mac80211 (STA)")
 993
 994     phy = hapd.get_driver_status_field("phyname")
 995     keys = "/sys/kernel/debug/ieee80211/%s/keys" % phy
 996     try:
 997         for key in os.listdir(keys):
 998             keydir = os.path.join(keys, key)
 999             vals = mac80211_read_key(keydir)
1000             keyidx = int(vals['keyidx'])
1001             if keyidx == 6 or keyidx == 7:
1002                 ap_key = vals;
1003                 break
1004     except OSError as e:
1005         raise HwsimSkip("debugfs not supported in mac80211 (AP)")
1006
1007     if not sta_key:
1008         raise Exception("Could not find STA key information from debugfs")
1009     logger.info("STA key: " + str(sta_key))
1010
1011     if not ap_key:
1012         raise Exception("Could not find AP key information from debugfs")
1013     logger.info("AP key: " + str(ap_key))
1014
1015     if sta_key['key'] != ap_key['key']:
1016         raise Exception("AP and STA BIGTK mismatch")
1017
1018     if sta_key['keyidx'] != ap_key['keyidx']:
1019         raise Exception("AP and STA BIGTK keyidx mismatch")
1020
1021     if sta_key['algorithm'] != ap_key['algorithm']:
1022         raise Exception("AP and STA BIGTK algorithm mismatch")
1023
1024     replays = int(sta_key['replays'])
1025     icverrors = int(sta_key['icverrors'])
1026     if replays > 0 or icverrors > 0:
1027         raise Exception("STA reported errors: replays=%d icverrors=%d" % replays, icverrors)
1028
1029     rx_spec = int(sta_key['rx_spec'], base=16)
1030     if rx_spec < 3:
1031         raise Exception("STA did not update BIGTK receive counter sufficiently")
1032
1033     tx_spec = int(ap_key['tx_spec'], base=16)
1034     if tx_spec < 3:
1035         raise Exception("AP did not update BIGTK BIPN sufficiently")
1036
1037     valid_bip = wt.get_bss_counter('valid_bip_mmie', bssid)
1038     invalid_bip = wt.get_bss_counter('invalid_bip_mmie', bssid)
1039     missing_bip = wt.get_bss_counter('missing_bip_mmie', bssid)
1040     logger.info("wlantest BIP counters: valid=%d invalid=%d missing=%d" % (valid_bip, invalid_bip, missing_bip))
1041     if valid_bip < 0 or invalid_bip > 0 or missing_bip > 0:
1042         raise Exception("Unexpected wlantest BIP counters: valid=%d invalid=%d missing=%d" % (valid_bip, invalid_bip, missing_bip))