CASAN/packages/casan-control-panel/scripts/provider-auth-bridge.py

#!/usr/bin/env python3
"""Local-only bridge between CASAN Control Panel and official provider CLIs.

The bridge never reads credential files. It asks the installed Codex/Claude
CLI for a sanitized status and can launch their official browser login flow.
Only fixed commands are allowed and every request requires a generated bearer
token supplied to the backend container by local-full.sh.
"""
import argparse
import hashlib
import hmac
import json
import os
import shutil
import subprocess
import tempfile
import threading
import time
from collections import deque
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Dict, Optional


PROVIDERS = {
    "codex": {
        "label": "OpenAI Codex",
        "binary": "codex",
        "status": ["codex", "login", "status"],
        "login": ["codex", "login"],
    },
    "claude": {
        "label": "Anthropic Claude Code",
        "binary": "claude",
        "status": ["claude", "auth", "status", "--json"],
        "login": ["claude", "auth", "login", "--claudeai"],
    },
}
RUNNING: Dict[str, subprocess.Popen] = {}
LOCK = threading.Lock()
MODEL_GATES = {provider: threading.Semaphore(1) for provider in PROVIDERS}
MODEL_WINDOWS = {provider: deque() for provider in PROVIDERS}
RATE_LOCK = threading.Lock()
AUDIT_LOCK = threading.Lock()


def model_call_allowed(provider: str) -> bool:
    timestamp = time.monotonic()
    with RATE_LOCK:
        window = MODEL_WINDOWS[provider]
        while window and timestamp - window[0] > 600:
            window.popleft()
        if len(window) >= 10:
            return False
        window.append(timestamp)
        return True


