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 5import dbus 6import logging 7import subprocess 8import time 9 10from autotest_lib.client.cros.networking import shill_proxy 11 12 13class WifiProxy(shill_proxy.ShillProxy): 14 """Wrapper around shill dbus interface used by wifi tests.""" 15 16 17 def set_logging_for_wifi_test(self): 18 """Set the logging in shill for a test of wifi technology. 19 20 Set the log level to |ShillProxy.LOG_LEVEL_FOR_TEST| and the log scopes 21 to the ones defined in |ShillProxy.LOG_SCOPES_FOR_TEST| for 22 |ShillProxy.TECHNOLOGY_WIFI|. 23 24 """ 25 self.set_logging_for_test(self.TECHNOLOGY_WIFI) 26 27 28 def remove_all_wifi_entries(self): 29 """Iterate over all pushed profiles and remove WiFi entries.""" 30 profiles = self.get_profiles() 31 for profile in profiles: 32 profile_properties = profile.GetProperties(utf8_strings=True) 33 entries = profile_properties[self.PROFILE_PROPERTY_ENTRIES] 34 for entry_id in entries: 35 try: 36 entry = profile.GetEntry(entry_id) 37 except dbus.exceptions.DBusException as e: 38 logging.error('Unable to retrieve entry %s', entry_id) 39 continue 40 if entry[self.ENTRY_FIELD_TYPE] == 'wifi': 41 profile.DeleteEntry(entry_id) 42 43 44 def configure_wifi_service(self, ssid, security, security_parameters={}, 45 save_credentials=True, station_type=None, 46 hidden_network=False, guid=None, 47 autoconnect=None): 48 """Configure a WiFi service. 49 50 @param ssid string name of network to connect to. 51 @param security string type of security used in network (e.g. psk) 52 @param security_parameters dict of service property/value pairs that 53 make up the credentials and settings for the given security 54 type (e.g. the passphrase for psk security). 55 @param save_credentials bool True if we should save EAP credentials. 56 @param station_type string one of SUPPORTED_WIFI_STATION_TYPES. 57 @param hidden_network bool True when the SSID is not broadcasted. 58 @param guid string unique identifier for network. 59 @param autoconnect bool or None. None indicates that this should not 60 be set one way or the other, while a boolean indicates a desired 61 value. 62 63 """ 64 # |mode| is derived from the station type we're attempting to join. It 65 # does not refer to the 802.11x (802.11a/b/g/n) type. It refers to a 66 # shill connection mode. 67 mode = self.SUPPORTED_WIFI_STATION_TYPES[station_type] 68 config_params = {self.SERVICE_PROPERTY_TYPE: 'wifi', 69 self.SERVICE_PROPERTY_HIDDEN: hidden_network, 70 self.SERVICE_PROPERTY_SSID: ssid, 71 self.SERVICE_PROPERTY_SECURITY_CLASS: security, 72 self.SERVICE_PROPERTY_MODE: mode} 73 if autoconnect is not None: 74 config_params[self.SERVICE_PROPERTY_AUTOCONNECT] = autoconnect 75 config_params.update(security_parameters) 76 if guid is not None: 77 config_params[self.SERVICE_PROPERTY_GUID] = guid 78 try: 79 self.configure_service(config_params) 80 except dbus.exceptions.DBusException as e: 81 logging.error('Caught an error while configuring a WiFi ' 82 'service: %r', e) 83 return False 84 85 logging.info('Configured service: %s', ssid) 86 return True 87 88 89 def connect_to_wifi_network(self, 90 ssid, 91 security, 92 security_parameters, 93 save_credentials, 94 station_type=None, 95 hidden_network=False, 96 guid=None, 97 autoconnect=None, 98 discovery_timeout_seconds=15, 99 association_timeout_seconds=15, 100 configuration_timeout_seconds=15): 101 """ 102 Connect to a WiFi network with the given association parameters. 103 104 @param ssid string name of network to connect to. 105 @param security string type of security used in network (e.g. psk) 106 @param security_parameters dict of service property/value pairs that 107 make up the credentials and settings for the given security 108 type (e.g. the passphrase for psk security). 109 @param save_credentials bool True if we should save EAP credentials. 110 @param station_type string one of SUPPORTED_WIFI_STATION_TYPES. 111 @param hidden_network bool True when the SSID is not broadcasted. 112 @param guid string unique identifier for network. 113 @param discovery_timeout_seconds float timeout for service discovery. 114 @param association_timeout_seconds float timeout for service 115 association. 116 @param configuration_timeout_seconds float timeout for DHCP 117 negotiations. 118 @param autoconnect: bool or None. None indicates that this should not 119 be set one way or the other, while a boolean indicates a desired 120 value. 121 @return (successful, discovery_time, association_time, 122 configuration_time, reason) 123 where successful is True iff the operation succeeded, *_time is 124 the time spent waiting for each transition, and reason is a string 125 which may contain a meaningful description of failures. 126 127 """ 128 logging.info('Attempting to connect to %s', ssid) 129 service_proxy = None 130 start_time = time.time() 131 discovery_time = -1.0 132 association_time = -1.0 133 configuration_time = -1.0 134 if station_type not in self.SUPPORTED_WIFI_STATION_TYPES: 135 return (False, discovery_time, association_time, 136 configuration_time, 137 'FAIL(Invalid station type specified.)') 138 139 # |mode| is derived from the station type we're attempting to join. It 140 # does not refer to the 802.11x (802.11a/b/g/n) type. It refers to a 141 # shill connection mode. 142 mode = self.SUPPORTED_WIFI_STATION_TYPES[station_type] 143 144 if hidden_network: 145 logging.info('Configuring %s as a hidden network.', ssid) 146 if not self.configure_wifi_service( 147 ssid, security, save_credentials=save_credentials, 148 station_type=station_type, hidden_network=True, 149 autoconnect=autoconnect): 150 return (False, discovery_time, association_time, 151 configuration_time, 152 'FAIL(Failed to configure hidden SSID)') 153 154 logging.info('Configured hidden service: %s', ssid) 155 156 157 logging.info('Discovering...') 158 discovery_params = {self.SERVICE_PROPERTY_TYPE: 'wifi', 159 self.SERVICE_PROPERTY_NAME: ssid, 160 self.SERVICE_PROPERTY_SECURITY_CLASS: security, 161 self.SERVICE_PROPERTY_MODE: mode} 162 while time.time() - start_time < discovery_timeout_seconds: 163 discovery_time = time.time() - start_time 164 service_object = self.find_matching_service(discovery_params) 165 if service_object: 166 try: 167 service_properties = service_object.GetProperties( 168 utf8_strings=True) 169 except dbus.exceptions.DBusException: 170 # This usually means the service handle has become invalid. 171 # Which is sort of like not getting a handle back from 172 # find_matching_service in the first place. 173 continue 174 strength = self.dbus2primitive( 175 service_properties[self.SERVICE_PROPERTY_STRENGTH]) 176 if strength > 0: 177 logging.info('Discovered service: %s. Strength: %r.', 178 ssid, strength) 179 break 180 181 # This is spammy, but shill handles that for us. 182 self.manager.RequestScan('wifi') 183 time.sleep(self.POLLING_INTERVAL_SECONDS) 184 else: 185 return (False, discovery_time, association_time, 186 configuration_time, 'FAIL(Discovery timed out)') 187 188 # At this point, we know |service| is in the service list. Attempt 189 # to connect it, and watch the states roll by. 190 logging.info('Connecting...') 191 try: 192 for service_property, value in security_parameters.iteritems(): 193 service_object.SetProperty(service_property, value) 194 if guid is not None: 195 service_object.SetProperty(self.SERVICE_PROPERTY_GUID, guid) 196 if autoconnect is not None: 197 service_object.SetProperty(self.SERVICE_PROPERTY_AUTOCONNECT, 198 autoconnect) 199 service_object.Connect() 200 logging.info('Called connect on service') 201 except dbus.exceptions.DBusException, e: 202 logging.error('Caught an error while trying to connect: %s', 203 e.get_dbus_message()) 204 return (False, discovery_time, association_time, 205 configuration_time, 'FAIL(Failed to call connect)') 206 207 logging.info('Associating...') 208 result = self.wait_for_property_in( 209 service_object, 210 self.SERVICE_PROPERTY_STATE, 211 ('configuration', 'ready', 'portal', 'online'), 212 association_timeout_seconds) 213 (successful, _, association_time) = result 214 if not successful: 215 return (False, discovery_time, association_time, 216 configuration_time, 'FAIL(Association timed out)') 217 218 logging.info('Associated with service: %s', ssid) 219 220 logging.info('Configuring...') 221 result = self.wait_for_property_in( 222 service_object, 223 self.SERVICE_PROPERTY_STATE, 224 ('ready', 'portal', 'online'), 225 configuration_timeout_seconds) 226 (successful, _, configuration_time) = result 227 if not successful: 228 return (False, discovery_time, association_time, 229 configuration_time, 'FAIL(Configuration timed out)') 230 231 logging.info('Configured service: %s', ssid) 232 233 # Great success! 234 logging.info('Connected to WiFi service.') 235 return (True, discovery_time, association_time, configuration_time, 236 'SUCCESS(Connection successful)') 237 238 239 def disconnect_from_wifi_network(self, ssid, timeout=None): 240 """Disconnect from the specified WiFi network. 241 242 Method will succeed if it observes the specified network in the idle 243 state after calling Disconnect. 244 245 @param ssid string name of network to disconnect. 246 @param timeout float number of seconds to wait for idle. 247 @return tuple(success, duration, reason) where: 248 success is a bool (True on success). 249 duration is a float number of seconds the operation took. 250 reason is a string containing an informative error on failure. 251 252 """ 253 if timeout is None: 254 timeout = self.SERVICE_DISCONNECT_TIMEOUT 255 service_description = {self.SERVICE_PROPERTY_TYPE: 'wifi', 256 self.SERVICE_PROPERTY_NAME: ssid} 257 service = self.find_matching_service(service_description) 258 if service is None: 259 return (False, 260 0.0, 261 'Failed to disconnect from %s, service not found.' % ssid) 262 263 service.Disconnect() 264 result = self.wait_for_property_in(service, 265 self.SERVICE_PROPERTY_STATE, 266 ('idle',), 267 timeout) 268 (successful, final_state, duration) = result 269 message = 'Success.' 270 if not successful: 271 message = ('Failed to disconnect from %s, ' 272 'timed out in state: %s.' % (ssid, final_state)) 273 return (successful, duration, message) 274 275 276 def configure_bgscan(self, interface, method=None, short_interval=None, 277 long_interval=None, signal=None): 278 """Configures bgscan parameters for wpa_supplicant. 279 280 @param interface string name of interface to configure (e.g. 'mlan0'). 281 @param method string bgscan method (e.g. 'none'). 282 @param short_interval int short scanning interval. 283 @param long_interval int normal scanning interval. 284 @param signal int signal threshold. 285 286 """ 287 device = self.find_object('Device', {'Name': interface}) 288 if device is None: 289 logging.error('No device found with name: %s', interface) 290 return False 291 292 attributes = {'ScanInterval': (dbus.UInt16, long_interval), 293 'BgscanMethod': (dbus.String, method), 294 'BgscanShortInterval': (dbus.UInt16, short_interval), 295 'BgscanSignalThreshold': (dbus.Int32, signal)} 296 for k, (type_cast, value) in attributes.iteritems(): 297 if value is None: 298 continue 299 300 # 'default' is defined in: 301 # client/common_lib/cros/network/xmlrpc_datatypes.py 302 # but we don't have access to that file here. 303 if value == 'default': 304 device.ClearProperty(k) 305 else: 306 device.SetProperty(k, type_cast(value)) 307 return True 308 309 310 def get_active_wifi_SSIDs(self): 311 """@return list of string SSIDs with at least one BSS we've scanned.""" 312 properties = self.manager.GetProperties(utf8_strings=True) 313 services = [self.get_dbus_object(self.DBUS_TYPE_SERVICE, path) 314 for path in properties[self.MANAGER_PROPERTY_SERVICES]] 315 wifi_services = [] 316 for service in services: 317 try: 318 service_properties = self.dbus2primitive(service.GetProperties( 319 utf8_strings=True)) 320 except dbus.exceptions.DBusException as e: 321 pass # Probably the service disappeared before GetProperties(). 322 logging.debug('Considering service with properties: %r', 323 service_properties) 324 service_type = service_properties[self.SERVICE_PROPERTY_TYPE] 325 strength = service_properties[self.SERVICE_PROPERTY_STRENGTH] 326 if service_type == 'wifi' and strength > 0: 327 # Note that this may cause terrible things if the SSID 328 # is not a valid ASCII string. 329 ssid = service_properties[self.SERVICE_PROPERTY_HEX_SSID] 330 logging.info('Found active WiFi service: %s', ssid) 331 wifi_services.append(ssid.decode('hex')) 332 return wifi_services 333 334 335 def wait_for_service_states(self, ssid, states, timeout_seconds): 336 """Wait for a service (ssid) to achieve one of a number of states. 337 338 @param ssid string name of network for whose state we're waiting. 339 @param states tuple states for which to wait. 340 @param timeout_seconds seconds to wait for property to be achieved 341 @return tuple(successful, final_value, duration) 342 where successful is True iff we saw one of |states|, final_value 343 is the final state we saw, and duration is how long we waited to 344 see that value. 345 346 """ 347 discovery_params = {self.SERVICE_PROPERTY_TYPE: 'wifi', 348 self.SERVICE_PROPERTY_NAME: ssid} 349 start_time = time.time() 350 service_object = None 351 while time.time() - start_time < timeout_seconds: 352 service_object = self.find_matching_service(discovery_params) 353 if service_object: 354 break 355 356 time.sleep(self.POLLING_INTERVAL_SECONDS) 357 else: 358 logging.error('Timed out waiting for %s states', ssid) 359 return False, 'unknown', timeout_seconds 360 361 return self.wait_for_property_in( 362 service_object, 363 self.SERVICE_PROPERTY_STATE, 364 states, 365 timeout_seconds - (time.time() - start_time)) 366