diff --git a/.env.example b/.env.example index eb159466..8445bac4 100644 --- a/.env.example +++ b/.env.example @@ -23,6 +23,16 @@ OIDC_CLOCK_SKEW_SECONDS=5 BACKEND_PORT=18420 +# Optional dedicated MCP resource server. The localhost URI is development +# only; production must publish the externally reachable HTTPS /mcp URI and +# configure the authorization server to issue tokens for that exact resource. +LINEAGEWEAVE_MCP_PORT=18421 +LINEAGEWEAVE_MCP_RESOURCE_URI=http://localhost:18421/mcp +# Browser Origins are denied unless explicitly listed. Non-browser MCP clients +# may omit Origin. Comma-separated exact origins; never use `*`. +LINEAGEWEAVE_MCP_ALLOWED_ORIGINS= +LINEAGEWEAVE_MCP_REQUESTS_PER_MINUTE=30 + # Optional. Empty = every LLM/vision channel is unavailable (Null client, # dropped and renormalized -- never a placeholder score). Point these at a # running contextual-orchestrator to turn the channels on. diff --git a/CHANGELOG.d/2.18.0-authenticated-global-ask-mcp.md b/CHANGELOG.d/2.18.0-authenticated-global-ask-mcp.md new file mode 100644 index 00000000..2e4d0bd0 --- /dev/null +++ b/CHANGELOG.d/2.18.0-authenticated-global-ask-mcp.md @@ -0,0 +1,23 @@ +# 2.18.0 — Authenticated Global Ask MCP + +## Added + +- Added a separately deployable, stateless MCP 2026-07-28 Streamable HTTP resource server with one read-only `global_ask` tool. +- Added `server/discover`, per-request MCP `_meta`, required routing headers, deterministic private/no-cache tool-list hints, `resultType=complete`, and server identity metadata. +- Added RFC 9728 protected-resource metadata, exact MCP resource audience validation, non-empty exact JWKS `kid` selection, and browser Origin allow-listing. +- Reused the existing Global Ask authorization/evidence assembler and contextual-orchestrator channel without bearer-token passthrough. +- Added structured tool output containing answer text, citations, provenance-bearing citation evidence, bounded source ids, and explicit no-evidence next action. +- Added distributed per-account invocation rate limiting through the existing Valkey service without logging question or answer text. + +## Security + +- MCP access is denied unless the OIDC subject maps to a provisioned LineageWeave account with `post_read`; per-row corporate-entity ABAC remains in force before source normalization. +- Inbound MCP bearer credentials are never forwarded to contextual-orchestrator or the browser-facing REST API. +- Unknown browser Origins fail closed; header/body protocol mismatches and unsupported revisions fail closed; JSON-RPC batching is rejected. +- Non-loopback MCP resource identifiers require HTTPS. + +## Known prerequisite blockers + +- This change is stacked on PR #264 and must not merge before the #258 → #260 → #261 → #262 → #263 → #264 prerequisite chain is ready. +- PR #258 still has an unresolved review thread in the accumulated base. +- PR #264 still has an unresolved analysis-run cutoff propagation review finding. diff --git a/backend/app/mcp_server.py b/backend/app/mcp_server.py new file mode 100644 index 00000000..fa578443 --- /dev/null +++ b/backend/app/mcp_server.py @@ -0,0 +1,662 @@ +"""Authenticated Model Context Protocol endpoint for LineageWeave Global Ask. + +The server implements the stateless MCP protocol revision 2026-07-28 over +Streamable HTTP. It exposes one read-only ``global_ask`` tool and reuses +LineageWeave's persisted authorization and evidence contracts instead of +forwarding the caller's bearer token to another service. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import time +from contextlib import asynccontextmanager +from dataclasses import dataclass +from typing import Any +from urllib.parse import urlsplit + +import asyncpg +import jwt +import redis.asyncio as redis +from fastapi import FastAPI, HTTPException, Request, Response, status +from fastapi.responses import JSONResponse +from jwt.algorithms import RSAAlgorithm + +from backend.app.activity_stream import create_valkey_client +from backend.app.auth import CurrentAccount +from backend.app.config import Settings, load_settings +from backend.app.db import create_pool +from backend.app.post_chat_ingestion import gather_global_chat_sources +from lineageweave.http_client import HttpClientError, get_json +from lineageweave.post_chat import ( + ContextualOrchestratorPostChatClient, + cited_post_evidence, + cited_post_summaries, +) + +_PROTOCOL_VERSION = "2026-07-28" +_SERVER_NAME = "lineageweave" +_SERVER_VERSION = "2.18.0" +_TOOL_NAME = "global_ask" +_LOG = logging.getLogger("lineageweave.mcp") +_JWKS_CACHE: dict[str, dict[str, Any]] = {} + +_META_PROTOCOL_VERSION = "io.modelcontextprotocol/protocolVersion" +_META_CLIENT_INFO = "io.modelcontextprotocol/clientInfo" +_META_CLIENT_CAPABILITIES = "io.modelcontextprotocol/clientCapabilities" +_META_SERVER_INFO = "io.modelcontextprotocol/serverInfo" + +_HEADER_MISMATCH = -32020 +_UNSUPPORTED_PROTOCOL_VERSION = -32022 +_SERVER_INFO = {"name": _SERVER_NAME, "version": _SERVER_VERSION} + +_GLOBAL_ASK_OUTPUT_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": { + "answer_text": {"type": "string"}, + "cited_post_ids": {"type": "array", "items": {"type": "string"}}, + "cited_posts": {"type": "array", "items": {"type": "object"}}, + "cited_post_evidence": {"type": "array", "items": {"type": "object"}}, + "source_post_ids": {"type": "array", "items": {"type": "string"}}, + "next_action": {"type": ["string", "null"]}, + }, + "required": [ + "answer_text", + "cited_post_ids", + "cited_posts", + "cited_post_evidence", + "source_post_ids", + "next_action", + ], + "additionalProperties": False, +} + +_GLOBAL_ASK_TOOL: dict[str, Any] = { + "name": _TOOL_NAME, + "title": "LineageWeave Global Ask", + "description": ( + "Ask a question over only the LineageWeave source posts and persisted " + "business evidence the authenticated account is authorized to read. " + "Returns source-grounded citations and never fabricates unavailable evidence." + ), + "inputSchema": { + "type": "object", + "properties": { + "question": { + "type": "string", + "minLength": 1, + "maxLength": 4000, + "description": "Question to answer from authorized LineageWeave evidence.", + } + }, + "required": ["question"], + "additionalProperties": False, + }, + "outputSchema": _GLOBAL_ASK_OUTPUT_SCHEMA, + "annotations": { + "readOnlyHint": True, + "destructiveHint": False, + "idempotentHint": True, + "openWorldHint": False, + }, +} + + +@dataclass(frozen=True) +class McpRuntimeSettings: + """MCP-specific settings layered on shared LineageWeave settings.""" + + resource_uri: str + allowed_origins: frozenset[str] + requests_per_minute: int + + +def load_mcp_settings() -> McpRuntimeSettings: + """Load the canonical MCP resource and bounded transport policy.""" + resource_uri = os.environ.get( + "LINEAGEWEAVE_MCP_RESOURCE_URI", "http://localhost:18421/mcp" + ).strip() + parsed = urlsplit(resource_uri) + if parsed.scheme not in {"http", "https"} or not parsed.netloc or parsed.fragment or parsed.query: + raise ValueError( + "LINEAGEWEAVE_MCP_RESOURCE_URI must be an absolute http(s) URI without query or fragment" + ) + loopback = parsed.hostname in {"localhost", "127.0.0.1", "::1"} + if parsed.scheme != "https" and not loopback: + raise ValueError("LINEAGEWEAVE_MCP_RESOURCE_URI must use HTTPS outside loopback development") + allowed_origins = frozenset( + origin.strip() + for origin in os.environ.get("LINEAGEWEAVE_MCP_ALLOWED_ORIGINS", "").split(",") + if origin.strip() + ) + try: + requests_per_minute = int(os.environ.get("LINEAGEWEAVE_MCP_REQUESTS_PER_MINUTE", "30")) + except ValueError as exc: + raise ValueError("LINEAGEWEAVE_MCP_REQUESTS_PER_MINUTE must be an integer") from exc + if not 1 <= requests_per_minute <= 600: + raise ValueError("LINEAGEWEAVE_MCP_REQUESTS_PER_MINUTE must be between 1 and 600") + return McpRuntimeSettings( + resource_uri=resource_uri, + allowed_origins=allowed_origins, + requests_per_minute=requests_per_minute, + ) + + +def _resource_metadata_url(mcp_settings: McpRuntimeSettings) -> str: + """Return the canonical RFC 9728 metadata URL for the configured resource.""" + parsed = urlsplit(mcp_settings.resource_uri) + path = parsed.path if parsed.path.startswith("/") else f"/{parsed.path}" + return f"{parsed.scheme}://{parsed.netloc}/.well-known/oauth-protected-resource{path}" + + +def _validate_transport_target(request: Request, mcp_settings: McpRuntimeSettings) -> None: + """Validate browser Origin and canonical Host to bound DNS-rebinding surface.""" + origin = request.headers.get("origin") + if origin is not None and origin not in mcp_settings.allowed_origins: + raise HTTPException(status.HTTP_403_FORBIDDEN, "MCP Origin is not allowed") + canonical_host = urlsplit(mcp_settings.resource_uri).netloc.lower() + request_host = request.headers.get("host", "").lower() + if request_host and request_host != canonical_host: + raise HTTPException(status.HTTP_400_BAD_REQUEST, "MCP Host does not match the configured resource") + + +def _mcp_jwks(settings: Settings) -> dict[str, Any]: + """Fetch and cache the configured OIDC provider's JWKS.""" + cache_key = settings.oidc_issuer + cached = _JWKS_CACHE.get(cache_key) + if cached is not None: + return cached + try: + if settings.oidc_jwks_uri_override: + jwks_uri = settings.oidc_jwks_uri_override + else: + metadata = get_json(settings.oidc_discovery_uri, timeout=10) + jwks_uri = metadata.get("jwks_uri") + if not isinstance(jwks_uri, str) or not jwks_uri.strip(): + raise ValueError("OIDC discovery document has no jwks_uri") + cached = get_json(jwks_uri, timeout=10) + except (HttpClientError, OSError, ValueError) as exc: + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "could not fetch the configured OIDC signing keys", + ) from exc + if not isinstance(cached, dict) or not isinstance(cached.get("keys"), list): + raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "OIDC JWKS has no key set") + _JWKS_CACHE[cache_key] = cached + return cached + + +def _mcp_signing_key(jwks: dict[str, Any], token: str): + """Require a non-empty JWT ``kid`` and an exact RSA signing-key match.""" + try: + header = jwt.get_unverified_header(token) + except jwt.PyJWTError as exc: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid access-token header") from exc + kid = header.get("kid") + if not isinstance(kid, str) or not kid.strip(): + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "access token must include a non-empty kid") + for key in jwks.get("keys", []): + if not isinstance(key, dict) or key.get("kid") != kid: + continue + if key.get("kty") not in (None, "RSA") or key.get("alg") not in (None, "RS256"): + continue + try: + return RSAAlgorithm.from_jwk(json.dumps(key)) + except (KeyError, TypeError, ValueError) as exc: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "matching JWKS key is invalid") from exc + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "no JWKS key matched the access-token kid") + + +def _decode_mcp_access_token( + token: str, + settings: Settings, + mcp_settings: McpRuntimeSettings, +) -> dict[str, Any]: + """Validate signature, issuer, expiry and this MCP resource audience.""" + try: + signing_key = _mcp_signing_key(_mcp_jwks(settings), token) + claims = jwt.decode( + token, + key=signing_key, + algorithms=["RS256"], + issuer=settings.oidc_issuer, + audience=mcp_settings.resource_uri, + leeway=settings.oidc_clock_skew_seconds, + ) + except HTTPException: + raise + except jwt.PyJWTError as exc: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid MCP access token") from exc + subject = claims.get("sub") + if not isinstance(subject, str) or not subject.strip(): + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "MCP access token has no subject") + return claims + + +async def _resolve_account(pool: asyncpg.Pool, subject: str) -> CurrentAccount: + """Resolve authorization from LineageWeave DB state, never token attributes.""" + async with pool.acquire() as conn: + account_row = await conn.fetchrow( + "select user_account_id, display_name, preferred_locale " + "from user_account where external_subject_id = $1", + subject, + ) + if account_row is None: + raise HTTPException( + status.HTTP_403_FORBIDDEN, + "token is valid but no LineageWeave account is provisioned for this subject", + ) + entity_rows = await conn.fetch( + "select corporate_entity_id from account_affiliation where user_account_id = $1", + account_row["user_account_id"], + ) + permission_rows = await conn.fetch( + """ + select distinct rp.permission_code + from account_role_assignment ara + join role_permission rp on rp.access_role_id = ara.access_role_id + where ara.user_account_id = $1 + """, + account_row["user_account_id"], + ) + account = CurrentAccount( + user_account_id=str(account_row["user_account_id"]), + external_subject_id=subject, + display_name=account_row["display_name"], + preferred_locale=account_row["preferred_locale"], + corporate_entity_ids=frozenset(str(row["corporate_entity_id"]) for row in entity_rows), + permission_codes=frozenset(str(row["permission_code"]) for row in permission_rows), + ) + if not account.has_permission("post_read"): + raise HTTPException(status.HTTP_403_FORBIDDEN, "account lacks the post_read permission") + return account + + +async def _authenticate(request: Request, pool: asyncpg.Pool) -> CurrentAccount: + """Authenticate one MCP HTTP request and resolve its persisted account policy.""" + authorization = request.headers.get("authorization", "") + scheme, _, token = authorization.partition(" ") + if scheme.lower() != "bearer" or not token.strip(): + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Bearer access token required") + claims = _decode_mcp_access_token(token.strip(), load_settings(), load_mcp_settings()) + return await _resolve_account(pool, str(claims["sub"])) + + +async def _check_rate_limit(client: redis.Redis, account_id: str, limit: int) -> None: + """Apply a distributed fixed-window MCP rate limit using existing Valkey.""" + bucket = int(time.time() // 60) + key = f"mcp:rate:{account_id}:{bucket}" + script = """ + local count = redis.call('INCR', KEYS[1]) + if count == 1 then redis.call('EXPIRE', KEYS[1], 120) end + return count + """ + count = int(await client.eval(script, 1, key)) + if count > limit: + raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, "MCP tool rate limit exceeded") + + +def _can_see_post(account: CurrentAccount, post: Any) -> bool: + """Match the product ABAC rule used by the browser Global Ask flow.""" + if post["visibility_code"] == "public": + return True + return str(post["corporate_entity_id"]) in account.corporate_entity_ids + + +def _post_chat_client(settings: Settings): + """Build the same contextual-orchestrator-only chat channel as the product.""" + if not (settings.orchestrator_base_url and settings.orchestrator_api_key): + return None + return ContextualOrchestratorPostChatClient( + base_url=settings.orchestrator_base_url, + api_key=settings.orchestrator_api_key, + ) + + +async def _global_ask( + pool: asyncpg.Pool, + account: CurrentAccount, + question: str, +) -> dict[str, Any]: + """Run Global Ask against only evidence visible to ``account``.""" + normalized_question = question.strip() + if not normalized_question or len(normalized_question) > 4000: + raise ValueError("question must contain between 1 and 4000 characters") + client = _post_chat_client(load_settings()) + if client is None: + raise RuntimeError("Global Ask is unavailable because contextual-orchestrator is not configured") + async with pool.acquire() as conn: + sources = await gather_global_chat_sources( + conn, + lambda row: _can_see_post(account, row), + account.corporate_entity_ids, + question=normalized_question, + ) + if not sources: + return { + "answer_text": "", + "cited_post_ids": [], + "cited_posts": [], + "cited_post_evidence": [], + "source_post_ids": [], + "next_action": "No authorized source posts are available for this question.", + } + try: + answer = await asyncio.to_thread(client.answer, normalized_question, sources) + except (HttpClientError, KeyError, OSError, ValueError) as exc: + raise RuntimeError("contextual-orchestrator returned no complete evidence object") from exc + cited_ids = list(answer.cited_post_ids) + result = { + "answer_text": answer.answer_text, + "cited_post_ids": cited_ids, + "cited_posts": cited_post_summaries(sources, cited_ids), + "cited_post_evidence": cited_post_evidence(sources, cited_ids), + "source_post_ids": [source.post_id for source in sources], + "next_action": None, + } + _LOG.info( + "mcp_global_ask account=%s sources=%d citations=%d question_chars=%d", + account.user_account_id, + len(result["source_post_ids"]), + len(cited_ids), + len(normalized_question), + ) + return result + + +def _jsonrpc_result(request_id: Any, payload: dict[str, Any]) -> dict[str, Any]: + """Build a complete 2026-era result with server identity metadata.""" + return { + "jsonrpc": "2.0", + "id": request_id, + "result": { + **payload, + "resultType": "complete", + "_meta": {_META_SERVER_INFO: _SERVER_INFO}, + }, + } + + +def _jsonrpc_error( + request_id: Any, + code: int, + message: str, + *, + data: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Build a JSON-RPC error response.""" + error: dict[str, Any] = {"code": code, "message": message} + if data is not None: + error["data"] = data + return {"jsonrpc": "2.0", "id": request_id, "error": error} + + +def _unsupported_version_error(request_id: Any, requested: Any) -> dict[str, Any]: + """Build the final 2026 UnsupportedProtocolVersion wire shape.""" + return _jsonrpc_error( + request_id, + _UNSUPPORTED_PROTOCOL_VERSION, + "Unsupported protocol version", + data={"supported": [_PROTOCOL_VERSION], "requested": str(requested or "")}, + ) + + +def _request_envelope(message: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: + """Validate the stateless per-request MCP metadata envelope.""" + params = message.get("params") + if params is None: + params = {} + if not isinstance(params, dict): + raise ValueError("params must be an object") + meta = params.get("_meta") + if not isinstance(meta, dict): + raise ValueError("params._meta is required") + protocol_version = meta.get(_META_PROTOCOL_VERSION) + if protocol_version != _PROTOCOL_VERSION: + raise RuntimeError(str(protocol_version or "")) + client_capabilities = meta.get(_META_CLIENT_CAPABILITIES) + if not isinstance(client_capabilities, dict): + raise ValueError("clientCapabilities must be an object") + client_info = meta.get(_META_CLIENT_INFO) + if client_info is not None: + if not isinstance(client_info, dict): + raise ValueError("clientInfo must be an object when present") + if not isinstance(client_info.get("name"), str) or not isinstance(client_info.get("version"), str): + raise ValueError("clientInfo name and version must be strings") + return params, meta + + +def _expected_mcp_name(method: Any, params: dict[str, Any]) -> str | None: + """Return the standardized Mcp-Name source for name-bearing methods.""" + if method == "tools/call": + name = params.get("name") + return name if isinstance(name, str) else None + if method in {"resources/read", "prompts/get"}: + source = params.get("uri") if method == "resources/read" else params.get("name") + return source if isinstance(source, str) else None + return None + + +def _validate_transport_headers(request: Request, message: dict[str, Any]) -> JSONResponse | None: + """Validate 2026 protocol/version/method/name headers against the body.""" + request_id = message.get("id") + params = message.get("params") if isinstance(message.get("params"), dict) else {} + meta = params.get("_meta") if isinstance(params, dict) else None + body_version = meta.get(_META_PROTOCOL_VERSION) if isinstance(meta, dict) else None + header_version = request.headers.get("mcp-protocol-version") + + if header_version != body_version: + return JSONResponse( + _jsonrpc_error(request_id, _HEADER_MISMATCH, "MCP-Protocol-Version header mismatch"), + status_code=status.HTTP_400_BAD_REQUEST, + ) + if header_version != _PROTOCOL_VERSION: + return JSONResponse( + _unsupported_version_error(request_id, header_version), + status_code=status.HTTP_400_BAD_REQUEST, + ) + + method = message.get("method") + method_header = request.headers.get("mcp-method") + if not isinstance(method, str) or method_header != method: + return JSONResponse( + _jsonrpc_error(request_id, _HEADER_MISMATCH, "Mcp-Method header mismatch"), + status_code=status.HTTP_400_BAD_REQUEST, + ) + expected_name = _expected_mcp_name(method, params) + name_header = request.headers.get("mcp-name") + if expected_name is None: + if name_header is not None: + return JSONResponse( + _jsonrpc_error(request_id, _HEADER_MISMATCH, "unexpected Mcp-Name header"), + status_code=status.HTTP_400_BAD_REQUEST, + ) + elif name_header != expected_name: + return JSONResponse( + _jsonrpc_error(request_id, _HEADER_MISMATCH, "Mcp-Name header mismatch"), + status_code=status.HTTP_400_BAD_REQUEST, + ) + return None + + +async def _dispatch_message( + message: Any, + pool: asyncpg.Pool, + valkey: redis.Redis, + account: CurrentAccount, + mcp_settings: McpRuntimeSettings, +) -> tuple[dict[str, Any] | None, int]: + """Dispatch one stateless, non-batched JSON-RPC 2.0 MCP message.""" + if not isinstance(message, dict) or message.get("jsonrpc") != "2.0": + request_id = message.get("id") if isinstance(message, dict) else None + return _jsonrpc_error(request_id, -32600, "Invalid Request"), status.HTTP_200_OK + method = message.get("method") + request_id = message.get("id") + if request_id is None: + return None, status.HTTP_202_ACCEPTED + try: + params, _meta = _request_envelope(message) + except RuntimeError as exc: + return _unsupported_version_error(request_id, str(exc)), status.HTTP_400_BAD_REQUEST + except ValueError as exc: + return _jsonrpc_error(request_id, -32602, str(exc)), status.HTTP_200_OK + + if method == "server/discover": + return ( + _jsonrpc_result( + request_id, + { + "supportedVersions": [_PROTOCOL_VERSION], + "capabilities": {"tools": {"listChanged": False}}, + "instructions": "Use global_ask only for source-grounded LineageWeave questions.", + "ttlMs": 0, + "cacheScope": "private", + }, + ), + status.HTTP_200_OK, + ) + if method == "ping": + return _jsonrpc_result(request_id, {}), status.HTTP_200_OK + if method == "tools/list": + return ( + _jsonrpc_result( + request_id, + {"tools": [_GLOBAL_ASK_TOOL], "ttlMs": 0, "cacheScope": "private"}, + ), + status.HTTP_200_OK, + ) + if method == "tools/call": + if params.get("name") != _TOOL_NAME: + return _jsonrpc_error(request_id, -32602, "Unknown tool"), status.HTTP_200_OK + arguments = params.get("arguments") + if not isinstance(arguments, dict) or set(arguments) != {"question"}: + return ( + _jsonrpc_error(request_id, -32602, "global_ask requires only the question argument"), + status.HTTP_200_OK, + ) + question = arguments.get("question") + if not isinstance(question, str) or not question.strip() or len(question.strip()) > 4000: + return ( + _jsonrpc_error( + request_id, + -32602, + "question must contain between 1 and 4000 characters", + ), + status.HTTP_200_OK, + ) + await _check_rate_limit(valkey, account.user_account_id, mcp_settings.requests_per_minute) + try: + structured = await _global_ask(pool, account, question) + except RuntimeError as exc: + return ( + _jsonrpc_result( + request_id, + {"content": [{"type": "text", "text": str(exc)}], "isError": True}, + ), + status.HTTP_200_OK, + ) + text = json.dumps(structured, ensure_ascii=False, separators=(",", ":")) + return ( + _jsonrpc_result( + request_id, + { + "content": [{"type": "text", "text": text}], + "structuredContent": structured, + "isError": False, + }, + ), + status.HTTP_200_OK, + ) + return _jsonrpc_error(request_id, -32601, f"Method not found: {method}"), status.HTTP_404_NOT_FOUND + + +@asynccontextmanager +async def lifespan(app: FastAPI): + """Create shared DB and Valkey clients; MCP protocol state stays per request.""" + shared = load_settings() + app.state.pool = await create_pool(shared.database_url) + app.state.valkey = create_valkey_client(shared.valkey_url) + try: + yield + finally: + await app.state.pool.close() + await app.state.valkey.aclose() + + +mcp_app = FastAPI(title="LineageWeave MCP", lifespan=lifespan) + + +@mcp_app.get("/healthz") +async def healthz() -> dict[str, str]: + """Process liveness without leaking authentication or evidence state.""" + return {"status": "ok"} + + +@mcp_app.get("/.well-known/oauth-protected-resource") +@mcp_app.get("/.well-known/oauth-protected-resource/mcp") +async def protected_resource_metadata() -> dict[str, Any]: + """Publish RFC 9728 resource metadata for MCP client discovery.""" + settings = load_settings() + mcp_settings = load_mcp_settings() + return { + "resource": mcp_settings.resource_uri, + "authorization_servers": [settings.oidc_issuer], + "bearer_methods_supported": ["header"], + } + + +@mcp_app.get("/mcp") +async def mcp_get() -> Response: + """No subscription stream is exposed in this read-only first slice.""" + return Response(status_code=status.HTTP_405_METHOD_NOT_ALLOWED, headers={"Allow": "POST"}) + + +@mcp_app.post("/mcp") +async def mcp_post(request: Request) -> Response: + """Handle one authenticated MCP 2026-07-28 Streamable HTTP request.""" + mcp_settings = load_mcp_settings() + _validate_transport_target(request, mcp_settings) + try: + message = await request.json() + except (json.JSONDecodeError, UnicodeDecodeError, ValueError): + return JSONResponse(_jsonrpc_error(None, -32700, "Parse error"), status_code=200) + if isinstance(message, list): + return JSONResponse( + _jsonrpc_error(None, -32600, "JSON-RPC batching is not supported"), + status_code=200, + ) + if not isinstance(message, dict): + return JSONResponse(_jsonrpc_error(None, -32600, "Invalid Request"), status_code=200) + header_error = _validate_transport_headers(request, message) + if header_error is not None: + return header_error + try: + account = await _authenticate(request, request.app.state.pool) + except HTTPException as exc: + if exc.status_code == status.HTTP_401_UNAUTHORIZED: + return JSONResponse( + {"detail": exc.detail}, + status_code=exc.status_code, + headers={ + "WWW-Authenticate": ( + 'Bearer resource_metadata="' + _resource_metadata_url(mcp_settings) + '"' + ) + }, + ) + raise + response_message, http_status = await _dispatch_message( + message, + request.app.state.pool, + request.app.state.valkey, + account, + mcp_settings, + ) + if response_message is None: + return Response(status_code=http_status) + return JSONResponse(response_message, status_code=http_status) + + +app = mcp_app diff --git a/docs/MCP_SERVER.md b/docs/MCP_SERVER.md new file mode 100644 index 00000000..4f9f5432 --- /dev/null +++ b/docs/MCP_SERVER.md @@ -0,0 +1,113 @@ +# LineageWeave MCP server + +LineageWeave exposes the Buyer Global Ask workflow as a separately deployable, authenticated MCP resource server. + +## Surface + +The first release exposes one read-only tool: + +```text +global_ask(question) +``` + +It uses the same persisted `post_read` permission, per-post corporate-entity ABAC predicate, bounded Global Ask source assembler, evidence, and contextual-orchestrator answer channel as the Buyer product. It does not expose SQL, unrestricted post bodies, admin operations, tickets, analysis-run writes, or arbitrary graph queries. + +## Protocol + +The server targets **MCP 2026-07-28**. The protocol core is stateless: there is no `initialize` handshake or protocol session. Clients may probe `server/discover`; every request otherwise stands alone. + +Every POST must carry: + +```text +MCP-Protocol-Version: 2026-07-28 +Mcp-Method: +``` + +`tools/call` also carries: + +```text +Mcp-Name: global_ask +``` + +The JSON body carries the same protocol version and client capabilities in `params._meta`. LineageWeave rejects header/body mismatches and unsupported protocol revisions. `tools/list` returns `ttlMs=0` and `cacheScope=private`; successful responses include `resultType=complete` and server identity in result `_meta`. + +## Run locally + +```bash +uv sync --frozen --extra backend --extra dev +uv run uvicorn backend.app.mcp_server:app --host 127.0.0.1 --port 18421 +``` + +Local development endpoints: + +```text +MCP endpoint: http://localhost:18421/mcp +Protected-resource metadata: http://localhost:18421/.well-known/oauth-protected-resource/mcp +``` + +Production must publish an HTTPS MCP resource URI and configure the authorization server to issue access tokens whose audience includes that exact resource URI. + +## Configuration + +```text +LINEAGEWEAVE_MCP_RESOURCE_URI +- canonical OAuth resource identifier +- local default: http://localhost:18421/mcp +- HTTPS required outside loopback development + +LINEAGEWEAVE_MCP_ALLOWED_ORIGINS +- comma-separated exact browser Origins permitted to reach /mcp +- empty means requests carrying Origin are rejected +- non-browser clients may omit Origin + +LINEAGEWEAVE_MCP_REQUESTS_PER_MINUTE +- distributed per-account tool-call limit stored in existing Valkey +- default 30, allowed range 1..600 +``` + +The MCP verifier additionally requires a non-empty exact RSA/RS256 JWT `kid`, matching issuer, matching MCP resource audience, normal JWT time validation, a provisioned LineageWeave account, and persisted `post_read`. Corporate affiliations and permissions are loaded from PostgreSQL; token-side business attributes cannot widen evidence access. + +## Authorization-server requirement + +The MCP process is an OAuth protected resource, not an authorization server. It publishes RFC 9728 protected-resource metadata. The configured Keyverse/OIDC authorization server must support the MCP client authorization flow and mint a resource-bound access token for `LINEAGEWEAVE_MCP_RESOURCE_URI`. + +If the authorization server has not been configured for that resource, MCP authentication must fail rather than accepting a frontend token by disabling audience verification. Authorization-client configuration should follow the current MCP 2026-07-28 issuer-validation and client-registration guidance. + +## Codex + +Configure Codex to connect to the externally reachable LineageWeave `/mcp` endpoint and complete the OAuth flow discovered through the protected-resource metadata. OpenAI's operational guidance allows MCP OAuth credentials to be stored in the operating-system keyring. Do not commit bearer tokens or place them in `AGENTS.md`. + +The exact client registration belongs in Keyverse/identity configuration; do not add a Codex-specific authentication bypass to LineageWeave. + +## Result contract + +A successful tool call returns MCP text content plus `structuredContent`: + +```json +{ + "answer_text": "...", + "cited_post_ids": ["..."], + "cited_posts": [{"post_id": "...", "post_title": "..."}], + "cited_post_evidence": [{"post_id": "..."}], + "source_post_ids": ["..."], + "next_action": null +} +``` + +When no authorized source exists, the answer is empty and `next_action` explains that no authorized evidence is available. A missing or incomplete contextual-orchestrator result is a tool execution error; LineageWeave does not manufacture an answer. + +## Security properties + +- No inbound MCP bearer token is passed to contextual-orchestrator or the Buyer REST API. +- Authorization happens before source normalization and model context assembly. +- Browser Origin validation fails closed. +- Questions are bounded to 4,000 characters. +- The tool catalog contains no write tool. +- Valkey rate limiting works across MCP replicas without introducing MCP protocol session state. +- Audit logging records opaque account id and counts only, not question text, answer text, source bodies, or credentials. + +## Stack prerequisites + +This feature is stacked on PR #264. It does not permit bypassing existing review or merge gates. The accumulated Buyer stack must first resolve the #258 review blocker and #264 analysis-run cutoff propagation finding, then revalidate exact-head checks in stack order. + +See ADR 0090 and `docs/doctoring/MCP_REFERENCES.md` for normative protocol and OAuth traceability. diff --git a/docs/adr/0090-authenticated-global-ask-mcp.md b/docs/adr/0090-authenticated-global-ask-mcp.md new file mode 100644 index 00000000..0a206f5c --- /dev/null +++ b/docs/adr/0090-authenticated-global-ask-mcp.md @@ -0,0 +1,74 @@ +# ADR 0090: Authenticated Global Ask MCP resource server + +- Status: Accepted +- Date: 2026-08-20 + +## Context + +LineageWeave already has an authenticated buyer-facing Global Ask flow (`POST /api/ask`) that assembles a bounded evidence set only after the caller's `post_read` RBAC and per-row corporate-entity ABAC checks. The answer is produced only through `contextual-orchestrator`, and citations identify the source posts used by the answer (ADR 0039). + +External agent clients such as Codex need the same capability without receiving database credentials or a privileged service token and without bypassing the existing evidence boundary. A remote MCP server creates a distinct OAuth protected-resource boundary: the access token must be intended for the MCP resource itself, browser origins must not be able to reach it through an unrelated origin, and an inbound bearer token must not be forwarded to another API. + +The accumulated Buyer stack contains independent blockers that this ADR does not hide or override: PR #258 has an unresolved static-analysis review thread, and PR #264 has an unresolved review finding that DAG navigation can discard analysis-run cutoff context. This MCP slice is stacked on #264 and must not merge ahead of its prerequisite chain. + +## Decision + +LineageWeave exposes a **separate FastAPI MCP resource server** at `backend.app.mcp_server:app` rather than adding MCP transport behavior to the browser-facing application. + +The first MCP surface contains exactly one tool: + +```text +global_ask(question) +``` + +The tool is read-only and returns grounded answer text, cited post identifiers and summaries, citation evidence, the bounded source set considered, and an explicit next action when no authorized evidence exists. It reuses the same Global Ask source assembler and `ContextualOrchestratorPostChatClient`. It does **not** forward the MCP access token to contextual-orchestrator or call the Buyer REST endpoint with that token. + +### Protocol profile + +The server targets the current MCP protocol revision `2026-07-28` over stateless Streamable HTTP. + +- There is no `initialize`/`initialized` handshake and no `Mcp-Session-Id`. +- Each request carries the protocol revision and client capabilities in `params._meta`. +- Clients may call `server/discover` to learn the supported revision and capabilities. +- Every POST requires `MCP-Protocol-Version` and `Mcp-Method`; `tools/call` additionally requires `Mcp-Name` matching `params.name`. +- Header/body mismatches fail with `HeaderMismatch` (`-32020`); unsupported revisions fail with `UnsupportedProtocolVersion` (`-32022`). +- `tools/list` returns deterministic tool definitions with explicit `ttlMs=0` and `cacheScope=private`. +- Successful 2026-era results carry `resultType=complete` and `_meta.io.modelcontextprotocol/serverInfo`. +- `global_ask` advertises `readOnlyHint=true`, `destructiveHint=false`, `idempotentHint=true`, and `openWorldHint=false`. +- JSON-RPC batching is rejected. No subscription stream or MRTR flow is needed for this first read-only tool. + +### OAuth protected-resource boundary + +The MCP process is an OAuth protected resource, not an authorization server. It publishes RFC 9728 protected-resource metadata and advertises the configured LineageWeave/Keyverse OIDC issuer. `LINEAGEWEAVE_MCP_RESOURCE_URI` is the canonical resource identifier. + +Every protected request must satisfy all of the following before evidence is touched: + +1. Bearer authentication is present. +2. JWT `kid` is non-empty and exactly selects one acceptable RSA/RS256 JWKS key; there is no first-key fallback. +3. Signature, issuer, expiry/not-before semantics, and bounded clock skew validate. +4. JWT audience includes the exact MCP resource URI. +5. `sub` resolves to a provisioned `user_account`. +6. persisted roles grant `post_read`. +7. corporate-entity affiliations are loaded from LineageWeave, not trusted from token-side business claims. +8. each source still passes the product's row-level ABAC predicate before normalization or LLM context assembly. + +A 401 challenge includes the RFC 9728 `resource_metadata` location. The bearer token is never passed through to downstream APIs. The authorization client/issuer side must also follow the 2026-07-28 authorization hardening, including issuer validation and the migration away from Dynamic Client Registration toward Client ID Metadata Documents where supported. + +### Origin, transport, bounds, and audit + +When an HTTP `Origin` header is present it must exactly match `LINEAGEWEAVE_MCP_ALLOWED_ORIGINS`; an unknown browser origin fails closed. Non-browser clients may omit `Origin`. Production resource identifiers must use HTTPS; plain HTTP is accepted only for loopback development. + +Questions are limited to 4,000 characters. Tool calls use the existing Valkey service for a distributed fixed-window per-account rate limit, preserving the stateless/horizontally scalable MCP request model. Audit logging records only opaque account id, question length, considered-source count, and citation count; question text, answer text, bearer tokens, and source bodies are not logged by the MCP audit path. + +## Consequences + +- Codex and other conforming MCP clients can authenticate to a dedicated LineageWeave resource and invoke Global Ask without database or orchestrator credentials. +- The tool cannot expand evidence visibility beyond `post_read` plus the existing ABAC contract. +- A frontend login token is not automatically an MCP token: the authorization server must mint a token for the exact MCP resource audience. +- No write tools, arbitrary SQL, unrestricted graph traversal, admin actions, ticket changes, analysis-run starts, or unrestricted post-body resources are exposed. +- The MCP request path is stateless across replicas; only authorization/evidence data in PostgreSQL and bounded rate counters in Valkey are shared. +- This ADR does not cure the Buyer application's separate audience-validation weakness or #264 temporal-cutoff navigation defect. Those remain prerequisite-stack work. + +## References + +See `docs/doctoring/MCP_REFERENCES.md` for normative protocol and OAuth references in APA 7th format. diff --git a/docs/doctoring/MCP_REFERENCES.md b/docs/doctoring/MCP_REFERENCES.md new file mode 100644 index 00000000..425be202 --- /dev/null +++ b/docs/doctoring/MCP_REFERENCES.md @@ -0,0 +1,35 @@ +# MCP standards references + +This bibliography supports ADR 0090 and the authenticated LineageWeave MCP resource server. Product claims are pinned to the current MCP revision or the referenced RFC rather than remembered protocol behavior. + +## Product decision crosswalk + +| Product decision | Source | +|---|---| +| Stateless core, no `initialize`, optional `server/discover` | MCP 2026-07-28 specification release | +| Per-request `_meta`, server identity, `resultType` | MCP 2026-07-28 SDK migration guidance | +| Required `MCP-Protocol-Version`, `Mcp-Method`, `Mcp-Name` headers | MCP 2026-07-28 Streamable HTTP specification | +| Cache hints on `tools/list` | MCP 2026-07-28 specification release | +| OAuth issuer hardening and DCR deprecation direction | MCP 2026-07-28 authorization changes; RFC 9207 | +| Protected-resource metadata and `WWW-Authenticate` discovery | RFC 9728 | +| Resource-bound access-token audience | RFC 8707 | +| Authorization-server discovery metadata | RFC 8414 | +| Codex MCP OAuth credentials may be stored in an OS keyring | OpenAI Codex operational security guidance | + +## APA 7th references + +Campbell, B., Bradley, J., & Tschofenig, H. (2020). *Resource indicators for OAuth 2.0* (RFC 8707). Internet Engineering Task Force. https://doi.org/10.17487/RFC8707 + +Jones, M. B., Hunt, P., & Parecki, A. (2025). *OAuth 2.0 protected resource metadata* (RFC 9728). Internet Engineering Task Force. https://doi.org/10.17487/RFC9728 + +Jones, M., Sakimura, N., & Bradley, J. (2018). *OAuth 2.0 authorization server metadata* (RFC 8414). Internet Engineering Task Force. https://doi.org/10.17487/RFC8414 + +Jones, M., Bradley, J., & Sakimura, N. (2022). *OAuth 2.0 authorization server issuer identification* (RFC 9207). Internet Engineering Task Force. https://doi.org/10.17487/RFC9207 + +Model Context Protocol. (2026, July 28). *The 2026-07-28 specification*. Model Context Protocol Blog. https://blog.modelcontextprotocol.io/posts/2026-07-28/ + +Model Context Protocol. (2026). *Streamable HTTP: Protocol revision 2026-07-28*. https://github.com/modelcontextprotocol/modelcontextprotocol/blob/main/docs/specification/2026-07-28/basic/transports/streamable-http.mdx + +Model Context Protocol. (2026). *Supporting protocol revision 2026-07-28*. MCP TypeScript SDK. https://ts.sdk.modelcontextprotocol.io/v2/migration/support-2026-07-28 + +OpenAI. (2026). *Running Codex safely at OpenAI*. https://openai.com/index/running-codex-safely/ diff --git a/frontend/package.json b/frontend/package.json index 7a697d0c..f216081b 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "2.17.0", + "version": "2.18.0", "type": "module", "scripts": { "dev": "vite", diff --git a/pyproject.toml b/pyproject.toml index a651b6b8..98d877ea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "2.17.0" +version = "2.18.0" description = "Reconstructs git-branch-style lineage DAGs from scattered short records using multi-channel score fusion and LLM adjudication." readme = "README.md" license = { text = "MIT" } diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py new file mode 100644 index 00000000..eb40ebc1 --- /dev/null +++ b/tests/test_mcp_server.py @@ -0,0 +1,431 @@ +from __future__ import annotations + +import asyncio +import base64 +import json + +import pytest +from starlette.requests import Request + +import backend.app.mcp_server as mcp +from backend.app.auth import CurrentAccount + + +def _segment(value: dict) -> str: + raw = json.dumps(value, separators=(",", ":")).encode() + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode() + + +def _unsigned_token(header: dict) -> str: + return f"{_segment(header)}.{_segment({'sub': 'subject'})}.signature" + + +def _account() -> CurrentAccount: + return CurrentAccount( + user_account_id="account-1", + external_subject_id="subject-1", + display_name="Demo Analyst", + preferred_locale="ko-KR", + corporate_entity_ids=frozenset({"corp-demo"}), + permission_codes=frozenset({"post_read"}), + ) + + +def _request(*, origin: str | None = None, headers: dict[str, str] | None = None) -> Request: + raw_headers: list[tuple[bytes, bytes]] = [] + if origin is not None: + raw_headers.append((b"origin", origin.encode())) + for name, value in (headers or {}).items(): + raw_headers.append((name.lower().encode(), value.encode())) + return Request( + { + "type": "http", + "http_version": "1.1", + "method": "POST", + "scheme": "https", + "path": "/mcp", + "raw_path": b"/mcp", + "query_string": b"", + "headers": raw_headers, + "client": ("127.0.0.1", 1234), + "server": ("lineage.example", 443), + } + ) + + +def _meta() -> dict: + return { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": {"name": "test-client", "version": "1.0.0"}, + "io.modelcontextprotocol/clientCapabilities": {}, + } + + +def _message(method: str, *, request_id: int = 1, **params) -> dict: + return { + "jsonrpc": "2.0", + "id": request_id, + "method": method, + "params": {**params, "_meta": _meta()}, + } + + +class FakeValkey: + def __init__(self, counts: list[int] | None = None) -> None: + self.counts = list(counts or [1]) + self.calls: list[tuple[str, int, str]] = [] + + async def eval(self, script: str, key_count: int, key: str) -> int: + self.calls.append((script, key_count, key)) + return self.counts.pop(0) + + +def test_mcp_settings_validate_resource_and_rate(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LINEAGEWEAVE_MCP_RESOURCE_URI", "https://lineage.example/mcp") + monkeypatch.setenv( + "LINEAGEWEAVE_MCP_ALLOWED_ORIGINS", "https://codex.example, https://admin.example" + ) + monkeypatch.setenv("LINEAGEWEAVE_MCP_REQUESTS_PER_MINUTE", "17") + settings = mcp.load_mcp_settings() + assert settings.resource_uri == "https://lineage.example/mcp" + assert settings.allowed_origins == frozenset({"https://codex.example", "https://admin.example"}) + assert settings.requests_per_minute == 17 + + monkeypatch.setenv("LINEAGEWEAVE_MCP_RESOURCE_URI", "file:///etc/passwd") + with pytest.raises(ValueError, match="absolute http"): + mcp.load_mcp_settings() + + monkeypatch.setenv("LINEAGEWEAVE_MCP_RESOURCE_URI", "http://lineage.example/mcp") + with pytest.raises(ValueError, match="HTTPS"): + mcp.load_mcp_settings() + + monkeypatch.setenv("LINEAGEWEAVE_MCP_RESOURCE_URI", "http://localhost:18421/mcp") + assert mcp.load_mcp_settings().resource_uri == "http://localhost:18421/mcp" + + monkeypatch.setenv("LINEAGEWEAVE_MCP_RESOURCE_URI", "https://lineage.example/mcp") + monkeypatch.setenv("LINEAGEWEAVE_MCP_REQUESTS_PER_MINUTE", "0") + with pytest.raises(ValueError, match="between 1 and 600"): + mcp.load_mcp_settings() + + monkeypatch.setenv("LINEAGEWEAVE_MCP_REQUESTS_PER_MINUTE", "not-a-number") + with pytest.raises(ValueError, match="must be an integer"): + mcp.load_mcp_settings() + + +def test_transport_target_checks_origin_and_host() -> None: + settings = mcp.McpRuntimeSettings( + resource_uri="https://lineage.example/mcp", + allowed_origins=frozenset({"https://codex.example"}), + requests_per_minute=30, + ) + mcp._validate_transport_target(_request(), settings) + mcp._validate_transport_target( + _request(origin="https://codex.example", headers={"Host": "lineage.example"}), settings + ) + with pytest.raises(Exception) as origin_error: + mcp._validate_transport_target(_request(origin="https://evil.example"), settings) + assert getattr(origin_error.value, "status_code", None) == 403 + with pytest.raises(Exception) as host_error: + mcp._validate_transport_target(_request(headers={"Host": "evil.example"}), settings) + assert getattr(host_error.value, "status_code", None) == 400 + + +def test_resource_metadata_url_uses_canonical_resource_not_request_host() -> None: + settings = mcp.McpRuntimeSettings( + resource_uri="https://lineage.example/public/mcp", + allowed_origins=frozenset(), + requests_per_minute=30, + ) + assert ( + mcp._resource_metadata_url(settings) + == "https://lineage.example/.well-known/oauth-protected-resource/public/mcp" + ) + + +def test_signing_key_requires_nonempty_exact_kid(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(mcp.RSAAlgorithm, "from_jwk", lambda value: ("key", value)) + jwks = { + "keys": [ + {"kid": "first", "kty": "RSA", "alg": "RS256", "n": "x", "e": "AQAB"}, + {"kid": "wanted", "kty": "RSA", "alg": "RS256", "n": "y", "e": "AQAB"}, + ] + } + key = mcp._mcp_signing_key(jwks, _unsigned_token({"alg": "RS256", "kid": "wanted"})) + assert key[0] == "key" + assert '"kid": "wanted"' in key[1] + + with pytest.raises(Exception) as missing: + mcp._mcp_signing_key(jwks, _unsigned_token({"alg": "RS256"})) + assert getattr(missing.value, "status_code", None) == 401 + + with pytest.raises(Exception) as unknown: + mcp._mcp_signing_key(jwks, _unsigned_token({"alg": "RS256", "kid": "unknown"})) + assert getattr(unknown.value, "status_code", None) == 401 + + +def test_access_token_decode_binds_audience(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict[str, object] = {} + monkeypatch.setattr(mcp, "_mcp_jwks", lambda settings: {"keys": []}) + monkeypatch.setattr(mcp, "_mcp_signing_key", lambda jwks, token: "signing-key") + + def fake_decode(token, **kwargs): + captured.update(kwargs) + return {"sub": "subject-1"} + + monkeypatch.setattr(mcp.jwt, "decode", fake_decode) + shared = type( + "SettingsStub", + (), + {"oidc_issuer": "https://id.example", "oidc_clock_skew_seconds": 5}, + )() + runtime = mcp.McpRuntimeSettings( + resource_uri="https://lineage.example/mcp", + allowed_origins=frozenset(), + requests_per_minute=30, + ) + claims = mcp._decode_mcp_access_token("token", shared, runtime) + assert claims["sub"] == "subject-1" + assert captured["issuer"] == "https://id.example" + assert captured["audience"] == "https://lineage.example/mcp" + assert captured["algorithms"] == ["RS256"] + + +def test_access_token_requires_subject(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(mcp, "_mcp_jwks", lambda settings: {"keys": []}) + monkeypatch.setattr(mcp, "_mcp_signing_key", lambda jwks, token: "signing-key") + monkeypatch.setattr(mcp.jwt, "decode", lambda *args, **kwargs: {}) + shared = type( + "SettingsStub", + (), + {"oidc_issuer": "https://id.example", "oidc_clock_skew_seconds": 5}, + )() + runtime = mcp.McpRuntimeSettings( + resource_uri="https://lineage.example/mcp", + allowed_origins=frozenset(), + requests_per_minute=30, + ) + with pytest.raises(Exception) as error: + mcp._decode_mcp_access_token("token", shared, runtime) + assert getattr(error.value, "status_code", None) == 401 + + +def test_request_envelope_requires_current_protocol_and_capabilities() -> None: + params, meta = mcp._request_envelope(_message("tools/list")) + assert params["_meta"] is meta + assert meta["io.modelcontextprotocol/protocolVersion"] == "2026-07-28" + + missing = _message("tools/list") + del missing["params"]["_meta"]["io.modelcontextprotocol/clientCapabilities"] + with pytest.raises(ValueError, match="clientCapabilities"): + mcp._request_envelope(missing) + + old = _message("tools/list") + old["params"]["_meta"]["io.modelcontextprotocol/protocolVersion"] = "2025-06-18" + with pytest.raises(RuntimeError): + mcp._request_envelope(old) + + +def test_transport_headers_follow_final_error_precedence() -> None: + message = _message("tools/call", name="global_ask", arguments={"question": "Q"}) + valid = _request( + headers={ + "MCP-Protocol-Version": "2026-07-28", + "Mcp-Method": "tools/call", + "Mcp-Name": "global_ask", + } + ) + assert mcp._validate_transport_headers(valid, message) is None + + mismatch = _request( + headers={ + "MCP-Protocol-Version": "2025-06-18", + "Mcp-Method": "tools/call", + "Mcp-Name": "global_ask", + } + ) + response = mcp._validate_transport_headers(mismatch, message) + assert response.status_code == 400 + assert b'"code":-32020' in response.body + + old_message = _message("tools/call", name="global_ask", arguments={"question": "Q"}) + old_message["params"]["_meta"]["io.modelcontextprotocol/protocolVersion"] = "2025-06-18" + old_header = _request( + headers={ + "MCP-Protocol-Version": "2025-06-18", + "Mcp-Method": "tools/call", + "Mcp-Name": "global_ask", + } + ) + response = mcp._validate_transport_headers(old_header, old_message) + assert response.status_code == 400 + assert b'"code":-32022' in response.body + assert b'"supported":["2026-07-28"]' in response.body + assert b'"requested":"2025-06-18"' in response.body + + +def test_discover_and_tool_catalog_are_stateless_current_protocol() -> None: + runtime = mcp.McpRuntimeSettings( + resource_uri="https://lineage.example/mcp", + allowed_origins=frozenset(), + requests_per_minute=30, + ) + discover, discover_status = asyncio.run( + mcp._dispatch_message( + _message("server/discover"), object(), FakeValkey(), _account(), runtime + ) + ) + assert discover_status == 200 + assert discover["result"]["supportedVersions"] == ["2026-07-28"] + assert discover["result"]["resultType"] == "complete" + assert discover["result"]["ttlMs"] == 0 + assert discover["result"]["cacheScope"] == "private" + assert discover["result"]["_meta"]["io.modelcontextprotocol/serverInfo"]["name"] == "lineageweave" + + tools, tools_status = asyncio.run( + mcp._dispatch_message(_message("tools/list"), object(), FakeValkey(), _account(), runtime) + ) + assert tools_status == 200 + tool = tools["result"]["tools"][0] + assert tool["name"] == "global_ask" + assert tool["annotations"]["readOnlyHint"] is True + assert tool["annotations"]["openWorldHint"] is False + assert tool["outputSchema"] == mcp._GLOBAL_ASK_OUTPUT_SCHEMA + assert tools["result"]["ttlMs"] == 0 + assert tools["result"]["cacheScope"] == "private" + + +def test_global_ask_tool_returns_structured_and_text_content(monkeypatch: pytest.MonkeyPatch) -> None: + runtime = mcp.McpRuntimeSettings( + resource_uri="https://lineage.example/mcp", + allowed_origins=frozenset(), + requests_per_minute=30, + ) + expected = { + "answer_text": "Grounded answer", + "cited_post_ids": ["post-1"], + "cited_posts": [{"post_id": "post-1", "post_title": "Evidence"}], + "cited_post_evidence": [{"post_id": "post-1", "facts": ["source record"]}], + "source_post_ids": ["post-1"], + "next_action": None, + } + + async def fake_ask(pool, account, question): + assert account.user_account_id == "account-1" + assert question == "What changed?" + return expected + + async def no_limit(client, account_id, limit): + assert account_id == "account-1" + assert limit == 30 + + monkeypatch.setattr(mcp, "_global_ask", fake_ask) + monkeypatch.setattr(mcp, "_check_rate_limit", no_limit) + response, http_status = asyncio.run( + mcp._dispatch_message( + _message( + "tools/call", + request_id=7, + name="global_ask", + arguments={"question": "What changed?"}, + ), + object(), + FakeValkey(), + _account(), + runtime, + ) + ) + assert http_status == 200 + result = response["result"] + assert result["structuredContent"] == expected + assert json.loads(result["content"][0]["text"]) == expected + assert result["isError"] is False + assert result["resultType"] == "complete" + + +def test_tool_validation_execution_failure_and_unknown_method_are_distinct( + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime = mcp.McpRuntimeSettings( + resource_uri="https://lineage.example/mcp", + allowed_origins=frozenset(), + requests_per_minute=30, + ) + unknown_tool, unknown_status = asyncio.run( + mcp._dispatch_message( + _message("tools/call", name="write_everything", arguments={}), + object(), + FakeValkey(), + _account(), + runtime, + ) + ) + assert unknown_status == 200 + assert unknown_tool["error"]["code"] == -32602 + + async def no_limit(client, account_id, limit): + return None + + async def unavailable(pool, account, question): + raise RuntimeError("orchestrator unavailable") + + monkeypatch.setattr(mcp, "_check_rate_limit", no_limit) + monkeypatch.setattr(mcp, "_global_ask", unavailable) + failed, failed_status = asyncio.run( + mcp._dispatch_message( + _message( + "tools/call", + name="global_ask", + arguments={"question": "Question"}, + ), + object(), + FakeValkey(), + _account(), + runtime, + ) + ) + assert failed_status == 200 + assert "error" not in failed + assert failed["result"]["isError"] is True + assert failed["result"]["content"][0]["text"] == "orchestrator unavailable" + + missing, missing_status = asyncio.run( + mcp._dispatch_message( + _message("resources/list"), object(), FakeValkey(), _account(), runtime + ) + ) + assert missing_status == 404 + assert missing["error"]["code"] == -32601 + + +def test_rate_limit_uses_valkey_and_rejects_over_limit(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(mcp.time, "time", lambda: 600.0) + allowed = FakeValkey([2]) + asyncio.run(mcp._check_rate_limit(allowed, "account", 2)) + assert allowed.calls[0][1] == 1 + assert allowed.calls[0][2] == "mcp:rate:account:10" + + denied = FakeValkey([3]) + with pytest.raises(Exception) as error: + asyncio.run(mcp._check_rate_limit(denied, "account", 2)) + assert getattr(error.value, "status_code", None) == 429 + + +def test_invalid_jsonrpc_and_notification_behavior() -> None: + runtime = mcp.McpRuntimeSettings( + resource_uri="https://lineage.example/mcp", + allowed_origins=frozenset(), + requests_per_minute=30, + ) + invalid, invalid_status = asyncio.run( + mcp._dispatch_message({"id": 1}, object(), FakeValkey(), _account(), runtime) + ) + assert invalid_status == 200 + assert invalid["error"]["code"] == -32600 + + notification = _message("notifications/custom") + notification.pop("id") + response, response_status = asyncio.run( + mcp._dispatch_message(notification, object(), FakeValkey(), _account(), runtime) + ) + assert response is None + assert response_status == 202