Coverage for src/lexigram/auth/di/sub_providers/authorization_provider.py: 94%

1# lexigram/auth/providers/authorization_provider.py 

2"""Authorization provider - handles role-based access control and permissions.""" 

3 

4from __future__ import annotations 

5 

6from typing import TYPE_CHECKING, Annotated, Any, cast 

7 

8from lexigram.auth.authz.service import AuthorizationService 

9from lexigram.contracts import ( 

10 AuthorizerProtocol, 

11 HealthCheckResult, 

12 HealthStatus, 

13 ProviderPriority, 

14) 

15from lexigram.di.decorators import inject 

16from lexigram.di.markers import Inject 

17from lexigram.di.provider import Provider 

18from lexigram.logging import get_logger 

19 

20if TYPE_CHECKING: 

21 from lexigram.auth.config import AuthConfig 

22 from lexigram.contracts.core.di import ( 

23 ContainerRegistrarProtocol, 

24 ContainerResolverProtocol, 

25 ) 

26 

27logger = get_logger(__name__) 

28 

29 

30@inject 

31class AuthorizationProvider(Provider): 

32 """Role-based access control and permission management.""" 

33 

34 def __init__( 

35 self, 

36 config: Annotated[AuthConfig, Inject] | None = None, 

37 initial_roles: dict[str, Any] | None = None, 

38 **kwargs: Any, 

39 ) -> None: 

40 super().__init__(name="authorization", priority=ProviderPriority.SECURITY) 

41 self._config = config 

42 self.initial_roles = initial_roles or {} 

43 

44 @property 

45 def auth_config(self) -> AuthConfig | None: 

46 return cast("AuthConfig | None", self._config) 

47 

48 async def register(self, container: ContainerRegistrarProtocol) -> None: 

49 """Register authorization services with the container.""" 

50 ttl = ( 

51 getattr(self.auth_config.rbac, "permission_cache_ttl", 300.0) 

52 if self.auth_config and self.auth_config.rbac 

53 else 300.0 

54 ) 

55 auth_service = AuthorizationService(permission_cache_ttl=ttl) 

56 

57 if self.initial_roles: 

58 auth_service.set_roles(self.initial_roles) 

59 

60 container.singleton(AuthorizerProtocol, lambda: auth_service) 

61 container.singleton(AuthorizationService, lambda: auth_service) 

62 

63 logger.info( 

64 "AuthorizationProvider registered with %d initial roles", 

65 len(self.initial_roles), 

66 ) 

67 

68 async def boot(self, container: ContainerResolverProtocol) -> None: 

69 """Initialize authorization provider.""" 

70 logger.info("AuthorizationProvider started") 

71 

72 async def shutdown(self) -> None: 

73 """Shutdown authorization provider.""" 

74 logger.info("AuthorizationProvider shutdown") 

75 

76 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

77 """Check authorization provider health.""" 

78 return HealthCheckResult( 

79 component=self.name, 

80 status=HealthStatus.HEALTHY, 

81 details={ 

82 "service": "authorization", 

83 "initial_roles_count": len(self.initial_roles), 

84 }, 

85 ) 

86 

87 

88__all__ = [ 

89 "AuthorizationProvider", 

90 "logger", 

91]