obsidian-vault/ligbox-ops-platform/deploy/vm123-finance-stack/openpanel-community-bridge/bridge.py

#!/usr/bin/env python3
"""OpenPanel Community → FOSSBilling API bridge (opencli backend). LAN only."""
from __future__ import annotations

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

HOST = os.environ.get("BRIDGE_HOST", "0.0.0.0")
PORT = int(os.environ.get("BRIDGE_PORT", "18087"))
ADMIN_USER = os.environ.get("BRIDGE_ADMIN_USER", "ligboxadmin")
ADMIN_PASS = os.environ.get("BRIDGE_ADMIN_PASS", "LbOpen805353")
TOKEN = os.environ.get("BRIDGE_TOKEN", "ligbox-community-bridge-token")


def run_opencli(*args: str, timeout: int = 120) -> tuple[int, str, str]:
    cmd = ["opencli", *args]
    proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    return proc.returncode, proc.stdout.strip(), proc.stderr.strip()


class Handler(BaseHTTPRequestHandler):
    server_version = "OpenPanelCommunityBridge/1.0"

    def log_message(self, fmt: str, *args) -> None:
        sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args))

    def _read_json(self) -> dict:
        length = int(self.headers.get("Content-Length", 0))
        if not length:
            return {}
        return json.loads(self.rfile.read(length).decode("utf-8"))

    def _send(self, code: int, payload: dict) -> None:
        body = json.dumps(payload).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _auth_ok(self) -> bool:
        auth = self.headers.get("Authorization", "")
        if auth == f"Bearer {TOKEN}":
            return True
        return False

    def do_POST(self) -> None:
        path = urlparse(self.path).path.rstrip("/") or "/"
        if path == "/api":
            data = self._read_json()
            if data.get("username") == ADMIN_USER and data.get("password") == ADMIN_PASS:
                self._send(200, {"access_token": TOKEN})
            else:
                self._send(401, {"error": "Invalid credentials"})
            return

        if not self._auth_ok():
            self._send(401, {"error": "Unauthorized"})
            return

        if path == "/api/users":
            data = self._read_json()
            username = data.get("username", "")
            password = data.get("password", "")
            email = data.get("email", "")
            plan = data.get("plan_name", "ligbox-site-cms")
            if not re.fullmatch(r"[a-z][a-z0-9]{2,15}", username):
                self._send(400, {"success": False, "error": f"Invalid username: {username}"})
                return
            code, out, err = run_opencli(
                "user-add", username, password, email, plan, "--no-sentinel"
            )
            msg = out or err
            if code == 0 or "Successfully added user" in msg:
                self._send(200, {
                    "success": True,
                    "response": {"message": msg or f"Successfully added user {username}"},
                })
            else:
                self._send(500, {"success": False, "error": msg})
            return

        self._send(404, {"error": "This api route does not exist."})

    def do_GET(self) -> None:
        path = urlparse(self.path).path.rstrip("/") or "/"
        if path == "/api":
            if not self._auth_ok():
                self._send(401, {"error": "Unauthorized"})
                return
            self._send(200, {"message": "API is working!"})
            return
        self._send(404, {"error": "This api route does not exist."})

    def do_PATCH(self) -> None:
        if not self._auth_ok():
            self._send(401, {"error": "Unauthorized"})
            return
        path = urlparse(self.path).path
        m = re.match(r"^/api/users/([a-z0-9]+)$", path)
        if not m:
            self._send(404, {"error": "This api route does not exist."})
            return
        username = m.group(1)
        data = self._read_json()
        action = data.get("action")
        if action == "suspend":
            code, out, err = run_opencli("user-suspend", username)
        elif action == "unsuspend":
            code, out, err = run_opencli("user-unsuspend", username)
        elif "password" in data:
            code, out, err = run_opencli("user-password", username, data["password"])
        else:
            self._send(400, {"success": False, "error": "Unknown action"})
            return
        if code == 0:
            self._send(200, {"success": True})
        else:
            self._send(500, {"success": False, "error": out or err})

    def do_DELETE(self) -> None:
        if not self._auth_ok():
            self._send(401, {"error": "Unauthorized"})
            return
        path = urlparse(self.path).path
        m = re.match(r"^/api/users/([a-z0-9]+)$", path)
        if not m:
            self._send(404, {"error": "This api route does not exist."})
            return
        username = m.group(1)
        code, out, err = run_opencli("user-delete", username, "-y")
        if code == 0:
            self._send(200, {"success": True})
        else:
            self._send(500, {"success": False, "error": out or err})


def main() -> None:
    httpd = ThreadingHTTPServer((HOST, PORT), Handler)
    print(f"OpenPanel Community bridge on http://{HOST}:{PORT}", flush=True)
    httpd.serve_forever()


if __name__ == "__main__":
    main()