life-echo/api/app/features/auth/google.py

"""Google OpenID Connect token verification."""

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

import httpx
import jwt
from jwt import PyJWKClient

from app.core.config import settings
from app.core.errors import ProviderError, ServiceUnavailableError
from app.core.logging import get_logger
from app.features.auth.service_errors import AuthError

logger = get_logger(__name__)

GOOGLE_JWKS_URL = "https://www.googleapis.com/oauth2/v3/certs"
GOOGLE_ISSUERS = {"accounts.google.com", "https://accounts.google.com"}

_jwks_client = PyJWKClient(GOOGLE_JWKS_URL)


@dataclass(frozen=True)
class GoogleIdentity:
    subject: str
    email: str
    email_verified: bool
    name: str
    picture: str | None
    audience: str = ""


def _configured_client_ids() -> list[str]:
    raw = (settings.google_oauth_client_ids or "").strip()
    return [part.strip() for part in raw.split(",") if part.strip()]


def verify_google_id_token(id_token: str) -> GoogleIdentity:
    """Verify a Google ID token and return the minimal identity claims."""
    client_ids = _configured_client_ids()
    if not client_ids:
        raise ServiceUnavailableError("Google 登录尚未配置")

    verifier_url = (settings.google_token_verifier_url or "").strip()
    if verifier_url:
        return _verify_google_id_token_remotely(id_token, client_ids, verifier_url)

    return _verify_google_id_token_locally(id_token, client_ids)


def _verify_google_id_token_locally(
    id_token: str,
    client_ids: list[str],
) -> GoogleIdentity:
    try:
        signing_key = _jwks_client.get_signing_key_from_jwt(id_token)
        payload = jwt.decode(
            id_token,
            signing_key.key,
            algorithms=["RS256"],
            audience=client_ids,
            options={"verify_iss": False},
        )
    except jwt.InvalidAudienceError as exc:
        raise AuthError("Google 登录客户端不匹配", "INVALID_TOKEN") from exc
    except jwt.ExpiredSignatureError as exc:
        raise AuthError("Google 登录凭证已过期，请重试", "INVALID_TOKEN") from exc
    except jwt.InvalidTokenError as exc:
        raise AuthError("Google 登录凭证无效", "INVALID_TOKEN") from exc
    except Exception as exc:
        logger.exception("Google JWKS/token verification failed: {}", exc)
        raise ProviderError("Google 登录校验暂时不可用，请稍后重试") from exc

    issuer = str(payload.get("iss") or "")
    if issuer not in GOOGLE_ISSUERS:
        raise AuthError("Google 登录凭证来源无效", "INVALID_TOKEN")

    audience = _matched_audience(payload, client_ids)
    return _identity_from_claims(payload, audience=audience)


def _verify_google_id_token_remotely(
    id_token: str,
    client_ids: list[str],
    verifier_url: str,
) -> GoogleIdentity:
    secret = (settings.google_token_verifier_secret or "").strip()
    if not secret:
        raise ServiceUnavailableError("Google 登录校验代理未配置")

    try:
        response = httpx.post(
            verifier_url,
            json={"id_token": id_token},
            headers={
                "Accept": "application/json",
                "Authorization": f"Bearer {secret}",
            },
            timeout=settings.google_token_verifier_timeout_seconds or 5.0,
        )
    except (httpx.TimeoutException, httpx.RequestError) as exc:
        logger.warning("Google token verifier request failed: {}", exc)
        raise ProviderError("Google 登录校验暂时不可用，请稍后重试") from exc

    try:
        data = response.json()
    except ValueError as exc:
        logger.warning(
            "Google token verifier returned non-JSON response: status={}",
            response.status_code,
        )
        raise ProviderError("Google 登录校验暂时不可用，请稍后重试") from exc

    if not isinstance(data, dict):
        logger.warning(
            "Google token verifier returned invalid JSON shape: status={}",
            response.status_code,
        )
        raise ProviderError("Google 登录校验暂时不可用，请稍后重试")

    if response.status_code != 200:
        _raise_remote_verifier_error(response.status_code, data)

    audience = str(data.get("audience") or "").strip()
    if audience not in client_ids:
        raise AuthError("Google 登录客户端不匹配", "INVALID_TOKEN")

    return _identity_from_claims(data, audience=audience)


def _raise_remote_verifier_error(
    status_code: int,
    data: Mapping[str, Any],
) -> None:
    code = str(data.get("error") or "")
    if code == "INVALID_TOKEN" or (status_code in (400, 401) and not code):
        raise AuthError("Google 登录凭证无效", "INVALID_TOKEN")
    if code == "INVALID_REQUEST":
        raise AuthError("Google 登录凭证无效", "INVALID_TOKEN")
    if code in {"CONFIGURATION_ERROR", "UNAUTHORIZED"}:
        raise ServiceUnavailableError("Google 登录校验代理配置错误")

    logger.warning(
        "Google token verifier returned error: status={} code={}",
        status_code,
        code or "<empty>",
    )
    raise ProviderError("Google 登录校验暂时不可用，请稍后重试")


def _matched_audience(payload: Mapping[str, Any], client_ids: list[str]) -> str:
    raw_audience = payload.get("aud")
    audiences = (
        raw_audience
        if isinstance(raw_audience, list)
        else [raw_audience]
        if raw_audience is not None
        else []
    )
    for audience in audiences:
        audience_str = str(audience or "").strip()
        if audience_str in client_ids:
            return audience_str
    raise AuthError("Google 登录客户端不匹配", "INVALID_TOKEN")


def _identity_from_claims(
    payload: Mapping[str, Any],
    *,
    audience: str,
) -> GoogleIdentity:
    subject = str(payload.get("sub") or payload.get("subject") or "").strip()
    email = str(payload.get("email") or "").strip().lower()
    email_verified = (
        payload.get("email_verified") is True
        or str(payload.get("email_verified")).lower() == "true"
    )
    if not subject or not email or not email_verified:
        raise AuthError("Google 账号邮箱未验证，无法登录", "INVALID_TOKEN")

    return GoogleIdentity(
        subject=subject,
        email=email,
        email_verified=email_verified,
        name=str(payload.get("name") or "").strip(),
        picture=str(payload.get("picture") or "").strip() or None,
        audience=audience,
    )