test(proxy): behavior-pinning matrix for tier-2/3 key + team management endpoints (#28620)

* test(proxy): add create_scratch_actor harness helper

Adds create_scratch_actor() to the management behavior-suite conftest and
extends create_scratch_team() with team_member_permissions / models kwargs,
needed by the PR3 team-key-permission and team-model matrices. The new
helper mints a scratch-prefixed user + verification token (+ org
memberships), all reclaimed by the existing scratch-prefix teardown.

* test(proxy): pin /key block, unblock, health, aliases behavior

Adds behavior-pinning matrices for POST /key/block, POST /key/unblock,
POST /key/health, and GET /key/aliases. Pins that the management-route gate
401s ORG_ADMIN-role callers before _check_key_admin_access runs, the
block/unblock round-trip on the blocked column, missing-key 404, and the
_apply_non_admin_alias_scope visibility rules for /key/aliases.

* test(proxy): pin /key/bulk_update + /team/key/bulk_update behavior

Adds behavior-pinning matrices for POST /key/bulk_update (PROXY_ADMIN-only;
ORG_ADMIN stopped 401 at the route gate, INTERNAL_USER-role 403 at the
handler) and POST /team/key/bulk_update (team-member-permission gate keyed
on KEY_UPDATE). Pins batch semantics: empty/over-cap 400, per-key failure
isolation into failed_updates, all_keys_in_team broadcast, and no-keys 404.
Adds an optional key_alias arg to create_scratch_key for multi-key scenarios.

* test(proxy): pin /key SA-generate, v2-info, reset-spend behavior

Adds behavior-pinning matrices for POST /key/service-account/generate
(team-membership + team-member-permission gating; SA keys carry no user_id),
POST /v2/key/info (per-key _can_user_query_key_info silently drops invisible
keys), and POST /key/{key}/reset_spend (PROXY_ADMIN or team admin only;
missing key 404, reset-value 400). Pins that ORG_ADMIN-role callers are
stopped 401 at the management-route gate on the two non-info routes.

* test(proxy): close PR1/PR2 key-side deferred coverage gaps

Closes the four key-side gaps deferred from PR1/PR2:
- 404 on missing key for /key/update and /key/delete (not 401/403)
- denied /key/update leaves max_budget/tpm_limit/rpm_limit untouched
- /key/regenerate enforces litellm.upperbound_key_generate_params (#26340)
- /key/list key_alias substring vs exact (admin-only) + team_id filter,
  and a non-admin filtering a foreign team is 403

* test(proxy): pin /team block, unblock, available, filter/ui, members/me

Adds behavior-pinning matrices for POST /team/block + /team/unblock
(management-route gate fronts _verify_team_access; reachable only by
PROXY_ADMIN and an org admin of the team's own org), GET /team/available
(default empty path), GET /team/filter/ui (route-gated PROXY-ADMIN-only
despite the handler having no gate), and GET /team/{team_id}/members/me
(caller resolves its own membership; non-member 404, no-user_id key 400).

* test(proxy): pin /team model add/delete + permissions endpoints

Adds behavior-pinning matrices for POST /team/model/add + /team/model/delete
(route-gated PROXY-ADMIN-only; missing team 404), GET /team/permissions_list +
POST /team/permissions_update (self-managed; proxy/team/org admin pass), and
POST /team/permissions_bulk_update (PROXY_ADMIN-only). Pins the deliberate
divergence that the available-team self-join grants read access via
permissions_list but never write access via permissions_update.

* test(proxy): pin /team delete, bulk_member_add, v2/list, daily/activity

Adds behavior-pinning matrices for POST /team/delete (per-team
_verify_team_access; batch aborts whole on a missing id), POST
/team/bulk_member_add (route-gated PROXY-ADMIN-only; empty/over-cap 400),
GET /v2/team/list (_enforce_list_team_v2_access — bare query 401s regular
users, org-scoped for org admins) and GET /team/daily/activity (non-member
team_ids filter 404, the VERIA-43 fix).

* test(proxy): add route-coverage gate + close team org-relocation gap

Adds test_route_coverage.py (PR3.M1): parses every @router route literal
from the two management-endpoint source files and asserts each is exercised
by >=1 behavior-suite scenario — a permanent regression guard for future
routes. Closes the last PR1/PR2 deferred gap: the /team/update org-relocation
allowed branch, exercised by a dual-org-admin minted via create_scratch_actor.
test_team_model uses literal route URLs so the coverage parser resolves them.

* test(proxy): bound plain route params to one path segment in coverage gate

Plain path params ({team_id}) now compile to [^/?]+ instead of [^?]+, so a
parameter cannot span '/'. Starlette ':path' params still match across '/'.
Keeps the route-coverage guard from falsely reporting a future multi-segment
route as covered. All 37 routes remain covered.
This commit is contained in:
yuneng-jiang
2026-05-22 11:24:41 -07:00
committed by GitHub
parent 985574b6be
commit f62ae93e13
26 changed files with 2342 additions and 11 deletions
+76 -1
View File
@@ -11,6 +11,7 @@ import pytest_asyncio
import yaml
from prisma import Json
from litellm.proxy.utils import hash_token
MASTER_KEY = "sk-1234"
SCRATCH_PREFIX = "scratch-"
@@ -106,12 +107,19 @@ async def create_scratch_key(
user_id: str,
team_id: Optional[str] = None,
organization_id: Optional[str] = None,
key_alias: 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).
key_alias defaults to scratch_prefix; pass a distinct scratch-prefixed
alias when a single scenario needs more than one key (/key/generate
enforces unique aliases).
"""
body: Dict[str, Any] = {"key_alias": scratch_prefix, "user_id": user_id}
body: Dict[str, Any] = {
"key_alias": key_alias or scratch_prefix,
"user_id": user_id,
}
if team_id is not None:
body["team_id"] = team_id
if organization_id is not None:
@@ -132,6 +140,8 @@ async def create_scratch_team(
organization_id: Optional[str] = None,
admin_user_ids: Optional[list] = None,
member_user_ids: Optional[list] = None,
team_member_permissions: Optional[list] = None,
models: Optional[list] = None,
) -> str:
"""Raw-seed a scratch-tagged team row; returns its team_id.
@@ -142,6 +152,9 @@ async def create_scratch_team(
members_with_roles JSON, so a raw-seeded team exercises them exactly as
a /team/new-created team would. team_id must start with the scratch
prefix so the `scratch` fixture reclaims the row.
team_member_permissions / models seed the matching raw columns — needed
by the team-key-permission and team-model matrices.
"""
admin_user_ids = list(admin_user_ids or [])
member_user_ids = list(member_user_ids or [])
@@ -157,10 +170,72 @@ async def create_scratch_team(
}
if organization_id is not None:
data["organization_id"] = organization_id
if team_member_permissions is not None:
data["team_member_permissions"] = team_member_permissions
if models is not None:
data["models"] = models
await prisma.db.litellm_teamtable.create(data=data)
return team_id
@dataclass(frozen=True)
class SeededActor:
user_id: str
cleartext: str
hashed: str
async def create_scratch_actor(
prisma,
scratch_prefix: str,
*,
user_role: str,
org_admin_of: tuple = (),
organization_id: Optional[str] = None,
suffix: str = "actor",
) -> SeededActor:
"""Mint a scratch-prefixed user + verification token (+ org memberships).
Reclaimed by the existing `scratch` teardown, which sweeps
litellm_usertable, litellm_verificationtoken, and
litellm_organizationmembership by scratch prefix — no bespoke cleanup
needed. Does NOT write litellm_teammembership against world teams: the
teardown reclaims that table only by team_id prefix, so a scratch actor
needing team membership must join a scratch team instead. The cleartext
is hashed with the real hash_token so the key authenticates end-to-end;
models=[] satisfies LiteLLM_VerificationTokenView.
"""
user_id = f"{scratch_prefix}-{suffix}"
cleartext = "sk-" + uuid.uuid4().hex
hashed = hash_token(cleartext)
await prisma.db.litellm_usertable.create(
data={
"user_id": user_id,
"user_role": user_role,
"organization_id": organization_id,
}
)
token_data: Dict[str, Any] = {
"token": hashed,
"key_name": f"{scratch_prefix}-{suffix}-key",
"key_alias": f"{scratch_prefix}-{suffix}-alias",
"user_id": user_id,
"models": [],
}
if organization_id is not None:
token_data["organization_id"] = organization_id
await prisma.db.litellm_verificationtoken.create(data=token_data)
for org_id in org_admin_of:
await prisma.db.litellm_organizationmembership.create(
data={
"user_id": user_id,
"organization_id": org_id,
"user_role": "org_admin",
}
)
return SeededActor(user_id=user_id, cleartext=cleartext, hashed=hashed)
@pytest_asyncio.fixture
async def scratch(prisma):
handle = Scratch(prefix=f"{SCRATCH_PREFIX}{uuid.uuid4().hex[:12]}")
@@ -0,0 +1,119 @@
import uuid
from typing import FrozenSet
import pytest
from litellm.proxy.utils import hash_token
from .actors import TEAM_ALPHA, TEAM_BETA, Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# GET /key/aliases scopes non-admins via _apply_non_admin_alias_scope: a
# non-admin sees an alias only if it owns the key (user_id match) or the key
# belongs to one of its teams. PROXY_ADMIN sees every alias. The seeded keys:
# own — owned by INTERNAL_USER, no team -> user_id scope only
# alpha — owned by OWNER, team TEAM_ALPHA -> team scope for alpha members
# beta — owned by CROSS_ORG_USER, TEAM_BETA
async def _seed_alias_keys(prisma, prefix: str, world) -> dict:
spec = {
"own": (Actor.INTERNAL_USER, None),
"alpha": (Actor.OWNER, TEAM_ALPHA),
"beta": (Actor.CROSS_ORG_USER, TEAM_BETA),
}
out = {}
for tag, (owner, team_id) in spec.items():
alias = f"{prefix}-{tag}"
data = {
"token": hash_token("sk-" + uuid.uuid4().hex),
"key_name": f"{prefix}-{tag}-key",
"key_alias": alias,
"user_id": world.keys[owner].user_id,
"models": [],
}
if team_id is not None:
data["team_id"] = team_id
await prisma.db.litellm_verificationtoken.create(data=data)
out[tag] = alias
return out
async def _fetch_aliases(proxy_client, caller_cleartext: str, query: str) -> set:
resp = await proxy_client.get(
f"/key/aliases?{query}&size=100",
headers={"Authorization": f"Bearer {caller_cleartext}"},
)
assert resp.status_code == 200, resp.text
return set(resp.json()["aliases"])
# ORG_ADMIN-role callers are stopped 401 by the management-route gate before
# the handler runs — /key/aliases carries no org context. Every other actor
# reaches the handler and is scoped by _apply_non_admin_alias_scope.
_VISIBILITY = {
Actor.PROXY_ADMIN: (200, frozenset({"own", "alpha", "beta"})),
Actor.ORG_ADMIN: (401, None),
Actor.TEAM_ADMIN: (200, frozenset({"alpha"})),
Actor.INTERNAL_USER: (200, frozenset({"own", "alpha"})),
Actor.OWNER: (200, frozenset({"alpha"})),
Actor.UNRELATED_SAME_ORG: (200, frozenset({"alpha"})),
Actor.CROSS_ORG_USER: (200, frozenset({"beta"})),
Actor.SERVICE_ACCOUNT: (200, frozenset({"alpha"})),
Actor.ORG_B_ADMIN: (401, None),
}
@pytest.mark.parametrize(
"actor,expected_status,expected_tags",
[(a, s, t) for a, (s, t) in _VISIBILITY.items()],
ids=[a.value for a in _VISIBILITY],
)
async def test_key_aliases_visibility(
actor: Actor,
expected_status: int,
expected_tags: FrozenSet[str],
proxy_client,
prisma,
scratch,
world,
):
aliases = await _seed_alias_keys(prisma, scratch.prefix, world)
known = {v: k for k, v in aliases.items()}
resp = await proxy_client.get(
f"/key/aliases?search={scratch.prefix}&size=100",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
)
assert (
resp.status_code == expected_status
), f"{actor.value}: {resp.status_code} {resp.text}"
if expected_status != 200:
return
visible = {known[a] for a in resp.json()["aliases"] if a in known}
assert visible == set(
expected_tags
), f"{actor.value}: expected {sorted(expected_tags)}, got {sorted(visible)}"
async def test_key_aliases_team_id_filter(proxy_client, prisma, scratch, world):
"""team_id filter narrows the result to keys of that team."""
aliases = await _seed_alias_keys(prisma, scratch.prefix, world)
returned = await _fetch_aliases(
proxy_client,
world.keys[Actor.PROXY_ADMIN].cleartext,
f"search={scratch.prefix}&team_id={TEAM_ALPHA}",
)
assert returned & set(aliases.values()) == {aliases["alpha"]}
async def test_key_aliases_search_filter(proxy_client, prisma, scratch, world):
"""search is a case-insensitive substring match on key_alias."""
aliases = await _seed_alias_keys(prisma, scratch.prefix, world)
returned = await _fetch_aliases(
proxy_client,
world.keys[Actor.PROXY_ADMIN].cleartext,
f"search={aliases['beta']}",
)
assert returned & set(aliases.values()) == {aliases["beta"]}
@@ -0,0 +1,159 @@
import uuid
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")
# POST /key/block + /key/unblock. PROXY_ADMIN bypasses. ORG_ADMIN-role callers
# are stopped 401 by the management-route gate BEFORE the handler runs — the
# body carries no organization_id, so the gate has no org context and falls
# back to proxy-admin-only. The handler's own _check_key_admin_access org-admin
# branch is therefore unreachable via these routes. INTERNAL_USER-role callers
# do reach _check_key_admin_access: a team admin of the key's team passes (200);
# everyone else (incl. a teamless "self" key with no team to admin) is 403.
_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/cross_org_user", Actor.CROSS_ORG_USER, "self", 403),
("owner/proxy_admin", Actor.PROXY_ADMIN, "owner", 200),
("owner/org_admin", Actor.ORG_ADMIN, "owner", 401),
("owner/team_admin", Actor.TEAM_ADMIN, "owner", 200),
("owner/internal_user", Actor.INTERNAL_USER, "owner", 403),
("owner/owner", Actor.OWNER, "owner", 403),
("owner/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "owner", 403),
("owner/cross_org_user", Actor.CROSS_ORG_USER, "owner", 403),
("owner/service_account", Actor.SERVICE_ACCOUNT, "owner", 403),
("owner/org_b_admin", Actor.ORG_B_ADMIN, "owner", 401),
("cross_org/proxy_admin", Actor.PROXY_ADMIN, "cross_org", 200),
("cross_org/org_admin", Actor.ORG_ADMIN, "cross_org", 401),
("cross_org/team_admin", Actor.TEAM_ADMIN, "cross_org", 403),
("cross_org/cross_org_user", Actor.CROSS_ORG_USER, "cross_org", 403),
("cross_org/org_b_admin", Actor.ORG_B_ADMIN, "cross_org", 401),
]
async def _seed_target(proxy_client, seeder, scratch_prefix, world, shape, caller):
if shape == "self":
return await create_scratch_key(
proxy_client, seeder, scratch_prefix, user_id=caller.user_id
)
if shape == "owner":
return await create_scratch_key(
proxy_client,
seeder,
scratch_prefix,
user_id=world.keys[Actor.OWNER].user_id,
team_id=TEAM_ALPHA,
)
if shape == "cross_org":
return await create_scratch_key(
proxy_client,
seeder,
scratch_prefix,
user_id=world.keys[Actor.CROSS_ORG_USER].user_id,
team_id=TEAM_BETA,
)
pytest.fail(f"unknown shape={shape}") # pragma: no cover
@pytest.mark.parametrize("route", ["block", "unblock"])
@pytest.mark.parametrize(
"actor,shape,expected_status",
[(a, sh, s) for (_id, a, sh, s) in _SCENARIOS],
ids=[s[0] for s in _SCENARIOS],
)
async def test_key_block_unblock_authz_matrix(
route: str,
actor: Actor,
shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
caller = world.keys[actor]
seeder = world.keys[Actor.PROXY_ADMIN].cleartext
target_cleartext = await _seed_target(
proxy_client, seeder, scratch.prefix, world, shape, caller
)
target_hashed = hash_token(target_cleartext)
# /unblock starts from a blocked row so a 200 is observable as True->False.
if route == "unblock":
await prisma.db.litellm_verificationtoken.update(
where={"token": target_hashed}, data={"blocked": True}
)
resp = await proxy_client.post(
f"/key/{route}",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"key": target_cleartext},
)
assert (
resp.status_code == expected_status
), f"{route} {actor.value} {shape}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_verificationtoken.find_unique(
where={"token": target_hashed}
)
assert row is not None
# A never-blocked key reads back blocked=None; treat that as not-blocked.
if expected_status == 200:
assert bool(row.blocked) is (route == "block")
else:
# A denial leaves the blocked column at its pre-request value.
assert bool(row.blocked) is (route == "unblock"), "denied but blocked mutated"
async def test_key_block_unblock_round_trip(proxy_client, prisma, scratch, world):
"""PROXY_ADMIN block then unblock flips the blocked column True then False."""
admin = world.keys[Actor.PROXY_ADMIN]
target = await create_scratch_key(
proxy_client, admin.cleartext, scratch.prefix, user_id=admin.user_id
)
hashed = hash_token(target)
headers = {"Authorization": f"Bearer {admin.cleartext}"}
blocked = await proxy_client.post(
"/key/block", headers=headers, json={"key": target}
)
assert blocked.status_code == 200, blocked.text
row = await prisma.db.litellm_verificationtoken.find_unique(where={"token": hashed})
assert row is not None and row.blocked is True
unblocked = await proxy_client.post(
"/key/unblock", headers=headers, json={"key": target}
)
assert unblocked.status_code == 200, unblocked.text
row = await prisma.db.litellm_verificationtoken.find_unique(where={"token": hashed})
assert row is not None and row.blocked is False
@pytest.mark.parametrize("route", ["block", "unblock"])
@pytest.mark.parametrize(
"actor", [Actor.PROXY_ADMIN, Actor.TEAM_ADMIN], ids=["proxy_admin", "team_admin"]
)
async def test_key_block_unblock_missing_key_returns_404(
route: str, actor: Actor, proxy_client, world
):
"""A well-formed but unseeded key is 404 — not 401/403 — for both the
PROXY_ADMIN existence check and the non-admin _check_key_admin_access path."""
caller = world.keys[actor]
missing = "sk-" + uuid.uuid4().hex
resp = await proxy_client.post(
f"/key/{route}",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"key": missing},
)
assert (
resp.status_code == 404
), f"{route} {actor.value}: {resp.status_code} {resp.text}"
@@ -0,0 +1,123 @@
import uuid
import pytest
from litellm.proxy.utils import hash_token
from .actors import Actor
from .conftest import create_scratch_key
pytestmark = pytest.mark.asyncio(loop_scope="session")
_MARKER_BUDGET = 42.0
# POST /key/bulk_update is PROXY_ADMIN-only. The handler's own gate is
# user_role != PROXY_ADMIN -> 403, but ORG_ADMIN-role callers never reach it:
# the management-route gate 401s them first (the body carries no org context,
# and /key/bulk_update is an internal_user route, not an org-admin one).
# INTERNAL_USER-role callers clear the route gate and hit the handler's 403.
_MATRIX = [
("proxy_admin", Actor.PROXY_ADMIN, 200),
("org_admin", Actor.ORG_ADMIN, 401),
("team_admin", Actor.TEAM_ADMIN, 403),
("internal_user", Actor.INTERNAL_USER, 403),
("owner", Actor.OWNER, 403),
("unrelated_same_org", Actor.UNRELATED_SAME_ORG, 403),
("cross_org_user", Actor.CROSS_ORG_USER, 403),
("service_account", Actor.SERVICE_ACCOUNT, 403),
]
@pytest.mark.parametrize(
"actor,expected_status",
[(a, s) for (_id, a, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_key_bulk_update_authz_matrix(
actor: Actor,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
caller = world.keys[actor]
seeder = world.keys[Actor.PROXY_ADMIN].cleartext
target = await create_scratch_key(
proxy_client, seeder, scratch.prefix, user_id=caller.user_id
)
hashed = hash_token(target)
resp = await proxy_client.post(
"/key/bulk_update",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"keys": [{"key": target, "max_budget": _MARKER_BUDGET}]},
)
assert (
resp.status_code == expected_status
), f"{actor.value}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_verificationtoken.find_unique(where={"token": hashed})
assert row is not None
if expected_status == 200:
body = resp.json()
assert len(body["successful_updates"]) == 1
assert body["failed_updates"] == []
assert row.max_budget == _MARKER_BUDGET
else:
assert row.max_budget != _MARKER_BUDGET, "denied but key mutated"
async def test_key_bulk_update_empty_keys_is_400(proxy_client, world):
"""An empty batch is rejected 400 before any per-key processing."""
resp = await proxy_client.post(
"/key/bulk_update",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"keys": []},
)
assert resp.status_code == 400, resp.text
async def test_key_bulk_update_over_max_batch_is_400(proxy_client, world):
"""A batch larger than the 500-key cap is rejected 400."""
items = [{"key": "sk-" + uuid.uuid4().hex} for _ in range(501)]
resp = await proxy_client.post(
"/key/bulk_update",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"keys": items},
)
assert resp.status_code == 400, resp.text
async def test_key_bulk_update_per_key_failure_is_isolated(
proxy_client, prisma, scratch, world
):
"""One bad key in the batch does not abort the others — it lands in
failed_updates while the valid key is still updated."""
admin = world.keys[Actor.PROXY_ADMIN]
valid = await create_scratch_key(
proxy_client, admin.cleartext, scratch.prefix, user_id=admin.user_id
)
missing = "sk-" + uuid.uuid4().hex
resp = await proxy_client.post(
"/key/bulk_update",
headers={"Authorization": f"Bearer {admin.cleartext}"},
json={
"keys": [
{"key": valid, "max_budget": _MARKER_BUDGET},
{"key": missing, "max_budget": _MARKER_BUDGET},
]
},
)
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["total_requested"] == 2
assert len(body["successful_updates"]) == 1
assert len(body["failed_updates"]) == 1
row = await prisma.db.litellm_verificationtoken.find_unique(
where={"token": hash_token(valid)}
)
assert row is not None and row.max_budget == _MARKER_BUDGET
@@ -1,3 +1,5 @@
import uuid
import pytest
from litellm.proxy.utils import hash_token
@@ -99,3 +101,13 @@ async def test_key_delete_authz_matrix(
else:
assert row is not None, f"{actor.value}: denied but row vanished"
assert auth_check.status_code == 200
async def test_key_delete_missing_key_is_404(proxy_client, world):
"""Deleting a key absent from the DB is a 404 — not 401/403."""
resp = await proxy_client.post(
"/key/delete",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"keys": ["sk-" + uuid.uuid4().hex]},
)
assert resp.status_code == 404, resp.text
@@ -0,0 +1,24 @@
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /key/health has no role gate — it reflects the caller's OWN key logging
# metadata. The world keys carry no "logging" metadata, so every authenticated
# actor gets 200 with key="healthy". This pins auth-required + route coverage.
@pytest.mark.parametrize("actor", list(Actor), ids=[a.value for a in Actor])
async def test_key_health_each_actor_is_healthy(actor: Actor, proxy_client, world):
caller = world.keys[actor]
resp = await proxy_client.post(
"/key/health",
headers={"Authorization": f"Bearer {caller.cleartext}"},
)
assert resp.status_code == 200, f"{actor.value}: {resp.status_code} {resp.text}"
assert resp.json()["key"] == "healthy"
async def test_key_health_requires_auth(proxy_client):
resp = await proxy_client.post("/key/health")
assert resp.status_code == 401, resp.text
@@ -0,0 +1,82 @@
import uuid
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /v2/key/info resolves the posted keys, then drops any key the caller
# cannot see via _can_user_query_key_info — silently, no 403. A non-admin sees
# a key it owns (user_id match) or a key whose team it belongs to. The world's
# TEAM_ALPHA members all see each other's keys; CROSS_ORG_USER and the org
# admins see only their own. The request is posted with every world key, and
# the returned info set is asserted to equal the visible subset.
_ALPHA_KEYS = frozenset(
{
Actor.TEAM_ADMIN,
Actor.INTERNAL_USER,
Actor.OWNER,
Actor.UNRELATED_SAME_ORG,
Actor.SERVICE_ACCOUNT,
}
)
_VISIBILITY = {
Actor.PROXY_ADMIN: frozenset(Actor),
Actor.ORG_ADMIN: frozenset({Actor.ORG_ADMIN}),
Actor.TEAM_ADMIN: _ALPHA_KEYS,
Actor.INTERNAL_USER: _ALPHA_KEYS,
Actor.OWNER: _ALPHA_KEYS,
Actor.UNRELATED_SAME_ORG: _ALPHA_KEYS,
Actor.SERVICE_ACCOUNT: _ALPHA_KEYS,
Actor.CROSS_ORG_USER: frozenset({Actor.CROSS_ORG_USER}),
Actor.ORG_B_ADMIN: frozenset({Actor.ORG_B_ADMIN}),
}
@pytest.mark.parametrize(
"actor,expected_visible",
list(_VISIBILITY.items()),
ids=[a.value for a in _VISIBILITY],
)
async def test_key_info_v2_visibility(actor, expected_visible, proxy_client, world):
caller = world.keys[actor]
user_id_to_actor = {world.keys[a].user_id: a for a in Actor}
resp = await proxy_client.post(
"/v2/key/info",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"keys": [world.keys[a].cleartext for a in Actor]},
)
assert resp.status_code == 200, f"{actor.value}: {resp.status_code} {resp.text}"
visible = {
user_id_to_actor[entry["user_id"]]
for entry in resp.json()["info"]
if entry.get("user_id") in user_id_to_actor
}
assert visible == set(expected_visible), (
f"{actor.value}: expected {sorted(a.value for a in expected_visible)}, "
f"got {sorted(a.value for a in visible)}"
)
async def test_key_info_v2_no_body_is_422(proxy_client, world):
"""A request with no body is a 422 — the handler has no keys to resolve."""
resp = await proxy_client.post(
"/v2/key/info",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
)
assert resp.status_code == 422, resp.text
async def test_key_info_v2_unknown_key_returns_empty_info(proxy_client, world):
"""Keys that resolve to no rows yield an empty info list, not an error."""
resp = await proxy_client.post(
"/v2/key/info",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"keys": ["sk-" + uuid.uuid4().hex]},
)
assert resp.status_code == 200, resp.text
assert resp.json()["info"] == []
@@ -2,7 +2,10 @@ from typing import FrozenSet
import pytest
from .actors import Actor
from litellm.proxy.utils import hash_token
from .actors import TEAM_ALPHA, Actor
from .conftest import create_scratch_key
pytestmark = pytest.mark.asyncio(loop_scope="session")
@@ -61,3 +64,108 @@ async def test_key_list_visibility(
f"{actor.value}: expected {sorted(a.value for a in expected_visible)}, "
f"got {sorted(a.value for a in visible_seeded)}"
)
async def _list_hashes(proxy_client, caller_cleartext: str, query: str) -> set:
resp = await proxy_client.get(
f"/key/list?{query}&size=100",
headers={"Authorization": f"Bearer {caller_cleartext}"},
)
assert resp.status_code == 200, resp.text
hashes: set = set()
for entry in resp.json().get("keys", []):
tok = entry.get("token") if isinstance(entry, dict) else entry
if tok:
hashes.add(tok)
return hashes
async def test_key_list_admin_key_alias_substring_match(proxy_client, scratch, world):
"""A PROXY_ADMIN's key_alias filter is a case-insensitive substring match;
a narrower fragment selects the subset whose alias contains it."""
admin = world.keys[Actor.PROXY_ADMIN]
a = await create_scratch_key(
proxy_client,
admin.cleartext,
scratch.prefix,
user_id=admin.user_id,
key_alias=f"{scratch.prefix}-sub-a",
)
b = await create_scratch_key(
proxy_client,
admin.cleartext,
scratch.prefix,
user_id=admin.user_id,
key_alias=f"{scratch.prefix}-sub-b",
)
seeded = {hash_token(a), hash_token(b)}
broad = await _list_hashes(
proxy_client, admin.cleartext, f"key_alias={scratch.prefix}-sub"
)
assert broad & seeded == seeded
narrow = await _list_hashes(
proxy_client, admin.cleartext, f"key_alias={scratch.prefix}-sub-a"
)
assert narrow & seeded == {hash_token(a)}
async def test_key_list_non_admin_key_alias_is_exact_match(
proxy_client, scratch, world
):
"""A non-admin's key_alias filter is exact-match only — substring filtering
is restricted to admins. The full alias matches; a fragment does not."""
caller = world.keys[Actor.INTERNAL_USER]
alias = f"{scratch.prefix}-exact"
key = await create_scratch_key(
proxy_client,
world.keys[Actor.PROXY_ADMIN].cleartext,
scratch.prefix,
user_id=caller.user_id,
key_alias=alias,
)
key_hash = hash_token(key)
exact = await _list_hashes(proxy_client, caller.cleartext, f"key_alias={alias}")
assert key_hash in exact
fragment = await _list_hashes(
proxy_client, caller.cleartext, f"key_alias={scratch.prefix}-exac"
)
assert key_hash not in fragment
async def test_key_list_team_id_filter(proxy_client, scratch, world):
"""A team_id filter narrows the listing to keys of that team."""
admin = world.keys[Actor.PROXY_ADMIN]
team_key = await create_scratch_key(
proxy_client,
admin.cleartext,
scratch.prefix,
user_id=world.keys[Actor.OWNER].user_id,
team_id=TEAM_ALPHA,
key_alias=f"{scratch.prefix}-team",
)
no_team_key = await create_scratch_key(
proxy_client,
admin.cleartext,
scratch.prefix,
user_id=admin.user_id,
key_alias=f"{scratch.prefix}-noteam",
)
hashes = await _list_hashes(proxy_client, admin.cleartext, f"team_id={TEAM_ALPHA}")
assert hash_token(team_key) in hashes
assert hash_token(no_team_key) not in hashes
async def test_key_list_non_admin_cannot_filter_other_team(proxy_client, world):
"""A non-admin filtering by a team it does not belong to is rejected 403."""
resp = await proxy_client.get(
f"/key/list?team_id={world.team_beta_id}",
headers={
"Authorization": f"Bearer {world.keys[Actor.INTERNAL_USER].cleartext}"
},
)
assert resp.status_code == 403, resp.text
@@ -1,5 +1,10 @@
import litellm
import pytest
from litellm.types.proxy.management_endpoints.ui_sso import (
LiteLLM_UpperboundKeyGenerateParams,
)
from .actors import TEAM_ALPHA, TEAM_BETA, Actor
from .conftest import create_scratch_key
@@ -115,3 +120,46 @@ async def test_key_path_regenerate_smoke(proxy_client, scratch, world):
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
async def test_key_regenerate_enforces_upperbound_key_params(
proxy_client, scratch, world, monkeypatch
):
"""Regenerate runs _enforce_upperbound_key_params: a max_budget above
litellm.upperbound_key_generate_params is rejected 400, a value within the
bound is accepted. Pins #26340 (db8ef44323) — regenerate previously
bypassed the upperbound. upperbound_key_generate_params is module-level
litellm.* state, so monkeypatch save/restores it."""
admin = world.keys[Actor.PROXY_ADMIN]
over_key = await create_scratch_key(
proxy_client,
admin.cleartext,
scratch.prefix,
user_id=admin.user_id,
key_alias=f"{scratch.prefix}-over",
)
within_key = await create_scratch_key(
proxy_client,
admin.cleartext,
scratch.prefix,
user_id=admin.user_id,
key_alias=f"{scratch.prefix}-within",
)
monkeypatch.setattr(
litellm,
"upperbound_key_generate_params",
LiteLLM_UpperboundKeyGenerateParams(max_budget=100.0),
)
headers = {"Authorization": f"Bearer {admin.cleartext}"}
over = await proxy_client.post(
"/key/regenerate", headers=headers, json={"key": over_key, "max_budget": 500.0}
)
assert over.status_code == 400, over.text
within = await proxy_client.post(
"/key/regenerate",
headers=headers,
json={"key": within_key, "max_budget": 50.0},
)
assert within.status_code == 200, within.text
@@ -0,0 +1,136 @@
import uuid
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")
_SEED_SPEND = 5.0
_RESET_TO = 2.0
# POST /key/{key}/reset_spend. The target key is pre-seeded with spend=5.0 so
# reset_to=2.0 always clears _validate_reset_spend_value (which runs before
# authz). _check_proxy_or_team_admin_for_key then allows only PROXY_ADMIN or a
# team admin of the key's team — there is no org-admin branch, and a teamless
# "self" key has no team to admin. ORG_ADMIN-role callers are stopped 401 at
# the management-route gate before the handler runs.
_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/cross_org_user", Actor.CROSS_ORG_USER, "self", 403),
("team_alpha/proxy_admin", Actor.PROXY_ADMIN, "team_alpha", 200),
("team_alpha/org_admin", Actor.ORG_ADMIN, "team_alpha", 401),
("team_alpha/team_admin", Actor.TEAM_ADMIN, "team_alpha", 200),
("team_alpha/internal_user", Actor.INTERNAL_USER, "team_alpha", 403),
("team_alpha/owner", Actor.OWNER, "team_alpha", 403),
("team_alpha/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "team_alpha", 403),
("team_alpha/cross_org_user", Actor.CROSS_ORG_USER, "team_alpha", 403),
("team_alpha/service_account", Actor.SERVICE_ACCOUNT, "team_alpha", 403),
("team_alpha/org_b_admin", Actor.ORG_B_ADMIN, "team_alpha", 401),
("team_beta/proxy_admin", Actor.PROXY_ADMIN, "team_beta", 200),
("team_beta/org_admin", Actor.ORG_ADMIN, "team_beta", 401),
("team_beta/team_admin", Actor.TEAM_ADMIN, "team_beta", 403),
("team_beta/cross_org_user", Actor.CROSS_ORG_USER, "team_beta", 403),
("team_beta/org_b_admin", Actor.ORG_B_ADMIN, "team_beta", 401),
]
async def _seed_target(proxy_client, seeder, prefix, world, shape, caller) -> str:
if shape == "self":
return await create_scratch_key(
proxy_client, seeder, prefix, user_id=caller.user_id
)
if shape == "team_alpha":
return await create_scratch_key(
proxy_client,
seeder,
prefix,
user_id=world.keys[Actor.OWNER].user_id,
team_id=TEAM_ALPHA,
)
if shape == "team_beta":
return await create_scratch_key(
proxy_client,
seeder,
prefix,
user_id=world.keys[Actor.CROSS_ORG_USER].user_id,
team_id=TEAM_BETA,
)
pytest.fail(f"unknown shape={shape}") # pragma: no cover
@pytest.mark.parametrize(
"actor,shape,expected_status",
[(a, sh, s) for (_id, a, sh, s) in _SCENARIOS],
ids=[s[0] for s in _SCENARIOS],
)
async def test_key_reset_spend_authz_matrix(
actor: Actor,
shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
caller = world.keys[actor]
seeder = world.keys[Actor.PROXY_ADMIN].cleartext
target = await _seed_target(
proxy_client, seeder, scratch.prefix, world, shape, caller
)
hashed = hash_token(target)
await prisma.db.litellm_verificationtoken.update(
where={"token": hashed}, data={"spend": _SEED_SPEND}
)
resp = await proxy_client.post(
f"/key/{target}/reset_spend",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"reset_to": _RESET_TO},
)
assert (
resp.status_code == expected_status
), f"{actor.value} {shape}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_verificationtoken.find_unique(where={"token": hashed})
assert row is not None
if expected_status == 200:
assert row.spend == _RESET_TO
else:
assert row.spend == _SEED_SPEND, "denied but spend reset"
@pytest.mark.parametrize(
"actor", [Actor.PROXY_ADMIN, Actor.TEAM_ADMIN], ids=["proxy_admin", "team_admin"]
)
async def test_key_reset_spend_missing_key_is_404(actor: Actor, proxy_client, world):
"""A well-formed but unseeded key is 404 before any spend validation."""
resp = await proxy_client.post(
f"/key/sk-{uuid.uuid4().hex}/reset_spend",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
json={"reset_to": 0.0},
)
assert resp.status_code == 404, resp.text
async def test_key_reset_spend_above_current_spend_is_400(
proxy_client, prisma, scratch, world
):
"""reset_to above the key's current spend is rejected 400."""
admin = world.keys[Actor.PROXY_ADMIN]
target = await create_scratch_key(
proxy_client, admin.cleartext, scratch.prefix, user_id=admin.user_id
)
resp = await proxy_client.post(
f"/key/{target}/reset_spend",
headers={"Authorization": f"Bearer {admin.cleartext}"},
json={"reset_to": 1.0},
)
assert resp.status_code == 400, resp.text
@@ -0,0 +1,98 @@
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /key/service-account/generate. PROXY_ADMIN always passes. ORG_ADMIN-role
# callers are stopped 401 by the management-route gate (the body carries a
# team_id but no organization_id, so the org-admin route branch never matches).
# INTERNAL_USER-role callers reach the handler: a team admin of the target team
# passes (200); a "user"-role member is 401 (no service-account-generate
# permission); a non-member is 400 ("not assigned to team"). A request with no
# team_id is 400 ("team_id is required") for every actor that reaches the handler.
_SCENARIOS = [
("own/proxy_admin", Actor.PROXY_ADMIN, "own", 200),
("own/org_admin", Actor.ORG_ADMIN, "own", 401),
("own/team_admin", Actor.TEAM_ADMIN, "own", 200),
("own/internal_user", Actor.INTERNAL_USER, "own", 401),
("own/owner", Actor.OWNER, "own", 401),
("own/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "own", 401),
("own/cross_org_user", Actor.CROSS_ORG_USER, "own", 400),
("own/service_account", Actor.SERVICE_ACCOUNT, "own", 401),
("own/org_b_admin", Actor.ORG_B_ADMIN, "own", 401),
("cross_org/proxy_admin", Actor.PROXY_ADMIN, "cross_org", 200),
("cross_org/org_admin", Actor.ORG_ADMIN, "cross_org", 401),
("cross_org/team_admin", Actor.TEAM_ADMIN, "cross_org", 400),
("cross_org/internal_user", Actor.INTERNAL_USER, "cross_org", 400),
("cross_org/cross_org_user", Actor.CROSS_ORG_USER, "cross_org", 401),
("cross_org/org_b_admin", Actor.ORG_B_ADMIN, "cross_org", 401),
("none/proxy_admin", Actor.PROXY_ADMIN, "none", 400),
("none/org_admin", Actor.ORG_ADMIN, "none", 401),
("none/team_admin", Actor.TEAM_ADMIN, "none", 400),
("none/internal_user", Actor.INTERNAL_USER, "none", 400),
("none/cross_org_user", Actor.CROSS_ORG_USER, "none", 400),
]
@pytest.mark.parametrize(
"actor,team_target,expected_status",
[(a, t, s) for (_id, a, t, s) in _SCENARIOS],
ids=[s[0] for s in _SCENARIOS],
)
async def test_key_service_account_generate_authz_matrix(
actor: Actor,
team_target: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
caller = world.keys[actor]
team_id = {
"own": world.team_alpha_id,
"cross_org": world.team_beta_id,
"none": None,
}[team_target]
body = {"key_alias": scratch.prefix}
if team_id is not None:
body["team_id"] = team_id
resp = await proxy_client.post(
"/key/service-account/generate",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json=body,
)
assert (
resp.status_code == expected_status
), f"{actor.value} {team_target}: {resp.status_code} {resp.text}"
rows = await prisma.db.litellm_verificationtoken.find_many(
where={"key_alias": scratch.prefix}
)
if expected_status == 200:
assert len(rows) == 1
# A service-account key belongs to the team, not a user.
assert rows[0].user_id is None
assert rows[0].team_id == team_id
else:
assert rows == [], f"{actor.value}: denied but key row leaked"
async def test_key_service_account_generate_unknown_team_is_400(
proxy_client, prisma, scratch, world
):
"""A team_id absent from the database is rejected 400."""
resp = await proxy_client.post(
"/key/service-account/generate",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"key_alias": scratch.prefix, "team_id": scratch.tag("no-such-team")},
)
assert resp.status_code == 400, resp.text
rows = await prisma.db.litellm_verificationtoken.find_many(
where={"key_alias": scratch.prefix}
)
assert rows == []
@@ -1,3 +1,5 @@
import uuid
import pytest
from litellm.proxy.utils import hash_token
@@ -98,3 +100,85 @@ async def test_key_update_authz_matrix(
assert row.models == [MARKER_MODEL]
else:
assert row.models != [MARKER_MODEL], "denied but row mutated"
async def _seed_shape(proxy_client, seeder, prefix, world, shape, caller) -> str:
if shape == "self":
return await create_scratch_key(
proxy_client, seeder, prefix, user_id=caller.user_id
)
if shape == "owner":
return await create_scratch_key(
proxy_client,
seeder,
prefix,
user_id=world.keys[Actor.OWNER].user_id,
team_id=TEAM_ALPHA,
)
if shape == "cross_org":
return await create_scratch_key(
proxy_client,
seeder,
prefix,
user_id=world.keys[Actor.CROSS_ORG_USER].user_id,
team_id=TEAM_BETA,
)
pytest.fail(f"unknown shape={shape}") # pragma: no cover
async def test_key_update_missing_key_is_404(proxy_client, world):
"""An update targeting a key absent from the DB is a 404 — not 401/403."""
resp = await proxy_client.post(
"/key/update",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"key": "sk-" + uuid.uuid4().hex, "models": [MARKER_MODEL]},
)
assert resp.status_code == 404, resp.text
# A denied /key/update must not partially apply: the budget/limit columns are
# left untouched. Each scenario is a denial cell from the matrix above.
_DENIED_BUDGET = [
("team_admin/self", Actor.TEAM_ADMIN, "self", 403),
("internal_user/owner", Actor.INTERNAL_USER, "owner", 403),
("cross_org_user/cross_org", Actor.CROSS_ORG_USER, "cross_org", 401),
]
@pytest.mark.parametrize(
"actor,target_shape,expected_status",
[(a, t, s) for (_id, a, t, s) in _DENIED_BUDGET],
ids=[s[0] for s in _DENIED_BUDGET],
)
async def test_key_update_denied_does_not_touch_budget_counters(
actor: Actor,
target_shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
caller = world.keys[actor]
seeder = world.keys[Actor.PROXY_ADMIN].cleartext
target = await _seed_shape(
proxy_client, seeder, scratch.prefix, world, target_shape, caller
)
target_hashed = hash_token(target)
resp = await proxy_client.post(
"/key/update",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"key": target, "max_budget": 999.0, "tpm_limit": 888, "rpm_limit": 777},
)
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
assert row.max_budget is None, "denied but max_budget applied"
assert row.tpm_limit is None, "denied but tpm_limit applied"
assert row.rpm_limit is None, "denied but rpm_limit applied"
@@ -0,0 +1,91 @@
"""PR3.M1 — codified route coverage.
Every route declared in the two management-endpoint source files must be
exercised by at least one behavior-suite scenario. This is a permanent
regression guard: a future route added without a behavior test fails CI here,
the same way test_no_management_imports.py codifies the G3 import grep.
"""
import ast
import pathlib
import re
REPO_ROOT = pathlib.Path(__file__).resolve().parents[3]
SOURCE_FILES = [
REPO_ROOT / "litellm/proxy/management_endpoints/key_management_endpoints.py",
REPO_ROOT / "litellm/proxy/management_endpoints/team_endpoints.py",
]
TEST_DIR = pathlib.Path(__file__).resolve().parent
SELF = pathlib.Path(__file__).resolve()
# Captures the route literal from `@router.<method>("<literal>"` — `\s*` spans
# newlines so multi-line decorators are matched too.
_ROUTE_DECORATOR = re.compile(
r"@router\.(?:get|post|put|delete|patch)\(\s*[\"']([^\"']+)[\"']"
)
def _source_routes() -> set:
routes: set = set()
for path in SOURCE_FILES:
routes.update(_ROUTE_DECORATOR.findall(path.read_text()))
return routes
def _route_to_regex(route: str) -> re.Pattern:
# A plain path param ({team_id}) matches a single path segment; a Starlette
# ':path' param ({key:path}) matches across '/'. Keeping plain params
# slash-bounded stops a loose regex from falsely reporting a future
# multi-segment route as already covered.
pattern = ["^"]
pos = 0
for match in re.finditer(r"\{([^}]+)\}", route):
pattern.append(re.escape(route[pos : match.start()]))
pattern.append("[^?]+" if match.group(1).endswith(":path") else "[^/?]+")
pos = match.end()
pattern.append(re.escape(route[pos:]) + "$")
return re.compile("".join(pattern))
def _test_urls() -> set:
"""Every request-URL string literal across the behavior test suite.
f-strings are reconstructed with each interpolation collapsed to a single
placeholder char, so f"/key/{target}/regenerate" becomes /key/X/regenerate.
Query strings are dropped — coverage is a path-level property.
"""
urls: set = set()
for path in sorted(TEST_DIR.glob("test_*.py")):
if path.resolve() == SELF:
continue
tree = ast.parse(path.read_text())
for node in ast.walk(tree):
literal = None
if isinstance(node, ast.Constant) and isinstance(node.value, str):
literal = node.value
elif isinstance(node, ast.JoinedStr):
chunks = []
for value in node.values:
if isinstance(value, ast.Constant) and isinstance(value.value, str):
chunks.append(value.value)
else:
chunks.append("X") # interpolated path / query segment
literal = "".join(chunks)
if literal and literal.startswith("/"):
urls.add(literal.split("?", 1)[0])
return urls
def test_every_management_route_has_a_behavior_scenario():
routes = _source_routes()
assert routes, "no @router routes parsed — the decorator regex is stale"
urls = _test_urls()
uncovered = sorted(
route
for route in routes
if not any(_route_to_regex(route).match(url) for url in urls)
)
assert (
not uncovered
), "management routes with no behavior-suite scenario:\n " + "\n ".join(uncovered)
@@ -1,13 +1,16 @@
import pytest
from .conftest import MASTER_KEY, SCRATCH_PREFIX
from litellm.proxy._types import LitellmUserRoles
from .actors import ORG_A, ORG_B
from .conftest import MASTER_KEY, SCRATCH_PREFIX, create_scratch_actor
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.
# The minting tests run in file order, then _b runs after their fixture
# teardown and asserts no scratch row survived in any reclaimed table. A leak
# in either direction fails _b on the next collection.
async def test_a_scratch_key_lands_in_db(proxy_client, prisma, scratch):
@@ -24,8 +27,35 @@ async def test_a_scratch_key_lands_in_db(proxy_client, prisma, scratch):
assert len(rows) == 1
async def test_a2_scratch_actor_lands_in_db(proxy_client, prisma, scratch):
actor = await create_scratch_actor(
prisma,
scratch.prefix,
user_role=LitellmUserRoles.ORG_ADMIN.value,
org_admin_of=(ORG_A, ORG_B),
)
user_row = await prisma.db.litellm_usertable.find_unique(
where={"user_id": actor.user_id}
)
assert user_row is not None
info = await proxy_client.get(
"/key/info", headers={"Authorization": f"Bearer {actor.cleartext}"}
)
assert info.status_code == 200, info.text
memberships = await prisma.db.litellm_organizationmembership.find_many(
where={"user_id": actor.user_id}
)
assert {m.organization_id for m in memberships} == {ORG_A, ORG_B}
async def test_b_scratch_namespace_is_clean(prisma):
rows = await prisma.db.litellm_verificationtoken.find_many(
tokens = await prisma.db.litellm_verificationtoken.find_many(
where={"key_alias": {"startswith": SCRATCH_PREFIX}}
)
assert rows == []
users = await prisma.db.litellm_usertable.find_many(
where={"user_id": {"startswith": SCRATCH_PREFIX}}
)
memberships = await prisma.db.litellm_organizationmembership.find_many(
where={"user_id": {"startswith": SCRATCH_PREFIX}}
)
assert tokens == [] and users == [] and memberships == []
@@ -0,0 +1,21 @@
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# GET /team/available lists teams from
# litellm.default_internal_user_params["available_teams"]. The behavior world
# configures no available_teams, so the handler returns [] for every actor
# before it even reads the caller — this is the route-coverage + default-path
# pin. /team/available is an info route, so every authenticated actor reaches
# the handler.
@pytest.mark.parametrize("actor", list(Actor), ids=[a.value for a in Actor])
async def test_team_available_default_is_empty(actor: Actor, proxy_client, world):
resp = await proxy_client.get(
"/team/available",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
)
assert resp.status_code == 200, f"{actor.value}: {resp.status_code} {resp.text}"
assert resp.json() == []
@@ -0,0 +1,114 @@
import pytest
from .actors import Actor
from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /team/block + /team/unblock. The handler gate is _verify_team_access
# (proxy admin / team admin / org admin), but the management-route gate fronts
# it: the request carries the team's organization_id so an org admin of that
# org clears the gate's org-scoped branch. A team admin is an INTERNAL_USER
# and these are not internal_user routes, so a team admin can never reach the
# handler — only PROXY_ADMIN and an org admin of the team's own org pass.
_MATRIX = [
("alpha/proxy_admin", Actor.PROXY_ADMIN, "alpha", 200),
("alpha/org_admin", Actor.ORG_ADMIN, "alpha", 200),
("alpha/team_admin", Actor.TEAM_ADMIN, "alpha", 401),
("alpha/internal_user", Actor.INTERNAL_USER, "alpha", 401),
("alpha/owner", Actor.OWNER, "alpha", 401),
("alpha/unrelated_same_org", Actor.UNRELATED_SAME_ORG, "alpha", 401),
("alpha/cross_org_user", Actor.CROSS_ORG_USER, "alpha", 401),
("alpha/service_account", Actor.SERVICE_ACCOUNT, "alpha", 401),
("alpha/org_b_admin", Actor.ORG_B_ADMIN, "alpha", 401),
("beta/proxy_admin", Actor.PROXY_ADMIN, "beta", 200),
("beta/org_admin", Actor.ORG_ADMIN, "beta", 401),
("beta/org_b_admin", Actor.ORG_B_ADMIN, "beta", 200),
]
async def _seed_target(prisma, world, shape: str, team_id: str) -> str:
"""Raw-seed the scratch target team; returns its organization_id."""
org_id = world.org_a_id if shape == "alpha" else world.org_b_id
await create_scratch_team(prisma, team_id, organization_id=org_id)
return org_id
@pytest.mark.parametrize("route", ["block", "unblock"])
@pytest.mark.parametrize(
"actor,shape,expected_status",
[(a, sh, s) for (_id, a, sh, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_block_unblock_authz_matrix(
route: str,
actor: Actor,
shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
org_id = await _seed_target(prisma, world, shape, scratch.prefix)
caller = world.keys[actor]
# /unblock starts from a blocked row so a 200 is observable as True->False.
if route == "unblock":
await prisma.db.litellm_teamtable.update(
where={"team_id": scratch.prefix}, data={"blocked": True}
)
resp = await proxy_client.post(
f"/team/{route}",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"team_id": scratch.prefix, "organization_id": org_id},
)
assert (
resp.status_code == expected_status
), f"{route} {actor.value} {shape}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert bool(row.blocked) is (route == "block")
else:
assert bool(row.blocked) is (route == "unblock"), "denied but blocked mutated"
async def test_team_block_unblock_round_trip(proxy_client, prisma, scratch, world):
"""PROXY_ADMIN block then unblock flips the blocked column True then False."""
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
headers = {"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"}
blocked = await proxy_client.post(
"/team/block", headers=headers, json={"team_id": scratch.prefix}
)
assert blocked.status_code == 200, blocked.text
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None and row.blocked is True
unblocked = await proxy_client.post(
"/team/unblock", headers=headers, json={"team_id": scratch.prefix}
)
assert unblocked.status_code == 200, unblocked.text
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None and row.blocked is False
@pytest.mark.parametrize("route", ["block", "unblock"])
async def test_team_block_unblock_missing_team_is_404(route: str, proxy_client, world):
"""A team_id absent from the DB is 404 — the existence check precedes authz."""
resp = await proxy_client.post(
f"/team/{route}",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"team_id": "behavior-pin-no-such-team"},
)
assert resp.status_code == 404, resp.text
@@ -0,0 +1,105 @@
import pytest
from .actors import Actor
from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
def _member_ids(row) -> list:
return [m["user_id"] for m in (row.members_with_roles or [])]
async def test_team_bulk_member_add_proxy_admin_adds_explicit_members(
proxy_client, prisma, scratch, world
):
"""PROXY_ADMIN bulk-adds an explicit member list to a scratch team."""
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
new_member = scratch.tag("m1")
resp = await proxy_client.post(
"/team/bulk_member_add",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={
"team_id": scratch.prefix,
"members": [{"user_id": new_member, "role": "user"}],
},
)
assert resp.status_code == 200, resp.text
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None and new_member in _member_ids(row)
async def test_team_bulk_member_add_empty_members_is_400(
proxy_client, prisma, scratch, world
):
"""An empty member list (with all_users unset) is rejected 400."""
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
resp = await proxy_client.post(
"/team/bulk_member_add",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"team_id": scratch.prefix, "members": []},
)
assert resp.status_code == 400, resp.text
async def test_team_bulk_member_add_over_max_batch_is_400(
proxy_client, prisma, scratch, world
):
"""A member list larger than the 500-member cap is rejected 400."""
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
members = [
{"user_id": f"{scratch.prefix}-u{i}", "role": "user"} for i in range(501)
]
resp = await proxy_client.post(
"/team/bulk_member_add",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"team_id": scratch.prefix, "members": members},
)
assert resp.status_code == 400, resp.text
@pytest.mark.parametrize(
"actor",
[Actor.TEAM_ADMIN, Actor.INTERNAL_USER],
ids=["team_admin", "internal_user"],
)
async def test_team_bulk_member_add_non_admin_is_401(
actor: Actor, proxy_client, prisma, scratch, world
):
"""/team/bulk_member_add is neither an internal_user nor a self-managed
route — a non-proxy-admin with no org context is 401 at the route gate."""
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
resp = await proxy_client.post(
"/team/bulk_member_add",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
json={
"team_id": scratch.prefix,
"members": [{"user_id": scratch.tag("m"), "role": "user"}],
},
)
assert resp.status_code == 401, f"{actor.value}: {resp.status_code} {resp.text}"
async def test_team_bulk_member_add_all_users_proxy_admin(
proxy_client, prisma, scratch, world
):
"""all_users=True pulls every user in the DB into the team. The route is
reachable only by PROXY_ADMIN (the route gate 401s every other actor — even
an org admin with organization_id in the body), so the handler's own
all_users PROXY_ADMIN gate is never the deciding check at the boundary."""
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
resp = await proxy_client.post(
"/team/bulk_member_add",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"team_id": scratch.prefix, "all_users": True},
)
assert resp.status_code == 200, resp.text
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
member_ids = _member_ids(row)
# every world actor is a user in the DB, so all are now team members
assert world.keys[Actor.INTERNAL_USER].user_id in member_ids
@@ -0,0 +1,63 @@
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# GET /team/daily/activity. A proxy admin (admin view) sees activity for any
# team. A non-admin is scoped to user_info.teams: a bare query defaults to its
# own teams (200), and an explicit team_ids filter naming a team it does not
# belong to is 404 (the VERIA-43 fix). Org admins have no team memberships, so
# they behave like a non-member for any specific team.
_MEMBERS = {
"alpha": {
Actor.TEAM_ADMIN,
Actor.INTERNAL_USER,
Actor.OWNER,
Actor.UNRELATED_SAME_ORG,
Actor.SERVICE_ACCOUNT,
},
"beta": {Actor.CROSS_ORG_USER},
}
def _expected(actor: Actor, team: str) -> int:
if team == "none" or actor == Actor.PROXY_ADMIN:
return 200
return 200 if actor in _MEMBERS.get(team, set()) else 404
_CASES = [
(f"{team}/{actor.value}", actor, team, _expected(actor, team))
for team in ("none", "alpha", "beta")
for actor in Actor
]
# start_date / end_date are required by the handler — pin only the team-scope
# authz, not the date validation.
_DATES = "start_date=2024-01-01&end_date=2024-12-31"
@pytest.mark.parametrize(
"actor,team,expected_status",
[(a, t, s) for (_id, a, t, s) in _CASES],
ids=[c[0] for c in _CASES],
)
async def test_team_daily_activity_matrix(
actor: Actor, team: str, expected_status: int, proxy_client, world
):
query = _DATES
if team == "alpha":
query += f"&team_ids={world.team_alpha_id}"
elif team == "beta":
query += f"&team_ids={world.team_beta_id}"
resp = await proxy_client.get(
f"/team/daily/activity?{query}",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
)
assert (
resp.status_code == expected_status
), f"{actor.value} -> {team}: {resp.status_code} {resp.text}"
@@ -0,0 +1,78 @@
import pytest
from .actors import Actor
from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
# POST /team/delete runs per-team _verify_team_access. The request carries the
# team's organization_id so an org admin of that org clears the management-
# route gate; a team admin is an INTERNAL_USER on a non-internal_user route,
# so a team admin never reaches the handler. Only PROXY_ADMIN and an org admin
# of the team's own org can delete it.
_MATRIX = [
("alpha/proxy_admin", Actor.PROXY_ADMIN, "alpha", 200),
("alpha/org_admin", Actor.ORG_ADMIN, "alpha", 200),
("alpha/team_admin", Actor.TEAM_ADMIN, "alpha", 401),
("alpha/internal_user", Actor.INTERNAL_USER, "alpha", 401),
("alpha/cross_org_user", Actor.CROSS_ORG_USER, "alpha", 401),
("alpha/org_b_admin", Actor.ORG_B_ADMIN, "alpha", 401),
("beta/proxy_admin", Actor.PROXY_ADMIN, "beta", 200),
("beta/org_admin", Actor.ORG_ADMIN, "beta", 401),
("beta/org_b_admin", Actor.ORG_B_ADMIN, "beta", 200),
]
@pytest.mark.parametrize(
"actor,shape,expected_status",
[(a, sh, s) for (_id, a, sh, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_delete_authz_matrix(
actor: Actor,
shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
org_id = world.org_a_id if shape == "alpha" else world.org_b_id
await create_scratch_team(prisma, scratch.prefix, organization_id=org_id)
caller = world.keys[actor]
resp = await proxy_client.post(
"/team/delete",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"team_ids": [scratch.prefix], "organization_id": org_id},
)
assert (
resp.status_code == expected_status
), f"{actor.value} {shape}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
if expected_status == 200:
assert row is None, "deleted but team row survives"
else:
assert row is not None, "denied but team row vanished"
async def test_team_delete_batch_with_missing_id_deletes_nothing(
proxy_client, prisma, scratch, world
):
"""A batch is validated whole before any deletion: one missing team_id
fails the request 404 and the accessible team in the batch survives."""
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
resp = await proxy_client.post(
"/team/delete",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"team_ids": [scratch.prefix, "behavior-pin-no-such-team"]},
)
assert resp.status_code == 404, resp.text
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None, "batch aborted but the accessible team was deleted"
@@ -0,0 +1,39 @@
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# GET /team/filter/ui (ui_view_teams) — include_in_schema=False. The handler
# body has no role/org check and never reads user_api_key_dict, but the
# endpoint is still effectively PROXY-ADMIN-only as its docstring claims: the
# management-route gate fronts it (not an internal_user / info / org-admin
# route) and 401s every non-proxy-admin before the handler runs. PROXY_ADMIN
# reaches the unscoped find_many and sees teams across every org.
@pytest.mark.parametrize("actor", list(Actor), ids=[a.value for a in Actor])
async def test_team_filter_ui_is_proxy_admin_only(actor: Actor, proxy_client, world):
resp = await proxy_client.get(
"/team/filter/ui",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
)
expected = 200 if actor == Actor.PROXY_ADMIN else 401
assert (
resp.status_code == expected
), f"{actor.value}: {resp.status_code} {resp.text}"
async def test_team_filter_ui_proxy_admin_sees_cross_org_teams(proxy_client, world):
"""The handler runs an unscoped query — PROXY_ADMIN sees teams from every
org, including the three seeded world teams."""
resp = await proxy_client.get(
"/team/filter/ui",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
)
assert resp.status_code == 200, resp.text
team_ids = {t.get("team_id") for t in resp.json() if isinstance(t, dict)}
assert {
world.team_alpha_id,
world.team_beta_id,
world.team_gamma_id,
} <= team_ids
@@ -0,0 +1,217 @@
import uuid
import pytest
from litellm.proxy._types import KeyManagementRoutes
from litellm.proxy.utils import hash_token
from .actors import Actor
from .conftest import create_scratch_key, create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
_MARKER_BUDGET = 42.0
_KEY_UPDATE = KeyManagementRoutes.KEY_UPDATE.value
# POST /team/key/bulk_update — PROXY_ADMIN bypasses; otherwise
# can_team_member_execute_key_management_endpoint runs with route=KEY_UPDATE.
# A team admin always passes; a "user"-role member passes only when the team's
# team_member_permissions grants /key/update; a non-member is 401. ORG_ADMIN is
# stopped 401 at the management-route gate before the handler (the body has a
# team_id but no organization_id, so the org-admin route branch never matches).
_MATRIX = [
("admin/proxy_admin", Actor.PROXY_ADMIN, "admin", 200),
("admin/internal_user", Actor.INTERNAL_USER, "admin", 200),
("member_allowed/internal_user", Actor.INTERNAL_USER, "member_allowed", 200),
("member_denied/internal_user", Actor.INTERNAL_USER, "member_denied", 401),
("nonmember/internal_user", Actor.INTERNAL_USER, "nonmember", 401),
("nonmember/org_admin", Actor.ORG_ADMIN, "nonmember", 401),
("nonmember/proxy_admin", Actor.PROXY_ADMIN, "nonmember", 200),
]
async def _seed_team_key(prisma, proxy_client, prefix: str, world, shape: str) -> str:
"""Raw-seed the scratch team for `shape`, return a team key's cleartext."""
internal = world.keys[Actor.INTERNAL_USER].user_id
owner = world.keys[Actor.OWNER].user_id
if shape == "admin":
await create_scratch_team(
prisma, prefix, organization_id=world.org_a_id, admin_user_ids=[internal]
)
key_owner = internal
elif shape == "member_allowed":
await create_scratch_team(
prisma,
prefix,
organization_id=world.org_a_id,
admin_user_ids=[owner],
member_user_ids=[internal],
team_member_permissions=[_KEY_UPDATE],
)
key_owner = owner
elif shape == "member_denied":
await create_scratch_team(
prisma,
prefix,
organization_id=world.org_a_id,
admin_user_ids=[owner],
member_user_ids=[internal],
team_member_permissions=[],
)
key_owner = owner
elif shape == "nonmember":
await create_scratch_team(
prisma, prefix, organization_id=world.org_a_id, admin_user_ids=[owner]
)
key_owner = owner
else:
pytest.fail(f"unknown shape={shape}") # pragma: no cover
return await create_scratch_key(
proxy_client,
world.keys[Actor.PROXY_ADMIN].cleartext,
prefix,
user_id=key_owner,
team_id=prefix,
)
@pytest.mark.parametrize(
"actor,shape,expected_status",
[(a, sh, s) for (_id, a, sh, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_key_bulk_update_authz_matrix(
actor: Actor,
shape: str,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
key = await _seed_team_key(prisma, proxy_client, scratch.prefix, world, shape)
hashed = hash_token(key)
caller = world.keys[actor]
resp = await proxy_client.post(
"/team/key/bulk_update",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={
"team_id": scratch.prefix,
"key_ids": [key],
"update_fields": {"max_budget": _MARKER_BUDGET},
},
)
assert (
resp.status_code == expected_status
), f"{actor.value} {shape}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_verificationtoken.find_unique(where={"token": hashed})
assert row is not None
if expected_status == 200:
assert len(resp.json()["successful_updates"]) == 1
assert row.max_budget == _MARKER_BUDGET
else:
assert row.max_budget != _MARKER_BUDGET, "denied but key mutated"
async def test_team_key_bulk_update_requires_team_id(
proxy_client, prisma, scratch, world
):
"""An empty team_id is rejected 400."""
resp = await proxy_client.post(
"/team/key/bulk_update",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={
"team_id": "",
"key_ids": ["sk-" + uuid.uuid4().hex],
"update_fields": {"max_budget": _MARKER_BUDGET},
},
)
assert resp.status_code == 400, resp.text
async def test_team_key_bulk_update_all_keys_in_team(
proxy_client, prisma, scratch, world
):
"""all_keys_in_team=True broadcasts the update to every key in the team."""
admin = world.keys[Actor.PROXY_ADMIN].cleartext
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
keys = [
await create_scratch_key(
proxy_client,
admin,
scratch.prefix,
user_id=world.keys[Actor.OWNER].user_id,
team_id=scratch.prefix,
key_alias=f"{scratch.prefix}-k{i}",
)
for i in range(2)
]
resp = await proxy_client.post(
"/team/key/bulk_update",
headers={"Authorization": f"Bearer {admin}"},
json={
"team_id": scratch.prefix,
"all_keys_in_team": True,
"update_fields": {"max_budget": _MARKER_BUDGET},
},
)
assert resp.status_code == 200, resp.text
assert len(resp.json()["successful_updates"]) == 2
for key in keys:
row = await prisma.db.litellm_verificationtoken.find_unique(
where={"token": hash_token(key)}
)
assert row is not None and row.max_budget == _MARKER_BUDGET
async def test_team_key_bulk_update_no_keys_found_is_404(
proxy_client, prisma, scratch, world
):
"""all_keys_in_team=True on a team with no keys is a top-level 404."""
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
resp = await proxy_client.post(
"/team/key/bulk_update",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={
"team_id": scratch.prefix,
"all_keys_in_team": True,
"update_fields": {"max_budget": _MARKER_BUDGET},
},
)
assert resp.status_code == 404, resp.text
async def test_team_key_bulk_update_missing_key_is_isolated(
proxy_client, prisma, scratch, world
):
"""A key_id absent from the team lands in failed_updates; the batch still
returns 200 and the real key is updated."""
admin = world.keys[Actor.PROXY_ADMIN].cleartext
real = await _seed_team_key(
prisma, proxy_client, scratch.prefix, world, "nonmember"
)
missing = "sk-" + uuid.uuid4().hex
resp = await proxy_client.post(
"/team/key/bulk_update",
headers={"Authorization": f"Bearer {admin}"},
json={
"team_id": scratch.prefix,
"key_ids": [real, missing],
"update_fields": {"max_budget": _MARKER_BUDGET},
},
)
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["total_requested"] == 2
assert len(body["successful_updates"]) == 1
assert len(body["failed_updates"]) == 1
row = await prisma.db.litellm_verificationtoken.find_unique(
where={"token": hash_token(real)}
)
assert row is not None and row.max_budget == _MARKER_BUDGET
@@ -0,0 +1,141 @@
from typing import FrozenSet, Optional
import pytest
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
def _seeded(team_ids: set, world) -> set:
known = {
world.team_alpha_id: "alpha",
world.team_beta_id: "beta",
world.team_gamma_id: "gamma",
}
return {known[t] for t in team_ids if t in known}
async def _v2_team_ids(proxy_client, caller_cleartext: str, extra: str = "") -> set:
"""Walk every /v2/team/list page and collect the returned team_ids."""
ids: set = set()
page = 1
while True:
resp = await proxy_client.get(
f"/v2/team/list?page={page}&page_size=100{extra}",
headers={"Authorization": f"Bearer {caller_cleartext}"},
)
assert resp.status_code == 200, resp.text
body = resp.json()
teams = body.get("teams", []) or []
for t in teams:
tid = t.get("team_id") if isinstance(t, dict) else None
if tid:
ids.add(tid)
if page * 100 >= (body.get("total") or 0) or not teams:
return ids
page += 1
# GET /v2/team/list is an info route reachable by every actor, but
# _enforce_list_team_v2_access still gates a BARE query: a proxy admin sees
# all teams, an org admin sees its orgs' teams, and a regular user — who has
# passed no user_id filter — is rejected 401 ("only admins can query all
# teams"). A regular user must scope the query to its own user_id.
_BARE = [
("proxy_admin", Actor.PROXY_ADMIN, 200, frozenset({"alpha", "beta", "gamma"})),
("org_admin", Actor.ORG_ADMIN, 200, frozenset({"alpha", "gamma"})),
("org_b_admin", Actor.ORG_B_ADMIN, 200, frozenset({"beta"})),
("team_admin", Actor.TEAM_ADMIN, 401, None),
("internal_user", Actor.INTERNAL_USER, 401, None),
("owner", Actor.OWNER, 401, None),
("unrelated_same_org", Actor.UNRELATED_SAME_ORG, 401, None),
("cross_org_user", Actor.CROSS_ORG_USER, 401, None),
("service_account", Actor.SERVICE_ACCOUNT, 401, None),
]
@pytest.mark.parametrize(
"actor,expected_status,expected_visible",
[(a, s, v) for (_id, a, s, v) in _BARE],
ids=[s[0] for s in _BARE],
)
async def test_team_list_v2_bare(
actor: Actor,
expected_status: int,
expected_visible: Optional[FrozenSet[str]],
proxy_client,
world,
):
caller = world.keys[actor]
if expected_status != 200:
resp = await proxy_client.get(
"/v2/team/list",
headers={"Authorization": f"Bearer {caller.cleartext}"},
)
assert resp.status_code == expected_status, resp.text
return
visible = _seeded(await _v2_team_ids(proxy_client, caller.cleartext), world)
assert visible == set(
expected_visible
), f"{actor.value}: expected {sorted(expected_visible)}, got {sorted(visible)}"
# A regular user scoping the query to its own user_id is allowed, and sees
# exactly the teams it belongs to.
_OWN = {
Actor.TEAM_ADMIN: frozenset({"alpha"}),
Actor.INTERNAL_USER: frozenset({"alpha"}),
Actor.OWNER: frozenset({"alpha"}),
Actor.UNRELATED_SAME_ORG: frozenset({"alpha"}),
Actor.CROSS_ORG_USER: frozenset({"beta"}),
Actor.SERVICE_ACCOUNT: frozenset({"alpha"}),
}
@pytest.mark.parametrize(
"actor,expected_visible", list(_OWN.items()), ids=[a.value for a in _OWN]
)
async def test_team_list_v2_own_user_id_query(
actor: Actor, expected_visible: FrozenSet[str], proxy_client, world
):
caller = world.keys[actor]
visible = _seeded(
await _v2_team_ids(
proxy_client, caller.cleartext, f"&user_id={caller.user_id}"
),
world,
)
assert visible == set(
expected_visible
), f"{actor.value}: expected {sorted(expected_visible)}, got {sorted(visible)}"
async def test_team_list_v2_user_id_filter_other_user_is_401(proxy_client, world):
"""A regular user filtering by another user's user_id is rejected 401."""
resp = await proxy_client.get(
f"/v2/team/list?user_id={world.keys[Actor.OWNER].user_id}",
headers={
"Authorization": f"Bearer {world.keys[Actor.INTERNAL_USER].cleartext}"
},
)
assert resp.status_code == 401, resp.text
async def test_team_list_v2_org_filter_foreign_org_is_403(proxy_client, world):
"""An org admin filtering by an organization it does not administer is 403."""
resp = await proxy_client.get(
f"/v2/team/list?organization_id={world.org_b_id}",
headers={"Authorization": f"Bearer {world.keys[Actor.ORG_ADMIN].cleartext}"},
)
assert resp.status_code == 403, resp.text
async def test_team_list_v2_invalid_status_is_400(proxy_client, world):
"""status accepts only 'deleted' — any other value is 400."""
resp = await proxy_client.get(
"/v2/team/list?status=bogus",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
)
assert resp.status_code == 400, resp.text
@@ -0,0 +1,83 @@
import uuid
import pytest
from litellm.proxy.utils import hash_token
from .actors import Actor
pytestmark = pytest.mark.asyncio(loop_scope="session")
# GET /team/{team_id}/members/me resolves the CALLER's own membership row.
# A caller that is not a member of the team is 404 — even PROXY_ADMIN, which
# is not in any seeded team. The route is self-managed, so every actor reaches
# the handler. TEAM_GAMMA has no members, so every actor is 404 there.
_MEMBERS = {
"alpha": {
Actor.TEAM_ADMIN,
Actor.INTERNAL_USER,
Actor.OWNER,
Actor.UNRELATED_SAME_ORG,
Actor.SERVICE_ACCOUNT,
},
"beta": {Actor.CROSS_ORG_USER},
"gamma": set(),
}
_CASES = [
(f"{team}/{actor.value}", actor, team, 200 if actor in members else 404)
for team, members in _MEMBERS.items()
for actor in Actor
]
@pytest.mark.parametrize(
"actor,team,expected_status",
[(a, t, s) for (_id, a, t, s) in _CASES],
ids=[c[0] for c in _CASES],
)
async def test_team_member_me_matrix(
actor: Actor, team: str, expected_status: int, proxy_client, world
):
team_id = {
"alpha": world.team_alpha_id,
"beta": world.team_beta_id,
"gamma": world.team_gamma_id,
}[team]
caller = world.keys[actor]
resp = await proxy_client.get(
f"/team/{team_id}/members/me",
headers={"Authorization": f"Bearer {caller.cleartext}"},
)
assert (
resp.status_code == expected_status
), f"{actor.value} -> {team}: {resp.status_code} {resp.text}"
if expected_status == 200:
body = resp.json()
assert body["user_id"] == caller.user_id
assert body["team_id"] == team_id
async def test_team_member_me_team_key_without_user_id_is_400(
proxy_client, prisma, scratch, world
):
"""A key with no associated user_id (a team / service-account key) cannot
resolve 'me' — the caller has no identity to look up — so it is 400."""
cleartext = "sk-" + uuid.uuid4().hex
await prisma.db.litellm_verificationtoken.create(
data={
"token": hash_token(cleartext),
"key_name": f"{scratch.prefix}-teamkey",
"key_alias": f"{scratch.prefix}-teamkey",
"team_id": world.team_alpha_id,
"models": [],
}
)
resp = await proxy_client.get(
f"/team/{world.team_alpha_id}/members/me",
headers={"Authorization": f"Bearer {cleartext}"},
)
assert resp.status_code == 400, resp.text
@@ -0,0 +1,78 @@
import pytest
from .actors import Actor
from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
_MARKER_MODEL = "behavior-pin-team-model-marker"
_ROUTE_URL = {"add": "/team/model/add", "delete": "/team/model/delete"}
# POST /team/model/add + /team/model/delete. The handler gate is PROXY_ADMIN
# or team admin or org admin, but the management-route gate fronts it — these
# are neither internal_user nor org-admin nor info routes, so every
# non-proxy-admin is 401 before the handler runs. Only PROXY_ADMIN reaches the
# handler, making the team-admin / org-admin handler branches unreachable here.
_MATRIX = [
("proxy_admin", Actor.PROXY_ADMIN, 200),
("org_admin", Actor.ORG_ADMIN, 401),
("team_admin", Actor.TEAM_ADMIN, 401),
("internal_user", Actor.INTERNAL_USER, 401),
("owner", Actor.OWNER, 401),
("unrelated_same_org", Actor.UNRELATED_SAME_ORG, 401),
("cross_org_user", Actor.CROSS_ORG_USER, 401),
("service_account", Actor.SERVICE_ACCOUNT, 401),
("org_b_admin", Actor.ORG_B_ADMIN, 401),
]
@pytest.mark.parametrize("route", ["add", "delete"])
@pytest.mark.parametrize(
"actor,expected_status",
[(a, s) for (_id, a, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_model_authz_matrix(
route: str,
actor: Actor,
expected_status: int,
proxy_client,
prisma,
scratch,
world,
):
initial = [] if route == "add" else [_MARKER_MODEL]
await create_scratch_team(
prisma, scratch.prefix, organization_id=world.org_a_id, models=initial
)
caller = world.keys[actor]
resp = await proxy_client.post(
_ROUTE_URL[route],
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"team_id": scratch.prefix, "models": [_MARKER_MODEL]},
)
assert (
resp.status_code == expected_status
), f"{route} {actor.value}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert (_MARKER_MODEL in row.models) is (route == "add")
else:
assert list(row.models) == initial, "denied but models mutated"
@pytest.mark.parametrize("route", ["add", "delete"])
async def test_team_model_missing_team_is_404(route: str, proxy_client, world):
"""A team_id absent from the DB is 404 — the existence check precedes authz."""
resp = await proxy_client.post(
_ROUTE_URL[route],
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"team_id": "behavior-pin-no-such-team", "models": [_MARKER_MODEL]},
)
assert resp.status_code == 404, resp.text
@@ -0,0 +1,170 @@
import litellm
import pytest
from litellm.proxy._types import KeyManagementRoutes
from .actors import Actor
from .conftest import create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
_PERM = KeyManagementRoutes.KEY_INFO.value
# GET /team/permissions_list and POST /team/permissions_update are self-managed
# routes, so every actor reaches the handler. Both grant access to PROXY_ADMIN,
# the team admin, or an org admin of the team's org. The scratch team is in
# ORG_A with TEAM_ADMIN as its team admin.
_MATRIX = [
("proxy_admin", Actor.PROXY_ADMIN, 200),
("org_admin", Actor.ORG_ADMIN, 200),
("team_admin", Actor.TEAM_ADMIN, 200),
("internal_user", Actor.INTERNAL_USER, 403),
("owner", Actor.OWNER, 403),
("unrelated_same_org", Actor.UNRELATED_SAME_ORG, 403),
("cross_org_user", Actor.CROSS_ORG_USER, 403),
("service_account", Actor.SERVICE_ACCOUNT, 403),
("org_b_admin", Actor.ORG_B_ADMIN, 403),
]
async def _seed_team(prisma, scratch_prefix, world) -> None:
await create_scratch_team(
prisma,
scratch_prefix,
organization_id=world.org_a_id,
admin_user_ids=[world.keys[Actor.TEAM_ADMIN].user_id],
member_user_ids=[
world.keys[Actor.INTERNAL_USER].user_id,
world.keys[Actor.OWNER].user_id,
world.keys[Actor.UNRELATED_SAME_ORG].user_id,
world.keys[Actor.SERVICE_ACCOUNT].user_id,
],
)
@pytest.mark.parametrize(
"actor,expected_status",
[(a, s) for (_id, a, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_permissions_list_authz_matrix(
actor: Actor, expected_status: int, proxy_client, prisma, scratch, world
):
await _seed_team(prisma, scratch.prefix, world)
resp = await proxy_client.get(
f"/team/permissions_list?team_id={scratch.prefix}",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
)
assert (
resp.status_code == expected_status
), f"{actor.value}: {resp.status_code} {resp.text}"
if expected_status == 200:
assert resp.json()["team_id"] == scratch.prefix
@pytest.mark.parametrize(
"actor,expected_status",
[(a, s) for (_id, a, s) in _MATRIX],
ids=[s[0] for s in _MATRIX],
)
async def test_team_permissions_update_authz_matrix(
actor: Actor, expected_status: int, proxy_client, prisma, scratch, world
):
await _seed_team(prisma, scratch.prefix, world)
resp = await proxy_client.post(
"/team/permissions_update",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
json={"team_id": scratch.prefix, "team_member_permissions": [_PERM]},
)
assert (
resp.status_code == expected_status
), f"{actor.value}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert _PERM in (row.team_member_permissions or [])
else:
assert _PERM not in (row.team_member_permissions or []), "denied but mutated"
async def test_team_permissions_available_team_self_join_divergence(
proxy_client, prisma, scratch, world, monkeypatch
):
"""permissions_list honours the available-team self-join — a non-admin can
READ an available team's permissions — but permissions_update deliberately
does not: the same caller is 403 on update. default_internal_user_params is
module-level litellm.* state, so monkeypatch save/restores it."""
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
monkeypatch.setattr(
litellm, "default_internal_user_params", {"available_teams": [scratch.prefix]}
)
caller = world.keys[Actor.CROSS_ORG_USER] # non-admin, unrelated to the team
listed = await proxy_client.get(
f"/team/permissions_list?team_id={scratch.prefix}",
headers={"Authorization": f"Bearer {caller.cleartext}"},
)
assert listed.status_code == 200, listed.text
updated = await proxy_client.post(
"/team/permissions_update",
headers={"Authorization": f"Bearer {caller.cleartext}"},
json={"team_id": scratch.prefix, "team_member_permissions": [_PERM]},
)
assert updated.status_code == 403, updated.text
# POST /team/permissions_bulk_update is PROXY_ADMIN-only. ORG_ADMIN-role
# callers are stopped 401 by the management-route gate; INTERNAL_USER-role
# callers, on a route that is neither internal_user nor self-managed, are 401
# there too — only PROXY_ADMIN reaches the handler's own admin gate.
_BULK_MATRIX = [
("proxy_admin", Actor.PROXY_ADMIN, 200),
("org_admin", Actor.ORG_ADMIN, 401),
("team_admin", Actor.TEAM_ADMIN, 401),
("internal_user", Actor.INTERNAL_USER, 401),
("cross_org_user", Actor.CROSS_ORG_USER, 401),
("org_b_admin", Actor.ORG_B_ADMIN, 401),
]
@pytest.mark.parametrize(
"actor,expected_status",
[(a, s) for (_id, a, s) in _BULK_MATRIX],
ids=[s[0] for s in _BULK_MATRIX],
)
async def test_team_permissions_bulk_update_authz_matrix(
actor: Actor, expected_status: int, proxy_client, prisma, scratch, world
):
await create_scratch_team(prisma, scratch.prefix, organization_id=world.org_a_id)
resp = await proxy_client.post(
"/team/permissions_bulk_update",
headers={"Authorization": f"Bearer {world.keys[actor].cleartext}"},
json={"team_ids": [scratch.prefix], "permissions": [_PERM]},
)
assert (
resp.status_code == expected_status
), f"{actor.value}: {resp.status_code} {resp.text}"
row = await prisma.db.litellm_teamtable.find_unique(
where={"team_id": scratch.prefix}
)
assert row is not None
if expected_status == 200:
assert _PERM in (row.team_member_permissions or [])
else:
assert _PERM not in (row.team_member_permissions or []), "denied but mutated"
async def test_team_permissions_bulk_update_no_selector_is_400(proxy_client, world):
"""Neither team_ids nor apply_to_all_teams is a 400."""
resp = await proxy_client.post(
"/team/permissions_bulk_update",
headers={"Authorization": f"Bearer {world.keys[Actor.PROXY_ADMIN].cleartext}"},
json={"permissions": [_PERM]},
)
assert resp.status_code == 400, resp.text
@@ -1,7 +1,9 @@
import pytest
from litellm.proxy._types import LitellmUserRoles
from .actors import Actor
from .conftest import create_scratch_team
from .conftest import create_scratch_actor, create_scratch_team
pytestmark = pytest.mark.asyncio(loop_scope="session")
@@ -130,8 +132,8 @@ async def test_team_update_requires_proxy_admin_without_org_context(
# in ORG_A; each scenario relocates it to ORG_B. PROXY_ADMIN bypasses;
# ORG_B_ADMIN clears the route gate (dest-org admin) but fails
# _verify_team_access on the source team (403); the rest fail the route gate
# (401). The relocation-allowed branch needs a caller who is org admin of both
# orgs — no seeded actor is, so it is left to a later slice.
# (401). The relocation-*allowed* branch (caller is org admin of both orgs) is
# covered by test_team_update_org_relocation_allowed_for_dual_org_admin below.
_RELOCATION = [
("proxy_admin", Actor.PROXY_ADMIN, 200),
("org_b_admin", Actor.ORG_B_ADMIN, 403),
@@ -174,3 +176,34 @@ async def test_team_update_org_relocation_gate(
assert row.organization_id == world.org_b_id
else:
assert row.organization_id == world.org_a_id, "denied but team relocated"
async def test_team_update_org_relocation_allowed_for_dual_org_admin(
proxy_client, prisma, scratch, world
):
"""Relocation-allowed branch: a caller who is org admin of BOTH the source
and destination org may relocate a team between them. Completes the
_RELOCATION matrix, whose allowed branch PR2 left open — no seeded actor is
a dual-org admin, so one is minted with create_scratch_actor."""
actor = await create_scratch_actor(
prisma,
scratch.prefix,
user_role=LitellmUserRoles.ORG_ADMIN.value,
org_admin_of=(world.org_a_id, world.org_b_id),
)
team_id = await create_scratch_team(
prisma, scratch.tag("team"), organization_id=world.org_a_id
)
resp = await proxy_client.post(
"/team/update",
headers={"Authorization": f"Bearer {actor.cleartext}"},
json={"team_id": team_id, "organization_id": world.org_b_id},
)
assert resp.status_code == 200, resp.text
row = await prisma.db.litellm_teamtable.find_unique(where={"team_id": team_id})
assert row is not None
assert (
row.organization_id == world.org_b_id
), "dual-org admin relocation not applied"