#!/usr/bin/env python3 """Translate an internal relay key into the owner's Claude OAuth credential.""" from __future__ import annotations import hmac import json import os from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Final import httpx from routing_catalog import resolve_route HOST: Final = os.environ.get("HERMES_CLAUDE_BROKER_HOST", "0.0.0.0") PORT: Final = int(os.environ.get("HERMES_CLAUDE_BROKER_PORT", "9006")) UPSTREAM: Final = os.environ.get( "HERMES_CLAUDE_BROKER_UPSTREAM", "https://api.anthropic.com" ).rstrip("/") MAX_BODY_BYTES: Final = int( os.environ.get("HERMES_CLAUDE_BROKER_MAX_BODY", str(64 << 20)) ) READ_TIMEOUT_SECONDS: Final = float( os.environ.get("HERMES_CLAUDE_BROKER_READ_TIMEOUT", "900") ) ALLOWED_PATHS: Final = { "/v1/messages", "/v1/messages/count_tokens", "/v1/models", } REQUIRED_BETAS: Final = ( "interleaved-thinking-2025-05-14", "fine-grained-tool-streaming-2025-05-14", "claude-code-20250219", "oauth-2025-04-20", ) ROUTED_MODEL_PREFIX: Final = "route/claude/" def _translate_model(body: bytes) -> bytes: """Translate a Switchyard effort-qualified target into a Claude model.""" if not body: return body try: payload = json.loads(body) except (TypeError, ValueError, json.JSONDecodeError): return body if not isinstance(payload, dict): return body model = payload.get("model") if isinstance(model, str) and model.startswith(ROUTED_MODEL_PREFIX): payload["model"] = resolve_route(model) return json.dumps(payload, separators=(",", ":")).encode("utf-8") return body def _read_secret(env_name: str, file_env_name: str) -> str: """Read a secret from an environment value or a mounted file.""" value = os.environ.get(env_name, "").strip() if value: return value path = os.environ.get(file_env_name, "").strip() if not path: return "" try: return Path(path).read_text(encoding="utf-8").strip() except OSError: return "" def _relay_key() -> str: """Return the shared internal key without caching rotated file contents.""" return _read_secret( "HERMES_CLAUDE_BROKER_KEY", "HERMES_CLAUDE_BROKER_KEY_FILE" ) def _oauth_token() -> str: """Return the current owner OAuth token or fail closed.""" token = _read_secret( "CLAUDE_CODE_OAUTH_TOKEN", "HERMES_CLAUDE_OAUTH_TOKEN_FILE" ) if not token: raise RuntimeError("owner Claude authentication is unavailable") return token def _authorized(authorization: str | None, api_key: str | None) -> bool: """Accept Switchyard's x-api-key or an internal Bearer relay key.""" expected = _relay_key() if not expected: return False candidates = [api_key or ""] if authorization and authorization.startswith("Bearer "): candidates.append(authorization[7:].strip()) return any(candidate and hmac.compare_digest(candidate, expected) for candidate in candidates) def _merge_betas(incoming: str | None) -> str: """Preserve requested Anthropic betas while adding Claude Code OAuth betas.""" values: list[str] = [] for value in (*((incoming or "").split(",")), *REQUIRED_BETAS): value = value.strip() if value and value not in values: values.append(value) return ",".join(values) class Handler(BaseHTTPRequestHandler): """Stream Anthropic responses while keeping the OAuth token server-side.""" server_version = "HermesClaudeOAuthBroker/1" def log_message(self, format: str, *args: object) -> None: """Avoid logging paths or headers that could contain sensitive metadata.""" return def _json(self, status: int, value: dict[str, object]) -> None: body = json.dumps(value, separators=(",", ":")).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _check_auth(self) -> bool: if _authorized( self.headers.get("Authorization"), self.headers.get("x-api-key") ): return True self._json(401, {"error": {"type": "authentication_error", "message": "unauthorized"}}) return False def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API if self.path == "/health": try: _oauth_token() except RuntimeError as exc: self._json(503, {"ok": False, "error": str(exc)}) return self._json(200, {"ok": True, "provider": "anthropic-oauth"}) return if self.path not in ALLOWED_PATHS: self._json(404, {"error": {"type": "not_found", "message": "not found"}}) return if not self._check_auth(): return self._proxy(b"") def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API if self.path not in ALLOWED_PATHS: self._json(404, {"error": {"type": "not_found", "message": "not found"}}) return if not self._check_auth(): return try: length = int(self.headers.get("Content-Length", "0")) except ValueError: length = -1 if length < 0 or length > MAX_BODY_BYTES: self._json(413, {"error": {"type": "request_too_large", "message": "request too large"}}) return self._proxy(_translate_model(self.rfile.read(length))) def _proxy(self, body: bytes) -> None: """Forward one bounded request and stream its response unchanged.""" response_started = False try: token = _oauth_token() headers = { "Accept": self.headers.get("Accept", "application/json"), "Authorization": f"Bearer {token}", "anthropic-version": self.headers.get( "anthropic-version", "2023-06-01" ), "anthropic-beta": _merge_betas(self.headers.get("anthropic-beta")), "Content-Type": self.headers.get("Content-Type", "application/json"), "User-Agent": "claude-code/2.1.226 (external, cli)", "x-app": "cli", } timeout = httpx.Timeout(30.0, read=READ_TIMEOUT_SECONDS) with httpx.Client(timeout=timeout) as client: with client.stream( self.command, f"{UPSTREAM}{self.path}", headers=headers, content=body or None, ) as response: self.send_response(response.status_code) for name, value in response.headers.items(): if name.lower() in { "content-type", "cache-control", "request-id", "retry-after", "anthropic-ratelimit-requests-limit", "anthropic-ratelimit-requests-remaining", "anthropic-ratelimit-requests-reset", "anthropic-ratelimit-tokens-limit", "anthropic-ratelimit-tokens-remaining", "anthropic-ratelimit-tokens-reset", }: self.send_header(name, value) self.send_header("Connection", "close") self.end_headers() response_started = True for chunk in response.iter_bytes(): if chunk: self.wfile.write(chunk) self.wfile.flush() except (RuntimeError, httpx.HTTPError, OSError) as exc: # Once streaming headers have crossed the wire, an upstream failure # can only terminate the stream. Sending a second HTTP response # would corrupt the Anthropic event stream seen by Switchyard. if not response_started and not self.wfile.closed: self._json( 503, {"error": {"type": "provider_unavailable", "message": str(exc)}}, ) finally: self.close_connection = True class Server(ThreadingHTTPServer): """Threaded server whose request workers do not block shutdown.""" daemon_threads = True if __name__ == "__main__": Server((HOST, PORT), Handler).serve_forever()