Coverage for src/lexigram/auth/di/sub_providers/google_oauth_provider.py: 83%

1"""Google OAuth provider — first-class Google token verification support.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Annotated, Any 

6 

7from lexigram.auth.authn.google_oauth import ( 

8 GOOGLE_ISSUERS, 

9 GOOGLE_JWKS_URL, 

10 GOOGLE_TOKENINFO_URL, 

11 GOOGLE_USERINFO_URL, 

12 GoogleOAuthService, 

13) 

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

15from lexigram.contracts.exceptions import ConfigurationError 

16from lexigram.di.decorators import inject 

17from lexigram.di.markers import Inject 

18from lexigram.di.provider import Provider 

19from lexigram.logging import get_logger 

20 

21if TYPE_CHECKING: 

22 from lexigram.auth.config import AuthConfig 

23 from lexigram.contracts.core.di import ( 

24 ContainerRegistrarProtocol, 

25 ContainerResolverProtocol, 

26 ) 

27 

28logger = get_logger(__name__) 

29 

30 

31@inject 

32class GoogleOAuthProvider(Provider): 

33 """Registers a first-class Google OAuth verification service.""" 

34 

35 def __init__( 

36 self, 

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

38 google_oauth: dict[str, str] | None = None, 

39 http_client: Any | None = None, 

40 **kwargs: Any, 

41 ) -> None: 

42 super().__init__(name="google_oauth", priority=ProviderPriority.SECURITY) 

43 self._config = google_oauth or ( 

44 config.oauth2_providers.get("google", {}) if config else {} 

45 ) 

46 self._http_client = http_client 

47 self._service: GoogleOAuthService | None = None 

48 

49 @property 

50 def service(self) -> GoogleOAuthService | None: 

51 """Return the registered Google OAuth service, if any.""" 

52 return self._service 

53 

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

55 """Register the Google OAuth verifier service.""" 

56 client_id = self._config.get("client_id") 

57 if not client_id: 

58 raise ConfigurationError( 

59 "Google OAuth configuration is missing client_id; set " 

60 "auth.oauth2_providers.google.client_id or pass google_oauth={...}" 

61 ) 

62 

63 jwks_url = self._config.get("jwks_url") or None 

64 tokeninfo_url = self._config.get("tokeninfo_url") or None 

65 userinfo_url = self._config.get("userinfo_url") or None 

66 issuer_values = self._config.get("issuer") or self._config.get("issuers") 

67 allowed_issuers: tuple[str, ...] 

68 if isinstance(issuer_values, str): 

69 allowed_issuers = (issuer_values,) 

70 elif isinstance(issuer_values, list): 

71 allowed_issuers = tuple(str(value) for value in issuer_values if value) 

72 else: 

73 allowed_issuers = () 

74 

75 jwks_cache_ttl_seconds = int(self._config.get("jwks_cache_ttl_seconds", 300)) 

76 

77 service = GoogleOAuthService( 

78 client_id=client_id, 

79 http_client=self._http_client, 

80 jwks_url=jwks_url or GOOGLE_JWKS_URL, 

81 tokeninfo_url=tokeninfo_url or GOOGLE_TOKENINFO_URL, 

82 userinfo_url=userinfo_url or GOOGLE_USERINFO_URL, 

83 allowed_issuers=allowed_issuers or GOOGLE_ISSUERS, 

84 jwks_cache_ttl_seconds=jwks_cache_ttl_seconds, 

85 ) 

86 self._service = service 

87 container.singleton(GoogleOAuthService, lambda: service) 

88 logger.info( 

89 "google_oauth.registered", 

90 configured=True, 

91 client_id_present=bool(client_id), 

92 ) 

93 

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

95 """Initialize Google OAuth support.""" 

96 logger.info("GoogleOAuthProvider started") 

97 

98 async def shutdown(self) -> None: 

99 """Shutdown Google OAuth support.""" 

100 logger.info("GoogleOAuthProvider shutdown") 

101 

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

103 """Check Google OAuth provider health.""" 

104 configured = self._service is not None 

105 return HealthCheckResult( 

106 component=self.name, 

107 status=HealthStatus.HEALTHY if configured else HealthStatus.DEGRADED, 

108 details={ 

109 "service": "google_oauth", 

110 "configured": configured, 

111 "client_id_present": bool(self._config.get("client_id")), 

112 "jwks_url": self._config.get("jwks_url") 

113 or "https://www.googleapis.com/oauth2/v3/certs", 

114 "userinfo_url": self._config.get("userinfo_url") 

115 or "https://www.googleapis.com/oauth2/v3/userinfo", 

116 }, 

117 ) 

118 

119 

120__all__ = ["GoogleOAuthProvider"]