ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-20 20:44:17 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix(google): replace deprecated /oauth2/v1/tokeninfo with /oauth2/v3/userinfo (#3603)

Browse source 
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
This commit is contained in:
AIKAWA Shigechika 2026-03-25 23:39:34 +09:00 • committed by GitHub
commit c3f0223bbb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 369 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

100
src/fastmcp/server/auth/providers/google.py
View file

@ -57,8 +57,11 @@ def _normalize_google_scope(scope: str) -> str:
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.
Google OAuth tokens are opaque (not JWTs), so we verify them by calling
Google's tokeninfo endpoint with the access token as a query parameter.
This returns the OAuth app ID (``aud``), granted scopes, and expiry time.
User profile data (name, picture, etc.) is fetched separately from the
v2 userinfo endpoint when the token is valid.
"""
def __init__(
@ -87,16 +90,23 @@ class GoogleTokenVerifier(TokenVerifier):
self._http_client = http_client
async def verify_token(self, token: str) -> AccessToken | None:
"""Verify Google OAuth token by calling Google's tokeninfo API."""
"""Verify a Google OAuth token using the tokeninfo endpoint.
Calls ``https://oauth2.googleapis.com/tokeninfo?access_token=TOKEN``
to validate the token and retrieve the OAuth app ID (``aud``), granted
scopes, and expiry time. On success, fetches user profile data from
the v2 userinfo endpoint to populate name, picture, and locale claims.
"""
try:
async with (
contextlib.nullcontext(self._http_client)
if self._http_client is not None
else httpx.AsyncClient(timeout=self.timeout_seconds)
) as client:
# Use Google's tokeninfo endpoint to validate the token
# Step 1: Verify token via tokeninfo endpoint.
# Returns aud (OAuth app ID), scope (space-separated), expires_in, sub, email.
response = await client.get(
"https://www.googleapis.com/oauth2/v1/tokeninfo",
"https://oauth2.googleapis.com/tokeninfo",
params={"access_token": token},
headers={"User-Agent": "FastMCP-Google-OAuth"},
)
@ -108,19 +118,23 @@ class GoogleTokenVerifier(TokenVerifier):
)
return None
token_info = response.json()
token_data = 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")
# aud is the OAuth app ID (client_id / audience)
aud = token_data.get("aud")
if not aud:
logger.debug("Google tokeninfo missing 'aud' claim")
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()
]
# sub is required (unique Google user ID)
sub = token_data.get("sub")
if not sub:
logger.debug("Google tokeninfo missing 'sub' claim")
return None
# Parse scopes directly from the tokeninfo response (space-separated)
scope_str = token_data.get("scope", "")
token_scopes = scope_str.split() if scope_str else []
# Check required scopes
if self.required_scopes:
@ -134,46 +148,46 @@ class GoogleTokenVerifier(TokenVerifier):
)
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)
# Compute expiry from expires_in (seconds until expiry)
expires_at: int | None = None
expires_in = token_data.get("expires_in")
if expires_in is not None:
with contextlib.suppress(ValueError, TypeError):
expires_at = int(time.time()) + int(expires_in)
# Calculate expiration time
expires_at = None
if expires_in:
expires_at = int(time.time() + int(expires_in))
# Step 2: Fetch user profile from v2 userinfo endpoint.
# tokeninfo provides auth data; userinfo provides name, picture, locale.
user_data: dict = {}
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 profile: %s", e)
# Create AccessToken with Google user info
access_token = AccessToken(
token=token,
client_id=token_info.get(
"audience", "unknown"
), # Use audience as client_id
client_id=aud,
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"),
"sub": sub,
"aud": aud,
"email": token_data.get("email") or user_data.get("email"),
"email_verified": token_data.get("email_verified")
or user_data.get("verified_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,
"google_user_data": user_data or None,
},
)
logger.debug("Google token verified successfully")

303
tests/server/auth/providers/test_google.py
View file

