tests/hwsim/test_mbo.py

   1 # MBO tests
   2 # Copyright (c) 2016, Intel Deutschland GmbH
   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 logging
   9 logger = logging.getLogger()
  10
  11 import hostapd
  12 import os
  13 import time
  14
  15 import hostapd
  16 from tshark import run_tshark
  17 from utils import *
  18
  19 def set_reg(country_code, apdev0=None, apdev1=None, dev0=None):
  20     if apdev0:
  21         hostapd.cmd_execute(apdev0, ['iw', 'reg', 'set', country_code])
  22     if apdev1:
  23         hostapd.cmd_execute(apdev1, ['iw', 'reg', 'set', country_code])
  24     if dev0:
  25         dev0.cmd_execute(['iw', 'reg', 'set', country_code])
  26
  27 def run_mbo_supp_oper_classes(dev, apdev, hapd, hapd2, country, freq_list=None,
  28                               disable_ht=False, disable_vht=False):
  29     """MBO and supported operating classes"""
  30     addr = dev[0].own_addr()
  31
  32     res2 = None
  33     res5 = None
  34
  35     dev[0].flush_scan_cache()
  36     dev[0].dump_monitor()
  37
  38     logger.info("Country: " + country)
  39     dev[0].note("Setting country code " + country)
  40     set_reg(country, apdev[0], apdev[1], dev[0])
  41     for j in range(5):
  42         ev = dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=5)
  43         if ev is None:
  44             raise Exception("No regdom change event")
  45         if "alpha2=" + country in ev:
  46             break
  47     dev[0].dump_monitor()
  48     dev[1].dump_monitor()
  49     dev[2].dump_monitor()
  50     _disable_ht = "1" if disable_ht else "0"
  51     _disable_vht = "1" if disable_vht else "0"
  52     if hapd:
  53         hapd.set("country_code", country)
  54         hapd.enable()
  55         dev[0].scan_for_bss(hapd.own_addr(), 5180, force_scan=True)
  56         dev[0].connect("test-wnm-mbo", key_mgmt="NONE", scan_freq="5180",
  57                        freq_list=freq_list, disable_ht=_disable_ht,
  58                        disable_vht=_disable_vht)
  59         sta = hapd.get_sta(addr)
  60         res5 = sta['supp_op_classes'][2:]
  61         dev[0].wait_regdom(country_ie=True)
  62         time.sleep(0.1)
  63         hapd.disable()
  64         time.sleep(0.1)
  65         dev[0].request("REMOVE_NETWORK all")
  66         dev[0].request("ABORT_SCAN")
  67         dev[0].wait_disconnected()
  68         dev[0].dump_monitor()
  69
  70     hapd2.set("country_code", country)
  71     hapd2.enable()
  72     dev[0].scan_for_bss(hapd2.own_addr(), 2412, force_scan=True)
  73     dev[0].connect("test-wnm-mbo-2", key_mgmt="NONE", scan_freq="2412",
  74                    freq_list=freq_list, disable_ht=_disable_ht,
  75                    disable_vht=_disable_vht)
  76     sta = hapd2.get_sta(addr)
  77     res2 = sta['supp_op_classes'][2:]
  78     dev[0].wait_regdom(country_ie=True)
  79     time.sleep(0.1)
  80     hapd2.disable()
  81     time.sleep(0.1)
  82     dev[0].request("REMOVE_NETWORK all")
  83     dev[0].request("ABORT_SCAN")
  84     dev[0].wait_disconnected()
  85     dev[0].dump_monitor()
  86
  87     return res2, res5
  88
  89 def run_mbo_supp_oper_class(dev, apdev, country, expected, inc5,
  90                             freq_list=None, disable_ht=False,
  91                             disable_vht=False):
  92     if inc5:
  93         params = {'ssid': "test-wnm-mbo",
  94                   'mbo': '1',
  95                   "country_code": "US",
  96                   'ieee80211d': '1',
  97                   "ieee80211n": "1",
  98                   "hw_mode": "a",
  99                   "channel": "36"}
 100         hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
 101     else:
 102         hapd = None
 103
 104     params = {'ssid': "test-wnm-mbo-2",
 105               'mbo': '1',
 106               "country_code": "US",
 107               'ieee80211d': '1',
 108               "ieee80211n": "1",
 109               "hw_mode": "g",
 110               "channel": "1"}
 111     hapd2 = hostapd.add_ap(apdev[1], params, no_enable=True)
 112
 113     try:
 114         dev[0].request("STA_AUTOCONNECT 0")
 115         res2, res5 = run_mbo_supp_oper_classes(dev, apdev, hapd, hapd2, country,
 116                                                freq_list=freq_list,
 117                                                disable_ht=disable_ht,
 118                                                disable_vht=disable_vht)
 119     finally:
 120         dev[0].dump_monitor()
 121         dev[0].request("STA_AUTOCONNECT 1")
 122         wait_regdom_changes(dev[0])
 123         country1 = dev[0].get_driver_status_field("country")
 124         logger.info("Country code at the end (1): " + country1)
 125         set_reg("00", apdev[0], apdev[1], dev[0])
 126         country2 = dev[0].get_driver_status_field("country")
 127         logger.info("Country code at the end (2): " + country2)
 128         for i in range(5):
 129             ev = dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=1)
 130             if ev is None or "init=USER type=WORLD" in ev:
 131                 break
 132         wait_regdom_changes(dev[0])
 133         country3 = dev[0].get_driver_status_field("country")
 134         logger.info("Country code at the end (3): " + country3)
 135         if country3 != "00":
 136             clear_country(dev)
 137
 138     # For now, allow operating class 129 to be missing since not all
 139     # installed regdb files include the 160 MHz channels.
 140     expected2 = expected.replace('808182', '8082')
 141     # For now, allow operating classes 121-123 to be missing since not all
 142     # installed regdb files include the related US DFS channels.
 143     expected2 = expected2.replace('78797a7b7c', '787c')
 144     expected3 = expected
 145     # For now, allow operating classes 124-127 to be missing for Finland
 146     # since they were added only recently in regdb.
 147     if country == "FI":
 148         expected3 = expected3.replace("7b7c7d7e7f80", "7b80")
 149     if res2 != expected and res2 != expected2 and res2 != expected3:
 150         raise Exception("Unexpected supp_op_class string (country=%s, 2.4 GHz): %s (expected: %s)" % (country, res2, expected))
 151     if inc5 and res5 != expected and res5 != expected2 and res5 != expected3:
 152         raise Exception("Unexpected supp_op_class string (country=%s, 5 GHz): %s (expected: %s)" % (country, res5, expected))
 153
 154 def test_mbo_supp_oper_classes_za(dev, apdev):
 155     """MBO and supported operating classes (ZA)"""
 156     run_mbo_supp_oper_class(dev, apdev, "ZA",
 157                             "515354737475767778797a7b808182", True)
 158
 159 def test_mbo_supp_oper_classes_fi(dev, apdev):
 160     """MBO and supported operating classes (FI)"""
 161     run_mbo_supp_oper_class(dev, apdev, "FI",
 162                             "515354737475767778797a7b7c7d7e7f808182", True)
 163
 164 def test_mbo_supp_oper_classes_us(dev, apdev):
 165     """MBO and supported operating classes (US)"""
 166     run_mbo_supp_oper_class(dev, apdev, "US",
 167                             "515354737475767778797a7b7c7d7e7f808182", True)
 168
 169 def test_mbo_supp_oper_classes_jp(dev, apdev):
 170     """MBO and supported operating classes (JP)"""
 171     run_mbo_supp_oper_class(dev, apdev, "JP",
 172                             "51525354737475767778797a7b808182", True)
 173
 174 def test_mbo_supp_oper_classes_bd(dev, apdev):
 175     """MBO and supported operating classes (BD)"""
 176     run_mbo_supp_oper_class(dev, apdev, "BD",
 177                             "5153547c7d7e7f80", False)
 178
 179 def test_mbo_supp_oper_classes_sy(dev, apdev):
 180     """MBO and supported operating classes (SY)"""
 181     run_mbo_supp_oper_class(dev, apdev, "SY",
 182                             "515354", False)
 183
 184 def test_mbo_supp_oper_classes_us_freq_list(dev, apdev):
 185     """MBO and supported operating classes (US) - freq_list"""
 186     run_mbo_supp_oper_class(dev, apdev, "US", "515354", False,
 187                             freq_list="2412 2437 2462")
 188
 189 def test_mbo_supp_oper_classes_us_disable_ht(dev, apdev):
 190     """MBO and supported operating classes (US) - disable_ht"""
 191     run_mbo_supp_oper_class(dev, apdev, "US", "517376797c7d", False,
 192                             disable_ht=True)
 193
 194 def test_mbo_supp_oper_classes_us_disable_vht(dev, apdev):
 195     """MBO and supported operating classes (US) - disable_vht"""
 196     run_mbo_supp_oper_class(dev, apdev, "US",
 197                             "515354737475767778797a7b7c7d7e7f", False,
 198                             disable_vht=True)
 199
 200 def test_mbo_assoc_disallow(dev, apdev, params):
 201     """MBO and association disallowed"""
 202     hapd1 = hostapd.add_ap(apdev[0], {"ssid": "MBO", "mbo": "1"})
 203     hapd2 = hostapd.add_ap(apdev[1], {"ssid": "MBO", "mbo": "1"})
 204
 205     logger.debug("Set mbo_assoc_disallow with invalid value")
 206     if "FAIL" not in hapd1.request("SET mbo_assoc_disallow 2"):
 207         raise Exception("Set mbo_assoc_disallow for AP1 succeeded unexpectedly with value 2")
 208
 209     logger.debug("Disallow associations to AP1 and allow association to AP2")
 210     if "OK" not in hapd1.request("SET mbo_assoc_disallow 1"):
 211         raise Exception("Failed to set mbo_assoc_disallow for AP1")
 212     if "OK" not in hapd2.request("SET mbo_assoc_disallow 0"):
 213         raise Exception("Failed to set mbo_assoc_disallow for AP2")
 214
 215     dev[0].connect("MBO", key_mgmt="NONE", scan_freq="2412")
 216
 217     out = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
 218                      "wlan.fc.type == 0 && wlan.fc.type_subtype == 0x00",
 219                      wait=False)
 220     if "Destination address: " + hapd1.own_addr() in out:
 221         raise Exception("Association request sent to disallowed AP")
 222
 223     timestamp = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
 224                            "wlan.fc.type_subtype == 0x00",
 225                            display=['frame.time'], wait=False)
 226
 227     logger.debug("Allow associations to AP1 and disallow associations to AP2")
 228     if "OK" not in hapd1.request("SET mbo_assoc_disallow 0"):
 229         raise Exception("Failed to set mbo_assoc_disallow for AP1")
 230     if "OK" not in hapd2.request("SET mbo_assoc_disallow 1"):
 231         raise Exception("Failed to set mbo_assoc_disallow for AP2")
 232
 233     dev[0].request("DISCONNECT")
 234     dev[0].wait_disconnected()
 235
 236     # Force new scan, so the assoc_disallowed indication is updated */
 237     dev[0].request("FLUSH")
 238
 239     dev[0].connect("MBO", key_mgmt="NONE", scan_freq="2412")
 240
 241     filter = 'wlan.fc.type == 0 && wlan.fc.type_subtype == 0x00 && frame.time > "' + timestamp.rstrip() + '"'
 242     out = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
 243                      filter, wait=False)
 244     if "Destination address: " + hapd2.own_addr() in out:
 245         raise Exception("Association request sent to disallowed AP 2")
 246
 247 def test_mbo_assoc_disallow_ignore(dev, apdev):
 248     """MBO and ignoring disallowed association"""
 249     try:
 250         _test_mbo_assoc_disallow_ignore(dev, apdev)
 251     finally:
 252         dev[0].request("SCAN_INTERVAL 5")
 253
 254 def _test_mbo_assoc_disallow_ignore(dev, apdev):
 255     hapd1 = hostapd.add_ap(apdev[0], {"ssid": "MBO", "mbo": "1"})
 256     if "OK" not in hapd1.request("SET mbo_assoc_disallow 1"):
 257         raise Exception("Failed to set mbo_assoc_disallow for AP1")
 258
 259     if "OK" not in dev[0].request("SCAN_INTERVAL 1"):
 260         raise Exception("Failed to set scan interval")
 261     dev[0].connect("MBO", key_mgmt="NONE", scan_freq="2412", wait_connect=False)
 262     ev = dev[0].wait_event(["CTRL-EVENT-NETWORK-NOT-FOUND"], timeout=10)
 263     if ev is None:
 264         raise Exception("CTRL-EVENT-NETWORK-NOT-FOUND not seen")
 265
 266     if "OK" not in dev[0].request("SET ignore_assoc_disallow 1"):
 267         raise Exception("Failed to set ignore_assoc_disallow")
 268     ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"], timeout=10)
 269     if ev is None:
 270         raise Exception("CTRL-EVENT-ASSOC-REJECT not seen")
 271     if "status_code=17" not in ev:
 272         raise Exception("Unexpected association reject reason: " + ev)
 273
 274     if "OK" not in hapd1.request("SET mbo_assoc_disallow 0"):
 275         raise Exception("Failed to set mbo_assoc_disallow for AP1")
 276     dev[0].wait_connected()
 277
 278 @remote_compatible
 279 def test_mbo_cell_capa_update(dev, apdev):
 280     """MBO cellular data capability update"""
 281     ssid = "test-wnm-mbo"
 282     params = {'ssid': ssid, 'mbo': '1'}
 283     hapd = hostapd.add_ap(apdev[0], params)
 284     bssid = apdev[0]['bssid']
 285     if "OK" not in dev[0].request("SET mbo_cell_capa 1"):
 286         raise Exception("Failed to set STA as cellular data capable")
 287
 288     dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 289
 290     addr = dev[0].own_addr()
 291     sta = hapd.get_sta(addr)
 292     if 'mbo_cell_capa' not in sta or sta['mbo_cell_capa'] != '1':
 293         raise Exception("mbo_cell_capa missing after association")
 294
 295     if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 296         raise Exception("Failed to set STA as cellular data not-capable")
 297     # Duplicate update for additional code coverage
 298     if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 299         raise Exception("Failed to set STA as cellular data not-capable")
 300
 301     time.sleep(0.2)
 302     sta = hapd.get_sta(addr)
 303     if 'mbo_cell_capa' not in sta:
 304         raise Exception("mbo_cell_capa missing after update")
 305     if sta['mbo_cell_capa'] != '3':
 306         raise Exception("mbo_cell_capa not updated properly")
 307
 308 @remote_compatible
 309 def test_mbo_cell_capa_update_pmf(dev, apdev):
 310     """MBO cellular data capability update with PMF required"""
 311     ssid = "test-wnm-mbo"
 312     passphrase = "12345678"
 313     params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase)
 314     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 315     params["ieee80211w"] = "2"
 316     params['mbo'] = '1'
 317     hapd = hostapd.add_ap(apdev[0], params)
 318     bssid = apdev[0]['bssid']
 319     if "OK" not in dev[0].request("SET mbo_cell_capa 1"):
 320         raise Exception("Failed to set STA as cellular data capable")
 321
 322     dev[0].connect(ssid, psk=passphrase, key_mgmt="WPA-PSK-SHA256",
 323                    proto="WPA2", ieee80211w="2", scan_freq="2412")
 324     hapd.wait_sta()
 325
 326     addr = dev[0].own_addr()
 327     sta = hapd.get_sta(addr)
 328     if 'mbo_cell_capa' not in sta or sta['mbo_cell_capa'] != '1':
 329         raise Exception("mbo_cell_capa missing after association")
 330
 331     if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 332         raise Exception("Failed to set STA as cellular data not-capable")
 333
 334     time.sleep(0.2)
 335     sta = hapd.get_sta(addr)
 336     if 'mbo_cell_capa' not in sta:
 337         raise Exception("mbo_cell_capa missing after update")
 338     if sta['mbo_cell_capa'] != '3':
 339         raise Exception("mbo_cell_capa not updated properly")
 340
 341 def test_mbo_wnm_token_wrap(dev, apdev):
 342     """MBO WNM token wrap around"""
 343     ssid = "test-wnm-mbo"
 344     params = {'ssid': ssid, 'mbo': '1'}
 345     hapd = hostapd.add_ap(apdev[0], params)
 346     bssid = apdev[0]['bssid']
 347
 348     dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 349
 350     # Trigger transmission of 256 WNM-Notification frames to wrap around the
 351     # 8-bit mbo_wnm_token counter.
 352     for i in range(128):
 353         if "OK" not in dev[0].request("SET mbo_cell_capa 1"):
 354             raise Exception("Failed to set STA as cellular data capable")
 355         if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 356             raise Exception("Failed to set STA as cellular data not-capable")
 357
 358 @remote_compatible
 359 def test_mbo_non_pref_chan(dev, apdev):
 360     """MBO non-preferred channel list"""
 361     ssid = "test-wnm-mbo"
 362     params = {'ssid': ssid, 'mbo': '1'}
 363     hapd = hostapd.add_ap(apdev[0], params)
 364     bssid = apdev[0]['bssid']
 365     if "FAIL" not in dev[0].request("SET non_pref_chan 81:7:200:99"):
 366         raise Exception("Invalid non_pref_chan value accepted")
 367     if "FAIL" not in dev[0].request("SET non_pref_chan 81:15:200:3"):
 368         raise Exception("Invalid non_pref_chan value accepted")
 369     if "FAIL" not in dev[0].request("SET non_pref_chan 81:7:200:3 81:7:201:3"):
 370         raise Exception("Invalid non_pref_chan value accepted")
 371     if "OK" not in dev[0].request("SET non_pref_chan 81:7:200:3"):
 372         raise Exception("Failed to set non-preferred channel list")
 373     if "OK" not in dev[0].request("SET non_pref_chan 81:7:200:1 81:9:100:2"):
 374         raise Exception("Failed to set non-preferred channel list")
 375
 376     dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 377
 378     addr = dev[0].own_addr()
 379     sta = hapd.get_sta(addr)
 380     logger.debug("STA: " + str(sta))
 381     if 'non_pref_chan[0]' not in sta:
 382         raise Exception("Missing non_pref_chan[0] value (assoc)")
 383     if sta['non_pref_chan[0]'] != '81:200:1:7':
 384         raise Exception("Unexpected non_pref_chan[0] value (assoc)")
 385     if 'non_pref_chan[1]' not in sta:
 386         raise Exception("Missing non_pref_chan[1] value (assoc)")
 387     if sta['non_pref_chan[1]'] != '81:100:2:9':
 388         raise Exception("Unexpected non_pref_chan[1] value (assoc)")
 389     if 'non_pref_chan[2]' in sta:
 390         raise Exception("Unexpected non_pref_chan[2] value (assoc)")
 391
 392     if "OK" not in dev[0].request("SET non_pref_chan 81:9:100:2"):
 393         raise Exception("Failed to update non-preferred channel list")
 394     time.sleep(0.1)
 395     sta = hapd.get_sta(addr)
 396     logger.debug("STA: " + str(sta))
 397     if 'non_pref_chan[0]' not in sta:
 398         raise Exception("Missing non_pref_chan[0] value (update 1)")
 399     if sta['non_pref_chan[0]'] != '81:100:2:9':
 400         raise Exception("Unexpected non_pref_chan[0] value (update 1)")
 401     if 'non_pref_chan[1]' in sta:
 402         raise Exception("Unexpected non_pref_chan[1] value (update 1)")
 403
 404     if "OK" not in dev[0].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"):
 405         raise Exception("Failed to update non-preferred channel list")
 406     time.sleep(0.1)
 407     sta = hapd.get_sta(addr)
 408     logger.debug("STA: " + str(sta))
 409     if 'non_pref_chan[0]' not in sta:
 410         raise Exception("Missing non_pref_chan[0] value (update 2)")
 411     if sta['non_pref_chan[0]'] != '81:100:1:7,5':
 412         raise Exception("Unexpected non_pref_chan[0] value (update 2)")
 413     if 'non_pref_chan[1]' not in sta:
 414         raise Exception("Missing non_pref_chan[1] value (update 2)")
 415     if sta['non_pref_chan[1]'] != '81:100:2:9,10,8':
 416         raise Exception("Unexpected non_pref_chan[1] value (update 2)")
 417     if 'non_pref_chan[2]' in sta:
 418         raise Exception("Unexpected non_pref_chan[2] value (update 2)")
 419
 420     if "OK" not in dev[0].request("SET non_pref_chan 81:5:90:2 82:14:91:2"):
 421         raise Exception("Failed to update non-preferred channel list")
 422     time.sleep(0.1)
 423     sta = hapd.get_sta(addr)
 424     logger.debug("STA: " + str(sta))
 425     if 'non_pref_chan[0]' not in sta:
 426         raise Exception("Missing non_pref_chan[0] value (update 3)")
 427     if sta['non_pref_chan[0]'] != '81:90:2:5':
 428         raise Exception("Unexpected non_pref_chan[0] value (update 3)")
 429     if 'non_pref_chan[1]' not in sta:
 430         raise Exception("Missing non_pref_chan[1] value (update 3)")
 431     if sta['non_pref_chan[1]'] != '82:91:2:14':
 432         raise Exception("Unexpected non_pref_chan[1] value (update 3)")
 433     if 'non_pref_chan[2]' in sta:
 434         raise Exception("Unexpected non_pref_chan[2] value (update 3)")
 435
 436     if "OK" not in dev[0].request("SET non_pref_chan "):
 437         raise Exception("Failed to update non-preferred channel list")
 438     time.sleep(0.1)
 439     sta = hapd.get_sta(addr)
 440     logger.debug("STA: " + str(sta))
 441     if 'non_pref_chan[0]' in sta:
 442         raise Exception("Unexpected non_pref_chan[0] value (update 4)")
 443
 444 @remote_compatible
 445 def test_mbo_sta_supp_op_classes(dev, apdev):
 446     """MBO STA supported operating classes"""
 447     ssid = "test-wnm-mbo"
 448     params = {'ssid': ssid, 'mbo': '1'}
 449     hapd = hostapd.add_ap(apdev[0], params)
 450
 451     dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 452
 453     addr = dev[0].own_addr()
 454     sta = hapd.get_sta(addr)
 455     logger.debug("STA: " + str(sta))
 456     if 'supp_op_classes' not in sta:
 457         raise Exception("No supp_op_classes")
 458     supp = bytearray(binascii.unhexlify(sta['supp_op_classes']))
 459     if supp[0] != 81:
 460         raise Exception("Unexpected current operating class %d" % supp[0])
 461     if 115 not in supp:
 462         raise Exception("Operating class 115 missing")
 463
 464 def test_mbo_failures(dev, apdev):
 465     """MBO failure cases"""
 466     ssid = "test-wnm-mbo"
 467     params = {'ssid': ssid, 'mbo': '1'}
 468     hapd = hostapd.add_ap(apdev[0], params)
 469
 470     with alloc_fail(dev[0], 1, "wpas_mbo_ie"):
 471         dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 472
 473     with alloc_fail(dev[0], 1, "wpas_mbo_send_wnm_notification"):
 474         if "OK" not in dev[0].request("SET mbo_cell_capa 1"):
 475             raise Exception("Failed to set STA as cellular data capable")
 476     with fail_test(dev[0], 1, "wpas_mbo_send_wnm_notification"):
 477         if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 478             raise Exception("Failed to set STA as cellular data not-capable")
 479     with alloc_fail(dev[0], 1, "wpas_mbo_update_non_pref_chan"):
 480         if "FAIL" not in dev[0].request("SET non_pref_chan 81:7:200:3"):
 481             raise Exception("non_pref_chan value accepted during OOM")
 482     with alloc_fail(dev[0], 2, "wpas_mbo_update_non_pref_chan"):
 483         if "FAIL" not in dev[0].request("SET non_pref_chan 81:7:200:3"):
 484             raise Exception("non_pref_chan value accepted during OOM")
 485
 486 def test_mbo_wnm_bss_tm_ie_parsing(dev, apdev):
 487     """MBO BSS transition request MBO IE parsing"""
 488     ssid = "test-wnm-mbo"
 489     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 490     hapd = hostapd.add_ap(apdev[0], params)
 491     bssid = apdev[0]['bssid']
 492     addr = dev[0].own_addr()
 493     dev[0].connect(ssid, psk="12345678", key_mgmt="WPA-PSK",
 494                    proto="WPA2", ieee80211w="0", scan_freq="2412")
 495
 496     dev[0].request("SET ext_mgmt_frame_handling 1")
 497     hdr = "d0003a01" + addr.replace(':', '') + bssid.replace(':', '') + bssid.replace(':', '') + "3000"
 498     btm_hdr = "0a070100030001"
 499
 500     tests = [("Truncated attribute in MBO IE", "dd06506f9a160101"),
 501              ("Unexpected cell data capa attribute length in MBO IE",
 502               "dd09506f9a160501030500"),
 503              ("Unexpected transition reason attribute length in MBO IE",
 504               "dd06506f9a160600"),
 505              ("Unexpected assoc retry delay attribute length in MBO IE",
 506               "dd0c506f9a160100080200000800"),
 507              ("Unknown attribute id 255 in MBO IE",
 508               "dd06506f9a16ff00")]
 509
 510     for test, mbo_ie in tests:
 511         logger.info(test)
 512         dev[0].request("NOTE " + test)
 513         frame = hdr + btm_hdr + mbo_ie
 514         if "OK" not in dev[0].request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=" + frame):
 515             raise Exception("MGMT_RX_PROCESS failed")
 516
 517     logger.info("Unexpected association retry delay")
 518     dev[0].request("NOTE Unexpected association retry delay")
 519     btm_hdr = "0a070108030001112233445566778899aabbcc"
 520     mbo_ie = "dd08506f9a1608020000"
 521     frame = hdr + btm_hdr + mbo_ie
 522     if "OK" not in dev[0].request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=" + frame):
 523         raise Exception("MGMT_RX_PROCESS failed")
 524
 525     dev[0].request("SET ext_mgmt_frame_handling 0")
 526
 527 def test_mbo_without_pmf(dev, apdev):
 528     """MBO and WPA2 without PMF"""
 529     ssid = "test-wnm-mbo"
 530     params = {'ssid': ssid, 'mbo': '1', "wpa": '2',
 531               "wpa_key_mgmt": "WPA-PSK", "rsn_pairwise": "CCMP",
 532               "wpa_passphrase": "12345678"}
 533     try:
 534         # "MBO: PMF needs to be enabled whenever using WPA2 with MBO"
 535         hostapd.add_ap(apdev[0], params)
 536         raise Exception("AP setup succeeded unexpectedly")
 537     except Exception as e:
 538         if "Failed to enable hostapd" in str(e):
 539             pass
 540         else:
 541             raise
 542
 543 def test_mbo_without_pmf_workaround(dev, apdev):
 544     """MBO and WPA2 without PMF on misbehaving AP"""
 545     ssid = "test-wnm-mbo"
 546     params = {'ssid': ssid, "wpa": '2',
 547               "wpa_key_mgmt": "WPA-PSK", "rsn_pairwise": "CCMP",
 548               "wpa_passphrase": "12345678",
 549               "vendor_elements": "dd07506f9a16010100"}
 550     hapd = hostapd.add_ap(apdev[0], params)
 551     dev[0].connect(ssid, psk="12345678", key_mgmt="WPA-PSK",
 552                    proto="WPA2", ieee80211w="1", scan_freq="2412")
 553     hapd.wait_sta()
 554     sta = hapd.get_sta(dev[0].own_addr())
 555     ext_capab = bytearray(binascii.unhexlify(sta['ext_capab']))
 556     if ext_capab[2] & 0x08:
 557         raise Exception("STA did not disable BSS Transition capability")
 558
 559 def check_mbo_anqp(dev, bssid, cell_data_conn_pref):
 560     if "OK" not in dev.request("ANQP_GET " + bssid + " 272,mbo:2"):
 561         raise Exception("ANQP_GET command failed")
 562
 563     ev = dev.wait_event(["GAS-QUERY-START"], timeout=5)
 564     if ev is None:
 565         raise Exception("GAS query start timed out")
 566
 567     ev = dev.wait_event(["GAS-QUERY-DONE"], timeout=10)
 568     if ev is None:
 569         raise Exception("GAS query timed out")
 570
 571     if cell_data_conn_pref is not None:
 572         ev = dev.wait_event(["RX-MBO-ANQP"], timeout=1)
 573         if ev is None or "cell_conn_pref" not in ev:
 574             raise Exception("Did not receive MBO Cellular Data Connection Preference")
 575         if cell_data_conn_pref != int(ev.split('=')[1]):
 576             raise Exception("Unexpected cell_conn_pref value: " + ev)
 577
 578     dev.dump_monitor()
 579
 580 def test_mbo_anqp(dev, apdev):
 581     """MBO ANQP"""
 582     params = {'ssid': "test-wnm-mbo",
 583               'mbo': '1',
 584               'interworking': '1',
 585               'mbo_cell_data_conn_pref': '1'}
 586     hapd = hostapd.add_ap(apdev[0], params)
 587     bssid = hapd.own_addr()
 588
 589     dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
 590     check_mbo_anqp(dev[0], bssid, 1)
 591
 592     hapd.set('mbo_cell_data_conn_pref', '255')
 593     check_mbo_anqp(dev[0], bssid, 255)
 594
 595     hapd.set('mbo_cell_data_conn_pref', '-1')
 596     check_mbo_anqp(dev[0], bssid, None)