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
  13 import hostapd
  14 from wlantest import Wlantest
  15
  16 def test_wnm_bss_transition_mgmt(dev, apdev):
  17     """WNM BSS Transition Management"""
  18     params = { "ssid": "test-wnm",
  19                "time_advertisement": "2",
  20                "time_zone": "EST5",
  21                "wnm_sleep_mode": "1",
  22                "bss_transition": "1" }
  23     hostapd.add_ap(apdev[0]['ifname'], params)
  24
  25     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
  26     dev[0].request("WNM_BSS_QUERY 0")
  27
  28 def test_wnm_disassoc_imminent(dev, apdev):
  29     """WNM Disassociation Imminent"""
  30     params = { "ssid": "test-wnm",
  31                "time_advertisement": "2",
  32                "time_zone": "EST5",
  33                "wnm_sleep_mode": "1",
  34                "bss_transition": "1" }
  35     hostapd.add_ap(apdev[0]['ifname'], params)
  36     hapd = hostapd.Hostapd(apdev[0]['ifname'])
  37
  38     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
  39     addr = dev[0].p2p_interface_addr()
  40     hapd.request("DISASSOC_IMMINENT " + addr + " 10")
  41     ev = dev[0].wait_event(["WNM: Disassociation Imminent"])
  42     if ev is None:
  43         raise Exception("Timeout while waiting for disassociation imminent")
  44     if "Disassociation Timer 10" not in ev:
  45         raise Exception("Unexpected disassociation imminent contents")
  46     ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"])
  47     if ev is None:
  48         raise Exception("Timeout while waiting for re-connection scan")
  49
  50 def test_wnm_ess_disassoc_imminent(dev, apdev):
  51     """WNM ESS Disassociation Imminent"""
  52     params = { "ssid": "test-wnm",
  53                "time_advertisement": "2",
  54                "time_zone": "EST5",
  55                "wnm_sleep_mode": "1",
  56                "bss_transition": "1" }
  57     hostapd.add_ap(apdev[0]['ifname'], params)
  58     hapd = hostapd.Hostapd(apdev[0]['ifname'])
  59
  60     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
  61     addr = dev[0].p2p_interface_addr()
  62     hapd.request("ESS_DISASSOC " + addr + " 10 http://example.com/session-info")
  63     ev = dev[0].wait_event(["ESS-DISASSOC-IMMINENT"])
  64     if ev is None:
  65         raise Exception("Timeout while waiting for ESS disassociation imminent")
  66     if "0 1024 http://example.com/session-info" not in ev:
  67         raise Exception("Unexpected ESS disassociation imminent message contents")
  68     ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"])
  69     if ev is None:
  70         raise Exception("Timeout while waiting for re-connection scan")
  71
  72 def test_wnm_ess_disassoc_imminent_pmf(dev, apdev):
  73     """WNM ESS Disassociation Imminent"""
  74     params = hostapd.wpa2_params("test-wnm-rsn", "12345678")
  75     params["wpa_key_mgmt"] = "WPA-PSK-SHA256";
  76     params["ieee80211w"] = "2";
  77     params["bss_transition"] = "1"
  78     hostapd.add_ap(apdev[0]['ifname'], params)
  79     hapd = hostapd.Hostapd(apdev[0]['ifname'])
  80
  81     dev[0].connect("test-wnm-rsn", psk="12345678", ieee80211w="2",
  82                    key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
  83     addr = dev[0].p2p_interface_addr()
  84     hapd.request("ESS_DISASSOC " + addr + " 10 http://example.com/session-info")
  85     ev = dev[0].wait_event(["ESS-DISASSOC-IMMINENT"])
  86     if ev is None:
  87         raise Exception("Timeout while waiting for ESS disassociation imminent")
  88     if "1 1024 http://example.com/session-info" not in ev:
  89         raise Exception("Unexpected ESS disassociation imminent message contents")
  90     ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"])
  91     if ev is None:
  92         raise Exception("Timeout while waiting for re-connection scan")
  93
  94 def check_wnm_sleep_mode_enter_exit(hapd, dev, interval=None, tfs_req=None):
  95     addr = dev.p2p_interface_addr()
  96     sta = hapd.get_sta(addr)
  97     if "[WNM_SLEEP_MODE]" in sta['flags']:
  98         raise Exception("Station unexpectedly in WNM-Sleep Mode")
  99     logger.info("Going to WNM Sleep Mode")
 100     extra = ""
 101     if interval is not None:
 102         extra += " interval=" + str(interval)
 103     if tfs_req:
 104         extra += " tfs_req=" + tfs_req
 105     if "OK" not in dev.request("WNM_SLEEP enter" + extra):
 106         raise Exception("WNM_SLEEP failed")
 107     time.sleep(0.5)
 108     sta = hapd.get_sta(addr)
 109     if "[WNM_SLEEP_MODE]" not in sta['flags']:
 110         raise Exception("Station failed to enter WNM-Sleep Mode")
 111     logger.info("Waking up from WNM Sleep Mode")
 112     dev.request("WNM_SLEEP exit")
 113     time.sleep(0.5)
 114     sta = hapd.get_sta(addr)
 115     if "[WNM_SLEEP_MODE]" in sta['flags']:
 116         raise Exception("Station failed to exit WNM-Sleep Mode")
 117
 118 def test_wnm_sleep_mode_open(dev, apdev):
 119     """WNM Sleep Mode - open"""
 120     params = { "ssid": "test-wnm",
 121                "time_advertisement": "2",
 122                "time_zone": "EST5",
 123                "wnm_sleep_mode": "1",
 124                "bss_transition": "1" }
 125     hostapd.add_ap(apdev[0]['ifname'], params)
 126     hapd = hostapd.Hostapd(apdev[0]['ifname'])
 127
 128     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
 129     check_wnm_sleep_mode_enter_exit(hapd, dev[0])
 130     check_wnm_sleep_mode_enter_exit(hapd, dev[0], interval=100)
 131     check_wnm_sleep_mode_enter_exit(hapd, dev[0], tfs_req="5b17010001130e110000071122334455661122334455661234")
 132
 133 def test_wnm_sleep_mode_rsn(dev, apdev):
 134     """WNM Sleep Mode - RSN"""
 135     params = hostapd.wpa2_params("test-wnm-rsn", "12345678")
 136     params["time_advertisement"] = "2"
 137     params["time_zone"] = "EST5"
 138     params["wnm_sleep_mode"] = "1"
 139     params["bss_transition"] = "1"
 140     hostapd.add_ap(apdev[0]['ifname'], params)
 141     hapd = hostapd.Hostapd(apdev[0]['ifname'])
 142
 143     dev[0].connect("test-wnm-rsn", psk="12345678", scan_freq="2412")
 144     check_wnm_sleep_mode_enter_exit(hapd, dev[0])
 145
 146 def test_wnm_sleep_mode_rsn_pmf(dev, apdev):
 147     """WNM Sleep Mode - RSN with PMF"""
 148     wt = Wlantest()
 149     wt.flush()
 150     wt.add_passphrase("12345678")
 151     params = hostapd.wpa2_params("test-wnm-rsn", "12345678")
 152     params["wpa_key_mgmt"] = "WPA-PSK-SHA256";
 153     params["ieee80211w"] = "2";
 154     params["time_advertisement"] = "2"
 155     params["time_zone"] = "EST5"
 156     params["wnm_sleep_mode"] = "1"
 157     params["bss_transition"] = "1"
 158     hostapd.add_ap(apdev[0]['ifname'], params)
 159     hapd = hostapd.Hostapd(apdev[0]['ifname'])
 160
 161     dev[0].connect("test-wnm-rsn", psk="12345678", ieee80211w="2",
 162                    key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
 163     check_wnm_sleep_mode_enter_exit(hapd, dev[0])
 164
 165 MGMT_SUBTYPE_ACTION = 13
 166 ACTION_CATEG_WNM = 10
 167 WNM_ACT_BSS_TM_REQ = 7
 168 WNM_ACT_BSS_TM_RESP = 8
 169
 170 def bss_tm_req(dst, src, dialog_token=1, req_mode=0, disassoc_timer=0,
 171                validity_interval=1):
 172     msg = {}
 173     msg['fc'] = MGMT_SUBTYPE_ACTION << 4
 174     msg['da'] = dst
 175     msg['sa'] = src
 176     msg['bssid'] = src
 177     msg['payload'] = struct.pack("<BBBBHB",
 178                                  ACTION_CATEG_WNM, WNM_ACT_BSS_TM_REQ,
 179                                  dialog_token, req_mode, disassoc_timer,
 180                                  validity_interval)
 181     return msg
 182
 183 def rx_bss_tm_resp(hapd, expect_dialog=None, expect_status=None):
 184     for i in range(0, 100):
 185         resp = hapd.mgmt_rx()
 186         if resp is None:
 187             raise Exception("No BSS TM Response received")
 188         if resp['subtype'] == MGMT_SUBTYPE_ACTION:
 189             break
 190     if i == 99:
 191         raise Exception("Not an Action frame")
 192     payload = resp['payload']
 193     if len(payload) < 2 + 3:
 194         raise Exception("Too short payload")
 195     (category, action) = struct.unpack('BB', payload[0:2])
 196     if category != ACTION_CATEG_WNM or action != WNM_ACT_BSS_TM_RESP:
 197         raise Exception("Not a BSS TM Response")
 198     pos = payload[2:]
 199     (dialog, status, bss_term_delay) = struct.unpack('BBB', pos[0:3])
 200     resp['dialog'] = dialog
 201     resp['status'] = status
 202     resp['bss_term_delay'] = bss_term_delay
 203     pos = pos[3:]
 204     if len(pos) >= 6 and status == 0:
 205         resp['target_bssid'] = binascii.hexlify(pos[0:6])
 206         pos = pos[6:]
 207     resp['candidates'] = pos
 208     if expect_dialog is not None and dialog != expect_dialog:
 209         raise Exception("Unexpected dialog token")
 210     if expect_status is not None and status != expect_status:
 211         raise Exception("Unexpected status code %d" % status)
 212     return resp
 213
 214 def except_ack(hapd):
 215     ev = hapd.wait_event(["MGMT-TX-STATUS"], timeout=5)
 216     if ev is None:
 217         raise Exception("Missing TX status")
 218     if "ok=1" not in ev:
 219         raise Exception("Action frame not acknowledged")
 220
 221 def test_wnm_bss_tm_req(dev, apdev):
 222     """BSS Transition Management Request"""
 223     params = { "ssid": "test-wnm", "bss_transition": "1" }
 224     hapd = hostapd.add_ap(apdev[0]['ifname'], params)
 225     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
 226     hapd2 = hostapd.add_ap(apdev[1]['ifname'], params)
 227
 228     hapd.set("ext_mgmt_frame_handling", "1")
 229
 230     # truncated BSS TM Request
 231     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 232                      req_mode=0x08)
 233     req['payload'] = struct.pack("<BBBBH",
 234                                  ACTION_CATEG_WNM, WNM_ACT_BSS_TM_REQ,
 235                                  1, 0, 0)
 236     hapd.mgmt_tx(req)
 237     except_ack(hapd)
 238
 239     # no disassociation and no candidate list
 240     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 241                      dialog_token=2)
 242     hapd.mgmt_tx(req)
 243     resp = rx_bss_tm_resp(hapd, expect_dialog=2, expect_status=1)
 244
 245     # truncated BSS Termination Duration
 246     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 247                      req_mode=0x08)
 248     hapd.mgmt_tx(req)
 249     except_ack(hapd)
 250
 251     # BSS Termination Duration with TSF=0 and Duration=10
 252     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 253                      req_mode=0x08, dialog_token=3)
 254     req['payload'] += struct.pack("<BBQH", 4, 10, 0, 10)
 255     hapd.mgmt_tx(req)
 256     resp = rx_bss_tm_resp(hapd, expect_dialog=3, expect_status=1)
 257
 258     # truncated Session Information URL
 259     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 260                      req_mode=0x10)
 261     hapd.mgmt_tx(req)
 262     except_ack(hapd)
 263     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 264                      req_mode=0x10)
 265     req['payload'] += struct.pack("<BBB", 3, 65, 66)
 266     hapd.mgmt_tx(req)
 267     except_ack(hapd)
 268
 269     # Session Information URL
 270     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 271                      req_mode=0x10, dialog_token=4)
 272     req['payload'] += struct.pack("<BBB", 2, 65, 66)
 273     hapd.mgmt_tx(req)
 274     resp = rx_bss_tm_resp(hapd, expect_dialog=4, expect_status=0)
 275
 276     # Preferred Candidate List without any entries
 277     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 278                      req_mode=0x01, dialog_token=5)
 279     hapd.mgmt_tx(req)
 280     resp = rx_bss_tm_resp(hapd, expect_dialog=5, expect_status=1)
 281
 282     # Preferred Candidate List with a truncated entry
 283     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 284                      req_mode=0x01)
 285     req['payload'] += struct.pack("<BB", 52, 1)
 286     hapd.mgmt_tx(req)
 287     except_ack(hapd)
 288
 289     # Preferred Candidate List with a too short entry
 290     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 291                      req_mode=0x01, dialog_token=6)
 292     req['payload'] += struct.pack("<BB", 52, 0)
 293     hapd.mgmt_tx(req)
 294     resp = rx_bss_tm_resp(hapd, expect_dialog=6, expect_status=1)
 295
 296     # Preferred Candidate List with a non-matching entry
 297     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 298                      req_mode=0x01, dialog_token=6)
 299     req['payload'] += struct.pack("<BB6BLBBB", 52, 13,
 300                                   1, 2, 3, 4, 5, 6,
 301                                   0, 81, 1, 7)
 302     hapd.mgmt_tx(req)
 303     resp = rx_bss_tm_resp(hapd, expect_dialog=6, expect_status=1)
 304
 305     # Preferred Candidate List with a truncated subelement
 306     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 307                      req_mode=0x01, dialog_token=7)
 308     req['payload'] += struct.pack("<BB6BLBBBBB", 52, 13 + 2,
 309                                   1, 2, 3, 4, 5, 6,
 310                                   0, 81, 1, 7,
 311                                   1, 1)
 312     hapd.mgmt_tx(req)
 313     resp = rx_bss_tm_resp(hapd, expect_dialog=7, expect_status=1)
 314
 315     # Preferred Candidate List with lots of invalid optional subelements
 316     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 317                      req_mode=0x01, dialog_token=8)
 318     subelems = struct.pack("<BBHB", 1, 3, 0, 100)
 319     subelems += struct.pack("<BBB", 2, 1, 65)
 320     subelems += struct.pack("<BB", 3, 0)
 321     subelems += struct.pack("<BBQB", 4, 9, 0, 10)
 322     subelems += struct.pack("<BBHLB", 5, 7, 0, 0, 0)
 323     subelems += struct.pack("<BB", 66, 0)
 324     subelems += struct.pack("<BBBBBB", 70, 4, 0, 0, 0, 0)
 325     subelems += struct.pack("<BB", 71, 0)
 326     req['payload'] += struct.pack("<BB6BLBBB", 52, 13 + len(subelems),
 327                                   1, 2, 3, 4, 5, 6,
 328                                   0, 81, 1, 7) + subelems
 329     hapd.mgmt_tx(req)
 330     resp = rx_bss_tm_resp(hapd, expect_dialog=8, expect_status=1)
 331
 332     # Preferred Candidate List with lots of valid optional subelements (twice)
 333     req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'],
 334                      req_mode=0x01, dialog_token=8)
 335     # TSF Information
 336     subelems = struct.pack("<BBHH", 1, 4, 0, 100)
 337     # Condensed Country String
 338     subelems += struct.pack("<BBBB", 2, 2, 65, 66)
 339     # BSS Transition Candidate Preference
 340     subelems += struct.pack("<BBB", 3, 1, 100)
 341     # BSS Termination Duration
 342     subelems += struct.pack("<BBQH", 4, 10, 0, 10)
 343     # Bearing
 344     subelems += struct.pack("<BBHLH", 5, 8, 0, 0, 0)
 345     # Measurement Pilot Transmission
 346     subelems += struct.pack("<BBBBB", 66, 3, 0, 0, 0)
 347     # RM Enabled Capabilities
 348     subelems += struct.pack("<BBBBBBB", 70, 5, 0, 0, 0, 0, 0)
 349     # Multiple BSSID
 350     subelems += struct.pack("<BBBB", 71, 2, 0, 0)
 351     req['payload'] += struct.pack("<BB6BLBBB", 52, 13 + len(subelems) * 2,
 352                                   1, 2, 3, 4, 5, 6,
 353                                   0, 81, 1, 7) + subelems + subelems
 354     hapd.mgmt_tx(req)
 355     resp = rx_bss_tm_resp(hapd, expect_dialog=8, expect_status=1)
 356
 357 def test_wnm_bss_keep_alive(dev, apdev):
 358     """WNM keep-alive"""
 359     params = { "ssid": "test-wnm",
 360                "ap_max_inactivity": "1" }
 361     hostapd.add_ap(apdev[0]['ifname'], params)
 362
 363     dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412")
 364     time.sleep(2)