tests/hwsim/test_gas.py

   1 #!/usr/bin/python
   2 #
   3 # GAS tests
   4 # Copyright (c) 2013, Qualcomm Atheros, Inc.
   5 #
   6 # This software may be distributed under the terms of the BSD license.
   7 # See README for more details.
   8
   9 import time
  10 import binascii
  11 import logging
  12 logger = logging.getLogger()
  13 import re
  14 import struct
  15
  16 import hostapd
  17
  18 def hs20_ap_params():
  19     params = hostapd.wpa2_params(ssid="test-gas")
  20     params['wpa_key_mgmt'] = "WPA-EAP"
  21     params['ieee80211w'] = "1"
  22     params['ieee8021x'] = "1"
  23     params['auth_server_addr'] = "127.0.0.1"
  24     params['auth_server_port'] = "1812"
  25     params['auth_server_shared_secret'] = "radius"
  26     params['interworking'] = "1"
  27     params['access_network_type'] = "14"
  28     params['internet'] = "1"
  29     params['asra'] = "0"
  30     params['esr'] = "0"
  31     params['uesa'] = "0"
  32     params['venue_group'] = "7"
  33     params['venue_type'] = "1"
  34     params['venue_name'] = [ "eng:Example venue", "fin:Esimerkkipaikka" ]
  35     params['roaming_consortium'] = [ "112233", "1020304050", "010203040506",
  36                                      "fedcba" ]
  37     params['domain_name'] = "example.com,another.example.com"
  38     params['nai_realm'] = [ "0,example.com,13[5:6],21[2:4][5:7]",
  39                             "0,another.example.com" ]
  40     params['anqp_3gpp_cell_net'] = "244,91"
  41     return params
  42
  43 def start_ap(ap):
  44     params = hs20_ap_params()
  45     params['hessid'] = ap['bssid']
  46     hostapd.add_ap(ap['ifname'], params)
  47     return hostapd.Hostapd(ap['ifname'])
  48
  49 def get_gas_response(dev, bssid, info, allow_fetch_failure=False):
  50     exp = r'<.>(GAS-RESPONSE-INFO) addr=([0-9a-f:]*) dialog_token=([0-9]*) status_code=([0-9]*) resp_len=([\-0-9]*)'
  51     res = re.split(exp, info)
  52     if len(res) < 6:
  53         raise Exception("Could not parse GAS-RESPONSE-INFO")
  54     if res[2] != bssid:
  55         raise Exception("Unexpected BSSID in response")
  56     token = res[3]
  57     status = res[4]
  58     if status != "0":
  59         raise Exception("GAS query failed")
  60     resp_len = res[5]
  61     if resp_len == "-1":
  62         raise Exception("GAS query reported invalid response length")
  63     if int(resp_len) > 2000:
  64         raise Exception("Unexpected long GAS response")
  65
  66     resp = dev.request("GAS_RESPONSE_GET " + bssid + " " + token)
  67     if "FAIL" in resp:
  68         if allow_fetch_failure:
  69             logger.debug("GAS response was not available anymore")
  70             return
  71         raise Exception("Could not fetch GAS response")
  72     if len(resp) != int(resp_len) * 2:
  73         raise Exception("Unexpected GAS response length")
  74     logger.debug("GAS response: " + resp)
  75
  76 def test_gas_generic(dev, apdev):
  77     """Generic GAS query"""
  78     bssid = apdev[0]['bssid']
  79     params = hs20_ap_params()
  80     params['hessid'] = bssid
  81     hostapd.add_ap(apdev[0]['ifname'], params)
  82
  83     dev[0].scan()
  84     req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000101")
  85     if "FAIL" in req:
  86         raise Exception("GAS query request rejected")
  87     ev = dev[0].wait_event(["GAS-RESPONSE-INFO"], timeout=10)
  88     if ev is None:
  89         raise Exception("GAS query timed out")
  90     get_gas_response(dev[0], bssid, ev)
  91
  92 def test_gas_concurrent_scan(dev, apdev):
  93     """Generic GAS queries with concurrent scan operation"""
  94     bssid = apdev[0]['bssid']
  95     params = hs20_ap_params()
  96     params['hessid'] = bssid
  97     hostapd.add_ap(apdev[0]['ifname'], params)
  98
  99     dev[0].scan()
 100
 101     logger.info("Request concurrent operations")
 102     req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000101")
 103     if "FAIL" in req:
 104         raise Exception("GAS query request rejected")
 105     req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000801")
 106     if "FAIL" in req:
 107         raise Exception("GAS query request rejected")
 108     dev[0].request("SCAN")
 109     req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000201")
 110     if "FAIL" in req:
 111         raise Exception("GAS query request rejected")
 112     req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000501")
 113     if "FAIL" in req:
 114         raise Exception("GAS query request rejected")
 115
 116     responses = 0
 117     for i in range(0, 5):
 118         ev = dev[0].wait_event(["GAS-RESPONSE-INFO", "CTRL-EVENT-SCAN-RESULTS"],
 119                                timeout=10)
 120         if ev is None:
 121             raise Exception("Operation timed out")
 122         if "GAS-RESPONSE-INFO" in ev:
 123             responses = responses + 1
 124             get_gas_response(dev[0], bssid, ev, allow_fetch_failure=True)
 125
 126     if responses != 4:
 127         raise Exception("Unexpected number of GAS responses")
 128
 129 def test_gas_concurrent_connect(dev, apdev):
 130     """Generic GAS queries with concurrent connection operation"""
 131     bssid = apdev[0]['bssid']
 132     params = hs20_ap_params()
 133     params['hessid'] = bssid
 134     hostapd.add_ap(apdev[0]['ifname'], params)
 135
 136     dev[0].scan()
 137
 138     logger.debug("Start concurrent connect and GAS request")
 139     dev[0].connect("test-gas", key_mgmt="WPA-EAP", eap="TTLS",
 140                    identity="DOMAIN\mschapv2 user", anonymous_identity="ttls",
 141                    password="password", phase2="auth=MSCHAPV2",
 142                    ca_cert="auth_serv/ca.pem", wait_connect=False)
 143     req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000101")
 144     if "FAIL" in req:
 145         raise Exception("GAS query request rejected")
 146
 147     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED", "GAS-RESPONSE-INFO"],
 148                            timeout=20)
 149     if ev is None:
 150         raise Exception("Operation timed out")
 151     if "CTRL-EVENT-CONNECTED" not in ev:
 152         raise Exception("Unexpected operation order")
 153
 154     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED", "GAS-RESPONSE-INFO"],
 155                            timeout=20)
 156     if ev is None:
 157         raise Exception("Operation timed out")
 158     if "GAS-RESPONSE-INFO" not in ev:
 159         raise Exception("Unexpected operation order")
 160     get_gas_response(dev[0], bssid, ev)
 161
 162     dev[0].request("DISCONNECT")
 163     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
 164     if ev is None:
 165         raise Exception("Disconnection timed out")
 166
 167     logger.debug("Wait six seconds for expiration of connect-without-scan")
 168     time.sleep(6)
 169
 170     logger.debug("Start concurrent GAS request and connect")
 171     req = dev[0].request("GAS_REQUEST " + bssid + " 00 000102000101")
 172     if "FAIL" in req:
 173         raise Exception("GAS query request rejected")
 174     dev[0].request("RECONNECT")
 175
 176     ev = dev[0].wait_event(["GAS-RESPONSE-INFO"], timeout=10)
 177     if ev is None:
 178         raise Exception("Operation timed out")
 179     get_gas_response(dev[0], bssid, ev)
 180
 181     ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"], timeout=20)
 182     if ev is None:
 183         raise Exception("No new scan results reported")
 184
 185     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
 186     if ev is None:
 187         raise Exception("Operation timed out")
 188     if "CTRL-EVENT-CONNECTED" not in ev:
 189         raise Exception("Unexpected operation order")
 190
 191 def test_gas_fragment(dev, apdev):
 192     """GAS fragmentation"""
 193     hapd = start_ap(apdev[0])
 194     hapd.set("gas_frag_limit", "50")
 195
 196     dev[0].scan(freq="2412")
 197     dev[0].request("FETCH_ANQP")
 198     for i in range(0, 6):
 199         ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
 200         if ev is None:
 201             raise Exception("Operation timed out")
 202
 203 def test_gas_comeback_delay(dev, apdev):
 204     """GAS fragmentation"""
 205     hapd = start_ap(apdev[0])
 206     hapd.set("gas_comeback_delay", "500")
 207
 208     dev[0].scan(freq="2412")
 209     dev[0].request("FETCH_ANQP")
 210     for i in range(0, 6):
 211         ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
 212         if ev is None:
 213             raise Exception("Operation timed out")
 214
 215 def expect_gas_result(dev, result):
 216     ev = dev.wait_event(["GAS-QUERY-DONE"], timeout=10)
 217     if ev is None:
 218         raise Exception("GAS query timed out")
 219     if "result=" + result not in ev:
 220         raise Exception("Unexpected GAS query result")
 221
 222 def anqp_get(dev, bssid, id):
 223     dev.request("ANQP_GET " + bssid + " " + str(id))
 224     ev = dev.wait_event(["GAS-QUERY-START"], timeout=5)
 225     if ev is None:
 226         raise Exception("GAS query start timed out")
 227
 228 def test_gas_timeout(dev, apdev):
 229     """GAS timeout"""
 230     hapd = start_ap(apdev[0])
 231     bssid = apdev[0]['bssid']
 232
 233     dev[0].scan(freq="2412")
 234     hapd.set("ext_mgmt_frame_handling", "1")
 235
 236     anqp_get(dev[0], bssid, 263)
 237
 238     ev = hapd.wait_event(["MGMT-RX"], timeout=5)
 239     if ev is None:
 240         raise Exception("MGMT RX wait timed out")
 241
 242     expect_gas_result(dev[0], "TIMEOUT")
 243
 244 MGMT_SUBTYPE_ACTION = 13
 245 ACTION_CATEG_PUBLIC = 4
 246
 247 GAS_INITIAL_REQUEST = 10
 248 GAS_INITIAL_RESPONSE = 11
 249 GAS_COMEBACK_REQUEST = 12
 250 GAS_COMEBACK_RESPONSE = 13
 251 GAS_ACTIONS = [ GAS_INITIAL_REQUEST, GAS_INITIAL_RESPONSE,
 252                 GAS_COMEBACK_REQUEST, GAS_COMEBACK_RESPONSE ]
 253
 254 def anqp_adv_proto():
 255     return struct.pack('BBBB', 108, 2, 127, 0)
 256
 257 def anqp_initial_resp(dialog_token, status_code):
 258     return struct.pack('<BBBHH', ACTION_CATEG_PUBLIC, GAS_INITIAL_RESPONSE,
 259                        dialog_token, status_code, 0) + anqp_adv_proto()
 260
 261 def anqp_comeback_resp(dialog_token):
 262     return struct.pack('<BBBHBH', ACTION_CATEG_PUBLIC, GAS_COMEBACK_RESPONSE,
 263                        dialog_token, 0, 0, 0) + anqp_adv_proto()
 264
 265 def gas_rx(hapd):
 266     count = 0
 267     while count < 30:
 268         count = count + 1
 269         query = hapd.mgmt_rx()
 270         if query is None:
 271             raise Exception("Action frame not received")
 272         if query['subtype'] != MGMT_SUBTYPE_ACTION:
 273             continue
 274         payload = query['payload']
 275         if len(payload) < 2:
 276             continue
 277         (category, action) = struct.unpack('BB', payload[0:2])
 278         if category != ACTION_CATEG_PUBLIC or action not in GAS_ACTIONS:
 279             continue
 280         return query
 281     raise Exception("No Action frame received")
 282
 283 def parse_gas(payload):
 284     pos = payload
 285     (category, action, dialog_token) = struct.unpack('BBB', pos[0:3])
 286     if category != ACTION_CATEG_PUBLIC:
 287         return None
 288     if action not in GAS_ACTIONS:
 289         return None
 290     gas = {}
 291     gas['action'] = action
 292     pos = pos[3:]
 293
 294     if len(pos) < 1:
 295         return None
 296
 297     gas['dialog_token'] = dialog_token
 298     return gas
 299
 300 def action_response(req):
 301     resp = {}
 302     resp['fc'] = req['fc']
 303     resp['da'] = req['sa']
 304     resp['sa'] = req['da']
 305     resp['bssid'] = req['bssid']
 306     return resp
 307
 308 def test_gas_invalid_response_type(dev, apdev):
 309     """GAS invalid response type"""
 310     hapd = start_ap(apdev[0])
 311     bssid = apdev[0]['bssid']
 312
 313     dev[0].scan(freq="2412")
 314     hapd.set("ext_mgmt_frame_handling", "1")
 315
 316     anqp_get(dev[0], bssid, 263)
 317
 318     query = gas_rx(hapd)
 319     gas = parse_gas(query['payload'])
 320
 321     resp = action_response(query)
 322     # GAS Comeback Response instead of GAS Initial Response
 323     resp['payload'] = anqp_comeback_resp(gas['dialog_token']) + struct.pack('<H', 0)
 324     hapd.mgmt_tx(resp)
 325     ev = hapd.wait_event(["MGMT-TX-STATUS"], timeout=5)
 326     if ev is None:
 327         raise Exception("Missing TX status for GAS response")
 328     if "ok=1" not in ev:
 329         raise Exception("GAS response not acknowledged")
 330
 331     # station drops the invalid frame, so this needs to result in GAS timeout
 332     expect_gas_result(dev[0], "TIMEOUT")
 333
 334 def test_gas_failure_status_code(dev, apdev):
 335     """GAS failure status code"""
 336     hapd = start_ap(apdev[0])
 337     bssid = apdev[0]['bssid']
 338
 339     dev[0].scan(freq="2412")
 340     hapd.set("ext_mgmt_frame_handling", "1")
 341
 342     anqp_get(dev[0], bssid, 263)
 343
 344     query = gas_rx(hapd)
 345     gas = parse_gas(query['payload'])
 346
 347     resp = action_response(query)
 348     resp['payload'] = anqp_initial_resp(gas['dialog_token'], 61) + struct.pack('<H', 0)
 349     hapd.mgmt_tx(resp)
 350     ev = hapd.wait_event(["MGMT-TX-STATUS"], timeout=5)
 351     if ev is None:
 352         raise Exception("Missing TX status for GAS response")
 353     if "ok=1" not in ev:
 354         raise Exception("GAS response not acknowledged")
 355
 356     expect_gas_result(dev[0], "FAILURE")