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