Your Name
ea8eb2b5f9
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
137 lines
4.4 KiB
Python
137 lines
4.4 KiB
Python
#!/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()
|