src/auth/google.py

"""Google OAuth2 (authorization-code flow) — no third-party dep.

Flow:
  1. GET /auth/google           -> 302 redirect to Google's consent page
  2. Google redirects back to /auth/google/callback?code=...
  3. We exchange `code` for tokens (POST to oauth2.googleapis.com/token)
  4. We fetch the user (openid userinfo endpoint)
  5. On success: set session cookie, redirect to dashboard.html

Docs: https://developers.google.com/identity/openid-connect
"""
from __future__ import annotations   # defer annotation evaluation (PEP 563)

import os   # read GOOGLE_* env vars
import urllib.parse   # build the authorize URL query string

import httpx   # async HTTP for the token exchange + user fetch

GOOGLE_API = "https://oauth2.googleapis.com"   # token endpoint host
AUTHORIZE_URL = "https://accounts.google.com/o/oauth2/v2/auth"   # the user-facing consent page
TOKEN_URL = GOOGLE_API + "/token"   # code -> tokens exchange
USERINFO_URL = "https://openidconnect.googleapis.com/v1/userinfo"   # current user info (OpenID)

# Read the OAuth app config from env (set as HF Space secrets).
CLIENT_ID = os.environ.get("GOOGLE_CLIENT_ID", "")   # public app id (from Google Cloud Console)
CLIENT_SECRET = os.environ.get("GOOGLE_CLIENT_SECRET", "")   # app secret (never shipped to browser)
REDIRECT_URI = os.environ.get("GOOGLE_REDIRECT_URI", "")   # our callback URL (must be in Authorized redirect URIs)
HOSTED_DOMAIN = os.environ.get("GOOGLE_HOSTED_DOMAIN", "").strip()   # optional: restrict to a Google Workspace domain (hd param)

# Scopes: openid = id token; email = email address; profile = name/picture.
SCOPES = "openid email profile"   # request identity, email, and basic profile


def is_configured() -> bool:   # is Google login enabled on this deployment?
    """True only if all required env vars are set (used to show/hide the button)."""
    return bool(CLIENT_ID and CLIENT_SECRET and REDIRECT_URI and os.environ.get("SESSION_SECRET"))   # all four required


def authorize_url(state: str) -> str:   # build the Google authorize redirect URL
    """Build the full authorize URL with query params for the OAuth redirect."""
    params = {   # query string parameters
        "client_id": CLIENT_ID,   # our app id
        "redirect_uri": REDIRECT_URI,   # where Google sends the user back
        "response_type": "code",   # authorization-code flow (not implicit)
        "scope": SCOPES,   # what user data we request
        "state": state,   # CSRF token echoed back by Google
        "prompt": "consent",   # always show the consent screen (allows account picker + re-login)
        "access_type": "online",   # we don't need refresh tokens (stateless sessions)
    }
    if HOSTED_DOMAIN:   # restrict to a specific Google Workspace domain
        params["hd"] = HOSTED_DOMAIN   # Google enforces this server-side
    return AUTHORIZE_URL + "?" + urllib.parse.urlencode(params)   # encode + concatenate


async def exchange_code(code: str, client: httpx.AsyncClient) -> dict | None:   # code -> tokens
    """Exchange an authorization code for tokens. Returns None on failure."""
    data = {   # token endpoint form params
        "client_id": CLIENT_ID,   # app id
        "client_secret": CLIENT_SECRET,   # app secret (server-side only)
        "grant_type": "authorization_code",   # the OAuth grant type
        "code": code,   # the code from the callback
        "redirect_uri": REDIRECT_URI,   # must match the authorize request exactly
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}   # Google wants form-encoded
    try:   # network call to Google
        r = await client.post(TOKEN_URL, data=data, headers=headers, timeout=15.0)   # POST exchange
    except Exception:   # network error / timeout
        return None   # caller treats as login failure
    if r.status_code != 200:   # Google rejected the code (expired, reused, mismatched redirect)
        return None   # caller shows a login error
    return r.json()   # {access_token, id_token, token_type, scope, ...}


async def fetch_user(access_token: str, client: httpx.AsyncClient) -> dict | None:   # get the user profile
    """Fetch the OpenID userinfo with the access token. Returns None on failure."""
    headers = {"Authorization": f"Bearer {access_token}"}   # Bearer auth with the OAuth token
    try:   # network call
        r = await client.get(USERINFO_URL, headers=headers, timeout=15.0)   # GET userinfo
    except Exception:   # network error
        return None   # treat as failure
    if r.status_code != 200:   # token invalid / revoked
        return None   # reject
    return r.json()   # {sub, email, email_verified, name, picture, ...}