fix(auth_v2): close token-claim privilege escalation and related hardening

Token-flows and credential-flows security review fixes:

- H1 (privilege escalation): a validly-signed token could self-assert
  platform_admin and arbitrary teams via the roles/groups claims. Roles from a
  token are now filtered to a per-provider allowlist (OIDCProviderConfig.allowed_roles,
  default empty = none) with platform-level roles gated behind an explicit
  allow_platform_roles flag; groups become authoritative TeamIdentity only when they
  resolve to a provisioned SCIM Group in the store. Introspection responses carry no
  roles (no per-provider policy applies).
- M1: enforce iss on introspection (OAuth2IntrospectionConfig.issuer) in addition to aud.
- M2: bound JWKS refetch with cache_jwk_set + a 300s lifespan and add a 10s PyJWKClient
  timeout, so an unknown-kid stream can't amplify into per-request network fetches.
- M3: require https for issuer/jwks_uri/introspection_endpoint (loopback excepted for dev).
- rbac: anchor the Casbin act matcher (^(...)$) so a "GET" policy can't grant "GETX".
- basic auth: switch the reference store to pbkdf2_hmac-sha256 (600k iterations); the
  verifier protocol still lets deployments plug argon2/bcrypt.
- LOW: generic invalid_token description instead of echoing PyJWT internals; guard the
  introspection response.json() and require active to be boolean true.

litellm/proxy/auth_v2/authenticators.py

@@ -23,7 +23,6 @@ from .config import (
     OAuth2IntrospectionConfig,
     TrustedProxyConfig,
 )
-from .oidc.config import OIDCProviderConfig
 from .models import (
     AuthMethod,
     ClientCertificate,
@@ -32,8 +31,23 @@ from .models import (
     SecuritySchemeType,
 )
 from .network import ip_in_trusted_proxies
+from .oidc.config import OIDCProviderConfig
+from .rbac import Role
 
 AT_JWT_TYPES = {"at+jwt", "application/at+jwt"}
+_PLATFORM_ROLE_VALUES = {Role.PLATFORM_ADMIN.value, Role.PLATFORM_VIEWER.value}
+
+
+def _apply_role_policy(claims: Dict[str, Any], provider: OIDCProviderConfig) -> None:
+    raw = claims.get("roles")
+    if not isinstance(raw, list):
+        claims["roles"] = []
+        return
+    allowed = set(provider.allowed_roles)
+    filtered = [role for role in raw if role in allowed]
+    if not provider.allow_platform_roles:
+        filtered = [role for role in filtered if role not in _PLATFORM_ROLE_VALUES]
+    claims["roles"] = filtered
 
 
 @runtime_checkable
@@ -48,10 +62,15 @@ class BasicAuthVerifier(Protocol):
     def verify(self, username: str, password: str) -> bool: ...
 
 
+_PBKDF2_ITERATIONS = 600_000
+
+
 def hash_basic_password(password: str, salt: Optional[str] = None) -> str:
     salt = salt or secrets.token_hex(16)
-    digest = hashlib.sha256(bytes.fromhex(salt) + password.encode()).hexdigest()
-    return f"{salt}${digest}"
+    digest = hashlib.pbkdf2_hmac(
+        "sha256", password.encode(), bytes.fromhex(salt), _PBKDF2_ITERATIONS
+    ).hex()
+    return f"pbkdf2_sha256${_PBKDF2_ITERATIONS}${salt}${digest}"
 
 
 class InMemoryBasicAuthStore:
@@ -62,11 +81,11 @@ class InMemoryBasicAuthStore:
         stored = self._credentials.get(username)
         if stored is None:
             return False
-        salt, _, expected = stored.partition("$")
         try:
-            candidate = hashlib.sha256(
-                bytes.fromhex(salt) + password.encode()
-            ).hexdigest()
+            _algorithm, iterations, salt, expected = stored.split("$")
+            candidate = hashlib.pbkdf2_hmac(
+                "sha256", password.encode(), bytes.fromhex(salt), int(iterations)
+            ).hex()
         except ValueError:
             return False
         return hmac.compare_digest(candidate, expected)
@@ -132,7 +151,13 @@ class JWTVerifier:
         jwks_uri = (
             str(provider.jwks_uri) if provider.jwks_uri else self._discover_jwks()
         )
-        self._jwks_client = PyJWKClient(jwks_uri, cache_keys=True)
+        self._jwks_client = PyJWKClient(
+            jwks_uri,
+            cache_keys=True,
+            cache_jwk_set=True,
+            lifespan=300,
+            timeout=10,
+        )
 
     def _discover_jwks(self) -> str:
         import httpx
@@ -166,7 +191,7 @@ class JWTVerifier:
                 options={"verify_exp": True, "require": ["exp", "iss", "aud"]},
             )
         except jwt.PyJWTError as exc:
