diff --git a/litellm/proxy/control_plane/app.py b/litellm/proxy/control_plane/app.py index a01ba31b6b8..3236a83fef5 100644 --- a/litellm/proxy/control_plane/app.py +++ b/litellm/proxy/control_plane/app.py @@ -6,6 +6,8 @@ from typing import Final, Protocol from fastapi import FastAPI from starlette.types import ASGIApp, Message, Receive, Scope, Send +from litellm.proxy.control_plane.audit.dependencies import get_audit_log +from litellm.proxy.control_plane.audit.router import router as audit_router from litellm.proxy.control_plane.auth.dependencies import get_token_verifier from litellm.proxy.control_plane.auth.router import get_login_service from litellm.proxy.control_plane.auth.router import router as auth_router @@ -24,6 +26,7 @@ from litellm.proxy.control_plane.dataplanes.dependencies import ( from litellm.proxy.control_plane.dataplanes.router import router as dataplanes_router from litellm.proxy.control_plane.fleet.router import router as fleet_router from litellm.proxy.control_plane.health.router import router as health_router +from litellm.proxy.control_plane.proxy.router import router as proxy_router from litellm.proxy.control_plane.usage.router import router as usage_router from litellm.proxy.middleware.request_size_limit_middleware import RequestSizeLimitMiddleware @@ -92,10 +95,13 @@ def create_app(container: Container, license_check: License) -> FastAPI: app.dependency_overrides[get_control_plane_settings] = lambda: container.settings app.dependency_overrides[get_dataplane_client] = lambda: container.dataplane_client app.dependency_overrides[get_fanout] = lambda: container.fanout + app.dependency_overrides[get_audit_log] = lambda: container.audit_log install_exception_handlers(app) app.include_router(health_router) app.include_router(auth_router, prefix=container.settings.api_prefix) app.include_router(dataplanes_router, prefix=container.settings.api_prefix) app.include_router(usage_router, prefix=container.settings.api_prefix) app.include_router(fleet_router, prefix=container.settings.api_prefix) + app.include_router(proxy_router, prefix=container.settings.api_prefix) + app.include_router(audit_router, prefix=container.settings.api_prefix) return app diff --git a/litellm/proxy/control_plane/audit/__init__.py b/litellm/proxy/control_plane/audit/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/proxy/control_plane/audit/dependencies.py b/litellm/proxy/control_plane/audit/dependencies.py new file mode 100644 index 00000000000..c260c4a65b3 --- /dev/null +++ b/litellm/proxy/control_plane/audit/dependencies.py @@ -0,0 +1,5 @@ +from litellm.proxy.control_plane.audit.log import AuditLog + + +def get_audit_log() -> AuditLog: + raise RuntimeError("get_audit_log is not bound, build the app with create_app") diff --git a/litellm/proxy/control_plane/audit/log.py b/litellm/proxy/control_plane/audit/log.py new file mode 100644 index 00000000000..63605af048e --- /dev/null +++ b/litellm/proxy/control_plane/audit/log.py @@ -0,0 +1,105 @@ +"""Who did what on which dataplane through the control plane. Bounded in memory plus a structured log line.""" + +import json +from collections import deque +from collections.abc import Callable +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from functools import reduce +from typing import Final, Literal, TypeAlias +from urllib.parse import unquote + +from litellm._logging import verbose_proxy_logger + +Decision: TypeAlias = Literal["allowed", "denied", "unreachable"] +DEFAULT_CAPACITY: Final = 10_000 +KEY_PREFIX: Final = "sk-" +REDACTED_KEY: Final = "sk-[redacted]" +REDACTION_DECODE_ROUNDS: Final = 8 +FINAL_KEY_ACTIONS: Final = frozenset({"reset_spend", "regenerate"}) + + +def _decode_segment(segment: str, remaining: int = REDACTION_DECODE_ROUNDS) -> str: + return reduce(lambda decoded, _: unquote(decoded), range(remaining), segment) + + +def redact_path(path: str) -> str: + """A virtual key in a path (``/key/sk-.../reset_spend``) must never reach the log or the audit API.""" + segments: Final = path.split("/") + key_index: Final = next( + (index for index, segment in enumerate(segments) if _decode_segment(segment).lower().startswith(KEY_PREFIX)), + None, + ) + if key_index is None: + return path + final_action: Final = segments[-1] if segments[-1] in FINAL_KEY_ACTIONS else None + return "/".join((*segments[:key_index], REDACTED_KEY, *((final_action,) if final_action is not None else ()))) + + +@dataclass(frozen=True, slots=True) +class AuditEvent: + at: datetime + actor: str + actor_email: str | None + dataplane_id: str + method: str + path: str + decision: Decision + status: int | None + credential: str | None + detail: str | None + + +def utc_now() -> datetime: + return datetime.now(UTC) + + +@dataclass(frozen=True, slots=True) +class AuditLog: + events: deque[AuditEvent] # mutable-ok: a bounded ring buffer of events is the whole point + now: Callable[[], datetime] = utc_now + + def record( + self, + actor: str, + actor_email: str | None, + dataplane_id: str, + method: str, + path: str, + decision: Decision, + status: int | None = None, + credential: str | None = None, + detail: str | None = None, + ) -> AuditEvent: + event: Final = AuditEvent( + at=self.now(), + actor=actor, + actor_email=actor_email, + dataplane_id=dataplane_id, + method=method, + path=redact_path(path), + decision=decision, + status=status, + credential=credential, + detail=detail, + ) + self.events.append(event) + verbose_proxy_logger.info("control_plane.audit %s", json.dumps(asdict(event), default=str)) + return event + + def recent( + self, limit: int, dataplane_ids: frozenset[str] | None = None, actor: str | None = None + ) -> tuple[AuditEvent, ...]: + """``dataplane_ids`` of ``None`` means every dataplane, an empty set means nothing is visible.""" + newest_first: Final = reversed(self.events) + matching: Final = ( + event + for event in newest_first + if (dataplane_ids is None or event.dataplane_id in dataplane_ids) + and (actor is None or event.actor == actor) + ) + return tuple(event for _, event in zip(range(limit), matching, strict=False)) + + +def build_audit_log(capacity: int = DEFAULT_CAPACITY, now: Callable[[], datetime] = utc_now) -> AuditLog: + return AuditLog(events=deque(maxlen=capacity), now=now) diff --git a/litellm/proxy/control_plane/audit/router.py b/litellm/proxy/control_plane/audit/router.py new file mode 100644 index 00000000000..06b2fb0bc8c --- /dev/null +++ b/litellm/proxy/control_plane/audit/router.py @@ -0,0 +1,50 @@ +from typing import Annotated, Final + +from fastapi import APIRouter, Depends, Query, Security + +from litellm.proxy.control_plane.audit.dependencies import get_audit_log +from litellm.proxy.control_plane.audit.log import AuditLog +from litellm.proxy.control_plane.auth.dependencies import require_scopes +from litellm.proxy.control_plane.auth.rbac import DataplaneBinding, Principal, Scope +from litellm.proxy.control_plane.common.exceptions import ControlPlaneException, forbidden +from litellm.types.proxy.control_plane.audit import AuditEventResponse, AuditListResponse + +router: Final = APIRouter(prefix="/audit", tags=["audit"]) + + +def visible_dataplanes(binding: DataplaneBinding, requested: str | None) -> frozenset[str] | None: + """What this caller may read: everything when fleet-wide, else only bound ids, and a requested id must be bound.""" + if requested is not None and not (binding.all or requested in binding.ids): + raise ControlPlaneException(forbidden(f"Dataplane {requested!r} is not bound to this caller.")) + if requested is not None: + return frozenset({requested}) + return None if binding.all else binding.ids + + +@router.get("", response_model=AuditListResponse) +async def list_audit( + principal: Annotated[Principal, Security(require_scopes, scopes=[Scope.AUDIT_READ.value])], + audit: Annotated[AuditLog, Depends(get_audit_log)], + limit: Annotated[int, Query(ge=1, le=1000)] = 100, + dataplane: Annotated[str | None, Query(description="Only events against this dataplane id.")] = None, + actor: Annotated[str | None, Query(description="Only events by this control plane subject.")] = None, +) -> AuditListResponse: + """Newest first, what every control plane caller did on the dataplanes this caller is bound to.""" + visible: Final = visible_dataplanes(principal.dataplanes, dataplane) + return AuditListResponse( + items=tuple( + AuditEventResponse( + at=event.at, + actor=event.actor, + actor_email=event.actor_email, + dataplane_id=event.dataplane_id, + method=event.method, + path=event.path, + decision=event.decision, + status=event.status, + credential=event.credential, + detail=event.detail, + ) + for event in audit.recent(limit, visible, actor) + ) + ) diff --git a/litellm/proxy/control_plane/common/fanout.py b/litellm/proxy/control_plane/common/fanout.py index bde3624a201..dc527dd78dd 100644 --- a/litellm/proxy/control_plane/common/fanout.py +++ b/litellm/proxy/control_plane/common/fanout.py @@ -7,7 +7,12 @@ from typing import Final, Generic, TypeVar from pydantic import TypeAdapter, ValidationError -from litellm.proxy.control_plane.dataplanes.client import CallResult, DataplaneReply, DataplaneUnreachable +from litellm.proxy.control_plane.dataplanes.client import ( + CallResult, + DataplaneReply, + DataplaneResponseTooLarge, + DataplaneUnreachable, +) from litellm.proxy.control_plane.dataplanes.credentials import CredentialUnavailable from litellm.proxy.control_plane.dataplanes.registry import Dataplane @@ -82,6 +87,8 @@ def parse_reply(result: CallResult, shape: TypeAdapter[M]) -> M | Failure: return Failure(f"HTTP {status} with the {credential} credential") case DataplaneUnreachable(detail=detail, attempts=attempts): return Failure(f"unreachable after {attempts} attempt(s): {detail}") + case DataplaneResponseTooLarge(limit_bytes=limit_bytes): + return Failure(f"response exceeded {limit_bytes} bytes") case CredentialUnavailable(reason=reason): return Failure(f"no usable credential: {reason}") diff --git a/litellm/proxy/control_plane/container.py b/litellm/proxy/control_plane/container.py index e73212c2b19..268506d1dc0 100644 --- a/litellm/proxy/control_plane/container.py +++ b/litellm/proxy/control_plane/container.py @@ -2,6 +2,7 @@ import time from collections.abc import Mapping, Sequence +from contextlib import AbstractAsyncContextManager from dataclasses import dataclass from datetime import timedelta from types import MappingProxyType @@ -13,6 +14,7 @@ from litellm.llms.custom_httpx.http_handler import ( AsyncHTTPHandler, get_async_httpx_client, # pyright: ignore[reportUnknownVariableType] # params is an untyped dict upstream ) +from litellm.proxy.control_plane.audit.log import AuditLog, build_audit_log from litellm.proxy.control_plane.auth.jwt import JwksKeyLookup, SessionTokenCodec, TokenVerifier, build_token_verifier from litellm.proxy.control_plane.auth.login import LoginService, LoginThrottle from litellm.proxy.control_plane.auth.rbac import ( @@ -41,6 +43,7 @@ class Container: registry: DataplaneRegistry dataplane_client: DataplaneClient fanout: Fanout + audit_log: AuditLog class InvalidRolePolicy(ValueError): @@ -85,7 +88,7 @@ class HandlerTransport: handler: AsyncHTTPHandler - async def request( + def stream( self, method: str, url: str, @@ -94,9 +97,9 @@ class HandlerTransport: json: dict[str, object] | None = None, # mutable-ok: httpx takes dict headers: dict[str, str] | None = None, # mutable-ok: httpx takes dict timeout: float, - ) -> httpx.Response: + ) -> AbstractAsyncContextManager[httpx.Response]: httpx_params: Final = tuple(params) if params is not None else None - return await self.handler.client.request( + return self.handler.client.stream( method, url, params=httpx_params, json=json, headers=headers, timeout=timeout, follow_redirects=False ) @@ -146,4 +149,5 @@ def build_container( retry=RetryPolicy(attempts=settings.dataplane_retry_attempts), ), fanout=Fanout(concurrency=settings.fanout_concurrency), + audit_log=build_audit_log(settings.audit_capacity), ) diff --git a/litellm/proxy/control_plane/dataplanes/client.py b/litellm/proxy/control_plane/dataplanes/client.py index c662fadffbf..2bed1c12a53 100644 --- a/litellm/proxy/control_plane/dataplanes/client.py +++ b/litellm/proxy/control_plane/dataplanes/client.py @@ -6,7 +6,8 @@ a management passthrough can tolerate. Retrying is this module's job and only ev import asyncio import random -from collections.abc import Awaitable, Callable, Mapping, Sequence +from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence +from contextlib import AbstractAsyncContextManager, AsyncExitStack from dataclasses import dataclass from typing import Final, Literal, Protocol, TypeAlias @@ -17,9 +18,12 @@ from litellm._logging import verbose_proxy_logger from litellm.proxy.control_plane.dataplanes.credentials import Caller, CredentialKind, CredentialUnavailable from litellm.proxy.control_plane.dataplanes.registry import Dataplane -Method: TypeAlias = Literal["GET", "POST", "PUT", "PATCH", "DELETE"] +Method: TypeAlias = Literal["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE"] +WRITE_METHODS: Final = frozenset({"POST", "PUT", "PATCH", "DELETE"}) Sleeper: TypeAlias = Callable[[float], Awaitable[None]] RETRYABLE_STATUSES: Final = frozenset({502, 503, 504}) +DEFAULT_MAX_RESPONSE_BYTES: Final = 64 * 1024 * 1024 +_STREAM_CHUNK_BYTES: Final = 1024 * 1024 _JSON: Final = TypeAdapter(object) @@ -27,7 +31,7 @@ class HttpTransport(Protocol): """The slice of ``httpx.AsyncClient.request`` the client needs, so tests inject a fake. Raises only httpx errors and never retries on its own.""" - async def request( + def stream( self, method: str, url: str, @@ -36,7 +40,7 @@ class HttpTransport(Protocol): json: dict[str, object] | None = None, # mutable-ok: httpx takes dict headers: dict[str, str] | None = None, # mutable-ok: httpx takes dict timeout: float, - ) -> httpx.Response: ... + ) -> AbstractAsyncContextManager[httpx.Response]: ... @dataclass(frozen=True, slots=True) @@ -44,6 +48,10 @@ class DataplaneReply: status: int body: object credential: CredentialKind + raw_content: bytes | None = None + content_type: str | None = None + content_disposition: str | None = None + cache_control: str | None = None @dataclass(frozen=True, slots=True) @@ -52,13 +60,22 @@ class DataplaneUnreachable: attempts: int -CallResult: TypeAlias = DataplaneReply | DataplaneUnreachable | CredentialUnavailable +@dataclass(frozen=True, slots=True) +class DataplaneResponseTooLarge: + limit_bytes: int + + +CallResult: TypeAlias = DataplaneReply | DataplaneUnreachable | DataplaneResponseTooLarge | CredentialUnavailable @dataclass(frozen=True, slots=True) class _RawReply: status: int body: object + raw_content: bytes | None + content_type: str | None + content_disposition: str | None + cache_control: str | None @dataclass(frozen=True, slots=True) @@ -72,17 +89,46 @@ class RetryPolicy: return capped * (0.5 + jitter / 2) -def _parse_body(response: httpx.Response) -> object: +def _parse_body(content: bytes, response: httpx.Response) -> tuple[object, bool]: try: - return _JSON.validate_python(response.json()) + return _JSON.validate_json(content), True except ValueError: - return response.text + return content.decode(response.charset_encoding or "utf-8", errors="replace"), False + + +async def _read_chunks( + chunks: AsyncIterator[bytes], + limit_bytes: int, + collected: tuple[bytes, ...], + total_bytes: int, +) -> bytes | None: + try: + chunk: Final = await anext(chunks) + except StopAsyncIteration: + return b"".join(collected) + next_total: Final = total_bytes + len(chunk) + if next_total > limit_bytes: + return None + return await _read_chunks(chunks, limit_bytes, (*collected, chunk), next_total) + + +async def _read_bounded(response: httpx.Response, limit_bytes: int) -> bytes | None: + chunks: Final = response.aiter_bytes(chunk_size=_STREAM_CHUNK_BYTES).__aiter__() + return await _read_chunks(chunks, limit_bytes, (), 0) + + +async def _close_quietly(stack: AsyncExitStack) -> None: + try: + await stack.aclose() + except httpx.HTTPError as error: + verbose_proxy_logger.debug("Control-plane dataplane stream teardown failed: %s", error) @dataclass(frozen=True, slots=True) class DataplaneClient: transport: HttpTransport timeout_seconds: float + max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES retry: RetryPolicy = RetryPolicy() sleep: Sleeper = asyncio.sleep jitter: Callable[[], float] = random.random @@ -99,6 +145,18 @@ class DataplaneClient: async def post(self, dataplane: Dataplane, path: str, caller: Caller, json: Mapping[str, object]) -> CallResult: return await self._call("POST", dataplane, path, caller, params=None, json=json) + async def request( + self, + method: Method, + dataplane: Dataplane, + path: str, + caller: Caller, + params: Sequence[tuple[str, str]] | None = None, + json: Mapping[str, object] | None = None, + ) -> CallResult: + """Any management verb. Only GET is retried, every write goes out exactly once.""" + return await self._call(method, dataplane, path, caller, params=params, json=json) + async def _call( self, method: Method, @@ -116,10 +174,20 @@ class DataplaneClient: outcome: Final = await self._attempt( method, url, credential.headers, params, json, attempt=0, attempts=attempts ) + if isinstance(outcome, DataplaneResponseTooLarge): + return outcome return ( outcome if isinstance(outcome, DataplaneUnreachable) - else DataplaneReply(outcome.status, outcome.body, credential.kind) + else DataplaneReply( + outcome.status, + outcome.body, + credential.kind, + outcome.raw_content, + outcome.content_type, + outcome.content_disposition, + outcome.cache_control, + ) ) async def _attempt( @@ -131,8 +199,10 @@ class DataplaneClient: json: Mapping[str, object] | None, attempt: int, attempts: int, - ) -> _RawReply | DataplaneUnreachable: + ) -> _RawReply | DataplaneUnreachable | DataplaneResponseTooLarge: outcome: Final = await self._send(method, url, headers, params, json) + if isinstance(outcome, DataplaneResponseTooLarge): + return outcome last: Final = attempt + 1 >= attempts if last or not _retryable(outcome): return outcome if isinstance(outcome, _RawReply) else DataplaneUnreachable(outcome.detail, attempt + 1) @@ -146,25 +216,50 @@ class DataplaneClient: headers: Mapping[str, str], params: Sequence[tuple[str, str]] | None, json: Mapping[str, object] | None, - ) -> _RawReply | DataplaneUnreachable: - try: - response: Final = await self.transport.request( - method, - url, - params=params, - json=None if json is None else dict(json), - headers=dict(headers), - timeout=self.timeout_seconds, + ) -> _RawReply | DataplaneUnreachable | DataplaneResponseTooLarge: + async with AsyncExitStack() as stack: + try: + response: Final = await stack.enter_async_context( + self.transport.stream( + method, + url, + params=params, + json=None if json is None else dict(json), + headers=dict(headers), + timeout=self.timeout_seconds, + ) + ) + content: Final = await _read_bounded(response, self.max_response_bytes) + except httpx.HTTPError as error: + verbose_proxy_logger.debug("Control-plane dataplane transport failed: %s", error) + await _close_quietly(stack) + return DataplaneUnreachable(_describe(error), attempts=1) + content_type: Final = response.headers.get("content-type") + is_json: Final = content_type is not None and "json" in content_type.lower() + body, parsed_as_json = _parse_body(content, response) if content is not None else ("", False) + parsed_json_response: Final = is_json and parsed_as_json + outcome: Final = ( + DataplaneResponseTooLarge(self.max_response_bytes) + if content is None + else _RawReply( + response.status_code, + body, + None if parsed_json_response else content, + None if parsed_json_response else content_type, + response.headers.get("content-disposition"), + response.headers.get("cache-control"), + ) ) - except httpx.HTTPError as error: - verbose_proxy_logger.debug("Control-plane dataplane transport failed: %s", error) - return DataplaneUnreachable(_describe(error), attempts=1) - return _RawReply(response.status_code, _parse_body(response)) + deferred_exit: Final = stack.pop_all() + await _close_quietly(deferred_exit) + return outcome def _describe(error: httpx.HTTPError) -> str: return type(error).__name__ -def _retryable(outcome: _RawReply | DataplaneUnreachable) -> bool: - return isinstance(outcome, DataplaneUnreachable) or outcome.status in RETRYABLE_STATUSES +def _retryable(outcome: _RawReply | DataplaneUnreachable | DataplaneResponseTooLarge) -> bool: + return isinstance(outcome, DataplaneUnreachable) or ( + isinstance(outcome, _RawReply) and outcome.status in RETRYABLE_STATUSES + ) diff --git a/litellm/proxy/control_plane/dataplanes/router.py b/litellm/proxy/control_plane/dataplanes/router.py index d23928cac64..b1a95405579 100644 --- a/litellm/proxy/control_plane/dataplanes/router.py +++ b/litellm/proxy/control_plane/dataplanes/router.py @@ -6,7 +6,12 @@ from fastapi import APIRouter, Depends, Security from litellm.proxy.control_plane.auth.dependencies import require_scopes from litellm.proxy.control_plane.auth.rbac import Principal, Scope from litellm.proxy.control_plane.common.exceptions import ControlPlaneException, bad_gateway -from litellm.proxy.control_plane.dataplanes.client import DataplaneClient, DataplaneReply, DataplaneUnreachable +from litellm.proxy.control_plane.dataplanes.client import ( + DataplaneClient, + DataplaneReply, + DataplaneResponseTooLarge, + DataplaneUnreachable, +) from litellm.proxy.control_plane.dataplanes.credentials import Caller, CredentialUnavailable from litellm.proxy.control_plane.dataplanes.dependencies import ( dataplane_access, @@ -65,6 +70,10 @@ async def dataplane_health( return DataplaneHealthResponse(id=dataplane.id, status="degraded", http_status=status, detail=str(body)) case DataplaneUnreachable(detail=detail): return DataplaneHealthResponse(id=dataplane.id, status="unreachable", http_status=None, detail=detail) + case DataplaneResponseTooLarge(limit_bytes=limit_bytes): + return DataplaneHealthResponse( + id=dataplane.id, status="degraded", http_status=None, detail=f"response exceeded {limit_bytes} bytes" + ) case CredentialUnavailable(reason=reason): return DataplaneHealthResponse(id=dataplane.id, status="degraded", http_status=None, detail=reason) @@ -101,5 +110,7 @@ async def dataplane_whoami( ) case DataplaneUnreachable(detail=detail, attempts=attempts): raise ControlPlaneException(bad_gateway(f"Dataplane unreachable after {attempts} attempt(s): {detail}")) + case DataplaneResponseTooLarge(limit_bytes=limit_bytes): + raise ControlPlaneException(bad_gateway(f"Dataplane response exceeded the {limit_bytes}-byte limit.")) case CredentialUnavailable(reason=reason): raise ControlPlaneException(bad_gateway(f"No usable credential for this dataplane: {reason}")) diff --git a/litellm/proxy/control_plane/proxy/__init__.py b/litellm/proxy/control_plane/proxy/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/proxy/control_plane/proxy/allowlist.py b/litellm/proxy/control_plane/proxy/allowlist.py new file mode 100644 index 00000000000..6f4026bb794 --- /dev/null +++ b/litellm/proxy/control_plane/proxy/allowlist.py @@ -0,0 +1,110 @@ +"""Which dataplane routes the passthrough may carry. Management only, never auth bootstrap or inference.""" + +from dataclasses import dataclass +from typing import Final +from urllib.parse import unquote + +MANAGEMENT_PREFIXES: Final = ( + "/key/", + "/v2/key/", + "/user/", + "/v2/user/", + "/team/", + "/v2/team/", + "/organization/", + "/v2/organization/", + "/customer/", + "/end_user/", + "/budget/", + "/tag/", + "/model/", + "/v1/model/info", + "/v2/model/", + "/model_group/", + "/model_access_group/", + "/access_group/", + "/credentials", + "/credential", + "/guardrails/", + "/v2/guardrails/", + "/v1/mcp/server", + "/vector_store/", + "/spend/", + "/global/spend", + "/user/daily/activity", + "/team/daily/activity", + "/tag/daily/activity", + "/audit", + "/management/v1/", + "/active/callbacks", + "/callbacks", + "/alerting/", + "/router/settings", + "/settings", +) + +HEALTH_READ_ROUTES: Final = frozenset({"/health/readiness", "/health/liveliness", "/health/liveness", "/health/latest"}) + +DENIED_PREFIXES: Final = ( + "/health/drain", + "/login", + "/logout", + "/sso/", + "/token", + "/jwt/", + "/oauth/", + "/invitation/", + "/onboarding/", + "/session/", + "/key/service-account/generate", + "/guardrails/apply_guardrail", + "/config/", + "/debug/", + "/utils/", +) + + +@dataclass(frozen=True, slots=True) +class Allowed: + path: str + + +@dataclass(frozen=True, slots=True) +class Denied: + path: str + reason: str + + +def _matches(path: str, prefix: str) -> bool: + """Route boundary aware: ``/audit`` matches ``/audit`` and ``/audit/x`` but never ``/auditor``.""" + base: Final = prefix.rstrip("/") + return path == base or path.startswith(base + "/") + + +def _normalize(raw_path: str) -> str | None: + """Decode once and refuse anything still encoded, so the path we check is the path httpx sends.""" + decoded: Final = unquote(raw_path) + if any(reserved in decoded for reserved in ("%", "?", "#", "\\")): + return None + path: Final = "/" + decoded.lstrip("/") + return path[:-1] if path != "/" and path.endswith("/") and not path.endswith("//") else path + + +def decide(raw_path: str) -> Allowed | Denied: + path: Final = _normalize(raw_path) + if path is None: + return Denied("/" + raw_path.lstrip("/"), "encoded or reserved path characters are not allowed") + segments: Final = path.split("/")[1:] + if any(segment in {".", ".."} for segment in segments) or "" in segments[:-1]: + return Denied(path, "path traversal is not allowed") + if any(_matches(path, prefix) for prefix in DENIED_PREFIXES): + return Denied( + path, "auth bootstrap, debug, lifecycle and guardrail apply routes never go through the control plane" + ) + if _matches(path, "/v1/mcp/server") and "oauth" in segments: + return Denied(path, "OAuth routes never go through the control plane") + if path in HEALTH_READ_ROUTES: + return Allowed(path) + if not any(_matches(path, prefix) for prefix in MANAGEMENT_PREFIXES): + return Denied(path, "only management routes are proxied, inference goes straight to the dataplane") + return Allowed(path) diff --git a/litellm/proxy/control_plane/proxy/router.py b/litellm/proxy/control_plane/proxy/router.py new file mode 100644 index 00000000000..d4dcb40b252 --- /dev/null +++ b/litellm/proxy/control_plane/proxy/router.py @@ -0,0 +1,253 @@ +"""Authenticated management passthrough: one allowlisted dataplane route, as the caller, audited.""" + +from collections.abc import Mapping +from typing import Annotated, Final + +from fastapi import APIRouter, Depends, Request, Security +from fastapi.responses import JSONResponse, Response +from fastapi.security import SecurityScopes +from pydantic import TypeAdapter, ValidationError + +from litellm.proxy.control_plane.audit.dependencies import get_audit_log +from litellm.proxy.control_plane.audit.log import AuditLog, redact_path +from litellm.proxy.control_plane.auth.dependencies import get_principal, require_scopes +from litellm.proxy.control_plane.auth.rbac import Principal, Scope +from litellm.proxy.control_plane.common.exceptions import ControlPlaneException, bad_gateway, error_body, forbidden +from litellm.proxy.control_plane.dataplanes.client import ( + DataplaneClient, + DataplaneReply, + DataplaneResponseTooLarge, + DataplaneUnreachable, + Method, +) +from litellm.proxy.control_plane.dataplanes.credentials import Caller, CredentialUnavailable +from litellm.proxy.control_plane.dataplanes.dependencies import ( + dataplane_access, + get_caller, + get_dataplane_client, + get_registry, +) +from litellm.proxy.control_plane.dataplanes.registry import Dataplane, DataplaneRegistry +from litellm.proxy.control_plane.proxy.allowlist import Denied, decide + +DATAPLANE_HEADER: Final = "X-LiteLLM-Dataplane" +READ_METHODS: Final = ("GET", "HEAD") +WRITE_METHODS: Final = ("POST", "PUT", "PATCH", "DELETE") + +router: Final = APIRouter(tags=["proxy"]) + +_BODY: Final = TypeAdapter(Mapping[str, object]) +_METHOD: Final[TypeAdapter[Method]] = TypeAdapter(Method) + + +def bad_request(detail: str) -> ControlPlaneException: + return ControlPlaneException(error_body("bad-request", "Bad Request", 400, detail)) + + +async def _json_body(request: Request) -> Mapping[str, object] | None: + raw: Final = await request.body() + if not raw: + return None + try: + return _BODY.validate_json(raw) + except ValidationError: + raise bad_request("Request body must be a JSON object.") from None + + +async def _forward( + request: Request, + path: str, + principal: Principal, + dataplane: Dataplane, + caller: Caller, + client: DataplaneClient, + audit: AuditLog, +) -> Response: + method: Final = _METHOD.validate_python(request.method) + decision: Final = decide(path) + if isinstance(decision, Denied): + audit.record( + principal.subject, principal.email, dataplane.id, method, decision.path, "denied", detail=decision.reason + ) + raise ControlPlaneException(forbidden(f"{redact_path(decision.path)}: {decision.reason}.")) + result: Final = await client.request( + method, + dataplane, + decision.path, + caller, + params=tuple(request.query_params.multi_items()), + json=await _body_or_audit(request, principal, dataplane, method, decision.path, audit), + ) + match result: + case DataplaneReply(status=status, credential=credential) if 300 <= status < 400: + audit.record( + principal.subject, + principal.email, + dataplane.id, + method, + decision.path, + "denied", + status, + credential, + detail="dataplane answered with a redirect, which the control plane never follows", + ) + raise ControlPlaneException(bad_gateway("Dataplane answered with a redirect, which is not followed.")) + case DataplaneReply( + status=status, + body=body, + credential=credential, + raw_content=raw_content, + content_type=content_type, + content_disposition=content_disposition, + cache_control=cache_control, + ): + audit.record( + principal.subject, principal.email, dataplane.id, method, decision.path, "allowed", status, credential + ) + if raw_content is not None: + headers: Final = { + **({"Content-Type": content_type} if content_type is not None else {}), + **({"Content-Disposition": content_disposition} if content_disposition is not None else {}), + **({"Cache-Control": cache_control} if cache_control is not None else {}), + DATAPLANE_HEADER: dataplane.id, + } + return Response(content=raw_content, status_code=status, headers=headers) + return JSONResponse( + status_code=status, + content=body, + headers={ + **({"Cache-Control": cache_control} if cache_control is not None else {}), + **({"Content-Disposition": content_disposition} if content_disposition is not None else {}), + DATAPLANE_HEADER: dataplane.id, + }, + ) + case DataplaneResponseTooLarge(limit_bytes=limit_bytes): + audit.record( + principal.subject, + principal.email, + dataplane.id, + method, + decision.path, + "unreachable", + 502, + detail="dataplane response exceeded the configured byte limit", + ) + raise ControlPlaneException( + error_body( + "dataplane-response-too-large", + "Bad Gateway", + 502, + f"Dataplane response exceeded the {limit_bytes}-byte limit.", + ) + ) + case DataplaneUnreachable(detail=detail, attempts=attempts): + audit.record( + principal.subject, principal.email, dataplane.id, method, decision.path, "unreachable", detail=detail + ) + raise ControlPlaneException(bad_gateway(f"Dataplane unreachable after {attempts} attempt(s): {detail}")) + case CredentialUnavailable(reason=reason): + audit.record( + principal.subject, principal.email, dataplane.id, method, decision.path, "unreachable", detail=reason + ) + raise ControlPlaneException(bad_gateway(f"No usable credential for this dataplane: {reason}")) + + +async def _body_or_audit( + request: Request, + principal: Principal, + dataplane: Dataplane, + method: Method, + path: str, + audit: AuditLog, +) -> Mapping[str, object] | None: + try: + return await _json_body(request) + except ControlPlaneException as error: + audit.record( + principal.subject, + principal.email, + dataplane.id, + method, + path, + "denied", + error.body.status, + detail=error.body.detail, + ) + raise + + +async def _passthrough_dataplane_access( + request: Request, + dataplane_id: str, + path: str, + principal: Annotated[Principal, Depends(get_principal)], + registry: Annotated[DataplaneRegistry, Depends(get_registry)], + audit: Annotated[AuditLog, Depends(get_audit_log)], +) -> Dataplane: + try: + return await dataplane_access(dataplane_id, principal, registry) + except ControlPlaneException as error: + if error.body.status == 403: + audit.record( + principal.subject, + principal.email, + dataplane_id, + request.method, + "/" + path.lstrip("/"), + "denied", + error.body.status, + detail=error.body.detail, + ) + raise + + +async def _passthrough_scope( + security_scopes: SecurityScopes, + request: Request, + dataplane_id: str, + path: str, + principal: Annotated[Principal, Depends(get_principal)], + audit: Annotated[AuditLog, Depends(get_audit_log)], +) -> Principal: + try: + return await require_scopes(security_scopes, principal) + except ControlPlaneException as error: + audit.record( + principal.subject, + principal.email, + dataplane_id, + request.method, + "/" + path.lstrip("/"), + "denied", + error.body.status, + detail=error.body.detail, + ) + raise + + +@router.api_route("/dataplanes/{dataplane_id}/proxy/{path:path}", methods=list(READ_METHODS)) +async def proxy_read( + request: Request, + path: str, + principal: Annotated[Principal, Security(_passthrough_scope, scopes=[Scope.PROXY_READ.value])], + dataplane: Annotated[Dataplane, Depends(_passthrough_dataplane_access)], + caller: Annotated[Caller, Depends(get_caller)], + client: Annotated[DataplaneClient, Depends(get_dataplane_client)], + audit: Annotated[AuditLog, Depends(get_audit_log)], +) -> Response: + """Read one allowlisted management route on a dataplane as the caller.""" + return await _forward(request, path, principal, dataplane, caller, client, audit) + + +@router.api_route("/dataplanes/{dataplane_id}/proxy/{path:path}", methods=list(WRITE_METHODS)) +async def proxy_write( + request: Request, + path: str, + principal: Annotated[Principal, Security(_passthrough_scope, scopes=[Scope.PROXY_WRITE.value])], + dataplane: Annotated[Dataplane, Depends(_passthrough_dataplane_access)], + caller: Annotated[Caller, Depends(get_caller)], + client: Annotated[DataplaneClient, Depends(get_dataplane_client)], + audit: Annotated[AuditLog, Depends(get_audit_log)], +) -> Response: + """Write through one allowlisted management route on a dataplane as the caller, sent exactly once.""" + return await _forward(request, path, principal, dataplane, caller, client, audit) diff --git a/litellm/proxy/control_plane/settings.py b/litellm/proxy/control_plane/settings.py index 030c602b153..af158be6015 100644 --- a/litellm/proxy/control_plane/settings.py +++ b/litellm/proxy/control_plane/settings.py @@ -112,6 +112,7 @@ class ControlPlaneSettings(BaseSettings): dataplane_retry_attempts: int = 3 fanout_concurrency: int = Field(default=16, ge=1) usage_max_window_days: int = Field(default=366, ge=1) + audit_capacity: int = Field(default=10_000, ge=1) api_prefix: str = "/api/v1" @field_validator("master_key") diff --git a/litellm/types/proxy/control_plane/audit.py b/litellm/types/proxy/control_plane/audit.py new file mode 100644 index 00000000000..aab75106dd5 --- /dev/null +++ b/litellm/types/proxy/control_plane/audit.py @@ -0,0 +1,29 @@ +from datetime import datetime +from typing import Literal, TypeAlias + +from pydantic import ConfigDict + +from litellm.types.utils import LiteLLMBaseModel + +AuditDecision: TypeAlias = Literal["allowed", "denied", "unreachable"] + + +class AuditEventResponse(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + at: datetime + actor: str + actor_email: str | None + dataplane_id: str + method: str + path: str + decision: AuditDecision + status: int | None + credential: str | None + detail: str | None + + +class AuditListResponse(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True) + + items: tuple[AuditEventResponse, ...] diff --git a/tests/unit/proxy/control_plane/audit/__init__.py b/tests/unit/proxy/control_plane/audit/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/proxy/control_plane/audit/test_log.py b/tests/unit/proxy/control_plane/audit/test_log.py new file mode 100644 index 00000000000..c1c1030d25a --- /dev/null +++ b/tests/unit/proxy/control_plane/audit/test_log.py @@ -0,0 +1,79 @@ +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from itertools import count +from typing import Final + +import pytest + +from litellm.proxy.control_plane.audit.log import build_audit_log, redact_path + +T0: Final = datetime(2026, 10, 13, 12, 0, tzinfo=UTC) + + +def _ticking() -> Callable[[], datetime]: + ticks: Final = count() + return lambda: T0 + timedelta(seconds=next(ticks)) + + +def test_recent_is_newest_first_filtered_and_limited() -> None: + audit: Final = build_audit_log(capacity=10, now=_ticking()) + audit.record("alice", None, "eu", "GET", "/key/list", "allowed", 200, "service_token") + audit.record("bob", "bob@x.test", "us", "POST", "/key/generate", "allowed", 200, "oidc_exchange") + audit.record("alice", None, "us", "DELETE", "/key/delete", "denied", detail="nope") + assert [(e.actor, e.path) for e in audit.recent(10)] == [ + ("alice", "/key/delete"), + ("bob", "/key/generate"), + ("alice", "/key/list"), + ] + assert [e.path for e in audit.recent(10, dataplane_ids=frozenset({"us"}), actor="alice")] == ["/key/delete"] + assert [e.path for e in audit.recent(1)] == ["/key/delete"] + assert audit.recent(10)[0].at == T0 + timedelta(seconds=2) + + +def test_the_ring_buffer_drops_the_oldest_event_at_capacity() -> None: + audit: Final = build_audit_log(capacity=2, now=_ticking()) + for path in ("/a", "/b", "/c"): + audit.record("alice", None, "eu", "GET", path, "allowed", 200) + assert [e.path for e in audit.recent(10)] == ["/c", "/b"] + + +def test_virtual_keys_in_paths_are_redacted_before_they_are_stored_or_logged() -> None: + audit: Final = build_audit_log(capacity=10, now=_ticking()) + event: Final = audit.record( + "alice", None, "eu", "POST", "/key/sk-live-123/reset_spend", "denied", detail="no scope" + ) + assert event.path == "/key/sk-[redacted]/reset_spend" + assert audit.recent(10)[0].path == "/key/sk-[redacted]/reset_spend" + assert audit.record("alice", None, "eu", "GET", "/key/list", "allowed", 200).path == "/key/list" + + +def test_recent_with_an_empty_dataplane_set_shows_nothing_and_none_shows_everything() -> None: + audit: Final = build_audit_log(capacity=10, now=_ticking()) + audit.record("alice", None, "eu", "GET", "/key/list", "allowed", 200) + audit.record("alice", None, "us", "GET", "/key/list", "allowed", 200) + assert audit.recent(10, dataplane_ids=frozenset()) == () + assert [e.dataplane_id for e in audit.recent(10, dataplane_ids=None)] == ["us", "eu"] + assert [e.dataplane_id for e in audit.recent(10, dataplane_ids=frozenset({"eu", "us"}))] == ["us", "eu"] + + +@pytest.mark.parametrize( + "path", + [ + "/key/SK-live-123/reset_spend", + "/key/sk%2Dlive-123/reset_spend", + "/key/sk%252Dlive-123/reset_spend", + ], +) +def test_encoded_virtual_keys_are_redacted_after_repeated_decoding(path: str) -> None: + assert redact_path(path) == "/key/sk-[redacted]/reset_spend" + + +@pytest.mark.parametrize( + ("path", "expected"), + [ + ("/key/sk-live-123/team/secret/reset_spend", "/key/sk-[redacted]/reset_spend"), + ("/key/sk-live-123/team/secret", "/key/sk-[redacted]"), + ], +) +def test_virtual_key_paths_redact_all_segments_before_a_known_final_action(path: str, expected: str) -> None: + assert redact_path(path) == expected diff --git a/tests/unit/proxy/control_plane/dataplanes/test_client.py b/tests/unit/proxy/control_plane/dataplanes/test_client.py index 04117798460..83f82404526 100644 --- a/tests/unit/proxy/control_plane/dataplanes/test_client.py +++ b/tests/unit/proxy/control_plane/dataplanes/test_client.py @@ -1,4 +1,5 @@ -from collections.abc import Iterator, Sequence +from collections.abc import AsyncIterator, Iterator, Sequence +from contextlib import asynccontextmanager from dataclasses import dataclass, field from typing import Final @@ -8,6 +9,7 @@ import pytest from litellm.proxy.control_plane.dataplanes.client import ( DataplaneClient, DataplaneReply, + DataplaneResponseTooLarge, DataplaneUnreachable, RetryPolicy, ) @@ -24,8 +26,10 @@ from tests.unit.proxy.control_plane.dataplanes.conftest import caller class FakeTransport: # mutable-ok: records calls for assertions outcomes: Iterator[httpx.Response | Exception] calls: list[tuple[str, str, dict[str, str] | None]] = field(default_factory=list) + close_error: httpx.HTTPError | None = None - async def request( + @asynccontextmanager + async def stream( self, method: str, url: str, @@ -34,12 +38,23 @@ class FakeTransport: # mutable-ok: records calls for assertions json: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> httpx.Response: + ) -> AsyncIterator[httpx.Response]: self.calls.append((method, url, headers)) outcome: Final = next(self.outcomes) if isinstance(outcome, Exception): raise outcome - return outcome + yield outcome + if self.close_error is not None: + raise self.close_error + + +class _ReadErrorStream(httpx.AsyncByteStream): + async def __aiter__(self) -> AsyncIterator[bytes]: + yield b'{"partial":' + raise httpx.ReadError("failed while reading") + + async def aclose(self) -> None: + return None @dataclass @@ -54,15 +69,25 @@ def _response(status: int, body: object = None) -> httpx.Response: return httpx.Response(status, json=body, request=httpx.Request("GET", "https://eu.test/x")) +def _read_error_response() -> httpx.Response: + return httpx.Response( + 200, + headers={"Content-Type": "application/json"}, + stream=_ReadErrorStream(), + request=httpx.Request("GET", "https://eu.test/user/info"), + ) + + def _dataplane(token: str | None = "sk-svc") -> Dataplane: chain: Final = CredentialChain(() if token is None else (ServiceTokenImpersonation(token),)) return Dataplane(id="eu", name="eu", url="https://eu.test", credential=chain, credential_kinds=chain.kinds) -def _client(transport: FakeTransport, clock: Clock) -> DataplaneClient: +def _client(transport: FakeTransport, clock: Clock, max_response_bytes: int = 64 * 1024 * 1024) -> DataplaneClient: return DataplaneClient( transport, timeout_seconds=1.0, + max_response_bytes=max_response_bytes, retry=RetryPolicy(attempts=3, base_delay_seconds=0.1), sleep=clock.sleep, jitter=lambda: 1.0, @@ -84,6 +109,84 @@ async def test_get_injects_the_credential_and_parses_json() -> None: assert clock.slept == [] +@pytest.mark.asyncio +async def test_streamed_non_json_body_without_charset_is_forwarded_as_text() -> None: + response: Final = httpx.Response( + 200, + headers={"Content-Type": "text/plain"}, + stream=httpx.ByteStream(b"upstream text"), + request=httpx.Request("GET", "https://eu.test/user/info"), + ) + transport: Final = FakeTransport(iter([response])) + + reply: Final = await _client(transport, Clock()).get(_dataplane(), "/user/info", caller("session")) + + assert reply == DataplaneReply( + 200, + "upstream text", + "service_token", + raw_content=b"upstream text", + content_type="text/plain", + ) + + +@pytest.mark.asyncio +async def test_response_body_over_limit_is_rejected_while_streaming() -> None: + transport: Final = FakeTransport(iter([_response(200, {"value": "large"})])) + reply: Final = await _client(transport, Clock(), max_response_bytes=4).get( + _dataplane(), "/user/info", caller("session") + ) + assert reply == DataplaneResponseTooLarge(4) + + +@pytest.mark.asyncio +async def test_response_too_large_survives_stream_close_error_without_retry() -> None: + transport: Final = FakeTransport( + iter( + [ + _response(200, {"value": "large"}), + _response(200, {"value": "large"}), + _response(200, {"value": "large"}), + ] + ), + close_error=httpx.ReadError("close failed"), + ) + clock: Final = Clock() + + reply: Final = await _client(transport, clock, max_response_bytes=4).get( + _dataplane(), "/user/info", caller("session") + ) + + assert reply == DataplaneResponseTooLarge(4) + assert len(transport.calls) == 1 + assert clock.slept == [] + + +@pytest.mark.asyncio +async def test_reply_survives_stream_close_error() -> None: + transport: Final = FakeTransport( + iter([_response(200, {"ok": True}), _response(200, {"ok": True}), _response(200, {"ok": True})]), + close_error=httpx.ReadError("close failed"), + ) + + reply: Final = await _client(transport, Clock()).get(_dataplane(), "/user/info", caller("session")) + + assert reply == DataplaneReply(200, {"ok": True}, "service_token") + assert len(transport.calls) == 1 + + +@pytest.mark.asyncio +async def test_get_retries_mid_body_read_error() -> None: + transport: Final = FakeTransport(iter([_read_error_response(), _read_error_response(), _read_error_response()])) + clock: Final = Clock() + + reply: Final = await _client(transport, clock).get(_dataplane(), "/user/info", caller("session")) + + assert reply == DataplaneUnreachable("ReadError", attempts=3) + assert len(transport.calls) == 3 + assert clock.slept == [pytest.approx(0.1), pytest.approx(0.2)] + + @pytest.mark.asyncio async def test_get_retries_transient_failures_with_capped_exponential_backoff() -> None: transport, clock = FakeTransport(iter([httpx.ConnectError("down"), _response(503), _response(200, "up")])), Clock() @@ -106,7 +209,7 @@ async def test_post_is_never_retried_and_client_errors_are_replies_not_failures( result: Final = await _client(transport, clock).post( _dataplane(), "/key/generate", caller("session"), json={"a": 1} ) - assert result == DataplaneReply(502, "", "service_token") + assert result == DataplaneReply(502, "", "service_token", raw_content=b"") assert len(transport.calls) == 1 and clock.slept == [] forbidden: Final = FakeTransport(iter([_response(403, {"error": "no"})])) denied: Final = await _client(forbidden, clock).get(_dataplane(), "/x", caller("session")) diff --git a/tests/unit/proxy/control_plane/dataplanes/test_router.py b/tests/unit/proxy/control_plane/dataplanes/test_router.py index f133e282dbc..7d8869c70f0 100644 --- a/tests/unit/proxy/control_plane/dataplanes/test_router.py +++ b/tests/unit/proxy/control_plane/dataplanes/test_router.py @@ -1,4 +1,5 @@ -from collections.abc import Iterator +from collections.abc import AsyncIterator, Iterator +from contextlib import asynccontextmanager from dataclasses import dataclass, field from datetime import timedelta from pathlib import Path @@ -31,7 +32,8 @@ class FakeTransport: # mutable-ok: records calls for assertions outcomes: Iterator[httpx.Response | Exception] calls: list[tuple[str, dict[str, str] | None]] = field(default_factory=list) - async def request( + @asynccontextmanager + async def stream( self, method: str, url: str, @@ -40,13 +42,13 @@ class FakeTransport: # mutable-ok: records calls for assertions json: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> httpx.Response: + ) -> AsyncIterator[httpx.Response]: assert method == "GET", "router never writes" self.calls.append((url, headers)) outcome: Final = next(self.outcomes) if isinstance(outcome, Exception): raise outcome - return outcome + yield outcome async def _no_exchange(config: OAuthTokenExchangeConfig, subject_token: str) -> str: diff --git a/tests/unit/proxy/control_plane/proxy/__init__.py b/tests/unit/proxy/control_plane/proxy/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/proxy/control_plane/proxy/test_allowlist.py b/tests/unit/proxy/control_plane/proxy/test_allowlist.py new file mode 100644 index 00000000000..0a77a4b0389 --- /dev/null +++ b/tests/unit/proxy/control_plane/proxy/test_allowlist.py @@ -0,0 +1,116 @@ +from typing import Final + +import pytest + +from litellm.proxy.control_plane.proxy.allowlist import Allowed, Denied, decide + + +@pytest.mark.parametrize( + "path", + [ + "/key/generate", + "key/list", + "/team/new", + "/user/info", + "/model/info", + "/v1/model/info", + "/model_group/info", + "/management/v1/keys", + "/audit", + "/spend/logs", + "/router/settings", + ], +) +def test_management_routes_are_allowed_with_a_normalised_leading_slash(path: str) -> None: + assert decide(path) == Allowed("/" + path.lstrip("/")) + + +@pytest.mark.parametrize( + ("path", "fragment"), + [ + ("/v1/chat/completions", "inference"), + ("/chat/completions", "inference"), + ("/v1/embeddings", "inference"), + ("/v1/messages", "inference"), + ("/v1/mcp/server/oauth/client/register", "OAuth"), + ("/v1/mcp/server/oauth/client/callback", "OAuth"), + ("/guardrails/apply_guardrail", "guardrail apply"), + ("/guardrails/apply_guardrail/123", "guardrail apply"), + ("/mcp/tools/call", "inference"), + ("/login", "auth bootstrap"), + ("/sso/key/generate", "auth bootstrap"), + ("/key/service-account/generate", "auth bootstrap"), + ("/key%2Fservice-account%2Fgenerate", "auth bootstrap"), + ("/key/service-account/generate/", "auth bootstrap"), + ("/key/service-account/generate%2F", "auth bootstrap"), + ("/config/yaml", "auth bootstrap"), + ("/key/../config/yaml", "traversal"), + ("/key//info", "traversal"), + ("/health/drain", "lifecycle"), + ("/health/drain/", "lifecycle"), + ], +) +def test_inference_auth_and_traversal_are_denied(path: str, fragment: str) -> None: + decision: Final = decide(path) + assert isinstance(decision, Denied), f"{path} was allowed" + assert fragment in decision.reason + + +@pytest.mark.parametrize( + "path", + [ + "/key%2F..%2Flogin", + "/key/%2e%2e/login", + "/key/%252e%252e/login", + "/key/./info", + "/key//info", + "/key/list/.", + "/key/list/..", + "/key/\\../login", + "/credentialsx", + "/auditor/logs", + "/router_settings", + "/healthz", + "/settings_export", + ], +) +def test_encoded_dotted_or_boundary_breaking_paths_are_denied(path: str) -> None: + assert isinstance(decide(path), Denied), path + + +@pytest.mark.parametrize("path", ["/credentials", "/credentials/by_name/x", "/audit", "/audit/1", "/settings"]) +def test_bare_management_routes_match_only_on_a_route_boundary(path: str) -> None: + assert decide(path) == Allowed(path) + + +@pytest.mark.parametrize("path", ["/health/readiness", "/health/liveliness", "/health/liveness", "/health/latest"]) +def test_safe_health_routes_are_allowed(path: str) -> None: + assert decide(path) == Allowed(path) + + +@pytest.mark.parametrize( + "path", + [ + "/health", + "/health/services", + "/health/test_connection", + "/health/readiness/details", + "/health/liveness/details", + ], +) +def test_health_routes_that_probe_or_expose_details_are_denied(path: str) -> None: + assert isinstance(decide(path), Denied) + + +def test_decided_path_is_the_decoded_path_the_dataplane_will_receive() -> None: + assert decide("key/info%3Fx") == Denied("/key/info%3Fx", "encoded or reserved path characters are not allowed") + assert decide("/team/my%20team/info") == Allowed("/team/my team/info") + + +def test_single_trailing_slash_is_removed_without_allowing_empty_segments() -> None: + root: Final = decide("/") + assert root == Denied("/", "only management routes are proxied, inference goes straight to the dataplane") + assert decide("key/list/") == Allowed("/key/list") + assert isinstance(decide("//x"), Denied) + assert isinstance(decide("/a//b"), Denied) + assert isinstance(decide("/key/list//"), Denied) diff --git a/tests/unit/proxy/control_plane/proxy/test_router.py b/tests/unit/proxy/control_plane/proxy/test_router.py new file mode 100644 index 00000000000..0a53e0c382b --- /dev/null +++ b/tests/unit/proxy/control_plane/proxy/test_router.py @@ -0,0 +1,326 @@ +from typing import Final + +import httpx + +from litellm.proxy.control_plane.auth.rbac import EDITOR_ROLE, USAGE_VIEWER_ROLE, VIEWER_ROLE +from litellm.proxy.control_plane.dataplanes.client import DataplaneClient, RetryPolicy +from litellm.proxy.control_plane.dataplanes.dependencies import get_dataplane_client +from tests.unit.proxy.control_plane.support import ( + response, + rig, +) + +EU = "https://eu.test" + + +def test_get_is_forwarded_as_the_caller_with_query_and_the_dataplane_status_comes_back() -> None: + client, transport, minter = rig({f"{EU}/key/list": [response(200, {"keys": ["k1"], "total_count": 1})]}) + reply: Final = client.get("/api/v1/dataplanes/eu/proxy/key/list?page=2&size=5", headers=minter.headers("eu")) + assert (reply.status_code, reply.json()) == (200, {"keys": ["k1"], "total_count": 1}) + assert reply.headers["X-LiteLLM-Dataplane"] == "eu" + url, params, headers = transport.calls[0] + assert (url, params) == (f"{EU}/key/list", (("page", "2"), ("size", "5"))) + assert headers is not None and headers["X-LiteLLM-Impersonate-User"] == "alice" + + +def test_get_with_a_trailing_slash_reaches_the_dataplane_without_it() -> None: + client, transport, minter = rig({f"{EU}/key/list": [response(200, {"keys": ["k1"], "total_count": 1})]}) + + reply: Final = client.get("/api/v1/dataplanes/eu/proxy/key/list/", headers=minter.headers("eu")) + + assert (reply.status_code, reply.json()) == (200, {"keys": ["k1"], "total_count": 1}) + assert transport.calls[0][0] == f"{EU}/key/list" + + +def test_oversized_dataplane_response_is_a_control_plane_bad_gateway() -> None: + client, transport, minter = rig({f"{EU}/key/list": [response(200, {"oversized": "body"})]}) + client.app.dependency_overrides[get_dataplane_client] = lambda: DataplaneClient( + transport, timeout_seconds=1, max_response_bytes=4, retry=RetryPolicy(attempts=1) + ) + + reply: Final = client.get("/api/v1/dataplanes/eu/proxy/key/list", headers=minter.headers("eu")) + + assert reply.status_code == 502 + assert reply.json()["type"].endswith("dataplane-response-too-large") + + +def test_scope_denied_passthrough_reads_and_writes_are_audited() -> None: + client, transport, minter = rig({}) + headers: Final = minter.headers("eu", role=USAGE_VIEWER_ROLE) + + read: Final = client.get("/api/v1/dataplanes/eu/proxy/key/sk-live-secret/info", headers=headers) + write: Final = client.post("/api/v1/dataplanes/eu/proxy/key/sk-live-secret/regenerate", headers=headers) + + assert (read.status_code, write.status_code) == (403, 403) + assert transport.calls == [] + trail: Final = client.get("/api/v1/audit", headers=minter.headers(role="admin")).json()["items"] + assert { + (event["actor"], event["dataplane_id"], event["method"], event["path"], event["decision"]) for event in trail + } == { + ("alice", "eu", "GET", "/key/sk-[redacted]", "denied"), + ("alice", "eu", "POST", "/key/sk-[redacted]/regenerate", "denied"), + } + assert all("Missing scope" in event["detail"] for event in trail) + + +def test_get_preserves_repeated_query_parameters_in_order() -> None: + client, transport, minter = rig({f"{EU}/key/list": [response(200, {})]}) + reply: Final = client.get( + "/api/v1/dataplanes/eu/proxy/key/list?team_id=first&team_id=second", + headers=minter.headers("eu"), + ) + assert reply.status_code == 200 + assert transport.calls[0][1] == (("team_id", "first"), ("team_id", "second")) + + +def test_writes_need_proxy_write_go_out_once_and_are_audited() -> None: + client, transport, minter = rig( + { + f"{EU}/key/generate": [response(200, {"key": "sk-new", "token_id": "t1"})], + f"{EU}/team/update": [httpx.ReadTimeout("slow")], + f"{EU}/key/delete": [response(400, {"error": {"message": "no such key"}})], + } + ) + viewer: Final = minter.headers("eu", role=VIEWER_ROLE) + editor: Final = minter.headers("eu", role=EDITOR_ROLE) + denied: Final = client.post("/api/v1/dataplanes/eu/proxy/key/generate", json={"key_alias": "x"}, headers=viewer) + assert denied.status_code == 403 and "proxy:write" in denied.json()["detail"] + created: Final = client.post("/api/v1/dataplanes/eu/proxy/key/generate", json={"key_alias": "x"}, headers=editor) + assert (created.status_code, created.json()["token_id"]) == (200, "t1") + assert transport.writes == [("POST", f"{EU}/key/generate", {"key_alias": "x"})] + timed_out: Final = client.patch("/api/v1/dataplanes/eu/proxy/team/update", json={"team_id": "t"}, headers=editor) + assert timed_out.status_code == 502 and "1 attempt" in timed_out.json()["detail"] + failed: Final = client.delete("/api/v1/dataplanes/eu/proxy/key/delete", headers=editor) + assert (failed.status_code, failed.json()) == (400, {"error": {"message": "no such key"}}) + trail: Final = client.get("/api/v1/audit", headers=minter.headers(role="admin")).json()["items"] + assert [(e["method"], e["path"], e["decision"], e["status"]) for e in trail] == [ + ("DELETE", "/key/delete", "allowed", 400), + ("PATCH", "/team/update", "unreachable", None), + ("POST", "/key/generate", "allowed", 200), + ("POST", "/key/generate", "denied", 403), + ] + assert trail[2]["actor"] == "alice" and trail[2]["credential"] == "service_token" + + +def test_passthrough_write_accepts_a_200_kilobyte_body() -> None: + client, transport, minter = rig({f"{EU}/key/generate": [response(200, {"accepted": True})]}) + payload: Final = {"key_alias": "x" * (200 * 1024)} + + reply: Final = client.post( + "/api/v1/dataplanes/eu/proxy/key/generate", + json=payload, + headers=minter.headers("eu", role=EDITOR_ROLE), + ) + + assert (reply.status_code, reply.json()) == (200, {"accepted": True}) + assert transport.writes == [("POST", f"{EU}/key/generate", payload)] + + +def test_denied_paths_are_403_audited_and_never_reach_the_dataplane() -> None: + client, transport, minter = rig({}) + blocked: Final = client.post( + "/api/v1/dataplanes/eu/proxy/v1/chat/completions", json={"model": "x"}, headers=minter.headers(role=EDITOR_ROLE) + ) + assert blocked.status_code == 403 and "inference" in blocked.json()["detail"] + assert transport.calls == [] + trail: Final = client.get("/api/v1/audit?dataplane=eu", headers=minter.headers(role="admin")).json()["items"] + assert [(e["path"], e["decision"]) for e in trail] == [("/v1/chat/completions", "denied")] + + +def test_a_non_object_body_is_400_and_unbound_dataplane_is_403() -> None: + client, _, minter = rig({}) + bad: Final = client.post( + "/api/v1/dataplanes/eu/proxy/key/generate", + content=b"[1,2]", + headers={**minter.headers(role=EDITOR_ROLE), "Content-Type": "application/json"}, + ) + assert bad.status_code == 400 + unbound: Final = client.get("/api/v1/dataplanes/us/proxy/key/list", headers=minter.headers("eu")) + assert unbound.status_code == 403 + trail: Final = client.get("/api/v1/audit", headers=minter.headers(role="admin")).json()["items"] + assert [(event["method"], event["path"], event["decision"], event["status"]) for event in trail] == [ + ("GET", "/key/list", "denied", 403), + ("POST", "/key/generate", "denied", 400), + ] + + +def test_audit_read_needs_its_scope_and_filters_by_actor() -> None: + client, _, minter = rig({f"{EU}/key/list": [response(200, {})]}) + client.get("/api/v1/dataplanes/eu/proxy/key/list", headers=minter.headers("eu")) + assert client.get("/api/v1/audit", headers=minter.headers(role=EDITOR_ROLE)).status_code == 403 + assert client.get("/api/v1/audit?actor=bob", headers=minter.headers(role="admin")).json()["items"] == [] + assert len(client.get("/api/v1/audit?actor=alice", headers=minter.headers(role="admin")).json()["items"]) == 1 + + +def test_audit_reads_are_clamped_to_the_callers_dataplane_binding() -> None: + client, _, minter = rig({f"{EU}/key/list": [response(200, {})], "https://us.test/key/list": [response(200, {})]}) + client.get("/api/v1/dataplanes/eu/proxy/key/list", headers=minter.headers("eu")) + client.get("/api/v1/dataplanes/us/proxy/key/list", headers=minter.headers("us")) + eu_auditor: Final = minter.headers("eu", role="admin") + unfiltered: Final = client.get("/api/v1/audit", headers=eu_auditor).json()["items"] + assert [e["dataplane_id"] for e in unfiltered] == ["eu"] + assert client.get("/api/v1/audit?dataplane=us", headers=eu_auditor).status_code == 403 + assert len(client.get("/api/v1/audit?dataplane=eu", headers=eu_auditor).json()["items"]) == 1 + fleet: Final = client.get("/api/v1/audit", headers=minter.headers(role="admin")).json()["items"] + assert sorted(e["dataplane_id"] for e in fleet) == ["eu", "us"] + + +def test_a_redirecting_dataplane_is_a_502_and_the_redirect_is_never_followed() -> None: + client, transport, minter = rig( + {f"{EU}/v1/mcp/server/tools/call": [httpx.Response(302, headers={"Location": "http://127.0.0.1:9/"})]} + ) + reply: Final = client.get( + "/api/v1/dataplanes/eu/proxy/v1/mcp/server/tools/call", headers=minter.headers("eu") + ) + assert reply.status_code == 502 and "redirect" in reply.json()["detail"] + assert [url for url, _, _ in transport.calls] == [f"{EU}/v1/mcp/server/tools/call"] + trail: Final = client.get("/api/v1/audit", headers=minter.headers(role="admin")).json()["items"] + assert (trail[0]["decision"], trail[0]["status"]) == ("denied", 302) + + +def test_key_bearing_paths_are_redacted_in_the_audit_trail() -> None: + client, _, minter = rig({f"{EU}/key/sk-live-123/reset_spend": [response(200, {})]}) + assert ( + client.post( + "/api/v1/dataplanes/eu/proxy/key/sk-live-123/reset_spend", + json={}, + headers=minter.headers("eu", role=EDITOR_ROLE), + ).status_code + == 200 + ) + trail: Final = client.get("/api/v1/audit", headers=minter.headers(role="admin")).json()["items"] + assert trail[0]["path"] == "/key/sk-[redacted]/reset_spend" + + +def test_a_denied_key_bearing_path_is_redacted_in_the_403_too() -> None: + client, _, minter = rig({}) + denied: Final = client.get("/api/v1/dataplanes/eu/proxy/key/sk-live-123/x%25y", headers=minter.headers("eu")) + assert denied.status_code == 403 and "sk-live-123" not in denied.text and "sk-[redacted]" in denied.text + + +def test_non_json_management_replies_preserve_the_body_and_headers() -> None: + content: Final = b"user_id,email\nu1,user@example.com\n" + reply: Final = httpx.Response( + 200, + content=content, + headers={ + "Content-Type": "text/csv; charset=utf-8", + "Content-Disposition": 'attachment; filename="users.csv"', + "Cache-Control": "private, max-age=0", + }, + request=httpx.Request("GET", f"{EU}/user/export"), + ) + client, _, minter = rig({f"{EU}/user/export": [reply]}) + + response: Final = client.get("/api/v1/dataplanes/eu/proxy/user/export", headers=minter.headers("eu")) + + assert response.status_code == 200 + assert response.content == content + assert response.headers["content-type"] == "text/csv; charset=utf-8" + assert response.headers["content-disposition"] == 'attachment; filename="users.csv"' + assert response.headers["cache-control"] == "private, max-age=0" + assert response.headers["X-LiteLLM-Dataplane"] == "eu" + + +def test_empty_management_replies_without_content_type_preserve_status_and_headers() -> None: + reply: Final = httpx.Response( + 202, + content=b"", + headers={"Cache-Control": "private, max-age=0"}, + request=httpx.Request("GET", f"{EU}/user/export"), + ) + client, _, minter = rig({f"{EU}/user/export": [reply]}) + + response: Final = client.get("/api/v1/dataplanes/eu/proxy/user/export", headers=minter.headers("eu")) + + assert (response.status_code, response.content) == (202, b"") + assert response.headers["cache-control"] == "private, max-age=0" + assert response.headers["X-LiteLLM-Dataplane"] == "eu" + + +def test_empty_json_management_reply_preserves_empty_body_and_content_type() -> None: + reply: Final = httpx.Response( + 200, + content=b"", + headers={"Content-Type": "application/json"}, + request=httpx.Request("GET", f"{EU}/user/export"), + ) + client, _, minter = rig({f"{EU}/user/export": [reply]}) + + response: Final = client.get("/api/v1/dataplanes/eu/proxy/user/export", headers=minter.headers("eu")) + + assert (response.status_code, response.content, response.headers["content-type"]) == ( + 200, + b"", + "application/json", + ) + + +def test_invalid_json_management_reply_preserves_exact_body_and_content_type() -> None: + content: Final = b'{"invalid":' + reply: Final = httpx.Response( + 200, + content=content, + headers={"Content-Type": "application/json"}, + request=httpx.Request("GET", f"{EU}/user/export"), + ) + client, _, minter = rig({f"{EU}/user/export": [reply]}) + + response: Final = client.get("/api/v1/dataplanes/eu/proxy/user/export", headers=minter.headers("eu")) + + assert (response.status_code, response.content, response.headers["content-type"]) == ( + 200, + content, + "application/json", + ) + + +def test_json_management_replies_forward_cache_control() -> None: + reply: Final = httpx.Response( + 200, + json={"users": []}, + headers={"Cache-Control": "private, max-age=0"}, + request=httpx.Request("GET", f"{EU}/user/list"), + ) + client, _, minter = rig({f"{EU}/user/list": [reply]}) + + response: Final = client.get("/api/v1/dataplanes/eu/proxy/user/list", headers=minter.headers("eu")) + + assert response.json() == {"users": []} + assert response.headers["cache-control"] == "private, max-age=0" + + +def test_json_management_replies_forward_content_disposition() -> None: + reply: Final = httpx.Response( + 200, + json={"users": []}, + headers={"Content-Disposition": 'attachment; filename="users.json"'}, + request=httpx.Request("GET", f"{EU}/user/export"), + ) + client, _, minter = rig({f"{EU}/user/export": [reply]}) + + response: Final = client.get("/api/v1/dataplanes/eu/proxy/user/export", headers=minter.headers("eu")) + + assert response.json() == {"users": []} + assert response.headers["content-disposition"] == 'attachment; filename="users.json"' + + +def test_head_is_forwarded_as_a_read_and_returns_no_body() -> None: + reply: Final = httpx.Response(200, request=httpx.Request("HEAD", f"{EU}/key/list")) + client, transport, minter = rig({f"{EU}/key/list": [reply]}) + + response: Final = client.head("/api/v1/dataplanes/eu/proxy/key/list", headers=minter.headers("eu")) + + assert (response.status_code, response.content) == (200, b"") + assert transport.writes == [] + trail: Final = client.get("/api/v1/audit", headers=minter.headers(role="admin")).json()["items"] + assert (trail[0]["method"], trail[0]["decision"]) == ("HEAD", "allowed") + + +def test_guardrail_apply_is_denied_even_under_a_management_prefix() -> None: + client, transport, minter = rig({}) + + reply: Final = client.get("/api/v1/dataplanes/eu/proxy/guardrails/apply_guardrail", headers=minter.headers("eu")) + + assert reply.status_code == 403 and "guardrail apply" in reply.json()["detail"] + assert transport.calls == [] diff --git a/tests/unit/proxy/control_plane/support.py b/tests/unit/proxy/control_plane/support.py index 0a199ab2f34..4c875d43340 100644 --- a/tests/unit/proxy/control_plane/support.py +++ b/tests/unit/proxy/control_plane/support.py @@ -1,7 +1,8 @@ """Shared rig for control plane route tests: a routed fake transport and a licensed app over two dataplanes.""" from collections import defaultdict -from collections.abc import Iterator, Mapping, Sequence +from collections.abc import AsyncIterator, Iterator, Mapping, Sequence +from contextlib import asynccontextmanager from dataclasses import dataclass, field from datetime import timedelta from typing import Final @@ -34,8 +35,10 @@ class RoutedTransport: # mutable-ok: records calls for assertions routes: Mapping[str, Iterator[httpx.Response | Exception]] calls: list[tuple[str, tuple[tuple[str, str], ...] | None, dict[str, str] | None]] = field(default_factory=list) + writes: list[tuple[str, str, dict[str, object] | None]] = field(default_factory=list) - async def request( + @asynccontextmanager + async def stream( self, method: str, url: str, @@ -44,9 +47,13 @@ class RoutedTransport: # mutable-ok: records calls for assertions json: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> httpx.Response: - assert method == "GET", "fleet reads never write" + ) -> AsyncIterator[httpx.Response]: self.calls.append((url, None if params is None else tuple(params), headers)) + if method not in {"GET", "HEAD"}: + self.writes.append((method, url, json)) + yield self._answer(url) + + def _answer(self, url: str) -> httpx.Response: outcome: Final = next(self.routes[url], None) if outcome is None: raise AssertionError(f"no scripted outcome left for {url}") @@ -76,14 +83,10 @@ class Minter: def rig( - routes: Mapping[str, list[httpx.Response | Exception]], usage_max_window_days: int = 366 + routes: Mapping[str, list[httpx.Response | Exception]], **overrides: object ) -> tuple[TestClient, RoutedTransport, Minter]: settings: Final = ControlPlaneSettings.model_validate( - { - "LITELLM_MASTER_KEY": MASTER_KEY, - "dataplanes": DATAPLANES, - "usage_max_window_days": usage_max_window_days, - } + {"LITELLM_MASTER_KEY": MASTER_KEY, "dataplanes": DATAPLANES, **overrides} ) queues: defaultdict[str, Iterator[httpx.Response | Exception]] = defaultdict(lambda: iter(())) queues.update({url: iter(outcomes) for url, outcomes in routes.items()}) diff --git a/tests/unit/proxy/control_plane/test_container.py b/tests/unit/proxy/control_plane/test_container.py index 1688df093ee..4735b6af098 100644 --- a/tests/unit/proxy/control_plane/test_container.py +++ b/tests/unit/proxy/control_plane/test_container.py @@ -3,11 +3,14 @@ from typing import Final, cast import httpx import pytest +from pydantic import TypeAdapter from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler from litellm.proxy.control_plane.container import HandlerTransport, _shared_http_handler from litellm.proxy.control_plane.settings import ControlPlaneSettings +_JSON_COUNT: Final = TypeAdapter(dict[str, int]) + def test_shared_http_handler_does_not_follow_redirects() -> None: handler: Final = _shared_http_handler( @@ -35,9 +38,11 @@ class _RecyclingHandler: # mutable-ok: counts how often the pool is re-read async def test_transport_rereads_the_handlers_pool_on_every_request() -> None: handler: Final = _RecyclingHandler() transport: Final = HandlerTransport(cast(AsyncHTTPHandler, handler)) # cast-ok: only the client property is used - first: Final = await transport.request("GET", "https://dp.test/health", timeout=1.0) - second: Final = await transport.request("GET", "https://dp.test/health", timeout=1.0) - assert (first.json(), second.json()) == ({"n": 1}, {"n": 2}), "a pool closed by eviction must not be reused" + async with transport.stream("GET", "https://dp.test/health", timeout=1.0) as first_response: + first: Final = _JSON_COUNT.validate_json(await first_response.aread()) + async with transport.stream("GET", "https://dp.test/health", timeout=1.0) as second_response: + second: Final = _JSON_COUNT.validate_json(await second_response.aread()) + assert (first, second) == ({"n": 1}, {"n": 2}), "a pool closed by eviction must not be reused" @dataclass @@ -59,5 +64,6 @@ class _RedirectingHandler: # mutable-ok: counts the hops a following pool would async def test_transport_never_follows_a_dataplane_redirect_even_on_a_following_pool() -> None: handler: Final = _RedirectingHandler() transport: Final = HandlerTransport(cast(AsyncHTTPHandler, handler)) # cast-ok: only the client property is used - reply: Final = await transport.request("GET", "https://dp.test/v1/mcp/server/oauth/x/authorize", timeout=1.0) - assert reply.status_code == 302 and handler.hops == ["https://dp.test/v1/mcp/server/oauth/x/authorize"] + async with transport.stream("GET", "https://dp.test/v1/mcp/server/oauth/x/authorize", timeout=1.0) as response: + status: Final = response.status_code + assert status == 302 and handler.hops == ["https://dp.test/v1/mcp/server/oauth/x/authorize"]