commit a2a1f36092fafc4d48914ec7e963c50d140a46e5 Author: xuejin.wang Date: Thu Oct 8 15:35:20 2026 +0800 便携 DHCP 服务器 v1.0: 单文件 DHCP 引擎 + tkinter GUI - 完整 DHCP 流程: DISCOVER/OFFER/REQUEST/ACK/NAK/DECLINE/RELEASE/INFORM - 租约管理: 搜索/筛选/详情栏/解除/批量清除/CSV导出 - 保留地址: MAC→IP 绑定、独立租期、在线状态、导入导出 - 静态绑定/PXE 支持/冲突检测/租约持久化/防火墙一键放行 - 协议自检 25 项 + GUI 冒烟 21 项全部通过 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..dd43339 --- /dev/null +++ b/.gitignore @@ -0,0 +1,14 @@ +# 运行时生成的本地数据(不进版本库) +dhcp_config.json +dhcp_leases.json +__pycache__/ + +# 打包产物(体积大, 不入 git) +*.exe +dist/ +build/ +*.spec + +# 系统杂项 +Thumbs.db +desktop.ini diff --git a/dhcp_server_app.py b/dhcp_server_app.py new file mode 100644 index 0000000..cea06bc --- /dev/null +++ b/dhcp_server_app.py @@ -0,0 +1,2090 @@ +# -*- coding: utf-8 -*- +""" +便携 DHCP 服务器 (Portable DHCP Server) v1.0 +============================================ + +单文件、零第三方依赖(仅 Python 标准库)、带图形界面的轻量 DHCP 服务器， +适合在 Windows 电脑上为测试设备、摄像头、开发板、PXE 网络启动等临时分配 IP。 + +功能: + * 完整 DHCP 流程: DISCOVER / OFFER / REQUEST / ACK / NAK / DECLINE / RELEASE / INFORM + * 可视化配置: 地址池 / 子网掩码 / 网关 / DNS / 域名 / 租约时长 + * 静态 IP 绑定 (MAC -> IP) + * 分配前 Ping 冲突检测 + * 租约持久化(重启服务后客户端尽量保持原 IP) + * 实时日志、客户端列表、CSV 导出、防火墙一键放行 + * PXE 支持: TFTP 服务器(option 66) + 启动文件(option 67) + +运行: + python dhcp_server_app.py 启动图形界面 + python dhcp_server_app.py --selftest 运行内置自检(不需要管理员) + +注意: + * Windows 防火墙可能拦截发往 UDP 67 的入站报文, 请使用界面上的 + 「防火墙放行」按钮(需要管理员权限)。 + * 若启动失败提示端口被占用, 通常是系统开启了「Internet 连接共享 / 移动热点」, + 其内置的 DHCP 分配器占用了 UDP 67。 +""" + +import csv +import ctypes +import ipaddress +import json +import os +import queue +import re +import socket +import struct +import subprocess +import sys +import threading +import time +from datetime import datetime + +import tkinter as tk +from tkinter import filedialog, messagebox, ttk + +# --------------------------------------------------------------------------- +# DHCP 常量 +# --------------------------------------------------------------------------- + +MAGIC_COOKIE = b'\x63\x82\x53\x63' # 99.130.83.99 + +# 报文类型 +MSG_DISCOVER = 1 +MSG_OFFER = 2 +MSG_REQUEST = 3 +MSG_DECLINE = 4 +MSG_ACK = 5 +MSG_NAK = 6 +MSG_RELEASE = 7 +MSG_INFORM = 8 + +MSG_NAMES = { + 1: 'DISCOVER', 2: 'OFFER', 3: 'REQUEST', 4: 'DECLINE', + 5: 'ACK', 6: 'NAK', 7: 'RELEASE', 8: 'INFORM', +} + +# 常用 DHCP 选项 +OPT_SUBNET_MASK = 1 +OPT_ROUTER = 3 +OPT_DNS = 6 +OPT_HOSTNAME = 12 +OPT_DOMAIN_NAME = 15 +OPT_REQUESTED_IP = 50 +OPT_LEASE_TIME = 51 +OPT_MSG_TYPE = 53 +OPT_SERVER_ID = 54 +OPT_PARAM_REQ_LIST = 55 +OPT_RENEWAL_TIME = 58 +OPT_REBINDING_TIME = 59 +OPT_CLIENT_ID = 61 +OPT_TFTP_SERVER = 66 +OPT_BOOTFILE_NAME = 67 +OPT_END = 255 + +DHCP_SERVER_PORT = 67 +DHCP_CLIENT_PORT = 68 + +STATE_NAMES = {'offer': '预分配', 'active': '已分配', 'expired': '已过期'} + + +def mkopt(code, value): + """构造一个 DHCP 选项 (code, length, value)。""" + if isinstance(value, str): + value = value.encode('utf-8', 'replace') + return bytes([code, len(value)]) + value + + +def ip_to_bytes(ip): + return socket.inet_aton(ip if ip else '0.0.0.0') + + +def bytes_to_ip(b): + return socket.inet_ntoa(b) if b else '0.0.0.0' + + +def norm_mac(text): + """把各种 MAC 写法统一为 AA:BB:CC:DD:EE:FF；不合法返回 None。""" + if not text: + return None + t = text.strip().replace('-', ':').replace('.', ':').replace(' ', '') + if re.fullmatch(r'([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}', t): + return t.upper() + t2 = t.replace(':', '') + if re.fullmatch(r'[0-9A-Fa-f]{12}', t2): + return ':'.join(t2[i:i + 2] for i in range(0, 12, 2)).upper() + return None + + +def mac_to_bytes(mac): + return bytes.fromhex(mac.replace(':', '').replace('-', '')) + + +def format_secs(secs): + """把秒数格式化为 'x天x小时x分'。""" + secs = int(secs) + if secs < 0: + return '已过期' + d, r = divmod(secs, 86400) + h, r = divmod(r, 3600) + m, _ = divmod(r, 60) + if d: + return f'{d}天{h}小时' + if h: + return f'{h}小时{m}分' + return f'{m}分{secs % 60}秒' + + +def list_local_ipv4(): + """枚举本机所有 IPv4 地址。""" + ips = set() + try: + for ip in socket.gethostbyname_ex(socket.gethostname())[2]: + if ':' not in ip: + ips.add(ip) + except OSError: + pass + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + s.connect(('8.8.8.8', 80)) + ips.add(s.getsockname()[0]) + finally: + s.close() + except OSError: + pass + ips.discard('0.0.0.0') + return sorted(ips) + + +def is_admin(): + """当前进程是否具有管理员权限。""" + try: + return ctypes.windll.shell32.IsUserAnAdmin() != 0 + except Exception: + return True + + +# --------------------------------------------------------------------------- +# DHCP 报文解析 / 构造 +# --------------------------------------------------------------------------- + +class DhcpRequest: + """已解析的 DHCP 请求报文。""" + + __slots__ = ('op', 'htype', 'hlen', 'hops', 'xid_raw', 'xid', 'flags_raw', + 'flags', 'broadcast', 'ciaddr_raw', 'ciaddr', 'yiaddr', + 'siaddr', 'giaddr_raw', 'giaddr', 'chaddr_raw', 'chaddr_mac', + 'sname', 'bootfile', 'options', 'hostname', 'client_id_mac', + 'requested_ip') + + def __repr__(self): + return (f'') + + +def parse_dhcp_packet(data): + """解析 DHCP 报文; 非 DHCP 报文返回 None。""" + if len(data) < 240 or data[236:240] != MAGIC_COOKIE: + return None + req = DhcpRequest() + req.op = data[0] + req.htype = data[1] or 1 + req.hlen = data[2] or 6 + req.hops = data[3] + req.xid_raw = data[4:8] + req.xid = struct.unpack('!I', data[4:8])[0] + req.flags_raw = data[10:12] + req.flags = struct.unpack('!H', data[10:12])[0] + req.broadcast = bool(req.flags & 0x8000) + req.ciaddr_raw = data[12:16] + req.ciaddr = bytes_to_ip(data[12:16]) + req.yiaddr = bytes_to_ip(data[16:20]) + req.siaddr = bytes_to_ip(data[20:24]) + req.giaddr_raw = data[24:28] + req.giaddr = bytes_to_ip(data[24:28]) + req.chaddr_raw = data[28:44] + req.chaddr_mac = ':'.join(f'{b:02X}' for b in data[28:28 + req.hlen]) + req.sname = data[44:108].split(b'\x00')[0] + req.bootfile = data[108:236].split(b'\x00')[0] + + # 选项区 + options = {} + i = 240 + n = len(data) + while i < n: + code = data[i] + if code == 0: # pad + i += 1 + continue + if code == OPT_END: # end + break + if i + 1 >= n: + break + length = data[i + 1] + options[code] = data[i + 2:i + 2 + length] + i += 2 + length + req.options = options + + # 主机名 (option 12) + raw = options.get(OPT_HOSTNAME) + req.hostname = '' + if raw: + try: + req.hostname = raw.decode('utf-8') + except UnicodeDecodeError: + req.hostname = raw.decode('gbk', 'replace') + req.hostname = req.hostname.strip()[:40] + + # 客户端标识 (option 61): type(1) + 标识, type=1 时即 MAC + req.client_id_mac = None + cid = options.get(OPT_CLIENT_ID) + if cid and len(cid) == 7 and cid[0] == 1: + req.client_id_mac = ':'.join(f'{b:02X}' for b in cid[1:]) + + raw = options.get(OPT_REQUESTED_IP) + req.requested_ip = bytes_to_ip(raw) if raw and len(raw) == 4 else None + return req + + +def parse_ip_option(value): + """把 4 字节选项值转成 IP 字符串, 失败返回 None。""" + if value and len(value) == 4: + return socket.inet_ntoa(value) + return None + + +def build_reply(req, msg_type, offered_ip, cfg, lease_secs): + """根据请求构造应答报文 (BOOTREPLY)。""" + pkt = bytearray(240) + pkt[0] = 2 # op = BOOTREPLY + pkt[1] = req.htype + pkt[2] = req.hlen + pkt[3] = 0 + pkt[4:8] = req.xid_raw + pkt[10:12] = req.flags_raw + # RENEWING/REBINDING/INFORM 时回显 ciaddr + renewing = (req.ciaddr_raw != b'\x00\x00\x00\x00' + and OPT_SERVER_ID not in req.options) + if renewing or msg_type == MSG_INFORM: + pkt[12:16] = req.ciaddr_raw + pkt[16:20] = ip_to_bytes(offered_ip) + pkt[20:24] = ip_to_bytes(cfg['server_ip']) + pkt[24:28] = req.giaddr_raw + pkt[28:44] = req.chaddr_raw + + bootfile = (cfg.get('bootfile') or '')[:128] + if bootfile: + pkt[108:108 + len(bootfile)] = bootfile.encode('utf-8', 'replace') + + pkt[236:240] = MAGIC_COOKIE + + opts = bytearray() + opts += mkopt(OPT_MSG_TYPE, bytes([msg_type])) + opts += mkopt(OPT_SERVER_ID, ip_to_bytes(cfg['server_ip'])) + + if msg_type == MSG_NAK: + opts += bytes([OPT_END]) + pkt += opts + return bytes(pkt) + + if msg_type != MSG_INFORM and lease_secs: + lease_secs = int(lease_secs) + opts += mkopt(OPT_LEASE_TIME, struct.pack('!I', lease_secs)) + opts += mkopt(OPT_RENEWAL_TIME, struct.pack('!I', lease_secs // 2)) + opts += mkopt(OPT_REBINDING_TIME, struct.pack('!I', lease_secs * 7 // 8)) + if cfg.get('mask'): + opts += mkopt(OPT_SUBNET_MASK, ip_to_bytes(cfg['mask'])) + if cfg.get('gateway'): + opts += mkopt(OPT_ROUTER, ip_to_bytes(cfg['gateway'])) + if cfg.get('dns'): + for dns in cfg['dns']: + opts += mkopt(OPT_DNS, ip_to_bytes(dns)) + if cfg.get('domain'): + opts += mkopt(OPT_DOMAIN_NAME, cfg['domain']) + if cfg.get('tftp'): + opts += mkopt(OPT_TFTP_SERVER, cfg['tftp']) + if cfg.get('bootfile'): + opts += mkopt(OPT_BOOTFILE_NAME, cfg['bootfile']) + + opts += bytes([OPT_END]) + pkt += opts + if len(pkt) < 300: # BOOTP 报文至少 300 字节 + pkt += b'\x00' * (300 - len(pkt)) + return bytes(pkt) + + +def make_client_packet(msg_type, mac, xid, *, ciaddr='0.0.0.0', + requested_ip=None, server_id=None, hostname=None, + flags=0): + """构造客户端报文(用于内置自检)。mac 为 'AA:BB:...' 形式。""" + mac_bytes = mac_to_bytes(mac) + pkt = bytearray(240) + pkt[0] = 1 + pkt[1] = 1 + pkt[2] = 6 + pkt[4:8] = struct.pack('!I', xid) + pkt[10:12] = struct.pack('!H', flags) + pkt[12:16] = ip_to_bytes(ciaddr) + pkt[28:34] = mac_bytes + pkt[236:240] = MAGIC_COOKIE + opts = bytearray() + opts += mkopt(OPT_MSG_TYPE, bytes([msg_type])) + opts += mkopt(OPT_CLIENT_ID, b'\x01' + mac_bytes) + if requested_ip: + opts += mkopt(OPT_REQUESTED_IP, ip_to_bytes(requested_ip)) + if server_id: + opts += mkopt(OPT_SERVER_ID, ip_to_bytes(server_id)) + if hostname: + opts += mkopt(OPT_HOSTNAME, hostname) + opts += mkopt(OPT_PARAM_REQ_LIST, bytes([1, 3, 6, 15, 51])) + opts += bytes([OPT_END]) + pkt += opts + return bytes(pkt) + + +# --------------------------------------------------------------------------- +# DHCP 服务器引擎 +# --------------------------------------------------------------------------- + +class DhcpServerEngine: + """DHCP 服务引擎: 在独立线程中监听 UDP 端口并处理租约。""" + + OFFER_HOLD_SECS = 20 # OFFER 未确认的保留时间 + SWEEP_INTERVAL = 15 # 租约扫描周期 + SAVE_INTERVAL = 60 # 租约落盘周期 + + def __init__(self, config, events, lease_path=None): + self.cfg = dict(config) + self.events = events # queue.Queue, 事件回调 + self.lease_path = lease_path + self.port = int(self.cfg.get('port', DHCP_SERVER_PORT)) + self.leases = {} # ip -> {mac, hostname, state, expires, source} + self.mac_map = {} # mac -> ip + self.conflicts = set() # 已知冲突/被占用 IP + self.static = {} # mac -> ip + self.static_notes = {} # mac -> 备注 + self.static_lease_secs = {} # mac -> 自定义租约秒数(0=使用全局默认) + self._lock = threading.RLock() + self.set_static_bindings(self.cfg.get('static_bindings') or []) + self.sock = None + self.running = False + self.started_at = None + self.counters = {'discover': 0, 'offer': 0, 'request': 0, + 'ack': 0, 'nak': 0, 'release': 0, 'decline': 0} + self._thread = None + self._last_sweep = 0.0 + self._last_save = 0.0 + self._dirty = False + + # ---- 事件 --------------------------------------------------------------- + + def _emit(self, *event): + try: + self.events.put(event) + except Exception: + pass + + def _log(self, text, level='packet'): + self._emit('log', level, text) + + # ---- 生命周期 ------------------------------------------------------------ + + def start(self): + if self.running: + return + self._load_leases() + self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + except OSError: + pass + try: + self.sock.bind(('', self.port)) + except OSError: + self.sock.close() + self.sock = None + raise + self.sock.settimeout(0.5) + self.running = True + self.started_at = time.time() + self._thread = threading.Thread(target=self._loop, name='dhcp-engine', daemon=True) + self._thread.start() + self._log(f'DHCP 服务已启动: {self.cfg["server_ip"]} · UDP {self.port} · ' + f'地址池 {self.cfg["pool_start"]} ~ {self.cfg["pool_end"]}', 'ok') + self._emit('update') + + def stop(self, reason='手动停止'): + if not self.running: + return + self.running = False + if self._thread: + self._thread.join(timeout=5) + self._thread = None + if self.sock: + try: + self.sock.close() + except OSError: + pass + self.sock = None + self._save_leases() + self._log(f'DHCP 服务已停止({reason})，租约已保存。', 'warn') + self._emit('update') + + def _loop(self): + while self.running: + try: + data, addr = self.sock.recvfrom(4096) + except socket.timeout: + self._periodic() + continue + except OSError: + break # socket 被关闭 + try: + self._handle(data, addr) + except Exception as exc: # 单个报文出错不影响服务 + self._log(f'处理报文异常: {exc!r}', 'err') + self._periodic() + self._save_leases() + + def _periodic(self): + now = time.time() + if now - self._last_sweep >= self.SWEEP_INTERVAL: + self._last_sweep = now + if self._sweep_expired(): + self._emit('update') + if now - self._last_save >= self.SAVE_INTERVAL and self._dirty: + self._last_save = now + self._save_leases() + + # ---- 租约维护 ------------------------------------------------------------ + + def _sweep_expired(self): + """把过期租约标记为 expired, 返回是否有变化。""" + now = time.time() + changed = False + with self._lock: + for ip, lease in self.leases.items(): + if lease['state'] == 'active' and lease['expires'] <= now: + lease['state'] = 'expired' + changed = True + self._dirty = True + self._log(f'[租约过期] {lease["mac"]} 的 {ip} 租约到期', 'info') + return changed + + def _set_lease(self, ip, mac, hostname, state, source): + with self._lock: + old = self.leases.get(ip) + if old and old['mac'] != mac: + self.mac_map.pop(old['mac'], None) + self.leases[ip] = {'mac': mac, 'hostname': hostname or '', + 'state': state, 'expires': 0.0, + 'lease_secs': 0, 'source': source, + 'updated': time.time()} + self.mac_map[mac] = ip + self._dirty = True + + def _remove_lease(self, ip, new_state=None): + with self._lock: + lease = self.leases.pop(ip, None) + if lease: + self.mac_map.pop(lease['mac'], None) + self._dirty = True + if new_state: + self._log(f'租约 {ip} ({lease["mac"]}) 状态 -> {STATE_NAMES.get(new_state, new_state)}', 'info') + return lease + + def set_static_bindings(self, bindings): + """bindings: [{'mac':..., 'ip':..., 'note':..., 'lease_hours':...}, ...]""" + table = {} + notes = {} + lease_secs = {} + for item in bindings or []: + mac = norm_mac(item.get('mac', '')) + ip = item.get('ip', '') + if mac and ip: + table[mac] = ip + notes[mac] = (item.get('note') or '').strip() + try: + hours = int(item.get('lease_hours') or 0) + if 1 <= hours <= 8760: + lease_secs[mac] = hours * 3600 + except (TypeError, ValueError): + pass + with self._lock: + self.static = table + self.static_notes = notes + self.static_lease_secs = lease_secs + + def _lease_secs_for(self, mac): + """该 MAC 的租约时长(秒): 保留地址可单独自定义, 否则用全局默认。""" + secs = self.static_lease_secs.get(mac, 0) + return secs if secs else int(self.cfg['lease_hours']) * 3600 + + # ---- 地址分配 ------------------------------------------------------------ + + def _excluded_ips(self): + """不可分配的地址: 服务器自身 / 网关 / 静态绑定 / 网络号与广播地址。""" + excl = {self.cfg['server_ip']} + if self.cfg.get('gateway'): + excl.add(self.cfg['gateway']) + excl.update(self.static.values()) + try: + net = ipaddress.ip_network(f'{self.cfg["server_ip"]}/{self.cfg["mask"]}', strict=False) + excl.add(str(net.network_address)) + excl.add(str(net.broadcast_address)) + except ValueError: + pass + return excl + + def _candidate_ips(self, mac, requested_ip=None, hostname=''): + """按优先级返回候选 IP 列表(未做冲突检测)。""" + pool_lo = int(ipaddress.ip_address(self.cfg['pool_start'])) + pool_hi = int(ipaddress.ip_address(self.cfg['pool_end'])) + excl = self._excluded_ips() + now = time.time() + with self._lock: + # 1) 静态绑定 + if mac in self.static: + return [self.static[mac]] + # 2) 粘性: 同一 MAC 历史租约 + sticky = self.mac_map.get(mac) + if sticky: + lease = self.leases.get(sticky) + if lease and sticky not in excl and sticky not in self.conflicts: + if lease['state'] in ('active', 'offer', 'expired'): + return [sticky] + # 3) 客户端点名的地址(合法且空闲) + if requested_ip: + ip_obj = None + try: + ip_obj = ipaddress.ip_address(requested_ip) + except ValueError: + ip_obj = None + if ip_obj and pool_lo <= int(ip_obj) <= pool_hi \ + and requested_ip not in excl: + lease = self.leases.get(requested_ip) + if lease is None or lease['mac'] == mac or lease['state'] == 'expired': + if requested_ip not in self.conflicts: + return [requested_ip] + # 4) 顺序扫描地址池 + candidates = [] + for n in range(pool_lo, pool_hi + 1): + ip = str(ipaddress.ip_address(n)) + if ip in excl or ip in self.conflicts: + continue + lease = self.leases.get(ip) + if lease and lease['state'] in ('active', 'offer') and lease['mac'] != mac: + continue + if lease and lease['state'] == 'offer' and lease['mac'] == mac \ + and lease['expires'] > now: + return [ip] # 自己的未决 OFFER + candidates.append(ip) + if len(candidates) >= 3: + break + return candidates + + def _ip_in_use(self, ip): + """Ping 检测地址是否已被其他设备占用。""" + if not self.cfg.get('check_conflict', True): + return False + try: + out = subprocess.run(['ping', '-n', '1', '-w', '500', ip], + capture_output=True, timeout=3) + text = (out.stdout or b'').decode('gbk', 'ignore') + return 'TTL' in text.upper() + except Exception: + return False + + def _allocate(self, mac, hostname='', requested_ip=None): + """为 MAC 选择一个 IP; 无可用地址返回 None。""" + candidates = self._candidate_ips(mac, requested_ip, hostname) + for ip in candidates: + if self._ip_in_use(ip): + if ip not in self.conflicts: + self.conflicts.add(ip) + self._log(f'[冲突检测] {ip} 已被其他设备占用(Ping 有响应)，跳过分配', 'warn') + continue + with self._lock: + lease = self.leases.get(ip) + if lease and lease['mac'] != mac and lease['state'] in ('active', 'offer'): + continue # 刚刚被别人抢走 + return ip + return None + + # ---- 报文处理 ------------------------------------------------------------ + + def _handle(self, data, addr): + req = parse_dhcp_packet(data) + if req is None or req.op != 1: + return + raw_type = req.options.get(OPT_MSG_TYPE) + if not raw_type: + return # 纯 BOOTP, 忽略 + msg_type = raw_type[0] + + mac = req.client_id_mac or req.chaddr_mac + host = f'主机名 {req.hostname}' if req.hostname else '无主机名' + + if msg_type == MSG_DISCOVER: + self._on_discover(req, addr, mac, host) + elif msg_type == MSG_REQUEST: + self._on_request(req, addr, mac, host) + elif msg_type == MSG_DECLINE: + self._on_decline(req, mac, host) + elif msg_type == MSG_RELEASE: + self._on_release(req, mac, host) + elif msg_type == MSG_INFORM: + self._on_inform(req, addr, mac, host) + + def _reply(self, req, addr, msg_type, offered_ip, lease_secs): + data = build_reply(req, msg_type, offered_ip, self.cfg, lease_secs) + giaddr = req.giaddr + if giaddr != '0.0.0.0': # 经过中继 + target = (giaddr, DHCP_SERVER_PORT) + elif addr[0] not in ('0.0.0.0', '255.255.255.255'): + target = (addr[0], addr[1]) # 对端已有 IP(测试/中继场景) + elif req.broadcast or req.ciaddr == '0.0.0.0': + target = ('255.255.255.255', DHCP_CLIENT_PORT) + else: + target = (req.ciaddr, DHCP_CLIENT_PORT) + try: + self.sock.sendto(data, target) + except OSError as exc: + self._log(f'发送应答失败({target[0]}:{target[1]}): {exc}', 'err') + + def _on_discover(self, req, addr, mac, host): + self.counters['discover'] += 1 + ip = self._allocate(mac, req.hostname, req.requested_ip) + if ip is None: + self._log(f'[DISCOVER] {mac} ({host}) 请求地址，但地址池已无可用 IP', 'warn') + return + self._set_lease(ip, mac, req.hostname, 'offer', + 'static' if mac in self.static else 'dynamic') + with self._lock: + self.leases[ip]['expires'] = time.time() + self.OFFER_HOLD_SECS + self.counters['offer'] += 1 + lease_secs = self._lease_secs_for(mac) + with self._lock: + self.leases[ip]['lease_secs'] = lease_secs + self._log(f'[DISCOVER] {mac} ({host}) -> 预分配 {ip}，已回复 OFFER') + self._reply(req, addr, MSG_OFFER, ip, lease_secs) + self._emit('update') + + def _on_request(self, req, addr, mac, host): + self.counters['request'] += 1 + lease_secs = self._lease_secs_for(mac) + server_id = parse_ip_option(req.options.get(OPT_SERVER_ID)) + requested = req.requested_ip or (req.ciaddr if req.ciaddr != '0.0.0.0' else None) + + # 客户端选择了其他 DHCP 服务器 -> 撤销我们的预分配 + if server_id and server_id != self.cfg['server_ip']: + with self._lock: + ours = self.mac_map.get(mac) + if ours and self.leases.get(ours, {}).get('state') == 'offer': + self._remove_lease(ours) + self._log(f'[REQUEST] {mac} 选择了其他服务器 {server_id}，已撤销预分配 {ours}', 'info') + self._emit('update') + return + + with self._lock: + lease = self.leases.get(requested) if requested else None + static_ip = self.static.get(mac) + + granted = None + if static_ip: + granted = static_ip # 静态绑定优先 + elif lease and lease['mac'] == mac and lease['state'] in ('offer', 'active'): + granted = requested # 正常确认/续租 + elif requested: + # INIT-REBOOT: 服务重启后客户端直接请求旧地址 + pool_lo = int(ipaddress.ip_address(self.cfg['pool_start'])) + pool_hi = int(ipaddress.ip_address(self.cfg['pool_end'])) + excl = self._excluded_ips() + try: + in_pool = pool_lo <= int(ipaddress.ip_address(requested)) <= pool_hi + except ValueError: + in_pool = False + if in_pool and requested not in excl and requested not in self.conflicts \ + and not self._ip_in_use(requested): + granted = requested + + if granted: + if static_ip and granted != requested and requested: + self._log(f'[REQUEST] {mac} 请求 {requested}，按静态绑定改配 {static_ip}', 'warn') + self._set_lease(granted, mac, req.hostname, 'active', + 'static' if static_ip == granted else 'dynamic') + with self._lock: + self.leases[granted]['expires'] = time.time() + lease_secs + self.leases[granted]['lease_secs'] = lease_secs + self.counters['ack'] += 1 + self._log(f'[REQUEST] {mac} ({host}) 确认租约 {granted}，' + f'有效期 {format_secs(lease_secs)}，已回复 ACK', 'ok') + self._reply(req, addr, MSG_ACK, granted, lease_secs) + else: + self.counters['nak'] += 1 + self._log(f'[REQUEST] {mac} 请求 {requested or "?"} 不可用，已回复 NAK', 'warn') + self._reply(req, addr, MSG_NAK, None, 0) + self._emit('update') + + def _on_decline(self, req, mac, host): + self.counters['decline'] += 1 + ip = req.requested_ip + if ip: + self.conflicts.add(ip) + self._remove_lease(ip) + self._log(f'[DECLINE] {mac} ({host}) 报告 {ip} 存在地址冲突，已加入冲突名单', 'warn') + self._emit('update') + + def _on_release(self, req, mac, host): + self.counters['release'] += 1 + ip = req.ciaddr + with self._lock: + lease = self.leases.get(ip) + if lease and lease['mac'] == mac: + self._remove_lease(ip) + self._log(f'[RELEASE] {mac} ({host}) 主动释放了 {ip}', 'info') + self._emit('update') + else: + ours = self.mac_map.get(mac) + if ours: + self._remove_lease(ours) + self._log(f'[RELEASE] {mac} 释放了 {ours}', 'info') + self._emit('update') + + def _on_inform(self, req, addr, mac, host): + self._log(f'[INFORM] {mac} ({host}) 请求网络参数，已回复 ACK(仅参数)', 'info') + self._reply(req, addr, MSG_ACK, None, 0) + + # ---- 对外查询 ------------------------------------------------------------ + + def get_rows(self): + """客户端列表快照。""" + now = time.time() + rows = [] + with self._lock: + for ip in sorted(self.leases, key=lambda x: int(ipaddress.ip_address(x))): + lease = self.leases[ip] + remain = lease['expires'] - now if lease['state'] == 'active' else 0 + rows.append({ + 'ip': ip, + 'mac': lease['mac'], + 'hostname': lease['hostname'], + 'state': STATE_NAMES.get(lease['state'], lease['state']), + 'source': '静态绑定' if lease.get('source') == 'static' else '动态分配', + 'expires': (datetime.fromtimestamp(lease['expires']).strftime('%Y-%m-%d %H:%M:%S') + if lease['state'] == 'active' else '—'), + 'remain': format_secs(remain) if lease['state'] == 'active' else '—', + 'note': self.static_notes.get(lease['mac'], ''), + 'last_seen': datetime.fromtimestamp(lease['updated']).strftime('%m-%d %H:%M:%S'), + '_state': lease['state'], + }) + return rows + + def stats(self): + try: + pool_lo = int(ipaddress.ip_address(self.cfg['pool_start'])) + pool_hi = int(ipaddress.ip_address(self.cfg['pool_end'])) + except ValueError: + pool_lo = pool_hi = 0 + pool_size = max(0, pool_hi - pool_lo + 1) + with self._lock: + active = sum(1 for l in self.leases.values() if l['state'] == 'active') + offered = sum(1 for l in self.leases.values() if l['state'] == 'offer') + expired = sum(1 for l in self.leases.values() if l['state'] == 'expired') + static_in_pool = sum(1 for ip in self.static.values() + if pool_lo <= int(ipaddress.ip_address(ip)) <= pool_hi) + conflicts = len(self.conflicts) + available = max(0, pool_size - active - offered - static_in_pool - conflicts) + return {'pool_size': pool_size, 'active': active, 'offered': offered, + 'expired': expired, 'conflicts': conflicts, 'available': available, + 'static': len(self.static)} + + def remove_lease(self, ip): + """管理员手动解除租约。""" + with self._lock: + lease = self.leases.get(ip) + if lease: + self._remove_lease(ip) + self._log(f'已手动解除 {ip} ({lease["mac"]}) 的租约', 'warn') + self._emit('update') + return True + return False + + def lease_holder(self, ip): + """查询某 IP 当前租约归属; 无租约返回 None。""" + with self._lock: + lease = self.leases.get(ip) + return (lease['mac'], lease['state']) if lease else None + + def remove_expired(self): + """清除所有已过期租约, 返回数量。""" + with self._lock: + to_del = [ip for ip, l in self.leases.items() if l['state'] == 'expired'] + for ip in to_del: + self._remove_lease(ip) + if to_del: + self._log(f'已清除 {len(to_del)} 条过期租约', 'warn') + self._emit('update') + return len(to_del) + + def clear_all_leases(self): + """清空所有租约记录(保留地址与冲突名单不受影响), 返回数量。""" + with self._lock: + n = len(self.leases) + self.leases.clear() + self.mac_map.clear() + self._dirty = True + if n: + self._log(f'已清空全部 {n} 条租约记录', 'warn') + self._emit('update') + return n + + def get_reserved_summary(self): + """保留地址概览: 在线 / 离线 / 从未上线。""" + rows = [] + with self._lock: + for mac, ip in self.static.items(): + if self.leases.get(ip, {}).get('mac') == mac \ + and self.leases[ip]['state'] == 'active': + state = '在线' + elif mac in self.mac_map: + state = '离线' + else: + state = '从未上线' + rows.append({'mac': mac, 'ip': ip, + 'note': self.static_notes.get(mac, ''), + 'lease_hours': self.static_lease_secs.get(mac, 0) // 3600, + 'status': state}) + return rows + + def clear_conflicts(self): + with self._lock: + n = len(self.conflicts) + self.conflicts.clear() + if n: + self._log(f'已清空冲突名单({n} 条)', 'info') + self._emit('update') + + # ---- 租约持久化 ------------------------------------------------------------ + + def _load_leases(self): + if not self.lease_path or not os.path.exists(self.lease_path): + return + try: + with open(self.lease_path, 'r', encoding='utf-8') as f: + items = json.load(f) + except Exception as exc: + self._log(f'读取租约文件失败: {exc}', 'warn') + return + now = time.time() + loaded = 0 + for item in items: + ip, mac = item.get('ip'), norm_mac(item.get('mac', '')) + if not ip or not mac: + continue + if mac in self.static and self.static[mac] != ip: + continue + expires = float(item.get('expires', 0)) + state = item.get('state', 'active') + if state == 'active' and expires <= now: + state = 'expired' + with self._lock: + self.leases[ip] = {'mac': mac, 'hostname': item.get('hostname', ''), + 'state': state, 'expires': expires, + 'source': 'static' if mac in self.static else 'dynamic', + 'updated': now} + self.mac_map[mac] = ip + loaded += 1 + if loaded: + self._log(f'已从磁盘恢复 {loaded} 条租约记录', 'info') + + def _save_leases(self): + if not self.lease_path: + return + with self._lock: + items = [{'ip': ip, 'mac': l['mac'], 'hostname': l['hostname'], + 'expires': l['expires'], 'state': l['state']} + for ip, l in self.leases.items() if l['state'] != 'offer'] + try: + tmp = self.lease_path + '.tmp' + with open(tmp, 'w', encoding='utf-8') as f: + json.dump(items, f, ensure_ascii=False, indent=1) + os.replace(tmp, self.lease_path) + self._dirty = False + except Exception as exc: + self._log(f'保存租约失败: {exc}', 'warn') + + +# --------------------------------------------------------------------------- +# 图形界面 +# --------------------------------------------------------------------------- + +APP_TITLE = '便携 DHCP 服务器 v1.0' +FW_RULE_NAME = 'PortableDhcpServer-UDP67' + + +class DhcpGui: + + def __init__(self): + self.root = tk.Tk() + self.root.title(APP_TITLE) + self.root.geometry('1060x720') + self.root.minsize(960, 640) + # exe(PyInstaller --onefile) 时 __file__ 指向临时解压目录, + # 配置/租约必须落在 exe 同级目录, 否则退出后丢失 + if getattr(sys, 'frozen', False): + self.base = os.path.dirname(sys.executable) + else: + try: + self.base = os.path.dirname(os.path.abspath(__file__)) + except NameError: + self.base = os.getcwd() + self.cfg_path = os.path.join(self.base, 'dhcp_config.json') + self.lease_path = os.path.join(self.base, 'dhcp_leases.json') + + self.events = queue.Queue() + self.engine = None + self._config_widgets = [] + + self._build_ui() + self._load_config() + self._log('欢迎使用便携 DHCP 服务器。请先确认「服务网卡 IP」与地址池设置，再点击「启动服务」。', 'info') + self._log('提示: 首次使用建议点击「防火墙放行」(需管理员权限)，否则客户端的请求可能被 Windows 拦截。', 'info') + + self.root.protocol('WM_DELETE_WINDOW', self._on_close) + self.root.after(150, self._poll_events) + self.root.after(2000, self._auto_refresh) + + # ---- 界面搭建 ------------------------------------------------------------ + + def _build_ui(self): + style = ttk.Style(self.root) + try: + style.theme_use('vista') + except tk.TclError: + pass + + root = ttk.Frame(self.root, padding=8) + root.pack(fill='both', expand=True) + root.columnconfigure(0, weight=1) + root.rowconfigure(2, weight=1) + + # === 配置区 === + cfgbox = ttk.LabelFrame(root, text=' 服务配置 ', padding=(10, 6)) + cfgbox.grid(row=0, column=0, sticky='ew') + for c in (1, 3, 5, 7): + cfgbox.columnconfigure(c, weight=1) + + def add_row(r, defn): + col = 0 + for label, key, widget, width in defn: + ttk.Label(cfgbox, text=label).grid(row=r, column=col, sticky='e', padx=(6, 3), pady=3) + widget.grid(row=r, column=col + 1, sticky='ew', padx=(0, 10)) + if width: + widget.configure(width=width) + self._config_widgets.append(widget) + if key: + setattr(self, key, widget) + col += 2 + + # 第 1 行 + self.e_server_ip = ttk.Combobox(cfgbox, values=list_local_ipv4()) + add_row(0, [('服务网卡 IP', 'e_server_ip', self.e_server_ip, 18), + ('子网掩码', 'e_mask', ttk.Entry(cfgbox), 16)]) + ttk.Button(cfgbox, text='刷新本机IP', width=10, + command=self._refresh_local_ips).grid(row=0, column=6, padx=(0, 6)) + + # 第 2 行 + add_row(1, [('地址池起始', 'e_pool_start', ttk.Entry(cfgbox), 18), + ('地址池结束', 'e_pool_end', ttk.Entry(cfgbox), 16)]) + + # 第 3 行 + add_row(2, [('默认网关', 'e_gateway', ttk.Entry(cfgbox), 18), + ('DNS 服务器(逗号分隔)', 'e_dns', ttk.Entry(cfgbox), 30)]) + + # 第 4 行 + add_row(3, [('域名(可选)', 'e_domain', ttk.Entry(cfgbox), 18), + ('租约时长(小时)', 'e_lease', ttk.Spinbox(cfgbox, from_=1, to=8760), 16)]) + + # 第 5 行 + add_row(4, [('TFTP 服务器(可选,PXE)', 'e_tftp', ttk.Entry(cfgbox), 18), + ('启动文件名(可选,PXE)', 'e_bootfile', ttk.Entry(cfgbox), 30)]) + + # 第 6 行: 选项 + optrow = ttk.Frame(cfgbox) + optrow.grid(row=5, column=0, columnspan=8, sticky='w', pady=(4, 0)) + self.v_conflict = tk.BooleanVar(value=True) + self.v_persist = tk.BooleanVar(value=True) + self.v_autoscroll = tk.BooleanVar(value=True) + ttk.Checkbutton(optrow, text='分配前 Ping 检测地址冲突', + variable=self.v_conflict).pack(side='left', padx=(4, 14)) + ttk.Checkbutton(optrow, text='租约保存到磁盘(重启服务保留分配)', + variable=self.v_persist).pack(side='left', padx=(0, 14)) + ttk.Button(optrow, text='按网卡 IP 自动填充默认值', command=self._autofill).pack(side='left') + + # === 控制区 === + ctrl = ttk.Frame(root) + ctrl.grid(row=1, column=0, sticky='ew', pady=(8, 6)) + self.status_canvas = tk.Canvas(ctrl, width=14, height=14, highlightthickness=0) + self.status_canvas.pack(side='left') + self._set_status(False) + self.status_label = ttk.Label(ctrl, text='服务未启动', width=22, anchor='w') + self.status_label.pack(side='left', padx=(6, 18)) + self.btn_start = ttk.Button(ctrl, text='▶ 启动服务', command=self._start_server) + self.btn_start.pack(side='left', padx=3) + self.btn_stop = ttk.Button(ctrl, text='■ 停止服务', command=self._stop_server, state='disabled') + self.btn_stop.pack(side='left', padx=3) + ttk.Separator(ctrl, orient='vertical').pack(side='left', fill='y', padx=10) + ttk.Button(ctrl, text='保存配置', command=self._save_config).pack(side='left', padx=3) + self.btn_fw = ttk.Button(ctrl, text='防火墙放行 UDP 67', command=self._fw_add) + self.btn_fw.pack(side='left', padx=3) + ttk.Button(ctrl, text='移除放行', command=self._fw_del).pack(side='left', padx=3) + ttk.Separator(ctrl, orient='vertical').pack(side='left', fill='y', padx=10) + self.admin_label = ttk.Label(ctrl, text='') + self.admin_label.pack(side='left', padx=(0, 4)) + self._refresh_admin_status() + + # === 选项卡 === + nb = ttk.Notebook(root) + nb.grid(row=2, column=0, sticky='nsew') + + # -- 日志页 -- + log_frame = ttk.Frame(nb, padding=4) + nb.add(log_frame, text=' 运行日志 ') + self.txt = tk.Text(log_frame, wrap='none', font=('Consolas', 10), + state='disabled', bg='#fafafa', relief='flat') + ys = ttk.Scrollbar(log_frame, orient='vertical', command=self.txt.yview) + xs = ttk.Scrollbar(log_frame, orient='horizontal', command=self.txt.xview) + self.txt.configure(yscrollcommand=ys.set, xscrollcommand=xs.set) + self.txt.grid(row=0, column=0, sticky='nsew') + ys.grid(row=0, column=1, sticky='ns') + xs.grid(row=1, column=0, sticky='ew') + log_frame.rowconfigure(0, weight=1) + log_frame.columnconfigure(0, weight=1) + for tag, color in (('info', '#333333'), ('packet', '#1a5fb4'), + ('ok', '#26a269'), ('warn', '#b45800'), ('err', '#c01c28')): + self.txt.tag_configure(tag, foreground=color) + logbar = ttk.Frame(log_frame) + logbar.grid(row=2, column=0, columnspan=2, sticky='ew', pady=(4, 0)) + ttk.Checkbutton(logbar, text='自动滚动', variable=self.v_autoscroll).pack(side='left') + ttk.Button(logbar, text='清空日志', command=self._clear_log).pack(side='left', padx=10) + ttk.Button(logbar, text='清空冲突名单', command=self._clear_conflicts_clicked).pack(side='left') + + # -- 客户端页 -- + cli_frame = ttk.Frame(nb, padding=4) + nb.add(cli_frame, text=' 客户端列表 ') + cols = ('ip', 'mac', 'hostname', 'state', 'source', 'expires', 'remain', 'note', 'last_seen') + heads = ('IP 地址', 'MAC 地址', '主机名', '状态', '来源', '租约到期', '剩余有效期', '备注', '最近活动') + widths = (120, 150, 130, 70, 80, 150, 90, 110, 110) + + clitop = ttk.Frame(cli_frame) + clitop.grid(row=0, column=0, columnspan=2, sticky='ew', pady=(0, 4)) + self.var_query = tk.StringVar() + self.var_query.trace_add('write', lambda *a: self._refresh_clients()) + ttk.Label(clitop, text='搜索:').pack(side='left') + ttk.Entry(clitop, textvariable=self.var_query, width=22).pack(side='left', padx=(4, 12)) + ttk.Label(clitop, text='状态:').pack(side='left') + self.var_state = tk.StringVar(value='全部') + st_combo = ttk.Combobox(clitop, textvariable=self.var_state, state='readonly', width=12, + values=('全部', '已分配', '预分配', '已过期')) + st_combo.pack(side='left', padx=(4, 12)) + st_combo.bind('<>', lambda e: self._refresh_clients()) + ttk.Button(clitop, text='刷新', command=self._refresh_clients).pack(side='left') + + self.cli_tree = ttk.Treeview(cli_frame, columns=cols, show='headings', height=13) + for c, h, w in zip(cols, heads, widths): + self.cli_tree.heading(c, text=h) + self.cli_tree.column(c, width=w, anchor='w') + cys = ttk.Scrollbar(cli_frame, orient='vertical', command=self.cli_tree.yview) + self.cli_tree.configure(yscrollcommand=cys.set) + self.cli_tree.grid(row=1, column=0, sticky='nsew') + cys.grid(row=1, column=1, sticky='ns') + cli_frame.rowconfigure(1, weight=1) + cli_frame.columnconfigure(0, weight=1) + + self.cli_detail = ttk.Label(cli_frame, text='未选择客户端', foreground='#555', anchor='w') + self.cli_detail.grid(row=2, column=0, columnspan=2, sticky='ew', pady=(4, 0)) + + clibar = ttk.Frame(cli_frame) + clibar.grid(row=3, column=0, columnspan=2, sticky='ew', pady=(4, 0)) + ttk.Button(clibar, text='解除选中租约', command=self._remove_selected_lease).pack(side='left') + ttk.Button(clibar, text='设为保留地址…', command=self._reserve_selected).pack(side='left', padx=6) + ttk.Button(clibar, text='导出 CSV', command=self._export_csv).pack(side='left', padx=6) + ttk.Button(clibar, text='清空过期租约', command=self._clear_expired_clicked).pack(side='left', padx=6) + ttk.Button(clibar, text='清空全部租约', command=self._clear_all_clicked).pack(side='left', padx=6) + ttk.Label(clibar, text='双击 MAC 复制 · 右键更多操作', foreground='#666').pack(side='left', padx=10) + self.cli_tree.bind('', self._on_tree_double) + self.cli_tree.bind('', self._on_client_menu) + self.cli_tree.bind('<>', self._on_client_select) + + # -- 保留地址页 -- + st_frame = ttk.Frame(nb, padding=4) + nb.add(st_frame, text=' 保留地址 (MAC→IP) ') + scol = ('mac', 'ip', 'note', 'hours', 'status') + shead = ('MAC 地址', '固定 IP', '备注', '租约(小时)', '在线状态') + swidth = (150, 130, 130, 100, 90) + self.st_tree = ttk.Treeview(st_frame, columns=scol, show='headings', height=9) + for c, h, w in zip(scol, shead, swidth): + self.st_tree.heading(c, text=h) + self.st_tree.column(c, width=w, anchor='w') + sysv = ttk.Scrollbar(st_frame, orient='vertical', command=self.st_tree.yview) + self.st_tree.configure(yscrollcommand=sysv.set) + self.st_tree.grid(row=0, column=0, sticky='nsew') + sysv.grid(row=0, column=1, sticky='ns') + st_frame.rowconfigure(0, weight=1) + st_frame.columnconfigure(0, weight=1) + stbar = ttk.Frame(st_frame) + stbar.grid(row=1, column=0, columnspan=2, sticky='ew', pady=(6, 0)) + ttk.Button(stbar, text='新增保留地址…', command=self._add_static).pack(side='left') + ttk.Button(stbar, text='编辑选中', command=self._edit_static).pack(side='left', padx=6) + ttk.Button(stbar, text='删除选中', command=self._del_static).pack(side='left', padx=6) + ttk.Button(stbar, text='导入 CSV', command=self._import_static).pack(side='left', padx=6) + ttk.Button(stbar, text='导出 CSV', command=self._export_static).pack(side='left', padx=6) + ttk.Button(stbar, text='清空列表', command=self._clear_static).pack(side='left', padx=6) + ttk.Label(st_frame, text='提示: 保留地址立即生效(无需重启服务)；「租约(小时)」可为该设备单独指定租期，留空用全局默认；' + '「在线状态」依据当前租约实时判断。', + foreground='#666').grid(row=2, column=0, columnspan=2, sticky='w', pady=(6, 0)) + + # === 底部状态栏 === + status = ttk.Frame(root, relief='groove', padding=(8, 4)) + status.grid(row=3, column=0, sticky='ew') + self.stats_label = ttk.Label(status, text='地址池: — · 已分配 0 · 预分配 0 · 冲突 0 · 可用 —') + self.stats_label.pack(side='left') + self.pool_canvas = tk.Canvas(status, height=12, width=220, highlightthickness=0, bg='#e8e8e8') + self.pool_canvas.pack(side='left', padx=12) + ttk.Label(status, text='监听: UDP 67').pack(side='right') + + # ---- 状态/辅助 ------------------------------------------------------------ + + def _set_status(self, running, text=None): + self.status_canvas.delete('all') + color = '#26a269' if running else '#c01c28' + self.status_canvas.create_oval(2, 2, 13, 13, fill=color, outline=color) + + def _set_ui_running(self, running): + state = 'disabled' if running else 'normal' + for w in self._config_widgets: + try: + w.configure(state=state) + except tk.TclError: + pass + self.btn_start.configure(state='disabled' if running else 'normal') + self.btn_stop.configure(state='normal' if running else 'disabled') + self._set_status(running) + self.status_label.configure(text='服务运行中 · UDP 67' if running else '服务未启动') + + def _refresh_admin_status(self): + if is_admin(): + self.admin_label.configure(text='[管理员权限]', foreground='#26a269') + else: + self.admin_label.configure(text='[普通权限 · 防火墙放行需管理员]', foreground='#b45800') + + def _refresh_local_ips(self): + ips = list_local_ipv4() + self.e_server_ip.configure(values=ips) + if ips and not self.e_server_ip.get(): + self.e_server_ip.set(ips[0]) + self._log(f'本机 IPv4 地址: {", ".join(ips) if ips else "未检测到"}', 'info') + + def _log(self, text, level='info'): + ts = time.strftime('%H:%M:%S') + self.txt.configure(state='normal') + self.txt.insert('end', f'[{ts}] {text}\n', level) + lines = int(self.txt.index('end-1c').split('.')[0]) + if lines > 3000: + self.txt.delete('1.0', f'{lines - 2500}.0') + if self.v_autoscroll.get(): + self.txt.see('end') + self.txt.configure(state='disabled') + + def _clear_log(self): + self.txt.configure(state='normal') + self.txt.delete('1.0', 'end') + self.txt.configure(state='disabled') + + def _clear_conflicts_clicked(self): + if self.engine and self.engine.running: + self.engine.clear_conflicts() + else: + messagebox.showinfo('提示', '服务未运行') + + # ---- 配置读写 ------------------------------------------------------------ + + def _collect_config(self): + def get_ip(widget, name, required=True): + text = widget.get().strip() + if not text: + if required: + raise ValueError(f'「{name}」不能为空') + return '' + try: + return str(ipaddress.ip_address(text)) + except ValueError: + raise ValueError(f'「{name}」不是合法 IP: {text}') + + cfg = { + 'server_ip': get_ip(self.e_server_ip, '服务网卡 IP'), + 'mask': get_ip(self.e_mask, '子网掩码'), + 'pool_start': get_ip(self.e_pool_start, '地址池起始'), + 'pool_end': get_ip(self.e_pool_end, '地址池结束'), + 'gateway': get_ip(self.e_gateway, '默认网关', required=False), + 'dns': [], + 'domain': self.e_domain.get().strip(), + 'lease_hours': 12, + 'tftp': self.e_tftp.get().strip(), + 'bootfile': self.e_bootfile.get().strip(), + 'check_conflict': bool(self.v_conflict.get()), + 'persist_leases': bool(self.v_persist.get()), + 'port': DHCP_SERVER_PORT, + 'static_bindings': self._static_rows(), + } + for item in self.e_dns.get().replace('，', ',').replace(' ', ',').split(','): + item = item.strip() + if item: + try: + cfg['dns'].append(str(ipaddress.ip_address(item))) + except ValueError: + raise ValueError(f'DNS 地址不合法: {item}') + try: + hours = int(float(self.e_lease.get())) + if not 1 <= hours <= 8760: + raise ValueError + except ValueError: + raise ValueError('租约时长必须为 1~8760 之间的整数(小时)') + cfg['lease_hours'] = hours + if int(ipaddress.ip_address(cfg['pool_start'])) > int(ipaddress.ip_address(cfg['pool_end'])): + raise ValueError('地址池起始不能大于结束') + pool_count = int(ipaddress.ip_address(cfg['pool_end'])) - int(ipaddress.ip_address(cfg['pool_start'])) + 1 + if pool_count > 65536: + raise ValueError('地址池过大(超过 65536 个地址)') + for row in cfg['static_bindings']: + ip = row['ip'] + if int(ipaddress.ip_address(ip)) < int(ipaddress.ip_address(cfg['pool_start'])) \ + or int(ipaddress.ip_address(ip)) > int(ipaddress.ip_address(cfg['pool_end'])): + if not messagebox.askyesno('确认', f'静态绑定 {ip} 不在地址池范围内，' + f'仍要继续吗？(地址池外绑定是允许的)'): + raise ValueError('用户取消') + break + # 网段一致性提醒 + try: + net = ipaddress.ip_network(f'{cfg["server_ip"]}/{cfg["mask"]}', strict=False) + outside = [t for t in (cfg['pool_start'], cfg['pool_end'], cfg['gateway']) + if t and ipaddress.ip_address(t) not in net] + if outside: + if not messagebox.askyesno('网段提醒', + f'以下地址与服务网卡不在同一网段:\n{", ".join(outside)}\n\n' + f'是否仍要启动？(客户端可能无法正常通信)'): + raise ValueError('用户取消') + except ValueError as exc: + if str(exc) == '用户取消': + raise + return cfg + + def _static_rows(self): + rows = [] + for iid in self.st_tree.get_children(): + vals = self.st_tree.item(iid, 'values') + mac = norm_mac(str(vals[0])) if vals else None + ip = str(vals[1]).strip() if len(vals) > 1 else '' + if mac and ip: + hours = 0 + if len(vals) > 3 and str(vals[3]).strip().isdigit(): + hours = int(str(vals[3]).strip()) + rows.append({'mac': mac, 'ip': ip, + 'note': str(vals[2]) if len(vals) > 2 else '', + 'lease_hours': hours}) + return rows + + def _sync_static_to_engine(self): + if self.engine and self.engine.running: + self.engine.set_static_bindings(self._static_rows()) + + def _save_config(self): + try: + cfg = self._collect_config() + except ValueError as exc: + messagebox.showwarning('配置有误', str(exc)) + return + data = dict(cfg) + data['static_bindings'] = self._static_rows() + data['saved_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + try: + with open(self.cfg_path, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + except OSError as exc: + messagebox.showerror('保存失败', f'无法写入配置文件:\n{exc}') + return + self._log(f'配置已保存到 {os.path.basename(self.cfg_path)}', 'ok') + + def _load_config(self): + defaults = self._default_config() + data = dict(defaults) + if os.path.exists(self.cfg_path): + try: + with open(self.cfg_path, 'r', encoding='utf-8') as f: + data.update(json.load(f)) + except Exception as exc: + self._log(f'读取配置失败, 使用默认值: {exc}', 'warn') + self.e_server_ip.set(data.get('server_ip', '')) + self.e_mask.insert(0, data.get('mask', '')) + self.e_pool_start.insert(0, data.get('pool_start', '')) + self.e_pool_end.insert(0, data.get('pool_end', '')) + self.e_gateway.insert(0, data.get('gateway', '')) + self.e_dns.insert(0, ', '.join(data.get('dns', []))) + self.e_domain.insert(0, data.get('domain', '')) + self.e_lease.delete(0, 'end') + self.e_lease.insert(0, str(data.get('lease_hours', 12))) + self.e_tftp.insert(0, data.get('tftp', '')) + self.e_bootfile.insert(0, data.get('bootfile', '')) + self.v_conflict.set(bool(data.get('check_conflict', True))) + self.v_persist.set(bool(data.get('persist_leases', True))) + for item in data.get('static_bindings', []): + mac = norm_mac(item.get('mac', '')) + if mac: + hours = int(item.get('lease_hours') or 0) + self.st_tree.insert('', 'end', values=( + mac, item.get('ip', ''), item.get('note', ''), + hours if hours else '')) + + def _default_config(self): + ips = list_local_ipv4() + base = ips[0] if ips else '192.168.1.10' + prefix = base.rsplit('.', 1)[0] + return { + 'server_ip': base, + 'mask': '255.255.255.0', + 'pool_start': f'{prefix}.100', + 'pool_end': f'{prefix}.200', + 'gateway': f'{prefix}.1', + 'dns': ['223.5.5.5', '114.114.114.114'], + 'domain': '', + 'lease_hours': 12, + 'tftp': '', + 'bootfile': '', + 'check_conflict': True, + 'persist_leases': True, + 'static_bindings': [], + } + + def _autofill(self): + ip = self.e_server_ip.get().strip() + try: + ipaddress.ip_address(ip) + except ValueError: + messagebox.showwarning('提示', '请先填写合法的「服务网卡 IP」') + return + prefix = ip.rsplit('.', 1)[0] + self.e_mask.delete(0, 'end') + self.e_mask.insert(0, '255.255.255.0') + self.e_pool_start.delete(0, 'end') + self.e_pool_start.insert(0, f'{prefix}.100') + self.e_pool_end.delete(0, 'end') + self.e_pool_end.insert(0, f'{prefix}.200') + self.e_gateway.delete(0, 'end') + self.e_gateway.insert(0, f'{prefix}.1') + if not self.e_dns.get().strip(): + self.e_dns.insert(0, '223.5.5.5, 114.114.114.114') + + # ---- 服务控制 ------------------------------------------------------------ + + def _start_server(self): + if self.engine and self.engine.running: + return + try: + cfg = self._collect_config() + except ValueError as exc: + messagebox.showwarning('配置有误', str(exc)) + return + lease_path = self.lease_path if cfg.get('persist_leases', True) else None + engine = DhcpServerEngine(cfg, self.events, lease_path=lease_path) + try: + engine.start() + except OSError as exc: + code = getattr(exc, 'winerror', None) or getattr(exc, 'errno', None) + hint = '' + if code in (10013, 10048): + hint = ('\n\n常见原因: 系统开启了「Internet 连接共享 / 移动热点」，' + '其内置 DHCP 分配器占用了 UDP 67。\n' + '可运行: net stop SharedAccess 关闭(需管理员)，\n' + '或用 netstat -ano -p udp | findstr :67 查看占用进程。') + messagebox.showerror('启动失败', f'{exc}{hint}') + self._log(f'启动失败: {exc}{hint}', 'err') + return + self.engine = engine + self._set_ui_running(True) + self._refresh_clients() + self._refresh_stats() + + def _stop_server(self): + if self.engine and self.engine.running: + self.engine.stop() + self._set_ui_running(False) + + # ---- 事件循环 ------------------------------------------------------------ + + def _poll_events(self): + try: + while True: + event = self.events.get_nowait() + kind = event[0] + if kind == 'log': + self._log(event[2], event[1]) + elif kind == 'update': + self._refresh_clients() + self._refresh_stats() + elif kind == 'stopped': + self._set_ui_running(False) + except queue.Empty: + pass + self.root.after(150, self._poll_events) + + def _auto_refresh(self): + if self.engine and self.engine.running: + self._refresh_clients() + self._refresh_stats() + self._refresh_reserved() + self.root.after(2000, self._auto_refresh) + + # ---- 客户端列表 ------------------------------------------------------------ + + def _refresh_clients(self): + tree = self.cli_tree + selected = {str(tree.item(i)['values'][0]) for i in tree.selection()} + tree.delete(*tree.get_children()) + if not self.engine: + return + q = self.var_query.get().strip().lower() + state = self.var_state.get() + for row in self.engine.get_rows(): + if state != '全部' and row['state'] != state: + continue + if q and not any(q in str(row.get(k)).lower() + for k in ('ip', 'mac', 'hostname', 'note')): + continue + iid = tree.insert('', 'end', values=( + row['ip'], row['mac'], row['hostname'] or '—', row['state'], + row['source'], row['expires'], row['remain'], + row['note'] or '', row['last_seen'])) + if row['ip'] in selected: + tree.selection_add(iid) + if row['_state'] == 'active': + tree.item(iid, tags=('active',)) + elif row['_state'] == 'offer': + tree.item(iid, tags=('offer',)) + tree.tag_configure('active', foreground='#1a5fb4') + tree.tag_configure('offer', foreground='#888888') + + def _refresh_stats(self): + if not self.engine: + self.stats_label.configure(text='地址池: — · 已分配 0 · 预分配 0 · 冲突 0 · 可用 —') + self._draw_pool_bar(0, 1) + return + s = self.engine.stats() + self.stats_label.configure( + text=f'地址池: {s["pool_size"]} · 已分配 {s["active"]} · 预分配 {s["offered"]} · ' + f'过期 {s["expired"]} · 冲突 {s["conflicts"]} · 静态绑定 {s["static"]} · 可用 {s["available"]}') + self._draw_pool_bar(s['active'] + s['offered'], max(1, s['pool_size'])) + + def _draw_pool_bar(self, used, total): + c = self.pool_canvas + c.delete('all') + width = int(c['width']) + frac = min(1.0, used / total) if total else 0 + c.create_rectangle(0, 0, width, 12, fill='#e8e8e8', outline='') + c.create_rectangle(0, 0, int(width * frac), 12, + fill='#c01c28' if frac > 0.9 else '#26a269', outline='') + c.create_text(width // 2, 6, text=f'{used}/{total}', + font=('Microsoft YaHei UI', 8), fill='#333') + + def _remove_selected_lease(self): + sel = self.cli_tree.selection() + if not sel: + messagebox.showinfo('提示', '请先在列表中选择要解除的租约') + return + ip = str(self.cli_tree.item(sel[0])['values'][0]) + if self.engine and self.engine.running: + if messagebox.askyesno('确认', f'确定解除 {ip} 的租约吗？\n该客户端下次续租时会被拒绝。'): + self.engine.remove_lease(ip) + else: + messagebox.showinfo('提示', '服务未运行') + + # ---- 客户端选择 / 右键 / 保留 ------------------------------------------------- + + def _on_client_select(self, event=None): + sel = self.cli_tree.selection() + if not sel: + self.cli_detail.configure(text='未选择客户端') + return + vals = self.cli_tree.item(sel[0])['values'] + ip, mac, host, state = vals[0], vals[1], vals[2], vals[3] + source, expires, remain = vals[4], vals[5], vals[6] + note, last = vals[7], vals[8] + self.cli_detail.configure( + text=f'{ip} · {mac} · {host} · {state} · {source} · 到期 {expires} · 剩余 {remain} · 最近 {last}' + + (f' · 备注 {note}' if str(note) else '')) + + def _on_client_menu(self, event): + iid = self.cli_tree.identify_row(event.y) + if not iid: + return + self.cli_tree.selection_set(iid) + vals = self.cli_tree.item(iid)['values'] + menu = tk.Menu(self.root, tearoff=0) + menu.add_command(label='复制 MAC', command=lambda: self._copy_text(str(vals[1]))) + menu.add_command(label='复制 IP', command=lambda: self._copy_text(str(vals[0]))) + menu.add_separator() + menu.add_command(label='解除租约', command=self._remove_selected_lease) + menu.add_command(label='设为保留地址…', command=self._reserve_selected) + try: + menu.tk_popup(event.x_root, event.y_root) + finally: + menu.grab_release() + + def _copy_text(self, text): + self.root.clipboard_clear() + self.root.clipboard_append(text) + self._log(f'已复制: {text}', 'info') + + def _reserve_selected(self): + """把选中客户端直接加入保留地址。""" + sel = self.cli_tree.selection() + if not sel: + messagebox.showinfo('提示', '请先在列表中选择客户端') + return + vals = self.cli_tree.item(sel[0])['values'] + mac, ip = str(vals[1]), str(vals[0]) + saved = self._edit_static_dialog({'mac': mac, 'ip': ip, '_title': '设为保留地址'}) + if not saved: + return + try: + self._insert_static_row(**saved) + except ValueError as exc: + messagebox.showwarning('提示', str(exc)) + return + self._sync_static_to_engine() + self._log(f'已设为保留地址: {saved["mac"]} -> {saved["ip"]}', 'ok') + self._refresh_clients() + self._refresh_reserved() + + def _clear_expired_clicked(self): + if not (self.engine and self.engine.running): + messagebox.showinfo('提示', '服务未运行') + return + if messagebox.askyesno('确认', '确定清除所有「已过期」状态的租约记录吗？'): + self.engine.remove_expired() + self._refresh_clients() + self._refresh_stats() + + def _clear_all_clicked(self): + if not (self.engine and self.engine.running): + messagebox.showinfo('提示', '服务未运行') + return + if messagebox.askyesno('确认', '将清空全部租约记录(所有客户端需重新获取 IP)，\n保留地址与冲突名单不受影响。确定吗？'): + self.engine.clear_all_leases() + self._refresh_clients() + self._refresh_stats() + self._refresh_reserved() + + def _export_csv(self): + rows = self.engine.get_rows() if self.engine else [] + if not rows: + messagebox.showinfo('提示', '当前没有可导出的客户端') + return + path = filedialog.asksaveasfilename( + defaultextension='.csv', initialfile=f'dhcp客户端-{time.strftime("%Y%m%d-%H%M%S")}.csv', + filetypes=[('CSV 文件', '*.csv')], parent=self.root) + if not path: + return + try: + with open(path, 'w', newline='', encoding='utf-8-sig') as f: + writer = csv.writer(f) + writer.writerow(['IP 地址', 'MAC 地址', '主机名', '状态', '来源', '租约到期', '剩余有效期']) + for r in rows: + writer.writerow([r['ip'], r['mac'], r['hostname'], r['state'], + r['source'], r['expires'], r['remain']]) + except OSError as exc: + messagebox.showerror('导出失败', str(exc)) + return + self._log(f'客户端列表已导出到 {path}', 'ok') + messagebox.showinfo('导出成功', f'已导出 {len(rows)} 条记录到:\n{path}') + + def _on_tree_double(self, event): + region = self.cli_tree.identify('region', event.x, event.y) + if region != 'cell': + return + col = self.cli_tree.identify_column(event.x) + if col != '#2': # MAC 列 + return + row = self.cli_tree.identify_row(event.y) + if row: + mac = str(self.cli_tree.item(row)['values'][1]) + self.root.clipboard_clear() + self.root.clipboard_append(mac) + self._log(f'已复制 MAC: {mac}', 'info') + + # ---- 保留地址 ------------------------------------------------------------ + + def _edit_static_dialog(self, prefill=None): + """保留地址编辑对话框; 返回 {mac, ip, note, lease_hours} 或 None。""" + prefill = prefill or {} + dlg = tk.Toplevel(self.root) + dlg.title(prefill.get('_title') or '新增保留地址') + dlg.transient(self.root) + dlg.resizable(False, False) + frm = ttk.Frame(dlg, padding=12) + frm.pack(fill='both', expand=True) + fields = [ + ('MAC 地址', prefill.get('mac', '')), + ('固定 IP', prefill.get('ip', '')), + ('备注(可选)', prefill.get('note', '')), + ('租约时长(小时, 可选)', str(prefill.get('lease_hours') or '')), + ] + entries = {} + for r, (label, value) in enumerate(fields): + ttk.Label(frm, text=label).grid(row=r, column=0, sticky='e', padx=(0, 8), pady=4) + e = ttk.Entry(frm, width=30) + e.insert(0, value) + e.grid(row=r, column=1, sticky='ew', pady=4) + entries[r] = e + result = {} + + def on_ok(): + mac = norm_mac(entries[0].get()) + ip = entries[1].get().strip() + note = entries[2].get().strip() + hours_text = entries[3].get().strip() + if not mac: + messagebox.showwarning('提示', 'MAC 地址格式不正确，示例: AA:BB:CC:DD:EE:FF', parent=dlg) + return + try: + ip = str(ipaddress.ip_address(ip)) + except ValueError: + messagebox.showwarning('提示', f'IP 地址不合法: {ip}', parent=dlg) + return + hours = 0 + if hours_text: + try: + hours = int(hours_text) + if not 1 <= hours <= 8760: + raise ValueError + except ValueError: + messagebox.showwarning('提示', '租约时长必须为 1~8760 的整数小时，或留空使用默认值', parent=dlg) + return + result.update({'mac': mac, 'ip': ip, 'note': note, 'lease_hours': hours}) + dlg.destroy() + + btns = ttk.Frame(frm) + btns.grid(row=len(fields), column=0, columnspan=2, pady=(10, 0)) + ttk.Button(btns, text='确定', command=on_ok).pack(side='left', padx=6) + ttk.Button(btns, text='取消', command=dlg.destroy).pack(side='left', padx=6) + frm.columnconfigure(1, weight=1) + dlg.grab_set() + self.root.wait_window(dlg) + return result or None + + def _insert_static_row(self, mac, ip, note, lease_hours): + """向保留地址表插入一行; 重复或冲突时抛 ValueError。""" + for row_iid in self.st_tree.get_children(): + vals = self.st_tree.item(row_iid)['values'] + if norm_mac(str(vals[0])) == mac: + raise ValueError(f'MAC {mac} 已存在保留地址') + if str(vals[1]) == ip: + raise ValueError(f'IP {ip} 已被其他保留地址占用') + if self.engine: + holder = self.engine.lease_holder(ip) + if holder and holder[0] != mac and holder[1] in ('active', 'offer'): + if not messagebox.askyesno( + '冲突确认', + f'IP {ip} 当前已被 {holder[0]} 占用租约。\n' + f'保留给 {mac} 后，原设备续租将被拒绝并改分配其他地址。\n继续吗？'): + raise ValueError('用户取消') + self.st_tree.insert('', 'end', values=( + mac, ip, note, lease_hours if lease_hours else '')) + + def _add_static(self): + saved = self._edit_static_dialog() + if not saved: + return + try: + self._insert_static_row(**saved) + except ValueError as exc: + messagebox.showwarning('提示', str(exc)) + return + self._sync_static_to_engine() + self._log(f'新增保留地址: {saved["mac"]} -> {saved["ip"]}', 'ok') + self._refresh_clients() + self._refresh_reserved() + + def _edit_static(self): + sel = self.st_tree.selection() + if not sel: + messagebox.showinfo('提示', '请先选中要编辑的保留地址') + return + vals = self.st_tree.item(sel[0])['values'] + prefill = {'mac': str(vals[0]), 'ip': str(vals[1]), + 'note': str(vals[2]) if len(vals) > 2 else '', + 'lease_hours': int(str(vals[3])) if len(vals) > 3 and str(vals[3]).isdigit() else 0, + '_title': '编辑保留地址'} + saved = self._edit_static_dialog(prefill) + if not saved: + return + old_mac = norm_mac(str(vals[0])) + if old_mac == saved['mac']: + # 同一 MAC: 直接原位更新(只允许改 IP/备注/租期) + self.st_tree.item(sel[0], values=( + saved['mac'], saved['ip'], saved['note'], + saved['lease_hours'] if saved['lease_hours'] else '')) + self._log(f'已更新保留地址: {saved["mac"]} -> {saved["ip"]}', 'ok') + else: + try: + self._insert_static_row(**saved) + except ValueError as exc: + messagebox.showwarning('提示', str(exc)) + return + self.st_tree.delete(sel[0]) + self._sync_static_to_engine() + self._refresh_clients() + self._refresh_reserved() + + def _del_static(self): + sel = self.st_tree.selection() + if not sel: + messagebox.showinfo('提示', '请先选中要删除的保留地址') + return + for iid in sel: + vals = self.st_tree.item(iid)['values'] + self.st_tree.delete(iid) + self._log(f'删除保留地址: {vals[0]} -> {vals[1]}', 'info') + self._sync_static_to_engine() + self._refresh_clients() + self._refresh_reserved() + + def _clear_static(self): + if not self.st_tree.get_children(): + messagebox.showinfo('提示', '保留地址列表为空') + return + if not messagebox.askyesno('确认', '确定清空所有保留地址吗？(不影响当前租约)'): + return + n = len(self.st_tree.get_children()) + self.st_tree.delete(*self.st_tree.get_children()) + self._sync_static_to_engine() + self._log(f'已清空 {n} 条保留地址', 'warn') + self._refresh_reserved() + + def _export_static(self): + rows = self._static_rows() + if not rows: + messagebox.showinfo('提示', '没有可导出的保留地址') + return + path = filedialog.asksaveasfilename( + defaultextension='.csv', initialfile=f'保留地址-{time.strftime("%Y%m%d")}.csv', + filetypes=[('CSV 文件', '*.csv')], parent=self.root) + if not path: + return + try: + with open(path, 'w', newline='', encoding='utf-8-sig') as f: + writer = csv.writer(f) + writer.writerow(['MAC 地址', '固定 IP', '备注', '租约时长(小时)']) + for r in rows: + writer.writerow([r['mac'], r['ip'], r['note'], + r['lease_hours'] if r['lease_hours'] else '']) + except OSError as exc: + messagebox.showerror('导出失败', str(exc)) + return + self._log(f'保留地址已导出到 {path}', 'ok') + messagebox.showinfo('导出成功', f'已导出 {len(rows)} 条到:\n{path}') + + def _import_static(self): + path = filedialog.askopenfilename( + filetypes=[('CSV 文件', '*.csv')], parent=self.root) + if not path: + return + added, skipped = 0, [] + try: + with open(path, 'r', encoding='utf-8-sig') as f: + reader = csv.reader(f) + for row in reader: + if not row or not row[0].strip(): + continue + mac = norm_mac(row[0]) + if not mac or len(row) < 2: + skipped.append(row[0].strip() or '空行') + continue + try: + ip = str(ipaddress.ip_address(row[1].strip())) + except ValueError: + skipped.append(f'{mac}(IP不合法)') + continue + note = row[2].strip() if len(row) > 2 else '' + hours = 0 + if len(row) > 3 and row[3].strip().isdigit(): + hours = max(0, min(8760, int(row[3].strip()))) + try: + self._insert_static_row(mac, ip, note, hours) + added += 1 + except ValueError as exc: + skipped.append(str(exc)) + except OSError as exc: + messagebox.showerror('导入失败', str(exc)) + return + if added: + self._sync_static_to_engine() + self._refresh_reserved() + self._log(f'保留地址导入完成: 新增 {added} 条', 'ok') + message = f'导入完成: 新增 {added} 条。' + if skipped: + message += f'\n跳过 {len(skipped)} 条:\n' + '\n'.join(skipped[:10]) + messagebox.showinfo('导入结果', message) + + def _refresh_reserved(self): + """刷新保留地址的在线状态列。""" + summary = {} + if self.engine and self.engine.running: + summary = {r['mac']: r['status'] for r in self.engine.get_reserved_summary()} + for iid in self.st_tree.get_children(): + vals = list(self.st_tree.item(iid)['values']) + while len(vals) < 5: + vals.append('') + mac = norm_mac(str(vals[0])) + status = summary.get(mac, '') + vals[4] = status + self.st_tree.item(iid, values=tuple(vals)) + if status == '在线': + self.st_tree.item(iid, tags=('online',)) + elif status == '离线': + self.st_tree.item(iid, tags=('offline',)) + else: + self.st_tree.item(iid, tags=()) + self.st_tree.tag_configure('online', foreground='#26a269') + self.st_tree.tag_configure('offline', foreground='#b45800') + + # ---- 防火墙 ------------------------------------------------------------ + + def _run_netsh(self, args): + try: + out = subprocess.run(['netsh', 'advfirewall', 'firewall'] + args, + capture_output=True, timeout=30) + text = ((out.stdout or b'') + (out.stderr or b'')).decode('gbk', 'ignore').strip() + return out.returncode == 0, text + except Exception as exc: + return False, str(exc) + + def _fw_add(self): + if not is_admin(): + messagebox.showwarning('需要管理员权限', + '请右键以管理员身份运行本程序(或用管理员打开的终端启动)，\n' + '然后再点击「防火墙放行」。') + return + if not messagebox.askyesno( + '确认操作', + '即将添加 Windows 防火墙入站规则:\n\n' + f' 规则名: {FW_RULE_NAME}\n 协议/端口: UDP 67(本机监听端口)\n 动作: 允许\n\n' + '作用: 允许局域网设备发来的 DHCP 请求进入本机。\n确定添加吗？'): + return + ok, text = self._run_netsh(['add', 'rule', f'name={FW_RULE_NAME}', 'dir=in', + 'action=allow', 'protocol=UDP', 'localport=67', + 'profile=any']) + if ok: + self._log('防火墙规则已添加: 允许 UDP 67 入站', 'ok') + messagebox.showinfo('成功', '防火墙规则已添加。') + else: + self._log(f'防火墙规则添加失败: {text}', 'err') + messagebox.showerror('失败', text or 'netsh 执行失败') + + def _fw_del(self): + if not is_admin(): + messagebox.showwarning('需要管理员权限', '请以管理员身份运行本程序后再操作。') + return + if not messagebox.askyesno('确认操作', + f'即将删除防火墙入站规则 {FW_RULE_NAME}。\n确定删除吗？'): + return + ok, text = self._run_netsh(['delete', 'rule', f'name={FW_RULE_NAME}']) + if ok: + self._log('防火墙规则已删除', 'info') + messagebox.showinfo('成功', '防火墙规则已删除。') + else: + self._log(f'防火墙规则删除失败: {text}', 'err') + messagebox.showerror('失败', text or 'netsh 执行失败') + + # ---- 退出 ------------------------------------------------------------ + + def _on_close(self): + try: + if self.engine and self.engine.running: + self.engine.stop('窗口关闭') + finally: + self.root.destroy() + + def run(self): + self.root.mainloop() + + +# --------------------------------------------------------------------------- +# 内置自检 (python dhcp_server_app.py --selftest) +# --------------------------------------------------------------------------- + +def run_selftest(): + print('=' * 62) + print('便携 DHCP 服务器 - 内置自检') + print('=' * 62) + failures = [] + + def check(name, cond, detail=''): + mark = 'PASS' if cond else 'FAIL' + print(f' [{mark}] {name}' + (f' ({detail})' if detail else '')) + if not cond: + failures.append(name) + + # ---- 1. 报文解析 ---- + pkt = make_client_packet(MSG_DISCOVER, 'AA:BB:CC:DD:EE:01', 0x1234ABCD, + hostname='TEST-PC') + req = parse_dhcp_packet(pkt) + check('解析: 识别 DISCOVER', req is not None and + req.options.get(OPT_MSG_TYPE) == bytes([MSG_DISCOVER])) + check('解析: MAC 地址', req is not None and req.chaddr_mac == 'AA:BB:CC:DD:EE:01') + check('解析: 事务 ID', req is not None and req.xid == 0x1234ABCD) + check('解析: 主机名', req is not None and req.hostname == 'TEST-PC') + check('解析: 客户端标识', req is not None and req.client_id_mac == 'AA:BB:CC:DD:EE:01') + + # ---- 2. 应答构造 ---- + cfg = {'server_ip': '192.168.10.1', 'mask': '255.255.255.0', + 'pool_start': '192.168.10.100', 'pool_end': '192.168.10.102', + 'gateway': '192.168.10.1', 'dns': ['223.5.5.5'], 'domain': 'test', + 'lease_hours': 2, 'tftp': '', 'bootfile': ''} + reply = build_reply(req, MSG_OFFER, '192.168.10.100', cfg, 7200) + r2 = parse_dhcp_packet(reply) + check('构造: OFFER 可被解析', r2 is not None) + check('构造: yiaddr', r2 is not None and r2.yiaddr == '192.168.10.100') + check('构造: 报文类型', r2 is not None and + r2.options.get(OPT_MSG_TYPE) == bytes([MSG_OFFER])) + check('构造: 服务器标识', r2 is not None and + parse_ip_option(r2.options.get(OPT_SERVER_ID)) == '192.168.10.1') + check('构造: 租约时长', r2 is not None and + struct.unpack('!I', r2.options.get(OPT_LEASE_TIME))[0] == 7200) + check('构造: 最小长度', len(reply) >= 300) + + # ---- 3. 端到端交互(高位端口, 不需要管理员) ---- + test_port, test_client_port = 1067, 1068 + cfg2 = {'server_ip': '127.0.0.1', 'mask': '255.255.255.0', + 'pool_start': '10.77.0.10', 'pool_end': '10.77.0.12', + 'gateway': '10.77.0.1', 'dns': ['223.5.5.5'], 'domain': '', + 'lease_hours': 1, 'tftp': '', 'bootfile': '', 'check_conflict': False, + 'port': test_port, 'static_bindings': []} + events = queue.Queue() + engine = DhcpServerEngine(cfg2, events, lease_path=None) + engine.start() + time.sleep(0.2) + + client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + client.bind(('127.0.0.1', test_client_port)) + client.settimeout(2) + + def xchg(pkt_bytes): + client.sendto(pkt_bytes, ('127.0.0.1', test_port)) + try: + data, _ = client.recvfrom(2048) + return parse_dhcp_packet(data) + except socket.timeout: + return None + + # 3.1 DISCOVER -> OFFER + mac1 = 'AA:BB:CC:00:00:01' + off = xchg(make_client_packet(MSG_DISCOVER, mac1, 0x11110001, hostname='dev-1')) + check('交互: DISCOVER 得到 OFFER', off is not None and + off.options.get(OPT_MSG_TYPE) == bytes([MSG_OFFER])) + ip1 = off.yiaddr if off else None + check('交互: OFFER 分配了池内 IP', ip1 is not None and ip1.startswith('10.77.0.'), str(ip1)) + + # 3.2 REQUEST -> ACK + ack = xchg(make_client_packet(MSG_REQUEST, mac1, 0x11110002, + requested_ip=ip1, server_id='127.0.0.1')) + check('交互: REQUEST 得到 ACK', ack is not None and + ack.options.get(OPT_MSG_TYPE) == bytes([MSG_ACK])) + check('交互: ACK 地址一致', ack is not None and ack.yiaddr == ip1) + + # 3.3 续租 + ack2 = xchg(make_client_packet(MSG_REQUEST, mac1, 0x11110003, ciaddr=ip1)) + check('交互: 续租得到 ACK', ack2 is not None and + ack2.options.get(OPT_MSG_TYPE) == bytes([MSG_ACK])) + + # 3.4 第二个客户端获得不同 IP + mac2 = 'AA:BB:CC:00:00:02' + off2 = xchg(make_client_packet(MSG_DISCOVER, mac2, 0x22220001)) + check('交互: 第二客户端获得不同 IP', off2 is not None and off2.yiaddr != ip1, + str(off2.yiaddr if off2 else None)) + + # 3.5 非法请求 -> NAK + nak = xchg(make_client_packet(MSG_REQUEST, 'AA:BB:CC:00:00:99', 0x33330001, + requested_ip='10.99.9.9', server_id='127.0.0.1')) + check('交互: 非法请求得到 NAK', nak is not None and + nak.options.get(OPT_MSG_TYPE) == bytes([MSG_NAK])) + + # 3.6 DECLINE -> 冲突名单 + xchg(make_client_packet(MSG_DECLINE, mac2, 0x22220002, requested_ip=off2.yiaddr)) + check('交互: DECLINE 加入冲突名单', off2 is not None and off2.yiaddr in engine.conflicts) + + # 3.7 RELEASE + xchg(make_client_packet(MSG_RELEASE, mac1, 0x11110004, ciaddr=ip1)) + rows = {r['ip'] for r in engine.get_rows()} + check('交互: RELEASE 后租约移除', ip1 not in rows) + + # 3.8 静态绑定 + 自定义租约 + engine.set_static_bindings([{'mac': 'AA:BB:CC:00:00:55', 'ip': '10.77.0.50', + 'note': 'camera', 'lease_hours': 2}]) + off3 = xchg(make_client_packet(MSG_DISCOVER, 'AA:BB:CC:00:00:55', 0x44440001)) + check('交互: 静态绑定生效', off3 is not None and off3.yiaddr == '10.77.0.50') + check('交互: 保留地址自定义租期(2h)', off3 is not None and + struct.unpack('!I', off3.options.get(OPT_LEASE_TIME))[0] == 7200) + check('交互: 保留地址状态查询', any( + r['mac'] == 'AA:BB:CC:00:00:55' for r in engine.get_reserved_summary())) + + # 3.9 地址池耗尽 (池 3 个: 10.77.0.10~12, 其中 .11 已加入冲突名单) + got = [] + for i in range(6): + rep = xchg(make_client_packet(MSG_DISCOVER, f'AA:55:00:00:00:{i:02X}', 0x55550000 + i)) + if rep is not None and rep.options.get(OPT_MSG_TYPE) == bytes([MSG_OFFER]): + got.append(rep.yiaddr) + unique = set(got) + check('交互: 池耗尽时不再分配重复地址', len(unique) == len(got) and len(got) <= 3, + f'分配 {len(got)} 个: {sorted(unique)}') + + # 3.10 统计 + st = engine.stats() + check('统计: 池大小', st['pool_size'] == 3, str(st)) + check('统计: 可用数不为负', st['available'] >= 0, str(st)) + + # 3.11 批量租约管理 + engine._set_lease('10.77.0.12', 'AA:BB:CC:00:00:77', 'stale', 'expired', 'dynamic') + n_expired = engine.remove_expired() + check('管理: 清除过期租约', n_expired >= 1, str(n_expired)) + n_clear = engine.clear_all_leases() + check('管理: 清空全部租约', n_clear >= 0, str(n_clear)) + + client.close() + engine.stop('自检结束') + while not events.empty(): + events.get_nowait() + + print('-' * 62) + if failures: + print(f'自检结果: {len(failures)} 项失败 -> {", ".join(failures)}') + return 1 + print('自检结果: 全部通过') + return 0 + + +def main(): + if '--selftest' in sys.argv: + sys.exit(run_selftest()) + if '--version' in sys.argv: + print(APP_TITLE) + return + app = DhcpGui() + app.run() + + +if __name__ == '__main__': + main() diff --git a/使用说明.md b/使用说明.md new file mode 100644 index 0000000..ef012de --- /dev/null +++ b/使用说明.md @@ -0,0 +1,118 @@ +# 便携 DHCP 服务器 使用说明 + +一个运行在 Windows 电脑上的轻量 DHCP 服务器（单文件、免安装、图形界面）， +用于给局域网里的测试设备临时分配 IP：开发板、IP 摄像头、交换机、PXE 网络启动等。 + +## 一、环境要求（二选一） + +- **直接用 exe（推荐）**：Windows 7 及以上，双击即用，无需安装任何东西 +- **源码方式**：需 Python 3.10+（仅用标准库，无需第三方包） + 安装时勾选 **Add python.exe to PATH** + +## 二、快速开始 + +1. 双击 **`便携DHCP服务器.exe`**（首次启动约需几秒自解压，属正常现象） + 或源码方式：双击 `启动DHCP服务器.bat` +2. 确认「服务网卡 IP」是接入测试网络的网卡地址（可点「刷新本机IP」） +3. 设置地址池，例如 `192.168.10.100 ~ 192.168.10.200` +4. 点击 **▶ 启动服务** +5. 把待分配 IP 的设备接进同一个交换机/Wi-Fi，设为自动获取 IP 即可 + +> 排错：窗口一闪而过时，在命令行手动运行 `python dhcp_server_app.py` 查看报错。 +> 自检：`python dhcp_server_app.py --selftest` 可验证协议引擎是否正常。 + +## 三、三大管理界面 + +**客户端列表**：顶部搜索框可按 IP / MAC / 主机名 / 备注实时过滤；状态下拉可只看 +已分配 / 预分配 / 已过期。选中任意行，下方详情栏显示该客户端完整信息； +右键弹出快捷菜单（复制、解除租约、设为保留地址）。 + +**租约管理**：列表工具栏提供「解除选中租约」「清空过期租约」「清空全部租约」， +均带确认弹窗。解除租约后该设备下次续租会被拒绝并重新申请地址。 + +**保留地址（MAC→IP）**：点「新增保留地址」填写 MAC、固定 IP、备注和可选的 +独立租约时长；立即生效无需重启服务。「在线状态」列实时显示该设备是否在线。 +在客户端列表中选中某行点「设为保留地址」，可把当前分配的 IP 直接固定给该设备。 +支持导入 / 导出 CSV，方便批量维护。 + +## 四、首次使用必看：防火墙 + +客户端的 DHCP 请求是发往本机 **UDP 67** 的入站流量，默认会被 Windows 防火墙拦截。 + +- 右键 `启动DHCP服务器.bat` → **以管理员身份运行** +- 在程序里点击 **「防火墙放行 UDP 67」**（会弹窗确认） +- 之后日常使用无需再管（也可随时点「移除放行」撤销） + +对应的手动命令（管理员）： + +``` +netsh advfirewall firewall add rule name="PortableDhcpServer-UDP67" dir=in action=allow protocol=UDP localport=67 profile=any +netsh advfirewall firewall delete rule name="PortableDhcpServer-UDP67" +``` + +## 五、功能说明 + +| 功能 | 说明 | +| --- | --- | +| 地址池 | 起始~结束范围内的顺序分配；自动跳过网关、本机 IP、网络号、广播地址 | +| 客户端列表 | 实时查看 IP / MAC / 主机名 / 状态 / 来源 / 租约到期 / 剩余有效期 / 备注 / 最近活动；支持关键字搜索与状态筛选；选中行在下方显示完整详情 | +| 租约管理 | 解除选中租约、一键清除全部过期租约、清空全部租约；到期自动回收；同一设备尽量保持原 IP；停止服务时保存到磁盘 | +| 保留地址 | 按 MAC 固定 IP（即 DHCP 保留地址）；支持备注和**单独自定义租约时长**；显示在线/离线/从未上线状态；可编辑、导入/导出 CSV、从客户端列表一键转为保留 | +| 客户端操作 | 右键菜单：复制 MAC / 复制 IP / 解除租约 / 设为保留地址；双击 MAC 复制到剪贴板 | +| 冲突检测 | 分配前先 Ping 候选地址，占用或已被 DECLINE 的地址自动跳过并记入冲突名单 | +| 导出 CSV | 客户端列表、保留地址均可导出为 Excel 可读的 CSV | +| PXE 支持 | 填写 TFTP 服务器(option 66) 和启动文件名(option 67) 即可用于网络启动 | + +## 六、常见问题 + +**1. 启动失败：端口被占用（10013/10048）** +通常是系统开启了「Internet 连接共享 / 移动热点」，其内置 DHCP 分配器占了 UDP 67。 +用管理员运行 `net stop SharedAccess` 关闭，或 +`netstat -ano -p udp | findstr :67` 查看占用进程。 + +**2. 客户端拿不到 IP** +- 防火墙没放行（见第四节） +- 设备和电脑不在同一个二层网络（DHCP 依赖广播，不能跨路由器/跨 VPN） +- 网段填错：地址池必须与网卡 IP 同网段 + +**3. 多网卡电脑** +服务监听所有网卡的 UDP 67，但「服务网卡 IP」决定下发的网关/子网参数。 +插着 VPN 或虚拟网卡时请选对网卡，建议只保留接测试网的网卡。 + +**4. 和路由器的 DHCP 冲突？** +同一网段只能有一个 DHCP 服务器，否则客户端会随机拿到二者之一。 +要么关闭路由器 DHCP，要么把电脑接到独立交换机做纯测试网。 + +## 七、文件说明 + +| 文件 | 作用 | +| --- | --- | +| `便携DHCP服务器.exe` | **独立可执行程序**，免 Python，可单独拷贝到其他电脑使用 | +| `dhcp_server_app.py` | 主程序源码（引擎 + 界面） | +| `启动DHCP服务器.bat` | 源码方式一键启动（需装 Python） | +| `dhcp_config.json` | 点「保存配置」后生成的配置文件（exe 生成时保存在其所在目录） | +| `dhcp_leases.json` | 运行时自动生成的租约记录（可删除，删除后重新分配） | + +> exe 使用提示：配置与租约文件保存在 exe **同一目录**，拷贝 exe 时同目录 +> 的这两个 json 可一并拷贝以保留配置。首次运行若 Windows SmartScreen 提示 +> "已保护你的电脑"，点「更多信息」→「仍要运行」即可（自打包程序未签名属正常现象）。 + +## 八、重新打包 exe（可选） + +当前 `便携DHCP服务器.exe` 已打包完成。修改源码后如需重新打包： + +``` +python -m venv .temp\dhcppkg +.temp\dhcppkg\Scripts\python.exe -m pip install pyinstaller +.temp\dhcppkg\Scripts\pyinstaller.exe --onefile --windowed --name 便携DHCP服务器 dhcp_server_app.py +``` + +生成的 `dist\便携DHCP服务器.exe` 可单独拷贝使用。 + +## 九、使用须知 + +- 本工具面向**测试/实验网络**，请勿接入生产网络与正式 DHCP 服务混用 +- DHCP 基于 UDP 广播，仅适用于与电脑**同一物理网段**的设备 +- 程序不做鉴权，同一网段内任何设备都能申请到地址，注意使用场景 + +> AI生成 \ No newline at end of file diff --git a/启动DHCP服务器.bat b/启动DHCP服务器.bat new file mode 100644 index 0000000..452e424 --- /dev/null +++ b/启动DHCP服务器.bat @@ -0,0 +1,33 @@ +@echo off +rem ============================================ +rem Я DHCP - һ +rem ӷǽ, Ҽļѡ +rem "ԹԱ", ڳ"ǽ" +rem ============================================ +cd /d "%~dp0" + +where pythonw >nul 2>nul +if not errorlevel 1 ( + start "" pythonw "dhcp_server_app.py" + exit /b 0 +) +where python >nul 2>nul +if not errorlevel 1 ( + start "" python "dhcp_server_app.py" + exit /b 0 +) +where py >nul 2>nul +if not errorlevel 1 ( + start "" py "dhcp_server_app.py" + exit /b 0 +) + +echo [ʾ] δ⵽ Python +echo Ȱװ Python 3.10 ߰汾(װʱѡ "Add python.exe to PATH"): +echo https://www.python.org/downloads/ +echo. +echo Ŵʾ: 󴰿һ, ֶ +echo python dhcp_server_app.py +echo ɿ屨Ϣ +echo. +pause