197 lines
7 KiB
Python
197 lines
7 KiB
Python
#!/usr/bin/python
|
|
#
|
|
# GAS tests
|
|
# Copyright (c) 2013, Qualcomm Atheros, Inc.
|
|
#
|
|
# This software may be distributed under the terms of the BSD license.
|
|
# See README for more details.
|
|
|
|
import time
|
|
import logging
|
|
logger = logging.getLogger()
|
|
import re
|
|
|
|
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"
|
|
return params
|
|
|
|
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()
|
|
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)
|
|
|
|
dev[0].scan()
|
|
|
|
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].request("SCAN")
|
|
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()
|
|
|
|
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)
|
|
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)
|
|
|
|
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"""
|
|
bssid = apdev[0]['bssid']
|
|
params = hs20_ap_params()
|
|
params['hessid'] = bssid
|
|
hostapd.add_ap(apdev[0]['ifname'], params)
|
|
hapd = hostapd.Hostapd(apdev[0]['ifname'])
|
|
hapd.set("gas_frag_limit", "50")
|
|
|
|
dev[0].scan()
|
|
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")
|