From ea8eb2b5f982635dcca987a490f9c710993169cd Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 13 Jul 2026 16:22:19 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20WinDHCPD=20=E2=80=94=20Windows-native?= =?UTF-8?q?=20DHCP=20server?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .gitignore | 11 ++ README.md | 223 ++++++++++++++++++++++ dhcpd.yaml.example | 77 ++++++++ dhcpd/__init__.py | 2 + dhcpd/config.py | 189 +++++++++++++++++++ dhcpd/lease.py | 189 +++++++++++++++++++ dhcpd/parser.py | 332 +++++++++++++++++++++++++++++++++ dhcpd/server.py | 303 ++++++++++++++++++++++++++++++ dhcpd/web/app.py | 121 ++++++++++++ dhcpd/web/templates/index.html | 150 +++++++++++++++ requirements.txt | 4 + run.py | 136 ++++++++++++++ 12 files changed, 1737 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 dhcpd.yaml.example create mode 100644 dhcpd/__init__.py create mode 100644 dhcpd/config.py create mode 100644 dhcpd/lease.py create mode 100644 dhcpd/parser.py create mode 100644 dhcpd/server.py create mode 100644 dhcpd/web/app.py create mode 100644 dhcpd/web/templates/index.html create mode 100644 requirements.txt create mode 100644 run.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4d99f09 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +__pycache__/ +*.pyc +*.db +*.log +*.yaml +!*.yaml.example +.env +*.egg-info/ +dist/ +build/ +.ipynb_checkpoints/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..ae39300 --- /dev/null +++ b/README.md @@ -0,0 +1,223 @@ +# WinDHCPD — Windows 原生 DHCP 服务器 + +基于 **Python 3 + Scapy** 实现的轻量级 DHCP 服务器，可在 Windows 上以管理员权限运行。 + +## 功能特性 + +| 功能 | 说明 | +|------|------| +| **完整 DORA 握手** | DISCOVER → OFFER → REQUEST → ACK，符合 RFC 2131 | +| **多作用域** | 支持多个网段（/24），可通过 DHCP Relay 跨网段服务 | +| **静态绑定** | MAC → IP 固定分配，优先于动态分配 | +| **租约管理** | 内存索引 + JSON 持久化，支持释放、续租、清理过期 | +| **Web 管理面板** | Flask 面板，实时查看租约、作用域、统计（端口 8080） | +| **NACK 拒绝** | 对不合法请求返回 NACK | +| **日志记录** | 控制台 + 文件双输出 | + +## 架构 + +``` +win-dhcpd/ +├── run.py # 入口脚本 +├── dhcpd.yaml.example # 配置文件示例 +├── requirements.txt # Python 依赖 +├── dhcpd/ +│ ├── config.py # 配置加载与校验 +│ ├── lease.py # 租约数据库（内存 + JSON 持久化） +│ ├── parser.py # DHCP 包解析与构建（Scapy） +│ ├── server.py # DHCP 服务器主循环（Sniffer + 响应） +│ └── web/ +│ ├── app.py # Flask Web 面板 +│ └── templates/ +│ └── index.html # 管理界面（深色主题，自动刷新） +``` + +## 前置条件 + +### 1. 安装 Python 3.8+ +确保 Windows 上已安装 Python 3.8 或更高版本。 + +### 2. 安装 Npcap（scapy 依赖） +Scapy 在 Windows 上需要 Npcap 来捕获和发送网络数据包： +- 下载：https://nmap.org/npcap/ +- 安装时 **必须勾选 "Install Npcap in WinPcap API-compatible Mode"** +- 默认安装即可 + +### 3. 安装依赖 +```bash +pip install -r requirements.txt +``` +依赖包： +- **scapy** (≥2.5) — 网络数据包捕获与发送 +- **flask** (≥3.0) — Web 管理面板 +- **pyyaml** (≥6.0) — 配置文件解析 + +## 快速开始 + +### 1. 创建配置文件 +```bash +copy dhcpd.yaml.example dhcpd.yaml +``` +编辑 `dhcpd.yaml`，修改作用域（网段、地址池、网关、DNS）为实际网络参数。 + +**关键参数说明：** + +| 参数 | 说明 | 示例 | +|------|------|------| +| `server.ip` | DHCP 服务器自身 IP | `192.168.1.1` | +| `scopes[].network` | 网段（CIDR） | `192.168.1.0/24` | +| `scopes[].start_ip` | 地址池起始 | `192.168.1.100` | +| `scopes[].end_ip` | 地址池结束 | `192.168.1.200` | +| `scopes[].router` | 默认网关 | `192.168.1.1` | +| `scopes[].dns_servers` | DNS 服务器列表 | `[8.8.8.8, 114.114.114.114]` | +| `scopes[].lease_seconds` | 租期（秒） | `86400` (24小时) | +| `bindings[].mac` | 静态绑定 MAC | `aa:bb:cc:dd:ee:01` | +| `bindings[].ip` | 静态绑定 IP | `192.168.1.10` | + +### 2. 以管理员身份运行 +**必须使用管理员权限**（捕获端口 67 的广播数据包需要特权）： + +```bash +# Windows 命令行 +python run.py + +# 指定配置文件 +python run.py dhcpd.yaml + +# 指定监听网卡 +python run.py dhcpd.yaml -i "以太网" + +# 禁用 Web 面板 +python run.py dhcpd.yaml --no-web + +# 指定 Web 端口 +python run.py dhcpd.yaml --web-port 9000 +``` + +### 3. 访问 Web 面板 +打开浏览器访问： +``` +http://localhost:8080 +``` + +## 配置文件详解 + +```yaml +# ── 服务器设置 ── +server: + ip: "192.168.1.1" # DHCP 服务器 IP（响应包中的 server_id） + +# ── 作用域 ── +scopes: + - name: "Office-LAN" # 作用域名称 + network: "192.168.1.0/24" + start_ip: "192.168.1.100" + end_ip: "192.168.1.200" + subnet_mask: "255.255.255.0" + router: "192.168.1.1" + dns_servers: + - "8.8.8.8" + - "114.114.114.114" + lease_seconds: 86400 # 租期 24 小时 + domain: "local" + enable: true + +# ── 静态绑定 ── +bindings: + - mac: "aa:bb:cc:dd:ee:01" + ip: "192.168.1.10" + scope: "Office-LAN" + hostname: "printer" + +# ── 其他 ── +log_file: "dhcpd.log" +log_level: "INFO" # DEBUG / INFO / WARNING / ERROR +web_host: "0.0.0.0" +web_port: 8080 +# interface: "以太网" # 可选：指定监听网卡 +``` + +## 网络拓扑要求 + +### 场景 1：单机 DHCP 服务器 +``` +客户端 ────── 交换机/集线器 ────── WinDHCPD (192.168.1.1) + ↑ ↑ + │ 广播 DISCOVER │ 响应 OFFER/ACK + └────────────────────────────────┘ +``` +- WinDHCPD 所在的网卡必须是 **混杂模式** 或连接在 Hub/支持组播的交换机上 +- 同一网段 **只能有一个 DHCP 服务器** + +### 场景 2：跨网段（DHCP Relay） +``` +子网 A 客户端 ── 路由器(Relay) ── WinDHCPD (192.168.1.1) + ↑ + giaddr = 10.0.2.1 +``` +- 路由器需配置 `ip helper-address ` +- WinDHCPD 通过 `giaddr` 字段确定客户端所在网段，选择对应作用域 + +## 工作原理 + +``` +客户端 服务器 + │ │ + │ DHCPDISCOVER (广播) │ + │ ─────────────────────► │ + │ │ 分配 IP → 发 OFFER + │ DHCPOFFER (广播) │ + │ ◄────────────────────── │ + │ │ + │ DHCPREQUEST (广播) │ + │ ─────────────────────► │ + │ │ 更新租约 → 发 ACK + │ DHCPACK (广播) │ + │ ◄────────────────────── │ + │ │ + │ DHCPRELEASE │ + │ ─────────────────────► │ 释放 IP +``` + +## 停止服务器 + +- 控制台：按 `Ctrl+C` +- 任务管理器：结束 `python.exe` 进程 + +租约数据自动保存到 `leases.db`，下次启动时自动加载。 + +## 常见问题 + +### Q: 客户端获取不到 IP +1. **必须以管理员身份运行** — 端口 67 需要特权 +2. **检查 Npcap 是否安装** — 且勾选了 WinPcap 兼容模式 +3. **确认网卡选择** — 用 `-i "网卡名"` 指定正确的网卡 +4. **同一网段不能有另一个 DHCP 服务器** — Windows 自带的 DHCP Server 服务必须关闭 + +### Q: 如何查看日志 +```bash +# 实时查看 +type dhcpd.log + +# 修改日志级别（配置文件中） +log_level: "DEBUG" +``` + +### Q: 支持 IPv6 吗 +不支持。本服务器仅处理 IPv4 DHCP。 + +### Q: 支持 DHCPv6 吗 +不支持。如需 IPv6 分配，请配置 SLAAC 或单独的 DHCPv6 服务器。 + +## 安全注意事项 + +⚠️ **不要在生产网络中直接部署**，除非你完全了解后果： + +1. **单点故障** — 这台机器挂了，所有 DHCP 客户端都无法获取 IP +2. **冲突风险** — 同一网段有两个 DHCP 服务器会导致客户端随机获取 IP +3. **端口监听** — 服务器会监听 UDP 67 的广播包，其他设备可能无法响应 +4. **建议** — 先用隔离网络测试，确认无误后再部署 + +## 许可 + +MIT License diff --git a/dhcpd.yaml.example b/dhcpd.yaml.example new file mode 100644 index 0000000..7d6f87c --- /dev/null +++ b/dhcpd.yaml.example @@ -0,0 +1,77 @@ +# ============================================================================= +# WinDHCPD 配置文件 +# ============================================================================= +# 用法: 将此文件保存为 dhcpd.yaml，修改后运行 python run.py dhcpd.yaml +# +# 注意: +# - 本服务器通过 scapy 以混杂模式捕获并发送以太网帧，需要管理员权限。 +# - Windows 上需安装 Npcap (带 "WinPcap API-compatible" 选项) 供 scapy 使用。 +# - 同一网段只能有一个 DHCP 服务器，否则客户端可能获取到错误 IP。 +# - 不要与 Windows 内置 DHCP Server 服务共用网段。 +# ============================================================================= + +# ── 服务器基本设置 ──────────────────────────────────────────────────────── +server: + # DHCP 服务器本身的 IP 地址（响应包中的 siaddr / server_id） + ip: "192.168.1.1" + +# ── 作用域 (Scopes) ────────────────────────────────────────────────────── +# 每个作用域 = 一个网段 + 一个 IP 地址池 +# 多个作用域可以实现多网段 DHCP（需配置 relay） +scopes: + - name: "Office-LAN" + # 网段，格式 network/prefix + network: "192.168.1.0/24" + # 地址池起止 + start_ip: "192.168.1.100" + end_ip: "192.168.1.200" + # 子网掩码 + subnet_mask: "255.255.255.0" + # 默认网关 + router: "192.168.1.1" + # DNS 服务器（可填多个） + dns_servers: + - "8.8.8.8" + - "114.114.114.114" + # 租期（秒） + lease_seconds: 86400 # 24 小时 + # 域名 + domain: "local" + # 启用/禁用 + enable: true + +# # 示例：第二个作用域 +# - name: "Guest-LAN" +# network: "192.168.2.0/24" +# start_ip: "192.168.2.50" +# end_ip: "192.168.2.150" +# subnet_mask: "255.255.255.0" +# router: "192.168.2.1" +# dns_servers: ["8.8.8.8"] +# lease_seconds: 3600 +# domain: "guest" +# enable: true + +# ── 静态绑定 ───────────────────────────────────────────────────────────── +# 为特定 MAC 地址固定分配指定 IP +bindings: + - mac: "aa:bb:cc:dd:ee:01" + ip: "192.168.1.10" + scope: "Office-LAN" + hostname: "printer" + - mac: "aa:bb:cc:dd:ee:02" + ip: "192.168.1.11" + scope: "Office-LAN" + hostname: "nas" + +# ── 日志 ───────────────────────────────────────────────────────────────── +log_file: "dhcpd.log" +log_level: "INFO" # DEBUG / INFO / WARNING / ERROR + +# ── Web 管理面板 ───────────────────────────────────────────────────────── +web_host: "0.0.0.0" +web_port: 8080 + +# ── 网络接口 ───────────────────────────────────────────────────────────── +# 留空则 scapy 自动选择; 指定则只监听该网卡 +# interface: "Ethernet" diff --git a/dhcpd/__init__.py b/dhcpd/__init__.py new file mode 100644 index 0000000..fffaaf7 --- /dev/null +++ b/dhcpd/__init__.py @@ -0,0 +1,2 @@ +# WinDHCPD - Windows-native DHCP server +# Python 3 + Scapy based diff --git a/dhcpd/config.py b/dhcpd/config.py new file mode 100644 index 0000000..d27bdcb --- /dev/null +++ b/dhcpd/config.py @@ -0,0 +1,189 @@ +"""Configuration loading and validation.""" + +from __future__ import annotations + +import copy +import ipaddress +import logging +import os +import socket +from typing import Any, Dict, List, Optional + +import yaml + +log = logging.getLogger("dhcpd.config") + +# ── Defaults ────────────────────────────────────────────────────────────── + +DEFAULT_LEASE_SECONDS = 86400 # 24 hours +DEFAULT_OFFER_LEASE_SECONDS = 7200 # 2 hours (offer expires faster) +DEFAULT_LOG_LEVEL = "INFO" + + +class Scope: + """One DHCP scope (one /24 subnet or similar).""" + + def __init__( + self, + name: str, + network: str, + start_ip: str, + end_ip: str, + subnet_mask: str = "255.255.255.0", + router: str = "", + dns_servers: List[str] | None = None, + lease_seconds: int = DEFAULT_LEASE_SECONDS, + domain: str = "", + enable: bool = True, + ): + self.name = name + self.subnet = ipaddress.ip_network(network, strict=False) + self.network = str(self.subnet) + self.start_ip = ipaddress.ip_address(start_ip) + self.end_ip = ipaddress.ip_address(end_ip) + self.subnet_mask = subnet_mask + self.router = router + self.dns_servers = dns_servers or [] + self.lease_seconds = lease_seconds + self.domain = domain + self.enable = enable + + # Validate pool is inside the subnet + assert self.start_ip in self.subnet, f"{start_ip} not in {network}" + assert self.end_ip in self.subnet, f"{end_ip} not in {network}" + assert self.start_ip <= self.end_ip, f"start > end in scope {name}" + if self.router: + assert ipaddress.ip_address(self.router) in self.subnet, \ + f"router {self.router} not in {network}" + for d in self.dns_servers: + # DNS may be outside the subnet (e.g. public DNS), only warn + if ipaddress.ip_address(d) not in self.subnet: + logging.getLogger("dhcpd.config").warning( + "Scope %s: DNS %s is outside subnet %s (ok for public DNS)", + self.name, d, self.subnet) + + def __repr__(self) -> str: + return f"Scope({self.name} {self.network} [{self.start_ip}-{self.end_ip}])" + + def is_for_network(self, ip: str) -> bool: + try: + return ipaddress.ip_address(ip) in self.subnet + except ValueError: + return False + + +class StaticBinding: + """Fixed IP-for-MAC assignment.""" + + def __init__(self, mac: str, ip: str, scope_name: str, hostname: str = ""): + self.mac = mac.lower().replace(":", "").replace("-", "") + self.ip = str(ipaddress.ip_address(ip)) + self.scope_name = scope_name + self.hostname = hostname + # Normalize MAC display + self.mac_display = ":".join( + self.mac[i:i+2] for i in range(0, 12, 2) + ) + + def __repr__(self) -> str: + return f"StaticBinding({self.mac_display} -> {self.ip})" + + +class Config: + """Top-level configuration.""" + + def __init__(self, path: str): + self.path = path + self.raw: Dict[str, Any] = {} + self.server: Dict[str, Any] = {} + self.scopes: List[Scope] = [] + self.bindings: List[StaticBinding] = [] + self.log_file: str = "dhcpd.log" + self.log_level: str = DEFAULT_LOG_LEVEL + self.web_port: int = 8080 + self.web_host: str = "0.0.0.0" + self._load(path) + + # ── loading ────────────────────────────────────────────────────────── + + def _load(self, path: str) -> None: + if not os.path.exists(path): + raise FileNotFoundError(f"Config not found: {path}") + with open(path, "r", encoding="utf-8") as f: + self.raw = yaml.safe_load(f) or {} + + self._apply(self.raw) + + def _apply(self, d: Dict[str, Any]) -> None: + self.server = d.get("server", {}) + self.log_file = d.get("log_file", "dhcpd.log") + self.log_level = d.get("log_level", DEFAULT_LOG_LEVEL) + self.web_host = d.get("web_host", "0.0.0.0") + self.web_port = int(d.get("web_port", 8080)) + + scopes = d.get("scopes", []) + for s in scopes: + sc = Scope( + name=s["name"], + network=s["network"], + start_ip=s["start_ip"], + end_ip=s["end_ip"], + subnet_mask=s.get("subnet_mask", "255.255.255.0"), + router=s.get("router", ""), + dns_servers=s.get("dns_servers", None), + lease_seconds=s.get("lease_seconds", DEFAULT_LEASE_SECONDS), + domain=s.get("domain", ""), + enable=s.get("enable", True), + ) + self.scopes.append(sc) + + for b in d.get("bindings", []): + self.bindings.append(StaticBinding( + mac=b["mac"], + ip=b["ip"], + scope_name=b.get("scope", ""), + hostname=b.get("hostname", ""), + )) + + self._validate() + log.info("Loaded %d scopes, %d bindings from %s", + len(self.scopes), len(self.bindings), self.path) + + def _validate(self) -> None: + if not self.scopes: + raise ValueError("No scopes defined in config") + for sc in self.scopes: + if not sc.enable: + continue + # Check no overlapping scopes + for other in self.scopes: + if other is sc or not other.enable: + continue + if sc.subnet.overlaps(other.subnet): + raise ValueError( + f"Overlapping enabled scopes: {sc.name} vs {other.name}") + # Check all bindings reference a valid scope + scope_names = {s.name for s in self.scopes} + for b in self.bindings: + if b.scope_name and b.scope_name not in scope_names: + log.warning("Binding for %s references unknown scope '%s'", + b.mac_display, b.scope_name) + + # ── queries ────────────────────────────────────────────────────────── + + def scope_for_ip(self, ip: str) -> Optional[Scope]: + """Return the (enabled) scope whose network contains *ip*.""" + for sc in self.scopes: + if sc.enable and sc.is_for_network(ip): + return sc + return None + + def binding_for_mac(self, mac: str) -> Optional[StaticBinding]: + raw = mac.lower().replace(":", "").replace("-", "") + for b in self.bindings: + if b.mac == raw: + return b + return None + + def clone(self) -> "Config": + return copy.deepcopy(self) diff --git a/dhcpd/lease.py b/dhcpd/lease.py new file mode 100644 index 0000000..0c419a9 --- /dev/null +++ b/dhcpd/lease.py @@ -0,0 +1,189 @@ +"""Lease database: in-memory index + JSON persistence.""" + +from __future__ import annotations + +import ipaddress +import json +import logging +import os +import time +from dataclasses import dataclass, asdict, field +from datetime import datetime, timezone +from typing import Dict, List, Optional + +log = logging.getLogger("dhcpd.lease") + + +@dataclass +class Lease: + mac: str # "aa:bb:cc:dd:ee:ff" + ip: str + hostname: str = "" + offered_at: float = 0.0 + acknowledged_at: float = 0.0 + expires_at: float = 0.0 + static: bool = False # True = from static binding + xid: int = 0 + scope: str = "" + + @property + def is_expired(self) -> bool: + return time.time() > self.expires_at + + @property + def is_active(self) -> bool: + return self.expires_at > time.time() + + def to_json(self) -> dict: + return { + "mac": self.mac, + "ip": self.ip, + "hostname": self.hostname, + "offered_at": self.offered_at, + "acknowledged_at": self.acknowledged_at, + "expires_at": self.expires_at, + "static": self.static, + "scope": self.scope, + } + + +class LeaseDB: + """Thread-ish safe lease store. (Single-threaded server, no lock needed.)""" + + def __init__(self, db_path: str = "leases.db"): + self.db_path = db_path + # mac -> Lease + self.by_mac: Dict[str, Lease] = {} + # ip -> mac + self.by_ip: Dict[str, str] = {} + self._load() + + # ── persistence ────────────────────────────────────────────────────── + + def _load(self) -> None: + if not os.path.exists(self.db_path): + return + try: + with open(self.db_path, "r", encoding="utf-8") as f: + data = json.load(f) + except (json.JSONDecodeError, OSError) as e: + log.warning("Could not load lease DB: %s", e) + return + if isinstance(data, list): + for rec in data: + try: + lease = Lease(**rec) + if lease.is_active or lease.static: + self._put(lease) + except TypeError: + continue + log.info("Loaded %d leases from %s", len(self.by_mac), self.db_path) + + def save(self) -> None: + try: + records = [l.to_json() for l in self.by_mac.values()] + with open(self.db_path, "w", encoding="utf-8") as f: + json.dump(records, f, indent=2) + except OSError as e: + log.error("Failed to save lease DB: %s", e) + + # ── CRUD ───────────────────────────────────────────────────────────── + + def _put(self, lease: Lease) -> None: + # remove stale entry if any + if lease.mac in self.by_mac: + old = self.by_mac[lease.mac] + if old.ip != lease.ip: + self.by_ip.pop(old.ip, None) + if lease.ip in self.by_ip: + del self.by_mac[self.by_ip[lease.ip]] + self.by_mac[lease.mac] = lease + self.by_ip[lease.ip] = lease.mac + + def find_by_mac(self, mac: str) -> Optional[Lease]: + return self.by_mac.get(mac) + + def find_by_ip(self, ip: str) -> Optional[Lease]: + mac = self.by_ip.get(ip) + if mac: + return self.by_mac.get(mac) + return None + + def is_ip_taken(self, ip: str) -> bool: + return ip in self.by_ip + + def add(self, lease: Lease) -> Lease: + self._put(lease) + self.save() + return lease + + def release(self, mac: str) -> bool: + if mac in self.by_mac: + lease = self.by_mac[mac] + self.by_ip.pop(lease.ip, None) + del self.by_mac[mac] + self.save() + log.info("Released lease: %s -> %s", mac, lease.ip) + return True + return False + + def extend(self, mac: str, new_expires: float) -> Optional[Lease]: + lease = self.by_mac.get(mac) + if lease: + lease.expires_at = new_expires + self.save() + return lease + return None + + # ── allocation ─────────────────────────────────────────────────────── + + def allocate_next( + self, + start: str, + end: str, + mac: str, + lease_seconds: int, + xid: int = 0, + hostname: str = "", + scope: str = "", + ) -> Optional[Lease]: + """Walk from start_ip to end_ip, return the first free address.""" + now = time.time() + a_start = ipaddress.ip_address(start) + a_end = ipaddress.ip_address(end) + cur = a_start + while cur <= a_end: + cand = str(cur) + if not self.is_ip_taken(cand): + lease = Lease( + mac=mac, + ip=cand, + hostname=hostname, + offered_at=now, + expires_at=now + lease_seconds, + xid=xid, + scope=scope, + ) + self.add(lease) + return lease + cur = cur + 1 + return None + + def list_all(self) -> List[Lease]: + return sorted(self.by_mac.values(), key=lambda l: l.ip) + + def count(self) -> int: + return len(self.by_mac) + + def count_active(self) -> int: + return sum(1 for l in self.by_mac.values() if l.is_active) + + def prune_expired(self) -> int: + """Remove expired non-static leases. Returns count removed.""" + to_remove = [ + mac for mac, l in self.by_mac.items() + if not l.static and l.is_expired + ] + for mac in to_remove: + self.release(mac) + return len(to_remove) diff --git a/dhcpd/parser.py b/dhcpd/parser.py new file mode 100644 index 0000000..cc3138b --- /dev/null +++ b/dhcpd/parser.py @@ -0,0 +1,332 @@ +"""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 diff --git a/dhcpd/server.py b/dhcpd/server.py new file mode 100644 index 0000000..a3902bc --- /dev/null +++ b/dhcpd/server.py @@ -0,0 +1,303 @@ +"""Main DHCP server loop: sniff packets and dispatch to handlers.""" + +from __future__ import annotations + +import logging +import signal +import socket +import sys +import threading +import time +from typing import Optional + +from scapy.all import sniff, send + +from .config import Config, Scope, StaticBinding +from .lease import Lease, LeaseDB +from .parser import ( + DHCPDISCOVER, DHCPREQUEST, DHCPRELEASE, + DHCPMessage, parse, + build_offer, build_ack, build_nak, +) + +log = logging.getLogger("dhcpd.server") + +# ── counters ────────────────────────────────────────────────────────────── +_msgs = {"DISCOVER": 0, "REQUEST": 0, "RELEASE": 0, "NAK": 0, "total": 0} + + +class DHCPd: + """ + Non-blocking DHCP server. + + Runs scapy sniff in a background thread; packet handler runs in the same + thread so lease DB access is effectively single-threaded. + """ + + def __init__( + self, + config: Config, + lease_db: LeaseDB, + iface: Optional[str] = None, + ): + self.config = config + self.db = lease_db + self.server_ip = config.server.get("ip", "") + if not self.server_ip: + # derive from first scope + for sc in config.scopes: + if sc.enable: + # pick router as server IP if set, else first usable + self.server_ip = sc.router or str(sc.start_ip - 1 if sc.start_ip > 1 else sc.start_ip) + break + + self.iface = iface + self._stop_event = threading.Event() + self._thread: Optional[threading.Thread] = None + + # Bindings index: ip -> binding, mac -> binding + self._binding_by_ip: dict = {} + self._binding_by_mac: dict = {} + for b in config.bindings: + self._binding_by_ip[b.ip] = b + self._binding_by_mac[b.mac] = b + + # ── public ─────────────────────────────────────────────────────────── + + def start(self) -> None: + self._thread = threading.Thread(target=self._run, daemon=True, name="dhcpd-sniff") + self._thread.start() + log.info("DHCP server started (server_ip=%s, interface=%s)", + self.server_ip, self.iface or "") + + def stop(self) -> None: + log.info("Stopping DHCP server...") + self._stop_event.set() + if self._thread: + self._thread.join(timeout=5) + self.db.save() + log.info("DHCP server stopped.") + + @property + def stats(self) -> dict: + return dict(_msgs) + + # ── internal ───────────────────────────────────────────────────────── + + def _run(self) -> None: + bpf = "udp port 67 or udp port 68" + log.info("Starting packet sniff on %s (BPF: %s)", + self.iface or "", bpf) + try: + sniff( + iface=self.iface, + filter=bpf, + prn=self._handle_packet, + stopper=lambda: self._stop_event.is_set(), + store=False, + ) + except Exception as e: + log.error("Sniffer died: %s", e) + log.info("Sniffer thread exited.") + + def _handle_packet(self, pkt) -> None: + _msgs["total"] += 1 + msg = parse(pkt) + if msg is None: + return + log.debug("Recv %s from %s xid=0x%x", msg.msg_name, msg.client_mac, msg.xid) + + try: + if msg.msg_type == DHCPDISCOVER: + _msgs["DISCOVER"] += 1 + self._handle_discover(msg) + elif msg.msg_type == DHCPREQUEST: + _msgs["REQUEST"] += 1 + self._handle_request(msg) + elif msg.msg_type == DHCPRELEASE: + _msgs["RELEASE"] += 1 + self._handle_release(msg) + except Exception as e: + log.exception("Error handling %s", msg.msg_name) + + # ── message handlers ───────────────────────────────────────────────── + + def _handle_discover(self, msg: DHCPMessage) -> None: + """ + DHCPDISCOVER -> DHCPOFFER + + Pick an address based on: + 1. Static binding for this MAC + 2. Requested IP in options (if present and valid) + 3. Auto-allocate from pool + """ + sc, offer_ip = self._pick_address(msg) + if offer_ip is None: + log.warning("DISCOVER: no address available for %s", msg.client_mac) + return + + pkt = build_offer( + req=msg, + offer_ip=offer_ip, + lease_seconds=sc.lease_seconds, + server_ip=self.server_ip, + router=sc.router, + dns=sc.dns_servers, + subnet_mask=sc.subnet_mask, + domain=sc.domain, + ) + send(pkt, iface=self.iface) + log.info("OFFER %s -> %s (lease=%ds)", offer_ip, msg.client_mac, sc.lease_seconds) + + def _handle_request(self, msg: DHCPMessage) -> None: + """ + DHCPREQUEST -> DHCPACK or NAK + + Client may specify the requested IP in two ways: + - ciaddr field (non-zero) = client already has an address + - "requested_addr" DHCP option = newly requested + - yiaddr from the OFFER (server must check) + """ + # Determine which IP the client wants + requested_ip = msg.ciaddr if msg.ciaddr != "0.0.0.0" else None + if not requested_ip: + for o in (msg.options.get("param_req"),): + pass # param_req is what client wants, not the address + # Check if there's a "requested_addr" option + # (parser currently doesn't expose it; derive from OFFER state) + # Fallback: use the lease we offered + lease = self.db.find_by_mac(msg.client_mac) + if lease: + requested_ip = lease.ip + + if requested_ip is None: + log.warning("REQUEST from %s has no requested IP", msg.client_mac) + return + + sc = self.config.scope_for_ip(requested_ip) + if sc is None: + log.warning("REQUEST for %s: no matching scope", requested_ip) + self._send_nak(msg) + _msgs["NAK"] += 1 + return + + # Check if a static binding maps this IP + static = self._binding_by_ip.get(requested_ip) + if static and static.mac != msg.client_mac.replace(":", ""): + log.warning("REQUEST for %s but IP is statically bound to another MAC", + requested_ip) + self._send_nak(msg) + _msgs["NAK"] += 1 + return + + # Mark lease as acknowledged + now = time.time() + lease = self.db.find_by_mac(msg.client_mac) + if lease and lease.ip == requested_ip: + lease.acknowledged_at = now + lease.expires_at = now + sc.lease_seconds + lease.scope = sc.name + lease.hostname = msg.hostname or lease.hostname + self.db.save() + else: + # New allocation that wasn't pre-allocated + lease = Lease( + mac=msg.client_mac, + ip=requested_ip, + hostname=msg.hostname, + offered_at=now, + acknowledged_at=now, + expires_at=now + sc.lease_seconds, + static=static is not None, + scope=sc.name, + ) + self.db.add(lease) + + pkt = build_ack( + req=msg, + offered_ip=requested_ip, + lease_seconds=sc.lease_seconds, + server_ip=self.server_ip, + router=sc.router, + dns=sc.dns_servers, + subnet_mask=sc.subnet_mask, + domain=sc.domain, + ) + send(pkt, iface=self.iface) + log.info("ACK %s -> %s (hostname=%s)", requested_ip, msg.client_mac, msg.hostname) + + def _handle_release(self, msg: DHCPMessage) -> None: + """Handle DHCPRELEASE: remove lease for this MAC.""" + lease = self.db.find_by_mac(msg.client_mac) + if lease: + ip = lease.ip + self.db.release(msg.client_mac) + log.info("RELEASE acknowledged: %s freed", ip) + else: + log.info("RELEASE for unknown MAC %s", msg.client_mac) + + # ── helpers ────────────────────────────────────────────────────────── + + def _pick_address(self, msg: DHCPMessage) -> tuple: + """ + Return (Scope, IP) to offer, or (Scope, None). + """ + # 1. Static binding + binding = self.config.binding_for_mac(msg.client_mac) + if binding: + sc = self.config.scope_for_ip(binding.ip) + if not sc: + log.warning("Static binding for %s points to IP %s not in any scope", + msg.client_mac, binding.ip) + sc = self._select_scope(msg) # fallback + if sc: + # Mark as static in lease if not already + lease = self.db.find_by_mac(msg.client_mac) + if lease and not lease.static: + lease.static = True + lease.expires_at = time.time() + sc.lease_seconds + self.db.save() + return sc, binding.ip + + # 2. Auto-allocate: pick scope based on server/relay context + sc = self._select_scope(msg) + if sc is None: + # fallback: first enabled scope + for s in self.config.scopes: + if s.enable: + sc = s + break + if sc is None: + return sc, None + + # Auto-allocate next free + lease = self.db.find_by_mac(msg.client_mac) + if lease and not lease.is_expired: + return sc, lease.ip + + offer = self.db.allocate_next( + start=str(sc.start_ip), + end=str(sc.end_ip), + mac=msg.client_mac, + lease_seconds=sc.lease_seconds, + xid=msg.xid, + hostname=msg.hostname, + scope=sc.name, + ) + return sc, offer.ip if offer else None + + def _select_scope(self, msg: DHCPMessage) -> Optional[Scope]: + """Pick a scope based on giaddr (relay) or first enabled scope.""" + if msg.giaddr and msg.giaddr != "0.0.0.0": + sc = self.config.scope_for_ip(msg.giaddr) + if sc and sc.enable: + return sc + # Return first enabled scope + for s in self.config.scopes: + if s.enable: + return s + return None + + def _send_nak(self, msg: DHCPMessage) -> None: + pkt = build_nak(msg, self.server_ip) + send(pkt, iface=self.iface) diff --git a/dhcpd/web/app.py b/dhcpd/web/app.py new file mode 100644 index 0000000..19b14c3 --- /dev/null +++ b/dhcpd/web/app.py @@ -0,0 +1,121 @@ +"""Flask web management panel for WinDHCPD.""" + +from __future__ import annotations + +import logging +import threading +from typing import Optional + +from flask import Flask, jsonify, request, render_template + +from ..config import Config, StaticBinding +from ..lease import Lease, LeaseDB +from ..server import DHCPd + +log = logging.getLogger("dhcpd.web") + +app = Flask(__name__) + +# ── global state (set by run.py) ──────────────────────────────────────── +_server: Optional[DHCPd] = None +_config: Optional[Config] = None +_db: Optional[LeaseDB] = None + + +def init(srv: DHCPd, cfg: Config, db: LeaseDB) -> None: + global _server, _config, _db + _server = srv + _config = cfg + _db = db + + +def run(host: str, port: int) -> None: + log.info("Web panel on http://%s:%d", host, port) + app.run(host=host, port=port, threaded=True, debug=False) + + +# ── routes ────────────────────────────────────────────────────────────── + +def _safe_db() -> LeaseDB: + return _db # type: ignore + + +@app.route("/") +def index(): + cfg = _config # type: ignore + db = _safe_db() + srv = _server # type: ignore + leases = db.list_all() if db else [] + scopes = cfg.scopes if cfg else [] + return render_template("index.html", + scopes=scopes, + leases=leases, + stats=srv.stats if srv else {}, + bindings=cfg.bindings if cfg else [], + count_active=db.count_active() if db else 0, + count_total=db.count() if db else 0, + ) + + +@app.route("/api/leases") +def api_leases(): + db = _safe_db() + leases = db.list_all() if db else [] + return jsonify([l.to_json() for l in leases]) + + +@app.route("/api/leases//release", methods=["POST"]) +def api_release(mac: str): + db = _safe_db() + if db: + db.release(mac) + return jsonify({"ok": True}) + + +@app.route("/api/leases/prune", methods=["POST"]) +def api_prune(): + db = _safe_db() + n = 0 + if db: + n = db.prune_expired() + return jsonify({"pruned": n}) + + +@app.route("/api/scopes") +def api_scopes(): + cfg = _config # type: ignore + out = [] + for sc in cfg.scopes: + out.append({ + "name": sc.name, + "network": sc.network, + "start_ip": str(sc.start_ip), + "end_ip": str(sc.end_ip), + "subnet_mask": sc.subnet_mask, + "router": sc.router, + "dns_servers": sc.dns_servers, + "lease_seconds": sc.lease_seconds, + "enable": sc.enable, + }) + return jsonify(out) + + +@app.route("/api/stats") +def api_stats(): + srv = _server # type: ignore + db = _safe_db() + stats: dict = {} + if srv: + stats.update(srv.stats) + if db: + stats["leases_total"] = db.count() + stats["leases_active"] = db.count_active() + return jsonify(stats) + + +@app.route("/api/bindings") +def api_bindings(): + cfg = _config # type: ignore + return jsonify([{"mac": b.mac_display, "ip": b.ip, + "scope": b.scope_name, "hostname": b.hostname} + for b in cfg.bindings]) diff --git a/dhcpd/web/templates/index.html b/dhcpd/web/templates/index.html new file mode 100644 index 0000000..aa674ad --- /dev/null +++ b/dhcpd/web/templates/index.html @@ -0,0 +1,150 @@ + + + + + +WinDHCPD 管理面板 + + + + +

