refactor(auth_v2): move has_required_scopes onto Principal

Scope checking is identity state, so it belongs on the Principal rather
than a standalone authorization/scopes.py helper. Callers now use
principal.has_required_scopes(security_scopes).

litellm/proxy/auth_v2/authorization/scopes.py

@@ -1,12 +0,0 @@
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from fastapi.security import SecurityScopes
-
-if TYPE_CHECKING:
-    from litellm.proxy.auth_v2.models import Principal
-
-
-def has_required_scopes(security_scopes: SecurityScopes, principal: "Principal") -> bool:
-    return set(security_scopes.scopes).issubset(set(principal.scopes))

litellm/proxy/auth_v2/models.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
 from enum import Enum
-from typing import Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 from urllib.parse import urlparse
 
 from pydantic import BaseModel, ConfigDict, Field
 
 from litellm.proxy.auth_v2.authorization import Role
 
+if TYPE_CHECKING:
+    from fastapi.security import SecurityScopes
+
 _LOOPBACK_HOSTS = {"localhost", "127.0.0.1", "::1"}
 
 
@@ -123,3 +126,6 @@ class Principal(BaseModel):
     credential_ref: CredentialRef = Field(default_factory=CredentialRef)
     network: NetworkContext = Field(default_factory=NetworkContext)
     claims: Dict[str, Any] = Field(default_factory=dict)
+
+    def has_required_scopes(self, security_scopes: SecurityScopes) -> bool:
+        return set(security_scopes.scopes).issubset(self.scopes)

litellm/proxy/auth_v2/security.py

@@ -18,7 +18,6 @@ from litellm.proxy.auth_v2.authorization import (
     Authorizer,
     RBACEngine,
     Role,
-    has_required_scopes,
 )
 from litellm.proxy.auth_v2.authenticators.session import SessionAuthenticator
 from litellm.proxy.auth_v2.resolvers import IdentityResolver
@@ -38,7 +37,9 @@ _REDIS_ENV_SIGNALS = (
 )
 
 
-def _open_session_store(namespace: str, *, default_ttl: int) -> SessionStore[SessionValue]:
+def _open_session_store(
+    namespace: str, *, default_ttl: int
+) -> SessionStore[SessionValue]:
     """Build the session/login-state store for ``namespace``.
 
     Uses Redis when configured via the environment (required so state is shared
@@ -102,7 +103,9 @@ class AuthSecurity:
         chain.append(SessionAuthenticator(config.session.cookie, self.session_store))
         self.authenticators = chain
 
-    async def principal(self, security_scopes: SecurityScopes, request: Request) -> Principal:
+    async def principal(
+        self, security_scopes: SecurityScopes, request: Request
+    ) -> Principal:
         """Resolve the caller to a Principal, enforcing scheme OR and required scopes."""
         credential = None
         for authenticator in self.authenticators:
@@ -113,8 +116,10 @@ class AuthSecurity:
             raise errors.unauthenticated(_combined_challenge(self.authenticators))
 
         resolved = await self.resolver.resolve(credential)
-        principal = resolved.model_copy(update={"network": resolve_network_context(request, self.config.network)})
-        if not has_required_scopes(security_scopes, principal):
+        principal = resolved.model_copy(
+            update={"network": resolve_network_context(request, self.config.network)}
+        )
+        if not principal.has_required_scopes(security_scopes):
             raise errors.insufficient_scope()
         return principal
 

tests/test_litellm/proxy/auth_v2/test_rbac.py

@@ -3,8 +3,8 @@ from __future__ import annotations
 import pytest
 from fastapi.security import SecurityScopes
 
+from litellm.proxy.auth_v2.authorization import RBACEngine, Role
 from litellm.proxy.auth_v2.models import AuthMethod, Principal, PrincipalType
-from litellm.proxy.auth_v2.rbac import RBACEngine, Role, has_required_scopes
 
 
 def _principal(*, scopes=None, roles=None) -> Principal:
@@ -24,17 +24,17 @@ def _principal(*, scopes=None, roles=None) -> Principal:
 
 def test_required_scopes_is_subset_check():
     principal = _principal(scopes=["models:read", "chat:write", "scim:write"])
-    assert has_required_scopes(SecurityScopes(["models:read"]), principal)
-    assert has_required_scopes(SecurityScopes(["models:read", "chat:write"]), principal)
+    assert principal.has_required_scopes(SecurityScopes(["models:read"]))
+    assert principal.has_required_scopes(SecurityScopes(["models:read", "chat:write"]))
 
 
 def test_missing_required_scope_fails():
     principal = _principal(scopes=["models:read"])
-    assert not has_required_scopes(SecurityScopes(["chat:write"]), principal)
+    assert not principal.has_required_scopes(SecurityScopes(["chat:write"]))
 
 
 def test_empty_required_scopes_always_passes():
-    assert has_required_scopes(SecurityScopes([]), _principal())
+    assert _principal().has_required_scopes(SecurityScopes([]))
 
 
 # --------------------------------------------------------------------------- #
@@ -159,7 +159,7 @@ def test_act_matcher_is_anchored(tmp_path):
     ],
 )
 def test_filter_claim_roles(roles, allowed, allow_platform, expected):
-    from litellm.proxy.auth_v2.rbac import filter_claim_roles
+    from litellm.proxy.auth_v2.authorization import filter_claim_roles
 
     assert filter_claim_roles(roles, allowed, allow_platform) == expected