Amazing-Mirror/fragattacks
1
0
Fork 0
mirror of https://github.com/vanhoefm/fragattacks.git synced 2024-12-01 11:48:23 -05:00
Code Issues Projects Releases Wiki Activity
46dea617ae
fragattacks/tests/hwsim/test_pmksa_cache.py
Jouni Malinen c4668009d0 tests: Remove unnecessary use of sudo from test cases 
run-tests.py is running as root, so sudo does not need to be used
anymore from within each test case.

Signed-off-by: Jouni Malinen <jouni@qca.qualcomm.com>
2015-02-07 15:37:13 +02:00

693 lines
28 KiB
Python
Raw Blame History

# 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 logging
logger = logging.getLogger()
import subprocess
import time
import hostapd
from wpasupplicant import WpaSupplicant
from utils import alloc_fail
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]['ifname'], 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]['ifname'], 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_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]['ifname'], 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]['ifname'], 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]['ifname'], 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]['ifname'], 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]['ifname'], 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]['ifname'], params)
bssid2 = apdev[1]['bssid']
wpas.dump_monitor()
logger.info("Roam to AP2")
wpas.scan_for_bss(bssid2, freq="2412")
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")
hostapd.add_ap(apdev[0]['ifname'], 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")
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]['ifname'], 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]['ifname'], 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(dev, apdev):
"""RSN pre-authentication to generate PMKSA cache entry"""
try:
params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
params['bridge'] = 'ap-br0'
hostapd.add_ap(apdev[0]['ifname'], params)
subprocess.call(['brctl', 'setfd', 'ap-br0', '0'])
subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
eap_connect(dev[0], apdev[0], "PAX", "pax.user@example.com",
password_hex="0123456789abcdef0123456789abcdef")
params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
params['bridge'] = 'ap-br0'
params['rsn_preauth'] = '1'
params['rsn_preauth_interfaces'] = 'ap-br0'
hostapd.add_ap(apdev[1]['ifname'], 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")
finally:
subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'down'])
subprocess.call(['brctl', 'delbr', 'ap-br0'])
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]['ifname'], 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]['ifname'], 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]['ifname'], 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]['ifname'], 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")
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.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]['ifname'], 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")
sta.request("ROAM " + bssid2)
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")
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]['ifname'], params)
bssid = apdev[0]['bssid']
wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
wpas.interface_add("wlan5")
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]['ifname'], 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")
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:
subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'down'])
subprocess.call(['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'
hostapd.add_ap(apdev[0]['ifname'], params)
subprocess.call(['brctl', 'setfd', 'ap-br0', '0'])
subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
eap_connect(dev[0], apdev[0], "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]['ifname'], params)
bssid1 = apdev[1]['bssid']
tests = [ (1, "rsn_preauth_receive"),
(2, "rsn_preauth_receive"),
(1, "rsn_preauth_send") ]
for test in tests:
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()
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]['ifname'], 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]['ifname'], 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()
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")
hostapd.add_ap(apdev[0]['ifname'], params)
eap_connect(dev[0], apdev[0], "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")
hostapd.add_ap(apdev[0]['ifname'], params)
eap_connect(dev[0], apdev[0], "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)
Reference in New Issue View Git Blame Copy Permalink