API Reference

This section provides references for the primary classes in the pydhcp package.

DhcpMessage

pydhcp.packet.message.DhcpMessage dataclass

Source code in src/pydhcp/packet/message.py

@_data.dataclass
class DhcpMessage:
    MIN_LEGAL_SIZE = _const.DHCP_MIN_LEGAL_PACKET_SIZE - _const.UDP_MIN_PACKET_SIZE
    MAGIC_COOKIE: _ty.ClassVar[bytes] = 0x63825363.to_bytes(4, "big")
    """The first four octets of the 'options' field of the DHCP message decimal values: 99, 130, 83 and 99"""

    op: _enum.OpCode  # One byte
    """Message op code / message type"""
    htype: _enum.HardwareAddressType
    """Hardware address type, see ARP section in "Assigned Numbers" RFC"""
    hlen: int  # One byte
    """Hardware address length"""
    hops: int  # One byte
    """Client sets to zero, optionally used by relay agents when booting via a relay agent."""
    xid: int  # 4 bytes
    """Transaction ID, a random number chosen by the
    client, used by the client and server to associate
    messages and responses between a client and a server."""
    secs: _dt.timedelta  # 2 bytes
    """Filled in by client, seconds elapsed since client
    began address acquisition or renewal process."""
    flags: _enum.Flags  # 2 bytes
    """Only use for the BROADCAST flag in clients"""
    ciaddr: _net.IPv4
    """Client IP address; only filled in if client is in
    BOUND, RENEW or REBINDING state and can respond
    to ARP requests."""
    yiaddr: _net.IPv4
    """'your' (client) IP address."""
    siaddr: _net.IPv4
    """IP address of next server to use in bootstrap;
    returned in DHCPOFFER, DHCPACK by server."""
    giaddr: _net.IPv4
    """Relay agent IP address, used in booting via a
    relay agent."""
    chaddr: bytes  # 16 bytes
    """Client hardware address."""
    sname: str  # 64 bytes
    """Optional server host name, null terminated string."""
    file: str  # 128 bytes
    """Boot file name, null terminated string; "generic"
    name or null in DHCPDISCOVER, fully qualified
    directory-path name in DHCPOFFER."""
    options: DhcpOptions
    """Optional parameters field."""

    def to_mapping(self) -> dict[str, _ty.Any]:
        options: dict[str, _ty.Any] = {}
        for code, value in self.options._options.items():
            try:
                code_obj = self.options._codemap.from_code(code)
                key = code_obj.label()
                option_type = code_obj.get_type()
                decoded = option_type._dhcp_decode(value)
                option_value = decoded.__json__() if isinstance(decoded, _type.DhcpOptionType) else decoded
            except Exception:
                key = str(code)
                option_value = _type.Bytes(value).__json__()
            options[key] = _enum_name(option_value)

        return {
            "op": self.op.name,
            "htype": self.htype.name,
            "hlen": self.hlen,
            "hops": self.hops,
            "xid": self.xid,
            "secs": int(self.secs.total_seconds()),
            "flags": self.flags.name,
            "ciaddr": str(self.ciaddr),
            "yiaddr": str(self.yiaddr),
            "siaddr": str(self.siaddr),
            "giaddr": str(self.giaddr),
            "chaddr": self.chaddr.hex(":").upper(),
            "sname": self.sname,
            "file": self.file,
            "options": options,
        }

    @classmethod
    def from_mapping(cls, data: _ty.Mapping[str, _ty.Any]) -> "DhcpMessage":
        options = DhcpOptions()
        raw_options = data.get("options", {})
        if not isinstance(raw_options, _ty.Mapping):
            raise TypeError("options must be a mapping")

        for raw_code, raw_value in raw_options.items():
            code = _coerce_option_code(raw_code, options._codemap)
            try:
                code_obj = options._codemap.from_code(code)
                option_type = code_obj.get_type()
                value = _coerce_option_value(option_type, raw_value)
                options[code] = value
            except Exception:
                if isinstance(raw_value, str):
                    raw_bytes = bytearray.fromhex(_strip_hex_text(raw_value))
                elif isinstance(raw_value, (bytes, bytearray, memoryview)):
                    raw_bytes = bytearray(raw_value)
                else:
                    raise TypeError(f"Unsupported value for unknown option {raw_code!r}") from None
                options[code] = raw_bytes

        return cls(
            op=_coerce_enum_value(_enum.OpCode, data["op"]),
            htype=_coerce_enum_value(_enum.HardwareAddressType, data["htype"]),
            hlen=_coerce_int(data["hlen"]),
            hops=_coerce_int(data["hops"]),
            xid=_coerce_int(data["xid"]),
            secs=_dt.timedelta(seconds=_coerce_int(data["secs"])),
            flags=_coerce_enum_value(_enum.Flags, data["flags"]),
            ciaddr=_net.IPv4(data["ciaddr"]),
            yiaddr=_net.IPv4(data["yiaddr"]),
            siaddr=_net.IPv4(data["siaddr"]),
            giaddr=_net.IPv4(data["giaddr"]),
            chaddr=_coerce_chaddr(data["chaddr"]),
            sname=str(data["sname"]),
            file=str(data["file"]),
            options=options,
        )

    @classmethod
    def decode(cls, data: _ty.Union[bytes, bytearray, memoryview]) -> "DhcpMessage":
        if not isinstance(data, memoryview):
            data = memoryview(data)
        if len(data) < _FIXED_HEADER_SIZE:
            raise ValueError(
                f"Packet is too short for DHCP fixed header: got {len(data)} bytes, need at least {_FIXED_HEADER_SIZE}"
            )
        if len(data) < _MAGIC_COOKIE_END:
            raise ValueError(
                f"Packet is too short for DHCP magic cookie at offset 236: got {len(data)} bytes, need at least {_MAGIC_COOKIE_END}"
            )
        (
            op,
            htype,
            hlen,
            hops,
            xid,
            secs,
            flags,
            ciaddr,
            yiaddr,
            siaddr,
            giaddr,
        ) = _HEADER_STRUCT.unpack_from(data, 0)
        if hlen > 16:
            raise ValueError(f"Hardware address length {hlen} exceeds maximum of 16")
        op = _enum.OpCode(op)
        try:
            htype = _enum.HardwareAddressType(htype)
        except ValueError:
            LOGGER.warning(f"Unknown hardware type {htype}, using ETHERNET")
            htype = _enum.HardwareAddressType.ETHERNET
        secs = _dt.timedelta(seconds=secs)
        flags = _enum.Flags(flags & _enum.Flags.BROADCAST.value)
        ciaddr = _net.IPv4(ciaddr)
        yiaddr = _net.IPv4(yiaddr)
        siaddr = _net.IPv4(siaddr)
        giaddr = _net.IPv4(giaddr)
        chaddr = data[28 : 28 + hlen].tobytes()
        sname_data = data[44:108]
        file_data = data[108:236]

        if cls.MAGIC_COOKIE != data[236:240]:
            raise ValueError(
                f"Invalid magic cookie at offset 236: expected {cls.MAGIC_COOKIE.hex()}, got {data[236:240].hex()}"
            )

        options = DhcpOptions()
        remaining_opts = options.decode(data[240:], base_offset=240)
        if remaining_opts and remaining_opts[0] != 255:
            raise ValueError(f"Bad options terminator: expected 255 (END), got {remaining_opts[0]}")

        overload = options.get(
            DhcpOptionCode.OPTION_OVERLOAD,
            default=_type.OptionOverload.NONE,
            decode=_type.OptionOverload,
        )

        # rfc3396 order
        if overload is not None and bool(overload.value & _type.OptionOverload.FILE.value):
            options.decode(file_data, base_offset=108)
            file_raw: _ty.Optional[memoryview] = None
        else:
            file_raw = file_data

        if overload is not None and bool(overload.value & _type.OptionOverload.SNAME.value):
            options.decode(sname_data, base_offset=44)
            sname_raw: _ty.Optional[memoryview] = None
        else:
            sname_raw = sname_data

        sname_str: str = ""
        if sname_raw is not None:
            sname_str = sname_raw.tobytes().split(_NULL, 1)[0].decode()
        else:
            tftp_val = options.get(
                DhcpOptionCode.TFTP_SERVER, default="", decode=_type.String
            )
            if tftp_val is not None:
                sname_str = str(tftp_val)

        file_str: str = ""
        if file_raw is not None:
            file_str = file_raw.tobytes().split(_NULL, 1)[0].decode()
        else:
            bootfile_val = options.get(
                DhcpOptionCode.BOOTFILE_NAME, default="", decode=_type.String
            )
            if bootfile_val is not None:
                file_str = str(bootfile_val)

        # opts -> file -> sname

        return DhcpMessage(
            op,
            htype,
            hlen,
            hops,
            xid,
            secs,
            flags,
            ciaddr,
            yiaddr,
            siaddr,
            giaddr,
            chaddr,
            sname_str,
            file_str,
            options,
        )

    def encode(self, max_packetsize: int = _const.DHCP_MIN_LEGAL_PACKET_SIZE) -> bytearray:
        max_packetsize = int(max_packetsize or _const.DHCP_MIN_LEGAL_PACKET_SIZE)
        max_options_field_size = max_packetsize - 264 - len(self.MAGIC_COOKIE)
        if max_options_field_size < 0:
            raise ValueError(f"{max_packetsize} is too small for a DHCP packet")

        options = DhcpOptions(self.options._codemap)
        options._options = _ty.OrderedDict(
            (code, bytearray(value)) for code, value in self.options._options.items()
        )
        sname_bytes: _ty.Union[bytes, bytearray] = self.sname.encode()
        file_bytes: _ty.Union[bytes, bytearray] = self.file.encode()
        options_field: _ty.Union[bytes, bytearray] = options.encode()
        if len(options_field) > max_options_field_size + 128 + 64:
            raise OverflowError("DHCP options exceed maximum packet size")
        elif len(options_field) > max_options_field_size + 128:
            if self.file and DhcpOptionCode.BOOTFILE_NAME not in options:
                options[DhcpOptionCode.BOOTFILE_NAME] = self.file
                options._options.move_to_end(
                    int(DhcpOptionCode.BOOTFILE_NAME), False
                )
            if self.sname and DhcpOptionCode.TFTP_SERVER not in options:
                options[DhcpOptionCode.TFTP_SERVER] = self.sname
                options._options.move_to_end(
                    int(DhcpOptionCode.TFTP_SERVER), False
                )
            overload = _type.OptionOverload.BOTH
        elif len(options_field) > max_options_field_size + 64:
            if self.file and DhcpOptionCode.BOOTFILE_NAME not in options:
                options[DhcpOptionCode.BOOTFILE_NAME] = self.file
                options._options.move_to_end(
                    int(DhcpOptionCode.BOOTFILE_NAME), False
                )
            overload = _type.OptionOverload.FILE
        elif len(options_field) > max_options_field_size:
            if self.sname and DhcpOptionCode.TFTP_SERVER not in options:
                options[DhcpOptionCode.TFTP_SERVER] = self.sname
                options._options.move_to_end(
                    int(DhcpOptionCode.TFTP_SERVER), False
                )
            overload = _type.OptionOverload.SNAME
        else:
            overload = _type.OptionOverload.NONE

        try:
            options._options.move_to_end(
                int(DhcpOptionCode.DHCP_MESSAGE_TYPE), False
            )
        except KeyError:
            pass

        if overload is not _type.OptionOverload.NONE:
            options._options[
                int(DhcpOptionCode.OPTION_OVERLOAD)
            ] = bytearray([overload.value])
            options._options.move_to_end(
                int(DhcpOptionCode.OPTION_OVERLOAD), False
            )
            options_field, leftover = options.partial_encode(max_options_field_size)

            if bool(overload.value & _type.OptionOverload.FILE.value) and leftover is not None:
                file_bytes, leftover = leftover.partial_encode(128)
            if bool(overload.value & _type.OptionOverload.SNAME.value) and leftover is not None:
                sname_bytes, leftover = leftover.partial_encode(64)

        data = bytearray(28)
        _HEADER_STRUCT.pack_into(
            data,
            0,
            self.op.value,
            int(self.htype),
            self.hlen,
            self.hops,
            self.xid,
            min(0xFFFF, max(0, int(self.secs.total_seconds()))),
            self.flags.value,
            int(self.ciaddr),
            int(self.yiaddr),
            int(self.siaddr),
            int(self.giaddr),
        )
        data.extend(self.chaddr.ljust(16, b"\x00")[:16])
        data.extend(sname_bytes.ljust(64, b"\x00")[:64])
        data.extend(file_bytes.ljust(128, b"\x00")[:128])
        data.extend(self.MAGIC_COOKIE)
        data.extend(options_field)
        return data

    def client_id(self, func: _ty.Optional[_ty.Callable[["DhcpMessage"], bytearray]] = None) -> str:
        cid = self.options.get(DhcpOptionCode.CLIENT_IDENTIFIER, decode=False)
        if not cid:
            if func:
                cid = func(self)
            if not cid:
                cid = bytearray([self.htype.value])
                cid.extend(self.chaddr)
        return cid.hex(":").upper()

    def dumps(self, codemap: _ty.Optional[type[BaseDhcpOptionCode]] = None) -> str:
        lines = []
        for name, value in [
            ("OP", self.op.name),
            ("Time Since Boot", str(self.secs)),
            ("Hops", str(self.hops)),
            ("Transaction ID", str(self.xid)),
            ("Flags", self.flags.name),
            ("Client Current Address", str(self.ciaddr)),
            ("Allocated Address", str(self.yiaddr)),
            ("Gateway Address", str(self.giaddr)),
            ("Hardware Address", f"{self.htype.name}({self.htype.dumps(self.chaddr)})"),
            ("Server Address", str(self.siaddr)),
            ("Next Server", str(self.siaddr)),
            ("Server Host Name", self.sname),
            ("Bootfile", self.file),
        ]:
            lines.append(f"{name: <40}: {value}")
        lines.append(f"OPTIONS:")
        for code, opt_val in self.options.items(decoded=codemap or True):
            if isinstance(opt_val, list):
                decoded_str = "\n".join([repr(i) for i in opt_val])
            else:
                decoded_str = repr(opt_val)
            decoded_lines = decoded_str.splitlines()
            SPACE = " " * 42
            if decoded_lines:
                first = _tw.fill(
                    decoded_lines[0], width=100, initial_indent="", subsequent_indent=SPACE
                )
            else:
                first = ""
            lines.append(f"  {repr(code): <38}: {first}")
            for line in decoded_lines[1:]:
                lines.append(
                    _tw.fill(
                        line, width=100, initial_indent=SPACE, subsequent_indent=SPACE
                    )
                )

        return "\n".join(lines)

    def log_str(self, src: _ty.Any, dst: _ty.Any) -> str:
        return (
            f"{self.op.name} XID={self.xid:08X} Src: {src} Dst: {dst}\n"
            f"{self.dumps()}"
        )

    def __contains__(self, __key: object) -> bool:
        return self.options.__contains__(__key)

    def log(self, src: _ty.Any, dst: _ty.Any, level: int) -> None:
        header = f"{'#' * 10} {self.op.name} XID={self.xid:08X} Src: {src} Dst: {dst} {'#' * 10}"
        LOGGER.log(level, f"\n{header}\n{self.dumps()}\n{'#' * len(header)}")

MAGIC_COOKIE = 1669485411.to_bytes(4, 'big') class-attribute

The first four octets of the 'options' field of the DHCP message decimal values: 99, 130, 83 and 99

chaddr instance-attribute

Client hardware address.

ciaddr instance-attribute

Client IP address; only filled in if client is in BOUND, RENEW or REBINDING state and can respond to ARP requests.

file instance-attribute

Boot file name, null terminated string; "generic" name or null in DHCPDISCOVER, fully qualified directory-path name in DHCPOFFER.

flags instance-attribute

Only use for the BROADCAST flag in clients

giaddr instance-attribute

Relay agent IP address, used in booting via a relay agent.

hlen instance-attribute

Hardware address length

hops instance-attribute

Client sets to zero, optionally used by relay agents when booting via a relay agent.

htype instance-attribute

Hardware address type, see ARP section in "Assigned Numbers" RFC

op instance-attribute

Message op code / message type

options instance-attribute

Optional parameters field.

secs instance-attribute

Filled in by client, seconds elapsed since client began address acquisition or renewal process.

siaddr instance-attribute

IP address of next server to use in bootstrap; returned in DHCPOFFER, DHCPACK by server.

sname instance-attribute

Optional server host name, null terminated string.

xid instance-attribute

Transaction ID, a random number chosen by the client, used by the client and server to associate messages and responses between a client and a server.

yiaddr instance-attribute

'your' (client) IP address.

DhcpOptions

pydhcp.options.DhcpOptions

Bases: MutableMapping[int, bytearray]

Source code in src/pydhcp/options/__init__.py

