krackattacks/tests/hwsim/test_ap_config.py

# hostapd configuration tests
# Copyright (c) 2014-2016, Jouni Malinen <j@w1.fi>
#
# This software may be distributed under the terms of the BSD license.
# See README for more details.

import os
import signal
import time

from remotehost import remote_compatible
import hostapd
from utils import alloc_fail

@remote_compatible
def test_ap_config_errors(dev, apdev):
    """Various hostapd configuration errors"""

    # IEEE 802.11d without country code
    params = { "ssid": "foo", "ieee80211d": "1" }
    hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
    if "FAIL" not in hapd.request("ENABLE"):
        raise Exception("Unexpected ENABLE success (ieee80211d without country_code)")
    hostapd.remove_bss(apdev[0])

    # IEEE 802.11h without IEEE 802.11d
    params = { "ssid": "foo", "ieee80211h": "1" }
    hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
    if "FAIL" not in hapd.request("ENABLE"):
        raise Exception("Unexpected ENABLE success (ieee80211h without ieee80211d")
    hostapd.remove_bss(apdev[0])

    # Power Constraint without IEEE 802.11d
    params = { "ssid": "foo", "local_pwr_constraint": "1" }
    hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
    if "FAIL" not in hapd.request("ENABLE"):
        raise Exception("Unexpected ENABLE success (local_pwr_constraint without ieee80211d)")
    hostapd.remove_bss(apdev[0])

    # Spectrum management without Power Constraint
    params = { "ssid": "foo", "spectrum_mgmt_required": "1" }
    hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
    if "FAIL" not in hapd.request("ENABLE"):
        raise Exception("Unexpected ENABLE success (spectrum_mgmt_required without local_pwr_constraint)")
    hostapd.remove_bss(apdev[0])

    # IEEE 802.1X without authentication server
    params = { "ssid": "foo", "ieee8021x": "1" }
    hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
    if "FAIL" not in hapd.request("ENABLE"):
        raise Exception("Unexpected ENABLE success (ieee8021x)")
    hostapd.remove_bss(apdev[0])

    # RADIUS-PSK without macaddr_acl=2
    params = hostapd.wpa2_params(ssid="foo", passphrase="12345678")
    params["wpa_psk_radius"] = "1"
    hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
    if "FAIL" not in hapd.request("ENABLE"):
        raise Exception("Unexpected ENABLE success (wpa_psk_radius)")
    hostapd.remove_bss(apdev[0])

    # FT without NAS-Identifier
    params = { "wpa": "2",
               "wpa_key_mgmt": "FT-PSK",
               "rsn_pairwise": "CCMP",
               "wpa_passphrase": "12345678" }
    hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
    if "FAIL" not in hapd.request("ENABLE"):
        raise Exception("Unexpected ENABLE success (FT without nas_identifier)")
    hostapd.remove_bss(apdev[0])

    # Hotspot 2.0 without WPA2/CCMP
    params = hostapd.wpa2_params(ssid="foo")
    params['wpa_key_mgmt'] = "WPA-EAP"
    params['ieee8021x'] = "1"
    params['auth_server_addr'] = "127.0.0.1"
    params['auth_server_port'] = "1812"
    params['auth_server_shared_secret'] = "radius"
    params['interworking'] = "1"
    params['hs20'] = "1"
    params['wpa'] = "1"
    hapd = hostapd.add_ap(apdev[0], params, no_enable=True)
    if "FAIL" not in hapd.request("ENABLE"):
        raise Exception("Unexpected ENABLE success (HS 2.0 without WPA2/CCMP)")
    hostapd.remove_bss(apdev[0])

def test_ap_config_reload(dev, apdev, params):
    """hostapd configuration reload"""
    hapd = hostapd.add_ap(apdev[0], { "ssid": "foo" })
    hapd.set("ssid", "foobar")
    with open(os.path.join(params['logdir'], 'hostapd-test.pid'), "r") as f:
        pid = int(f.read())
    os.kill(pid, signal.SIGHUP)
    time.sleep(0.1)
    dev[0].connect("foobar", key_mgmt="NONE", scan_freq="2412")
    hapd.set("ssid", "foo")
    os.kill(pid, signal.SIGHUP)
    dev[0].wait_disconnected()
    dev[0].request("DISCONNECT")

def test_ap_config_reload_file(dev, apdev, params):
    """hostapd configuration reload from file"""
    hapd = hostapd.add_iface(apdev[0], "bss-1.conf")
    hapd.enable()
    hapd.set("ssid", "foobar")
    with open(os.path.join(params['logdir'], 'hostapd-test.pid'), "r") as f:
        pid = int(f.read())
    os.kill(pid, signal.SIGHUP)
    time.sleep(0.1)
    dev[0].connect("foobar", key_mgmt="NONE", scan_freq="2412")
    hapd.set("ssid", "foo")
    os.kill(pid, signal.SIGHUP)
    dev[0].wait_disconnected()
    dev[0].request("DISCONNECT")

def test_ap_config_reload_before_enable(dev, apdev, params):
    """hostapd configuration reload before enable"""
    hapd = hostapd.add_iface(apdev[0], "bss-1.conf")
    with open(os.path.join(params['logdir'], 'hostapd-test.pid'), "r") as f:
        pid = int(f.read())
    os.kill(pid, signal.SIGHUP)
    hapd.ping()

