src/fastmcp/server/auth/providers/google.py

"""Google OAuth provider for FastMCP.

This module provides a complete Google OAuth integration that's ready to use
with just a client ID and client secret. It handles all the complexity of
Google's OAuth flow, token validation, and user management.

Example:
    ```python
    from fastmcp import FastMCP
    from fastmcp.server.auth.providers.google import GoogleProvider

    # Simple Google OAuth protection
    auth = GoogleProvider(
        client_id="your-google-client-id.apps.googleusercontent.com",
        client_secret="your-google-client-secret"
    )

    mcp = FastMCP("My Protected Server", auth=auth)
    ```
"""

from __future__ import annotations

import time

import httpx
from pydantic import AnyHttpUrl, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

from fastmcp.server.auth import TokenVerifier
from fastmcp.server.auth.auth import AccessToken
from fastmcp.server.auth.proxy import OAuthProxy
from fastmcp.server.auth.registry import register_provider
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import NotSet, NotSetT

logger = get_logger(__name__)


class GoogleProviderSettings(BaseSettings):
    """Settings for Google OAuth provider."""

    model_config = SettingsConfigDict(
        env_prefix="FASTMCP_SERVER_AUTH_GOOGLE_",
        env_file=".env",
        extra="ignore",
    )

    client_id: str | None = None
    client_secret: SecretStr | None = None
    base_url: AnyHttpUrl | str | None = None
    redirect_path: str | None = None
    required_scopes: list[str] | None = None
    timeout_seconds: int | None = None


class GoogleTokenVerifier(TokenVerifier):
    """Token verifier for Google OAuth tokens.

    Google OAuth tokens are opaque (not JWTs), so we verify them
    by calling Google's tokeninfo API to check if they're valid and get user info.
    """

    def __init__(
        self,
        *,
        required_scopes: list[str] | None = None,
        timeout_seconds: int = 10,
    ):
        """Initialize the Google token verifier.

        Args:
            required_scopes: Required OAuth scopes (e.g., ['openid', 'email'])
            timeout_seconds: HTTP request timeout
        """
        super().__init__(required_scopes=required_scopes)
        self.timeout_seconds = timeout_seconds

    async def verify_token(self, token: str) -> AccessToken | None:
        """Verify Google OAuth token by calling Google's tokeninfo API."""
        try:
            async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
                # Use Google's tokeninfo endpoint to validate the token
                response = await client.get(
                    "https://www.googleapis.com/oauth2/v1/tokeninfo",
                    params={"access_token": token},
                    headers={"User-Agent": "FastMCP-Google-OAuth"},
                )

                if response.status_code != 200:
                    logger.debug(
                        "Google token verification failed: %d",
                        response.status_code,
                    )
                    return None

                token_info = response.json()

                # Check if token is expired
                expires_in = token_info.get("expires_in")
                if expires_in and int(expires_in) <= 0:
                    logger.debug("Google token has expired")
                    return None

                # Extract scopes from token info
                scope_string = token_info.get("scope", "")
                token_scopes = [
                    scope.strip() for scope in scope_string.split(" ") if scope.strip()
                ]

                # Check required scopes
                if self.required_scopes:
                    token_scopes_set = set(token_scopes)
                    required_scopes_set = set(self.required_scopes)
                    if not required_scopes_set.issubset(token_scopes_set):
                        logger.debug(
                            "Google token missing required scopes. Has %d, needs %d",
                            len(token_scopes_set),
                            len(required_scopes_set),
                        )
                        return None

                # Get additional user info if we have the right scopes
                user_data = {}
                if "openid" in token_scopes or "profile" in token_scopes:
                    try:
                        userinfo_response = await client.get(
                            "https://www.googleapis.com/oauth2/v2/userinfo",
                            headers={
                                "Authorization": f"Bearer {token}",
                                "User-Agent": "FastMCP-Google-OAuth",
                            },
                        )
                        if userinfo_response.status_code == 200:
                            user_data = userinfo_response.json()
                    except Exception as e:
                        logger.debug("Failed to fetch Google user info: %s", e)

                # Calculate expiration time
                expires_at = None
                if expires_in:
                    expires_at = int(time.time() + int(expires_in))

                # Create AccessToken with Google user info
                access_token = AccessToken(
                    token=token,
                    client_id=token_info.get(
                        "audience", "unknown"
                    ),  # Use audience as client_id
                    scopes=token_scopes,
                    expires_at=expires_at,
                    claims={
                        "sub": user_data.get("id")
                        or token_info.get("user_id", "unknown"),
                        "email": user_data.get("email"),
                        "name": user_data.get("name"),
                        "picture": user_data.get("picture"),
                        "given_name": user_data.get("given_name"),
                        "family_name": user_data.get("family_name"),
                        "locale": user_data.get("locale"),
                        "google_user_data": user_data,
                        "google_token_info": token_info,
                    },
                )
                logger.debug("Google token verified successfully")
                return access_token

        except httpx.RequestError as e:
            logger.debug("Failed to verify Google token: %s", e)
            return None
        except Exception as e:
            logger.debug("Google token verification error: %s", e)
            return None


