Merge d7feabe880 into f415aa88da

litellm/proxy/control_plane/app.py

@@ -4,6 +4,8 @@ from typing import Final, Protocol
 
 from fastapi import FastAPI
 
+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
@@ -13,6 +15,7 @@ from litellm.proxy.control_plane.dataplanes.dependencies import get_dataplane_cl
 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
 
 TITLE: Final = "LiteLLM Control Plane"
@@ -43,10 +46,13 @@ def create_app(container: Container, license_check: License) -> FastAPI:
     app.dependency_overrides[get_registry] = lambda: container.registry
     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,89 @@
+"""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 typing import Final, Literal, TypeAlias
+
+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]"
+
+
+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."""
+    return "/".join(REDACTED_KEY if segment.startswith(KEY_PREFIX) else segment for segment in path.split("/"))
+
+
+@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/container.py

@@ -13,6 +13,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 +42,7 @@ class Container:
     registry: DataplaneRegistry
     dataplane_client: DataplaneClient
     fanout: Fanout
+    audit_log: AuditLog
 
 
 class InvalidRolePolicy(ValueError):
@@ -145,4 +147,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

@@ -17,6 +17,7 @@ from litellm.proxy.control_plane.dataplanes.credentials import Caller, Credentia
 from litellm.proxy.control_plane.dataplanes.registry import Dataplane
 
 Method: TypeAlias = Literal["GET", "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})
 _JSON: Final = TypeAdapter(object)
@@ -98,6 +99,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,

litellm/proxy/control_plane/proxy/allowlist.py

@@ -0,0 +1,101 @@
+"""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/",
+    "/health",
+    "/active/callbacks",
+    "/callbacks",
+    "/alerting/",
+    "/router_settings",
+    "/settings",
+)
+
+DENIED_PREFIXES: Final = (
+    "/health/drain",
+    "/login",
+    "/logout",
+    "/sso/",
+    "/token",
+    "/jwt/",
+    "/oauth/",
+    "/invitation/",
+    "/onboarding/",
+    "/session/",
+    "/key/service-account/generate",
+    "/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
+    return "/" + decoded.lstrip("/")
+
+
+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 and lifecycle routes never go through the control plane")
+    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,130 @@
+"""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
+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 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,
+    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
+from litellm.proxy.control_plane.dataplanes.registry import Dataplane
+from litellm.proxy.control_plane.proxy.allowlist import Denied, decide
+
+DATAPLANE_HEADER: Final = "X-LiteLLM-Dataplane"
+READ_METHODS: Final = ("GET",)
+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,
+) -> JSONResponse:
+    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 _json_body(request),
+    )
+    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):
+            audit.record(
+                principal.subject, principal.email, dataplane.id, method, decision.path, "allowed", status, credential
+            )
+            return JSONResponse(status_code=status, content=body, headers={DATAPLANE_HEADER: dataplane.id})
+        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}"))
+
+
+@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(require_scopes, scopes=[Scope.PROXY_READ.value])],
+    dataplane: Annotated[Dataplane, Depends(dataplane_access)],
+    caller: Annotated[Caller, Depends(get_caller)],
+    client: Annotated[DataplaneClient, Depends(get_dataplane_client)],
+    audit: Annotated[AuditLog, Depends(get_audit_log)],
+) -> JSONResponse:
+    """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(require_scopes, scopes=[Scope.PROXY_WRITE.value])],
+    dataplane: Annotated[Dataplane, Depends(dataplane_access)],
+    caller: Annotated[Caller, Depends(get_caller)],
+    client: Annotated[DataplaneClient, Depends(get_dataplane_client)],
+    audit: Annotated[AuditLog, Depends(get_audit_log)],
+) -> JSONResponse:
+    """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

@@ -107,6 +107,7 @@ class ControlPlaneSettings(BaseSettings):
     dataplane_timeout_seconds: float = DEFAULT_DATAPLANE_TIMEOUT_SECONDS
     dataplane_retry_attempts: int = 3
     fanout_concurrency: int = 16
