| # Copyright 2014 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 logging |
| import time |
| |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.common_lib.cros import avahi_utils |
| from autotest_lib.client.common_lib.cros.network import interface |
| from autotest_lib.client.common_lib.cros.network import iw_runner |
| from autotest_lib.client.common_lib.cros.network import netblock |
| from autotest_lib.client.common_lib.cros.network import ping_runner |
| from autotest_lib.client.common_lib.cros.network import xmlrpc_security_types |
| from autotest_lib.client.common_lib.cros.tendo import peerd_config |
| from autotest_lib.client.common_lib.cros.tendo import buffet_config |
| from autotest_lib.client.common_lib.cros.tendo import privet_helper |
| from autotest_lib.server import site_linux_router |
| from autotest_lib.server import test |
| from autotest_lib.server.cros.network import hostap_config |
| from autotest_lib.server.cros.network import wifi_client |
| |
| |
| PASSPHRASE = 'chromeos' |
| |
| PRIVET_AP_STARTUP_TIMEOUT_SECONDS = 30 |
| PRIVET_MDNS_RECORD_TIMEOUT_SECONDS = 10 |
| PRIVET_CONNECT_TIMEOUT_SECONDS = 30 |
| |
| POLLING_PERIOD = 0.5 |
| |
| |
| class buffet_PrivetSetupFlow(test.test): |
| """This test validates the privet pairing/authentication/setup flow.""" |
| version = 1 |
| |
| def warmup(self, host, router_hostname=None): |
| self._router = None |
| self._shill_xmlrpc_proxy = None |
| config = buffet_config.BuffetConfig( |
| log_verbosity=3, |
| enable_ping=True, |
| disable_pairing_security=True, |
| device_whitelist='any', |
| options={'wifi_bootstrap_mode': 'automatic'}) |
| config.restart_with_config(host=host) |
| self._router = site_linux_router.build_router_proxy( |
| test_name=self.__class__.__name__, |
| client_hostname=host.hostname, |
| router_addr=router_hostname, |
| enable_avahi=True) |
| self._shill_xmlrpc_proxy = wifi_client.get_xmlrpc_proxy(host) |
| # Cleans up profiles, wifi credentials, sandboxes our new credentials. |
| self._shill_xmlrpc_proxy.init_test_network_state() |
| peerd_config.PeerdConfig(verbosity_level=3).restart_with_config( |
| host=host) |
| |
| |
| def cleanup(self, host): |
| if self._shill_xmlrpc_proxy is not None: |
| self._shill_xmlrpc_proxy.clean_profiles() |
| if self._router is not None: |
| self._router.close() |
| buffet_config.naive_restart(host=host) |
| |
| |
| def run_once(self, host): |
| helper = privet_helper.PrivetHelper(host=host) |
| logging.info('Looking for privet bootstrapping network from DUT.') |
| scan_interface = self._router.get_wlanif(2437, 'managed') |
| self._router.host.run('%s link set %s up' % |
| (self._router.cmd_ip, scan_interface)) |
| start_time = time.time() |
| privet_bss = None |
| while time.time() - start_time < PRIVET_AP_STARTUP_TIMEOUT_SECONDS: |
| bss_list = self._router.iw_runner.scan(scan_interface) |
| for bss in bss_list or []: |
| if helper.is_softap_ssid(bss.ssid): |
| privet_bss = bss |
| if privet_bss is None: |
| raise error.TestFail('Device did not start soft AP in time.') |
| self._router.release_interface(scan_interface) |
| |
| # Get the netblock of the interface running the AP. |
| dut_iw_runner = iw_runner.IwRunner(remote_host=host) |
| devs = dut_iw_runner.list_interfaces(desired_if_type='AP') |
| if not devs: |
| raise error.TestFail('No AP devices on DUT?') |
| ap_interface = interface.Interface(devs[0].if_name, host=host) |
| ap_netblock = netblock.from_addr(ap_interface.ipv4_address_and_prefix) |
| |
| # Set up an AP on the router in the 5Ghz range with WPA2 security. |
| wpa_config = xmlrpc_security_types.WPAConfig( |
| psk=PASSPHRASE, |
| wpa_mode=xmlrpc_security_types.WPAConfig.MODE_PURE_WPA2, |
| wpa2_ciphers=[xmlrpc_security_types.WPAConfig.CIPHER_CCMP]) |
| router_conf = hostap_config.HostapConfig( |
| frequency=5240, security_config=wpa_config, |
| mode=hostap_config.HostapConfig.MODE_11N_PURE) |
| self._router.hostap_configure(router_conf) |
| |
| # Connect the other interface on the router to the AP on the client |
| # at a hardcoded IP address. |
| self._router.configure_managed_station( |
| privet_bss.ssid, privet_bss.frequency, |
| ap_netblock.get_addr_in_block(200)) |
| station_interface = self._router.get_station_interface(instance=0) |
| logging.debug('Set up station on %s', station_interface) |
| self._router.ping(ping_runner.PingConfig(ap_netblock.addr, count=3)) |
| |
| logging.info('Looking for privet webserver in mDNS records.') |
| start_time = time.time() |
| while time.time() - start_time < PRIVET_MDNS_RECORD_TIMEOUT_SECONDS: |
| all_records = avahi_utils.avahi_browse(host=self._router.host) |
| records = [record for record in all_records |
| if (record.interface == station_interface and |
| record.record_type == '_privet._tcp')] |
| if records: |
| break |
| time.sleep(POLLING_PERIOD) |
| if not records: |
| raise error.TestFail('Did not find privet mDNS records in time.') |
| if len(records) > 1: |
| raise error.TestFail('Should not see multiple privet records.') |
| privet_record = records[0] |
| # TODO(wiley) pull the HTTPs port number out of the /info API. |
| helper = privet_helper.PrivetdHelper( |
| host=self._router.host, |
| hostname=privet_record.address, |
| http_port=int(privet_record.port)) |
| helper.ping_server() |
| |
| # Now configure the client with WiFi credentials. |
| auth_token = helper.privet_auth() |
| ssid = self._router.get_ssid() |
| data = helper.setup_add_wifi_credentials(ssid, PASSPHRASE) |
| helper.setup_start(data, auth_token) |
| |
| logging.info('Waiting for DUT to connect to router network.') |
| start_time = time.time() |
| # Wait for the DUT to take down the AP. |
| while time.time() - start_time < PRIVET_CONNECT_TIMEOUT_SECONDS: |
| if not dut_iw_runner.list_interfaces(desired_if_type='AP'): |
| break |
| time.sleep(POLLING_PERIOD) |
| else: |
| raise error.TestFail('Timeout waiting for DUT to take down AP.') |
| |
| # But we should be able to ping the client from the router's AP. |
| while time.time() - start_time < PRIVET_CONNECT_TIMEOUT_SECONDS: |
| if dut_iw_runner.list_interfaces(desired_if_type='managed'): |
| break |
| time.sleep(POLLING_PERIOD) |
| else: |
| raise error.TestFail('Timeout waiting for DUT managerd interface.') |
| |
| while time.time() - start_time < PRIVET_CONNECT_TIMEOUT_SECONDS: |
| devs = dut_iw_runner.list_interfaces(desired_if_type='managed') |
| if devs: |
| managed_interface = interface.Interface(devs[0].if_name, |
| host=host) |
| # Check if we have an IP yet. |
| if managed_interface.ipv4_address_and_prefix: |
| break |
| time.sleep(POLLING_PERIOD) |
| else: |
| raise error.TestFail('Timeout waiting for DUT managerd interface.') |
| |
| managed_netblock = netblock.from_addr( |
| managed_interface.ipv4_address_and_prefix) |
| while time.time() - start_time < PRIVET_CONNECT_TIMEOUT_SECONDS: |
| PING_COUNT = 3 |
| result = self._router.ping( |
| ping_runner.PingConfig(managed_netblock.addr, |
| ignore_result=True, |
| count=PING_COUNT)) |
| if result.received == PING_COUNT: |
| break |
| time.sleep(POLLING_PERIOD) |
| else: |
| raise error.TestFail('Timeout before ping was successful.') |
| |
| # And buffet should think it is online as well. |
| helper = privet_helper.PrivetdHelper( |
| host=host, hostname=managed_netblock.addr, |
| http_port=int(privet_record.port)) |
| helper.ping_server() |
| if not helper.wifi_setup_was_successful(ssid, auth_token): |
| raise error.TestFail('Device claims to be offline, but is online.') |
