Accounts Phase 2: Google sign-in (OAuth 2.0 / OIDC)
- oauth_google.py (stdlib): PKCE, auth URL, code exchange, ID-token claim validation (iss/aud/exp/email_verified — token comes straight from Google's token endpoint over TLS, so no signature re-verify / JWKS needed). - API: GET /api/auth/google/start (302 to Google, PKCE + signed state cookie binding the flow to the browser) and /callback (CSRF-checked state, exchange, find-or-create by verified email → links to an existing magic-link account, session cookie, redirect home). Errors land on /auth/verify?error=google. - SignIn modal: "Continue with Google" + an "or email link" divider. - 112 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,101 @@
|
||||
"""Google Sign-In (OAuth 2.0 / OpenID Connect) — stdlib only.
|
||||
|
||||
Authorization-code flow with PKCE. The ID token is received directly from
|
||||
Google's token endpoint over TLS (server-to-server), so per Google's own
|
||||
guidance we validate its claims (iss/aud/exp/email_verified) without re-verifying
|
||||
the RS256 signature — no JWT/JWKS dependency required.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
|
||||
AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
|
||||
TOKEN_URL = "https://oauth2.googleapis.com/token"
|
||||
_VALID_ISS = {"accounts.google.com", "https://accounts.google.com"}
|
||||
|
||||
|
||||
def configured() -> bool:
|
||||
return bool(
|
||||
os.environ.get("GOODNEWS_GOOGLE_CLIENT_ID")
|
||||
and os.environ.get("GOODNEWS_GOOGLE_CLIENT_SECRET")
|
||||
)
|
||||
|
||||
|
||||
def _b64url(raw: bytes) -> str:
|
||||
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
|
||||
|
||||
|
||||
def new_pkce() -> tuple[str, str]:
|
||||
"""Return (code_verifier, code_challenge) for S256 PKCE."""
|
||||
verifier = _b64url(secrets.token_bytes(48))
|
||||
challenge = _b64url(hashlib.sha256(verifier.encode("ascii")).digest())
|
||||
return verifier, challenge
|
||||
|
||||
|
||||
def auth_url(redirect_uri: str, state: str, code_challenge: str) -> str:
|
||||
params = {
|
||||
"client_id": os.environ["GOODNEWS_GOOGLE_CLIENT_ID"],
|
||||
"redirect_uri": redirect_uri,
|
||||
"response_type": "code",
|
||||
"scope": "openid email profile",
|
||||
"state": state,
|
||||
"code_challenge": code_challenge,
|
||||
"code_challenge_method": "S256",
|
||||
"access_type": "online",
|
||||
"prompt": "select_account",
|
||||
}
|
||||
return AUTH_URL + "?" + urllib.parse.urlencode(params)
|
||||
|
||||
|
||||
def exchange_code(code: str, redirect_uri: str, code_verifier: str) -> dict:
|
||||
data = urllib.parse.urlencode(
|
||||
{
|
||||
"code": code,
|
||||
"client_id": os.environ["GOODNEWS_GOOGLE_CLIENT_ID"],
|
||||
"client_secret": os.environ["GOODNEWS_GOOGLE_CLIENT_SECRET"],
|
||||
"redirect_uri": redirect_uri,
|
||||
"grant_type": "authorization_code",
|
||||
"code_verifier": code_verifier,
|
||||
}
|
||||
).encode("ascii")
|
||||
request = urllib.request.Request(
|
||||
TOKEN_URL,
|
||||
data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
||||
)
|
||||
with urllib.request.urlopen(request, timeout=15) as response:
|
||||
return json.loads(response.read())
|
||||
|
||||
|
||||
def _decode_jwt_payload(token: str) -> dict:
|
||||
parts = token.split(".")
|
||||
if len(parts) != 3:
|
||||
raise ValueError("malformed id_token")
|
||||
payload = parts[1]
|
||||
payload += "=" * (-len(payload) % 4)
|
||||
return json.loads(base64.urlsafe_b64decode(payload))
|
||||
|
||||
|
||||
def verify_id_token(id_token: str) -> dict:
|
||||
"""Validate an ID token's claims (not its signature — see module docstring).
|
||||
|
||||
Returns {"sub", "email", "name"} or raises ValueError.
|
||||
"""
|
||||
claims = _decode_jwt_payload(id_token)
|
||||
if claims.get("iss") not in _VALID_ISS:
|
||||
raise ValueError("unexpected issuer")
|
||||
if claims.get("aud") != os.environ.get("GOODNEWS_GOOGLE_CLIENT_ID"):
|
||||
raise ValueError("audience mismatch")
|
||||
if datetime.now(timezone.utc).timestamp() > float(claims.get("exp", 0)):
|
||||
raise ValueError("id_token expired")
|
||||
if not claims.get("email") or claims.get("email_verified") not in (True, "true"):
|
||||
raise ValueError("email not verified")
|
||||
return {"sub": str(claims["sub"]), "email": claims["email"], "name": claims.get("name")}
|
||||
Reference in New Issue
Block a user