tests/hwsim/test_ap_hs20.py

   1 # Hotspot 2.0 tests
   2 # Copyright (c) 2013-2014, Jouni Malinen <j@w1.fi>
   3 #
   4 # This software may be distributed under the terms of the BSD license.
   5 # See README for more details.
   6
   7 import time
   8 import subprocess
   9 import logging
  10 logger = logging.getLogger()
  11 import os
  12 import os.path
  13 import subprocess
  14
  15 import hostapd
  16 from wlantest import Wlantest
  17
  18 def hs20_ap_params(ssid="test-hs20"):
  19     params = hostapd.wpa2_params(ssid=ssid)
  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['hs20'] = "1"
  41     params['hs20_wan_metrics'] = "01:8000:1000:80:240:3000"
  42     params['hs20_conn_capab'] = [ "1:0:2", "6:22:1", "17:5060:0" ]
  43     params['hs20_operating_class'] = "5173"
  44     params['anqp_3gpp_cell_net'] = "244,91"
  45     return params
  46
  47 def check_auto_select(dev, bssid):
  48     dev.request("INTERWORKING_SELECT auto freq=2412")
  49     ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
  50     if ev is None:
  51         raise Exception("Connection timed out")
  52     if bssid not in ev:
  53         raise Exception("Connected to incorrect network")
  54     dev.request("REMOVE_NETWORK all")
  55
  56 def interworking_select(dev, bssid, type=None, no_match=False, freq=None):
  57     dev.dump_monitor()
  58     freq_extra = " freq=" + freq if freq else ""
  59     dev.request("INTERWORKING_SELECT" + freq_extra)
  60     ev = dev.wait_event(["INTERWORKING-AP", "INTERWORKING-NO-MATCH"],
  61                         timeout=15)
  62     if ev is None:
  63         raise Exception("Network selection timed out");
  64     if no_match:
  65         if "INTERWORKING-NO-MATCH" not in ev:
  66             raise Exception("Unexpected network match")
  67         return
  68     if "INTERWORKING-NO-MATCH" in ev:
  69         raise Exception("Matching network not found")
  70     if bssid and bssid not in ev:
  71         raise Exception("Unexpected BSSID in match")
  72     if type and "type=" + type not in ev:
  73         raise Exception("Network type not recognized correctly")
  74
  75 def check_sp_type(dev, sp_type):
  76     type = dev.get_status_field("sp_type")
  77     if type is None:
  78         raise Exception("sp_type not available")
  79     if type != sp_type:
  80         raise Exception("sp_type did not indicate home network")
  81
  82 def hlr_auc_gw_available():
  83     if not os.path.exists("/tmp/hlr_auc_gw.sock"):
  84         logger.info("No hlr_auc_gw available");
  85         return False
  86     if not os.path.exists("../../hostapd/hlr_auc_gw"):
  87         logger.info("No hlr_auc_gw available");
  88         return False
  89     return True
  90
  91 def interworking_ext_sim_connect(dev, bssid, method):
  92     dev.request("INTERWORKING_CONNECT " + bssid)
  93     interworking_ext_sim_auth(dev, method)
  94
  95 def interworking_ext_sim_auth(dev, method):
  96     ev = dev.wait_event(["CTRL-EVENT-EAP-METHOD"], timeout=15)
  97     if ev is None:
  98         raise Exception("Network connected timed out")
  99     if "(" + method + ")" not in ev:
 100         raise Exception("Unexpected EAP method selection")
 101
 102     ev = dev.wait_event(["CTRL-REQ-SIM"], timeout=15)
 103     if ev is None:
 104         raise Exception("Wait for external SIM processing request timed out")
 105     p = ev.split(':', 2)
 106     if p[1] != "GSM-AUTH":
 107         raise Exception("Unexpected CTRL-REQ-SIM type")
 108     id = p[0].split('-')[3]
 109     rand = p[2].split(' ')[0]
 110
 111     res = subprocess.check_output(["../../hostapd/hlr_auc_gw",
 112                                    "-m",
 113                                    "auth_serv/hlr_auc_gw.milenage_db",
 114                                    "GSM-AUTH-REQ 232010000000000 " + rand])
 115     if "GSM-AUTH-RESP" not in res:
 116         raise Exception("Unexpected hlr_auc_gw response")
 117     resp = res.split(' ')[2].rstrip()
 118
 119     dev.request("CTRL-RSP-SIM-" + id + ":GSM-AUTH:" + resp)
 120     ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
 121     if ev is None:
 122         raise Exception("Connection timed out")
 123
 124 def interworking_connect(dev, bssid, method):
 125     dev.request("INTERWORKING_CONNECT " + bssid)
 126     interworking_auth(dev, method)
 127
 128 def interworking_auth(dev, method):
 129     ev = dev.wait_event(["CTRL-EVENT-EAP-METHOD"], timeout=15)
 130     if ev is None:
 131         raise Exception("Network connected timed out")
 132     if "(" + method + ")" not in ev:
 133         raise Exception("Unexpected EAP method selection")
 134
 135     ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
 136     if ev is None:
 137         raise Exception("Connection timed out")
 138
 139 def check_probe_resp(wt, bssid_unexpected, bssid_expected):
 140     if bssid_unexpected:
 141         count = wt.get_bss_counter("probe_response", bssid_unexpected)
 142         if count > 0:
 143             raise Exception("Unexpected Probe Response frame from AP")
 144
 145     if bssid_expected:
 146         count = wt.get_bss_counter("probe_response", bssid_expected)
 147         if count == 0:
 148             raise Exception("No Probe Response frame from AP")
 149
 150 def test_ap_anqp_sharing(dev, apdev):
 151     """ANQP sharing within ESS and explicit unshare"""
 152     bssid = apdev[0]['bssid']
 153     params = hs20_ap_params()
 154     params['hessid'] = bssid
 155     hostapd.add_ap(apdev[0]['ifname'], params)
 156
 157     bssid2 = apdev[1]['bssid']
 158     params = hs20_ap_params()
 159     params['hessid'] = bssid
 160     params['nai_realm'] = [ "0,example.com,13[5:6],21[2:4][5:7]" ]
 161     hostapd.add_ap(apdev[1]['ifname'], params)
 162
 163     dev[0].hs20_enable()
 164     id = dev[0].add_cred_values({ 'realm': "example.com", 'username': "test",
 165                                   'password': "secret",
 166                                   'domain': "example.com" })
 167     logger.info("Normal network selection with shared ANQP results")
 168     interworking_select(dev[0], None, "home", freq="2412")
 169     dev[0].dump_monitor()
 170
 171     res1 = dev[0].get_bss(bssid)
 172     res2 = dev[0].get_bss(bssid2)
 173     if res1['anqp_nai_realm'] != res2['anqp_nai_realm']:
 174         raise Exception("ANQP results were not shared between BSSes")
 175
 176     logger.info("Explicit ANQP request to unshare ANQP results")
 177     dev[0].request("ANQP_GET " + bssid + " 263")
 178     ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
 179     if ev is None:
 180         raise Exception("ANQP operation timed out")
 181
 182     dev[0].request("ANQP_GET " + bssid2 + " 263")
 183     ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
 184     if ev is None:
 185         raise Exception("ANQP operation timed out")
 186
 187     res1 = dev[0].get_bss(bssid)
 188     res2 = dev[0].get_bss(bssid2)
 189     if res1['anqp_nai_realm'] == res2['anqp_nai_realm']:
 190         raise Exception("ANQP results were not unshared")
 191
 192 def test_ap_nai_home_realm_query(dev, apdev):
 193     """NAI Home Realm Query"""
 194     bssid = apdev[0]['bssid']
 195     params = hs20_ap_params()
 196     params['nai_realm'] = [ "0,example.com,13[5:6],21[2:4][5:7]",
 197                             "0,another.example.org" ]
 198     hostapd.add_ap(apdev[0]['ifname'], params)
 199
 200     dev[0].scan(freq="2412")
 201     dev[0].request("HS20_GET_NAI_HOME_REALM_LIST " + bssid + " realm=example.com")
 202     ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
 203     if ev is None:
 204         raise Exception("ANQP operation timed out")
 205     nai1 = dev[0].get_bss(bssid)['anqp_nai_realm']
 206     dev[0].dump_monitor()
 207
 208     dev[0].request("ANQP_GET " + bssid + " 263")
 209     ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
 210     if ev is None:
 211         raise Exception("ANQP operation timed out")
 212     nai2 = dev[0].get_bss(bssid)['anqp_nai_realm']
 213
 214     if len(nai1) >= len(nai2):
 215         raise Exception("Unexpected NAI Realm list response lengths")
 216     if "example.com".encode('hex') not in nai1:
 217         raise Exception("Home realm not reported")
 218     if "example.org".encode('hex') in nai1:
 219         raise Exception("Non-home realm reported")
 220     if "example.com".encode('hex') not in nai2:
 221         raise Exception("Home realm not reported in wildcard query")
 222     if "example.org".encode('hex') not in nai2:
 223         raise Exception("Non-home realm not reported in wildcard query ")
 224
 225 def test_ap_interworking_scan_filtering(dev, apdev):
 226     """Interworking scan filtering with HESSID and access network type"""
 227     bssid = apdev[0]['bssid']
 228     params = hs20_ap_params()
 229     ssid = "test-hs20-ap1"
 230     params['ssid'] = ssid
 231     params['hessid'] = bssid
 232     hostapd.add_ap(apdev[0]['ifname'], params)
 233
 234     bssid2 = apdev[1]['bssid']
 235     params = hs20_ap_params()
 236     ssid2 = "test-hs20-ap2"
 237     params['ssid'] = ssid2
 238     params['hessid'] = bssid2
 239     params['access_network_type'] = "1"
 240     del params['venue_group']
 241     del params['venue_type']
 242     hostapd.add_ap(apdev[1]['ifname'], params)
 243
 244     dev[0].hs20_enable()
 245
 246     wt = Wlantest()
 247     wt.flush()
 248
 249     logger.info("Check probe request filtering based on HESSID")
 250
 251     dev[0].request("SET hessid " + bssid2)
 252     dev[0].scan(freq="2412")
 253     time.sleep(0.03)
 254     check_probe_resp(wt, bssid, bssid2)
 255
 256     logger.info("Check probe request filtering based on access network type")
 257
 258     wt.clear_bss_counters(bssid)
 259     wt.clear_bss_counters(bssid2)
 260     dev[0].request("SET hessid 00:00:00:00:00:00")
 261     dev[0].request("SET access_network_type 14")
 262     dev[0].scan(freq="2412")
 263     time.sleep(0.03)
 264     check_probe_resp(wt, bssid2, bssid)
 265
 266     wt.clear_bss_counters(bssid)
 267     wt.clear_bss_counters(bssid2)
 268     dev[0].request("SET hessid 00:00:00:00:00:00")
 269     dev[0].request("SET access_network_type 1")
 270     dev[0].scan(freq="2412")
 271     time.sleep(0.03)
 272     check_probe_resp(wt, bssid, bssid2)
 273
 274     logger.info("Check probe request filtering based on HESSID and ANT")
 275
 276     wt.clear_bss_counters(bssid)
 277     wt.clear_bss_counters(bssid2)
 278     dev[0].request("SET hessid " + bssid)
 279     dev[0].request("SET access_network_type 14")
 280     dev[0].scan(freq="2412")
 281     time.sleep(0.03)
 282     check_probe_resp(wt, bssid2, bssid)
 283
 284     wt.clear_bss_counters(bssid)
 285     wt.clear_bss_counters(bssid2)
 286     dev[0].request("SET hessid " + bssid2)
 287     dev[0].request("SET access_network_type 14")
 288     dev[0].scan(freq="2412")
 289     time.sleep(0.03)
 290     check_probe_resp(wt, bssid, None)
 291     check_probe_resp(wt, bssid2, None)
 292
 293     wt.clear_bss_counters(bssid)
 294     wt.clear_bss_counters(bssid2)
 295     dev[0].request("SET hessid " + bssid)
 296     dev[0].request("SET access_network_type 1")
 297     dev[0].scan(freq="2412")
 298     time.sleep(0.03)
 299     check_probe_resp(wt, bssid, None)
 300     check_probe_resp(wt, bssid2, None)
 301
 302 def test_ap_hs20_select(dev, apdev):
 303     """Hotspot 2.0 network selection"""
 304     bssid = apdev[0]['bssid']
 305     params = hs20_ap_params()
 306     params['hessid'] = bssid
 307     hostapd.add_ap(apdev[0]['ifname'], params)
 308
 309     dev[0].hs20_enable()
 310     id = dev[0].add_cred_values({ 'realm': "example.com", 'username': "test",
 311                                   'password': "secret",
 312                                   'domain': "example.com" })
 313     interworking_select(dev[0], bssid, "home")
 314
 315     dev[0].remove_cred(id)
 316     id = dev[0].add_cred_values({ 'realm': "example.com", 'username': "test",
 317                                   'password': "secret",
 318                                   'domain': "no.match.example.com" })
 319     interworking_select(dev[0], bssid, "roaming", freq="2412")
 320
 321     dev[0].set_cred_quoted(id, "realm", "no.match.example.com");
 322     interworking_select(dev[0], bssid, no_match=True, freq="2412")
 323
 324     bssid2 = apdev[1]['bssid']
 325     params = hs20_ap_params()
 326     params['nai_realm'] = [ "0,example.org,21" ]
 327     params['hessid'] = bssid2
 328     params['domain_name'] = "example.org"
 329     hostapd.add_ap(apdev[1]['ifname'], params)
 330     dev[0].remove_cred(id)
 331     id = dev[0].add_cred_values({ 'realm': "example.org", 'username': "test",
 332                                   'password': "secret",
 333                                   'domain': "example.org" })
 334     interworking_select(dev[0], bssid2, "home", freq="2412")
 335
 336 def hs20_simulated_sim(dev, ap, method):
 337     bssid = ap['bssid']
 338     params = hs20_ap_params()
 339     params['hessid'] = bssid
 340     params['anqp_3gpp_cell_net'] = "555,444"
 341     params['domain_name'] = "wlan.mnc444.mcc555.3gppnetwork.org"
 342     hostapd.add_ap(ap['ifname'], params)
 343
 344     dev.hs20_enable()
 345     dev.add_cred_values({ 'imsi': "555444-333222111", 'eap': method,
 346                           'milenage': "5122250214c33e723a5dd523fc145fc0:981d464c7c52eb6e5036234984ad0bcf:000000000123"})
 347     interworking_select(dev, "home", freq="2412")
 348     interworking_connect(dev, bssid, method)
 349     check_sp_type(dev, "home")
 350
 351 def test_ap_hs20_sim(dev, apdev):
 352     """Hotspot 2.0 with simulated SIM and EAP-SIM"""
 353     if not hlr_auc_gw_available():
 354         return "skip"
 355     hs20_simulated_sim(dev[0], apdev[0], "SIM")
 356     dev[0].request("INTERWORKING_SELECT auto freq=2412")
 357     ev = dev[0].wait_event(["INTERWORKING-ALREADY-CONNECTED"], timeout=15)
 358     if ev is None:
 359         raise Exception("Timeout on already-connected event")
 360
 361 def test_ap_hs20_aka(dev, apdev):
 362     """Hotspot 2.0 with simulated USIM and EAP-AKA"""
 363     if not hlr_auc_gw_available():
 364         return "skip"
 365     hs20_simulated_sim(dev[0], apdev[0], "AKA")
 366
 367 def test_ap_hs20_aka_prime(dev, apdev):
 368     """Hotspot 2.0 with simulated USIM and EAP-AKA'"""
 369     if not hlr_auc_gw_available():
 370         return "skip"
 371     hs20_simulated_sim(dev[0], apdev[0], "AKA'")
 372
 373 def test_ap_hs20_ext_sim(dev, apdev):
 374     """Hotspot 2.0 with external SIM processing"""
 375     if not hlr_auc_gw_available():
 376         return "skip"
 377     bssid = apdev[0]['bssid']
 378     params = hs20_ap_params()
 379     params['hessid'] = bssid
 380     params['anqp_3gpp_cell_net'] = "232,01"
 381     params['domain_name'] = "wlan.mnc001.mcc232.3gppnetwork.org"
 382     hostapd.add_ap(apdev[0]['ifname'], params)
 383
 384     dev[0].hs20_enable()
 385     dev[0].request("SET external_sim 1")
 386     dev[0].add_cred_values({ 'imsi': "23201-0000000000", 'eap': "SIM" })
 387     interworking_select(dev[0], "home", freq="2412")
 388     interworking_ext_sim_connect(dev[0], bssid, "SIM")
 389     check_sp_type(dev[0], "home")
 390
 391 def test_ap_hs20_ext_sim_roaming(dev, apdev):
 392     """Hotspot 2.0 with external SIM processing in roaming network"""
 393     if not hlr_auc_gw_available():
 394         return "skip"
 395     bssid = apdev[0]['bssid']
 396     params = hs20_ap_params()
 397     params['hessid'] = bssid
 398     params['anqp_3gpp_cell_net'] = "244,91;310,026;232,01;234,56"
 399     params['domain_name'] = "wlan.mnc091.mcc244.3gppnetwork.org"
 400     hostapd.add_ap(apdev[0]['ifname'], params)
 401
 402     dev[0].hs20_enable()
 403     dev[0].request("SET external_sim 1")
 404     dev[0].add_cred_values({ 'imsi': "23201-0000000000", 'eap': "SIM" })
 405     interworking_select(dev[0], "roaming", freq="2412")
 406     interworking_ext_sim_connect(dev[0], bssid, "SIM")
 407     check_sp_type(dev[0], "roaming")
 408
 409 def test_ap_hs20_username(dev, apdev):
 410     """Hotspot 2.0 connection in username/password credential"""
 411     bssid = apdev[0]['bssid']
 412     params = hs20_ap_params()
 413     params['hessid'] = bssid
 414     hostapd.add_ap(apdev[0]['ifname'], params)
 415
 416     dev[0].hs20_enable()
 417     id = dev[0].add_cred_values({ 'realm': "example.com",
 418                                   'username': "hs20-test",
 419                                   'password': "password",
 420                                   'ca_cert': "auth_serv/ca.pem",
 421                                   'domain': "example.com",
 422                                   'update_identifier': "1234" })
 423     interworking_select(dev[0], bssid, "home", freq="2412")
 424     interworking_connect(dev[0], bssid, "TTLS")
 425     check_sp_type(dev[0], "home")
 426     status = dev[0].get_status()
 427     if status['pairwise_cipher'] != "CCMP":
 428         raise Exception("Unexpected pairwise cipher")
 429     if status['hs20'] != "2":
 430         raise Exception("Unexpected HS 2.0 support indication")
 431
 432     dev[1].connect("test-hs20", key_mgmt="WPA-EAP", eap="TTLS",
 433                    identity="hs20-test", password="password",
 434                    ca_cert="auth_serv/ca.pem", phase2="auth=MSCHAPV2",
 435                    scan_freq="2412")
 436
 437 def eap_test(dev, ap, eap_params, method, user):
 438     bssid = ap['bssid']
 439     params = hs20_ap_params()
 440     params['nai_realm'] = [ "0,example.com," + eap_params ]
 441     hostapd.add_ap(ap['ifname'], params)
 442
 443     dev.hs20_enable()
 444     dev.add_cred_values({ 'realm': "example.com",
 445                           'username': user,
 446                           'password': "password" })
 447     interworking_select(dev, bssid, freq="2412")
 448     interworking_connect(dev, bssid, method)
 449
 450 def test_ap_hs20_eap_peap_mschapv2(dev, apdev):
 451     """Hotspot 2.0 connection with PEAP/MSCHAPV2"""
 452     eap_test(dev[0], apdev[0], "25[3:26]", "PEAP", "user")
 453
 454 def test_ap_hs20_eap_peap_gtc(dev, apdev):
 455     """Hotspot 2.0 connection with PEAP/GTC"""
 456     eap_test(dev[0], apdev[0], "25[3:6]", "PEAP", "user")
 457
 458 def test_ap_hs20_eap_ttls_chap(dev, apdev):
 459     """Hotspot 2.0 connection with TTLS/CHAP"""
 460     eap_test(dev[0], apdev[0], "21[2:2]", "TTLS", "chap user")
 461
 462 def test_ap_hs20_eap_ttls_mschap(dev, apdev):
 463     """Hotspot 2.0 connection with TTLS/MSCHAP"""
 464     eap_test(dev[0], apdev[0], "21[2:3]", "TTLS", "mschap user")
 465
 466 def test_ap_hs20_eap_ttls_eap_mschapv2(dev, apdev):
 467     """Hotspot 2.0 connection with TTLS/EAP-MSCHAPv2"""
 468     eap_test(dev[0], apdev[0], "21[3:26]", "TTLS", "user")
 469
 470 def test_ap_hs20_eap_fast_mschapv2(dev, apdev):
 471     """Hotspot 2.0 connection with FAST/EAP-MSCHAPV2"""
 472     eap_test(dev[0], apdev[0], "43[3:26]", "FAST", "user")
 473
 474 def test_ap_hs20_eap_fast_gtc(dev, apdev):
 475     """Hotspot 2.0 connection with FAST/EAP-GTC"""
 476     eap_test(dev[0], apdev[0], "43[3:6]", "FAST", "user")
 477
 478 def test_ap_hs20_eap_tls(dev, apdev):
 479     """Hotspot 2.0 connection with EAP-TLS"""
 480     bssid = apdev[0]['bssid']
 481     params = hs20_ap_params()
 482     params['nai_realm'] = [ "0,example.com,13[5:6]" ]
 483     hostapd.add_ap(apdev[0]['ifname'], params)
 484
 485     dev[0].hs20_enable()
 486     dev[0].add_cred_values({ 'realm': "example.com",
 487                              'username': "certificate-user",
 488                              'ca_cert': "auth_serv/ca.pem",
 489                              'client_cert': "auth_serv/user.pem",
 490                              'private_key': "auth_serv/user.key"})
 491     interworking_select(dev[0], bssid, freq="2412")
 492     interworking_connect(dev[0], bssid, "TLS")
 493
 494 def test_ap_hs20_nai_realms(dev, apdev):
 495     """Hotspot 2.0 connection and multiple NAI realms and TTLS/PAP"""
 496     bssid = apdev[0]['bssid']
 497     params = hs20_ap_params()
 498     params['hessid'] = bssid
 499     params['nai_realm'] = [ "0,no.match.here;example.com;no.match.here.either,21[2:1][5:7]" ]
 500     hostapd.add_ap(apdev[0]['ifname'], params)
 501
 502     dev[0].hs20_enable()
 503     id = dev[0].add_cred_values({ 'realm': "example.com",
 504                                   'username': "pap user",
 505                                   'password': "password",
 506                                   'domain': "example.com" })
 507     interworking_select(dev[0], bssid, "home", freq="2412")
 508     interworking_connect(dev[0], bssid, "TTLS")
 509     check_sp_type(dev[0], "home")
 510
 511 def test_ap_hs20_roaming_consortium(dev, apdev):
 512     """Hotspot 2.0 connection based on roaming consortium match"""
 513     bssid = apdev[0]['bssid']
 514     params = hs20_ap_params()
 515     params['hessid'] = bssid
 516     hostapd.add_ap(apdev[0]['ifname'], params)
 517
 518     dev[0].hs20_enable()
 519     for consortium in [ "112233", "1020304050", "010203040506", "fedcba" ]:
 520         id = dev[0].add_cred_values({ 'username': "user",
 521                                       'password': "password",
 522                                       'domain': "example.com",
 523                                       'roaming_consortium': consortium,
 524                                       'eap': "PEAP" })
 525         interworking_select(dev[0], bssid, "home", freq="2412")
 526         interworking_connect(dev[0], bssid, "PEAP")
 527         check_sp_type(dev[0], "home")
 528         dev[0].request("INTERWORKING_SELECT auto freq=2412")
 529         ev = dev[0].wait_event(["INTERWORKING-ALREADY-CONNECTED"], timeout=15)
 530         if ev is None:
 531             raise Exception("Timeout on already-connected event")
 532         dev[0].remove_cred(id)
 533
 534 def test_ap_hs20_username_roaming(dev, apdev):
 535     """Hotspot 2.0 connection in username/password credential (roaming)"""
 536     bssid = apdev[0]['bssid']
 537     params = hs20_ap_params()
 538     params['nai_realm'] = [ "0,example.com,13[5:6],21[2:4][5:7]",
 539                             "0,roaming.example.com,21[2:4][5:7]",
 540                             "0,another.example.com" ]
 541     params['domain_name'] = "another.example.com"
 542     params['hessid'] = bssid
 543     hostapd.add_ap(apdev[0]['ifname'], params)
 544
 545     dev[0].hs20_enable()
 546     id = dev[0].add_cred_values({ 'realm': "roaming.example.com",
 547                                   'username': "hs20-test",
 548                                   'password': "password",
 549                                   'domain': "example.com" })
 550     interworking_select(dev[0], bssid, "roaming", freq="2412")
 551     interworking_connect(dev[0], bssid, "TTLS")
 552     check_sp_type(dev[0], "roaming")
 553
 554 def test_ap_hs20_username_unknown(dev, apdev):
 555     """Hotspot 2.0 connection in username/password credential (no domain in cred)"""
 556     bssid = apdev[0]['bssid']
 557     params = hs20_ap_params()
 558     params['hessid'] = bssid
 559     hostapd.add_ap(apdev[0]['ifname'], params)
 560
 561     dev[0].hs20_enable()
 562     id = dev[0].add_cred_values({ 'realm': "example.com",
 563                                   'username': "hs20-test",
 564                                   'password': "password" })
 565     interworking_select(dev[0], bssid, "unknown", freq="2412")
 566     interworking_connect(dev[0], bssid, "TTLS")
 567     check_sp_type(dev[0], "unknown")
 568
 569 def test_ap_hs20_username_unknown2(dev, apdev):
 570     """Hotspot 2.0 connection in username/password credential (no domain advertized)"""
 571     bssid = apdev[0]['bssid']
 572     params = hs20_ap_params()
 573     params['hessid'] = bssid
 574     del params['domain_name']
 575     hostapd.add_ap(apdev[0]['ifname'], params)
 576
 577     dev[0].hs20_enable()
 578     id = dev[0].add_cred_values({ 'realm': "example.com",
 579                                   'username': "hs20-test",
 580                                   'password': "password",
 581                                   'domain': "example.com" })
 582     interworking_select(dev[0], bssid, "unknown", freq="2412")
 583     interworking_connect(dev[0], bssid, "TTLS")
 584     check_sp_type(dev[0], "unknown")
 585
 586 def test_ap_hs20_gas_while_associated(dev, apdev):
 587     """Hotspot 2.0 connection with GAS query while associated"""
 588     bssid = apdev[0]['bssid']
 589     params = hs20_ap_params()
 590     params['hessid'] = bssid
 591     hostapd.add_ap(apdev[0]['ifname'], params)
 592
 593     dev[0].hs20_enable()
 594     id = dev[0].add_cred_values({ 'realm': "example.com",
 595                                   'username': "hs20-test",
 596                                   'password': "password",
 597                                   'domain': "example.com" })
 598     interworking_select(dev[0], bssid, "home", freq="2412")
 599     interworking_connect(dev[0], bssid, "TTLS")
 600
 601     logger.info("Verifying GAS query while associated")
 602     dev[0].request("FETCH_ANQP")
 603     for i in range(0, 6):
 604         ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
 605         if ev is None:
 606             raise Exception("Operation timed out")
 607
 608 def test_ap_hs20_gas_while_associated_with_pmf(dev, apdev):
 609     """Hotspot 2.0 connection with GAS query while associated and using PMF"""
 610     bssid = apdev[0]['bssid']
 611     params = hs20_ap_params()
 612     params['hessid'] = bssid
 613     hostapd.add_ap(apdev[0]['ifname'], params)
 614
 615     bssid2 = apdev[1]['bssid']
 616     params = hs20_ap_params()
 617     params['hessid'] = bssid2
 618     params['nai_realm'] = [ "0,no-match.example.org,13[5:6],21[2:4][5:7]" ]
 619     hostapd.add_ap(apdev[1]['ifname'], params)
 620
 621     dev[0].hs20_enable()
 622     dev[0].request("SET pmf 2")
 623     id = dev[0].add_cred_values({ 'realm': "example.com",
 624                                   'username': "hs20-test",
 625                                   'password': "password",
 626                                   'domain': "example.com" })
 627     interworking_select(dev[0], bssid, "home", freq="2412")
 628     interworking_connect(dev[0], bssid, "TTLS")
 629
 630     logger.info("Verifying GAS query while associated")
 631     dev[0].request("FETCH_ANQP")
 632     for i in range(0, 2 * 6):
 633         ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
 634         if ev is None:
 635             raise Exception("Operation timed out")
 636
 637 def test_ap_hs20_gas_frag_while_associated(dev, apdev):
 638     """Hotspot 2.0 connection with fragmented GAS query while associated"""
 639     bssid = apdev[0]['bssid']
 640     params = hs20_ap_params()
 641     params['hessid'] = bssid
 642     hostapd.add_ap(apdev[0]['ifname'], params)
 643     hapd = hostapd.Hostapd(apdev[0]['ifname'])
 644     hapd.set("gas_frag_limit", "50")
 645
 646     dev[0].hs20_enable()
 647     id = dev[0].add_cred_values({ 'realm': "example.com",
 648                                   'username': "hs20-test",
 649                                   'password': "password",
 650                                   'domain': "example.com" })
 651     interworking_select(dev[0], bssid, "home", freq="2412")
 652     interworking_connect(dev[0], bssid, "TTLS")
 653
 654     logger.info("Verifying GAS query while associated")
 655     dev[0].request("FETCH_ANQP")
 656     for i in range(0, 6):
 657         ev = dev[0].wait_event(["RX-ANQP"], timeout=5)
 658         if ev is None:
 659             raise Exception("Operation timed out")
 660
 661 def test_ap_hs20_multiple_connects(dev, apdev):
 662     """Hotspot 2.0 connection through multiple network selections"""
 663     bssid = apdev[0]['bssid']
 664     params = hs20_ap_params()
 665     params['hessid'] = bssid
 666     hostapd.add_ap(apdev[0]['ifname'], params)
 667
 668     dev[0].hs20_enable()
 669     values = { 'realm': "example.com",
 670                'username': "hs20-test",
 671                'password': "password",
 672                'domain': "example.com" }
 673     id = dev[0].add_cred_values(values)
 674
 675     for i in range(0, 3):
 676         logger.info("Starting Interworking network selection")
 677         dev[0].request("INTERWORKING_SELECT auto freq=2412")
 678         while True:
 679             ev = dev[0].wait_event(["INTERWORKING-NO-MATCH",
 680                                     "INTERWORKING-ALREADY-CONNECTED",
 681                                     "CTRL-EVENT-CONNECTED"], timeout=15)
 682             if ev is None:
 683                 raise Exception("Connection timed out")
 684             if "INTERWORKING-NO-MATCH" in ev:
 685                 raise Exception("Matching AP not found")
 686             if "CTRL-EVENT-CONNECTED" in ev:
 687                 break
 688             if i == 2 and "INTERWORKING-ALREADY-CONNECTED" in ev:
 689                 break
 690         if i == 0:
 691             dev[0].request("DISCONNECT")
 692         dev[0].dump_monitor()
 693
 694     networks = dev[0].list_networks()
 695     if len(networks) > 1:
 696         raise Exception("Duplicated network block detected")
 697
 698 def test_ap_hs20_disallow_aps(dev, apdev):
 699     """Hotspot 2.0 connection and disallow_aps"""
 700     bssid = apdev[0]['bssid']
 701     params = hs20_ap_params()
 702     params['hessid'] = bssid
 703     hostapd.add_ap(apdev[0]['ifname'], params)
 704
 705     dev[0].hs20_enable()
 706     values = { 'realm': "example.com",
 707                'username': "hs20-test",
 708                'password': "password",
 709                'domain': "example.com" }
 710     id = dev[0].add_cred_values(values)
 711
 712     logger.info("Verify disallow_aps bssid")
 713     dev[0].request("SET disallow_aps bssid " + bssid.translate(None, ':'))
 714     dev[0].request("INTERWORKING_SELECT auto")
 715     ev = dev[0].wait_event(["INTERWORKING-NO-MATCH"], timeout=15)
 716     if ev is None:
 717         raise Exception("Network selection timed out")
 718     dev[0].dump_monitor()
 719
 720     logger.info("Verify disallow_aps ssid")
 721     dev[0].request("SET disallow_aps ssid 746573742d68733230")
 722     dev[0].request("INTERWORKING_SELECT auto freq=2412")
 723     ev = dev[0].wait_event(["INTERWORKING-NO-MATCH"], timeout=15)
 724     if ev is None:
 725         raise Exception("Network selection timed out")
 726     dev[0].dump_monitor()
 727
 728     logger.info("Verify disallow_aps clear")
 729     dev[0].request("SET disallow_aps ")
 730     interworking_select(dev[0], bssid, "home", freq="2412")
 731
 732     dev[0].request("SET disallow_aps bssid " + bssid.translate(None, ':'))
 733     ret = dev[0].request("INTERWORKING_CONNECT " + bssid)
 734     if "FAIL" not in ret:
 735         raise Exception("INTERWORKING_CONNECT to disallowed BSS not rejected")
 736
 737 def policy_test(dev, ap, values, only_one=True):
 738     dev.dump_monitor()
 739     if ap:
 740         logger.info("Verify network selection to AP " + ap['ifname'])
 741         bssid = ap['bssid']
 742     else:
 743         logger.info("Verify network selection")
 744         bssid = None
 745     dev.hs20_enable()
 746     id = dev.add_cred_values(values)
 747     dev.request("INTERWORKING_SELECT auto freq=2412")
 748     events = []
 749     while True:
 750         ev = dev.wait_event(["INTERWORKING-AP", "INTERWORKING-NO-MATCH",
 751                              "INTERWORKING-BLACKLISTED",
 752                              "INTERWORKING-SELECTED"], timeout=15)
 753         if ev is None:
 754             raise Exception("Network selection timed out")
 755         events.append(ev)
 756         if "INTERWORKING-NO-MATCH" in ev:
 757             raise Exception("Matching AP not found")
 758         if bssid and only_one and "INTERWORKING-AP" in ev and bssid not in ev:
 759             raise Exception("Unexpected AP claimed acceptable")
 760         if "INTERWORKING-SELECTED" in ev:
 761             if bssid and bssid not in ev:
 762                 raise Exception("Selected incorrect BSS")
 763             break
 764
 765     ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
 766     if ev is None:
 767         raise Exception("Connection timed out")
 768     if bssid and bssid not in ev:
 769         raise Exception("Connected to incorrect BSS")
 770
 771     conn_bssid = dev.get_status_field("bssid")
 772     if bssid and conn_bssid != bssid:
 773         raise Exception("bssid information points to incorrect BSS")
 774
 775     dev.remove_cred(id)
 776     dev.dump_monitor()
 777     return events
 778
 779 def default_cred(domain=None):
 780     cred = { 'realm': "example.com",
 781              'username': "hs20-test",
 782              'password': "password" }
 783     if domain:
 784         cred['domain'] = domain
 785     return cred
 786
 787 def test_ap_hs20_prefer_home(dev, apdev):
 788     """Hotspot 2.0 required roaming consortium"""
 789     params = hs20_ap_params()
 790     params['domain_name'] = "example.org"
 791     hostapd.add_ap(apdev[0]['ifname'], params)
 792
 793     params = hs20_ap_params()
 794     params['ssid'] = "test-hs20-other"
 795     params['domain_name'] = "example.com"
 796     hostapd.add_ap(apdev[1]['ifname'], params)
 797
 798     values = default_cred()
 799     values['domain'] = "example.com"
 800     policy_test(dev[0], apdev[1], values, only_one=False)
 801     values['domain'] = "example.org"
 802     policy_test(dev[0], apdev[0], values, only_one=False)
 803
 804 def test_ap_hs20_req_roaming_consortium(dev, apdev):
 805     """Hotspot 2.0 required roaming consortium"""
 806     params = hs20_ap_params()
 807     hostapd.add_ap(apdev[0]['ifname'], params)
 808
 809     params = hs20_ap_params()
 810     params['ssid'] = "test-hs20-other"
 811     params['roaming_consortium'] = [ "223344" ]
 812     hostapd.add_ap(apdev[1]['ifname'], params)
 813
 814     values = default_cred()
 815     values['required_roaming_consortium'] = "223344"
 816     policy_test(dev[0], apdev[1], values)
 817     values['required_roaming_consortium'] = "112233"
 818     policy_test(dev[0], apdev[0], values)
 819
 820     id = dev[0].add_cred()
 821     dev[0].set_cred(id, "required_roaming_consortium", "112233")
 822     dev[0].set_cred(id, "required_roaming_consortium", "112233445566778899aabbccddeeff")
 823
 824     for val in [ "", "1", "11", "1122", "1122334", "112233445566778899aabbccddeeff00" ]:
 825         if "FAIL" not in dev[0].request('SET_CRED {} required_roaming_consortium {}'.format(id, val)):
 826             raise Exception("Invalid roaming consortium value accepted: " + val)
 827
 828 def test_ap_hs20_excluded_ssid(dev, apdev):
 829     """Hotspot 2.0 exclusion based on SSID"""
 830     params = hs20_ap_params()
 831     params['roaming_consortium'] = [ "223344" ]
 832     params['anqp_3gpp_cell_net'] = "555,444"
 833     hostapd.add_ap(apdev[0]['ifname'], params)
 834
 835     params = hs20_ap_params()
 836     params['ssid'] = "test-hs20-other"
 837     params['roaming_consortium'] = [ "223344" ]
 838     params['anqp_3gpp_cell_net'] = "555,444"
 839     hostapd.add_ap(apdev[1]['ifname'], params)
 840
 841     values = default_cred()
 842     values['excluded_ssid'] = "test-hs20"
 843     events = policy_test(dev[0], apdev[1], values)
 844     ev = [e for e in events if "INTERWORKING-BLACKLISTED " + apdev[0]['bssid'] in e]
 845     if len(ev) != 1:
 846         raise Exception("Excluded network not reported")
 847     values['excluded_ssid'] = "test-hs20-other"
 848     events = policy_test(dev[0], apdev[0], values)
 849     ev = [e for e in events if "INTERWORKING-BLACKLISTED " + apdev[1]['bssid'] in e]
 850     if len(ev) != 1:
 851         raise Exception("Excluded network not reported")
 852
 853     values = default_cred()
 854     values['roaming_consortium'] = "223344"
 855     values['eap'] = "TTLS"
 856     values['phase2'] = "auth=MSCHAPV2"
 857     values['excluded_ssid'] = "test-hs20"
 858     events = policy_test(dev[0], apdev[1], values)
 859     ev = [e for e in events if "INTERWORKING-BLACKLISTED " + apdev[0]['bssid'] in e]
 860     if len(ev) != 1:
 861         raise Exception("Excluded network not reported")
 862
 863     values = { 'imsi': "555444-333222111", 'eap': "SIM",
 864                'milenage': "5122250214c33e723a5dd523fc145fc0:981d464c7c52eb6e5036234984ad0bcf:000000000123",
 865                'excluded_ssid': "test-hs20" }
 866     events = policy_test(dev[0], apdev[1], values)
 867     ev = [e for e in events if "INTERWORKING-BLACKLISTED " + apdev[0]['bssid'] in e]
 868     if len(ev) != 1:
 869         raise Exception("Excluded network not reported")
 870
 871 def test_ap_hs20_roam_to_higher_prio(dev, apdev):
 872     """Hotspot 2.0 and roaming from current to higher priority network"""
 873     bssid = apdev[0]['bssid']
 874     params = hs20_ap_params(ssid="test-hs20-visited")
 875     params['domain_name'] = "visited.example.org"
 876     hostapd.add_ap(apdev[0]['ifname'], params)
 877
 878     dev[0].hs20_enable()
 879     id = dev[0].add_cred_values({ 'realm': "example.com",
 880                                   'username': "hs20-test",
 881                                   'password': "password",
 882                                   'domain': "example.com" })
 883     logger.info("Connect to the only network option")
 884     interworking_select(dev[0], bssid, "roaming", freq="2412")
 885     dev[0].dump_monitor()
 886     interworking_connect(dev[0], bssid, "TTLS")
 887
 888     logger.info("Start another AP (home operator) and reconnect")
 889     bssid2 = apdev[1]['bssid']
 890     params = hs20_ap_params(ssid="test-hs20-home")
 891     params['domain_name'] = "example.com"
 892     hostapd.add_ap(apdev[1]['ifname'], params)
 893
 894     dev[0].request("INTERWORKING_SELECT auto freq=2412")
 895     ev = dev[0].wait_event(["INTERWORKING-NO-MATCH",
 896                             "INTERWORKING-ALREADY-CONNECTED",
 897                             "CTRL-EVENT-CONNECTED"], timeout=15)
 898     if ev is None:
 899         raise Exception("Connection timed out")
 900     if "INTERWORKING-NO-MATCH" in ev:
 901         raise Exception("Matching AP not found")
 902     if "INTERWORKING-ALREADY-CONNECTED" in ev:
 903         raise Exception("Unexpected AP selected")
 904     if bssid2 not in ev:
 905         raise Exception("Unexpected BSSID after reconnection")
 906
 907 def test_ap_hs20_domain_suffix_match(dev, apdev):
 908     """Hotspot 2.0 and domain_suffix_match"""
 909     bssid = apdev[0]['bssid']
 910     params = hs20_ap_params()
 911     hostapd.add_ap(apdev[0]['ifname'], params)
 912
 913     dev[0].hs20_enable()
 914     id = dev[0].add_cred_values({ 'realm': "example.com",
 915                                   'username': "hs20-test",
 916                                   'password': "password",
 917                                   'domain': "example.com",
 918                                   'domain_suffix_match': "w1.fi" })
 919     interworking_select(dev[0], bssid, "home", freq="2412")
 920     dev[0].dump_monitor()
 921     interworking_connect(dev[0], bssid, "TTLS")
 922     dev[0].request("REMOVE_NETWORK all")
 923     dev[0].dump_monitor()
 924
 925     dev[0].set_cred_quoted(id, "domain_suffix_match", "no-match.example.com")
 926     interworking_select(dev[0], bssid, "home", freq="2412")
 927     dev[0].dump_monitor()
 928     dev[0].request("INTERWORKING_CONNECT " + bssid)
 929     ev = dev[0].wait_event(["CTRL-EVENT-EAP-TLS-CERT-ERROR"])
 930     if ev is None:
 931         raise Exception("TLS certificate error not reported")
 932     if "Domain suffix mismatch" not in ev:
 933         raise Exception("Domain suffix mismatch not reported")
 934
 935 def test_ap_hs20_roaming_partner_preference(dev, apdev):
 936     """Hotspot 2.0 and roaming partner preference"""
 937     params = hs20_ap_params()
 938     params['domain_name'] = "roaming.example.org"
 939     hostapd.add_ap(apdev[0]['ifname'], params)
 940
 941     params = hs20_ap_params()
 942     params['ssid'] = "test-hs20-other"
 943     params['domain_name'] = "roaming.example.net"
 944     hostapd.add_ap(apdev[1]['ifname'], params)
 945
 946     logger.info("Verify default vs. specified preference")
 947     values = default_cred()
 948     values['roaming_partner'] = "roaming.example.net,1,127,*"
 949     policy_test(dev[0], apdev[1], values, only_one=False)
 950     values['roaming_partner'] = "roaming.example.net,1,129,*"
 951     policy_test(dev[0], apdev[0], values, only_one=False)
 952
 953     logger.info("Verify partial FQDN match")
 954     values['roaming_partner'] = "example.net,0,0,*"
 955     policy_test(dev[0], apdev[1], values, only_one=False)
 956     values['roaming_partner'] = "example.net,0,255,*"
 957     policy_test(dev[0], apdev[0], values, only_one=False)
 958
 959 def test_ap_hs20_max_bss_load(dev, apdev):
 960     """Hotspot 2.0 and maximum BSS load"""
 961     params = hs20_ap_params()
 962     params['bss_load_test'] = "12:200:20000"
 963     hostapd.add_ap(apdev[0]['ifname'], params)
 964
 965     params = hs20_ap_params()
 966     params['ssid'] = "test-hs20-other"
 967     params['bss_load_test'] = "5:20:10000"
 968     hostapd.add_ap(apdev[1]['ifname'], params)
 969
 970     logger.info("Verify maximum BSS load constraint")
 971     values = default_cred()
 972     values['domain'] = "example.com"
 973     values['max_bss_load'] = "100"
 974     events = policy_test(dev[0], apdev[1], values, only_one=False)
 975
 976     ev = [e for e in events if "INTERWORKING-AP " + apdev[0]['bssid'] in e]
 977     if len(ev) != 1 or "over_max_bss_load=1" not in ev[0]:
 978         raise Exception("Maximum BSS Load case not noticed")
 979     ev = [e for e in events if "INTERWORKING-AP " + apdev[1]['bssid'] in e]
 980     if len(ev) != 1 or "over_max_bss_load=1" in ev[0]:
 981         raise Exception("Maximum BSS Load case reported incorrectly")
 982
 983     logger.info("Verify maximum BSS load does not prevent connection")
 984     values['max_bss_load'] = "1"
 985     events = policy_test(dev[0], None, values)
 986
 987     ev = [e for e in events if "INTERWORKING-AP " + apdev[0]['bssid'] in e]
 988     if len(ev) != 1 or "over_max_bss_load=1" not in ev[0]:
 989         raise Exception("Maximum BSS Load case not noticed")
 990     ev = [e for e in events if "INTERWORKING-AP " + apdev[1]['bssid'] in e]
 991     if len(ev) != 1 or "over_max_bss_load=1" not in ev[0]:
 992         raise Exception("Maximum BSS Load case not noticed")
 993
 994 def test_ap_hs20_max_bss_load2(dev, apdev):
 995     """Hotspot 2.0 and maximum BSS load with one AP not advertising"""
 996     params = hs20_ap_params()
 997     params['bss_load_test'] = "12:200:20000"
 998     hostapd.add_ap(apdev[0]['ifname'], params)
 999
