Coverage for src/lexigram/auth/di/provider.py: 0%

1"""DI provider for lexigram-auth. 

2 

3This provider wires the auth subsystem with ClockProtocol, IdGeneratorProtocol, 

4TracerProtocol, and MetricsCollectorProtocol for testability and hardening. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

11from lexigram.auth.authn.password_hasher import Argon2idKeyDerivation 

12from lexigram.auth.config import AuthConfig 

13from lexigram.contracts.core import HealthCheckResult, HealthStatus, ProviderPriority 

14from lexigram.contracts.core.health import HealthCheckCategory 

15from lexigram.contracts.security.protocols import KeyDerivationProtocol 

16from lexigram.di.provider import Provider 

17from lexigram.logging import get_logger 

18 

19if TYPE_CHECKING: 

20 from lexigram.contracts.core.di import ( 

21 ContainerRegistrarProtocol, 

22 ContainerResolverProtocol, 

23 ) 

24 

25logger = get_logger(__name__) 

26 

27__all__ = ["AuthProvider"] 

28 

29 

30class AuthProvider(Provider): 

31 """Authentication provider for Lexigram Framework. 

32 

33 This provider wires: 

34 - ClockProtocol for token/session timestamps 

35 - IdGeneratorProtocol for session/token IDs 

36 - TracerProtocol for observability 

37 - MetricsCollectorProtocol for metrics 

38 - KeyDerivationProtocol (Argon2id) for key derivation 

39 """ 

40 

41 name = "auth" 

42 priority = ProviderPriority.SECURITY 

43 

44 config_key: str | None = "auth" 

45 config_model: type | None = AuthConfig 

46 

47 def __init__(self, config: AuthConfig | None = None) -> None: 

48 super().__init__() 

49 self._config = config or AuthConfig() 

50 

51 @classmethod 

52 def from_config(cls, config: AuthConfig, **context: Any) -> AuthProvider: 

53 return cls(config=config) 

54 

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

56 """Register auth services with the DI container.""" 

57 container.singleton(AuthConfig, instance=self._config) 

58 

59 container.singleton( 

60 KeyDerivationProtocol, 

61 factory=lambda _: Argon2idKeyDerivation(config=self._config.password), 

62 ) 

63 

64 if self._config.relay_verification: 

65 from lexigram.auth.authn.apikeys import APIKeyManager 

66 from lexigram.auth.authn.relay import RelayApiKeyVerifier 

67 from lexigram.contracts.ai.relay import RelayAuthVerifierProtocol 

68 from lexigram.contracts.auth import APIKeyRepositoryProtocol 

69 

70 if not container.has(APIKeyRepositoryProtocol): 

71 logger.warning( 

72 "relay_verifier_skipped", 

73 reason="APIKeyRepositoryProtocol not bound", 

74 ) 

75 return 

76 container.singleton( 

77 RelayAuthVerifierProtocol, 

78 factory=lambda c: RelayApiKeyVerifier( 

79 manager=APIKeyManager(repo=c.resolve(APIKeyRepositoryProtocol)) 

80 ), 

81 ) 

82 

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

84 """Boot the auth provider.""" 

85 

86 async def shutdown(self) -> None: 

87 """Shutdown the auth provider.""" 

88 

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

90 """Return health status of the auth provider.""" 

91 return HealthCheckResult( 

92 component=self.name, 

93 status=HealthStatus.HEALTHY, 

94 category=HealthCheckCategory.READINESS, 

95 )