1 # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 import os 6 import subprocess 7 8 from autotest_lib.client.bin import utils 9 from autotest_lib.client.common_lib.cros import site_eap_certs 10 11 class HostapdServer(object): 12 """Hostapd server instance wrapped in a context manager. 13 14 Simple interface to starting and controlling a hsotapd instance. 15 This can be combined with a virtual-ethernet setup to test 802.1x 16 on a wired interface. 17 18 Example usage: 19 with hostapd_server.HostapdServer(interface='veth_master') as hostapd: 20 hostapd.send_eap_packets() 21 22 """ 23 CONFIG_TEMPLATE = """ 24 interface=%(interface)s 25 driver=%(driver)s 26 logger_syslog=-1 27 logger_syslog_level=2 28 logger_stdout=-1 29 logger_stdout_level=2 30 dump_file=%(config_directory)s/hostapd.dump 31 ctrl_interface=%(config_directory)s/%(control_directory)s 32 ieee8021x=1 33 eapol_key_index_workaround=0 34 eap_server=1 35 eap_user_file=%(config_directory)s/%(user_file)s 36 ca_cert=%(config_directory)s/%(ca_cert)s 37 server_cert=%(config_directory)s/%(server_cert)s 38 private_key=%(config_directory)s/%(server_key)s 39 use_pae_group_addr=1 40 eap_reauth_period=10 41 """ 42 CA_CERTIFICATE_FILE = 'ca.crt' 43 CONFIG_FILE = 'hostapd.conf' 44 CONTROL_DIRECTORY = 'hostapd.ctl' 45 EAP_PASSWORD = 'password' 46 EAP_PHASE2 = 'MSCHAPV2' 47 EAP_TYPE = 'PEAP' 48 EAP_USERNAME = 'test' 49 HOSTAPD_EXECUTABLE = 'hostapd' 50 HOSTAPD_CLIENT_EXECUTABLE = 'hostapd_cli' 51 SERVER_CERTIFICATE_FILE = 'server.crt' 52 SERVER_PRIVATE_KEY_FILE = 'server.key' 53 USER_AUTHENTICATION_TEMPLATE = """* %(type)s 54 "%(username)s"\t%(phase2)s\t"%(password)s"\t[2] 55 """ 56 USER_FILE = 'hostapd.eap_user' 57 # This is the default group MAC address to which EAP challenges 58 # are sent, absent any prior knowledge of a specific client on 59 # the link. 60 PAE_NEAREST_ADDRESS = '01:80:c2:00:00:03' 61 62 def __init__(self, 63 interface=None, 64 driver='wired', 65 config_directory='/tmp/hostapd-test'): 66 super(HostapdServer, self).__init__() 67 self._interface = interface 68 self._config_directory = config_directory 69 self._control_directory = '%s/%s' % (self._config_directory, 70 self.CONTROL_DIRECTORY) 71 self._driver = driver 72 self._process = None 73 74 75 def __enter__(self): 76 self.start() 77 return self 78 79 80 def __exit__(self, exception, value, traceback): 81 self.stop() 82 83 84 def write_config(self): 85 """Write out a hostapd configuration file-set based on the caller 86 supplied parameters. 87 88 @return the file name of the top-level configuration file written. 89 90 """ 91 if not os.path.exists(self._config_directory): 92 os.mkdir(self._config_directory) 93 config_params = { 94 'ca_cert': self.CA_CERTIFICATE_FILE, 95 'config_directory' : self._config_directory, 96 'control_directory': self.CONTROL_DIRECTORY, 97 'driver': self._driver, 98 'interface': self._interface, 99 'server_cert': self.SERVER_CERTIFICATE_FILE, 100 'server_key': self.SERVER_PRIVATE_KEY_FILE, 101 'user_file': self.USER_FILE 102 } 103 authentication_params = { 104 'password': self.EAP_PASSWORD, 105 'phase2': self.EAP_PHASE2, 106 'username': self.EAP_USERNAME, 107 'type': self.EAP_TYPE 108 } 109 for filename, contents in ( 110 ( self.CA_CERTIFICATE_FILE, site_eap_certs.ca_cert_1 ), 111 ( self.CONFIG_FILE, self.CONFIG_TEMPLATE % config_params), 112 ( self.SERVER_CERTIFICATE_FILE, site_eap_certs.server_cert_1 ), 113 ( self.SERVER_PRIVATE_KEY_FILE, 114 site_eap_certs.server_private_key_1 ), 115 ( self.USER_FILE, 116 self.USER_AUTHENTICATION_TEMPLATE % authentication_params )): 117 config_file = '%s/%s' % (self._config_directory, filename) 118 with open(config_file, 'w') as f: 119 f.write(contents) 120 return '%s/%s' % (self._config_directory, self.CONFIG_FILE) 121 122 123 def start(self): 124 """Start the hostap server.""" 125 config_file = self.write_config() 126 self._process = subprocess.Popen( 127 [self.HOSTAPD_EXECUTABLE, '-dd', config_file]) 128 129 130 def stop(self): 131 """Stop the hostapd server.""" 132 if self._process: 133 self._process.terminate() 134 self._process.wait() 135 self._process = None 136 137 138 def running(self): 139 """Tests whether the hostapd process is still running. 140 141 @return True if the hostapd process is still running, False otherwise. 142 143 """ 144 if not self._process: 145 return False 146 147 if self._process.poll() != None: 148 # We have essentially reaped the proces, and it is no more. 149 self._process = None 150 return False 151 152 return True 153 154 155 def send_eap_packets(self): 156 """Start sending EAP packets to the nearest neighbor.""" 157 self.send_command('new_sta %s' % self.PAE_NEAREST_ADDRESS) 158 159 160 def get_client_mib(self, client_mac_address): 161 """Get a dict representing the MIB properties for |client_mac_address|. 162 163 @param client_mac_address string MAC address of the client. 164 @return dict containing mib properties. 165 166 """ 167 # Expected output of "hostapd cli <client_mac_address>": 168 # 169 # Selected interface 'veth_master' 170 # b6:f1:39:1d:ad:10 171 # dot1xPaePortNumber=0 172 # dot1xPaePortProtocolVersion=2 173 # [...] 174 result = self.send_command('sta %s' % client_mac_address) 175 client_mib = {} 176 found_client = False 177 for line in result.splitlines(): 178 if found_client: 179 parts = line.split('=', 1) 180 if len(parts) == 2: 181 client_mib[parts[0]] = parts[1] 182 elif line == client_mac_address: 183 found_client = True 184 return client_mib 185 186 187 def send_command(self, command): 188 """Send a command to the hostapd instance. 189 190 @param command string containing the command to run on hostapd. 191 @return string output of the command. 192 193 """ 194 return utils.system_output('%s -p %s %s' % 195 (self.HOSTAPD_CLIENT_EXECUTABLE, 196 self._control_directory, command)) 197 198 199 def client_has_authenticated(self, client_mac_address): 200 """Return whether |client_mac_address| has successfully authenticated. 201 202 @param client_mac_address string MAC address of the client. 203 @return True if client is authenticated. 204 205 """ 206 mib = self.get_client_mib(client_mac_address) 207 return mib.get('dot1xAuthAuthSuccessesWhileAuthenticating', '') == '1' 208 209 210 def client_has_logged_off(self, client_mac_address): 211 """Return whether |client_mac_address| has logged-off. 212 213 @param client_mac_address string MAC address of the client. 214 @return True if client has logged off. 215 216 """ 217 mib = self.get_client_mib(client_mac_address) 218 return mib.get('dot1xAuthAuthEapLogoffWhileAuthenticated', '') == '1' 219
