client/cros/hostapd_server.py

# 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'