libeap/tests/hwsim/test_wnm.py

   1 # WNM tests
   2 # Copyright (c) 2013-2014, 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 import binascii
   8 import struct
   9 import time
  10 import logging
  11 logger = logging.getLogger()
  12 import subprocess
  13
  14 import hostapd
  15 from wlantest import Wlantest
  16
  17 def test_wnm_bss_transition_mgmt(dev, apdev):
  18     """WNM BSS Transition Management"""
  19     params = { "ssid": "test-wnm",
  20                "time_advertisement": "2",
  21                "time_zone": "EST5",
  22                "wnm_sleep_mode": "1",
  23                "bss_transition": "1" }
  24     hostapd.add_ap(apdev[0]['ifname'], params)
  25
  26     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
  27     dev[0].request("WNM_BSS_QUERY 0")
  28
  29 def test_wnm_disassoc_imminent(dev, apdev):
  30     """WNM Disassociation Imminent"""
  31     params = { "ssid": "test-wnm",
  32                "time_advertisement": "2",
  33                "time_zone": "EST5",
  34                "wnm_sleep_mode": "1",
  35                "bss_transition": "1" }
  36     hostapd.add_ap(apdev[0]['ifname'], params)
  37     hapd = hostapd.Hostapd(apdev[0]['ifname'])
  38
  39     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
  40     addr = dev[0].p2p_interface_addr()
  41     hapd.request("DISASSOC_IMMINENT " + addr + " 10")
  42     ev = dev[0].wait_event(["WNM: Disassociation Imminent"])
  43     if ev is None:
  44         raise Exception("Timeout while waiting for disassociation imminent")
  45     if "Disassociation Timer 10" not in ev:
  46         raise Exception("Unexpected disassociation imminent contents")
  47     ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"])
  48     if ev is None:
  49         raise Exception("Timeout while waiting for re-connection scan")
  50
  51 def test_wnm_ess_disassoc_imminent(dev, apdev):
  52     """WNM ESS Disassociation Imminent"""
  53     params = { "ssid": "test-wnm",
  54                "time_advertisement": "2",
  55                "time_zone": "EST5",
  56                "wnm_sleep_mode": "1",
  57                "bss_transition": "1" }
  58     hostapd.add_ap(apdev[0]['ifname'], params)
  59     hapd = hostapd.Hostapd(apdev[0]['ifname'])
  60
  61     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
  62     addr = dev[0].p2p_interface_addr()
  63     hapd.request("ESS_DISASSOC " + addr + " 10 http://example.com/session-info")
  64     ev = dev[0].wait_event(["ESS-DISASSOC-IMMINENT"])
  65     if ev is None:
  66         raise Exception("Timeout while waiting for ESS disassociation imminent")
  67     if "0 1024 http://example.com/session-info" not in ev:
  68         raise Exception("Unexpected ESS disassociation imminent message contents")
  69     ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"])
  70     if ev is None:
  71         raise Exception("Timeout while waiting for re-connection scan")
  72
  73 def test_wnm_ess_disassoc_imminent_pmf(dev, apdev):
  74     """WNM ESS Disassociation Imminent"""
  75     params = hostapd.wpa2_params("test-wnm-rsn", "12345678")
  76     params["wpa_key_mgmt"] = "WPA-PSK-SHA256";
  77     params["ieee80211w"] = "2";
  78     params["bss_transition"] = "1"
  79     hostapd.add_ap(apdev[0]['ifname'], params)
  80     hapd = hostapd.Hostapd(apdev[0]['ifname'])
  81
  82     dev[0].connect("test-wnm-rsn", psk="12345678", ieee80211w="2",
  83                    key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
  84     addr = dev[0].p2p_interface_addr()
  85     hapd.request("ESS_DISASSOC " + addr + " 10 http://example.com/session-info")
  86     ev = dev[0].wait_event(["ESS-DISASSOC-IMMINENT"])
  87     if ev is None:
  88         raise Exception("Timeout while waiting for ESS disassociation imminent")
  89     if "1 1024 http://example.com/session-info" not in ev:
  90         raise Exception("Unexpected ESS disassociation imminent message contents")
  91     ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"])
  92     if ev is None:
  93         raise Exception("Timeout while waiting for re-connection scan")
  94
  95 def check_wnm_sleep_mode_enter_exit(hapd, dev, interval=None, tfs_req=None):
  96     addr = dev.p2p_interface_addr()
  97     sta = hapd.get_sta(addr)
  98     if "[WNM_SLEEP_MODE]" in sta['flags']:
  99         raise Exception("Station unexpectedly in WNM-Sleep Mode")
 100
 101     logger.info("Going to WNM Sleep Mode")
 102     extra = ""
 103     if interval is not None:
 104         extra += " interval=" + str(interval)
 105     if tfs_req:
 106         extra += " tfs_req=" + tfs_req
 107     if "OK" not in dev.request("WNM_SLEEP enter" + extra):
 108         raise Exception("WNM_SLEEP failed")
 109     ok = False
 110     for i in range(20):
 111         time.sleep(0.1)
 112         sta = hapd.get_sta(addr)
 113         if "[WNM_SLEEP_MODE]" in sta['flags']:
 114             ok = True
 115             break
 116     if not ok:
 117         raise Exception("Station failed to enter WNM-Sleep Mode")
 118
 119     logger.info("Waking up from WNM Sleep Mode")
 120     ok = False
 121     dev.request("WNM_SLEEP exit")
 122     for i in range(20):
 123         time.sleep(0.1)
 124         sta = hapd.get_sta(addr)
 125         if "[WNM_SLEEP_MODE]" not in sta['flags']:
 126             ok = True
 127             break
 128     if not ok:
 129         raise Exception("Station failed to exit WNM-Sleep Mode")
 130
 131 def test_wnm_sleep_mode_open(dev, apdev):
 132     """WNM Sleep Mode - open"""
 133     params = { "ssid": "test-wnm",
 134                "time_advertisement": "2",
 135                "time_zone": "EST5",
 136                "wnm_sleep_mode": "1",
 137                "bss_transition": "1" }
 138     hostapd.add_ap(apdev[0]['ifname'], params)
 139     hapd = hostapd.Hostapd(apdev[0]['ifname'])
 140
 141     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
 142     ev = hapd.wait_event([ "AP-STA-CONNECTED" ], timeout=5)
 143     if ev is None:
 144         raise Exception("No connection event received from hostapd")
 145     check_wnm_sleep_mode_enter_exit(hapd, dev[0])
 146     check_wnm_sleep_mode_enter_exit(hapd, dev[0], interval=100)
 147     check_wnm_sleep_mode_enter_exit(hapd, dev[0], tfs_req="5b17010001130e110000071122334455661122334455661234")
 148
 149     cmds = [ "foo",
 150              "exit tfs_req=123 interval=10",
 151              "enter tfs_req=qq interval=10" ]
 152     for cmd in cmds:
 153         if "FAIL" not in dev[0].request("WNM_SLEEP " + cmd):
 154             raise Exception("Invalid WNM_SLEEP accepted")
 155
 156 def test_wnm_sleep_mode_rsn(dev, apdev):
 157     """WNM Sleep Mode - RSN"""
 158     params = hostapd.wpa2_params("test-wnm-rsn", "12345678")
 159     params["time_advertisement"] = "2"
 160     params["time_zone"] = "EST5"
 161     params["wnm_sleep_mode"] = "1"
 162     params["bss_transition"] = "1"
 163     hostapd.add_ap(apdev[0]['ifname'], params)
 164     hapd = hostapd.Hostapd(apdev[0]['ifname'])
 165
 166     dev[0].connect("test-wnm-rsn", psk="12345678", scan_freq="2412")
 167     ev = hapd.wait_event([ "AP-STA-CONNECTED" ], timeout=5)
 168     if ev is None:
 169         raise Exception("No connection event received from hostapd")
 170     check_wnm_sleep_mode_enter_exit(hapd, dev[0])
 171
 172 def test_wnm_sleep_mode_rsn_pmf(dev, apdev):
 173     """WNM Sleep Mode - RSN with PMF"""
 174     wt = Wlantest()
 175     wt.flush()
 176     wt.add_passphrase("12345678")
 177     params = hostapd.wpa2_params("test-wnm-rsn", "12345678")
 178     params["wpa_key_mgmt"] = "WPA-PSK-SHA256";
 179     params["ieee80211w"] = "2";
 180     params["time_advertisement"] = "2"
 181     params["time_zone"] = "EST5"
 182     params["wnm_sleep_mode"] = "1"
 183     params["bss_transition"] = "1"
 184     hostapd.add_ap(apdev[0]['ifname'], params)
 185     hapd = hostapd.Hostapd(apdev[0]['ifname'])
 186
 187     dev[0].connect("test-wnm-rsn", psk="12345678", ieee80211w="2",
 188                    key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
 189     ev = hapd.wait_event([ "AP-STA-CONNECTED" ], timeout=5)
 190     if ev is None:
 191         raise Exception("No connection event received from hostapd")
 192     check_wnm_sleep_mode_enter_exit(hapd, dev[0])
 193
 194 MGMT_SUBTYPE_ACTION = 13
 195 ACTION_CATEG_WNM = 10
 196 WNM_ACT_BSS_TM_REQ = 7
 197 WNM_ACT_BSS_TM_RESP = 8
 198
 199 def bss_tm_req(dst, src, dialog_token=1, req_mode=0, disassoc_timer=0,
 200                validity_interval=1):
 201     msg = {}
 202     msg['fc'] = MGMT_SUBTYPE_ACTION << 4
 203     msg['da'] = dst
 204     msg['sa'] = src
 205     msg['bssid'] = src
 206     msg['payload'] = struct.pack("<BBBBHB",
 207                                  ACTION_CATEG_WNM, WNM_ACT_BSS_TM_REQ,
 208                                  dialog_token, req_mode, disassoc_timer,
 209                                  validity_interval)
 210     return msg
 211
 212 def rx_bss_tm_resp(hapd, expect_dialog=None, expect_status=None):
 213     for i in range(0, 100):
 214         resp = hapd.mgmt_rx()
 215         if resp is None:
 216             raise Exception("No BSS TM Response received")
 217         if resp['subtype'] == MGMT_SUBTYPE_ACTION:
 218             break
 219     if i == 99:
 220         raise Exception("Not an Action frame")
 221     payload = resp['payload']
 222     if len(payload) < 2 + 3:
 223         raise Exception("Too short payload")
 224     (category, action) = struct.unpack('BB', payload[0:2])
 225     if category != ACTION_CATEG_WNM or action != WNM_ACT_BSS_TM_RESP:
 226         raise Exception("Not a BSS TM Response")
 227     pos = payload[2:]
 228     (dialog, status, bss_term_delay) = struct.unpack('BBB', pos[0:3])
 229     resp['dialog'] = dialog
 230     resp['status'] = status
 231     resp['bss_term_delay'] = bss_term_delay
 232     pos = pos[3:]
 233     if len(pos) >= 6 and status == 0:
 234         resp['target_bssid'] = binascii.hexlify(pos[0:6])
 235         pos = pos[6:]
 236     resp['candidates'] = pos
 237     if expect_dialog is not None and dialog != expect_dialog:
 238         raise Exception("Unexpected dialog token")
 239     if expect_status is not None and status != expect_status:
 240         raise Exception("Unexpected status code %d" % status)
 241     return resp
 242
 243 def expect_ack(hapd):
 244     ev = hapd.wait_event(["MGMT-TX-STATUS"], timeout=5)
 245     if ev is None:
 246         raise Exception("Missing TX status")
 247     if "ok=1" not in ev:
 248         raise Exception("Action frame not acknowledged")
 249
 250 def test_wnm_bss_tm_req(dev, apdev):
 251     """BSS Transition Management Request"""
 252     params = { "ssid": "test-wnm", "bss_transition": "1" }
 253     hapd = hostapd.add_ap(apdev[0]['ifname'], params)
 254     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
 255     hapd2 = hostapd.add_ap(apdev[1]['ifname'], params)
 256
 257     hapd.set("ext_mgmt_frame_handling", "1")
 258
 259     # truncated BSS TM Request
 260     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 261                      req_mode=0x08)
 262     req['payload'] = struct.pack("<BBBBH",
 263                                  ACTION_CATEG_WNM, WNM_ACT_BSS_TM_REQ,
 264                                  1, 0, 0)
 265     hapd.mgmt_tx(req)
 266     expect_ack(hapd)
 267
 268     # no disassociation and no candidate list
 269     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 270                      dialog_token=2)
 271     hapd.mgmt_tx(req)
 272     resp = rx_bss_tm_resp(hapd, expect_dialog=2, expect_status=1)
 273
 274     # truncated BSS Termination Duration
 275     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 276                      req_mode=0x08)
 277     hapd.mgmt_tx(req)
 278     expect_ack(hapd)
 279
 280     # BSS Termination Duration with TSF=0 and Duration=10
 281     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 282                      req_mode=0x08, dialog_token=3)
 283     req['payload'] += struct.pack("<BBQH", 4, 10, 0, 10)
 284     hapd.mgmt_tx(req)
 285     resp = rx_bss_tm_resp(hapd, expect_dialog=3, expect_status=1)
 286
 287     # truncated Session Information URL
 288     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 289                      req_mode=0x10)
 290     hapd.mgmt_tx(req)
 291     expect_ack(hapd)
 292     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 293                      req_mode=0x10)
 294     req['payload'] += struct.pack("<BBB", 3, 65, 66)
 295     hapd.mgmt_tx(req)
 296     expect_ack(hapd)
 297
 298     # Session Information URL
 299     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 300                      req_mode=0x10, dialog_token=4)
 301     req['payload'] += struct.pack("<BBB", 2, 65, 66)
 302     hapd.mgmt_tx(req)
 303     resp = rx_bss_tm_resp(hapd, expect_dialog=4, expect_status=0)
 304
 305     # Preferred Candidate List without any entries
 306     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 307                      req_mode=0x01, dialog_token=5)
 308     hapd.mgmt_tx(req)
 309     resp = rx_bss_tm_resp(hapd, expect_dialog=5, expect_status=7)
 310
 311     # Preferred Candidate List with a truncated entry
 312     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 313                      req_mode=0x01)
 314     req['payload'] += struct.pack("<BB", 52, 1)
 315     hapd.mgmt_tx(req)
 316     expect_ack(hapd)
 317
 318     # Preferred Candidate List with a too short entry
 319     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 320                      req_mode=0x01, dialog_token=6)
 321     req['payload'] += struct.pack("<BB", 52, 0)
 322     hapd.mgmt_tx(req)
 323     resp = rx_bss_tm_resp(hapd, expect_dialog=6, expect_status=7)
 324
 325     # Preferred Candidate List with a non-matching entry
 326     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 327                      req_mode=0x01, dialog_token=6)
 328     req['payload'] += struct.pack("<BB6BLBBB", 52, 13,
 329                                   1, 2, 3, 4, 5, 6,
 330                                   0, 81, 1, 7)
 331     hapd.mgmt_tx(req)
 332     resp = rx_bss_tm_resp(hapd, expect_dialog=6, expect_status=7)
 333
 334     # Preferred Candidate List with a truncated subelement
 335     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 336                      req_mode=0x01, dialog_token=7)
 337     req['payload'] += struct.pack("<BB6BLBBBBB", 52, 13 + 2,
 338                                   1, 2, 3, 4, 5, 6,
 339                                   0, 81, 1, 7,
 340                                   1, 1)
 341     hapd.mgmt_tx(req)
 342     resp = rx_bss_tm_resp(hapd, expect_dialog=7, expect_status=7)
 343
 344     # Preferred Candidate List with lots of invalid optional subelements
 345     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 346                      req_mode=0x01, dialog_token=8)
 347     subelems = struct.pack("<BBHB", 1, 3, 0, 100)
 348     subelems += struct.pack("<BBB", 2, 1, 65)
 349     subelems += struct.pack("<BB", 3, 0)
 350     subelems += struct.pack("<BBQB", 4, 9, 0, 10)
 351     subelems += struct.pack("<BBHLB", 5, 7, 0, 0, 0)
 352     subelems += struct.pack("<BB", 66, 0)
 353     subelems += struct.pack("<BBBBBB", 70, 4, 0, 0, 0, 0)
 354     subelems += struct.pack("<BB", 71, 0)
 355     req['payload'] += struct.pack("<BB6BLBBB", 52, 13 + len(subelems),
 356                                   1, 2, 3, 4, 5, 6,
 357                                   0, 81, 1, 7) + subelems
 358     hapd.mgmt_tx(req)
 359     resp = rx_bss_tm_resp(hapd, expect_dialog=8, expect_status=7)
 360
 361     # Preferred Candidate List with lots of valid optional subelements (twice)
 362     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 363                      req_mode=0x01, dialog_token=8)
 364     # TSF Information
 365     subelems = struct.pack("<BBHH", 1, 4, 0, 100)
 366     # Condensed Country String
 367     subelems += struct.pack("<BBBB", 2, 2, 65, 66)
 368     # BSS Transition Candidate Preference
 369     subelems += struct.pack("<BBB", 3, 1, 100)
 370     # BSS Termination Duration
 371     subelems += struct.pack("<BBQH", 4, 10, 0, 10)
 372     # Bearing
 373     subelems += struct.pack("<BBHLH", 5, 8, 0, 0, 0)
 374     # Measurement Pilot Transmission
 375     subelems += struct.pack("<BBBBB", 66, 3, 0, 0, 0)
 376     # RM Enabled Capabilities
 377     subelems += struct.pack("<BBBBBBB", 70, 5, 0, 0, 0, 0, 0)
 378     # Multiple BSSID
 379     subelems += struct.pack("<BBBB", 71, 2, 0, 0)
 380     req['payload'] += struct.pack("<BB6BLBBB", 52, 13 + len(subelems) * 2,
 381                                   1, 2, 3, 4, 5, 6,
 382                                   0, 81, 1, 7) + subelems + subelems
 383     hapd.mgmt_tx(req)
 384     resp = rx_bss_tm_resp(hapd, expect_dialog=8, expect_status=7)
 385
 386 def test_wnm_bss_keep_alive(dev, apdev):
 387     """WNM keep-alive"""
 388     params = { "ssid": "test-wnm",
 389                "ap_max_inactivity": "1" }
 390     hapd = hostapd.add_ap(apdev[0]['ifname'], params)
 391
 392     addr = dev[0].p2p_interface_addr()
 393     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
 394     start = hapd.get_sta(addr)
 395     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=2)
 396     if ev is not None:
 397         raise Exception("Unexpected disconnection")
 398     end = hapd.get_sta(addr)
 399     if int(end['rx_packets']) <= int(start['rx_packets']):
 400         raise Exception("No keep-alive packets received")
 401     try:
 402         # Disable client keep-alive so that hostapd will verify connection
 403         # with client poll
 404         dev[0].request("SET no_keep_alive 1")
 405         for i in range(60):
 406             sta = hapd.get_sta(addr)
 407             logger.info("timeout_next=%s rx_packets=%s tx_packets=%s" % (sta['timeout_next'], sta['rx_packets'], sta['tx_packets']))
 408             if i > 1 and sta['timeout_next'] != "NULLFUNC POLL" and int(sta['tx_packets']) > int(end['tx_packets']):
 409                 break
 410             ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.5)
 411             if ev is not None:
 412                 raise Exception("Unexpected disconnection (client poll expected)")
 413     finally:
 414         dev[0].request("SET no_keep_alive 0")
 415     if int(sta['tx_packets']) <= int(end['tx_packets']):
 416         raise Exception("No client poll packet seen")
 417
 418 def test_wnm_bss_tm(dev, apdev):
 419     """WNM BSS Transition Management"""
 420     try:
 421         hapd = None
 422         hapd2 = None
 423         params = { "ssid": "test-wnm",
 424                    "country_code": "FI",
 425                    "ieee80211d": "1",
 426                    "hw_mode": "g",
 427                    "channel": "1",
 428                    "bss_transition": "1" }
 429         hapd = hostapd.add_ap(apdev[0]['ifname'], params)
 430
 431         id = dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
 432         dev[0].set_network(id, "scan_freq", "")
 433
 434         params = { "ssid": "test-wnm",
 435                    "country_code": "FI",
 436                    "ieee80211d": "1",
 437                    "hw_mode": "a",
 438                    "channel": "36",
 439                    "bss_transition": "1" }
 440         hapd2 = hostapd.add_ap(apdev[1]['ifname'], params)
 441
 442         addr = dev[0].p2p_interface_addr()
 443         dev[0].dump_monitor()
 444
 445         logger.info("No neighbor list entries")
 446         if "OK" not in hapd.request("BSS_TM_REQ " + addr):
 447             raise Exception("BSS_TM_REQ command failed")
 448         ev = hapd.wait_event(['BSS-TM-RESP'], timeout=10)
 449         if ev is None:
 450             raise Exception("No BSS Transition Management Response")
 451         if addr not in ev:
 452             raise Exception("Unexpected BSS Transition Management Response address")
 453         if "status_code=0" in ev:
 454             raise Exception("BSS transition accepted unexpectedly")
 455         dev[0].dump_monitor()
 456
 457         logger.info("Neighbor list entry, but not claimed as Preferred Candidate List")
 458         if "OK" not in hapd.request("BSS_TM_REQ " + addr + " neighbor=11:22:33:44:55:66,0x0000,81,3,7"):
 459             raise Exception("BSS_TM_REQ command failed")
 460         ev = hapd.wait_event(['BSS-TM-RESP'], timeout=10)
 461         if ev is None:
 462             raise Exception("No BSS Transition Management Response")
 463         if "status_code=0" in ev:
 464             raise Exception("BSS transition accepted unexpectedly")
 465         dev[0].dump_monitor()
 466
 467         logger.info("Preferred Candidate List (no matching neighbor) without Disassociation Imminent")
 468         if "OK" not in hapd.request("BSS_TM_REQ " + addr + " pref=1 neighbor=11:22:33:44:55:66,0x0000,81,3,7,0301ff neighbor=22:33:44:55:66:77,0x0000,1,36,7 neighbor=00:11:22:33:44:55,0x0000,81,4,7,03010a"):
 469             raise Exception("BSS_TM_REQ command failed")
 470         ev = hapd.wait_event(['BSS-TM-RESP'], timeout=10)
 471         if ev is None:
 472             raise Exception("No BSS Transition Management Response")
 473         if "status_code=0" in ev:
 474             raise Exception("BSS transition accepted unexpectedly")
 475         ev = dev[0].wait_event(["CTRL-EVENT-SCAN-STARTED"], timeout=5)
 476         if ev is None:
 477             raise Exception("No scan started")
 478         dev[0].dump_monitor()
 479
 480         logger.info("Preferred Candidate List (matching neighbor for another BSS) without Disassociation Imminent")
 481         if "OK" not in hapd.request("BSS_TM_REQ " + addr + " pref=1 abridged=1 valid_int=255 neighbor=" + apdev[1]['bssid'] + ",0x0000,115,36,7,0301ff"):
 482             raise Exception("BSS_TM_REQ command failed")
 483         ev = hapd.wait_event(['BSS-TM-RESP'], timeout=10)
 484         if ev is None:
 485             raise Exception("No BSS Transition Management Response")
 486         if "status_code=0" not in ev:
 487             raise Exception("BSS transition request was not accepted: " + ev)
 488         if "target_bssid=" + apdev[1]['bssid'] not in ev:
 489             raise Exception("Unexpected target BSS: " + ev)
 490         dev[0].wait_connected(timeout=15, error="No reassociation seen")
 491         if apdev[1]['bssid'] not in ev:
 492             raise Exception("Unexpected reassociation target: " + ev)
 493         ev = dev[0].wait_event(["CTRL-EVENT-SCAN-STARTED"], timeout=0.1)
 494         if ev is not None:
 495             raise Exception("Unexpected scan started")
 496         dev[0].dump_monitor()
 497
 498         logger.info("Preferred Candidate List with two matches, no roam needed")
 499         if "OK" not in hapd2.request("BSS_TM_REQ " + addr + " pref=1 abridged=1 valid_int=255 neighbor=" + apdev[0]['bssid'] + ",0x0000,81,1,7,030101 neighbor=" + apdev[1]['bssid'] + ",0x0000,115,36,7,0301ff"):
 500             raise Exception("BSS_TM_REQ command failed")
 501         ev = hapd2.wait_event(['BSS-TM-RESP'], timeout=10)
 502         if ev is None:
 503             raise Exception("No BSS Transition Management Response")
 504         if "status_code=0" not in ev:
 505             raise Exception("BSS transition request was not accepted: " + ev)
 506         if "target_bssid=" + apdev[1]['bssid'] not in ev:
 507             raise Exception("Unexpected target BSS: " + ev)
 508         ev = dev[0].wait_event(["CTRL-EVENT-SCAN-STARTED"], timeout=0.1)
 509         if ev is not None:
 510             raise Exception("Unexpected scan started")
 511         ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=0.5)
 512         if ev is not None:
 513             raise Exception("Unexpected reassociation");
 514     finally:
 515         dev[0].request("DISCONNECT")
 516         if hapd:
 517             hapd.request("DISABLE")
 518         if hapd2:
 519             hapd2.request("DISABLE")
 520         subprocess.call(['iw', 'reg', 'set', '00'])
 521         dev[0].flush_scan_cache()
 522
 523 def start_wnm_tm(ap, country, dev):
 524     params = { "ssid": "test-wnm",
 525                "country_code": country,
 526                "ieee80211d": "1",
 527                "hw_mode": "g",
 528                "channel": "1",
 529                "bss_transition": "1" }
 530     hapd = hostapd.add_ap(ap['ifname'], params)
 531     id = dev.connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
 532     dev.dump_monitor()
 533     dev.set_network(id, "scan_freq", "")
 534     return hapd, id
 535
 536 def stop_wnm_tm(hapd, dev):
 537     dev.request("DISCONNECT")
 538     try:
 539         dev.wait_disconnected()
 540     except:
 541         pass
 542     if hapd:
 543         hapd.request("DISABLE")
 544     subprocess.call(['iw', 'reg', 'set', '00'])
 545     dev.flush_scan_cache()
 546
 547 def wnm_bss_tm_check(hapd, dev, data):
 548     addr = dev.p2p_interface_addr()
 549     if "OK" not in hapd.request("BSS_TM_REQ " + addr + " " + data):
 550         raise Exception("BSS_TM_REQ command failed")
 551     ev = dev.wait_event(["CTRL-EVENT-SCAN-STARTED"], timeout=5)
 552     if ev is None:
 553         raise Exception("No scan started")
 554     ev = dev.wait_event(["CTRL-EVENT-SCAN-RESULTS"], 15)
 555     if ev is None:
 556         raise Exception("Scan did not complete")
 557
 558     ev = hapd.wait_event(['BSS-TM-RESP'], timeout=10)
 559     if ev is None:
 560         raise Exception("No BSS Transition Management Response")
 561     if "status_code=7" not in ev:
 562         raise Exception("Unexpected response: " + ev)
 563
 564 def test_wnm_bss_tm_country_us(dev, apdev):
 565     """WNM BSS Transition Management (US)"""
 566     try:
 567         hapd = None
 568         hapd, id = start_wnm_tm(apdev[0], "US", dev[0])
 569
 570         logger.info("Preferred Candidate List (no matching neighbor, known channels)")
 571         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=11:22:33:44:55:66,0x0000,12,3,7,0301ff neighbor=00:11:22:33:44:55,0x0000,2,52,7,03010a neighbor=00:11:22:33:44:57,0x0000,4,100,7 neighbor=00:11:22:33:44:59,0x0000,3,149,7 neighbor=00:11:22:33:44:5b,0x0000,34,1,7 neighbor=00:11:22:33:44:5d,0x0000,5,149,7")
 572
 573         # Make the test take less time by limiting full scans
 574         dev[0].set_network(id, "scan_freq", "2412")
 575         logger.info("Preferred Candidate List (no matching neighbor, unknown channels)")
 576         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=11:22:33:44:55:66,0x0000,12,0,7,0301ff neighbor=22:33:44:55:66:77,0x0000,12,12,7 neighbor=00:11:22:33:44:55,0x0000,2,35,7,03010a neighbor=00:11:22:33:44:56,0x0000,2,65,7 neighbor=00:11:22:33:44:57,0x0000,4,99,7 neighbor=00:11:22:33:44:58,0x0000,4,145,7")
 577
 578         logger.info("Preferred Candidate List (no matching neighbor, unknown channels 2)")
 579         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=00:11:22:33:44:59,0x0000,3,148,7 neighbor=00:11:22:33:44:5a,0x0000,3,162,7 neighbor=00:11:22:33:44:5b,0x0000,34,0,7 neighbor=00:11:22:33:44:5c,0x0000,34,4,7 neighbor=00:11:22:33:44:5d,0x0000,5,148,7 neighbor=00:11:22:33:44:5e,0x0000,5,166,7 neighbor=00:11:22:33:44:5f,0x0000,0,0,7")
 580     finally:
 581         stop_wnm_tm(hapd, dev[0])
 582
 583 def test_wnm_bss_tm_country_fi(dev, apdev):
 584     """WNM BSS Transition Management (FI)"""
 585     addr = dev[0].p2p_interface_addr()
 586     try:
 587         hapd = None
 588         hapd, id = start_wnm_tm(apdev[0], "FI", dev[0])
 589
 590         logger.info("Preferred Candidate List (no matching neighbor, known channels)")
 591         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=11:22:33:44:55:66,0x0000,4,3,7,0301ff neighbor=00:11:22:33:44:55,0x0000,1,36,7,03010a neighbor=00:11:22:33:44:57,0x0000,3,100,7 neighbor=00:11:22:33:44:59,0x0000,17,149,7 neighbor=00:11:22:33:44:5c,0x0000,18,1,7")
 592
 593         # Make the test take less time by limiting full scans
 594         dev[0].set_network(id, "scan_freq", "2412")
 595         logger.info("Preferred Candidate List (no matching neighbor, unknown channels)")
 596         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=00:11:22:33:44:00,0x0000,4,0,7 neighbor=00:11:22:33:44:01,0x0000,4,14,7 neighbor=00:11:22:33:44:02,0x0000,1,35,7 neighbor=00:11:22:33:44:03,0x0000,1,65,7 neighbor=00:11:22:33:44:04,0x0000,3,99,7 neighbor=00:11:22:33:44:05,0x0000,3,141,7 neighbor=00:11:22:33:44:06,0x0000,17,148,7 neighbor=00:11:22:33:44:07,0x0000,17,170,7 neighbor=00:11:22:33:44:08,0x0000,18,0,7 neighbor=00:11:22:33:44:09,0x0000,18,5,7")
 597
 598         logger.info("Preferred Candidate List (no matching neighbor, unknown channels 2)")
 599         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=00:11:22:33:44:00,0x0000,0,0,7")
 600     finally:
 601         stop_wnm_tm(hapd, dev[0])
 602
 603 def test_wnm_bss_tm_country_jp(dev, apdev):
 604     """WNM BSS Transition Management (JP)"""
 605     addr = dev[0].p2p_interface_addr()
 606     try:
 607         hapd = None
 608         hapd, id = start_wnm_tm(apdev[0], "JP", dev[0])
 609
 610         logger.info("Preferred Candidate List (no matching neighbor, known channels)")
 611         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=11:22:33:44:55:66,0x0000,30,3,7,0301ff neighbor=00:11:22:33:44:55,0x0000,31,14,7,03010a neighbor=00:11:22:33:44:57,0x0000,1,36,7 neighbor=00:11:22:33:44:59,0x0000,34,100,7 neighbor=00:11:22:33:44:5c,0x0000,59,1,7")
 612
 613         # Make the test take less time by limiting full scans
 614         dev[0].set_network(id, "scan_freq", "2412")
 615         logger.info("Preferred Candidate List (no matching neighbor, unknown channels)")
 616         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=11:22:33:44:55:66,0x0000,30,0,7,0301ff neighbor=22:33:44:55:66:77,0x0000,30,14,7 neighbor=00:11:22:33:44:56,0x0000,31,13,7 neighbor=00:11:22:33:44:57,0x0000,1,33,7 neighbor=00:11:22:33:44:58,0x0000,1,65,7 neighbor=00:11:22:33:44:5a,0x0000,34,99,7 neighbor=00:11:22:33:44:5b,0x0000,34,141,7 neighbor=00:11:22:33:44:5d,0x0000,59,0,7 neighbor=00:11:22:33:44:5e,0x0000,59,4,7 neighbor=00:11:22:33:44:5f,0x0000,0,0,7")
 617     finally:
 618         stop_wnm_tm(hapd, dev[0])
 619
 620 def test_wnm_bss_tm_country_cn(dev, apdev):
 621     """WNM BSS Transition Management (CN)"""
 622     addr = dev[0].p2p_interface_addr()
 623     try:
 624         hapd = None
 625         hapd, id = start_wnm_tm(apdev[0], "CN", dev[0])
 626
 627         logger.info("Preferred Candidate List (no matching neighbor, known channels)")
 628         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=11:22:33:44:55:66,0x0000,7,3,7,0301ff neighbor=00:11:22:33:44:55,0x0000,1,36,7,03010a neighbor=00:11:22:33:44:57,0x0000,3,149,7 neighbor=00:11:22:33:44:59,0x0000,6,149,7")
 629
 630         # Make the test take less time by limiting full scans
 631         dev[0].set_network(id, "scan_freq", "2412")
 632         logger.info("Preferred Candidate List (no matching neighbor, unknown channels)")
 633         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=11:22:33:44:55:66,0x0000,7,0,7,0301ff neighbor=22:33:44:55:66:77,0x0000,7,14,7 neighbor=00:11:22:33:44:56,0x0000,1,35,7 neighbor=00:11:22:33:44:57,0x0000,1,65,7 neighbor=00:11:22:33:44:58,0x0000,3,148,7 neighbor=00:11:22:33:44:5a,0x0000,3,166,7 neighbor=00:11:22:33:44:5f,0x0000,0,0,7")
 634     finally:
 635         stop_wnm_tm(hapd, dev[0])
 636
 637 def test_wnm_bss_tm_global(dev, apdev):
 638     """WNM BSS Transition Management (global)"""
 639     addr = dev[0].p2p_interface_addr()
 640     try:
 641         hapd = None
 642         hapd, id = start_wnm_tm(apdev[0], "XX", dev[0])
 643
 644         logger.info("Preferred Candidate List (no matching neighbor, known channels)")
 645         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=11:22:33:44:55:66,0x0000,81,3,7,0301ff neighbor=00:11:22:33:44:55,0x0000,82,14,7,03010a neighbor=00:11:22:33:44:57,0x0000,83,1,7 neighbor=00:11:22:33:44:59,0x0000,115,36,7 neighbor=00:11:22:33:44:5a,0x0000,121,100,7 neighbor=00:11:22:33:44:5c,0x0000,124,149,7 neighbor=00:11:22:33:44:5d,0x0000,125,149,7 neighbor=00:11:22:33:44:5e,0x0000,128,42,7 neighbor=00:11:22:33:44:5f,0x0000,129,50,7 neighbor=00:11:22:33:44:60,0x0000,180,1,7")
 646
 647         # Make the test take less time by limiting full scans
 648         dev[0].set_network(id, "scan_freq", "2412")
 649         logger.info("Preferred Candidate List (no matching neighbor, unknown channels)")
 650         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=00:11:22:33:44:00,0x0000,81,0,7 neighbor=00:11:22:33:44:01,0x0000,81,14,7 neighbor=00:11:22:33:44:02,0x0000,82,13,7 neighbor=00:11:22:33:44:03,0x0000,83,0,7 neighbor=00:11:22:33:44:04,0x0000,83,14,7 neighbor=00:11:22:33:44:05,0x0000,115,35,7 neighbor=00:11:22:33:44:06,0x0000,115,65,7 neighbor=00:11:22:33:44:07,0x0000,121,99,7 neighbor=00:11:22:33:44:08,0x0000,121,141,7 neighbor=00:11:22:33:44:09,0x0000,124,148,7")
 651
 652         logger.info("Preferred Candidate List (no matching neighbor, unknown channels 2)")
 653         wnm_bss_tm_check(hapd, dev[0], "pref=1 neighbor=00:11:22:33:44:00,0x0000,124,162,7 neighbor=00:11:22:33:44:01,0x0000,125,148,7 neighbor=00:11:22:33:44:02,0x0000,125,170,7 neighbor=00:11:22:33:44:03,0x0000,128,35,7 neighbor=00:11:22:33:44:04,0x0000,128,162,7 neighbor=00:11:22:33:44:05,0x0000,129,49,7 neighbor=00:11:22:33:44:06,0x0000,129,115,7 neighbor=00:11:22:33:44:07,0x0000,180,0,7 neighbor=00:11:22:33:44:08,0x0000,180,5,7 neighbor=00:11:22:33:44:09,0x0000,0,0,7")
 654     finally:
 655         stop_wnm_tm(hapd, dev[0])