diff --git a/changelog.d/track-a-passthrough.md b/changelog.d/track-a-passthrough.md new file mode 100644 index 000000000..b9ecd3e99 --- /dev/null +++ b/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). diff --git a/migrations/postgres/019_add_agent_to_request_logs.sql b/migrations/postgres/019_add_agent_to_request_logs.sql new file mode 100644 index 000000000..f3c961cd6 --- /dev/null +++ b/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; diff --git a/migrations/sqlite/019_add_agent_to_request_logs.sql b/migrations/sqlite/019_add_agent_to_request_logs.sql new file mode 100644 index 000000000..fec758f2b --- /dev/null +++ b/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; diff --git a/pyproject.toml b/pyproject.toml index dbaa8830f..24a7b08e9 100644 --- a/pyproject.toml +++ b/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] diff --git a/scripts/run_e2e.sh b/scripts/run_e2e.sh index 3b7dbb719..ac9d21b02 100755 --- a/scripts/run_e2e.sh +++ b/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..." diff --git a/scripts/start_mock_gateway.py b/scripts/start_mock_gateway.py index 96c0b652f..080778d89 100644 --- a/scripts/start_mock_gateway.py +++ b/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() diff --git a/src/luthien_proxy/main.py b/src/luthien_proxy/main.py index 152a6501c..b04a6f4fe 100644 --- a/src/luthien_proxy/main.py +++ b/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") diff --git a/src/luthien_proxy/passthrough_auth.py b/src/luthien_proxy/passthrough_auth.py new file mode 100644 index 000000000..b153e3e7f --- /dev/null +++ b/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"}, + ) diff --git a/src/luthien_proxy/passthrough_routes.py b/src/luthien_proxy/passthrough_routes.py new file mode 100644 index 000000000..d4616f840 --- /dev/null +++ b/src/luthien_proxy/passthrough_routes.py @@ -0,0 +1,333 @@ +"""Passthrough routes for /openai/*, /gemini/*, /anthropic/* prefixes. + +Bridges OpenAI, Gemini, and Anthropic traffic through the Luthien gateway. +Injects server-side API keys, strips internal x-luthien-* headers from outbound, +and streams responses for streaming endpoints. +""" + +from __future__ import annotations + +import json +import logging +import os +import uuid +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +import httpx +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import Response, StreamingResponse + +from luthien_proxy.passthrough_auth import verify_passthrough_token, verify_strict_client_key +from luthien_proxy.request_log.recorder import create_recorder +from luthien_proxy.utils.constants import MAX_REQUEST_PAYLOAD_BYTES + +router = APIRouter() +logger = logging.getLogger(__name__) + +UPSTREAM_BASES = { + "openai": "https://api.openai.com", + "gemini": "https://generativelanguage.googleapis.com", + "anthropic": "https://api.anthropic.com", +} + + +def _upstream_base(provider: str) -> str: + env_overrides: dict[str, str | None] = { + "openai": os.environ.get("OPENAI_BASE_URL"), + "gemini": os.environ.get("GEMINI_BASE_URL"), + "anthropic": os.environ.get("ANTHROPIC_BASE_URL"), + } + return env_overrides.get(provider) or UPSTREAM_BASES[provider] + + +# Headers stripped from inbound before forwarding (httpx sets these itself) +_STRIP_INBOUND = frozenset({"host", "content-length"}) + +# Auth headers stripped from all outbound — re-injected per provider below +_STRIP_AUTH = frozenset({"authorization", "x-api-key", "x-anthropic-api-key", "x-goog-api-key"}) + +# Headers stripped from upstream responses that httpx has already handled +# (httpx auto-decompresses, so forwarding content-encoding would mismatch the body) +_STRIP_RESPONSE = frozenset({"content-encoding", "content-length"}) + +HOP_BY_HOP_HEADERS = frozenset( + { + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailer", + "transfer-encoding", + "upgrade", + } +) +DANGEROUS_RESPONSE_HEADERS = frozenset({"set-cookie", "server", "x-powered-by"}) + +# Hop-by-hop and connection-scoped headers must not be forwarded upstream +# (RFC 7230 §6.1). Also strip cookie — none of the three providers consume it +# and forwarding it could leak browser session data. +_STRIP_INBOUND_HOP_BY_HOP = HOP_BY_HOP_HEADERS | frozenset({"cookie"}) + + +def get_streaming_client(request: Request) -> httpx.AsyncClient: + return request.app.state.passthrough_streaming_client + + +def get_buffered_client(request: Request) -> httpx.AsyncClient: + return request.app.state.passthrough_buffered_client + + +def _is_streaming(path: str, body: bytes) -> bool: + if ":streamGenerateContent" in path: + return True + try: + data = json.loads(body) + return data.get("stream") is True + except (json.JSONDecodeError, AttributeError, ValueError): + return False + + +def _build_outbound_headers(request: Request, provider: str) -> dict[str, str]: + """Build headers for the upstream request. + + Strips internal x-luthien-* headers and replaces auth headers with + server-side credentials for OpenAI/Gemini. Anthropic alias forwards + the client's auth as-is. + """ + headers: dict[str, str] = {} + for k, v in request.headers.items(): + k_lower = k.lower() + if k_lower in _STRIP_INBOUND: + continue + if k_lower in _STRIP_AUTH: + continue + if k_lower in _STRIP_INBOUND_HOP_BY_HOP: + continue + if k_lower.startswith("x-luthien-"): + continue + headers[k_lower] = v + + if provider == "openai": + key = os.environ.get("OPENAI_API_KEY", "") + if not key: + raise HTTPException(status_code=503, detail="OpenAI API key not configured on server") + headers["authorization"] = f"Bearer {key}" + elif provider == "gemini": + key = os.environ.get("GOOGLE_API_KEY", "") + if not key: + raise HTTPException(status_code=503, detail="Google API key not configured on server") + headers["x-goog-api-key"] = key + elif provider == "anthropic": + # Forward the client's own auth — this is an alias to the existing /v1/ + for auth_header in ("authorization", "x-api-key", "x-anthropic-api-key"): + if (val := request.headers.get(auth_header)) is not None: + headers[auth_header] = val + + if "user-agent" not in headers: + headers["user-agent"] = "luthien-passthrough/0.1" + + return headers + + +def _safe_response_headers(response_headers: httpx.Headers) -> dict[str, str]: + return { + k: v + for k, v in response_headers.items() + if k.lower() not in HOP_BY_HOP_HEADERS + and k.lower() not in DANGEROUS_RESPONSE_HEADERS + and k.lower() not in _STRIP_RESPONSE + } + + +def _sanitize_url(url: str, provider: str) -> str: + """Strip auth query params from the URL before logging. + + Gemini accepts ?key= as an alternative to the x-goog-api-key header. + We strip it from outbound requests, but the inbound URL (from the client) + may still carry it — redact before storing in request_logs. + """ + if provider != "gemini" or "key=" not in url: + return url + + parts = urlsplit(url) + params = [(k, v) for k, v in parse_qsl(parts.query) if k.lower() != "key"] + return urlunsplit(parts._replace(query=urlencode(params))) + + +async def _handle_passthrough(request: Request, provider: str, path: str) -> Response: + content_length = request.headers.get("content-length") + if content_length: + try: + if int(content_length) > MAX_REQUEST_PAYLOAD_BYTES: + raise HTTPException(status_code=413, detail="Request payload too large") + except ValueError as exc: + raise HTTPException(status_code=400, detail="Invalid content-length header") from exc + + body = await request.body() + + if len(body) > MAX_REQUEST_PAYLOAD_BYTES: + raise HTTPException(status_code=413, detail="Request payload too large") + upstream_url = f"{_upstream_base(provider)}/{path}" + if request.url.query: + query = request.url.query + if provider == "gemini": + # Strip ?key= to prevent clients from bypassing server-injected auth; + # the server key is injected via x-goog-api-key header instead. + params = [(k, v) for k, v in parse_qsl(query) if k.lower() != "key"] + query = urlencode(params) + if query: + upstream_url = f"{upstream_url}?{query}" + + headers = _build_outbound_headers(request, provider) + + deps = getattr(request.app.state, "dependencies", None) + recorder = create_recorder( + db_pool=deps.db_pool if deps is not None else None, + transaction_id=str(uuid.uuid4()), + enabled=deps.enable_request_logging if deps is not None else False, + ) + streaming = _is_streaming(path, body) + recorder.record_inbound_request( + method=request.method, + url=_sanitize_url(str(request.url), provider), + headers=dict(request.headers), + body=None, + session_id=request.headers.get("x-luthien-session-id"), + agent=request.headers.get("x-luthien-agent"), + model=request.headers.get("x-luthien-model"), + endpoint=f"/{provider}/{path}", + is_streaming=streaming, + ) + + streaming_client = get_streaming_client(request) + buffered_client = get_buffered_client(request) + + if streaming: + # We must peek at the upstream response status before committing an HTTP + # status to the client. StreamingResponse locks in 200 the moment it is + # returned, so a 4xx/5xx from upstream would be silently misreported. + # Strategy: open the upstream connection, read the status + headers, then + # either return a plain Response for non-2xx or hand off to a generator + # for 2xx streaming. + try: + upstream_cm = streaming_client.stream( + request.method, + upstream_url, + headers=headers, + content=body or None, + ) + response = await upstream_cm.__aenter__() + except httpx.RequestError as exc: + logger.warning("Streaming passthrough error for %s/%s: %s", provider, path, repr(exc)) + recorder.record_inbound_response(status=502, error=repr(exc)) + recorder.flush() + raise HTTPException(status_code=502, detail="Failed to connect to upstream API") + + if response.status_code >= 300: + # Non-2xx: buffer the error body and return a plain Response so the + # client sees the real status code. + try: + error_body = await response.aread() + finally: + await upstream_cm.__aexit__(None, None, None) + recorder.record_inbound_response(status=response.status_code) + recorder.flush() + return Response( + content=error_body, + status_code=response.status_code, + headers=_safe_response_headers(response.headers), + ) + + # 2xx: stream the body. Forward the upstream Content-Type so clients + # that branch on it (e.g. Gemini JSON vs SSE) get the right value. + # Also forward other safe upstream headers (rate-limit, request-id, etc.) + # to match the behaviour of the buffered path. + upstream_content_type = response.headers.get("content-type", "text/event-stream") + safe_headers = { + k: v for k, v in _safe_response_headers(response.headers).items() if k.lower() != "content-type" + } + + async def stream_chunks(): + status = response.status_code + error: str | None = None + try: + async for chunk in response.aiter_bytes(): + yield chunk + except httpx.RequestError as exc: + logger.warning("Streaming passthrough error for %s/%s: %s", provider, path, repr(exc)) + status = 502 + error = repr(exc) + # Mid-stream failure: the client has already received 200 OK and + # a partial body. We can't change the HTTP status at this point; + # the truncated stream is the only signal available to the client. + finally: + try: + await upstream_cm.__aexit__(None, None, None) + except Exception: + logger.warning("Error closing upstream connection for %s/%s", provider, path, exc_info=True) + recorder.record_inbound_response(status=status, error=error) + recorder.flush() + + return StreamingResponse( + stream_chunks(), + status_code=response.status_code, + media_type=upstream_content_type, + headers=safe_headers, + ) + + try: + response = await buffered_client.request( + request.method, + upstream_url, + headers=headers, + content=body or None, + ) + except httpx.RequestError as exc: + logger.warning("Buffered passthrough error for %s/%s: %s", provider, path, repr(exc)) + recorder.record_inbound_response(status=502, error=repr(exc)) + recorder.flush() + raise HTTPException(status_code=502, detail="Failed to connect to upstream API") + + recorder.record_inbound_response(status=response.status_code) + recorder.flush() + + safe_headers = _safe_response_headers(response.headers) + return Response( + content=response.content, + status_code=response.status_code, + headers=safe_headers, + ) + + +@router.api_route("/openai/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"]) +async def openai_passthrough( + request: Request, + path: str, + _token: str = Depends(verify_strict_client_key), +) -> Response: + # Track A bridge passthrough — replaced by native pipeline in Track B (#563-569) + return await _handle_passthrough(request, "openai", path) + + +@router.api_route("/gemini/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"]) +async def gemini_passthrough( + request: Request, + path: str, + _token: str = Depends(verify_strict_client_key), +) -> Response: + # Track A bridge passthrough — replaced by native pipeline in Track B (#563-569) + return await _handle_passthrough(request, "gemini", path) + + +@router.api_route("/anthropic/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"]) +async def anthropic_alias_passthrough( + request: Request, + path: str, + _token: str = Depends(verify_passthrough_token), +) -> Response: + # Track A bridge passthrough — replaced by native pipeline in Track B (#563-569) + return await _handle_passthrough(request, "anthropic", path) + + +__all__ = ["router"] diff --git a/src/luthien_proxy/request_log/recorder.py b/src/luthien_proxy/request_log/recorder.py index f42fafe25..c18442b45 100644 --- a/src/luthien_proxy/request_log/recorder.py +++ b/src/luthien_proxy/request_log/recorder.py @@ -55,6 +55,8 @@ class _PendingLog: is_streaming: bool = False endpoint: str | None = None error: str | None = None + agent: str | None = None + populated: bool = False async def _insert_log_row( @@ -79,13 +81,13 @@ async def _insert_log_row( http_method, url, request_headers, request_body, response_status, response_headers, response_body, started_at, completed_at, duration_ms, - model, is_streaming, endpoint, error + model, is_streaming, endpoint, error, agent ) VALUES ( $1, $2, $3, $4, $5, $6::jsonb, $7::jsonb, $8, $9::jsonb, $10::jsonb, to_timestamp($11), to_timestamp($12), $13, - $14, $15, $16, $17 + $14, $15, $16, $17, $18 ) """, pending.transaction_id, @@ -105,6 +107,7 @@ async def _insert_log_row( pending.is_streaming, pending.endpoint, pending.error, + pending.agent, ) except Exception as exc: raise DatabaseWriteError( @@ -141,8 +144,9 @@ def record_inbound_request( method: str, url: str, headers: dict[str, str], - body: dict[str, Any], + body: dict[str, Any] | None, session_id: str | None = None, + agent: str | None = None, model: str | None = None, is_streaming: bool = False, endpoint: str | None = None, @@ -153,6 +157,7 @@ def record_inbound_request( self._inbound.request_headers = sanitize_headers(headers) self._inbound.request_body = body self._inbound.session_id = session_id + self._inbound.agent = agent self._inbound.model = model self._inbound.is_streaming = is_streaming self._inbound.endpoint = endpoint @@ -191,10 +196,12 @@ def record_outbound_request( self._outbound.url = url self._outbound.request_body = body self._outbound.session_id = self._inbound.session_id + self._outbound.agent = self._inbound.agent self._outbound.model = model self._outbound.is_streaming = is_streaming self._outbound.endpoint = endpoint self._outbound.started_at = time.time() + self._outbound.populated = True def record_outbound_response( self, @@ -239,6 +246,11 @@ async def _write_logs(self) -> None: try: async with self._db_pool.connection() as conn: for pending in (self._inbound, self._outbound): + if not pending.populated and pending.direction == "outbound": + # Outbound side was never populated — skip rather than + # inserting a fully-NULL row. Passthrough requests only + # record the inbound side; this is the expected path for them. + continue await _insert_log_row(conn, pending, self._serialize_body) except DatabaseWriteError as exc: RequestLogRecorder.dropped_writes += 1 @@ -265,8 +277,9 @@ def record_inbound_request( # noqa: D102, ARG002 method: str, url: str, headers: dict[str, str], - body: dict[str, Any], + body: dict[str, Any] | None, session_id: str | None = None, + agent: str | None = None, model: str | None = None, is_streaming: bool = False, endpoint: str | None = None, diff --git a/src/luthien_proxy/request_log/sanitize.py b/src/luthien_proxy/request_log/sanitize.py index 79bf4810a..477bd737f 100644 --- a/src/luthien_proxy/request_log/sanitize.py +++ b/src/luthien_proxy/request_log/sanitize.py @@ -10,6 +10,7 @@ "authorization", "x-api-key", "x-anthropic-api-key", + "x-goog-api-key", "proxy-authorization", "cookie", "set-cookie", diff --git a/src/luthien_proxy/utils/sqlite_migrations/019_add_agent_to_request_logs.sql b/src/luthien_proxy/utils/sqlite_migrations/019_add_agent_to_request_logs.sql new file mode 100644 index 000000000..fec758f2b --- /dev/null +++ b/src/luthien_proxy/utils/sqlite_migrations/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; diff --git a/tests/luthien_proxy/e2e_tests/conftest.py b/tests/luthien_proxy/e2e_tests/conftest.py index c25317a97..491d63bd7 100644 --- a/tests/luthien_proxy/e2e_tests/conftest.py +++ b/tests/luthien_proxy/e2e_tests/conftest.py @@ -21,6 +21,18 @@ from dotenv import load_dotenv from tests.luthien_proxy.e2e_tests.mock_anthropic.responses import text_response as _text_response from tests.luthien_proxy.e2e_tests.mock_anthropic.server import MockAnthropicServer +from tests.luthien_proxy.e2e_tests.mock_gemini.server import ( + DEFAULT_MOCK_PORT as DEFAULT_GEMINI_MOCK_PORT, +) +from tests.luthien_proxy.e2e_tests.mock_gemini.server import ( + MockGeminiServer, +) +from tests.luthien_proxy.e2e_tests.mock_openai.server import ( + DEFAULT_MOCK_PORT as DEFAULT_OPENAI_MOCK_PORT, +) +from tests.luthien_proxy.e2e_tests.mock_openai.server import ( + MockOpenAIServer, +) # === Repository Root Finding === @@ -123,22 +135,37 @@ def mock_anthropic_port(mock_anthropic: MockAnthropicServer) -> int: return mock_anthropic.port -@pytest.fixture(autouse=True) -def _reset_mock_server(request): - """Drain the mock queue and clear request history before each mock_e2e test. +@pytest.fixture(scope="session") +def mock_openai_server(): + port = int(os.getenv("MOCK_OPENAI_PORT", str(DEFAULT_OPENAI_MOCK_PORT))) + server = MockOpenAIServer(port=port) + server.start() + yield server + server.stop() - Only activates for tests marked with @pytest.mark.mock_e2e so real-API - tests don't trigger an unnecessary mock server startup. - Note: tests that use the mock_anthropic fixture directly without the mock_e2e - marker will NOT get a clean queue reset — add the marker to avoid stale state. - """ +@pytest.fixture(scope="session") +def mock_gemini_server(): + port = int(os.getenv("MOCK_GEMINI_PORT", str(DEFAULT_GEMINI_MOCK_PORT))) + server = MockGeminiServer(port=port) + server.start() + yield server + server.stop() + + +@pytest.fixture(autouse=True) +def _reset_mock_server(request): if not request.node.get_closest_marker("mock_e2e"): yield return server: MockAnthropicServer = request.getfixturevalue("mock_anthropic") server.drain_queue() server.clear_requests() + for srv_fixture in ("mock_openai_server", "mock_gemini_server"): + if srv_fixture in request.fixturenames: + srv = request.getfixturevalue(srv_fixture) + srv.drain_queue() + srv.clear_requests() yield diff --git a/tests/luthien_proxy/e2e_tests/mock_gemini/__init__.py b/tests/luthien_proxy/e2e_tests/mock_gemini/__init__.py new file mode 100644 index 000000000..cc457a775 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/mock_gemini/__init__.py @@ -0,0 +1,8 @@ +"""Mock Gemini API server for e2e testing without real API calls.""" + +from .server import DEFAULT_MOCK_PORT, MockGeminiServer + +__all__ = [ + "MockGeminiServer", + "DEFAULT_MOCK_PORT", +] diff --git a/tests/luthien_proxy/e2e_tests/mock_gemini/conftest.py b/tests/luthien_proxy/e2e_tests/mock_gemini/conftest.py new file mode 100644 index 000000000..cf6dee4fc --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/mock_gemini/conftest.py @@ -0,0 +1,10 @@ +import pytest +from tests.luthien_proxy.e2e_tests.mock_gemini.server import MockGeminiServer + + +@pytest.fixture(scope="session") +def mock_gemini_server(): + server = MockGeminiServer() + server.start() + yield server + server.stop() diff --git a/tests/luthien_proxy/e2e_tests/mock_gemini/server.py b/tests/luthien_proxy/e2e_tests/mock_gemini/server.py new file mode 100644 index 000000000..3bf2fa9a9 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/mock_gemini/server.py @@ -0,0 +1,182 @@ +"""Mock Gemini API server for e2e testing. + +Implements the subset of the Gemini API needed by luthien-proxy passthrough tests: + POST /v1beta/models/{model}:generateContent → JSON response + POST /v1beta/models/{model}:streamGenerateContent → SSE stream + +Follows the same pattern as mock_anthropic/server.py: dedicated background thread +with its own event loop, FIFO response queue, thread-safe request/header capture. + +Usage: + server = MockGeminiServer() + server.start() + + server.enqueue({"candidates": [{"content": {"parts": [{"text": "Hi"}], "role": "model"}}]}) + + server.stop() + +If no response is enqueued, a default canned response is returned. +""" + +import asyncio +import json +import logging +import queue +import threading + +from aiohttp import web + +logger = logging.getLogger(__name__) + +DEFAULT_MOCK_PORT = 18890 + +_DEFAULT_RESPONSE = { + "candidates": [ + { + "content": {"parts": [{"text": "mock response"}], "role": "model"}, + "finishReason": "STOP", + "index": 0, + } + ], + "usageMetadata": {"promptTokenCount": 10, "candidatesTokenCount": 5, "totalTokenCount": 15}, +} + + +class MockGeminiServer: + """Minimal Gemini API mock backed by aiohttp. + + Runs in a dedicated background thread with its own event loop so it remains + responsive regardless of what the pytest event loop is doing. + + Responses are consumed from a FIFO queue. Enqueue a dict response before + each test request; extra requests fall back to the default canned response. + + For streaming (:streamGenerateContent), the response dict is emitted as a + single SSE data chunk. To emit multiple chunks, enqueue a list of dicts. + """ + + def __init__(self, port: int = DEFAULT_MOCK_PORT): + self._port = port + self._queue: queue.SimpleQueue = queue.SimpleQueue() + self._thread: threading.Thread | None = None + self._loop: asyncio.AbstractEventLoop | None = None + self._runner: web.AppRunner | None = None + self._ready = threading.Event() + self._stop_event: asyncio.Event | None = None + self._received_requests: list[dict] = [] + self._received_headers: list[dict[str, str]] = [] + self._requests_lock = threading.Lock() + + @property + def port(self) -> int: + return self._port + + @property + def base_url(self) -> str: + return f"http://localhost:{self._port}" + + def enqueue(self, response: dict | list) -> None: + self._queue.put(response) + + def last_request(self) -> dict | None: + with self._requests_lock: + return self._received_requests[-1] if self._received_requests else None + + def received_requests(self) -> list[dict]: + with self._requests_lock: + return list(self._received_requests) + + def last_request_headers(self) -> dict[str, str] | None: + with self._requests_lock: + return self._received_headers[-1] if self._received_headers else None + + def received_request_headers(self) -> list[dict[str, str]]: + with self._requests_lock: + return list(self._received_headers) + + def clear_requests(self) -> None: + with self._requests_lock: + self._received_requests.clear() + self._received_headers.clear() + + def drain_queue(self) -> None: + while True: + try: + self._queue.get_nowait() + except queue.Empty: + break + + def start(self) -> None: + self._thread = threading.Thread(target=self._run_loop, daemon=True, name="mock-gemini") + self._thread.start() + self._ready.wait(timeout=10) + if not self._ready.is_set(): + raise RuntimeError("MockGeminiServer failed to start within 10 seconds") + logger.info(f"MockGeminiServer ready on port {self._port}") + + def stop(self) -> None: + if self._loop and self._stop_event: + self._loop.call_soon_threadsafe(self._stop_event.set) + if self._thread: + self._thread.join(timeout=5) + + def _run_loop(self) -> None: + self._loop = asyncio.new_event_loop() + asyncio.set_event_loop(self._loop) + try: + self._loop.run_until_complete(self._serve()) + finally: + self._loop.close() + + async def _serve(self) -> None: + self._stop_event = asyncio.Event() + app = web.Application() + app.router.add_post(r"/v1beta/models/{model}:generateContent", self._handle_generate) + app.router.add_post(r"/v1beta/models/{model}:streamGenerateContent", self._handle_stream_generate) + self._runner = web.AppRunner(app, access_log=None) + await self._runner.setup() + site = web.TCPSite(self._runner, "0.0.0.0", self._port) + await site.start() + self._ready.set() + await self._stop_event.wait() + await self._runner.cleanup() + + def _record_request(self, body: dict, headers: dict[str, str] | None = None) -> None: + with self._requests_lock: + self._received_requests.append(body) + self._received_headers.append(headers or {}) + + def _next_response(self) -> dict | list: + try: + return self._queue.get_nowait() + except queue.Empty: + return _DEFAULT_RESPONSE + + async def _handle_generate(self, request: web.Request) -> web.Response: + body = await request.json() + self._record_request(body, dict(request.headers)) + enqueued = self._next_response() + data = enqueued[0] if isinstance(enqueued, list) else enqueued + return web.Response(body=json.dumps(data), content_type="application/json") + + async def _handle_stream_generate(self, request: web.Request) -> web.StreamResponse: + body = await request.json() + self._record_request(body, dict(request.headers)) + enqueued = self._next_response() + chunks = enqueued if isinstance(enqueued, list) else [enqueued] + + response = web.StreamResponse( + headers={ + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + } + ) + await response.prepare(request) + + for chunk in chunks: + line = f"data: {json.dumps(chunk)}\n\n" + await response.write(line.encode()) + + await response.write_eof() + return response diff --git a/tests/luthien_proxy/e2e_tests/mock_gemini/test_smoke.py b/tests/luthien_proxy/e2e_tests/mock_gemini/test_smoke.py new file mode 100644 index 000000000..6da08a9f6 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/mock_gemini/test_smoke.py @@ -0,0 +1,113 @@ +import json + +import httpx + + +def test_mock_gemini_non_streaming(mock_gemini_server): + base = mock_gemini_server.base_url + mock_gemini_server.clear_requests() + + mock_gemini_server.enqueue( + { + "candidates": [ + { + "content": {"parts": [{"text": "Hello"}], "role": "model"}, + "finishReason": "STOP", + "index": 0, + } + ], + "usageMetadata": {"promptTokenCount": 10, "candidatesTokenCount": 5, "totalTokenCount": 15}, + } + ) + + response = httpx.post( + f"{base}/v1beta/models/gemini-1.5-flash:generateContent", + headers={"x-goog-api-key": "test-gemini-key"}, + json={"contents": [{"role": "user", "parts": [{"text": "hi"}]}]}, + timeout=10, + ) + + assert response.status_code == 200 + data = response.json() + assert data["candidates"][0]["content"]["parts"][0]["text"] == "Hello" + assert data["candidates"][0]["finishReason"] == "STOP" + + headers = mock_gemini_server.last_request_headers() + assert headers is not None + assert "test-gemini-key" in headers.get("x-goog-api-key", "") + + +def test_mock_gemini_streaming(mock_gemini_server): + base = mock_gemini_server.base_url + mock_gemini_server.clear_requests() + + mock_gemini_server.enqueue( + [ + { + "candidates": [ + { + "content": {"parts": [{"text": "Hello"}], "role": "model"}, + "finishReason": "STOP", + "index": 0, + } + ] + } + ] + ) + + with httpx.stream( + "POST", + f"{base}/v1beta/models/gemini-1.5-flash:streamGenerateContent", + headers={"x-goog-api-key": "test-stream-key"}, + json={"contents": [{"role": "user", "parts": [{"text": "hi"}]}]}, + timeout=10, + ) as resp: + assert resp.status_code == 200 + assert "text/event-stream" in resp.headers.get("content-type", "") + + lines = list(resp.iter_lines()) + + data_lines = [line for line in lines if line.startswith("data:")] + assert len(data_lines) >= 1 + + parsed = json.loads(data_lines[0][len("data:") :].strip()) + assert parsed["candidates"][0]["content"]["parts"][0]["text"] == "Hello" + + headers = mock_gemini_server.last_request_headers() + assert headers is not None + assert "test-stream-key" in headers.get("x-goog-api-key", "") + + +def test_mock_gemini_default_response(mock_gemini_server): + base = mock_gemini_server.base_url + mock_gemini_server.drain_queue() + mock_gemini_server.clear_requests() + + response = httpx.post( + f"{base}/v1beta/models/gemini-1.5-flash:generateContent", + headers={"x-goog-api-key": "any-key"}, + json={"contents": [{"role": "user", "parts": [{"text": "hi"}]}]}, + timeout=10, + ) + + assert response.status_code == 200 + data = response.json() + assert data["candidates"][0]["content"]["parts"][0]["text"] == "mock response" + + +def test_mock_gemini_captures_headers(mock_gemini_server): + base = mock_gemini_server.base_url + mock_gemini_server.clear_requests() + + httpx.post( + f"{base}/v1beta/models/gemini-1.5-flash:generateContent", + headers={"x-goog-api-key": "my-gemini-key", "X-Custom": "custom-value"}, + json={"contents": [{"role": "user", "parts": [{"text": "hi"}]}]}, + timeout=10, + ) + + all_headers = mock_gemini_server.received_request_headers() + assert len(all_headers) >= 1 + last = mock_gemini_server.last_request_headers() + assert "my-gemini-key" in last.get("x-goog-api-key", "") + assert last.get("X-Custom") == "custom-value" diff --git a/tests/luthien_proxy/e2e_tests/mock_openai/__init__.py b/tests/luthien_proxy/e2e_tests/mock_openai/__init__.py new file mode 100644 index 000000000..172a886c3 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/mock_openai/__init__.py @@ -0,0 +1,8 @@ +"""Mock OpenAI API server for e2e testing without real API calls.""" + +from .server import DEFAULT_MOCK_PORT, MockOpenAIServer + +__all__ = [ + "MockOpenAIServer", + "DEFAULT_MOCK_PORT", +] diff --git a/tests/luthien_proxy/e2e_tests/mock_openai/conftest.py b/tests/luthien_proxy/e2e_tests/mock_openai/conftest.py new file mode 100644 index 000000000..882bcaaeb --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/mock_openai/conftest.py @@ -0,0 +1,10 @@ +import pytest +from tests.luthien_proxy.e2e_tests.mock_openai.server import MockOpenAIServer + + +@pytest.fixture(scope="session") +def mock_openai_server(): + server = MockOpenAIServer() + server.start() + yield server + server.stop() diff --git a/tests/luthien_proxy/e2e_tests/mock_openai/server.py b/tests/luthien_proxy/e2e_tests/mock_openai/server.py new file mode 100644 index 000000000..6634dc353 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/mock_openai/server.py @@ -0,0 +1,255 @@ +"""Mock OpenAI Chat Completions server for e2e testing. + +Implements the subset of the OpenAI API needed by luthien-proxy passthrough tests: + POST /v1/chat/completions → JSON response or SSE stream (OpenAI format) + +Follows the same pattern as mock_anthropic/server.py: dedicated background thread +with its own event loop, FIFO response queue, thread-safe request/header capture. + +Usage: + server = MockOpenAIServer() + server.start() + + server.enqueue({"choices": [{"message": {"role": "assistant", "content": "Hi"}}]}) + + server.stop() + +If no response is enqueued, a default canned response is returned. +""" + +import asyncio +import json +import logging +import queue +import threading +import time + +from aiohttp import web + +logger = logging.getLogger(__name__) + +DEFAULT_MOCK_PORT = 18889 + +_DEFAULT_RESPONSE = { + "id": "chatcmpl-default", + "object": "chat.completion", + "created": 0, + "model": "gpt-4o", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "mock response"}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, +} + +_DEFAULT_STREAMING_CHUNKS = [ + { + "id": "chatcmpl-default", + "object": "chat.completion.chunk", + "created": 0, + "model": "gpt-4o", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "mock"}, "finish_reason": None}], + }, + { + "id": "chatcmpl-default", + "object": "chat.completion.chunk", + "created": 0, + "model": "gpt-4o", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + }, +] + + +class MockOpenAIServer: + """Minimal OpenAI Chat Completions mock backed by aiohttp. + + Runs in a dedicated background thread with its own event loop so it remains + responsive regardless of what the pytest event loop is doing. + + Responses are consumed from a FIFO queue. Each enqueued item may be: + - dict: for non-streaming, returned as JSON; for streaming, treated as + a single-chunk SSE stream ending with [DONE] + - list[dict]: for streaming, each dict is emitted as one SSE data line + followed by [DONE] + + The ``stream`` field in the request body determines the response format. + If the enqueued item has ``"_streaming_chunks"`` key (list), those chunks + are always used for streaming regardless of request body. + """ + + def __init__(self, port: int = DEFAULT_MOCK_PORT): + self._port = port + self._queue: queue.SimpleQueue = queue.SimpleQueue() + self._thread: threading.Thread | None = None + self._loop: asyncio.AbstractEventLoop | None = None + self._runner: web.AppRunner | None = None + self._ready = threading.Event() + self._stop_event: asyncio.Event | None = None + self._received_requests: list[dict] = [] + self._received_headers: list[dict[str, str]] = [] + self._requests_lock = threading.Lock() + + @property + def port(self) -> int: + return self._port + + @property + def base_url(self) -> str: + return f"http://localhost:{self._port}" + + def enqueue(self, response: dict | list) -> None: + self._queue.put(response) + + def last_request(self) -> dict | None: + with self._requests_lock: + return self._received_requests[-1] if self._received_requests else None + + def received_requests(self) -> list[dict]: + with self._requests_lock: + return list(self._received_requests) + + def last_request_headers(self) -> dict[str, str] | None: + with self._requests_lock: + return self._received_headers[-1] if self._received_headers else None + + def received_request_headers(self) -> list[dict[str, str]]: + with self._requests_lock: + return list(self._received_headers) + + def clear_requests(self) -> None: + with self._requests_lock: + self._received_requests.clear() + self._received_headers.clear() + + def drain_queue(self) -> None: + while True: + try: + self._queue.get_nowait() + except queue.Empty: + break + + def start(self) -> None: + self._thread = threading.Thread(target=self._run_loop, daemon=True, name="mock-openai") + self._thread.start() + self._ready.wait(timeout=10) + if not self._ready.is_set(): + raise RuntimeError("MockOpenAIServer failed to start within 10 seconds") + logger.info(f"MockOpenAIServer ready on port {self._port}") + + def stop(self) -> None: + if self._loop and self._stop_event: + self._loop.call_soon_threadsafe(self._stop_event.set) + if self._thread: + self._thread.join(timeout=5) + + def _run_loop(self) -> None: + self._loop = asyncio.new_event_loop() + asyncio.set_event_loop(self._loop) + try: + self._loop.run_until_complete(self._serve()) + finally: + self._loop.close() + + async def _serve(self) -> None: + self._stop_event = asyncio.Event() + app = web.Application() + app.router.add_post("/v1/chat/completions", self._handle_chat_completions) + self._runner = web.AppRunner(app, access_log=None) + await self._runner.setup() + site = web.TCPSite(self._runner, "0.0.0.0", self._port) + await site.start() + self._ready.set() + await self._stop_event.wait() + await self._runner.cleanup() + + def _record_request(self, body: dict, headers: dict[str, str] | None = None) -> None: + with self._requests_lock: + self._received_requests.append(body) + self._received_headers.append(headers or {}) + + def _next_response(self) -> dict | list: + try: + return self._queue.get_nowait() + except queue.Empty: + return _DEFAULT_RESPONSE + + async def _handle_chat_completions(self, request: web.Request) -> web.StreamResponse | web.Response: + body = await request.json() + self._record_request(body, dict(request.headers)) + + enqueued = self._next_response() + want_stream = body.get("stream", False) + + if want_stream: + return await self._stream_response(enqueued, request) + return self._json_response(enqueued) + + def _json_response(self, enqueued: dict | list) -> web.Response: + if isinstance(enqueued, list): + content = "" + for chunk in enqueued: + for choice in chunk.get("choices", []): + content += choice.get("delta", {}).get("content", "") + data = { + "id": "chatcmpl-test", + "object": "chat.completion", + "created": int(time.time()), + "model": "gpt-4o", + "choices": [ + {"index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": "stop"} + ], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + } + else: + data = enqueued + return web.Response(body=json.dumps(data), content_type="application/json") + + async def _stream_response(self, enqueued: dict | list, request: web.Request) -> web.StreamResponse: + if isinstance(enqueued, list): + chunks = enqueued + else: + content = "" + for choice in enqueued.get("choices", []): + msg = choice.get("message", {}) + content += msg.get("content", "") + created = enqueued.get("created", int(time.time())) + model = enqueued.get("model", "gpt-4o") + cid = enqueued.get("id", "chatcmpl-test") + chunks = [ + { + "id": cid, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [ + {"index": 0, "delta": {"role": "assistant", "content": content}, "finish_reason": None} + ], + }, + { + "id": cid, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + }, + ] + + response = web.StreamResponse( + headers={ + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + } + ) + await response.prepare(request) + + for chunk in chunks: + line = f"data: {json.dumps(chunk)}\n\n" + await response.write(line.encode()) + + await response.write(b"data: [DONE]\n\n") + await response.write_eof() + return response diff --git a/tests/luthien_proxy/e2e_tests/mock_openai/test_smoke.py b/tests/luthien_proxy/e2e_tests/mock_openai/test_smoke.py new file mode 100644 index 000000000..0ae689102 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/mock_openai/test_smoke.py @@ -0,0 +1,133 @@ +import json + +import httpx + + +def test_mock_openai_non_streaming(mock_openai_server): + base = mock_openai_server.base_url + mock_openai_server.clear_requests() + + mock_openai_server.enqueue( + { + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 1234567890, + "model": "gpt-4o", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "Hello"}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + } + ) + + response = httpx.post( + f"{base}/v1/chat/completions", + headers={"Authorization": "Bearer sk-test-key"}, + json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}, + timeout=10, + ) + + assert response.status_code == 200 + data = response.json() + assert data["object"] == "chat.completion" + assert data["choices"][0]["message"]["content"] == "Hello" + assert data["choices"][0]["finish_reason"] == "stop" + + headers = mock_openai_server.last_request_headers() + assert headers is not None + assert "sk-test-key" in headers.get("Authorization", "") + + +def test_mock_openai_streaming(mock_openai_server): + base = mock_openai_server.base_url + mock_openai_server.clear_requests() + + mock_openai_server.enqueue( + [ + { + "id": "chatcmpl-test", + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "gpt-4o", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "Hello"}, "finish_reason": None}], + }, + { + "id": "chatcmpl-test", + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "gpt-4o", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + }, + ] + ) + + with httpx.stream( + "POST", + f"{base}/v1/chat/completions", + headers={"Authorization": "Bearer sk-stream-key"}, + json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}], "stream": True}, + timeout=10, + ) as resp: + assert resp.status_code == 200 + assert "text/event-stream" in resp.headers.get("content-type", "") + + lines = list(resp.iter_lines()) + + data_lines = [line for line in lines if line.startswith("data:")] + assert any("[DONE]" in line for line in data_lines) + + content_chunks = [] + for line in data_lines: + payload = line[len("data:") :].strip() + if payload == "[DONE]": + continue + chunk = json.loads(payload) + for choice in chunk.get("choices", []): + delta = choice.get("delta", {}) + if "content" in delta: + content_chunks.append(delta["content"]) + + assert "".join(content_chunks) == "Hello" + + headers = mock_openai_server.last_request_headers() + assert headers is not None + assert "sk-stream-key" in headers.get("Authorization", "") + + +def test_mock_openai_default_response(mock_openai_server): + base = mock_openai_server.base_url + mock_openai_server.drain_queue() + mock_openai_server.clear_requests() + + response = httpx.post( + f"{base}/v1/chat/completions", + headers={"Authorization": "Bearer any-key"}, + json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}, + timeout=10, + ) + + assert response.status_code == 200 + data = response.json() + assert data["choices"][0]["message"]["content"] == "mock response" + + +def test_mock_openai_captures_headers(mock_openai_server): + base = mock_openai_server.base_url + mock_openai_server.clear_requests() + + httpx.post( + f"{base}/v1/chat/completions", + headers={"Authorization": "Bearer my-secret-key", "X-Custom": "custom-value"}, + json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}, + timeout=10, + ) + + all_headers = mock_openai_server.received_request_headers() + assert len(all_headers) >= 1 + last = mock_openai_server.last_request_headers() + assert "my-secret-key" in last.get("Authorization", "") + assert last.get("X-Custom") == "custom-value" diff --git a/tests/luthien_proxy/e2e_tests/sqlite/_boot.py b/tests/luthien_proxy/e2e_tests/sqlite/_boot.py index a7bc2ecfa..f0d5167b2 100644 --- a/tests/luthien_proxy/e2e_tests/sqlite/_boot.py +++ b/tests/luthien_proxy/e2e_tests/sqlite/_boot.py @@ -18,6 +18,7 @@ import time from collections.abc import Iterator from contextlib import ExitStack, contextmanager +from dataclasses import dataclass import uvicorn @@ -27,6 +28,12 @@ from luthien_proxy.utils.migration_check import check_migrations +@dataclass(frozen=True) +class BootedSqliteGateway: + url: str + db_path: str + + def free_port() -> int: with socket.socket() as s: s.bind(("", 0)) @@ -41,10 +48,10 @@ def boot_sqlite_gateway( mock_anthropic_url: str, tmp_prefix: str, thread_name: str, -) -> Iterator[str]: +) -> Iterator[BootedSqliteGateway]: """Spin up an in-process SQLite gateway pointed at a mock Anthropic server. - Yields the gateway base URL. Cleanup runs on both normal exit and any + Yields a BootedSqliteGateway with url and db_path. Cleanup runs on both normal exit and any failure during setup — `ExitStack` registers each rollback as the matching resource is acquired, so a raise from `check_migrations()`, `create_app()`, or the gateway-startup wait still tears down everything that was set up. @@ -70,7 +77,9 @@ def boot_sqlite_gateway( loop.run_until_complete(check_migrations(db_pool)) - old_env: dict[str, str | None] = {k: os.environ.get(k) for k in ("ANTHROPIC_BASE_URL", "ANTHROPIC_API_KEY")} + old_env: dict[str, str | None] = { + k: os.environ.get(k) for k in ("ANTHROPIC_BASE_URL", "ANTHROPIC_API_KEY", "ENABLE_REQUEST_LOGGING") + } def restore_env() -> None: for k, v in old_env.items(): @@ -84,6 +93,7 @@ def restore_env() -> None: os.environ["ANTHROPIC_BASE_URL"] = mock_anthropic_url os.environ["ANTHROPIC_API_KEY"] = "mock-key" + os.environ["ENABLE_REQUEST_LOGGING"] = "true" clear_settings_cache() app = create_app( @@ -116,4 +126,4 @@ def stop_server() -> None: else: raise RuntimeError(f"SQLite gateway ({thread_name}) did not start") - yield f"http://127.0.0.1:{port}" + yield BootedSqliteGateway(url=f"http://127.0.0.1:{port}", db_path=os.path.join(tmp_dir, "test.db")) diff --git a/tests/luthien_proxy/e2e_tests/sqlite/conftest.py b/tests/luthien_proxy/e2e_tests/sqlite/conftest.py index 45d0ddedf..09e9e57bc 100644 --- a/tests/luthien_proxy/e2e_tests/sqlite/conftest.py +++ b/tests/luthien_proxy/e2e_tests/sqlite/conftest.py @@ -6,7 +6,11 @@ Run: uv run pytest -m sqlite_e2e tests/luthien_proxy/e2e_tests/sqlite/ -v --timeout=30 """ +import os + import pytest +from tests.luthien_proxy.e2e_tests.mock_gemini.server import MockGeminiServer +from tests.luthien_proxy.e2e_tests.mock_openai.server import MockOpenAIServer from tests.luthien_proxy.e2e_tests.sqlite._boot import boot_sqlite_gateway _API_KEY = "test-sqlite-key" @@ -14,16 +18,25 @@ @pytest.fixture(scope="session") -def sqlite_gateway_url(mock_anthropic): - """Start a SQLite-backed gateway on a random port. Returns the base URL.""" +def _sqlite_booted(mock_anthropic): with boot_sqlite_gateway( api_key=_API_KEY, admin_key=_ADMIN_API_KEY, mock_anthropic_url=f"http://localhost:{mock_anthropic.port}", tmp_prefix="luthien_sqlite_e2e_", thread_name="sqlite-gateway", - ) as url: - yield url + ) as booted: + yield booted + + +@pytest.fixture(scope="session") +def sqlite_gateway_url(_sqlite_booted): + return _sqlite_booted.url + + +@pytest.fixture(scope="session") +def sqlite_db_path(_sqlite_booted): + return _sqlite_booted.db_path # --- Fixture overrides --- @@ -44,3 +57,37 @@ def api_key(): @pytest.fixture(scope="session") def admin_api_key(): return _ADMIN_API_KEY + + +@pytest.fixture(scope="session") +def mock_openai_server(): + server = MockOpenAIServer() + server.start() + old_url = os.environ.get("OPENAI_BASE_URL") + old_key = os.environ.get("OPENAI_API_KEY") + os.environ["OPENAI_BASE_URL"] = server.base_url + os.environ["OPENAI_API_KEY"] = "mock-openai-key" + yield server + server.stop() + for k, v in (("OPENAI_BASE_URL", old_url), ("OPENAI_API_KEY", old_key)): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + +@pytest.fixture(scope="session") +def mock_gemini_server(): + server = MockGeminiServer() + server.start() + old_url = os.environ.get("GEMINI_BASE_URL") + old_key = os.environ.get("GOOGLE_API_KEY") + os.environ["GEMINI_BASE_URL"] = server.base_url + os.environ["GOOGLE_API_KEY"] = "mock-google-key" + yield server + server.stop() + for k, v in (("GEMINI_BASE_URL", old_url), ("GOOGLE_API_KEY", old_key)): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v diff --git a/tests/luthien_proxy/e2e_tests/sqlite/test_activity_stream.py b/tests/luthien_proxy/e2e_tests/sqlite/test_activity_stream.py index a2cdcaae1..0415c311c 100644 --- a/tests/luthien_proxy/e2e_tests/sqlite/test_activity_stream.py +++ b/tests/luthien_proxy/e2e_tests/sqlite/test_activity_stream.py @@ -32,15 +32,14 @@ def mock_server(): @pytest.fixture(scope="module") def gateway_url(mock_server): - """Boot an in-process SQLite gateway with no Redis.""" with boot_sqlite_gateway( api_key=_API_KEY, admin_key=_ADMIN_KEY, mock_anthropic_url=f"http://127.0.0.1:{mock_server.port}", tmp_prefix="luthien_activity_e2e_", thread_name="activity-gateway", - ) as url: - yield url + ) as booted: + yield booted.url @pytest.mark.asyncio diff --git a/tests/luthien_proxy/e2e_tests/sqlite/test_passthrough_routes.py b/tests/luthien_proxy/e2e_tests/sqlite/test_passthrough_routes.py new file mode 100644 index 000000000..0b2cfcdb7 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/sqlite/test_passthrough_routes.py @@ -0,0 +1,136 @@ +import asyncio + +import aiosqlite +import httpx +import pytest + +pytestmark = pytest.mark.sqlite_e2e + + +@pytest.mark.asyncio +async def test_openai_headers_persist_to_request_logs(sqlite_gateway_url, sqlite_db_path, api_key, mock_openai_server): + mock_openai_server.clear_requests() + async with httpx.AsyncClient() as client: + response = await client.post( + f"{sqlite_gateway_url}/openai/v1/chat/completions", + headers={ + "Authorization": f"Bearer {api_key}", + "x-luthien-session-id": "persist-test-session", + "x-luthien-agent": "build", + "x-luthien-model": "gpt-4o", + }, + json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}, + ) + assert response.status_code == 200 + + await asyncio.sleep(0.5) + + async with aiosqlite.connect(sqlite_db_path) as db: + cursor = await db.execute( + "SELECT session_id, agent, model, endpoint FROM request_logs" + " WHERE direction='inbound' AND session_id='persist-test-session'" + " ORDER BY created_at DESC LIMIT 1" + ) + row = await cursor.fetchone() + + assert row is not None, "No log row found — ENABLE_REQUEST_LOGGING not set or INSERT failed" + assert row[0] == "persist-test-session" + assert row[1] == "build" + assert row[2] == "gpt-4o" + assert "/openai/" in row[3] + + +@pytest.mark.asyncio +async def test_openai_missing_luthien_headers_null_columns( + sqlite_gateway_url, sqlite_db_path, api_key, mock_openai_server +): + mock_openai_server.clear_requests() + async with httpx.AsyncClient() as client: + response = await client.post( + f"{sqlite_gateway_url}/openai/v1/chat/completions", + headers={ + "Authorization": f"Bearer {api_key}", + "x-luthien-model": "null-test-marker", + }, + json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}, + ) + assert response.status_code == 200 + + await asyncio.sleep(0.5) + + async with aiosqlite.connect(sqlite_db_path) as db: + cursor = await db.execute( + "SELECT session_id, agent FROM request_logs" + " WHERE direction='inbound' AND model='null-test-marker'" + " ORDER BY created_at DESC LIMIT 1" + ) + row = await cursor.fetchone() + + assert row is not None, "No log row found" + assert row[0] is None, "session_id should be NULL when x-luthien-session-id is absent" + assert row[1] is None, "agent should be NULL when x-luthien-agent is absent" + + +def test_openai_missing_auth_returns_401(sqlite_gateway_url): + response = httpx.post( + f"{sqlite_gateway_url}/openai/v1/chat/completions", + json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}, + timeout=10, + ) + assert response.status_code == 401 + + +def test_openai_server_key_not_leaked_to_upstream(sqlite_gateway_url, api_key, mock_openai_server): + mock_openai_server.clear_requests() + httpx.post( + f"{sqlite_gateway_url}/openai/v1/chat/completions", + headers={"Authorization": f"Bearer {api_key}"}, + json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}, + timeout=10, + ) + captured = mock_openai_server.last_request_headers() + assert captured is not None + outbound_auth = next((v for k, v in captured.items() if k.lower() == "authorization"), "") + assert api_key not in outbound_auth, "Proxy key must not reach upstream" + + +def test_gemini_api_key_injected(sqlite_gateway_url, api_key, mock_gemini_server): + mock_gemini_server.clear_requests() + httpx.post( + f"{sqlite_gateway_url}/gemini/v1beta/models/gemini-1.5-flash:generateContent", + headers={"Authorization": f"Bearer {api_key}"}, + json={"contents": [{"parts": [{"text": "hi"}]}]}, + timeout=10, + ) + captured = mock_gemini_server.last_request_headers() + assert captured is not None + lower_keys = {k.lower() for k in captured} + assert "x-goog-api-key" in lower_keys, "Google API key header not injected" + + +def test_luthien_headers_stripped_from_outbound(sqlite_gateway_url, api_key, mock_openai_server): + mock_openai_server.clear_requests() + httpx.post( + f"{sqlite_gateway_url}/openai/v1/chat/completions", + headers={ + "Authorization": f"Bearer {api_key}", + "x-luthien-session-id": "strip-test-session", + "x-luthien-agent": "build", + "x-luthien-model": "gpt-4o", + "x-luthien-provider": "openai", + }, + json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}, + timeout=10, + ) + captured = mock_openai_server.last_request_headers() + assert captured is not None + leaked = [k for k in captured if k.lower().startswith("x-luthien-")] + assert len(leaked) == 0, f"x-luthien-* headers leaked to upstream: {leaked}" + + +def test_anthropic_alias_route_is_registered(sqlite_gateway_url): + response = httpx.get( + f"{sqlite_gateway_url}/anthropic/v1/models", + timeout=5.0, + ) + assert response.status_code == 401 diff --git a/tests/luthien_proxy/e2e_tests/sqlite/test_request_logs_schema.py b/tests/luthien_proxy/e2e_tests/sqlite/test_request_logs_schema.py new file mode 100644 index 000000000..1d1c12b02 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/sqlite/test_request_logs_schema.py @@ -0,0 +1,13 @@ +import aiosqlite +import pytest + +pytestmark = pytest.mark.sqlite_e2e + + +@pytest.mark.asyncio +async def test_request_logs_has_session_id_and_agent(sqlite_db_path): + async with aiosqlite.connect(sqlite_db_path) as db: + cursor = await db.execute("PRAGMA table_info(request_logs)") + columns = {row[1] for row in await cursor.fetchall()} + assert "session_id" in columns, "session_id missing — migration 008 regression" + assert "agent" in columns, "agent missing — migration 019 not applied" diff --git a/tests/luthien_proxy/e2e_tests/test_passthrough_streaming.py b/tests/luthien_proxy/e2e_tests/test_passthrough_streaming.py new file mode 100644 index 000000000..5f6f3a483 --- /dev/null +++ b/tests/luthien_proxy/e2e_tests/test_passthrough_streaming.py @@ -0,0 +1,142 @@ +import httpx +import pytest +from tests.luthien_proxy.e2e_tests.mock_anthropic.server import MockAnthropicServer +from tests.luthien_proxy.e2e_tests.mock_gemini.server import MockGeminiServer +from tests.luthien_proxy.e2e_tests.mock_openai.server import MockOpenAIServer + +pytestmark = pytest.mark.mock_e2e + +_OPENAI_REQUEST = { + "model": "gpt-4o", + "messages": [{"role": "user", "content": "hi"}], +} + +_GEMINI_REQUEST = { + "contents": [{"parts": [{"text": "hi"}]}], +} + +_ANTHROPIC_REQUEST = { + "model": "claude-haiku-4-5", + "max_tokens": 10, + "messages": [{"role": "user", "content": "hi"}], +} + + +@pytest.mark.asyncio +async def test_openai_streaming_delivers_sse_chunks( + mock_openai_server: MockOpenAIServer, + gateway_healthy, + gateway_url: str, + auth_headers: dict, +): + chunks: list[bytes] = [] + async with httpx.AsyncClient(timeout=15.0) as client: + async with client.stream( + "POST", + f"{gateway_url}/openai/v1/chat/completions", + headers=auth_headers, + json={**_OPENAI_REQUEST, "stream": True}, + ) as response: + assert response.status_code == 200 + async for chunk in response.aiter_bytes(): + if chunk: + chunks.append(chunk) + + assert chunks, "No chunks received" + all_data = b"".join(chunks) + assert b"data:" in all_data + assert b"[DONE]" in all_data + + +@pytest.mark.asyncio +async def test_gemini_streaming_delivers_sse_chunks( + mock_gemini_server: MockGeminiServer, + gateway_healthy, + gateway_url: str, + auth_headers: dict, +): + chunks: list[bytes] = [] + async with httpx.AsyncClient(timeout=15.0) as client: + async with client.stream( + "POST", + f"{gateway_url}/gemini/v1beta/models/gemini-1.5-flash:streamGenerateContent", + headers=auth_headers, + json=_GEMINI_REQUEST, + ) as response: + assert response.status_code == 200 + async for chunk in response.aiter_bytes(): + if chunk: + chunks.append(chunk) + + assert chunks, "No chunks received" + all_data = b"".join(chunks) + assert b"data:" in all_data + + +@pytest.mark.asyncio +async def test_anthropic_alias_streaming( + mock_anthropic: MockAnthropicServer, + gateway_healthy, + gateway_url: str, + auth_headers: dict, +): + from tests.luthien_proxy.e2e_tests.mock_anthropic.responses import stream_response + + mock_anthropic.enqueue(stream_response("alias reply")) + + lines: list[str] = [] + async with httpx.AsyncClient(timeout=15.0) as client: + async with client.stream( + "POST", + f"{gateway_url}/anthropic/v1/messages", + headers={**auth_headers, "anthropic-version": "2023-06-01"}, + json={**_ANTHROPIC_REQUEST, "stream": True}, + ) as response: + assert response.status_code == 200 + async for line in response.aiter_lines(): + lines.append(line) + + event_types = {line[len("event: ") :].strip() for line in lines if line.startswith("event: ")} + assert "message_start" in event_types + assert "message_stop" in event_types + + +@pytest.mark.asyncio +async def test_openai_non_streaming_returns_buffered_json( + mock_openai_server: MockOpenAIServer, + gateway_healthy, + gateway_url: str, + auth_headers: dict, +): + async with httpx.AsyncClient(timeout=15.0) as client: + response = await client.post( + f"{gateway_url}/openai/v1/chat/completions", + headers=auth_headers, + json=_OPENAI_REQUEST, + ) + + assert response.status_code == 200 + data = response.json() + assert "choices" in data + assert data["choices"][0]["message"]["content"] + + +@pytest.mark.asyncio +async def test_gemini_non_streaming_returns_buffered_json( + mock_gemini_server: MockGeminiServer, + gateway_healthy, + gateway_url: str, + auth_headers: dict, +): + async with httpx.AsyncClient(timeout=15.0) as client: + response = await client.post( + f"{gateway_url}/gemini/v1beta/models/gemini-1.5-flash:generateContent", + headers=auth_headers, + json=_GEMINI_REQUEST, + ) + + assert response.status_code == 200 + data = response.json() + assert "candidates" in data + parts = data["candidates"][0]["content"]["parts"] + assert any(p.get("text") for p in parts) diff --git a/tests/luthien_proxy/unit_tests/request_log/test_recorder.py b/tests/luthien_proxy/unit_tests/request_log/test_recorder.py index 2d987cfc8..a87d5ceb0 100644 --- a/tests/luthien_proxy/unit_tests/request_log/test_recorder.py +++ b/tests/luthien_proxy/unit_tests/request_log/test_recorder.py @@ -42,6 +42,7 @@ def test_pending_log_defaults(self) -> None: assert log.direction == "inbound" assert log.transaction_id == "txn-123" assert log.session_id is None + assert log.agent is None assert log.http_method is None assert log.url is None assert log.request_headers is None @@ -168,6 +169,7 @@ def test_record_inbound_request_stores_all_fields(self) -> None: headers=headers, body=body, session_id="sess-456", + agent="opencode/1.0", model="gpt-4-turbo", is_streaming=True, endpoint="/v1/messages/completions", @@ -176,6 +178,7 @@ def test_record_inbound_request_stores_all_fields(self) -> None: assert recorder._inbound.http_method == "POST" assert recorder._inbound.url == "http://example.com/api" assert recorder._inbound.session_id == "sess-456" + assert recorder._inbound.agent == "opencode/1.0" assert recorder._inbound.model == "gpt-4-turbo" assert recorder._inbound.is_streaming is True assert recorder._inbound.endpoint == "/v1/messages/completions" @@ -280,7 +283,6 @@ def test_record_outbound_request_inherits_session_id_from_inbound(self) -> None: db_pool = MagicMock(spec=DatabasePool) recorder = RequestLogRecorder(db_pool, "txn-123") - # Set inbound with session_id recorder.record_inbound_request( method="POST", url="http://example.com/api", @@ -288,12 +290,37 @@ def test_record_outbound_request_inherits_session_id_from_inbound(self) -> None: body={}, session_id="my-session-id", ) - - # Record outbound without specifying session_id recorder.record_outbound_request(body={}) assert recorder._outbound.session_id == "my-session-id" + def test_record_outbound_request_inherits_agent_from_inbound(self) -> None: + """record_outbound_request copies agent from inbound log.""" + db_pool = MagicMock(spec=DatabasePool) + recorder = RequestLogRecorder(db_pool, "txn-123") + + recorder.record_inbound_request( + method="POST", + url="http://example.com/api", + headers={}, + body={}, + agent="opencode/1.0", + ) + recorder.record_outbound_request(body={}) + + assert recorder._outbound.agent == "opencode/1.0" + + def test_record_inbound_request_agent_defaults_to_none(self) -> None: + """agent defaults to None when not provided.""" + db_pool = MagicMock(spec=DatabasePool) + recorder = RequestLogRecorder(db_pool, "txn-123") + + recorder.record_inbound_request(method="GET", url="http://example.com", headers={}, body={}) + + assert recorder._inbound.agent is None + recorder.record_outbound_request(body={}) + assert recorder._outbound.agent is None + def test_record_outbound_request_updates_started_at(self) -> None: """record_outbound_request sets started_at to current time.""" db_pool = MagicMock(spec=DatabasePool) @@ -861,3 +888,100 @@ async def test_success_does_not_raise(self) -> None: await _insert_log_row(conn, self._make_pending(), lambda b: None) conn.execute.assert_called_once() + + +class TestPendingLogPopulatedFlag: + def test_populated_defaults_false(self) -> None: + log = _PendingLog(direction="outbound", transaction_id="txn-1") + assert log.populated is False + + def test_record_outbound_request_sets_populated(self) -> None: + recorder = RequestLogRecorder.__new__(RequestLogRecorder) + recorder._transaction_id = "txn-1" + recorder._inbound = _PendingLog(direction="inbound", transaction_id="txn-1") + recorder._outbound = _PendingLog(direction="outbound", transaction_id="txn-1") + + recorder.record_outbound_request(body={}, method="POST") + + assert recorder._outbound.populated is True + + def test_record_outbound_response_does_not_set_populated(self) -> None: + recorder = RequestLogRecorder.__new__(RequestLogRecorder) + recorder._transaction_id = "txn-2" + recorder._inbound = _PendingLog(direction="inbound", transaction_id="txn-2") + recorder._outbound = _PendingLog( + direction="outbound", + transaction_id="txn-2", + started_at=__import__("time").time(), + ) + + recorder.record_outbound_response(status=200) + + assert recorder._outbound.populated is False + + @pytest.mark.asyncio + async def test_write_logs_skips_unpopulated_outbound(self) -> None: + recorder = RequestLogRecorder.__new__(RequestLogRecorder) + recorder._transaction_id = "txn-skip" + recorder._inbound = _PendingLog( + direction="inbound", + transaction_id="txn-skip", + http_method="POST", + populated=True, + ) + recorder._outbound = _PendingLog(direction="outbound", transaction_id="txn-skip") + + insert_calls: list[str] = [] + + async def fake_insert(conn, pending, serialize): + insert_calls.append(pending.direction) + + with patch("luthien_proxy.request_log.recorder._insert_log_row", side_effect=fake_insert): + mock_pool = MagicMock() + mock_conn = AsyncMock() + mock_pool.connection = MagicMock( + return_value=AsyncMock( + __aenter__=AsyncMock(return_value=mock_conn), + __aexit__=AsyncMock(return_value=None), + ) + ) + recorder._db_pool = mock_pool + await recorder._write_logs() + + assert insert_calls == ["inbound"] + + @pytest.mark.asyncio + async def test_write_logs_includes_populated_outbound(self) -> None: + recorder = RequestLogRecorder.__new__(RequestLogRecorder) + recorder._transaction_id = "txn-both" + recorder._inbound = _PendingLog( + direction="inbound", + transaction_id="txn-both", + http_method="POST", + populated=True, + ) + recorder._outbound = _PendingLog( + direction="outbound", + transaction_id="txn-both", + http_method="POST", + populated=True, + ) + + insert_calls: list[str] = [] + + async def fake_insert(conn, pending, serialize): + insert_calls.append(pending.direction) + + with patch("luthien_proxy.request_log.recorder._insert_log_row", side_effect=fake_insert): + mock_pool = MagicMock() + mock_conn = AsyncMock() + mock_pool.connection = MagicMock( + return_value=AsyncMock( + __aenter__=AsyncMock(return_value=mock_conn), + __aexit__=AsyncMock(return_value=None), + ) + ) + recorder._db_pool = mock_pool + await recorder._write_logs() + + assert insert_calls == ["inbound", "outbound"] diff --git a/tests/luthien_proxy/unit_tests/request_log/test_sanitize.py b/tests/luthien_proxy/unit_tests/request_log/test_sanitize.py index b6828e1be..148645375 100644 --- a/tests/luthien_proxy/unit_tests/request_log/test_sanitize.py +++ b/tests/luthien_proxy/unit_tests/request_log/test_sanitize.py @@ -50,6 +50,20 @@ def test_x_anthropic_api_key_mixed_case_redacted(self): assert result["X-Anthropic-API-Key"] == "[REDACTED]" +class TestGoogApiKeyHeader: + """Tests that x-goog-api-key is treated as sensitive.""" + + def test_x_goog_api_key_redacted(self): + headers = {"x-goog-api-key": "AIzaSyD-some-gemini-key"} + result = sanitize_headers(headers) + assert result["x-goog-api-key"] == "[REDACTED]" + + def test_x_goog_api_key_mixed_case_redacted(self): + headers = {"X-Goog-API-Key": "AIzaSyD-some-gemini-key"} + result = sanitize_headers(headers) + assert result["X-Goog-API-Key"] == "[REDACTED]" + + class TestCaseInsensitivity: """Tests that header name matching is case-insensitive.""" diff --git a/tests/luthien_proxy/unit_tests/test_passthrough_auth.py b/tests/luthien_proxy/unit_tests/test_passthrough_auth.py new file mode 100644 index 000000000..1857ae58e --- /dev/null +++ b/tests/luthien_proxy/unit_tests/test_passthrough_auth.py @@ -0,0 +1,340 @@ +"""Unit tests for passthrough_auth dependency.""" + +from unittest.mock import MagicMock + +import pytest +from fastapi import HTTPException +from fastapi.security import HTTPAuthorizationCredentials + +from luthien_proxy.credential_manager import AuthConfig, AuthMode, CredentialManager +from luthien_proxy.passthrough_auth import verify_passthrough_token, verify_strict_client_key + + +def _make_request(headers: dict | None = None) -> MagicMock: + req = MagicMock() + req.headers = headers or {} + return req + + +@pytest.mark.asyncio +async def test_passthrough_mode_accepts_any_token(): + """PASSTHROUGH mode: any token is accepted.""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.PASSTHROUGH, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="any-token") + result = await verify_passthrough_token( + request=_make_request(), + credentials=credentials, + api_key=None, + credential_manager=cred_manager, + ) + + assert result == "any-token" + + +@pytest.mark.asyncio +async def test_passthrough_mode_accepts_no_token(): + """PASSTHROUGH mode: missing token is also accepted.""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.PASSTHROUGH, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + result = await verify_passthrough_token( + request=_make_request(), + credentials=None, + api_key=None, + credential_manager=cred_manager, + ) + + assert result == "" + + +@pytest.mark.asyncio +async def test_client_key_mode_accepts_matching_token(): + """CLIENT_KEY mode: token matching CLIENT_API_KEY is accepted.""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.CLIENT_KEY, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="sk-test-key") + result = await verify_passthrough_token( + request=_make_request(), + credentials=credentials, + api_key="sk-test-key", + credential_manager=cred_manager, + ) + + assert result == "sk-test-key" + + +@pytest.mark.asyncio +async def test_client_key_mode_rejects_mismatched_token(): + """CLIENT_KEY mode: token not matching CLIENT_API_KEY is rejected.""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.CLIENT_KEY, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="wrong-token") + with pytest.raises(HTTPException) as exc_info: + await verify_passthrough_token( + request=_make_request(), + credentials=credentials, + api_key="sk-test-key", + credential_manager=cred_manager, + ) + + assert exc_info.value.status_code == 401 + assert "Invalid bearer token" in exc_info.value.detail + + +@pytest.mark.asyncio +async def test_client_key_mode_rejects_missing_token(): + """CLIENT_KEY mode: missing token is rejected.""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.CLIENT_KEY, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + with pytest.raises(HTTPException) as exc_info: + await verify_passthrough_token( + request=_make_request(), + credentials=None, + api_key="sk-test-key", + credential_manager=cred_manager, + ) + + assert exc_info.value.status_code == 401 + assert "Missing bearer token" in exc_info.value.detail + + +@pytest.mark.asyncio +async def test_both_mode_accepts_matching_client_key(): + """BOTH mode: token matching CLIENT_API_KEY is accepted.""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.BOTH, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="sk-test-key") + result = await verify_passthrough_token( + request=_make_request(), + credentials=credentials, + api_key="sk-test-key", + credential_manager=cred_manager, + ) + + assert result == "sk-test-key" + + +@pytest.mark.asyncio +async def test_both_mode_accepts_any_other_token(): + """BOTH mode: any token (not matching CLIENT_API_KEY) is also accepted (passthrough path).""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.BOTH, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="user-token") + result = await verify_passthrough_token( + request=_make_request(), + credentials=credentials, + api_key="sk-test-key", + credential_manager=cred_manager, + ) + + assert result == "user-token" + + +@pytest.mark.asyncio +async def test_both_mode_rejects_missing_token(): + """BOTH mode: missing token is rejected.""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.BOTH, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + with pytest.raises(HTTPException) as exc_info: + await verify_passthrough_token( + request=_make_request(), + credentials=None, + api_key="sk-test-key", + credential_manager=cred_manager, + ) + + assert exc_info.value.status_code == 401 + assert "Missing bearer token" in exc_info.value.detail + + +@pytest.mark.asyncio +async def test_no_credential_manager_defaults_to_client_key(): + """When credential_manager is None, default to CLIENT_KEY mode.""" + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="sk-test-key") + result = await verify_passthrough_token( + request=_make_request(), + credentials=credentials, + api_key="sk-test-key", + credential_manager=None, + ) + + assert result == "sk-test-key" + + +@pytest.mark.asyncio +async def test_no_credential_manager_rejects_mismatched_token(): + """When credential_manager is None, reject token not matching api_key.""" + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="wrong-token") + with pytest.raises(HTTPException) as exc_info: + await verify_passthrough_token( + request=_make_request(), + credentials=credentials, + api_key="sk-test-key", + credential_manager=None, + ) + + assert exc_info.value.status_code == 401 + + +@pytest.mark.asyncio +async def test_x_api_key_header_accepted_in_passthrough_mode(): + """x-api-key header is accepted as an alternative to Bearer token.""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.PASSTHROUGH, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + result = await verify_passthrough_token( + request=_make_request({"x-api-key": "sk-from-header"}), + credentials=None, + api_key=None, + credential_manager=cred_manager, + ) + + assert result == "sk-from-header" + + +@pytest.mark.asyncio +async def test_x_anthropic_api_key_header_accepted_in_passthrough_mode(): + """x-anthropic-api-key header is accepted as an alternative to Bearer token.""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.PASSTHROUGH, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + result = await verify_passthrough_token( + request=_make_request({"x-anthropic-api-key": "sk-anthropic-key"}), + credentials=None, + api_key=None, + credential_manager=cred_manager, + ) + + assert result == "sk-anthropic-key" + + +@pytest.mark.asyncio +async def test_open_proxy_closed_no_client_api_key(): + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="any-token") + with pytest.raises(HTTPException) as exc_info: + await verify_strict_client_key(credentials=credentials, api_key=None) + assert exc_info.value.status_code == 401 + + +@pytest.mark.asyncio +async def test_open_proxy_closed_wrong_token(): + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="wrong-key") + with pytest.raises(HTTPException) as exc_info: + await verify_strict_client_key(credentials=credentials, api_key="correct-key") + assert exc_info.value.status_code == 401 + + +@pytest.mark.asyncio +async def test_valid_client_key_accepted(): + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="correct-key") + result = await verify_strict_client_key(credentials=credentials, api_key="correct-key") + assert result == "correct-key" + + +@pytest.mark.asyncio +async def test_timing_safe_comparison(): + """Token comparison uses secrets.compare_digest (timing-safe).""" + cred_manager = MagicMock(spec=CredentialManager) + cred_manager.config = AuthConfig( + auth_mode=AuthMode.CLIENT_KEY, + validate_credentials=False, + valid_cache_ttl_seconds=3600, + invalid_cache_ttl_seconds=60, + ) + + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="sk-test-key") + result = await verify_passthrough_token( + request=_make_request(), + credentials=credentials, + api_key="sk-test-key", + credential_manager=cred_manager, + ) + + assert result == "sk-test-key" + + +@pytest.mark.asyncio +async def test_strict_client_key_rejects_when_env_unset(): + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="any-token") + + with pytest.raises(HTTPException) as exc_info: + await verify_strict_client_key(credentials=credentials, api_key=None) + + assert exc_info.value.status_code == 401 + + +@pytest.mark.asyncio +async def test_strict_client_key_rejects_wrong_token(): + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="sk-wrong-key") + + with pytest.raises(HTTPException) as exc_info: + await verify_strict_client_key(credentials=credentials, api_key="sk-correct-key") + + assert exc_info.value.status_code == 401 + + +@pytest.mark.asyncio +async def test_strict_client_key_accepts_matching_token(): + credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="sk-correct-key") + + result = await verify_strict_client_key(credentials=credentials, api_key="sk-correct-key") + + assert result == "sk-correct-key" diff --git a/tests/luthien_proxy/unit_tests/test_passthrough_routes.py b/tests/luthien_proxy/unit_tests/test_passthrough_routes.py new file mode 100644 index 000000000..be0a5de7e --- /dev/null +++ b/tests/luthien_proxy/unit_tests/test_passthrough_routes.py @@ -0,0 +1,810 @@ +from __future__ import annotations + +import tempfile +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from luthien_proxy.passthrough_auth import verify_passthrough_token, verify_strict_client_key +from luthien_proxy.passthrough_routes import router +from luthien_proxy.request_log.recorder import NoOpRequestLogRecorder, RequestLogRecorder +from luthien_proxy.utils.db import DatabasePool + + +@pytest.fixture(autouse=True) +def _patch_provider_keys(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "test-openai-key") + monkeypatch.setenv("GOOGLE_API_KEY", "test-google-key") + + +def _make_buffered_client(status_code: int = 200, content: bytes = b"{}", headers: dict | None = None) -> MagicMock: + mock_response = MagicMock() + mock_response.status_code = status_code + mock_response.content = content + mock_response.headers = headers or {} + mock_client = MagicMock() + mock_client.request = AsyncMock(return_value=mock_response) + return mock_client + + +def _make_app(deps=None, buffered_client=None) -> FastAPI: + app = FastAPI() + app.include_router(router) + app.dependency_overrides[verify_passthrough_token] = lambda: "tok" + app.dependency_overrides[verify_strict_client_key] = lambda: "tok" + if deps is not None: + app.state.dependencies = deps + app.state.passthrough_buffered_client = buffered_client or _make_buffered_client() + app.state.passthrough_streaming_client = MagicMock() + return app + + +def _make_deps(*, enable_request_logging: bool = True) -> MagicMock: + deps = MagicMock() + deps.db_pool = MagicMock(spec=DatabasePool) + deps.enable_request_logging = enable_request_logging + return deps + + +class TestPassthroughRecorderCreation: + def test_creates_noop_recorder_when_logging_disabled(self) -> None: + deps = _make_deps(enable_request_logging=False) + app = _make_app(deps) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_recorder = MagicMock(spec=NoOpRequestLogRecorder) + mock_create.return_value = mock_recorder + + client = TestClient(app, raise_server_exceptions=True) + client.post("/openai/v1/chat/completions", json={"model": "gpt-4o", "messages": []}) + + mock_create.assert_called_once() + assert mock_create.call_args.kwargs["enabled"] is False + + def test_creates_noop_recorder_when_no_deps(self) -> None: + app = _make_app(deps=None) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_recorder = MagicMock(spec=NoOpRequestLogRecorder) + mock_create.return_value = mock_recorder + + client = TestClient(app, raise_server_exceptions=True) + client.post("/openai/v1/chat/completions", json={"model": "gpt-4o", "messages": []}) + + mock_create.assert_called_once() + assert mock_create.call_args.kwargs["enabled"] is False + assert mock_create.call_args.kwargs["db_pool"] is None + + +class TestPassthroughLuthienHeadersPersisted: + def _make_request_and_capture(self, headers: dict[str, str], body: dict) -> tuple[MagicMock, dict]: + deps = _make_deps(enable_request_logging=True) + app = _make_app(deps) + + captured: dict = {} + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_recorder = MagicMock(spec=RequestLogRecorder) + mock_create.return_value = mock_recorder + + def capture_inbound(**kwargs): + captured.update(kwargs) + + mock_recorder.record_inbound_request.side_effect = capture_inbound + + client = TestClient(app, raise_server_exceptions=True) + client.post("/openai/v1/chat/completions", json=body, headers=headers) + + return mock_recorder, captured + + def test_session_id_passed_to_recorder(self) -> None: + _, captured = self._make_request_and_capture( + headers={"x-luthien-session-id": "sess-abc"}, + body={"model": "gpt-4o", "messages": []}, + ) + assert captured["session_id"] == "sess-abc" + + def test_agent_passed_to_recorder(self) -> None: + _, captured = self._make_request_and_capture( + headers={"x-luthien-agent": "opencode/1.0"}, + body={"model": "gpt-4o", "messages": []}, + ) + assert captured["agent"] == "opencode/1.0" + + def test_model_passed_to_recorder(self) -> None: + _, captured = self._make_request_and_capture( + headers={"x-luthien-model": "gpt-4o"}, + body={"model": "gpt-4o", "messages": []}, + ) + assert captured["model"] == "gpt-4o" + + def test_missing_luthien_headers_yields_none_values(self) -> None: + _, captured = self._make_request_and_capture( + headers={}, + body={"model": "gpt-4o", "messages": []}, + ) + assert captured["session_id"] is None + assert captured["agent"] is None + assert captured["model"] is None + + def test_all_luthien_headers_passed_together(self) -> None: + _, captured = self._make_request_and_capture( + headers={ + "x-luthien-session-id": "sess-xyz", + "x-luthien-agent": "opencode/2.0", + "x-luthien-model": "gpt-4o-mini", + }, + body={"model": "gpt-4o-mini", "messages": []}, + ) + assert captured["session_id"] == "sess-xyz" + assert captured["agent"] == "opencode/2.0" + assert captured["model"] == "gpt-4o-mini" + + +class TestPassthroughRecorderFlush: + def test_flush_called_after_buffered_response(self) -> None: + deps = _make_deps(enable_request_logging=True) + mock_buffered = _make_buffered_client(status_code=201, content=b'{"id": "chatcmpl-123"}') + app = _make_app(deps, buffered_client=mock_buffered) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_recorder = MagicMock(spec=RequestLogRecorder) + mock_create.return_value = mock_recorder + + client = TestClient(app, raise_server_exceptions=True) + client.post("/openai/v1/chat/completions", json={"model": "gpt-4o", "messages": []}) + + mock_recorder.record_inbound_response.assert_called_once_with(status=201) + mock_recorder.flush.assert_called_once() + + def test_flush_called_on_upstream_error(self) -> None: + import httpx + + deps = _make_deps(enable_request_logging=True) + mock_buffered = MagicMock() + mock_buffered.request = AsyncMock(side_effect=httpx.ConnectError("refused")) + app = _make_app(deps, buffered_client=mock_buffered) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_recorder = MagicMock(spec=RequestLogRecorder) + mock_create.return_value = mock_recorder + + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/anthropic/v1/messages", + json={"model": "claude-haiku-4-5", "max_tokens": 10, "messages": []}, + headers={"anthropic-version": "2023-06-01"}, + ) + + assert response.status_code == 502 + mock_recorder.record_inbound_response.assert_called_once() + args = mock_recorder.record_inbound_response.call_args.kwargs + assert args["status"] == 502 + mock_recorder.flush.assert_called_once() + + +@pytest.mark.parametrize( + "sensitive_header", + [ + "authorization", + "x-api-key", + "x-anthropic-api-key", + "x-goog-api-key", + ], +) +def test_sensitive_headers_are_passed_to_recorder_raw(sensitive_header: str) -> None: + deps = _make_deps(enable_request_logging=True) + app = _make_app(deps) + captured_headers: dict[str, str] = {} + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_recorder = MagicMock(spec=RequestLogRecorder) + mock_create.return_value = mock_recorder + + def capture(**kwargs): + captured_headers.update(kwargs.get("headers", {})) + + mock_recorder.record_inbound_request.side_effect = capture + + client = TestClient(app) + client.post( + "/openai/v1/chat/completions", + json={"model": "gpt-4o", "messages": []}, + headers={sensitive_header: "secret-value"}, + ) + + assert sensitive_header in captured_headers + + +@pytest.fixture +def policy_config_file(): + config_content = """ +policy: + class: "luthien_proxy.policies.noop_policy:NoOpPolicy" + config: {} +""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: + f.write(config_content) + config_path = f.name + + yield config_path + Path(config_path).unlink(missing_ok=True) + + +@pytest.fixture +def mock_db_pool(): + mock = AsyncMock() + mock_pool = AsyncMock() + mock_pool.fetchrow = AsyncMock(return_value=None) + mock.get_pool = AsyncMock(return_value=mock_pool) + mock.close = AsyncMock() + mock.is_sqlite = False + return mock + + +class TestBodySizeLimit: + def test_body_size_limit_413(self) -> None: + app = _make_app() + + with patch("luthien_proxy.passthrough_routes.MAX_REQUEST_PAYLOAD_BYTES", 5): + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/openai/v1/chat/completions", + content=b"123456", + headers={"content-type": "application/octet-stream"}, + ) + + assert response.status_code == 413 + + def test_body_size_limit_normal_passes(self) -> None: + app = _make_app() + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/openai/v1/chat/completions", + json={"model": "gpt-4o", "messages": []}, + ) + + assert response.status_code != 413 + + +class TestHopByHopHeaderStripping: + def test_hop_by_hop_stripped(self) -> None: + upstream_headers = { + "transfer-encoding": "chunked", + "set-cookie": "x=y", + "server": "nginx", + "content-type": "application/json", + } + mock_buffered = _make_buffered_client(headers=upstream_headers) + app = _make_app(buffered_client=mock_buffered) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + client = TestClient(app, raise_server_exceptions=False) + response = client.post("/openai/v1/chat/completions", json={"model": "gpt-4o", "messages": []}) + + assert "transfer-encoding" not in response.headers + assert "set-cookie" not in response.headers + assert "server" not in response.headers + + def test_essential_headers_preserved(self) -> None: + upstream_headers = { + "content-type": "application/json", + "x-request-id": "req-123", + } + mock_buffered = _make_buffered_client(headers=upstream_headers) + app = _make_app(buffered_client=mock_buffered) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + client = TestClient(app, raise_server_exceptions=False) + response = client.post("/openai/v1/chat/completions", json={"model": "gpt-4o", "messages": []}) + + assert response.headers.get("content-type", "").startswith("application/json") + assert response.headers.get("x-request-id") == "req-123" + + +class TestLifespanHttpxClients: + def test_lifespan_closes_httpx_clients(self, policy_config_file, mock_db_pool) -> None: + from luthien_proxy.main import create_app as main_create_app + + app = main_create_app( + api_key=None, + admin_key=None, + db_pool=mock_db_pool, + redis_client=None, + startup_policy_path=policy_config_file, + ) + + with TestClient(app): + streaming_client = app.state.passthrough_streaming_client + buffered_client = app.state.passthrough_buffered_client + assert not streaming_client.is_closed + assert not buffered_client.is_closed + + assert streaming_client.is_closed + assert buffered_client.is_closed + + +class TestAnthropicBaseUrl: + def test_default_anthropic_base_has_no_v1_suffix(self) -> None: + from luthien_proxy.passthrough_routes import UPSTREAM_BASES + + base = UPSTREAM_BASES["anthropic"] + assert not base.endswith("/v1"), ( + f"UPSTREAM_BASES['anthropic'] must not include /v1 (got {base!r}); " + "a request to /anthropic/v1/messages would become /v1/v1/messages upstream" + ) + assert base == "https://api.anthropic.com" + + def test_upstream_url_construction_no_double_v1(self) -> None: + from luthien_proxy.passthrough_routes import UPSTREAM_BASES + + base = UPSTREAM_BASES["anthropic"] + path = "v1/messages" + constructed = f"{base}/{path}" + assert constructed == "https://api.anthropic.com/v1/messages" + assert "/v1/v1/" not in constructed + + +class TestStreamingUpstreamError: + def _make_streaming_app(self, streaming_client: MagicMock, deps=None) -> FastAPI: + app = FastAPI() + app.include_router(router) + app.dependency_overrides[verify_passthrough_token] = lambda: "tok" + app.dependency_overrides[verify_strict_client_key] = lambda: "tok" + if deps is not None: + app.state.dependencies = deps + app.state.passthrough_buffered_client = _make_buffered_client() + app.state.passthrough_streaming_client = streaming_client + return app + + def _make_streaming_client( + self, status_code: int, content: bytes = b"", content_type: str = "text/event-stream" + ) -> MagicMock: + mock_response = MagicMock() + mock_response.status_code = status_code + mock_response.headers = {"content-type": content_type} + + async def aread(): + return content + + async def aiter_bytes(): + yield content + + mock_response.aread = aread + mock_response.aiter_bytes = aiter_bytes + + cm = AsyncMock() + cm.__aenter__ = AsyncMock(return_value=mock_response) + cm.__aexit__ = AsyncMock(return_value=None) + + mock_client = MagicMock() + mock_client.stream = MagicMock(return_value=cm) + return mock_client + + @pytest.mark.parametrize("upstream_status", [401, 429, 500, 503]) + def test_streaming_upstream_error_returns_real_status(self, upstream_status: int) -> None: + error_body = b'{"error": "upstream error"}' + streaming_client = self._make_streaming_client( + status_code=upstream_status, + content=error_body, + content_type="application/json", + ) + app = self._make_streaming_app(streaming_client) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock(spec=NoOpRequestLogRecorder) + with patch.dict("os.environ", {"ANTHROPIC_BASE_URL": "http://mock-upstream"}): + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/anthropic/v1/messages", + json={"model": "claude-haiku-4-5", "max_tokens": 10, "messages": [], "stream": True}, + headers={"anthropic-version": "2023-06-01"}, + ) + + assert response.status_code == upstream_status, ( + f"Expected {upstream_status} from upstream to be forwarded to client, got {response.status_code}" + ) + + def test_streaming_2xx_returns_streaming_response(self) -> None: + sse_chunk = b"data: {}\n\ndata: [DONE]\n\n" + streaming_client = self._make_streaming_client( + status_code=200, + content=sse_chunk, + content_type="text/event-stream", + ) + app = self._make_streaming_app(streaming_client) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock(spec=NoOpRequestLogRecorder) + with patch.dict("os.environ", {"ANTHROPIC_BASE_URL": "http://mock-upstream"}): + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/anthropic/v1/messages", + json={"model": "claude-haiku-4-5", "max_tokens": 10, "messages": [], "stream": True}, + headers={"anthropic-version": "2023-06-01"}, + ) + + assert response.status_code == 200 + assert b"DONE" in response.content + + def test_streaming_forwards_upstream_content_type(self) -> None: + streaming_client = self._make_streaming_client( + status_code=200, + content=b'[{"candidates": []}]', + content_type="application/json", + ) + app = self._make_streaming_app(streaming_client) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock(spec=NoOpRequestLogRecorder) + with patch.dict("os.environ", {"GEMINI_BASE_URL": "http://mock-upstream", "GOOGLE_API_KEY": "test-key"}): + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/gemini/v1beta/models/gemini-1.5-flash:streamGenerateContent", + json={"contents": [{"parts": [{"text": "hi"}]}]}, + ) + + assert "application/json" in response.headers.get("content-type", "") + + +class TestMissingServerKey: + def test_openai_missing_key_returns_503(self) -> None: + app = _make_app() + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock(spec=NoOpRequestLogRecorder) + with patch.dict("os.environ", {}, clear=True): + env = {k: v for k, v in __import__("os").environ.items() if k != "OPENAI_API_KEY"} + with patch.dict("os.environ", env, clear=True): + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/openai/v1/chat/completions", + json={"model": "gpt-4o", "messages": []}, + ) + + assert response.status_code == 503 + + def test_gemini_missing_key_returns_503(self) -> None: + app = _make_app() + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock(spec=NoOpRequestLogRecorder) + env = {k: v for k, v in __import__("os").environ.items() if k != "GOOGLE_API_KEY"} + with patch.dict("os.environ", env, clear=True): + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/gemini/v1beta/models/gemini-1.5-flash:generateContent", + json={"contents": [{"parts": [{"text": "hi"}]}]}, + ) + + assert response.status_code == 503 + + +class TestContentEncodingStripped: + def test_content_encoding_stripped_from_buffered_response(self) -> None: + upstream_headers = { + "content-type": "application/json", + "content-encoding": "gzip", + "x-request-id": "req-123", + } + mock_buffered = _make_buffered_client(status_code=200, content=b'{"ok": true}', headers=upstream_headers) + app = _make_app(buffered_client=mock_buffered) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + client = TestClient(app, raise_server_exceptions=False) + response = client.post("/openai/v1/chat/completions", json={"model": "gpt-4o", "messages": []}) + + assert response.status_code == 200 + assert "content-encoding" not in response.headers + + def test_safe_headers_still_forwarded(self) -> None: + upstream_headers = { + "content-type": "application/json", + "x-request-id": "req-456", + } + mock_buffered = _make_buffered_client(status_code=200, content=b'{"ok": true}', headers=upstream_headers) + app = _make_app(buffered_client=mock_buffered) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + client = TestClient(app, raise_server_exceptions=False) + response = client.post("/openai/v1/chat/completions", json={"model": "gpt-4o", "messages": []}) + + assert response.headers.get("x-request-id") == "req-456" + + +class TestInvalidContentLength: + def test_non_numeric_content_length_returns_400(self) -> None: + app = _make_app() + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/openai/v1/chat/completions", + content=b'{"model": "gpt-4o"}', + headers={"content-type": "application/json", "content-length": "abc"}, + ) + + assert response.status_code == 400 + + def test_valid_content_length_passes(self) -> None: + app = _make_app() + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/openai/v1/chat/completions", + json={"model": "gpt-4o", "messages": []}, + ) + + assert response.status_code != 400 + + +class TestGeminiKeyQueryStripping: + def test_key_param_stripped_from_gemini_query(self) -> None: + captured_urls: list[str] = [] + mock_buffered = MagicMock() + + async def fake_request(method, url, **kwargs): + captured_urls.append(url) + resp = MagicMock() + resp.status_code = 200 + resp.content = b"{}" + resp.headers = {"content-type": "application/json"} + return resp + + mock_buffered.request = fake_request + app = _make_app(buffered_client=mock_buffered) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + with patch.dict("os.environ", {"GEMINI_BASE_URL": "http://mock-upstream"}): + client = TestClient(app, raise_server_exceptions=False) + client.post( + "/gemini/v1beta/models/gemini-1.5-flash:generateContent?key=CLIENT_SECRET&alt=json", + json={"contents": []}, + ) + + assert captured_urls, "No upstream request was made" + upstream_url = captured_urls[0] + assert "key=CLIENT_SECRET" not in upstream_url + assert "alt=json" in upstream_url + + def test_non_key_params_preserved_for_gemini(self) -> None: + captured_urls: list[str] = [] + mock_buffered = MagicMock() + + async def fake_request(method, url, **kwargs): + captured_urls.append(url) + resp = MagicMock() + resp.status_code = 200 + resp.content = b"{}" + resp.headers = {"content-type": "application/json"} + return resp + + mock_buffered.request = fake_request + app = _make_app(buffered_client=mock_buffered) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + with patch.dict("os.environ", {"GEMINI_BASE_URL": "http://mock-upstream"}): + client = TestClient(app, raise_server_exceptions=False) + client.post( + "/gemini/v1beta/models/gemini-1.5-flash:generateContent?alt=sse", + json={"contents": []}, + ) + + assert captured_urls + assert "alt=sse" in captured_urls[0] + + def test_key_param_not_stripped_for_openai(self) -> None: + captured_urls: list[str] = [] + mock_buffered = MagicMock() + + async def fake_request(method, url, **kwargs): + captured_urls.append(url) + resp = MagicMock() + resp.status_code = 200 + resp.content = b"{}" + resp.headers = {"content-type": "application/json"} + return resp + + mock_buffered.request = fake_request + app = _make_app(buffered_client=mock_buffered) + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock() + with patch.dict("os.environ", {"OPENAI_BASE_URL": "http://mock-upstream"}): + client = TestClient(app, raise_server_exceptions=False) + client.post( + "/openai/v1/chat/completions?key=somevalue", + json={"model": "gpt-4o", "messages": []}, + ) + + assert captured_urls + assert "key=somevalue" in captured_urls[0] + + +class TestStreamingResponseHeaders: + def _make_streaming_client_with_headers(self, status_code: int, content: bytes, headers: dict) -> MagicMock: + mock_response = MagicMock() + mock_response.status_code = status_code + mock_response.headers = headers + + async def aread(): + return content + + async def aiter_bytes(): + yield content + + mock_response.aread = aread + mock_response.aiter_bytes = aiter_bytes + + cm = AsyncMock() + cm.__aenter__ = AsyncMock(return_value=mock_response) + cm.__aexit__ = AsyncMock(return_value=None) + + mock_client = MagicMock() + mock_client.stream = MagicMock(return_value=cm) + return mock_client + + def test_streaming_2xx_forwards_upstream_headers(self) -> None: + streaming_client = self._make_streaming_client_with_headers( + status_code=200, + content=b"data: {}\n\ndata: [DONE]\n\n", + headers={ + "content-type": "text/event-stream", + "x-request-id": "req-stream-123", + "anthropic-ratelimit-requests-remaining": "99", + }, + ) + app = FastAPI() + app.include_router(router) + app.dependency_overrides[verify_passthrough_token] = lambda: "tok" + app.dependency_overrides[verify_strict_client_key] = lambda: "tok" + app.state.passthrough_buffered_client = _make_buffered_client() + app.state.passthrough_streaming_client = streaming_client + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock(spec=NoOpRequestLogRecorder) + with patch.dict("os.environ", {"ANTHROPIC_BASE_URL": "http://mock-upstream"}): + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/anthropic/v1/messages", + json={"model": "claude-haiku-4-5", "max_tokens": 10, "messages": [], "stream": True}, + headers={"anthropic-version": "2023-06-01"}, + ) + + assert response.status_code == 200 + assert response.headers.get("x-request-id") == "req-stream-123" + assert response.headers.get("anthropic-ratelimit-requests-remaining") == "99" + + def test_streaming_2xx_strips_dangerous_headers(self) -> None: + streaming_client = self._make_streaming_client_with_headers( + status_code=200, + content=b"data: {}\n\ndata: [DONE]\n\n", + headers={ + "content-type": "text/event-stream", + "set-cookie": "session=evil", + "server": "nginx", + "transfer-encoding": "chunked", + }, + ) + app = FastAPI() + app.include_router(router) + app.dependency_overrides[verify_passthrough_token] = lambda: "tok" + app.dependency_overrides[verify_strict_client_key] = lambda: "tok" + app.state.passthrough_buffered_client = _make_buffered_client() + app.state.passthrough_streaming_client = streaming_client + + with patch("luthien_proxy.passthrough_routes.create_recorder") as mock_create: + mock_create.return_value = MagicMock(spec=NoOpRequestLogRecorder) + with patch.dict("os.environ", {"ANTHROPIC_BASE_URL": "http://mock-upstream"}): + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/anthropic/v1/messages", + json={"model": "claude-haiku-4-5", "max_tokens": 10, "messages": [], "stream": True}, + headers={"anthropic-version": "2023-06-01"}, + ) + + assert "set-cookie" not in response.headers + assert "server" not in response.headers + assert "transfer-encoding" not in response.headers + + +class TestWriteLogsSkipsEmptyOutbound: + @pytest.mark.asyncio + async def test_only_inbound_row_written_for_passthrough(self) -> None: + from luthien_proxy.request_log.recorder import RequestLogRecorder, _PendingLog + + recorder = RequestLogRecorder.__new__(RequestLogRecorder) + recorder._transaction_id = "test-txn" + recorder._inbound = _PendingLog( + direction="inbound", + transaction_id="test-txn", + http_method="POST", + url="http://gateway/openai/v1/chat/completions", + ) + recorder._outbound = _PendingLog( + direction="outbound", + transaction_id="test-txn", + ) + + insert_calls: list[str] = [] + + async def fake_insert(conn, pending, serialize): + insert_calls.append(pending.direction) + + with patch("luthien_proxy.request_log.recorder._insert_log_row", side_effect=fake_insert): + mock_pool = MagicMock() + mock_conn = AsyncMock() + mock_pool.connection = MagicMock( + return_value=AsyncMock( + __aenter__=AsyncMock(return_value=mock_conn), + __aexit__=AsyncMock(return_value=None), + ) + ) + recorder._db_pool = mock_pool + await recorder._write_logs() + + assert insert_calls == ["inbound"], f"Expected only inbound row, got: {insert_calls}" + + +class TestIsStreaming: + def test_empty_body_is_not_streaming(self) -> None: + from luthien_proxy.passthrough_routes import _is_streaming + + assert _is_streaming("v1/chat/completions", b"") is False + + def test_non_json_body_is_not_streaming(self) -> None: + from luthien_proxy.passthrough_routes import _is_streaming + + assert _is_streaming("v1/chat/completions", b"not json at all") is False + + def test_stream_true_bool_is_streaming(self) -> None: + import json + + from luthien_proxy.passthrough_routes import _is_streaming + + body = json.dumps({"model": "gpt-4o", "stream": True}).encode() + assert _is_streaming("v1/chat/completions", body) is True + + def test_stream_false_bool_is_not_streaming(self) -> None: + import json + + from luthien_proxy.passthrough_routes import _is_streaming + + body = json.dumps({"model": "gpt-4o", "stream": False}).encode() + assert _is_streaming("v1/chat/completions", body) is False + + def test_stream_string_true_is_not_streaming(self) -> None: + import json + + from luthien_proxy.passthrough_routes import _is_streaming + + body = json.dumps({"model": "gpt-4o", "stream": "true"}).encode() + assert _is_streaming("v1/chat/completions", body) is False + + def test_stream_generate_content_path_is_streaming(self) -> None: + from luthien_proxy.passthrough_routes import _is_streaming + + assert _is_streaming("v1beta/models/gemini-1.5-flash:streamGenerateContent", b"{}") is True + + def test_no_stream_key_is_not_streaming(self) -> None: + import json + + from luthien_proxy.passthrough_routes import _is_streaming + + body = json.dumps({"model": "gpt-4o", "messages": []}).encode() + assert _is_streaming("v1/chat/completions", body) is False diff --git a/uv.lock b/uv.lock index 2d0ec8150..e7badadc2 100644 --- a/uv.lock +++ b/uv.lock @@ -1042,6 +1042,7 @@ dependencies = [ { name = "click" }, { name = "cryptography" }, { name = "httpx" }, + { name = "httpx-sse" }, { name = "jsonschema" }, { name = "litellm", extra = ["proxy"] }, { name = "opentelemetry-api" }, @@ -1084,6 +1085,7 @@ requires-dist = [ { name = "click", specifier = ">=8.1.0" }, { name = "cryptography", specifier = ">=44.0.0" }, { name = "httpx", specifier = ">=0.28.1" }, + { name = "httpx-sse", specifier = ">=0.4" }, { name = "jsonschema", specifier = ">=4.17.0" }, { name = "litellm", extras = ["proxy"], specifier = ">=1.81.0,!=1.82.7,!=1.82.8" }, { name = "opentelemetry-api", specifier = ">=1.20.0" },