Added DHCPRELEASE, DHCPINFORM and option 82 suboptions

README.md

@@ -1,6 +1,6 @@
 # DHCP Python
 
-Version 0.1.4
+Version 0.1.4b
 
 A Python implementation of a DHCP client and the tools to manipulate DHCP packets. Includes:
 

dhcppython/client.py

@@ -10,7 +10,7 @@
 from .exceptions import DHCPClientError
 
 
-COL_LEN = 80
+COL_LEN = 80-3
 
 Lease = collections.namedtuple(
  "Lease", ["discover", "offer", "request", "ack", "time", "server"]
@@ -150,6 +150,101 @@ def receive_ack(self, tx_id: int, verbosity: int) -> Optional[packet.DHCPPacket]
  print("Did not receive ack packet")
  return ack
 
+ def send_release(
+ self, server: str, release_packet: packet.DHCPPacket, verbosity: int
+ ):
+ self.send(server, self.send_to_port, release_packet.asbytes, verbosity)
+
+ def send_inform(
+ self, server: str, inform_packet: packet.DHCPPacket, verbosity: int
+ ):
+ self.send(server, self.send_to_port, inform_packet.asbytes, verbosity)
+
+ def send_decline(
+ self, server: str, decline_packet: packet.DHCPPacket, verbosity: int
+ ):
+ self.send(server, self.send_to_port, decline_packet.asbytes, verbosity)
+
+ def put_release(
+ self,
+ mac_addr: Optional[str] = None,
+ server: str = "255.255.255.255",
+ relay: Optional[str] = None,
+ client: Optional[str] = None,
+ broadcast: bool = True, 
+ options_list: Optional[options.OptionList] = None,
+ verbose: int = 0, 
+ ):
+ logging.debug("Synthetizing release packet")
+ release = packet.DHCPPacket.Release(
+ mac_addr,
+ use_broadcast=broadcast,
+ option_list=options_list,
+ relay=relay,
+ client=client,
+ )
+ tx_id = release.xid
+ if verbose > 1:
+ print("RELEASE Packet")
+ print(format_dhcp_packet(release))
+ logging.debug(f"Constructed release packet: {release}")
+ logging.debug(f"Sending release packet to {server} with {tx_id=}")
+ self.send_release(server=server, release_packet=release, verbosity=verbose)
+
+ def put_inform(
+ self,
+ mac_addr: Optional[str] = None,
+ broadcast: bool = True,
+ relay: Optional[str] = None,
+ client: Optional[str] = None,
+ server: str = "255.255.255.255",
+ ip_protocol: int = 4,
+ options_list: Optional[options.OptionList] = None,
+ verbose: int = 0,
+ ):
+ logging.debug("Synthetizing inform packet")
+ inform = packet.DHCPPacket.Inform(
+ mac_addr,
+ use_broadcast=broadcast,
+ option_list=options_list,
+ client_ip=client,
+ relay=relay,
+ )
+ tx_id = inform.xid
+ if verbose > 1:
+ print("INFORM Packet")
+ print(format_dhcp_packet(inform))
+ logging.debug(f"Constructed inform packet: {inform}")
+ logging.debug(f"Sending inform packet to {server} with {tx_id=}")
+ self.send_inform(server=server, inform_packet=inform, verbosity=verbose)
+
+ def put_decline(
+ self,
+ mac_addr: Optional[str] = None,
+ server: str = "255.255.255.255",
+ relay: Optional[str] = None,
+ client: Optional[str] = None,
+ broadcast: bool = True, 
+ options_list: Optional[options.OptionList] = None,
+ verbose: int = 0, 
+ ):
+ logging.debug("Synthetizing decline packet")
+ decline = packet.DHCPPacket.Decline(
+ mac_addr,
+ use_broadcast=broadcast,
+ option_list=options_list,
+ relay=relay,
+ client=client,
+ )
+ tx_id = decline.xid
+ if verbose > 1:
+ print("INFORM Packet")
+ print(format_dhcp_packet(decline))
+ logging.debug(f"Constructed decline packet: {decline}")
+ logging.debug(f"Sending decline packet to {server} with {tx_id=}")
+
+ self.send_decline(server=server, decline_packet=decline, verbosity=verbose)
+
  def get_lease(
  self,
  mac_addr: Optional[str] = None,

dhcppython/options.py

@@ -1784,12 +1784,48 @@ class RelayAgentInformation(StrOption):
  """
  Option 82
 
- Relay Agent Information
+ Relay Agent Information is comprised of suboptions with basically same 
+ structure as a string option.
+
  """
 
  code = 82
  key = "relay_agent_info"
 
+ code82 = {
+ 1: "circuit_id",
+ 2: "remote_id",
+ }
+ key82 = {
+ "circuit_id": 1,
+ "remote_id": 2,
+ }
+
+ @property
+ def value(self) -> Dict[str, Dict[str, str]]:
+
+ if self._value is None:
+ self._value = {}
+ data = self.data
+ while data:
+ code, length = struct.unpack(">BB",data[:2])
+ data0 = data[2:2+length]
+ self._value[self.code82[code]] = data0.decode()
+ data = data[2+length:]
+ return self._value
+
+ @classmethod
+ def from_value(cls, value):
+ data = b""
+ for raisub in ["circuit_id","remote_id"]:
+ if raisub in value[cls.key]: 
+ subval = value[cls.key][raisub]
+ data0 = subval.encode()
+ data += struct.pack(">Bb", cls.key82[raisub], len(data0)) + \
+ struct.pack(">" + "B" * len(data0), *data0)
+ return cls(cls.code, len(data), data)
+
+
 
 class UnknownOption(BinOption):
  """

dhcppython/packet.py

@@ -430,3 +430,148 @@ def Ack(
  fname,
  option_list,
  )
+
+ @classmethod
+ def Release(
+ cls,
+ mac_addr: str,
+ seconds: int = 0,
+ tx_id: int = 0,
+ use_broadcast: bool = True,
+ relay: Optional[str] = None,
+ client: Optional[str] = None,
+ sname: bytes = b"",
+ fname: bytes = b"",
+ option_list: Optional[options.OptionList] = None,
+ ):
+ """
+ Convenient constructor for a DHCP release packet.
+ """
+ if len(mac_addr.split(":")) != 6 or len(mac_addr) != 17:
+ raise DHCPValueError(
+ "MAC address must consist of 6 octets delimited by ':'"
+ )
+ option_list = option_list if option_list else options.OptionList()
+ option_list.insert(0, options.options.short_value_to_object(53, "DHCPRELEASE"))
+ relay_ip = ipaddress.IPv4Address(relay or 0)
+ client_ip = ipaddress.IPv4Address(client or 0)
+ return cls(
+ "BOOTREQUEST",
+ cls.htype_map[1], # 10 mb ethernet
+ 6, # 6 byte hardware addr
+ 0, # clients should set this to 0
+ tx_id or random.getrandbits(32),
+ seconds,
+ 0b1000_0000_0000_0000 if use_broadcast else 0,
+ # ciaddr - client address
+ #ipaddress.IPv4Address(0),
+ client_ip,
+ # yiaddr - "your address", address being proposed by server
+ #ipaddress.IPv4Address(0),
+ client_ip,
+ # siaddr - servr address
+ ipaddress.IPv4Address(0),
+ # giaddr - gateway address = relay address
+ relay_ip,
+ mac_addr,
+ sname,
+ fname,
+ option_list,
+ )
+
+ @classmethod
+ def Inform(
+ cls,
+ mac_addr: str,
+ seconds: int = 0,
+ tx_id: int = 0,
+ use_broadcast: bool = True,
+ relay: Optional[str] = None,
+ client_ip: Optional[str] = None,
+ sname: bytes = b"",
+ fname: bytes = b"",
+ option_list: Optional[options.OptionList] = None,
+ ):
+ """
+ Convenient constructor for a DHCP inform packet.
+ """
+ if len(mac_addr.split(":")) != 6 or len(mac_addr) != 17:
+ raise DHCPValueError(
+ "MAC address must consist of 6 octets delimited by ':'"
+ )
+ option_list = option_list if option_list else options.OptionList()
+ option_list.insert(0, options.options.short_value_to_object(53, "DHCPINFORM"))
+ relay_ip = ipaddress.IPv4Address(relay or 0)
+ client_ip = ipaddress.IPv4Address(client_ip or 0)
+ return cls(
+ "BOOTREQUEST",
+ cls.htype_map[1], # 10 mb ethernet
+ 6, # 6 byte hardware addr
+ 0, # clients should set this to 0
+ tx_id or random.getrandbits(32),
+ seconds,
+ 0b1000_0000_0000_0000 if use_broadcast else 0,
+ # ciaddr - client address
+ #ipaddress.IPv4Address(0),
+ client_ip,
+ # yiaddr - "your address", address being proposed by server
+ #ipaddress.IPv4Address(0),
+ client_ip,
+ # siaddr - servr address
+ ipaddress.IPv4Address(0),
+ # giaddr - gateway address = relay address
+ relay_ip,
+ mac_addr,
+ sname,
+ fname,
+ option_list,
+ )
+ @classmethod
+ def Decline(
+ cls,
+ mac_addr: str,
+ seconds: int = 0,
+ tx_id: int = 0,
+ use_broadcast: bool = True,
+ relay: Optional[str] = None,
+ client: Optional[str] = None,
+ sname: bytes = b"",
+ fname: bytes = b"",
+ option_list: Optional[options.OptionList] = None,
+ ):
+ """
+ Convenient constructor for a DHCP decline packet.
+ """
+ if len(mac_addr.split(":")) != 6 or len(mac_addr) != 17:
+ raise DHCPValueError(
+ "MAC address must consist of 6 octets delimited by ':'"
+ )
+ option_list = option_list if option_list else options.OptionList()
+ option_list.insert(0, options.options.short_value_to_object(53, "DHCPDECLINE"))
+ relay_ip = ipaddress.IPv4Address(relay or 0)
+ client_ip = ipaddress.IPv4Address(client or 0)
+ return cls(
+ "BOOTREQUEST",
+ cls.htype_map[1], # 10 mb ethernet
+ 6, # 6 byte hardware addr
+ 0, # clients should set this to 0
+ tx_id or random.getrandbits(32),
+ seconds,
+ 0b1000_0000_0000_0000 if use_broadcast else 0,
+ # ciaddr - client address
+ #ipaddress.IPv4Address(0),
+ client_ip,
+ # yiaddr - "your address", address being proposed by server
+ #ipaddress.IPv4Address(0),
+ client_ip,
+ # siaddr - servr address
+ ipaddress.IPv4Address(0),
+ # giaddr - gateway address = relay address
+ relay_ip,
+ mac_addr,
+ sname,
+ fname,
+ option_list,
+ )
+
+

dhcppython/utils.py

@@ -48,6 +48,28 @@ def random_mac(num_bytes: int = 6, delimiter: str = ":") -> str:
  ["".join(random.choices(VALID_HEX, k=2)) for i in range(num_bytes)]
  )
 
+def random_mac_prefix(num_bytes: int = 6, delimiter: str = ":", prefix: str = "") -> str:
+ """
+ Generates an 6 byte long MAC address with predefined prefix
+
+ >>> random_mac_prefix(prefix="00:0A:A0")
+ '00:0A:A0:85:A4:EF'
+ """
+
+ if not prefix:
+ return random_mac(num_bytes=num_bytes, delimiter=delimiter)
+ else:
+ prefix = prefix.upper()
+ prefix = prefix.replace(":","").replace("-","").replace(".","")
+ if not set(prefix).issubset(set(VALID_HEX)):
+ raise TypeError(f"Wrong characters in MAC prefix: {prefix}")
+ if len(prefix)%2 != 0:
+ raise ValueError(f"Even number of HEX characters expected: {prefix}")
+ prefix_list = [ prefix[i]+prefix[i+1] for i in range(0, len(prefix), 2)]
+
+ return delimiter.join(prefix_list+
+ ["".join(random.choices(VALID_HEX, k=2)) for i in range(num_bytes-len(prefix)//2)]
+ )
 
 def is_mac_addr(mac_addr: str) -> bool:
  """

setup.py

@@ -15,7 +15,7 @@
 EMAIL = ''
 AUTHOR = 'Victor Frazao'
 REQUIRES_PYTHON = '>=3.8.0'
-VERSION = '0.1.4'
+VERSION = '0.1.4b'
 
 # What packages are required for this module to be executed?
 REQUIRED = []