def audit_model_call(path: str, provider: str, status: str, prompt: str, latency_ms: int) -> None:
    if not path:
        return
    record = {
        "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "provider": provider,
        "status": status,
        "prompt_hash": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
        "prompt_characters": len(prompt),
        "latency_ms": latency_ms,
    }
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with AUDIT_LOCK, open(path, "a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, ensure_ascii=False) + "\n")
        handle.flush()
    os.chmod(path, 0o600)


def command_status(provider: str) -> dict:
    config = PROVIDERS[provider]
    available = shutil.which(config["binary"]) is not None
    with LOCK:
        process = RUNNING.get(provider)
        authenticating = bool(process and process.poll() is None)
        if process and process.poll() is not None:
            RUNNING.pop(provider, None)
    if not available:
        return {
            "id": provider,
            "label": config["label"],
            "available": False,
            "loggedIn": False,
            "authenticating": False,
            "authMethod": "unavailable",
        }
    try:
        result = subprocess.run(
            config["status"], capture_output=True, text=True, timeout=8,
            env={**os.environ, "NO_COLOR": "1"},
        )
    except (OSError, subprocess.TimeoutExpired):
        return {
            "id": provider,
            "label": config["label"],
            "available": True,
            "loggedIn": False,
            "authenticating": authenticating,
            "authMethod": "unknown",
        }
    if provider == "claude":
        try:
            payload = json.loads(result.stdout or "{}")
        except ValueError:
            payload = {}
        logged_in = result.returncode == 0 and bool(payload.get("loggedIn"))
        method = str(payload.get("authMethod") or "none")
    else:
        text = (result.stdout + result.stderr).lower()
        logged_in = result.returncode == 0 and "logged in" in text
        method = "chatgpt" if "chatgpt" in text else ("api" if "api" in text and logged_in else "none")
    return {
        "id": provider,
        "label": config["label"],
        "available": True,
        "loggedIn": logged_in,
        "authenticating": authenticating and not logged_in,
        "authMethod": method,
    }


def start_login(provider: str) -> dict:
    status = command_status(provider)
    if not status["available"]:
        return {"success": False, "reason": "cli_not_installed", "provider": status}
    if status["loggedIn"]:
        return {"success": True, "reason": "already_logged_in", "provider": status}
    with LOCK:
        current = RUNNING.get(provider)
        if current and current.poll() is None:
            return {"success": True, "reason": "login_in_progress", "provider": {**status, "authenticating": True}}
        process = subprocess.Popen(
            PROVIDERS[provider]["login"],
            stdin=subprocess.DEVNULL,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
            env={**os.environ, "NO_COLOR": "1"},
        )
        RUNNING[provider] = process
    return {"success": True, "reason": "browser_login_started", "provider": command_status(provider)}


def run_account_command(provider: str, prompt: str, directory: str):
    if provider == "codex":
        output_path = os.path.join(directory, "last-message.txt")
        command = [
            "codex", "exec", "--ephemeral", "--ignore-user-config", "--ignore-rules",
            "--skip-git-repo-check", "--sandbox", "read-only", "--color", "never",
            "--cd", directory, "--output-last-message", output_path, "-",
        ]
        result = subprocess.run(
            command, input=prompt, capture_output=True, text=True, timeout=300,
            env={**os.environ, "NO_COLOR": "1"},
        )
        text = ""
        if result.returncode == 0 and os.path.isfile(output_path):
            with open(output_path, encoding="utf-8") as handle:
                text = handle.read().strip()
        return result, text, {}, "codex-account-default"

    command = [
        "claude", "--print", "--output-format", "json", "--permission-mode", "plan",
        "--tools", "", "--safe-mode", "--no-session-persistence",
    ]
    result = subprocess.run(
        command, input=prompt, capture_output=True, text=True, timeout=300, cwd=directory,
        env={**os.environ, "NO_COLOR": "1"},
    )
    try:
        payload = json.loads(result.stdout or "{}")
    except ValueError:
        payload = {}
    text = str(payload.get("result") or "").strip()
    raw_usage = payload.get("usage") if isinstance(payload.get("usage"), dict) else {}
    usage = {
        "input_tokens": int(raw_usage.get("input_tokens") or 0),
        "output_tokens": int(raw_usage.get("output_tokens") or 0),
    }
    return result, text, usage, str(payload.get("model") or "claude-account-default")


def generate_with_account(provider: str, prompt: str, audit_path: str) -> dict:
    started = time.monotonic()
    status = command_status(provider)
    if not status["available"] or not status["loggedIn"]:
        audit_model_call(audit_path, provider, "provider_not_logged_in", prompt, 0)
        return {"success": False, "reason": "provider_not_logged_in"}
    if not prompt or len(prompt) > 24000:
        audit_model_call(audit_path, provider, "prompt_length_invalid", prompt, 0)
        return {"success": False, "reason": "prompt_length_invalid"}
    if not model_call_allowed(provider):
        audit_model_call(audit_path, provider, "rate_limited", prompt, 0)
        return {"success": False, "reason": "account_model_rate_limited"}
    gate = MODEL_GATES[provider]
    if not gate.acquire(blocking=False):
        audit_model_call(audit_path, provider, "busy", prompt, 0)
        return {"success": False, "reason": "account_model_busy"}
    try:
        with tempfile.TemporaryDirectory(prefix="casan-account-model-") as directory:
            result, text, usage, model = run_account_command(provider, prompt, directory)
    except (OSError, subprocess.TimeoutExpired):
        audit_model_call(audit_path, provider, "unreachable", prompt, int((time.monotonic() - started) * 1000))
        return {"success": False, "reason": "account_model_unreachable"}
    finally:
        gate.release()
    if result.returncode != 0 or not text:
        audit_model_call(audit_path, provider, "failed", prompt, int((time.monotonic() - started) * 1000))
        return {"success": False, "reason": "account_model_failed"}
    audit_model_call(audit_path, provider, "success", prompt, int((time.monotonic() - started) * 1000))
    return {
        "success": True,
        "provider": provider,
        "model": model,
        "text": text,
        "usage": usage,
        "latency_ms": int((time.monotonic() - started) * 1000),
    }


class BridgeHandler(BaseHTTPRequestHandler):
    server_version = "CASANAuthBridge/1.0"

    def log_message(self, _format: str, *_args) -> None:
        return

    def send_json(self, status: int, payload: dict) -> None:
        encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        self.send_header("Cache-Control", "no-store")
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("X-Frame-Options", "DENY")
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header("Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'")
        self.end_headers()
        self.wfile.write(encoded)

    def authorized(self) -> bool:
        host = self.headers.get("Host", "").split(":", 1)[0].lower()
        if host not in {"127.0.0.1", "localhost", "host.docker.internal"}:
            return False
        expected = self.server.bridge_token  # type: ignore[attr-defined]
        supplied = self.headers.get("X-CASAN-Bridge-Token", "")
        return bool(expected) and hmac.compare_digest(expected, supplied)

    def read_json(self) -> dict:
        if self.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json":
            return {}
        try:
            length = int(self.headers.get("Content-Length", "0"))
        except ValueError:
            return {}
        if length < 1 or length > 100000:
            return {}
        try:
            payload = json.loads(self.rfile.read(length).decode("utf-8"))
        except (UnicodeDecodeError, ValueError):
            return {}
        return payload if isinstance(payload, dict) else {}

    def do_GET(self) -> None:
        if self.path == "/healthz":
            self.send_json(200, {"status": "ok"})
            return
        if not self.authorized():
            self.send_json(401, {"success": False, "reason": "unauthorized"})
            return
        if self.path == "/v1/auth/providers":
            self.send_json(200, {"success": True, "providers": [command_status(provider) for provider in PROVIDERS]})
            return
        self.send_json(404, {"success": False, "reason": "not_found"})

    def do_POST(self) -> None:
        if not self.authorized():
            self.send_json(401, {"success": False, "reason": "unauthorized"})
            return
        parts = [part for part in self.path.split("/") if part]
        if len(parts) == 4 and parts[:2] == ["v1", "auth"] and parts[3] == "login" and parts[2] in PROVIDERS:
            result = start_login(parts[2])
            self.send_json(202 if result["success"] else 503, result)
            return
        if len(parts) == 4 and parts[:2] == ["v1", "models"] and parts[3] == "generate" and parts[2] in PROVIDERS:
            body = self.read_json()
            result = generate_with_account(parts[2], str(body.get("prompt") or ""), self.server.audit_log)  # type: ignore[attr-defined]
            self.send_json(200 if result["success"] else 503, result)
            return
        self.send_json(404, {"success": False, "reason": "not_found"})


def read_token(path: str) -> str:
    try:
        with open(path, encoding="utf-8") as handle:
            token = handle.read().strip()
    except OSError:
        raise SystemExit("AUTH_BRIDGE_TOKEN_MISSING")
    if len(token) < 32:
        raise SystemExit("AUTH_BRIDGE_TOKEN_WEAK")
    return token


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--bind", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=20130)
    parser.add_argument("--token-file", required=True)
    parser.add_argument("--audit-log", default="")
    args = parser.parse_args()
    server = ThreadingHTTPServer((args.bind, args.port), BridgeHandler)
    server.bridge_token = read_token(args.token_file)  # type: ignore[attr-defined]
    server.audit_log = os.path.abspath(args.audit_log) if args.audit_log else ""  # type: ignore[attr-defined]
    server.daemon_threads = True
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())