class DhcpOptions(_ty.MutableMapping[int, bytearray]):
    def __init__(self, codemap: _ty.Optional[type[BaseDhcpOptionCode]] = None) -> None:
        if codemap is None:
            codemap = DhcpOptionCode
        self._codemap = codemap
        if codemap is DhcpOptionCode:
            DhcpOptionCode.ensure_registered()
        self._options: _ty.OrderedDict[int, bytearray] = _ty.OrderedDict()

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}({list(self._options.keys())})"

    def decode(self, options: memoryview, base_offset: int = 0) -> memoryview:
        offset = base_offset
        while options:
            code = options[0]
            if code == 0:
                options = options[1:]
                offset += 1
                continue

            if code == 255:
                break

            if len(options) < 2:
                LOGGER.warning(f"Option {code} at offset {offset} is truncated (cannot read length)")
                options = options[len(options):]
                break

            length = options[1]
            remaining = len(options) - 2
            if length > remaining:
                LOGGER.warning(
                    f"Option {code} at offset {offset} claims {length} bytes but only {remaining} available"
                )
                data = options[2:]
                self._options.setdefault(code, bytearray()).extend(data)
                options = options[len(options):]
                continue

            next_idx = 2 + length
            data = options[2:next_idx]
            options = options[next_idx:]
            offset += next_idx
            self._options.setdefault(code, bytearray()).extend(data)
        return options

    def partial_encode(self, maxsize: _ty.Optional[float], word_size: int = 1) -> tuple[bytearray, _ty.Optional["DhcpOptions"]]:
        if maxsize is None:
            maxsize = _inf

        if word_size <= 0:
            raise ValueError(f"Invalid Options Word Size")

        endbytes = b"\xff" + b"\x00" * (word_size - 1)

        if maxsize < max(word_size * 2, 4):
            raise ValueError(f"Invalid Options Max Size")

        tofill = maxsize - word_size
        options = bytearray()
        _extraoptions: _ty.OrderedDict[int, bytearray] = _ty.OrderedDict()

        for code, option in self._options.items():
            if not tofill or tofill < 3:
                _extraoptions[code] = option
                continue

            options.append(int(code))
            opt_view = memoryview(option)
            tofill -= 1

            while opt_view and tofill >= word_size:
                slice_data = opt_view[: int(min(255, tofill))]
                _len = len(slice_data)
                options.append(_len)
                options.extend(slice_data)
                options.extend(b"\x00" * (word_size - _len))
                opt_view = opt_view[_len:]
                tofill -= _len
            if opt_view:
                _extraoptions[code] = bytearray(opt_view)

        options.extend(endbytes)
        if _extraoptions:
            leftover = DhcpOptions(self._codemap)
            leftover._options = _extraoptions
        else:
            leftover = None
        return options, leftover

    def encode(self, word_size: int = 1) -> bytearray:
        encoded, _ = self.partial_encode(None, word_size)
        return encoded

    def __getitem__(self, _key: int) -> bytearray:
        return self._options[_key]

    @_ty.overload  # type: ignore[override]
    def get(self, __key: int, default: _ty.Any = None, *, decode: type[T]) -> T | None:
        ...

    @_ty.overload
    def get(
        self, __key: int, default: _ty.Any = None, *, decode: _ty.Callable[[bytearray], _R]
    ) -> _R | None:
        ...

    @_ty.overload
    def get(
        self, __key: int, default: _ty.Any = None, *, decode: _ty.Literal[True]
    ) -> DhcpOptionType | None:
        ...

    @_ty.overload
    def get(
        self, __key: int, default: _ty.Any = None, *, decode: _ty.Literal[False]
    ) -> bytearray | None:
        ...

    @_ty.overload
    def get(self, __key: int, default: _ty.Any = None) -> DhcpOptionType | None:
        ...

    def get(
        self,
        __key: int,
        default: _ty.Any = None,
        decode: _ty.Union[bool, type[DhcpOptionType], _ty.Callable[[bytearray], _ty.Any]] = True,
    ) -> _ty.Any:
        value = self._options.get(__key, _const.MISSING)
        if value is _const.MISSING:
            return default
        assert isinstance(value, bytearray)
        if decode:
            target_decoder: _ty.Union[type[DhcpOptionType], _ty.Callable[[bytearray], _ty.Any]]
            if decode is True:
                target_decoder = self._codemap.from_code(__key).get_type()
            else:
                target_decoder = decode

            if isinstance(target_decoder, _builtins.type) and issubclass(target_decoder, DhcpOptionType):
                return target_decoder._dhcp_decode(value)
            return _ty.cast(_ty.Callable[[bytearray], _ty.Any], target_decoder)(value)
        else:
            return value

    def _ensuretype(self, option: _ty.Union[DhcpOption, tuple[int, _ty.Any]]) -> DhcpOption:
        if isinstance(option, DhcpOption):
            return option
        return self._codemap.normalize(*option)

    def append(self, option: _ty.Union[DhcpOption, tuple[int, _ty.Any]]) -> None:
        opt = self._ensuretype(option)
        opt.value._dhcp_write(self._options.setdefault(int(opt.code), bytearray()))

    def replace(self, option: _ty.Union[DhcpOption, tuple[int, _ty.Any]]) -> None:
        opt = self._ensuretype(option)
        self[int(opt.code)] = opt.value

    def __setitem__(self, __key: int, __value: _ty.Any) -> None:
        if not isinstance(__value, (bytes, memoryview, bytearray, DhcpOptionType)):
            __value = self._codemap.from_code(__key).get_type()(__value)  # type: ignore[call-arg]
        if isinstance(__value, DhcpOptionType):
            data = self._options.setdefault(__key, bytearray())
            data.clear()
            __value._dhcp_write(data)
        else:
            if not isinstance(__value, bytearray):
                __value = bytearray(__value)
            self._options[__key] = __value

    def __delitem__(self, __key: int) -> None:
        return self._options.__delitem__(__key)

    def __len__(self) -> int:
        return len(self._options)

    def __iter__(self) -> _ty.Iterator[int]:
        return self._options.__iter__()

    @_ty.overload  # type: ignore[override]
    def items(self) -> _ty.ItemsView[BaseDhcpOptionCode, DhcpOptionType]:
        ...

    @_ty.overload
    def items(self, decoded: _ty.Literal[False]) -> _ty.ItemsView[int, bytearray]:
        ...

    @_ty.overload
    def items(
        self, decoded: _ty.Literal[True]
    ) -> _ty.ItemsView[BaseDhcpOptionCode, DhcpOptionType]:
        ...

    @_ty.overload
    def items(self, decoded: type[C]) -> _ty.ItemsView[C, DhcpOptionType]:
        ...

    def items(self, decoded: _ty.Union[bool, type[BaseDhcpOptionCode]] = True) -> _ty.Any:
        items = self._options.items()
        if decoded is True:
            decoded = self._codemap
        if decoded:
            items = [decoded.decode(*option) for option in items]  # type: ignore
        return items

    def __contains__(self, __key: object) -> bool:
        return self._options.__contains__(__key)

Option Types

pydhcp.options.type

Boolean

Bases: DhcpOptionType, int

Boolean option encoded as a single octet.

Source code in src/pydhcp/options/type/scalar.py

class Boolean(DhcpOptionType, int):
    """Boolean option encoded as a single octet."""
    def __new__(cls, val: _ty.Any) -> Self:
        if val:
            val = 1
        else:
            val = 0
        return super().__new__(cls, val)

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        return cls(option[0]), 1

    def _dhcp_write(self, data: bytearray) -> int:
        data.append(self)
        return 1

    @classmethod
    def _dhcp_len_hint(cls) -> int | None:
        return 1

    def __repr__(self) -> str:
        return f"Boolean({bool(self)!r})"

    def __json__(self) -> bool:
        return self.__bool__()

Bytes

Bases: DhcpOptionType, bytes

Opaque byte payload.

Source code in src/pydhcp/options/type/scalar.py

class Bytes(DhcpOptionType, bytes):
    """Opaque byte payload."""
    def __new__(cls, src: _ty.Optional[_ty.Union[bytes, bytearray, memoryview, str]] = None) -> Self:
        if isinstance(src, str):
            return cls.fromhex(src)
        if src is None:
            return super().__new__(cls)
        return super().__new__(cls, src)

    def __repr__(self) -> str:
        return str(bytes(self))

    def __str__(self) -> str:
        return self.hex().upper()

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        return cls(option), len(option)

    def _dhcp_write(self, data: bytearray) -> int:
        data.extend(self)
        return len(self)

    def __json__(self) -> str:
        return self.hex()

CccApReqApRepBackoffRetry

Bases: CccAsReqAsRepBackoffRetry

CCC sub-option 5 AP-REQ/AP-REP backoff and retry tuple.

Source code in src/pydhcp/options/ccc.py

class CccApReqApRepBackoffRetry(CccAsReqAsRepBackoffRetry):
    """CCC sub-option 5 AP-REQ/AP-REP backoff and retry tuple."""

CccAsReqAsRepBackoffRetry

Bases: DhcpOptionType

CCC sub-option 4 AS-REQ/AS-REP backoff and retry tuple.

Source code in src/pydhcp/options/ccc.py

class CccAsReqAsRepBackoffRetry(DhcpOptionType):
    """CCC sub-option 4 AS-REQ/AS-REP backoff and retry tuple."""

    def __init__(self, initial_timeout: _ty.Any, maximum_timeout: _ty.Any, maximum_retry_count: _ty.Any) -> None:
        self.initial_timeout = int(initial_timeout)
        self.maximum_timeout = int(maximum_timeout)
        self.maximum_retry_count = int(maximum_retry_count)

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        if len(option) != 12:
            raise ValueError(f"{cls.__name__} option must contain 12 bytes")
        return (
            cls(
                int.from_bytes(option[0:4], "big"),
                int.from_bytes(option[4:8], "big"),
                int.from_bytes(option[8:12], "big"),
            ),
            12,
        )

    def _dhcp_write(self, data: bytearray) -> int:
        data.extend(self.initial_timeout.to_bytes(4, "big"))
        data.extend(self.maximum_timeout.to_bytes(4, "big"))
        data.extend(self.maximum_retry_count.to_bytes(4, "big"))
        return 12

    def __repr__(self) -> str:
        return (
            f"{type(self).__name__}(initial_timeout={self.initial_timeout!r}, "
            f"maximum_timeout={self.maximum_timeout!r}, "
            f"maximum_retry_count={self.maximum_retry_count!r})"
        )

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, CccAsReqAsRepBackoffRetry):
            return NotImplemented
        return (
            self.initial_timeout,
            self.maximum_timeout,
            self.maximum_retry_count,
        ) == (
            other.initial_timeout,
            other.maximum_timeout,
            other.maximum_retry_count,
        )

    def __json__(self) -> list[int]:
        return [self.initial_timeout, self.maximum_timeout, self.maximum_retry_count]

CccKdcServerAddressList

Bases: List[IPv4Address]

CCC sub-option 10 KDC server address list.

Source code in src/pydhcp/options/ccc.py

class CccKdcServerAddressList(List[IPv4Address]):
    """CCC sub-option 10 KDC server address list."""

CccKerberosRealmName

Bases: _CccDomainText

CCC Kerberos realm payload without DNS compression.

Source code in src/pydhcp/options/ccc.py

class CccKerberosRealmName(_CccDomainText):
    """CCC Kerberos realm payload without DNS compression."""

    def __new__(cls, value: _ty.Any) -> Self:
        return super().__new__(cls, str(value).upper())

CccOption

Bases: DhcpOptionType, list[CccSubOption]

CCC option container preserving unknown sub-options.

Source code in src/pydhcp/options/ccc.py

class CccOption(DhcpOptionType, list[CccSubOption]):
    """CCC option container preserving unknown sub-options."""

    def __init__(self, *items: _ty.Any):
        if len(items) == 1 and isinstance(items[0], list):
            self.extend(items[0])
            return
        for item in items:
            self.append(item)

    @classmethod
    def _normalize(cls, item: _ty.Any) -> CccSubOption:
        if isinstance(item, CccSubOption):
            return item
        code, value = item
        record_type = _CCC_SUBOPTION_TYPES.get(int(code), CccSubOption)
        return record_type(int(code), value)

    def append(self, item: _ty.Any) -> None:
        return list.append(self, self._normalize(item))

    def extend(self, __iterable: _ty.Iterable[_ty.Any]) -> None:
        list.extend(self, [self._normalize(item) for item in __iterable])

    @classmethod
    def _read_record(cls, option: memoryview) -> tuple[CccSubOption, int]:
        if len(option) < 2:
            raise ValueError(f"{cls.__name__} option is truncated")
        code = option[0]
        length = option[1]
        if len(option) < 2 + length:
            raise ValueError(f"{cls.__name__} option is truncated")
        payload = option[2 : 2 + length]
        record_type = _CCC_SUBOPTION_TYPES.get(code, CccSubOption)
        return record_type._from_payload(code, payload), 2 + length

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        self = cls()
        idx = 0
        size = len(option)
        while idx < size:
            record, read = cls._read_record(option[idx:])
            self.append(record)
            idx += read
        return self, size

    def _dhcp_write(self, data: bytearray) -> int:
        written = 0
        for item in self:
            written += item._dhcp_write(data)
        return written

    def __json__(self) -> list[list[_ty.Any]]:
        return [item.__json__() for item in self]

CccPrimaryDhcpServerAddress

Bases: IPv4Address

CCC sub-option 1 primary DHCP server address.

Source code in src/pydhcp/options/ccc.py

class CccPrimaryDhcpServerAddress(IPv4Address):
    """CCC sub-option 1 primary DHCP server address."""

CccProvisioningServerAddress

Bases: DhcpOptionType

CCC sub-option 3 tagged union for IPv4 address or FQDN.

Source code in src/pydhcp/options/ccc.py

class CccProvisioningServerAddress(DhcpOptionType):
    """CCC sub-option 3 tagged union for IPv4 address or FQDN."""

    def __init__(self, value: _ty.Any) -> None:
        self.kind, self.value = self._normalize(value)

    @staticmethod
    def _normalize(value: _ty.Any) -> tuple[str, _ty.Any]:
        if isinstance(value, CccProvisioningServerAddress):
            return value.kind, value.value
        if isinstance(value, (list, tuple)) and len(value) == 2:
            kind, payload = value
            kind_name = str(kind).lower()
            if kind_name in {"ipv4", "address"}:
                return "ipv4", IPv4Address(payload)
            if kind_name in {"fqdn", "domain"}:
                return "fqdn", CccProvisioningServerFqdn(payload)
            raise ValueError("CCC provisioning server address kind must be ipv4 or fqdn")
        if isinstance(value, _net.IPv4):
            return "ipv4", IPv4Address(value)
        if isinstance(value, str):
            try:
                return "ipv4", IPv4Address(value)
            except Exception:
                return "fqdn", CccProvisioningServerFqdn(value)
        if isinstance(value, CccProvisioningServerFqdn):
            return "fqdn", value
        return "ipv4", IPv4Address(value)

    @classmethod
    def _read_payload(cls, payload: memoryview) -> "CccProvisioningServerAddress":
        if len(payload) < 1:
            raise ValueError("CCC provisioning server address is truncated")
        kind = payload[0]
        if kind == 1:
            if len(payload) != 5:
                raise ValueError("CCC provisioning server IPv4 payload must be 5 bytes")
            return cls(("ipv4", IPv4Address(payload[1:5].tobytes())))
        if kind == 0:
            text, read = _decode_no_compression_domain(payload, 1)
            if 1 + read != len(payload):
                raise ValueError("CCC provisioning server FQDN payload is truncated")
            return cls(("fqdn", CccProvisioningServerFqdn(text)))
        raise ValueError(f"CCC provisioning server address kind {kind} is unsupported")

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple["CccProvisioningServerAddress", int]:
        return cls._read_payload(option), len(option)

    def _dhcp_write(self, data: bytearray) -> int:
        data.append(1 if self.kind == "ipv4" else 0)
        if self.kind == "ipv4":
            ipv4_payload = _ty.cast(IPv4Address, self.value)
            data.extend(ipv4_payload.packed)
            return 5
        fqdn_payload = _ty.cast(CccProvisioningServerFqdn, self.value)
        encoded = fqdn_payload._dhcp_encode()
        data.extend(encoded)
        return len(encoded) + 1

    def __repr__(self) -> str:
        return f"{type(self).__name__}(kind={self.kind!r}, value={self.value!r})"

    def __json__(self) -> list[_ty.Any]:
        return [self.kind, self.value.__json__() if isinstance(self.value, DhcpOptionType) else str(self.value)]

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, CccProvisioningServerAddress):
            return NotImplemented
        return (self.kind, self.value) == (other.kind, other.value)

CccProvisioningServerFqdn

Bases: _CccDomainText

CCC provisioning server FQDN payload without DNS compression.

Source code in src/pydhcp/options/ccc.py

class CccProvisioningServerFqdn(_CccDomainText):
    """CCC provisioning server FQDN payload without DNS compression."""

CccProvisioningTimer

Bases: U8

CCC sub-option 8 provisioning timer value.

Source code in src/pydhcp/options/ccc.py

class CccProvisioningTimer(U8):
    """CCC sub-option 8 provisioning timer value."""

CccSecondaryDhcpServerAddress

Bases: IPv4Address

CCC sub-option 2 secondary DHCP server address.

Source code in src/pydhcp/options/ccc.py

class CccSecondaryDhcpServerAddress(IPv4Address):
    """CCC sub-option 2 secondary DHCP server address."""

CccSecurityTicketControl

Bases: DhcpOptionType, int

CCC sub-option 9 security ticket control mask.

Source code in src/pydhcp/options/ccc.py

class CccSecurityTicketControl(DhcpOptionType, int):
    """CCC sub-option 9 security ticket control mask."""

    def __new__(cls, value: _ty.Any) -> Self:
        return int.__new__(cls, int(value))

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        if len(option) != 2:
            raise ValueError(f"{cls.__name__} option must contain 2 bytes")
        return cls(int.from_bytes(option, "big")), 2

    def _dhcp_write(self, data: bytearray) -> int:
        if self < 0 or self > 0xFFFF:
            raise ValueError(f"{type(self).__name__} mask must fit in 16 bits")
        if int(self) & ~0x0003:
            raise ValueError(f"{type(self).__name__} reserved bits 2-15 must be zero")
        data.extend(int(self).to_bytes(2, "big"))
        return 2

    def __repr__(self) -> str:
        return f"{type(self).__name__}({int(self)!r})"

    def __json__(self) -> int:
        return int(self)

CccSubOption

Bases: DhcpOptionType

Typed CCC sub-option record.

Source code in src/pydhcp/options/ccc.py

class CccSubOption(DhcpOptionType):
    """Typed CCC sub-option record."""

    _PAYLOAD_TYPE: type[DhcpOptionType] = Bytes

    def __init__(self, code: int, value: _ty.Any) -> None:
        self.code = int(code)
        self.value = self._normalize_value(self.code, value)

    @classmethod
    def _normalize_value(cls, code: int, value: _ty.Any) -> _ty.Any:
        if isinstance(value, DhcpOptionType):
            return value
        return Bytes(value)

    @classmethod
    def _read_payload(cls, payload: memoryview) -> _ty.Any:
        return Bytes(payload)

    def _write_payload(self, data: bytearray) -> int:
        payload = self.value
        if isinstance(payload, DhcpOptionType):
            return payload._dhcp_write(data)
        payload_bytes = Bytes(payload)
        data.extend(payload_bytes)
        return len(payload_bytes)

    @classmethod
    def _from_payload(cls: type[Self], code: int, payload: memoryview) -> Self:
        return cls(code, cls._read_payload(payload))

    def _dhcp_write(self, data: bytearray) -> int:
        payload = bytearray()
        payload_len = self._write_payload(payload)
        if payload_len > 255:
            raise ValueError(f"{type(self).__name__} entry exceeds 255 bytes")
        data.append(self.code)
        data.append(payload_len)
        data.extend(payload)
        return payload_len + 2

    def __repr__(self) -> str:
        return f"{type(self).__name__}(code={self.code!r}, value={self.value!r})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, CccSubOption):
            return NotImplemented
        return (self.code, self.value) == (other.code, other.value)

    def __json__(self) -> list[_ty.Any]:
        value = self.value
        if isinstance(value, DhcpOptionType):
            value = value.__json__()
        else:
            value = Bytes(value).__json__()
        return [self.code, value]

CccTicketGrantingServerUtilization

Bases: Boolean

CCC sub-option 7 ticket-granting server utilization toggle.

Source code in src/pydhcp/options/ccc.py

class CccTicketGrantingServerUtilization(Boolean):
    """CCC sub-option 7 ticket-granting server utilization toggle."""

ClasslessRoute

Bases: DhcpOptionType

RFC 3442 classless static route entry.

Source code in src/pydhcp/options/type/net.py

class ClasslessRoute(DhcpOptionType):
    """RFC 3442 classless static route entry."""
    def __init__(self, gateway: _IP, network: _Network) -> None:
        self.gateway = _IP(gateway)
        self.network = _Network(network)

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        if len(option) < 1:
            raise ValueError("ClasslessRoute option is truncated: missing prefix length")
        cidr = option[0]
        if cidr > 32:
            raise ValueError(f"ClasslessRoute prefix length {cidr} exceeds 32")
        last = 1 + (cidr + 7) // 8
        if len(option) < last + 4:
            raise ValueError(
                f"ClasslessRoute option is truncated: needs {last + 4} bytes, got {len(option)}"
            )
        net_bytes = option[1:last].tobytes() + b"\x00\x00\x00\x00"
        network = _Network((net_bytes[:4], cidr))
        gateway = _IP(option[last : last + 4].tobytes())
        return cls(gateway, network), last + 4

    def _dhcp_write(self, data: bytearray) -> int:
        cidr = self.network.prefixlen
        last = (cidr + 7) // 8
        network = self.network.network_address.packed[:last]
        data.append(cidr)
        data.extend(network)
        data.extend(self.gateway.packed)
        return last + 5

    def __repr__(self) -> str:
        return f"ClasslessRoute(gateway={self.gateway}, network={self.network})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, ClasslessRoute):
            return NotImplemented
        return (self.gateway, self.network) == (other.gateway, other.network)

    def __json__(self) -> list[_ty.Any]:
        return [str(self.gateway), str(self.network)]

