feat(track-a): multi-provider passthrough routes + 4 security fixes

changelog.d/track-a-passthrough.md

@@ -0,0 +1,14 @@
+# Track A: Multi-provider Passthrough Routes
+
+Adds `/openai/{path}`, `/gemini/{path}`, and `/anthropic/{path}` passthrough routes to the gateway.
+
+## Security fixes
+- **Open proxy closed**: `/openai` and `/gemini` routes now require strict `CLIENT_API_KEY` match (regardless of global `AUTH_MODE`)
+- **Body size limit**: Enforces `MAX_REQUEST_PAYLOAD_BYTES` on passthrough requests
+- **Lifespan-managed clients**: httpx clients are now created/closed via FastAPI lifespan (no resource leaks)
+- **Hop-by-hop header stripping**: Response headers `transfer-encoding`, `set-cookie`, `server`, etc. are stripped before forwarding to clients
+
+## Database
+- Migration 019: adds `agent` column to `request_logs` table
+
+**Depends on PR-A** (httpx-sse dependency).

migrations/postgres/019_add_agent_to_request_logs.sql

@@ -0,0 +1,4 @@
+-- Migration 019: Add agent column to request_logs
+-- Track A bridge: captures x-luthien-agent header from opencode-luthien plugin
+-- Indexing to be reviewed in Track B based on usage patterns
+ALTER TABLE request_logs ADD COLUMN IF NOT EXISTS agent TEXT NULL;

migrations/sqlite/019_add_agent_to_request_logs.sql

@@ -0,0 +1,4 @@
+-- Migration 019: Add agent column to request_logs
+-- Track A bridge: captures x-luthien-agent header from opencode-luthien plugin
+-- Indexing to be reviewed in Track B based on usage patterns
+ALTER TABLE request_logs ADD COLUMN agent TEXT;

pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
     "anthropic>=0.84.0",
     "aiohttp>=3.9.0",
     "sentry-sdk[fastapi]>=2.54.0",
+    "httpx-sse>=0.4",
 ]
 
 [tool.hatch.version]

scripts/run_e2e.sh

@@ -305,12 +305,12 @@ run_mock() {
 import json, sys
 try:
     d = json.load(open('$config_json'))
-    print(d['gateway_url'], d['mock_port'], d['api_key'], d['admin_api_key'])
+    print(d['gateway_url'], d['mock_port'], d.get('mock_openai_port', 18889), d.get('mock_gemini_port', 18890), d['api_key'], d['admin_api_key'])
 except (json.JSONDecodeError, KeyError) as e:
     print(f'Invalid config JSON: {e}', file=sys.stderr)
     sys.exit(1)
 ")" || { fail "Failed to parse gateway config"; rm -f "$config_json"; return 1; }
-    read -r gw_url mock_port gw_api_key gw_admin_key <<< "$config_vals"
+    read -r gw_url mock_port mock_openai_port mock_gemini_port gw_api_key gw_admin_key <<< "$config_vals"
     rm -f "$config_json"
 
     ok "Gateway ready at $gw_url (mock on port $mock_port)"
@@ -320,6 +320,8 @@ except (json.JSONDecodeError, KeyError) as e:
     export E2E_ADMIN_API_KEY="$gw_admin_key"
     export MOCK_ANTHROPIC_PORT="$mock_port"
     export MOCK_ANTHROPIC_HOST="localhost"
+    export MOCK_OPENAI_PORT="$mock_openai_port"
+    export MOCK_GEMINI_PORT="$mock_gemini_port"
     export ENABLE_REQUEST_LOGGING="true"
 
     info "Running tests..."

scripts/start_mock_gateway.py

@@ -44,6 +44,9 @@ def main():
     # from the environment to bind to the same port.
     mock_port = int(os.getenv("MOCK_ANTHROPIC_PORT", "0")) or _free_port()
 
+    openai_mock_port = int(os.getenv("MOCK_OPENAI_PORT", "0")) or _free_port()
+    gemini_mock_port = int(os.getenv("MOCK_GEMINI_PORT", "0")) or _free_port()
+
     # Create SQLite gateway
     gateway_port = _free_port()
     tmp_dir = tempfile.mkdtemp(prefix="luthien_mock_e2e_")
@@ -74,6 +77,8 @@ def main():
 
     os.environ["ANTHROPIC_BASE_URL"] = f"http://localhost:{mock_port}"
     os.environ["ANTHROPIC_API_KEY"] = "mock-key"
+    os.environ["OPENAI_BASE_URL"] = f"http://localhost:{openai_mock_port}"
+    os.environ["GEMINI_BASE_URL"] = f"http://localhost:{gemini_mock_port}"
 
     app = create_app(
         api_key=api_key,
@@ -107,6 +112,8 @@ def main():
         "api_key": api_key,
         "admin_api_key": admin_api_key,
         "mock_port": mock_port,
+        "mock_openai_port": openai_mock_port,
+        "mock_gemini_port": gemini_mock_port,
     }
     print(json.dumps(info))
     sys.stdout.flush()

src/luthien_proxy/main.py

@@ -11,6 +11,7 @@
 from collections.abc import MutableMapping
 from contextlib import asynccontextmanager
 
+import httpx
 import litellm
 import uvicorn
 from fastapi import FastAPI, Request
@@ -41,6 +42,7 @@
 )
 from luthien_proxy.observability.redis_event_publisher import RedisEventPublisher
 from luthien_proxy.observability.sentry import init_sentry
