diff --git a/litellm/proxy/auth_v2/authorization/scopes.py b/litellm/proxy/auth_v2/authorization/scopes.py deleted file mode 100644 index f761b337d3a..00000000000 --- a/litellm/proxy/auth_v2/authorization/scopes.py +++ /dev/null @@ -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)) diff --git a/litellm/proxy/auth_v2/models.py b/litellm/proxy/auth_v2/models.py index 046442b0ec9..c94417489a5 100644 --- a/litellm/proxy/auth_v2/models.py +++ b/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) diff --git a/litellm/proxy/auth_v2/security.py b/litellm/proxy/auth_v2/security.py index 87f15da4251..f713372eabf 100644 --- a/litellm/proxy/auth_v2/security.py +++ b/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 diff --git a/tests/test_litellm/proxy/auth_v2/test_rbac.py b/tests/test_litellm/proxy/auth_v2/test_rbac.py index 7f6ca77f4b1..f6ddbbb67d7 100644 --- a/tests/test_litellm/proxy/auth_v2/test_rbac.py +++ b/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