+    audit_capacity: int = 10_000
     api_prefix: str = "/api/v1"
 
     @property

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,50 @@
+from datetime import UTC, datetime, timedelta
+from itertools import count
+
+from litellm.proxy.control_plane.audit.log import build_audit_log
+
+T0 = datetime(2026, 10, 13, 12, 0, tzinfo=UTC)
+
+
+def _ticking():
+    ticks = count()
+    return lambda: T0 + timedelta(seconds=next(ticks))
+
+
+def test_recent_is_newest_first_filtered_and_limited() -> None:
+    audit = 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 = 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 = build_audit_log(capacity=10, now=_ticking())
+    event = 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 = 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"]

tests/unit/proxy/control_plane/proxy/test_allowlist.py

@@ -0,0 +1,80 @@
+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",
+        "/management/v1/keys",
+        "/audit",
+        "/spend/logs",
+    ],
+)
+def test_management_routes_are_allowed_with_a_normalised_leading_slash(path) -> 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"),
+        ("/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, fragment) -> None:
+    decision = 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",
+        "/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", "/health", "/settings"]
+)
+def test_bare_management_routes_match_only_on_a_route_boundary(path: str) -> None:
+    assert decide(path) == Allowed(path)
+
+
+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")

tests/unit/proxy/control_plane/proxy/test_router.py

@@ -0,0 +1,128 @@
+import httpx
+
+from litellm.proxy.control_plane.auth.rbac import EDITOR_ROLE, VIEWER_ROLE
+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 = 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_preserves_repeated_query_parameters_in_order() -> None:
+    client, transport, minter = rig({f"{EU}/key/list": [response(200, {})]})
+    reply = 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 = minter.headers("eu", role=VIEWER_ROLE)
+    editor = minter.headers("eu", role=EDITOR_ROLE)
+    denied = 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 = 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 = 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 = client.delete("/api/v1/dataplanes/eu/proxy/key/delete", headers=editor)
+    assert (failed.status_code, failed.json()) == (400, {"error": {"message": "no such key"}})
+    trail = 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),
+    ]
+    assert trail[2]["actor"] == "alice" and trail[2]["credential"] == "service_token"
+
+
+def test_denied_paths_are_403_audited_and_never_reach_the_dataplane() -> None:
+    client, transport, minter = rig({})
+    blocked = 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 = 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 = 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
+    assert client.get("/api/v1/dataplanes/us/proxy/key/list", headers=minter.headers("eu")).status_code == 403
+
+
+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 = minter.headers("eu", role="admin")
+    unfiltered = 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 = 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/oauth/x/authorize": [httpx.Response(302, headers={"Location": "http://127.0.0.1:9/"})]}
+    )
+    reply = client.get("/api/v1/dataplanes/eu/proxy/v1/mcp/server/oauth/x/authorize", 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/oauth/x/authorize"]
+    trail = 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 = 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 = 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

tests/unit/proxy/control_plane/support.py

@@ -34,6 +34,7 @@ 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(
         self,
@@ -45,8 +46,12 @@ class RoutedTransport:  # mutable-ok: records calls for assertions
         headers: dict[str, str] | None = None,
         timeout: float,
     ) -> httpx.Response:
-        assert method == "GET", "fleet reads never write"
         self.calls.append((url, None if params is None else tuple(params), headers))
+        if method != "GET":
+            self.writes.append((method, url, json))
+        return self._answer(url)
+
+    def _answer(self, url: str) -> httpx.Response:
         outcome = next(self.routes[url], None)
         if outcome is None:
             raise AssertionError(f"no scripted outcome left for {url}")
@@ -75,8 +80,12 @@ class Minter:
         return {"Authorization": f"Bearer {self.codec.mint(principal)}"}
 
 
-def rig(routes: Mapping[str, list[httpx.Response | Exception]]) -> tuple[TestClient, RoutedTransport, Minter]:
-    settings = ControlPlaneSettings.model_validate({"LITELLM_MASTER_KEY": MASTER_KEY, "dataplanes": DATAPLANES})
+def rig(
+    routes: Mapping[str, list[httpx.Response | Exception]], **overrides: object
+) -> tuple[TestClient, RoutedTransport, Minter]:
+    settings = ControlPlaneSettings.model_validate(
+        {"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()})
     transport = RoutedTransport(queues)