+from luthien_proxy.passthrough_routes import router as passthrough_router
 from luthien_proxy.pipeline.upstream_headers import validate_upstream_headers_at_startup
 from luthien_proxy.policy_manager import PolicyManager
 from luthien_proxy.rate_limit import TokenBucketRateLimiter
@@ -385,17 +387,35 @@ async def lifespan(app: FastAPI):
         app.state.dependencies = _dependencies
         logger.info("Dependencies container initialized")
 
+        app.state.passthrough_streaming_client = httpx.AsyncClient(
+            timeout=httpx.Timeout(connect=10.0, read=300.0, write=10.0, pool=30.0)
+        )
+        app.state.passthrough_buffered_client = httpx.AsyncClient(timeout=120.0)
+        # Two separate clients: streaming needs a long read timeout (300s) for
+        # token-by-token SSE; buffered uses 120s to accommodate long non-streaming
+        # generations (extended thinking, large max_tokens) without 502ing.
+        logger.info("Passthrough httpx clients created")
+        logger.info(
+            "/anthropic/* passthrough route is active. Requests to /anthropic/v1/... "
+            "bypass the policy chain (no judges, no transformations). "
+            "This is a temporary Track A bridge — see Track B (#563-569)."
+        )
+
         yield
 
         # Shutdown
-        # Webhook sender goes first: stop() drains in-flight tasks within the
-        # configured window, then cancels survivors and aclose()s the httpx
-        # client. After this returns, _stopped=True silently no-ops any
-        # in-flight request that reaches fire_and_forget — this is the
-        # at-most-once semantics we documented. If you reorder this so
-        # webhook.stop() runs after anthropic_client_cache.close_all() or
-        # before request handling has fully drained, fire_and_forget calls
-        # could land against an already-closed httpx client.
+        # Passthrough httpx clients are independent of the webhook sender and
+        # can be closed first.
+        await app.state.passthrough_streaming_client.aclose()
+        await app.state.passthrough_buffered_client.aclose()
+        # Webhook sender: stop() drains in-flight tasks within the configured
+        # window, then cancels survivors and aclose()s the httpx client. After
+        # this returns, _stopped=True silently no-ops any in-flight request
+        # that reaches fire_and_forget — this is the at-most-once semantics we
+        # documented. If you reorder this so webhook.stop() runs after
+        # anthropic_client_cache.close_all() or before request handling has
+        # fully drained, fire_and_forget calls could land against an
+        # already-closed httpx client.
         await _webhook_sender.stop()
         if _purger is not None:
             await _purger.stop()
@@ -450,6 +470,7 @@ async def dispatch(self, request: Request, call_next):
     app.include_router(history_routes.router)  # /history/* (conversation history UI)
     app.include_router(history_routes.api_router)  # /api/history/* (conversation history API)
     app.include_router(request_log_router)  # /request-logs/* (HTTP-level logging)
+    app.include_router(passthrough_router)  # /openai/*, /gemini/*, /anthropic/* (Track A bridge)
 
     # Simple utility endpoints
     @app.get("/health")

