Coverage for src/lexigram/auth/authz/scopes.py: 100%

1"""OAuth2 scopes and scope management""" 

2 

3from __future__ import annotations 

4 

5from enum import StrEnum 

6import threading 

7 

8 

9class OAuthScope(StrEnum): 

10 """Standard OAuth2 scopes""" 

11 

12 OPENID = "openid" 

13 EMAIL = "email" 

14 PROFILE = "profile" 

15 ADDRESS = "address" 

16 PHONE = "phone" 

17 

18 # Custom scopes for applications 

19 READ = "read" 

20 WRITE = "write" 

21 DELETE = "delete" 

22 ADMIN = "admin" 

23 

24 

25class ScopeManager: 

26 """Manages OAuth2 scopes and their mappings. 

27 

28 Thread-safe class for managing scope-to-permission mappings. 

29 

30 .. note:: 

31 Register as a singleton through ``AuthorizationProvider``. 

32 """ 

33 

34 def __init__(self) -> None: 

35 self._lock = threading.Lock() 

36 self.scope_permissions: dict[str, set[str]] = { 

37 OAuthScope.READ: {"read"}, 

38 OAuthScope.WRITE: {"read", "write"}, 

39 OAuthScope.DELETE: {"read", "write", "delete"}, 

40 OAuthScope.ADMIN: {"read", "write", "delete", "admin"}, 

41 } 

42 

43 def get_scope_permissions(self, scope: str) -> set[str]: 

44 """Get permissions associated with a scope.""" 

45 with self._lock: 

46 return set(self.scope_permissions.get(scope, set())) 

47 

48 def get_scopes_for_permissions(self, permissions: list[str]) -> set[str]: 

49 """Get minimum scopes required for permissions.""" 

50 with self._lock: 

51 required_scopes = set() 

52 for perm in permissions: 

53 for scope, scope_perms in self.scope_permissions.items(): 

54 if perm in scope_perms: 

55 required_scopes.add(scope) 

56 return required_scopes 

57 

58 def validate_scopes( 

59 self, 

60 requested_scopes: list[str], 

61 allowed_scopes: list[str], 

62 ) -> list[str]: 

63 """Validate requested scopes against allowed scopes.""" 

64 with self._lock: 

65 allowed_set = set(allowed_scopes) 

66 return list( 

67 filter(lambda scope: scope in allowed_set, requested_scopes), 

68 ) 

69 

70 def expand_scope_permissions(self, scopes: list[str]) -> set[str]: 

71 """Expand scopes to their associated permissions.""" 

72 with self._lock: 

73 permissions = set() 

74 for scope in scopes: 

75 permissions.update(self.scope_permissions.get(scope, set())) 

76 return permissions 

77 

78 

79__all__ = [ 

80 "OAuthScope", 

81 "ScopeManager", 

82]