# 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.
|
|
"""
|
Programmable testing DHCP server.
|
|
Simple DHCP server you can program with expectations of future packets and
|
responses to those packets. The server is basically a thin wrapper around a
|
server socket with some utility logic to make setting up tests easier. To write
|
a test, you start a server, construct a sequence of handling rules.
|
|
Handling rules let you set up expectations of future packets of certain types.
|
Handling rules are processed in order, and only the first remaining handler
|
handles a given packet. In theory you could write the entire test into a single
|
handling rule and keep an internal state machine for how far that handler has
|
gotten through the test. This would be poor style however. Correct style is to
|
write (or reuse) a handler for each packet the server should see, leading us to
|
a happy land where any conceivable packet handler has already been written for
|
us.
|
|
Example usage:
|
|
# Start up the DHCP server, which will ignore packets until a test is started
|
server = DhcpTestServer(interface="veth_master")
|
server.start()
|
|
# Given a list of handling rules, start a test with a 30 sec timeout.
|
handling_rules = []
|
handling_rules.append(DhcpHandlingRule_RespondToDiscovery(intended_ip,
|
intended_subnet_mask,
|
dhcp_server_ip,
|
lease_time_seconds)
|
server.start_test(handling_rules, 30.0)
|
|
# Trigger DHCP clients to do various test related actions
|
...
|
|
# Get results
|
server.wait_for_test_to_finish()
|
if (server.last_test_passed):
|
...
|
else:
|
...
|
|
|
Note that if you make changes, make sure that the tests in dhcp_unittest.py
|
still pass.
|
"""
|
|
import logging
|
import socket
|
import threading
|
import time
|
import traceback
|
|
from autotest_lib.client.cros import dhcp_packet
|
from autotest_lib.client.cros import dhcp_handling_rule
|
|
# From socket.h
|
SO_BINDTODEVICE = 25
|
|
class DhcpTestServer(threading.Thread):
|
def __init__(self,
|
interface=None,
|
ingress_address="<broadcast>",
|
ingress_port=67,
|
broadcast_address="255.255.255.255",
|
broadcast_port=68):
|
super(DhcpTestServer, self).__init__()
|
self._mutex = threading.Lock()
|
self._ingress_address = ingress_address
|
self._ingress_port = ingress_port
|
self._broadcast_port = broadcast_port
|
self._broadcast_address = broadcast_address
|
self._socket = None
|
self._interface = interface
|
self._stopped = False
|
self._test_in_progress = False
|
self._last_test_passed = False
|
self._test_timeout = 0
|
self._handling_rules = []
|
self._logger = logging.getLogger("dhcp.test_server")
|
self._exception = None
|
self.daemon = False
|
|
@property
|
def stopped(self):
|
with self._mutex:
|
return self._stopped
|
|
@property
|
def is_healthy(self):
|
with self._mutex:
|
return self._socket is not None
|
|
@property
|
def test_in_progress(self):
|
with self._mutex:
|
return self._test_in_progress
|
|
@property
|
def last_test_passed(self):
|
with self._mutex:
|
return self._last_test_passed
|
|
@property
|
def current_rule(self):
|
"""
|
Return the currently active DhcpHandlingRule.
|
"""
|
with self._mutex:
|
return self._handling_rules[0]
|
|
def start(self):
|
"""
|
Start the DHCP server. Only call this once.
|
"""
|
if self.is_alive():
|
return False
|
self._logger.info("DhcpTestServer started; opening sockets.")
|
try:
|
self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
self._logger.info("Opening socket on '%s' port %d." %
|
(self._ingress_address, self._ingress_port))
|
self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
if self._interface is not None:
|
self._logger.info("Binding to %s" % self._interface)
|
self._socket.setsockopt(socket.SOL_SOCKET,
|
SO_BINDTODEVICE,
|
self._interface)
|
self._socket.bind((self._ingress_address, self._ingress_port))
|
# Wait 100 ms for a packet, then return, thus keeping the thread
|
# active but mostly idle.
|
self._socket.settimeout(0.1)
|
except socket.error, socket_error:
|
self._logger.error("Socket error: %s." % str(socket_error))
|
self._logger.error(traceback.format_exc())
|
if not self._socket is None:
|
self._socket.close()
|
self._socket = None
|
self._logger.error("Failed to open server socket. Aborting.")
|
return
|
super(DhcpTestServer, self).start()
|
|
def stop(self):
|
"""
|
Stop the DHCP server and free its socket.
|
"""
|
with self._mutex:
|
self._stopped = True
|
|
def start_test(self, handling_rules, test_timeout_seconds):
|
"""
|
Start a new test using |handling_rules|. The server will call the
|
test successfull if it receives a RESPONSE_IGNORE_SUCCESS (or
|
RESPONSE_RESPOND_SUCCESS) from a handling_rule before
|
|test_timeout_seconds| passes. If the timeout passes without that
|
message, the server runs out of handling rules, or a handling rule
|
return RESPONSE_FAIL, the test is ended and marked as not passed.
|
|
All packets received before start_test() is called are received and
|
ignored.
|
"""
|
with self._mutex:
|
self._test_timeout = time.time() + test_timeout_seconds
|
self._handling_rules = handling_rules
|
self._test_in_progress = True
|
self._last_test_passed = False
|
self._exception = None
|
|
def wait_for_test_to_finish(self):
|
"""
|
Block on the test finishing in a CPU friendly way. Timeouts, successes,
|
and failures count as finishes.
|
"""
|
while self.test_in_progress:
|
time.sleep(0.1)
|
if self._exception:
|
raise self._exception
|
|
def abort_test(self):
|
"""
|
Abort a test prematurely, counting the test as a failure.
|
"""
|
with self._mutex:
|
self._logger.info("Manually aborting test.")
|
self._end_test_unsafe(False)
|
|
def _teardown(self):
|
with self._mutex:
|
self._socket.close()
|
self._socket = None
|
|
def _end_test_unsafe(self, passed):
|
if not self._test_in_progress:
|
return
|
if passed:
|
self._logger.info("DHCP server says test passed.")
|
else:
|
self._logger.info("DHCP server says test failed.")
|
self._test_in_progress = False
|
self._last_test_passed = passed
|
|
def _send_response_unsafe(self, packet):
|
if packet is None:
|
self._logger.error("Handling rule failed to return a packet.")
|
return False
|
self._logger.debug("Sending response: %s" % packet)
|
binary_string = packet.to_binary_string()
|
if binary_string is None or len(binary_string) < 1:
|
self._logger.error("Packet failed to serialize to binary string.")
|
return False
|
|
self._socket.sendto(binary_string,
|
(self._broadcast_address, self._broadcast_port))
|
return True
|
|
def _loop_body(self):
|
with self._mutex:
|
if self._test_in_progress and self._test_timeout < time.time():
|
# The test has timed out, so we abort it. However, we should
|
# continue to accept packets, so we fall through.
|
self._logger.error("Test in progress has timed out.")
|
self._end_test_unsafe(False)
|
try:
|
data, _ = self._socket.recvfrom(1024)
|
self._logger.info("Server received packet of length %d." %
|
len(data))
|
except socket.timeout:
|
# No packets available, lets return and see if the server has
|
# been shut down in the meantime.
|
return
|
|
# Receive packets when no test is in progress, just don't process
|
# them.
|
if not self._test_in_progress:
|
return
|
|
packet = dhcp_packet.DhcpPacket(byte_str=data)
|
if not packet.is_valid:
|
self._logger.warning("Server received an invalid packet over a "
|
"DHCP port?")
|
return
|
|
logging.debug("Server received a DHCP packet: %s." % packet)
|
if len(self._handling_rules) < 1:
|
self._logger.info("No handling rule for packet: %s." %
|
str(packet))
|
self._end_test_unsafe(False)
|
return
|
|
handling_rule = self._handling_rules[0]
|
response_code = handling_rule.handle(packet)
|
logging.info("Handler gave response: %d" % response_code)
|
if response_code & dhcp_handling_rule.RESPONSE_POP_HANDLER:
|
self._handling_rules.pop(0)
|
|
if response_code & dhcp_handling_rule.RESPONSE_HAVE_RESPONSE:
|
for response_instance in range(
|
handling_rule.response_packet_count):
|
response = handling_rule.respond(packet)
|
if not self._send_response_unsafe(response):
|
self._logger.error(
|
"Failed to send packet, ending test.")
|
self._end_test_unsafe(False)
|
return
|
|
if response_code & dhcp_handling_rule.RESPONSE_TEST_FAILED:
|
self._logger.info("Handling rule %s rejected packet %s." %
|
(handling_rule, packet))
|
self._end_test_unsafe(False)
|
return
|
|
if response_code & dhcp_handling_rule.RESPONSE_TEST_SUCCEEDED:
|
self._end_test_unsafe(True)
|
return
|
|
def run(self):
|
"""
|
Main method of the thread. Never call this directly, since it assumes
|
some setup done in start().
|
"""
|
with self._mutex:
|
if self._socket is None:
|
self._logger.error("Failed to create server socket, exiting.")
|
return
|
|
self._logger.info("DhcpTestServer entering handling loop.")
|
while not self.stopped:
|
try:
|
self._loop_body()
|
# Python does not have waiting queues on Lock objects.
|
# Give other threads a change to hold the mutex by
|
# forcibly releasing the GIL while we sleep.
|
time.sleep(0.01)
|
except Exception as e:
|
with self._mutex:
|
self._end_test_unsafe(False)
|
self._exception = e
|
with self._mutex:
|
self._end_test_unsafe(False)
|
self._logger.info("DhcpTestServer closing sockets.")
|
self._teardown()
|
self._logger.info("DhcpTestServer exiting.")
|
