#!/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 hmac import json import os import shutil import subprocess import tempfile import threading import time 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_LOCKS = {provider: threading.Lock() for provider in PROVIDERS} 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 generate_with_account(provider: str, prompt: str) -> dict: status = command_status(provider) if not status["available"] or not status["loggedIn"]: return {"success": False, "reason": "provider_not_logged_in"} if not prompt or len(prompt) > 24000: return {"success": False, "reason": "prompt_length_invalid"} started = time.monotonic() with MODEL_LOCKS[provider], tempfile.TemporaryDirectory(prefix="casan-account-model-") as directory: try: 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() usage = {} model = "codex-account-default" else: 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), } model = str(payload.get("model") or "claude-account-default") except (OSError, subprocess.TimeoutExpired): return {"success": False, "reason": "account_model_unreachable"} if result.returncode != 0 or not text: return {"success": False, "reason": "account_model_failed"} 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.end_headers() self.wfile.write(encoded) def authorized(self) -> bool: 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: 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.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) args = parser.parse_args() server = ThreadingHTTPServer((args.bind, args.port), BridgeHandler) server.bridge_token = read_token(args.token_file) # type: ignore[attr-defined] try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())