ClientIdentifier

Bases: Bytes

RFC 2132 client identifier.

Source code in src/pydhcp/options/type/scalar.py

class ClientIdentifier(Bytes):
    """RFC 2132 client identifier."""
    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        if len(option) < 2:
            raise ValueError(option)
        return super()._dhcp_read(option)

    def __repr__(self) -> str:
        ty_val = self[0]
        addr = self[1:]
        ty_str = str(ty_val)
        try:
            from ...packet.enums import HardwareAddressType

            ty_str = HardwareAddressType(ty_val).name
        except ValueError:
            ...
        maybe = f"{ty_str}({addr.hex(':').upper()})"

        return f"{maybe}|{self}"

    def __str__(self) -> str:
        return self.hex(":").upper()

DhcpOptionCodes

Bases: List[_C]

List of option codes used by parameter-request-list style options.

Source code in src/pydhcp/options/type/base.py

class DhcpOptionCodes(List[_C]):  # type: ignore[type-var]
    """List of option codes used by parameter-request-list style options."""
    @classmethod
    def _normalize(cls, item: _ty.Any) -> _ty.Any:
        ty = cls._args_[0]
        if isinstance(item, _ty.cast(_ty.Any, ty)):
            return item
        try:
            return _ty.cast(_ty.Any, ty)(item)
        except (TypeError, ValueError):
            ...
        item_int = int(item)
        if item_int > 255:
            raise ValueError()
        return item_int

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple["Self", int]:
        return cls(option.tolist()), len(option)

    def _dhcp_write(self, data: bytearray) -> int:
        data.extend(_ty.cast(_ty.Iterable[int], self))
        return len(self)

DhcpOptionType

Protocol for DHCP option payload codecs.

Implementations decode with _dhcp_read, encode with _dhcp_write, and may advertise a fixed size with _dhcp_len_hint. The encode/decode pair should round-trip the same Python value.

Source code in src/pydhcp/options/type/base.py

class DhcpOptionType:
    """Protocol for DHCP option payload codecs.

    Implementations decode with `_dhcp_read`, encode with `_dhcp_write`, and may
    advertise a fixed size with `_dhcp_len_hint`. The encode/decode pair should
    round-trip the same Python value.
    """
    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple["Self", int]:
        raise NotImplementedError()

    def _dhcp_write(self, buffer: bytearray) -> int:
        raise NotImplementedError()

    def _dhcp_encode(self) -> bytes:
        encoded = bytearray()
        _wrote = self._dhcp_write(encoded)
        return bytes(encoded)

    def __json__(self) -> _ty.Any:
        return self

    @classmethod
    def _dhcp_len_hint(cls) -> int | None:
        return None

    @classmethod
    def _dhcp_decode(cls, option: memoryview | bytes | bytearray) -> "Self":
        hint = cls._dhcp_len_hint()
        todecode = len(option)
        option = memoryview(option) if not isinstance(option, memoryview) else option
        if hint:
            if todecode != hint:
                raise ValueError("Wrong option size")
        decoded, read = cls._dhcp_read(option)
        if read != todecode:
            raise ValueError("Couldnt decode whole option")
        return decoded

DomainList

Bases: DhcpOptionType, list[str]

RFC 1035 domain-name list with compression support.

Source code in src/pydhcp/options/type/net.py

class DomainList(DhcpOptionType, list[str]):
    """RFC 1035 domain-name list with compression support."""
    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        view = memoryview(option)
        self = cls()
        if not option:
            return self, 0
        components: dict[int, str | int | None] = _ty.OrderedDict()
        domains: list[int] = [0]
        id = 0
        size = len(view)
        while id < size:
            ptr_or_len = view[id]
            id += 1
            if ptr_or_len == 0x00:
                components[id - 1] = None
                if id < size:
                    domains.append(id)
                continue
            is_ptr = ptr_or_len & 0xC0
            if is_ptr:
                if is_ptr != 0xC0:
                    raise ValueError()
                components[id - 1] = ((0x3F & ptr_or_len) << 8) | view[id]
                id += 1
                if id < size:
                    domains.append(id)
            else:
                dc = view[id : ptr_or_len + id]
                if len(dc) != ptr_or_len:
                    raise ValueError()
                components[id - 1] = dc.tobytes().decode()
                id += ptr_or_len

        def get_dn(id: int) -> list[str]:
            result = []
            for idx, dc in components.items():
                if idx >= id:
                    if dc is None:
                        break
                    elif isinstance(dc, int):
                        result.extend(get_dn(dc))
                        break
                    result.append(dc)
            return result

        for domain in domains:
            self.append(".".join(get_dn(domain)))
        return self, len(option)

    def _dhcp_write(self, _data: bytearray) -> int:
        components: list[tuple[list[str], int]] = []
        data = bytearray()
        for domain_str in self:
            domain = domain_str.split(".")
            unique = domain
            parent: _ty.Optional[tuple[int, int]] = None
            for cn, cidx in components:
                pair = 1
                while pair < len(domain):
                    if domain[-pair:] != cn[-pair:]:
                        break
                    pair += 1
                pair -= 1
                if pair:
                    if parent:
                        _, _pair = parent
                        if pair <= _pair:
                            continue
                    for n in cn[:-pair]:
                        cidx += 1 + len(n)

                    parent = cidx, pair
                    unique = domain[:-pair]

            components.append((domain, len(data)))
            for comp in unique:
                data.append(len(comp))
                data.extend(comp.encode())
            if parent is None:
                data.append(0x00)
            else:
                data.extend((0xC000 | parent[0]).to_bytes(2, byteorder="big"))
        _data.extend(data)
        return len(data)

I32

Bases: FixedLengthInteger

Signed 32-bit integer.

Source code in src/pydhcp/options/type/scalar.py

class I32(FixedLengthInteger):
    """Signed 32-bit integer."""
    NUMBER_OF_BYTES = 4
    SIGNED = True

IPv4Address

Bases: DhcpOptionType, IPv4

A single IPv4 address carried in network byte order.

Source code in src/pydhcp/options/type/net.py

class IPv4Address(DhcpOptionType, _IP):
    """A single IPv4 address carried in network byte order."""
    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        return cls(option[:4].tobytes()), 4

    def _dhcp_write(self, data: bytearray) -> int:
        data.extend(self.packed)
        return 4

    @classmethod
    def _dhcp_len_hint(cls) -> int | None:
        return 4

    def __repr__(self) -> str:
        return str(self)

    def __json__(self) -> str:
        return str(self)

List

Bases: DhcpOptionType, list[_T]

Typed DHCP option list container.

Source code in src/pydhcp/options/type/base.py

class List(DhcpOptionType, list[_T], metaclass=_utils.GenericMeta):
    """Typed DHCP option list container."""
    _args_: _ty.ClassVar[tuple[_T]]

    def __init__(self, *items: _ty.Any):
        for _items in items:
            self.extend(_items if isinstance(_items, (tuple, list)) else (_items,))

    @classmethod
    def _normalize(cls, item: _ty.Any) -> _T:
        ty = cls._args_[0]
        if isinstance(item, _ty.cast(_ty.Any, ty)):
            return _ty.cast(_T, item)
        return _ty.cast(_T, _ty.cast(_ty.Any, ty)(item))

    def __setitem__(self, idx: _ty.Any, item: _T) -> None:  # type: ignore[override]
        return list.__setitem__(self, idx, self._normalize(item))

    def append(self, item: _T) -> None:
        return list.append(self, self._normalize(item))

    def extend(self, __iterable: Iterable[_T]) -> None:
        list.extend(
            self,
            [self._normalize(item) for item in __iterable],
        )

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple["Self", int]:
        _l = len(option)
        self = cls()
        ty = self._args_[0]
        while option:
            item, l = ty._dhcp_read(option)
            self.append(item)
            option = option[l:]
        return self, _l

    def _dhcp_write(self, data: bytearray) -> int:
        written = 0
        for item in self:
            written += item._dhcp_write(data)
        return written

    def __json__(self) -> list[_ty.Any]:
        return [item.__json__() for item in self]

MoSFqdnList

Bases: _MoSOptionBase

RFC 5678 MoS option carrying FQDN sub-options.

Source code in src/pydhcp/options/type/mos.py

class MoSFqdnList(_MoSOptionBase):
    """RFC 5678 MoS option carrying FQDN sub-options."""

    _RECORD_TYPE = MoSFqdnRecord

MoSFqdnRecord

Bases: _MoSFqdnSubOption

RFC 5678 MoS sub-option record carrying FQDN label sequences.

Source code in src/pydhcp/options/type/mos.py

class MoSFqdnRecord(_MoSFqdnSubOption):
    """RFC 5678 MoS sub-option record carrying FQDN label sequences."""

MoSIpv4AddressList

Bases: _MoSOptionBase

RFC 5678 MoS option carrying IPv4 address sub-options.

Source code in src/pydhcp/options/type/mos.py

class MoSIpv4AddressList(_MoSOptionBase):
    """RFC 5678 MoS option carrying IPv4 address sub-options."""

    _RECORD_TYPE = MoSIpv4AddressRecord

MoSIpv4AddressRecord

Bases: _MoSIpv4AddressSubOption

RFC 5678 MoS sub-option record carrying IPv4 addresses.

Source code in src/pydhcp/options/type/mos.py

class MoSIpv4AddressRecord(_MoSIpv4AddressSubOption):
    """RFC 5678 MoS sub-option record carrying IPv4 addresses."""

OptionOverload

Bases: DhcpOptionType, IntFlag

RFC 2132 option-overload selector.

Source code in src/pydhcp/options/type/scalar.py

class OptionOverload(DhcpOptionType, _enum.IntFlag):
    """RFC 2132 option-overload selector."""
    NONE = 0
    FILE = 1
    SNAME = 2
    BOTH = FILE | SNAME

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        option_part = option[:1]
        if len(option_part) != 1:
            raise ValueError()
        return cls(option_part[0]), 1

    def _dhcp_write(self, data: bytearray) -> int:
        data.append(self.value)
        return 1

    @classmethod
    def _dhcp_len_hint(cls) -> int | None:
        return 1

PolicyFilter

Bases: _IPv4PairList

List of IPv4 destination/mask pairs for policy filtering.

Source code in src/pydhcp/options/type/net.py

class PolicyFilter(_IPv4PairList):
    """List of IPv4 destination/mask pairs for policy filtering."""

RdnssSelection

Bases: DhcpOptionType

RFC 6731 RDNSS selection payload.

Source code in src/pydhcp/options/type/net.py

class RdnssSelection(DhcpOptionType):
    """RFC 6731 RDNSS selection payload."""

    def __init__(self, flags: int, primary: _IP, secondary: _IP, domains: DomainList | None = None) -> None:
        self.flags = int(flags)
        self.primary = _IP(primary)
        self.secondary = _IP(secondary)
        self.domains = self._normalize_domains(domains or [])

    @staticmethod
    def _normalize_domains(domains: _ty.Iterable[str]) -> DomainList:
        normalized = DomainList(domains)
        if normalized and normalized[-1] == "":
            normalized.pop()
        return normalized

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        if len(option) < 9:
            raise ValueError(f"{cls.__name__} option is truncated")
        flags = option[0]
        primary = _IP(option[1:5].tobytes())
        secondary = _IP(option[5:9].tobytes())
        domains, read = DomainList._dhcp_read(option[9:])
        return cls(flags, primary, secondary, domains), 9 + read

    def _dhcp_write(self, data: bytearray) -> int:
        encoded = self.domains._dhcp_encode()
        data.append(self.flags)
        data.extend(self.primary.packed)
        data.extend(self.secondary.packed)
        data.extend(encoded)
        return 9 + len(encoded)

    def __repr__(self) -> str:
        return (
            f"RdnssSelection(flags={self.flags!r}, primary={self.primary}, "
            f"secondary={self.secondary}, domains={self.domains!r})"
        )

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, RdnssSelection):
            return NotImplemented
        return (
            self.flags,
            self.primary,
            self.secondary,
            list(self.domains),
        ) == (
            other.flags,
            other.primary,
            other.secondary,
            list(other.domains),
        )

    def __json__(self) -> list[_ty.Any]:
        return [self.flags, str(self.primary), str(self.secondary), self.domains.__json__()]

RelayAgentInformation

Bases: EncapsulatedOptions

RFC 3046 relay-agent sub-options.

Source code in src/pydhcp/options/type/vendor.py

class RelayAgentInformation(EncapsulatedOptions):
    """RFC 3046 relay-agent sub-options."""

StaticRoute

Bases: _IPv4PairList

List of IPv4 destination/router pairs for static routing.

Source code in src/pydhcp/options/type/net.py

class StaticRoute(_IPv4PairList):
    """List of IPv4 destination/router pairs for static routing."""

    @classmethod
    def _normalize(cls, item: _ty.Any) -> tuple[_IP, _IP]:
        left, right = super()._normalize(item)
        if left == _IP("0.0.0.0"):
            raise ValueError("StaticRoute does not allow a default-route destination")
        return left, right

String

Bases: DhcpOptionType, str

RFC 2132 NVT-ASCII string with null termination on the wire.

Source code in src/pydhcp/options/type/scalar.py

class String(DhcpOptionType, str):
    """RFC 2132 NVT-ASCII string with null termination on the wire."""
    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        text, _, _ = option.tobytes().partition(b"\x00")
        try:
            decoded_text = text.decode("utf-8")
        except UnicodeDecodeError:
            LOGGER.warning(f"Option contains invalid UTF-8: {text.hex()}")
            decoded_text = text.decode("utf-8", errors="replace")
        return cls(decoded_text), len(option)

    def _dhcp_write(self, data: bytearray) -> int:
        text = self.encode()
        data.extend(text)
        return len(text)

U16

Bases: FixedLengthInteger

Unsigned 16-bit integer.

Source code in src/pydhcp/options/type/scalar.py

class U16(FixedLengthInteger):
    """Unsigned 16-bit integer."""
    NUMBER_OF_BYTES = 2
    SIGNED = False

U32

Bases: FixedLengthInteger

Unsigned 32-bit integer.

Source code in src/pydhcp/options/type/scalar.py

class U32(FixedLengthInteger):
    """Unsigned 32-bit integer."""
    NUMBER_OF_BYTES = 4
    SIGNED = False

U8

Bases: FixedLengthInteger

Unsigned 8-bit integer.

Source code in src/pydhcp/options/type/scalar.py

class U8(FixedLengthInteger):
    """Unsigned 8-bit integer."""
    NUMBER_OF_BYTES = 1
    SIGNED = False

UriList

Bases: DhcpOptionType, list[str]

List of UTF-8 URIs encoded as repeated U16-length-prefixed entries.

Source code in src/pydhcp/options/type/scalar.py

class UriList(DhcpOptionType, list[str]):
    """List of UTF-8 URIs encoded as repeated U16-length-prefixed entries."""

    def __init__(self, *items: _ty.Any):
        if len(items) == 1 and isinstance(items[0], list):
            self.extend(items[0])
            return
        for item in items:
            self.append(item)

    @classmethod
    def _normalize(cls, item: _ty.Any) -> str:
        if isinstance(item, str):
            return item
        return str(item)

    def append(self, item: _ty.Any) -> None:
        return list.append(self, self._normalize(item))

    def extend(self, __iterable: Iterable[_ty.Any]) -> None:
        list.extend(self, [self._normalize(item) for item in __iterable])

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        self = cls()
        idx = 0
        size = len(option)
        while idx < size:
            if idx + 2 > size:
                raise ValueError(f"{cls.__name__} option is truncated")
            length = int.from_bytes(option[idx : idx + 2], "big")
            idx += 2
            if idx + length > size:
                raise ValueError(f"{cls.__name__} option is truncated")
            payload = option[idx : idx + length].tobytes()
            try:
                decoded = payload.decode("utf-8")
            except UnicodeDecodeError as exc:
                raise ValueError(f"{cls.__name__} option contains invalid UTF-8") from exc
            self.append(decoded)
            idx += length
        return self, size

    def _dhcp_write(self, data: bytearray) -> int:
        written = 0
        for item in self:
            encoded = item.encode("utf-8")
            if len(encoded) > 0xFFFF:
                raise ValueError(f"{type(self).__name__} entry exceeds 65535 bytes")
            data.extend(len(encoded).to_bytes(2, "big"))
            data.extend(encoded)
            written += len(encoded) + 2
        return written

    def __json__(self) -> list[str]:
        return list(self)

UserClass

Bases: _LengthPrefixedOpaqueList

RFC 3004 user-class opaque byte list.

Source code in src/pydhcp/options/type/vendor.py

class UserClass(_LengthPrefixedOpaqueList):
    """RFC 3004 user-class opaque byte list."""

VendorSpecificInformation

Bases: Bytes

Opaque vendor-specific payload for option 43.

Source code in src/pydhcp/options/type/vendor.py

class VendorSpecificInformation(Bytes):
    """Opaque vendor-specific payload for option 43."""

ViVendorClass

Bases: DhcpOptionType, list[ViVendorClassRecord]

RFC 3925 vendor-identifying vendor class records.

Source code in src/pydhcp/options/type/vendor.py

class ViVendorClass(DhcpOptionType, list[ViVendorClassRecord]):
    """RFC 3925 vendor-identifying vendor class records."""

    def __init__(self, *items: _ty.Any):
        if len(items) == 1 and isinstance(items[0], list):
            self.extend(items[0])
            return
        for item in items:
            self.append(item)

    @classmethod
    def _normalize(cls, item: _ty.Any) -> ViVendorClassRecord:
        if isinstance(item, ViVendorClassRecord):
            return item
        enterprise_number, value = item
        return ViVendorClassRecord(enterprise_number, value)

    def append(self, item: _ty.Any) -> None:
        return list.append(self, self._normalize(item))

    def extend(self, __iterable: Iterable[_ty.Any]) -> None:
        list.extend(self, [self._normalize(item) for item in __iterable])

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        self = cls()
        idx = 0
        size = len(option)
        while idx < size:
            record, read = ViVendorClassRecord._dhcp_read(option[idx:])
            self.append(record)
            idx += read
        return self, size

    def _dhcp_write(self, data: bytearray) -> int:
        written = 0
        for item in self:
            written += item._dhcp_write(data)
        return written

    def __json__(self) -> list[list[_ty.Any]]:
        return [item.__json__() for item in self]

ViVendorSpecificInformation

Bases: DhcpOptionType, list[ViVendorSpecificInformationRecord]

RFC 3925 vendor-identifying vendor-specific information records.

Source code in src/pydhcp/options/type/vendor.py