src/luthien_proxy/passthrough_auth.py

@@ -0,0 +1,125 @@
+"""Passthrough auth dependency for /openai/, /gemini/, /anthropic/ routes.
+
+Validates bearer tokens without building Anthropic-specific Credential objects.
+The existing Anthropic auth chain (get_request_credential, verify_token) is
+untouched — this is a parallel, simpler dep for passthrough routes only.
+"""
+
+import secrets
+
+from fastapi import Depends, HTTPException, Request, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from luthien_proxy.credential_manager import AuthMode, CredentialManager
+from luthien_proxy.dependencies import get_api_key, get_credential_manager
+
+_bearer = HTTPBearer(auto_error=False)
+
+
+def _extract_token(request: Request, credentials: HTTPAuthorizationCredentials | None) -> str | None:
+    """Extract auth token from Bearer header or Anthropic SDK-style API key headers."""
+    if credentials:
+        return credentials.credentials
+    for header in ("x-api-key", "x-anthropic-api-key"):
+        val = request.headers.get(header)
+        if val:
+            return val
+    return None
+
+
+async def verify_passthrough_token(
+    request: Request,
+    credentials: HTTPAuthorizationCredentials | None = Depends(_bearer),
+    api_key: str | None = Depends(get_api_key),
+    credential_manager: CredentialManager | None = Depends(get_credential_manager),
+) -> str:
+    """Validate bearer token for passthrough routes.
+
+    Returns the raw token string on success.
+    Raises 401 on invalid/missing token (when required by auth mode).
+
+    Auth mode semantics:
+    - PASSTHROUGH: any token accepted (client's own key forwarded upstream)
+    - CLIENT_KEY: only the configured CLIENT_API_KEY is accepted
+    - BOTH: CLIENT_API_KEY accepted, or any token (passthrough path)
+
+    Accepts Authorization: Bearer, x-api-key, or x-anthropic-api-key headers
+    so Anthropic SDK clients (which default to x-api-key) work without changes.
+    """
+    token = _extract_token(request, credentials)
+
+    # Determine auth mode
+    if credential_manager is None:
+        auth_mode = AuthMode.CLIENT_KEY
+    else:
+        auth_mode = credential_manager.config.auth_mode
+
+    if auth_mode == AuthMode.PASSTHROUGH:
+        # Any token (or no token) is accepted; client's own key forwarded upstream
+        return token or ""
+
+    # CLIENT_KEY or BOTH mode: validate against configured CLIENT_API_KEY
+    if not token:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing bearer token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    if api_key and secrets.compare_digest(token, api_key):
+        return token
+
+    if auth_mode == AuthMode.BOTH:
+        # In BOTH mode, also accept any token (passthrough path)
+        return token
+
+    # CLIENT_KEY mode: token did not match
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid bearer token",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+
+async def verify_strict_client_key(
+    credentials: HTTPAuthorizationCredentials | None = Depends(_bearer),
+    api_key: str | None = Depends(get_api_key),
+) -> str:
+    """Validate bearer token for /openai and /gemini passthrough routes.
+
+    Unlike verify_passthrough_token (which has a PASSTHROUGH/BOTH mode that accepts any
+    token), this function ALWAYS requires an exact match against CLIENT_API_KEY.
+
+    Threat model: /openai and /gemini inject server-side API keys (OPENAI_API_KEY,
+    GOOGLE_API_KEY) into every outbound request. An unauthenticated or loosely-authenticated
+    caller could burn the operator's API credits on third-party providers. Strict
+    CLIENT_API_KEY enforcement ensures only trusted clients can trigger these calls,
+    regardless of the gateway's global AUTH_MODE setting.
+
+    Returns the token on success. Raises HTTP 401 if CLIENT_API_KEY is unset or if
+    the supplied token does not match via timing-safe comparison.
+    """
+    configured_key = api_key
+    if not configured_key:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="CLIENT_API_KEY is not configured — /openai and /gemini passthrough is disabled",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    token = credentials.credentials if credentials else None
+    if not token:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing bearer token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    if secrets.compare_digest(token, configured_key):
+        return token
+
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid bearer token",
+        headers={"WWW-Authenticate": "Bearer"},
+    )