cellular-proxy/scripts/admin-api.py

#!/usr/bin/env python3
"""轻量管理 API：用面板 secret 鉴权，配置代理账号密码。

默认监听 127.0.0.1:9091，UI 经同源时用相对路径 /cpxy-admin/*
（由 nginx 反代或浏览器直连时配置）。

为免引入 nginx，默认让 UI 调 http://host:9091（与 9090 同机）。
"""
from __future__ import annotations

import json
import os
import re
import subprocess
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse

INSTALL_DIR = Path(os.environ.get("INSTALL_DIR", "/opt/cellular-proxy"))
SETTINGS = Path(os.environ.get("CPXY_SETTINGS", str(INSTALL_DIR / "etc" / "settings.conf")))
LISTEN_HOST = os.environ.get("ADMIN_LISTEN_HOST", "0.0.0.0")
LISTEN_PORT = int(os.environ.get("ADMIN_LISTEN_PORT", "9091"))
APPLY_SCRIPT = Path(os.environ.get("CPXY_APPLY_AUTH", str(INSTALL_DIR / "scripts" / "apply-proxy-auth.sh")))


def load_settings(path: Path) -> dict:
    data = {}
    if not path.is_file():
        return data
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        data[k.strip()] = v.strip()
    return data


def mask(s: str) -> str:
    if not s:
        return ""
    if len(s) <= 2:
        return "***"
    return s[0] + "***" + s[-1]


class Handler(BaseHTTPRequestHandler):
    server_version = "cellular-proxy-admin/1.0"

    def log_message(self, fmt, *args):
        sys.stderr.write("[admin-api] " + (fmt % args) + "\n")

    def _cors(self):
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")

    def _json(self, code: int, obj):
        body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
        self.send_response(code)
        self._cors()
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _auth_ok(self) -> bool:
        conf = load_settings(SETTINGS)
        secret = conf.get("PANEL_SECRET", "")
        if not secret:
            return False
        auth = self.headers.get("Authorization", "")
        if auth.startswith("Bearer "):
            return auth[7:].strip() == secret
        if auth.startswith("Bearer"):
            return auth[6:].strip() == secret
        # 兼容 query ?secret=
        return False

    def do_OPTIONS(self):
        self.send_response(204)
        self._cors()
        self.end_headers()

    def do_GET(self):
        path = urlparse(self.path).path.rstrip("/") or "/"
        if path in ("/health", "/cpxy-admin/health"):
            self._json(200, {"ok": True})
            return
        if path not in ("/proxy-auth", "/cpxy-admin/proxy-auth"):
            self._json(404, {"error": "not found"})
            return
        if not self._auth_ok():
            self._json(401, {"message": "Unauthorized"})
            return
        conf = load_settings(SETTINGS)
        user = conf.get("PROXY_USER", "")
        pw = conf.get("PROXY_PASS", "")
        self._json(
            200,
            {
                "proxy_user": user,
                "proxy_pass_set": bool(pw),
                "proxy_pass_masked": mask(pw) if pw else "",
                "auth_enabled": bool(user),
                "port": conf.get("PROXY_MIXED_PORT", "7890"),
            },
        )

    def do_POST(self):
        path = urlparse(self.path).path.rstrip("/") or "/"
        if path not in ("/proxy-auth", "/cpxy-admin/proxy-auth"):
            self._json(404, {"error": "not found"})
            return
        if not self._auth_ok():
            self._json(401, {"message": "Unauthorized"})
            return
        length = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(length) if length else b"{}"
        try:
            body = json.loads(raw.decode("utf-8") or "{}")
        except json.JSONDecodeError:
            self._json(400, {"error": "invalid json"})
            return

        clear = bool(body.get("clear"))
        user = (body.get("user") or body.get("proxy_user") or "").strip()
        password = body.get("pass") or body.get("password") or body.get("proxy_pass") or ""

        if not APPLY_SCRIPT.is_file():
            self._json(500, {"error": f"apply script missing: {APPLY_SCRIPT}"})
            return

        cmd = ["bash", str(APPLY_SCRIPT)]
        if clear or (not user and not password):
            cmd.append("--clear")
        else:
            if not user:
                self._json(400, {"error": "user required (or set clear:true)"})
                return
            if password is None:
                self._json(400, {"error": "pass required"})
                return
            # 基本字符约束，避免注入
            if not re.fullmatch(r"[A-Za-z0-9_@./+\-]{1,64}", user):
                self._json(400, {"error": "user 仅允许字母数字及 _@.+/-"})
                return
            if len(str(password)) > 128:
                self._json(400, {"error": "password too long"})
                return
            cmd.extend(["--user", user, "--pass", str(password)])

        try:
            r = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=60,
                env={**os.environ, "INSTALL_DIR": str(INSTALL_DIR)},
            )
        except subprocess.TimeoutExpired:
            self._json(504, {"error": "apply timeout"})
            return

        out = (r.stdout or "") + (r.stderr or "")
        if r.returncode != 0:
            self._json(500, {"error": "apply failed", "detail": out[-2000:]})
            return

        conf = load_settings(SETTINGS)
        self._json(
            200,
            {
                "ok": True,
                "proxy_user": conf.get("PROXY_USER", ""),
                "proxy_pass_set": bool(conf.get("PROXY_PASS")),
                "auth_enabled": bool(conf.get("PROXY_USER")),
                "log": out[-1500:],
            },
        )


def main():
    if not SETTINGS.is_file():
        print(f"warn: settings not found yet: {SETTINGS}", file=sys.stderr)
    httpd = ThreadingHTTPServer((LISTEN_HOST, LISTEN_PORT), Handler)
    print(f"cellular-proxy admin API on {LISTEN_HOST}:{LISTEN_PORT}", flush=True)
    httpd.serve_forever()


if __name__ == "__main__":
    main()