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 = []