| # RADIUS tests |
| # Copyright (c) 2013-2015, Jouni Malinen <j@w1.fi> |
| # |
| # This software may be distributed under the terms of the BSD license. |
| # See README for more details. |
| |
| import hashlib |
| import hmac |
| import logging |
| logger = logging.getLogger() |
| import os |
| import select |
| import struct |
| import subprocess |
| import threading |
| import time |
| |
| import hostapd |
| from utils import HwsimSkip, require_under_vm |
| |
| def connect(dev, ssid, wait_connect=True): |
| dev.connect(ssid, key_mgmt="WPA-EAP", scan_freq="2412", |
| eap="PSK", identity="psk.user@example.com", |
| password_hex="0123456789abcdef0123456789abcdef", |
| wait_connect=wait_connect) |
| |
| def test_radius_auth_unreachable(dev, apdev): |
| """RADIUS Authentication server unreachable""" |
| params = hostapd.wpa2_eap_params(ssid="radius-auth") |
| params['auth_server_port'] = "18139" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| hapd = hostapd.Hostapd(apdev[0]['ifname']) |
| connect(dev[0], "radius-auth", wait_connect=False) |
| ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED"]) |
| if ev is None: |
| raise Exception("Timeout on EAP start") |
| logger.info("Checking for RADIUS retries") |
| time.sleep(4) |
| mib = hapd.get_mib() |
| if "radiusAuthClientAccessRequests" not in mib: |
| raise Exception("Missing MIB fields") |
| if int(mib["radiusAuthClientAccessRetransmissions"]) < 1: |
| raise Exception("Missing RADIUS Authentication retransmission") |
| if int(mib["radiusAuthClientPendingRequests"]) < 1: |
| raise Exception("Missing pending RADIUS Authentication request") |
| |
| def test_radius_auth_unreachable2(dev, apdev): |
| """RADIUS Authentication server unreachable (2)""" |
| subprocess.call(['ip', 'ro', 'replace', '192.168.213.17', 'dev', 'lo']) |
| params = hostapd.wpa2_eap_params(ssid="radius-auth") |
| params['auth_server_addr'] = "192.168.213.17" |
| params['auth_server_port'] = "18139" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| hapd = hostapd.Hostapd(apdev[0]['ifname']) |
| subprocess.call(['ip', 'ro', 'del', '192.168.213.17', 'dev', 'lo']) |
| connect(dev[0], "radius-auth", wait_connect=False) |
| ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED"]) |
| if ev is None: |
| raise Exception("Timeout on EAP start") |
| logger.info("Checking for RADIUS retries") |
| time.sleep(4) |
| mib = hapd.get_mib() |
| if "radiusAuthClientAccessRequests" not in mib: |
| raise Exception("Missing MIB fields") |
| if int(mib["radiusAuthClientAccessRetransmissions"]) < 1: |
| raise Exception("Missing RADIUS Authentication retransmission") |
| |
| def test_radius_auth_unreachable3(dev, apdev): |
| """RADIUS Authentication server initially unreachable, but then available""" |
| subprocess.call(['ip', 'ro', 'replace', 'blackhole', '192.168.213.18']) |
| params = hostapd.wpa2_eap_params(ssid="radius-auth") |
| params['auth_server_addr'] = "192.168.213.18" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| hapd = hostapd.Hostapd(apdev[0]['ifname']) |
| connect(dev[0], "radius-auth", wait_connect=False) |
| ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED"]) |
| if ev is None: |
| raise Exception("Timeout on EAP start") |
| subprocess.call(['ip', 'ro', 'del', 'blackhole', '192.168.213.18']) |
| time.sleep(0.1) |
| dev[0].request("DISCONNECT") |
| hapd.set('auth_server_addr_replace', '127.0.0.1') |
| dev[0].request("RECONNECT") |
| |
| dev[0].wait_connected() |
| |
| def test_radius_acct_unreachable(dev, apdev): |
| """RADIUS Accounting server unreachable""" |
| params = hostapd.wpa2_eap_params(ssid="radius-acct") |
| params['acct_server_addr'] = "127.0.0.1" |
| params['acct_server_port'] = "18139" |
| params['acct_server_shared_secret'] = "radius" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| hapd = hostapd.Hostapd(apdev[0]['ifname']) |
| connect(dev[0], "radius-acct") |
| logger.info("Checking for RADIUS retries") |
| time.sleep(4) |
| mib = hapd.get_mib() |
| if "radiusAccClientRetransmissions" not in mib: |
| raise Exception("Missing MIB fields") |
| if int(mib["radiusAccClientRetransmissions"]) < 2: |
| raise Exception("Missing RADIUS Accounting retransmissions") |
| if int(mib["radiusAccClientPendingRequests"]) < 2: |
| raise Exception("Missing pending RADIUS Accounting requests") |
| |
| def test_radius_acct_unreachable2(dev, apdev): |
| """RADIUS Accounting server unreachable(2)""" |
| subprocess.call(['ip', 'ro', 'replace', '192.168.213.17', 'dev', 'lo']) |
| params = hostapd.wpa2_eap_params(ssid="radius-acct") |
| params['acct_server_addr'] = "192.168.213.17" |
| params['acct_server_port'] = "18139" |
| params['acct_server_shared_secret'] = "radius" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| hapd = hostapd.Hostapd(apdev[0]['ifname']) |
| subprocess.call(['ip', 'ro', 'del', '192.168.213.17', 'dev', 'lo']) |
| connect(dev[0], "radius-acct") |
| logger.info("Checking for RADIUS retries") |
| time.sleep(4) |
| mib = hapd.get_mib() |
| if "radiusAccClientRetransmissions" not in mib: |
| raise Exception("Missing MIB fields") |
| if int(mib["radiusAccClientRetransmissions"]) < 1 and int(mib["radiusAccClientPendingRequests"]) < 1: |
| raise Exception("Missing pending or retransmitted RADIUS Accounting requests") |
| |
| def test_radius_acct_unreachable3(dev, apdev): |
| """RADIUS Accounting server initially unreachable, but then available""" |
| require_under_vm() |
| subprocess.call(['ip', 'ro', 'replace', 'blackhole', '192.168.213.18']) |
| as_hapd = hostapd.Hostapd("as") |
| as_mib_start = as_hapd.get_mib(param="radius_server") |
| params = hostapd.wpa2_eap_params(ssid="radius-acct") |
| params['acct_server_addr'] = "192.168.213.18" |
| params['acct_server_port'] = "1813" |
| params['acct_server_shared_secret'] = "radius" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| hapd = hostapd.Hostapd(apdev[0]['ifname']) |
| connect(dev[0], "radius-acct") |
| subprocess.call(['ip', 'ro', 'del', 'blackhole', '192.168.213.18']) |
| time.sleep(0.1) |
| dev[0].request("DISCONNECT") |
| hapd.set('acct_server_addr_replace', '127.0.0.1') |
| dev[0].request("RECONNECT") |
| dev[0].wait_connected() |
| time.sleep(1) |
| as_mib_end = as_hapd.get_mib(param="radius_server") |
| req_s = int(as_mib_start['radiusAccServTotalResponses']) |
| req_e = int(as_mib_end['radiusAccServTotalResponses']) |
| if req_e <= req_s: |
| raise Exception("Unexpected RADIUS server acct MIB value") |
| |
| def test_radius_acct_unreachable4(dev, apdev): |
| """RADIUS Accounting server unreachable and multiple STAs""" |
| params = hostapd.wpa2_eap_params(ssid="radius-acct") |
| params['acct_server_addr'] = "127.0.0.1" |
| params['acct_server_port'] = "18139" |
| params['acct_server_shared_secret'] = "radius" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| hapd = hostapd.Hostapd(apdev[0]['ifname']) |
| for i in range(20): |
| connect(dev[0], "radius-acct") |
| dev[0].request("REMOVE_NETWORK all") |
| dev[0].wait_disconnected() |
| |
| def test_radius_acct(dev, apdev): |
| """RADIUS Accounting""" |
| as_hapd = hostapd.Hostapd("as") |
| as_mib_start = as_hapd.get_mib(param="radius_server") |
| params = hostapd.wpa2_eap_params(ssid="radius-acct") |
| params['acct_server_addr'] = "127.0.0.1" |
| params['acct_server_port'] = "1813" |
| params['acct_server_shared_secret'] = "radius" |
| params['radius_auth_req_attr'] = [ "126:s:Operator", "77:s:testing" ] |
| params['radius_acct_req_attr'] = [ "126:s:Operator", "77:s:testing" ] |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| hapd = hostapd.Hostapd(apdev[0]['ifname']) |
| connect(dev[0], "radius-acct") |
| dev[1].connect("radius-acct", key_mgmt="WPA-EAP", scan_freq="2412", |
| eap="PAX", identity="test-class", |
| password_hex="0123456789abcdef0123456789abcdef") |
| dev[2].connect("radius-acct", key_mgmt="WPA-EAP", |
| eap="GPSK", identity="gpsk-cui", |
| password="abcdefghijklmnop0123456789abcdef", |
| scan_freq="2412") |
| logger.info("Checking for RADIUS counters") |
| count = 0 |
| while True: |
| mib = hapd.get_mib() |
| if int(mib['radiusAccClientResponses']) >= 3: |
| break |
| time.sleep(0.1) |
| count += 1 |
| if count > 10: |
| raise Exception("Did not receive Accounting-Response packets") |
| |
| if int(mib['radiusAccClientRetransmissions']) > 0: |
| raise Exception("Unexpected Accounting-Request retransmission") |
| |
| as_mib_end = as_hapd.get_mib(param="radius_server") |
| |
| req_s = int(as_mib_start['radiusAccServTotalRequests']) |
| req_e = int(as_mib_end['radiusAccServTotalRequests']) |
| if req_e < req_s + 2: |
| raise Exception("Unexpected RADIUS server acct MIB value") |
| |
| acc_s = int(as_mib_start['radiusAuthServAccessAccepts']) |
| acc_e = int(as_mib_end['radiusAuthServAccessAccepts']) |
| if acc_e < acc_s + 1: |
| raise Exception("Unexpected RADIUS server auth MIB value") |
| |
| def test_radius_acct_pmksa_caching(dev, apdev): |
| """RADIUS Accounting with PMKSA caching""" |
| as_hapd = hostapd.Hostapd("as") |
| as_mib_start = as_hapd.get_mib(param="radius_server") |
| params = hostapd.wpa2_eap_params(ssid="radius-acct") |
| 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) |
| connect(dev[0], "radius-acct") |
| dev[1].connect("radius-acct", key_mgmt="WPA-EAP", scan_freq="2412", |
| eap="PAX", identity="test-class", |
| password_hex="0123456789abcdef0123456789abcdef") |
| for d in [ dev[0], dev[1] ]: |
| d.request("REASSOCIATE") |
| d.wait_connected(timeout=15, error="Reassociation timed out") |
| |
| count = 0 |
| while True: |
| mib = hapd.get_mib() |
| if int(mib['radiusAccClientResponses']) >= 4: |
| break |
| time.sleep(0.1) |
| count += 1 |
| if count > 10: |
| raise Exception("Did not receive Accounting-Response packets") |
| |
| if int(mib['radiusAccClientRetransmissions']) > 0: |
| raise Exception("Unexpected Accounting-Request retransmission") |
| |
| as_mib_end = as_hapd.get_mib(param="radius_server") |
| |
| req_s = int(as_mib_start['radiusAccServTotalRequests']) |
| req_e = int(as_mib_end['radiusAccServTotalRequests']) |
| if req_e < req_s + 2: |
| raise Exception("Unexpected RADIUS server acct MIB value") |
| |
| acc_s = int(as_mib_start['radiusAuthServAccessAccepts']) |
| acc_e = int(as_mib_end['radiusAuthServAccessAccepts']) |
| if acc_e < acc_s + 1: |
| raise Exception("Unexpected RADIUS server auth MIB value") |
| |
| def test_radius_acct_interim(dev, apdev): |
| """RADIUS Accounting interim update""" |
| as_hapd = hostapd.Hostapd("as") |
| params = hostapd.wpa2_eap_params(ssid="radius-acct") |
| params['acct_server_addr'] = "127.0.0.1" |
| params['acct_server_port'] = "1813" |
| params['acct_server_shared_secret'] = "radius" |
| params['radius_acct_interim_interval'] = "1" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| hapd = hostapd.Hostapd(apdev[0]['ifname']) |
| connect(dev[0], "radius-acct") |
| logger.info("Checking for RADIUS counters") |
| as_mib_start = as_hapd.get_mib(param="radius_server") |
| time.sleep(3.1) |
| as_mib_end = as_hapd.get_mib(param="radius_server") |
| req_s = int(as_mib_start['radiusAccServTotalRequests']) |
| req_e = int(as_mib_end['radiusAccServTotalRequests']) |
| if req_e < req_s + 3: |
| raise Exception("Unexpected RADIUS server acct MIB value") |
| |
| def test_radius_acct_interim_unreachable(dev, apdev): |
| """RADIUS Accounting interim update with unreachable server""" |
| params = hostapd.wpa2_eap_params(ssid="radius-acct") |
| params['acct_server_addr'] = "127.0.0.1" |
| params['acct_server_port'] = "18139" |
| params['acct_server_shared_secret'] = "radius" |
| params['radius_acct_interim_interval'] = "1" |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| start = hapd.get_mib() |
| connect(dev[0], "radius-acct") |
| logger.info("Waiting for interium accounting updates") |
| time.sleep(3.1) |
| end = hapd.get_mib() |
| req_s = int(start['radiusAccClientTimeouts']) |
| req_e = int(end['radiusAccClientTimeouts']) |
| if req_e < req_s + 2: |
| raise Exception("Unexpected RADIUS server acct MIB value") |
| |
| def send_and_check_reply(srv, req, code, error_cause=0): |
| reply = srv.SendPacket(req) |
| logger.debug("RADIUS response from hostapd") |
| for i in reply.keys(): |
| logger.debug("%s: %s" % (i, reply[i])) |
| if reply.code != code: |
| raise Exception("Unexpected response code") |
| if error_cause: |
| if 'Error-Cause' not in reply: |
| raise Exception("Missing Error-Cause") |
| if reply['Error-Cause'][0] != error_cause: |
| raise Exception("Unexpected Error-Cause: {}".format(reply['Error-Cause'])) |
| |
| def test_radius_das_disconnect(dev, apdev): |
| """RADIUS Dynamic Authorization Extensions - Disconnect""" |
| try: |
| import pyrad.client |
| import pyrad.packet |
| import pyrad.dictionary |
| import radius_das |
| except ImportError: |
| raise HwsimSkip("No pyrad modules available") |
| |
| params = hostapd.wpa2_eap_params(ssid="radius-das") |
| params['radius_das_port'] = "3799" |
| params['radius_das_client'] = "127.0.0.1 secret" |
| params['radius_das_require_event_timestamp'] = "1" |
| params['own_ip_addr'] = "127.0.0.1" |
| params['nas_identifier'] = "nas.example.com" |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| connect(dev[0], "radius-das") |
| addr = dev[0].p2p_interface_addr() |
| sta = hapd.get_sta(addr) |
| id = sta['dot1xAuthSessionId'] |
| |
| 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 |
| |
| logger.info("Disconnect-Request with incorrect secret") |
| req = radius_das.DisconnectPacket(dict=dict, secret="incorrect", |
| User_Name="foo", |
| NAS_Identifier="localhost", |
| Event_Timestamp=int(time.time())) |
| logger.debug(req) |
| try: |
| reply = srv.SendPacket(req) |
| raise Exception("Unexpected response to Disconnect-Request") |
| except pyrad.client.Timeout: |
| logger.info("Disconnect-Request with incorrect secret properly ignored") |
| |
| logger.info("Disconnect-Request without Event-Timestamp") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| User_Name="psk.user@example.com") |
| logger.debug(req) |
| try: |
| reply = srv.SendPacket(req) |
| raise Exception("Unexpected response to Disconnect-Request") |
| except pyrad.client.Timeout: |
| logger.info("Disconnect-Request without Event-Timestamp properly ignored") |
| |
| logger.info("Disconnect-Request with non-matching Event-Timestamp") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| User_Name="psk.user@example.com", |
| Event_Timestamp=123456789) |
| logger.debug(req) |
| try: |
| reply = srv.SendPacket(req) |
| raise Exception("Unexpected response to Disconnect-Request") |
| except pyrad.client.Timeout: |
| logger.info("Disconnect-Request with non-matching Event-Timestamp properly ignored") |
| |
| logger.info("Disconnect-Request with unsupported attribute") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| User_Name="foo", |
| User_Password="foo", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 401) |
| |
| logger.info("Disconnect-Request with invalid Calling-Station-Id") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| User_Name="foo", |
| Calling_Station_Id="foo", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 407) |
| |
| logger.info("Disconnect-Request with mismatching User-Name") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| User_Name="foo", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503) |
| |
| logger.info("Disconnect-Request with mismatching Calling-Station-Id") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| Calling_Station_Id="12:34:56:78:90:aa", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503) |
| |
| logger.info("Disconnect-Request with mismatching Acct-Session-Id") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| Acct_Session_Id="12345678-87654321", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503) |
| |
| logger.info("Disconnect-Request with mismatching Acct-Session-Id (len)") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| Acct_Session_Id="12345678", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503) |
| |
| logger.info("Disconnect-Request with mismatching Acct-Multi-Session-Id") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| Acct_Multi_Session_Id="12345678+87654321", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503) |
| |
| logger.info("Disconnect-Request with mismatching Acct-Multi-Session-Id (len)") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| Acct_Multi_Session_Id="12345678", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503) |
| |
| logger.info("Disconnect-Request with no session identification attributes") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 503) |
| |
| ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1) |
| if ev is not None: |
| raise Exception("Unexpected disconnection") |
| |
| logger.info("Disconnect-Request with mismatching NAS-IP-Address") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_IP_Address="192.168.3.4", |
| Acct_Session_Id=id, |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 403) |
| |
| logger.info("Disconnect-Request with mismatching NAS-Identifier") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_Identifier="unknown.example.com", |
| Acct_Session_Id=id, |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, 403) |
| |
| ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1) |
| if ev is not None: |
| raise Exception("Unexpected disconnection") |
| |
| logger.info("Disconnect-Request with matching Acct-Session-Id") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_IP_Address="127.0.0.1", |
| NAS_Identifier="nas.example.com", |
| Acct_Session_Id=id, |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| dev[0].wait_disconnected(timeout=10) |
| dev[0].wait_connected(timeout=10, error="Re-connection timed out") |
| |
| logger.info("Disconnect-Request with matching Acct-Multi-Session-Id") |
| sta = hapd.get_sta(addr) |
| multi_sess_id = sta['authMultiSessionId'] |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_IP_Address="127.0.0.1", |
| NAS_Identifier="nas.example.com", |
| Acct_Multi_Session_Id=multi_sess_id, |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| dev[0].wait_disconnected(timeout=10) |
| dev[0].wait_connected(timeout=10, error="Re-connection timed out") |
| |
| logger.info("Disconnect-Request with matching User-Name") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_Identifier="nas.example.com", |
| User_Name="psk.user@example.com", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| dev[0].wait_disconnected(timeout=10) |
| dev[0].wait_connected(timeout=10, error="Re-connection timed out") |
| |
| logger.info("Disconnect-Request with matching Calling-Station-Id") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_IP_Address="127.0.0.1", |
| Calling_Station_Id=addr, |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| dev[0].wait_disconnected(timeout=10) |
| ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED", "CTRL-EVENT-CONNECTED"]) |
| if ev is None: |
| raise Exception("Timeout while waiting for re-connection") |
| if "CTRL-EVENT-EAP-STARTED" not in ev: |
| raise Exception("Unexpected skipping of EAP authentication in reconnection") |
| dev[0].wait_connected(timeout=10, error="Re-connection timed out") |
| |
| logger.info("Disconnect-Request with matching Calling-Station-Id and non-matching CUI") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| Calling_Station_Id=addr, |
| Chargeable_User_Identity="foo@example.com", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, error_cause=503) |
| |
| logger.info("Disconnect-Request with matching CUI") |
| dev[1].connect("radius-das", key_mgmt="WPA-EAP", |
| eap="GPSK", identity="gpsk-cui", |
| password="abcdefghijklmnop0123456789abcdef", |
| scan_freq="2412") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| Chargeable_User_Identity="gpsk-chargeable-user-identity", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| dev[1].wait_disconnected(timeout=10) |
| dev[1].wait_connected(timeout=10, error="Re-connection timed out") |
| |
| ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1) |
| if ev is not None: |
| raise Exception("Unexpected disconnection") |
| |
| connect(dev[2], "radius-das") |
| |
| logger.info("Disconnect-Request with matching User-Name - multiple sessions matching") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_Identifier="nas.example.com", |
| User_Name="psk.user@example.com", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, error_cause=508) |
| |
| logger.info("Disconnect-Request with User-Name matching multiple sessions, Calling-Station-Id only one") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_Identifier="nas.example.com", |
| Calling_Station_Id=addr, |
| User_Name="psk.user@example.com", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| dev[0].wait_disconnected(timeout=10) |
| dev[0].wait_connected(timeout=10, error="Re-connection timed out") |
| |
| ev = dev[2].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1) |
| if ev is not None: |
| raise Exception("Unexpected disconnection") |
| |
| logger.info("Disconnect-Request with matching Acct-Multi-Session-Id after disassociation") |
| sta = hapd.get_sta(addr) |
| multi_sess_id = sta['authMultiSessionId'] |
| dev[0].request("DISCONNECT") |
| dev[0].wait_disconnected(timeout=10) |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_IP_Address="127.0.0.1", |
| NAS_Identifier="nas.example.com", |
| Acct_Multi_Session_Id=multi_sess_id, |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| dev[0].request("RECONNECT") |
| ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED"], timeout=15) |
| if ev is None: |
| raise Exception("Timeout on EAP start") |
| dev[0].wait_connected(timeout=15) |
| |
| logger.info("Disconnect-Request with matching User-Name after disassociation") |
| dev[0].request("DISCONNECT") |
| dev[0].wait_disconnected(timeout=10) |
| dev[2].request("DISCONNECT") |
| dev[2].wait_disconnected(timeout=10) |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_IP_Address="127.0.0.1", |
| NAS_Identifier="nas.example.com", |
| User_Name="psk.user@example.com", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| logger.info("Disconnect-Request with matching CUI after disassociation") |
| dev[1].request("DISCONNECT") |
| dev[1].wait_disconnected(timeout=10) |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_IP_Address="127.0.0.1", |
| NAS_Identifier="nas.example.com", |
| Chargeable_User_Identity="gpsk-chargeable-user-identity", |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| logger.info("Disconnect-Request with matching Calling-Station-Id after disassociation") |
| dev[0].request("RECONNECT") |
| ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED"], timeout=15) |
| if ev is None: |
| raise Exception("Timeout on EAP start") |
| dev[0].wait_connected(timeout=15) |
| dev[0].request("DISCONNECT") |
| dev[0].wait_disconnected(timeout=10) |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_IP_Address="127.0.0.1", |
| NAS_Identifier="nas.example.com", |
| Calling_Station_Id=addr, |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectACK) |
| |
| logger.info("Disconnect-Request with mismatching Calling-Station-Id after disassociation") |
| req = radius_das.DisconnectPacket(dict=dict, secret="secret", |
| NAS_IP_Address="127.0.0.1", |
| NAS_Identifier="nas.example.com", |
| Calling_Station_Id=addr, |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.DisconnectNAK, error_cause=503) |
| |
| def test_radius_das_coa(dev, apdev): |
| """RADIUS Dynamic Authorization Extensions - CoA""" |
| try: |
| import pyrad.client |
| import pyrad.packet |
| import pyrad.dictionary |
| import radius_das |
| except ImportError: |
| raise HwsimSkip("No pyrad modules available") |
| |
| params = hostapd.wpa2_eap_params(ssid="radius-das") |
| 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]['ifname'], params) |
| connect(dev[0], "radius-das") |
| addr = dev[0].p2p_interface_addr() |
| sta = hapd.get_sta(addr) |
| id = sta['dot1xAuthSessionId'] |
| |
| 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 |
| |
| # hostapd does not currently support CoA-Request, so NAK is expected |
| logger.info("CoA-Request with matching Acct-Session-Id") |
| req = radius_das.CoAPacket(dict=dict, secret="secret", |
| Acct_Session_Id=id, |
| Event_Timestamp=int(time.time())) |
| send_and_check_reply(srv, req, pyrad.packet.CoANAK, error_cause=405) |
| |
| def test_radius_ipv6(dev, apdev): |
| """RADIUS connection over IPv6""" |
| params = {} |
| params['ssid'] = 'as' |
| params['beacon_int'] = '2000' |
| params['radius_server_clients'] = 'auth_serv/radius_clients_ipv6.conf' |
| params['radius_server_ipv6'] = '1' |
| params['radius_server_auth_port'] = '18129' |
| params['radius_server_acct_port'] = '18139' |
| params['eap_server'] = '1' |
| params['eap_user_file'] = 'auth_serv/eap_user.conf' |
| params['ca_cert'] = 'auth_serv/ca.pem' |
| params['server_cert'] = 'auth_serv/server.pem' |
| params['private_key'] = 'auth_serv/server.key' |
| hostapd.add_ap(apdev[1]['ifname'], params) |
| |
| params = hostapd.wpa2_eap_params(ssid="radius-ipv6") |
| params['auth_server_addr'] = "::0" |
| params['auth_server_port'] = "18129" |
| params['acct_server_addr'] = "::0" |
| params['acct_server_port'] = "18139" |
| params['acct_server_shared_secret'] = "radius" |
| params['own_ip_addr'] = "::0" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| connect(dev[0], "radius-ipv6") |
| |
| def test_radius_macacl(dev, apdev): |
| """RADIUS MAC ACL""" |
| params = hostapd.radius_params() |
| params["ssid"] = "radius" |
| params["macaddr_acl"] = "2" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| dev[0].connect("radius", key_mgmt="NONE", scan_freq="2412") |
| |
| def test_radius_macacl_acct(dev, apdev): |
| """RADIUS MAC ACL and accounting enabled""" |
| params = hostapd.radius_params() |
| params["ssid"] = "radius" |
| params["macaddr_acl"] = "2" |
| params['acct_server_addr'] = "127.0.0.1" |
| params['acct_server_port'] = "1813" |
| params['acct_server_shared_secret'] = "radius" |
| hostapd.add_ap(apdev[0]['ifname'], params) |
| dev[0].connect("radius", key_mgmt="NONE", scan_freq="2412") |
| dev[1].connect("radius", key_mgmt="NONE", scan_freq="2412") |
| dev[1].request("DISCONNECT") |
| dev[1].wait_disconnected() |
| dev[1].request("RECONNECT") |
| |
| def test_radius_failover(dev, apdev): |
| """RADIUS Authentication and Accounting server failover""" |
| subprocess.call(['ip', 'ro', 'replace', '192.168.213.17', 'dev', 'lo']) |
| as_hapd = hostapd.Hostapd("as") |
| as_mib_start = as_hapd.get_mib(param="radius_server") |
| params = hostapd.wpa2_eap_params(ssid="radius-failover") |
| params["auth_server_addr"] = "192.168.213.17" |
| params["auth_server_port"] = "1812" |
| params["auth_server_shared_secret"] = "testing" |
| params['acct_server_addr'] = "192.168.213.17" |
| params['acct_server_port'] = "1813" |
| params['acct_server_shared_secret'] = "testing" |
| params['radius_retry_primary_interval'] = "20" |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params, no_enable=True) |
| hapd.set("auth_server_addr", "127.0.0.1") |
| hapd.set("auth_server_port", "1812") |
| hapd.set("auth_server_shared_secret", "radius") |
| hapd.set('acct_server_addr', "127.0.0.1") |
| hapd.set('acct_server_port', "1813") |
| hapd.set('acct_server_shared_secret', "radius") |
| hapd.enable() |
| ev = hapd.wait_event(["AP-ENABLED", "AP-DISABLED"], timeout=30) |
| if ev is None: |
| raise Exception("AP startup timed out") |
| if "AP-ENABLED" not in ev: |
| raise Exception("AP startup failed") |
| start = os.times()[4] |
| |
| try: |
| subprocess.call(['ip', 'ro', 'replace', 'prohibit', '192.168.213.17']) |
| dev[0].request("SET EAPOL::authPeriod 5") |
| connect(dev[0], "radius-failover", wait_connect=False) |
| dev[0].wait_connected(timeout=20) |
| finally: |
| dev[0].request("SET EAPOL::authPeriod 30") |
| subprocess.call(['ip', 'ro', 'del', '192.168.213.17']) |
| |
| as_mib_end = as_hapd.get_mib(param="radius_server") |
| req_s = int(as_mib_start['radiusAccServTotalRequests']) |
| req_e = int(as_mib_end['radiusAccServTotalRequests']) |
| if req_e <= req_s: |
| raise Exception("Unexpected RADIUS server acct MIB value") |
| |
| end = os.times()[4] |
| try: |
| subprocess.call(['ip', 'ro', 'replace', 'prohibit', '192.168.213.17']) |
| dev[1].request("SET EAPOL::authPeriod 5") |
| if end - start < 21: |
| time.sleep(21 - (end - start)) |
| connect(dev[1], "radius-failover", wait_connect=False) |
| dev[1].wait_connected(timeout=20) |
| finally: |
| dev[1].request("SET EAPOL::authPeriod 30") |
| subprocess.call(['ip', 'ro', 'del', '192.168.213.17']) |
| |
| def run_pyrad_server(srv, t_events): |
| srv.RunWithStop(t_events) |
| |
| def test_radius_protocol(dev, apdev): |
| """RADIUS Authentication protocol tests with a fake server""" |
| try: |
| import pyrad.server |
| import pyrad.packet |
| import pyrad.dictionary |
| except ImportError: |
| raise HwsimSkip("No pyrad modules available") |
| |
| class TestServer(pyrad.server.Server): |
| def _HandleAuthPacket(self, pkt): |
| pyrad.server.Server._HandleAuthPacket(self, pkt) |
| logger.info("Received authentication request") |
| reply = self.CreateReplyPacket(pkt) |
| reply.code = pyrad.packet.AccessAccept |
| if self.t_events['msg_auth'].is_set(): |
| logger.info("Add Message-Authenticator") |
| if self.t_events['wrong_secret'].is_set(): |
| logger.info("Use incorrect RADIUS shared secret") |
| pw = "incorrect" |
| else: |
| pw = reply.secret |
| hmac_obj = hmac.new(pw) |
| hmac_obj.update(struct.pack("B", reply.code)) |
| hmac_obj.update(struct.pack("B", reply.id)) |
| |
| # reply attributes |
| reply.AddAttribute("Message-Authenticator", |
| "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00") |
| attrs = reply._PktEncodeAttributes() |
| |
| # Length |
| flen = 4 + 16 + len(attrs) |
| hmac_obj.update(struct.pack(">H", flen)) |
| hmac_obj.update(pkt.authenticator) |
| hmac_obj.update(attrs) |
| if self.t_events['double_msg_auth'].is_set(): |
| logger.info("Include two Message-Authenticator attributes") |
| else: |
| del reply[80] |
| reply.AddAttribute("Message-Authenticator", hmac_obj.digest()) |
| self.SendReplyPacket(pkt.fd, reply) |
| |
| def RunWithStop(self, t_events): |
| self._poll = select.poll() |
| self._fdmap = {} |
| self._PrepareSockets() |
| self.t_events = t_events |
| |
| while not t_events['stop'].is_set(): |
| for (fd, event) in self._poll.poll(1000): |
| if event == select.POLLIN: |
| try: |
| fdo = self._fdmap[fd] |
| self._ProcessInput(fdo) |
| except ServerPacketError as err: |
| logger.info("pyrad server dropping packet: " + str(err)) |
| except pyrad.packet.PacketError as err: |
| logger.info("pyrad server received invalid packet: " + str(err)) |
| else: |
| logger.error("Unexpected event in pyrad server main loop") |
| |
| srv = TestServer(dict=pyrad.dictionary.Dictionary("dictionary.radius"), |
| authport=18138, acctport=18139) |
| srv.hosts["127.0.0.1"] = pyrad.server.RemoteHost("127.0.0.1", |
| "radius", |
| "localhost") |
| srv.BindToAddress("") |
| t_events = {} |
| t_events['stop'] = threading.Event() |
| t_events['msg_auth'] = threading.Event() |
| t_events['wrong_secret'] = threading.Event() |
| t_events['double_msg_auth'] = threading.Event() |
| t = threading.Thread(target=run_pyrad_server, args=(srv, t_events)) |
| t.start() |
| |
| try: |
| params = hostapd.wpa2_eap_params(ssid="radius-test") |
| params['auth_server_port'] = "18138" |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| connect(dev[0], "radius-test", wait_connect=False) |
| ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED"], timeout=15) |
| if ev is None: |
| raise Exception("Timeout on EAP start") |
| time.sleep(1) |
| dev[0].request("REMOVE_NETWORK all") |
| time.sleep(0.1) |
| dev[0].dump_monitor() |
| t_events['msg_auth'].set() |
| t_events['wrong_secret'].set() |
| connect(dev[0], "radius-test", wait_connect=False) |
| time.sleep(1) |
| dev[0].request("REMOVE_NETWORK all") |
| time.sleep(0.1) |
| dev[0].dump_monitor() |
| t_events['wrong_secret'].clear() |
| connect(dev[0], "radius-test", wait_connect=False) |
| time.sleep(1) |
| dev[0].request("REMOVE_NETWORK all") |
| time.sleep(0.1) |
| dev[0].dump_monitor() |
| t_events['double_msg_auth'].set() |
| connect(dev[0], "radius-test", wait_connect=False) |
| time.sleep(1) |
| finally: |
| t_events['stop'].set() |
| t.join() |
| |
| def test_radius_psk(dev, apdev): |
| """WPA2 with PSK from RADIUS""" |
| try: |
| import pyrad.server |
| import pyrad.packet |
| import pyrad.dictionary |
| except ImportError: |
| raise HwsimSkip("No pyrad modules available") |
| |
| class TestServer(pyrad.server.Server): |
| def _HandleAuthPacket(self, pkt): |
| pyrad.server.Server._HandleAuthPacket(self, pkt) |
| logger.info("Received authentication request") |
| reply = self.CreateReplyPacket(pkt) |
| reply.code = pyrad.packet.AccessAccept |
| a = "\xab\xcd" |
| secret = reply.secret |
| if self.t_events['long'].is_set(): |
| p = b'\x10' + "0123456789abcdef" + 15 * b'\x00' |
| b = hashlib.md5(secret + pkt.authenticator + a).digest() |
| pp = bytearray(p[0:16]) |
| bb = bytearray(b) |
| cc = bytearray(pp[i] ^ bb[i] for i in range(len(bb))) |
| |
| b = hashlib.md5(reply.secret + bytes(cc)).digest() |
| pp = bytearray(p[16:32]) |
| bb = bytearray(b) |
| cc += bytearray(pp[i] ^ bb[i] for i in range(len(bb))) |
| |
| data = '\x00' + a + bytes(cc) |
| else: |
| p = b'\x08' + "12345678" + 7 * b'\x00' |
| b = hashlib.md5(secret + pkt.authenticator + a).digest() |
| pp = bytearray(p) |
| bb = bytearray(b) |
| cc = bytearray(pp[i] ^ bb[i] for i in range(len(bb))) |
| data = '\x00' + a + bytes(cc) |
| reply.AddAttribute("Tunnel-Password", data) |
| self.SendReplyPacket(pkt.fd, reply) |
| |
| def RunWithStop(self, t_events): |
| self._poll = select.poll() |
| self._fdmap = {} |
| self._PrepareSockets() |
| self.t_events = t_events |
| |
| while not t_events['stop'].is_set(): |
| for (fd, event) in self._poll.poll(1000): |
| if event == select.POLLIN: |
| try: |
| fdo = self._fdmap[fd] |
| self._ProcessInput(fdo) |
| except ServerPacketError as err: |
| logger.info("pyrad server dropping packet: " + str(err)) |
| except pyrad.packet.PacketError as err: |
| logger.info("pyrad server received invalid packet: " + str(err)) |
| else: |
| logger.error("Unexpected event in pyrad server main loop") |
| |
| srv = TestServer(dict=pyrad.dictionary.Dictionary("dictionary.radius"), |
| authport=18138, acctport=18139) |
| srv.hosts["127.0.0.1"] = pyrad.server.RemoteHost("127.0.0.1", |
| "radius", |
| "localhost") |
| srv.BindToAddress("") |
| t_events = {} |
| t_events['stop'] = threading.Event() |
| t_events['long'] = threading.Event() |
| t = threading.Thread(target=run_pyrad_server, args=(srv, t_events)) |
| t.start() |
| |
| try: |
| ssid = "test-wpa2-psk" |
| params = hostapd.radius_params() |
| params['ssid'] = ssid |
| params["wpa"] = "2" |
| params["wpa_key_mgmt"] = "WPA-PSK" |
| params["rsn_pairwise"] = "CCMP" |
| params['macaddr_acl'] = '2' |
| params['wpa_psk_radius'] = '2' |
| params['auth_server_port'] = "18138" |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| dev[0].connect(ssid, psk="12345678", scan_freq="2412") |
| t_events['long'].set() |
| dev[1].connect(ssid, psk="0123456789abcdef", scan_freq="2412") |
| finally: |
| t_events['stop'].set() |
| t.join() |
| |
| def test_radius_auth_force_client_addr(dev, apdev): |
| """RADIUS client address specified""" |
| params = hostapd.wpa2_eap_params(ssid="radius-auth") |
| params['radius_client_addr'] = "127.0.0.1" |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| connect(dev[0], "radius-auth") |
| |
| def test_radius_auth_force_invalid_client_addr(dev, apdev): |
| """RADIUS client address specified and invalid address""" |
| params = hostapd.wpa2_eap_params(ssid="radius-auth") |
| #params['radius_client_addr'] = "10.11.12.14" |
| params['radius_client_addr'] = "1::2" |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| connect(dev[0], "radius-auth", wait_connect=False) |
| ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED"]) |
| if ev is None: |
| raise Exception("Timeout on EAP start") |
| ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=1) |
| if ev is not None: |
| raise Exception("Unexpected connection") |
| |
| def add_message_auth(req): |
| req.authenticator = req.CreateAuthenticator() |
| hmac_obj = hmac.new(req.secret) |
| hmac_obj.update(struct.pack("B", req.code)) |
| hmac_obj.update(struct.pack("B", req.id)) |
| |
| # request attributes |
| req.AddAttribute("Message-Authenticator", |
| "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00") |
| attrs = req._PktEncodeAttributes() |
| |
| # Length |
| flen = 4 + 16 + len(attrs) |
| hmac_obj.update(struct.pack(">H", flen)) |
| hmac_obj.update(req.authenticator) |
| hmac_obj.update(attrs) |
| del req[80] |
| req.AddAttribute("Message-Authenticator", hmac_obj.digest()) |
| |
| def test_radius_server_failures(dev, apdev): |
| """RADIUS server failure cases""" |
| try: |
| import pyrad.client |
| import pyrad.packet |
| import pyrad.dictionary |
| except ImportError: |
| raise HwsimSkip("No pyrad modules available") |
| |
| dict = pyrad.dictionary.Dictionary("dictionary.radius") |
| client = pyrad.client.Client(server="127.0.0.1", authport=1812, |
| secret="radius", dict=dict) |
| client.retries = 1 |
| client.timeout = 1 |
| |
| # unexpected State |
| req = client.CreateAuthPacket(code=pyrad.packet.AccessRequest, |
| User_Name="foo") |
| req['State'] = 'foo-state' |
| add_message_auth(req) |
| reply = client.SendPacket(req) |
| if reply.code != pyrad.packet.AccessReject: |
| raise Exception("Unexpected RADIUS response code " + str(reply.code)) |
| |
| # no EAP-Message |
| req = client.CreateAuthPacket(code=pyrad.packet.AccessRequest, |
| User_Name="foo") |
| add_message_auth(req) |
| try: |
| reply = client.SendPacket(req) |
| raise Exception("Unexpected response") |
| except pyrad.client.Timeout: |
| pass |
| |
| def test_ap_vlan_wpa2_psk_radius_required(dev, apdev): |
| """AP VLAN with WPA2-PSK and RADIUS attributes required""" |
| try: |
| import pyrad.server |
| import pyrad.packet |
| import pyrad.dictionary |
| except ImportError: |
| raise HwsimSkip("No pyrad modules available") |
| |
| class TestServer(pyrad.server.Server): |
| def _HandleAuthPacket(self, pkt): |
| pyrad.server.Server._HandleAuthPacket(self, pkt) |
| logger.info("Received authentication request") |
| reply = self.CreateReplyPacket(pkt) |
| reply.code = pyrad.packet.AccessAccept |
| secret = reply.secret |
| if self.t_events['extra'].is_set(): |
| reply.AddAttribute("Chargeable-User-Identity", "test-cui") |
| reply.AddAttribute("User-Name", "test-user") |
| if self.t_events['long'].is_set(): |
| reply.AddAttribute("Tunnel-Type", 13) |
| reply.AddAttribute("Tunnel-Medium-Type", 6) |
| reply.AddAttribute("Tunnel-Private-Group-ID", "1") |
| self.SendReplyPacket(pkt.fd, reply) |
| |
| def RunWithStop(self, t_events): |
| self._poll = select.poll() |
| self._fdmap = {} |
| self._PrepareSockets() |
| self.t_events = t_events |
| |
| while not t_events['stop'].is_set(): |
| for (fd, event) in self._poll.poll(1000): |
| if event == select.POLLIN: |
| try: |
| fdo = self._fdmap[fd] |
| self._ProcessInput(fdo) |
| except ServerPacketError as err: |
| logger.info("pyrad server dropping packet: " + str(err)) |
| except pyrad.packet.PacketError as err: |
| logger.info("pyrad server received invalid packet: " + str(err)) |
| else: |
| logger.error("Unexpected event in pyrad server main loop") |
| |
| srv = TestServer(dict=pyrad.dictionary.Dictionary("dictionary.radius"), |
| authport=18138, acctport=18139) |
| srv.hosts["127.0.0.1"] = pyrad.server.RemoteHost("127.0.0.1", |
| "radius", |
| "localhost") |
| srv.BindToAddress("") |
| t_events = {} |
| t_events['stop'] = threading.Event() |
| t_events['long'] = threading.Event() |
| t_events['extra'] = threading.Event() |
| t = threading.Thread(target=run_pyrad_server, args=(srv, t_events)) |
| t.start() |
| |
| try: |
| ssid = "test-wpa2-psk" |
| params = hostapd.radius_params() |
| params['ssid'] = ssid |
| params["wpa"] = "2" |
| params["wpa_key_mgmt"] = "WPA-PSK" |
| params["rsn_pairwise"] = "CCMP" |
| params['macaddr_acl'] = '2' |
| params['dynamic_vlan'] = "2" |
| params['wpa_passphrase'] = '0123456789abcdefghi' |
| params['auth_server_port'] = "18138" |
| hapd = hostapd.add_ap(apdev[0]['ifname'], params) |
| |
| logger.info("connecting without VLAN") |
| dev[0].connect(ssid, psk="0123456789abcdefghi", scan_freq="2412", |
| wait_connect=False) |
| ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED", |
| "CTRL-EVENT-SSID-TEMP-DISABLED"], timeout=20) |
| if ev is None: |
| raise Exception("Timeout on connection attempt") |
| if "CTRL-EVENT-CONNECTED" in ev: |
| raise Exception("Unexpected success without vlan parameters") |
| logger.info("connecting without VLAN failed as expected") |
| |
| logger.info("connecting without VLAN (CUI/User-Name)") |
| t_events['extra'].set() |
| dev[1].connect(ssid, psk="0123456789abcdefghi", scan_freq="2412", |
| wait_connect=False) |
| ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED", |
| "CTRL-EVENT-SSID-TEMP-DISABLED"], timeout=20) |
| if ev is None: |
| raise Exception("Timeout on connection attempt") |
| if "CTRL-EVENT-CONNECTED" in ev: |
| raise Exception("Unexpected success without vlan parameters(2)") |
| logger.info("connecting without VLAN failed as expected(2)") |
| t_events['extra'].clear() |
| |
| t_events['long'].set() |
| logger.info("connecting with VLAN") |
| dev[2].connect(ssid, psk="0123456789abcdefghi", scan_freq="2412", |
| wait_connect=False) |
| ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED", |
| "CTRL-EVENT-SSID-TEMP-DISABLED"], timeout=20) |
| if ev is None: |
| raise Exception("Timeout on connection attempt") |
| if "CTRL-EVENT-SSID-TEMP-DISABLED" in ev: |
| raise Exception("Unexpected failure with vlan parameters") |
| logger.info("connecting with VLAN succeeded as expected") |
| finally: |
| t_events['stop'].set() |
| t.join() |