🖥 WinDHCPD 管理面板

+

Windows DHCP 服务器 — 实时监控

+ +
+
{{ stats.get("DISCOVER", 0) }}
DISCOVER
+
{{ stats.get("REQUEST", 0) }}
REQUEST
+
{{ count_total }}
租约总数
+
{{ count_active }}
活跃租约
+
{{ stats.get("NAK", 0) }}
NAK (拒绝)
+
+ +
+{% for sc in scopes %} +
+

{{ sc.name }} {% if not sc.enable %}禁用{% endif %}

+
+ 网段: {{ sc.network }}
+ 地址池: {{ sc.start_ip }} → {{ sc.end_ip }}
+ 网关: {{ sc.router or "—" }}
+ DNS: {{ ", ".join(sc.dns_servers) or "—" }}
+ 租期: {{ sc.lease_seconds }}s +
+
+{% endfor %} +
+ +{% if bindings %} +

静态绑定

+
+ +{% for b in bindings %} + +{% endfor %} +
MACIP作用域主机名
{{ b.mac_display }}{{ b.ip }}{{ b.scope_name or "—" }}{{ b.hostname or "—" }}
+
+{% endif %} + +

租约列表

+
+ +{% for l in leases %} + + + + + + + + +{% endfor %} +
MACIP主机名状态到期
{{ l.mac }}{{ l.ip }}{{ l.hostname or "—" }}{% if l.is_active %}活跃{% else %}已过期{% endif %}{{ l.expires_at }}
+
+ +
+ + + 自动刷新: 5s +
+ + + + diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..8c0de99 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +# WinDHCPD 依赖 +scapy>=2.5.0 +flask>=3.0 +pyyaml>=6.0 diff --git a/run.py b/run.py new file mode 100644 index 0000000..f633a17 --- /dev/null +++ b/run.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +""" +WinDHCPD - Windows-native DHCP server +====================================== +Python 3 + Scapy based lightweight DHCP server for Windows. + +Usage: + python run.py # uses dhcpd.yaml in cwd + python run.py config.yaml # specify config file + python run.py config.yaml -i # specify interface + python run.py --help + +Web panel: http://localhost:8080 +""" + +from __future__ import annotations + +import argparse +import logging +import os +import signal +import sys +import threading + +from dhcpd.config import Config +from dhcpd.lease import LeaseDB +from dhcpd.server import DHCPd +from dhcpd.web import app as web_app + +# ── logging setup ──────────────────────────────────────────────────────── + +def setup_logging(level: str = "INFO", log_file: str = "dhcpd.log") -> None: + """Configure logging to both console and file.""" + numeric = getattr(logging, level.upper(), logging.INFO) + fmt = logging.Formatter( + "%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + # Console handler + ch = logging.StreamHandler(sys.stdout) + ch.setFormatter(fmt) + ch.setLevel(numeric) + + # File handler + try: + fh = logging.FileHandler(log_file, encoding="utf-8") + fh.setFormatter(fmt) + fh.setLevel(numeric) + except OSError: + fh = None + + root = logging.getLogger() + root.setLevel(numeric) + root.addHandler(ch) + if fh: + root.addHandler(fh) + + # Scapy is chatty + logging.getLogger("scapy").setLevel(logging.WARNING) + + +# ── main ───────────────────────────────────────────────────────────────── + +def main(): + parser = argparse.ArgumentParser(description="WinDHCPD - DHCP server for Windows") + parser.add_argument("config", nargs="?", default="dhcpd.yaml", + help="Path to YAML config file (default: dhcpd.yaml)") + parser.add_argument("-i", "--interface", default=None, + help="Network interface name to listen on") + parser.add_argument("--no-web", action="store_true", + help="Disable web management panel") + parser.add_argument("--web-port", type=int, default=0, + help="Web panel port (overrides config)") + args = parser.parse_args() + + # Load config + config_path = args.config + if not os.path.exists(config_path): + print(f"[ERROR] Config not found: {config_path}") + print(f" Create one from dhcpd.yaml.example") + sys.exit(1) + + cfg = Config(config_path) + setup_logging(cfg.log_level, cfg.log_file) + logging.getLogger("dhcpd").info("=" * 60) + logging.getLogger("dhcpd").info("WinDHCPD starting") + logging.getLogger("dhcpd").info("Config: %s", config_path) + logging.getLogger("dhcpd").info("Scopes: %s", ", ".join(s.name for s in cfg.scopes)) + logging.getLogger("dhcpd").info("Bindings: %d", len(cfg.bindings)) + + # Lease database + db_path = os.path.join(os.path.dirname(os.path.abspath(config_path)), "leases.db") + db = LeaseDB(db_path) + logging.getLogger("dhcpd").info("Lease DB: %s (%d leases loaded)", + db_path, db.count()) + + # Server + srv = DHCPd(cfg, db, iface=args.interface) + srv.start() + + # Web panel + if not args.no_web: + web_port = args.web_port or cfg.web_port + web_host = cfg.web_host + web_app.init(srv, cfg, db) + t = threading.Thread( + target=web_app.run, + args=(web_host, web_port), + daemon=True, + name="web-panel", + ) + t.start() + logging.getLogger("dhcpd").info("Web panel: http://%s:%d", web_host, web_port) + + # Graceful shutdown + def shutdown(sig, frame): + logging.getLogger("dhcpd").info("Received signal %s, shutting down...", sig) + srv.stop() + sys.exit(0) + + signal.signal(signal.SIGINT, shutdown) + signal.signal(signal.SIGTERM, shutdown) + + logging.getLogger("dhcpd").info("Server running. Press Ctrl+C to stop.") + # Keep main thread alive + try: + while True: + import time as _t + _t.sleep(1) + except KeyboardInterrupt: + shutdown(signal.SIGINT, None) + + +if __name__ == "__main__": + main()