test(auth_v2): pin the token-claim hardening and provisioning security fixes

Regression coverage for the security review fixes (H1/M1/M3/S7):
- resolver: a deactivated SCIM user (active=False) is rejected 403; claims
  whose keys start with "_" never surface on the Principal
- authenticators: H1 privilege escalation - a self-asserted token role grants
  nothing without a per-provider allowlist, the allowlist filters roles, and
  platform-level roles need an explicit allow_platform_roles gate
- rbac: the Casbin act matcher is anchored, so a "GET" policy does not grant
  "GETX"
- scim: PATCH that targets the read-only id (replace, remove, no-path replace)
  is rejected 400 with the record's id unchanged; unauthenticated/under-scoped
  requests render a SCIM Error body (401/403); /Schemas is a ListResponse
  envelope
- saml: a replayed signed assertion is rejected 401 (single-use), and an
  unsolicited IdP-initiated response is rejected 401 when allow_unsolicited is
  off (default secure)

Full auth_v2 suite: 165 passing.
This commit is contained in:
Yassin Kortam 2026-06-10 19:58:44 -07:00
parent 99efd314f7
commit 200f674b94
5 changed files with 196 additions and 1 deletions

View file

@ -483,3 +483,63 @@ def test_build_authenticators_includes_mtls_when_enabled():
config = AuthConfig(mutual_tls=MutualTLSConfig(enabled=True))
types = [type(a) for a in build_authenticators(config)]
assert types[-1] is MutualTLSAuthenticator
# --------------------------------------------------------------------------- #
# H1: token role claims are gated by the provider allowlist (privilege escalation)
# --------------------------------------------------------------------------- #
def _bearer_roles(public_key, token_factory, roles, **provider_kwargs):
provider = OIDCProviderConfig(
issuer=TEST_ISSUER, audience=[TEST_AUDIENCE], **provider_kwargs
)
auth = HttpAuthenticator(
HttpBasicConfig(),
[JWTVerifier(provider, jwks_client=FakeJwksClient(public_key))],
)
token = token_factory.mint(roles=roles)
return auth, make_request(headers={"authorization": f"Bearer {token}"})
async def test_token_roles_are_dropped_without_allowlist(rsa_keypair, token_factory):
_, public_key = rsa_keypair
auth, request = _bearer_roles(
public_key, token_factory, ["platform_admin", "org_admin"]
)
credential = await auth.authenticate(request)
# default allowed_roles=[] -> a self-asserted role grants nothing
assert credential.claims["roles"] == []
async def test_token_roles_filtered_to_allowlist(rsa_keypair, token_factory):
_, public_key = rsa_keypair
auth, request = _bearer_roles(
public_key,
token_factory,
["platform_admin", "org_admin"],
allowed_roles=["org_admin"],
)
credential = await auth.authenticate(request)
# org_admin is allowed; platform_admin is dropped (and not in the allowlist anyway)
assert credential.claims["roles"] == ["org_admin"]
async def test_platform_role_requires_explicit_gate(rsa_keypair, token_factory):
_, public_key = rsa_keypair
gated_auth, gated_req = _bearer_roles(
public_key, token_factory, ["platform_admin"], allowed_roles=["platform_admin"]
)
# allowed but platform gate off -> still dropped
assert (await gated_auth.authenticate(gated_req)).claims["roles"] == []
open_auth, open_req = _bearer_roles(
public_key,
token_factory,
["platform_admin"],
allowed_roles=["platform_admin"],
allow_platform_roles=True,
)
assert (await open_auth.authenticate(open_req)).claims["roles"] == [
"platform_admin"
]

View file

@ -126,3 +126,13 @@ def test_csv_policy_overrides_defaults(tmp_path):
assert not engine.enforce(
_principal(roles=[Role.PLATFORM_ADMIN]), "/anything", "GET"
)
def test_act_matcher_is_anchored(tmp_path):
# a "GET" policy must not grant a superstring act like "GETX" (regexMatch ^(...)$)
policy = tmp_path / "policy.csv"
policy.write_text("p, platform_viewer, /x, GET\n")
engine = RBACEngine(policy_path=str(policy))
viewer = _principal(roles=[Role.PLATFORM_VIEWER])
assert engine.enforce(viewer, "/x", "GET")
assert not engine.enforce(viewer, "/x", "GETX")

View file

