feat: 一条 curl 自动装/升（未装安装、已装升级）

2026-07-21 14:23:31 +00:00
commit b2a7332f86
20 changed files with 3459 additions and 0 deletions
@@ -0,0 +1,193 @@
+#!/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()