Merge remote-tracking branch 'origin/litellm_internal_staging' into pragnyan/fix-websearch-nim-model-prefix-23970

.github/workflows/test-unit-proxy-mgmt-behavior.yml

@@ -0,0 +1,34 @@
+name: "Unit Tests: Proxy Management-Endpoint Behavior Pinning"
+
+on:
+  pull_request:
+    branches:
+      - main
+      - litellm_internal_staging
+      - litellm_oss_branch
+      - "litellm_**"
+
+permissions:
+  contents: read
+  id-token: write
+  pull-requests: write
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  proxy-mgmt-behavior:
+    uses: ./.github/workflows/_test-unit-services-base.yml
+    with:
+      test-path: tests/proxy_behavior
+      # workers=0 (no xdist): the world seed is a single shared Postgres
+      # state — two xdist workers both call seed_world() and race on the
+      # ``behavior-pin-budget`` row, producing UniqueViolation + cascading
+      # missing-membership FK failures. The whole suite is ~7s sequentially,
+      # so the cost of disabling parallelism here is negligible.
+      workers: 0
+      reruns: 0
+      enable-postgres: true
+      artifact-name: proxy-mgmt-behavior
+      timeout-minutes: 15

litellm/integrations/opentelemetry.py

@@ -1611,11 +1611,11 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
                     "masked_entity_count", safe_dumps(masked_entity_count)
                 )
 
-            self.safe_set_attribute(
-                span=guardrail_span,
-                key="guardrail_response",
-                value=guardrail_information.get("guardrail_response"),
-            )
+            guardrail_response = guardrail_information.get("guardrail_response")
+            if guardrail_response is not None:
+                guardrail_span.set_attribute(
+                    "guardrail_response", safe_dumps(guardrail_response)
+                )
 
             self._set_team_attributes_from_kwargs(guardrail_span, kwargs)
 

litellm/proxy/auth/user_api_key_auth.py

@@ -12,7 +12,7 @@ import fnmatch
 import re
 import secrets
 from datetime import datetime, timezone
-from typing import Any, Iterator, List, Optional, Tuple, Union, cast
+from typing import Any, Dict, Iterator, List, Optional, Tuple, Union, cast
 
 import fastapi
 from fastapi import HTTPException, Request, WebSocket, status