class ViVendorSpecificInformation(DhcpOptionType, list[ViVendorSpecificInformationRecord]):
    """RFC 3925 vendor-identifying vendor-specific information records."""

    def __init__(self, *items: _ty.Any):
        if len(items) == 1 and isinstance(items[0], list):
            self.extend(items[0])
            return
        for item in items:
            self.append(item)

    @classmethod
    def _normalize(cls, item: _ty.Any) -> ViVendorSpecificInformationRecord:
        if isinstance(item, ViVendorSpecificInformationRecord):
            return item
        enterprise_number, value = item
        return ViVendorSpecificInformationRecord(enterprise_number, value)

    def append(self, item: _ty.Any) -> None:
        return list.append(self, self._normalize(item))

    def extend(self, __iterable: Iterable[_ty.Any]) -> None:
        list.extend(self, [self._normalize(item) for item in __iterable])

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        self = cls()
        idx = 0
        size = len(option)
        while idx < size:
            record, read = ViVendorSpecificInformationRecord._dhcp_read(option[idx:])
            self.append(record)
            idx += read
        return self, size

    def _dhcp_write(self, data: bytearray) -> int:
        written = 0
        for item in self:
            written += item._dhcp_write(data)
        return written

    def __json__(self) -> list[list[_ty.Any]]:
        return [item.__json__() for item in self]

DhcpServer

pydhcp.server.DhcpServer

Bases: DhcpListener

Source code in src/pydhcp/server.py

class DhcpServer(_Base):
    DEFAULT_PORTS = (_enum.DhcpPort.SERVER,)

    def __init__(
        self,
        listen: ListenSpec = None,
        select_timeout: _ty.Optional[float] = None,
        max_packet_size: _ty.Optional[int] = None,
        lease_backend: _ty.Optional[LeaseBackend] = None,
        per_interface: bool | None = None,
    ) -> None:
        super().__init__(
            listen=listen,
            select_timeout=select_timeout,
            max_packet_size=max_packet_size,
            per_interface=per_interface,
        )
        from .lease import InMemoryLeaseBackend
        self.lease_backend = lease_backend or InMemoryLeaseBackend()

    def acquire_lease(self, client_id: str, server_id: _net.IPv4, msg: DhcpMessage) -> _ty.Optional[DhcpLease]:
        """Return a lease for a client message.

        The base implementation is intentionally small: it renews existing leases and
        allocates only when the client supplies `REQUESTED_IP` or `ciaddr`. Override
        this method to implement address pools, reservations, policy checks, or custom
        response options.
        """
        _server = next(_net.host_ip_interfaces(lambda interface: interface.ip == server_id), None)
        if _server is None:
            return None

        existing = self.lease_backend.lookup(client_id)
        if existing:
            requested_ttl = msg.options.get(DhcpOptionCode.IP_ADDRESS_LEASE_TIME, decode=_type.U32)
            ttl = int(requested_ttl) if requested_ttl is not None else 3600
            renewed = self.lease_backend.renew(client_id, ttl)
            if renewed:
                self.metrics.leases_renewed += 1
                return renewed
            return existing

        requested_ip = msg.options.get(
            DhcpOptionCode.REQUESTED_IP, decode=_type.IPv4Address
        )
        requested_ttl = msg.options.get(DhcpOptionCode.IP_ADDRESS_LEASE_TIME, decode=_type.U32)
        ttl = int(requested_ttl) if requested_ttl is not None else 3600

        ip: _ty.Optional[_net.IPv4] = None
        if requested_ip:
            ip = requested_ip
        elif msg.ciaddr != _net.WILDCARD_IPv4:
            ip = msg.ciaddr

        if ip is None:
            return None

        options = DhcpOptions()
        options[DhcpOptionCode.SUBNET_MASK] = _server.network.netmask
        options[DhcpOptionCode.BROADCAST_ADDRESS] = _server.network.broadcast_address
        options[DhcpOptionCode.ROUTER] = [server_id]
        options[DhcpOptionCode.DNS] = [server_id]

        LOGGER.debug(f"[XID={msg.xid:08x}] Allocating {ip} for {client_id}")
        lease = self.lease_backend.allocate(client_id, ip, ttl, options)
        if lease is not None:
            self.metrics.leases_allocated += 1
        return lease

    def release_lease(self, client_id: str, server_id: _net.IPv4, msg: DhcpMessage) -> None:
        """Release any lease associated with `client_id`.

        Override this method when lease release needs to update an external store,
        quarantine declined addresses, or emit custom audit records.
        """
        if self.lease_backend.release(client_id):
            self.metrics.leases_released += 1

    def get_inform_options(self, server_id: _net.IPv4, msg: DhcpMessage) -> DhcpOptions:
        """Return configuration options for DHCPINFORM responses.

        DHCPINFORM does not allocate an address. Override this method when clients
        should receive site-specific options without touching lease allocation.
        """
        options = DhcpOptions()
        _server = next(_net.host_ip_interfaces(lambda interface: interface.ip == server_id), None)
        if _server is not None:
            options[DhcpOptionCode.SUBNET_MASK] = _server.network.netmask
            options[DhcpOptionCode.BROADCAST_ADDRESS] = _server.network.broadcast_address
            options[DhcpOptionCode.ROUTER] = [server_id]
            options[DhcpOptionCode.DNS] = [server_id]
        return options

    def handle(
        self,
        msg: DhcpMessage,
        context: RequestContext,
    ) -> None:
        if msg.op != _enum.OpCode.BOOTREQUEST:
            LOGGER.warning(
                f"[XID={msg.xid:08x}] Received a reply msg from {context.client} ignoring it."
            )
            return
        client_id = msg.client_id()
        msg_ty = msg.options.get(DhcpOptionCode.DHCP_MESSAGE_TYPE)
        msg_ty_name = msg_ty.name if (msg_ty is not None and hasattr(msg_ty, "name")) else str(msg_ty)
        LOGGER.debug(f"[XID={msg.xid:08x}] Received {msg_ty_name} from {context.client.ip}")
        server_id: _ty.Optional[_net.IPv4] = msg.options.get(
            DhcpOptionCode.SERVER_IDENTIFIER, decode=_type.IPv4Address
        )
        msg_ty = msg.options.get(DhcpOptionCode.DHCP_MESSAGE_TYPE)
        actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)

        if server_id is not None and server_id != actual_server_id:
            if msg_ty is _enum.DhcpMessageType.DHCPREQUEST:
                self.release_lease(client_id, server_id, msg)
            else:
                LOGGER.warning(
                    f"[XID={msg.xid:08x}] Received a message for {server_id} by {context.client}|{client_id} at {actual_server_id} ignoring"
                )
            return

        if msg_ty is _enum.DhcpMessageType.DHCPDISCOVER:
            self.handle_discover(msg, context)
        elif msg_ty is _enum.DhcpMessageType.DHCPREQUEST:
            self.handle_request(msg, context)
        elif msg_ty is _enum.DhcpMessageType.DHCPDECLINE:
            self.handle_decline(msg, context)
        elif msg_ty is _enum.DhcpMessageType.DHCPRELEASE:
            self.handle_release(msg, context)
        elif msg_ty is _enum.DhcpMessageType.DHCPINFORM:
            self.handle_inform(msg, context)
        else:
            LOGGER.warning(
                f"[XID={msg.xid:08x}] Received a DHCP Message with message type: {msg_ty} from: {context.client}|{client_id} at: {actual_server_id}, which we don't handle"
            )

    def handle_discover(self, msg: DhcpMessage, context: RequestContext) -> None:
        """Handle DHCPDISCOVER by offering a lease returned from `acquire_lease`."""
        client_id = msg.client_id()
        actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
        LOGGER.info(f"[XID={msg.xid:08x}] DHCPDISCOVER from {context.client}|{client_id}")
        lease = self.acquire_lease(client_id, actual_server_id, msg)
        if not lease:
            LOGGER.info(
                f"[XID={msg.xid:08x}] No lease available for {context.client}|{client_id} at {actual_server_id} ignoring"
            )
            return
        resp = self._create_response(msg, lease, actual_server_id, _enum.DhcpMessageType.DHCPOFFER)
        self._filter_and_send(msg, resp, context, _enum.DhcpMessageType.DHCPOFFER)

    def handle_request(self, msg: DhcpMessage, context: RequestContext) -> None:
        """Handle DHCPREQUEST by ACKing or NAKing the lease returned from `acquire_lease`."""
        client_id = msg.client_id()
        actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
        LOGGER.info(f"[XID={msg.xid:08x}] DHCPREQUEST from {context.client}|{client_id}")
        lease = self.acquire_lease(client_id, actual_server_id, msg)
        if not lease:
            LOGGER.info(
                f"[XID={msg.xid:08x}] No lease available for {context.client}|{client_id} at {actual_server_id} ignoring"
            )
            return
        ip_req: _ty.Optional[_net.IPv4] = msg.options.get(
            DhcpOptionCode.REQUESTED_IP, decode=_type.IPv4Address
        )
        if not ip_req:
            ip_req = msg.ciaddr
        if ip_req == lease.ip:
            resp_ty = _enum.DhcpMessageType.DHCPACK
        else:
            resp_ty = _enum.DhcpMessageType.DHCPNAK
        resp = self._create_response(msg, lease, actual_server_id, resp_ty)
        self._filter_and_send(msg, resp, context, resp_ty)

    def handle_decline(self, msg: DhcpMessage, context: RequestContext) -> None:
        """Handle DHCPDECLINE by releasing the client's lease through `release_lease`."""
        client_id = msg.client_id()
        actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
        LOGGER.warning(f"[XID={msg.xid:08x}] DHCPDECLINE from {context.client}|{client_id}")
        self.release_lease(client_id, actual_server_id, msg)

    def handle_release(self, msg: DhcpMessage, context: RequestContext) -> None:
        """Handle DHCPRELEASE by releasing the client's lease through `release_lease`."""
        client_id = msg.client_id()
        actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
        LOGGER.info(f"[XID={msg.xid:08x}] DHCPRELEASE from {context.client}|{client_id}")
        self.release_lease(client_id, actual_server_id, msg)

    def handle_inform(self, msg: DhcpMessage, context: RequestContext) -> None:
        """Handle DHCPINFORM without requiring address allocation."""
        client_id = msg.client_id()
        actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
        LOGGER.info(f"[XID={msg.xid:08x}] DHCPINFORM from {context.client}|{client_id}")
        lease = self.acquire_lease(client_id, actual_server_id, msg)
        if not lease:
            lease = DhcpLease(
                _net.WILDCARD_IPv4,
                _inf,
                self.get_inform_options(actual_server_id, msg),
            )
        resp = self._create_response(msg, lease, actual_server_id, _enum.DhcpMessageType.DHCPACK)
        if DhcpOptionCode.IP_ADDRESS_LEASE_TIME in resp.options:
            del resp.options[DhcpOptionCode.IP_ADDRESS_LEASE_TIME]
        resp.yiaddr = _net.WILDCARD_IPv4
        self._filter_and_send(msg, resp, context, _enum.DhcpMessageType.DHCPACK)

    def _create_response(
        self,
        msg: DhcpMessage,
        lease: DhcpLease,
        actual_server_id: _net.IPv4,
        resp_ty: _enum.DhcpMessageType,
    ) -> DhcpMessage:
        resp = DhcpMessage(**msg.__dict__.copy())
        resp.options = lease.options
        resp.op = _enum.OpCode.BOOTREPLY
        resp.hops = 0
        resp.secs = _dt.timedelta(seconds=0)
        if lease.ip:
            if lease.expires is None or lease.expires == _inf or not isinstance(lease.expires, _dt.datetime):
                expires = _const.INFINITE_LEASE_TIME
            else:
                expires = int((lease.expires - _dt.datetime.now()).total_seconds())
                expires = min(expires, _const.INFINITE_LEASE_TIME)
            if expires > 0:
                resp.options[DhcpOptionCode.IP_ADDRESS_LEASE_TIME] = expires
                resp.yiaddr = lease.ip
        resp.options[DhcpOptionCode.SERVER_IDENTIFIER] = actual_server_id
        resp.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = resp_ty
        relay_info = msg.options.get(DhcpOptionCode.RELAY_AGENT_INFORMATION, decode=False)
        if relay_info is not None:
            resp.options[DhcpOptionCode.RELAY_AGENT_INFORMATION] = relay_info
        return resp

    def _filter_and_send(
        self,
        msg: DhcpMessage,
        resp: DhcpMessage,
        context: RequestContext,
        resp_ty: _enum.DhcpMessageType,
    ) -> None:
        requests_params_raw = msg.options.get(
            DhcpOptionCode.PARAMETER_REQUEST_LIST,
            decode=_type.DhcpOptionCodes[DhcpOptionCode],
        )
        requests_params: _ty.List[DhcpOptionCode] = []
        if requests_params_raw:
            requests_params = [
                *requests_params_raw,
                DhcpOptionCode.IP_ADDRESS_LEASE_TIME,
                DhcpOptionCode.SERVER_IDENTIFIER,
            ]
        if resp_ty is _enum.DhcpMessageType.DHCPNAK:
            requests_params = [
                DhcpOptionCode.DHCP_MESSAGE,
                DhcpOptionCode.CLIENT_IDENTIFIER,
                DhcpOptionCode.VENDOR_CLASS_IDENTIFIER,
                DhcpOptionCode.SERVER_IDENTIFIER,
            ]
            resp.options[DhcpOptionCode.CLIENT_IDENTIFIER] = bytearray.fromhex(
                msg.client_id().replace(":", "")
            )
        if requests_params:
            def _paramfilter(opt: tuple[int, bytearray]) -> bool:
                return opt[0] in requests_params
            resp.options._options = _ty.OrderedDict(
                filter(_paramfilter, resp.options.items(decoded=False))
            )
        resp.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = resp_ty

        max_size_opt = msg.options.get(
            DhcpOptionCode.MAXIMUM_DHCP_MESSAGE_SIZE,
            default=_const.DHCP_MIN_LEGAL_PACKET_SIZE,
            decode=_type.U16,
        )
        max_size = int(max_size_opt) if max_size_opt is not None else _const.DHCP_MIN_LEGAL_PACKET_SIZE
        data = resp.encode(max_size)

        dest: _net.IPv4
        dest_port: int = context.client.port

        if msg.giaddr != _net.WILDCARD_IPv4:
            dest = msg.giaddr
            dest_port = 67 if context.client.port == 68 else context.client.port
        elif msg.ciaddr != _net.WILDCARD_IPv4:
            dest = msg.ciaddr
        elif msg.flags is _enum.Flags.BROADCAST:
            dest = _net.IPv4("255.255.255.255")
        else:
            if resp.yiaddr != _net.WILDCARD_IPv4:
                dest = resp.yiaddr
            else:
                dest = _net.IPv4("255.255.255.255")

        resp.log(context.interface.ip, _net.SocketAddress(dest, dest_port), _logging.INFO)
        if __debug__:
            _check = DhcpMessage.decode(memoryview(data))
            _check.log(
                context.interface.ip, _net.SocketAddress(dest, dest_port), _logging.DEBUG
            )
        context.transport.send(data, dest, dest_port, context.client_mac)
        self.metrics.packets_sent += 1

acquire_lease(client_id, server_id, msg)

Return a lease for a client message.

The base implementation is intentionally small: it renews existing leases and allocates only when the client supplies REQUESTED_IP or ciaddr. Override this method to implement address pools, reservations, policy checks, or custom response options.

Source code in src/pydhcp/server.py

def acquire_lease(self, client_id: str, server_id: _net.IPv4, msg: DhcpMessage) -> _ty.Optional[DhcpLease]:
    """Return a lease for a client message.

    The base implementation is intentionally small: it renews existing leases and
    allocates only when the client supplies `REQUESTED_IP` or `ciaddr`. Override
    this method to implement address pools, reservations, policy checks, or custom
    response options.
    """
    _server = next(_net.host_ip_interfaces(lambda interface: interface.ip == server_id), None)
    if _server is None:
        return None

    existing = self.lease_backend.lookup(client_id)
    if existing:
        requested_ttl = msg.options.get(DhcpOptionCode.IP_ADDRESS_LEASE_TIME, decode=_type.U32)
        ttl = int(requested_ttl) if requested_ttl is not None else 3600
        renewed = self.lease_backend.renew(client_id, ttl)
        if renewed:
            self.metrics.leases_renewed += 1
            return renewed
        return existing

    requested_ip = msg.options.get(
        DhcpOptionCode.REQUESTED_IP, decode=_type.IPv4Address
    )
    requested_ttl = msg.options.get(DhcpOptionCode.IP_ADDRESS_LEASE_TIME, decode=_type.U32)
    ttl = int(requested_ttl) if requested_ttl is not None else 3600

    ip: _ty.Optional[_net.IPv4] = None
    if requested_ip:
        ip = requested_ip
    elif msg.ciaddr != _net.WILDCARD_IPv4:
        ip = msg.ciaddr

    if ip is None:
        return None

    options = DhcpOptions()
    options[DhcpOptionCode.SUBNET_MASK] = _server.network.netmask
    options[DhcpOptionCode.BROADCAST_ADDRESS] = _server.network.broadcast_address
    options[DhcpOptionCode.ROUTER] = [server_id]
    options[DhcpOptionCode.DNS] = [server_id]

    LOGGER.debug(f"[XID={msg.xid:08x}] Allocating {ip} for {client_id}")
    lease = self.lease_backend.allocate(client_id, ip, ttl, options)
    if lease is not None:
        self.metrics.leases_allocated += 1
    return lease

get_inform_options(server_id, msg)

Return configuration options for DHCPINFORM responses.

DHCPINFORM does not allocate an address. Override this method when clients should receive site-specific options without touching lease allocation.

Source code in src/pydhcp/server.py

def get_inform_options(self, server_id: _net.IPv4, msg: DhcpMessage) -> DhcpOptions:
    """Return configuration options for DHCPINFORM responses.

    DHCPINFORM does not allocate an address. Override this method when clients
    should receive site-specific options without touching lease allocation.
    """
    options = DhcpOptions()
    _server = next(_net.host_ip_interfaces(lambda interface: interface.ip == server_id), None)
    if _server is not None:
        options[DhcpOptionCode.SUBNET_MASK] = _server.network.netmask
        options[DhcpOptionCode.BROADCAST_ADDRESS] = _server.network.broadcast_address
        options[DhcpOptionCode.ROUTER] = [server_id]
        options[DhcpOptionCode.DNS] = [server_id]
    return options

handle_decline(msg, context)

Handle DHCPDECLINE by releasing the client's lease through release_lease.

Source code in src/pydhcp/server.py

def handle_decline(self, msg: DhcpMessage, context: RequestContext) -> None:
    """Handle DHCPDECLINE by releasing the client's lease through `release_lease`."""
    client_id = msg.client_id()
    actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
    LOGGER.warning(f"[XID={msg.xid:08x}] DHCPDECLINE from {context.client}|{client_id}")
    self.release_lease(client_id, actual_server_id, msg)

handle_discover(msg, context)

Handle DHCPDISCOVER by offering a lease returned from acquire_lease.

Source code in src/pydhcp/server.py

def handle_discover(self, msg: DhcpMessage, context: RequestContext) -> None:
    """Handle DHCPDISCOVER by offering a lease returned from `acquire_lease`."""
    client_id = msg.client_id()
    actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
    LOGGER.info(f"[XID={msg.xid:08x}] DHCPDISCOVER from {context.client}|{client_id}")
    lease = self.acquire_lease(client_id, actual_server_id, msg)
    if not lease:
        LOGGER.info(
            f"[XID={msg.xid:08x}] No lease available for {context.client}|{client_id} at {actual_server_id} ignoring"
        )
        return
    resp = self._create_response(msg, lease, actual_server_id, _enum.DhcpMessageType.DHCPOFFER)
    self._filter_and_send(msg, resp, context, _enum.DhcpMessageType.DHCPOFFER)

