# Copyright (c) 2012 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 os import re import time from autotest_lib.client.bin import local_host from autotest_lib.client.bin import utils from autotest_lib.client.common_lib import error from autotest_lib.client.common_lib.cros.network import interface # Flag file used to tell backchannel script it's okay to run. BACKCHANNEL_FILE = '/mnt/stateful_partition/etc/enable_backchannel_network' # Backchannel interface name. BACKCHANNEL_IFACE_NAME = 'eth_test' # Script that handles backchannel heavy lifting. BACKCHANNEL_SCRIPT = '/usr/local/lib/flimflam/test/backchannel' class Backchannel(object): """Wrap backchannel in a context manager so it can be used with with. Example usage: with backchannel.Backchannel(): block The backchannel will be torn down whether or not 'block' throws. """ def __init__(self, host=None, *args, **kwargs): self.args = args self.kwargs = kwargs self.gateway = None self.interface = None if host is not None: self.host = host else: self.host = local_host.LocalHost() self._run = self.host.run def __enter__(self): self.setup(*self.args, **self.kwargs) return self def __exit__(self, exception, value, traceback): self.teardown() return False def setup(self, create_ssh_routes=True): """ Enables the backchannel interface. @param create_ssh_routes: If True set up routes so that all existing SSH sessions will remain open. @returns True if the backchannel is already set up, or was set up by this call, otherwise False. """ # If the backchannel interface is already up there's nothing # for us to do. if self._is_test_iface_running(): return True # Retrieve the gateway for the default route. try: # Poll here until we have route information. # If shill was recently started, it will take some time before # DHCP gives us an address. line = utils.poll_for_condition( lambda: self._get_default_route(), exception=utils.TimeoutError( 'Timed out waiting for route information'), timeout=30) self.gateway, self.interface = line.strip().split(' ') # Retrieve list of open ssh sessions so we can reopen # routes afterward. if create_ssh_routes: out = self._run( "netstat -tanp | grep :22 | " "grep ESTABLISHED | awk '{print $5}'").stdout # Extract IP from IP:PORT listing. Uses set to remove # duplicates. open_ssh = list(set(item.strip().split(':')[0] for item in out.split('\n') if item.strip())) # Build a command that will set up the test interface and add # ssh routes in one shot. This is necessary since we'll lose # connectivity to a remote host between these steps. cmd = '%s setup %s' % (BACKCHANNEL_SCRIPT, self.interface) if create_ssh_routes: for ip in open_ssh: # Add route using the pre-backchannel gateway. cmd += '&& %s reach %s %s' % (BACKCHANNEL_SCRIPT, ip, self.gateway) self._run(cmd) # Make sure we have a route to the gateway before continuing. logging.info('Waiting for route to gateway %s', self.gateway) utils.poll_for_condition( lambda: self._is_route_ready(), exception=utils.TimeoutError('Timed out waiting for route'), timeout=30) except Exception, e: logging.error(e) return False finally: # Remove backchannel file flag so system reverts to normal # on reboot. if os.path.isfile(BACKCHANNEL_FILE): os.remove(BACKCHANNEL_FILE) return True def teardown(self): """Tears down the backchannel.""" if self.interface: self._run('%s teardown %s' % (BACKCHANNEL_SCRIPT, self.interface)) # Hack around broken Asix network adaptors that may flake out when we # bring them up and down (crbug.com/349264). # TODO(thieule): Remove this when the adaptor/driver is fixed # (crbug.com/350172). try: if self.gateway: logging.info('Waiting for route restore to gateway %s', self.gateway) utils.poll_for_condition( lambda: self._is_route_ready(), exception=utils.TimeoutError( 'Timed out waiting for route'), timeout=30) except utils.TimeoutError: if self.host is None: self._reset_usb_ethernet_device() def is_using_ethernet(self): """ Checks to see if the backchannel is using an ethernet device. @returns True if the backchannel is using an ethernet device. """ result = self._run( 'ethtool %s' % BACKCHANNEL_IFACE_NAME, ignore_status=True) if result.exit_status: return False match = re.search('Port: (.+)', result.stdout) return match and _is_ethernet_port(match.group(1)) def _reset_usb_ethernet_device(self): try: # Use the absolute path to the USB device instead of accessing it # via the path with the interface name because once we # deauthorize the USB device, the interface name will be gone. usb_authorized_path = os.path.realpath( '/sys/class/net/%s/device/../authorized' % self.interface) logging.info('Reset ethernet device at %s', usb_authorized_path) utils.system('echo 0 > %s' % usb_authorized_path) time.sleep(10) utils.system('echo 1 > %s' % usb_authorized_path) except error.CmdError: pass def _get_default_route(self): """Retrieves default route information.""" cmd = "route -n | awk '/^0.0.0.0/ { print $2, $8 }'" return self._run(cmd).stdout.split('\n')[0] def _is_test_iface_running(self): """Checks whether the test interface is running.""" return interface.Interface(BACKCHANNEL_IFACE_NAME).is_link_operational() def _is_route_ready(self): """Checks for a route to the specified destination.""" dest = self.gateway result = self._run('ping -c 1 %s' % dest, ignore_status=True) if result.exit_status: logging.warning('Route to %s is not ready.', dest) return False logging.info('Route to %s is ready.', dest) return True def is_network_iface_running(name): """ Checks to see if the interface is running. @param name: Name of the interface to check. @returns True if the interface is running. """ try: out = utils.system_output('ip link show dev %s' % name) except error.CmdError, e: logging.info(e) return False return out.find('state UP') >= 0 def _is_ethernet_port(port): # Some versions of ethtool may report the full name. ETHTOOL_PORT_TWISTED_PAIR = 'TP' ETHTOOL_PORT_TWISTED_PAIR_FULL = 'Twisted Pair' ETHTOOL_PORT_MEDIA_INDEPENDENT_INTERFACE = 'MII' ETHTOOL_PORT_MEDIA_INDEPENDENT_INTERFACE_FULL = \ 'Media Independent Interface' return port in [ETHTOOL_PORT_TWISTED_PAIR, ETHTOOL_PORT_TWISTED_PAIR_FULL, ETHTOOL_PORT_MEDIA_INDEPENDENT_INTERFACE, ETHTOOL_PORT_MEDIA_INDEPENDENT_INTERFACE_FULL] def is_backchannel_using_ethernet(): """ Checks to see if the backchannel is using an ethernet device. @returns True if the backchannel is using an ethernet device. """ ethtool_output = utils.system_output( 'ethtool %s' % BACKCHANNEL_IFACE_NAME, ignore_status=True) match = re.search('Port: (.+)', ethtool_output) return match and _is_ethernet_port(match.group(1))