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 alloc_fail, fail_test
  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):
  28     """MBO and supported operating classes"""
  29     addr = dev[0].own_addr()
  30
  31     res2 = None
  32     res5 = None
  33
  34     dev[0].flush_scan_cache()
  35     dev[0].dump_monitor()
  36
  37     logger.info("Country: " + country)
  38     set_reg(country, apdev[0], apdev[1], dev[0])
  39     for j in range(5):
  40         ev = dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=5)
  41         if ev is None:
  42             raise Exception("No regdom change event")
  43         if "alpha2=" + country in ev:
  44             break
  45     dev[0].dump_monitor()
  46     dev[1].dump_monitor()
  47     dev[2].dump_monitor()
  48     if hapd:
  49         hapd.set("country_code", country)
  50         hapd.enable()
  51         dev[0].scan_for_bss(hapd.own_addr(), 5180, force_scan=True)
  52         dev[0].connect("test-wnm-mbo", key_mgmt="NONE", scan_freq="5180")
  53         sta = hapd.get_sta(addr)
  54         res5 = sta['supp_op_classes'][2:]
  55         dev[0].request("REMOVE_NETWORK all")
  56         hapd.disable()
  57         dev[0].wait_disconnected()
  58         dev[0].dump_monitor()
  59
  60     hapd2.set("country_code", country)
  61     hapd2.enable()
  62     dev[0].scan_for_bss(hapd2.own_addr(), 2412, force_scan=True)
  63     dev[0].connect("test-wnm-mbo-2", key_mgmt="NONE", scan_freq="2412")
  64     sta = hapd2.get_sta(addr)
  65     res2 = sta['supp_op_classes'][2:]
  66     dev[0].request("REMOVE_NETWORK all")
  67     hapd2.disable()
  68     dev[0].wait_disconnected()
  69     dev[0].dump_monitor()
  70
  71     return res2, res5
  72
  73 def test_mbo_supp_oper_classes(dev, apdev):
  74     """MBO and supported operating classes"""
  75     params = { 'ssid': "test-wnm-mbo",
  76                'mbo': '1',
  77                "country_code": "US",
  78                'ieee80211d': '1',
  79                "ieee80211n": "1",
  80                "hw_mode": "a",
  81                "channel": "36" }
  82     hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
  83
  84     params = { 'ssid': "test-wnm-mbo-2",
  85                'mbo': '1',
  86                "country_code": "US",
  87                'ieee80211d': '1',
  88                "ieee80211n": "1",
  89                "hw_mode": "g",
  90                "channel": "1" }
  91     hapd2 = hostapd.add_ap(apdev[1], params, no_enable=True)
  92
  93     try:
  94         za2, za5 = run_mbo_supp_oper_classes(dev, apdev, hapd, hapd2, "ZA")
  95         fi2, fi5 = run_mbo_supp_oper_classes(dev, apdev, hapd, hapd2, "FI")
  96         us2, us5 = run_mbo_supp_oper_classes(dev, apdev, hapd, hapd2, "US")
  97         jp2, jp5 = run_mbo_supp_oper_classes(dev, apdev, hapd, hapd2, "JP")
  98         bd2, bd5 = run_mbo_supp_oper_classes(dev, apdev, None, hapd2, "BD")
  99         kz2, kz5 = run_mbo_supp_oper_classes(dev, apdev, None, hapd2, "KZ")
 100     finally:
 101         dev[0].dump_monitor()
 102         set_reg("00", apdev[0], apdev[1], dev[0])
 103         ev = dev[0].wait_event(["CTRL-EVENT-REGDOM-CHANGE"], timeout=1)
 104
 105     za = "515354737475767778797a7b808182"
 106     fi = "515354737475767778797a7b808182"
 107     us = "515354737475767778797a7b7c7d7e7f808182"
 108     jp = "51525354737475767778797a7b808182"
 109     bd = "5153547c7d7e7f80"
 110     kz = "515354"
 111
 112     tests = [ ("ZA", za, za2, za5, True),
 113               ("FI", fi, fi2, fi5, True),
 114               ("US", us, us2, us5, True),
 115               ("JP", jp, jp2, jp5, True),
 116               ("BD", bd, bd2, bd5, False),
 117               ("KZ", kz, kz2, kz5, False) ]
 118     for country, expected, res2, res5, inc5 in tests:
 119         # For now, allow operating class 129 to be missing since not all
 120         # installed regdb files include the 160 MHz channels.
 121         expected2 = expected.replace('808182', '8082')
 122         # For now, allow operating classes 121-123 to be missing since not all
 123         # installed regdb files include the related US DFS channels.
 124         expected2 = expected2.replace('78797a7b7c', '787c')
 125         if res2 != expected and res2 != expected2:
 126             raise Exception("Unexpected supp_op_class string (country=%s, 2.4 GHz): %s (expected: %s)" % (country, res2, expected))
 127         if inc5 and res5 != expected and res5 != expected2:
 128             raise Exception("Unexpected supp_op_class string (country=%s, 5 GHz): %s (expected: %s)" % (country, res5, expected))
 129
 130 def test_mbo_assoc_disallow(dev, apdev, params):
 131     hapd1 = hostapd.add_ap(apdev[0], { "ssid": "MBO", "mbo": "1" })
 132     hapd2 = hostapd.add_ap(apdev[1], { "ssid": "MBO", "mbo": "1" })
 133
 134     logger.debug("Set mbo_assoc_disallow with invalid value")
 135     if "FAIL" not in hapd1.request("SET mbo_assoc_disallow 2"):
 136         raise Exception("Set mbo_assoc_disallow for AP1 succeeded unexpectedly with value 2")
 137
 138     logger.debug("Disallow associations to AP1 and allow association to AP2")
 139     if "OK" not in hapd1.request("SET mbo_assoc_disallow 1"):
 140         raise Exception("Failed to set mbo_assoc_disallow for AP1")
 141     if "OK" not in hapd2.request("SET mbo_assoc_disallow 0"):
 142         raise Exception("Failed to set mbo_assoc_disallow for AP2")
 143
 144     dev[0].connect("MBO", key_mgmt="NONE", scan_freq="2412")
 145
 146     out = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
 147                      "wlan.fc.type == 0 && wlan.fc.type_subtype == 0x00",
 148                      wait=False)
 149     if "Destination address: " + hapd1.own_addr() in out:
 150         raise Exception("Association request sent to disallowed AP")
 151
 152     timestamp = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
 153                            "wlan.fc.type_subtype == 0x00",
 154                            display=['frame.time'], wait=False)
 155
 156     logger.debug("Allow associations to AP1 and disallow assications to AP2")
 157     if "OK" not in hapd1.request("SET mbo_assoc_disallow 0"):
 158         raise Exception("Failed to set mbo_assoc_disallow for AP1")
 159     if "OK" not in hapd2.request("SET mbo_assoc_disallow 1"):
 160         raise Exception("Failed to set mbo_assoc_disallow for AP2")
 161
 162     dev[0].request("DISCONNECT")
 163     dev[0].wait_disconnected()
 164
 165     # Force new scan, so the assoc_disallowed indication is updated */
 166     dev[0].request("FLUSH")
 167
 168     dev[0].connect("MBO", key_mgmt="NONE", scan_freq="2412")
 169
 170     filter = 'wlan.fc.type == 0 && wlan.fc.type_subtype == 0x00 && frame.time > "' + timestamp.rstrip() + '"'
 171     out = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
 172                      filter, wait=False)
 173     if "Destination address: " + hapd2.own_addr() in out:
 174         raise Exception("Association request sent to disallowed AP 2")
 175
 176 @remote_compatible
 177 def test_mbo_cell_capa_update(dev, apdev):
 178     """MBO cellular data capability update"""
 179     ssid = "test-wnm-mbo"
 180     params = { 'ssid': ssid, 'mbo': '1' }
 181     hapd = hostapd.add_ap(apdev[0], params)
 182     bssid = apdev[0]['bssid']
 183     if "OK" not in dev[0].request("SET mbo_cell_capa 1"):
 184         raise Exception("Failed to set STA as cellular data capable")
 185
 186     dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 187
 188     addr = dev[0].own_addr()
 189     sta = hapd.get_sta(addr)
 190     if 'mbo_cell_capa' not in sta or sta['mbo_cell_capa'] != '1':
 191         raise Exception("mbo_cell_capa missing after association")
 192
 193     if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 194         raise Exception("Failed to set STA as cellular data not-capable")
 195     # Duplicate update for additional code coverage
 196     if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 197         raise Exception("Failed to set STA as cellular data not-capable")
 198
 199     time.sleep(0.2)
 200     sta = hapd.get_sta(addr)
 201     if 'mbo_cell_capa' not in sta:
 202         raise Exception("mbo_cell_capa missing after update")
 203     if sta['mbo_cell_capa'] != '3':
 204         raise Exception("mbo_cell_capa not updated properly")
 205
 206 @remote_compatible
 207 def test_mbo_cell_capa_update_pmf(dev, apdev):
 208     """MBO cellular data capability update with PMF required"""
 209     ssid = "test-wnm-mbo"
 210     passphrase = "12345678"
 211     params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase)
 212     params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
 213     params["ieee80211w"] = "2"
 214     params['mbo'] = '1'
 215     hapd = hostapd.add_ap(apdev[0], params)
 216     bssid = apdev[0]['bssid']
 217     if "OK" not in dev[0].request("SET mbo_cell_capa 1"):
 218         raise Exception("Failed to set STA as cellular data capable")
 219
 220     dev[0].connect(ssid, psk=passphrase, key_mgmt="WPA-PSK-SHA256",
 221                    proto="WPA2", ieee80211w="2", scan_freq="2412")
 222
 223     addr = dev[0].own_addr()
 224     sta = hapd.get_sta(addr)
 225     if 'mbo_cell_capa' not in sta or sta['mbo_cell_capa'] != '1':
 226         raise Exception("mbo_cell_capa missing after association")
 227
 228     if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 229         raise Exception("Failed to set STA as cellular data not-capable")
 230
 231     time.sleep(0.2)
 232     sta = hapd.get_sta(addr)
 233     if 'mbo_cell_capa' not in sta:
 234         raise Exception("mbo_cell_capa missing after update")
 235     if sta['mbo_cell_capa'] != '3':
 236         raise Exception("mbo_cell_capa not updated properly")
 237
 238 def test_mbo_wnm_token_wrap(dev, apdev):
 239     """MBO WNM token wrap around"""
 240     ssid = "test-wnm-mbo"
 241     params = { 'ssid': ssid, 'mbo': '1' }
 242     hapd = hostapd.add_ap(apdev[0], params)
 243     bssid = apdev[0]['bssid']
 244
 245     dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 246
 247     # Trigger transmission of 256 WNM-Notification frames to wrap around the
 248     # 8-bit mbo_wnm_token counter.
 249     for i in range(128):
 250         if "OK" not in dev[0].request("SET mbo_cell_capa 1"):
 251             raise Exception("Failed to set STA as cellular data capable")
 252         if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 253             raise Exception("Failed to set STA as cellular data not-capable")
 254
 255 @remote_compatible
 256 def test_mbo_non_pref_chan(dev, apdev):
 257     """MBO non-preferred channel list"""
 258     ssid = "test-wnm-mbo"
 259     params = { 'ssid': ssid, 'mbo': '1' }
 260     hapd = hostapd.add_ap(apdev[0], params)
 261     bssid = apdev[0]['bssid']
 262     if "FAIL" not in dev[0].request("SET non_pref_chan 81:7:200:99"):
 263         raise Exception("Invalid non_pref_chan value accepted")
 264     if "FAIL" not in dev[0].request("SET non_pref_chan 81:15:200:3"):
 265         raise Exception("Invalid non_pref_chan value accepted")
 266     if "FAIL" not in dev[0].request("SET non_pref_chan 81:7:200:3 81:7:201:3"):
 267         raise Exception("Invalid non_pref_chan value accepted")
 268     if "OK" not in dev[0].request("SET non_pref_chan 81:7:200:3"):
 269         raise Exception("Failed to set non-preferred channel list")
 270     if "OK" not in dev[0].request("SET non_pref_chan 81:7:200:1 81:9:100:2"):
 271         raise Exception("Failed to set non-preferred channel list")
 272
 273     dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 274
 275     addr = dev[0].own_addr()
 276     sta = hapd.get_sta(addr)
 277     logger.debug("STA: " + str(sta))
 278     if 'non_pref_chan[0]' not in sta:
 279         raise Exception("Missing non_pref_chan[0] value (assoc)")
 280     if sta['non_pref_chan[0]'] != '81:200:1:7':
 281         raise Exception("Unexpected non_pref_chan[0] value (assoc)")
 282     if 'non_pref_chan[1]' not in sta:
 283         raise Exception("Missing non_pref_chan[1] value (assoc)")
 284     if sta['non_pref_chan[1]'] != '81:100:2:9':
 285         raise Exception("Unexpected non_pref_chan[1] value (assoc)")
 286     if 'non_pref_chan[2]' in sta:
 287         raise Exception("Unexpected non_pref_chan[2] value (assoc)")
 288
 289     if "OK" not in dev[0].request("SET non_pref_chan 81:9:100:2"):
 290         raise Exception("Failed to update non-preferred channel list")
 291     time.sleep(0.1)
 292     sta = hapd.get_sta(addr)
 293     logger.debug("STA: " + str(sta))
 294     if 'non_pref_chan[0]' not in sta:
 295         raise Exception("Missing non_pref_chan[0] value (update 1)")
 296     if sta['non_pref_chan[0]'] != '81:100:2:9':
 297         raise Exception("Unexpected non_pref_chan[0] value (update 1)")
 298     if 'non_pref_chan[1]' in sta:
 299         raise Exception("Unexpected non_pref_chan[1] value (update 1)")
 300
 301     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"):
 302         raise Exception("Failed to update non-preferred channel list")
 303     time.sleep(0.1)
 304     sta = hapd.get_sta(addr)
 305     logger.debug("STA: " + str(sta))
 306     if 'non_pref_chan[0]' not in sta:
 307         raise Exception("Missing non_pref_chan[0] value (update 2)")
 308     if sta['non_pref_chan[0]'] != '81:100:1:7,5':
 309         raise Exception("Unexpected non_pref_chan[0] value (update 2)")
 310     if 'non_pref_chan[1]' not in sta:
 311         raise Exception("Missing non_pref_chan[1] value (update 2)")
 312     if sta['non_pref_chan[1]'] != '81:100:2:9,10,8':
 313         raise Exception("Unexpected non_pref_chan[1] value (update 2)")
 314     if 'non_pref_chan[2]' in sta:
 315         raise Exception("Unexpected non_pref_chan[2] value (update 2)")
 316
 317     if "OK" not in dev[0].request("SET non_pref_chan 81:5:90:2 82:14:91:2"):
 318         raise Exception("Failed to update non-preferred channel list")
 319     time.sleep(0.1)
 320     sta = hapd.get_sta(addr)
 321     logger.debug("STA: " + str(sta))
 322     if 'non_pref_chan[0]' not in sta:
 323         raise Exception("Missing non_pref_chan[0] value (update 3)")
 324     if sta['non_pref_chan[0]'] != '81:90:2:5':
 325         raise Exception("Unexpected non_pref_chan[0] value (update 3)")
 326     if 'non_pref_chan[1]' not in sta:
 327         raise Exception("Missing non_pref_chan[1] value (update 3)")
 328     if sta['non_pref_chan[1]'] != '82:91:2:14':
 329         raise Exception("Unexpected non_pref_chan[1] value (update 3)")
 330     if 'non_pref_chan[2]' in sta:
 331         raise Exception("Unexpected non_pref_chan[2] value (update 3)")
 332
 333     if "OK" not in dev[0].request("SET non_pref_chan "):
 334         raise Exception("Failed to update non-preferred channel list")
 335     time.sleep(0.1)
 336     sta = hapd.get_sta(addr)
 337     logger.debug("STA: " + str(sta))
 338     if 'non_pref_chan[0]' in sta:
 339         raise Exception("Unexpected non_pref_chan[0] value (update 4)")
 340
 341 @remote_compatible
 342 def test_mbo_sta_supp_op_classes(dev, apdev):
 343     """MBO STA supported operating classes"""
 344     ssid = "test-wnm-mbo"
 345     params = { 'ssid': ssid, 'mbo': '1' }
 346     hapd = hostapd.add_ap(apdev[0], params)
 347
 348     dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 349
 350     addr = dev[0].own_addr()
 351     sta = hapd.get_sta(addr)
 352     logger.debug("STA: " + str(sta))
 353     if 'supp_op_classes' not in sta:
 354         raise Exception("No supp_op_classes")
 355     supp = bytearray(sta['supp_op_classes'].decode("hex"))
 356     if supp[0] != 81:
 357         raise Exception("Unexpected current operating class %d" % supp[0])
 358     if 115 not in supp:
 359         raise Exception("Operating class 115 missing")
 360
 361 def test_mbo_failures(dev, apdev):
 362     """MBO failure cases"""
 363     ssid = "test-wnm-mbo"
 364     params = { 'ssid': ssid, 'mbo': '1' }
 365     hapd = hostapd.add_ap(apdev[0], params)
 366
 367     with alloc_fail(dev[0], 1, "wpas_mbo_ie"):
 368         dev[0].connect(ssid, key_mgmt="NONE", scan_freq="2412")
 369
 370     with alloc_fail(dev[0], 1, "wpas_mbo_send_wnm_notification"):
 371         if "OK" not in dev[0].request("SET mbo_cell_capa 1"):
 372             raise Exception("Failed to set STA as cellular data capable")
 373     with fail_test(dev[0], 1, "wpas_mbo_send_wnm_notification"):
 374         if "OK" not in dev[0].request("SET mbo_cell_capa 3"):
 375             raise Exception("Failed to set STA as cellular data not-capable")
 376     with alloc_fail(dev[0], 1, "wpas_mbo_update_non_pref_chan"):
 377         if "FAIL" not in dev[0].request("SET non_pref_chan 81:7:200:3"):
 378             raise Exception("non_pref_chan value accepted during OOM")
 379     with alloc_fail(dev[0], 2, "wpas_mbo_update_non_pref_chan"):
 380         if "FAIL" not in dev[0].request("SET non_pref_chan 81:7:200:3"):
 381             raise Exception("non_pref_chan value accepted during OOM")
 382
 383 def test_mbo_wnm_bss_tm_ie_parsing(dev, apdev):
 384     """MBO BSS transition request MBO IE parsing"""
 385     ssid = "test-wnm-mbo"
 386     params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
 387     hapd = hostapd.add_ap(apdev[0], params)
 388     bssid = apdev[0]['bssid']
 389     addr = dev[0].own_addr()
 390     dev[0].connect(ssid, psk="12345678", key_mgmt="WPA-PSK",
 391                    proto="WPA2", ieee80211w="0", scan_freq="2412")
 392
 393     dev[0].request("SET ext_mgmt_frame_handling 1")
 394     hdr = "d0003a01" + addr.replace(':', '') + bssid.replace(':', '') + bssid.replace(':', '') + "3000"
 395     btm_hdr = "0a070100030001"
 396
 397     tests = [ ("Truncated attribute in MBO IE", "dd06506f9a160101"),
 398               ("Unexpected cell data capa attribute length in MBO IE",
 399                "dd09506f9a160501030500"),
 400               ("Unexpected transition reason attribute length in MBO IE",
 401                "dd06506f9a160600"),
 402               ("Unexpected assoc retry delay attribute length in MBO IE",
 403                "dd0c506f9a160100080200000800"),
 404               ("Unknown attribute id 255 in MBO IE",
 405                "dd06506f9a16ff00") ]
 406
 407     for test, mbo_ie in tests:
 408         logger.info(test)
 409         dev[0].request("NOTE " + test)
 410         frame = hdr + btm_hdr + mbo_ie
 411         if "OK" not in dev[0].request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=" + frame):
 412             raise Exception("MGMT_RX_PROCESS failed")
 413
 414     logger.info("Unexpected association retry delay")
 415     dev[0].request("NOTE Unexpected association retry delay")
 416     btm_hdr = "0a070108030001112233445566778899aabbcc"
 417     mbo_ie = "dd08506f9a1608020000"
 418     frame = hdr + btm_hdr + mbo_ie
 419     if "OK" not in dev[0].request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=" + frame):
 420         raise Exception("MGMT_RX_PROCESS failed")
 421
 422     dev[0].request("SET ext_mgmt_frame_handling 0")