| #!/usr/bin/python |
| |
| # 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 socket |
| import sys |
| import time |
| |
| from autotest_lib.client.cros import dhcp_handling_rule |
| from autotest_lib.client.cros import dhcp_packet |
| from autotest_lib.client.cros import dhcp_test_server |
| |
| 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" |
| |
| TEST_DOMAIN_SEARCH_LIST_PARSED = ("eng.google.com", "marketing.google.com") |
| |
| # At this time, we don't support the compression allowed in the RFC. |
| # This is correct and sufficient for our purposes. |
| TEST_DOMAIN_SEARCH_LIST_EXPECTED = \ |
| "\x03eng\x06google\x03com\x00\x09marketing\x06google\x03com\x00" |
| |
| def bin2hex(byte_str, justification=20): |
| """ |
| Turn big hex strings into prettier strings of hex bytes. Group those hex |
| bytes into lines justification bytes long. |
| """ |
| chars = ["x" + (hex(ord(c))[2:].zfill(2)) for c in byte_str] |
| groups = [] |
| for i in xrange(0, len(chars), justification): |
| groups.append("".join(chars[i:i+justification])) |
| return "\n".join(groups) |
| |
| def test_packet_serialization(): |
| log_file = open(TEST_DATA_PATH_PREFIX + "dhcp_discovery.log", "rb") |
| binary_discovery_packet = log_file.read() |
| log_file.close() |
| discovery_packet = dhcp_packet.DhcpPacket(byte_str=binary_discovery_packet) |
| if not discovery_packet.is_valid: |
| return False |
| generated_string = discovery_packet.to_binary_string() |
| if generated_string is None: |
| print "Failed to generate string from packet object." |
| return False |
| if generated_string != binary_discovery_packet: |
| print "Packets didn't match: " |
| print "Generated: \n%s" % bin2hex(generated_string) |
| print "Expected: \n%s" % bin2hex(binary_discovery_packet) |
| return False |
| 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) |
| # Order matters too. |
| parsed_domains = tuple(parsed_domains) |
| if parsed_domains != TEST_DOMAIN_SEARCH_LIST_PARSED: |
| print ("Parsed binary domain list and got %s but expected %s" % |
| (parsed_domains, TEST_DOMAIN_SEARCH_LIST_EXPECTED)) |
| return False |
| print "test_domain_search_list_parsing PASSED" |
| return True |
| |
| def test_domain_search_list_serialization(): |
| byte_string = dhcp_packet.DomainListOption.pack( |
| TEST_DOMAIN_SEARCH_LIST_PARSED) |
| if byte_string != TEST_DOMAIN_SEARCH_LIST_EXPECTED: |
| # Turn the strings into printable hex strings on a single line. |
| pretty_actual = bin2hex(byte_string, 100) |
| pretty_expected = bin2hex(TEST_DOMAIN_SEARCH_LIST_EXPECTED, 100) |
| print ("Expected to serialize %s to %s but instead got %s." % |
| (TEST_DOMAIN_SEARCH_LIST_PARSED, pretty_expected, pretty_actual)) |
| return False |
| print "test_domain_search_list_serialization PASSED" |
| return True |
| |
| def receive_packet(a_socket, timeout_seconds=1.0): |
| data = None |
| start_time = time.time() |
| while data is None and start_time + timeout_seconds > time.time(): |
| try: |
| data, _ = a_socket.recvfrom(1024) |
| except socket.timeout: |
| pass # We expect many timeouts. |
| if data is None: |
| print "Timed out before we received a response from the server." |
| return None |
| |
| print "Client received a packet of length %d from the server." % len(data) |
| packet = dhcp_packet.DhcpPacket(byte_str=data) |
| if not packet.is_valid: |
| print "Received an invalid response from DHCP server." |
| return None |
| |
| return packet |
| |
| def test_simple_server_exchange(server): |
| intended_ip = "127.0.0.42" |
| subnet_mask = "255.255.255.0" |
| server_ip = "127.0.0.1" |
| lease_time_seconds = 60 |
| test_timeout = 3.0 |
| mac_addr = "\x01\x02\x03\x04\x05\x06" |
| # Build up our packets and have them request some default option values, |
| # like the IP we're being assigned and the address of the server assigning |
| # it. |
| discovery_message = dhcp_packet.DhcpPacket.create_discovery_packet(mac_addr) |
| discovery_message.set_option( |
| dhcp_packet.OPTION_PARAMETER_REQUEST_LIST, |
| dhcp_packet.OPTION_VALUE_PARAMETER_REQUEST_LIST_DEFAULT) |
| request_message = dhcp_packet.DhcpPacket.create_request_packet( |
| discovery_message.transaction_id, |
| mac_addr) |
| request_message.set_option( |
| dhcp_packet.OPTION_PARAMETER_REQUEST_LIST, |
| dhcp_packet.OPTION_VALUE_PARAMETER_REQUEST_LIST_DEFAULT) |
| # This is the pool of settings the DHCP server will seem to draw from to |
| # answer queries from the client. This information is written into packets |
| # through the handling rules. |
| dhcp_server_config = { |
| dhcp_packet.OPTION_SERVER_ID : 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, |
| } |
| # Build up the handling rules for the server and start the test. |
| rules = [] |
| rules.append(dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery( |
| intended_ip, |
| server_ip, |
| dhcp_server_config)) |
| rules.append(dhcp_handling_rule.DhcpHandlingRule_RespondToRequest( |
| intended_ip, |
| server_ip, |
| dhcp_server_config)) |
| rules[-1].is_final_handler = True |
| server.start_test(rules, test_timeout) |
| # Because we don't want to require root permissions to run these tests, |
| # listen on the loopback device, don't broadcast, and don't use reserved |
| # ports (like the actual DHCP ports). Use 8068/8067 instead. |
| client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| client_socket.bind(("127.0.0.1", 8068)) |
| client_socket.settimeout(0.1) |
| client_socket.sendto(discovery_message.to_binary_string(), |
| (server_ip, 8067)) |
| |
| offer_packet = receive_packet(client_socket) |
| if offer_packet is None: |
| return False |
| |
| if (offer_packet.message_type != |
| dhcp_packet.OPTION_VALUE_DHCP_MESSAGE_TYPE_OFFER): |
| print "Type of DHCP response is not offer." |
| return False |
| |
| if offer_packet.get_field(dhcp_packet.FIELD_YOUR_IP) != intended_ip: |
| print "Server didn't offer the IP we expected." |
| return False |
| |
| print "Offer looks good to the client, sending request." |
| # In real tests, dhcpcd formats all the DISCOVERY and REQUEST messages. In |
| # our unit test, we have to do this ourselves. |
| request_message.set_option( |
| dhcp_packet.OPTION_SERVER_ID, |
| offer_packet.get_option(dhcp_packet.OPTION_SERVER_ID)) |
| request_message.set_option( |
| dhcp_packet.OPTION_SUBNET_MASK, |
| offer_packet.get_option(dhcp_packet.OPTION_SUBNET_MASK)) |
| request_message.set_option( |
| dhcp_packet.OPTION_IP_LEASE_TIME, |
| offer_packet.get_option(dhcp_packet.OPTION_IP_LEASE_TIME)) |
| request_message.set_option( |
| dhcp_packet.OPTION_REQUESTED_IP, |
| offer_packet.get_option(dhcp_packet.OPTION_REQUESTED_IP)) |
| # Send the REQUEST message. |
| client_socket.sendto(request_message.to_binary_string(), |
| (server_ip, 8067)) |
| ack_packet = receive_packet(client_socket) |
| if ack_packet is None: |
| return False |
| |
| if (ack_packet.message_type != |
| dhcp_packet.OPTION_VALUE_DHCP_MESSAGE_TYPE_ACK): |
| print "Type of DHCP response is not acknowledgement." |
| return False |
| |
| if offer_packet.get_field(dhcp_packet.FIELD_YOUR_IP) != intended_ip: |
| print "Server didn't give us the IP we expected." |
| return False |
| |
| print "Waiting for the server to finish." |
| server.wait_for_test_to_finish() |
| print "Server agrees that the test is over." |
| if not server.last_test_passed: |
| print "Server is unhappy with the test result." |
| return False |
| |
| print "test_simple_server_exchange PASSED." |
| return True |
| |
| def test_server_dialogue(): |
| server = dhcp_test_server.DhcpTestServer(ingress_address="127.0.0.1", |
| ingress_port=8067, |
| broadcast_address="127.0.0.1", |
| broadcast_port=8068) |
| server.start() |
| ret = False |
| if server.is_healthy: |
| ret = test_simple_server_exchange(server) |
| else: |
| print "Server isn't healthy, aborting." |
| print "Sending server stop() signal." |
| server.stop() |
| print "Stop signal sent." |
| return ret |
| |
| def run_tests(): |
| logger = logging.getLogger("dhcp") |
| logger.setLevel(logging.DEBUG) |
| stream_handler = logging.StreamHandler() |
| 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() |
| if retval: |
| print "All tests PASSED." |
| return 0 |
| else: |
| print "Some tests FAILED" |
| return -1 |
| |
| if __name__ == "__main__": |
| sys.exit(run_tests()) |