AI-Agent/win-dhcpd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ea8eb2b5f982635dcca987a490f9c710993169cd
win-dhcpd/dhcpd/parser.py
T
Your Name ea8eb2b5f9 feat: WinDHCPD — Windows-native DHCP server 
Python 3 + Scapy DHCP server with:
- Full DORA handshake (DISCOVER/OFFER/REQUEST/ACK)
- Multi-scope support with DHCP relay
- Static MAC-to-IP bindings
- Lease persistence (JSON) with auto-prune
- NACK for invalid requests
- Flask web management panel (dark theme, live refresh)
- Admin-privileged scapy sniff on UDP 67
2026-07-13 16:22:19 +08:00

333 lines
8.8 KiB
Python
Raw Blame History

"""DHCP packet parsing using Scapy."""
from __future__ import annotations
import logging
from typing import Optional, Tuple
from scapy.all import Ether, IP, UDP, BOOTP, DHCP
log = logging.getLogger("dhcpd.parser")
# DHCP message types (RFC 2132)
DHCPDISCOVER = 1
DHCPOFFER = 2
DHCPREQUEST = 3
DHCPDECLINE = 4
DHCPACK = 5
DHCPNAK = 6
DHCPRELEASE = 7
DHCPINFORM = 8
MSG_NAMES = {
DHCPDISCOVER: "DISCOVER",
DHCPOFFER: "OFFER",
DHCPREQUEST: "REQUEST",
DHCPDECLINE: "DECLINE",
DHCPACK: "ACK",
DHCPNAK: "NAK",
DHCPRELEASE: "RELEASE",
DHCPINFORM: "INFORM",
}
class DHCPMessage:
"""Normalized view of a parsed DHCP packet."""
def __init__(
self,
msg_type: int,
xid: int,
client_mac: str,
client_ip: str = "0.0.0.0",
your_ip: str = "0.0.0.0",
server_ip: str = "0.0.0.0",
server_id: str = "0.0.0.0",
giaddr: str = "0.0.0.0",
ciaddr: str = "0.0.0.0",
options: dict | None = None,
hostname: str = "",
src_mac: str = "",
src_ip: str = "",
dst_ip: str = "",
):
self.msg_type = msg_type
self.msg_name = MSG_NAMES.get(msg_type, f"TYPE-{msg_type}")
self.xid = xid
self.client_mac = client_mac.upper()
self.client_ip = client_ip
self.your_ip = your_ip
self.server_ip = server_ip
self.server_id = server_id
self.giaddr = giaddr
self.ciaddr = ciaddr
self.options = options or {}
self.hostname = hostname
self.src_mac = src_mac
self.src_ip = src_ip
self.dst_ip = dst_ip
@property
def is_broadcast(self) -> bool:
return self.dst_ip in ("255.255.255.255", "0.0.0.0") or self.dst_mac == "ff:ff:ff:ff:ff:ff"
@property
def dst_mac(self) -> str:
return self.dst_ip # placeholder
def __repr__(self) -> str:
return (f"{self.msg_name} xid=0x{self.xid:08x} "
f"mac={self.client_mac} ci={self.client_ip} "
f"hostname={self.hostname!r}")
def _get_option(options: list | None, name: str) -> any:
if not options:
return None
for o in options:
if o.name == name:
return o
return None
def parse(pkt) -> Optional[DHCPMessage]:
"""Parse a scapy packet; return normalized DHCPMessage or None."""
if not pkt.haslayer(DHCP):
return None
dh = pkt[DHCP]
bp = pkt[BOOTP]
msg_type = None
hostname = ""
server_id = "0.0.0.0"
options = {}
# Scapy stores DHCP options as:
# ("option-name", value) or
# (code_int, value_bytes) or
# "end" (terminator)
for o in dh.options:
if o == "end":
continue
if isinstance(o, tuple):
name = o[0]
val = o[1]
else:
continue # skip unknown format
# Map option codes to names
_CODE_NAMES = {53: "message-type", 54: "server_id", 1: "subnet_mask",
3: "router", 6: "domain_name_server", 15: "domain_name",
51: "lease_time", 55: "parameter_request_list",
50: "requested_addr", 12: "hostname"}
if name == "message-type" or (isinstance(name, int) and name == 53):
msg_type = val
elif name == "hostname" or (isinstance(name, int) and name == 12):
hostname = val.decode("utf-8", "ignore") if isinstance(val, bytes) else str(val)
elif name == "server_id" or (isinstance(name, int) and name == 54):
server_id = val
elif name == "subnet_mask" or (isinstance(name, int) and name == 1):
options["subnet_mask"] = val
elif name in ("router",) or (isinstance(name, int) and name == 3):
options["router"] = val
elif name == "domain_name_server" or (isinstance(name, int) and name == 6):
options["dns"] = val
elif name == "domain_name" or (isinstance(name, int) and name == 15):
options["domain"] = val
elif name == "lease_time" or (isinstance(name, int) and name == 51):
options["lease_time"] = val
elif name == "parameter_request_list" or (isinstance(name, int) and name == 55):
options["param_req"] = val
if msg_type is None:
return None
try:
client_mac_raw = bp.chaddr
except Exception:
client_mac_raw = bytes(16)
client_mac = ":".join(f"{b:02x}" for b in client_mac_raw[:6])
try:
ip_layer = pkt[IP]
except Exception:
ip_layer = None
try:
ether = pkt[Ether]
src_mac = ether.src
except Exception:
src_mac = client_mac
# BOOTP fields
yiaddr = bp.yiaddr or "0.0.0.0"
siaddr = bp.siaddr or "0.0.0.0"
giaddr = bp.giaddr or "0.0.0.0"
ciaddr = bp.ciaddr or "0.0.0.0"
if ip_layer:
src_ip = ip_layer.src
dst_ip = ip_layer.dst
else:
src_ip = siaddr
dst_ip = "255.255.255.255"
return DHCPMessage(
msg_type=msg_type,
xid=bp.xid,
client_mac=client_mac,
client_ip=ciaddr,
your_ip=yiaddr,
server_ip=siaddr,
server_id=server_id,
giaddr=giaddr,
ciaddr=ciaddr,
options=options,
hostname=hostname,
src_mac=src_mac,
src_ip=src_ip,
dst_ip=dst_ip,
)
def build_offer(
req: DHCPMessage,
offer_ip: str,
lease_seconds: int,
server_ip: str,
router: str = "",
dns: list = None,
subnet_mask: str = "255.255.255.0",
domain: str = "",
) -> object:
"""Build a scapy OFFER packet to reply to *req*."""
dns = dns or []
xid = req.xid
client_mac = req.client_mac
# Convert MAC hex to bytes
mac_bytes = bytes.fromhex(client_mac.replace(":", ""))[:6]
chaddr = mac_bytes + bytes(10)
options = [
("message-type", DHCPOFFER),
("server_id", server_ip),
("subnet_mask", subnet_mask),
("lease_time", lease_seconds),
("router", router),
]
if dns:
options.append(("domain_name_server", dns))
if domain:
options.append(("domain_name", domain))
dst_ip = "255.255.255.255" # broadcast by default
if req.giaddr and req.giaddr != "0.0.0.0":
dst_ip = req.giaddr # relay agent
pkt = (
Ether(dst="ff:ff:ff:ff:ff:ff")
/ IP(src=server_ip, dst=dst_ip)
/ UDP(sport=67, dport=68)
/ BOOTP(
op=2, # reply
xid=xid,
chaddr=chaddr,
yiaddr=offer_ip,
siaddr=server_ip,
giaddr=req.giaddr,
)
/ DHCP(options=options)
)
return pkt
def build_ack(
req: DHCPMessage,
offered_ip: str,
lease_seconds: int,
server_ip: str,
router: str = "",
dns: list = None,
subnet_mask: str = "255.255.255.0",
domain: str = "",
) -> object:
"""Build a scapy ACK packet. Same structure as OFFER but message-type=ACK."""
dns = dns or []
xid = req.xid
client_mac = req.client_mac
mac_bytes = bytes.fromhex(client_mac.replace(":", ""))[:6]
chaddr = mac_bytes + bytes(10)
options = [
("message-type", DHCPACK),
("server_id", server_ip),
("subnet_mask", subnet_mask),
("lease_time", lease_seconds),
("router", router),
]
if dns:
options.append(("domain_name_server", dns))
if domain:
options.append(("domain_name", domain))
dst_ip = "255.255.255.255"
if req.giaddr and req.giaddr != "0.0.0.0":
dst_ip = req.giaddr
pkt = (
Ether(dst="ff:ff:ff:ff:ff:ff")
/ IP(src=server_ip, dst=dst_ip)
/ UDP(sport=67, dport=68)
/ BOOTP(
op=2,
xid=xid,
chaddr=chaddr,
yiaddr=offered_ip,
siaddr=server_ip,
giaddr=req.giaddr,
)
/ DHCP(options=options)
)
return pkt
def build_nak(req: DHCPMessage, server_ip: str) -> object:
"""Build a NAK to reject a request."""
xid = req.xid
client_mac = req.client_mac
mac_bytes = bytes.fromhex(client_mac.replace(":", ""))[:6]
chaddr = mac_bytes + bytes(10)
options = [
("message-type", DHCPNAK),
("server_id", server_ip),
]
dst_ip = "255.255.255.255"
if req.giaddr and req.giaddr != "0.0.0.0":
dst_ip = req.giaddr
pkt = (
Ether(dst="ff:ff:ff:ff:ff:ff")
/ IP(src=server_ip, dst=dst_ip)
/ UDP(sport=67, dport=68)
/ BOOTP(
op=2,
xid=xid,
chaddr=chaddr,
yiaddr="0.0.0.0",
siaddr=server_ip,
giaddr=req.giaddr,
)
/ DHCP(options=options)
)
return pkt
def build_release_ack(req: DHCPMessage, server_ip: str) -> object:
"""Optional lightweight ACK to a RELEASE (many servers just silently accept)."""
return None # Most DHCP servers don't respond to RELEASE
Reference in New Issue View Git Blame Copy Permalink