# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import os
import subprocess
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib.cros import site_eap_certs
class HostapdServer(object):
"""Hostapd server instance wrapped in a context manager.
Simple interface to starting and controlling a hsotapd instance.
This can be combined with a virtual-ethernet setup to test 802.1x
on a wired interface.
Example usage:
with hostapd_server.HostapdServer(interface='veth_master') as hostapd:
hostapd.send_eap_packets()
"""
CONFIG_TEMPLATE = """
interface=%(interface)s
driver=%(driver)s
logger_syslog=-1
logger_syslog_level=2
logger_stdout=-1
logger_stdout_level=2
dump_file=%(config_directory)s/hostapd.dump
ctrl_interface=%(config_directory)s/%(control_directory)s
ieee8021x=1
eapol_key_index_workaround=0
eap_server=1
eap_user_file=%(config_directory)s/%(user_file)s
ca_cert=%(config_directory)s/%(ca_cert)s
server_cert=%(config_directory)s/%(server_cert)s
private_key=%(config_directory)s/%(server_key)s
use_pae_group_addr=1
eap_reauth_period=10
"""
CA_CERTIFICATE_FILE = 'ca.crt'
CONFIG_FILE = 'hostapd.conf'
CONTROL_DIRECTORY = 'hostapd.ctl'
EAP_PASSWORD = 'password'
EAP_PHASE2 = 'MSCHAPV2'
EAP_TYPE = 'PEAP'
EAP_USERNAME = 'test'
HOSTAPD_EXECUTABLE = 'hostapd'
HOSTAPD_CLIENT_EXECUTABLE = 'hostapd_cli'
SERVER_CERTIFICATE_FILE = 'server.crt'
SERVER_PRIVATE_KEY_FILE = 'server.key'
USER_AUTHENTICATION_TEMPLATE = """* %(type)s
"%(username)s"\t%(phase2)s\t"%(password)s"\t[2]
"""
USER_FILE = 'hostapd.eap_user'
# This is the default group MAC address to which EAP challenges
# are sent, absent any prior knowledge of a specific client on
# the link.
PAE_NEAREST_ADDRESS = '01:80:c2:00:00:03'
def __init__(self,
interface=None,
driver='wired',
config_directory='/tmp/hostapd-test'):
super(HostapdServer, self).__init__()
self._interface = interface
self._config_directory = config_directory
self._control_directory = '%s/%s' % (self._config_directory,
self.CONTROL_DIRECTORY)
self._driver = driver
self._process = None
def __enter__(self):
self.start()
return self
def __exit__(self, exception, value, traceback):
self.stop()
def write_config(self):
"""Write out a hostapd configuration file-set based on the caller
supplied parameters.
@return the file name of the top-level configuration file written.
"""
if not os.path.exists(self._config_directory):
os.mkdir(self._config_directory)
config_params = {
'ca_cert': self.CA_CERTIFICATE_FILE,
'config_directory' : self._config_directory,
'control_directory': self.CONTROL_DIRECTORY,
'driver': self._driver,
'interface': self._interface,
'server_cert': self.SERVER_CERTIFICATE_FILE,
'server_key': self.SERVER_PRIVATE_KEY_FILE,
'user_file': self.USER_FILE
}
authentication_params = {
'password': self.EAP_PASSWORD,
'phase2': self.EAP_PHASE2,
'username': self.EAP_USERNAME,
'type': self.EAP_TYPE
}
for filename, contents in (
( self.CA_CERTIFICATE_FILE, site_eap_certs.ca_cert_1 ),
( self.CONFIG_FILE, self.CONFIG_TEMPLATE % config_params),
( self.SERVER_CERTIFICATE_FILE, site_eap_certs.server_cert_1 ),
( self.SERVER_PRIVATE_KEY_FILE,
site_eap_certs.server_private_key_1 ),
( self.USER_FILE,
self.USER_AUTHENTICATION_TEMPLATE % authentication_params )):
config_file = '%s/%s' % (self._config_directory, filename)
with open(config_file, 'w') as f:
f.write(contents)
return '%s/%s' % (self._config_directory, self.CONFIG_FILE)
def start(self):
"""Start the hostap server."""
config_file = self.write_config()
self._process = subprocess.Popen(
[self.HOSTAPD_EXECUTABLE, '-dd', config_file])
def stop(self):
"""Stop the hostapd server."""
if self._process:
self._process.terminate()
self._process.wait()
self._process = None
def running(self):
"""Tests whether the hostapd process is still running.
@return True if the hostapd process is still running, False otherwise.
"""
if not self._process:
return False
if self._process.poll() != None:
# We have essentially reaped the proces, and it is no more.
self._process = None
return False
return True
def send_eap_packets(self):
"""Start sending EAP packets to the nearest neighbor."""
self.send_command('new_sta %s' % self.PAE_NEAREST_ADDRESS)
def get_client_mib(self, client_mac_address):
"""Get a dict representing the MIB properties for |client_mac_address|.
@param client_mac_address string MAC address of the client.
@return dict containing mib properties.
"""
# Expected output of "hostapd cli <client_mac_address>":
#
# Selected interface 'veth_master'
# b6:f1:39:1d:ad:10
# dot1xPaePortNumber=0
# dot1xPaePortProtocolVersion=2
# [...]
result = self.send_command('sta %s' % client_mac_address)
client_mib = {}
found_client = False
for line in result.splitlines():
if found_client:
parts = line.split('=', 1)
if len(parts) == 2:
client_mib[parts[0]] = parts[1]
elif line == client_mac_address:
found_client = True
return client_mib
def send_command(self, command):
"""Send a command to the hostapd instance.
@param command string containing the command to run on hostapd.
@return string output of the command.
"""
return utils.system_output('%s -p %s %s' %
(self.HOSTAPD_CLIENT_EXECUTABLE,
self._control_directory, command))
def client_has_authenticated(self, client_mac_address):
"""Return whether |client_mac_address| has successfully authenticated.
@param client_mac_address string MAC address of the client.
@return True if client is authenticated.
"""
mib = self.get_client_mib(client_mac_address)
return mib.get('dot1xAuthAuthSuccessesWhileAuthenticating', '') == '1'
def client_has_logged_off(self, client_mac_address):
"""Return whether |client_mac_address| has logged-off.
@param client_mac_address string MAC address of the client.
@return True if client has logged off.
"""
mib = self.get_client_mib(client_mac_address)
return mib.get('dot1xAuthAuthEapLogoffWhileAuthenticated', '') == '1'