hostapd/tests/hwsim/test_gas.py

#!/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 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"
    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()
    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"""
    hapd = start_ap(apdev[0])
    hapd.set("gas_frag_limit", "50")

    dev[0].scan(freq="2412")
    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_comeback_delay(dev, apdev):
    """GAS fragmentation"""
    hapd = start_ap(apdev[0])
    hapd.set("gas_comeback_delay", "500")

    dev[0].scan(freq="2412")
    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 expect_gas_result(dev, result):
    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")

def anqp_get(dev, bssid, id):
    dev.request("ANQP_GET " + bssid + " " + str(id))
    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(freq="2412")
    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):
    return struct.pack('<BBBHH', ACTION_CATEG_PUBLIC, GAS_INITIAL_RESPONSE,
                       dialog_token, status_code, 0) + anqp_adv_proto()

def anqp_comeback_resp(dialog_token):
    return struct.pack('<BBBHBH', ACTION_CATEG_PUBLIC, GAS_COMEBACK_RESPONSE,
                       dialog_token, 0, 0, 0) + anqp_adv_proto()

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:
        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 test_gas_invalid_response_type(dev, apdev):
    """GAS invalid response type"""
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']

    dev[0].scan(freq="2412")
    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)
    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")

    # 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(freq="2412")
    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)
    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")

    expect_gas_result(dev[0], "FAILURE")