def test_ap_config_sigusr1(dev, apdev, params):
    """hostapd SIGUSR1"""
    hapd = hostapd.add_ap(apdev[0], { "ssid": "foobar" })
    with open(os.path.join(params['logdir'], 'hostapd-test.pid'), "r") as f:
        pid = int(f.read())
    os.kill(pid, signal.SIGUSR1)
    dev[0].connect("foobar", key_mgmt="NONE", scan_freq="2412")
    os.kill(pid, signal.SIGUSR1)

def test_ap_config_invalid_value(dev, apdev, params):
    """Ignoring invalid hostapd configuration parameter updates"""
    hapd = hostapd.add_ap(apdev[0], { "ssid": "test" }, no_enable=True)
    not_exist = "/tmp/hostapd-test/does-not-exist"
    tests = [ ("driver", "foobar"),
              ("ssid2", "Q"),
              ("macaddr_acl", "255"),
              ("accept_mac_file", not_exist),
              ("deny_mac_file", not_exist),
              ("eapol_version", "255"),
              ("eap_user_file", not_exist),
              ("wep_key_len_broadcast", "-1"),
              ("wep_key_len_unicast", "-1"),
              ("wep_rekey_period", "-1"),
              ("eap_rekey_period", "-1"),
              ("radius_client_addr", "foo"),
              ("acs_chan_bias", "-1:0.8"),
              ("acs_chan_bias", "1"),
              ("acs_chan_bias", "1:p"),
              ("acs_chan_bias", "1:-0.8"),
              ("acs_chan_bias", "1:0.8p"),
              ("dtim_period", "0"),
              ("bss_load_update_period", "-1"),
              ("send_probe_response", "255"),
              ("beacon_rate", "ht:-1"),
              ("beacon_rate", "ht:32"),
              ("beacon_rate", "vht:-1"),
              ("beacon_rate", "vht:10"),
              ("beacon_rate", "9"),
              ("beacon_rate", "10001"),
              ("vlan_file", not_exist),
              ("bss", ""),
              ("bssid", "foo"),
              ("extra_cred", not_exist),
              ("anqp_elem", "265"),
              ("anqp_elem", "265"),
              ("anqp_elem", "265:1"),
              ("anqp_elem", "265:1q"),
              ("fst_priority", ""),
              ("fils_cache_id", "q"),
              ("unknown-item", "foo") ]
    for field, val in tests:
        if "FAIL" not in hapd.request("SET %s %s" % (field, val)):
            raise Exception("Invalid %s accepted" % field)
    hapd.enable()
    dev[0].connect("test", key_mgmt="NONE", scan_freq="2412")

def test_ap_config_eap_user_file_parsing(dev, apdev, params):
    """hostapd eap_user_file parsing"""
    tmp = os.path.join(params['logdir'], 'ap_vlan_file_parsing.tmp')
    hapd = hostapd.add_ap(apdev[0], { "ssid": "foobar" })

    for i in range(2):
        if "OK" not in hapd.request("SET eap_user_file auth_serv/eap_user.conf"):
            raise Exception("eap_user_file rejected")

    tests = [ "#\n\n*\tTLS\nradius_accept_attr=:",
              "foo\n",
              "\"foo\n",
              "\"foo\"\n",
              "\"foo\" FOOBAR\n",
              "\"foo\" " + 10*"TLS," + "TLS \"\n",
              "\"foo\" TLS \nfoo\n",
              "\"foo\" PEAP hash:foo\n",
              "\"foo\" PEAP hash:8846f7eaee8fb117ad06bdd830b7586q\n",
              "\"foo\" PEAP 01020\n",
              "\"foo\" PEAP 010q\n",
              "\"foo\" TLS\nradius_accept_attr=123:x:012\n",
              "\"foo\" TLS\nradius_accept_attr=123:x:012q\n",
              "\"foo\" TLS\nradius_accept_attr=123:Q:01\n",
              "\"foo\" TLS\nradius_accept_attr=123\nfoo\n" ]
    for t in tests:
        with open(tmp, "w") as f:
            f.write(t)
        if "FAIL" not in hapd.request("SET eap_user_file " + tmp):
            raise Exception("Invalid eap_user_file accepted")

    tests = [ ("\"foo\" TLS\n", 2, "hostapd_config_read_eap_user"),
              ("\"foo\" PEAP \"foo\"\n", 3, "hostapd_config_read_eap_user"),
              ("\"foo\" PEAP hash:8846f7eaee8fb117ad06bdd830b75861\n", 3,
               "hostapd_config_read_eap_user"),
              ("\"foo\" PEAP 0102\n", 3, "hostapd_config_read_eap_user"),
              ("\"foo\" TLS\nradius_accept_attr=123\n", 1,
               "=hostapd_parse_radius_attr"),
              ("\"foo\" TLS\nradius_accept_attr=123\n", 1,
               "wpabuf_alloc;hostapd_parse_radius_attr"),
              ("\"foo\" TLS\nradius_accept_attr=123:s:foo\n", 2,
               "hostapd_parse_radius_attr"),
              ("\"foo\" TLS\nradius_accept_attr=123:x:0102\n", 2,
               "hostapd_parse_radius_attr"),
              ("\"foo\" TLS\nradius_accept_attr=123:d:1\n", 2,
               "hostapd_parse_radius_attr"),
              ("* TLS\n", 1, "hostapd_config_read_eap_user") ]
    for t, count, func in tests:
        with alloc_fail(hapd, count, func):
            with open(tmp, "w") as f:
                f.write(t)
            if "FAIL" not in hapd.request("SET eap_user_file " + tmp):
                raise Exception("eap_user_file accepted during OOM")