feat(proxy): auth_v2 slice 7 - OAuth2 token introspection authenticator

Completes the authn surface from the plan with an RFC 7662 introspection node
for opaque bearer tokens. Dispatched only for tokens that are neither virtual
keys nor JWTs, and only when an introspection endpoint is configured, so
unconfigured deployments fall through to a clean 401 rather than calling out.

The introspection response parsing is a framework-free core: a token is valid
only when explicitly active=true, the subject claim is required, and OAuth
scopes map to a litellm role via an explicit scope->role map (unmapped scopes
grant no role). The HTTP introspection call is isolated from that core.

Tests cover inactive/missing-active rejection, missing subject, and scope
(string and list) to role mapping.

litellm/proxy/auth/v2/authenticators.py

@@ -147,12 +147,97 @@ class JWTAuthenticator:
         )
 
 
-# Master key is matched first (exact compare), then virtual keys, then JWTs.
-# Dispatch is by credential shape, so the chain is deterministic, not "ask everyone".
+def _load_introspection_settings() -> Any:
+    import os
+
+    from litellm.proxy.proxy_server import general_settings
+
+    from .oauth2_introspection import IntrospectionSettings
+
+    cfg = (general_settings or {}).get("auth_v2_oauth2") or {}
+    endpoint = cfg.get("introspection_endpoint") or os.getenv(
+        "AUTH_V2_OAUTH2_INTROSPECTION_ENDPOINT"
+    )
+    if not endpoint:
+        return None
+    return IntrospectionSettings(
+        endpoint=endpoint,
+        client_id=cfg.get("client_id") or os.getenv("AUTH_V2_OAUTH2_CLIENT_ID"),
+        client_secret=cfg.get("client_secret")
+        or os.getenv("AUTH_V2_OAUTH2_CLIENT_SECRET"),
+        user_id_claim=cfg.get("user_id_claim", "sub"),
+        team_claim=cfg.get("team_claim"),
+        scope_claim=cfg.get("scope_claim", "scope"),
+        role_map=cfg.get("role_map") or {},
+    )
+
+
+class OAuth2IntrospectionAuthenticator:
+    """Validates an opaque bearer token via an RFC 7662 introspection endpoint."""
+
+    def can_handle(self, api_key: Optional[str]) -> bool:
+        if not isinstance(api_key, str) or not api_key:
+            return False
+        # Opaque token: not a virtual key, not a JWT. Only when introspection is
+        # configured, so unconfigured deployments fall through to a clean 401.
+        if api_key.startswith("sk-") or api_key.count(".") == 2:
+            return False
+        return _load_introspection_settings() is not None
+
+    async def authenticate(self, api_key: str, ctx: AuthContext) -> Any:
+        from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+
+        from .oauth2_introspection import (
+            OAuth2IntrospectionError,
+            parse_introspection_response,
+        )
+
+        settings = _load_introspection_settings()
+        data = await self._introspect(api_key, settings)
+        try:
+            identity = parse_introspection_response(data, settings)
+        except OAuth2IntrospectionError as e:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED, detail=f"auth_v2: {e}"
+            )
+
+        user_role = None
+        if identity.role is not None:
+            try:
+                user_role = LitellmUserRoles(identity.role)
+            except ValueError:
+                user_role = None
+
+        return UserAPIKeyAuth(
+            user_id=identity.user_id,
+            team_id=identity.team_id,
+            user_role=user_role,
+        )
+
+    async def _introspect(self, token: str, settings: Any) -> Any:
+        import httpx
+
+        auth = (
+            (settings.client_id, settings.client_secret)
+            if settings.client_id
+            else None
+        )
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                settings.endpoint, data={"token": token}, auth=auth
+            )
+            response.raise_for_status()
+            return response.json()
+
+
+# Master key first (exact compare), then virtual keys, then JWTs, then opaque
+# tokens via introspection. Dispatch is by credential shape, so the chain is
+# deterministic, not "ask everyone".
 AUTHENTICATORS: List[Authenticator] = [
     MasterKeyAuthenticator(),
     VirtualKeyAuthenticator(),
     JWTAuthenticator(),
+    OAuth2IntrospectionAuthenticator(),
 ]
 
 

litellm/proxy/auth/v2/oauth2_introspection.py

