feat: add web admin panel for node management
Add Flask panel with login, add/delete nodes, and share link copy. Generate sing-box config from SQLite; add uninstall script and clean install flow. Panel served at https://DOMAIN:8444 via nginx. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
+167
@@ -0,0 +1,167 @@
|
||||
#!/usr/bin/env python3
|
||||
"""jiedian 管理面板:登录、添加/删除节点、复制分享链接。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import secrets
|
||||
import subprocess
|
||||
from functools import wraps
|
||||
from pathlib import Path
|
||||
|
||||
from flask import (
|
||||
Flask,
|
||||
jsonify,
|
||||
redirect,
|
||||
render_template,
|
||||
request,
|
||||
session,
|
||||
url_for,
|
||||
)
|
||||
|
||||
from db import add_node, delete_node, list_nodes, node_count, verify_admin
|
||||
from links import build_links, load_env
|
||||
|
||||
ROOT = Path(os.environ.get("JIEDIAN_ROOT", Path(__file__).resolve().parents[1]))
|
||||
SECRET_FILE = ROOT / "data" / ".panel_secret"
|
||||
RENDER_SCRIPT = ROOT / "scripts" / "render-server.py"
|
||||
|
||||
|
||||
def _secret_key() -> str:
|
||||
if SECRET_FILE.exists():
|
||||
return SECRET_FILE.read_text(encoding="utf-8").strip()
|
||||
SECRET_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
key = secrets.token_hex(32)
|
||||
SECRET_FILE.write_text(key, encoding="utf-8")
|
||||
SECRET_FILE.chmod(0o600)
|
||||
return key
|
||||
|
||||
|
||||
app = Flask(__name__)
|
||||
app.secret_key = _secret_key()
|
||||
app.config.update(
|
||||
SESSION_COOKIE_HTTPONLY=True,
|
||||
SESSION_COOKIE_SAMESITE="Lax",
|
||||
PERMANENT_SESSION_LIFETIME=86400 * 7,
|
||||
)
|
||||
|
||||
|
||||
def login_required(view):
|
||||
@wraps(view)
|
||||
def wrapped(*args, **kwargs):
|
||||
if not session.get("user"):
|
||||
if request.path.startswith("/api/"):
|
||||
return jsonify({"error": "未登录"}), 401
|
||||
return redirect(url_for("login"))
|
||||
return view(*args, **kwargs)
|
||||
|
||||
return wrapped
|
||||
|
||||
|
||||
def apply_singbox() -> tuple[bool, str]:
|
||||
env = os.environ.copy()
|
||||
env["JIEDIAN_ROOT"] = str(ROOT)
|
||||
proc = subprocess.run(
|
||||
["python3", str(RENDER_SCRIPT)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
return False, proc.stderr or proc.stdout or "配置生成失败"
|
||||
restart = subprocess.run(["systemctl", "restart", "sing-box"], capture_output=True, text=True)
|
||||
if restart.returncode != 0:
|
||||
return False, restart.stderr or restart.stdout or "sing-box 重启失败"
|
||||
return True, "ok"
|
||||
|
||||
|
||||
@app.route("/login", methods=["GET", "POST"])
|
||||
def login():
|
||||
if session.get("user"):
|
||||
return redirect(url_for("dashboard"))
|
||||
error = None
|
||||
if request.method == "POST":
|
||||
username = request.form.get("username", "").strip()
|
||||
password = request.form.get("password", "")
|
||||
if verify_admin(username, password):
|
||||
session.permanent = True
|
||||
session["user"] = username
|
||||
return redirect(url_for("dashboard"))
|
||||
error = "用户名或密码错误"
|
||||
return render_template("login.html", error=error)
|
||||
|
||||
|
||||
@app.route("/logout")
|
||||
def logout():
|
||||
session.clear()
|
||||
return redirect(url_for("login"))
|
||||
|
||||
|
||||
@app.route("/")
|
||||
@login_required
|
||||
def dashboard():
|
||||
env = load_env()
|
||||
nodes = []
|
||||
for node in list_nodes():
|
||||
item = dict(node)
|
||||
item["links"] = build_links(node, env)
|
||||
nodes.append(item)
|
||||
return render_template(
|
||||
"dashboard.html",
|
||||
nodes=nodes,
|
||||
domain=env.get("DOMAIN", ""),
|
||||
vps_ip=env.get("VPS_IP", ""),
|
||||
)
|
||||
|
||||
|
||||
@app.route("/api/nodes", methods=["GET"])
|
||||
@login_required
|
||||
def api_list_nodes():
|
||||
env = load_env()
|
||||
data = []
|
||||
for node in list_nodes():
|
||||
item = {
|
||||
"id": node["id"],
|
||||
"name": node["name"],
|
||||
"uuid": node["uuid"],
|
||||
"created_at": node["created_at"],
|
||||
"links": build_links(node, env),
|
||||
}
|
||||
data.append(item)
|
||||
return jsonify(data)
|
||||
|
||||
|
||||
@app.route("/api/nodes", methods=["POST"])
|
||||
@login_required
|
||||
def api_add_node():
|
||||
body = request.get_json(silent=True) or {}
|
||||
name = (body.get("name") or request.form.get("name") or "新节点").strip()
|
||||
node = add_node(name)
|
||||
ok, msg = apply_singbox()
|
||||
if not ok:
|
||||
delete_node(node["id"])
|
||||
return jsonify({"error": msg}), 500
|
||||
env = load_env()
|
||||
return jsonify(
|
||||
{
|
||||
"id": node["id"],
|
||||
"name": node["name"],
|
||||
"links": build_links(node, env),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@app.route("/api/nodes/<int:node_id>", methods=["DELETE"])
|
||||
@login_required
|
||||
def api_delete_node(node_id: int):
|
||||
if node_count() <= 1:
|
||||
return jsonify({"error": "至少保留一个节点"}), 400
|
||||
if not delete_node(node_id):
|
||||
return jsonify({"error": "节点不存在"}), 404
|
||||
ok, msg = apply_singbox()
|
||||
if not ok:
|
||||
return jsonify({"error": msg}), 500
|
||||
return jsonify({"ok": True})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(host="127.0.0.1", port=5080)
|
||||
Reference in New Issue
Block a user