1000     params = hs20_ap_params()
1001     params['ssid'] = "test-hs20-other"
1002     hostapd.add_ap(apdev[1]['ifname'], params)
1003
1004     logger.info("Verify maximum BSS load constraint with AP advertisement")
1005     values = default_cred()
1006     values['domain'] = "example.com"
1007     values['max_bss_load'] = "100"
1008     events = policy_test(dev[0], apdev[1], values, only_one=False)
1009
1010     ev = [e for e in events if "INTERWORKING-AP " + apdev[0]['bssid'] in e]
1011     if len(ev) != 1 or "over_max_bss_load=1" not in ev[0]:
1012         raise Exception("Maximum BSS Load case not noticed")
1013     ev = [e for e in events if "INTERWORKING-AP " + apdev[1]['bssid'] in e]
1014     if len(ev) != 1 or "over_max_bss_load=1" in ev[0]:
1015         raise Exception("Maximum BSS Load case reported incorrectly")
1016
1017 def test_ap_hs20_multi_cred_sp_prio(dev, apdev):
1018     """Hotspot 2.0 multi-cred sp_priority"""
1019     if not hlr_auc_gw_available():
1020         return "skip"
1021     bssid = apdev[0]['bssid']
1022     params = hs20_ap_params()
1023     params['hessid'] = bssid
1024     del params['domain_name']
1025     params['anqp_3gpp_cell_net'] = "232,01"
1026     hostapd.add_ap(apdev[0]['ifname'], params)
1027
1028     dev[0].hs20_enable()
1029     dev[0].request("SET external_sim 1")
1030     id1 = dev[0].add_cred_values({ 'imsi': "23201-0000000000", 'eap': "SIM",
1031                                    'provisioning_sp': "example.com",
1032                                    'sp_priority' :"1" })
1033     id2 = dev[0].add_cred_values({ 'realm': "example.com",
1034                                    'username': "hs20-test",
1035                                    'password': "password",
1036                                    'domain': "example.com",
1037                                    'provisioning_sp': "example.com",
1038                                    'sp_priority': "2" })
1039     dev[0].dump_monitor()
1040     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1041     interworking_ext_sim_auth(dev[0], "SIM")
1042     check_sp_type(dev[0], "unknown")
1043     dev[0].request("REMOVE_NETWORK all")
1044
1045     dev[0].set_cred(id1, "sp_priority", "2")
1046     dev[0].set_cred(id2, "sp_priority", "1")
1047     dev[0].dump_monitor()
1048     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1049     interworking_auth(dev[0], "TTLS")
1050     check_sp_type(dev[0], "unknown")
1051
1052 def test_ap_hs20_multi_cred_sp_prio2(dev, apdev):
1053     """Hotspot 2.0 multi-cred sp_priority with two BSSes"""
1054     if not hlr_auc_gw_available():
1055         return "skip"
1056     bssid = apdev[0]['bssid']
1057     params = hs20_ap_params()
1058     params['hessid'] = bssid
1059     del params['nai_realm']
1060     del params['domain_name']
1061     params['anqp_3gpp_cell_net'] = "232,01"
1062     hostapd.add_ap(apdev[0]['ifname'], params)
1063
1064     bssid2 = apdev[1]['bssid']
1065     params = hs20_ap_params()
1066     params['ssid'] = "test-hs20-other"
1067     params['hessid'] = bssid2
1068     del params['domain_name']
1069     del params['anqp_3gpp_cell_net']
1070     hostapd.add_ap(apdev[1]['ifname'], params)
1071
1072     dev[0].hs20_enable()
1073     dev[0].request("SET external_sim 1")
1074     id1 = dev[0].add_cred_values({ 'imsi': "23201-0000000000", 'eap': "SIM",
1075                                    'provisioning_sp': "example.com",
1076                                    'sp_priority': "1" })
1077     id2 = dev[0].add_cred_values({ 'realm': "example.com",
1078                                    'username': "hs20-test",
1079                                    'password': "password",
1080                                    'domain': "example.com",
1081                                    'provisioning_sp': "example.com",
1082                                    'sp_priority': "2" })
1083     dev[0].dump_monitor()
1084     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1085     interworking_ext_sim_auth(dev[0], "SIM")
1086     check_sp_type(dev[0], "unknown")
1087     conn_bssid = dev[0].get_status_field("bssid")
1088     if conn_bssid != bssid:
1089         raise Exception("Connected to incorrect BSS")
1090     dev[0].request("REMOVE_NETWORK all")
1091
1092     dev[0].set_cred(id1, "sp_priority", "2")
1093     dev[0].set_cred(id2, "sp_priority", "1")
1094     dev[0].dump_monitor()
1095     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1096     interworking_auth(dev[0], "TTLS")
1097     check_sp_type(dev[0], "unknown")
1098     conn_bssid = dev[0].get_status_field("bssid")
1099     if conn_bssid != bssid2:
1100         raise Exception("Connected to incorrect BSS")
1101
1102 def check_conn_capab_selection(dev, type, missing):
1103     dev.request("INTERWORKING_SELECT freq=2412")
1104     ev = dev.wait_event(["INTERWORKING-AP"])
1105     if ev is None:
1106         raise Exception("Network selection timed out");
1107     if "type=" + type not in ev:
1108         raise Exception("Unexpected network type")
1109     if missing and "conn_capab_missing=1" not in ev:
1110         raise Exception("conn_capab_missing not reported")
1111     if not missing and "conn_capab_missing=1" in ev:
1112         raise Exception("conn_capab_missing reported unexpectedly")
1113
1114 def conn_capab_cred(domain=None, req_conn_capab=None):
1115     cred = default_cred(domain=domain)
1116     if req_conn_capab:
1117         cred['req_conn_capab'] = req_conn_capab
1118     return cred
1119
1120 def test_ap_hs20_req_conn_capab(dev, apdev):
1121     """Hotspot 2.0 network selection with req_conn_capab"""
1122     bssid = apdev[0]['bssid']
1123     params = hs20_ap_params()
1124     hostapd.add_ap(apdev[0]['ifname'], params)
1125
1126     dev[0].hs20_enable()
1127     logger.info("Not used in home network")
1128     values = conn_capab_cred(domain="example.com", req_conn_capab="6:1234")
1129     id = dev[0].add_cred_values(values)
1130     check_conn_capab_selection(dev[0], "home", False)
1131
1132     logger.info("Used in roaming network")
1133     dev[0].remove_cred(id)
1134     values = conn_capab_cred(domain="example.org", req_conn_capab="6:1234")
1135     id = dev[0].add_cred_values(values)
1136     check_conn_capab_selection(dev[0], "roaming", True)
1137
1138     logger.info("Verify that req_conn_capab does not prevent connection if no other network is available")
1139     check_auto_select(dev[0], bssid)
1140
1141     logger.info("Additional req_conn_capab checks")
1142
1143     dev[0].remove_cred(id)
1144     values = conn_capab_cred(domain="example.org", req_conn_capab="1:0")
1145     id = dev[0].add_cred_values(values)
1146     check_conn_capab_selection(dev[0], "roaming", True)
1147
1148     dev[0].remove_cred(id)
1149     values = conn_capab_cred(domain="example.org", req_conn_capab="17:5060")
1150     id = dev[0].add_cred_values(values)
1151     check_conn_capab_selection(dev[0], "roaming", True)
1152
1153     bssid2 = apdev[1]['bssid']
1154     params = hs20_ap_params(ssid="test-hs20b")
1155     params['hs20_conn_capab'] = [ "1:0:2", "6:22:1", "17:5060:0", "50:0:1" ]
1156     hostapd.add_ap(apdev[1]['ifname'], params)
1157
1158     dev[0].remove_cred(id)
1159     values = conn_capab_cred(domain="example.org", req_conn_capab="50")
1160     id = dev[0].add_cred_values(values)
1161     dev[0].set_cred(id, "req_conn_capab", "6:22")
1162     dev[0].request("INTERWORKING_SELECT freq=2412")
1163     for i in range(0, 2):
1164         ev = dev[0].wait_event(["INTERWORKING-AP"])
1165         if ev is None:
1166             raise Exception("Network selection timed out");
1167         if bssid in ev and "conn_capab_missing=1" not in ev:
1168             raise Exception("Missing protocol connection capability not reported")
1169         if bssid2 in ev and "conn_capab_missing=1" in ev:
1170             raise Exception("Protocol connection capability not reported correctly")
1171
1172 def test_ap_hs20_req_conn_capab_and_roaming_partner_preference(dev, apdev):
1173     """Hotspot 2.0 and req_conn_capab with roaming partner preference"""
1174     bssid = apdev[0]['bssid']
1175     params = hs20_ap_params()
1176     params['domain_name'] = "roaming.example.org"
1177     params['hs20_conn_capab'] = [ "1:0:2", "6:22:1", "17:5060:0", "50:0:1" ]
1178     hostapd.add_ap(apdev[0]['ifname'], params)
1179
1180     bssid2 = apdev[1]['bssid']
1181     params = hs20_ap_params(ssid="test-hs20-b")
1182     params['domain_name'] = "roaming.example.net"
1183     hostapd.add_ap(apdev[1]['ifname'], params)
1184
1185     values = default_cred()
1186     values['roaming_partner'] = "roaming.example.net,1,127,*"
1187     id = dev[0].add_cred_values(values)
1188     check_auto_select(dev[0], bssid2)
1189
1190     dev[0].set_cred(id, "req_conn_capab", "50")
1191     check_auto_select(dev[0], bssid)
1192
1193     dev[0].remove_cred(id)
1194     id = dev[0].add_cred_values(values)
1195     dev[0].set_cred(id, "req_conn_capab", "51")
1196     check_auto_select(dev[0], bssid2)
1197
1198 def check_bandwidth_selection(dev, type, below):
1199     dev.request("INTERWORKING_SELECT freq=2412")
1200     ev = dev.wait_event(["INTERWORKING-AP"])
1201     if ev is None:
1202         raise Exception("Network selection timed out");
1203     if "type=" + type not in ev:
1204         raise Exception("Unexpected network type")
1205     if below and "below_min_backhaul=1" not in ev:
1206         raise Exception("below_min_backhaul not reported")
1207     if not below and "below_min_backhaul=1" in ev:
1208         raise Exception("below_min_backhaul reported unexpectedly")
1209
1210 def bw_cred(domain=None, dl_home=None, ul_home=None, dl_roaming=None, ul_roaming=None):
1211     cred = default_cred(domain=domain)
1212     if dl_home:
1213         cred['min_dl_bandwidth_home'] = str(dl_home)
1214     if ul_home:
1215         cred['min_ul_bandwidth_home'] = str(ul_home)
1216     if dl_roaming:
1217         cred['min_dl_bandwidth_roaming'] = str(dl_roaming)
1218     if ul_roaming:
1219         cred['min_ul_bandwidth_roaming'] = str(ul_roaming)
1220     return cred
1221
1222 def test_ap_hs20_min_bandwidth_home(dev, apdev):
1223     """Hotspot 2.0 network selection with min bandwidth (home)"""
1224     bssid = apdev[0]['bssid']
1225     params = hs20_ap_params()
1226     hostapd.add_ap(apdev[0]['ifname'], params)
1227
1228     dev[0].hs20_enable()
1229     values = bw_cred(domain="example.com", dl_home=5490, ul_home=58)
1230     id = dev[0].add_cred_values(values)
1231     check_bandwidth_selection(dev[0], "home", False)
1232     dev[0].remove_cred(id)
1233
1234     values = bw_cred(domain="example.com", dl_home=5491, ul_home=58)
1235     id = dev[0].add_cred_values(values)
1236     check_bandwidth_selection(dev[0], "home", True)
1237     dev[0].remove_cred(id)
1238
1239     values = bw_cred(domain="example.com", dl_home=5490, ul_home=59)
1240     id = dev[0].add_cred_values(values)
1241     check_bandwidth_selection(dev[0], "home", True)
1242     dev[0].remove_cred(id)
1243
1244     values = bw_cred(domain="example.com", dl_home=5491, ul_home=59)
1245     id = dev[0].add_cred_values(values)
1246     check_bandwidth_selection(dev[0], "home", True)
1247     check_auto_select(dev[0], bssid)
1248
1249     bssid2 = apdev[1]['bssid']
1250     params = hs20_ap_params(ssid="test-hs20-b")
1251     params['hs20_wan_metrics'] = "01:8000:1000:1:1:3000"
1252     hostapd.add_ap(apdev[1]['ifname'], params)
1253
1254     check_auto_select(dev[0], bssid2)
1255
1256 def test_ap_hs20_min_bandwidth_roaming(dev, apdev):
1257     """Hotspot 2.0 network selection with min bandwidth (roaming)"""
1258     bssid = apdev[0]['bssid']
1259     params = hs20_ap_params()
1260     hostapd.add_ap(apdev[0]['ifname'], params)
1261
1262     dev[0].hs20_enable()
1263     values = bw_cred(domain="example.org", dl_roaming=5490, ul_roaming=58)
1264     id = dev[0].add_cred_values(values)
1265     check_bandwidth_selection(dev[0], "roaming", False)
1266     dev[0].remove_cred(id)
1267
1268     values = bw_cred(domain="example.org", dl_roaming=5491, ul_roaming=58)
1269     id = dev[0].add_cred_values(values)
1270     check_bandwidth_selection(dev[0], "roaming", True)
1271     dev[0].remove_cred(id)
1272
1273     values = bw_cred(domain="example.org", dl_roaming=5490, ul_roaming=59)
1274     id = dev[0].add_cred_values(values)
1275     check_bandwidth_selection(dev[0], "roaming", True)
1276     dev[0].remove_cred(id)
1277
1278     values = bw_cred(domain="example.org", dl_roaming=5491, ul_roaming=59)
1279     id = dev[0].add_cred_values(values)
1280     check_bandwidth_selection(dev[0], "roaming", True)
1281     check_auto_select(dev[0], bssid)
1282
1283     bssid2 = apdev[1]['bssid']
1284     params = hs20_ap_params(ssid="test-hs20-b")
1285     params['hs20_wan_metrics'] = "01:8000:1000:1:1:3000"
1286     hostapd.add_ap(apdev[1]['ifname'], params)
1287
1288     check_auto_select(dev[0], bssid2)
1289
1290 def test_ap_hs20_min_bandwidth_and_roaming_partner_preference(dev, apdev):
1291     """Hotspot 2.0 and minimum bandwidth with roaming partner preference"""
1292     bssid = apdev[0]['bssid']
1293     params = hs20_ap_params()
1294     params['domain_name'] = "roaming.example.org"
1295     params['hs20_wan_metrics'] = "01:8000:1000:1:1:3000"
1296     hostapd.add_ap(apdev[0]['ifname'], params)
1297
1298     bssid2 = apdev[1]['bssid']
1299     params = hs20_ap_params(ssid="test-hs20-b")
1300     params['domain_name'] = "roaming.example.net"
1301     hostapd.add_ap(apdev[1]['ifname'], params)
1302
1303     values = default_cred()
1304     values['roaming_partner'] = "roaming.example.net,1,127,*"
1305     id = dev[0].add_cred_values(values)
1306     check_auto_select(dev[0], bssid2)
1307
1308     dev[0].set_cred(id, "min_dl_bandwidth_roaming", "6000")
1309     check_auto_select(dev[0], bssid)
1310
1311     dev[0].set_cred(id, "min_dl_bandwidth_roaming", "10000")
1312     check_auto_select(dev[0], bssid2)
1313
1314 def test_ap_hs20_min_bandwidth_no_wan_metrics(dev, apdev):
1315     """Hotspot 2.0 network selection with min bandwidth but no WAN Metrics"""
1316     bssid = apdev[0]['bssid']
1317     params = hs20_ap_params()
1318     del params['hs20_wan_metrics']
1319     hostapd.add_ap(apdev[0]['ifname'], params)
1320
1321     dev[0].hs20_enable()
1322     values = bw_cred(domain="example.com", dl_home=10000, ul_home=10000,
1323                      dl_roaming=10000, ul_roaming=10000)
1324     dev[0].add_cred_values(values)
1325     check_bandwidth_selection(dev[0], "home", False)
1326
1327 def test_ap_hs20_deauth_req_ess(dev, apdev):
1328     """Hotspot 2.0 connection and deauthentication request for ESS"""
1329     dev[0].request("SET pmf 2")
1330     eap_test(dev[0], apdev[0], "21[3:26]", "TTLS", "user")
1331     dev[0].dump_monitor()
1332     addr = dev[0].p2p_interface_addr()
1333     hapd = hostapd.Hostapd(apdev[0]['ifname'])
1334     hapd.request("HS20_DEAUTH_REQ " + addr + " 1 120 http://example.com/")
1335     ev = dev[0].wait_event(["HS20-DEAUTH-IMMINENT-NOTICE"])
1336     if ev is None:
1337         raise Exception("Timeout on deauth imminent notice")
1338     if "1 120 http://example.com/" not in ev:
1339         raise Exception("Unexpected deauth imminent notice: " + ev)
1340     hapd.request("DEAUTHENTICATE " + addr)
1341     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
1342     if ev is None:
1343         raise Exception("Timeout on disconnection")
1344     if "[TEMP-DISABLED]" not in dev[0].list_networks()[0]['flags']:
1345         raise Exception("Network not marked temporarily disabled")
1346     ev = dev[0].wait_event(["SME: Trying to authenticate",
1347                             "Trying to associate",
1348                             "CTRL-EVENT-CONNECTED"], timeout=5)
1349     if ev is not None:
1350         raise Exception("Unexpected connection attempt")
1351
1352 def test_ap_hs20_deauth_req_bss(dev, apdev):
1353     """Hotspot 2.0 connection and deauthentication request for BSS"""
1354     dev[0].request("SET pmf 2")
1355     eap_test(dev[0], apdev[0], "21[3:26]", "TTLS", "user")
1356     dev[0].dump_monitor()
1357     addr = dev[0].p2p_interface_addr()
1358     hapd = hostapd.Hostapd(apdev[0]['ifname'])
1359     hapd.request("HS20_DEAUTH_REQ " + addr + " 0 120 http://example.com/")
1360     ev = dev[0].wait_event(["HS20-DEAUTH-IMMINENT-NOTICE"])
1361     if ev is None:
1362         raise Exception("Timeout on deauth imminent notice")
1363     if "0 120 http://example.com/" not in ev:
1364         raise Exception("Unexpected deauth imminent notice: " + ev)
1365     hapd.request("DEAUTHENTICATE " + addr + " reason=4")
1366     ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
1367     if ev is None:
1368         raise Exception("Timeout on disconnection")
1369     if "reason=4" not in ev:
1370         raise Exception("Unexpected disconnection reason")
1371     if "[TEMP-DISABLED]" not in dev[0].list_networks()[0]['flags']:
1372         raise Exception("Network not marked temporarily disabled")
1373     ev = dev[0].wait_event(["SME: Trying to authenticate",
1374                             "Trying to associate",
1375                             "CTRL-EVENT-CONNECTED"], timeout=5)
1376     if ev is not None:
1377         raise Exception("Unexpected connection attempt")
1378
1379 def test_ap_hs20_osen(dev, apdev):
1380     """Hotspot 2.0 OSEN connection"""
1381     params = { 'ssid': "osen",
1382                'osen': "1",
1383                'auth_server_addr': "127.0.0.1",
1384                'auth_server_port': "1812",
1385                'auth_server_shared_secret': "radius" }
1386     hostapd.add_ap(apdev[0]['ifname'], params)
1387
1388     dev[0].connect("osen", proto="OSEN", key_mgmt="OSEN", pairwise="CCMP",
1389                    group="GTK_NOT_USED",
1390                    eap="WFA-UNAUTH-TLS", identity="osen@example.com",
1391                    ca_cert="auth_serv/ca.pem",
1392                    scan_freq="2412")
1393
1394 def test_ap_hs20_network_preference(dev, apdev):
1395     """Hotspot 2.0 network selection with preferred home network"""
1396     bssid = apdev[0]['bssid']
1397     params = hs20_ap_params()
1398     hostapd.add_ap(apdev[0]['ifname'], params)
1399
1400     dev[0].hs20_enable()
1401     values = { 'realm': "example.com",
1402                'username': "hs20-test",
1403                'password': "password",
1404                'domain': "example.com" }
1405     dev[0].add_cred_values(values)
1406
1407     id = dev[0].add_network()
1408     dev[0].set_network_quoted(id, "ssid", "home")
1409     dev[0].set_network_quoted(id, "psk", "12345678")
1410     dev[0].set_network(id, "priority", "1")
1411     dev[0].request("ENABLE_NETWORK %s no-connect" % id)
1412
1413     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1414     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
1415     if ev is None:
1416         raise Exception("Connection timed out")
1417     if bssid not in ev:
1418         raise Exception("Unexpected network selected")
1419
1420     bssid2 = apdev[1]['bssid']
1421     params = hostapd.wpa2_params(ssid="home", passphrase="12345678")
1422     hostapd.add_ap(apdev[1]['ifname'], params)
1423
1424     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1425     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED",
1426                             "INTERWORKING-ALREADY-CONNECTED" ], timeout=15)
1427     if ev is None:
1428         raise Exception("Connection timed out")
1429     if "INTERWORKING-ALREADY-CONNECTED" in ev:
1430         raise Exception("No roam to higher priority network")
1431     if bssid2 not in ev:
1432         raise Exception("Unexpected network selected")
1433
1434 def test_ap_hs20_network_preference2(dev, apdev):
1435     """Hotspot 2.0 network selection with preferred credential"""
1436     bssid2 = apdev[1]['bssid']
1437     params = hostapd.wpa2_params(ssid="home", passphrase="12345678")
1438     hostapd.add_ap(apdev[1]['ifname'], params)
1439
1440     dev[0].hs20_enable()
1441     values = { 'realm': "example.com",
1442                'username': "hs20-test",
1443                'password': "password",
1444                'domain': "example.com",
1445                'priority': "1" }
1446     dev[0].add_cred_values(values)
1447
1448     id = dev[0].add_network()
1449     dev[0].set_network_quoted(id, "ssid", "home")
1450     dev[0].set_network_quoted(id, "psk", "12345678")
1451     dev[0].request("ENABLE_NETWORK %s no-connect" % id)
1452
1453     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1454     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
1455     if ev is None:
1456         raise Exception("Connection timed out")
1457     if bssid2 not in ev:
1458         raise Exception("Unexpected network selected")
1459
1460     bssid = apdev[0]['bssid']
1461     params = hs20_ap_params()
1462     hostapd.add_ap(apdev[0]['ifname'], params)
1463
1464     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1465     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED",
1466                             "INTERWORKING-ALREADY-CONNECTED" ], timeout=15)
1467     if ev is None:
1468         raise Exception("Connection timed out")
1469     if "INTERWORKING-ALREADY-CONNECTED" in ev:
1470         raise Exception("No roam to higher priority network")
1471     if bssid not in ev:
1472         raise Exception("Unexpected network selected")
1473
1474 def test_ap_hs20_network_preference3(dev, apdev):
1475     """Hotspot 2.0 network selection with two credential (one preferred)"""
1476     bssid = apdev[0]['bssid']
1477     params = hs20_ap_params()
1478     hostapd.add_ap(apdev[0]['ifname'], params)
1479
1480     bssid2 = apdev[1]['bssid']
1481     params = hs20_ap_params(ssid="test-hs20b")
1482     params['nai_realm'] = "0,example.org,13[5:6],21[2:4][5:7]"
1483     hostapd.add_ap(apdev[1]['ifname'], params)
1484
1485     dev[0].hs20_enable()
1486     values = { 'realm': "example.com",
1487                'username': "hs20-test",
1488                'password': "password",
1489                'priority': "1" }
1490     dev[0].add_cred_values(values)
1491     values = { 'realm': "example.org",
1492                'username': "hs20-test",
1493                'password': "password" }
1494     id = dev[0].add_cred_values(values)
1495
1496     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1497     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
1498     if ev is None:
1499         raise Exception("Connection timed out")
1500     if bssid not in ev:
1501         raise Exception("Unexpected network selected")
1502
1503     dev[0].set_cred(id, "priority", "2")
1504     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1505     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED",
1506                             "INTERWORKING-ALREADY-CONNECTED" ], timeout=15)
1507     if ev is None:
1508         raise Exception("Connection timed out")
1509     if "INTERWORKING-ALREADY-CONNECTED" in ev:
1510         raise Exception("No roam to higher priority network")
1511     if bssid2 not in ev:
1512         raise Exception("Unexpected network selected")
1513
1514 def test_ap_hs20_network_preference4(dev, apdev):
1515     """Hotspot 2.0 network selection with username vs. SIM credential"""
1516     bssid = apdev[0]['bssid']
1517     params = hs20_ap_params()
1518     hostapd.add_ap(apdev[0]['ifname'], params)
1519
1520     bssid2 = apdev[1]['bssid']
1521     params = hs20_ap_params(ssid="test-hs20b")
1522     params['hessid'] = bssid2
1523     params['anqp_3gpp_cell_net'] = "555,444"
1524     params['domain_name'] = "wlan.mnc444.mcc555.3gppnetwork.org"
1525     hostapd.add_ap(apdev[1]['ifname'], params)
1526
1527     dev[0].hs20_enable()
1528     values = { 'realm': "example.com",
1529                'username': "hs20-test",
1530                'password': "password",
1531                'priority': "1" }
1532     dev[0].add_cred_values(values)
1533     values = { 'imsi': "555444-333222111",
1534                'eap': "SIM",
1535                'milenage': "5122250214c33e723a5dd523fc145fc0:981d464c7c52eb6e5036234984ad0bcf:000000000123" }
1536     id = dev[0].add_cred_values(values)
1537
1538     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1539     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
1540     if ev is None:
1541         raise Exception("Connection timed out")
1542     if bssid not in ev:
1543         raise Exception("Unexpected network selected")
1544
1545     dev[0].set_cred(id, "priority", "2")
1546     dev[0].request("INTERWORKING_SELECT auto freq=2412")
1547     ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED",
1548                             "INTERWORKING-ALREADY-CONNECTED" ], timeout=15)
1549     if ev is None:
1550         raise Exception("Connection timed out")
1551     if "INTERWORKING-ALREADY-CONNECTED" in ev:
1552         raise Exception("No roam to higher priority network")
1553     if bssid2 not in ev:
1554         raise Exception("Unexpected network selected")
1555
1556 def test_ap_hs20_fetch_osu(dev, apdev):
1557     """Hotspot 2.0 OSU provider and icon fetch"""
1558     bssid = apdev[0]['bssid']
1559     params = hs20_ap_params()
1560     params['hs20_icon'] = "128:80:zxx:image/png:w1fi_logo:w1fi_logo.png"
1561     params['osu_ssid'] = '"HS 2.0 OSU open"'
1562     params['osu_method_list'] = "1"
1563     params['osu_friendly_name'] = [ "eng:Test OSU", "fin:Testi-OSU" ]
1564     params['osu_icon'] = "w1fi_logo"
1565     params['osu_service_desc'] = [ "eng:Example services", "fin:Esimerkkipalveluja" ]
1566     params['osu_server_uri'] = "https://example.com/osu/"
1567     hostapd.add_ap(apdev[0]['ifname'], params)
1568
1569     bssid2 = apdev[1]['bssid']
1570     params = hs20_ap_params(ssid="test-hs20b")
1571     params['hessid'] = bssid2
1572     params['hs20_icon'] = "128:80:zxx:image/png:w1fi_logo:w1fi_logo.png"
1573     params['osu_ssid'] = '"HS 2.0 OSU OSEN"'
1574     params['osu_method_list'] = "0"
1575     params['osu_friendly_name'] = [ "eng:Test2 OSU", "fin:Testi2-OSU" ]
1576     params['osu_icon'] = "w1fi_logo"
1577     params['osu_service_desc'] = [ "eng:Example services2", "fin:Esimerkkipalveluja2" ]
1578     params['osu_server_uri'] = "https://example.org/osu/"
1579     hostapd.add_ap(apdev[1]['ifname'], params)
1580
1581     with open("w1fi_logo.png", "r") as f:
1582         orig_logo = f.read()
1583     dev[0].hs20_enable()
1584     dir = "/tmp/osu-fetch"
1585     if os.path.isdir(dir):
1586        files = [ f for f in os.listdir(dir) if f.startswith("osu-") ]
1587        for f in files:
1588            os.remove(dir + "/" + f)
1589     else:
1590         try:
1591             os.makedirs(dir)
1592         except:
1593             pass
1594     try:
1595         dev[0].request("SET osu_dir " + dir)
1596         dev[0].request("FETCH_OSU")
1597         icons = 0
1598         while True:
1599             ev = dev[0].wait_event(["OSU provider fetch completed",
1600                                     "RX-HS20-ANQP-ICON"], timeout=15)
1601             if ev is None:
1602                 raise Exception("Timeout on OSU fetch")
1603             if "OSU provider fetch completed" in ev:
1604                 break
1605             if "RX-HS20-ANQP-ICON" in ev:
1606                 with open(ev.split(' ')[1], "r") as f:
1607                     logo = f.read()
1608                     if logo == orig_logo:
1609                         icons += 1
1610
1611         with open(dir + "/osu-providers.txt", "r") as f:
1612             prov = f.read()
1613         if "OSU-PROVIDER " + bssid not in prov:
1614             raise Exception("Missing OSU_PROVIDER")
1615         if "OSU-PROVIDER " + bssid2 not in prov:
1616             raise Exception("Missing OSU_PROVIDER")
1617     finally:
1618         files = [ f for f in os.listdir(dir) if f.startswith("osu-") ]
1619         for f in files:
1620             os.remove(dir + "/" + f)
1621         os.rmdir(dir)
1622
1623     if icons != 2:
1624         raise Exception("Unexpected number of icons fetched")