handle_inform(msg, context)

Handle DHCPINFORM without requiring address allocation.

Source code in src/pydhcp/server.py

def handle_inform(self, msg: DhcpMessage, context: RequestContext) -> None:
    """Handle DHCPINFORM without requiring address allocation."""
    client_id = msg.client_id()
    actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
    LOGGER.info(f"[XID={msg.xid:08x}] DHCPINFORM from {context.client}|{client_id}")
    lease = self.acquire_lease(client_id, actual_server_id, msg)
    if not lease:
        lease = DhcpLease(
            _net.WILDCARD_IPv4,
            _inf,
            self.get_inform_options(actual_server_id, msg),
        )
    resp = self._create_response(msg, lease, actual_server_id, _enum.DhcpMessageType.DHCPACK)
    if DhcpOptionCode.IP_ADDRESS_LEASE_TIME in resp.options:
        del resp.options[DhcpOptionCode.IP_ADDRESS_LEASE_TIME]
    resp.yiaddr = _net.WILDCARD_IPv4
    self._filter_and_send(msg, resp, context, _enum.DhcpMessageType.DHCPACK)

handle_release(msg, context)

Handle DHCPRELEASE by releasing the client's lease through release_lease.

Source code in src/pydhcp/server.py

def handle_release(self, msg: DhcpMessage, context: RequestContext) -> None:
    """Handle DHCPRELEASE by releasing the client's lease through `release_lease`."""
    client_id = msg.client_id()
    actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
    LOGGER.info(f"[XID={msg.xid:08x}] DHCPRELEASE from {context.client}|{client_id}")
    self.release_lease(client_id, actual_server_id, msg)

handle_request(msg, context)

Handle DHCPREQUEST by ACKing or NAKing the lease returned from acquire_lease.

Source code in src/pydhcp/server.py

def handle_request(self, msg: DhcpMessage, context: RequestContext) -> None:
    """Handle DHCPREQUEST by ACKing or NAKing the lease returned from `acquire_lease`."""
    client_id = msg.client_id()
    actual_server_id = _ty.cast(_net.IPv4, context.interface.ip)
    LOGGER.info(f"[XID={msg.xid:08x}] DHCPREQUEST from {context.client}|{client_id}")
    lease = self.acquire_lease(client_id, actual_server_id, msg)
    if not lease:
        LOGGER.info(
            f"[XID={msg.xid:08x}] No lease available for {context.client}|{client_id} at {actual_server_id} ignoring"
        )
        return
    ip_req: _ty.Optional[_net.IPv4] = msg.options.get(
        DhcpOptionCode.REQUESTED_IP, decode=_type.IPv4Address
    )
    if not ip_req:
        ip_req = msg.ciaddr
    if ip_req == lease.ip:
        resp_ty = _enum.DhcpMessageType.DHCPACK
    else:
        resp_ty = _enum.DhcpMessageType.DHCPNAK
    resp = self._create_response(msg, lease, actual_server_id, resp_ty)
    self._filter_and_send(msg, resp, context, resp_ty)

release_lease(client_id, server_id, msg)

Release any lease associated with client_id.

Override this method when lease release needs to update an external store, quarantine declined addresses, or emit custom audit records.

Source code in src/pydhcp/server.py

def release_lease(self, client_id: str, server_id: _net.IPv4, msg: DhcpMessage) -> None:
    """Release any lease associated with `client_id`.

    Override this method when lease release needs to update an external store,
    quarantine declined addresses, or emit custom audit records.
    """
    if self.lease_backend.release(client_id):
        self.metrics.leases_released += 1

DhcpClient

pydhcp.client.DhcpClient

Bases: DhcpListener

Small DHCPv4 packet client for tests and troubleshooting.

The base client builds and sends DHCP client messages, then queues matching replies. It does not configure operating-system network interfaces.

Source code in src/pydhcp/client.py

class DhcpClient(DhcpListener):
    """Small DHCPv4 packet client for tests and troubleshooting.

    The base client builds and sends DHCP client messages, then queues matching
    replies. It does not configure operating-system network interfaces.
    """

    DEFAULT_PORTS = (_enum.DhcpPort.CLIENT,)

    def __init__(
        self,
        listen: ListenSpec = None,
        select_timeout: float | None = None,
        max_packet_size: int | None = None,
        per_interface: bool | None = None,
    ) -> None:
        super().__init__(
            listen=listen,
            select_timeout=select_timeout,
            max_packet_size=max_packet_size,
            per_interface=per_interface,
        )
        self._replies: _queue.Queue[tuple[DhcpMessage, RequestContext]] = _queue.Queue()
        self._pending_xids: set[int] = set()

    def build_discover(
        self,
        chaddr: bytes,
        *,
        xid: int | None = None,
        client_identifier: bytes | bytearray | None = None,
        parameter_request_list: _ty.Iterable[DhcpOptionCode] | None = None,
        broadcast: bool = True,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=broadcast)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPDISCOVER
        self._add_client_options(msg, client_identifier, parameter_request_list)
        return msg

    def build_request(
        self,
        chaddr: bytes,
        *,
        xid: int | None = None,
        requested_ip: _net.IPv4 | str | None = None,
        server_identifier: _net.IPv4 | str | None = None,
        ciaddr: _net.IPv4 | str | None = None,
        client_identifier: bytes | bytearray | None = None,
        parameter_request_list: _ty.Iterable[DhcpOptionCode] | None = None,
        broadcast: bool = True,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=broadcast)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPREQUEST
        if ciaddr is not None:
            msg.ciaddr = _net.IPv4(ciaddr)
        if requested_ip is not None:
            msg.options[DhcpOptionCode.REQUESTED_IP] = _net.IPv4(requested_ip)
        if server_identifier is not None:
            msg.options[DhcpOptionCode.SERVER_IDENTIFIER] = _net.IPv4(server_identifier)
        self._add_client_options(msg, client_identifier, parameter_request_list)
        return msg

    def build_inform(
        self,
        chaddr: bytes,
        *,
        ciaddr: _net.IPv4 | str,
        xid: int | None = None,
        client_identifier: bytes | bytearray | None = None,
        parameter_request_list: _ty.Iterable[DhcpOptionCode] | None = None,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=False)
        msg.ciaddr = _net.IPv4(ciaddr)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPINFORM
        self._add_client_options(msg, client_identifier, parameter_request_list)
        return msg

    def build_release(
        self,
        chaddr: bytes,
        *,
        ciaddr: _net.IPv4 | str,
        server_identifier: _net.IPv4 | str | None = None,
        xid: int | None = None,
        client_identifier: bytes | bytearray | None = None,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=False)
        msg.ciaddr = _net.IPv4(ciaddr)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPRELEASE
        if server_identifier is not None:
            msg.options[DhcpOptionCode.SERVER_IDENTIFIER] = _net.IPv4(server_identifier)
        self._add_client_options(msg, client_identifier, None)
        return msg

    def build_decline(
        self,
        chaddr: bytes,
        *,
        requested_ip: _net.IPv4 | str,
        server_identifier: _net.IPv4 | str | None = None,
        xid: int | None = None,
        client_identifier: bytes | bytearray | None = None,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=True)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPDECLINE
        msg.options[DhcpOptionCode.REQUESTED_IP] = _net.IPv4(requested_ip)
        if server_identifier is not None:
            msg.options[DhcpOptionCode.SERVER_IDENTIFIER] = _net.IPv4(server_identifier)
        self._add_client_options(msg, client_identifier, None)
        return msg

    def send(
        self,
        message: DhcpMessage,
        destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
        port: int = int(_enum.DhcpPort.SERVER),
    ) -> int:
        if not self._sockets:
            self.bind()
        transport = UdpTransport(self._sockets[0])
        data = message.encode(_const.DHCP_MIN_LEGAL_PACKET_SIZE)
        sent = transport.send(data, _net.IPv4(destination), int(port), message.chaddr)
        self._pending_xids.add(message.xid)
        self.metrics.packets_sent += 1
        return sent

    def _wait_for(
        self, xid: int, msg_type: _enum.DhcpMessageType, timeout: float
    ) -> DhcpMessage | None:
        deadline = _time.monotonic() + timeout
        while True:
            remaining = deadline - _time.monotonic()
            if remaining <= 0:
                return None
            reply = self.next_reply(timeout=remaining)
            if reply is None:
                return None
            msg, _context = reply
            if msg.xid != xid:
                continue
            if msg.options.get(DhcpOptionCode.DHCP_MESSAGE_TYPE) is not msg_type:
                continue
            return msg

    def discover_offer(
        self,
        chaddr: bytes,
        *,
        timeout: float = 2.0,
        retries: int = 2,
        destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
        port: int = int(_enum.DhcpPort.SERVER),
        **discover_kwargs: _ty.Any,
    ) -> DhcpMessage | None:
        """Broadcast DHCPDISCOVER and return the first DHCPOFFER, or None."""
        discover = self.build_discover(chaddr, **discover_kwargs)
        for _attempt in range(retries + 1):
            self.send(discover, destination, port)
            offer = self._wait_for(discover.xid, _enum.DhcpMessageType.DHCPOFFER, timeout)
            if offer is not None:
                return offer
        return None

    def dora(
        self,
        chaddr: bytes,
        *,
        timeout: float = 2.0,
        retries: int = 2,
        destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
        port: int = int(_enum.DhcpPort.SERVER),
        broadcast: bool = True,
        **discover_kwargs: _ty.Any,
    ) -> DhcpMessage | None:
        """Run a full DISCOVER/OFFER/REQUEST/ACK exchange and return the DHCPACK, or None."""
        offer = self.discover_offer(
            chaddr,
            timeout=timeout,
            retries=retries,
            destination=destination,
            port=port,
            broadcast=broadcast,
            **discover_kwargs,
        )
        if offer is None:
            return None
        server_identifier = offer.options.get(
            DhcpOptionCode.SERVER_IDENTIFIER, decode=_type.IPv4Address
        )
        request = self.build_request(
            chaddr,
            xid=offer.xid,
            requested_ip=offer.yiaddr,
            server_identifier=server_identifier,
            broadcast=broadcast,
        )
        for _attempt in range(retries + 1):
            self.send(request, destination, port)
            ack = self._wait_for(request.xid, _enum.DhcpMessageType.DHCPACK, timeout)
            if ack is not None:
                return ack
        return None

    def handle(self, msg: DhcpMessage, context: RequestContext) -> None:
        if msg.op != _enum.OpCode.BOOTREPLY:
            return
        if self._pending_xids and msg.xid not in self._pending_xids:
            return
        self._replies.put((msg, context))
        self.on_reply(msg, context)

    def on_reply(self, msg: DhcpMessage, context: RequestContext) -> None:
        """Hook called after a BOOTREPLY is accepted and queued."""

    def next_reply(self, timeout: float | None = None) -> tuple[DhcpMessage, RequestContext] | None:
        try:
            return self._replies.get(timeout=timeout)
        except _queue.Empty:
            return None

    def drain_replies(self) -> list[tuple[DhcpMessage, RequestContext]]:
        replies: list[tuple[DhcpMessage, RequestContext]] = []
        while True:
            try:
                replies.append(self._replies.get_nowait())
            except _queue.Empty:
                return replies

    def _base_request(
        self,
        chaddr: bytes,
        *,
        xid: int | None,
        broadcast: bool,
    ) -> DhcpMessage:
        return DhcpMessage(
            op=_enum.OpCode.BOOTREQUEST,
            htype=_enum.HardwareAddressType.ETHERNET,
            hlen=len(chaddr),
            hops=0,
            xid=_secrets.randbits(32) if xid is None else int(xid),
            secs=_dt.timedelta(seconds=0),
            flags=_enum.Flags.BROADCAST if broadcast else _enum.Flags.UNICAST,
            ciaddr=_net.WILDCARD_IPv4,
            yiaddr=_net.WILDCARD_IPv4,
            siaddr=_net.WILDCARD_IPv4,
            giaddr=_net.WILDCARD_IPv4,
            chaddr=bytes(chaddr),
            sname="",
            file="",
            options=DhcpOptions(),
        )

    def _add_client_options(
        self,
        msg: DhcpMessage,
        client_identifier: bytes | bytearray | None,
        parameter_request_list: _ty.Iterable[DhcpOptionCode] | None,
    ) -> None:
        if client_identifier is not None:
            msg.options[DhcpOptionCode.CLIENT_IDENTIFIER] = bytearray(client_identifier)
        if parameter_request_list is not None:
            msg.options[DhcpOptionCode.PARAMETER_REQUEST_LIST] = list(parameter_request_list)

discover_offer(chaddr, *, timeout=2.0, retries=2, destination=_net.IPv4('255.255.255.255'), port=int(_enum.DhcpPort.SERVER), **discover_kwargs)

Broadcast DHCPDISCOVER and return the first DHCPOFFER, or None.

Source code in src/pydhcp/client.py

def discover_offer(
    self,
    chaddr: bytes,
    *,
    timeout: float = 2.0,
    retries: int = 2,
    destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
    port: int = int(_enum.DhcpPort.SERVER),
    **discover_kwargs: _ty.Any,
) -> DhcpMessage | None:
    """Broadcast DHCPDISCOVER and return the first DHCPOFFER, or None."""
    discover = self.build_discover(chaddr, **discover_kwargs)
    for _attempt in range(retries + 1):
        self.send(discover, destination, port)
        offer = self._wait_for(discover.xid, _enum.DhcpMessageType.DHCPOFFER, timeout)
        if offer is not None:
            return offer
    return None

dora(chaddr, *, timeout=2.0, retries=2, destination=_net.IPv4('255.255.255.255'), port=int(_enum.DhcpPort.SERVER), broadcast=True, **discover_kwargs)

Run a full DISCOVER/OFFER/REQUEST/ACK exchange and return the DHCPACK, or None.

Source code in src/pydhcp/client.py

def dora(
    self,
    chaddr: bytes,
    *,
    timeout: float = 2.0,
    retries: int = 2,
    destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
    port: int = int(_enum.DhcpPort.SERVER),
    broadcast: bool = True,
    **discover_kwargs: _ty.Any,
) -> DhcpMessage | None:
    """Run a full DISCOVER/OFFER/REQUEST/ACK exchange and return the DHCPACK, or None."""
    offer = self.discover_offer(
        chaddr,
        timeout=timeout,
        retries=retries,
        destination=destination,
        port=port,
        broadcast=broadcast,
        **discover_kwargs,
    )
    if offer is None:
        return None
    server_identifier = offer.options.get(
        DhcpOptionCode.SERVER_IDENTIFIER, decode=_type.IPv4Address
    )
    request = self.build_request(
        chaddr,
        xid=offer.xid,
        requested_ip=offer.yiaddr,
        server_identifier=server_identifier,
        broadcast=broadcast,
    )
    for _attempt in range(retries + 1):
        self.send(request, destination, port)
        ack = self._wait_for(request.xid, _enum.DhcpMessageType.DHCPACK, timeout)
        if ack is not None:
            return ack
    return None

on_reply(msg, context)

Hook called after a BOOTREPLY is accepted and queued.

Source code in src/pydhcp/client.py

def on_reply(self, msg: DhcpMessage, context: RequestContext) -> None:
    """Hook called after a BOOTREPLY is accepted and queued."""

DhcpRelay

pydhcp.relay.DhcpRelay

Bases: DhcpListener

RFC 1542 / RFC 2131 4.1 / RFC 3046 DHCP relay agent.

Forwards client broadcasts (received on port 67, same as a server) to one or more configured upstream DHCP servers, stamping giaddr and incrementing hops. Forwards server replies (unicast back to this relay) on to the client, applying the same destination rules a server uses on its own client-facing side (broadcast flag, else yiaddr/ciaddr).

Source code in src/pydhcp/relay.py

class DhcpRelay(_Base):
    """RFC 1542 / RFC 2131 4.1 / RFC 3046 DHCP relay agent.

    Forwards client broadcasts (received on port 67, same as a server) to one or
    more configured upstream DHCP servers, stamping `giaddr` and incrementing
    `hops`. Forwards server replies (unicast back to this relay) on to the
    client, applying the same destination rules a server uses on its own
    client-facing side (broadcast flag, else `yiaddr`/`ciaddr`).
    """

    DEFAULT_PORTS = (_enum.DhcpPort.SERVER,)

    def __init__(
        self,
        listen: ListenSpec = None,
        server_addresses: _ty.Sequence[ServerAddress] = (),
        max_hops: int = 16,
        insert_relay_agent_info: bool = False,
        circuit_id: _ty.Optional[bytes] = None,
        remote_id: _ty.Optional[bytes] = None,
        select_timeout: _ty.Optional[float] = None,
        max_packet_size: _ty.Optional[int] = None,
        per_interface: bool | None = None,
    ) -> None:
        if not server_addresses:
            raise ValueError("DhcpRelay requires at least one server address")
        super().__init__(
            listen=listen,
            select_timeout=select_timeout,
            max_packet_size=max_packet_size,
            per_interface=per_interface,
        )
        self.server_addresses = [_normalize_server_address(a) for a in server_addresses]
        self.max_hops = max_hops
        self.insert_relay_agent_info = insert_relay_agent_info
        self.circuit_id = circuit_id
        self.remote_id = remote_id
        self._pending_clients: dict[int, _net.SocketAddress] = {}

    def handle(self, msg: DhcpMessage, context: RequestContext) -> None:
        if msg.op == _enum.OpCode.BOOTREQUEST:
            self._forward_to_servers(msg, context)
        elif msg.op == _enum.OpCode.BOOTREPLY:
            self._forward_to_client(msg, context)
        else:
            LOGGER.warning(f"[XID={msg.xid:08x}] Received message with unknown op {msg.op}, ignoring.")

    def _forward_to_servers(self, msg: DhcpMessage, context: RequestContext) -> None:
        forwarded = DhcpMessage(**msg.__dict__.copy())
        forwarded.hops = msg.hops + 1
        if forwarded.hops > self.max_hops:
            LOGGER.warning(
                f"[XID={msg.xid:08x}] Dropping request from {context.client}: hop count {forwarded.hops} exceeds max_hops={self.max_hops}"
            )
            self.metrics.packets_dropped_hop_limit += 1
            return

        if forwarded.giaddr == _net.WILDCARD_IPv4:
            forwarded.giaddr = _ty.cast(_net.IPv4, context.interface.ip)

        self._pending_clients[msg.xid] = context.client
        self._insert_relay_agent_info(forwarded)

        data = forwarded.encode()
        for server_ip, server_port in self.server_addresses:
            forwarded.log(context.interface.ip, _net.SocketAddress(server_ip, server_port), _logging.INFO)
            context.transport.send(data, server_ip, server_port, msg.chaddr)
            self.metrics.packets_sent += 1

    def _insert_relay_agent_info(self, msg: DhcpMessage) -> None:
        if not self.insert_relay_agent_info:
            return
        if DhcpOptionCode.RELAY_AGENT_INFORMATION in msg.options:
            LOGGER.warning(
                f"[XID={msg.xid:08x}] Request already carries RELAY_AGENT_INFORMATION, passing through unmodified."
            )
            return
        suboptions: list[tuple[int, bytes]] = []
        if self.circuit_id is not None:
            suboptions.append((1, self.circuit_id))
        if self.remote_id is not None:
            suboptions.append((2, self.remote_id))
        if suboptions:
            msg.options[DhcpOptionCode.RELAY_AGENT_INFORMATION] = _type.RelayAgentInformation(suboptions)

    def _forward_to_client(self, msg: DhcpMessage, context: RequestContext) -> None:
        original_client = self._pending_clients.pop(msg.xid, None)
        client_port = original_client.port if original_client is not None else int(_enum.DhcpPort.CLIENT)

        dest: _net.IPv4
        if msg.flags is _enum.Flags.BROADCAST:
            dest = _net.IPv4("255.255.255.255")
        elif msg.ciaddr != _net.WILDCARD_IPv4:
            dest = msg.ciaddr
        elif msg.yiaddr != _net.WILDCARD_IPv4:
            dest = msg.yiaddr
        else:
            dest = _net.IPv4("255.255.255.255")

        data = msg.encode()
        msg.log(context.interface.ip, _net.SocketAddress(dest, client_port), _logging.INFO)
        context.transport.send(data, dest, client_port, msg.chaddr)
        self.metrics.packets_sent += 1

