# WNM tests # Copyright (c) 2013-2014, Jouni Malinen <j@w1.fi> # # This software may be distributed under the terms of the BSD license. # See README for more details. import binascii import struct import time import logging logger = logging.getLogger() import hostapd from wlantest import Wlantest def test_wnm_bss_transition_mgmt(dev, apdev): """WNM BSS Transition Management""" params = { "ssid": "test-wnm", "time_advertisement": "2", "time_zone": "EST5", "wnm_sleep_mode": "1", "bss_transition": "1" } hostapd.add_ap(apdev[0]['ifname'], params) dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412") dev[0].request("WNM_BSS_QUERY 0") def test_wnm_disassoc_imminent(dev, apdev): """WNM Disassociation Imminent""" params = { "ssid": "test-wnm", "time_advertisement": "2", "time_zone": "EST5", "wnm_sleep_mode": "1", "bss_transition": "1" } hostapd.add_ap(apdev[0]['ifname'], params) hapd = hostapd.Hostapd(apdev[0]['ifname']) dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412") addr = dev[0].p2p_interface_addr() hapd.request("DISASSOC_IMMINENT " + addr + " 10") ev = dev[0].wait_event(["WNM: Disassociation Imminent"]) if ev is None: raise Exception("Timeout while waiting for disassociation imminent") if "Disassociation Timer 10" not in ev: raise Exception("Unexpected disassociation imminent contents") ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"]) if ev is None: raise Exception("Timeout while waiting for re-connection scan") def test_wnm_ess_disassoc_imminent(dev, apdev): """WNM ESS Disassociation Imminent""" params = { "ssid": "test-wnm", "time_advertisement": "2", "time_zone": "EST5", "wnm_sleep_mode": "1", "bss_transition": "1" } hostapd.add_ap(apdev[0]['ifname'], params) hapd = hostapd.Hostapd(apdev[0]['ifname']) dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412") addr = dev[0].p2p_interface_addr() hapd.request("ESS_DISASSOC " + addr + " 10 http://example.com/session-info") ev = dev[0].wait_event(["ESS-DISASSOC-IMMINENT"]) if ev is None: raise Exception("Timeout while waiting for ESS disassociation imminent") if "0 1024 http://example.com/session-info" not in ev: raise Exception("Unexpected ESS disassociation imminent message contents") ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"]) if ev is None: raise Exception("Timeout while waiting for re-connection scan") def test_wnm_ess_disassoc_imminent_pmf(dev, apdev): """WNM ESS Disassociation Imminent""" params = hostapd.wpa2_params("test-wnm-rsn", "12345678") params["wpa_key_mgmt"] = "WPA-PSK-SHA256"; params["ieee80211w"] = "2"; params["bss_transition"] = "1" hostapd.add_ap(apdev[0]['ifname'], params) hapd = hostapd.Hostapd(apdev[0]['ifname']) dev[0].connect("test-wnm-rsn", psk="12345678", ieee80211w="2", key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412") addr = dev[0].p2p_interface_addr() hapd.request("ESS_DISASSOC " + addr + " 10 http://example.com/session-info") ev = dev[0].wait_event(["ESS-DISASSOC-IMMINENT"]) if ev is None: raise Exception("Timeout while waiting for ESS disassociation imminent") if "1 1024 http://example.com/session-info" not in ev: raise Exception("Unexpected ESS disassociation imminent message contents") ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"]) if ev is None: raise Exception("Timeout while waiting for re-connection scan") def check_wnm_sleep_mode_enter_exit(hapd, dev, interval=None, tfs_req=None): addr = dev.p2p_interface_addr() sta = hapd.get_sta(addr) if "[WNM_SLEEP_MODE]" in sta['flags']: raise Exception("Station unexpectedly in WNM-Sleep Mode") logger.info("Going to WNM Sleep Mode") extra = "" if interval is not None: extra += " interval=" + str(interval) if tfs_req: extra += " tfs_req=" + tfs_req if "OK" not in dev.request("WNM_SLEEP enter" + extra): raise Exception("WNM_SLEEP failed") time.sleep(0.5) sta = hapd.get_sta(addr) if "[WNM_SLEEP_MODE]" not in sta['flags']: raise Exception("Station failed to enter WNM-Sleep Mode") logger.info("Waking up from WNM Sleep Mode") dev.request("WNM_SLEEP exit") time.sleep(0.5) sta = hapd.get_sta(addr) if "[WNM_SLEEP_MODE]" in sta['flags']: raise Exception("Station failed to exit WNM-Sleep Mode") def test_wnm_sleep_mode_open(dev, apdev): """WNM Sleep Mode - open""" params = { "ssid": "test-wnm", "time_advertisement": "2", "time_zone": "EST5", "wnm_sleep_mode": "1", "bss_transition": "1" } hostapd.add_ap(apdev[0]['ifname'], params) hapd = hostapd.Hostapd(apdev[0]['ifname']) dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412") check_wnm_sleep_mode_enter_exit(hapd, dev[0]) check_wnm_sleep_mode_enter_exit(hapd, dev[0], interval=100) check_wnm_sleep_mode_enter_exit(hapd, dev[0], tfs_req="5b17010001130e110000071122334455661122334455661234") def test_wnm_sleep_mode_rsn(dev, apdev): """WNM Sleep Mode - RSN""" params = hostapd.wpa2_params("test-wnm-rsn", "12345678") params["time_advertisement"] = "2" params["time_zone"] = "EST5" params["wnm_sleep_mode"] = "1" params["bss_transition"] = "1" hostapd.add_ap(apdev[0]['ifname'], params) hapd = hostapd.Hostapd(apdev[0]['ifname']) dev[0].connect("test-wnm-rsn", psk="12345678", scan_freq="2412") check_wnm_sleep_mode_enter_exit(hapd, dev[0]) def test_wnm_sleep_mode_rsn_pmf(dev, apdev): """WNM Sleep Mode - RSN with PMF""" wt = Wlantest() wt.flush() wt.add_passphrase("12345678") params = hostapd.wpa2_params("test-wnm-rsn", "12345678") params["wpa_key_mgmt"] = "WPA-PSK-SHA256"; params["ieee80211w"] = "2"; params["time_advertisement"] = "2" params["time_zone"] = "EST5" params["wnm_sleep_mode"] = "1" params["bss_transition"] = "1" hostapd.add_ap(apdev[0]['ifname'], params) hapd = hostapd.Hostapd(apdev[0]['ifname']) dev[0].connect("test-wnm-rsn", psk="12345678", ieee80211w="2", key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412") check_wnm_sleep_mode_enter_exit(hapd, dev[0]) MGMT_SUBTYPE_ACTION = 13 ACTION_CATEG_WNM = 10 WNM_ACT_BSS_TM_REQ = 7 WNM_ACT_BSS_TM_RESP = 8 def bss_tm_req(dst, src, dialog_token=1, req_mode=0, disassoc_timer=0, validity_interval=1): msg = {} msg['fc'] = MGMT_SUBTYPE_ACTION << 4 msg['da'] = dst msg['sa'] = src msg['bssid'] = src msg['payload'] = struct.pack("<BBBBHB", ACTION_CATEG_WNM, WNM_ACT_BSS_TM_REQ, dialog_token, req_mode, disassoc_timer, validity_interval) return msg def rx_bss_tm_resp(hapd, expect_dialog=None, expect_status=None): for i in range(0, 100): resp = hapd.mgmt_rx() if resp is None: raise Exception("No BSS TM Response received") if resp['subtype'] == MGMT_SUBTYPE_ACTION: break if i == 99: raise Exception("Not an Action frame") payload = resp['payload'] if len(payload) < 2 + 3: raise Exception("Too short payload") (category, action) = struct.unpack('BB', payload[0:2]) if category != ACTION_CATEG_WNM or action != WNM_ACT_BSS_TM_RESP: raise Exception("Not a BSS TM Response") pos = payload[2:] (dialog, status, bss_term_delay) = struct.unpack('BBB', pos[0:3]) resp['dialog'] = dialog resp['status'] = status resp['bss_term_delay'] = bss_term_delay pos = pos[3:] if len(pos) >= 6 and status == 0: resp['target_bssid'] = binascii.hexlify(pos[0:6]) pos = pos[6:] resp['candidates'] = pos if expect_dialog is not None and dialog != expect_dialog: raise Exception("Unexpected dialog token") if expect_status is not None and status != expect_status: raise Exception("Unexpected status code %d" % status) return resp def except_ack(hapd): ev = hapd.wait_event(["MGMT-TX-STATUS"], timeout=5) if ev is None: raise Exception("Missing TX status") if "ok=1" not in ev: raise Exception("Action frame not acknowledged") def test_wnm_bss_tm_req(dev, apdev): """BSS Transition Management Request""" params = { "ssid": "test-wnm", "bss_transition": "1" } hapd = hostapd.add_ap(apdev[0]['ifname'], params) dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412") hapd2 = hostapd.add_ap(apdev[1]['ifname'], params) hapd.set("ext_mgmt_frame_handling", "1") # truncated BSS TM Request req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x08) req['payload'] = struct.pack("<BBBBH", ACTION_CATEG_WNM, WNM_ACT_BSS_TM_REQ, 1, 0, 0) hapd.mgmt_tx(req) except_ack(hapd) # no disassociation and no candidate list req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], dialog_token=2) hapd.mgmt_tx(req) resp = rx_bss_tm_resp(hapd, expect_dialog=2, expect_status=1) # truncated BSS Termination Duration req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x08) hapd.mgmt_tx(req) except_ack(hapd) # BSS Termination Duration with TSF=0 and Duration=10 req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x08, dialog_token=3) req['payload'] += struct.pack("<BBQH", 4, 10, 0, 10) hapd.mgmt_tx(req) resp = rx_bss_tm_resp(hapd, expect_dialog=3, expect_status=1) # truncated Session Information URL req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x10) hapd.mgmt_tx(req) except_ack(hapd) req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x10) req['payload'] += struct.pack("<BBB", 3, 65, 66) hapd.mgmt_tx(req) except_ack(hapd) # Session Information URL req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x10, dialog_token=4) req['payload'] += struct.pack("<BBB", 2, 65, 66) hapd.mgmt_tx(req) resp = rx_bss_tm_resp(hapd, expect_dialog=4, expect_status=0) # Preferred Candidate List without any entries req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x01, dialog_token=5) hapd.mgmt_tx(req) resp = rx_bss_tm_resp(hapd, expect_dialog=5, expect_status=1) # Preferred Candidate List with a truncated entry req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x01) req['payload'] += struct.pack("<BB", 52, 1) hapd.mgmt_tx(req) except_ack(hapd) # Preferred Candidate List with a too short entry req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x01, dialog_token=6) req['payload'] += struct.pack("<BB", 52, 0) hapd.mgmt_tx(req) resp = rx_bss_tm_resp(hapd, expect_dialog=6, expect_status=1) # Preferred Candidate List with a non-matching entry req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x01, dialog_token=6) req['payload'] += struct.pack("<BB6BLBBB", 52, 13, 1, 2, 3, 4, 5, 6, 0, 81, 1, 7) hapd.mgmt_tx(req) resp = rx_bss_tm_resp(hapd, expect_dialog=6, expect_status=1) # Preferred Candidate List with a truncated subelement req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x01, dialog_token=7) req['payload'] += struct.pack("<BB6BLBBBBB", 52, 13 + 2, 1, 2, 3, 4, 5, 6, 0, 81, 1, 7, 1, 1) hapd.mgmt_tx(req) resp = rx_bss_tm_resp(hapd, expect_dialog=7, expect_status=1) # Preferred Candidate List with lots of invalid optional subelements req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x01, dialog_token=8) subelems = struct.pack("<BBHB", 1, 3, 0, 100) subelems += struct.pack("<BBB", 2, 1, 65) subelems += struct.pack("<BB", 3, 0) subelems += struct.pack("<BBQB", 4, 9, 0, 10) subelems += struct.pack("<BBHLB", 5, 7, 0, 0, 0) subelems += struct.pack("<BB", 66, 0) subelems += struct.pack("<BBBBBB", 70, 4, 0, 0, 0, 0) subelems += struct.pack("<BB", 71, 0) req['payload'] += struct.pack("<BB6BLBBB", 52, 13 + len(subelems), 1, 2, 3, 4, 5, 6, 0, 81, 1, 7) + subelems hapd.mgmt_tx(req) resp = rx_bss_tm_resp(hapd, expect_dialog=8, expect_status=1) # Preferred Candidate List with lots of valid optional subelements (twice) req = bss_tm_req(dev[0].p2p_interface_addr(), apdev[0]['bssid'], req_mode=0x01, dialog_token=8) # TSF Information subelems = struct.pack("<BBHH", 1, 4, 0, 100) # Condensed Country String subelems += struct.pack("<BBBB", 2, 2, 65, 66) # BSS Transition Candidate Preference subelems += struct.pack("<BBB", 3, 1, 100) # BSS Termination Duration subelems += struct.pack("<BBQH", 4, 10, 0, 10) # Bearing subelems += struct.pack("<BBHLH", 5, 8, 0, 0, 0) # Measurement Pilot Transmission subelems += struct.pack("<BBBBB", 66, 3, 0, 0, 0) # RM Enabled Capabilities subelems += struct.pack("<BBBBBBB", 70, 5, 0, 0, 0, 0, 0) # Multiple BSSID subelems += struct.pack("<BBBB", 71, 2, 0, 0) req['payload'] += struct.pack("<BB6BLBBB", 52, 13 + len(subelems) * 2, 1, 2, 3, 4, 5, 6, 0, 81, 1, 7) + subelems + subelems hapd.mgmt_tx(req) resp = rx_bss_tm_resp(hapd, expect_dialog=8, expect_status=1) def test_wnm_bss_keep_alive(dev, apdev): """WNM keep-alive""" params = { "ssid": "test-wnm", "ap_max_inactivity": "1" } hostapd.add_ap(apdev[0]['ifname'], params) dev[0].connect("test-wnm", key_mgmt="NONE", scan_freq="2412") time.sleep(2)