split resource server and auth server

Signed-off-by: Jesse Sanford <108698+jessesanford@users.noreply.github.com>

Author: jessesanford

examples/servers/proxy_oauth/auth_server.py

@@ -0,0 +1,204 @@
+# pyright: reportMissingImports=false
+import logging
+import os
+import time
+
+from dotenv import load_dotenv  # type: ignore
+from mcp.server.auth.provider import AccessToken, OAuthToken
+from mcp.server.auth.providers.transparent_proxy import (
+    ProxySettings,  # type: ignore
+    TransparentOAuthProxyProvider,
+    ProxyTokenHandler,
+)
+from mcp.server.auth.routes import cors_middleware, create_auth_routes
+from mcp.server.auth.settings import ClientRegistrationOptions
+from pydantic import AnyHttpUrl
+from starlette.applications import Starlette
+from starlette.requests import Request  # type: ignore
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Route
+from uvicorn import Config, Server
+
+# Load environment variables from .env if present
+load_dotenv()
+
+# Configure logging after .env so LOG_LEVEL can come from environment
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
+
+logging.basicConfig(
+    level=LOG_LEVEL,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+
+# Dedicated logger for this server module
+logger = logging.getLogger("proxy_oauth.auth_server")
+
+# Suppress noisy INFO messages from the FastMCP low-level server unless we are
+# explicitly running in DEBUG mode. These logs (e.g. "Processing request of type
+# ListToolsRequest") are helpful for debugging but clutter normal output.
+
+_mcp_lowlevel_logger = logging.getLogger("mcp.server.lowlevel.server")
+if LOG_LEVEL == "DEBUG":
+    # In full debug mode, allow the library to emit its detailed logs
+    _mcp_lowlevel_logger.setLevel(logging.DEBUG)
+else:
+    # Otherwise, only warnings and above
+    _mcp_lowlevel_logger.setLevel(logging.WARNING)
+
+# ----------------------------------------------------------------------------
+# Environment configuration
+# ----------------------------------------------------------------------------
+# Load and validate settings from the environment (uses .env automatically)
+settings = ProxySettings.load()
+
+# Upstream endpoints (fully-qualified URLs)
+UPSTREAM_AUTHORIZE: str = str(settings.upstream_authorize)
+UPSTREAM_TOKEN: str = str(settings.upstream_token)
+UPSTREAM_JWKS_URI = settings.jwks_uri
+# Derive base URL from the authorize endpoint for convenience / tests
+UPSTREAM_BASE: str = UPSTREAM_AUTHORIZE.rsplit("/", 1)[0]
+
+# Client credentials & defaults
+CLIENT_ID: str = settings.client_id or "demo-client-id"
+CLIENT_SECRET = settings.client_secret
+DEFAULT_SCOPE: str = settings.default_scope
+
+# Optional audience passthrough (not part of ProxySettings yet)
+AUDIENCE = os.getenv("PROXY_AUDIENCE")
+
+# Metadata URL (only used if we need to fetch from upstream)
+UPSTREAM_METADATA = f"{UPSTREAM_BASE}/.well-known/oauth-authorization-server"
+
+# ---------------------------------------------------------------------------
+# Logging helpers
+# ---------------------------------------------------------------------------
+
+
+def _mask_secret(secret: str | None) -> str | None:  # noqa: D401
+    """Return a masked version of the given secret.
+
+    The first and last four characters are preserved (if available) and the
+    middle section is replaced by asterisks. If the secret is shorter than
+    eight characters, the entire value is replaced by ``*``.
+    """
+
+    if not secret:
+        return None
+
+    if len(secret) <= 8:
+        return "*" * len(secret)
+
+    return f"{secret[:4]}{'*' * (len(secret) - 8)}{secret[-4:]}"
+
+
+# Consolidated configuration (with sensitive data redacted)
+_masked_settings = settings.model_dump(exclude_none=True).copy()
+
+if "client_secret" in _masked_settings:
+    _masked_settings["client_secret"] = _mask_secret(_masked_settings["client_secret"])
+
+# Log configuration at *debug* level only so it can be enabled when needed
+logger.debug("[Auth Proxy Config] %s", _masked_settings)
+
+# Server host/port
+AUTH_SERVER_PORT = int(os.getenv("AUTH_SERVER_PORT", "9000"))
+AUTH_SERVER_HOST = os.getenv("AUTH_SERVER_HOST", "localhost")
+AUTH_SERVER_URL = os.getenv(
+    "AUTH_SERVER_URL", f"http://{AUTH_SERVER_HOST}:{AUTH_SERVER_PORT}"
+)
+
+# ----------------------------------------------------------------------------
+# Auth Server
+# ----------------------------------------------------------------------------
+
+# Create auth provider
+oauth_provider = TransparentOAuthProxyProvider(settings=settings)
+
+# Enable client registration
+client_registration_options = ClientRegistrationOptions(
+    enabled=True,
+    valid_scopes=["openid"],
+    default_scopes=["openid"],
+)
+
+# Create auth routes
+routes = create_auth_routes(
+    provider=oauth_provider,
+    issuer_url=AnyHttpUrl(AUTH_SERVER_URL),
+    service_documentation_url=None,
+    client_registration_options=client_registration_options,
+    revocation_options=None,
+)
+
+# Add token endpoint handler
+# We need to replace any existing token endpoint route
+routes = [r for r in routes if not (hasattr(r, "path") and r.path == "/token")]
+
+# Create token handler and add it to routes
+proxy_token_handler = ProxyTokenHandler(oauth_provider)
+routes.append(Route("/token", endpoint=proxy_token_handler.handle, methods=["POST"]))
+
+# Add token introspection endpoint for Resource Servers
+async def introspect_handler(request: Request) -> Response:
+    """
+    Token introspection endpoint for Resource Servers.
+
+    Resource Servers call this endpoint to validate tokens without
+    needing direct access to token storage.
+    """
+    form = await request.form()
+    token = form.get("token")
+    if not token or not isinstance(token, str):
+        return JSONResponse({"active": False}, status_code=400)
+
+    # For the transparent proxy, we don't actually validate tokens
+    # Just create a dummy AccessToken like the provider does
+    access_token = AccessToken(
+        token=token, client_id=str(CLIENT_ID), scopes=[DEFAULT_SCOPE], expires_at=None
+    )
+
+    return JSONResponse(
+        {
+            "active": True,
+            "client_id": access_token.client_id,
+            "scope": " ".join(access_token.scopes),
+            "exp": access_token.expires_at,
+            "iat": int(time.time()),
+            "token_type": "Bearer",
+            "aud": access_token.resource,  # RFC 8707 audience claim
+        }
+    )
+
+
+routes.append(
+    Route(
+        "/introspect",
+        endpoint=cors_middleware(introspect_handler, ["POST", "OPTIONS"]),
+        methods=["POST", "OPTIONS"],
+    )
+)
+
+# Create Starlette app with routes
+auth_app = Starlette(routes=routes)
+
+
+async def run_server():
+    """Run the Authorization Server."""
+    config = Config(
+        auth_app,
+        host=AUTH_SERVER_HOST,
+        port=AUTH_SERVER_PORT,
+        log_level="info",
+    )
+    server = Server(config)
+
+    logger.info(f"🚀 MCP Authorization Server running on {AUTH_SERVER_URL}")
+
+    await server.serve()
+
+
+if __name__ == "__main__":
+    import asyncio
+
+    asyncio.run(run_server())

examples/servers/proxy_oauth/token_verifier.py

@@ -0,0 +1,118 @@
+"""Example token verifier implementation using OAuth 2.0 Token Introspection."""
+
+import logging
+from typing import Any
+
+from mcp.server.auth.provider import AccessToken, TokenVerifier
+from mcp.shared.auth_utils import check_resource_allowed, resource_url_from_server_url
+
+logger = logging.getLogger(__name__)
+
+
+class IntrospectionTokenVerifier(TokenVerifier):
+    """Example token verifier that uses OAuth 2.0 Token Introspection (RFC 7662).
+
+    This is a simple example implementation for demonstration purposes.
+    Production implementations should consider:
+    - Connection pooling and reuse
+    - More sophisticated error handling
+    - Rate limiting and retry logic
+    - Comprehensive configuration options
+    """
+
+    def __init__(
+        self,
+        introspection_endpoint: str,
+        server_url: str,
+        validate_resource: bool = False,
+    ):
+        self.introspection_endpoint = introspection_endpoint
+        self.server_url = server_url
+        self.validate_resource = validate_resource
+        self.resource_url = resource_url_from_server_url(server_url)
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Verify token via introspection endpoint."""
+        import httpx
+
+        # Validate URL to prevent SSRF attacks
+        if not self.introspection_endpoint.startswith(
+            ("https://", "http://localhost", "http://127.0.0.1")
+        ):
+            logger.warning(
+                f"Rejecting introspection endpoint with unsafe scheme: "
+                f"{self.introspection_endpoint}"
+            )
+            return None
+
+        # Configure secure HTTP client
+        timeout = httpx.Timeout(10.0, connect=5.0)
+        limits = httpx.Limits(max_connections=10, max_keepalive_connections=5)
+
+        async with httpx.AsyncClient(
+            timeout=timeout,
+            limits=limits,
+            verify=True,  # Enforce SSL verification
+        ) as client:
+            try:
+                response = await client.post(
+                    self.introspection_endpoint,
+                    data={"token": token},
+                    headers={"Content-Type": "application/x-www-form-urlencoded"},
+                )
+
+                if response.status_code != 200:
+                    logger.debug(
+                        f"Token introspection returned status {response.status_code}"
+                    )
+                    return None
+
+                data = response.json()
+                if not data.get("active", False):
+                    return None
+
+                # RFC 8707 resource validation (only when --oauth-strict is set)
+                if self.validate_resource and not self._validate_resource(data):
+                    logger.warning(
+                        f"Token resource validation failed. Expected: "
+                        f"{self.resource_url}"
+                    )
+                    return None
+
+                return AccessToken(
+                    token=token,
+                    client_id=data.get("client_id", "unknown"),
+                    scopes=data.get("scope", "").split() if data.get("scope") else [],
+                    expires_at=data.get("exp"),
+                    resource=data.get("aud"),  # Include resource in token
+                )
+            except Exception as e:
+                logger.warning(f"Token introspection failed: {e}")
+                return None
+
+    def _validate_resource(self, token_data: dict[str, Any]) -> bool:
+        """Validate token was issued for this resource server."""
+        if not self.server_url or not self.resource_url:
+            return False  # Fail if strict validation requested but URLs missing
+
+        # Check 'aud' claim first (standard JWT audience)
+        aud = token_data.get("aud")
+        if isinstance(aud, list):
+            for audience in aud:
+                if self._is_valid_resource(audience):
+                    return True
+            return False
+        elif aud:
+            return self._is_valid_resource(aud)
+
+        # No resource binding - invalid per RFC 8707
+        return False
+
+    def _is_valid_resource(self, resource: str) -> bool:
+        """Check if resource matches this server using hierarchical matching."""
+        if not self.resource_url:
+            return False
+
+        return check_resource_allowed(
+            requested_resource=self.resource_url, configured_resource=resource
+        )