DhcpCapture

pydhcp.capture.DhcpCapture

Bases: DhcpListener

Source code in src/pydhcp/capture.py

class DhcpCapture(DhcpListener):
    def __init__(
        self,
        listen: ListenSpec = None,
        packet_filter: str | CapturePredicate | None = None,
        sink: CaptureSink | None = None,
        hook: CaptureHook | None = None,
        hook_fail_fast: bool = False,
        select_timeout: float | None = None,
        max_packet_size: int | None = None,
        per_interface: bool | None = None,
    ) -> None:
        super().__init__(
            listen=listen,
            select_timeout=select_timeout,
            max_packet_size=max_packet_size,
            per_interface=per_interface,
        )
        self.packet_filter = (
            compile_capture_filter(packet_filter)
            if isinstance(packet_filter, str) or packet_filter is None
            else packet_filter
        )
        self.sink = sink
        self.hook = hook
        self.hook_fail_fast = hook_fail_fast
        self.accepted_count = 0

    def handle(self, msg: DhcpMessage, context: RequestContext) -> None:
        event = CaptureEvent(
            message=msg,
            context=context,
            captured_at=_dt.datetime.now(tz=_dt.timezone.utc),
        )
        if not self.packet_filter(event):
            return
        self.accepted_count += 1
        if self.sink is not None:
            self.sink(event)
        if self.hook is not None:
            try:
                self.hook(event)
            except Exception:
                LOGGER.exception("Capture hook failed")
                if self.hook_fail_fast:
                    raise

CaptureEvent

pydhcp.capture.CaptureEvent dataclass

Source code in src/pydhcp/capture.py

@_data.dataclass(frozen=True)
class CaptureEvent:
    message: DhcpMessage
    context: RequestContext
    captured_at: _dt.datetime

    @property
    def source(self) -> _net.SocketAddress:
        return self.context.client

    @property
    def destination(self) -> _net.SocketAddress:
        local_ip = self.context.local_ip or _ty.cast(_net.IPv4, self.context.interface.ip)
        return _net.SocketAddress(local_ip, 0)

    @property
    def message_type(self) -> str:
        value = self.message.options.get(DhcpOptionCode.DHCP_MESSAGE_TYPE)
        if value is None:
            return "UNKNOWN"
        return value.name if hasattr(value, "name") else str(value)

    @property
    def client_id(self) -> str:
        return self.message.client_id()

    @property
    def xid(self) -> str:
        return f"{self.message.xid:08X}"

    def format_filename(self, pattern: str, format: str) -> str:
        values = {
            "client_id": _sanitize_filename_value(self.client_id),
            "timestamp": _sanitize_filename_value(self.captured_at.strftime("%Y%m%dT%H%M%S.%fZ")),
            "msg_type": _sanitize_filename_value(self.message_type),
            "xid": _sanitize_filename_value(self.xid),
            "format": _sanitize_filename_value(format),
        }
        return pattern.format(**values)

AsyncDhcpServer

pydhcp.server.AsyncDhcpServer

Bases: AsyncDhcpListener, DhcpServer

Source code in src/pydhcp/server.py

class AsyncDhcpServer(_AsyncBase, DhcpServer):  # type: ignore[misc]
    DEFAULT_PORTS = (_enum.DhcpPort.SERVER,)

    def __init__(
        self,
        listen: ListenSpec = None,
        max_packet_size: _ty.Optional[int] = None,
        lease_backend: _ty.Optional[LeaseBackend] = None,
        per_interface: bool | None = None,
    ) -> None:
        _AsyncBase.__init__(self, listen=listen, max_packet_size=max_packet_size, per_interface=per_interface)
        from .lease import InMemoryLeaseBackend
        self.lease_backend = lease_backend or InMemoryLeaseBackend()

    def handle(self, msg: DhcpMessage, context: RequestContext) -> None:
        DhcpServer.handle(self, msg, context)

DhcpListener

pydhcp.listener.DhcpListener

Source code in src/pydhcp/listener.py

class DhcpListener:
    DEFAULT_PORTS: _ty.Sequence[int] = tuple(p.value for p in _enum.DhcpPort)

    def __init__(
        self,
        listen: ListenSpec = None,
        select_timeout: float | None = None,
        max_packet_size: int | None = _const.UDP_MAX_PACKET_SIZE,
        per_interface: bool | None = None,
    ) -> None:
        self._max_packet_size = max_packet_size or _const.UDP_MAX_PACKET_SIZE
        if listen is None:
            listen = "*"
        self._pktinfo = (
            per_interface is not True
            and hasattr(_socket.socket, "recvmsg")
            and hasattr(_socket, "IP_PKTINFO")
            and _listen_uses_wildcard(listen)
        )
        self._listen = _parselisteners(listen, self.DEFAULT_PORTS, expand_wildcard=not self._pktinfo)
        self._per_interface = per_interface
        self._sockets: list[_socket.socket] = []
        self._select_timeout = select_timeout or 1
        self._cancelleation_token: _thread.Event | None = None
        self.metrics = DhcpMetrics()

    def handle(self, msg: DhcpMessage, context: RequestContext) -> None:
        pass

    def bind(self) -> None:
        active = {_net.SocketAddress(socket): socket for socket in self._sockets}
        _listen = []
        for address in self._listen:
            _listen.append(address)
            if address in active:
                continue
            LOGGER.info(f"Listening on: {address}")
            try:
                socket = address.listen(
                    _socket.AF_INET,
                    _socket.SOCK_DGRAM,
                    _socket.IPPROTO_UDP,
                    options=[
                        _net.SocketOption(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1),
                        _net.SocketOption(_socket.SOL_SOCKET, _socket.SO_BROADCAST, 1),
                    ],
                )
                if self._pktinfo and address.ip == _net.WILDCARD_IPv4:
                    if IP_PKTINFO is not None:
                        socket.setsockopt(_socket.IPPROTO_IP, IP_PKTINFO, 1)
            except OSError as e:
                # netimps recognises the POSIX errnos *and* the Windows
                # WinError codes, which differ; the DHCP-specific suggestion
                # is appended rather than replacing the generic diagnosis.
                hint = _netimps.bind_error_hint(e, address.port)
                if hint is None:
                    raise
                if isinstance(e, PermissionError) or "permission" in hint.lower():
                    raise PermissionError(
                        f"{hint}. Try 6767 for testing."
                    ) from e
                if "in use" in hint:
                    raise OSError(
                        e.errno, f"{hint}; try port {address.port + 1000}."
                    ) from e
                raise OSError(e.errno, hint) from e
            self._sockets.append(socket)
        for address, socket in active.items():
            if address not in _listen:
                self._sockets.remove(socket)
                try:
                    socket.close()
                except:
                    pass

    def stop(self) -> None:
        if self._cancelleation_token is not None:
            self._cancelleation_token.set()

    def wait(self) -> None:
        while self._cancelleation_token is not None:
            self._cancelleation_token.wait(self._select_timeout)

    def start(self, cancellation_token: _thread.Event | None = None) -> _thread.Thread | None:
        if not self._cancelleation_token:
            thread = _thread.Thread(target=self.listen, args=())
            self._cancelleation_token = cancellation_token or _thread.Event()
            import signal

            def stop(*args: _ty.Any) -> None:
                self.stop()
                LOGGER.info("Stopped listening due to Ctrl-C")

            signal.signal(signal.SIGINT, stop)
            thread.start()
            return thread
        return None

    def listen(self) -> None:
        self.bind()
        listen = True
        rlist: list[_socket.socket]
        buffer = bytearray(self._max_packet_size)
        view = memoryview(buffer)
        if self._cancelleation_token is None:
            self._cancelleation_token = _thread.Event()
        try:
            while listen and not self._cancelleation_token.is_set():
                rlist, _, _ = _select.select(
                    list(self._sockets), [], [], self._select_timeout
                )
                if self._cancelleation_token.is_set():
                    break
                for socket in rlist:
                    try:
                        if self._pktinfo and hasattr(socket, "recvmsg"):
                            if CMSG_SPACE is None or IP_PKTINFO is None:
                                raise RuntimeError("packet info support unavailable")
                            data, ancdata, _, client_tuple = socket.recvmsg(
                                self._max_packet_size,
                                CMSG_SPACE(_struct.calcsize("=I4s4s")),
                            )
                            size = len(data)
                            local_ip = None
                            ifindex = None
                            for level, ctype, cdata in ancdata:
                                if level == _socket.IPPROTO_IP and ctype == IP_PKTINFO:
                                    ifindex, dst1, _ = _struct.unpack(
                                        "=I4s4s", cdata[: _struct.calcsize("=I4s4s")]
                                    )
                                    local_ip = _net.IPv4(_socket.inet_ntoa(dst1))
                                    break
                            msg = DhcpMessage.decode(memoryview(data))
                            client = _net.SocketAddress(*client_tuple)
                            interface = _resolve_interface(socket)
                            pkt_transport = PktInfoUdpTransport(socket)
                            context = RequestContext(
                                transport=pkt_transport,
                                interface=interface,
                                client=client,
                                client_mac=msg.chaddr,
                                ifindex=ifindex,
                                local_ip=local_ip,
                            )
                            pkt_transport.ifindex = ifindex
                            pkt_transport.local_ip = local_ip
                        else:
                            size, client_tuple = socket.recvfrom_into(view, self._max_packet_size)
                            client = _net.SocketAddress(*client_tuple)
                            msg = DhcpMessage.decode(view[:size])
                            interface = _resolve_interface(socket)
                            transport = UdpTransport(socket)
                            context = RequestContext(
                                transport=transport,
                                interface=interface,
                                client=client,
                                client_mac=msg.chaddr,
                            )
                        self.metrics.packets_received += 1
                        msg.log(client, _net.SocketAddress(socket), _logging.DEBUG)
                        self.handle(msg, context)
                    except Exception as e:
                        if isinstance(e, KeyboardInterrupt):
                            raise e
                        LOGGER.error(
                            f"Encounter error handling request: {e.__class__.__name__} | {e}"
                        )
        except KeyboardInterrupt:
            LOGGER.info("Stopped listening due to Ctrl-C")
            self._cancelleation_token.set()
        finally:
            self._cancelleation_token = None

AsyncDhcpListener

pydhcp.listener.AsyncDhcpListener

Source code in src/pydhcp/listener.py

class AsyncDhcpListener:
    DEFAULT_PORTS: _ty.Sequence[int] = tuple(p.value for p in _enum.DhcpPort)

    def __init__(
        self,
        listen: ListenSpec = None,
        max_packet_size: _ty.Optional[int] = None,
        per_interface: bool | None = None,
    ) -> None:
        self._max_packet_size = max_packet_size or _const.UDP_MAX_PACKET_SIZE
        if listen is None:
            listen = "*"
        self._pktinfo = False
        self._listen = _parselisteners(listen, self.DEFAULT_PORTS, expand_wildcard=True)
        self._per_interface = per_interface
        self._sockets: list[_socket.socket] = []
        self._transports: list[_asyncio.DatagramTransport] = []
        self.metrics = DhcpMetrics()

    def handle(self, msg: DhcpMessage, context: RequestContext) -> None:
        pass

    def bind(self) -> None:
        active = {_net.SocketAddress(socket): socket for socket in self._sockets}
        _listen = []
        for address in self._listen:
            _listen.append(address)
            if address in active:
                continue
            LOGGER.info(f"Listening on (async): {address}")
            socket = address.listen(
                _socket.AF_INET,
                _socket.SOCK_DGRAM,
                _socket.IPPROTO_UDP,
                options=[
                    _net.SocketOption(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1),
                    _net.SocketOption(_socket.SOL_SOCKET, _socket.SO_BROADCAST, 1),
                ],
            )
            self._sockets.append(socket)
        for address, socket in active.items():
            if address not in _listen:
                self._sockets.remove(socket)
                try:
                    socket.close()
                except Exception:
                    pass

    async def start(self) -> None:
        self.bind()
        loop = _asyncio.get_running_loop()
        for sock in self._sockets:
            transport, _ = await loop.create_datagram_endpoint(
                lambda: _DhcpDatagramProtocol(self, sock),
                sock=sock
            )
            self._transports.append(transport)

    async def stop(self) -> None:
        for transport in self._transports:
            transport.close()
        self._transports.clear()
        for sock in self._sockets:
            try:
                sock.close()
            except Exception:
                pass
        self._sockets.clear()

DhcpMetrics

Every DhcpListener (and therefore DhcpServer, DhcpClient) owns its own metrics: DhcpMetrics instance — counters are per-instance, not global, so running multiple listeners in one process (e.g. in tests) never cross-contaminates counts. Call .snapshot() for a plain dict[str, int].

pydhcp.metrics.DhcpMetrics

Source code in src/pydhcp/metrics.py

class DhcpMetrics:
    def __init__(self) -> None:
        self.packets_received = 0
        self.packets_sent = 0
        self.leases_allocated = 0
        self.leases_renewed = 0
        self.leases_released = 0
        self.packets_dropped_hop_limit = 0

    def reset(self) -> None:
        self.packets_received = 0
        self.packets_sent = 0
        self.leases_allocated = 0
        self.leases_renewed = 0
        self.leases_released = 0
        self.packets_dropped_hop_limit = 0

    def snapshot(self) -> _ty.Dict[str, int]:
        return {
            "packets_received": self.packets_received,
            "packets_sent": self.packets_sent,
            "leases_allocated": self.leases_allocated,
            "leases_renewed": self.leases_renewed,
            "leases_released": self.leases_released,
            "packets_dropped_hop_limit": self.packets_dropped_hop_limit,
        }

pydhcp.client

pydhcp.client

DhcpClient

Bases: DhcpListener

Small DHCPv4 packet client for tests and troubleshooting.

The base client builds and sends DHCP client messages, then queues matching replies. It does not configure operating-system network interfaces.

Source code in src/pydhcp/client.py

