jiedian/panel/app.py

#!/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 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

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,
)

_panel_path = os.environ.get("PANEL_PATH", "").strip().strip("/")
if _panel_path:
    app.config["SESSION_COOKIE_PATH"] = f"/{_panel_path}/"
    app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)


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)