fix: harden mcp auth and user lookup (#111)

* fix: harden mcp auth and user lookup

* fix: restore mcp verifier fallback

Author: mplemay

examples/mcp/README.md

@@ -1,7 +1,8 @@
 # Belgie MCP + OAuth Example
 
 This example hosts the Belgie OAuth authorization server and an MCP resource server on a **single FastAPI app**.
-The MCP server validates bearer tokens by introspecting against the co-located OAuth server.
+The MCP server validates bearer tokens against the co-located OAuth provider state when available, and falls back to
+HTTP introspection for external OAuth deployments.
 
 ## Setup
 
@@ -34,8 +35,12 @@ The app runs at `http://localhost:8000`.
 ## Notes
 
 - The MCP server is mounted at `/mcp` and accepts both `/mcp` and `/mcp/`.
+- Streamable HTTP transport security is derived from the configured MCP resource URL, so production hosts no longer
+  need a custom `host="localhost"` override.
 - OAuth discovery serving (`/.well-known/oauth-authorization-server*` and
   `/.well-known/oauth-protected-resource*`) is owned by `OAuthServerPlugin`.
 - Configure `OAuthServer.resources=[OAuthResource(prefix="/mcp", ...)]` so protected
   resource metadata is published at the RFC9728 well-known endpoint.
+- `SimpleOAuthProvider` keeps clients and tokens in memory, so deploys and restarts invalidate previously issued
+  tokens.
 - The example uses SQLite and will create `./belgie_mcp_example.db` in the working directory.

examples/mcp/main.py

@@ -207,7 +207,7 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
 
 app.include_router(belgie.router)
 
-_ = mcp_plugin.mount_streamable_http(app, mcp_server, host="localhost")
+_ = mcp_plugin.mount_streamable_http(app, mcp_server)
 
 
 @mcp_server.tool()

packages/belgie-mcp/src/belgie_mcp/__tests__/test_plugin.py

@@ -2,17 +2,24 @@
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from types import SimpleNamespace
+from urllib.parse import parse_qs, urlparse
+from uuid import uuid4
 
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
+from pydantic import AnyUrl
 
 pytest.importorskip("mcp")
 
 from belgie_core.core.settings import BelgieSettings
 from belgie_mcp.plugin import Mcp, McpPlugin
 from belgie_mcp.verifier import BelgieOAuthTokenVerifier
+from belgie_oauth_server.models import OAuthClientMetadata
+from belgie_oauth_server.plugin import OAuthServerPlugin
+from belgie_oauth_server.provider import AccessToken as OAuthAccessToken, AuthorizationParams, SimpleOAuthProvider
 from belgie_oauth_server.settings import OAuthServer
+from mcp.server.auth.provider import AccessToken
 from mcp.server.mcpserver import MCPServer
 
 
@@ -103,6 +110,43 @@ def test_mcp_plugin_defaults_base_url_from_belgie_settings() -> None:
     assert str(plugin.auth.resource_server_url) == "https://example.com/mcp"
 
 
+@pytest.mark.asyncio
+async def test_mcp_plugin_verifier_uses_linked_oauth_plugin_provider() -> None:
+    settings = OAuthServer(
+        base_url="https://auth.local",
+        redirect_uris=["http://localhost/callback"],
+        client_id="client",
+        client_secret="secret",
+        default_scope="user",
+    )
+    provider = SimpleOAuthProvider(settings, issuer_url=str(settings.issuer_url))
+    oauth_plugin = OAuthServerPlugin(_belgie_settings(), settings)
+    oauth_plugin._provider = provider
+    plugin = McpPlugin(
+        _belgie_settings(),
+        Mcp(
+            oauth=settings,
+            server_url="https://mcp.local/mcp",
+        ),
+    )
+    _ = plugin.router(SimpleNamespace(plugins=[oauth_plugin, plugin]))
+    token_value, stored_token = await _issue_dynamic_client_access_token(
+        provider,
+        user_id=str(uuid4()),
+        resource="https://mcp.local/mcp",
+    )
+
+    token = await plugin.token_verifier.verify_token(token_value)
+
+    assert token == AccessToken(
+        token=token_value,
+        client_id=stored_token.client_id,
+        scopes=["user"],
+        expires_at=stored_token.expires_at,
+        resource="https://mcp.local/mcp",
+    )
+
+
 def test_mount_streamable_http_accepts_alias_path_without_redirect() -> None:
     settings = OAuthServer(
         base_url="https://auth.local",
@@ -121,7 +165,7 @@ def test_mount_streamable_http_accepts_alias_path_without_redirect() -> None:
     server = MCPServer(name="Belgie MCP")
     app = _build_test_app(plugin, server)
 
-    with TestClient(app, base_url="http://localhost:8000") as client:
+    with TestClient(app, base_url="https://example.com") as client:
         alias_response = client.post(
             "/mcp",
             headers={"Content-Type": "application/json"},
@@ -142,6 +186,81 @@ def test_mount_streamable_http_accepts_alias_path_without_redirect() -> None:
     assert alias_response.json() == mounted_response.json()
 
 
+@pytest.mark.parametrize("host_header", ["localhost:8000", "127.0.0.1:8000", "[::1]:8000"])
+def test_mount_streamable_http_allows_loopback_hosts(host_header: str) -> None:
+    plugin = _build_plugin(server_url="http://localhost:8000/mcp")
+    server = MCPServer(
+        name="Belgie MCP",
+        auth=plugin.auth,
+        token_verifier=_AllowingTokenVerifier(),
+    )
+    app = _build_test_app(plugin, server)
+
+    with TestClient(app, base_url="http://localhost:8000") as client:
+        response = client.post(
+            "/mcp",
+            headers={
+                "Authorization": "Bearer valid-token",
+                "Host": host_header,
+            },
+            json=_build_initialize_request(),
+            follow_redirects=False,
+        )
+
+    assert response.status_code == 200
+    assert response.headers["content-type"].startswith("text/event-stream")
+    assert response.headers.get("mcp-session-id")
+
+
+def test_mount_streamable_http_allows_configured_external_host() -> None:
+    plugin = _build_plugin(server_url="https://example.com/mcp")
+    server = MCPServer(
+        name="Belgie MCP",
+        auth=plugin.auth,
+        token_verifier=_AllowingTokenVerifier(),
+    )
+    app = _build_test_app(plugin, server)
+
+    with TestClient(app, base_url="https://example.com") as client:
+        response = client.post(
+            "/mcp",
+            headers={
+                "Authorization": "Bearer valid-token",
+                "Host": "example.com",
+            },
+            json=_build_initialize_request(),
+            follow_redirects=False,
+        )
+
+    assert response.status_code == 200
+    assert response.headers["content-type"].startswith("text/event-stream")
+    assert response.headers.get("mcp-session-id")
+
+
+def test_mount_streamable_http_rejects_mismatched_host() -> None:
+    plugin = _build_plugin(server_url="http://localhost:8000/mcp")
+    server = MCPServer(
+        name="Belgie MCP",
+        auth=plugin.auth,
+        token_verifier=_AllowingTokenVerifier(),
+    )
+    app = _build_test_app(plugin, server)
+
+    with TestClient(app, base_url="http://localhost:8000") as client:
+        response = client.post(
+            "/mcp",
+            headers={
+                "Authorization": "Bearer valid-token",
+                "Host": "example.com",
+            },
+            json=_build_initialize_request(),
+            follow_redirects=False,
+        )
+
+    assert response.status_code == 421
+    assert response.text == "Invalid Host header"
+
+
 def test_mount_streamable_http_preserves_auth_middleware() -> None:
     settings = OAuthServer(
         base_url="https://auth.local",
@@ -165,7 +284,7 @@ def test_mount_streamable_http_preserves_auth_middleware() -> None:
     )
     app = _build_test_app(plugin, server)
 
-    with TestClient(app, base_url="http://localhost:8000") as client:
+    with TestClient(app, base_url="https://example.com") as client:
         alias_response = client.post(
             "/mcp",
             headers={"Authorization": "Bearer alias-token"},
@@ -192,7 +311,7 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
             yield
 
     app = FastAPI(lifespan=lifespan)
-    _ = plugin.mount_streamable_http(app, server, host="localhost")
+    _ = plugin.mount_streamable_http(app, server)
     return app
 
 
@@ -202,3 +321,81 @@ class _StubTokenVerifier:
 
     async def verify_token(self, token: str) -> None:
         self.tokens.append(token)
+
+
+@dataclass(slots=True)
+class _AllowingTokenVerifier:
+    async def verify_token(self, token: str) -> AccessToken:
+        return AccessToken(
+            token=token,
+            client_id="client",
+            scopes=["user"],
+        )
+
+
+def _build_plugin(*, server_url: str) -> McpPlugin:
+    settings = OAuthServer(
+        base_url="https://auth.local",
+        redirect_uris=["http://localhost/callback"],
+        client_id="client",
+        client_secret="secret",
+        default_scope="user",
+    )
+    return McpPlugin(
+        _belgie_settings(),
+        Mcp(
+            oauth=settings,
+            server_url=server_url,
+        ),
+    )
+
+
+def _build_initialize_request() -> dict[str, object]:
+    return {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "initialize",
+        "params": {
+            "protocolVersion": "2025-03-26",
+            "capabilities": {},
+            "clientInfo": {"name": "test-client", "version": "1"},
+        },
+    }
+
+
+async def _issue_dynamic_client_access_token(
+    provider: SimpleOAuthProvider,
+    *,
+    user_id: str | None = None,
+    resource: str | None = None,
+) -> tuple[str, OAuthAccessToken]:
+    client = await provider.register_client(
+        OAuthClientMetadata(
+            redirect_uris=[AnyUrl("http://localhost:6274/oauth/callback")],
+            grant_types=["authorization_code", "refresh_token"],
+            response_types=["code"],
+            scope="user",
+            token_endpoint_auth_method="none",
+        ),
+    )
+    state = await provider.authorize(
+        client,
+        AuthorizationParams(
+            state=None,
+            scopes=["user"],
+            code_challenge="test-challenge",
+            redirect_uri=AnyUrl("http://localhost:6274/oauth/callback"),
+            redirect_uri_provided_explicitly=True,
+            resource=resource,
+            user_id=user_id,
+            session_id=str(uuid4()),
+        ),
+    )
+    redirect = await provider.issue_authorization_code(state)
+    code = parse_qs(urlparse(redirect).query)["code"][0]
+    authorization_code = await provider.load_authorization_code(code)
+    assert authorization_code is not None
+    token_response = await provider.exchange_authorization_code(authorization_code)
+    stored_token = await provider.load_access_token(token_response.access_token)
+    assert stored_token is not None
+    return token_response.access_token, stored_token