status = dev[0].get_status()
if status['pairwise_cipher'] != "CCMP":
raise Exception("Unexpected pairwise cipher")
- if status['hs20'] != "2":
+ if status['hs20'] != "3":
raise Exception("Unexpected HS 2.0 support indication")
dev[1].connect("test-hs20", key_mgmt="WPA-EAP", eap="TTLS",
status = wpas.get_status()
if status['pairwise_cipher'] != "CCMP":
raise Exception("Unexpected pairwise cipher")
- if status['hs20'] != "2":
+ if status['hs20'] != "3":
raise Exception("Unexpected HS 2.0 support indication")
def test_ap_hs20_auto_interworking(dev, apdev):
status = dev[0].get_status()
if status['pairwise_cipher'] != "CCMP":
raise Exception("Unexpected pairwise cipher")
- if status['hs20'] != "2":
+ if status['hs20'] != "3":
raise Exception("Unexpected HS 2.0 support indication")
@remote_compatible
raise Exception("Scan timed out")
logger.info("Scan completed")
-def eap_test(dev, ap, eap_params, method, user):
+def eap_test(dev, ap, eap_params, method, user, release=0):
bssid = ap['bssid']
params = hs20_ap_params()
params['nai_realm'] = [ "0,example.com," + eap_params ]
+ if release > 0:
+ params['hs20_release'] = str(release)
hostapd.add_ap(ap, params)
dev.hs20_enable()
raise Exception("Timeout on already-connected event")
dev[0].remove_cred(id)
+def test_ap_hs20_max_roaming_consortiums(dev, apdev):
+ """Maximum number of cred roaming_consortiums"""
+ id = dev[0].add_cred()
+ consortium = (36*",ffffff")[1:]
+ if "OK" not in dev[0].request('SET_CRED %d roaming_consortiums "%s"' % (id, consortium)):
+ raise Exception("Maximum number of consortium OIs rejected")
+ consortium = (37*",ffffff")[1:]
+ if "FAIL" not in dev[0].request('SET_CRED %d roaming_consortiums "%s"' % (id, consortium)):
+ raise Exception("Over maximum number of consortium OIs accepted")
+ dev[0].remove_cred(id)
+
def test_ap_hs20_roaming_consortium_invalid(dev, apdev):
"""Hotspot 2.0 connection and invalid roaming consortium ANQP-element"""
bssid = apdev[0]['bssid']
"""Hotspot 2.0 connection and deauthentication request without PMF"""
check_eap_capa(dev[0], "MSCHAPV2")
dev[0].request("SET pmf 0")
- eap_test(dev[0], apdev[0], "21[3:26]", "TTLS", "user")
+ eap_test(dev[0], apdev[0], "21[3:26]", "TTLS", "user", release=1)
dev[0].dump_monitor()
+ id = int(dev[0].get_status_field("id"))
+ dev[0].set_network(id, "ieee80211w", "0")
+ dev[0].request("DISCONNECT")
+ dev[0].wait_disconnected()
+ dev[0].select_network(id, freq=2412)
+ dev[0].wait_connected()
addr = dev[0].own_addr()
hapd = hostapd.Hostapd(apdev[0]['ifname'])
hapd.request("HS20_DEAUTH_REQ " + addr + " 1 120 http://example.com/")
scan_freq="2412", wait_connect=False)
dev[0].flush_scan_cache()
dev[0].connect("osen", proto="OSEN", key_mgmt="OSEN", pairwise="CCMP",
- group="GTK_NOT_USED",
+ group="GTK_NOT_USED CCMP",
eap="WFA-UNAUTH-TLS", identity="osen@example.com",
ca_cert="auth_serv/ca.pem",
scan_freq="2412")
wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
wpas.interface_add("wlan5", drv_params="force_connect_cmd=1")
wpas.connect("osen", proto="OSEN", key_mgmt="OSEN", pairwise="CCMP",
- group="GTK_NOT_USED",
+ group="GTK_NOT_USED CCMP",
eap="WFA-UNAUTH-TLS", identity="osen@example.com",
ca_cert="auth_serv/ca.pem",
scan_freq="2412")
wpas.request("DISCONNECT")
+def test_ap_hs20_osen_single_ssid(dev, apdev):
+ """Hotspot 2.0 OSEN-single-SSID connection"""
+ bssid = apdev[0]['bssid']
+ params = hs20_ap_params()
+ params['wpa_key_mgmt'] = "WPA-EAP OSEN"
+ params['hessid'] = bssid
+ hapd = hostapd.add_ap(apdev[0], params)
+
+ # RSN-OSEN (for OSU)
+ dev[0].connect("test-hs20", proto="OSEN", key_mgmt="OSEN", pairwise="CCMP",
+ group="CCMP GTK_NOT_USED",
+ eap="WFA-UNAUTH-TLS", identity="osen@example.com",
+ ca_cert="auth_serv/ca.pem", ieee80211w='2',
+ scan_freq="2412")
+ # RSN-EAP (for data connection)
+ dev[1].connect("test-hs20", key_mgmt="WPA-EAP", eap="TTLS",
+ identity="hs20-test", password="password",
+ ca_cert="auth_serv/ca.pem", phase2="auth=MSCHAPV2",
+ pairwise="CCMP", group="CCMP",
+ ieee80211w='2', scan_freq="2412")
+
+ res = dev[0].get_bss(apdev[0]['bssid'])['flags']
+ if "[WPA2-EAP+OSEN-CCMP]" not in res:
+ raise Exception("OSEN not reported in BSS")
+ if "[WEP]" in res:
+ raise Exception("WEP reported in BSS")
+ res = dev[0].request("SCAN_RESULTS")
+ if "[WPA2-EAP+OSEN-CCMP]" not in res:
+ raise Exception("OSEN not reported in SCAN_RESULTS")
+
+ hwsim_utils.test_connectivity(dev[1], hapd)
+ hwsim_utils.test_connectivity(dev[0], hapd, broadcast=False)
+ hwsim_utils.test_connectivity(dev[0], hapd, timeout=1,
+ success_expected=False)
+
def test_ap_hs20_network_preference(dev, apdev):
"""Hotspot 2.0 network selection with preferred home network"""
check_eap_capa(dev[0], "MSCHAPV2")
break
icon += base64.b64decode(res)
pos += 1000
- hex = binascii.hexlify(icon)
+ hex = binascii.hexlify(icon).decode()
if not hex.startswith("0009696d6167652f706e677d1d"):
raise Exception("Unexpected beacon binary header: " + hex)
with open('w1fi_logo.png', 'r') as f:
os.remove(dir + "/" + f)
os.rmdir(dir)
+def test_ap_hs20_fetch_osu_single_ssid(dev, apdev):
+ """Hotspot 2.0 OSU provider and single SSID"""
+ bssid = apdev[0]['bssid']
+ params = hs20_ap_params()
+ params['hs20_icon'] = "128:80:zxx:image/png:w1fi_logo:w1fi_logo-no-file.png"
+ params['osu_ssid'] = '"HS 2.0 OSU open"'
+ params['osu_method_list'] = "1"
+ params['osu_friendly_name'] = [ "eng:Test OSU", "fin:Testi-OSU" ]
+ params['osu_nai2'] = "osen@example.com"
+ params['osu_icon'] = "w1fi_logo"
+ params['osu_service_desc'] = [ "eng:Example services", "fin:Esimerkkipalveluja" ]
+ params['osu_server_uri'] = "https://example.com/osu/"
+ params['wpa_key_mgmt'] = "WPA-EAP OSEN"
+ hostapd.add_ap(apdev[0], params)
+
+ dev[0].hs20_enable()
+ dir = "/tmp/osu-fetch"
+ if os.path.isdir(dir):
+ files = [ f for f in os.listdir(dir) if f.startswith("osu-") ]
+ for f in files:
+ os.remove(dir + "/" + f)
+ else:
+ try:
+ os.makedirs(dir)
+ except:
+ pass
+ dev[0].scan_for_bss(bssid, freq="2412")
+ try:
+ dev[0].request("SET osu_dir " + dir)
+ dev[0].request("FETCH_OSU")
+ ev = dev[0].wait_event(["OSU provider fetch completed"], timeout=30)
+ if ev is None:
+ raise Exception("Timeout on OSU fetch")
+ osu_ssid = False
+ osu_ssid2 = False
+ osu_nai = False
+ osu_nai2 = False
+ with open(os.path.join(dir, "osu-providers.txt"), "r") as f:
+ for l in f.readlines():
+ logger.info(l.strip())
+ if l.strip() == "osu_ssid=HS 2.0 OSU open":
+ osu_ssid = True
+ if l.strip() == "osu_ssid2=test-hs20":
+ osu_ssid2 = True
+ if l.strip().startswith("osu_nai="):
+ osu_nai = True
+ if l.strip() == "osu_nai2=osen@example.com":
+ osu_nai2 = True
+ if not osu_ssid:
+ raise Exception("osu_ssid not reported")
+ if not osu_ssid2:
+ raise Exception("osu_ssid2 not reported")
+ if osu_nai:
+ raise Exception("osu_nai reported unexpectedly")
+ if not osu_nai2:
+ raise Exception("osu_nai2 not reported")
+ finally:
+ files = [ f for f in os.listdir(dir) if f.startswith("osu-") ]
+ for f in files:
+ os.remove(dir + "/" + f)
+ os.rmdir(dir)
+
+def test_ap_hs20_fetch_osu_single_ssid2(dev, apdev):
+ """Hotspot 2.0 OSU provider and single SSID (two OSU providers)"""
+ bssid = apdev[0]['bssid']
+ params = hs20_ap_params()
+ params['hs20_icon'] = "128:80:zxx:image/png:w1fi_logo:w1fi_logo-no-file.png"
+ params['osu_ssid'] = '"HS 2.0 OSU open"'
+ params['osu_method_list'] = "1"
+ params['osu_friendly_name'] = [ "eng:Test OSU", "fin:Testi-OSU" ]
+ params['osu_nai2'] = "osen@example.com"
+ params['osu_icon'] = "w1fi_logo"
+ params['osu_service_desc'] = [ "eng:Example services", "fin:Esimerkkipalveluja" ]
+ params['osu_server_uri'] = "https://example.com/osu/"
+ params['wpa_key_mgmt'] = "WPA-EAP OSEN"
+ hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
+
+ hapd.set('osu_server_uri', 'https://another.example.com/osu/')
+ hapd.set('osu_method_list', "1")
+ hapd.set('osu_nai2', "osen@another.example.com")
+ hapd.enable()
+
+ dev[0].hs20_enable()
+ dir = "/tmp/osu-fetch"
+ if os.path.isdir(dir):
+ files = [ f for f in os.listdir(dir) if f.startswith("osu-") ]
+ for f in files:
+ os.remove(dir + "/" + f)
+ else:
+ try:
+ os.makedirs(dir)
+ except:
+ pass
+ dev[0].scan_for_bss(bssid, freq="2412")
+ try:
+ dev[0].request("SET osu_dir " + dir)
+ dev[0].request("FETCH_OSU")
+ ev = dev[0].wait_event(["OSU provider fetch completed"], timeout=30)
+ if ev is None:
+ raise Exception("Timeout on OSU fetch")
+ osu_ssid = False
+ osu_ssid2 = False
+ osu_nai = False
+ osu_nai2 = False
+ osu_nai2b = False
+ with open(os.path.join(dir, "osu-providers.txt"), "r") as f:
+ for l in f.readlines():
+ logger.info(l.strip())
+ if l.strip() == "osu_ssid=HS 2.0 OSU open":
+ osu_ssid = True
+ if l.strip() == "osu_ssid2=test-hs20":
+ osu_ssid2 = True
+ if l.strip().startswith("osu_nai="):
+ osu_nai = True
+ if l.strip() == "osu_nai2=osen@example.com":
+ osu_nai2 = True
+ if l.strip() == "osu_nai2=osen@another.example.com":
+ osu_nai2b = True
+ if not osu_ssid:
+ raise Exception("osu_ssid not reported")
+ if not osu_ssid2:
+ raise Exception("osu_ssid2 not reported")
+ if osu_nai:
+ raise Exception("osu_nai reported unexpectedly")
+ if not osu_nai2:
+ raise Exception("osu_nai2 not reported")
+ if not osu_nai2b:
+ raise Exception("osu_nai2b not reported")
+ finally:
+ files = [ f for f in os.listdir(dir) if f.startswith("osu-") ]
+ for f in files:
+ os.remove(dir + "/" + f)
+ os.rmdir(dir)
+
def get_icon(dev, bssid, iconname):
icon = ""
pos = 0
bss = dev[0].get_bss(bssid)
if "hs20_operator_icon_metadata" not in bss:
raise Exception("hs20_operator_icon_metadata missing from BSS entry")
- if bss["hs20_operator_icon_metadata"] != binascii.hexlify(value):
- print binascii.hexlify(value)
+ if bss["hs20_operator_icon_metadata"] != binascii.hexlify(value).decode():
+ print(binascii.hexlify(value))
raise Exception("Unexpected hs20_operator_icon_metadata value: " +
bss["hs20_operator_icon_metadata"])
params['r1_key_holder'] = "000102030405"
params["mobility_domain"] = "a1b2"
params["reassociation_deadline"] = "1000"
- hostapd.add_ap(apdev[0], params)
+ hapd = hostapd.add_ap(apdev[0], params)
dev[0].hs20_enable()
id = dev[0].add_cred_values({ 'realm': "example.com",
'update_identifier': "1234" })
interworking_select(dev[0], bssid, "home", freq="2412")
interworking_connect(dev[0], bssid, "TTLS")
+ dev[0].dump_monitor()
+ key_mgmt = dev[0].get_status_field("key_mgmt")
+ if key_mgmt != "FT-EAP":
+ raise Exception("Unexpected key_mgmt: " + key_mgmt)
+ # speed up testing by avoiding unnecessary scanning of other channels
+ nid = dev[0].get_status_field("id")
+ dev[0].set_network(nid, "scan_freq", "2412")
+
+ params = hs20_ap_params()
+ hapd2 = hostapd.add_ap(apdev[1], params)
+
+ hapd.disable()
+ ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
+ if ev is None:
+ raise Exception("Disconnection not reported")
+ ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=5)
+ if ev is None:
+ raise Exception("Connection to AP2 not reported")
+ key_mgmt = dev[0].get_status_field("key_mgmt")
+ if key_mgmt != "WPA2/IEEE 802.1X/EAP":
+ raise Exception("Unexpected key_mgmt: " + key_mgmt)
def test_ap_hs20_remediation_sql(dev, apdev, params):
"""Hotspot 2.0 connection and remediation required using SQLite for user DB"""
os.remove(dbfile)
dev[0].request("SET pmf 0")
+def test_ap_hs20_sim_provisioning(dev, apdev, params):
+ """Hotspot 2.0 AAA server behavior for SIM provisioning"""
+ check_eap_capa(dev[0], "SIM")
+ try:
+ import sqlite3
+ except ImportError:
+ raise HwsimSkip("No sqlite3 module available")
+ dbfile = os.path.join(params['logdir'], "ap_hs20_sim_provisioning-eap-user.db")
+ try:
+ os.remove(dbfile)
+ except:
+ pass
+ con = sqlite3.connect(dbfile)
+ with con:
+ cur = con.cursor()
+ cur.execute("CREATE TABLE users(identity TEXT PRIMARY KEY, methods TEXT, password TEXT, remediation TEXT, phase2 INTEGER, last_msk TEXT)")
+ cur.execute("CREATE TABLE wildcards(identity TEXT PRIMARY KEY, methods TEXT)")
+ cur.execute("INSERT INTO wildcards(identity,methods) VALUES ('1','SIM')")
+ cur.execute("CREATE TABLE authlog(timestamp TEXT, session TEXT, nas_ip TEXT, username TEXT, note TEXT)")
+ cur.execute("CREATE TABLE current_sessions(mac_addr TEXT PRIMARY KEY, identity TEXT, start_time TEXT, nas TEXT, hs20_t_c_filtering BOOLEAN, waiting_coa_ack BOOLEAN, coa_ack_received BOOLEAN)")
+
+ try:
+ params = { "ssid": "as", "beacon_int": "2000",
+ "radius_server_clients": "auth_serv/radius_clients.conf",
+ "radius_server_auth_port": '18128',
+ "eap_server": "1",
+ "eap_user_file": "sqlite:" + dbfile,
+ "eap_sim_db": "unix:/tmp/hlr_auc_gw.sock",
+ "ca_cert": "auth_serv/ca.pem",
+ "server_cert": "auth_serv/server.pem",
+ "private_key": "auth_serv/server.key",
+ "hs20_sim_provisioning_url":
+ "https://example.org/?hotspot2dot0-mobile-identifier-hash=",
+ "subscr_remediation_method": "1" }
+ hostapd.add_ap(apdev[1], params)
+
+ bssid = apdev[0]['bssid']
+ params = hs20_ap_params()
+ params['auth_server_port'] = "18128"
+ hostapd.add_ap(apdev[0], params)
+
+ dev[0].request("SET pmf 1")
+ dev[0].hs20_enable()
+ dev[0].connect("test-hs20", proto="RSN", key_mgmt="WPA-EAP", eap="SIM",
+ ieee80211w="1",
+ identity="1232010000000000",
+ password="90dca4eda45b53cf0f12d7c9c3bc6a89:cb9cccc4b9258e6dca4760379fb82581",
+ scan_freq="2412", update_identifier="54321")
+ ev = dev[0].wait_event(["HS20-SUBSCRIPTION-REMEDIATION"], timeout=0.5)
+ if ev is not None:
+ raise Exception("Unexpected subscription remediation notice")
+ dev[0].request("REMOVE_NETWORK all")
+ dev[0].wait_disconnected()
+ dev[0].dump_monitor()
+
+ dev[0].connect("test-hs20", proto="RSN", key_mgmt="WPA-EAP", eap="SIM",
+ ieee80211w="1",
+ identity="1232010000000000",
+ password="90dca4eda45b53cf0f12d7c9c3bc6a89:cb9cccc4b9258e6dca4760379fb82581",
+ scan_freq="2412", update_identifier="0")
+ ev = dev[0].wait_event(["HS20-SUBSCRIPTION-REMEDIATION"], timeout=5)
+ if ev is None:
+ raise Exception("Timeout on subscription remediation notice")
+ if " 1 https://example.org/?hotspot2dot0-mobile-identifier-hash=" not in ev:
+ raise Exception("Unexpected subscription remediation event contents: " + ev)
+ id_hash = ev.split(' ')[2].split('=')[1]
+
+ with con:
+ cur = con.cursor()
+ cur.execute("SELECT * from authlog")
+ rows = cur.fetchall()
+ if len(rows) < 1:
+ raise Exception("No authlog entries")
+
+ with con:
+ cur = con.cursor()
+ cur.execute("SELECT * from sim_provisioning")
+ rows = cur.fetchall()
+ if len(rows) != 1:
+ raise Exeception("Unexpected number of rows in sim_provisioning (%d; expected %d)" % (len(rows), 1))
+ logger.info("sim_provisioning: " + str(rows))
+ if len(rows[0][0]) != 32:
+ raise Exception("Unexpected mobile_identifier_hash length in DB")
+ if rows[0][1] != "232010000000000":
+ raise Exception("Unexpected IMSI in DB")
+ if rows[0][2] != dev[0].own_addr():
+ raise Exception("Unexpected MAC address in DB")
+ if rows[0][0] != id_hash:
+ raise Exception("hotspot2dot0-mobile-identifier-hash mismatch")
+ finally:
+ dev[0].request("SET pmf 0")
+
def test_ap_hs20_external_selection(dev, apdev):
"""Hotspot 2.0 connection using external network selection and creation"""
check_eap_capa(dev[0], "MSCHAPV2")
dev[0].hs20_enable()
dev[0].connect("test-hs20", proto="RSN", key_mgmt="WPA-EAP", eap="TTLS",
+ ieee80211w="1",
identity="hs20-test", password="password",
ca_cert="auth_serv/ca.pem", phase2="auth=MSCHAPV2",
scan_freq="2412", update_identifier="54321",
roaming_consortium_selection="1020304050")
- if dev[0].get_status_field("hs20") != "2":
+ if dev[0].get_status_field("hs20") != "3":
raise Exception("Unexpected hs20 indication")
network_id = dev[0].get_status_field("id")
sel = dev[0].get_network(network_id, "roaming_consortium_selection")
pkt = build_ns(src_ll=addr0, ip_src="aaaa:bbbb:cccc::2",
ip_dst="ff02::1:ff00:2", target="aaaa:bbbb:cccc::2",
opt=src_ll_opt0)
- if "OK" not in dev[0].request("DATA_TEST_FRAME " + binascii.hexlify(pkt)):
+ if "OK" not in dev[0].request("DATA_TEST_FRAME " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
pkt = build_ns(src_ll=addr1, ip_src="aaaa:bbbb:dddd::2",
ip_dst="ff02::1:ff00:2", target="aaaa:bbbb:dddd::2",
opt=src_ll_opt1)
- if "OK" not in dev[1].request("DATA_TEST_FRAME " + binascii.hexlify(pkt)):
+ if "OK" not in dev[1].request("DATA_TEST_FRAME " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
pkt = build_ns(src_ll=addr1, ip_src="aaaa:bbbb:eeee::2",
ip_dst="ff02::1:ff00:2", target="aaaa:bbbb:eeee::2",
opt=src_ll_opt1)
- if "OK" not in dev[1].request("DATA_TEST_FRAME " + binascii.hexlify(pkt)):
+ if "OK" not in dev[1].request("DATA_TEST_FRAME " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
matches = get_permanent_neighbors("ap-br0")
pkt = build_ns(src_ll=addr0, ip_src="aaaa:bbbb:cccc::2",
ip_dst="ff02::1:ff00:2", target="aaaa:bbbb:cccc::2",
opt=src_ll_opt0)
- if "OK" not in dev[0].request("DATA_TEST_FRAME " + binascii.hexlify(pkt)):
+ if "OK" not in dev[0].request("DATA_TEST_FRAME " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
pkt = build_ra(src_ll=apdev[0]['bssid'], ip_src="aaaa:bbbb:cccc::33",
ip_dst="ff01::1")
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
pkt = build_na(src_ll=apdev[0]['bssid'], ip_src="aaaa:bbbb:cccc::44",
ip_dst="ff01::1", target="aaaa:bbbb:cccc::55")
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
pkt = build_dhcp_ack(dst_ll="ff:ff:ff:ff:ff:ff", src_ll=bssid,
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="192.168.1.123", chaddr=addr0)
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
# another copy for additional code coverage
pkt = build_dhcp_ack(dst_ll=addr0, src_ll=bssid,
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="192.168.1.123", chaddr=addr0)
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
matches = get_permanent_neighbors("ap-br0")
def ip_checksum(buf):
sum = 0
if len(buf) & 0x01:
- buf += '\0x00'
+ buf += '\x00'
for i in range(0, len(buf), 2):
val, = struct.unpack('H', buf[i:i+2])
sum += val
pkt = build_ns(src_ll=src_ll, ip_src=ip_src, ip_dst=ip_dst, target=target,
opt=opt)
- if "OK" not in dev.request(cmd + binascii.hexlify(pkt)):
+ if "OK" not in dev.request(cmd + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
def build_na(src_ll, ip_src, ip_dst, target, opt=None, flags=0):
pkt = build_na(src_ll=src_ll, ip_src=ip_src, ip_dst=ip_dst, target=target,
opt=opt)
- if "OK" not in dev.request(cmd + binascii.hexlify(pkt)):
+ if "OK" not in dev.request(cmd + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
def build_dhcp_ack(dst_ll, src_ll, ip_src, ip_dst, yiaddr, chaddr,
subnet_mask="255.255.255.0", truncated_opt=False,
- wrong_magic=False, force_tot_len=None, no_dhcp=False):
+ wrong_magic=False, force_tot_len=None, no_dhcp=False,
+ udp_checksum=True):
_dst_ll = binascii.unhexlify(dst_ll.replace(':',''))
_src_ll = binascii.unhexlify(src_ll.replace(':',''))
proto = '\x08\x00'
payload = struct.pack('>BBBBL3BB', 2, 1, 6, 0, 12345, 0, 0, 0, 0)
payload += _ciaddr + _yiaddr + _siaddr + _giaddr + _chaddr + 192*'\x00'
- udp = struct.pack('>HHHH', 67, 68, 8 + len(payload), 0) + payload
+ if udp_checksum:
+ pseudohdr = _ip_src + _ip_dst + struct.pack('>BBH', 0, 17,
+ 8 + len(payload))
+ udphdr = struct.pack('>HHHH', 67, 68, 8 + len(payload), 0)
+ checksum, = struct.unpack('>H', ip_checksum(pseudohdr + udphdr + payload))
+ else:
+ checksum = 0
+ udp = struct.pack('>HHHH', 67, 68, 8 + len(payload), checksum) + payload
if force_tot_len:
tot_len = force_tot_len
pkt = build_arp(dst_ll=dst_ll, src_ll=src_ll, opcode=opcode,
sender_mac=sender_mac, sender_ip=sender_ip,
target_mac=target_mac, target_ip=target_ip)
- if "OK" not in dev.request(cmd + binascii.hexlify(pkt)):
+ if "OK" not in dev.request(cmd + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
def get_permanent_neighbors(ifname):
cmd = subprocess.Popen(['ip', 'nei'], stdout=subprocess.PIPE)
- res = cmd.stdout.read()
+ res = cmd.stdout.read().decode()
cmd.stdout.close()
return [ line for line in res.splitlines() if "PERMANENT" in line and ifname in line ]
time.sleep(0.1)
brcmd = subprocess.Popen(['brctl', 'show'], stdout=subprocess.PIPE)
- res = brcmd.stdout.read()
+ res = brcmd.stdout.read().decode()
brcmd.stdout.close()
logger.info("Bridge setup: " + res)
brcmd = subprocess.Popen(['brctl', 'showstp', 'ap-br0'],
stdout=subprocess.PIPE)
- res = brcmd.stdout.read()
+ res = brcmd.stdout.read().decode()
brcmd.stdout.close()
logger.info("Bridge showstp: " + res)
pkt = build_dhcp_ack(dst_ll="ff:ff:ff:ff:ff:ff", src_ll=bssid,
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="192.168.1.124", chaddr=addr0)
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
# Change address and verify unicast
pkt = build_dhcp_ack(dst_ll=addr0, src_ll=bssid,
ip_src="192.168.1.1", ip_dst="255.255.255.255",
- yiaddr="192.168.1.123", chaddr=addr0)
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ yiaddr="192.168.1.123", chaddr=addr0,
+ udp_checksum=False)
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
# Not-associated client MAC address
pkt = build_dhcp_ack(dst_ll="ff:ff:ff:ff:ff:ff", src_ll=bssid,
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="192.168.1.125", chaddr="22:33:44:55:66:77")
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
# No IP address
pkt = build_dhcp_ack(dst_ll=addr1, src_ll=bssid,
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="0.0.0.0", chaddr=addr1)
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
# Zero subnet mask
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="192.168.1.126", chaddr=addr1,
subnet_mask="0.0.0.0")
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
# Truncated option
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="192.168.1.127", chaddr=addr1,
truncated_opt=True)
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
# Wrong magic
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="192.168.1.128", chaddr=addr1,
wrong_magic=True)
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
# Wrong IPv4 total length
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="192.168.1.129", chaddr=addr1,
force_tot_len=1000)
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
# BOOTP
ip_src="192.168.1.1", ip_dst="255.255.255.255",
yiaddr="192.168.1.129", chaddr=addr1,
no_dhcp=True)
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
macs = get_bridge_macs("ap-br0")
try:
hwsim_utils.test_connectivity_iface(dev[0], hapd, "ap-br0")
- except Exception, e:
+ except Exception as e:
logger.info("test_connectibity_iface failed: " + str(e))
raise HwsimSkip("Assume kernel did not have the required patches for proxyarp")
hwsim_utils.test_connectivity_iface(dev[1], hapd, "ap-br0")
if ebtables:
cmd = subprocess.Popen(['ebtables', '-L', '--Lc'],
stdout=subprocess.PIPE)
- res = cmd.stdout.read()
+ res = cmd.stdout.read().decode()
cmd.stdout.close()
logger.info("ebtables results:\n" + res)
time.sleep(0.1)
brcmd = subprocess.Popen(['brctl', 'show'], stdout=subprocess.PIPE)
- res = brcmd.stdout.read()
+ res = brcmd.stdout.read().decode()
brcmd.stdout.close()
logger.info("Bridge setup: " + res)
brcmd = subprocess.Popen(['brctl', 'showstp', 'ap-br0'],
stdout=subprocess.PIPE)
- res = brcmd.stdout.read()
+ res = brcmd.stdout.read().decode()
brcmd.stdout.close()
logger.info("Bridge showstp: " + res)
try:
hwsim_utils.test_connectivity_iface(dev[0], hapd, "ap-br0")
- except Exception, e:
+ except Exception as e:
logger.info("test_connectibity_iface failed: " + str(e))
raise HwsimSkip("Assume kernel did not have the required patches for proxyarp")
hwsim_utils.test_connectivity_iface(dev[1], hapd, "ap-br0")
if ebtables:
cmd = subprocess.Popen(['ebtables', '-L', '--Lc'],
stdout=subprocess.PIPE)
- res = cmd.stdout.read()
+ res = cmd.stdout.read().decode()
cmd.stdout.close()
logger.info("ebtables results:\n" + res)
pkt = build_ra(src_ll=apdev[0]['bssid'], ip_src="aaaa:bbbb:cccc::33",
ip_dst="ff01::1")
with fail_test(hapd, 1, "x_snoop_mcast_to_ucast_convert_send"):
- if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt)):
+ if "OK" not in hapd.request("DATA_TEST_FRAME ifname=ap-br0 " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
wait_fail_trigger(dev[0], "GET_FAIL")
pkt = build_ns(src_ll=addr0, ip_src="aaaa:bbbb:cccc::2",
ip_dst="ff02::1:ff00:2", target="aaaa:bbbb:cccc::2",
opt=src_ll_opt0)
- if "OK" not in dev[0].request("DATA_TEST_FRAME " + binascii.hexlify(pkt)):
+ if "OK" not in dev[0].request("DATA_TEST_FRAME " + binascii.hexlify(pkt).decode()):
raise Exception("DATA_TEST_FRAME failed")
wait_fail_trigger(dev[0], "GET_ALLOC_FAIL")
'realm': "example.com",
'username': "user",
'eap': "PEAP" })
- dev[0].set_cred(id, "password", "ext:password");
+ dev[0].set_cred(id, "password", "ext:password")
interworking_select(dev[0], bssid, "home", freq=2412)
dev[0].dump_monitor()
dev[0].request("NOTE wpa_config_set(password)")
params['wpa'] = "3"
params['wpa_pairwise'] = "TKIP CCMP"
params['rsn_pairwise'] = "CCMP"
+ params['ieee80211w'] = "1"
#params['vendor_elements'] = 'dd07506f9a10140000'
params['vendor_elements'] = 'dd04506f9a10'
hostapd.add_ap(apdev[0], params)
dev[2].hs20_enable()
dev[2].scan_for_bss(bssid, freq="2412")
dev[2].connect("test-hs20-fake", key_mgmt="WPA-EAP", eap="TTLS",
+ ieee80211w="1",
proto="RSN", pairwise="CCMP",
identity="hs20-test", password="password",
ca_cert="auth_serv/ca.pem", phase2="auth=MSCHAPV2",
params['hessid'] = bssid
params['hs20_t_c_filename'] = 'terms-and-conditions'
params['hs20_t_c_timestamp'] = '123456789'
- params['hs20_t_c_server_url'] = 'https://example.com/t_and_c?addr=@1@&ap=123'
+
hostapd.add_ap(apdev[0], params)
dev[0].hs20_enable()
url = "https://example.com/t_and_c?addr=%s&ap=123" % dev[0].own_addr()
if url not in ev:
raise Exception("Unexpected URL: " + ev)
+
+def test_ap_hs20_terms_and_conditions_coa(dev, apdev):
+ """Hotspot 2.0 Terms and Conditions signaling - CoA"""
+ try:
+ import pyrad.client
+ import pyrad.packet
+ import pyrad.dictionary
+ import radius_das
+ except ImportError:
+ raise HwsimSkip("No pyrad modules available")
+
+ check_eap_capa(dev[0], "MSCHAPV2")
+ bssid = apdev[0]['bssid']
+ params = hs20_ap_params()
+ params['hessid'] = bssid
+ params['hs20_t_c_filename'] = 'terms-and-conditions'
+ params['hs20_t_c_timestamp'] = '123456789'
+ params['own_ip_addr'] = "127.0.0.1"
+ params['radius_das_port'] = "3799"
+ params['radius_das_client'] = "127.0.0.1 secret"
+ params['radius_das_require_event_timestamp'] = "1"
+ hapd = hostapd.add_ap(apdev[0], params)
+
+ dev[0].hs20_enable()
+ dev[0].connect("test-hs20", proto="RSN", key_mgmt="WPA-EAP", eap="TTLS",
+ identity="hs20-t-c-test", password="password",
+ ca_cert="auth_serv/ca.pem", phase2="auth=MSCHAPV2",
+ ieee80211w='2', scan_freq="2412")
+
+ ev = hapd.wait_event(["HS20-T-C-FILTERING-ADD"], timeout=5)
+ if ev is None:
+ raise Exception("Terms and Conditions filtering not enabled")
+ if ev.split(' ')[1] != dev[0].own_addr():
+ raise Exception("Unexpected STA address for filtering: " + ev)
+
+ ev = dev[0].wait_event(["HS20-T-C-ACCEPTANCE"], timeout=5)
+ if ev is None:
+ raise Exception("Terms and Conditions Acceptance notification not received")
+ url = "https://example.com/t_and_c?addr=%s&ap=123" % dev[0].own_addr()
+ if url not in ev:
+ raise Exception("Unexpected URL: " + ev)
+
+ dict = pyrad.dictionary.Dictionary("dictionary.radius")
+
+ srv = pyrad.client.Client(server="127.0.0.1", acctport=3799,
+ secret="secret", dict=dict)
+ srv.retries = 1
+ srv.timeout = 1
+
+ sta = hapd.get_sta(dev[0].own_addr())
+ multi_sess_id = sta['authMultiSessionId']
+
+ logger.info("CoA-Request with matching Acct-Session-Id")
+ vsa = binascii.unhexlify('00009f68090600000000')
+ req = radius_das.CoAPacket(dict=dict, secret="secret",
+ NAS_IP_Address="127.0.0.1",
+ Acct_Multi_Session_Id=multi_sess_id,
+ Chargeable_User_Identity="hs20-cui",
+ Event_Timestamp=int(time.time()),
+ Vendor_Specific=vsa)
+ reply = srv.SendPacket(req)
+ logger.debug("RADIUS response from hostapd")
+ for i in list(reply.keys()):
+ logger.debug("%s: %s" % (i, reply[i]))
+ if reply.code != pyrad.packet.CoAACK:
+ raise Exception("CoA-Request failed")
+
+ ev = hapd.wait_event(["HS20-T-C-FILTERING-REMOVE"], timeout=5)
+ if ev is None:
+ raise Exception("Terms and Conditions filtering not disabled")
+ if ev.split(' ')[1] != dev[0].own_addr():
+ raise Exception("Unexpected STA address for filtering: " + ev)
+
+def test_ap_hs20_terms_and_conditions_sql(dev, apdev, params):
+ """Hotspot 2.0 Terms and Conditions using SQLite for user DB"""
+ addr = dev[0].own_addr()
+ run_ap_hs20_terms_and_conditions_sql(dev, apdev, params,
+ "https://example.com/t_and_c?addr=@1@&ap=123",
+ "https://example.com/t_and_c?addr=" + addr + "&ap=123")
+
+def test_ap_hs20_terms_and_conditions_sql2(dev, apdev, params):
+ """Hotspot 2.0 Terms and Conditions using SQLite for user DB"""
+ addr = dev[0].own_addr()
+ run_ap_hs20_terms_and_conditions_sql(dev, apdev, params,
+ "https://example.com/t_and_c?addr=@1@",
+ "https://example.com/t_and_c?addr=" + addr)
+
+def run_ap_hs20_terms_and_conditions_sql(dev, apdev, params, url_template,
+ url_expected):
+ check_eap_capa(dev[0], "MSCHAPV2")
+ try:
+ import sqlite3
+ except ImportError:
+ raise HwsimSkip("No sqlite3 module available")
+ dbfile = os.path.join(params['logdir'], "eap-user.db")
+ try:
+ os.remove(dbfile)
+ except:
+ pass
+ con = sqlite3.connect(dbfile)
+ with con:
+ cur = con.cursor()
+ cur.execute("CREATE TABLE users(identity TEXT PRIMARY KEY, methods TEXT, password TEXT, remediation TEXT, phase2 INTEGER, t_c_timestamp INTEGER)")
+ cur.execute("CREATE TABLE wildcards(identity TEXT PRIMARY KEY, methods TEXT)")
+ cur.execute("INSERT INTO users(identity,methods,password,phase2) VALUES ('user-mschapv2','TTLS-MSCHAPV2','password',1)")
+ cur.execute("INSERT INTO wildcards(identity,methods) VALUES ('','TTLS,TLS')")
+ cur.execute("CREATE TABLE authlog(timestamp TEXT, session TEXT, nas_ip TEXT, username TEXT, note TEXT)")
+ cur.execute("CREATE TABLE pending_tc(mac_addr TEXT PRIMARY KEY, identity TEXT)")
+ cur.execute("CREATE TABLE current_sessions(mac_addr TEXT PRIMARY KEY, identity TEXT, start_time TEXT, nas TEXT, hs20_t_c_filtering BOOLEAN, waiting_coa_ack BOOLEAN, coa_ack_received BOOLEAN)")
+
+
+ try:
+ params = { "ssid": "as", "beacon_int": "2000",
+ "radius_server_clients": "auth_serv/radius_clients.conf",
+ "radius_server_auth_port": '18128',
+ "eap_server": "1",
+ "eap_user_file": "sqlite:" + dbfile,
+ "ca_cert": "auth_serv/ca.pem",
+ "server_cert": "auth_serv/server.pem",
+ "private_key": "auth_serv/server.key" }
+ params['hs20_t_c_server_url'] = url_template
+ authsrv = hostapd.add_ap(apdev[1], params)
+
+ bssid = apdev[0]['bssid']
+ params = hs20_ap_params()
+ params['auth_server_port'] = "18128"
+ params['hs20_t_c_filename'] = 'terms-and-conditions'
+ params['hs20_t_c_timestamp'] = '123456789'
+ params['own_ip_addr'] = "127.0.0.1"
+ params['radius_das_port'] = "3799"
+ params['radius_das_client'] = "127.0.0.1 radius"
+ params['radius_das_require_event_timestamp'] = "1"
+ params['disable_pmksa_caching'] = '1'
+ hapd = hostapd.add_ap(apdev[0], params)
+
+ dev[0].request("SET pmf 1")
+ dev[0].hs20_enable()
+ id = dev[0].add_cred_values({ 'realm': "example.com",
+ 'username': "user-mschapv2",
+ 'password': "password",
+ 'ca_cert': "auth_serv/ca.pem" })
+ interworking_select(dev[0], bssid, freq="2412")
+ interworking_connect(dev[0], bssid, "TTLS")
+
+ ev = hapd.wait_event(["HS20-T-C-FILTERING-ADD"], timeout=5)
+ if ev is None:
+ raise Exception("Terms and Conditions filtering not enabled")
+ hapd.dump_monitor()
+
+ ev = dev[0].wait_event(["HS20-T-C-ACCEPTANCE"], timeout=5)
+ if ev is None:
+ raise Exception("Terms and Conditions Acceptance notification not received")
+ url = ev.split(' ')[1]
+ if url != url_expected:
+ raise Exception("Unexpected URL delivered to the client: %s (expected %s)" % (url, url_expected))
+ dev[0].dump_monitor()
+
+ with con:
+ cur = con.cursor()
+ cur.execute("SELECT * from current_sessions")
+ rows = cur.fetchall()
+ if len(rows) != 1:
+ raise Exeception("Unexpected number of rows in current_sessions (%d; expected %d)" % (len(rows), 1))
+ logger.info("current_sessions: " + str(rows))
+
+ if "OK" not in authsrv.request("DAC_REQUEST coa %s t_c_clear" % dev[0].own_addr()):
+ raise Exception("DAC_REQUEST failed")
+
+ ev = hapd.wait_event(["HS20-T-C-FILTERING-REMOVE"], timeout=5)
+ if ev is None:
+ raise Exception("Terms and Conditions filtering not disabled")
+ if ev.split(' ')[1] != dev[0].own_addr():
+ raise Exception("Unexpected STA address for filtering: " + ev)
+
+ time.sleep(0.2)
+ with con:
+ cur = con.cursor()
+ cur.execute("SELECT * from current_sessions")
+ rows = cur.fetchall()
+ if len(rows) != 1:
+ raise Exeception("Unexpected number of rows in current_sessions (%d; expected %d)" % (len(rows), 1))
+ logger.info("current_sessions: " + str(rows))
+ if rows[0][4] != 0 or rows[0][5] != 0 or rows[0][6] != 1:
+ raise Exception("Unexpected current_sessions information after CoA-ACK")
+
+ dev[0].request("DISCONNECT")
+ dev[0].wait_disconnected()
+ dev[0].dump_monitor()
+
+ # Simulate T&C server operation on user reading the updated version
+ with con:
+ cur = con.cursor()
+ cur.execute("SELECT identity FROM pending_tc WHERE mac_addr='" +
+ dev[0].own_addr() + "'")
+ rows = cur.fetchall()
+ if len(rows) != 1:
+ raise Exception("No pending_tc entry found")
+ if rows[0][0] != 'user-mschapv2':
+ raise Exception("Unexpected pending_tc identity value")
+
+ cur.execute("UPDATE users SET t_c_timestamp=123456789 WHERE identity='user-mschapv2'")
+
+ dev[0].request("RECONNECT")
+ dev[0].wait_connected()
+
+ ev = hapd.wait_event(["HS20-T-C-FILTERING-ADD"], timeout=0.1)
+ if ev is not None:
+ raise Exception("Terms and Conditions filtering enabled unexpectedly")
+ hapd.dump_monitor()
+
+ ev = dev[0].wait_event(["HS20-T-C-ACCEPTANCE"], timeout=0.1)
+ if ev is not None:
+ raise Exception("Unexpected Terms and Conditions Acceptance notification")
+ dev[0].dump_monitor()
+
+ dev[0].request("DISCONNECT")
+ dev[0].wait_disconnected()
+ dev[0].dump_monitor()
+
+ # New T&C available
+ hapd.set('hs20_t_c_timestamp', '123456790')
+
+ dev[0].request("RECONNECT")
+ dev[0].wait_connected()
+
+ ev = hapd.wait_event(["HS20-T-C-FILTERING-ADD"], timeout=5)
+ if ev is None:
+ raise Exception("Terms and Conditions filtering not enabled")
+ hapd.dump_monitor()
+
+ ev = dev[0].wait_event(["HS20-T-C-ACCEPTANCE"], timeout=5)
+ if ev is None:
+ raise Exception("Terms and Conditions Acceptance notification not received (2)")
+ dev[0].dump_monitor()
+
+ dev[0].request("DISCONNECT")
+ dev[0].wait_disconnected()
+ dev[0].dump_monitor()
+
+ # Simulate T&C server operation on user reading the updated version
+ with con:
+ cur = con.cursor()
+ cur.execute("UPDATE users SET t_c_timestamp=123456790 WHERE identity='user-mschapv2'")
+
+ dev[0].request("RECONNECT")
+ dev[0].wait_connected()
+
+ ev = hapd.wait_event(["HS20-T-C-FILTERING-ADD"], timeout=0.1)
+ if ev is not None:
+ raise Exception("Terms and Conditions filtering enabled unexpectedly")
+ hapd.dump_monitor()
+
+ ev = dev[0].wait_event(["HS20-T-C-ACCEPTANCE"], timeout=0.1)
+ if ev is not None:
+ raise Exception("Unexpected Terms and Conditions Acceptance notification (2)")
+ dev[0].dump_monitor()
+ finally:
+ os.remove(dbfile)
+ dev[0].request("SET pmf 0")
+
+def test_ap_hs20_release_number_1(dev, apdev):
+ """Hotspot 2.0 with AP claiming support for Release 1"""
+ run_ap_hs20_release_number(dev, apdev, 1)
+
+def test_ap_hs20_release_number_2(dev, apdev):
+ """Hotspot 2.0 with AP claiming support for Release 2"""
+ run_ap_hs20_release_number(dev, apdev, 2)
+
+def test_ap_hs20_release_number_3(dev, apdev):
+ """Hotspot 2.0 with AP claiming support for Release 3"""
+ run_ap_hs20_release_number(dev, apdev, 3)
+
+def run_ap_hs20_release_number(dev, apdev, release):
+ check_eap_capa(dev[0], "MSCHAPV2")
+ eap_test(dev[0], apdev[0], "21[3:26][6:7][99:99]", "TTLS", "user",
+ release=release)
+ rel = dev[0].get_status_field('hs20')
+ if rel != str(release):
+ raise Exception("Unexpected release number indicated: " + rel)
+
+def test_ap_hs20_missing_pmf(dev, apdev):
+ """Hotspot 2.0 connection attempt without PMF"""
+ check_eap_capa(dev[0], "MSCHAPV2")
+ bssid = apdev[0]['bssid']
+ params = hs20_ap_params()
+ params['hessid'] = bssid
+ params['disable_dgaf'] = '1'
+ hostapd.add_ap(apdev[0], params)
+
+ dev[0].hs20_enable()
+ dev[0].connect("test-hs20", proto="RSN", key_mgmt="WPA-EAP", eap="TTLS",
+ ieee80211w="0",
+ identity="hs20-test", password="password",
+ ca_cert="auth_serv/ca.pem", phase2="auth=MSCHAPV2",
+ scan_freq="2412", update_identifier="54321",
+ roaming_consortium_selection="1020304050",
+ wait_connect=False)
+ ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"], timeout=10)
+ dev[0].request("DISCONNECT")
+ if ev is None:
+ raise Exception("Association rejection not reported")
+ if "status_code=31" not in ev:
+ raise Exception("Unexpected rejection reason: " + ev)
projects / thirdparty / hostap.git / blobdiff