autotest: dhcp: Add test for classless static routes
Add parsing and serialization of the DHCP classless static route
option. Add a unit test for the above. Add verification that
classless static route options are converted into a default route.
Add a test that uses the classless static route option.
BUG=chromium-os:25908
TEST=Unit test + run auto test
CQ-DEPEND=I6acecb09258229d84d4aa43372a1dc13fbac1df5
Change-Id: Id371841657c45bc3033fb5cc9944d8af5b032402
Reviewed-on: https://gerrit.chromium.org/gerrit/38204
Commit-Ready: Paul Stewart <[email protected]>
Reviewed-by: Paul Stewart <[email protected]>
Tested-by: Paul Stewart <[email protected]>
diff --git a/client/cros/dhcp_packet.py b/client/cros/dhcp_packet.py
index e5496d7..e218cf9 100644
--- a/client/cros/dhcp_packet.py
+++ b/client/cros/dhcp_packet.py
@@ -110,6 +110,53 @@
return [ord(c) for c in byte_string]
+class ClasslessStaticRoutesOption(Option):
+ """
+ This is a RFC 3442 compliant classless static route option parser and
+ serializer. The symbolic "value" packed and unpacked from this class
+ is a list (prefix_size, destination, router) tuples.
+ """
+
+ @staticmethod
+ def pack(value):
+ route_list = value
+ byte_string = ""
+ for prefix_size, destination, router in route_list:
+ byte_string += chr(prefix_size)
+ # Encode only the significant octets of the destination
+ # that fall within the prefix.
+ destination_address_count = (prefix_size + 7) / 8
+ destination_address = socket.inet_aton(destination)
+ byte_string += destination_address[:destination_address_count]
+ byte_string += socket.inet_aton(router)
+
+ return byte_string
+
+ @staticmethod
+ def unpack(byte_string):
+ route_list = []
+ offset = 0
+ while offset < len(byte_string):
+ prefix_size = ord(byte_string[offset])
+ destination_address_count = (prefix_size + 7) / 8
+ entry_end = offset + 1 + destination_address_count + 4
+ if entry_end > len(byte_string):
+ raise Exception("Classless domain list is corrupted.")
+ offset += 1
+ destination_address_end = offset + destination_address_count
+ destination_address = byte_string[offset:destination_address_end]
+ # Pad the destination address bytes with zero byte octets to
+ # fill out an IPv4 address.
+ destination_address += '\x00' * (4 - destination_address_count)
+ router_address = byte_string[destination_address_end:entry_end]
+ route_list.append((prefix_size,
+ socket.inet_ntoa(destination_address),
+ socket.inet_ntoa(router_address)))
+ offset = entry_end
+
+ return route_list
+
+
class DomainListOption(Option):
"""
This is a RFC 1035 compliant domain list option parser and serializer.
@@ -296,6 +343,8 @@
OPTION_TFTP_SERVER_NAME = RawOption("tftp_server_name", 66)
OPTION_BOOTFILE_NAME = RawOption("bootfile_name", 67)
OPTION_DNS_DOMAIN_SEARCH_LIST = DomainListOption("domain_search_list", 119)
+OPTION_CLASSLESS_STATIC_ROUTES = ClasslessStaticRoutesOption(
+ "classless_static_routes", 121)
# Unlike every other option, which are tuples like:
# <number, length in bytes, data>, the pad and end options are just
@@ -410,6 +459,7 @@
OPTION_TFTP_SERVER_NAME,
OPTION_BOOTFILE_NAME,
OPTION_DNS_DOMAIN_SEARCH_LIST,
+ OPTION_CLASSLESS_STATIC_ROUTES,
]
def get_dhcp_option_by_number(number):
diff --git a/client/cros/dhcp_test_base.py b/client/cros/dhcp_test_base.py
index 4056a72..62a06f8 100644
--- a/client/cros/dhcp_test_base.py
+++ b/client/cros/dhcp_test_base.py
@@ -276,6 +276,22 @@
"search list %s, but got %s instead." %
(expected_search_list, configured_search_list))
+ expected_routers = dhcp_options.get(dhcp_packet.OPTION_ROUTERS)
+ if (not expected_routers and
+ dhcp_options.get(dhcp_packet.OPTION_CLASSLESS_STATIC_ROUTES)):
+ classless_static_routes = dhcp_options[
+ dhcp_packet.OPTION_CLASSLESS_STATIC_ROUTES]
+ for prefix, destination, gateway in classless_static_routes:
+ if not prefix:
+ logging.info("Using %s as the default gateway" % gateway)
+ expected_routers = [ gateway ]
+ break
+ configured_router = dhcp_config.get(DHCPCD_KEY_GATEWAY)
+ if expected_routers and expected_routers[0] != configured_router:
+ raise error.TestFail("Expected to be configured with gateway %s, "
+ "but got %s instead." %
+ (expected_routers[0], configured_router))
+
self.server.wait_for_test_to_finish()
if not self.server.last_test_passed:
raise error.TestFail("Test server didn't get all the messages it "
diff --git a/client/cros/dhcp_unittest.py b/client/cros/dhcp_unittest.py
index d4b228a..907d132 100755
--- a/client/cros/dhcp_unittest.py
+++ b/client/cros/dhcp_unittest.py
@@ -15,6 +15,15 @@
TEST_DATA_PATH_PREFIX = "client/cros/dhcp_test_data/"
+TEST_CLASSLESS_STATIC_ROUTE_DATA = \
+ "\x12\x0a\x09\xc0\xac\x1f\x9b\x0a" \
+ "\x00\xc0\xa8\x00\xfe"
+
+TEST_CLASSLESS_STATIC_ROUTE_LIST_PARSED = [
+ (18, "10.9.192.0", "172.31.155.10"),
+ (0, "0.0.0.0", "192.168.0.254")
+ ]
+
TEST_DOMAIN_SEARCH_LIST_COMPRESSED = \
"\x03eng\x06google\x03com\x00\x09marketing\xC0\x04"
@@ -55,6 +64,31 @@
print "test_packet_serialization PASSED"
return True
+def test_classless_static_route_parsing():
+ parsed_routes = dhcp_packet.ClasslessStaticRoutesOption.unpack(
+ TEST_CLASSLESS_STATIC_ROUTE_DATA)
+ if parsed_routes != TEST_CLASSLESS_STATIC_ROUTE_LIST_PARSED:
+ print ("Parsed binary domain list and got %s but expected %s" %
+ (repr(parsed_routes),
+ repr(TEST_CLASSLESS_STATIC_ROUTE_LIST_PARSED)))
+ return False
+ print "test_classless_static_route_parsing PASSED"
+ return True
+
+def test_classless_static_route_serialization():
+ byte_string = dhcp_packet.ClasslessStaticRoutesOption.pack(
+ TEST_CLASSLESS_STATIC_ROUTE_LIST_PARSED)
+ if byte_string != TEST_CLASSLESS_STATIC_ROUTE_DATA:
+ # Turn the strings into printable hex strings on a single line.
+ pretty_actual = bin2hex(byte_string, 100)
+ pretty_expected = bin2hex(TEST_CLASSLESS_STATIC_ROUTE_DATA, 100)
+ print ("Expected to serialize %s to %s but instead got %s." %
+ (repr(TEST_CLASSLESS_STATIC_ROUTE_LIST_PARSED), pretty_expected,
+ pretty_actual))
+ return False
+ print "test_classless_static_route_serialization PASSED"
+ return True
+
def test_domain_search_list_parsing():
parsed_domains = dhcp_packet.DomainListOption.unpack(
TEST_DOMAIN_SEARCH_LIST_COMPRESSED)
@@ -227,6 +261,8 @@
stream_handler.setLevel(logging.DEBUG)
logger.addHandler(stream_handler)
retval = test_packet_serialization()
+ retval &= test_classless_static_route_parsing()
+ retval &= test_classless_static_route_serialization()
retval &= test_domain_search_list_parsing()
retval &= test_domain_search_list_serialization()
retval &= test_server_dialogue()
diff --git a/client/site_tests/network_DhcpClasslessStaticRoute/control b/client/site_tests/network_DhcpClasslessStaticRoute/control
new file mode 100644
index 0000000..b54f054
--- /dev/null
+++ b/client/site_tests/network_DhcpClasslessStaticRoute/control
@@ -0,0 +1,23 @@
+# 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.
+
+AUTHOR = "ChromeOS Team"
+NAME = "network_DhcpClasslessStaticRoute"
+PURPOSE = "Verify DHCP negotions can succeed with a classless static route"
+CRITERIA = """
+This test fails if dhcpcd is unable to attain a default route using
+the classless static route option (RFC 3442) instead of the normal
+gateway parameter.
+"""
+TIME = "SHORT"
+TEST_CATEGORY = "Functional"
+TEST_CLASS = "network"
+TEST_TYPE = "client"
+
+DOC = """
+ Tests that we can negotiate a lease on an IPv4 address via DHCP.
+
+"""
+
+job.run_test('network_DhcpClasslessStaticRoute')
diff --git a/client/site_tests/network_DhcpClasslessStaticRoute/network_DhcpClasslessStaticRoute.py b/client/site_tests/network_DhcpClasslessStaticRoute/network_DhcpClasslessStaticRoute.py
new file mode 100644
index 0000000..1a8600a
--- /dev/null
+++ b/client/site_tests/network_DhcpClasslessStaticRoute/network_DhcpClasslessStaticRoute.py
@@ -0,0 +1,50 @@
+# 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.
+
+from autotest_lib.client.cros import dhcp_packet
+from autotest_lib.client.cros import dhcp_test_base
+
+# 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"
+# Set the router IP address based on the created prefix.
+ROUTER_IP_SUFFIX = "0.0.0.254"
+
+class network_DhcpClasslessStaticRoute(dhcp_test_base.DhcpTestBase):
+ def test_body(self):
+ 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)
+ router_ip = dhcp_test_base.DhcpTestBase.rewrite_ip_suffix(
+ subnet_mask,
+ self.server_ip,
+ ROUTER_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 = [
+ "nyan.cat.google.com",
+ "fail.whale.google.com",
+ "zircon.encrusted.tweezers.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,
+ dhcp_packet.OPTION_CLASSLESS_STATIC_ROUTES : [
+ (0, "0.0.0.0", router_ip),
+ (24, "192.168.100.200", "192.168.80.254")
+ ]
+ }
+ self.negotiate_and_check_lease(dhcp_options)