@register_provider("Google")
class GoogleProvider(OAuthProxy):
    """Complete Google OAuth provider for FastMCP.

    This provider makes it trivial to add Google OAuth protection to any
    FastMCP server. Just provide your Google OAuth app credentials and
    a base URL, and you're ready to go.

    Features:
    - Transparent OAuth proxy to Google
    - Automatic token validation via Google's tokeninfo API
    - User information extraction from Google APIs
    - Minimal configuration required

    Example:
        ```python
        from fastmcp import FastMCP
        from fastmcp.server.auth.providers.google import GoogleProvider

        auth = GoogleProvider(
            client_id="123456789.apps.googleusercontent.com",
            client_secret="GOCSPX-abc123...",
            base_url="https://my-server.com"  # Optional, defaults to http://localhost:8000
        )

        mcp = FastMCP("My App", auth=auth)
        ```
    """

    def __init__(
        self,
        *,
        client_id: str | NotSetT = NotSet,
        client_secret: str | NotSetT = NotSet,
        base_url: AnyHttpUrl | str | NotSetT = NotSet,
        redirect_path: str | NotSetT = NotSet,
        required_scopes: list[str] | None | NotSetT = NotSet,
        timeout_seconds: int | NotSetT = NotSet,
    ):
        """Initialize Google OAuth provider.

        Args:
            client_id: Google OAuth client ID (e.g., "123456789.apps.googleusercontent.com")
            client_secret: Google OAuth client secret (e.g., "GOCSPX-abc123...")
            base_url: Public URL of your FastMCP server (for OAuth callbacks)
            redirect_path: Redirect path configured in Google OAuth app (defaults to "/oauth/callback")
            required_scopes: Required Google scopes (defaults to []). Common scopes include:
                - "openid" for OpenID Connect
                - "https://www.googleapis.com/auth/userinfo.email" for email access
                - "https://www.googleapis.com/auth/userinfo.profile" for profile info
            timeout_seconds: HTTP request timeout for Google API calls
        """
        settings = GoogleProviderSettings.model_validate(
            {
                k: v
                for k, v in {
                    "client_id": client_id,
                    "client_secret": client_secret,
                    "base_url": base_url,
                    "redirect_path": redirect_path,
                    "required_scopes": required_scopes,
                    "timeout_seconds": timeout_seconds,
                }.items()
                if v is not NotSet
            }
        )

        # Validate required settings
        if not settings.client_id:
            raise ValueError(
                "client_id is required - set via parameter or FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_ID"
            )
        if not settings.client_secret:
            raise ValueError(
                "client_secret is required - set via parameter or FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET"
            )

        # Apply defaults
        base_url_final = settings.base_url or "http://localhost:8000"
        redirect_path_final = settings.redirect_path or "/oauth/callback"
        timeout_seconds_final = settings.timeout_seconds or 10
        required_scopes_final = settings.required_scopes or []

        # Create Google token verifier
        token_verifier = GoogleTokenVerifier(
            required_scopes=required_scopes_final,
            timeout_seconds=timeout_seconds_final,
        )

        # Extract secret string from SecretStr
        client_secret_str = (
            settings.client_secret.get_secret_value() if settings.client_secret else ""
        )

        # Initialize OAuth proxy with Google endpoints
        super().__init__(
            upstream_authorization_endpoint="https://accounts.google.com/o/oauth2/v2/auth",
            upstream_token_endpoint="https://oauth2.googleapis.com/token",
            upstream_client_id=settings.client_id,
            upstream_client_secret=client_secret_str,
            token_verifier=token_verifier,
            base_url=base_url_final,
            redirect_path=redirect_path_final,
            issuer_url=base_url_final,  # We act as the issuer for client registration
        )

        logger.info(
            "Initialized Google OAuth provider for client %s with scopes: %s",
            settings.client_id,
            required_scopes_final,
        )