@ -1,7 +1,11 @@
"""Tests for Google OAuth provider."""
import re
import time
import pytest
from key_value.aio.stores.memory import MemoryStore
from pytest_httpx import HTTPXMock
from fastmcp.server.auth.providers.google import (
GOOGLE_SCOPE_ALIASES,
@ -231,3 +235,302 @@ class TestGoogleScopeNormalization:
"""Verify the alias map covers the known Google shorthands."""
assert "email" in GOOGLE_SCOPE_ALIASES
assert "profile" in GOOGLE_SCOPE_ALIASES
# Regex patterns for URL matching (tokeninfo uses query params)
_TOKENINFO_RE = re.compile(r"https://oauth2\.googleapis\.com/tokeninfo")
_USERINFO_RE = re.compile(r"https://www\.googleapis\.com/oauth2/v2/userinfo")
class TestGoogleTokenVerifier:
"""Test GoogleTokenVerifier.verify_token() using the tokeninfo endpoint."""
TOKENINFO_URL = "https://oauth2.googleapis.com/tokeninfo"
USERINFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo"
async def test_valid_token_openid_only(self, httpx_mock: HTTPXMock):
"""A token with only openid scope is accepted; client_id comes from 'aud'."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"sub": "12345",
"scope": "openid",
"expires_in": "3600",
},
)
httpx_mock.add_response(
url=_USERINFO_RE,
json={"sub": "12345"},
)
verifier = GoogleTokenVerifier()
result = await verifier.verify_token("valid-token")
assert result is not None
assert result.client_id == "123.apps.googleusercontent.com"
assert result.scopes == ["openid"]
assert result.expires_at is not None
assert result.claims["sub"] == "12345"
assert result.claims["aud"] == "123.apps.googleusercontent.com"
async def test_valid_token_with_email_and_profile(self, httpx_mock: HTTPXMock):
"""A token with email+profile scope returns correct scopes and profile claims."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"sub": "12345",
"email": "user@example.com",
"email_verified": "true",
"scope": "openid https://www.googleapis.com/auth/userinfo.email https://www.googleapis.com/auth/userinfo.profile",
"expires_in": "3600",
},
)
httpx_mock.add_response(
url=_USERINFO_RE,
json={
"sub": "12345",
"email": "user@example.com",
"verified_email": True,
"name": "Test User",
"picture": "https://example.com/photo.jpg",
"given_name": "Test",
"family_name": "User",
"locale": "en",
},
)
verifier = GoogleTokenVerifier()
result = await verifier.verify_token("valid-token")
assert result is not None
assert result.client_id == "123.apps.googleusercontent.com"
assert "openid" in result.scopes
assert "https://www.googleapis.com/auth/userinfo.email" in result.scopes
assert "https://www.googleapis.com/auth/userinfo.profile" in result.scopes
assert result.claims["email"] == "user@example.com"
assert result.claims["name"] == "Test User"
assert result.claims["picture"] == "https://example.com/photo.jpg"
async def test_expired_or_invalid_token_returns_none(self, httpx_mock: HTTPXMock):
"""HTTP 400 from tokeninfo endpoint causes verify_token to return None."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
status_code=400,
json={
"error": "invalid_token",
"error_description": "Token has been expired or revoked.",
},
)
verifier = GoogleTokenVerifier()
result = await verifier.verify_token("expired-token")
assert result is None
async def test_missing_aud_returns_none(self, httpx_mock: HTTPXMock):
"""A 200 response without 'aud' is rejected."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={"sub": "12345", "scope": "openid", "expires_in": "3600"},
)
verifier = GoogleTokenVerifier()
result = await verifier.verify_token("token-without-aud")
assert result is None
async def test_missing_sub_returns_none(self, httpx_mock: HTTPXMock):
"""A 200 response without 'sub' is rejected."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"scope": "openid",
"expires_in": "3600",
},
)
verifier = GoogleTokenVerifier()
result = await verifier.verify_token("token-without-sub")
assert result is None
async def test_required_scopes_satisfied(self, httpx_mock: HTTPXMock):
"""Token with required scopes passes the scope check."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"sub": "12345",
"email": "user@example.com",
"scope": "openid https://www.googleapis.com/auth/userinfo.email",
"expires_in": "3600",
},
)
httpx_mock.add_response(
url=_USERINFO_RE,
json={"sub": "12345", "email": "user@example.com"},
)
verifier = GoogleTokenVerifier(
required_scopes=["openid", "email"],
)
result = await verifier.verify_token("valid-token")
assert result is not None
async def test_required_scopes_not_satisfied_returns_none(
self, httpx_mock: HTTPXMock
):
"""Token without required scopes is rejected."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"sub": "12345",
"scope": "openid",
"expires_in": "3600",
},
)
verifier = GoogleTokenVerifier(
required_scopes=[
"openid",
"https://www.googleapis.com/auth/userinfo.email",
],
)
result = await verifier.verify_token("token-missing-email-scope")
assert result is None
async def test_uses_query_param_not_bearer_header(self, httpx_mock: HTTPXMock):
"""verify_token sends the token as a query parameter to tokeninfo, not a Bearer header."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"sub": "12345",
"scope": "openid",
"expires_in": "3600",
},
)
httpx_mock.add_response(
url=_USERINFO_RE,
json={"sub": "12345"},
)
verifier = GoogleTokenVerifier()
await verifier.verify_token("my-access-token")
requests = httpx_mock.get_requests()
tokeninfo_req = requests[0]
assert "access_token=my-access-token" in str(tokeninfo_req.url)
assert "Authorization" not in tokeninfo_req.headers
async def test_calls_tokeninfo_endpoint(self, httpx_mock: HTTPXMock):
"""verify_token calls the tokeninfo endpoint, not the userinfo endpoint, for verification."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"sub": "12345",
"scope": "openid",
"expires_in": "3600",
},
)
httpx_mock.add_response(
url=_USERINFO_RE,
json={"sub": "12345"},
)
verifier = GoogleTokenVerifier()
await verifier.verify_token("valid-token")
requests = httpx_mock.get_requests()
assert len(requests) >= 1
assert "tokeninfo" in str(requests[0].url)
assert "oauth2.googleapis.com" in str(requests[0].url)
async def test_expires_at_computed_from_expires_in(self, httpx_mock: HTTPXMock):
"""expires_at is set from expires_in returned by tokeninfo."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"sub": "12345",
"scope": "openid",
"expires_in": "3600",
},
)
httpx_mock.add_response(
url=_USERINFO_RE,
json={"sub": "12345"},
)
before = int(time.time())
verifier = GoogleTokenVerifier()
result = await verifier.verify_token("valid-token")
after = int(time.time())
assert result is not None
assert result.expires_at is not None
assert before + 3600 <= result.expires_at <= after + 3600
async def test_profile_data_fetched_from_userinfo(self, httpx_mock: HTTPXMock):
"""Profile data (name, picture, locale) comes from the v2 userinfo endpoint."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"sub": "12345",
"scope": "openid https://www.googleapis.com/auth/userinfo.profile",
"expires_in": "3600",
},
)
httpx_mock.add_response(
url=_USERINFO_RE,
json={
"sub": "12345",
"name": "Test User",
"picture": "https://example.com/photo.jpg",
"given_name": "Test",
"family_name": "User",
"locale": "en",
},
)
verifier = GoogleTokenVerifier()
result = await verifier.verify_token("valid-token")
assert result is not None
assert result.claims["name"] == "Test User"
assert result.claims["picture"] == "https://example.com/photo.jpg"
assert result.claims["given_name"] == "Test"
assert result.claims["family_name"] == "User"
assert result.claims["locale"] == "en"
async def test_non_openid_scopes_checked_correctly(self, httpx_mock: HTTPXMock):
"""Calendar scope is correctly checked from the tokeninfo scope string."""
httpx_mock.add_response(
url=_TOKENINFO_RE,
json={
"aud": "123.apps.googleusercontent.com",
"sub": "12345",
"scope": "openid https://www.googleapis.com/auth/calendar",
"expires_in": "3600",
},
)
httpx_mock.add_response(
url=_USERINFO_RE,
json={"sub": "12345"},
)
verifier = GoogleTokenVerifier(
required_scopes=["openid", "https://www.googleapis.com/auth/calendar"],
)
result = await verifier.verify_token("valid-token")
assert result is not None
assert "https://www.googleapis.com/auth/calendar" in result.scopes

18
tests/server/middleware/test_rate_limiting.py
View file

@ -421,22 +421,20 @@ class TestRateLimitingMiddlewareIntegration:
async def test_global_rate_limiting(self, rate_limit_server):
"""Test global rate limiting across all clients."""
rate_limit_server.add_middleware(
RateLimitingMiddleware(
max_requests_per_second=6.0,
burst_capacity=5, # 1 init + 2 list_tools + 2 calls before limit
global_limit=True, # Accounting for initialization and list_tools calls
)
middleware = RateLimitingMiddleware(
max_requests_per_second=0.001,
burst_capacity=1000,
global_limit=True,
)
rate_limit_server.add_middleware(middleware)
async with Client(rate_limit_server) as client:
# Use up the global capacity
await client.call_tool("quick_action", {"message": "1"})
await client.call_tool("quick_action", {"message": "2"})
# Should be globally rate limited
middleware.global_limiter.tokens = 0
with pytest.raises(ToolError, match="Global rate limit exceeded"):
await client.call_tool("quick_action", {"message": "3"})
await client.call_tool("quick_action", {"message": "blocked"})
async def test_rate_limiting_recovery_over_time(self, rate_limit_server):
"""Test that rate limiting allows requests again after time passes."""