client/site_tests/network_Ipv6SimpleNegotiation/network_Ipv6SimpleNegotiation.py

# Copyright (c) 2013 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 import utils
from autotest_lib.client.cros import dhcp_packet
from autotest_lib.client.cros import dhcp_test_base
from autotest_lib.client.cros import radvd_server

# Length of time the lease from the DHCP server is valid.
LEASE_TIME_SECONDS = 60
# We'll fill in the subnet and give this address to the client.
INTENDED_IP_SUFFIX = '0.0.0.101'

class network_Ipv6SimpleNegotiation(dhcp_test_base.DhcpTestBase):
    """
    The test subclass that implements IPv6 negotiation.  This test
    starts an IPv6 router, then completes a normal DHCP negotiation
    to bring up a stable connection.  It then performs a series of
    tests on the IPv6 addresses that the DUT should have also gained.
    """

    def _get_ip6_addresses(self):
        """
        Retrieve IPv6 addresses associated with the "client side" of the
        pseudo-interface pair.  Returns a dict keyed by the IPv6 address,
        with the values being the array of attribute strings that follow
        int the "ip addr show" output.  For example, a line containing:

            inet6 fe80::ae16:2dff:fe01:0203/64 scope link

        will turn into a dict key:

            'fe80::ae16:2dff:fe01:0203/64': [ 'scope', 'link' ]
        """
        addr_output = utils.system_output(
            "ip -6 addr show dev %s" % self.ethernet_pair.peer_interface_name)
        addresses = {}
        for line in addr_output.splitlines():
            parts = line.lstrip().split()
            if parts[0] != 'inet6' or 'deprecated' in parts:
                continue
            addresses[parts[1]] = parts[2:]
        return addresses

    def _get_link_address(self):
        """
        Retrieve the MAC address associated with the "client side" of the
        pseudo-interface pair.  For example, the "ip link show" output:

            link/ether 01:02:03:04:05:05 brd ff:ff:ff:ff:ff:ff

        will cause a return of "01:02:03:04:05:05"
        """
        addr_output = utils.system_output(
            'ip link show %s' % self.ethernet_pair.peer_interface_name)
        for line in addr_output.splitlines():
            parts = line.lstrip().split(' ')
            if parts[0] == 'link/ether':
                return parts[1]

    def negotiate_dhcp_lease(self):
        """
        Perform a DHCP negotiation.  Although this test isn't really meant
        to validate DHCP negotiation, we should go through this process so
        the connection manager keeps the interface up long enough for the
        IPv6 negotiation to complete reliably.
        """
        subnet_mask = self.ethernet_pair.interface_subnet_mask
        intended_ip = dhcp_test_base.DhcpTestBase.rewrite_ip_suffix(
                subnet_mask,
                self.server_ip,
                INTENDED_IP_SUFFIX)
        # Two real name servers, and a bogus one to be unpredictable.
        dns_servers = ['8.8.8.8', '8.8.4.4', '192.168.87.88']
        domain_name = 'corp.google.com'
        dns_search_list = [
                'you.can.pry.google.com',
                'my.pixel.google.com',
                'from.my.cold.dead.hands.google.com',
                ]
        # This is the pool of information the server will give out to the client
        # upon request.
        dhcp_options = {
                dhcp_packet.OPTION_SERVER_ID : self.server_ip,
                dhcp_packet.OPTION_SUBNET_MASK : subnet_mask,
                dhcp_packet.OPTION_IP_LEASE_TIME : LEASE_TIME_SECONDS,
                dhcp_packet.OPTION_REQUESTED_IP : intended_ip,
                dhcp_packet.OPTION_DNS_SERVERS : dns_servers,
                dhcp_packet.OPTION_DOMAIN_NAME : domain_name,
                dhcp_packet.OPTION_DNS_DOMAIN_SEARCH_LIST : dns_search_list,
                }
        self.negotiate_and_check_lease(dhcp_options)

    def verify_ipv6_addresses(self):
        """
        Perform various tests to validate the IPv6 addresses acquired by
        the client.
        """
        addresses = self._get_ip6_addresses()
        logging.info('Got addresses %r', addresses)
        global_addresses = [key for key in addresses
                            if 'global' in addresses[key]]

        if len(global_addresses) != 2:
            raise error.TestError('Expected 2 global address but got %d' %
                                  len(global_addresses))

        prefix = radvd_server.RADVD_DEFAULT_PREFIX
        prefix = prefix[:prefix.index('::')]
        for address in global_addresses:
            if not address.startswith(prefix):
                raise error.TestError('Global address %s does not start with '
                                      'expected prefix %s' %
                                      address, prefix)

        # One globally scoped address should be based on the last 3 octets
        # of the MAC adddress, while the other should not.  For example,
        # for MAC address "01:02:03:04:05:06", we should see an address
        # that ends with "4:506/64" (the "/64" is the default radvd suffix).
        link_parts = [int(b, 16) for b in self._get_link_address().split(':')]
        address_suffix = '%x:%x%s' % (link_parts[3],
                                      (link_parts[4] << 8) | link_parts[5],
                                      radvd_server.RADVD_DEFAULT_SUFFIX)
        mac_related_addresses = [addr for addr in global_addresses
                                 if addr.endswith(address_suffix)]
        if len(mac_related_addresses) != 1:
            raise error.TestError('Expected 1 mac-related global address but '
                                  'got %d' % len(mac_related_addresses))

        local_address_count = len(addresses) - len(global_addresses)
        if local_address_count <= 0:
            raise error.TestError('Expected at least 1 non-global address but '
                                  'got %d' % local_address_count)

    def test_body(self):
        """
        The main body for this test.
        """
        server = radvd_server.RadvdServer(self.ethernet_pair.interface_name)
        server.start_server()

        try:
            self.negotiate_dhcp_lease()

            # Wait a bit more for IPv6 negotiation to complete.
            time.sleep(radvd_server.RADVD_DEFAULT_MAX_ADV_INTERVAL)

            # In this time, we should have also acquired an IPv6 address.
            self.verify_ipv6_addresses()

        finally:
            server.stop_server()