| # 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 |
| from autotest_lib.client.cros.networking import shill_proxy |
| |
| # 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): |
| """Gets the list of client IPv6 addresses. |
| |
| 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): |
| """Get the client MAC address. |
| |
| 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): |
| """Verify IPv6 configuration. |
| |
| 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)) |
| mac_related_address = mac_related_addresses[0] |
| |
| 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) |
| |
| temporary_address = [addr for addr in global_addresses |
| if addr != mac_related_address][0] |
| self.verify_ipconfig_contains(temporary_address) |
| |
| |
| def verify_ipconfig_contains(self, address_and_prefix): |
| """Verify that shill has an IPConfig entry with the specified address. |
| |
| @param address_and_prefix string with address/prefix to search for. |
| |
| """ |
| address, prefix_str = address_and_prefix.split('/') |
| prefix = int(prefix_str) |
| for ipconfig in self.get_interface_ipconfig_objects( |
| self.ethernet_pair.peer_interface_name): |
| ipconfig_properties = shill_proxy.ShillProxy.dbus2primitive( |
| ipconfig.GetProperties(utf8_strings=True)) |
| if 'Method' not in ipconfig_properties: |
| continue |
| |
| if ipconfig_properties['Method'] != 'ipv6': |
| continue |
| |
| break |
| |
| else: |
| raise error.TestError('Found no IPv6 IPConfig entries') |
| |
| for property, value in (('Address', address), ('Prefixlen', prefix)): |
| if property not in ipconfig_properties: |
| raise error.TestError('IPv6 IPConfig entry does not ' |
| 'contain property %s' % property) |
| if ipconfig_properties[property] != value: |
| raise error.TestError('IPv6 IPConfig property %s does not ' |
| 'contain the expected value %s; ' |
| 'instead it is %s' % |
| (property, value, |
| ipconfig_properties[property])) |
| |
| |
| 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() |