class DhcpClient(DhcpListener):
    """Small DHCPv4 packet client for tests and troubleshooting.

    The base client builds and sends DHCP client messages, then queues matching
    replies. It does not configure operating-system network interfaces.
    """

    DEFAULT_PORTS = (_enum.DhcpPort.CLIENT,)

    def __init__(
        self,
        listen: ListenSpec = None,
        select_timeout: float | None = None,
        max_packet_size: int | None = None,
        per_interface: bool | None = None,
    ) -> None:
        super().__init__(
            listen=listen,
            select_timeout=select_timeout,
            max_packet_size=max_packet_size,
            per_interface=per_interface,
        )
        self._replies: _queue.Queue[tuple[DhcpMessage, RequestContext]] = _queue.Queue()
        self._pending_xids: set[int] = set()

    def build_discover(
        self,
        chaddr: bytes,
        *,
        xid: int | None = None,
        client_identifier: bytes | bytearray | None = None,
        parameter_request_list: _ty.Iterable[DhcpOptionCode] | None = None,
        broadcast: bool = True,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=broadcast)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPDISCOVER
        self._add_client_options(msg, client_identifier, parameter_request_list)
        return msg

    def build_request(
        self,
        chaddr: bytes,
        *,
        xid: int | None = None,
        requested_ip: _net.IPv4 | str | None = None,
        server_identifier: _net.IPv4 | str | None = None,
        ciaddr: _net.IPv4 | str | None = None,
        client_identifier: bytes | bytearray | None = None,
        parameter_request_list: _ty.Iterable[DhcpOptionCode] | None = None,
        broadcast: bool = True,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=broadcast)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPREQUEST
        if ciaddr is not None:
            msg.ciaddr = _net.IPv4(ciaddr)
        if requested_ip is not None:
            msg.options[DhcpOptionCode.REQUESTED_IP] = _net.IPv4(requested_ip)
        if server_identifier is not None:
            msg.options[DhcpOptionCode.SERVER_IDENTIFIER] = _net.IPv4(server_identifier)
        self._add_client_options(msg, client_identifier, parameter_request_list)
        return msg

    def build_inform(
        self,
        chaddr: bytes,
        *,
        ciaddr: _net.IPv4 | str,
        xid: int | None = None,
        client_identifier: bytes | bytearray | None = None,
        parameter_request_list: _ty.Iterable[DhcpOptionCode] | None = None,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=False)
        msg.ciaddr = _net.IPv4(ciaddr)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPINFORM
        self._add_client_options(msg, client_identifier, parameter_request_list)
        return msg

    def build_release(
        self,
        chaddr: bytes,
        *,
        ciaddr: _net.IPv4 | str,
        server_identifier: _net.IPv4 | str | None = None,
        xid: int | None = None,
        client_identifier: bytes | bytearray | None = None,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=False)
        msg.ciaddr = _net.IPv4(ciaddr)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPRELEASE
        if server_identifier is not None:
            msg.options[DhcpOptionCode.SERVER_IDENTIFIER] = _net.IPv4(server_identifier)
        self._add_client_options(msg, client_identifier, None)
        return msg

    def build_decline(
        self,
        chaddr: bytes,
        *,
        requested_ip: _net.IPv4 | str,
        server_identifier: _net.IPv4 | str | None = None,
        xid: int | None = None,
        client_identifier: bytes | bytearray | None = None,
    ) -> DhcpMessage:
        msg = self._base_request(chaddr, xid=xid, broadcast=True)
        msg.options[DhcpOptionCode.DHCP_MESSAGE_TYPE] = _enum.DhcpMessageType.DHCPDECLINE
        msg.options[DhcpOptionCode.REQUESTED_IP] = _net.IPv4(requested_ip)
        if server_identifier is not None:
            msg.options[DhcpOptionCode.SERVER_IDENTIFIER] = _net.IPv4(server_identifier)
        self._add_client_options(msg, client_identifier, None)
        return msg

    def send(
        self,
        message: DhcpMessage,
        destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
        port: int = int(_enum.DhcpPort.SERVER),
    ) -> int:
        if not self._sockets:
            self.bind()
        transport = UdpTransport(self._sockets[0])
        data = message.encode(_const.DHCP_MIN_LEGAL_PACKET_SIZE)
        sent = transport.send(data, _net.IPv4(destination), int(port), message.chaddr)
        self._pending_xids.add(message.xid)
        self.metrics.packets_sent += 1
        return sent

    def _wait_for(
        self, xid: int, msg_type: _enum.DhcpMessageType, timeout: float
    ) -> DhcpMessage | None:
        deadline = _time.monotonic() + timeout
        while True:
            remaining = deadline - _time.monotonic()
            if remaining <= 0:
                return None
            reply = self.next_reply(timeout=remaining)
            if reply is None:
                return None
            msg, _context = reply
            if msg.xid != xid:
                continue
            if msg.options.get(DhcpOptionCode.DHCP_MESSAGE_TYPE) is not msg_type:
                continue
            return msg

    def discover_offer(
        self,
        chaddr: bytes,
        *,
        timeout: float = 2.0,
        retries: int = 2,
        destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
        port: int = int(_enum.DhcpPort.SERVER),
        **discover_kwargs: _ty.Any,
    ) -> DhcpMessage | None:
        """Broadcast DHCPDISCOVER and return the first DHCPOFFER, or None."""
        discover = self.build_discover(chaddr, **discover_kwargs)
        for _attempt in range(retries + 1):
            self.send(discover, destination, port)
            offer = self._wait_for(discover.xid, _enum.DhcpMessageType.DHCPOFFER, timeout)
            if offer is not None:
                return offer
        return None

    def dora(
        self,
        chaddr: bytes,
        *,
        timeout: float = 2.0,
        retries: int = 2,
        destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
        port: int = int(_enum.DhcpPort.SERVER),
        broadcast: bool = True,
        **discover_kwargs: _ty.Any,
    ) -> DhcpMessage | None:
        """Run a full DISCOVER/OFFER/REQUEST/ACK exchange and return the DHCPACK, or None."""
        offer = self.discover_offer(
            chaddr,
            timeout=timeout,
            retries=retries,
            destination=destination,
            port=port,
            broadcast=broadcast,
            **discover_kwargs,
        )
        if offer is None:
            return None
        server_identifier = offer.options.get(
            DhcpOptionCode.SERVER_IDENTIFIER, decode=_type.IPv4Address
        )
        request = self.build_request(
            chaddr,
            xid=offer.xid,
            requested_ip=offer.yiaddr,
            server_identifier=server_identifier,
            broadcast=broadcast,
        )
        for _attempt in range(retries + 1):
            self.send(request, destination, port)
            ack = self._wait_for(request.xid, _enum.DhcpMessageType.DHCPACK, timeout)
            if ack is not None:
                return ack
        return None

    def handle(self, msg: DhcpMessage, context: RequestContext) -> None:
        if msg.op != _enum.OpCode.BOOTREPLY:
            return
        if self._pending_xids and msg.xid not in self._pending_xids:
            return
        self._replies.put((msg, context))
        self.on_reply(msg, context)

    def on_reply(self, msg: DhcpMessage, context: RequestContext) -> None:
        """Hook called after a BOOTREPLY is accepted and queued."""

    def next_reply(self, timeout: float | None = None) -> tuple[DhcpMessage, RequestContext] | None:
        try:
            return self._replies.get(timeout=timeout)
        except _queue.Empty:
            return None

    def drain_replies(self) -> list[tuple[DhcpMessage, RequestContext]]:
        replies: list[tuple[DhcpMessage, RequestContext]] = []
        while True:
            try:
                replies.append(self._replies.get_nowait())
            except _queue.Empty:
                return replies

    def _base_request(
        self,
        chaddr: bytes,
        *,
        xid: int | None,
        broadcast: bool,
    ) -> DhcpMessage:
        return DhcpMessage(
            op=_enum.OpCode.BOOTREQUEST,
            htype=_enum.HardwareAddressType.ETHERNET,
            hlen=len(chaddr),
            hops=0,
            xid=_secrets.randbits(32) if xid is None else int(xid),
            secs=_dt.timedelta(seconds=0),
            flags=_enum.Flags.BROADCAST if broadcast else _enum.Flags.UNICAST,
            ciaddr=_net.WILDCARD_IPv4,
            yiaddr=_net.WILDCARD_IPv4,
            siaddr=_net.WILDCARD_IPv4,
            giaddr=_net.WILDCARD_IPv4,
            chaddr=bytes(chaddr),
            sname="",
            file="",
            options=DhcpOptions(),
        )

    def _add_client_options(
        self,
        msg: DhcpMessage,
        client_identifier: bytes | bytearray | None,
        parameter_request_list: _ty.Iterable[DhcpOptionCode] | None,
    ) -> None:
        if client_identifier is not None:
            msg.options[DhcpOptionCode.CLIENT_IDENTIFIER] = bytearray(client_identifier)
        if parameter_request_list is not None:
            msg.options[DhcpOptionCode.PARAMETER_REQUEST_LIST] = list(parameter_request_list)

discover_offer(chaddr, *, timeout=2.0, retries=2, destination=_net.IPv4('255.255.255.255'), port=int(_enum.DhcpPort.SERVER), **discover_kwargs)

Broadcast DHCPDISCOVER and return the first DHCPOFFER, or None.

Source code in src/pydhcp/client.py

def discover_offer(
    self,
    chaddr: bytes,
    *,
    timeout: float = 2.0,
    retries: int = 2,
    destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
    port: int = int(_enum.DhcpPort.SERVER),
    **discover_kwargs: _ty.Any,
) -> DhcpMessage | None:
    """Broadcast DHCPDISCOVER and return the first DHCPOFFER, or None."""
    discover = self.build_discover(chaddr, **discover_kwargs)
    for _attempt in range(retries + 1):
        self.send(discover, destination, port)
        offer = self._wait_for(discover.xid, _enum.DhcpMessageType.DHCPOFFER, timeout)
        if offer is not None:
            return offer
    return None

dora(chaddr, *, timeout=2.0, retries=2, destination=_net.IPv4('255.255.255.255'), port=int(_enum.DhcpPort.SERVER), broadcast=True, **discover_kwargs)

Run a full DISCOVER/OFFER/REQUEST/ACK exchange and return the DHCPACK, or None.

Source code in src/pydhcp/client.py

def dora(
    self,
    chaddr: bytes,
    *,
    timeout: float = 2.0,
    retries: int = 2,
    destination: _net.IPv4 | str = _net.IPv4("255.255.255.255"),
    port: int = int(_enum.DhcpPort.SERVER),
    broadcast: bool = True,
    **discover_kwargs: _ty.Any,
) -> DhcpMessage | None:
    """Run a full DISCOVER/OFFER/REQUEST/ACK exchange and return the DHCPACK, or None."""
    offer = self.discover_offer(
        chaddr,
        timeout=timeout,
        retries=retries,
        destination=destination,
        port=port,
        broadcast=broadcast,
        **discover_kwargs,
    )
    if offer is None:
        return None
    server_identifier = offer.options.get(
        DhcpOptionCode.SERVER_IDENTIFIER, decode=_type.IPv4Address
    )
    request = self.build_request(
        chaddr,
        xid=offer.xid,
        requested_ip=offer.yiaddr,
        server_identifier=server_identifier,
        broadcast=broadcast,
    )
    for _attempt in range(retries + 1):
        self.send(request, destination, port)
        ack = self._wait_for(request.xid, _enum.DhcpMessageType.DHCPACK, timeout)
        if ack is not None:
            return ack
    return None

on_reply(msg, context)

Hook called after a BOOTREPLY is accepted and queued.

Source code in src/pydhcp/client.py

def on_reply(self, msg: DhcpMessage, context: RequestContext) -> None:
    """Hook called after a BOOTREPLY is accepted and queued."""

pydhcp.capture

pydhcp.capture

pydhcp.packet

pydhcp.packet

DhcpMessage dataclass

Source code in src/pydhcp/packet/message.py

@_data.dataclass
class DhcpMessage:
    MIN_LEGAL_SIZE = _const.DHCP_MIN_LEGAL_PACKET_SIZE - _const.UDP_MIN_PACKET_SIZE
    MAGIC_COOKIE: _ty.ClassVar[bytes] = 0x63825363.to_bytes(4, "big")
    """The first four octets of the 'options' field of the DHCP message decimal values: 99, 130, 83 and 99"""

    op: _enum.OpCode  # One byte
    """Message op code / message type"""
    htype: _enum.HardwareAddressType
    """Hardware address type, see ARP section in "Assigned Numbers" RFC"""
    hlen: int  # One byte
    """Hardware address length"""
    hops: int  # One byte
    """Client sets to zero, optionally used by relay agents when booting via a relay agent."""
    xid: int  # 4 bytes
    """Transaction ID, a random number chosen by the
    client, used by the client and server to associate
    messages and responses between a client and a server."""
    secs: _dt.timedelta  # 2 bytes
    """Filled in by client, seconds elapsed since client
    began address acquisition or renewal process."""
    flags: _enum.Flags  # 2 bytes
    """Only use for the BROADCAST flag in clients"""
    ciaddr: _net.IPv4
    """Client IP address; only filled in if client is in
    BOUND, RENEW or REBINDING state and can respond
    to ARP requests."""
    yiaddr: _net.IPv4
    """'your' (client) IP address."""
    siaddr: _net.IPv4
    """IP address of next server to use in bootstrap;
    returned in DHCPOFFER, DHCPACK by server."""
    giaddr: _net.IPv4
    """Relay agent IP address, used in booting via a
    relay agent."""
    chaddr: bytes  # 16 bytes
    """Client hardware address."""
    sname: str  # 64 bytes
    """Optional server host name, null terminated string."""
    file: str  # 128 bytes
    """Boot file name, null terminated string; "generic"
    name or null in DHCPDISCOVER, fully qualified
    directory-path name in DHCPOFFER."""
    options: DhcpOptions
    """Optional parameters field."""

    def to_mapping(self) -> dict[str, _ty.Any]:
        options: dict[str, _ty.Any] = {}
        for code, value in self.options._options.items():
            try:
                code_obj = self.options._codemap.from_code(code)
                key = code_obj.label()
                option_type = code_obj.get_type()
                decoded = option_type._dhcp_decode(value)
                option_value = decoded.__json__() if isinstance(decoded, _type.DhcpOptionType) else decoded
            except Exception:
                key = str(code)
                option_value = _type.Bytes(value).__json__()
            options[key] = _enum_name(option_value)

        return {
            "op": self.op.name,
            "htype": self.htype.name,
            "hlen": self.hlen,
            "hops": self.hops,
            "xid": self.xid,
            "secs": int(self.secs.total_seconds()),
            "flags": self.flags.name,
            "ciaddr": str(self.ciaddr),
            "yiaddr": str(self.yiaddr),
            "siaddr": str(self.siaddr),
            "giaddr": str(self.giaddr),
            "chaddr": self.chaddr.hex(":").upper(),
            "sname": self.sname,
            "file": self.file,
            "options": options,
        }

    @classmethod
    def from_mapping(cls, data: _ty.Mapping[str, _ty.Any]) -> "DhcpMessage":
        options = DhcpOptions()
        raw_options = data.get("options", {})
        if not isinstance(raw_options, _ty.Mapping):
            raise TypeError("options must be a mapping")

        for raw_code, raw_value in raw_options.items():
            code = _coerce_option_code(raw_code, options._codemap)
            try:
                code_obj = options._codemap.from_code(code)
                option_type = code_obj.get_type()
                value = _coerce_option_value(option_type, raw_value)
                options[code] = value
            except Exception:
                if isinstance(raw_value, str):
                    raw_bytes = bytearray.fromhex(_strip_hex_text(raw_value))
                elif isinstance(raw_value, (bytes, bytearray, memoryview)):
                    raw_bytes = bytearray(raw_value)
                else:
                    raise TypeError(f"Unsupported value for unknown option {raw_code!r}") from None
                options[code] = raw_bytes

        return cls(
            op=_coerce_enum_value(_enum.OpCode, data["op"]),
            htype=_coerce_enum_value(_enum.HardwareAddressType, data["htype"]),
            hlen=_coerce_int(data["hlen"]),
            hops=_coerce_int(data["hops"]),
            xid=_coerce_int(data["xid"]),
            secs=_dt.timedelta(seconds=_coerce_int(data["secs"])),
            flags=_coerce_enum_value(_enum.Flags, data["flags"]),
            ciaddr=_net.IPv4(data["ciaddr"]),
            yiaddr=_net.IPv4(data["yiaddr"]),
            siaddr=_net.IPv4(data["siaddr"]),
            giaddr=_net.IPv4(data["giaddr"]),
            chaddr=_coerce_chaddr(data["chaddr"]),
            sname=str(data["sname"]),
            file=str(data["file"]),
            options=options,
        )

    @classmethod
    def decode(cls, data: _ty.Union[bytes, bytearray, memoryview]) -> "DhcpMessage":
        if not isinstance(data, memoryview):
            data = memoryview(data)
        if len(data) < _FIXED_HEADER_SIZE:
            raise ValueError(
                f"Packet is too short for DHCP fixed header: got {len(data)} bytes, need at least {_FIXED_HEADER_SIZE}"
            )
        if len(data) < _MAGIC_COOKIE_END:
            raise ValueError(
                f"Packet is too short for DHCP magic cookie at offset 236: got {len(data)} bytes, need at least {_MAGIC_COOKIE_END}"
            )
        (
            op,
            htype,
            hlen,
            hops,
            xid,
            secs,
            flags,
            ciaddr,
            yiaddr,
            siaddr,
            giaddr,
        ) = _HEADER_STRUCT.unpack_from(data, 0)
        if hlen > 16:
            raise ValueError(f"Hardware address length {hlen} exceeds maximum of 16")
        op = _enum.OpCode(op)
        try:
            htype = _enum.HardwareAddressType(htype)
        except ValueError:
            LOGGER.warning(f"Unknown hardware type {htype}, using ETHERNET")
            htype = _enum.HardwareAddressType.ETHERNET
        secs = _dt.timedelta(seconds=secs)
        flags = _enum.Flags(flags & _enum.Flags.BROADCAST.value)
        ciaddr = _net.IPv4(ciaddr)
        yiaddr = _net.IPv4(yiaddr)
        siaddr = _net.IPv4(siaddr)
        giaddr = _net.IPv4(giaddr)
        chaddr = data[28 : 28 + hlen].tobytes()
        sname_data = data[44:108]
        file_data = data[108:236]

        if cls.MAGIC_COOKIE != data[236:240]:
            raise ValueError(
                f"Invalid magic cookie at offset 236: expected {cls.MAGIC_COOKIE.hex()}, got {data[236:240].hex()}"
            )

        options = DhcpOptions()
        remaining_opts = options.decode(data[240:], base_offset=240)
        if remaining_opts and remaining_opts[0] != 255:
            raise ValueError(f"Bad options terminator: expected 255 (END), got {remaining_opts[0]}")

        overload = options.get(
            DhcpOptionCode.OPTION_OVERLOAD,
            default=_type.OptionOverload.NONE,
            decode=_type.OptionOverload,
        )

        # rfc3396 order
        if overload is not None and bool(overload.value & _type.OptionOverload.FILE.value):
            options.decode(file_data, base_offset=108)
            file_raw: _ty.Optional[memoryview] = None
        else:
            file_raw = file_data

        if overload is not None and bool(overload.value & _type.OptionOverload.SNAME.value):
            options.decode(sname_data, base_offset=44)
            sname_raw: _ty.Optional[memoryview] = None
        else:
            sname_raw = sname_data

        sname_str: str = ""
        if sname_raw is not None:
            sname_str = sname_raw.tobytes().split(_NULL, 1)[0].decode()
        else:
            tftp_val = options.get(
                DhcpOptionCode.TFTP_SERVER, default="", decode=_type.String
            )
            if tftp_val is not None:
                sname_str = str(tftp_val)

        file_str: str = ""
        if file_raw is not None:
            file_str = file_raw.tobytes().split(_NULL, 1)[0].decode()
        else:
            bootfile_val = options.get(
                DhcpOptionCode.BOOTFILE_NAME, default="", decode=_type.String
            )
            if bootfile_val is not None:
                file_str = str(bootfile_val)

        # opts -> file -> sname

        return DhcpMessage(
            op,
            htype,
            hlen,
            hops,
            xid,
            secs,
            flags,
            ciaddr,
            yiaddr,
            siaddr,
            giaddr,
            chaddr,
            sname_str,
            file_str,
            options,
        )

    def encode(self, max_packetsize: int = _const.DHCP_MIN_LEGAL_PACKET_SIZE) -> bytearray:
        max_packetsize = int(max_packetsize or _const.DHCP_MIN_LEGAL_PACKET_SIZE)
        max_options_field_size = max_packetsize - 264 - len(self.MAGIC_COOKIE)
        if max_options_field_size < 0:
            raise ValueError(f"{max_packetsize} is too small for a DHCP packet")

        options = DhcpOptions(self.options._codemap)
        options._options = _ty.OrderedDict(
            (code, bytearray(value)) for code, value in self.options._options.items()
        )
        sname_bytes: _ty.Union[bytes, bytearray] = self.sname.encode()
        file_bytes: _ty.Union[bytes, bytearray] = self.file.encode()
        options_field: _ty.Union[bytes, bytearray] = options.encode()
        if len(options_field) > max_options_field_size + 128 + 64:
            raise OverflowError("DHCP options exceed maximum packet size")
        elif len(options_field) > max_options_field_size + 128:
            if self.file and DhcpOptionCode.BOOTFILE_NAME not in options:
                options[DhcpOptionCode.BOOTFILE_NAME] = self.file
                options._options.move_to_end(
                    int(DhcpOptionCode.BOOTFILE_NAME), False
                )
            if self.sname and DhcpOptionCode.TFTP_SERVER not in options:
                options[DhcpOptionCode.TFTP_SERVER] = self.sname
                options._options.move_to_end(
                    int(DhcpOptionCode.TFTP_SERVER), False
                )
            overload = _type.OptionOverload.BOTH
        elif len(options_field) > max_options_field_size + 64:
            if self.file and DhcpOptionCode.BOOTFILE_NAME not in options:
                options[DhcpOptionCode.BOOTFILE_NAME] = self.file
                options._options.move_to_end(
                    int(DhcpOptionCode.BOOTFILE_NAME), False
                )
            overload = _type.OptionOverload.FILE
        elif len(options_field) > max_options_field_size:
            if self.sname and DhcpOptionCode.TFTP_SERVER not in options:
                options[DhcpOptionCode.TFTP_SERVER] = self.sname
                options._options.move_to_end(
                    int(DhcpOptionCode.TFTP_SERVER), False
                )
            overload = _type.OptionOverload.SNAME
        else:
            overload = _type.OptionOverload.NONE

        try:
            options._options.move_to_end(
                int(DhcpOptionCode.DHCP_MESSAGE_TYPE), False
            )
        except KeyError:
            pass

        if overload is not _type.OptionOverload.NONE:
            options._options[
                int(DhcpOptionCode.OPTION_OVERLOAD)
            ] = bytearray([overload.value])
            options._options.move_to_end(
                int(DhcpOptionCode.OPTION_OVERLOAD), False
            )
            options_field, leftover = options.partial_encode(max_options_field_size)

            if bool(overload.value & _type.OptionOverload.FILE.value) and leftover is not None:
                file_bytes, leftover = leftover.partial_encode(128)
            if bool(overload.value & _type.OptionOverload.SNAME.value) and leftover is not None:
                sname_bytes, leftover = leftover.partial_encode(64)

        data = bytearray(28)
        _HEADER_STRUCT.pack_into(
            data,
            0,
            self.op.value,
            int(self.htype),
            self.hlen,
            self.hops,
            self.xid,
            min(0xFFFF, max(0, int(self.secs.total_seconds()))),
            self.flags.value,
            int(self.ciaddr),
            int(self.yiaddr),
            int(self.siaddr),
            int(self.giaddr),
        )
        data.extend(self.chaddr.ljust(16, b"\x00")[:16])
        data.extend(sname_bytes.ljust(64, b"\x00")[:64])
        data.extend(file_bytes.ljust(128, b"\x00")[:128])
        data.extend(self.MAGIC_COOKIE)
        data.extend(options_field)
        return data

    def client_id(self, func: _ty.Optional[_ty.Callable[["DhcpMessage"], bytearray]] = None) -> str:
        cid = self.options.get(DhcpOptionCode.CLIENT_IDENTIFIER, decode=False)
        if not cid:
            if func:
                cid = func(self)
            if not cid:
                cid = bytearray([self.htype.value])
                cid.extend(self.chaddr)
        return cid.hex(":").upper()

    def dumps(self, codemap: _ty.Optional[type[BaseDhcpOptionCode]] = None) -> str:
        lines = []
        for name, value in [
            ("OP", self.op.name),
            ("Time Since Boot", str(self.secs)),
            ("Hops", str(self.hops)),
            ("Transaction ID", str(self.xid)),
            ("Flags", self.flags.name),
            ("Client Current Address", str(self.ciaddr)),
            ("Allocated Address", str(self.yiaddr)),
            ("Gateway Address", str(self.giaddr)),
            ("Hardware Address", f"{self.htype.name}({self.htype.dumps(self.chaddr)})"),
            ("Server Address", str(self.siaddr)),
            ("Next Server", str(self.siaddr)),
            ("Server Host Name", self.sname),
            ("Bootfile", self.file),
        ]:
            lines.append(f"{name: <40}: {value}")
        lines.append(f"OPTIONS:")
        for code, opt_val in self.options.items(decoded=codemap or True):
            if isinstance(opt_val, list):
                decoded_str = "\n".join([repr(i) for i in opt_val])
            else:
                decoded_str = repr(opt_val)
            decoded_lines = decoded_str.splitlines()
            SPACE = " " * 42
            if decoded_lines:
                first = _tw.fill(
                    decoded_lines[0], width=100, initial_indent="", subsequent_indent=SPACE
                )
            else:
                first = ""
            lines.append(f"  {repr(code): <38}: {first}")
            for line in decoded_lines[1:]:
                lines.append(
                    _tw.fill(
                        line, width=100, initial_indent=SPACE, subsequent_indent=SPACE
                    )
                )

        return "\n".join(lines)

    def log_str(self, src: _ty.Any, dst: _ty.Any) -> str:
        return (
            f"{self.op.name} XID={self.xid:08X} Src: {src} Dst: {dst}\n"
            f"{self.dumps()}"
        )

    def __contains__(self, __key: object) -> bool:
        return self.options.__contains__(__key)

    def log(self, src: _ty.Any, dst: _ty.Any, level: int) -> None:
        header = f"{'#' * 10} {self.op.name} XID={self.xid:08X} Src: {src} Dst: {dst} {'#' * 10}"
        LOGGER.log(level, f"\n{header}\n{self.dumps()}\n{'#' * len(header)}")

