Coverage for src/lexigram/auth/authn/google_oauth.py: 65%

1"""Google OAuth verification and claim normalization helpers.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6import inspect 

7from typing import TYPE_CHECKING, Any, cast 

8 

9import jwt 

10 

11from lexigram.auth.exceptions import OAuth2Error 

12from lexigram.contracts.auth import VerifiedIdentityClaims 

13from lexigram.contracts.web import HTTPClientProtocol 

14from lexigram.logging import get_logger 

15from lexigram.serialization import dumps 

16 

17if TYPE_CHECKING: 

18 from lexigram.contracts.web import HttpResponse 

19 

20logger = get_logger(__name__) 

21 

22GOOGLE_ISSUERS: tuple[str, str] = ( 

23 "https://accounts.google.com", 

24 "accounts.google.com", 

25) 

26GOOGLE_JWKS_URL = "https://www.googleapis.com/oauth2/v3/certs" 

27GOOGLE_TOKENINFO_URL = "https://oauth2.googleapis.com/tokeninfo" 

28GOOGLE_USERINFO_URL = "https://www.googleapis.com/oauth2/v3/userinfo" 

29 

30 

31class GoogleOAuthService: 

32 """Verify Google OAuth tokens and normalize the verified claims.""" 

33 

34 def __init__( 

35 self, 

36 *, 

37 client_id: str, 

38 http_client: HTTPClientProtocol | None = None, 

39 jwks_url: str = GOOGLE_JWKS_URL, 

40 tokeninfo_url: str = GOOGLE_TOKENINFO_URL, 

41 userinfo_url: str = GOOGLE_USERINFO_URL, 

42 allowed_issuers: tuple[str, ...] = GOOGLE_ISSUERS, 

43 jwks_cache_ttl_seconds: int = 300, 

44 ) -> None: 

45 if not client_id: 

46 raise ValueError("Google OAuth client_id is required") 

47 self.client_id = client_id 

48 self.http_client = http_client 

49 self.jwks_url = jwks_url 

50 self.tokeninfo_url = tokeninfo_url 

51 self.userinfo_url = userinfo_url 

52 self.allowed_issuers = allowed_issuers 

53 self.jwks_cache_ttl_seconds = max(0, jwks_cache_ttl_seconds) 

54 self._jwks_cache: dict[str, Any] | None = None 

55 self._jwks_cached_at: datetime | None = None 

56 

57 async def verify_token(self, token: str) -> VerifiedIdentityClaims: 

58 """Verify a Google token, preferring ID-token JWKS validation. 

59 

60 Args: 

61 token: Google-issued ID token or access token. 

62 

63 Returns: 

64 Normalized, verified Google identity claims. 

65 

66 Raises: 

67 OAuth2Error: If the token cannot be verified or normalized. 

68 """ 

69 if self._looks_like_jwt(token): 

70 try: 

71 return await self.verify_id_token(token) 

72 except OAuth2Error: 

73 raise 

74 except (jwt.PyJWTError, ValueError, KeyError) as exc: 

75 logger.debug("google_id_token_verification_failed", error=str(exc)) 

76 raise OAuth2Error("Invalid Google ID token") from exc 

77 

78 return await self.verify_userinfo_token(token) 

79 

80 async def verify_id_token(self, token: str) -> VerifiedIdentityClaims: 

81 """Verify a Google ID token against Google's JWKS.""" 

82 header = jwt.get_unverified_header(token) 

83 kid = header.get("kid") 

84 alg = header.get("alg") or "RS256" 

85 

86 jwks = await self._get_jwks() 

87 key = self._select_jwk(jwks, kid) 

88 public_key: Any = jwt.algorithms.RSAAlgorithm.from_jwk(dumps(key).decode()) 

89 

90 try: 

91 payload = jwt.decode( 

92 token, 

93 key=public_key, 

94 algorithms=[cast("str", alg)], 

95 audience=self.client_id, 

96 options={ 

97 "require": ["exp", "iss", "sub", "aud"], 

98 }, 

99 ) 

100 except jwt.PyJWTError as exc: 

101 logger.warning( 

102 "google_id_token_decode_failed", 

103 error=str(exc), 

104 kid=kid, 

105 ) 

106 raise OAuth2Error("Invalid Google ID token") from exc 

107 

108 issuer = str(payload.get("iss") or "") 

109 if issuer not in self.allowed_issuers: 

110 raise OAuth2Error(f"Invalid Google token issuer: {issuer!r}") 

111 

112 email_verified = bool(payload.get("email_verified", False)) 

113 if not email_verified: 

114 raise OAuth2Error("Google email is not verified") 

115 

116 return self._claims_from_payload( 

117 payload, 

118 issuer=issuer, 

119 audience=str(payload.get("aud") or self.client_id), 

120 ) 

121 

122 async def verify_userinfo_token(self, token: str) -> VerifiedIdentityClaims: 

123 """Verify a Google access token via the userinfo endpoint.""" 

124 payload = await self._request_json( 

125 "GET", 

126 self.userinfo_url, 

127 headers={"Authorization": f"Bearer {token}"}, 

128 ) 

129 

130 issuer = str(payload.get("iss") or "accounts.google.com") 

131 email_verified = bool(payload.get("email_verified", False)) 

132 if not email_verified: 

133 raise OAuth2Error("Google email is not verified") 

134 

135 return self._claims_from_payload( 

136 payload, 

137 issuer=issuer, 

138 audience=self.client_id, 

139 ) 

140 

141 async def verify_tokeninfo(self, token: str) -> VerifiedIdentityClaims: 

142 """Verify a Google token using the tokeninfo endpoint fallback.""" 

