jiedian/panel/app.py

#!/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"
_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"] = "https"
    app.config["SESSION_COOKIE_SECURE"] = True
    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"] = "https"
        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 restart_singbox_async() -> None:
    subprocess.Popen(
        ["systemctl", "restart", "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
    restart_singbox_async()
    return True, "ok"


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/<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
    apply_singbox_background()
    return jsonify({"ok": True})


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5080, threaded=True)