Coverage for src/lexigram/auth/di/sub_providers/oauth2_provider.py: 56%

1# lexigram/auth/providers/oauth2_provider.py 

2"""OAuth2 provider - handles OAuth2/OIDC integration only.""" 

3 

4from __future__ import annotations 

5 

6from typing import TYPE_CHECKING, Annotated, Any 

7 

8from lexigram.auth.authn.oauth2 import ( 

9 OAuth2IdentityProvider as OAuth2IdentityProviderConfig, 

10) 

11from lexigram.auth.authn.oauth2 import OAuth2Manager 

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

13from lexigram.di.decorators import inject 

14from lexigram.di.markers import Inject 

15from lexigram.di.provider import Provider 

16from lexigram.logging import get_logger 

17 

18if TYPE_CHECKING: 

19 from lexigram.auth.config import AuthConfig 

20 from lexigram.auth.storage.oauth_identity_store import OAuthIdentityStore 

21 from lexigram.contracts.core.di import ( 

22 ContainerRegistrarProtocol, 

23 ContainerResolverProtocol, 

24 ) 

25 

26logger = get_logger(__name__) 

27 

28 

29@inject 

30class OAuth2Provider(Provider): 

31 """OAuth2/OIDC integration ONLY.""" 

32 

33 def __init__( 

34 self, 

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

36 oauth2_providers: dict[str, dict[str, str]] | None = None, 

37 oauth_identity_store: OAuthIdentityStore | None = None, 

38 http_client: Any | None = None, 

39 **kwargs: Any, 

40 ) -> None: 

41 super().__init__(name="oauth2", priority=ProviderPriority.SECURITY) 

42 self.oauth2_providers = oauth2_providers or ( 

43 config.oauth2_providers if config else {} 

44 ) 

45 self.oauth_identity_store: OAuthIdentityStore | None = oauth_identity_store 

46 self.http_client = http_client 

47 

48 @property 

49 def identity_resolver(self) -> OAuthIdentityStore | None: 

50 """Return the OAuth identity store for resolving external IDs to internal UUIDs. 

51 

52 This property exposes the OAuthIdentityStore which implements 

53 IdentityResolverProtocol for resolving OAuth external IDs. 

54 """ 

55 return self.oauth_identity_store 

56 

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

58 """Register OAuth2 services with the container.""" 

59 # Convert raw config dicts to OAuth2IdentityProviderConfig instances 

60 resolved_providers: dict[str, OAuth2IdentityProviderConfig] = { 

61 provider_name: OAuth2IdentityProviderConfig( 

62 name=provider_name, 

63 client_id=cfg.get("client_id", ""), 

64 client_secret=cfg.get("client_secret", ""), 

65 authorize_url=cfg.get("authorize_url", ""), 

66 access_token_url=cfg.get("access_token_url", ""), 

67 userinfo_url=cfg.get("userinfo_url", ""), 

68 scope=cfg.get("scope", "openid email profile"), 

69 redirect_uri=cfg.get("redirect_uri"), 

70 ) 

71 for provider_name, cfg in self.oauth2_providers.items() 

72 } 

73 

74 # OAuth2 manager 

75 oauth2_manager = OAuth2Manager( 

76 providers=resolved_providers, 

77 http_client=self.http_client, 

78 ) 

79 container.singleton(OAuth2Manager, lambda: oauth2_manager) 

80 

81 # Register OAuthIdentityStore for identity resolution 

82 # This enables resolving user IDs from OAuth external IDs 

83 from lexigram.auth.storage.oauth_identity_store import ( 

84 OAuthIdentityStore, 

85 ) 

86 

87 container.singleton(OAuthIdentityStore, lambda: self.oauth_identity_store) 

88 

89 # Register individual OAuth2 providers 

90 for name, cfg in self.oauth2_providers.items(): 

91 provider = OAuth2IdentityProviderConfig( 

92 name=name, 

93 client_id=cfg.get("client_id", ""), 

94 client_secret=cfg.get("client_secret", ""), 

95 authorize_url=cfg.get("authorize_url", ""), 

96 access_token_url=cfg.get("access_token_url", ""), 

97 userinfo_url=cfg.get("userinfo_url", ""), 

98 scope=cfg.get("scope", "openid email profile"), 

99 redirect_uri=cfg.get("redirect_uri"), 

100 require_pkce=cfg.get("require_pkce", "true").lower() == "true", 

101 ) 

102 container.singleton(f"oauth2.{name}", lambda p=provider: p) 

103 

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

105 """Initialize OAuth2 provider.""" 

106 logger.info("OAuth2Provider started") 

107 

108 async def shutdown(self) -> None: 

109 """Shutdown OAuth2 provider.""" 

110 logger.info("OAuth2Provider shutdown") 

111 

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

113 """Check OAuth2 provider health.""" 

114 return HealthCheckResult( 

115 component=self.name, 

116 status=HealthStatus.HEALTHY, 

117 details={ 

118 "service": "oauth2", 

119 "providers_count": len(self.oauth2_providers), 

120 "identity_store_type": type(self.oauth_identity_store).__name__, 

121 }, 

122 ) 

123 

124 

125__all__ = [ 

126 "OAuth2Provider", 

127 "logger", 

128]