@@ -333,8 +333,22 @@ def _apply_budget_limits_to_end_user_params(
 async def user_api_key_auth_websocket(websocket: WebSocket):
     # Accept the WebSocket connection
 
-    scope_headers = list(websocket.scope.get("headers") or [])
-    request = Request(scope={"type": "http", "headers": scope_headers})
+    ws_scope = websocket.scope or {}
+    scope_headers = list(ws_scope.get("headers") or [])
+    # ``get_request_route`` falls back to ``request.url.path`` when
+    # ``scope["path"]`` is absent. On WebSockets that fallback reads
+    # ``websocket.url``, which Starlette reconstructs from the (poisonable)
+    # Host header. Carry the ASGI scope's path / root_path so the lookup
+    # never reaches the fallback.
+    synthetic_scope: Dict[str, Any] = {
+        "type": "http",
+        "headers": scope_headers,
+        "path": ws_scope.get("path", ""),
+    }
+    for key in ("root_path", "app_root_path"):
+        if key in ws_scope:
+            synthetic_scope[key] = ws_scope[key]
+    request = Request(scope=synthetic_scope)
 
     request._url = websocket.url
 

litellm/proxy/common_utils/http_parsing_utils.py

@@ -13,6 +13,34 @@ from litellm.proxy.common_utils.callback_utils import (
 from litellm.types.router import Deployment
 
 
+_FORM_CONTENT_TYPES: frozenset[str] = frozenset(
+    {"application/x-www-form-urlencoded", "multipart/form-data"}
+)
+
+
+def _normalize_media_type(content_type: str) -> str:
+    """Return the bare media type per RFC 7231: strip params, trim, lowercase."""
+    if not content_type:
+        return ""
+    return content_type.split(";", 1)[0].strip().lower()
+
+
+def _is_form_content_type(content_type: str) -> bool:
+    """
+    True iff Starlette's ``request.form()`` will actually parse this body.
+
+    Substring matching ``"form"`` is unsafe: ``request.form()`` returns empty
+    ``FormData`` for non-canonical types without consuming the body, leaving
+    the auth-time pre-read and the handler's read seeing different payloads.
+    """
+    return _normalize_media_type(content_type) in _FORM_CONTENT_TYPES
+
+
+def _is_json_content_type(content_type: str) -> bool:
+    """True iff the body should be parsed as JSON."""
+    return _normalize_media_type(content_type) == "application/json"
+
+
 async def _read_request_body(request: Optional[Request]) -> Dict:
     """
     Safely read the request body and parse it as JSON.
@@ -37,8 +65,24 @@ async def _read_request_body(request: Optional[Request]) -> Dict:
         _request_headers: dict = _safe_get_request_headers(request=request)
         content_type = _request_headers.get("content-type", "")
 
-        if "form" in content_type:
-            parsed_body = dict(await request.form())
+        if _is_form_content_type(content_type):
+            try:
+                form_data = await request.form()
+            except Exception as e:
+                # ``request.form()`` raises on malformed multipart (missing
+                # boundary, malformed chunk encoding, …). Surface as 400 so
+                # the auth-time pre-read does not silently cache ``{}`` while
+                # a later raw-body re-read sees the original payload —
+                # banned-param checks must see the same body the handler
+                # acts on.
+                verbose_proxy_logger.error(f"Invalid form payload: {e}")
+                raise ProxyException(
+                    message=f"Invalid form payload: {e}",
+                    type="invalid_request_error",
+                    param="request_body",
+                    code=status.HTTP_400_BAD_REQUEST,
+                )
+            parsed_body = dict(form_data)
             if "metadata" in parsed_body and isinstance(parsed_body["metadata"], str):
                 parsed_body["metadata"] = json.loads(parsed_body["metadata"])
         else:
@@ -306,18 +350,13 @@ async def get_request_body(request: Request) -> Dict[str, Any]:
     Read the request body and parse it as JSON.
     """
     if request.method == "POST":
-        if request.headers.get("content-type", "") == "application/json":
+        content_type = request.headers.get("content-type", "")
+        if _is_json_content_type(content_type):
             return await _read_request_body(request)
-        elif "multipart/form-data" in request.headers.get(
-            "content-type", ""
-        ) or "application/x-www-form-urlencoded" in request.headers.get(
-            "content-type", ""
-        ):
+        elif _is_form_content_type(content_type):
             return await get_form_data(request)
         else:
-            raise ValueError(
-                f"Unsupported content type: {request.headers.get('content-type')}"
-            )
+            raise ValueError(f"Unsupported content type: {content_type}")
     return {}
 
 

pyproject.toml

@@ -287,6 +287,12 @@ paths_to_mutate = [
 ]
 tests_dir = [
     "tests/test_litellm/proxy/management_endpoints/",
+    # PR1 (key Tier-1) behavior-pinning suite. Manual mutmut runs
+    # (.github/workflows/mutation-test.yml) include this directory so the
+    # behavior matrix contributes to mutation-score signal alongside the
+    # legacy mock suite. See tests/proxy_behavior/management/README.md
+    # for the G5 triage protocol.
+    "tests/proxy_behavior/management/",
 ]
 also_copy = [
     "litellm/",

tests/llm_translation/realtime/base_realtime_tests.py

@@ -79,7 +79,7 @@ class RealTimeWebSocketClient:
 
     def _is_initial_event(self, msg_type: str) -> bool:
         """Check if message type is an initial connection event"""
-        # OpenAI sends "session.created", xAI sends "conversation.created"
+        # OpenAI and xAI send "session.created"; some providers send "conversation.created"
         return msg_type in ["session.created", "conversation.created"]
 
     async def receive_text(self):

tests/llm_translation/realtime/test_xai_realtime.py

@@ -19,8 +19,8 @@ class TestXAIRealtime(BaseRealtimeTest):
     """
     E2E tests for xAI Realtime API.
 
-    xAI's Grok Voice Agent API is OpenAI-compatible but uses:
-    - Different initial event: "conversation.created" instead of "session.created"
+    xAI's Grok Voice Agent API is OpenAI-compatible:
+    - Initial event: "session.created" (matches OpenAI)
     - Different endpoint: wss://api.x.ai/v1/realtime
     - Model: grok-4-1-fast-non-reasoning
     """
@@ -32,4 +32,4 @@ class TestXAIRealtime(BaseRealtimeTest):
         return "XAI_API_KEY"
 
     def get_initial_event_type(self) -> str:
-        return "conversation.created"
+        return "session.created"

tests/proxy_behavior/management/actors.py

@@ -0,0 +1,257 @@
+"""8-actor read-world seed for the authz matrix tests."""
+
+import enum
+import uuid
+from dataclasses import dataclass
+from typing import Any, Dict
+
+from prisma import Json
+
+from litellm.proxy._types import LitellmUserRoles
+from litellm.proxy.utils import PrismaClient, hash_token
+
+
+class Actor(str, enum.Enum):
+    PROXY_ADMIN = "proxy_admin"
+    ORG_ADMIN = "org_admin"
+    TEAM_ADMIN = "team_admin"
+    INTERNAL_USER = "internal_user"
+    OWNER = "owner"
+    UNRELATED_SAME_ORG = "unrelated_same_org"
+    CROSS_ORG_USER = "cross_org_user"
+    SERVICE_ACCOUNT = "service_account"
+
+
+PREFIX = "behavior-pin-"
+ORG_A = PREFIX + "org-a"
+ORG_B = PREFIX + "org-b"
+TEAM_ALPHA = PREFIX + "team-alpha"
+TEAM_BETA = PREFIX + "team-beta"
+BUDGET_ID = PREFIX + "budget"
+
+
+@dataclass(frozen=True)
+class SeededKey:
+    user_id: str
+    cleartext: str
+    hashed: str
+
+
+@dataclass(frozen=True)
+class World:
+    org_a_id: str
+    org_b_id: str
+    team_alpha_id: str
+    team_beta_id: str
+    keys: Dict[Actor, SeededKey]
+
+
+def _new_clear_key() -> str:
+    return "sk-" + uuid.uuid4().hex
+
+
+def _actor_profile() -> Dict[Actor, Dict[str, Any]]:
+    return {
+        Actor.PROXY_ADMIN: {
+            "user_role": LitellmUserRoles.PROXY_ADMIN.value,
+            "team_id": None,
+            "organization_id": None,
+        },
+        Actor.ORG_ADMIN: {
+            "user_role": LitellmUserRoles.ORG_ADMIN.value,
+            "team_id": None,
+            "organization_id": ORG_A,
+        },
+        Actor.TEAM_ADMIN: {
+            "user_role": LitellmUserRoles.INTERNAL_USER.value,
+            "team_id": TEAM_ALPHA,
+            "organization_id": ORG_A,
+        },
+        Actor.INTERNAL_USER: {
+            "user_role": LitellmUserRoles.INTERNAL_USER.value,
+            "team_id": TEAM_ALPHA,
+            "organization_id": ORG_A,
+        },
+        Actor.OWNER: {
+            "user_role": LitellmUserRoles.INTERNAL_USER.value,
+            "team_id": TEAM_ALPHA,
+            "organization_id": ORG_A,
+        },
+        Actor.UNRELATED_SAME_ORG: {
+            "user_role": LitellmUserRoles.INTERNAL_USER.value,
+            "team_id": TEAM_ALPHA,
+            "organization_id": ORG_A,
+        },
+        Actor.CROSS_ORG_USER: {
+            "user_role": LitellmUserRoles.INTERNAL_USER.value,
+            "team_id": TEAM_BETA,
+            "organization_id": ORG_B,
+        },
+        Actor.SERVICE_ACCOUNT: {
+            "user_role": LitellmUserRoles.INTERNAL_USER.value,
+            "team_id": TEAM_ALPHA,
+            "organization_id": ORG_A,
+        },
+    }
+
+
+async def _wipe_world(prisma: PrismaClient) -> None:
+    await prisma.db.litellm_verificationtoken.delete_many(
+        where={"user_id": {"startswith": PREFIX}}
+    )
+    await prisma.db.litellm_organizationmembership.delete_many(
+        where={"user_id": {"startswith": PREFIX}}
+    )
+    await prisma.db.litellm_teammembership.delete_many(
+        where={"user_id": {"startswith": PREFIX}}
+    )
+    await prisma.db.litellm_usertable.delete_many(
+        where={"user_id": {"startswith": PREFIX}}
+    )
+    await prisma.db.litellm_teamtable.delete_many(
+        where={"team_id": {"startswith": PREFIX}}
+    )
+    await prisma.db.litellm_organizationtable.delete_many(
+        where={"organization_id": {"startswith": PREFIX}}
+    )
+    await prisma.db.litellm_budgettable.delete_many(where={"budget_id": BUDGET_ID})
+
+
+async def seed_world(prisma: PrismaClient) -> World:
+    await _wipe_world(prisma)
+
+    await prisma.db.litellm_budgettable.create(
+        data={
+            "budget_id": BUDGET_ID,
+            "created_by": "behavior-pin-seeder",
+            "updated_by": "behavior-pin-seeder",
+        }
+    )
+
+    for org_id, alias in [(ORG_A, "alpha"), (ORG_B, "beta")]:
+        await prisma.db.litellm_organizationtable.create(
+            data={
+                "organization_id": org_id,
+                "organization_alias": alias,
+                "budget_id": BUDGET_ID,
+                "created_by": "behavior-pin-seeder",
+                "updated_by": "behavior-pin-seeder",
+            }
+        )
+
+    profiles = _actor_profile()
+    user_ids: Dict[Actor, str] = {actor: PREFIX + actor.value for actor in Actor}
+
+    for actor, profile in profiles.items():
+        teams_list = [profile["team_id"]] if profile["team_id"] else []
+        await prisma.db.litellm_usertable.create(
+            data={
+                "user_id": user_ids[actor],
+                "user_role": profile["user_role"],
+                "team_id": profile["team_id"],
+                "organization_id": profile["organization_id"],
+                "teams": teams_list,
+            }
+        )
+
+    # _get_user_in_team in key_management_endpoints.py walks members_with_roles
+    # (a JSON list of {user_id, role}), not the String[] members column —
+    # populate both to match what /team/new produces.
+    await prisma.db.litellm_teamtable.create(
+        data={
+            "team_id": TEAM_ALPHA,
+            "team_alias": "alpha-1",
+            "organization_id": ORG_A,
+            "admins": [user_ids[Actor.TEAM_ADMIN]],
+            "members": [
+                user_ids[Actor.TEAM_ADMIN],
+                user_ids[Actor.INTERNAL_USER],
+                user_ids[Actor.OWNER],
+                user_ids[Actor.UNRELATED_SAME_ORG],
+                user_ids[Actor.SERVICE_ACCOUNT],
+            ],
+            "members_with_roles": Json(
+                [
+                    {"user_id": user_ids[Actor.TEAM_ADMIN], "role": "admin"},
+                    {"user_id": user_ids[Actor.INTERNAL_USER], "role": "user"},
+                    {"user_id": user_ids[Actor.OWNER], "role": "user"},
+                    {"user_id": user_ids[Actor.UNRELATED_SAME_ORG], "role": "user"},
+                    {"user_id": user_ids[Actor.SERVICE_ACCOUNT], "role": "user"},
+                ]
+            ),
+        }
+    )
+    await prisma.db.litellm_teamtable.create(
+        data={
+            "team_id": TEAM_BETA,
+            "team_alias": "beta-1",
+            "organization_id": ORG_B,
+            "admins": [],
+            "members": [user_ids[Actor.CROSS_ORG_USER]],
+            "members_with_roles": Json(
+                [
+                    {"user_id": user_ids[Actor.CROSS_ORG_USER], "role": "user"},
+                ]
+            ),
+        }
+    )
+
+    for actor, org_id, role in [
+        (Actor.ORG_ADMIN, ORG_A, "org_admin"),
+        (Actor.TEAM_ADMIN, ORG_A, "internal_user"),
+        (Actor.INTERNAL_USER, ORG_A, "internal_user"),
+        (Actor.OWNER, ORG_A, "internal_user"),
+        (Actor.UNRELATED_SAME_ORG, ORG_A, "internal_user"),
+        (Actor.SERVICE_ACCOUNT, ORG_A, "internal_user"),
+        (Actor.CROSS_ORG_USER, ORG_B, "internal_user"),
+    ]:
+        await prisma.db.litellm_organizationmembership.create(
+            data={
+                "user_id": user_ids[actor],
+                "organization_id": org_id,
+                "user_role": role,
+            }
+        )
+
+    for actor, team_id in [
+        (Actor.TEAM_ADMIN, TEAM_ALPHA),
+        (Actor.INTERNAL_USER, TEAM_ALPHA),
+        (Actor.OWNER, TEAM_ALPHA),
+        (Actor.UNRELATED_SAME_ORG, TEAM_ALPHA),
+        (Actor.SERVICE_ACCOUNT, TEAM_ALPHA),
+        (Actor.CROSS_ORG_USER, TEAM_BETA),
+    ]:
+        await prisma.db.litellm_teammembership.create(
+            data={"user_id": user_ids[actor], "team_id": team_id}
+        )
+
+    keys: Dict[Actor, SeededKey] = {}
+    for actor, profile in profiles.items():
+        cleartext = _new_clear_key()
+        hashed = hash_token(cleartext)
+        token_data: Dict[str, Any] = {
+            "token": hashed,
+            "key_name": PREFIX + actor.value + "-key",
+            "user_id": user_ids[actor],
+            # LiteLLM_VerificationTokenView's models field rejects NULL even
+            # though the column is nullable in Postgres.
+            "models": [],
+        }
+        if profile["team_id"]:
+            token_data["team_id"] = profile["team_id"]
+        if profile["organization_id"]:
+            token_data["organization_id"] = profile["organization_id"]
+        if actor == Actor.SERVICE_ACCOUNT:
+            token_data["metadata"] = Json({"service_account_id": user_ids[actor]})
+        await prisma.db.litellm_verificationtoken.create(data=token_data)
+        keys[actor] = SeededKey(
+            user_id=user_ids[actor], cleartext=cleartext, hashed=hashed
+        )
+
+    return World(
+        org_a_id=ORG_A,
+        org_b_id=ORG_B,
+        team_alpha_id=TEAM_ALPHA,
+        team_beta_id=TEAM_BETA,
+        keys=keys,
+    )

tests/proxy_behavior/management/conftest.py

@@ -0,0 +1,156 @@
+"""Session-scoped async ASGI client for HTTP-boundary behavior tests."""
+
+import os
+import tempfile
+import uuid
+from dataclasses import dataclass
+from typing import Any, AsyncIterator, Dict, Optional
+
+import httpx
+import pytest_asyncio
+import yaml
+
+
+MASTER_KEY = "sk-1234"
+SCRATCH_PREFIX = "scratch-"
+
+
+def _write_minimal_proxy_config() -> str:
+    config = {
+        "general_settings": {"master_key": MASTER_KEY},
+        "litellm_settings": {},
+    }
+    database_url = os.environ.get("DATABASE_URL")
+    if database_url:
+        config["general_settings"]["database_url"] = database_url
+    f = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False)
+    yaml.dump(config, f)
+    f.close()
+    return f.name
+
+
+@pytest_asyncio.fixture(scope="session")
+async def proxy_app():
+    from litellm.proxy import proxy_server
+    from litellm.proxy.proxy_server import (
+        app,
+        cleanup_router_config_variables,
+        initialize,
+        proxy_startup_event,
+    )
+
+    cleanup_router_config_variables()
+    config_path = _write_minimal_proxy_config()
+
+    # proxy_startup_event re-reads master_key from LITELLM_MASTER_KEY and
+    # unconditionally overwrites the global, even when initialize() already
+    # set it from the config YAML. Force (not setdefault) both vars: an
+    # ambient LITELLM_MASTER_KEY with a different value would make the proxy
+    # authenticate on that key while the tests still send MASTER_KEY.
+    os.environ["LITELLM_MASTER_KEY"] = MASTER_KEY
+    os.environ["CONFIG_FILE_PATH"] = config_path
+
+    await initialize(config=config_path)
+
+    # /key/regenerate is gated behind premium_user; flipping it lets the matrix
+    # pin authz behavior instead of the licensing gate.
+    proxy_server.premium_user = True
+
+    async with proxy_startup_event(app):
+        proxy_server.premium_user = True  # lifespan re-runs _license_check
+        # The lifespan fires check_view_exists() as a background task; on a
+        # fresh DB the first auth call races it and resolves user_id=None.
+        if proxy_server.prisma_client is not None:
+            await proxy_server.prisma_client.check_view_exists()
+        yield app
+
+
+@pytest_asyncio.fixture(scope="session")
+async def proxy_client(proxy_app) -> AsyncIterator[httpx.AsyncClient]:
+    transport = httpx.ASGITransport(app=proxy_app)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="http://testserver"
+    ) as client:
+        yield client
+
+
+@pytest_asyncio.fixture(scope="session")
+async def prisma(proxy_app):
+    from litellm.proxy import proxy_server
+
+    assert proxy_server.prisma_client is not None
+    return proxy_server.prisma_client
+
+
+@pytest_asyncio.fixture(scope="session")
+async def world(prisma):
+    from .actors import seed_world
+
+    return await seed_world(prisma)
+
+
+@dataclass(frozen=True)
+class Scratch:
+    prefix: str
+
+    def tag(self, suffix: str = "") -> str:
+        return f"{self.prefix}-{suffix}" if suffix else self.prefix
+
+
+async def create_scratch_key(
+    proxy_client,
+    seeder_cleartext: str,
+    scratch_prefix: str,
+    *,
+    user_id: str,
+    team_id: Optional[str] = None,
+    organization_id: Optional[str] = None,
+) -> str:
+    """Seed a scratch-tagged key via /key/generate; returns its cleartext.
+
+    Shared by the write-scenario matrices (key update/regenerate/delete).
+    """
+    body: Dict[str, Any] = {"key_alias": scratch_prefix, "user_id": user_id}
+    if team_id is not None:
+        body["team_id"] = team_id
+    if organization_id is not None:
+        body["organization_id"] = organization_id
+    resp = await proxy_client.post(
+        "/key/generate",
+        headers={"Authorization": f"Bearer {seeder_cleartext}"},
+        json=body,
+    )
+    assert resp.status_code == 200, f"setup failed: {resp.text}"
+    return resp.json()["key"]
+
+
+@pytest_asyncio.fixture
+async def scratch(prisma):
+    handle = Scratch(prefix=f"{SCRATCH_PREFIX}{uuid.uuid4().hex[:12]}")
+    try:
+        yield handle
+    finally:
+        # Children before parents to avoid FK violations.
+        await prisma.db.litellm_verificationtoken.delete_many(
+            where={
+                "OR": [
+                    {"key_alias": {"startswith": handle.prefix}},
+                    {"key_name": {"startswith": handle.prefix}},
+                ]
+            }
+        )
+        await prisma.db.litellm_teammembership.delete_many(
+            where={"team_id": {"startswith": handle.prefix}}
+        )
+        await prisma.db.litellm_organizationmembership.delete_many(
+            where={"user_id": {"startswith": handle.prefix}}
+        )
+        await prisma.db.litellm_teamtable.delete_many(
+            where={"team_id": {"startswith": handle.prefix}}
+        )
+        await prisma.db.litellm_usertable.delete_many(
+            where={"user_id": {"startswith": handle.prefix}}
+        )
+        await prisma.db.litellm_budgettable.delete_many(
+            where={"budget_id": {"startswith": handle.prefix}}
+        )

tests/proxy_behavior/management/test_key_delete.py

@@ -0,0 +1,101 @@
+import pytest
+
+from litellm.proxy.utils import hash_token
+
+from .actors import TEAM_ALPHA, TEAM_BETA, Actor
+from .conftest import create_scratch_key
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+# Same-team peers can READ each other's keys (see test_key_info) but cannot
+# DELETE them — delete is stricter than read.
+_SCENARIOS = [
+    ("self/proxy_admin", Actor.PROXY_ADMIN, "self", 200),
+    ("self/org_admin", Actor.ORG_ADMIN, "self", 401),
+    ("self/team_admin", Actor.TEAM_ADMIN, "self", 200),
+    ("self/internal_user", Actor.INTERNAL_USER, "self", 200),
+    ("self/owner", Actor.OWNER, "self", 200),
+    ("self/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "self", 200),
+    ("self/cross_org_user", Actor.CROSS_ORG_USER, "self", 200),
+    ("self/service_account", Actor.SERVICE_ACCOUNT, "self", 200),
+    ("owner_target/proxy_admin", Actor.PROXY_ADMIN, "owner", 200),
+    ("owner_target/org_admin", Actor.ORG_ADMIN, "owner", 401),
+    ("owner_target/team_admin", Actor.TEAM_ADMIN, "owner", 200),
+    ("owner_target/internal_user", Actor.INTERNAL_USER, "owner", 403),
+    ("owner_target/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "owner", 403),
+    ("owner_target/cross_org_user", Actor.CROSS_ORG_USER, "owner", 403),
+    ("owner_target/service_account", Actor.SERVICE_ACCOUNT, "owner", 403),
+    ("cross_org_target/proxy_admin", Actor.PROXY_ADMIN, "cross_org", 200),
+    ("cross_org_target/org_admin", Actor.ORG_ADMIN, "cross_org", 401),
+    ("cross_org_target/team_admin", Actor.TEAM_ADMIN, "cross_org", 403),
+    ("cross_org_target/owner", Actor.OWNER, "cross_org", 403),
+    ("cross_org_target/cross_org_user", Actor.CROSS_ORG_USER, "cross_org", 200),
+    ("cross_org_target/service_account", Actor.SERVICE_ACCOUNT, "cross_org", 403),
+]
+
+
+@pytest.mark.parametrize(
+    "actor,target_shape,expected_status",
+    [(a, t, s) for (_id, a, t, s) in _SCENARIOS],
+    ids=[s[0] for s in _SCENARIOS],
+)
+async def test_key_delete_authz_matrix(
+    actor: Actor,
+    target_shape: str,
+    expected_status: int,
+    proxy_client,
+    prisma,
+    scratch,
+    world,
+):
+    caller = world.keys[actor]
+    seeder = world.keys[Actor.PROXY_ADMIN].cleartext
+
+    if target_shape == "self":
+        target_cleartext = await create_scratch_key(
+            proxy_client, seeder, scratch.prefix, user_id=caller.user_id
+        )
+    elif target_shape == "owner":
+        target_cleartext = await create_scratch_key(
+            proxy_client,
+            seeder,
+            scratch.prefix,
+            user_id=world.keys[Actor.OWNER].user_id,
+            team_id=TEAM_ALPHA,
+        )
+    elif target_shape == "cross_org":
+        target_cleartext = await create_scratch_key(
+            proxy_client,
+            seeder,
+            scratch.prefix,
+            user_id=world.keys[Actor.CROSS_ORG_USER].user_id,
+            team_id=TEAM_BETA,
+        )
+    else:
+        pytest.fail(f"unknown target_shape={target_shape}")
+
+    target_hashed = hash_token(target_cleartext)
+
+    resp = await proxy_client.post(
+        "/key/delete",
+        headers={"Authorization": f"Bearer {caller.cleartext}"},
+        json={"keys": [target_cleartext]},
+    )
+    assert (
+        resp.status_code == expected_status
+    ), f"{actor.value} {target_shape}: {resp.status_code} {resp.text}"
+
+    row = await prisma.db.litellm_verificationtoken.find_unique(
+        where={"token": target_hashed}
+    )
+    auth_check = await proxy_client.get(
+        "/key/info", headers={"Authorization": f"Bearer {target_cleartext}"}
+    )
+
+    if expected_status == 200:
+        # Hard- or soft-delete both produce a 401 on subsequent auth.
+        assert auth_check.status_code == 401
+    else:
+        assert row is not None, f"{actor.value}: denied but row vanished"
+        assert auth_check.status_code == 200

tests/proxy_behavior/management/test_key_generate.py

@@ -0,0 +1,70 @@
+from typing import Any, Dict
+
+import pytest
+
+from .actors import TEAM_ALPHA, TEAM_BETA, Actor
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+# (id, actor, body_extras, expected_status). Status codes pinned to observed
+# handler behavior — heterogeneous (200, 400, 401) because the handler routes
+# denials through three different gates (role gate, user_id mismatch, team
+# member permission).
+_SCENARIOS = [
+    ("self/proxy_admin", Actor.PROXY_ADMIN, {}, 200),
+    ("self/org_admin", Actor.ORG_ADMIN, {}, 401),
+    ("self/team_admin", Actor.TEAM_ADMIN, {}, 200),
+    ("self/internal_user", Actor.INTERNAL_USER, {}, 200),
+    ("self/owner", Actor.OWNER, {}, 200),
+    ("self/unrelated_same_org", Actor.UNRELATED_SAME_ORG, {}, 200),
+    ("self/cross_org_user", Actor.CROSS_ORG_USER, {}, 200),
+    ("self/service_account", Actor.SERVICE_ACCOUNT, {}, 200),
+    ("team_alpha/proxy_admin", Actor.PROXY_ADMIN, {"team_id": TEAM_ALPHA}, 200),
+    ("team_alpha/org_admin", Actor.ORG_ADMIN, {"team_id": TEAM_ALPHA}, 401),
+    ("team_alpha/team_admin", Actor.TEAM_ADMIN, {"team_id": TEAM_ALPHA}, 200),
+    ("team_alpha/internal_user", Actor.INTERNAL_USER, {"team_id": TEAM_ALPHA}, 401),
+    ("team_alpha/cross_org_user", Actor.CROSS_ORG_USER, {"team_id": TEAM_ALPHA}, 400),
+    ("team_beta/proxy_admin", Actor.PROXY_ADMIN, {"team_id": TEAM_BETA}, 200),
+    ("team_beta/org_admin", Actor.ORG_ADMIN, {"team_id": TEAM_BETA}, 401),
+    ("team_beta/team_admin", Actor.TEAM_ADMIN, {"team_id": TEAM_BETA}, 400),
+    ("team_beta/internal_user", Actor.INTERNAL_USER, {"team_id": TEAM_BETA}, 400),
+    ("team_beta/cross_org_user", Actor.CROSS_ORG_USER, {"team_id": TEAM_BETA}, 401),
+]
+
+
+@pytest.mark.parametrize(
+    "actor,body_extras,expected_status",
+    [(actor, body, expected) for (_id, actor, body, expected) in _SCENARIOS],
+    ids=[s[0] for s in _SCENARIOS],
+)
+async def test_key_generate_authz_matrix(
+    actor: Actor,
+    body_extras: Dict[str, Any],
+    expected_status: int,
+    proxy_client,
+    prisma,
+    scratch,
+    world,
+):
+    seeded = world.keys[actor]
+    body: Dict[str, Any] = {"key_alias": scratch.prefix, **body_extras}
+
+    resp = await proxy_client.post(
+        "/key/generate",
+        headers={"Authorization": f"Bearer {seeded.cleartext}"},
+        json=body,
+    )
+    assert (
+        resp.status_code == expected_status
+    ), f"{actor.value} {body!r} → {resp.status_code}: {resp.text}"
+
+    rows = await prisma.db.litellm_verificationtoken.find_many(
+        where={"key_alias": scratch.prefix}
+    )
+    if expected_status == 200:
+        cleartext = resp.json()["key"]
+        assert cleartext.startswith("sk-")
+        assert len(rows) == 1
+    else:
+        assert rows == [], f"{actor.value}: denied but row leaked"

tests/proxy_behavior/management/test_key_info.py

@@ -0,0 +1,74 @@
+import pytest
+
+from .actors import Actor
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+# (id, actor, target_actor, expected_status). Targets are 3 fixed seeded keys
+# representing the canonical relations: own, OWNER (same org_a/team_alpha),
+# and CROSS_ORG_USER (org_b/team_beta).
+#
+# Notable pinned behaviors (intentionally surfaced, not endorsed):
+#   - ORG_ADMIN 403s on individual key info even within its own org —
+#     visibility is "your own keys" + "your team's keys", not "your org's keys".
+#   - Same-team peers (internal_user, unrelated_same_org, service_account) DO
+#     see each other's keys.
+_SCENARIOS = [
+    ("own/proxy_admin", Actor.PROXY_ADMIN, Actor.PROXY_ADMIN, 200),
+    ("own/org_admin", Actor.ORG_ADMIN, Actor.ORG_ADMIN, 200),
+    ("own/team_admin", Actor.TEAM_ADMIN, Actor.TEAM_ADMIN, 200),
+    ("own/internal_user", Actor.INTERNAL_USER, Actor.INTERNAL_USER, 200),
+    ("own/owner", Actor.OWNER, Actor.OWNER, 200),
+    ("own/unrelated_same_org", Actor.UNRELATED_SAME_ORG, Actor.UNRELATED_SAME_ORG, 200),
+    ("own/cross_org_user", Actor.CROSS_ORG_USER, Actor.CROSS_ORG_USER, 200),
+    ("own/service_account", Actor.SERVICE_ACCOUNT, Actor.SERVICE_ACCOUNT, 200),
+    ("owner_key/proxy_admin", Actor.PROXY_ADMIN, Actor.OWNER, 200),
+    ("owner_key/org_admin", Actor.ORG_ADMIN, Actor.OWNER, 403),
+    ("owner_key/team_admin", Actor.TEAM_ADMIN, Actor.OWNER, 200),
+    ("owner_key/internal_user", Actor.INTERNAL_USER, Actor.OWNER, 200),
+    ("owner_key/owner", Actor.OWNER, Actor.OWNER, 200),
+    ("owner_key/unrelated_same_org", Actor.UNRELATED_SAME_ORG, Actor.OWNER, 200),
+    ("owner_key/cross_org_user", Actor.CROSS_ORG_USER, Actor.OWNER, 403),
+    ("owner_key/service_account", Actor.SERVICE_ACCOUNT, Actor.OWNER, 200),
+    ("cross_org/proxy_admin", Actor.PROXY_ADMIN, Actor.CROSS_ORG_USER, 200),
+    ("cross_org/org_admin", Actor.ORG_ADMIN, Actor.CROSS_ORG_USER, 403),
+    ("cross_org/team_admin", Actor.TEAM_ADMIN, Actor.CROSS_ORG_USER, 403),
+    ("cross_org/internal_user", Actor.INTERNAL_USER, Actor.CROSS_ORG_USER, 403),
+    ("cross_org/owner", Actor.OWNER, Actor.CROSS_ORG_USER, 403),
+    (
+        "cross_org/unrelated_same_org",
+        Actor.UNRELATED_SAME_ORG,
+        Actor.CROSS_ORG_USER,
+        403,
+    ),
+    ("cross_org/cross_org_user", Actor.CROSS_ORG_USER, Actor.CROSS_ORG_USER, 200),
+    ("cross_org/service_account", Actor.SERVICE_ACCOUNT, Actor.CROSS_ORG_USER, 403),
+]
+
+
+@pytest.mark.parametrize(
+    "actor,target_actor,expected_status",
+    [(a, t, s) for (_id, a, t, s) in _SCENARIOS],
+    ids=[s[0] for s in _SCENARIOS],
+)
+async def test_key_info_authz_matrix(
+    actor: Actor, target_actor: Actor, expected_status: int, proxy_client, world
+):
+    caller = world.keys[actor]
+    target = world.keys[target_actor]
+
+    resp = await proxy_client.get(
+        f"/key/info?key={target.cleartext}",
+        headers={"Authorization": f"Bearer {caller.cleartext}"},
+    )
+    assert (
+        resp.status_code == expected_status
+    ), f"{actor.value} → {target_actor.value}: {resp.status_code} {resp.text}"
+
+    if expected_status == 200:
+        body = resp.json()
+        # The handler echoes back whatever ?key was passed (cleartext here),
+        # so accept either form — info.user_id is the canonical identity check.
+        assert body.get("key") in (target.cleartext, target.hashed)
+        assert body["info"].get("user_id") == target.user_id

tests/proxy_behavior/management/test_key_list.py

@@ -0,0 +1,63 @@
+from typing import FrozenSet
+
+import pytest
+
+from .actors import Actor
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+# Pinned default visibility for /key/list (no filter params): each actor's
+# expected set of seeded actor keys.
+_VISIBILITY = {
+    Actor.PROXY_ADMIN: frozenset(Actor),
+    Actor.ORG_ADMIN: frozenset({Actor.ORG_ADMIN}),
+    Actor.TEAM_ADMIN: frozenset({Actor.TEAM_ADMIN}),
+    Actor.INTERNAL_USER: frozenset({Actor.INTERNAL_USER}),
+    Actor.OWNER: frozenset({Actor.OWNER}),
+    Actor.UNRELATED_SAME_ORG: frozenset({Actor.UNRELATED_SAME_ORG}),
+    Actor.CROSS_ORG_USER: frozenset({Actor.CROSS_ORG_USER}),
+    Actor.SERVICE_ACCOUNT: frozenset({Actor.SERVICE_ACCOUNT}),
+}
+
+
+async def _all_visible_hashes(proxy_client, caller_cleartext) -> set:
+    """Walk every /key/list page — size is capped at 100 by the endpoint, so a
+    single request can truncate PROXY_ADMIN's view on a non-fresh DB."""
+    hashes: set = set()
+    page = 1
+    while True:
+        resp = await proxy_client.get(
+            f"/key/list?page={page}&size=100",
+            headers={"Authorization": f"Bearer {caller_cleartext}"},
+        )
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        for entry in body.get("keys", []):
+            tok = entry.get("token") if isinstance(entry, dict) else entry
+            if tok:
+                hashes.add(tok)
+        if page >= (body.get("total_pages") or 1):
+            return hashes
+        page += 1
+
+
+@pytest.mark.parametrize(
+    "actor,expected_visible",
+    list(_VISIBILITY.items()),
+    ids=[a.value for a in _VISIBILITY],
+)
+async def test_key_list_visibility(
+    actor: Actor, expected_visible: FrozenSet[Actor], proxy_client, world
+):
+    caller = world.keys[actor]
+    hashed_to_actor = {world.keys[a].hashed: a for a in Actor}
+
+    returned_hashes = await _all_visible_hashes(proxy_client, caller.cleartext)
+    visible_seeded = {
+        hashed_to_actor[h] for h in returned_hashes if h in hashed_to_actor
+    }
+    assert visible_seeded == set(expected_visible), (
+        f"{actor.value}: expected {sorted(a.value for a in expected_visible)}, "
+        f"got {sorted(a.value for a in visible_seeded)}"
+    )

tests/proxy_behavior/management/test_key_regenerate.py

@@ -0,0 +1,117 @@
+import pytest
+
+from .actors import TEAM_ALPHA, TEAM_BETA, Actor
+from .conftest import create_scratch_key
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+# Most denials route through team_member_permission (401), unlike /key/update
+# which goes through user_id-mismatch (403). The matrix surfaces that
+# divergence between the two endpoints.
+_SCENARIOS = [
+    ("self/proxy_admin", Actor.PROXY_ADMIN, "self", 200),
+    ("self/org_admin", Actor.ORG_ADMIN, "self", 401),
+    ("self/team_admin", Actor.TEAM_ADMIN, "self", 200),
+    ("self/internal_user", Actor.INTERNAL_USER, "self", 200),
+    ("self/owner", Actor.OWNER, "self", 200),
+    ("self/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "self", 200),
+    ("self/cross_org_user", Actor.CROSS_ORG_USER, "self", 200),
+    ("self/service_account", Actor.SERVICE_ACCOUNT, "self", 200),
+    ("owner_target/proxy_admin", Actor.PROXY_ADMIN, "owner", 200),
+    ("owner_target/org_admin", Actor.ORG_ADMIN, "owner", 401),
+    ("owner_target/team_admin", Actor.TEAM_ADMIN, "owner", 200),
+    ("owner_target/internal_user", Actor.INTERNAL_USER, "owner", 401),
+    ("owner_target/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "owner", 401),
+    ("owner_target/cross_org_user", Actor.CROSS_ORG_USER, "owner", 401),
+    ("owner_target/service_account", Actor.SERVICE_ACCOUNT, "owner", 401),
+    ("cross_org_target/proxy_admin", Actor.PROXY_ADMIN, "cross_org", 200),
+    ("cross_org_target/org_admin", Actor.ORG_ADMIN, "cross_org", 401),
+    ("cross_org_target/team_admin", Actor.TEAM_ADMIN, "cross_org", 401),
+    ("cross_org_target/owner", Actor.OWNER, "cross_org", 401),
+    ("cross_org_target/cross_org_user", Actor.CROSS_ORG_USER, "cross_org", 401),
+    ("cross_org_target/service_account", Actor.SERVICE_ACCOUNT, "cross_org", 401),
+]
+
+
+async def _info(proxy_client, cleartext: str):
+    return await proxy_client.get(
+        "/key/info", headers={"Authorization": f"Bearer {cleartext}"}
+    )
+
+
+@pytest.mark.parametrize(
+    "actor,target_shape,expected_status",
+    [(a, t, s) for (_id, a, t, s) in _SCENARIOS],
+    ids=[s[0] for s in _SCENARIOS],
+)
+async def test_key_regenerate_authz_matrix(
+    actor: Actor,
+    target_shape: str,
+    expected_status: int,
+    proxy_client,
+    scratch,
+    world,
+):
+    caller = world.keys[actor]
+    seeder = world.keys[Actor.PROXY_ADMIN].cleartext
+
+    if target_shape == "self":
+        target_cleartext = await create_scratch_key(
+            proxy_client, seeder, scratch.prefix, user_id=caller.user_id
+        )
+    elif target_shape == "owner":
+        target_cleartext = await create_scratch_key(
+            proxy_client,
+            seeder,
+            scratch.prefix,
+            user_id=world.keys[Actor.OWNER].user_id,
+            team_id=TEAM_ALPHA,
+        )
+    elif target_shape == "cross_org":
+        target_cleartext = await create_scratch_key(
+            proxy_client,
+            seeder,
+            scratch.prefix,
+            user_id=world.keys[Actor.CROSS_ORG_USER].user_id,
+            team_id=TEAM_BETA,
+        )
+    else:
+        pytest.fail(f"unknown target_shape={target_shape}")
+
+    resp = await proxy_client.post(
+        "/key/regenerate",
+        headers={"Authorization": f"Bearer {caller.cleartext}"},
+        json={"key": target_cleartext},
+    )
+    assert (
+        resp.status_code == expected_status
+    ), f"{actor.value} {target_shape}: {resp.status_code} {resp.text}"
+
+    if expected_status == 200:
+        new_cleartext = resp.json()["key"]
+        assert new_cleartext.startswith("sk-") and new_cleartext != target_cleartext
+        assert (await _info(proxy_client, target_cleartext)).status_code == 401
+        assert (await _info(proxy_client, new_cleartext)).status_code == 200
+    else:
+        # Denied: rotation must not have leaked — old cleartext still works.
+        assert (await _info(proxy_client, target_cleartext)).status_code == 200
+
+
+async def test_key_path_regenerate_smoke(proxy_client, scratch, world):
+    """Pins that POST /key/{key:path}/regenerate shares the same handler."""
+    caller = world.keys[Actor.PROXY_ADMIN]
+    target_cleartext = await create_scratch_key(
+        proxy_client, caller.cleartext, scratch.prefix, user_id=caller.user_id
+    )
+
+    resp = await proxy_client.post(
+        f"/key/{target_cleartext}/regenerate",
+        headers={"Authorization": f"Bearer {caller.cleartext}"},
+        json={},
+    )
+    assert resp.status_code == 200, resp.text
+    new_cleartext = resp.json()["key"]
+    assert new_cleartext.startswith("sk-") and new_cleartext != target_cleartext
+    assert (await _info(proxy_client, target_cleartext)).status_code == 401
+    assert (await _info(proxy_client, new_cleartext)).status_code == 200

tests/proxy_behavior/management/test_key_update.py

@@ -0,0 +1,100 @@
+import pytest
+
+from litellm.proxy.utils import hash_token
+
+from .actors import TEAM_ALPHA, TEAM_BETA, Actor
+from .conftest import create_scratch_key
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+# (id, actor, target_shape, expected_status). Pinned against current gating:
+# proxy_admin bypasses; org_admin is blocked by an early role gate (401);
+# every other (INTERNAL_USER-roled) actor hits user_id-mismatch 403, no-team-
+# admin 403, or team_member_permission 401 depending on target / membership.
+_SCENARIOS = [
+    ("self/proxy_admin", Actor.PROXY_ADMIN, "self", 200),
+    ("self/org_admin", Actor.ORG_ADMIN, "self", 401),
+    ("self/team_admin", Actor.TEAM_ADMIN, "self", 403),
+    ("self/internal_user", Actor.INTERNAL_USER, "self", 403),
+    ("self/owner", Actor.OWNER, "self", 403),
+    ("self/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "self", 403),
+    ("self/cross_org_user", Actor.CROSS_ORG_USER, "self", 403),
+    ("self/service_account", Actor.SERVICE_ACCOUNT, "self", 403),
+    ("owner_target/proxy_admin", Actor.PROXY_ADMIN, "owner", 200),
+    ("owner_target/org_admin", Actor.ORG_ADMIN, "owner", 401),
+    ("owner_target/team_admin", Actor.TEAM_ADMIN, "owner", 403),
+    ("owner_target/internal_user", Actor.INTERNAL_USER, "owner", 403),
+    ("owner_target/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "owner", 403),
+    ("owner_target/cross_org_user", Actor.CROSS_ORG_USER, "owner", 403),
+    ("owner_target/service_account", Actor.SERVICE_ACCOUNT, "owner", 403),
+    ("cross_org_target/proxy_admin", Actor.PROXY_ADMIN, "cross_org", 200),
+    ("cross_org_target/org_admin", Actor.ORG_ADMIN, "cross_org", 401),
+    ("cross_org_target/team_admin", Actor.TEAM_ADMIN, "cross_org", 403),
+    ("cross_org_target/owner", Actor.OWNER, "cross_org", 403),
+    ("cross_org_target/cross_org_user", Actor.CROSS_ORG_USER, "cross_org", 401),
+    ("cross_org_target/service_account", Actor.SERVICE_ACCOUNT, "cross_org", 403),
+]
+
+MARKER_MODEL = "behavior-pin-update-marker-model"
+
+
+@pytest.mark.parametrize(
+    "actor,target_shape,expected_status",
+    [(a, t, s) for (_id, a, t, s) in _SCENARIOS],
+    ids=[s[0] for s in _SCENARIOS],
+)
+async def test_key_update_authz_matrix(
+    actor: Actor,
+    target_shape: str,
+    expected_status: int,
+    proxy_client,
+    prisma,
+    scratch,
+    world,
+):
+    caller = world.keys[actor]
+    seeder = world.keys[Actor.PROXY_ADMIN].cleartext
+
+    if target_shape == "self":
+        target_cleartext = await create_scratch_key(
+            proxy_client, seeder, scratch.prefix, user_id=caller.user_id
+        )
+    elif target_shape == "owner":
+        target_cleartext = await create_scratch_key(
+            proxy_client,
+            seeder,
+            scratch.prefix,
+            user_id=world.keys[Actor.OWNER].user_id,
+            team_id=TEAM_ALPHA,
+        )
+    elif target_shape == "cross_org":
+        target_cleartext = await create_scratch_key(
+            proxy_client,
+            seeder,
+            scratch.prefix,
+            user_id=world.keys[Actor.CROSS_ORG_USER].user_id,
+            team_id=TEAM_BETA,
+        )
+    else:
+        pytest.fail(f"unknown target_shape={target_shape}")
+
+    target_hashed = hash_token(target_cleartext)
+
+    resp = await proxy_client.post(
+        "/key/update",
+        headers={"Authorization": f"Bearer {caller.cleartext}"},
+        json={"key": target_cleartext, "models": [MARKER_MODEL]},
+    )
+    assert (
+        resp.status_code == expected_status
+    ), f"{actor.value} {target_shape}: {resp.status_code} {resp.text}"
+
+    row = await prisma.db.litellm_verificationtoken.find_unique(
+        where={"token": target_hashed}
+    )
+    assert row is not None
+    if expected_status == 200:
+        assert row.models == [MARKER_MODEL]
+    else:
+        assert row.models != [MARKER_MODEL], "denied but row mutated"

tests/proxy_behavior/management/test_no_management_imports.py

@@ -0,0 +1,46 @@
+import pathlib
+import re
+
+REPO_ROOT = pathlib.Path(__file__).resolve().parents[3]
+BEHAVIOR_DIR = REPO_ROOT / "tests" / "proxy_behavior"
+
+FORBIDDEN_IMPORT = re.compile(r"^\s*from\s+litellm\.proxy\.management_endpoints\b")
+FORBIDDEN_AUTH_MOCK = re.compile(
+    r"(?:mock\.[A-Za-z_]+|patch[a-z_]*)\([^)]*user_api_key_auth"
+)
+# This file is the only place the forbidden patterns appear as regex source;
+# exclude it so it can describe what it forbids.
+SELF = pathlib.Path(__file__).resolve()
+
+
+def _iter_py_files():
+    for path in BEHAVIOR_DIR.rglob("*.py"):
+        if path.resolve() != SELF:
+            yield path
+
+
+def _scan(pattern):
+    violations = []
+    for path in _iter_py_files():
+        for lineno, line in enumerate(path.read_text().splitlines(), start=1):
+            if pattern.search(line):
+                violations.append(
+                    f"{path.relative_to(REPO_ROOT)}:{lineno}: {line.strip()}"
+                )
+    return violations
+
+
+def test_no_management_endpoint_imports():
+    violations = _scan(FORBIDDEN_IMPORT)
+    assert not violations, (
+        "tests/proxy_behavior/ must not import from litellm.proxy.management_endpoints. "
+        "Violations:\n  " + "\n  ".join(violations)
+    )
+
+
+def test_no_user_api_key_auth_mocking():
+    violations = _scan(FORBIDDEN_AUTH_MOCK)
+    assert not violations, (
+        "tests/proxy_behavior/ must not mock user_api_key_auth. "
+        "Violations:\n  " + "\n  ".join(violations)
+    )

tests/proxy_behavior/management/test_scratch_teardown.py

@@ -0,0 +1,31 @@
+import pytest
+
+from .conftest import MASTER_KEY, SCRATCH_PREFIX
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+# The two tests run in file order: _a writes a scratch-tagged key and asserts
+# it lands; _b runs after _a's fixture teardown and asserts no scratch row
+# survived. A leak in either direction fails _b on the next collection.
+
+
+async def test_a_scratch_key_lands_in_db(proxy_client, prisma, scratch):
+    resp = await proxy_client.post(
+        "/key/generate",
+        headers={"Authorization": f"Bearer {MASTER_KEY}"},
+        json={"key_alias": scratch.prefix},
+    )
+    assert resp.status_code == 200, resp.text
+
+    rows = await prisma.db.litellm_verificationtoken.find_many(
+        where={"key_alias": scratch.prefix}
+    )
+    assert len(rows) == 1
+
+
+async def test_b_scratch_namespace_is_clean(prisma):
+    rows = await prisma.db.litellm_verificationtoken.find_many(
+        where={"key_alias": {"startswith": SCRATCH_PREFIX}}
+    )
+    assert rows == []

tests/proxy_behavior/management/test_smoke.py

@@ -0,0 +1,28 @@
+import pytest
+
+from .conftest import MASTER_KEY
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+async def test_liveliness(proxy_client):
+    resp = await proxy_client.get("/health/liveliness")
+    assert resp.status_code == 200
+
+
+async def test_key_generate_lands_in_db(proxy_client, prisma, scratch):
+    from litellm.proxy.utils import hash_token
+
+    resp = await proxy_client.post(
+        "/key/generate",
+        headers={"Authorization": f"Bearer {MASTER_KEY}"},
+        json={"key_alias": scratch.prefix},
+    )
+    assert resp.status_code == 200, resp.text
+    cleartext = resp.json()["key"]
+    assert cleartext.startswith("sk-")
+
+    hashed = hash_token(cleartext)
+    row = await prisma.db.litellm_verificationtoken.find_unique(where={"token": hashed})
+    assert row is not None
+    assert row.token == hashed != cleartext

tests/proxy_behavior/management/test_world_seed.py

@@ -0,0 +1,30 @@
+import pytest
+
+from .actors import Actor
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+@pytest.mark.parametrize("actor", list(Actor), ids=[a.value for a in Actor])
+async def test_each_actor_can_self_info(actor, proxy_client, world):
+    seeded = world.keys[actor]
+    resp = await proxy_client.get(
+        "/key/info",
+        headers={"Authorization": f"Bearer {seeded.cleartext}"},
+    )
+    assert resp.status_code == 200, f"{actor.value}: {resp.text}"
+    body = resp.json()
+    assert body.get("key") == seeded.hashed
+    assert body["info"].get("user_id") == seeded.user_id
+
+
+async def test_proxy_admin_actor_can_create_keys_for_others(proxy_client, world):
+    seeder = world.keys[Actor.PROXY_ADMIN]
+    target_user_id = world.keys[Actor.OWNER].user_id
+
+    resp = await proxy_client.post(
+        "/key/generate",
+        headers={"Authorization": f"Bearer {seeder.cleartext}"},
+        json={"key_alias": "smoke-proxy-admin-bypass", "user_id": target_user_id},
+    )
+    assert resp.status_code == 200, resp.text

tests/proxy_unit_tests/test_user_api_key_auth.py

@@ -915,6 +915,36 @@ async def test_user_api_key_auth_websocket():
         )
 
 
+@pytest.mark.asyncio
+async def test_user_api_key_auth_websocket_carries_asgi_path():
+    """
+    The synthetic Request must carry the ASGI scope's ``path`` so
+    ``get_request_route`` returns the real WebSocket path, not a value
+    reconstructed from the (Host-poisonable) ``websocket.url``.
+    """
+    from litellm.proxy.auth.user_api_key_auth import user_api_key_auth_websocket
+
+    mock_websocket = MagicMock(spec=WebSocket)
+    mock_websocket.query_params = {"model": "some_model"}
+    mock_websocket.headers = {"authorization": "Bearer some_api_key"}
+    mock_websocket.scope = {
+        "type": "websocket",
+        "path": "/v1/realtime",
+        "root_path": "",
+        "headers": [(b"authorization", b"Bearer some_api_key")],
+    }
+    mock_websocket.url = URL(url="/v1/realtime")
+
+    with patch(
+        "litellm.proxy.auth.user_api_key_auth.user_api_key_auth", autospec=True
+    ) as mock_user_api_key_auth:
+        await user_api_key_auth_websocket(mock_websocket)
+
+        request_arg = mock_user_api_key_auth.call_args.kwargs["request"]
+        assert request_arg.scope.get("path") == "/v1/realtime"
+        assert request_arg.scope.get("root_path") == ""
+
+
 @pytest.mark.parametrize("enforce_rbac", [True, False])
 @pytest.mark.asyncio
 async def test_jwt_user_api_key_auth_builder_enforce_rbac(enforce_rbac, monkeypatch):

tests/test_litellm/integrations/test_opentelemetry.py

@@ -66,7 +66,7 @@ class TestOpenTelemetryGuardrails(unittest.TestCase):
         mock_span.set_attribute.assert_any_call("guardrail_name", "test_guardrail")
         mock_span.set_attribute.assert_any_call("guardrail_mode", "input")
         mock_span.set_attribute.assert_any_call(
-            "guardrail_response", "filtered_content"
+            "guardrail_response", safe_dumps("filtered_content")
         )
         mock_span.set_attribute.assert_any_call(
             "masked_entity_count", safe_dumps({"CREDIT_CARD": 2})
@@ -87,6 +87,65 @@ class TestOpenTelemetryGuardrails(unittest.TestCase):
         # Verify that start_span was never called
         otel.tracer.start_span.assert_not_called()
 
+    @patch("litellm.integrations.opentelemetry.datetime")
+    def test_guardrail_response_dict_is_json_serialized(self, mock_datetime):
+        """Dict guardrail_response (e.g. OpenAI moderation result) must reach
+        the span as a JSON string so downstream pipelines can parse it for
+        metric extraction — this is the bug the PR fixes."""
+        otel = OpenTelemetry()
+        otel.tracer = MagicMock()
+        mock_span = MagicMock()
+        otel.tracer.start_span.return_value = mock_span
+
+        moderation_payload = {
+            "id": "modr-7740",
+            "model": "omni-moderation-latest",
+            "results": [{"categories": {"harassment": False}}],
+        }
+        guardrail_info = {
+            "guardrail_name": "test_guardrail",
+            "guardrail_mode": "input",
+            "guardrail_response": moderation_payload,
+            "start_time": 1609459200.0,
+            "end_time": 1609459201.0,
+        }
+        kwargs = {
+            "standard_logging_object": {"guardrail_information": [guardrail_info]}
+        }
+
+        otel._create_guardrail_span(kwargs=kwargs, context=None)
+
+        mock_span.set_attribute.assert_any_call(
+            "guardrail_response", safe_dumps(moderation_payload)
+        )
+
+    @patch("litellm.integrations.opentelemetry.datetime")
+    def test_guardrail_response_none_is_skipped(self, mock_datetime):
+        """When guardrail_response is None, the attribute must not be set —
+        guards against round-tripping ``"null"`` into traces."""
+        otel = OpenTelemetry()
+        otel.tracer = MagicMock()
+        mock_span = MagicMock()
+        otel.tracer.start_span.return_value = mock_span
+
+        guardrail_info = {
+            "guardrail_name": "test_guardrail",
+            "guardrail_mode": "input",
+            "guardrail_response": None,
+            "start_time": 1609459200.0,
+            "end_time": 1609459201.0,
+        }
+        kwargs = {
+            "standard_logging_object": {"guardrail_information": [guardrail_info]}
+        }
+
+        otel._create_guardrail_span(kwargs=kwargs, context=None)
+
+        attribute_keys = [
+            call.args[0] for call in mock_span.set_attribute.call_args_list
+        ]
+        self.assertNotIn("guardrail_response", attribute_keys)
+
 
 class TestOpenTelemetryTeamAttributesOnChildSpans(unittest.TestCase):
     """team_id / team_alias must land on every child span of a
@@ -1169,7 +1228,7 @@ class TestOpenTelemetry(unittest.TestCase):
         mock_span.set_attribute.assert_any_call("guardrail_name", "test_guardrail")
         mock_span.set_attribute.assert_any_call("guardrail_mode", "input")
         mock_span.set_attribute.assert_any_call(
-            "guardrail_response", "filtered_content"
+            "guardrail_response", safe_dumps("filtered_content")
         )
         mock_span.set_attribute.assert_any_call(
             "masked_entity_count", safe_dumps({"CREDIT_CARD": 2})

tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py

@@ -16,6 +16,7 @@ sys.path.insert(
 import litellm
 from litellm.proxy._types import ProxyException
 from litellm.proxy.common_utils.http_parsing_utils import (
+    _is_form_content_type,
     _read_request_body,
     _safe_get_request_headers,
     _safe_get_request_parsed_body,
@@ -853,3 +854,145 @@ class TestGetTagsFromRequestBodyStringCoerce:
 
         tags = get_tags_from_request_body({"metadata": {"tags": ["x"]}})
         assert tags == ["x"]
+
+
+class TestIsFormContentType:
+    @pytest.mark.parametrize(
+        "content_type",
+        [
+            "application/x-www-form-urlencoded",
+            "multipart/form-data",
+            "multipart/form-data; boundary=----WebKitFormBoundary",
+            "Application/X-WWW-Form-Urlencoded",
+            "  multipart/form-data  ",
+            "application/x-www-form-urlencoded; charset=utf-8",
+        ],
+    )
+    def test_form_types_match(self, content_type):
+        assert _is_form_content_type(content_type) is True
+
+    @pytest.mark.parametrize(
+        "content_type",
+        [
+            "",
+            "application/json",
+            "application/json; charset=utf-8",
+            "application/form-json",
+            "multiform/anything",
+            "application/json; xform=1",
+            "application/xml-with-form-data-but-not-actually",
+            "text/plain",
+            "form",
+        ],
+    )
+    def test_non_form_types_rejected(self, content_type):
+        assert _is_form_content_type(content_type) is False
+
+
+class TestReadRequestBodyNonCanonicalContentType:
+    """A JSON body with a ``"form"``-substring Content-Type must parse as JSON."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "content_type",
+        [
+            "application/form-json",
+            "application/json; xform=1",
+            "multiform/anything",
+        ],
+    )
+    async def test_json_body_with_formlike_content_type_parses_as_json(
+        self, content_type
+    ):
+        payload = {"user_config": {"model_list": []}, "model": "x"}
+
+        mock_request = MagicMock()
+        mock_request.body = AsyncMock(return_value=orjson.dumps(payload))
+        mock_request.form = AsyncMock(return_value={})
+        mock_request.headers = {"content-type": content_type}
+        mock_request.scope = {}
+
+        result = await _read_request_body(mock_request)
+        assert result == payload
+        mock_request.form.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_real_form_post_still_parsed_as_form(self):
+        mock_request = MagicMock()
+        mock_request.form = AsyncMock(return_value={"k": "v"})
+        mock_request.body = AsyncMock(return_value=b"")
+        mock_request.headers = {"content-type": "application/x-www-form-urlencoded"}
+        mock_request.scope = {}
+
+        result = await _read_request_body(mock_request)
+        assert result == {"k": "v"}
+        mock_request.form.assert_awaited_once()
+
+
+class TestReadRequestBodyFormParseFailure:
+    """
+    A failed ``request.form()`` parse (e.g. multipart with missing boundary)
+    must surface as a 400, not silently return ``{}`` — otherwise the
+    auth-time pre-read sees an empty body while a later raw-body re-read
+    sees the original payload, defeating every banned-param check.
+    """
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "raised_exception",
+        [
+            ValueError("Missing boundary in multipart."),
+            AssertionError("malformed chunk"),
+            RuntimeError("form parser exploded"),
+        ],
+    )
+    async def test_form_parse_failure_raises_400(self, raised_exception):
+        mock_request = MagicMock()
+        mock_request.form = AsyncMock(side_effect=raised_exception)
+        mock_request.headers = {"content-type": "multipart/form-data"}
+        mock_request.scope = {}
+
+        with pytest.raises(ProxyException) as exc_info:
+            await _read_request_body(mock_request)
+        assert str(exc_info.value.code) == "400"
+
+
+class TestGetRequestBody:
+    @pytest.mark.asyncio
+    async def test_json_with_charset_param_parses_as_json(self):
+        payload = {"k": "v"}
+        mock_request = MagicMock()
+        mock_request.method = "POST"
+        mock_request.body = AsyncMock(return_value=orjson.dumps(payload))
+        mock_request.headers = {"content-type": "application/json; charset=utf-8"}
+        mock_request.scope = {}
+
+        result = await get_request_body(mock_request)
+        assert result == payload
+
+    @pytest.mark.asyncio
+    async def test_form_post_routes_to_form_data(self):
+        mock_request = MagicMock()
+        mock_request.method = "POST"
+        mock_request.headers = {"content-type": "multipart/form-data; boundary=x"}
+        mock_request.form = AsyncMock(return_value={"k": "v"})
+        mock_request.scope = {}
+
+        result = await get_request_body(mock_request)
+        assert result == {"k": "v"}
+
+    @pytest.mark.asyncio
+    async def test_substring_match_no_longer_accepted(self):
+        mock_request = MagicMock()
+        mock_request.method = "POST"
+        mock_request.headers = {"content-type": "application/form-json"}
+        mock_request.scope = {}
+
+        with pytest.raises(ValueError, match="Unsupported content type"):
+            await get_request_body(mock_request)
+
+    @pytest.mark.asyncio
+    async def test_non_post_returns_empty(self):
+        mock_request = MagicMock()
+        mock_request.method = "GET"
+        assert await get_request_body(mock_request) == {}