@ -2,6 +2,7 @@ from __future__ import annotations
import pytest
from scim2_models import Group as ScimGroup
from scim2_models import User as ScimUser
from litellm.proxy.auth_v2.errors import AuthError
from litellm.proxy.auth_v2.models import (
@ -11,6 +12,7 @@ from litellm.proxy.auth_v2.models import (
Principal,
PrincipalType,
SecuritySchemeType,
UserIdentity,
)
from litellm.proxy.auth_v2.rbac import Role
from litellm.proxy.auth_v2.resolver import InMemoryIdentityStore, _hash_api_key
@ -137,3 +139,56 @@ async def test_mtls_credential_resolves_to_service_account():
assert principal.principal_type == PrincipalType.SERVICE_ACCOUNT
assert principal.user is None
assert principal.subject == "CN=svc-a,O=Co"
# --------------------------------------------------------------------------- #
# Deactivated users (M1) and claims scrubbing
# --------------------------------------------------------------------------- #
async def test_deactivated_user_is_rejected():
principal = Principal(
principal_type=PrincipalType.HUMAN,
subject="u-1",
auth_method=AuthMethod.API_KEY,
user=UserIdentity(id="u-1", email="u@example.com"),
)
store = InMemoryIdentityStore(
api_keys={_hash_api_key("sk-deact"): principal},
users={"u-1": ScimUser(id="u-1", user_name="u@example.com", active=False)},
)
with pytest.raises(AuthError) as exc:
await store.resolve(_api_key_credential("sk-deact"))
assert exc.value.status_code == 403
async def test_active_user_is_allowed():
principal = Principal(
principal_type=PrincipalType.HUMAN,
subject="u-2",
auth_method=AuthMethod.API_KEY,
user=UserIdentity(id="u-2", email="ok@example.com"),
)
store = InMemoryIdentityStore(
api_keys={_hash_api_key("sk-ok"): principal},
users={"u-2": ScimUser(id="u-2", user_name="ok@example.com", active=True)},
)
resolved = await store.resolve(_api_key_credential("sk-ok"))
assert resolved.subject == "u-2"
async def test_principal_claims_scrub_underscore_keys():
# internal underscore-prefixed claims (e.g. _raw_api_key) must never surface
# on the Principal built from a self-describing credential
store = InMemoryIdentityStore()
credential = Credential(
scheme=SecuritySchemeType.OPENID_CONNECT,
method=AuthMethod.OIDC,
subject="sub-x",
issuer="https://idp",
claims={"_raw_api_key": "leak", "_basic_password": "leak", "email": "e@x.com"},
)
principal = await store.resolve(credential)
assert "_raw_api_key" not in principal.claims
assert "_basic_password" not in principal.claims
assert principal.claims.get("email") == "e@x.com"

View file

@ -278,6 +278,36 @@ def test_acs_rejects_unsigned_assertion(saml_env):
assert store._users == {}
def test_acs_rejects_replayed_assertion(saml_env):
# a signed assertion is single-use; replaying it is rejected
app, _ = _build_app(saml_env)
client = TestClient(app)
response = saml_env.mint_response()
first = client.post(
"/auth/saml/acs", data={"SAMLResponse": response}, follow_redirects=False
)
assert first.status_code == 303
second = client.post(
"/auth/saml/acs", data={"SAMLResponse": response}, follow_redirects=False
)
assert second.status_code == 401
def test_acs_rejects_unsolicited_when_disabled(saml_env):
# default-secure: an IdP-initiated (no InResponseTo) response is rejected
disabled = saml_env.config.model_copy(update={"allow_unsolicited": False})
env = SamlEnv(config=disabled, idp=saml_env.idp)
app, store = _build_app(env)
client = TestClient(app)
response = client.post(
"/auth/saml/acs",
data={"SAMLResponse": env.mint_response()},
follow_redirects=False,
)
assert response.status_code == 401
assert store._users == {}
def test_acs_missing_response_is_rejected(saml_env):
app, _ = _build_app(saml_env)
client = TestClient(app)

View file

@ -162,7 +162,11 @@ def test_resource_types_lists_user_and_group(client):
def test_schemas_endpoint_returns_user_and_group(client):
response = client.get("/scim/v2/Schemas")
assert response.status_code == 200
assert response.json()["totalResults"] == 2
body = response.json()
assert body["totalResults"] == 2
# a ListResponse envelope, not a bare dict (regression for the envelope fix)
assert body["schemas"][0].endswith(":ListResponse")
assert len(body["Resources"]) == 2
# --------------------------------------------------------------------------- #
@ -178,6 +182,10 @@ def test_scim_requires_authentication():
)
assert response.status_code == 401
assert "WWW-Authenticate" in response.headers
# S7: auth failures are rendered as a SCIM Error, not the generic body
body = response.json()
assert body["schemas"] == [ERROR_SCHEMA]
assert body["status"] == "401"
def test_scim_requires_scim_write_scope():
@ -185,6 +193,38 @@ def test_scim_requires_scim_write_scope():
response = underscoped.get("/scim/v2/Users")
assert response.status_code == 403
assert "insufficient_scope" in response.headers.get("WWW-Authenticate", "")
body = response.json()
assert body["schemas"] == [ERROR_SCHEMA]
assert body["status"] == "403"
# --------------------------------------------------------------------------- #
# id is read-only: PATCH attempting to mutate it is rejected (RFC 7643)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"operation",
[
{"op": "replace", "path": "id", "value": "evil"},
{"op": "remove", "path": "id"},
{"op": "replace", "value": {"id": "evil", "displayName": "X"}},
],
)
def test_patch_id_mutation_is_rejected(client, operation):
user_id = _create_user(client).json()["id"]
response = client.patch(
f"/scim/v2/Users/{user_id}",
json={
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [operation],
},
)
assert response.status_code == 400
assert response.json()["schemas"] == [ERROR_SCHEMA]
# the record keeps its id; the attacker id never materializes
assert client.get(f"/scim/v2/Users/{user_id}").status_code == 200
assert client.get("/scim/v2/Users/evil").status_code == 404
# --------------------------------------------------------------------------- #