-            raise errors.invalid_token(str(exc)) from exc
+            raise errors.invalid_token("token verification failed") from exc
 
 
 async def _verify_jwt_off_loop(
@@ -202,6 +227,7 @@ async def _authenticate_bearer_jwt(
     if verifier is None:
         raise errors.invalid_token("no issuer match")
     claims = await _verify_jwt_off_loop(verifier, token, require_at_jwt=require_at_jwt)
+    _apply_role_policy(claims, verifier.provider)
     return _credential_from_claims(scheme, method, token, claims)
 
 
@@ -326,12 +352,18 @@ class OAuth2Authenticator:
         )
         if response.status_code != 200:
             raise errors.invalid_token("introspection failed")
-        body = response.json()
-        if not body.get("active"):
+        try:
+            body = response.json()
+        except ValueError as exc:
+            raise errors.invalid_token("introspection failed") from exc
+        if not isinstance(body, dict) or body.get("active") is not True:
             raise errors.invalid_token("token inactive")
         token_audience = _normalize_audience(body.get("aud"))
         if config.audience and not set(token_audience) & set(config.audience):
             raise errors.invalid_token("audience mismatch")
+        if config.issuer is not None and body.get("iss") != config.issuer:
+            raise errors.invalid_token("issuer mismatch")
+        claims = {key: value for key, value in body.items() if key != "roles"}
         return Credential(
             scheme=SecuritySchemeType.OAUTH2,
             method=AuthMethod.OAUTH2_INTROSPECTION,
@@ -339,7 +371,7 @@ class OAuth2Authenticator:
             issuer=body.get("iss"),
             audience=token_audience,
             scopes=_split_scope(body.get("scope")),
-            claims=body,
+            claims=claims,
         )
 
     def challenge(self) -> str:

litellm/proxy/auth_v2/config.py

@@ -1,8 +1,8 @@
 from typing import List, Optional
 
-from pydantic import AnyHttpUrl, BaseModel, Field, SecretStr
+from pydantic import AnyHttpUrl, BaseModel, Field, SecretStr, field_validator
 
-from .models import SecuritySchemeType
+from .models import SecuritySchemeType, require_secure_url
 from .oidc.config import OIDCProviderConfig
 from .saml.config import SAMLConfig
 from .session import SessionConfig
@@ -23,6 +23,13 @@ class OAuth2IntrospectionConfig(BaseModel):
     client_secret: SecretStr
     subject_field: str = "sub"
     audience: List[str] = Field(default_factory=list)
+    issuer: Optional[str] = None
+
+    @field_validator("introspection_endpoint")
+    @classmethod
+    def _endpoint_https(cls, value: AnyHttpUrl) -> AnyHttpUrl:
+        require_secure_url(str(value))
+        return value
 
 
 class MutualTLSConfig(BaseModel):

litellm/proxy/auth_v2/models.py

@@ -2,11 +2,21 @@ from __future__ import annotations
 
 from enum import Enum
 from typing import Any, Dict, List, Optional
+from urllib.parse import urlparse
 
 from pydantic import BaseModel, ConfigDict, Field
 
 from .rbac import Role
 
+_LOOPBACK_HOSTS = {"localhost", "127.0.0.1", "::1"}
+
+
+def require_secure_url(value: str) -> str:
+    host = urlparse(value).hostname or ""
+    if value.startswith("https://") or host in _LOOPBACK_HOSTS:
+        return value
+    raise ValueError(f"insecure URL, https is required (loopback excepted): {value}")
+
 
 class SecuritySchemeType(str, Enum):
     API_KEY = "apiKey"

litellm/proxy/auth_v2/oidc/config.py

@@ -1,6 +1,8 @@
 from typing import List, Optional
 
-from pydantic import AnyHttpUrl, BaseModel, Field, SecretStr
+from pydantic import AnyHttpUrl, BaseModel, Field, SecretStr, field_validator
+
+from ..models import require_secure_url
 
 
 class OIDCProviderConfig(BaseModel):
@@ -14,3 +16,17 @@ class OIDCProviderConfig(BaseModel):
     login_scopes: List[str] = Field(
         default_factory=lambda: ["openid", "email", "profile"]
     )
+    allowed_roles: List[str] = Field(default_factory=list)
+    allow_platform_roles: bool = False
+
+    @field_validator("issuer")
+    @classmethod
+    def _issuer_https(cls, value: str) -> str:
+        return require_secure_url(value)
+
+    @field_validator("jwks_uri")
+    @classmethod
+    def _jwks_https(cls, value: Optional[AnyHttpUrl]) -> Optional[AnyHttpUrl]:
+        if value is not None:
+            require_secure_url(str(value))
+        return value

litellm/proxy/auth_v2/rbac.py

@@ -39,7 +39,7 @@ g = _, _
 e = some(where (p.eft == allow))
 
 [matchers]
-m = g(r.sub, p.sub) && keyMatch2(r.obj, p.obj) && regexMatch(r.act, p.act)
+m = g(r.sub, p.sub) && keyMatch2(r.obj, p.obj) && regexMatch(r.act, "^(" + p.act + ")$")
 """
 
 _DEFAULT_GROUPING: List[Tuple[str, str]] = [

litellm/proxy/auth_v2/resolver.py

@@ -52,13 +52,6 @@ def _public_claims(claims: Dict[str, Any]) -> Dict[str, Any]:
     return {key: value for key, value in claims.items() if not key.startswith("_")}
 
 
-def _teams_from_claims(claims: Dict[str, Any]) -> List[TeamIdentity]:
-    groups = claims.get("groups", [])
-    if not isinstance(groups, list):
-        return []
-    return [TeamIdentity(id=str(group), name=str(group)) for group in groups]
-
-
 class InMemoryIdentityStore(IdentityResolver, ProvisioningStore):
     def __init__(
         self,
@@ -85,6 +78,28 @@ class InMemoryIdentityStore(IdentityResolver, ProvisioningStore):
         if user is not None and user.active is False:
             raise errors.account_disabled()
 
+    def _resolve_teams(self, claims: Dict[str, Any]) -> List[TeamIdentity]:
+        groups = claims.get("groups", [])
+        if not isinstance(groups, list):
+            return []
+        teams: List[TeamIdentity] = []
+        for group in groups:
+            scim_group = self._find_group(str(group))
+            if scim_group is not None:
+                teams.append(
+                    TeamIdentity(
+                        id=scim_group.id or str(group),
+                        name=scim_group.display_name or str(group),
+                    )
+                )
+        return teams
+
+    def _find_group(self, value: str) -> Optional[ScimGroup]:
+        for group in self._groups.values():
+            if group.id == value or group.display_name == value:
+                return group
+        return None
+
     def _lookup_scim_user(self, principal: Principal) -> Optional[ScimUser]:
         if principal.user is None:
             return None
@@ -138,7 +153,7 @@ class InMemoryIdentityStore(IdentityResolver, ProvisioningStore):
                 user_name=claims.get("preferred_username"),
                 display_name=claims.get("name"),
             ),
-            teams=_teams_from_claims(claims),
+            teams=self._resolve_teams(claims),
             roles=_roles_from_claims(claims),
             scopes=list(credential.scopes),
             auth_method=credential.method,