# 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.
|
|
"""
|
Tools for serializing and deserializing DHCP packets.
|
|
DhcpPacket is a class that represents a single DHCP packet and contains some
|
logic to create and parse binary strings containing on the wire DHCP packets.
|
|
While you could call the constructor explicitly, most users should use the
|
static factories to construct packets with reasonable default values in most of
|
the fields, even if those values are zeros.
|
|
For example:
|
|
packet = dhcp_packet.create_offer_packet(transaction_id,
|
hwmac_addr,
|
offer_ip,
|
server_ip)
|
socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
# Sending to the broadcast address needs special permissions.
|
socket.sendto(response_packet.to_binary_string(),
|
("255.255.255.255", 68))
|
|
Note that if you make changes, make sure that the tests in the bottom of this
|
file still pass.
|
"""
|
|
import collections
|
import logging
|
import random
|
import socket
|
import struct
|
|
|
def CreatePacketPieceClass(super_class, field_format):
|
class PacketPiece(super_class):
|
@staticmethod
|
def pack(value):
|
return struct.pack(field_format, value)
|
|
@staticmethod
|
def unpack(byte_string):
|
return struct.unpack(field_format, byte_string)[0]
|
return PacketPiece
|
|
"""
|
Represents an option in a DHCP packet. Options may or may not be present in any
|
given packet, depending on the configurations of the client and the server.
|
Using namedtuples as super classes gets us the comparison operators we want to
|
use these Options in dictionaries as keys. Below, we'll subclass Option to
|
reflect that different kinds of options serialize to on the wire formats in
|
different ways.
|
|
|name|
|
A human readable name for this option.
|
|
|number|
|
Every DHCP option has a number that goes into the packet to indicate
|
which particular option is being encoded in the next few bytes. This
|
property returns that number for each option.
|
"""
|
Option = collections.namedtuple("Option", ["name", "number"])
|
|
ByteOption = CreatePacketPieceClass(Option, "!B")
|
|
ShortOption = CreatePacketPieceClass(Option, "!H")
|
|
IntOption = CreatePacketPieceClass(Option, "!I")
|
|
class IpAddressOption(Option):
|
@staticmethod
|
def pack(value):
|
return socket.inet_aton(value)
|
|
@staticmethod
|
def unpack(byte_string):
|
return socket.inet_ntoa(byte_string)
|
|
|
class IpListOption(Option):
|
@staticmethod
|
def pack(value):
|
return "".join([socket.inet_aton(addr) for addr in value])
|
|
@staticmethod
|
def unpack(byte_string):
|
return [socket.inet_ntoa(byte_string[idx:idx+4])
|
for idx in range(0, len(byte_string), 4)]
|
|
|
class RawOption(Option):
|
@staticmethod
|
def pack(value):
|
return value
|
|
@staticmethod
|
def unpack(byte_string):
|
return byte_string
|
|
|
class ByteListOption(Option):
|
@staticmethod
|
def pack(value):
|
return "".join(chr(v) for v in value)
|
|
@staticmethod
|
def unpack(byte_string):
|
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.
|
There are some clever compression optimizations that it does not implement
|
for serialization, but correctly parses. This should be sufficient for
|
testing.
|
"""
|
# Various RFC's let you finish a domain name by pointing to an existing
|
# domain name rather than repeating the same suffix. All such pointers are
|
# two bytes long, specify the offset in the byte string, and begin with
|
# |POINTER_PREFIX| to distinguish them from normal characters.
|
POINTER_PREFIX = ord("\xC0")
|
|
@staticmethod
|
def pack(value):
|
domain_list = value
|
byte_string = ""
|
for domain in domain_list:
|
for part in domain.split("."):
|
byte_string += chr(len(part))
|
byte_string += part
|
byte_string += "\x00"
|
return byte_string
|
|
@staticmethod
|
def unpack(byte_string):
|
domain_list = []
|
offset = 0
|
try:
|
while offset < len(byte_string):
|
(new_offset, domain_parts) = DomainListOption._read_domain_name(
|
byte_string,
|
offset)
|
domain_name = ".".join(domain_parts)
|
domain_list.append(domain_name)
|
if new_offset <= offset:
|
raise Exception("Parsing logic error is letting domain "
|
"list parsing go on forever.")
|
offset = new_offset
|
except ValueError:
|
# Badly formatted packets are not necessarily test errors.
|
logging.warning("Found badly formatted DHCP domain search list")
|
return None
|
return domain_list
|
|
@staticmethod
|
def _read_domain_name(byte_string, offset):
|
"""
|
Recursively parse a domain name from a domain name list.
|
"""
|
parts = []
|
while True:
|
if offset >= len(byte_string):
|
raise ValueError("Domain list ended without a NULL byte.")
|
maybe_part_len = ord(byte_string[offset])
|
offset += 1
|
if maybe_part_len == 0:
|
# Domains are terminated with either a 0 or a pointer to a
|
# domain suffix within |byte_string|.
|
return (offset, parts)
|
elif ((maybe_part_len & DomainListOption.POINTER_PREFIX) ==
|
DomainListOption.POINTER_PREFIX):
|
if offset >= len(byte_string):
|
raise ValueError("Missing second byte of domain suffix "
|
"pointer.")
|
maybe_part_len &= ~DomainListOption.POINTER_PREFIX
|
pointer_offset = ((maybe_part_len << 8) +
|
ord(byte_string[offset]))
|
offset += 1
|
(_, more_parts) = DomainListOption._read_domain_name(
|
byte_string,
|
pointer_offset)
|
parts.extend(more_parts)
|
return (offset, parts)
|
else:
|
# That byte was actually the length of the next part, not a
|
# pointer back into the data.
|
part_len = maybe_part_len
|
if offset + part_len >= len(byte_string):
|
raise ValueError("Part of a domain goes beyond data "
|
"length.")
|
parts.append(byte_string[offset : offset + part_len])
|
offset += part_len
|
|
|
"""
|
Represents a required field in a DHCP packet. Similar to Option, we'll
|
subclass Field to reflect that different fields serialize to on the wire formats
|
in different ways.
|
|
|name|
|
A human readable name for this field.
|
|
|offset|
|
The |offset| for a field defines the starting byte of the field in the
|
binary packet string. |offset| is used during parsing, along with
|
|size| to extract the byte string of a field.
|
|
|size|
|
Fields in DHCP packets have a fixed size that must be respected. This
|
size property is used in parsing to indicate that |self._size| number of
|
bytes make up this field.
|
"""
|
Field = collections.namedtuple("Field", ["name", "offset", "size"])
|
|
ByteField = CreatePacketPieceClass(Field, "!B")
|
|
ShortField = CreatePacketPieceClass(Field, "!H")
|
|
IntField = CreatePacketPieceClass(Field, "!I")
|
|
HwAddrField = CreatePacketPieceClass(Field, "!16s")
|
|
ServerNameField = CreatePacketPieceClass(Field, "!64s")
|
|
BootFileField = CreatePacketPieceClass(Field, "!128s")
|
|
class IpAddressField(Field):
|
@staticmethod
|
def pack(value):
|
return socket.inet_aton(value)
|
|
@staticmethod
|
def unpack(byte_string):
|
return socket.inet_ntoa(byte_string)
|
|
|
# This is per RFC 2131. The wording doesn't seem to say that the packets must
|
# be this big, but that has been the historic assumption in implementations.
|
DHCP_MIN_PACKET_SIZE = 300
|
|
IPV4_NULL_ADDRESS = "0.0.0.0"
|
|
# These are required in every DHCP packet. Without these fields, the
|
# packet will not even pass DhcpPacket.is_valid
|
FIELD_OP = ByteField("op", 0, 1)
|
FIELD_HWTYPE = ByteField("htype", 1, 1)
|
FIELD_HWADDR_LEN = ByteField("hlen", 2, 1)
|
FIELD_RELAY_HOPS = ByteField("hops", 3, 1)
|
FIELD_TRANSACTION_ID = IntField("xid", 4, 4)
|
FIELD_TIME_SINCE_START = ShortField("secs", 8, 2)
|
FIELD_FLAGS = ShortField("flags", 10, 2)
|
FIELD_CLIENT_IP = IpAddressField("ciaddr", 12, 4)
|
FIELD_YOUR_IP = IpAddressField("yiaddr", 16, 4)
|
FIELD_SERVER_IP = IpAddressField("siaddr", 20, 4)
|
FIELD_GATEWAY_IP = IpAddressField("giaddr", 24, 4)
|
FIELD_CLIENT_HWADDR = HwAddrField("chaddr", 28, 16)
|
# The following two fields are considered "legacy BOOTP" fields but may
|
# sometimes be used by DHCP clients.
|
FIELD_LEGACY_SERVER_NAME = ServerNameField("servername", 44, 64);
|
FIELD_LEGACY_BOOT_FILE = BootFileField("bootfile", 108, 128);
|
FIELD_MAGIC_COOKIE = IntField("magic_cookie", 236, 4)
|
|
OPTION_TIME_OFFSET = IntOption("time_offset", 2)
|
OPTION_ROUTERS = IpListOption("routers", 3)
|
OPTION_SUBNET_MASK = IpAddressOption("subnet_mask", 1)
|
OPTION_TIME_SERVERS = IpListOption("time_servers", 4)
|
OPTION_NAME_SERVERS = IpListOption("name_servers", 5)
|
OPTION_DNS_SERVERS = IpListOption("dns_servers", 6)
|
OPTION_LOG_SERVERS = IpListOption("log_servers", 7)
|
OPTION_COOKIE_SERVERS = IpListOption("cookie_servers", 8)
|
OPTION_LPR_SERVERS = IpListOption("lpr_servers", 9)
|
OPTION_IMPRESS_SERVERS = IpListOption("impress_servers", 10)
|
OPTION_RESOURCE_LOC_SERVERS = IpListOption("resource_loc_servers", 11)
|
OPTION_HOST_NAME = RawOption("host_name", 12)
|
OPTION_BOOT_FILE_SIZE = ShortOption("boot_file_size", 13)
|
OPTION_MERIT_DUMP_FILE = RawOption("merit_dump_file", 14)
|
OPTION_DOMAIN_NAME = RawOption("domain_name", 15)
|
OPTION_SWAP_SERVER = IpAddressOption("swap_server", 16)
|
OPTION_ROOT_PATH = RawOption("root_path", 17)
|
OPTION_EXTENSIONS = RawOption("extensions", 18)
|
OPTION_INTERFACE_MTU = ShortOption("interface_mtu", 26)
|
OPTION_VENDOR_ENCAPSULATED_OPTIONS = RawOption(
|
"vendor_encapsulated_options", 43)
|
OPTION_REQUESTED_IP = IpAddressOption("requested_ip", 50)
|
OPTION_IP_LEASE_TIME = IntOption("ip_lease_time", 51)
|
OPTION_OPTION_OVERLOAD = ByteOption("option_overload", 52)
|
OPTION_DHCP_MESSAGE_TYPE = ByteOption("dhcp_message_type", 53)
|
OPTION_SERVER_ID = IpAddressOption("server_id", 54)
|
OPTION_PARAMETER_REQUEST_LIST = ByteListOption("parameter_request_list", 55)
|
OPTION_MESSAGE = RawOption("message", 56)
|
OPTION_MAX_DHCP_MESSAGE_SIZE = ShortOption("max_dhcp_message_size", 57)
|
OPTION_RENEWAL_T1_TIME_VALUE = IntOption("renewal_t1_time_value", 58)
|
OPTION_REBINDING_T2_TIME_VALUE = IntOption("rebinding_t2_time_value", 59)
|
OPTION_VENDOR_ID = RawOption("vendor_id", 60)
|
OPTION_CLIENT_ID = RawOption("client_id", 61)
|
OPTION_TFTP_SERVER_NAME = RawOption("tftp_server_name", 66)
|
OPTION_BOOTFILE_NAME = RawOption("bootfile_name", 67)
|
OPTION_FULLY_QUALIFIED_DOMAIN_NAME = RawOption("fqdn", 81)
|
OPTION_DNS_DOMAIN_SEARCH_LIST = DomainListOption("domain_search_list", 119)
|
OPTION_CLASSLESS_STATIC_ROUTES = ClasslessStaticRoutesOption(
|
"classless_static_routes", 121)
|
OPTION_WEB_PROXY_AUTO_DISCOVERY = RawOption("wpad", 252)
|
|
# Unlike every other option, which are tuples like:
|
# <number, length in bytes, data>, the pad and end options are just
|
# single bytes "\x00" and "\xff" (without length or data fields).
|
OPTION_PAD = 0
|
OPTION_END = 255
|
|
DHCP_COMMON_FIELDS = [
|
FIELD_OP,
|
FIELD_HWTYPE,
|
FIELD_HWADDR_LEN,
|
FIELD_RELAY_HOPS,
|
FIELD_TRANSACTION_ID,
|
FIELD_TIME_SINCE_START,
|
FIELD_FLAGS,
|
FIELD_CLIENT_IP,
|
FIELD_YOUR_IP,
|
FIELD_SERVER_IP,
|
FIELD_GATEWAY_IP,
|
FIELD_CLIENT_HWADDR,
|
]
|
|
DHCP_REQUIRED_FIELDS = DHCP_COMMON_FIELDS + [
|
FIELD_MAGIC_COOKIE,
|
]
|
|
DHCP_ALL_FIELDS = DHCP_COMMON_FIELDS + [
|
FIELD_LEGACY_SERVER_NAME,
|
FIELD_LEGACY_BOOT_FILE,
|
FIELD_MAGIC_COOKIE,
|
]
|
|
# The op field in an ipv4 packet is either 1 or 2 depending on
|
# whether the packet is from a server or from a client.
|
FIELD_VALUE_OP_CLIENT_REQUEST = 1
|
FIELD_VALUE_OP_SERVER_RESPONSE = 2
|
# 1 == 10mb ethernet hardware address type (aka MAC).
|
FIELD_VALUE_HWTYPE_10MB_ETH = 1
|
# MAC addresses are still 6 bytes long.
|
FIELD_VALUE_HWADDR_LEN_10MB_ETH = 6
|
FIELD_VALUE_MAGIC_COOKIE = 0x63825363
|
|
OPTIONS_START_OFFSET = 240
|
|
MessageType = collections.namedtuple('MessageType', 'name option_value')
|
# From RFC2132, the valid DHCP message types are:
|
MESSAGE_TYPE_UNKNOWN = MessageType('UNKNOWN', 0)
|
MESSAGE_TYPE_DISCOVERY = MessageType('DISCOVERY', 1)
|
MESSAGE_TYPE_OFFER = MessageType('OFFER', 2)
|
MESSAGE_TYPE_REQUEST = MessageType('REQUEST', 3)
|
MESSAGE_TYPE_DECLINE = MessageType('DECLINE', 4)
|
MESSAGE_TYPE_ACK = MessageType('ACK', 5)
|
MESSAGE_TYPE_NAK = MessageType('NAK', 6)
|
MESSAGE_TYPE_RELEASE = MessageType('RELEASE', 7)
|
MESSAGE_TYPE_INFORM = MessageType('INFORM', 8)
|
MESSAGE_TYPE_BY_NUM = [
|
None,
|
MESSAGE_TYPE_DISCOVERY,
|
MESSAGE_TYPE_OFFER,
|
MESSAGE_TYPE_REQUEST,
|
MESSAGE_TYPE_DECLINE,
|
MESSAGE_TYPE_ACK,
|
MESSAGE_TYPE_NAK,
|
MESSAGE_TYPE_RELEASE,
|
MESSAGE_TYPE_INFORM
|
]
|
|
OPTION_VALUE_PARAMETER_REQUEST_LIST_DEFAULT = [
|
OPTION_REQUESTED_IP.number,
|
OPTION_IP_LEASE_TIME.number,
|
OPTION_SERVER_ID.number,
|
OPTION_SUBNET_MASK.number,
|
OPTION_ROUTERS.number,
|
OPTION_DNS_SERVERS.number,
|
OPTION_HOST_NAME.number,
|
]
|
|
# These are possible options that may not be in every packet.
|
# Frequently, the client can include a bunch of options that indicate
|
# that it would like to receive information about time servers, routers,
|
# lpr servers, and much more, but the DHCP server can usually ignore
|
# those requests.
|
#
|
# Eventually, each option is encoded as:
|
# <option.number, option.size, [array of option.size bytes]>
|
# Unlike fields, which make up a fixed packet format, options can be in
|
# any order, except where they cannot. For instance, option 1 must
|
# follow option 3 if both are supplied. For this reason, potential
|
# options are in this list, and added to the packet in this order every
|
# time.
|
#
|
# size < 0 indicates that this is variable length field of at least
|
# abs(length) bytes in size.
|
DHCP_PACKET_OPTIONS = [
|
OPTION_TIME_OFFSET,
|
OPTION_ROUTERS,
|
OPTION_SUBNET_MASK,
|
OPTION_TIME_SERVERS,
|
OPTION_NAME_SERVERS,
|
OPTION_DNS_SERVERS,
|
OPTION_LOG_SERVERS,
|
OPTION_COOKIE_SERVERS,
|
OPTION_LPR_SERVERS,
|
OPTION_IMPRESS_SERVERS,
|
OPTION_RESOURCE_LOC_SERVERS,
|
OPTION_HOST_NAME,
|
OPTION_BOOT_FILE_SIZE,
|
OPTION_MERIT_DUMP_FILE,
|
OPTION_SWAP_SERVER,
|
OPTION_DOMAIN_NAME,
|
OPTION_ROOT_PATH,
|
OPTION_EXTENSIONS,
|
OPTION_INTERFACE_MTU,
|
OPTION_VENDOR_ENCAPSULATED_OPTIONS,
|
OPTION_REQUESTED_IP,
|
OPTION_IP_LEASE_TIME,
|
OPTION_OPTION_OVERLOAD,
|
OPTION_DHCP_MESSAGE_TYPE,
|
OPTION_SERVER_ID,
|
OPTION_PARAMETER_REQUEST_LIST,
|
OPTION_MESSAGE,
|
OPTION_MAX_DHCP_MESSAGE_SIZE,
|
OPTION_RENEWAL_T1_TIME_VALUE,
|
OPTION_REBINDING_T2_TIME_VALUE,
|
OPTION_VENDOR_ID,
|
OPTION_CLIENT_ID,
|
OPTION_TFTP_SERVER_NAME,
|
OPTION_BOOTFILE_NAME,
|
OPTION_FULLY_QUALIFIED_DOMAIN_NAME,
|
OPTION_DNS_DOMAIN_SEARCH_LIST,
|
OPTION_CLASSLESS_STATIC_ROUTES,
|
OPTION_WEB_PROXY_AUTO_DISCOVERY,
|
]
|
|
def get_dhcp_option_by_number(number):
|
for option in DHCP_PACKET_OPTIONS:
|
if option.number == number:
|
return option
|
return None
|
|
class DhcpPacket(object):
|
@staticmethod
|
def create_discovery_packet(hwmac_addr):
|
"""
|
Create a discovery packet.
|
|
Fill in fields of a DHCP packet as if it were being sent from
|
|hwmac_addr|. Requests subnet masks, broadcast addresses, router
|
addresses, dns addresses, domain search lists, client host name, and NTP
|
server addresses. Note that the offer packet received in response to
|
this packet will probably not contain all of that information.
|
"""
|
# MAC addresses are actually only 6 bytes long, however, for whatever
|
# reason, DHCP allocated 12 bytes to this field. Ease the burden on
|
# developers and hide this detail.
|
while len(hwmac_addr) < 12:
|
hwmac_addr += chr(OPTION_PAD)
|
|
packet = DhcpPacket()
|
packet.set_field(FIELD_OP, FIELD_VALUE_OP_CLIENT_REQUEST)
|
packet.set_field(FIELD_HWTYPE, FIELD_VALUE_HWTYPE_10MB_ETH)
|
packet.set_field(FIELD_HWADDR_LEN, FIELD_VALUE_HWADDR_LEN_10MB_ETH)
|
packet.set_field(FIELD_RELAY_HOPS, 0)
|
packet.set_field(FIELD_TRANSACTION_ID, random.getrandbits(32))
|
packet.set_field(FIELD_TIME_SINCE_START, 0)
|
packet.set_field(FIELD_FLAGS, 0)
|
packet.set_field(FIELD_CLIENT_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_YOUR_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_SERVER_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_GATEWAY_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_CLIENT_HWADDR, hwmac_addr)
|
packet.set_field(FIELD_MAGIC_COOKIE, FIELD_VALUE_MAGIC_COOKIE)
|
packet.set_option(OPTION_DHCP_MESSAGE_TYPE,
|
MESSAGE_TYPE_DISCOVERY.option_value)
|
return packet
|
|
@staticmethod
|
def create_offer_packet(transaction_id,
|
hwmac_addr,
|
offer_ip,
|
server_ip):
|
"""
|
Create an offer packet, given some fields that tie the packet to a
|
particular offer.
|
"""
|
packet = DhcpPacket()
|
packet.set_field(FIELD_OP, FIELD_VALUE_OP_SERVER_RESPONSE)
|
packet.set_field(FIELD_HWTYPE, FIELD_VALUE_HWTYPE_10MB_ETH)
|
packet.set_field(FIELD_HWADDR_LEN, FIELD_VALUE_HWADDR_LEN_10MB_ETH)
|
# This has something to do with relay agents
|
packet.set_field(FIELD_RELAY_HOPS, 0)
|
packet.set_field(FIELD_TRANSACTION_ID, transaction_id)
|
packet.set_field(FIELD_TIME_SINCE_START, 0)
|
packet.set_field(FIELD_FLAGS, 0)
|
packet.set_field(FIELD_CLIENT_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_YOUR_IP, offer_ip)
|
packet.set_field(FIELD_SERVER_IP, server_ip)
|
packet.set_field(FIELD_GATEWAY_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_CLIENT_HWADDR, hwmac_addr)
|
packet.set_field(FIELD_MAGIC_COOKIE, FIELD_VALUE_MAGIC_COOKIE)
|
packet.set_option(OPTION_DHCP_MESSAGE_TYPE,
|
MESSAGE_TYPE_OFFER.option_value)
|
return packet
|
|
@staticmethod
|
def create_request_packet(transaction_id,
|
hwmac_addr):
|
packet = DhcpPacket()
|
packet.set_field(FIELD_OP, FIELD_VALUE_OP_CLIENT_REQUEST)
|
packet.set_field(FIELD_HWTYPE, FIELD_VALUE_HWTYPE_10MB_ETH)
|
packet.set_field(FIELD_HWADDR_LEN, FIELD_VALUE_HWADDR_LEN_10MB_ETH)
|
# This has something to do with relay agents
|
packet.set_field(FIELD_RELAY_HOPS, 0)
|
packet.set_field(FIELD_TRANSACTION_ID, transaction_id)
|
packet.set_field(FIELD_TIME_SINCE_START, 0)
|
packet.set_field(FIELD_FLAGS, 0)
|
packet.set_field(FIELD_CLIENT_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_YOUR_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_SERVER_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_GATEWAY_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_CLIENT_HWADDR, hwmac_addr)
|
packet.set_field(FIELD_MAGIC_COOKIE, FIELD_VALUE_MAGIC_COOKIE)
|
packet.set_option(OPTION_DHCP_MESSAGE_TYPE,
|
MESSAGE_TYPE_REQUEST.option_value)
|
return packet
|
|
@staticmethod
|
def create_acknowledgement_packet(transaction_id,
|
hwmac_addr,
|
granted_ip,
|
server_ip):
|
packet = DhcpPacket()
|
packet.set_field(FIELD_OP, FIELD_VALUE_OP_SERVER_RESPONSE)
|
packet.set_field(FIELD_HWTYPE, FIELD_VALUE_HWTYPE_10MB_ETH)
|
packet.set_field(FIELD_HWADDR_LEN, FIELD_VALUE_HWADDR_LEN_10MB_ETH)
|
# This has something to do with relay agents
|
packet.set_field(FIELD_RELAY_HOPS, 0)
|
packet.set_field(FIELD_TRANSACTION_ID, transaction_id)
|
packet.set_field(FIELD_TIME_SINCE_START, 0)
|
packet.set_field(FIELD_FLAGS, 0)
|
packet.set_field(FIELD_CLIENT_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_YOUR_IP, granted_ip)
|
packet.set_field(FIELD_SERVER_IP, server_ip)
|
packet.set_field(FIELD_GATEWAY_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_CLIENT_HWADDR, hwmac_addr)
|
packet.set_field(FIELD_MAGIC_COOKIE, FIELD_VALUE_MAGIC_COOKIE)
|
packet.set_option(OPTION_DHCP_MESSAGE_TYPE,
|
MESSAGE_TYPE_ACK.option_value)
|
return packet
|
|
@staticmethod
|
def create_nak_packet(transaction_id, hwmac_addr):
|
"""
|
Create a negative acknowledge packet.
|
|
@param transaction_id: The DHCP transaction ID.
|
@param hwmac_addr: The client's MAC address.
|
"""
|
packet = DhcpPacket()
|
packet.set_field(FIELD_OP, FIELD_VALUE_OP_SERVER_RESPONSE)
|
packet.set_field(FIELD_HWTYPE, FIELD_VALUE_HWTYPE_10MB_ETH)
|
packet.set_field(FIELD_HWADDR_LEN, FIELD_VALUE_HWADDR_LEN_10MB_ETH)
|
# This has something to do with relay agents
|
packet.set_field(FIELD_RELAY_HOPS, 0)
|
packet.set_field(FIELD_TRANSACTION_ID, transaction_id)
|
packet.set_field(FIELD_TIME_SINCE_START, 0)
|
packet.set_field(FIELD_FLAGS, 0)
|
packet.set_field(FIELD_CLIENT_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_YOUR_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_SERVER_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_GATEWAY_IP, IPV4_NULL_ADDRESS)
|
packet.set_field(FIELD_CLIENT_HWADDR, hwmac_addr)
|
packet.set_field(FIELD_MAGIC_COOKIE, FIELD_VALUE_MAGIC_COOKIE)
|
packet.set_option(OPTION_DHCP_MESSAGE_TYPE,
|
MESSAGE_TYPE_NAK.option_value)
|
return packet
|
|
def __init__(self, byte_str=None):
|
"""
|
Create a DhcpPacket, filling in fields from a byte string if given.
|
|
Assumes that the packet starts at offset 0 in the binary string. This
|
includes the fields and options. Fields are different from options in
|
that we bother to decode these into more usable data types like
|
integers rather than keeping them as raw byte strings. Fields are also
|
required to exist, unlike options which may not.
|
|
Each option is encoded as a tuple <option number, length, data> where
|
option number is a byte indicating the type of option, length indicates
|
the number of bytes in the data for option, and data is a length array
|
of bytes. The only exceptions to this rule are the 0 and 255 options,
|
which have 0 data length, and no length byte. These tuples are then
|
simply appended to each other. This encoding is the same as the BOOTP
|
vendor extention field encoding.
|
"""
|
super(DhcpPacket, self).__init__()
|
self._options = {}
|
self._fields = {}
|
if byte_str is None:
|
return
|
if len(byte_str) < OPTIONS_START_OFFSET + 1:
|
logging.error("Invalid byte string for packet.")
|
return
|
for field in DHCP_ALL_FIELDS:
|
self._fields[field] = field.unpack(byte_str[field.offset :
|
field.offset +
|
field.size])
|
offset = OPTIONS_START_OFFSET
|
domain_search_list_byte_string = ""
|
while offset < len(byte_str) and ord(byte_str[offset]) != OPTION_END:
|
data_type = ord(byte_str[offset])
|
offset += 1
|
if data_type == OPTION_PAD:
|
continue
|
data_length = ord(byte_str[offset])
|
offset += 1
|
data = byte_str[offset: offset + data_length]
|
offset += data_length
|
option = get_dhcp_option_by_number(data_type)
|
if option is None:
|
logging.warning("Unsupported DHCP option found. "
|
"Option number: %d", data_type)
|
continue
|
if option == OPTION_DNS_DOMAIN_SEARCH_LIST:
|
# In a cruel twist of fate, the server is allowed to give
|
# multiple options with this number. The client is expected to
|
# concatenate the byte strings together and use it as a single
|
# value.
|
domain_search_list_byte_string += data
|
continue
|
option_value = option.unpack(data)
|
if option == OPTION_PARAMETER_REQUEST_LIST:
|
logging.info("Requested options: %s", str(option_value))
|
self._options[option] = option_value
|
if domain_search_list_byte_string:
|
self._options[OPTION_DNS_DOMAIN_SEARCH_LIST] = option_value
|
|
|
@property
|
def client_hw_address(self):
|
return self._fields.get(FIELD_CLIENT_HWADDR)
|
|
@property
|
def is_valid(self):
|
"""
|
Checks that we have (at a minimum) values for all the required fields,
|
and that the magic cookie is set correctly.
|
"""
|
for field in DHCP_REQUIRED_FIELDS:
|
if self._fields.get(field) is None:
|
logging.warning("Missing field %s in packet.", field)
|
return False
|
if self._fields[FIELD_MAGIC_COOKIE] != FIELD_VALUE_MAGIC_COOKIE:
|
return False
|
return True
|
|
@property
|
def message_type(self):
|
"""
|
Gets the value of the DHCP Message Type option in this packet.
|
|
If the option is not present, or the value of the option is not
|
recognized, returns MESSAGE_TYPE_UNKNOWN.
|
|
@returns The MessageType for this packet, or MESSAGE_TYPE_UNKNOWN.
|
"""
|
if (self._options.has_key(OPTION_DHCP_MESSAGE_TYPE) and
|
self._options[OPTION_DHCP_MESSAGE_TYPE] > 0 and
|
self._options[OPTION_DHCP_MESSAGE_TYPE] < len(MESSAGE_TYPE_BY_NUM)):
|
return MESSAGE_TYPE_BY_NUM[self._options[OPTION_DHCP_MESSAGE_TYPE]]
|
else:
|
return MESSAGE_TYPE_UNKNOWN
|
|
@property
|
def transaction_id(self):
|
return self._fields.get(FIELD_TRANSACTION_ID)
|
|
def get_field(self, field):
|
return self._fields.get(field)
|
|
def get_option(self, option):
|
return self._options.get(option)
|
|
def set_field(self, field, field_value):
|
self._fields[field] = field_value
|
|
def set_option(self, option, option_value):
|
self._options[option] = option_value
|
|
def to_binary_string(self):
|
if not self.is_valid:
|
return None
|
# A list of byte strings to be joined into a single string at the end.
|
data = []
|
offset = 0
|
for field in DHCP_ALL_FIELDS:
|
if field not in self._fields:
|
continue
|
field_data = field.pack(self._fields[field])
|
while offset < field.offset:
|
# This should only happen when we're padding the fields because
|
# we're not filling in legacy BOOTP stuff.
|
data.append("\x00")
|
offset += 1
|
data.append(field_data)
|
offset += field.size
|
# Last field processed is the magic cookie, so we're ready for options.
|
# Have to process options
|
for option in DHCP_PACKET_OPTIONS:
|
option_value = self._options.get(option)
|
if option_value is None:
|
continue
|
serialized_value = option.pack(option_value)
|
data.append(struct.pack("BB",
|
option.number,
|
len(serialized_value)))
|
offset += 2
|
data.append(serialized_value)
|
offset += len(serialized_value)
|
data.append(chr(OPTION_END))
|
offset += 1
|
while offset < DHCP_MIN_PACKET_SIZE:
|
data.append(chr(OPTION_PAD))
|
offset += 1
|
return "".join(data)
|
|
def __str__(self):
|
options = [k.name + "=" + str(v) for k, v in self._options.items()]
|
fields = [k.name + "=" + str(v) for k, v in self._fields.items()]
|
return "<DhcpPacket fields=%s, options=%s>" % (fields, options)
|
