Merge 198f711a7b into 115326356d

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

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")

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)

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)
+        )
+    )

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}")
 

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),
     )

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
+    )

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}"))

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)

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)

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")

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, ...]

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

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"))

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:

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)

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 == []

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()})

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"]