feat(oauth): add configurable token introspection for Google OAuth

.env.example

@@ -23,6 +23,11 @@ SSO_SCOPES=email,openid,profile,session:role-any
 # Get these from your provider's .well-known/openid-configuration endpoint
 SSO_AUTHORIZATION_URL=https://sso.example.com/realms/myrealm/protocol/openid-connect/auth
 SSO_TOKEN_URL=https://sso.example.com/realms/myrealm/protocol/openid-connect/token
+# rfc7662 = POST introspection (Keycloak, Auth0, Okta). tokeninfo = GET ?access_token= (Google)
+SSO_INTROSPECTION_MODE=rfc7662
+# For Google OAuth use tokeninfo mode instead:
+# SSO_INTROSPECTION_MODE=tokeninfo
+# SSO_INTROSPECTION_URL=https://oauth2.googleapis.com/tokeninfo
 SSO_INTROSPECTION_URL=https://sso.example.com/realms/myrealm/protocol/openid-connect/token/introspect
 
 # --- PostgreSQL (required only if ENABLE_AUTH=True, for token storage) ---

docs/authentication.md

@@ -148,6 +148,7 @@ SSO_INTROSPECTION_URL=https://sso.example.com/realms/myrealm/protocol/openid-con
 | `SSO_AUTHORIZATION_URL`     | Auth enabled  | `""`    | Provider's authorization endpoint               |
 | `SSO_TOKEN_URL`             | Auth enabled  | `""`    | Provider's token endpoint                       |
 | `SSO_INTROSPECTION_URL`     | Auth enabled  | `""`    | Provider's token introspection endpoint         |
+| `SSO_INTROSPECTION_MODE`    | Auth enabled  | `rfc7662` | Introspection strategy: `rfc7662` or `tokeninfo` |
 | `SESSION_SECRET`            | Production    | `None`  | Secret key for session middleware               |
 | `COMPATIBLE_WITH_CURSOR`    | Cursor IDE    | `False` | Enables Cursor-compatible OAuth2 flow           |
 | `POSTGRES_HOST`             | Auth enabled  | `None`  | PostgreSQL host for token storage               |
@@ -181,6 +182,23 @@ USE_EXTERNAL_BROWSER_AUTH=True   # or False for production
 
 > **Security note:** `COMPATIBLE_WITH_CURSOR=True` relaxes validation. Use it only for local development with Cursor, not in production.
 