143 payload = await self._request_json( 

144 "GET", 

145 self.tokeninfo_url, 

146 params={"id_token": token}, 

147 ) 

148 

149 issuer = str(payload.get("iss") or "") 

150 if issuer and issuer not in self.allowed_issuers: 

151 raise OAuth2Error(f"Invalid Google token issuer: {issuer!r}") 

152 

153 audience = str(payload.get("aud") or self.client_id) 

154 if audience != self.client_id: 

155 raise OAuth2Error("Google token audience mismatch") 

156 

157 email_verified = payload.get("email_verified") 

158 if email_verified is not None and not bool(email_verified): 

159 raise OAuth2Error("Google email is not verified") 

160 

161 return self._claims_from_payload( 

162 payload, 

163 issuer=issuer or None, 

164 audience=audience, 

165 ) 

166 

167 async def _get_jwks(self) -> dict[str, Any]: 

168 """Fetch Google's JWKS, caching it for a short TTL.""" 

169 now = datetime.now(UTC) 

170 if ( 

171 self._jwks_cache is not None 

172 and self._jwks_cached_at is not None 

173 and (now - self._jwks_cached_at).total_seconds() 

174 < self.jwks_cache_ttl_seconds 

175 ): 

176 return self._jwks_cache 

177 

178 jwks = await self._request_json("GET", self.jwks_url) 

179 self._jwks_cache = jwks 

180 self._jwks_cached_at = now 

181 return jwks 

182 

183 def _select_jwk(self, jwks: dict[str, Any], kid: str | None) -> dict[str, Any]: 

184 """Select the matching JWK for a token header.""" 

185 keys = jwks.get("keys") 

186 if not isinstance(keys, list) or not keys: 

187 raise OAuth2Error("Google JWKS payload is empty") 

188 

189 if kid: 

190 for key in keys: 

191 if isinstance(key, dict) and key.get("kid") == kid: 

192 return key 

193 

194 if len(keys) == 1 and isinstance(keys[0], dict): 

195 return cast("dict[str, Any]", keys[0]) 

196 

197 raise OAuth2Error("No matching Google signing key found") 

198 

199 async def _request_json( 

200 self, 

201 method: str, 

202 url: str, 

203 **kwargs: Any, 

204 ) -> dict[str, Any]: 

205 """Fetch JSON via the injected HTTP client or a temporary httpx client.""" 

206 if self.http_client is not None: 

207 response = await self.http_client.request(method, url, **kwargs) 

208 return await self._response_json(response) 

209 

210 import httpx 

211 

212 async with httpx.AsyncClient(timeout=10.0) as client: 

213 httpx_response = await client.request(method, url, **kwargs) 

214 httpx_response.raise_for_status() 

215 return cast("dict[str, Any]", httpx_response.json()) 

216 

217 async def _response_json(self, response: HttpResponse | Any) -> dict[str, Any]: 

218 """Normalise framework HTTP responses and test doubles to JSON dicts.""" 

219 status = getattr(response, "status", None) 

220 if status is None: 

221 status = getattr(response, "status_code", None) 

222 if isinstance(status, int) and status >= 400: 

223 raise OAuth2Error( 

224 f"Google request failed with HTTP {status}", 

225 ) 

226 

227 payload = getattr(response, "json", None) 

228 if callable(payload): 

229 payload = payload() 

230 if inspect.isawaitable(payload): 

231 payload = await payload 

232 if not isinstance(payload, dict): 

233 raise OAuth2Error("Google response did not contain JSON") 

234 return cast("dict[str, Any]", payload) 

235 

236 def _claims_from_payload( 

237 self, 

238 payload: dict[str, Any], 

239 *, 

240 issuer: str | None, 

241 audience: str | None, 

242 ) -> VerifiedIdentityClaims: 

243 """Convert Google payloads into normalized verified identity claims.""" 

244 expires_at = self._timestamp_to_datetime(payload.get("exp")) 

245 issued_at = self._timestamp_to_datetime(payload.get("iat")) 

246 provider_user_id_raw = ( 

247 payload.get("sub") or payload.get("id") or payload.get("provider_user_id") 

248 ) 

249 if not provider_user_id_raw: 

250 raise OAuth2Error("Google payload did not include a subject identifier") 

251 provider_user_id = str(provider_user_id_raw) 

252 

253 return VerifiedIdentityClaims( 

254 provider="google", 

255 provider_user_id=provider_user_id, 

256 email=payload.get("email"), 

257 email_verified=bool(payload.get("email_verified", False)), 

258 name=payload.get("name") or payload.get("given_name"), 

259 picture=payload.get("picture"), 

260 issuer=issuer, 

261 audience=audience, 

262 expires_at=expires_at, 

263 issued_at=issued_at, 

264 raw_data=dict(payload), 

265 ) 

266 

267 def _timestamp_to_datetime(self, value: Any) -> datetime | None: 

268 """Convert a numeric UNIX timestamp to UTC datetime.""" 

269 if value is None: 

270 return None 

271 try: 

272 return datetime.fromtimestamp(float(value), tz=UTC) 

273 except (TypeError, ValueError, OSError): 

274 return None 

275 

276 @staticmethod 

277 def _looks_like_jwt(token: str) -> bool: 

278 """Return True when the token resembles a JWT.""" 

279 return token.count(".") == 2 

280 

281 

282__all__ = [ 

283 "GOOGLE_ISSUERS", 

284 "GOOGLE_JWKS_URL", 

285 "GOOGLE_TOKENINFO_URL", 

286 "GOOGLE_USERINFO_URL", 

287 "GoogleOAuthService", 

288]