From 48b6bd990c477cce96d1bc3346c3f5e7ff501ea6 Mon Sep 17 00:00:00 2001 From: abhiskum Date: Sun, 7 Jun 2026 21:11:40 +0530 Subject: [PATCH 1/3] feat(oauth): add configurable token introspection strategies Google OAuth lacks RFC 7662 introspection, which caused bearer token validation to fail on protected routes. Introduce SSO_INTROSPECTION_MODE with pluggable rfc7662 and tokeninfo strategies so providers like Google can be configured explicitly without hardcoded URL detection. Co-authored-by: Cursor --- docs/authentication.md | 18 +++ template_mcp_server/src/oauth/handler.py | 43 ++---- .../src/oauth/introspection.py | 143 ++++++++++++++++++ template_mcp_server/src/settings.py | 14 +- tests/test_oauth_handler.py | 9 +- tests/test_oauth_introspection.py | 134 ++++++++++++++++ 6 files changed, 325 insertions(+), 36 deletions(-) create mode 100644 template_mcp_server/src/oauth/introspection.py create mode 100644 tests/test_oauth_introspection.py diff --git a/docs/authentication.md b/docs/authentication.md index 6501bf2..79349e7 100644 --- a/docs/authentication.md +++ b/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: diff --git a/template_mcp_server/src/oauth/handler.py b/template_mcp_server/src/oauth/handler.py index 182947a..3e4d6c8 100644 --- a/template_mcp_server/src/oauth/handler.py +++ b/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 diff --git a/template_mcp_server/src/oauth/introspection.py b/template_mcp_server/src/oauth/introspection.py new file mode 100644 index 0000000..60d02a5 --- /dev/null +++ b/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, + ) diff --git a/template_mcp_server/src/settings.py b/template_mcp_server/src/settings.py index a93a286..bc7df91 100644 --- a/template_mcp_server/src/settings.py +++ b/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={ diff --git a/tests/test_oauth_handler.py b/tests/test_oauth_handler.py index f8afa16..23f7728 100644 --- a/tests/test_oauth_handler.py +++ b/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 ): diff --git a/tests/test_oauth_introspection.py b/tests/test_oauth_introspection.py new file mode 100644 index 0000000..b3b54e6 --- /dev/null +++ b/tests/test_oauth_introspection.py @@ -0,0 +1,134 @@ +import time +from unittest.mock import Mock, patch + +import httpx + +from template_mcp_server.src.oauth.introspection import ( + Rfc7662Introspector, + TokenInfoIntrospector, + create_token_introspector, +) + + +class TestRfc7662Introspector: + @patch("template_mcp_server.src.oauth.introspection.httpx.post") + def test_introspect_success(self, mock_post): + mock_response = Mock() + mock_response.raise_for_status.return_value = None + mock_response.json.return_value = {"active": True, "sub": "user123"} + mock_post.return_value = mock_response + + introspector = Rfc7662Introspector( + url="https://sso.example.com/introspect", + client_id="client123", + client_secret="secret123", + ) + result = introspector.introspect("token123") + + mock_post.assert_called_once_with( + "https://sso.example.com/introspect", + data={ + "token": "token123", + "client_id": "client123", + "client_secret": "secret123", + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + timeout=10.0, + ) + assert result == {"active": True, "sub": "user123"} + + @patch("template_mcp_server.src.oauth.introspection.httpx.post") + def test_introspect_http_error(self, mock_post): + mock_post.side_effect = httpx.HTTPError("Connection failed") + + introspector = Rfc7662Introspector( + url="https://sso.example.com/introspect", + client_id="client123", + client_secret="secret123", + ) + result = introspector.introspect("token123") + + assert result["active"] is False + assert "Introspection failed" in result["error"] + + +class TestTokenInfoIntrospector: + @patch("template_mcp_server.src.oauth.introspection.httpx.get") + def test_introspect_success(self, mock_get): + mock_response = Mock() + mock_response.raise_for_status.return_value = None + mock_response.json.return_value = { + "sub": "123", + "email": "user@example.com", + "aud": "my-client-id", + "exp": str(int(time.time()) + 3600), + } + mock_get.return_value = mock_response + + introspector = TokenInfoIntrospector( + url="https://oauth2.googleapis.com/tokeninfo", + client_id="my-client-id", + ) + result = introspector.introspect("ya29.test") + + mock_get.assert_called_once_with( + "https://oauth2.googleapis.com/tokeninfo", + params={"access_token": "ya29.test"}, + timeout=10.0, + ) + assert result["active"] is True + assert result["sub"] == "123" + assert isinstance(result["exp"], int) + + @patch("template_mcp_server.src.oauth.introspection.httpx.get") + def test_introspect_audience_mismatch(self, mock_get): + mock_response = Mock() + mock_response.raise_for_status.return_value = None + mock_response.json.return_value = { + "sub": "123", + "email": "user@example.com", + "aud": "other-client-id", + "exp": str(int(time.time()) + 3600), + } + mock_get.return_value = mock_response + + introspector = TokenInfoIntrospector( + url="https://oauth2.googleapis.com/tokeninfo", + client_id="expected-client-id", + ) + result = introspector.introspect("ya29.test") + + assert result["active"] is False + assert result["error"] == "audience_mismatch" + + @patch("template_mcp_server.src.oauth.introspection.httpx.get") + def test_introspect_http_error(self, mock_get): + mock_get.side_effect = httpx.HTTPError("400 Bad Request") + + introspector = TokenInfoIntrospector( + url="https://oauth2.googleapis.com/tokeninfo", + ) + result = introspector.introspect("bad-token") + + assert result["active"] is False + assert "Introspection failed" in result["error"] + + +class TestCreateTokenIntrospector: + def test_create_rfc7662_introspector(self): + introspector = create_token_introspector( + "rfc7662", + "https://sso.example.com/introspect", + "client-id", + "client-secret", + ) + assert isinstance(introspector, Rfc7662Introspector) + + def test_create_tokeninfo_introspector(self): + introspector = create_token_introspector( + "tokeninfo", + "https://oauth2.googleapis.com/tokeninfo", + "client-id", + "client-secret", + ) + assert isinstance(introspector, TokenInfoIntrospector) From a643d5437e83f1c624af5cd46ae9a58e421ca8cf Mon Sep 17 00:00:00 2001 From: Abhishek Kumar Date: Sun, 7 Jun 2026 21:12:21 +0530 Subject: [PATCH 2/3] chore(oauth): document SSO_INTROSPECTION_MODE in .env.example --- .env.example | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.env.example b/.env.example index add3736..9386323 100644 --- a/.env.example +++ b/.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) --- From 06155efa8210bb00814a205235dc06171b0fba52 Mon Sep 17 00:00:00 2001 From: abhiskum Date: Sun, 7 Jun 2026 21:27:50 +0530 Subject: [PATCH 3/3] test(oauth): cover token introspection edge cases for Codecov Add tests for normalize_response branches and unexpected introspection errors so patch coverage meets the Codecov threshold. --- tests/test_oauth_introspection.py | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tests/test_oauth_introspection.py b/tests/test_oauth_introspection.py index b3b54e6..b072c1a 100644 --- a/tests/test_oauth_introspection.py +++ b/tests/test_oauth_introspection.py @@ -52,6 +52,22 @@ def test_introspect_http_error(self, mock_post): assert "Introspection failed" in result["error"] +class TestTokenInfoNormalizeResponse: + def test_passthrough_when_active_present(self): + data = {"active": False, "sub": "user123"} + assert TokenInfoIntrospector.normalize_response(data) == data + + def test_maps_error_to_inactive(self): + data = {"error": "invalid_token", "error_description": "Token expired"} + result = TokenInfoIntrospector.normalize_response(data) + assert result == {"active": False, **data} + + def test_maps_unrecognized_payload_to_inactive(self): + data = {"issued_to": "client@apps.googleusercontent.com"} + result = TokenInfoIntrospector.normalize_response(data) + assert result == {"active": False, **data} + + class TestTokenInfoIntrospector: @patch("template_mcp_server.src.oauth.introspection.httpx.get") def test_introspect_success(self, mock_get): @@ -113,6 +129,21 @@ def test_introspect_http_error(self, mock_get): assert result["active"] is False assert "Introspection failed" in result["error"] + @patch("template_mcp_server.src.oauth.introspection.httpx.get") + def test_introspect_unexpected_error(self, mock_get): + mock_response = Mock() + mock_response.raise_for_status.return_value = None + mock_response.json.side_effect = ValueError("invalid json") + mock_get.return_value = mock_response + + introspector = TokenInfoIntrospector( + url="https://oauth2.googleapis.com/tokeninfo", + ) + result = introspector.introspect("ya29.test") + + assert result["active"] is False + assert "Unexpected error" in result["error"] + class TestCreateTokenIntrospector: def test_create_rfc7662_introspector(self):