hostap/tests/hwsim/test_gas.py

# GAS tests
# Copyright (c) 2013, Qualcomm Atheros, Inc.
# Copyright (c) 2013-2015, Jouni Malinen <j@w1.fi>
#
# This software may be distributed under the terms of the BSD license.
# See README for more details.

from remotehost import remote_compatible
import time
import binascii
import logging
logger = logging.getLogger()
import os
import re
import struct

import hostapd
from wpasupplicant import WpaSupplicant
from tshark import run_tshark
from utils import alloc_fail, wait_fail_trigger, skip_with_fips, HwsimSkip
from hwsim import HWSimRadio

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']
    return hostapd.add_ap(ap, params)

def get_gas_response(dev, bssid, info, allow_fetch_failure=False,
                     extra_test=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")

    if extra_test:
        if "FAIL" not in dev.request("GAS_RESPONSE_GET " + bssid + " 123456"):
            raise Exception("Invalid dialog token accepted")
        if "FAIL-Invalid range" not in dev.request("GAS_RESPONSE_GET " + bssid + " " + token + " 10000,10001"):
            raise Exception("Invalid range accepted")
        if "FAIL-Invalid range" not in dev.request("GAS_RESPONSE_GET " + bssid + " " + token + " 0,10000"):
            raise Exception("Invalid range accepted")
        if "FAIL" not in dev.request("GAS_RESPONSE_GET " + bssid + " " + token + " 0"):
            raise Exception("Invalid GAS_RESPONSE_GET accepted")

        res1_2 = dev.request("GAS_RESPONSE_GET " + bssid + " " + token + " 1,2")
        res5_3 = dev.request("GAS_RESPONSE_GET " + bssid + " " + token + " 5,3")

    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)
    if extra_test:
        if resp[2:6] != res1_2:
            raise Exception("Unexpected response substring res1_2: " + res1_2)
        if resp[10:16] != res5_3:
            raise Exception("Unexpected response substring res5_3: " + res5_3)

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], params)

    cmds = ["foo",
            "00:11:22:33:44:55",
            "00:11:22:33:44:55 ",
            "00:11:22:33:44:55  ",
            "00:11:22:33:44:55 1",
            "00:11:22:33:44:55 1 1234",
            "00:11:22:33:44:55 qq",
            "00:11:22:33:44:55 qq 1234",
            "00:11:22:33:44:55 00      1",
            "00:11:22:33:44:55 00 123",
            "00:11:22:33:44:55 00 ",
            "00:11:22:33:44:55 00 qq"]
    for cmd in cmds:
        if "FAIL" not in dev[0].request("GAS_REQUEST " + cmd):
            raise Exception("Invalid GAS_REQUEST accepted: " + cmd)

    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, extra_test=True)

    if "FAIL" not in dev[0].request("GAS_RESPONSE_GET ff"):
        raise Exception("Invalid GAS_RESPONSE_GET accepted")

def test_gas_rand_ta(dev, apdev, params):
    """Generic GAS query with random TA"""
    flags = int(dev[0].get_driver_status_field('capa.flags'), 16)
    if flags & 0x0000400000000000 == 0:
        raise HwsimSkip("Driver does not support random GAS TA")

    try:
        _test_gas_rand_ta(dev, apdev, params['logdir'])
    finally:
        dev[0].request("SET gas_rand_mac_addr 0")