MAGIC_COOKIE = 1669485411.to_bytes(4, 'big') class-attribute

The first four octets of the 'options' field of the DHCP message decimal values: 99, 130, 83 and 99

chaddr instance-attribute

Client hardware address.

ciaddr instance-attribute

Client IP address; only filled in if client is in BOUND, RENEW or REBINDING state and can respond to ARP requests.

file instance-attribute

Boot file name, null terminated string; "generic" name or null in DHCPDISCOVER, fully qualified directory-path name in DHCPOFFER.

flags instance-attribute

Only use for the BROADCAST flag in clients

giaddr instance-attribute

Relay agent IP address, used in booting via a relay agent.

hlen instance-attribute

Hardware address length

hops instance-attribute

Client sets to zero, optionally used by relay agents when booting via a relay agent.

htype instance-attribute

Hardware address type, see ARP section in "Assigned Numbers" RFC

op instance-attribute

Message op code / message type

options instance-attribute

Optional parameters field.

secs instance-attribute

Filled in by client, seconds elapsed since client began address acquisition or renewal process.

siaddr instance-attribute

IP address of next server to use in bootstrap; returned in DHCPOFFER, DHCPACK by server.

sname instance-attribute

Optional server host name, null terminated string.

xid instance-attribute

Transaction ID, a random number chosen by the client, used by the client and server to associate messages and responses between a client and a server.

yiaddr instance-attribute

'your' (client) IP address.

DhcpMessageType

Bases: DhcpOptionType, IntEnum

DHCP message types

Source code in src/pydhcp/packet/enums.py

class DhcpMessageType(DhcpOptionType, _enum.IntEnum):
    """DHCP message types"""

    @classmethod
    def _dhcp_read(cls, option: memoryview) -> tuple[Self, int]:
        option_part = option[:1]
        if len(option_part) != 1:
            raise ValueError()
        return cls(option_part[0]), 1

    def _dhcp_write(self, data: bytearray) -> int:
        data.append(self.value)
        return 1

    @classmethod
    def _dhcp_len_hint(cls) -> int | None:
        return 1

    def __repr__(self) -> str:
        return self.name

    def __str__(self) -> str:
        return self.name

    DHCPDISCOVER = 1
    """ Client broadcast to locate available servers."""
    DHCPOFFER = 2
    """ Server to client in response to DHCPDISCOVER with offer of configuration parameters."""
    DHCPREQUEST = 3
    """Client message to servers either (a) requesting
    offered parameters from one server and implicitly
    declining offers from all others, (b) confirming
    correctness of previously allocated address after,
    e.g., system reboot, or (c) extending the lease on a
    particular network address."""
    DHCPDECLINE = 4
    """Client to server indicating network address is already
    in use."""
    DHCPACK = 5
    """Server to client with configuration parameters,
    including committed network address."""
    DHCPNAK = 6
    """Server to client indicating client's notion of network
    address is incorrect (e.g., client has moved to new
    subnet) or client's lease as expired"""
    DHCPRELEASE = 7
    """Client to server relinquishing network address and
    cancelling remaining lease."""
    DHCPINFORM = 8
    """ Client to server, asking only for local configuration
    parameters; client already has externally configured
    network address."""

    DHCPFORCERENEW = 9
    """Forces the client to the RENEW state. """
    DHCPLEASEQUERY = 10
    """The DHCPLEASEQUERY message is a new DHCP message type transmitted
   from a DHCP relay agent to a DHCP server.  A DHCPLEASEQUERY-aware
   relay agent sends the DHCPLEASEQUERY message when it needs to know
   the location of an IP endpoint.  The DHCPLEASEQUERY-aware DHCP server
   replies with a DHCPLEASEUNASSIGNED, DHCPLEASEACTIVE, or
   DHCPLEASEUNKNOWN message. """
    DHCPLEASEUNASSIGNED = 11
    """The DHCPLEASEUNASSIGNED is similar to a DHCPLEASEACTIVE message, but
   indicates that there is no currently active lease on the resultant IP
   address but that this DHCP server is authoritative for this IP
   address."""
    DHCPLEASEUNKNOWN = 12
    """The DHCPLEASEUNKNOWN message indicates that the DHCP server
   has no knowledge of the information specified in the query (e.g., IP
   address, MAC address, or Client-identifier option)."""
    DHCPLEASEACTIVE = 13
    """The DHCPLEASEACTIVE response to a
   DHCPLEASEQUERY message allows the relay agent to determine the IP
   endpoint location and the remaining duration of the IP address lease."""
    DHCPBULKLEASEQUERY = 14
    DHCPLEASEQUERYDONE = 15
    DHCPACTIVELEASEQUERY = 16
    DHCPLEASEQUERYSTATUS = 17
    DHCPTLS = 18

DHCPACK = 5 class-attribute instance-attribute

Server to client with configuration parameters, including committed network address.

DHCPDECLINE = 4 class-attribute instance-attribute

Client to server indicating network address is already in use.

DHCPDISCOVER = 1 class-attribute instance-attribute

Client broadcast to locate available servers.

DHCPFORCERENEW = 9 class-attribute instance-attribute

Forces the client to the RENEW state.

DHCPINFORM = 8 class-attribute instance-attribute

Client to server, asking only for local configuration parameters; client already has externally configured network address.

DHCPLEASEACTIVE = 13 class-attribute instance-attribute

The DHCPLEASEACTIVE response to a DHCPLEASEQUERY message allows the relay agent to determine the IP endpoint location and the remaining duration of the IP address lease.

DHCPLEASEQUERY = 10 class-attribute instance-attribute

The DHCPLEASEQUERY message is a new DHCP message type transmitted from a DHCP relay agent to a DHCP server. A DHCPLEASEQUERY-aware relay agent sends the DHCPLEASEQUERY message when it needs to know the location of an IP endpoint. The DHCPLEASEQUERY-aware DHCP server replies with a DHCPLEASEUNASSIGNED, DHCPLEASEACTIVE, or DHCPLEASEUNKNOWN message.

DHCPLEASEUNASSIGNED = 11 class-attribute instance-attribute

The DHCPLEASEUNASSIGNED is similar to a DHCPLEASEACTIVE message, but indicates that there is no currently active lease on the resultant IP address but that this DHCP server is authoritative for this IP address.

DHCPLEASEUNKNOWN = 12 class-attribute instance-attribute

The DHCPLEASEUNKNOWN message indicates that the DHCP server has no knowledge of the information specified in the query (e.g., IP address, MAC address, or Client-identifier option).

DHCPNAK = 6 class-attribute instance-attribute

Server to client indicating client's notion of network address is incorrect (e.g., client has moved to new subnet) or client's lease as expired

DHCPOFFER = 2 class-attribute instance-attribute

Server to client in response to DHCPDISCOVER with offer of configuration parameters.

DHCPRELEASE = 7 class-attribute instance-attribute

Client to server relinquishing network address and cancelling remaining lease.

DHCPREQUEST = 3 class-attribute instance-attribute

Client message to servers either (a) requesting offered parameters from one server and implicitly declining offers from all others, (b) confirming correctness of previously allocated address after, e.g., system reboot, or (c) extending the lease on a particular network address.

Flags

Bases: Flag

Source code in src/pydhcp/packet/enums.py

class Flags(_enum.Flag):
    UNICAST = 0
    BROADCAST = 1 << 15
    """Set by client that cant listen to unicast response as it doesnt have an ip yet"""

BROADCAST = 1 << 15 class-attribute instance-attribute

Set by client that cant listen to unicast response as it doesnt have an ip yet

OpCode

Bases: IntEnum

Specifies if the message originates from a server or client

Source code in src/pydhcp/packet/enums.py

class OpCode(_enum.IntEnum):
    """Specifies if the message originates from a server or client"""

    BOOTREQUEST = 1
    """DHCP message sent from a client to a server."""
    BOOTREPLY = 2
    """DHCP message sent from a server to a client."""

BOOTREPLY = 2 class-attribute instance-attribute

DHCP message sent from a server to a client.

BOOTREQUEST = 1 class-attribute instance-attribute

DHCP message sent from a client to a server.

pydhcp.network

pydhcp.network

MACAddress

Bases: MACAddress

A hardware address rendered the way DHCP tooling expects.

Only the presentation differs from :class:netimps.MACAddress: uppercase-hyphenated (00-11-22-33-44-55) rather than lowercase-colon, because that is the form this project's CLI and logs have always used. Parsing, comparison and hashing are inherited unchanged, so instances compare equal to the base type and interoperate with it as dict keys.

Note this is a display type. The wire hardware address (chaddr, option 61) is raw bytes throughout packet/ and never passes through here -- deliberately, since chaddr permits hlen up to 16 for non-Ethernet htype while a MAC is exactly 6.

Source code in src/pydhcp/network/__init__.py

class MACAddress(_netimps.MACAddress):
    """A hardware address rendered the way DHCP tooling expects.

    Only the *presentation* differs from :class:`netimps.MACAddress`:
    uppercase-hyphenated (``00-11-22-33-44-55``) rather than lowercase-colon,
    because that is the form this project's CLI and logs have always used.
    Parsing, comparison and hashing are inherited unchanged, so instances
    compare equal to the base type and interoperate with it as dict keys.

    Note this is a *display* type. The wire hardware address (``chaddr``,
    option 61) is raw ``bytes`` throughout ``packet/`` and never passes through
    here -- deliberately, since ``chaddr`` permits ``hlen`` up to 16 for
    non-Ethernet ``htype`` while a MAC is exactly 6.
    """

    def __str__(self) -> str:
        return self.as_str("-", upper=True)

    def hex(self, *args, **kwargs) -> str:
        """``bytes.hex`` passthrough.

        The base type is a value object exposing ``.packed`` rather than a
        ``bytes`` subclass, so this method is not inherited -- but callers
        (and tests) predating that reasonably expect it.
        """
        return self.packed.hex(*args, **kwargs)

hex(*args, **kwargs)

bytes.hex passthrough.

The base type is a value object exposing .packed rather than a bytes subclass, so this method is not inherited -- but callers (and tests) predating that reasonably expect it.

Source code in src/pydhcp/network/__init__.py

def hex(self, *args, **kwargs) -> str:
    """``bytes.hex`` passthrough.

    The base type is a value object exposing ``.packed`` rather than a
    ``bytes`` subclass, so this method is not inherited -- but callers
    (and tests) predating that reasonably expect it.
    """
    return self.packed.hex(*args, **kwargs)

SocketAddress

Bases: _SocketAddress

Source code in src/pydhcp/network/__init__.py

class SocketAddress(_SocketAddress):
    def __new__(
        cls,
        ip: _ty.Union[str, IPv4, _socket.socket],
        port: _ty.Optional[int] = None,
    ) -> "SocketAddress":
        if isinstance(ip, _socket.socket):
            ip_val, port_val = ip.getsockname()
        elif port is None:
            raise ValueError()
        else:
            ip_val, port_val = ip, port
        return super(SocketAddress, cls).__new__(cls, IPv4(ip_val), int(port_val))

    def compat(self) -> tuple[str, int]:
        return (str(self.ip), self.port)

    def __str__(self) -> str:
        return f"{self.ip}:{self.port}"

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(ip={self.ip}, port={self.port})"

    def listen(
        self,
        family: _socket.AddressFamily = _socket.AF_INET,
        kind: _socket.SocketKind = _socket.SOCK_DGRAM,
        proto: int = 0,
        fileno: _ty.Optional[int] = None,
        options: _ty.Iterable[SocketOption] = (),
    ) -> _socket.socket:
        """Create and bind a socket at this address.

        ``options`` are ``(level, name, value)`` triples applied before the
        bind. Delegates to :func:`netimps.bind`, which closes the socket before
        any exception propagates -- so a failed bind leaks nothing.

        ``reuse_address`` is left off: this has never set ``SO_REUSEADDR``
        implicitly, and turning it on now would let a socket bind a port still
        in ``TIME_WAIT``. Callers that want it pass it in ``options``, as
        ``listener.py`` does.
        """
        if fileno is not None:
            # netimps.bind() creates the socket itself, so an existing fd has
            # to keep the direct path.
            sock = _socket.socket(family, kind, proto, fileno)
            for opt in options:
                sock.setsockopt(*opt)
            sock.bind((str(self.ip), self.port))
            return sock

        return _netimps.bind(
            str(self.ip),
            self.port,
            family=family,
            kind=kind,
            reuse_address=False,
            options=tuple(options),
        )

listen(family=_socket.AF_INET, kind=_socket.SOCK_DGRAM, proto=0, fileno=None, options=())

Create and bind a socket at this address.

options are (level, name, value) triples applied before the bind. Delegates to :func:netimps.bind, which closes the socket before any exception propagates -- so a failed bind leaks nothing.

reuse_address is left off: this has never set SO_REUSEADDR implicitly, and turning it on now would let a socket bind a port still in TIME_WAIT. Callers that want it pass it in options, as listener.py does.

Source code in src/pydhcp/network/__init__.py

def listen(
    self,
    family: _socket.AddressFamily = _socket.AF_INET,
    kind: _socket.SocketKind = _socket.SOCK_DGRAM,
    proto: int = 0,
    fileno: _ty.Optional[int] = None,
    options: _ty.Iterable[SocketOption] = (),
) -> _socket.socket:
    """Create and bind a socket at this address.

    ``options`` are ``(level, name, value)`` triples applied before the
    bind. Delegates to :func:`netimps.bind`, which closes the socket before
    any exception propagates -- so a failed bind leaks nothing.

    ``reuse_address`` is left off: this has never set ``SO_REUSEADDR``
    implicitly, and turning it on now would let a socket bind a port still
    in ``TIME_WAIT``. Callers that want it pass it in ``options``, as
    ``listener.py`` does.
    """
    if fileno is not None:
        # netimps.bind() creates the socket itself, so an existing fd has
        # to keep the direct path.
        sock = _socket.socket(family, kind, proto, fileno)
        for opt in options:
            sock.setsockopt(*opt)
        sock.bind((str(self.ip), self.port))
        return sock

    return _netimps.bind(
        str(self.ip),
        self.port,
        family=family,
        kind=kind,
        reuse_address=False,
        options=tuple(options),
    )

host_ip_interfaces(filter=True, family=4)

Yield one :class:NetworkInterface per local address.

One entry per address, not per adapter, since callers here select and filter by address. filter defaults to excluding APIPA; pass False for everything, or a predicate of your own.

family defaults to 4: this is a DHCPv4 implementation, and the enumeration it replaced was IPv4-only, so yielding IPv6 addresses would silently change what existing callers iterate over. Pass None for both families or 6 for IPv6 only.

Enumeration comes from :func:netimps.iter_addresses, which reports real prefix lengths and human-readable adapter names on every platform.

Source code in src/pydhcp/network/__init__.py

def host_ip_interfaces(
    filter: _ty.Union[_ty.Callable[[NetworkInterface], bool], bool] = True,
    family: _ty.Optional[int] = 4,
) -> _ty.Iterator[NetworkInterface]:
    """Yield one :class:`NetworkInterface` per local address.

    One entry *per address*, not per adapter, since callers here select and
    filter by address. ``filter`` defaults to excluding APIPA; pass ``False``
    for everything, or a predicate of your own.

    ``family`` defaults to **4**: this is a DHCPv4 implementation, and the
    enumeration it replaced was IPv4-only, so yielding IPv6 addresses would
    silently change what existing callers iterate over. Pass ``None`` for both
    families or ``6`` for IPv6 only.

    Enumeration comes from :func:`netimps.iter_addresses`, which reports real
    prefix lengths and human-readable adapter names on every platform.
    """
    if filter is True:
        filter = lambda ni: ni.ip not in APIPA
    for iface, address in _netimps.iter_addresses(family=family):
        ni = NetworkInterface(
            name=iface.name,
            ip_interface=address,
            mac=MACAddress(iface.mac) if iface.mac else None,
        )
        if not filter or filter(ni):
            yield ni

pydhcp.lease

pydhcp.lease