@@ -0,0 +1,68 @@
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+class OAuth2IntrospectionError(Exception):
+    """Raised when an introspection response is inactive or unusable."""
+
+
+@dataclass
+class IntrospectionSettings:
+    endpoint: str
+    client_id: Optional[str] = None
+    client_secret: Optional[str] = None
+    user_id_claim: str = "sub"
+    team_claim: Optional[str] = None
+    scope_claim: str = "scope"
+    # Maps an OAuth scope value to a litellm role name.
+    role_map: Dict[str, str] = field(default_factory=dict)
+
+
+@dataclass
+class IntrospectionIdentity:
+    user_id: str
+    team_id: Optional[str] = None
+    role: Optional[str] = None
+
+
+def _scopes(raw: Any) -> List[str]:
+    if raw is None:
+        return []
+    if isinstance(raw, str):
+        return raw.split()
+    if isinstance(raw, (list, tuple)):
+        return [str(s) for s in raw]
+    return []
+
+
+def parse_introspection_response(
+    data: Dict[str, Any], settings: IntrospectionSettings
+) -> IntrospectionIdentity:
+    """Validate an RFC 7662 introspection response and map it to an identity.
+
+    Per the spec, ``active`` is the authoritative liveness flag; a token is only
+    valid when it is explicitly active.
+    """
+    if data.get("active") is not True:
+        raise OAuth2IntrospectionError("token is not active")
+
+    user_id = data.get(settings.user_id_claim) or data.get("sub")
+    if not user_id:
+        raise OAuth2IntrospectionError(
+            f"introspection response missing user id claim '{settings.user_id_claim}'"
+        )
+
+    team_id = data.get(settings.team_claim) if settings.team_claim else None
+
+    role: Optional[str] = None
+    for scope in _scopes(data.get(settings.scope_claim)):
+        mapped = settings.role_map.get(scope)
+        if mapped:
+            role = mapped
+            break
+
+    return IntrospectionIdentity(
+        user_id=str(user_id),
+        team_id=str(team_id) if team_id is not None else None,
+        role=role,
+    )

tests/test_litellm/proxy/auth/v2/test_oauth2_introspection.py

@@ -0,0 +1,64 @@
+import pytest
+
+from litellm.proxy.auth.v2.oauth2_introspection import (
+    IntrospectionSettings,
+    OAuth2IntrospectionError,
+    parse_introspection_response,
+)
+
+SETTINGS = IntrospectionSettings(endpoint="https://idp/introspect")
+
+
+def test_active_token_yields_identity():
+    ident = parse_introspection_response({"active": True, "sub": "u1"}, SETTINGS)
+    assert ident.user_id == "u1"
+
+
+def test_inactive_token_is_rejected():
+    with pytest.raises(OAuth2IntrospectionError):
+        parse_introspection_response({"active": False, "sub": "u1"}, SETTINGS)
+
+
+def test_missing_active_flag_is_rejected():
+    # Absence of active=True must not be treated as valid.
+    with pytest.raises(OAuth2IntrospectionError):
+        parse_introspection_response({"sub": "u1"}, SETTINGS)
+
+
+def test_missing_subject_is_rejected():
+    with pytest.raises(OAuth2IntrospectionError):
+        parse_introspection_response({"active": True}, SETTINGS)
+
+
+def test_scope_string_maps_to_role():
+    settings = IntrospectionSettings(
+        endpoint="x", role_map={"litellm:admin": "proxy_admin"}
+    )
+    ident = parse_introspection_response(
+        {"active": True, "sub": "u1", "scope": "openid litellm:admin"}, settings
+    )
+    assert ident.role == "proxy_admin"
+
+
+def test_scope_list_maps_to_role():
+    settings = IntrospectionSettings(endpoint="x", role_map={"a": "internal_user"})
+    ident = parse_introspection_response(
+        {"active": True, "sub": "u1", "scope": ["x", "a"]}, settings
+    )
+    assert ident.role == "internal_user"
+
+
+def test_unmapped_scopes_yield_no_role():
+    settings = IntrospectionSettings(endpoint="x", role_map={"a": "proxy_admin"})
+    ident = parse_introspection_response(
+        {"active": True, "sub": "u1", "scope": "b c"}, settings
+    )
+    assert ident.role is None
+
+
+def test_team_claim_is_extracted():
+    settings = IntrospectionSettings(endpoint="x", team_claim="team_id")
+    ident = parse_introspection_response(
+        {"active": True, "sub": "u1", "team_id": "eng"}, settings
+    )
+    assert ident.team_id == "eng"