123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116 |
- # WPA2-Enterprise PMKSA caching tests
- # Copyright (c) 2013-2014, Jouni Malinen <j@w1.fi>
- #
- # This software may be distributed under the terms of the BSD license.
- # See README for more details.
- import binascii
- import logging
- logger = logging.getLogger()
- import socket
- import struct
- import subprocess
- import time
- import hostapd
- import hwsim_utils
- from wpasupplicant import WpaSupplicant
- from utils import alloc_fail, HwsimSkip, wait_fail_trigger
- from test_ap_eap import eap_connect
- def test_pmksa_cache_on_roam_back(dev, apdev):
- """PMKSA cache to skip EAP on reassociation back to same AP"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- pmksa = dev[0].get_pmksa(bssid)
- if pmksa is None:
- raise Exception("No PMKSA cache entry created")
- if pmksa['opportunistic'] != '0':
- raise Exception("Unexpected opportunistic PMKSA cache entry")
- hostapd.add_ap(apdev[1], params)
- bssid2 = apdev[1]['bssid']
- dev[0].dump_monitor()
- logger.info("Roam to AP2")
- # It can take some time for the second AP to become ready to reply to Probe
- # Request frames especially under heavy CPU load, so allow couple of rounds
- # of scanning to avoid reporting errors incorrectly just because of scans
- # not having seen the target AP.
- for i in range(0, 10):
- dev[0].scan(freq="2412")
- if dev[0].get_bss(bssid2) is not None:
- break
- logger.info("Scan again to find target AP")
- dev[0].request("ROAM " + bssid2)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
- if ev is None:
- raise Exception("EAP success timed out")
- dev[0].wait_connected(timeout=10, error="Roaming timed out")
- pmksa2 = dev[0].get_pmksa(bssid2)
- if pmksa2 is None:
- raise Exception("No PMKSA cache entry found")
- if pmksa2['opportunistic'] != '0':
- raise Exception("Unexpected opportunistic PMKSA cache entry")
- dev[0].dump_monitor()
- logger.info("Roam back to AP1")
- dev[0].scan(freq="2412")
- dev[0].request("ROAM " + bssid)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- pmksa1b = dev[0].get_pmksa(bssid)
- if pmksa1b is None:
- raise Exception("No PMKSA cache entry found")
- if pmksa['pmkid'] != pmksa1b['pmkid']:
- raise Exception("Unexpected PMKID change for AP1")
- dev[0].dump_monitor()
- if "FAIL" in dev[0].request("PMKSA_FLUSH"):
- raise Exception("PMKSA_FLUSH failed")
- if dev[0].get_pmksa(bssid) is not None or dev[0].get_pmksa(bssid2) is not None:
- raise Exception("PMKSA_FLUSH did not remove PMKSA entries")
- dev[0].wait_disconnected(timeout=5)
- dev[0].wait_connected(timeout=15, error="Reconnection timed out")
- def test_pmksa_cache_and_reauth(dev, apdev):
- """PMKSA caching and EAPOL reauthentication"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- hostapd.add_ap(apdev[1], params)
- bssid2 = apdev[1]['bssid']
- dev[0].dump_monitor()
- logger.info("Roam to AP2")
- # It can take some time for the second AP to become ready to reply to Probe
- # Request frames especially under heavy CPU load, so allow couple of rounds
- # of scanning to avoid reporting errors incorrectly just because of scans
- # not having seen the target AP.
- for i in range(0, 10):
- dev[0].scan(freq="2412")
- if dev[0].get_bss(bssid2) is not None:
- break
- logger.info("Scan again to find target AP")
- dev[0].request("ROAM " + bssid2)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
- if ev is None:
- raise Exception("EAP success timed out")
- dev[0].wait_connected(timeout=10, error="Roaming timed out")
- dev[0].dump_monitor()
- logger.info("Roam back to AP1")
- dev[0].scan(freq="2412")
- dev[0].request("ROAM " + bssid)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- # Verify EAPOL reauthentication after PMKSA caching
- hapd.request("EAPOL_REAUTH " + dev[0].own_addr())
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED"], timeout=5)
- if ev is None:
- raise Exception("EAP authentication did not start")
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=5)
- if ev is None:
- raise Exception("EAP authentication did not succeed")
- def test_pmksa_cache_opportunistic_only_on_sta(dev, apdev):
- """Opportunistic PMKSA caching enabled only on station"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef", okc=True,
- scan_freq="2412")
- pmksa = dev[0].get_pmksa(bssid)
- if pmksa is None:
- raise Exception("No PMKSA cache entry created")
- if pmksa['opportunistic'] != '0':
- raise Exception("Unexpected opportunistic PMKSA cache entry")
- hostapd.add_ap(apdev[1], params)
- bssid2 = apdev[1]['bssid']
- dev[0].dump_monitor()
- logger.info("Roam to AP2")
- dev[0].scan(freq="2412")
- dev[0].request("ROAM " + bssid2)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
- if ev is None:
- raise Exception("EAP success timed out")
- dev[0].wait_connected(timeout=10, error="Roaming timed out")
- pmksa2 = dev[0].get_pmksa(bssid2)
- if pmksa2 is None:
- raise Exception("No PMKSA cache entry found")
- if pmksa2['opportunistic'] != '0':
- raise Exception("Unexpected opportunistic PMKSA cache entry")
- dev[0].dump_monitor()
- logger.info("Roam back to AP1")
- dev[0].scan(freq="2412")
- dev[0].request("ROAM " + bssid)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- pmksa1b = dev[0].get_pmksa(bssid)
- if pmksa1b is None:
- raise Exception("No PMKSA cache entry found")
- if pmksa['pmkid'] != pmksa1b['pmkid']:
- raise Exception("Unexpected PMKID change for AP1")
- def test_pmksa_cache_opportunistic(dev, apdev):
- """Opportunistic PMKSA caching"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- params['okc'] = "1"
- hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef", okc=True,
- scan_freq="2412")
- pmksa = dev[0].get_pmksa(bssid)
- if pmksa is None:
- raise Exception("No PMKSA cache entry created")
- if pmksa['opportunistic'] != '0':
- raise Exception("Unexpected opportunistic PMKSA cache entry")
- hostapd.add_ap(apdev[1], params)
- bssid2 = apdev[1]['bssid']
- dev[0].dump_monitor()
- logger.info("Roam to AP2")
- dev[0].scan(freq="2412")
- dev[0].request("ROAM " + bssid2)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- pmksa2 = dev[0].get_pmksa(bssid2)
- if pmksa2 is None:
- raise Exception("No PMKSA cache entry created")
- dev[0].dump_monitor()
- logger.info("Roam back to AP1")
- dev[0].scan(freq="2412")
- dev[0].request("ROAM " + bssid)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- pmksa1b = dev[0].get_pmksa(bssid)
- if pmksa1b is None:
- raise Exception("No PMKSA cache entry found")
- if pmksa['pmkid'] != pmksa1b['pmkid']:
- raise Exception("Unexpected PMKID change for AP1")
- def test_pmksa_cache_opportunistic_connect(dev, apdev):
- """Opportunistic PMKSA caching with connect API"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- params['okc'] = "1"
- hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
- wpas.interface_add("wlan5", drv_params="force_connect_cmd=1")
- wpas.connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef", okc=True,
- scan_freq="2412")
- pmksa = wpas.get_pmksa(bssid)
- if pmksa is None:
- raise Exception("No PMKSA cache entry created")
- if pmksa['opportunistic'] != '0':
- raise Exception("Unexpected opportunistic PMKSA cache entry")
- hostapd.add_ap(apdev[1], params)
- bssid2 = apdev[1]['bssid']
- wpas.dump_monitor()
- logger.info("Roam to AP2")
- wpas.scan_for_bss(bssid2, freq="2412", force_scan=True)
- wpas.request("ROAM " + bssid2)
- ev = wpas.wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- pmksa2 = wpas.get_pmksa(bssid2)
- if pmksa2 is None:
- raise Exception("No PMKSA cache entry created")
- wpas.dump_monitor()
- logger.info("Roam back to AP1")
- wpas.scan(freq="2412")
- wpas.request("ROAM " + bssid)
- ev = wpas.wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- pmksa1b = wpas.get_pmksa(bssid)
- if pmksa1b is None:
- raise Exception("No PMKSA cache entry found")
- if pmksa['pmkid'] != pmksa1b['pmkid']:
- raise Exception("Unexpected PMKID change for AP1")
- def test_pmksa_cache_expiration(dev, apdev):
- """PMKSA cache entry expiration"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- dev[0].request("SET dot11RSNAConfigPMKLifetime 10")
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- pmksa = dev[0].get_pmksa(bssid)
- if pmksa is None:
- raise Exception("No PMKSA cache entry created")
- logger.info("Wait for PMKSA cache entry to expire")
- ev = dev[0].wait_event(["WPA: Key negotiation completed",
- "CTRL-EVENT-DISCONNECTED"], timeout=15)
- if ev is None:
- raise Exception("No EAP reauthentication seen")
- if "CTRL-EVENT-DISCONNECTED" in ev:
- raise Exception("Unexpected disconnection")
- pmksa2 = dev[0].get_pmksa(bssid)
- if pmksa['pmkid'] == pmksa2['pmkid']:
- raise Exception("PMKID did not change")
- hwsim_utils.test_connectivity(dev[0], hapd)
- def test_pmksa_cache_expiration_disconnect(dev, apdev):
- """PMKSA cache entry expiration (disconnect)"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- dev[0].request("SET dot11RSNAConfigPMKLifetime 2")
- dev[0].request("SET dot11RSNAConfigPMKReauthThreshold 100")
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- pmksa = dev[0].get_pmksa(bssid)
- if pmksa is None:
- raise Exception("No PMKSA cache entry created")
- hapd.request("SET auth_server_shared_secret incorrect")
- logger.info("Wait for PMKSA cache entry to expire")
- ev = dev[0].wait_event(["WPA: Key negotiation completed",
- "CTRL-EVENT-DISCONNECTED"], timeout=15)
- if ev is None:
- raise Exception("No EAP reauthentication seen")
- if "CTRL-EVENT-DISCONNECTED" not in ev:
- raise Exception("Missing disconnection")
- hapd.request("SET auth_server_shared_secret radius")
- ev = dev[0].wait_event(["WPA: Key negotiation completed"], timeout=15)
- if ev is None:
- raise Exception("No EAP reauthentication seen")
- pmksa2 = dev[0].get_pmksa(bssid)
- if pmksa['pmkid'] == pmksa2['pmkid']:
- raise Exception("PMKID did not change")
- def test_pmksa_cache_and_cui(dev, apdev):
- """PMKSA cache and Chargeable-User-Identity"""
- params = hostapd.wpa2_eap_params(ssid="cui")
- params['radius_request_cui'] = '1'
- params['acct_server_addr'] = "127.0.0.1"
- params['acct_server_port'] = "1813"
- params['acct_server_shared_secret'] = "radius"
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- dev[0].connect("cui", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk-cui",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- pmksa = dev[0].get_pmksa(bssid)
- if pmksa is None:
- raise Exception("No PMKSA cache entry created")
- ev = hapd.wait_event([ "AP-STA-CONNECTED" ], timeout=5)
- if ev is None:
- raise Exception("No connection event received from hostapd")
- dev[0].dump_monitor()
- logger.info("Disconnect and reconnect to the same AP")
- dev[0].request("DISCONNECT")
- dev[0].wait_disconnected()
- dev[0].request("RECONNECT")
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Reconnect timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- pmksa1b = dev[0].get_pmksa(bssid)
- if pmksa1b is None:
- raise Exception("No PMKSA cache entry found")
- if pmksa['pmkid'] != pmksa1b['pmkid']:
- raise Exception("Unexpected PMKID change for AP1")
- dev[0].request("REAUTHENTICATE")
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
- if ev is None:
- raise Exception("EAP success timed out")
- for i in range(0, 20):
- state = dev[0].get_status_field("wpa_state")
- if state == "COMPLETED":
- break
- time.sleep(0.1)
- if state != "COMPLETED":
- raise Exception("Reauthentication did not complete")
- def test_pmksa_cache_preauth_auto(dev, apdev):
- """RSN pre-authentication based on pre-connection scan results"""
- try:
- run_pmksa_cache_preauth_auto(dev, apdev)
- finally:
- hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev',
- 'ap-br0', 'down', '2>', '/dev/null'],
- shell=True)
- hostapd.cmd_execute(apdev[0], ['brctl', 'delbr', 'ap-br0',
- '2>', '/dev/null'], shell=True)
- def run_pmksa_cache_preauth_auto(dev, apdev):
- params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
- params['bridge'] = 'ap-br0'
- params['rsn_preauth'] = '1'
- params['rsn_preauth_interfaces'] = 'ap-br0'
- hapd = hostapd.add_ap(apdev[0], params)
- hapd.cmd_execute(['brctl', 'setfd', 'ap-br0', '0'])
- hapd.cmd_execute(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
- hapd2 = hostapd.add_ap(apdev[1], params)
- eap_connect(dev[0], hapd, "PAX", "pax.user@example.com",
- password_hex="0123456789abcdef0123456789abcdef")
- found = False
- for i in range(20):
- time.sleep(0.5)
- res1 = dev[0].get_pmksa(apdev[0]['bssid'])
- res2 = dev[0].get_pmksa(apdev[1]['bssid'])
- if res1 and res2:
- found = True
- break
- if not found:
- raise Exception("The expected PMKSA cache entries not found")
- def generic_pmksa_cache_preauth(dev, apdev, extraparams, identity, databridge,
- force_disconnect=False):
- if not extraparams:
- extraparams = [{}, {}]
- try:
- params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
- params['bridge'] = 'ap-br0'
- for key, value in extraparams[0].iteritems():
- params[key] = value
- hapd = hostapd.add_ap(apdev[0], params)
- hapd.cmd_execute(['brctl', 'setfd', 'ap-br0', '0'])
- hapd.cmd_execute(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
- eap_connect(dev[0], hapd, "PAX", identity,
- password_hex="0123456789abcdef0123456789abcdef")
- # Verify connectivity in the correct VLAN
- hwsim_utils.test_connectivity_iface(dev[0], hapd, databridge)
- params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
- params['bridge'] = 'ap-br0'
- params['rsn_preauth'] = '1'
- params['rsn_preauth_interfaces'] = databridge
- for key, value in extraparams[1].iteritems():
- params[key] = value
- hostapd.add_ap(apdev[1], params)
- bssid1 = apdev[1]['bssid']
- dev[0].scan(freq="2412")
- success = False
- status_seen = False
- for i in range(0, 50):
- if not status_seen:
- status = dev[0].request("STATUS")
- if "Pre-authentication EAPOL state machines:" in status:
- status_seen = True
- time.sleep(0.1)
- pmksa = dev[0].get_pmksa(bssid1)
- if pmksa:
- success = True
- break
- if not success:
- raise Exception("No PMKSA cache entry created from pre-authentication")
- if not status_seen:
- raise Exception("Pre-authentication EAPOL status was not available")
- dev[0].scan(freq="2412")
- if "[WPA2-EAP-CCMP-preauth]" not in dev[0].request("SCAN_RESULTS"):
- raise Exception("Scan results missing RSN element info")
- dev[0].request("ROAM " + bssid1)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- pmksa2 = dev[0].get_pmksa(bssid1)
- if pmksa2 is None:
- raise Exception("No PMKSA cache entry")
- if pmksa['pmkid'] != pmksa2['pmkid']:
- raise Exception("Unexpected PMKID change")
- # Verify connectivity in the correct VLAN
- hwsim_utils.test_connectivity_iface(dev[0], hapd, databridge)
- if not force_disconnect:
- return
- # Disconnect the STA from both APs to avoid forceful ifdown by the
- # test script on a VLAN that this has an associated STA. That used to
- # trigger a mac80211 warning.
- dev[0].request("DISCONNECT")
- hapd.request("DISABLE")
- finally:
- hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev',
- 'ap-br0', 'down', '2>', '/dev/null'],
- shell=True)
- hostapd.cmd_execute(apdev[0], ['brctl', 'delbr', 'ap-br0',
- '2>', '/dev/null'], shell=True)
- def test_pmksa_cache_preauth(dev, apdev):
- """RSN pre-authentication to generate PMKSA cache entry"""
- generic_pmksa_cache_preauth(dev, apdev, None,
- "pax.user@example.com", "ap-br0")
- def test_pmksa_cache_preauth_per_sta_vif(dev, apdev):
- """RSN pre-authentication to generate PMKSA cache entry with per_sta_vif"""
- extraparams = [{}, {}]
- extraparams[0]['per_sta_vif'] = "1"
- extraparams[1]['per_sta_vif'] = "1"
- generic_pmksa_cache_preauth(dev, apdev, extraparams,
- "pax.user@example.com", "ap-br0")
- def test_pmksa_cache_preauth_vlan_enabled(dev, apdev):
- """RSN pre-authentication to generate PMKSA cache entry (dynamic_vlan optional but station without VLAN set)"""
- extraparams = [{}, {}]
- extraparams[0]['dynamic_vlan'] = '1'
- extraparams[1]['dynamic_vlan'] = '1'
- generic_pmksa_cache_preauth(dev, apdev, extraparams,
- "pax.user@example.com", "ap-br0")
- def test_pmksa_cache_preauth_vlan_enabled_per_sta_vif(dev, apdev):
- """RSN pre-authentication to generate PMKSA cache entry (dynamic_vlan optional but station without VLAN set, with per_sta_vif enabled)"""
- extraparams = [{}, {}]
- extraparams[0]['per_sta_vif'] = "1"
- extraparams[1]['per_sta_vif'] = "1"
- extraparams[0]['dynamic_vlan'] = '1'
- extraparams[1]['dynamic_vlan'] = '1'
- generic_pmksa_cache_preauth(dev, apdev, extraparams,
- "pax.user@example.com", "ap-br0")
- def test_pmksa_cache_preauth_vlan_used(dev, apdev):
- """RSN pre-authentication to generate PMKSA cache entry (station with VLAN set)"""
- run_pmksa_cache_preauth_vlan_used(dev, apdev, None, force_disconnect=True)
- def run_pmksa_cache_preauth_vlan_used(dev, apdev, extraparams=None,
- force_disconnect=False):
- try:
- subprocess.call(['brctl', 'addbr', 'brvlan1'])
- subprocess.call(['brctl', 'setfd', 'brvlan1', '0'])
- if not extraparams:
- extraparams = [{}, {}]
- extraparams[0]['dynamic_vlan'] = '1'
- extraparams[0]['vlan_file'] = 'hostapd.wlan3.vlan'
- extraparams[1]['dynamic_vlan'] = '1'
- extraparams[1]['vlan_file'] = 'hostapd.wlan4.vlan'
- generic_pmksa_cache_preauth(dev, apdev, extraparams,
- "vlan1", "brvlan1",
- force_disconnect=force_disconnect)
- finally:
- subprocess.call(['ip', 'link', 'set', 'dev', 'brvlan1', 'down'])
- subprocess.call(['ip', 'link', 'set', 'dev', 'wlan3.1', 'down'],
- stderr=open('/dev/null', 'w'))
- subprocess.call(['ip', 'link', 'set', 'dev', 'wlan4.1', 'down'],
- stderr=open('/dev/null', 'w'))
- subprocess.call(['brctl', 'delif', 'brvlan1', 'wlan3.1'],
- stderr=open('/dev/null', 'w'))
- subprocess.call(['brctl', 'delif', 'brvlan1', 'wlan4.1'],
- stderr=open('/dev/null', 'w'))
- subprocess.call(['brctl', 'delbr', 'brvlan1'])
- def test_pmksa_cache_preauth_vlan_used_per_sta_vif(dev, apdev):
- """RSN pre-authentication to generate PMKSA cache entry (station with VLAN set, per_sta_vif=1)"""
- extraparams = [{}, {}]
- extraparams[0]['per_sta_vif'] = "1"
- extraparams[1]['per_sta_vif'] = "1"
- run_pmksa_cache_preauth_vlan_used(dev, apdev, extraparams)
- def test_pmksa_cache_disabled(dev, apdev):
- """PMKSA cache disabling on AP"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- params['disable_pmksa_caching'] = '1'
- hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- hostapd.add_ap(apdev[1], params)
- bssid2 = apdev[1]['bssid']
- dev[0].dump_monitor()
- logger.info("Roam to AP2")
- dev[0].scan_for_bss(bssid2, freq="2412")
- dev[0].request("ROAM " + bssid2)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
- if ev is None:
- raise Exception("EAP success timed out")
- dev[0].wait_connected(timeout=10, error="Roaming timed out")
- dev[0].dump_monitor()
- logger.info("Roam back to AP1")
- dev[0].scan(freq="2412")
- dev[0].request("ROAM " + bssid)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=20)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-CONNECTED" in ev:
- raise Exception("EAP exchange missing")
- ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- def test_pmksa_cache_ap_expiration(dev, apdev):
- """PMKSA cache entry expiring on AP"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk-user-session-timeout",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- ev = hapd.wait_event([ "AP-STA-CONNECTED" ], timeout=5)
- if ev is None:
- raise Exception("No connection event received from hostapd")
- dev[0].request("DISCONNECT")
- time.sleep(5)
- dev[0].dump_monitor()
- dev[0].request("RECONNECT")
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=20)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-CONNECTED" in ev:
- raise Exception("EAP exchange missing")
- dev[0].wait_connected(timeout=20, error="Reconnect timed out")
- dev[0].dump_monitor()
- dev[0].wait_disconnected(timeout=20)
- dev[0].wait_connected(timeout=20, error="Reassociation timed out")
- def test_pmksa_cache_multiple_sta(dev, apdev):
- """PMKSA cache with multiple stations"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- for d in dev:
- d.flush_scan_cache()
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk-user-session-timeout",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- dev[1].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- dev[2].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk-user-session-timeout",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
- wpas.interface_add("wlan5")
- wpas.flush_scan_cache()
- wpas.connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- hostapd.add_ap(apdev[1], params)
- bssid2 = apdev[1]['bssid']
- logger.info("Roam to AP2")
- for sta in [ dev[1], dev[0], dev[2], wpas ]:
- sta.dump_monitor()
- sta.scan_for_bss(bssid2, freq="2412")
- if "OK" not in sta.request("ROAM " + bssid2):
- raise Exception("ROAM command failed (" + sta.ifname + ")")
- ev = sta.wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
- if ev is None:
- raise Exception("EAP success timed out")
- sta.wait_connected(timeout=10, error="Roaming timed out")
- sta.dump_monitor()
- logger.info("Roam back to AP1")
- for sta in [ dev[1], wpas, dev[0], dev[2] ]:
- sta.dump_monitor()
- sta.scan(freq="2412")
- sta.dump_monitor()
- sta.request("ROAM " + bssid)
- sta.wait_connected(timeout=10, error="Roaming timed out")
- sta.dump_monitor()
- time.sleep(4)
- logger.info("Roam back to AP2")
- for sta in [ dev[1], wpas, dev[0], dev[2] ]:
- sta.dump_monitor()
- sta.scan(freq="2412")
- sta.dump_monitor()
- sta.request("ROAM " + bssid2)
- sta.wait_connected(timeout=10, error="Roaming timed out")
- sta.dump_monitor()
- def test_pmksa_cache_opportunistic_multiple_sta(dev, apdev):
- """Opportunistic PMKSA caching with multiple stations"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- params['okc'] = "1"
- hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- for d in dev:
- d.flush_scan_cache()
- wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
- wpas.interface_add("wlan5")
- wpas.flush_scan_cache()
- for sta in [ dev[0], dev[1], dev[2], wpas ]:
- sta.connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef", okc=True,
- scan_freq="2412")
- hostapd.add_ap(apdev[1], params)
- bssid2 = apdev[1]['bssid']
- logger.info("Roam to AP2")
- for sta in [ dev[2], dev[0], wpas, dev[1] ]:
- sta.dump_monitor()
- sta.scan_for_bss(bssid2, freq="2412")
- if "OK" not in sta.request("ROAM " + bssid2):
- raise Exception("ROAM command failed")
- ev = sta.wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- pmksa2 = sta.get_pmksa(bssid2)
- if pmksa2 is None:
- raise Exception("No PMKSA cache entry created")
- sta.dump_monitor()
- logger.info("Roam back to AP1")
- for sta in [ dev[0], dev[1], dev[2], wpas ]:
- sta.dump_monitor()
- sta.scan_for_bss(bssid, freq="2412")
- sta.request("ROAM " + bssid)
- ev = sta.wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=10)
- if ev is None:
- raise Exception("Roaming with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange")
- def test_pmksa_cache_preauth_oom(dev, apdev):
- """RSN pre-authentication to generate PMKSA cache entry and OOM"""
- try:
- _test_pmksa_cache_preauth_oom(dev, apdev)
- finally:
- hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', 'ap-br0',
- 'down'])
- hostapd.cmd_execute(apdev[0], ['brctl', 'delbr', 'ap-br0'])
- def _test_pmksa_cache_preauth_oom(dev, apdev):
- params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
- params['bridge'] = 'ap-br0'
- hapd = hostapd.add_ap(apdev[0], params)
- hostapd.cmd_execute(apdev[0], ['brctl', 'setfd', 'ap-br0', '0'])
- hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
- eap_connect(dev[0], hapd, "PAX", "pax.user@example.com",
- password_hex="0123456789abcdef0123456789abcdef",
- bssid=apdev[0]['bssid'])
- params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
- params['bridge'] = 'ap-br0'
- params['rsn_preauth'] = '1'
- params['rsn_preauth_interfaces'] = 'ap-br0'
- hapd = hostapd.add_ap(apdev[1], params)
- bssid1 = apdev[1]['bssid']
- tests = [ (1, "rsn_preauth_receive"),
- (2, "rsn_preauth_receive"),
- (1, "rsn_preauth_send"),
- (1, "wpa_auth_pmksa_add_preauth;rsn_preauth_finished") ]
- for test in tests:
- hapd.request("DEAUTHENTICATE ff:ff:ff:ff:ff:ff")
- with alloc_fail(hapd, test[0], test[1]):
- dev[0].scan_for_bss(bssid1, freq="2412")
- if "OK" not in dev[0].request("PREAUTH " + bssid1):
- raise Exception("PREAUTH failed")
- success = False
- count = 0
- for i in range(50):
- time.sleep(0.1)
- pmksa = dev[0].get_pmksa(bssid1)
- if pmksa:
- success = True
- break
- state = hapd.request('GET_ALLOC_FAIL')
- if state.startswith('0:'):
- count += 1
- if count > 2:
- break
- logger.info("PMKSA cache success: " + str(success))
- dev[0].request("PMKSA_FLUSH")
- dev[0].wait_disconnected()
- dev[0].wait_connected()
- dev[0].dump_monitor()
- def test_pmksa_cache_size_limit(dev, apdev):
- """PMKSA cache size limit in wpa_supplicant"""
- try:
- _test_pmksa_cache_size_limit(dev, apdev)
- finally:
- try:
- hapd = hostapd.HostapdGlobal(apdev[0])
- hapd.flush()
- hapd.remove(apdev[0]['ifname'])
- except:
- pass
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- bssid = apdev[0]['bssid']
- params['bssid'] = bssid
- hostapd.add_ap(apdev[0], params)
- def _test_pmksa_cache_size_limit(dev, apdev):
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- id = dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412", only_add_network=True)
- for i in range(33):
- bssid = apdev[0]['bssid'][0:15] + "%02x" % i
- logger.info("Iteration with BSSID " + bssid)
- params['bssid'] = bssid
- hostapd.add_ap(apdev[0], params)
- dev[0].request("BSS_FLUSH 0")
- dev[0].scan_for_bss(bssid, freq=2412, only_new=True)
- dev[0].select_network(id)
- dev[0].wait_connected()
- dev[0].request("DISCONNECT")
- dev[0].wait_disconnected()
- dev[0].dump_monitor()
- entries = len(dev[0].request("PMKSA").splitlines()) - 1
- if i == 32:
- if entries != 32:
- raise Exception("Unexpected number of PMKSA entries after expected removal of the oldest entry")
- elif i + 1 != entries:
- raise Exception("Unexpected number of PMKSA entries")
- hapd = hostapd.HostapdGlobal(apdev[0])
- hapd.flush()
- hapd.remove(apdev[0]['ifname'])
- def test_pmksa_cache_preauth_timeout(dev, apdev):
- """RSN pre-authentication timing out"""
- try:
- _test_pmksa_cache_preauth_timeout(dev, apdev)
- finally:
- dev[0].request("SET dot11RSNAConfigSATimeout 60")
- def _test_pmksa_cache_preauth_timeout(dev, apdev):
- dev[0].request("SET dot11RSNAConfigSATimeout 1")
- params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
- hapd = hostapd.add_ap(apdev[0], params)
- eap_connect(dev[0], hapd, "PAX", "pax.user@example.com",
- password_hex="0123456789abcdef0123456789abcdef",
- bssid=apdev[0]['bssid'])
- if "OK" not in dev[0].request("PREAUTH f2:11:22:33:44:55"):
- raise Exception("PREAUTH failed")
- ev = dev[0].wait_event(["RSN: pre-authentication with"], timeout=5)
- if ev is None:
- raise Exception("No timeout event seen")
- if "timed out" not in ev:
- raise Exception("Unexpected event: " + ev)
- def test_pmksa_cache_preauth_wpas_oom(dev, apdev):
- """RSN pre-authentication OOM in wpa_supplicant"""
- params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
- hapd = hostapd.add_ap(apdev[0], params)
- eap_connect(dev[0], hapd, "PAX", "pax.user@example.com",
- password_hex="0123456789abcdef0123456789abcdef",
- bssid=apdev[0]['bssid'])
- for i in range(1, 11):
- with alloc_fail(dev[0], i, "rsn_preauth_init"):
- res = dev[0].request("PREAUTH f2:11:22:33:44:55").strip()
- logger.info("Iteration %d - PREAUTH command results: %s" % (i, res))
- for j in range(10):
- state = dev[0].request('GET_ALLOC_FAIL')
- if state.startswith('0:'):
- break
- time.sleep(0.05)
- def test_pmksa_cache_ctrl(dev, apdev):
- """PMKSA cache control interface operations"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- addr = dev[0].own_addr()
- dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- pmksa_sta = dev[0].get_pmksa(bssid)
- if pmksa_sta is None:
- raise Exception("No PMKSA cache entry created on STA")
- pmksa_ap = hapd.get_pmksa(addr)
- if pmksa_ap is None:
- raise Exception("No PMKSA cache entry created on AP")
- if pmksa_sta['pmkid'] != pmksa_ap['pmkid']:
- raise Exception("PMKID mismatch in PMKSA cache entries")
- if "OK" not in hapd.request("PMKSA_FLUSH"):
- raise Exception("PMKSA_FLUSH failed")
- pmksa_ap = hapd.get_pmksa(addr)
- if pmksa_ap is not None:
- raise Exception("PMKSA cache entry was not removed on AP")
- dev[0].request("DISCONNECT")
- dev[0].wait_disconnected()
- dev[0].request("RECONNECT")
- dev[0].wait_connected()
- pmksa_sta2 = dev[0].get_pmksa(bssid)
- if pmksa_sta2 is None:
- raise Exception("No PMKSA cache entry created on STA after reconnect")
- pmksa_ap2 = hapd.get_pmksa(addr)
- if pmksa_ap2 is None:
- raise Exception("No PMKSA cache entry created on AP after reconnect")
- if pmksa_sta2['pmkid'] != pmksa_ap2['pmkid']:
- raise Exception("PMKID mismatch in PMKSA cache entries after reconnect")
- if pmksa_sta2['pmkid'] == pmksa_sta['pmkid']:
- raise Exception("PMKID did not change after reconnect")
- def test_pmksa_cache_ctrl_events(dev, apdev):
- """PMKSA cache control interface events"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- id = dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412", wait_connect=False)
- ev = dev[0].wait_event(["PMKSA-CACHE-ADDED"], timeout=15)
- if ev is None:
- raise Exception("No PMKSA-CACHE-ADDED event")
- dev[0].wait_connected()
- items = ev.split(' ')
- if items[1] != bssid:
- raise Exception("BSSID mismatch: " + ev)
- if int(items[2]) != id:
- raise Exception("network_id mismatch: " + ev)
- dev[0].request("PMKSA_FLUSH")
- ev = dev[0].wait_event(["PMKSA-CACHE-REMOVED"], timeout=15)
- if ev is None:
- raise Exception("No PMKSA-CACHE-REMOVED event")
- dev[0].wait_disconnected()
- dev[0].request("DISCONNECT")
- items = ev.split(' ')
- if items[1] != bssid:
- raise Exception("BSSID mismatch: " + ev)
- if int(items[2]) != id:
- raise Exception("network_id mismatch: " + ev)
- def test_pmksa_cache_ctrl_ext(dev, apdev):
- """PMKSA cache control interface for external management"""
- params = hostapd.wpa2_eap_params(ssid="test-pmksa-cache")
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = apdev[0]['bssid']
- id = dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412")
- res1 = dev[0].request("PMKSA_GET %d" % id)
- logger.info("PMKSA_GET: " + res1)
- if "UNKNOWN COMMAND" in res1:
- raise HwsimSkip("PMKSA_GET not supported in the build")
- if bssid not in res1:
- raise Exception("PMKSA cache entry missing")
- hostapd.add_ap(apdev[1], params)
- bssid2 = apdev[1]['bssid']
- dev[0].scan_for_bss(bssid2, freq=2412, force_scan=True)
- dev[0].request("ROAM " + bssid2)
- dev[0].wait_connected()
- res2 = dev[0].request("PMKSA_GET %d" % id)
- logger.info("PMKSA_GET: " + res2)
- if bssid not in res2:
- raise Exception("PMKSA cache entry 1 missing")
- if bssid2 not in res2:
- raise Exception("PMKSA cache entry 2 missing")
- dev[0].request("REMOVE_NETWORK all")
- dev[0].wait_disconnected()
- dev[0].request("PMKSA_FLUSH")
- id = dev[0].connect("test-pmksa-cache", proto="RSN", key_mgmt="WPA-EAP",
- eap="GPSK", identity="gpsk user",
- password="abcdefghijklmnop0123456789abcdef",
- scan_freq="2412", only_add_network=True)
- res3 = dev[0].request("PMKSA_GET %d" % id)
- if res3 != '':
- raise Exception("Unexpected PMKSA cache entry remains: " + res3)
- res4 = dev[0].request("PMKSA_GET %d" % (id + 1234))
- if not res4.startswith('FAIL'):
- raise Exception("Unexpected PMKSA cache entry for unknown network: " + res4)
- for entry in res2.splitlines():
- if "OK" not in dev[0].request("PMKSA_ADD %d %s" % (id, entry)):
- raise Exception("Failed to add PMKSA entry")
- dev[0].select_network(id)
- ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",
- "CTRL-EVENT-CONNECTED"], timeout=15)
- if ev is None:
- raise Exception("Connection with the AP timed out")
- if "CTRL-EVENT-EAP-STARTED" in ev:
- raise Exception("Unexpected EAP exchange after external PMKSA cache restore")
- def test_rsn_preauth_processing(dev, apdev):
- """RSN pre-authentication processing on AP"""
- params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
- params['rsn_preauth'] = '1'
- params['rsn_preauth_interfaces'] = "lo"
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = hapd.own_addr()
- _bssid = binascii.unhexlify(bssid.replace(':', ''))
- eap_connect(dev[0], hapd, "PAX", "pax.user@example.com",
- password_hex="0123456789abcdef0123456789abcdef")
- addr = dev[0].own_addr()
- _addr = binascii.unhexlify(addr.replace(':', ''))
- sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW,
- socket.htons(0x88c7))
- sock.bind(("lo", socket.htons(0x88c7)))
- foreign = "\x02\x03\x04\x05\x06\x07"
- proto = "\x88\xc7"
- tests = []
- # RSN: too short pre-auth packet (len=14)
- tests += [ _bssid + foreign + proto ]
- # Not EAPOL-Start
- tests += [ _bssid + foreign + proto + struct.pack('>BBH', 0, 0, 0) ]
- # RSN: pre-auth for foreign address 02:03:04:05:06:07
- tests += [ foreign + foreign + proto + struct.pack('>BBH', 0, 0, 0) ]
- # RSN: pre-auth for already association STA 02:00:00:00:00:00
- tests += [ _bssid + _addr + proto + struct.pack('>BBH', 0, 0, 0) ]
- # New STA
- tests += [ _bssid + foreign + proto + struct.pack('>BBH', 0, 1, 1) ]
- # IEEE 802.1X: received EAPOL-Start from STA
- tests += [ _bssid + foreign + proto + struct.pack('>BBH', 0, 1, 0) ]
- # frame too short for this IEEE 802.1X packet
- tests += [ _bssid + foreign + proto + struct.pack('>BBH', 0, 1, 1) ]
- # EAPOL-Key - Dropped key data from unauthorized Supplicant
- tests += [ _bssid + foreign + proto + struct.pack('>BBH', 2, 3, 0) ]
- # EAPOL-Encapsulated-ASF-Alert
- tests += [ _bssid + foreign + proto + struct.pack('>BBH', 2, 4, 0) ]
- # unknown IEEE 802.1X packet type
- tests += [ _bssid + foreign + proto + struct.pack('>BBH', 2, 255, 0) ]
- for t in tests:
- sock.send(t)
- def test_rsn_preauth_local_errors(dev, apdev):
- """RSN pre-authentication and local errors on AP"""
- params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
- params['rsn_preauth'] = '1'
- params['rsn_preauth_interfaces'] = "lo"
- hapd = hostapd.add_ap(apdev[0], params)
- bssid = hapd.own_addr()
- _bssid = binascii.unhexlify(bssid.replace(':', ''))
- sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW,
- socket.htons(0x88c7))
- sock.bind(("lo", socket.htons(0x88c7)))
- foreign = "\x02\x03\x04\x05\x06\x07"
- foreign2 = "\x02\x03\x04\x05\x06\x08"
- proto = "\x88\xc7"
- with alloc_fail(hapd, 1, "ap_sta_add;rsn_preauth_receive"):
- sock.send(_bssid + foreign + proto + struct.pack('>BBH', 2, 1, 0))
- wait_fail_trigger(hapd, "GET_ALLOC_FAIL")
- with alloc_fail(hapd, 1, "eapol_auth_alloc;rsn_preauth_receive"):
- sock.send(_bssid + foreign + proto + struct.pack('>BBH', 2, 1, 0))
- wait_fail_trigger(hapd, "GET_ALLOC_FAIL")
- sock.send(_bssid + foreign + proto + struct.pack('>BBH', 2, 1, 0))
- with alloc_fail(hapd, 1, "eap_server_sm_init;ieee802_1x_new_station;rsn_preauth_receive"):
- sock.send(_bssid + foreign2 + proto + struct.pack('>BBH', 2, 1, 0))
- wait_fail_trigger(hapd, "GET_ALLOC_FAIL")
- sock.send(_bssid + foreign2 + proto + struct.pack('>BBH', 2, 1, 0))
- hapd.request("DISABLE")
- tests = [ (1, "=rsn_preauth_iface_add"),
- (2, "=rsn_preauth_iface_add"),
- (1, "l2_packet_init;rsn_preauth_iface_add"),
- (1, "rsn_preauth_iface_init"),
- (1, "rsn_preauth_iface_init") ]
- for count,func in tests:
- with alloc_fail(hapd, count, func):
- if "FAIL" not in hapd.request("ENABLE"):
- raise Exception("ENABLE succeeded unexpectedly")
- hapd.set("rsn_preauth_interfaces", "lo lo lo does-not-exist lo ")
- if "FAIL" not in hapd.request("ENABLE"):
- raise Exception("ENABLE succeeded unexpectedly")
- hapd.set("rsn_preauth_interfaces", " lo lo ")
- if "OK" not in hapd.request("ENABLE"):
- raise Exception("ENABLE failed")
- sock.send(_bssid + foreign + proto + struct.pack('>BBH', 2, 1, 0))
- sock.send(_bssid + foreign2 + proto + struct.pack('>BBH', 2, 1, 0))
|