def _test_gas_rand_ta(dev, apdev, logdir):
    bssid = apdev[0]['bssid']
    params = hs20_ap_params()
    params['hessid'] = bssid
    hostapd.add_ap(apdev[0], params)

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
    req = dev[0].request("SET gas_rand_mac_addr 1")
    if "FAIL" in req:
        raise Exception("Failed to set gas_rand_mac_addr")

    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, extra_test=True)

    out = run_tshark(os.path.join(logdir, "hwsim0.pcapng"),
                     "wlan_mgt.fixed.category_code == 4 && (wlan_mgt.fixed.publicact == 0x0a || wlan_mgt.fixed.publicact == 0x0b)",
                     display=["wlan.ta", "wlan.ra"])
    res = out.splitlines()
    if len(res) != 2:
        raise Exception("Unexpected number of GAS frames")
    req_ta = res[0].split('\t')[0]
    resp_ra = res[1].split('\t')[1]
    logger.info("Request TA: %s, Response RA: %s" % (req_ta, resp_ra))
    if req_ta != resp_ra:
        raise Exception("Request TA does not match response RA")
    if req_ta == dev[0].own_addr():
        raise Exception("Request TA was own permanent MAC address, not random")

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], 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"""
    skip_with_fips(dev[0])
    bssid = apdev[0]['bssid']
    params = hs20_ap_params()
    params['hessid'] = bssid
    hostapd.add_ap(apdev[0], 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")
    dev[0].wait_disconnected(timeout=5)

    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_connected(timeout=20, error="Operation tiemd out")
    if "CTRL-EVENT-CONNECTED" not in ev:
        raise Exception("Unexpected operation order")

def gas_fragment_and_comeback(dev, apdev, frag_limit=0, comeback_delay=0):
    hapd = start_ap(apdev)
    if frag_limit:
        hapd.set("gas_frag_limit", str(frag_limit))
    if comeback_delay:
        hapd.set("gas_comeback_delay", str(comeback_delay))

    dev.scan_for_bss(apdev['bssid'], freq="2412", force_scan=True)
    dev.request("FETCH_ANQP")
    ev = dev.wait_event(["GAS-QUERY-DONE"], timeout=5)
    if ev is None:
        raise Exception("No GAS-QUERY-DONE event")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected GAS result: " + ev)
    for i in range(0, 13):
        ev = dev.wait_event(["RX-ANQP", "RX-HS20-ANQP"], timeout=5)
        if ev is None:
            raise Exception("Operation timed out")
    ev = dev.wait_event(["ANQP-QUERY-DONE"], timeout=1)
    if ev is None:
        raise Exception("No ANQP-QUERY-DONE event")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected ANQP result: " + ev)

def test_gas_fragment(dev, apdev):
    """GAS fragmentation"""
    gas_fragment_and_comeback(dev[0], apdev[0], frag_limit=50)

def test_gas_fragment_mcc(dev, apdev):
    """GAS fragmentation with mac80211_hwsim MCC enabled"""
    with HWSimRadio(n_channels=2) as (radio, iface):
        wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
        wpas.interface_add(iface)
        gas_fragment_and_comeback(wpas, apdev[0], frag_limit=50)

def test_gas_fragment_with_comeback_delay(dev, apdev):
    """GAS fragmentation and comeback delay"""
    gas_fragment_and_comeback(dev[0], apdev[0], frag_limit=50,
                              comeback_delay=500)

def test_gas_fragment_with_comeback_delay_mcc(dev, apdev):
    """GAS fragmentation and comeback delay with mac80211_hwsim MCC enabled"""
    with HWSimRadio(n_channels=2) as (radio, iface):
        wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
        wpas.interface_add(iface)
        gas_fragment_and_comeback(wpas, apdev[0], frag_limit=50,
                                  comeback_delay=500)

def test_gas_comeback_delay(dev, apdev):
    """GAS comeback delay"""
    run_gas_comeback_delay(dev, apdev, 500)

def test_gas_comeback_delay_long(dev, apdev):
    """GAS long comeback delay"""
    run_gas_comeback_delay(dev, apdev, 2500)

def test_gas_comeback_delay_long2(dev, apdev):
    """GAS long comeback delay over default STA timeout"""
    run_gas_comeback_delay(dev, apdev, 6000)

def run_gas_comeback_delay(dev, apdev, delay):
    hapd = start_ap(apdev[0])
    hapd.set("gas_comeback_delay", str(delay))

    dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412", force_scan=True)
    dev[0].request("FETCH_ANQP")
    if "FAIL-BUSY" not in dev[0].request("SCAN"):
        raise Exception("SCAN accepted during FETCH_ANQP")
    for i in range(0, 6):
        ev = dev[0].wait_event(["RX-ANQP"], timeout=10)
        if ev is None:
            raise Exception("Operation timed out")

@remote_compatible
def test_gas_stop_fetch_anqp(dev, apdev):
    """Stop FETCH_ANQP operation"""
    hapd = start_ap(apdev[0])

    dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412", force_scan=True)
    hapd.set("ext_mgmt_frame_handling", "1")
    dev[0].request("FETCH_ANQP")
    dev[0].request("STOP_FETCH_ANQP")
    hapd.set("ext_mgmt_frame_handling", "0")
    ev = dev[0].wait_event(["RX-ANQP", "GAS-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("GAS-QUERY-DONE timed out")
    if "RX-ANQP" in ev:
        raise Exception("Unexpected ANQP response received")

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")

    ev = dev[0].wait_event(["ANQP-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("ANQP-QUERY-DONE event not seen")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected result: " + ev)

    if "OK" not in dev[0].request("ANQP_GET " + bssid + " hs20:3"):
        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-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(["ANQP-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("ANQP-QUERY-DONE event not seen")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected result: " + ev)

    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")

    logger.info("Attempt an MBO request with an AP that does not support MBO")
    if "OK" not in dev[0].request("ANQP_GET " + bssid + " 272,mbo:2"):
        raise Exception("ANQP_GET command failed (2)")

    ev = dev[0].wait_event(["GAS-QUERY-START"], timeout=5)
    if ev is None:
        raise Exception("GAS query start timed out (2)")

    ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("GAS query timed out (2)")

    cmds = ["",
            "foo",
            "00:11:22:33:44:55 258,hs20:-1",
            "00:11:22:33:44:55 258,hs20:0",
            "00:11:22:33:44:55 258,hs20:32",
            "00:11:22:33:44:55 hs20:-1",
            "00:11:22:33:44:55 hs20:0",
            "00:11:22:33:44:55 hs20:32",
            "00:11:22:33:44:55 mbo:-1",
            "00:11:22:33:44:55 mbo:0",
            "00:11:22:33:44:55 mbo:999",
            "00:11:22:33:44:55 mbo:1,258,mbo:2,mbo:3,259",
            "00:11:22:33:44:55",
            "00:11:22:33:44:55 ",
            "00:11:22:33:44:55 0",
            "00:11:22:33:44:55 1"]
    for cmd in cmds:
        if "FAIL" not in dev[0].request("ANQP_GET " + cmd):
            raise Exception("Invalid ANQP_GET accepted")

    cmds = ["",
            "foo",
            "00:11:22:33:44:55 -1",
            "00:11:22:33:44:55 0",
            "00:11:22:33:44:55 32",
            "00:11:22:33:44:55",
            "00:11:22:33:44:55 ",
            "00:11:22:33:44:55 0",
            "00:11:22:33:44:55 1"]
    for cmd in cmds:
        if "FAIL" not in dev[0].request("HS20_ANQP_GET " + cmd):
            raise Exception("Invalid HS20_ANQP_GET accepted")

def test_gas_anqp_get_no_scan(dev, apdev):
    """GAS/ANQP query without scan"""
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']
    if "OK" not in dev[0].request("ANQP_GET " + bssid + " freq=2412 258"):
        raise Exception("ANQP_GET command failed")
    ev = dev[0].wait_event(["ANQP-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("ANQP query timed out")
    dev[0].dump_monitor()

    if "OK" not in dev[0].request("ANQP_GET 02:11:22:33:44:55 freq=2417 258"):
        raise Exception("ANQP_GET command failed")
    ev = dev[0].wait_event(["ANQP-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("ANQP query timed out [2]")
    if "result=FAILURE" not in ev:
        raise Exception("Unexpected result: " + ev)

def test_gas_anqp_get_oom(dev, apdev):
    """GAS/ANQP query OOM"""
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
    with alloc_fail(dev[0], 1, "wpabuf_alloc;anqp_send_req"):
        if "FAIL" not in dev[0].request("ANQP_GET " + bssid + " 258,268,hs20:3,hs20:4"):
            raise Exception("ANQP_GET command accepted during OOM")
    with alloc_fail(dev[0], 1, "hs20_build_anqp_req;hs20_anqp_send_req"):
        if "FAIL" not in dev[0].request("HS20_ANQP_GET " + bssid + " 1"):
            raise Exception("HS20_ANQP_GET command accepted during OOM")
    with alloc_fail(dev[0], 1, "gas_query_req;hs20_anqp_send_req"):
        if "FAIL" not in dev[0].request("HS20_ANQP_GET " + bssid + " 1"):
            raise Exception("HS20_ANQP_GET command accepted during OOM")
    with alloc_fail(dev[0], 1, "=hs20_anqp_send_req"):
        if "FAIL" not in dev[0].request("REQ_HS20_ICON " + bssid + " w1fi_logo"):
            raise Exception("REQ_HS20_ICON command accepted during OOM")
    with alloc_fail(dev[0], 2, "=hs20_anqp_send_req"):
        if "FAIL" not in dev[0].request("REQ_HS20_ICON " + bssid + " w1fi_logo"):
            raise Exception("REQ_HS20_ICON command accepted during OOM")

def test_gas_anqp_icon_binary_proto(dev, apdev):
    """GAS/ANQP and icon binary protocol testing"""
    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")

    tests = ['010000', '01000000', '00000000', '00030000', '00020000',
             '00000100', '0001ff0100ee', '0001ff0200ee']
    for test in tests:
        dev[0].request("HS20_ICON_REQUEST " + bssid + " w1fi_logo")
        query = gas_rx(hapd)
        gas = parse_gas(query['payload'])
        resp = action_response(query)
        data = binascii.unhexlify(test)
        data = binascii.unhexlify('506f9a110b00') + data
        data = struct.pack('<HHH', len(data) + 4, 0xdddd, len(data)) + data
        resp['payload'] = anqp_initial_resp(gas['dialog_token'], 0) + data
        send_gas_resp(hapd, resp)
        expect_gas_result(dev[0], "SUCCESS")

def test_gas_anqp_hs20_proto(dev, apdev):
    """GAS/ANQP and Hotspot 2.0 element protocol testing"""
    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")

    tests = ['00', '0100', '0201', '0300', '0400', '0500', '0600', '0700',
             '0800', '0900', '0a00', '0b0000000000']
    for test in tests:
        dev[0].request("HS20_ANQP_GET " + bssid + " 3,4")
        query = gas_rx(hapd)
        gas = parse_gas(query['payload'])
        resp = action_response(query)
        data = binascii.unhexlify(test)
        data = binascii.unhexlify('506f9a11') + data
        data = struct.pack('<HHH', len(data) + 4, 0xdddd, len(data)) + data
        resp['payload'] = anqp_initial_resp(gas['dialog_token'], 0) + data
        send_gas_resp(hapd, resp)
        expect_gas_result(dev[0], "SUCCESS")

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

    if action == GAS_INITIAL_RESPONSE:
        if len(pos) < 4:
            return None
        (status_code, comeback_delay) = struct.unpack('<HH', pos[0:4])
        gas['status_code'] = status_code
        gas['comeback_delay'] = comeback_delay

    if action == GAS_COMEBACK_RESPONSE:
        if len(pos) < 5:
            return None
        (status_code, frag, comeback_delay) = struct.unpack('<HBH', pos[0:5])
        gas['status_code'] = status_code
        gas['frag'] = frag
        gas['comeback_delay'] = comeback_delay

    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 successfully
    # completed GAS exchange.
    expect_gas_result(dev[0], "SUCCESS")

    ev = dev[0].wait_event(["ANQP-QUERY-DONE"], timeout=5)
    if ev is None:
        raise Exception("ANQP-QUERY-DONE not reported")
    if "result=INVALID_FRAME" not in ev:
        raise Exception("Unexpected result: " + ev)

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 allow_gas_initial_req(hapd, dialog_token):
    msg = hapd.mgmt_rx(timeout=1)
    if msg is not None:
        gas = parse_gas(msg['payload'])
        if gas['action'] != GAS_INITIAL_REQUEST or dialog_token == gas['dialog_token']:
            raise Exception("Unexpected management frame")

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)
    allow_gas_initial_req(hapd, dialog_token)
    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)
    allow_gas_initial_req(hapd, dialog_token)
    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)
    allow_gas_initial_req(hapd, dialog_token)
    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], 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")

def test_gas_request_oom(dev, apdev):
    """GAS_REQUEST OOM"""
    bssid = apdev[0]['bssid']
    params = hs20_ap_params()
    params['hessid'] = bssid
    hostapd.add_ap(apdev[0], params)

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)

    with alloc_fail(dev[0], 1, "gas_build_req;gas_send_request"):
        if "FAIL" not in dev[0].request("GAS_REQUEST " + bssid + " 42"):
            raise Exception("GAS query request rejected")

    with alloc_fail(dev[0], 1, "gas_query_req;gas_send_request"):
        if "FAIL" not in dev[0].request("GAS_REQUEST " + bssid + " 42"):
            raise Exception("GAS query request rejected")

    with alloc_fail(dev[0], 1, "wpabuf_dup;gas_resp_cb"):
        if "OK" not in dev[0].request("GAS_REQUEST " + bssid + " 00 000102000101"):
            raise Exception("GAS query request rejected")
        ev = dev[0].wait_event(["GAS-RESPONSE-INFO"], timeout=10)
        if ev is None:
            raise Exception("No GAS response")
        if "status_code=0" not in ev:
            raise Exception("GAS response indicated a failure")

def test_gas_max_pending(dev, apdev):
    """GAS and maximum pending query limit"""
    hapd = start_ap(apdev[0])
    hapd.set("gas_frag_limit", "50")
    bssid = apdev[0]['bssid']

    wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
    wpas.interface_add("wlan5")
    if "OK" not in wpas.request("P2P_SET listen_channel 1"):
        raise Exception("Failed to set listen channel")
    if "OK" not in wpas.p2p_listen():
        raise Exception("Failed to start listen state")
    if "FAIL" in wpas.request("SET ext_mgmt_frame_handling 1"):
        raise Exception("Failed to enable external management frame handling")

    anqp_query = struct.pack('<HHHHHHHHHH', 256, 16, 257, 258, 260, 261, 262, 263, 264, 268)
    gas = struct.pack('<H', len(anqp_query)) + anqp_query

    for dialog_token in range(1, 10):
        msg = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_INITIAL_REQUEST,
                          dialog_token) + anqp_adv_proto() + gas
        req = "MGMT_TX {} {} freq=2412 wait_time=10 action={}".format(bssid, bssid, binascii.hexlify(msg).decode())
        if "OK" not in wpas.request(req):
            raise Exception("Could not send management frame")
        resp = wpas.mgmt_rx()
        if resp is None:
            raise Exception("MGMT-RX timeout")
        if 'payload' not in resp:
            raise Exception("Missing payload")
        gresp = parse_gas(resp['payload'])
        if gresp['dialog_token'] != dialog_token:
            raise Exception("Dialog token mismatch")
        status_code = gresp['status_code']
        if dialog_token < 9 and status_code != 0:
            raise Exception("Unexpected failure status code {} for dialog token {}".format(status_code, dialog_token))
        if dialog_token > 8 and status_code == 0:
            raise Exception("Unexpected success status code {} for dialog token {}".format(status_code, dialog_token))

def test_gas_no_pending(dev, apdev):
    """GAS and no pending query for comeback request"""
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']

    wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
    wpas.interface_add("wlan5")
    if "OK" not in wpas.request("P2P_SET listen_channel 1"):
        raise Exception("Failed to set listen channel")
    if "OK" not in wpas.p2p_listen():
        raise Exception("Failed to start listen state")
    if "FAIL" in wpas.request("SET ext_mgmt_frame_handling 1"):
        raise Exception("Failed to enable external management frame handling")

    msg = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_COMEBACK_REQUEST, 1)
    req = "MGMT_TX {} {} freq=2412 wait_time=10 action={}".format(bssid, bssid, binascii.hexlify(msg).decode())
    if "OK" not in wpas.request(req):
        raise Exception("Could not send management frame")
    resp = wpas.mgmt_rx()
    if resp is None:
        raise Exception("MGMT-RX timeout")
    if 'payload' not in resp:
        raise Exception("Missing payload")
    gresp = parse_gas(resp['payload'])
    status_code = gresp['status_code']
    if status_code != 60:
        raise Exception("Unexpected status code {} (expected 60)".format(status_code))

def test_gas_delete_at_deinit(dev, apdev):
    """GAS query deleted at deinit"""
    hapd = start_ap(apdev[0])
    hapd.set("gas_comeback_delay", "1000")
    bssid = apdev[0]['bssid']

    wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
    wpas.interface_add("wlan5")
    wpas.scan_for_bss(apdev[0]['bssid'], freq="2412", force_scan=True)
    wpas.request("ANQP_GET " + bssid + " 258")

    wpas.global_request("INTERFACE_REMOVE " + wpas.ifname)
    ev = wpas.wait_event(["GAS-QUERY-DONE"], timeout=2)
    del wpas
    if ev is None:
        raise Exception("GAS-QUERY-DONE not seen")
    if "result=DELETED_AT_DEINIT" not in ev:
        raise Exception("Unexpected result code: " + ev)

def test_gas_missing_payload(dev, apdev):
    """No action code in the query frame"""
    bssid = apdev[0]['bssid']
    params = hs20_ap_params()
    params['hessid'] = bssid
    hostapd.add_ap(apdev[0], params)

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)

    cmd = "MGMT_TX {} {} freq=2412 action=040A".format(bssid, bssid)
    if "FAIL" in dev[0].request(cmd):
        raise Exception("Could not send test Action frame")
    ev = dev[0].wait_event(["MGMT-TX-STATUS"], timeout=10)
    if ev is None:
        raise Exception("Timeout on MGMT-TX-STATUS")
    if "result=SUCCESS" not in ev:
        raise Exception("AP did not ack Action frame")

    cmd = "MGMT_TX {} {} freq=2412 action=04".format(bssid, bssid)
    if "FAIL" in dev[0].request(cmd):
        raise Exception("Could not send test Action frame")
    ev = dev[0].wait_event(["MGMT-TX-STATUS"], timeout=10)
    if ev is None:
        raise Exception("Timeout on MGMT-TX-STATUS")
    if "result=SUCCESS" not in ev:
        raise Exception("AP did not ack Action frame")

def test_gas_query_deinit(dev, apdev):
    """Pending GAS/ANQP query during deinit"""
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']

    wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
    wpas.interface_add("wlan5")

    wpas.scan_for_bss(bssid, freq="2412", force_scan=True)
    id = wpas.request("RADIO_WORK add block-work")
    if "OK" not in wpas.request("ANQP_GET " + bssid + " 258"):
        raise Exception("ANQP_GET command failed")

    ev = wpas.wait_event(["GAS-QUERY-START", "EXT-RADIO-WORK-START"], timeout=5)
    if ev is None:
        raise Exception("Timeout while waiting radio work to start")
    ev = wpas.wait_event(["GAS-QUERY-START", "EXT-RADIO-WORK-START"], timeout=5)
    if ev is None:
        raise Exception("Timeout while waiting radio work to start (2)")

    # Remove the interface while the gas-query radio work is still pending and
    # GAS query has not yet been started.
    wpas.interface_remove("wlan5")

@remote_compatible
def test_gas_anqp_oom_wpas(dev, apdev):
    """GAS/ANQP query and OOM in wpa_supplicant"""
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)

    with alloc_fail(dev[0], 1, "wpa_bss_anqp_alloc"):
        if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
            raise Exception("ANQP_GET command failed")
        ev = dev[0].wait_event(["ANQP-QUERY-DONE"], timeout=5)
        if ev is None:
            raise Exception("ANQP query did not complete")

    with alloc_fail(dev[0], 1, "gas_build_req"):
        if "FAIL" not in dev[0].request("ANQP_GET " + bssid + " 258"):
            raise Exception("Unexpected ANQP_GET command success (OOM)")

def test_gas_anqp_oom_hapd(dev, apdev):
    """GAS/ANQP query and OOM in hostapd"""
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)

    with alloc_fail(hapd, 1, "gas_build_resp"):
        # This query will time out due to the AP not sending a response (OOM).
        if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
            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")
        if "result=TIMEOUT" not in ev:
            raise Exception("Unexpected result: " + ev)

        ev = dev[0].wait_event(["ANQP-QUERY-DONE"], timeout=10)
        if ev is None:
            raise Exception("ANQP-QUERY-DONE event not seen")
        if "result=FAILURE" not in ev:
            raise Exception("Unexpected result: " + ev)

    with alloc_fail(hapd, 1, "gas_anqp_build_comeback_resp"):
        hapd.set("gas_frag_limit", "50")

        # The first attempt of this query will time out due to the AP not
        # sending a response (OOM), but the retry succeeds.
        dev[0].request("FETCH_ANQP")
        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")
        if "result=SUCCESS" not in ev:
            raise Exception("Unexpected result: " + ev)

        ev = dev[0].wait_event(["ANQP-QUERY-DONE"], timeout=10)
        if ev is None:
            raise Exception("ANQP-QUERY-DONE event not seen")
        if "result=SUCCESS" not in ev:
            raise Exception("Unexpected result: " + ev)

def test_gas_anqp_extra_elements(dev, apdev):
    """GAS/ANQP and extra ANQP elements"""
    geo_loc = "001052834d12efd2b08b9b4bf1cc2c00004104050000000000060100"
    civic_loc = "0000f9555302f50102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f202122232425262728292a2b2c2d2e2f303132333435363738393a3b3c3d3e3f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f606162636465666768696a6b6c6d6e6f707172737475767778797a7b7c7d7e7f808182838485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e9fa0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0f1f2f3f4f5"
    held_uri = "https://held.example.com/location"
    held = struct.pack('BBB', 0, 1 + len(held_uri), 1) + held_uri.encode()
    supl_fqdn = "supl.example.com"
    supl = struct.pack('BBB', 0, 1 + len(supl_fqdn), 1) + supl_fqdn.encode()
    public_id = binascii.hexlify(held + supl).decode()
    params = {"ssid": "gas/anqp",
              "interworking": "1",
              "anqp_elem": ["265:" + geo_loc,
                            "266:" + civic_loc,
                            "262:1122334455",
                            "267:" + public_id,
                            "279:01020304",
                            "60000:01",
                            "299:0102"]}
    hapd = hostapd.add_ap(apdev[0], params)
    bssid = apdev[0]['bssid']

    dev[0].flush_scan_cache()
    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
    if "OK" not in dev[0].request("ANQP_GET " + bssid + " 265,266"):
        raise Exception("ANQP_GET command failed")

    ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("GAS query timed out")

    bss = dev[0].get_bss(bssid)

    if 'anqp[265]' not in bss:
        raise Exception("AP Geospatial Location ANQP-element not seen")
    if bss['anqp[265]'] != geo_loc:
        raise Exception("Unexpected AP Geospatial Location ANQP-element value: " + bss['anqp[265]'])

    if 'anqp[266]' not in bss:
        raise Exception("AP Civic Location ANQP-element not seen")
    if bss['anqp[266]'] != civic_loc:
        raise Exception("Unexpected AP Civic Location ANQP-element value: " + bss['anqp[266]'])

    dev[1].scan_for_bss(bssid, freq="2412", force_scan=True)
    if "OK" not in dev[1].request("ANQP_GET " + bssid + " 257,258,259,260,261,262,263,264,265,267,268,269,270,271,272,273,274,275,276,277,278,279,280,281,282,283,284,285,286,287,288,289,290,291,292,293,294,295,296,297,298,299"):
        raise Exception("ANQP_GET command failed")

    ev = dev[1].wait_event(["GAS-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("GAS query timed out")

    bss = dev[1].get_bss(bssid)

    if 'anqp[265]' not in bss:
        raise Exception("AP Geospatial Location ANQP-element not seen")
    if bss['anqp[265]'] != geo_loc:
        raise Exception("Unexpected AP Geospatial Location ANQP-element value: " + bss['anqp[265]'])

    if 'anqp[266]' in bss:
        raise Exception("AP Civic Location ANQP-element unexpectedly seen")

    if 'anqp[267]' not in bss:
        raise Exception("AP Location Public Identifier ANQP-element not seen")
    if bss['anqp[267]'] != public_id:
        raise Exception("Unexpected AP Location Public Identifier ANQP-element value: " + bss['anqp[267]'])

    if 'anqp[279]' not in bss:
        raise Exception("ANQP-element Info ID 279 not seen")
    if bss['anqp[279]'] != "01020304":
        raise Exception("Unexpected AP ANQP-element Info ID 279 value: " + bss['anqp[279]'])

    if 'anqp[299]' not in bss:
        raise Exception("ANQP-element Info ID 299 not seen")
    if bss['anqp[299]'] != "0102":
        raise Exception("Unexpected AP ANQP-element Info ID 299 value: " + bss['anqp[299]'])

    if 'anqp_ip_addr_type_availability' not in bss:
        raise Exception("ANQP-element Info ID 292 not seen")
    if bss['anqp_ip_addr_type_availability'] != "1122334455":
        raise Exception("Unexpected AP ANQP-element Info ID 262 value: " + bss['anqp_ip_addr_type_availability'])

def test_gas_anqp_address3_not_assoc(dev, apdev, params):
    """GAS/ANQP query using IEEE 802.11 compliant Address 3 value when not associated"""
    try:
        _test_gas_anqp_address3_not_assoc(dev, apdev, params)
    finally:
        dev[0].request("SET gas_address3 0")

def _test_gas_anqp_address3_not_assoc(dev, apdev, params):
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']

    if "OK" not in dev[0].request("SET gas_address3 1"):
        raise Exception("Failed to set gas_address3")

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
    if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
        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(["ANQP-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("ANQP-QUERY-DONE event not seen")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected result: " + ev)

    out = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
                     "wlan_mgt.fixed.category_code == 4 && (wlan_mgt.fixed.publicact == 0x0a || wlan_mgt.fixed.publicact == 0x0b)",
                     display=["wlan.bssid"])
    res = out.splitlines()
    if len(res) != 2:
        raise Exception("Unexpected number of GAS frames")
    if res[0] != 'ff:ff:ff:ff:ff:ff':
        raise Exception("GAS request used unexpected Address3 field value: " + res[0])
    if res[1] != 'ff:ff:ff:ff:ff:ff':
        raise Exception("GAS response used unexpected Address3 field value: " + res[1])

def test_gas_anqp_address3_assoc(dev, apdev, params):
    """GAS/ANQP query using IEEE 802.11 compliant Address 3 value when associated"""
    try:
        _test_gas_anqp_address3_assoc(dev, apdev, params)
    finally:
        dev[0].request("SET gas_address3 0")

def _test_gas_anqp_address3_assoc(dev, apdev, params):
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']

    if "OK" not in dev[0].request("SET gas_address3 1"):
        raise Exception("Failed to set gas_address3")

    dev[0].scan_for_bss(bssid, freq="2412")
    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", scan_freq="2412")

    if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
        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(["ANQP-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("ANQP-QUERY-DONE event not seen")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected result: " + ev)

    out = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
                     "wlan_mgt.fixed.category_code == 4 && (wlan_mgt.fixed.publicact == 0x0a || wlan_mgt.fixed.publicact == 0x0b)",
                     display=["wlan.bssid"])
    res = out.splitlines()
    if len(res) != 2:
        raise Exception("Unexpected number of GAS frames")
    if res[0] != bssid:
        raise Exception("GAS request used unexpected Address3 field value: " + res[0])
    if res[1] != bssid:
        raise Exception("GAS response used unexpected Address3 field value: " + res[1])

def test_gas_anqp_address3_ap_forced(dev, apdev, params):
    """GAS/ANQP query using IEEE 802.11 compliant Address 3 value on AP"""
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']
    hapd.set("gas_address3", "1")

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
    if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
        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(["ANQP-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("ANQP-QUERY-DONE event not seen")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected result: " + ev)

    out = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
                     "wlan_mgt.fixed.category_code == 4 && (wlan_mgt.fixed.publicact == 0x0a || wlan_mgt.fixed.publicact == 0x0b)",
                     display=["wlan.bssid"])
    res = out.splitlines()
    if len(res) != 2:
        raise Exception("Unexpected number of GAS frames")
    if res[0] != bssid:
        raise Exception("GAS request used unexpected Address3 field value: " + res[0])
    if res[1] != 'ff:ff:ff:ff:ff:ff':
        raise Exception("GAS response used unexpected Address3 field value: " + res[1])

def test_gas_anqp_address3_ap_non_compliant(dev, apdev, params):
    """GAS/ANQP query using IEEE 802.11 non-compliant Address 3 (AP)"""
    try:
        _test_gas_anqp_address3_ap_non_compliant(dev, apdev, params)
    finally:
        dev[0].request("SET gas_address3 0")

def _test_gas_anqp_address3_ap_non_compliant(dev, apdev, params):
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']
    hapd.set("gas_address3", "2")

    if "OK" not in dev[0].request("SET gas_address3 1"):
        raise Exception("Failed to set gas_address3")

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
    if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
        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(["ANQP-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("ANQP-QUERY-DONE event not seen")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected result: " + ev)

    out = run_tshark(os.path.join(params['logdir'], "hwsim0.pcapng"),
                     "wlan_mgt.fixed.category_code == 4 && (wlan_mgt.fixed.publicact == 0x0a || wlan_mgt.fixed.publicact == 0x0b)",
                     display=["wlan.bssid"])
    res = out.splitlines()
    if len(res) != 2:
        raise Exception("Unexpected number of GAS frames")
    if res[0] != 'ff:ff:ff:ff:ff:ff':
        raise Exception("GAS request used unexpected Address3 field value: " + res[0])
    if res[1] != bssid:
        raise Exception("GAS response used unexpected Address3 field value: " + res[1])

def test_gas_anqp_address3_pmf(dev, apdev):
    """GAS/ANQP query using IEEE 802.11 compliant Address 3 value with PMF"""
    try:
        _test_gas_anqp_address3_pmf(dev, apdev)
    finally:
        dev[0].request("SET gas_address3 0")

def _test_gas_anqp_address3_pmf(dev, apdev):
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']
    hapd.set("gas_comeback_delay", "2")
    hapd.set("gas_address3", "1")

    if "OK" not in dev[0].request("SET gas_address3 1"):
        raise Exception("Failed to set gas_address3")

    dev[0].scan_for_bss(bssid, freq="2412")
    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", scan_freq="2412",
                   ieee80211w="2")

    if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
        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(["ANQP-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("ANQP-QUERY-DONE event not seen")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected result: " + ev)

    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")

def test_gas_prot_vs_not_prot(dev, apdev, params):
    """GAS/ANQP query protected vs. not protected"""
    hapd = start_ap(apdev[0])
    bssid = apdev[0]['bssid']

    dev[0].scan_for_bss(bssid, freq="2412")
    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", scan_freq="2412",
                   ieee80211w="2")

    if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
        raise Exception("ANQP_GET command failed")

    ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=5)
    if ev is None:
        raise Exception("No GAS-QUERY-DONE event")
    if "result=SUCCESS" not in ev:
        raise Exception("Unexpected GAS result: " + ev)

    # GAS: Drop unexpected unprotected GAS frame when PMF is enabled
    dev[0].request("SET ext_mgmt_frame_handling 1")
    res = dev[0].request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=d0003a010200000000000200000003000200000003001000040b00000005006c027f000000")
    dev[0].request("SET ext_mgmt_frame_handling 0")
    if "OK" not in res:
        raise Exception("MGMT_RX_PROCESS failed")

    dev[0].request("DISCONNECT")
    dev[0].wait_disconnected()

    # GAS: No pending query found for 02:00:00:00:03:00 dialog token 0
    dev[0].request("SET ext_mgmt_frame_handling 1")
    res = dev[0].request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=d0003a010200000000000200000003000200000003001000040b00000005006c027f000000")
    dev[0].request("SET ext_mgmt_frame_handling 0")
    if "OK" not in res:
        raise Exception("MGMT_RX_PROCESS failed")

    # GAS: Drop unexpected protected GAS frame when PMF is disabled
    dev[0].request("SET ext_mgmt_frame_handling 1")
    res = dev[0].request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=d0003a010200000000000200000003000200000003001000090b00000005006c027f000000")
    dev[0].request("SET ext_mgmt_frame_handling 0")
    if "OK" not in res:
        raise Exception("MGMT_RX_PROCESS failed")

def test_gas_failures(dev, apdev):
    """GAS failure cases"""
    hapd = start_ap(apdev[0])
    hapd.set("gas_comeback_delay", "5")
    bssid = apdev[0]['bssid']

    hapd2 = start_ap(apdev[1])
    bssid2 = apdev[1]['bssid']

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
    dev[0].scan_for_bss(bssid2, freq="2412")

    tests = [(bssid, "gas_build_req;gas_query_tx_comeback_req"),
             (bssid, "gas_query_tx;gas_query_tx_comeback_req"),
             (bssid, "gas_query_append;gas_query_rx_comeback"),
             (bssid2, "gas_query_append;gas_query_rx_initial"),
             (bssid2, "wpabuf_alloc_copy;gas_query_rx_initial"),
             (bssid, "gas_query_tx;gas_query_tx_initial_req")]
    for addr, func in tests:
        with alloc_fail(dev[0], 1, func):
            dev[0].request("ANQP_GET " + addr + " 258")
            ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=5)
            if ev is None:
                raise Exception("No GAS-QUERY-DONE seen")
            if "result=INTERNAL_ERROR" not in ev:
                raise Exception("Unexpected result code: " + ev)
        dev[0].dump_monitor()

    tests = ["=gas_query_req", "radio_add_work;gas_query_req"]
    for func in tests:
        with alloc_fail(dev[0], 1, func):
            if "FAIL" not in dev[0].request("ANQP_GET " + bssid + " 258"):
                raise Exception("ANQP_GET succeeded unexpectedly during OOM")
        dev[0].dump_monitor()

    wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
    wpas.interface_add("wlan5")
    wpas.scan_for_bss(bssid2, freq="2412")
    wpas.request("SET preassoc_mac_addr 1111")
    wpas.request("ANQP_GET " + bssid2 + " 258")
    ev = wpas.wait_event(["Failed to assign random MAC address for GAS"],
                         timeout=5)
    wpas.request("SET preassoc_mac_addr 0")
    if ev is None:
        raise Exception("No random MAC address error seen")

def test_gas_anqp_venue_url(dev, apdev):
    """GAS/ANQP and Venue URL"""
    venue_group = 1
    venue_type = 13
    venue_info = struct.pack('BB', venue_group, venue_type)
    lang1 = "eng"
    name1 = "Example venue"
    lang2 = "fin"
    name2 = "Esimerkkipaikka"
    venue1 = struct.pack('B', len(lang1 + name1)) + lang1.encode() + name1.encode()
    venue2 = struct.pack('B', len(lang2 + name2)) + lang2.encode() + name2.encode()
    venue_name = binascii.hexlify(venue_info + venue1 + venue2).decode()

    url1 = b"http://example.com/venue"
    url2 = b"https://example.org/venue-info/"
    duple1 = struct.pack('BB', 1 + len(url1), 1) + url1
    duple2 = struct.pack('BB', 1 + len(url2), 2) + url2
    venue_url = binascii.hexlify(duple1 + duple2).decode()

    params = {"ssid": "gas/anqp",
              "interworking": "1",
              "venue_group": str(venue_group),
              "venue_type": str(venue_type),
              "venue_name": [lang1 + ":" + name1, lang2 + ":" + name2],
              "anqp_elem": ["277:" + venue_url]}
    hapd = hostapd.add_ap(apdev[0], params)
    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 + " 257,258,277"):
        raise Exception("ANQP_GET command failed")

    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-VENUE-URL"], timeout=0.1)
    if ev is not None:
        raise Exception("Unexpected Venue URL indication without PMF")

    bss = dev[0].get_bss(bssid)

    if 'anqp_venue_name' not in bss:
        raise Exception("Venue Name ANQP-element not seen")
    if bss['anqp_venue_name'] != venue_name:
        raise Exception("Unexpected Venue Name ANQP-element value: " + bss['anqp_venue_name'])
    if 'anqp[277]' not in bss:
        raise Exception("Venue URL ANQP-element not seen")
    if bss['anqp[277]'] != venue_url:
        raise Exception("Unexpected Venue URL ANQP-element value: " + bss['anqp[277]'])

    if 'anqp_capability_list' not in bss:
        raise Exception("Capability List ANQP-element not seen")
    ids = struct.pack('<HHH', 257, 258, 277)
    if not bss['anqp_capability_list'].startswith(binascii.hexlify(ids).decode()):
        raise Exception("Unexpected Capability List ANQP-element value: " + bss['anqp_capability_list'])

def test_gas_anqp_venue_url2(dev, apdev):
    """GAS/ANQP and Venue URL (hostapd venue_url)"""
    venue_group = 1
    venue_type = 13
    venue_info = struct.pack('BB', venue_group, venue_type)
    lang1 = "eng"
    name1 = "Example venue"
    lang2 = "fin"
    name2 = "Esimerkkipaikka"
    venue1 = struct.pack('B', len(lang1 + name1)) + lang1.encode() + name1.encode()
    venue2 = struct.pack('B', len(lang2 + name2)) + lang2.encode() + name2.encode()
    venue_name = binascii.hexlify(venue_info + venue1 + venue2).decode()

    url1 = "http://example.com/venue"
    url2 = "https://example.org/venue-info/"
    duple1 = struct.pack('BB', 1 + len(url1.encode()), 1) + url1.encode()
    duple2 = struct.pack('BB', 1 + len(url2.encode()), 2) + url2.encode()
    venue_url = binascii.hexlify(duple1 + duple2).decode()

    params = {"ssid": "gas/anqp",
              "interworking": "1",
              "venue_group": str(venue_group),
              "venue_type": str(venue_type),
              "venue_name": [lang1 + ":" + name1, lang2 + ":" + name2],
              "venue_url": ["1:" + url1, "2:" + url2]}
    hapd = hostapd.add_ap(apdev[0], params)
    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 + " 257,258,277"):
        raise Exception("ANQP_GET command failed")

    ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("GAS query timed out")

    bss = dev[0].get_bss(bssid)

    if 'anqp_venue_name' not in bss:
        raise Exception("Venue Name ANQP-element not seen")
    if bss['anqp_venue_name'] != venue_name:
        raise Exception("Unexpected Venue Name ANQP-element value: " + bss['anqp_venue_name'])
    if 'anqp[277]' not in bss:
        raise Exception("Venue URL ANQP-element not seen")
    if bss['anqp[277]'] != venue_url:
        print(venue_url)
        raise Exception("Unexpected Venue URL ANQP-element value: " + bss['anqp[277]'])

    if 'anqp_capability_list' not in bss:
        raise Exception("Capability List ANQP-element not seen")
    ids = struct.pack('<HHH', 257, 258, 277)
    if not bss['anqp_capability_list'].startswith(binascii.hexlify(ids).decode()):
        raise Exception("Unexpected Capability List ANQP-element value: " + bss['anqp_capability_list'])

def test_gas_anqp_venue_url_pmf(dev, apdev):
    """GAS/ANQP and Venue URL with PMF"""
    venue_group = 1
    venue_type = 13
    venue_info = struct.pack('BB', venue_group, venue_type)
    lang1 = "eng"
    name1 = "Example venue"
    lang2 = "fin"
    name2 = "Esimerkkipaikka"
    venue1 = struct.pack('B', len(lang1 + name1)) + lang1.encode() + name1.encode()
    venue2 = struct.pack('B', len(lang2 + name2)) + lang2.encode() + name2.encode()
    venue_name = binascii.hexlify(venue_info + venue1 + venue2)

    url1 = "http://example.com/venue"
    url2 = "https://example.org/venue-info/"

    params = {"ssid": "gas/anqp/pmf",
              "wpa": "2",
              "wpa_key_mgmt": "WPA-PSK",
              "rsn_pairwise": "CCMP",
              "wpa_passphrase": "12345678",
              "ieee80211w": "2",
              "interworking": "1",
              "venue_group": str(venue_group),
              "venue_type": str(venue_type),
              "venue_name": [lang1 + ":" + name1, lang2 + ":" + name2],
              "venue_url": ["1:" + url1, "2:" + url2]}
    hapd = hostapd.add_ap(apdev[0], params)
    bssid = apdev[0]['bssid']

    dev[0].connect("gas/anqp/pmf", psk="12345678", ieee80211w="2",
                   scan_freq="2412")
    if "OK" not in dev[0].request("ANQP_GET " + bssid + " 277"):
        raise Exception("ANQP_GET command failed")

    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-VENUE-URL"], timeout=5)
    if ev is None:
        raise Exception("No Venue URL indication seen")
    if "1 " + url1 not in ev:
        raise Exception("Unexpected Venue URL information: " + ev)

    ev = dev[0].wait_event(["RX-VENUE-URL"], timeout=5)
    if ev is None:
        raise Exception("No Venue URL indication seen (2)")
    if "2 " + url2 not in ev:
        raise Exception("Unexpected Venue URL information (2): " + ev)

def test_gas_anqp_capab_list(dev, apdev):
    """GAS/ANQP and Capability List ANQP-element"""
    params = {"ssid": "gas/anqp",
              "interworking": "1"}
    params["anqp_elem"] = []
    for i in range(0, 400):
        if i not in [257]:
            params["anqp_elem"] += ["%d:010203" % i]
    hapd = hostapd.add_ap(apdev[0], params)
    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 + " 257"):
        raise Exception("ANQP_GET command failed")

    ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("GAS query timed out")

    bss = dev[0].get_bss(bssid)

    if 'anqp_capability_list' not in bss:
        raise Exception("Capability List ANQP-element not seen")
    val = bss['anqp_capability_list']
    logger.info("anqp_capability_list: " + val)
    ids = []
    while len(val) >= 4:
        id_bin = binascii.unhexlify(val[0:4])
        id = struct.unpack('<H', id_bin)[0]
        if id == 0xdddd:
            break
        ids.append(id)
        val = val[4:]
    logger.info("InfoIDs: " + str(ids))
    for i in range(257, 300):
        if i in [273, 274]:
            continue
        if i not in ids:
            raise Exception("Unexpected Capability List ANQP-element value (missing %d): %s" % (i, bss['anqp_capability_list']))

def test_gas_server_oom(dev, apdev):
    """GAS server OOM"""
    bssid = apdev[0]['bssid']
    params = hs20_ap_params()
    params['hessid'] = bssid
    params['gas_comeback_delay'] = "5"
    hapd = hostapd.add_ap(apdev[0], params)

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)

    tests = ["ap_sta_add;gas_dialog_create",
             "=gas_dialog_create",
             "wpabuf_alloc_copy;gas_serv_rx_gas_comeback_req"]
    for t in tests:
        with alloc_fail(hapd, 1, t):
            if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
                raise Exception("ANQP_GET command failed")
            ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=5)
            if ev is None:
                raise Exception("No GAS-QUERY-DONE seen")
            dev[0].dump_monitor()

    hapd.set("gas_comeback_delay", "0")

    tests = ["gas_serv_build_gas_resp_payload"]
    for t in tests:
        with alloc_fail(hapd, 1, t):
            if "OK" not in dev[0].request("ANQP_GET " + bssid + " 258"):
                raise Exception("ANQP_GET command failed")
            ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=5)
            if ev is None:
                raise Exception("No GAS-QUERY-DONE seen")
            dev[0].dump_monitor()

    with alloc_fail(hapd, 1,
                    "gas_build_initial_resp;gas_serv_rx_gas_initial_req"):
        req = dev[0].request("GAS_REQUEST " + bssid + " 42 000102000101")
        if "FAIL" in req:
            raise Exception("GAS query request rejected")
        ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=5)
        if ev is None:
            raise Exception("No GAS-QUERY-DONE seen")
        dev[0].dump_monitor()

    wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
    wpas.interface_add("wlan5")
    if "OK" not in wpas.request("P2P_SET listen_channel 1"):
        raise Exception("Failed to set listen channel")
    if "OK" not in wpas.p2p_listen():
        raise Exception("Failed to start listen state")
    if "FAIL" in wpas.request("SET ext_mgmt_frame_handling 1"):
        raise Exception("Failed to enable external management frame handling")

    msg = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_COMEBACK_REQUEST, 1)
    req = "MGMT_TX {} {} freq=2412 wait_time=10 action={}".format(bssid, bssid, binascii.hexlify(msg).decode())
    with alloc_fail(hapd, 1,
                    "gas_anqp_build_comeback_resp_buf;gas_serv_rx_gas_comeback_req"):
        if "OK" not in wpas.request(req):
            raise Exception("Could not send management frame")
        wait_fail_trigger(hapd, "GET_ALLOC_FAIL")

def test_gas_anqp_overrides(dev, apdev):
    """GAS and ANQP overrides"""
    params = {"ssid": "gas/anqp",
              "interworking": "1",
              "anqp_elem": ["257:111111",
                            "258:222222",
                            "260:333333",
                            "261:444444",
                            "262:555555",
                            "263:666666",
                            "264:777777",
                            "268:888888",
                            "275:999999"]}
    hapd = hostapd.add_ap(apdev[0], params)
    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 + " 257,258,260,261,262,263,264,268,275"):
        raise Exception("ANQP_GET command failed")

    ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=10)
    if ev is None:
        raise Exception("GAS query timed out")
    elems = 9
    capa = dev[0].get_capability("fils")
    if capa is None or "FILS" not in capa:
        # FILS Realm Info not supported in the build
        elems -= 1
    for i in range(elems):
        ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
        if ev is None:
            raise Exception("ANQP response not seen")

def test_gas_no_dialog_token_match(dev, apdev):
    """GAS and no dialog token match for comeback request"""
    hapd = start_ap(apdev[0])
    hapd.set("gas_frag_limit", "50")
    bssid = apdev[0]['bssid']

    wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
    wpas.interface_add("wlan5")
    if "OK" not in wpas.request("P2P_SET listen_channel 1"):
        raise Exception("Failed to set listen channel")
    if "OK" not in wpas.p2p_listen():
        raise Exception("Failed to start listen state")
    if "FAIL" in wpas.request("SET ext_mgmt_frame_handling 1"):
        raise Exception("Failed to enable external management frame handling")

    anqp_query = struct.pack('<HHHHHHHHHH', 256, 16, 257, 258, 260, 261, 262, 263, 264, 268)
    gas = struct.pack('<H', len(anqp_query)) + anqp_query

    dialog_token = 100
    msg = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_INITIAL_REQUEST,
                      dialog_token) + anqp_adv_proto() + gas
    req = "MGMT_TX {} {} freq=2412 wait_time=10 action={}".format(bssid, bssid, binascii.hexlify(msg).decode())
    if "OK" not in wpas.request(req):
        raise Exception("Could not send management frame")
    resp = wpas.mgmt_rx()
    if resp is None:
        raise Exception("MGMT-RX timeout")
    if 'payload' not in resp:
        raise Exception("Missing payload")
    gresp = parse_gas(resp['payload'])
    if gresp['dialog_token'] != dialog_token:
        raise Exception("Dialog token mismatch")
    status_code = gresp['status_code']
    if status_code != 0:
        raise Exception("Unexpected status code {}".format(status_code))

    msg = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_COMEBACK_REQUEST,
                      dialog_token + 1)
    req = "MGMT_TX {} {} freq=2412 wait_time=10 action={}".format(bssid, bssid, binascii.hexlify(msg).decode())
    if "OK" not in wpas.request(req):
        raise Exception("Could not send management frame")
    resp = wpas.mgmt_rx()
    if resp is None:
        raise Exception("MGMT-RX timeout")
    if 'payload' not in resp:
        raise Exception("Missing payload")
    gresp = parse_gas(resp['payload'])
    status_code = gresp['status_code']
    if status_code != 60:
        raise Exception("Unexpected failure status code {}".format(status_code))

def test_gas_vendor_spec_errors(dev, apdev):
    """GAS and vendor specific request error cases"""
    bssid = apdev[0]['bssid']
    params = hs20_ap_params()
    params['hessid'] = bssid
    params['osu_server_uri'] = "uri"
    params['hs20_icon'] = "32:32:eng:image/png:icon32:/tmp/icon32.png"
    del params['nai_realm']
    hapd = hostapd.add_ap(apdev[0], params)

    dev[0].scan_for_bss(bssid, freq="2412", force_scan=True)
    tests = ["00 12340000",
             "00 dddd0300506fff",
             "00 dddd0400506fffff",
             "00 dddd0400506f9aff",
             "00 dddd0400506f9a11",
             "00 dddd0600506f9a11ff00",
             "00 dddd0600506f9a110600",
             "00 dddd0600506f9a110600",
             "00 dddd0700506f9a11060000",
             "00 dddd0700506f9a110600ff",
             "00 dddd0800506f9a110600ff00",
             "00 dddd0900506f9a110600ff0000",
             "00 dddd0900506f9a110600ff0001",
             "00 dddd0900506f9a110600ffff00",
             "00 dddd0a00506f9a110600ff00013b",
             "00 dddd0700506f9a110100ff",
             "00 dddd0700506f9a11010008",
             "00 dddd14",
             "00 dddd1400506f9a11"]
    for t in tests:
        req = dev[0].request("GAS_REQUEST " + bssid + " " + t)
        if "FAIL" in req:
            raise Exception("GAS query request rejected")
        ev = dev[0].wait_event(["GAS-QUERY-START"], timeout=5)
        if ev is None:
            raise Exception("GAS query did not start")
        ev = dev[0].wait_event(["GAS-QUERY-DONE"], timeout=5)
        if ev is None:
            raise Exception("GAS query did not complete")
        if t == "00 dddd0600506f9a110600":
            hapd.set("nai_realm", "0,another.example.com")

    wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
    wpas.interface_add("wlan5")
    if "OK" not in wpas.request("P2P_SET listen_channel 1"):
        raise Exception("Failed to set listen channel")
    if "OK" not in wpas.p2p_listen():
        raise Exception("Failed to start listen state")
    if "FAIL" in wpas.request("SET ext_mgmt_frame_handling 1"):
        raise Exception("Failed to enable external management frame handling")

    anqp_query = struct.pack('<HHHHHHHHHH', 256, 16, 257, 258, 260, 261, 262, 263, 264, 268)
    gas = struct.pack('<H', len(anqp_query)) + anqp_query

    dialog_token = 100
    adv = struct.pack('BBBB', 109, 2, 0, 0)
    adv2 = struct.pack('BBB', 108, 1, 0)
    adv3 = struct.pack('BBBB', 108, 3, 0, 0)
    msg = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_INITIAL_REQUEST,
                      dialog_token) + adv + gas
    msg2 = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_INITIAL_REQUEST,
                       dialog_token) + adv2 + gas
    msg3 = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_INITIAL_REQUEST,
                       dialog_token) + adv3
    msg4 = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_INITIAL_REQUEST,
                       dialog_token) + anqp_adv_proto()
    msg5 = struct.pack('<BBB', ACTION_CATEG_PUBLIC, GAS_INITIAL_REQUEST,
                       dialog_token) + anqp_adv_proto() + struct.pack('<H', 1)
    msg6 = struct.pack('<BB', ACTION_CATEG_PUBLIC, GAS_COMEBACK_REQUEST)
    tests = [msg, msg2, msg3, msg4, msg5, msg6]
    for t in tests:
        req = "MGMT_TX {} {} freq=2412 wait_time=10 action={}".format(bssid, bssid, binascii.hexlify(t).decode())
        if "OK" not in wpas.request(req):
            raise Exception("Could not send management frame")
        ev = wpas.wait_event(["MGMT-TX-STATUS"], timeout=5)
        if ev is None:
            raise Exception("No ACK frame seen")