+## Google OAuth
+
+Google does **not** expose an [RFC 7662](https://datatracker.ietf.org/doc/html/rfc7662) token introspection endpoint. Use Google's tokeninfo endpoint instead:
+
+```bash
+SSO_INTROSPECTION_MODE=tokeninfo
+SSO_AUTHORIZATION_URL=https://accounts.google.com/o/oauth2/v2/auth
+SSO_TOKEN_URL=https://oauth2.googleapis.com/token
+SSO_INTROSPECTION_URL=https://oauth2.googleapis.com/tokeninfo
+SSO_SCOPES=https://www.googleapis.com/auth/userinfo.email,openid,https://www.googleapis.com/auth/userinfo.profile
+SSO_CALLBACK_URL=http://localhost:5001/auth/callback/oidc
+```
+
+Set `SSO_INTROSPECTION_MODE=tokeninfo` so the server validates tokens with `GET ?access_token=...` and normalizes the response to the RFC 7662 `{ "active": true }` shape expected by the auth middleware.
+
+> **Note:** Google's tokeninfo endpoint is intended for development and debugging. For production deployments, prefer an OIDC provider with proper introspection support (Keycloak, Auth0, Okta).
+
 ## Discovery Endpoints
 
 When auth is enabled, the server exposes two [RFC 8414](https://datatracker.ietf.org/doc/html/rfc8414) discovery endpoints:

template_mcp_server/src/oauth/handler.py

@@ -10,9 +10,9 @@
 import time
 from typing import Any, Dict, Optional
 
-import httpx
 from requests_oauthlib import OAuth2Session
 
+from template_mcp_server.src.oauth.introspection import create_token_introspector
 from template_mcp_server.src.settings import settings
 from template_mcp_server.utils.pylogger import get_python_logger
 
@@ -67,37 +67,18 @@ def get_access_token_from_refresh_token(refresh_token: str):
 
     @staticmethod
     def introspect_token(token: str) -> Dict[str, Any]:
-        """Introspect a token using the configured SSO introspection endpoint."""
-        introspection_url = settings.SSO_INTROSPECTION_URL
-
-        try:
-            response = httpx.post(
-                introspection_url,
-                data={
-                    "token": token,
-                    "client_id": settings.SSO_CLIENT_ID,
-                    "client_secret": settings.SSO_CLIENT_SECRET,
-                },
-                headers={"Content-Type": "application/x-www-form-urlencoded"},
-                timeout=10.0,
-            )
-            response.raise_for_status()
-
-            introspection_data = response.json()
-            logger.debug(f"Token introspection response: {introspection_data}")
-
-            return introspection_data
-
-        except httpx.HTTPError as e:
-            logger.error(f"Token introspection failed: {e}")
-            return {"active": False, "error": f"Introspection failed: {e}"}
-        except Exception as e:
-            logger.error(f"Unexpected error during token introspection: {e}")
-            return {"active": False, "error": f"Unexpected error: {e}"}
+        """Introspect a token using the configured SSO introspection strategy."""
+        introspector = create_token_introspector(
+            settings.SSO_INTROSPECTION_MODE,
+            settings.SSO_INTROSPECTION_URL,
+            settings.SSO_CLIENT_ID,
+            settings.SSO_CLIENT_SECRET,
+        )
+        return introspector.introspect(token)
 
     @staticmethod
     def verify_access_token(token: str) -> Optional[Dict[str, Any]]:
-        """Verify an access token using RedHat's introspection endpoint."""
+        """Verify an access token via the configured SSO introspection endpoint."""
         introspection_result = OAuth2Handler.introspect_token(token)
 
         if not introspection_result.get("active", False):
@@ -106,7 +87,7 @@ def verify_access_token(token: str) -> Optional[Dict[str, Any]]:
 
         # Check if token is expired
         exp = introspection_result.get("exp")
-        if exp and exp < time.time():
+        if exp is not None and int(exp) < time.time():
             logger.warning("Token has expired")
             return None
 
@@ -120,7 +101,7 @@ def verify_access_token(token: str) -> Optional[Dict[str, Any]]:
 
     @staticmethod
     def verify_authorization_header(auth_header: str) -> Optional[Dict[str, Any]]:
-        """Verify Authorization header with Bearer token using RedHat's introspection."""
+        """Verify Authorization header with Bearer token via SSO introspection."""
         if not auth_header or not auth_header.startswith("Bearer "):
             logger.warning("Invalid authorization header format")
             return None

template_mcp_server/src/oauth/introspection.py

@@ -0,0 +1,143 @@
+"""Pluggable token introspection strategies for OAuth SSO providers."""
+
+from abc import ABC, abstractmethod
+from typing import Any, Dict
+
+import httpx
+
+from template_mcp_server.utils.pylogger import get_python_logger
+
+logger = get_python_logger()
+
+INTROSPECTION_MODE_RFC7662 = "rfc7662"
+INTROSPECTION_MODE_TOKENINFO = "tokeninfo"
+SUPPORTED_INTROSPECTION_MODES = frozenset(
+    {INTROSPECTION_MODE_RFC7662, INTROSPECTION_MODE_TOKENINFO}
+)
+
+
+class TokenIntrospector(ABC):
+    """Validate access tokens against a configured SSO introspection endpoint."""
+
+    @abstractmethod
+    def introspect(self, token: str) -> Dict[str, Any]:
+        """Return RFC 7662-like introspection payload with an ``active`` flag."""
+
+
+class Rfc7662Introspector(TokenIntrospector):
+    """RFC 7662 POST introspection (Keycloak, Auth0, Okta, Red Hat SSO, etc.)."""
+
+    def __init__(
+        self,
+        url: str,
+        client_id: str,
+        client_secret: str,
+        timeout: float = 10.0,
+    ) -> None:
+        """Initialize RFC 7662 introspection client settings."""
+        self.url = url
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.timeout = timeout
+
+    def introspect(self, token: str) -> Dict[str, Any]:
+        """POST the token to the introspection endpoint and return the JSON payload."""
+        try:
+            response = httpx.post(
+                self.url,
+                data={
+                    "token": token,
+                    "client_id": self.client_id,
+                    "client_secret": self.client_secret,
+                },
+                headers={"Content-Type": "application/x-www-form-urlencoded"},
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+            data = response.json()
+            logger.debug(f"Token introspection response: {data}")
+            return data
+        except httpx.HTTPError as e:
+            logger.error(f"Token introspection failed: {e}")
+            return {"active": False, "error": f"Introspection failed: {e}"}
+        except Exception as e:
+            logger.error(f"Unexpected error during token introspection: {e}")
+            return {"active": False, "error": f"Unexpected error: {e}"}
+
+
+class TokenInfoIntrospector(TokenIntrospector):
+    """GET-based tokeninfo validation for providers without RFC 7662 introspection."""
+
+    def __init__(
+        self,
+        url: str,
+        client_id: str = "",
+        timeout: float = 10.0,
+    ) -> None:
+        """Initialize tokeninfo introspection client settings."""
+        self.url = url
+        self.client_id = client_id
+        self.timeout = timeout
+
+    @staticmethod
+    def normalize_response(data: Dict[str, Any], client_id: str = "") -> Dict[str, Any]:
+        """Map tokeninfo-style payloads to RFC 7662-like shape."""
+        if "active" in data:
+            return data
+
+        if data.get("error"):
+            return {"active": False, **data}
+
+        if "sub" in data or "email" in data:
+            normalized = dict(data)
+            normalized["active"] = True
+            normalized.setdefault("token_type", "Bearer")
+            if "exp" in normalized:
+                normalized["exp"] = int(normalized["exp"])
+
+            aud = normalized.get("aud") or normalized.get("azp")
+            if aud and client_id and aud != client_id:
+                return {
+                    "active": False,
+                    "error": "audience_mismatch",
+                    "aud": aud,
+                }
+
+            return normalized
+
+        return {"active": False, **data}
+
+    def introspect(self, token: str) -> Dict[str, Any]:
+        """Validate the token via tokeninfo and return a normalized payload."""
+        try:
+            response = httpx.get(
+                self.url,
+                params={"access_token": token},
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+            data = self.normalize_response(response.json(), self.client_id)
+            logger.debug(f"Token introspection response: {data}")
+            return data
+        except httpx.HTTPError as e:
+            logger.error(f"Token introspection failed: {e}")
+            return {"active": False, "error": f"Introspection failed: {e}"}
+        except Exception as e:
+            logger.error(f"Unexpected error during token introspection: {e}")
+            return {"active": False, "error": f"Unexpected error: {e}"}
+
+
+def create_token_introspector(
+    mode: str,
+    url: str,
+    client_id: str,
+    client_secret: str,
+) -> TokenIntrospector:
+    """Build the introspection strategy configured for the current SSO provider."""
+    if mode == INTROSPECTION_MODE_TOKENINFO:
+        return TokenInfoIntrospector(url=url, client_id=client_id)
+    return Rfc7662Introspector(
+        url=url,
+        client_id=client_id,
+        client_secret=client_secret,
+    )

template_mcp_server/src/settings.py

@@ -1,7 +1,7 @@
 """Settings for the Template MCP Server."""
 
 from functools import cached_property
-from typing import List, Optional
+from typing import List, Literal, Optional
 
 from dotenv import load_dotenv
 from pydantic import Field, model_validator
@@ -179,6 +179,18 @@ class Settings(BaseSettings):
             "description": "SSO token introspection endpoint URL",
         },
     )
+    SSO_INTROSPECTION_MODE: Literal["rfc7662", "tokeninfo"] = Field(
+        default="rfc7662",
+        json_schema_extra={
+            "env": "SSO_INTROSPECTION_MODE",
+            "description": (
+                "Token validation strategy: rfc7662 (POST introspection) or "
+                "tokeninfo (GET access_token query param for providers like Google)"
+            ),
+            "example": "rfc7662",
+            "enum": ["rfc7662", "tokeninfo"],
+        },
+    )
     SSO_SCOPES: str = Field(
         default="email,openid,profile,session:role-any",
         json_schema_extra={

tests/test_oauth_handler.py

@@ -121,11 +121,12 @@ def test_get_access_token_from_refresh_token(
         assert result == mock_token
 
     @patch("template_mcp_server.src.oauth.handler.settings")
-    @patch("template_mcp_server.src.oauth.handler.httpx.post")
+    @patch("template_mcp_server.src.oauth.introspection.httpx.post")
     def test_introspect_token_success(self, mock_post, mock_settings):
         """Test successful token introspection."""
         mock_settings.SSO_CLIENT_ID = "client123"
         mock_settings.SSO_CLIENT_SECRET = "secret123"
+        mock_settings.SSO_INTROSPECTION_MODE = "rfc7662"
 
         mock_response = Mock()
         mock_response.raise_for_status.return_value = None
@@ -147,7 +148,7 @@ def test_introspect_token_success(self, mock_post, mock_settings):
         assert result == {"active": True, "sub": "user123"}
 
     @patch("template_mcp_server.src.oauth.handler.settings")
-    @patch("template_mcp_server.src.oauth.handler.httpx.post")
+    @patch("template_mcp_server.src.oauth.introspection.httpx.post")
     def test_introspect_token_http_error(self, mock_post, mock_settings):
         """Test token introspection with HTTP error."""
         mock_settings.SSO_CLIENT_ID = "client123"
@@ -161,7 +162,7 @@ def test_introspect_token_http_error(self, mock_post, mock_settings):
         assert "Introspection failed" in result["error"]
 
     @patch("template_mcp_server.src.oauth.handler.settings")
-    @patch("template_mcp_server.src.oauth.handler.httpx.post")
+    @patch("template_mcp_server.src.oauth.introspection.httpx.post")
     def test_introspect_token_unexpected_error(self, mock_post, mock_settings):
         """Test token introspection with unexpected error."""
         mock_settings.SSO_CLIENT_ID = "client123"
@@ -270,7 +271,7 @@ class TestOAuth2HandlerIntegration:
 
     @patch("template_mcp_server.src.oauth.handler.settings")
     @patch("template_mcp_server.src.oauth.handler.OAuth2Session")
-    @patch("template_mcp_server.src.oauth.handler.httpx.post")
+    @patch("template_mcp_server.src.oauth.introspection.httpx.post")
     def test_full_oauth_flow_simulation(
         self, mock_post, mock_oauth_session, mock_settings
     ):