jeltz/hostap
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity
hostap/tests/hwsim/test_gas.py
Jouni Malinen 4ef7053119 tests: Make GAS test cases more robust 
Scan explicitly for the specific AP to work around issues where under
heavy CPU load, the single active scan round may miss the delayed Probe
Response from the AP. In addition, verify that ANQP_GET commands succeed
to make error cases clearer in the log.

Signed-off-by: Jouni Malinen <j@w1.fi>
2014-05-16 00:30:40 +03:00

648 lines
24 KiB
Python
Raw Blame History

# GAS tests
# Copyright (c) 2013, Qualcomm Atheros, Inc.
# 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 time
import binascii
import logging
logger = logging.getLogger()
import re
import struct
import hostapd
def hs20_ap_params():
params = hostapd.wpa2_params(ssid="test-gas")
params['wpa_key_mgmt'] = "WPA-EAP"
params['ieee80211w'] = "1"
params['ieee8021x'] = "1"
params['auth_server_addr'] = "127.0.0.1"
params['auth_server_port'] = "1812"
params['auth_server_shared_secret'] = "radius"
params['interworking'] = "1"
params['access_network_type'] = "14"
params['internet'] = "1"
params['asra'] = "0"
params['esr'] = "0"
params['uesa'] = "0"
params['venue_group'] = "7"
params['venue_type'] = "1"
params['venue_name'] = [ "eng:Example venue", "fin:Esimerkkipaikka" ]
params['roaming_consortium'] = [ "112233", "1020304050", "010203040506",
"fedcba" ]
params['domain_name'] = "example.com,another.example.com"
params['nai_realm'] = [ "0,example.com,13[5:6],21[2:4][5:7]",
"0,another.example.com" ]
params['anqp_3gpp_cell_net'] = "244,91"
params['network_auth_type'] = "02http://www.example.com/redirect/me/here/"
params['ipaddr_type_availability'] = "14"
params['hs20'] = "1"
params['hs20_oper_friendly_name'] = [ "eng:Example operator", "fin:Esimerkkioperaattori" ]
params['hs20_wan_metrics'] = "01:8000:1000:80:240:3000"
params['hs20_conn_capab'] = [ "1:0:2", "6:22:1", "17:5060:0" ]
params['hs20_operating_class'] = "5173"
return params
def start_ap(ap):
params = hs20_ap_params()
params['hessid'] = ap['bssid']
hostapd.add_ap(ap['ifname'], params)
return hostapd.Hostapd(ap['ifname'])
def get_gas_response(dev, bssid, info, allow_fetch_failure=False):
exp = r'<.>(GAS-RESPONSE-INFO) addr=([0-9a-f:]*) dialog_token=([0-9]*) status_code=([0-9]*) resp_len=([\-0-9]*)'
res = re.split(exp, info)
if len(res) < 6:
raise Exception("Could not parse GAS-RESPONSE-INFO")
if res[2] != bssid:
raise Exception("Unexpected BSSID in response")
token = res[3]
status = res[4]
if status != "0":
raise Exception("GAS query failed")
resp_len = res[5]
if resp_len == "-1":
raise Exception("GAS query reported invalid response length")
if int(resp_len) > 2000:
raise Exception("Unexpected long GAS response")
resp = dev.request("GAS_RESPONSE_GET " + bssid + " " + token)
if "FAIL" in resp:
if allow_fetch_failure:
logger.debug("GAS response was not available anymore")
return
raise Exception("Could not fetch GAS response")
if len(resp) != int(resp_len) * 2:
raise Exception("Unexpected GAS response length")
logger.debug("GAS response: " + resp)
def test_gas_generic(dev, apdev):
"""Generic GAS query"""
bssid = apdev[0]['bssid']
params = hs20_ap_params()
params['hessid'] = bssid
hostapd.add_ap(apdev[0]['ifname'], params)
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000101")
if "FAIL" in req:
raise Exception("GAS query request rejected")
ev = dev[0].wait_event(["GAS-RESPONSE-INFO"], timeout=10)
if ev is None:
raise Exception("GAS query timed out")
get_gas_response(dev[0], bssid, ev)
def test_gas_concurrent_scan(dev, apdev):
"""Generic GAS queries with concurrent scan operation"""
bssid = apdev[0]['bssid']
params = hs20_ap_params()
params['hessid'] = bssid
hostapd.add_ap(apdev[0]['ifname'], params)
# get BSS entry available to allow GAS query
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
logger.info("Request concurrent operations")
req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000101")
if "FAIL" in req:
raise Exception("GAS query request rejected")
req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000801")
if "FAIL" in req:
raise Exception("GAS query request rejected")
dev[0].scan(no_wait=True)
req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000201")
if "FAIL" in req:
raise Exception("GAS query request rejected")
req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000501")
if "FAIL" in req:
raise Exception("GAS query request rejected")
responses = 0
for i in range(0, 5):
ev = dev[0].wait_event(["GAS-RESPONSE-INFO", "CTRL-EVENT-SCAN-RESULTS"],
timeout=10)
if ev is None:
raise Exception("Operation timed out")
if "GAS-RESPONSE-INFO" in ev:
responses = responses + 1
get_gas_response(dev[0], bssid, ev, allow_fetch_failure=True)
if responses != 4:
raise Exception("Unexpected number of GAS responses")
def test_gas_concurrent_connect(dev, apdev):
"""Generic GAS queries with concurrent connection operation"""
bssid = apdev[0]['bssid']
params = hs20_ap_params()
params['hessid'] = bssid
hostapd.add_ap(apdev[0]['ifname'], params)
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
logger.debug("Start concurrent connect and GAS request")
dev[0].connect("test-gas", key_mgmt="WPA-EAP", eap="TTLS",
identity="DOMAIN\mschapv2 user", anonymous_identity="ttls",
password="password", phase2="auth=MSCHAPV2",
ca_cert="auth_serv/ca.pem", wait_connect=False,
scan_freq="2412")
req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000101")
if "FAIL" in req:
raise Exception("GAS query request rejected")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED", "GAS-RESPONSE-INFO"],
timeout=20)
if ev is None:
raise Exception("Operation timed out")
if "CTRL-EVENT-CONNECTED" not in ev:
raise Exception("Unexpected operation order")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED", "GAS-RESPONSE-INFO"],
timeout=20)
if ev is None:
raise Exception("Operation timed out")
if "GAS-RESPONSE-INFO" not in ev:
raise Exception("Unexpected operation order")
get_gas_response(dev[0], bssid, ev)
dev[0].request("DISCONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection timed out")
logger.debug("Wait six seconds for expiration of connect-without-scan")
time.sleep(6)
dev[0].dump_monitor()
logger.debug("Start concurrent GAS request and connect")
req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000101")
if "FAIL" in req:
raise Exception("GAS query request rejected")
dev[0].request("RECONNECT")
ev = dev[0].wait_event(["GAS-RESPONSE-INFO"], timeout=10)
if ev is None:
raise Exception("Operation timed out")
get_gas_response(dev[0], bssid, ev)
ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"], timeout=20)
if ev is None:
raise Exception("No new scan results reported")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
if ev is None:
raise Exception("Operation timed out")
if "CTRL-EVENT-CONNECTED" not in ev:
raise Exception("Unexpected operation order")
def test_gas_fragment(dev, apdev):
"""GAS fragmentation"""
hapd = start_ap(apdev[0])
hapd.set("gas_frag_limit", "50")
dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412", force_scan=True)
dev[0].request("FETCH_ANQP")
for i in range(0, 13):
ev = dev[0].wait_event(["RX-ANQP", "RX-HS20-ANQP"], timeout=5)
if ev is None:
raise Exception("Operation timed out")
def test_gas_comeback_delay(dev, apdev):
"""GAS fragmentation"""
hapd = start_ap(apdev[0])
hapd.set("gas_comeback_delay", "500")
dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412", force_scan=True)
dev[0].request("FETCH_ANQP")
for i in range(0, 6):
ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
if ev is None:
raise Exception("Operation timed out")
def test_gas_anqp_get(dev, apdev):
"""GAS/ANQP query for both IEEE 802.11 and Hotspot 2.0 elements"""
hapd = start_ap(apdev[0])
bssid = apdev[0]['bssid']
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258,268,hs20:3,hs20:4"):
raise Exception("ANQP_GET command failed")
ev = dev[0].wait_event(["GAS-QUERY-START"], timeout=5)
if ev is None:
raise Exception("GAS query start timed out")
ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=10)
if ev is None:
raise Exception("GAS query timed out")
ev = dev[0].wait_event(["RX-ANQP"], timeout=1)
if ev is None or "Venue Name" not in ev:
raise Exception("Did not receive Venue Name")
ev = dev[0].wait_event(["RX-ANQP"], timeout=1)
if ev is None or "Domain Name list" not in ev:
raise Exception("Did not receive Domain Name list")
ev = dev[0].wait_event(["RX-HS20-ANQP"], timeout=1)
if ev is None or "Operator Friendly Name" not in ev:
raise Exception("Did not receive Operator Friendly Name")
ev = dev[0].wait_event(["RX-HS20-ANQP"], timeout=1)
if ev is None or "WAN Metrics" not in ev:
raise Exception("Did not receive WAN Metrics")
if "OK" not in dev[0].request("HS20_ANQP_GET " + bssid + " 3,4"):
raise Exception("ANQP_GET command failed")
ev = dev[0].wait_event(["RX-HS20-ANQP"], timeout=1)
if ev is None or "Operator Friendly Name" not in ev:
raise Exception("Did not receive Operator Friendly Name")
ev = dev[0].wait_event(["RX-HS20-ANQP"], timeout=1)
if ev is None or "WAN Metrics" not in ev:
raise Exception("Did not receive WAN Metrics")
def expect_gas_result(dev, result, status=None):
ev = dev.wait_event(["GAS-QUERY-DONE"], timeout=10)
if ev is None:
raise Exception("GAS query timed out")
if "result=" + result not in ev:
raise Exception("Unexpected GAS query result")
if status and "status_code=" + str(status) + ' ' not in ev:
raise Exception("Unexpected GAS status code")
def anqp_get(dev, bssid, id):
if "OK" not in dev.request("ANQP_GET " + bssid + " " + str(id)):
raise Exception("ANQP_GET command failed")
ev = dev.wait_event(["GAS-QUERY-START"], timeout=5)
if ev is None:
raise Exception("GAS query start timed out")
def test_gas_timeout(dev, apdev):
"""GAS timeout"""
hapd = start_ap(apdev[0])
bssid = apdev[0]['bssid']
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
hapd.set("ext_mgmt_frame_handling", "1")
anqp_get(dev[0], bssid, 263)
ev = hapd.wait_event(["MGMT-RX"], timeout=5)
if ev is None:
raise Exception("MGMT RX wait timed out")
expect_gas_result(dev[0], "TIMEOUT")
MGMT_SUBTYPE_ACTION = 13
ACTION_CATEG_PUBLIC = 4
GAS_INITIAL_REQUEST = 10
GAS_INITIAL_RESPONSE = 11
GAS_COMEBACK_REQUEST = 12
GAS_COMEBACK_RESPONSE = 13
GAS_ACTIONS = [ GAS_INITIAL_REQUEST, GAS_INITIAL_RESPONSE,
GAS_COMEBACK_REQUEST, GAS_COMEBACK_RESPONSE ]
def anqp_adv_proto():
return struct.pack('BBBB', 108, 2, 127, 0)
def anqp_initial_resp(dialog_token, status_code, comeback_delay=0):
return struct.pack('<BBBHH', ACTION_CATEG_PUBLIC, GAS_INITIAL_RESPONSE,
dialog_token, status_code, comeback_delay) + anqp_adv_proto()
def anqp_comeback_resp(dialog_token, status_code=0, id=0, more=False, comeback_delay=0, bogus_adv_proto=False):
if more:
id |= 0x80
if bogus_adv_proto:
adv = struct.pack('BBBB', 108, 2, 127, 1)
else:
adv = anqp_adv_proto()
return struct.pack('<BBBHBH', ACTION_CATEG_PUBLIC, GAS_COMEBACK_RESPONSE,
dialog_token, status_code, id, comeback_delay) + adv
def gas_rx(hapd):
count = 0
while count < 30:
count = count + 1
query = hapd.mgmt_rx()
if query is None:
raise Exception("Action frame not received")
if query['subtype'] != MGMT_SUBTYPE_ACTION:
continue
payload = query['payload']
if len(payload) < 2:
continue
(category, action) = struct.unpack('BB', payload[0:2])
if category != ACTION_CATEG_PUBLIC or action not in GAS_ACTIONS:
continue
return query
raise Exception("No Action frame received")
def parse_gas(payload):
pos = payload
(category, action, dialog_token) = struct.unpack('BBB', pos[0:3])
if category != ACTION_CATEG_PUBLIC:
return None
if action not in GAS_ACTIONS:
return None
gas = {}
gas['action'] = action
pos = pos[3:]
if len(pos) < 1 and action != GAS_COMEBACK_REQUEST:
return None
gas['dialog_token'] = dialog_token
return gas
def action_response(req):
resp = {}
resp['fc'] = req['fc']
resp['da'] = req['sa']
resp['sa'] = req['da']
resp['bssid'] = req['bssid']
return resp
def send_gas_resp(hapd, resp):
hapd.mgmt_tx(resp)
ev = hapd.wait_event(["MGMT-TX-STATUS"], timeout=5)
if ev is None:
raise Exception("Missing TX status for GAS response")
if "ok=1" not in ev:
raise Exception("GAS response not acknowledged")
def test_gas_invalid_response_type(dev, apdev):
"""GAS invalid response type"""
hapd = start_ap(apdev[0])
bssid = apdev[0]['bssid']
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
hapd.set("ext_mgmt_frame_handling", "1")
anqp_get(dev[0], bssid, 263)
query = gas_rx(hapd)
gas = parse_gas(query['payload'])
resp = action_response(query)
# GAS Comeback Response instead of GAS Initial Response
resp['payload'] = anqp_comeback_resp(gas['dialog_token']) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
# station drops the invalid frame, so this needs to result in GAS timeout
expect_gas_result(dev[0], "TIMEOUT")
def test_gas_failure_status_code(dev, apdev):
"""GAS failure status code"""
hapd = start_ap(apdev[0])
bssid = apdev[0]['bssid']
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
hapd.set("ext_mgmt_frame_handling", "1")
anqp_get(dev[0], bssid, 263)
query = gas_rx(hapd)
gas = parse_gas(query['payload'])
resp = action_response(query)
resp['payload'] = anqp_initial_resp(gas['dialog_token'], 61) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
expect_gas_result(dev[0], "FAILURE")
def test_gas_malformed(dev, apdev):
"""GAS malformed response frames"""
hapd = start_ap(apdev[0])
bssid = apdev[0]['bssid']
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
hapd.set("ext_mgmt_frame_handling", "1")
anqp_get(dev[0], bssid, 263)
query = gas_rx(hapd)
gas = parse_gas(query['payload'])
resp = action_response(query)
resp['payload'] = struct.pack('<BBBH', ACTION_CATEG_PUBLIC,
GAS_COMEBACK_RESPONSE,
gas['dialog_token'], 0)
hapd.mgmt_tx(resp)
resp['payload'] = struct.pack('<BBBHB', ACTION_CATEG_PUBLIC,
GAS_COMEBACK_RESPONSE,
gas['dialog_token'], 0, 0)
hapd.mgmt_tx(resp)
hdr = struct.pack('<BBBHH', ACTION_CATEG_PUBLIC, GAS_INITIAL_RESPONSE,
gas['dialog_token'], 0, 0)
resp['payload'] = hdr + struct.pack('B', 108)
hapd.mgmt_tx(resp)
resp['payload'] = hdr + struct.pack('BB', 108, 0)
hapd.mgmt_tx(resp)
resp['payload'] = hdr + struct.pack('BB', 108, 1)
hapd.mgmt_tx(resp)
resp['payload'] = hdr + struct.pack('BB', 108, 255)
hapd.mgmt_tx(resp)
resp['payload'] = hdr + struct.pack('BBB', 108, 1, 127)
hapd.mgmt_tx(resp)
resp['payload'] = hdr + struct.pack('BBB', 108, 2, 127)
hapd.mgmt_tx(resp)
resp['payload'] = hdr + struct.pack('BBBB', 0, 2, 127, 0)
hapd.mgmt_tx(resp)
resp['payload'] = anqp_initial_resp(gas['dialog_token'], 0) + struct.pack('<H', 1)
hapd.mgmt_tx(resp)
resp['payload'] = anqp_initial_resp(gas['dialog_token'], 0) + struct.pack('<HB', 2, 0)
hapd.mgmt_tx(resp)
resp['payload'] = anqp_initial_resp(gas['dialog_token'], 0) + struct.pack('<H', 65535)
hapd.mgmt_tx(resp)
resp['payload'] = anqp_initial_resp(gas['dialog_token'], 0) + struct.pack('<HBB', 1, 0, 0)
hapd.mgmt_tx(resp)
# Station drops invalid frames, but the last of the responses is valid from
# GAS view point even though it has an extra octet in the end and the ANQP
# part of the response is not valid. This is reported as successfulyl
# completed GAS exchange.
expect_gas_result(dev[0], "SUCCESS")
def init_gas(hapd, bssid, dev):
anqp_get(dev, bssid, 263)
query = gas_rx(hapd)
gas = parse_gas(query['payload'])
dialog_token = gas['dialog_token']
resp = action_response(query)
resp['payload'] = anqp_initial_resp(dialog_token, 0, comeback_delay=1) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
query = gas_rx(hapd)
gas = parse_gas(query['payload'])
if gas['action'] != GAS_COMEBACK_REQUEST:
raise Exception("Unexpected request action")
if gas['dialog_token'] != dialog_token:
raise Exception("Unexpected dialog token change")
return query, dialog_token
def test_gas_malformed_comeback_resp(dev, apdev):
"""GAS malformed comeback response frames"""
hapd = start_ap(apdev[0])
bssid = apdev[0]['bssid']
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
hapd.set("ext_mgmt_frame_handling", "1")
logger.debug("Non-zero status code in comeback response")
query, dialog_token = init_gas(hapd, bssid, dev[0])
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token, status_code=2) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
expect_gas_result(dev[0], "FAILURE", status=2)
logger.debug("Different advertisement protocol in comeback response")
query, dialog_token = init_gas(hapd, bssid, dev[0])
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token, bogus_adv_proto=True) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
expect_gas_result(dev[0], "PEER_ERROR")
logger.debug("Non-zero frag id and comeback delay in comeback response")
query, dialog_token = init_gas(hapd, bssid, dev[0])
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token, id=1, comeback_delay=1) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
expect_gas_result(dev[0], "PEER_ERROR")
logger.debug("Unexpected frag id in comeback response")
query, dialog_token = init_gas(hapd, bssid, dev[0])
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token, id=1) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
expect_gas_result(dev[0], "PEER_ERROR")
logger.debug("Empty fragment and replay in comeback response")
query, dialog_token = init_gas(hapd, bssid, dev[0])
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token, more=True) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
query = gas_rx(hapd)
gas = parse_gas(query['payload'])
if gas['action'] != GAS_COMEBACK_REQUEST:
raise Exception("Unexpected request action")
if gas['dialog_token'] != dialog_token:
raise Exception("Unexpected dialog token change")
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
resp['payload'] = anqp_comeback_resp(dialog_token, id=1) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
expect_gas_result(dev[0], "SUCCESS")
logger.debug("Unexpected initial response when waiting for comeback response")
query, dialog_token = init_gas(hapd, bssid, dev[0])
resp = action_response(query)
resp['payload'] = anqp_initial_resp(dialog_token, 0) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
ev = hapd.wait_event(["MGMT-RX"], timeout=1)
if ev is not None:
raise Exception("Unexpected management frame")
expect_gas_result(dev[0], "TIMEOUT")
logger.debug("Too short comeback response")
query, dialog_token = init_gas(hapd, bssid, dev[0])
resp = action_response(query)
resp['payload'] = struct.pack('<BBBH', ACTION_CATEG_PUBLIC,
GAS_COMEBACK_RESPONSE, dialog_token, 0)
send_gas_resp(hapd, resp)
ev = hapd.wait_event(["MGMT-RX"], timeout=1)
if ev is not None:
raise Exception("Unexpected management frame")
expect_gas_result(dev[0], "TIMEOUT")
logger.debug("Too short comeback response(2)")
query, dialog_token = init_gas(hapd, bssid, dev[0])
resp = action_response(query)
resp['payload'] = struct.pack('<BBBHBB', ACTION_CATEG_PUBLIC,
GAS_COMEBACK_RESPONSE, dialog_token, 0, 0x80,
0)
send_gas_resp(hapd, resp)
ev = hapd.wait_event(["MGMT-RX"], timeout=1)
if ev is not None:
raise Exception("Unexpected management frame")
expect_gas_result(dev[0], "TIMEOUT")
logger.debug("Maximum comeback response fragment claiming more fragments")
query, dialog_token = init_gas(hapd, bssid, dev[0])
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token, more=True) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
for i in range(1, 129):
query = gas_rx(hapd)
gas = parse_gas(query['payload'])
if gas['action'] != GAS_COMEBACK_REQUEST:
raise Exception("Unexpected request action")
if gas['dialog_token'] != dialog_token:
raise Exception("Unexpected dialog token change")
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token, id=i, more=True) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
expect_gas_result(dev[0], "PEER_ERROR")
def test_gas_comeback_resp_additional_delay(dev, apdev):
"""GAS comeback response requesting additional delay"""
hapd = start_ap(apdev[0])
bssid = apdev[0]['bssid']
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
hapd.set("ext_mgmt_frame_handling", "1")
query, dialog_token = init_gas(hapd, bssid, dev[0])
for i in range(0, 2):
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token, status_code=95, comeback_delay=50) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
query = gas_rx(hapd)
gas = parse_gas(query['payload'])
if gas['action'] != GAS_COMEBACK_REQUEST:
raise Exception("Unexpected request action")
if gas['dialog_token'] != dialog_token:
raise Exception("Unexpected dialog token change")
resp = action_response(query)
resp['payload'] = anqp_comeback_resp(dialog_token, status_code=0) + struct.pack('<H', 0)
send_gas_resp(hapd, resp)
expect_gas_result(dev[0], "SUCCESS")
def test_gas_unknown_adv_proto(dev, apdev):
"""Unknown advertisement protocol id"""
bssid = apdev[0]['bssid']
params = hs20_ap_params()
params['hessid'] = bssid
hostapd.add_ap(apdev[0]['ifname'], params)
dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
req = dev[0].request("GAS_REQUEST " + bssid + " 42 000102000101")
if "FAIL" in req:
raise Exception("GAS query request rejected")
expect_gas_result(dev[0], "FAILURE", "59")
ev = dev[0].wait_event(["GAS-RESPONSE-INFO"], timeout=10)
if ev is None:
raise Exception("GAS query timed out")
exp = r'<.>(GAS-RESPONSE-INFO) addr=([0-9a-f:]*) dialog_token=([0-9]*) status_code=([0-9]*) resp_len=([\-0-9]*)'
res = re.split(exp, ev)
if len(res) < 6:
raise Exception("Could not parse GAS-RESPONSE-INFO")
if res[2] != bssid:
raise Exception("Unexpected BSSID in response")
status = res[4]
if status != "59":
raise Exception("Unexpected GAS-RESPONSE-INFO status")
Reference in a new issue View git blame Copy permalink