| from acts import asserts |
| from acts import base_test |
| from acts import signals |
| from acts.controllers import android_device |
| from acts.test_decorators import test_tracker_info |
| |
| from scapy.all import * |
| from threading import Event |
| from threading import Thread |
| import random |
| import time |
| import warnings |
| |
| |
| CLIENT_PORT = 68 |
| SERVER_PORT = 67 |
| BROADCAST_MAC = 'ff:ff:ff:ff:ff:ff' |
| INET4_ANY = '0.0.0.0' |
| NETADDR_PREFIX = '192.168.42.' |
| OTHER_NETADDR_PREFIX = '192.168.43.' |
| NETADDR_BROADCAST = '255.255.255.255' |
| SUBNET_BROADCAST = NETADDR_PREFIX + '255' |
| USB_CHARGE_MODE = 'svc usb setFunctions' |
| USB_TETHERING_MODE = 'svc usb setFunctions rndis' |
| DEVICE_IP_ADDRESS = 'ip address' |
| |
| |
| OFFER = 2 |
| REQUEST = 3 |
| ACK = 5 |
| NAK = 6 |
| |
| |
| class DhcpServerTest(base_test.BaseTestClass): |
| def setup_class(self): |
| self.dut = self.android_devices[0] |
| self.USB_TETHERED = False |
| self.next_hwaddr_index = 0 |
| self.stop_arp = Event() |
| |
| conf.checkIPaddr = 0 |
| conf.checkIPsrc = 0 |
| # Allow using non-67 server ports as long as client uses 68 |
| bind_layers(UDP, BOOTP, dport=CLIENT_PORT) |
| |
| iflist_before = get_if_list() |
| self._start_usb_tethering(self.dut) |
| self.iface = self._wait_for_new_iface(iflist_before) |
| self.real_hwaddr = get_if_raw_hwaddr(self.iface) |
| |
| # Start a thread to answer to all ARP "who-has" |
| thread = Thread(target=self._sniff_arp, args=(self.stop_arp,)) |
| thread.start() |
| |
| # Discover server IP and device hwaddr |
| hwaddr = self._next_hwaddr() |
| resp = self._get_response(self._make_discover(hwaddr)) |
| asserts.assert_false(None == resp, |
| "Device did not reply to first DHCP discover") |
| self.server_addr = getopt(resp, 'server_id') |
| self.dut_hwaddr = resp.getlayer(Ether).src |
| asserts.assert_false(None == self.server_addr, |
| "DHCP server did not specify server identifier") |
| # Ensure that we don't depend on assigned route/gateway on the host |
| conf.route.add(host=self.server_addr, dev=self.iface, gw="0.0.0.0") |
| |
| def setup_test(self): |
| # Some versions of scapy do not close the receive file properly |
| warnings.filterwarnings("ignore", category=ResourceWarning) |
| |
| bind_layers(UDP, BOOTP, dport=68) |
| self.hwaddr = self._next_hwaddr() |
| self.other_hwaddr = self._next_hwaddr() |
| self.cleanup_releases = [] |
| |
| def teardown_test(self): |
| for packet in self.cleanup_releases: |
| self._send(packet) |
| |
| def teardown_class(self): |
| self.stop_arp.set() |
| self._stop_usb_tethering(self.dut) |
| |
| def on_fail(self, test_name, begin_time): |
| self.dut.take_bug_report(test_name, begin_time) |
| |
| def _start_usb_tethering(self, dut): |
| """ Start USB tethering |
| |
| Args: |
| 1. dut - ad object |
| """ |
| self.log.info("Starting USB Tethering") |
| dut.stop_services() |
| dut.adb.shell(USB_TETHERING_MODE, ignore_status=True) |
| dut.adb.wait_for_device() |
| dut.start_services() |
| if 'rndis' not in dut.adb.shell(DEVICE_IP_ADDRESS): |
| raise signals.TestFailure('Unable to enable USB tethering.') |
| self.USB_TETHERED = True |
| |
| def _stop_usb_tethering(self, dut): |
| """ Stop USB tethering |
| |
| Args: |
| 1. dut - ad object |
| """ |
| self.log.info("Stopping USB Tethering") |
| dut.stop_services() |
| dut.adb.shell(USB_CHARGE_MODE) |
| dut.adb.wait_for_device() |
| dut.start_services() |
| self.USB_TETHERED = False |
| |
| def _wait_for_device(self, dut): |
| """ Wait for device to come back online |
| |
| Args: |
| 1. dut - ad object |
| """ |
| while dut.serial not in android_device.list_adb_devices(): |
| pass |
| dut.adb.wait_for_device() |
| |
| def _wait_for_new_iface(self, old_ifaces): |
| old_set = set(old_ifaces) |
| # Try 10 times to find a new interface with a 1s sleep every time |
| # (equivalent to a 9s timeout) |
| for i in range(0, 10): |
| new_ifaces = set(get_if_list()) - old_set |
| asserts.assert_true(len(new_ifaces) < 2, |
| "Too many new interfaces after turning on tethering") |
| if len(new_ifaces) == 1: |
| return new_ifaces.pop() |
| time.sleep(1) |
| asserts.fail("Timeout waiting for tethering interface on host") |
| |
| def _sniff_arp(self, stop_arp): |
| try: |
| sniff(iface=self.iface, filter='arp', prn=self._handle_arp, store=0, |
| stop_filter=lambda p: stop_arp.is_set()) |
| except: |
| # sniff may raise when tethering is disconnected. Ignore |
| # exceptions if stop was requested. |
| if not stop_arp.is_set(): |
| raise |
| |
| def _handle_arp(self, packet): |
| # Reply to all arp "who-has": say we have everything |
| if packet[ARP].op == ARP.who_has: |
| reply = ARP(op=ARP.is_at, hwsrc=self.real_hwaddr, psrc=packet.pdst, |
| hwdst=BROADCAST_MAC, pdst=SUBNET_BROADCAST) |
| sendp(Ether(dst=BROADCAST_MAC, src=self.real_hwaddr) / reply, |
| iface=self.iface, verbose=False) |
| |
| @test_tracker_info(uuid="a8712769-977a-4ee1-902f-90b3ba30b40c") |
| def test_config_assumptions(self): |
| resp = self._get_response(self._make_discover(self.hwaddr)) |
| asserts.assert_false(None == resp, "Device did not reply to discover") |
| asserts.assert_true(get_yiaddr(resp).startswith(NETADDR_PREFIX), |
| "Server does not use expected prefix") |
| |
| @test_tracker_info(uuid="e3761689-7d64-46b1-97ce-15f315eaf568") |
| def test_discover_broadcastbit(self): |
| resp = self._get_response( |
| self._make_discover(self.hwaddr, bcastbit=True)) |
| self._assert_offer(resp) |
| self._assert_broadcast(resp) |
| |
| @test_tracker_info(uuid="30a7ea7c-c20f-4c46-aaf2-96f19d8f8191") |
| def test_discover_bootpfields(self): |
| discover = self._make_discover(self.hwaddr) |
| resp = self._get_response(discover) |
| self._assert_offer(resp) |
| self._assert_unicast(resp) |
| bootp = assert_bootp_response(resp, discover) |
| asserts.assert_equal(INET4_ANY, bootp.ciaddr) |
| asserts.assert_equal(self.server_addr, bootp.siaddr) |
| asserts.assert_equal(INET4_ANY, bootp.giaddr) |
| asserts.assert_equal(self.hwaddr, get_chaddr(bootp)) |
| |
| @test_tracker_info(uuid="593f4051-516d-44fa-8834-7d485362f182") |
| def test_discover_relayed_broadcastbit(self): |
| giaddr = NETADDR_PREFIX + '123' |
| resp = self._get_response( |
| self._make_discover(self.hwaddr, giaddr=giaddr, bcastbit=True)) |
| self._assert_offer(resp) |
| self._assert_relayed(resp, giaddr) |
| self._assert_broadcastbit(resp) |
| |
| def _run_discover_paramrequestlist(self, params, unwanted_params): |
| params_opt = make_paramrequestlist_opt(params) |
| resp = self._get_response( |
| self._make_discover(self.hwaddr, options=[params_opt])) |
| |
| self._assert_offer(resp) |
| # List of requested params in response order |
| resp_opts = get_opt_labels(resp) |
| resp_requested_opts = [opt for opt in resp_opts if opt in params] |
| # All above params should be supported, order should be conserved |
| asserts.assert_equal(params, resp_requested_opts) |
| asserts.assert_equal(0, len(set(resp_opts) & set(unwanted_params))) |
| return resp |
| |
| @test_tracker_info(uuid="00a8a3f6-f143-47ff-a79b-482c607fb5b8") |
| def test_discover_paramrequestlist(self): |
| resp = self._run_discover_paramrequestlist( |
| ['subnet_mask', 'broadcast_address', 'router', 'name_server'], |
| unwanted_params=[]) |
| for opt in ['broadcast_address', 'router', 'name_server']: |
| asserts.assert_true(getopt(resp, opt).startswith(NETADDR_PREFIX), |
| opt + ' does not start with ' + NETADDR_PREFIX) |
| |
| subnet_mask = getopt(resp, 'subnet_mask') |
| asserts.assert_true(subnet_mask.startswith('255.255.'), |
| 'Unexpected subnet mask for /16+: ' + subnet_mask) |
| |
| @test_tracker_info(uuid="d1aad4a3-9eab-4900-aa6a-5b82a4a64f46") |
| def test_discover_paramrequestlist_rev(self): |
| # RFC2132 #9.8: "The DHCP server is not required to return the options |
| # in the requested order, but MUST try to insert the requested options |
| # in the order requested" |
| asserts.skip('legacy behavior not compliant: fixed order used') |
| self._run_discover_paramrequestlist( |
| ['name_server', 'router', 'broadcast_address', 'subnet_mask'], |
| unwanted_params=[]) |
| |
| @test_tracker_info(uuid="e3ae6335-8cc7-4bf1-bb58-67646b727f2b") |
| def test_discover_paramrequestlist_unwanted(self): |
| asserts.skip('legacy behavior always sends all parameters') |
| self._run_discover_paramrequestlist(['router', 'name_server'], |
| unwanted_params=['broadcast_address', 'subnet_mask']) |
| |
| def _assert_renews(self, request, addr, exp_time, resp_type=ACK): |
| # Sleep to test lease time renewal |
| time.sleep(3) |
| resp = self._get_response(request) |
| self._assert_type(resp, resp_type) |
| asserts.assert_equal(addr, get_yiaddr(resp)) |
| remaining_lease = getopt(resp, 'lease_time') |
| # Lease renewed: waited for 3s, lease time not decreased by more than 2 |
| asserts.assert_true(remaining_lease >= exp_time - 2, |
| 'Lease not renewed') |
| # Lease times should be consistent across offers/renewals |
| asserts.assert_true(remaining_lease <= exp_time + 2, |
| 'Lease time inconsistent') |
| return resp |
| |
| @test_tracker_info(uuid="d6b598b7-f443-4b5a-ba80-4af5d211cade") |
| def test_discover_assigned_ownaddress(self): |
| addr, siaddr, resp = self._request_address(self.hwaddr) |
| |
| lease_time = getopt(resp, 'lease_time') |
| server_id = getopt(resp, 'server_id') |
| asserts.assert_true(lease_time >= 60, "Lease time is too short") |
| asserts.assert_false(addr == INET4_ANY, "Assigned address is empty") |
| # Wait to test lease expiration time change |
| time.sleep(3) |
| |
| # New discover, same address |
| resp = self._assert_renews(self._make_discover(self.hwaddr), |
| addr, lease_time, resp_type=OFFER) |
| self._assert_unicast(resp, get_yiaddr(resp)) |
| self._assert_broadcastbit(resp, isset=False) |
| |
| @test_tracker_info(uuid="cbb07d77-912b-4269-bbbc-adba99779587") |
| def test_discover_assigned_otherhost(self): |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| |
| # New discover, same address, different client |
| resp = self._get_response(self._make_discover(self.other_hwaddr, |
| [('requested_addr', addr)])) |
| |
| self._assert_offer(resp) |
| asserts.assert_false(get_yiaddr(resp) == addr, |
| "Already assigned address offered") |
| self._assert_unicast(resp, get_yiaddr(resp)) |
| self._assert_broadcastbit(resp, isset=False) |
| |
| @test_tracker_info(uuid="3d2b3d2f-eb5f-498f-b887-3b4638cebf14") |
| def test_discover_requestaddress(self): |
| addr = NETADDR_PREFIX + '200' |
| resp = self._get_response(self._make_discover(self.hwaddr, |
| [('requested_addr', addr)])) |
| self._assert_offer(resp) |
| asserts.assert_equal(get_yiaddr(resp), addr) |
| |
| # Lease not committed: can request again |
| resp = self._get_response(self._make_discover(self.other_hwaddr, |
| [('requested_addr', addr)])) |
| self._assert_offer(resp) |
| asserts.assert_equal(get_yiaddr(resp), addr) |
| |
| @test_tracker_info(uuid="5ffd9d25-304e-434b-bedb-56ccf27dcebd") |
| def test_discover_requestaddress_wrongsubnet(self): |
| addr = OTHER_NETADDR_PREFIX + '200' |
| resp = self._get_response( |
| self._make_discover(self.hwaddr, [('requested_addr', addr)])) |
| self._assert_offer(resp) |
| self._assert_unicast(resp) |
| asserts.assert_false(get_yiaddr(resp) == addr, |
| 'Server offered invalid address') |
| |
| @test_tracker_info(uuid="f7d6a92f-9386-4b65-b6c1-d0a3f11213bf") |
| def test_discover_giaddr_outside_subnet(self): |
| giaddr = OTHER_NETADDR_PREFIX + '201' |
| resp = self._get_response( |
| self._make_discover(self.hwaddr, giaddr=giaddr)) |
| asserts.assert_equal(resp, None) |
| |
| @test_tracker_info(uuid="1348c79a-9203-4bb8-b33b-af80bacd17b1") |
| def test_discover_srcaddr_outside_subnet(self): |
| srcaddr = OTHER_NETADDR_PREFIX + '200' |
| resp = self._get_response( |
| self._make_discover(self.hwaddr, ip_src=srcaddr)) |
| self._assert_offer(resp) |
| asserts.assert_false(srcaddr == get_yiaddr(resp), |
| 'Server offered invalid address') |
| |
| @test_tracker_info(uuid="a03bb783-8665-4c66-9c0c-1bb02ddca07e") |
| def test_discover_requestaddress_giaddr_outside_subnet(self): |
| addr = NETADDR_PREFIX + '200' |
| giaddr = OTHER_NETADDR_PREFIX + '201' |
| req = self._make_discover(self.hwaddr, [('requested_addr', addr)], |
| ip_src=giaddr, giaddr=giaddr) |
| resp = self._get_response(req) |
| asserts.assert_equal(resp, None) |
| |
| @test_tracker_info(uuid="725956af-71e2-45d8-b8b3-402d21bfc7db") |
| def test_discover_knownaddress_giaddr_outside_subnet(self): |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| |
| # New discover, same client, through relay in invalid subnet |
| giaddr = OTHER_NETADDR_PREFIX + '200' |
| resp = self._get_response( |
| self._make_discover(self.hwaddr, giaddr=giaddr)) |
| asserts.assert_equal(resp, None) |
| |
| @test_tracker_info(uuid="2ee9d5b1-c15d-40c4-98e9-63202d1f1557") |
| def test_discover_knownaddress_giaddr_valid_subnet(self): |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| |
| # New discover, same client, through relay in valid subnet |
| giaddr = NETADDR_PREFIX + '200' |
| resp = self._get_response( |
| self._make_discover(self.hwaddr, giaddr=giaddr)) |
| self._assert_offer(resp) |
| self._assert_unicast(resp, giaddr) |
| self._assert_broadcastbit(resp, isset=False) |
| |
| @test_tracker_info(uuid="f43105a5-633a-417a-8a07-39bc36c493e7") |
| def test_request_unicast(self): |
| addr, siaddr, resp = self._request_address(self.hwaddr, bcast=False) |
| self._assert_unicast(resp, addr) |
| |
| @test_tracker_info(uuid="09f3c1c4-1202-4f85-a965-4d86aee069f3") |
| def test_request_bootpfields(self): |
| req_addr = NETADDR_PREFIX + '200' |
| req = self._make_request(self.hwaddr, req_addr, self.server_addr) |
| resp = self._get_response(req) |
| self._assert_ack(resp) |
| bootp = assert_bootp_response(resp, req) |
| asserts.assert_equal(INET4_ANY, bootp.ciaddr) |
| asserts.assert_equal(self.server_addr, bootp.siaddr) |
| asserts.assert_equal(INET4_ANY, bootp.giaddr) |
| asserts.assert_equal(self.hwaddr, get_chaddr(bootp)) |
| |
| @test_tracker_info(uuid="ec00d268-80cb-4be5-9771-2292cc7d2e18") |
| def test_request_selecting_inuse(self): |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| new_req = self._make_request(self.other_hwaddr, addr, siaddr) |
| resp = self._get_response(new_req) |
| self._assert_nak(resp) |
| self._assert_broadcast(resp) |
| bootp = assert_bootp_response(resp, new_req) |
| asserts.assert_equal(INET4_ANY, bootp.ciaddr) |
| asserts.assert_equal(INET4_ANY, bootp.yiaddr) |
| asserts.assert_equal(INET4_ANY, bootp.siaddr) |
| asserts.assert_equal(INET4_ANY, bootp.giaddr) |
| asserts.assert_equal(self.other_hwaddr, get_chaddr(bootp)) |
| asserts.assert_equal( |
| ['message-type', 'server_id', 56, 'end'], # 56 is "message" opt |
| get_opt_labels(bootp)) |
| asserts.assert_equal(self.server_addr, getopt(bootp, 'server_id')) |
| |
| @test_tracker_info(uuid="0643c179-3542-4297-9b06-8d86ff785e9c") |
| def test_request_selecting_wrongsiaddr(self): |
| addr = NETADDR_PREFIX + '200' |
| wrong_siaddr = NETADDR_PREFIX + '201' |
| asserts.assert_false(wrong_siaddr == self.server_addr, |
| 'Test assumption not met: server addr is ' + wrong_siaddr) |
| resp = self._get_response( |
| self._make_request(self.hwaddr, addr, siaddr=wrong_siaddr)) |
| asserts.assert_true(resp == None, |
| 'Received response for request with incorrect siaddr') |
| |
| @test_tracker_info(uuid="676beab2-4af8-4bf0-a4ad-c7626ae5987f") |
| def test_request_selecting_giaddr_outside_subnet(self): |
| addr = NETADDR_PREFIX + '200' |
| giaddr = OTHER_NETADDR_PREFIX + '201' |
| resp = self._get_response( |
| self._make_request(self.hwaddr, addr, siaddr=self.server_addr, |
| giaddr=giaddr)) |
| asserts.assert_equal(resp, None) |
| |
| @test_tracker_info(uuid="fe17df0c-2f41-416f-bb76-d75b74b63c0f") |
| def test_request_selecting_hostnameupdate(self): |
| addr = NETADDR_PREFIX + '123' |
| hostname1 = b'testhostname1' |
| hostname2 = b'testhostname2' |
| req = self._make_request(self.hwaddr, None, None, |
| options=[ |
| ('requested_addr', addr), |
| ('server_id', self.server_addr), |
| ('hostname', hostname1)]) |
| resp = self._get_response(req) |
| self._assert_ack(resp) |
| self._assert_unicast(resp, addr) |
| asserts.assert_equal(hostname1, getopt(req, 'hostname')) |
| |
| # Re-request with different hostname |
| setopt(req, 'hostname', hostname2) |
| resp = self._get_response(req) |
| self._assert_ack(resp) |
| self._assert_unicast(resp, addr) |
| asserts.assert_equal(hostname2, getopt(req, 'hostname')) |
| |
| def _run_initreboot(self, bcastbit): |
| addr, siaddr, resp = self._request_address(self.hwaddr) |
| exp = getopt(resp, 'lease_time') |
| |
| # init-reboot: siaddr is None |
| return self._assert_renews(self._make_request( |
| self.hwaddr, addr, siaddr=None, bcastbit=bcastbit), addr, exp) |
| |
| @test_tracker_info(uuid="263c91b9-cfe9-4f21-985d-b7046df80528") |
| def test_request_initreboot(self): |
| resp = self._run_initreboot(bcastbit=False) |
| self._assert_unicast(resp) |
| self._assert_broadcastbit(resp, isset=False) |
| |
| @test_tracker_info(uuid="f05dd60f-03dd-4e2b-8e58-80f4d752ad51") |
| def test_request_initreboot_broadcastbit(self): |
| resp = self._run_initreboot(bcastbit=True) |
| self._assert_broadcast(resp) |
| |
| @test_tracker_info(uuid="5563c616-2136-47f6-9151-4e28cbfe797c") |
| def test_request_initreboot_nolease(self): |
| # RFC2131 #4.3.2 |
| asserts.skip("legacy behavior not compliant") |
| addr = NETADDR_PREFIX + '123' |
| resp = self._get_response(self._make_request(self.hwaddr, addr, None)) |
| asserts.assert_equal(resp, None) |
| |
| @test_tracker_info(uuid="da5c5537-cb38-4a2e-828f-44bc97976fe5") |
| def test_request_initreboot_incorrectlease(self): |
| otheraddr = NETADDR_PREFIX + '123' |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| asserts.assert_false(addr == otheraddr, |
| "Test assumption not met: server assigned " + otheraddr) |
| |
| resp = self._get_response( |
| self._make_request(self.hwaddr, otheraddr, siaddr=None)) |
| self._assert_nak(resp) |
| self._assert_broadcast(resp) |
| |
| @test_tracker_info(uuid="ce42ba57-07be-427b-9cbd-5535c62b0120") |
| def test_request_initreboot_wrongnet(self): |
| resp = self._get_response(self._make_request(self.hwaddr, |
| OTHER_NETADDR_PREFIX + '1', siaddr=None)) |
| self._assert_nak(resp) |
| self._assert_broadcast(resp) |
| |
| def _run_rebinding(self, bcastbit, giaddr=INET4_ANY): |
| addr, siaddr, resp = self._request_address(self.hwaddr) |
| exp = getopt(resp, 'lease_time') |
| |
| # Rebinding: no siaddr or reqaddr |
| resp = self._assert_renews( |
| self._make_request(self.hwaddr, reqaddr=None, siaddr=None, |
| ciaddr=addr, giaddr=giaddr, ip_src=addr, |
| ip_dst=NETADDR_BROADCAST, bcastbit=bcastbit), |
| addr, exp) |
| return resp, addr |
| |
| @test_tracker_info(uuid="68bfcb25-5873-41ad-ad0a-bf22781534ca") |
| def test_request_rebinding(self): |
| resp, addr = self._run_rebinding(bcastbit=False) |
| self._assert_unicast(resp, addr) |
| self._assert_broadcastbit(resp, isset=False) |
| |
| @test_tracker_info(uuid="4c591536-8062-40ec-ae12-1ebe7dcad8e2") |
| def test_request_rebinding_relayed(self): |
| giaddr = NETADDR_PREFIX + '123' |
| resp, _ = self._run_rebinding(bcastbit=False, giaddr=giaddr) |
| self._assert_relayed(resp, giaddr) |
| self._assert_broadcastbit(resp, isset=False) |
| |
| @test_tracker_info(uuid="cee2668b-bd79-47d7-b358-8f9387d715b1") |
| def test_request_rebinding_inuse(self): |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| |
| resp = self._get_response(self._make_request( |
| self.other_hwaddr, reqaddr=None, siaddr=None, ciaddr=addr)) |
| self._assert_nak(resp) |
| self._assert_broadcast(resp) |
| |
| @test_tracker_info(uuid="d95d69b5-ab9a-42f5-8dd0-b9b6a6d960cc") |
| def test_request_rebinding_wrongaddr(self): |
| otheraddr = NETADDR_PREFIX + '123' |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| asserts.assert_false(addr == otheraddr, |
| "Test assumption not met: server assigned " + otheraddr) |
| |
| resp = self._get_response(self._make_request( |
| self.hwaddr, reqaddr=None, siaddr=siaddr, ciaddr=otheraddr)) |
| self._assert_nak(resp) |
| self._assert_broadcast(resp) |
| |
| @test_tracker_info(uuid="421a86b3-8779-4910-8050-7806536efabb") |
| def test_request_rebinding_wrongaddr_relayed(self): |
| otheraddr = NETADDR_PREFIX + '123' |
| relayaddr = NETADDR_PREFIX + '124' |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| asserts.assert_false(addr == otheraddr, |
| "Test assumption not met: server assigned " + otheraddr) |
| asserts.assert_false(addr == relayaddr, |
| "Test assumption not met: server assigned " + relayaddr) |
| |
| req = self._make_request(self.hwaddr, reqaddr=None, siaddr=None, |
| ciaddr=otheraddr, giaddr=relayaddr) |
| |
| resp = self._get_response(req) |
| self._assert_nak(resp) |
| self._assert_relayed(resp, relayaddr) |
| self._assert_broadcastbit(resp) |
| |
| @test_tracker_info(uuid="6ff1fab4-009a-4758-9153-0d9db63423da") |
| def test_release(self): |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| # Re-requesting fails |
| resp = self._get_response( |
| self._make_request(self.other_hwaddr, addr, siaddr)) |
| self._assert_nak(resp) |
| self._assert_broadcast(resp) |
| |
| # Succeeds after release |
| self._send(self._make_release(self.hwaddr, addr, siaddr)) |
| resp = self._get_response( |
| self._make_request(self.other_hwaddr, addr, siaddr)) |
| self._assert_ack(resp) |
| |
| @test_tracker_info(uuid="abb1a53e-6b6c-468f-88b9-ace9ca4d6593") |
| def test_release_noserverid(self): |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| |
| # Release without server_id opt is ignored |
| release = self._make_release(self.hwaddr, addr, siaddr) |
| removeopt(release, 'server_id') |
| self._send(release) |
| |
| # Not released: request fails |
| resp = self._get_response( |
| self._make_request(self.other_hwaddr, addr, siaddr)) |
| self._assert_nak(resp) |
| self._assert_broadcast(resp) |
| |
| @test_tracker_info(uuid="8415b69e-ae61-4474-8495-d783ba6818d1") |
| def test_release_wrongserverid(self): |
| addr, siaddr, _ = self._request_address(self.hwaddr) |
| |
| # Release with wrong server id |
| release = self._make_release(self.hwaddr, addr, siaddr) |
| setopt(release, 'server_id', addr) |
| self._send(release) |
| |
| # Not released: request fails |
| resp = self._get_response( |
| self._make_request(self.other_hwaddr, addr, siaddr)) |
| self._assert_nak(resp) |
| self._assert_broadcast(resp) |
| |
| @test_tracker_info(uuid="0858f678-3db2-4c12-a21b-6e16c5d7e7ce") |
| def test_unicast_l2l3(self): |
| reqAddr = NETADDR_PREFIX + '124' |
| resp = self._get_response(self._make_request( |
| self.hwaddr, reqAddr, siaddr=None)) |
| self._assert_unicast(resp) |
| str_hwaddr = format_hwaddr(self.hwaddr) |
| asserts.assert_equal(str_hwaddr, resp.getlayer(Ether).dst) |
| asserts.assert_equal(reqAddr, resp.getlayer(IP).dst) |
| asserts.assert_equal(CLIENT_PORT, resp.getlayer(UDP).dport) |
| |
| @test_tracker_info(uuid="bf05efe9-ee5b-46ba-9b3c-5a4441c13798") |
| def test_macos_10_13_3_discover(self): |
| params_opt = make_paramrequestlist_opt([ |
| 'subnet_mask', |
| 121, # Classless Static Route |
| 'router', |
| 'name_server', |
| 'domain', |
| 119, # Domain Search |
| 252, # Private/Proxy autodiscovery |
| 95, # LDAP |
| 'NetBIOS_server', |
| 46, # NetBIOS over TCP/IP Node Type |
| ]) |
| req = self._make_discover(self.hwaddr, |
| options=[ |
| params_opt, |
| ('max_dhcp_size', 1500), |
| # HW type Ethernet (0x01) |
| ('client_id', b'\x01' + self.hwaddr), |
| ('lease_time', 7776000), |
| ('hostname', b'test12-macbookpro'), |
| ], opts_padding=bytes(6)) |
| req.getlayer(BOOTP).secs = 2 |
| resp = self._get_response(req) |
| self._assert_standard_offer(resp) |
| |
| def _make_macos_10_13_3_paramrequestlist(self): |
| return make_paramrequestlist_opt([ |
| 'subnet_mask', |
| 121, # Classless Static Route |
| 'router', |
| 'name_server', |
| 'domain', |
| 119, # Domain Search |
| 252, # Private/Proxy autodiscovery |
| 95, # LDAP |
| 44, # NetBIOS over TCP/IP Name Server |
| 46, # NetBIOS over TCP/IP Node Type |
| ]) |
| |
| @test_tracker_info(uuid="bf05efe9-ee5b-46ba-9b3c-5a4441c13798") |
| def test_macos_10_13_3_discover(self): |
| req = self._make_discover(self.hwaddr, |
| options=[ |
| self._make_macos_10_13_3_paramrequestlist(), |
| ('max_dhcp_size', 1500), |
| # HW type Ethernet (0x01) |
| ('client_id', b'\x01' + self.hwaddr), |
| ('lease_time', 7776000), |
| ('hostname', b'test12-macbookpro'), |
| ], opts_padding=bytes(6)) |
| req.getlayer(BOOTP).secs = 2 |
| resp = self._get_response(req) |
| self._assert_offer(resp) |
| self._assert_standard_offer_or_ack(resp) |
| |
| @test_tracker_info(uuid="7acc796b-c4f1-46cc-8ffb-0a0efb05ae86") |
| def test_macos_10_13_3_request_selecting(self): |
| req = self._make_request(self.hwaddr, None, None, |
| options=[ |
| self._make_macos_10_13_3_paramrequestlist(), |
| ('max_dhcp_size', 1500), |
| # HW type Ethernet (0x01) |
| ('client_id', b'\x01' + self.hwaddr), |
| ('requested_addr', NETADDR_PREFIX + '109'), |
| ('server_id', self.server_addr), |
| ('hostname', b'test12-macbookpro'), |
| ]) |
| req.getlayer(BOOTP).secs = 5 |
| resp = self._get_response(req) |
| self._assert_ack(resp) |
| self._assert_standard_offer_or_ack(resp) |
| |
| # Note: macOS does not seem to do any rebinding (straight to discover) |
| @test_tracker_info(uuid="e8f0b60c-9ea3-4184-8426-151a395bff5b") |
| def test_macos_10_13_3_request_renewing(self): |
| req_ip = NETADDR_PREFIX + '109' |
| req = self._make_request(self.hwaddr, None, None, |
| ciaddr=req_ip, ip_src=req_ip, ip_dst=self.server_addr, options=[ |
| self._make_macos_10_13_3_paramrequestlist(), |
| ('max_dhcp_size', 1500), |
| # HW type Ethernet (0x01) |
| ('client_id', b'\x01' + self.hwaddr), |
| ('lease_time', 7776000), |
| ('hostname', b'test12-macbookpro'), |
| ], opts_padding=bytes(6)) |
| resp = self._get_response(req) |
| self._assert_ack(resp) |
| self._assert_standard_offer_or_ack(resp, renewing=True) |
| |
| def _make_win10_paramrequestlist(self): |
| return make_paramrequestlist_opt([ |
| 'subnet_mask', |
| 'router', |
| 'name_server', |
| 'domain', |
| 31, # Perform Router Discover |
| 33, # Static Route |
| 'vendor_specific', |
| 44, # NetBIOS over TCP/IP Name Server |
| 46, # NetBIOS over TCP/IP Node Type |
| 47, # NetBIOS over TCP/IP Scope |
| 121, # Classless Static Route |
| 249, # Private/Classless Static Route (MS) |
| 252, # Private/Proxy autodiscovery |
| ]) |
| |
| @test_tracker_info(uuid="11b3db9c-4cd7-4088-99dc-881f25ce4e76") |
| def test_win10_discover(self): |
| req = self._make_discover(self.hwaddr, bcastbit=True, |
| options=[ |
| # HW type Ethernet (0x01) |
| ('client_id', b'\x01' + self.hwaddr), |
| ('hostname', b'test120-w'), |
| ('vendor_class_id', b'MSFT 5.0'), |
| self._make_win10_paramrequestlist(), |
| ], opts_padding=bytes(11)) |
| req.getlayer(BOOTP).secs = 2 |
| resp = self._get_response(req) |
| self._assert_offer(resp) |
| self._assert_standard_offer_or_ack(resp, bcast=True) |
| |
| @test_tracker_info(uuid="4fe04e7f-c643-4a19-b15c-cf417b2c9410") |
| def test_win10_request_selecting(self): |
| req = self._make_request(self.hwaddr, None, None, bcastbit=True, |
| options=[ |
| ('max_dhcp_size', 1500), |
| # HW type Ethernet (0x01) |
| ('client_id', b'\x01' + self.hwaddr), |
| ('requested_addr', NETADDR_PREFIX + '109'), |
| ('server_id', self.server_addr), |
| ('hostname', b'test120-w'), |
| # Client Fully Qualified Domain Name |
| (81, b'\x00\x00\x00test120-w.ad.tst.example.com'), |
| ('vendor_class_id', b'MSFT 5.0'), |
| self._make_win10_paramrequestlist(), |
| ]) |
| resp = self._get_response(req) |
| self._assert_ack(resp) |
| self._assert_standard_offer_or_ack(resp, bcast=True) |
| |
| def _run_win10_request_renewing(self, bcast): |
| req_ip = NETADDR_PREFIX + '109' |
| req = self._make_request(self.hwaddr, None, None, bcastbit=bcast, |
| ciaddr=req_ip, ip_src=req_ip, |
| ip_dst=NETADDR_BROADCAST if bcast else self.server_addr, |
| options=[ |
| ('max_dhcp_size', 1500), |
| # HW type Ethernet (0x01) |
| ('client_id', b'\x01' + self.hwaddr), |
| ('hostname', b'test120-w'), |
| # Client Fully Qualified Domain Name |
| (81, b'\x00\x00\x00test120-w.ad.tst.example.com'), |
| ('vendor_class_id', b'MSFT 5.0'), |
| self._make_win10_paramrequestlist(), |
| ]) |
| resp = self._get_response(req) |
| self._assert_ack(resp) |
| self._assert_standard_offer_or_ack(resp, renewing=True, bcast=bcast) |
| |
| @test_tracker_info(uuid="1b23c9c7-cc94-42d0-83a6-f1b2bc125fb9") |
| def test_win10_request_renewing(self): |
| self._run_win10_request_renewing(bcast=False) |
| |
| @test_tracker_info(uuid="c846bd14-71fb-4492-a4d3-0aa5c2c35751") |
| def test_win10_request_rebinding(self): |
| self._run_win10_request_renewing(bcast=True) |
| |
| def _make_debian_paramrequestlist(self): |
| return make_paramrequestlist_opt([ |
| 'subnet_mask', |
| 'broadcast_address', |
| 'router', |
| 'name_server', |
| 119, # Domain Search |
| 'hostname', |
| 101, # TCode |
| 'domain', # NetBIOS over TCP/IP Name Server |
| 'vendor_specific', # NetBIOS over TCP/IP Node Type |
| 121, # Classless Static Route |
| 249, # Private/Classless Static Route (MS) |
| 33, # Static Route |
| 252, # Private/Proxy autodiscovery |
| 'NTP_server', |
| ]) |
| |
| @test_tracker_info(uuid="b0bb6ae7-07e6-4ecb-9a2f-db9c8146a3d5") |
| def test_debian_dhclient_4_3_5_discover(self): |
| req_ip = NETADDR_PREFIX + '109' |
| req = self._make_discover(self.hwaddr, |
| options=[ |
| ('requested_addr', req_ip), |
| ('hostname', b'test12'), |
| self._make_debian_paramrequestlist(), |
| ], opts_padding=bytes(26)) |
| resp = self._get_response(req) |
| self._assert_offer(resp) |
| # Don't test for hostname option: previous implementation would not |
| # set it in offer, which was not consistent with ack |
| self._assert_standard_offer_or_ack(resp, ignore_hostname=True) |
| asserts.assert_equal(req_ip, get_yiaddr(resp)) |
| |
| @test_tracker_info(uuid="d70bc043-84cb-4735-9123-c46c6d1ce5ac") |
| def test_debian_dhclient_4_3_5_request_selecting(self): |
| req = self._make_request(self.hwaddr, None, None, |
| options=[ |
| ('server_id', self.server_addr), |
| ('requested_addr', NETADDR_PREFIX + '109'), |
| ('hostname', b'test12'), |
| self._make_debian_paramrequestlist(), |
| ], opts_padding=bytes(20)) |
| resp = self._get_response(req) |
| self._assert_ack(resp) |
| self._assert_standard_offer_or_ack(resp, with_hostname=True) |
| |
| def _run_debian_renewing(self, bcast): |
| req_ip = NETADDR_PREFIX + '109' |
| req = self._make_request(self.hwaddr, None, None, |
| ciaddr=req_ip, ip_src=req_ip, |
| ip_dst=NETADDR_BROADCAST if bcast else self.server_addr, |
| options=[ |
| ('hostname', b'test12'), |
| self._make_debian_paramrequestlist(), |
| ], |
| opts_padding=bytes(32)) |
| resp = self._get_response(req) |
| self._assert_ack(resp) |
| self._assert_standard_offer_or_ack(resp, renewing=True, |
| with_hostname=True) |
| |
| @test_tracker_info(uuid="5e1e817d-9972-46ca-8d44-1e120bf1bafc") |
| def test_debian_dhclient_4_3_5_request_renewing(self): |
| self._run_debian_renewing(bcast=False) |
| |
| @test_tracker_info(uuid="b179a36d-910e-4006-a79a-11cc561b69db") |
| def test_debian_dhclient_4_3_5_request_rebinding(self): |
| self._run_debian_renewing(bcast=True) |
| |
| def _assert_standard_offer_or_ack(self, resp, renewing=False, bcast=False, |
| ignore_hostname=False, with_hostname=False): |
| # Responses to renew/rebind are always unicast to ciaddr even with |
| # broadcast flag set (RFC does not define this behavior, but this is |
| # more efficient and matches previous behavior) |
| if bcast and not renewing: |
| self._assert_broadcast(resp) |
| self._assert_broadcastbit(resp, isset=True) |
| else: |
| # Previous implementation would set the broadcast flag but send a |
| # unicast reply if (bcast and renewing). This was not consistent and |
| # new implementation consistently clears the flag. Not testing for |
| # broadcast flag value to maintain compatibility. |
| self._assert_unicast(resp) |
| |
| bootp_resp = resp.getlayer(BOOTP) |
| asserts.assert_equal(0, bootp_resp.hops) |
| if renewing: |
| asserts.assert_true(bootp_resp.ciaddr.startswith(NETADDR_PREFIX), |
| 'ciaddr does not start with expected prefix') |
| else: |
| asserts.assert_equal(INET4_ANY, bootp_resp.ciaddr) |
| asserts.assert_true(bootp_resp.yiaddr.startswith(NETADDR_PREFIX), |
| 'yiaddr does not start with expected prefix') |
| asserts.assert_true(bootp_resp.siaddr.startswith(NETADDR_PREFIX), |
| 'siaddr does not start with expected prefix') |
| asserts.assert_equal(INET4_ANY, bootp_resp.giaddr) |
| |
| opt_labels = get_opt_labels(bootp_resp) |
| # FQDN option 81 is not supported in new behavior |
| opt_labels = [opt for opt in opt_labels if opt != 81] |
| |
| # Expect exactly these options in this order |
| expected_opts = [ |
| 'message-type', 'server_id', 'lease_time', 'renewal_time', |
| 'rebinding_time', 'subnet_mask', 'broadcast_address', 'router', |
| 'name_server'] |
| if ignore_hostname: |
| opt_labels = [opt for opt in opt_labels if opt != 'hostname'] |
| elif with_hostname: |
| expected_opts.append('hostname') |
| expected_opts.extend(['vendor_specific', 'end']) |
| asserts.assert_equal(expected_opts, opt_labels) |
| |
| def _request_address(self, hwaddr, bcast=True): |
| resp = self._get_response(self._make_discover(hwaddr)) |
| self._assert_offer(resp) |
| addr = get_yiaddr(resp) |
| siaddr = getopt(resp, 'server_id') |
| resp = self._get_response(self._make_request(hwaddr, addr, siaddr, |
| ip_dst=(INET4_ANY if bcast else siaddr))) |
| self._assert_ack(resp) |
| return addr, siaddr, resp |
| |
| def _get_response(self, packet): |
| resp = srp1(packet, iface=self.iface, timeout=10, verbose=False) |
| bootp_resp = (resp or None) and resp.getlayer(BOOTP) |
| if bootp_resp != None and get_mess_type(bootp_resp) == ACK: |
| # Note down corresponding release for this request |
| release = self._make_release(bootp_resp.chaddr, bootp_resp.yiaddr, |
| getopt(bootp_resp, 'server_id')) |
| self.cleanup_releases.append(release) |
| return resp |
| |
| def _send(self, packet): |
| sendp(packet, iface=self.iface, verbose=False) |
| |
| def _assert_type(self, packet, tp): |
| asserts.assert_false(None == packet, "No packet") |
| asserts.assert_equal(tp, get_mess_type(packet)) |
| |
| def _assert_ack(self, packet): |
| self._assert_type(packet, ACK) |
| |
| def _assert_nak(self, packet): |
| self._assert_type(packet, NAK) |
| |
| def _assert_offer(self, packet): |
| self._assert_type(packet, OFFER) |
| |
| def _assert_broadcast(self, packet): |
| asserts.assert_false(None == packet, "No packet") |
| asserts.assert_equal(packet.getlayer(Ether).dst, BROADCAST_MAC) |
| asserts.assert_equal(packet.getlayer(IP).dst, NETADDR_BROADCAST) |
| self._assert_broadcastbit(packet) |
| |
| def _assert_broadcastbit(self, packet, isset=True): |
| mask = 0x8000 |
| flag = packet.getlayer(BOOTP).flags |
| asserts.assert_equal(flag & mask, mask if isset else 0) |
| |
| def _assert_unicast(self, packet, ipAddr=None): |
| asserts.assert_false(None == packet, "No packet") |
| asserts.assert_false(packet.getlayer(Ether).dst == BROADCAST_MAC, |
| "Layer 2 packet destination address was broadcast") |
| if ipAddr: |
| asserts.assert_equal(packet.getlayer(IP).dst, ipAddr) |
| |
| def _assert_relayed(self, packet, giaddr): |
| self._assert_unicast(packet, giaddr) |
| asserts.assert_equal(giaddr, packet.getlayer(BOOTP).giaddr, |
| 'Relayed response has invalid giaddr field') |
| |
| def _next_hwaddr(self): |
| addr = make_hwaddr(self.next_hwaddr_index) |
| self.next_hwaddr_index = self.next_hwaddr_index + 1 |
| return addr |
| |
| def _make_dhcp(self, src_hwaddr, options, ciaddr=INET4_ANY, |
| ip_src=INET4_ANY, ip_dst=NETADDR_BROADCAST, giaddr=INET4_ANY, |
| bcastbit=False): |
| broadcast = (ip_dst == NETADDR_BROADCAST) |
| ethernet = Ether(dst=(BROADCAST_MAC if broadcast else self.dut_hwaddr)) |
| ip = IP(src=ip_src, dst=ip_dst) |
| udp = UDP(sport=68, dport=SERVER_PORT) |
| bootp = BOOTP(chaddr=src_hwaddr, ciaddr=ciaddr, giaddr=giaddr, |
| flags=(0x8000 if bcastbit else 0), xid=random.randrange(0, 2**32)) |
| dhcp = DHCP(options=options) |
| return ethernet / ip / udp / bootp / dhcp |
| |
| def _make_discover(self, src_hwaddr, options = [], giaddr=INET4_ANY, |
| bcastbit=False, opts_padding=None, ip_src=INET4_ANY): |
| opts = [('message-type','discover')] |
| opts.extend(options) |
| opts.append('end') |
| if (opts_padding): |
| opts.append(opts_padding) |
| return self._make_dhcp(src_hwaddr, options=opts, giaddr=giaddr, |
| ip_dst=NETADDR_BROADCAST, bcastbit=bcastbit, ip_src=ip_src) |
| |
| def _make_request(self, src_hwaddr, reqaddr, siaddr, ciaddr=INET4_ANY, |
| ip_dst=None, ip_src=None, giaddr=INET4_ANY, bcastbit=False, |
| options=[], opts_padding=None): |
| if not ip_dst: |
| ip_dst = siaddr or INET4_ANY |
| |
| if not ip_src and ip_dst == INET4_ANY: |
| ip_src = INET4_ANY |
| elif not ip_src: |
| ip_src = (giaddr if not isempty(giaddr) |
| else ciaddr if not isempty(ciaddr) |
| else reqaddr) |
| # Kernel will not receive unicast UDP packets with empty ip_src |
| asserts.assert_false(ip_dst != INET4_ANY and isempty(ip_src), |
| "Unicast ip_src cannot be zero") |
| opts = [('message-type', 'request')] |
| if options: |
| opts.extend(options) |
| else: |
| if siaddr: |
| opts.append(('server_id', siaddr)) |
| if reqaddr: |
| opts.append(('requested_addr', reqaddr)) |
| opts.append('end') |
| if opts_padding: |
| opts.append(opts_padding) |
| return self._make_dhcp(src_hwaddr, options=opts, ciaddr=ciaddr, |
| ip_src=ip_src, ip_dst=ip_dst, giaddr=giaddr, bcastbit=bcastbit) |
| |
| def _make_release(self, src_hwaddr, addr, server_id): |
| opts = [('message-type', 'release'), ('server_id', server_id), 'end'] |
| return self._make_dhcp(src_hwaddr, opts, ciaddr=addr, ip_src=addr, |
| ip_dst=server_id) |
| |
| def assert_bootp_response(resp, req): |
| bootp = resp.getlayer(BOOTP) |
| asserts.assert_equal(2, bootp.op, 'Invalid BOOTP op') |
| asserts.assert_equal(1, bootp.htype, 'Invalid BOOTP htype') |
| asserts.assert_equal(6, bootp.hlen, 'Invalid BOOTP hlen') |
| asserts.assert_equal(0, bootp.hops, 'Invalid BOOTP hops') |
| asserts.assert_equal(req.getlayer(BOOTP).xid, bootp.xid, 'Invalid XID') |
| return bootp |
| |
| |
| def make_paramrequestlist_opt(params): |
| param_indexes = [DHCPRevOptions[opt][0] if isinstance(opt, str) else opt |
| for opt in params] |
| return tuple(['param_req_list'] + [ |
| opt.to_bytes(1, byteorder='big') if isinstance(opt, int) else opt |
| for opt in param_indexes]) |
| |
| |
| def isempty(addr): |
| return not addr or addr == INET4_ANY |
| |
| |
| def setopt(packet, optname, val): |
| dhcp = packet.getlayer(DHCP) |
| if optname in get_opt_labels(dhcp): |
| dhcp.options = [(optname, val) if opt[0] == optname else opt |
| for opt in dhcp.options] |
| else: |
| # Add before the last option (last option is "end") |
| dhcp.options.insert(len(dhcp.options) - 1, (optname, val)) |
| |
| |
| def getopt(packet, key): |
| opts = [opt[1] for opt in packet.getlayer(DHCP).options if opt[0] == key] |
| return opts[0] if opts else None |
| |
| |
| def removeopt(packet, key): |
| dhcp = packet.getlayer(DHCP) |
| dhcp.options = [opt for opt in dhcp.options if opt[0] != key] |
| |
| |
| def get_opt_labels(packet): |
| dhcp_resp = packet.getlayer(DHCP) |
| # end option is a single string, not a tuple. |
| return [opt if isinstance(opt, str) else opt[0] |
| for opt in dhcp_resp.options if opt != 'pad'] |
| |
| |
| def get_yiaddr(packet): |
| return packet.getlayer(BOOTP).yiaddr |
| |
| |
| def get_chaddr(packet): |
| # We use Ethernet addresses. Ignore address padding |
| return packet.getlayer(BOOTP).chaddr[:6] |
| |
| |
| def get_mess_type(packet): |
| return getopt(packet, 'message-type') |
| |
| |
| def make_hwaddr(index): |
| if index > 0xffff: |
| raise ValueError("Address index out of range") |
| return b'\x44\x85\x00\x00' + bytes([index >> 8, index & 0xff]) |
| |
| |
| def format_hwaddr(addr): |
| return ':'.join(['%02x' % c for c in addr]) |
