""" DHCP Web 管理界面 - Flask 主程序 - 用户认证（用户名/密码，session） - subnet/host 增删改，配置自动重载 - 租约查询、地址池使用情况 - 接口绑定配置 - 操作日志 设计：用户的元数据（subnet/host/操作日志/接口）全在 SQLite，实际配置写在 /etc/dhcp/dhcpd.conf。所有改动通过"Web -> 写文件 -> dhcpd -t -> systemctl reload" 流程。 """ import os import re import json from datetime import datetime, timedelta from functools import wraps from flask import (Flask, render_template, request, redirect, url_for, session, jsonify, flash, abort) from flask_sqlalchemy import SQLAlchemy from sqlalchemy import event from werkzeug.security import generate_password_hash, check_password_hash # 强制使用 pbkdf2:sha256，避免 scrypt 在某些环境下跨进程兼容性 bug _PASSWORD_METHOD = "pbkdf2:sha256" import dhcp_ops # --------------------------------------------------------------------------- # Flask & DB 配置 # --------------------------------------------------------------------------- BASE_DIR = os.path.dirname(os.path.abspath(__file__)) INSTANCE_DIR = os.path.join(BASE_DIR, "instance") os.makedirs(INSTANCE_DIR, exist_ok=True) app = Flask(__name__) app.config["SECRET_KEY"] = os.environ.get("DHCPMGR_SECRET", "change-me-on-first-deploy") app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{os.path.join(INSTANCE_DIR, 'dhcp.db')}" app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False app.config["JSON_AS_ASCII"] = False db = SQLAlchemy(app) # --------------------------------------------------------------------------- # 模型 # --------------------------------------------------------------------------- class User(db.Model): __tablename__ = "users" id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True, nullable=False) password_hash = db.Column(db.String(200), nullable=False) role = db.Column(db.String(20), default="admin") # 简单起见只有 admin / user created_at = db.Column(db.DateTime, default=lambda: datetime.utcnow()) last_login = db.Column(db.DateTime) def set_password(self, pw): self.password_hash = generate_password_hash(pw, method="pbkdf2:sha256") def check_password(self, pw): return check_password_hash(self.password_hash, pw) class OperationLog(db.Model): __tablename__ = "op_logs" id = db.Column(db.Integer, primary_key=True) timestamp = db.Column(db.DateTime, default=lambda: datetime.utcnow()) user = db.Column(db.String(80)) action = db.Column(db.String(50)) # create_subnet / update_subnet / delete_subnet / etc. target = db.Column(db.String(200)) # 子网网络、主机名等 detail = db.Column(db.Text) # JSON / 文本详情 result = db.Column(db.String(20)) # ok / error message = db.Column(db.Text) # --------------------------------------------------------------------------- # 辅助函数 # --------------------------------------------------------------------------- def current_user(): if "user_id" not in session: return None return db.session.get(User, session["user_id"]) def login_required(f): @wraps(f) def wrapper(*a, **kw): if "user_id" not in session: return redirect(url_for("login", next=request.path)) return f(*a, **kw) return wrapper def admin_required(f): @wraps(f) def wrapper(*a, **kw): u = current_user() if not u: return redirect(url_for("login", next=request.path)) if u.role != "admin": abort(403) return f(*a, **kw) return wrapper def log_op(action, target, detail, result, message): u = current_user() entry = OperationLog( user=(u.username if u else "anonymous"), action=action, target=target, detail=detail if isinstance(detail, str) else json.dumps(detail, ensure_ascii=False), result=result, message=message, ) db.session.add(entry) db.session.commit() return entry def get_subnets() -> list: block = dhcp_ops.read_managed_block() subs, _ = dhcp_ops.parse_managed_block(block) return subs def get_hosts() -> list: block = dhcp_ops.read_managed_block() _, hosts = dhcp_ops.parse_managed_block(block) return hosts def apply_and_reload(action: str, target: str, extra: dict = "", subnets_override=None, hosts_override=None): """ 通用的"改 conf -> 语法检查 -> reload" 流程。 subnets_override / hosts_override：调用方已修改内存中的对象，绕过从文件重读。 返回 (success, message) """ subnets = subnets_override if subnets_override is not None else get_subnets() hosts = hosts_override if hosts_override is not None else get_hosts() ok, msg = dhcp_ops.write_managed_block(subnets, hosts) if not ok: log_op(action, target, extra, "error", msg) return False, msg syntax_ok, syntax_msg = dhcp_ops.check_syntax() if not syntax_ok: log_op(action, target, extra, "error", f"语法错误: {syntax_msg}") return False, f"语法检查失败:\n{syntax_msg}\n（配置未生效，已备份到 backups/）" r_ok, r_msg = dhcp_ops.reload_service() if not r_ok: log_op(action, target, extra, "error", f"reload 失败: {r_msg}") return False, f"reload 失败: {r_msg}" log_op(action, target, extra, "ok", f"{msg}; 语法 OK; {r_msg}") return True, f"{msg}，{r_msg}" # --------------------------------------------------------------------------- # 上下文：菜单 / 当前用户 # --------------------------------------------------------------------------- @app.context_processor def inject_globals(): u = current_user() return dict( current_user=u, is_admin=(u.role == "admin" if u else False), now=datetime.utcnow(), ) # --------------------------------------------------------------------------- # 路由 # --------------------------------------------------------------------------- @app.route("/login", methods=["GET", "POST"]) def login(): if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "") user = User.query.filter_by(username=username).first() if user and user.check_password(password): session["user_id"] = user.id user.last_login = datetime.utcnow() db.session.commit() return redirect(request.args.get("next") or url_for("dashboard")) flash("用户名或密码错误", "error") return render_template("login.html") @app.route("/logout") def logout(): session.pop("user_id", None) return redirect(url_for("login")) @app.route("/") @login_required def dashboard(): subnets = get_subnets() hosts = get_hosts() leases = dhcp_ops.active_leases() svc = dhcp_ops.get_service_status() # 计算地址池使用率 pool_stats = [] lease_ips = {l.ip for l in leases} for s in subnets: try: from ipaddress import ip_network, ip_address net = ip_network(f"{s.network}/{s.netmask}", strict=False) start = ip_address(s.range_start) end = ip_address(s.range_end) pool_total = int(end) - int(start) + 1 used = sum(1 for ip in lease_ips if start <= ip_address(ip) <= end and ip_address(ip) in net) # 减去 host 静态绑定数 host_static_in_pool = sum( 1 for h in hosts if h.ip and s.range_start <= h.ip <= s.range_end ) pool_stats.append({ "subnet": f"{s.network}/{s.netmask}", "range": f"{s.range_start} - {s.range_end}", "total": pool_total, "used": used, "static": host_static_in_pool, "available": pool_total - used - host_static_in_pool, "util_pct": round(used / pool_total * 100, 1) if pool_total > 0 else 0, }) except Exception as e: pool_stats.append({"subnet": s.network, "error": str(e)}) return render_template("dashboard.html", subnets=subnets, hosts=hosts, leases=leases[:10], lease_total=len(leases), pool_stats=pool_stats, svc=svc) # ---------- subnet ---------- @app.route("/subnets") @login_required def subnet_list(): return render_template("subnets.html", subnets=get_subnets(), hosts=get_hosts()) @app.route("/subnets/new", methods=["GET", "POST"]) @admin_required def subnet_new(): if request.method == "POST": try: s = dhcp_ops.Subnet( network=request.form["network"].strip(), netmask=request.form["netmask"].strip(), routers=request.form["routers"].strip(), dns=request.form["dns"].strip(), range_start=request.form["range_start"].strip(), range_end=request.form["range_end"].strip(), default_lease=int(request.form.get("default_lease") or 600), max_lease=int(request.form.get("max_lease") or 7200), domain_name=request.form.get("domain_name", "").strip(), authoritative=bool(request.form.get("authoritative")), ) except (KeyError, ValueError) as e: flash(f"表单错误: {e}", "error") return render_template("subnet_form.html", subnet={}, form=request.form) errs = s.validate() if errs: for e in errs: flash(e, "error") return render_template("subnet_form.html", subnet={}, form=request.form) # 写入：先获取现有，追加，再走 apply_and_reload subnets = get_subnets() if any(x.network == s.network and x.netmask == s.netmask for x in subnets): flash(f"已存在相同 subnet {s.network}/{s.netmask}", "error") return render_template("subnet_form.html", subnet={}, form=request.form) subnets.append(s) ok, msg = apply_and_reload("create_subnet", s.network, asdict_safe(s), subnets_override=subnets) if ok: flash(f"subnet {s.network}/{s.netmask} 已创建，{msg}", "ok") return redirect(url_for("subnet_list")) flash(msg, "error") return render_template("subnet_form.html", subnet={}, form=request.form) return render_template("subnet_form.html", subnet={}, form={}) @app.route("/subnets///edit", methods=["GET", "POST"]) @admin_required def subnet_edit(network, netmask): subnets = get_subnets() target = next((s for s in subnets if s.network == network and s.netmask == netmask), None) if not target: abort(404) if request.method == "POST": try: new_s = dhcp_ops.Subnet( network=request.form["network"].strip(), netmask=request.form["netmask"].strip(), routers=request.form["routers"].strip(), dns=request.form["dns"].strip(), range_start=request.form["range_start"].strip(), range_end=request.form["range_end"].strip(), default_lease=int(request.form.get("default_lease") or 600), max_lease=int(request.form.get("max_lease") or 7200), domain_name=request.form.get("domain_name", "").strip(), authoritative=bool(request.form.get("authoritative")), ) except (KeyError, ValueError) as e: flash(f"表单错误: {e}", "error") return render_template("subnet_form.html", subnet=target, form=request.form) errs = new_s.validate() if errs: for e in errs: flash(e, "error") return render_template("subnet_form.html", subnet=new_s, form=request.form) subnets = [new_s if (s.network == network and s.netmask == netmask) else s for s in subnets] ok, msg = apply_and_reload( "update_subnet", new_s.network, {"old": f"{network}/{netmask}", "new": asdict_safe(new_s)}, subnets_override=subnets, ) if ok: flash(f"subnet 已更新，{msg}", "ok") return redirect(url_for("subnet_list")) flash(msg, "error") return render_template("subnet_form.html", subnet=new_s, form=request.form) return render_template("subnet_form.html", subnet=target, form=asdict_safe(target)) @app.route("/subnets///delete", methods=["POST"]) @admin_required def subnet_delete(network, netmask): subnets = [s for s in get_subnets() if not (s.network == network and s.netmask == netmask)] ok, msg = apply_and_reload( "delete_subnet", f"{network}/{netmask}", {"remaining_subnets": len(subnets)}, subnets_override=subnets, ) flash(f"已删除 subnet {network}/{netmask}，{msg}", "ok" if ok else "error") return redirect(url_for("subnet_list")) # ---------- host ---------- @app.route("/hosts") @login_required def host_list(): return render_template("hosts.html", hosts=get_hosts(), subnets=get_subnets()) @app.route("/hosts/new", methods=["GET", "POST"]) @admin_required def host_new(): if request.method == "POST": try: h = dhcp_ops.Host( name=request.form["name"].strip(), mac=request.form["mac"].strip().lower(), ip=request.form["ip"].strip(), subnet_network=request.form.get("subnet_network", "").strip(), ) except (KeyError, ValueError) as e: flash(f"表单错误: {e}", "error") return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets()) errs = h.validate() if errs: for e in errs: flash(e, "error") return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets()) hosts = get_hosts() if any(x.name == h.name for x in hosts): flash(f"已存在同名 host: {h.name}", "error") return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets()) if any(x.ip == h.ip for x in hosts): flash(f"IP {h.ip} 已被其他 host 占用", "error") return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets()) hosts.append(h) ok, msg = apply_and_reload("create_host", h.name, asdict_safe(h), hosts_override=hosts) if ok: flash(f"host {h.name} 已创建，{msg}", "ok") return redirect(url_for("host_list")) flash(msg, "error") return render_template("host_form.html", host={}, form=request.form, subnets=get_subnets()) return render_template("host_form.html", host={}, form={}, subnets=get_subnets()) @app.route("/hosts//edit", methods=["GET", "POST"]) @admin_required def host_edit(name): hosts = get_hosts() target = next((h for h in hosts if h.name == name), None) if not target: abort(404) if request.method == "POST": try: new_h = dhcp_ops.Host( name=request.form["name"].strip(), mac=request.form["mac"].strip().lower(), ip=request.form["ip"].strip(), subnet_network=request.form.get("subnet_network", "").strip(), ) except (KeyError, ValueError) as e: flash(f"表单错误: {e}", "error") return render_template("host_form.html", host=target, form=request.form, subnets=get_subnets()) errs = new_h.validate() if errs: for e in errs: flash(e, "error") return render_template("host_form.html", host=new_h, form=request.form, subnets=get_subnets()) # 名字变更则检查冲突 if new_h.name != name and any(h.name == new_h.name for h in hosts): flash(f"已存在同名 host: {new_h.name}", "error") return render_template("host_form.html", host=new_h, form=request.form, subnets=get_subnets()) hosts = [new_h if h.name == name else h for h in hosts] ok, msg = apply_and_reload("update_host", new_h.name, {"old": name, "new": asdict_safe(new_h)}, hosts_override=hosts) if ok: flash(f"host 已更新，{msg}", "ok") return redirect(url_for("host_list")) flash(msg, "error") return render_template("host_form.html", host=new_h, form=request.form, subnets=get_subnets()) return render_template("host_form.html", host=target, form=asdict_safe(target), subnets=get_subnets()) @app.route("/hosts//delete", methods=["POST"]) @admin_required def host_delete(name): hosts = [h for h in get_hosts() if h.name != name] ok, msg = apply_and_reload("delete_host", name, {"remaining_hosts": len(hosts)}, hosts_override=hosts) flash(f"已删除 host {name}，{msg}", "ok" if ok else "error") return redirect(url_for("host_list")) # ---------- leases ---------- @app.route("/leases") @login_required def lease_list(): leases = dhcp_ops.parse_leases() q = request.args.get("q", "").strip().lower() if q: leases = [l for l in leases if q in l.ip.lower() or q in l.mac.lower() or q in l.client_hostname.lower()] leases.sort(key=lambda l: l.remaining_seconds() or -1, reverse=True) return render_template("leases.html", leases=leases, q=q) # ---------- 网络扫描 (arp-scan) ---------- @app.route("/scan", methods=["GET", "POST"]) @login_required def scan(): """网络扫描 - 用 arp-scan 列出网段内所有活跃设备""" ifaces = dhcp_ops.get_interfaces() iface = request.values.get("iface") or (ifaces[0] if ifaces else "enp0s31f6-ovs") network = request.values.get("network") or dhcp_ops.get_local_network(iface) devices = [] scan_msg = "" did_scan = False if request.method == "POST" or request.args.get("auto") == "1": did_scan = True ok, scan_msg, devices = dhcp_ops.arp_scan(iface, network) if ok: # 交叉 dhcpd.leases 和 host 静态绑定 hosts = get_hosts() leases = dhcp_ops.parse_leases() devices = dhcp_ops.classify_devices(devices, hosts, leases) return render_template("scan.html", ifaces=ifaces, iface=iface, network=network, devices=devices, scan_msg=scan_msg, did_scan=did_scan) # ---------- config (interfaces + 原始 conf) ---------- @app.route("/config", methods=["GET", "POST"]) @admin_required def config_view(): if request.method == "POST": action = request.form.get("action") if action == "save_interfaces": ifaces = request.form.get("interfaces", "").strip().split() ok, msg = dhcp_ops.save_interfaces(ifaces) log_op("save_interfaces", ",".join(ifaces), "", "ok" if ok else "error", msg) flash(msg, "ok" if ok else "error") if ok: # 重新 reload 服务（需要重启以应用新 interface） r_ok, r_msg = dhcp_ops.restart_service() flash(r_msg, "ok" if r_ok else "error") log_op("restart_service", "", "", "ok" if r_ok else "error", r_msg) elif action == "reload": r_ok, r_msg = dhcp_ops.reload_service() flash(r_msg, "ok" if r_ok else "error") log_op("reload_service", "", "", "ok" if r_ok else "error", r_msg) elif action == "restart": r_ok, r_msg = dhcp_ops.restart_service() flash(r_msg, "ok" if r_ok else "error") log_op("restart_service", "", "", "ok" if r_ok else "error", r_msg) return redirect(url_for("config_view")) ifaces = dhcp_ops.get_interfaces() conf_content = "" if os.path.exists(dhcp_ops.DHCPD_CONF): with open(dhcp_ops.DHCPD_CONF) as f: conf_content = f.read() svc = dhcp_ops.get_service_status() syntax_ok, syntax_msg = dhcp_ops.check_syntax() return render_template("config.html", ifaces=ifaces, conf=conf_content, svc=svc, syntax_ok=syntax_ok, syntax_msg=syntax_msg) # ---------- 操作日志 ---------- @app.route("/logs") @admin_required def log_list(): page = request.args.get("page", 1, type=int) per = 50 q = OperationLog.query.order_by(OperationLog.id.desc()) total = q.count() logs = q.offset((page - 1) * per).limit(per).all() pages = (total + per - 1) // per return render_template("logs.html", logs=logs, page=page, pages=pages, total=total) # ---------- 用户管理（修改密码） ---------- @app.route("/profile", methods=["GET", "POST"]) @login_required def profile(): u = current_user() if request.method == "POST": old = request.form.get("old_password", "") new = request.form.get("new_password", "") if not u.check_password(old): flash("当前密码错误", "error") elif len(new) < 6: flash("新密码至少 6 位", "error") else: u.set_password(new) db.session.commit() flash("密码已更新", "ok") return redirect(url_for("profile")) return render_template("profile.html", user=u) # ---------- API（轻量查询） ---------- @app.route("/api/syntax_check") @admin_required def api_syntax(): ok, msg = dhcp_ops.check_syntax() return jsonify(ok=ok, msg=msg) # --------------------------------------------------------------------------- # 辅助：把 dataclass 转 dict 而非崩溃 # --------------------------------------------------------------------------- def asdict_safe(obj): try: from dataclasses import asdict return asdict(obj) except Exception: return {} def _format_duration(seconds): """把秒数格式化为 d/h/m/s 友好显示""" if seconds is None: return "—" seconds = int(seconds) if seconds < 0: return "已过期" if seconds < 60: return f"{seconds}s" if seconds < 3600: return f"{seconds // 60}m {seconds % 60}s" if seconds < 86400: h, rem = divmod(seconds, 3600) return f"{h}h {rem // 60}m" d, rem = divmod(seconds, 86400) h = rem // 3600 return f"{d}d {h}h" app.jinja_env.filters["format_duration"] = _format_duration # --------------------------------------------------------------------------- # 初始化 # --------------------------------------------------------------------------- def seed_admin(): """首次启动时创建默认管理员 admin/admin""" if User.query.count() == 0: admin = User(username="admin", role="admin") admin.set_password("admin") db.session.add(admin) db.session.commit() print("[init] created default user: admin / admin（请尽快修改）") def init_app(): with app.app_context(): db.create_all() seed_admin() if __name__ == "__main__": init_app() port = int(os.environ.get("PORT", "5100")) app.run(host="0.0.0.0", port=port, debug=False)