#!/usr/bin/env python3 """jiedian 管理面板：登录、添加/删除节点、复制分享链接。""" from __future__ import annotations import os import secrets import subprocess import threading from functools import wraps from pathlib import Path from flask import ( Flask, jsonify, redirect, render_template, request, session, url_for, ) from werkzeug.middleware.proxy_fix import ProxyFix from db import add_node, delete_node, list_nodes, node_count, verify_admin from links import build_links, load_env from stats import collect_node_stats 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" RENDER_XRAY_SCRIPT = ROOT / "scripts" / "render-xray.py" _apply_lock = threading.Lock() 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, ) _panel_path = os.environ.get("PANEL_PATH", "").strip().strip("/") _panel_domain = os.environ.get("PANEL_DOMAIN", "").strip() if _panel_path: app.config["SESSION_COOKIE_PATH"] = f"/{_panel_path}/" app.config["PREFERRED_URL_SCHEME"] = "http" app.wsgi_app = ProxyFix( app.wsgi_app, x_for=1, x_proto=1, x_host=0, x_prefix=0 ) class _PanelPrefixMiddleware: """始终用 PANEL_PATH 设置 SCRIPT_NAME，不依赖反代头是否完整。""" def __init__(self, app, prefix: str, domain: str = ""): self.app = app self.script_name = f"/{prefix.strip('/')}" self.domain = domain def __call__(self, environ, start_response): environ["SCRIPT_NAME"] = self.script_name if self.domain and not environ.get("HTTP_HOST"): environ["HTTP_HOST"] = self.domain if not environ.get("HTTP_X_FORWARDED_PROTO"): environ["HTTP_X_FORWARDED_PROTO"] = "http" return self.app(environ, start_response) if _panel_path: app.wsgi_app = _PanelPrefixMiddleware( app.wsgi_app, _panel_path, _panel_domain ) @app.context_processor def inject_panel_base() -> dict[str, str]: base = f"/{_panel_path}" if _panel_path else "" return {"panel_base": base} 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 render_singbox_config() -> 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 "sing-box 配置生成失败" return True, "ok" def render_xray_config() -> tuple[bool, str]: env = os.environ.copy() env["JIEDIAN_ROOT"] = str(ROOT) proc = subprocess.run( ["python3", str(RENDER_XRAY_SCRIPT)], capture_output=True, text=True, env=env, ) if proc.returncode != 0: return False, proc.stderr or proc.stdout or "Xray 配置生成失败" return True, "ok" def restart_services_async() -> None: """后台重启 sing-box 与 Xray。""" subprocess.Popen( ["systemctl", "restart", "xray", "sing-box"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) def apply_singbox() -> tuple[bool, str]: ok, msg = render_singbox_config() if not ok: return False, msg ok, msg = render_xray_config() if not ok: return False, msg restart_services_async() return True, "ok" def restart_singbox_async() -> None: restart_services_async() def apply_singbox_background(on_fail=None) -> None: """后台生成配置并重启 sing-box，避免阻塞 HTTP 请求导致 Nginx 503。""" def worker() -> None: with _apply_lock: ok, msg = apply_singbox() if not ok and on_fail: on_fail(msg) threading.Thread(target=worker, daemon=True).start() @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/stats", methods=["GET"]) @login_required def api_stats(): return jsonify(collect_node_stats()) @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) env = load_env() node_id = int(node["id"]) def on_fail(_msg: str) -> None: delete_node(node_id) apply_singbox_background(on_fail=on_fail) return jsonify( { "id": node["id"], "name": node["name"], "links": build_links(node, env), "pending": True, } ) @app.route("/api/nodes/", 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 apply_singbox_background() return jsonify({"ok": True}) if __name__ == "__main__": app.run(host="127.0.0.1", port=5080, threaded=True)