DHCP Service Bot
21e3d61871
New /scan page lists all active devices on the subnet (ARP scan): - 静态绑定: matches a host declaration - 动态分配: matches a dhcpd.leases entry - 未知 / 外部: nothing matches (manually configured or from other DHCP server) Click '转为静态绑定' to convert a discovered device to a static host binding. Also includes: - arp-scan wrapper in dhcp_ops.py (get_local_network, arp_scan, classify_devices) - /scan route with optional auto-refresh every 30s - README updated with arp-scan install step + new feature bullet
630 lines
23 KiB
Python
630 lines
23 KiB
Python
"""
|
||
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/<network>/<netmask>/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/<network>/<netmask>/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/<name>/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/<name>/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) |