"""ASGI middleware that authenticates MCP HTTP requests with a PAT. Sits in front of the FastMCP Streamable-HTTP app. Every request that isn't a CORS preflight is required to carry ``Authorization: Bearer helix_pat_<...>``; on success, the middleware populates the contextvars in :mod:`app.mcp.context` and the audit-aware contextvars in :mod:`app.auth.auth_context`, then forwards to the inner app. On any auth failure it short-circuits with a uniform 401 JSON envelope — matching the rest of the Helix API so clients can parse errors with the same code path. The PAT lookup uses an independent DB session so the auth path never touches a session that would later be reused by the tool handler. This keeps transaction lifetime tightly scoped: auth opens, reads, commits the ``last_used_*`` bookkeeping, closes. Defenses added in the launch-readiness audit (2026-05-05): * **Body-size cap** — reject requests whose ``Content-Length`` exceeds ``MCP_REQUEST_BODY_MAX_BYTES``. Without this a single PAT could OOM the uvicorn worker by streaming a multi-megabyte JSON-RPC payload. * **Per-PAT rate limit** — Redis fixed-window keyed by PAT id. Caps the audit-write amplification a runaway / compromised client can generate (every tool call writes ≥ 1 audit row). * **Per-IP failed-auth rate limit** — keyed by source IP, fired before the DB lookup. Stops a brute-force PAT-prefix scanner from churning through PAT hashes and from filling the auth-failure audit table. * **Auth-failure audit** — every rejected MCP request now writes a ``mcp.auth_failure`` audit row (rate-limited per-IP to prevent log flooding) so admins can detect token-abuse scans. * **Contextvar reset** — token-style ``var.set(...).reset(token)`` in a ``try/finally`` so contextvars never leak past the request task. """ from __future__ import annotations import asyncio import json import logging import re from uuid import UUID, uuid4 from app.auth.auth_context import ( current_actor_user_id, current_auth_method, current_pat_token_id, current_request_trace_id, ) from app.auth.service import is_effectively_departed from app.auth.tokens.service import PAT_PREFIX, PersonalAccessTokenService # ``async_session_factory`` is referenced via the module so test fixtures # can monkeypatch ``app.core.database.async_session_factory`` and have # the patch take effect on every subsequent middleware call. A direct # ``from app.core.database import async_session_factory`` would bind the # original at import time and silently bypass the patch — which would # defeat the MCP test at ``tests/test_auth/test_mcp_endpoint.py``. from app.core import database as _db from app.core.config import settings from app.mcp.context import ( mcp_client_ip, mcp_trace_id, mcp_user_agent, mcp_user_id, ) logger = logging.getLogger(__name__) def _real_ip_from_scope(scope) -> str: """Best-effort client IP extraction from an ASGI scope. Mirrors :func:`app.core.dependencies.get_real_ip`: only trusts X-Real-IP / X-Forwarded-For when the direct TCP peer is in TRUSTED_PROXY_IPS. Without this check any client reaching the MCP endpoint directly can spoof the per-IP auth-failure rate limiter and poison the audit log ``client_ip`` field. """ from app.core.net import is_trusted_proxy client = scope.get("client") peer = str(client[0]) if (client and isinstance(client, (tuple, list)) and len(client) >= 1) else "unknown" if not is_trusted_proxy(peer): return peer headers = {k.decode("latin1").lower(): v.decode("latin1") for k, v in scope.get("headers", [])} real_ip = headers.get("x-real-ip") if real_ip: return real_ip.strip() fwd = headers.get("x-forwarded-for") if fwd: # Pre-launch P1 fix (Round 45, CR-105 follow-up): pick the # RIGHTMOST element — the one nginx appended (= real peer IP # it observed). Leftmost is the client-supplied header value # which is freely spoofable, and would let MCP clients bypass # per-IP auth-failure rate limits by varying X-Forwarded-For # per attempt. parts = [p.strip() for p in fwd.split(",") if p.strip()] if parts: return parts[-1] return peer def _user_agent_from_scope(scope) -> str | None: for k, v in scope.get("headers", []): if k == b"user-agent": return v.decode("latin1", errors="replace") return None def _content_length_from_scope(scope) -> int | None: for k, v in scope.get("headers", []): if k == b"content-length": try: return int(v.decode("latin1")) except (ValueError, TypeError): return None return None async def _send_json( send, status: int, body: dict, *, www_authenticate: str | None = None, ) -> None: payload = json.dumps(body).encode("utf-8") headers = [ (b"content-type", b"application/json"), (b"content-length", str(len(payload)).encode("ascii")), ] if status == 401: challenge = www_authenticate or 'Bearer realm="helix-mcp"' headers.append((b"www-authenticate", challenge.encode("latin1"))) await send({ "type": "http.response.start", "status": status, "headers": headers, }) await send({"type": "http.response.body", "body": payload}) def _resource_metadata_url(scope) -> str: """RFC 9728 protected-resource-metadata URL for this MCP endpoint.""" from app.auth.oauth.metadata import origin_from_scope origin = origin_from_scope(scope) return f"{origin}/.well-known/oauth-protected-resource/api/v1/mcp" def _bearer_challenge(scope) -> str: """``WWW-Authenticate`` value advertising where to discover the OAuth authorization server. This is what lets a Claude / Cursor connector bootstrap the OAuth flow from a single unauthenticated probe.""" return ( 'Bearer realm="helix-mcp", ' f'resource_metadata="{_resource_metadata_url(scope)}"' ) def _trace_meta() -> dict: """Envelope ``meta`` carrying the request's real trace id. The outer ``TraceIdMiddleware`` stamps ``current_request_trace_id`` (and the ``X-Trace-ID`` response header) for every request, including ones routed into this mounted MCP app. Error envelopes previously hardcoded ``trace_id: None`` — so a user reporting an MCP failure quoted an X-Trace-ID that matched nothing in the body. (Launch audit 2026-06-10.) """ return {"trace_id": current_request_trace_id.get()} def _unauthorized(send, message: str, *, www_authenticate: str | None = None): return _send_json(send, 401, { "success": False, "data": None, "meta": _trace_meta(), "errors": [{"code": "UNAUTHORIZED", "message": message}], }, www_authenticate=www_authenticate) def _too_large(send) -> None: return _send_json(send, 413, { "success": False, "data": None, "meta": _trace_meta(), "errors": [{ "code": "PAYLOAD_TOO_LARGE", "message": ( f"Request body exceeds the " f"{settings.MCP_REQUEST_BODY_MAX_BYTES}-byte cap." ), }], }) _CLI_UA_RE = re.compile(r"helix-cli/(\d+)\.(\d+)\.(\d+)") def _release_tuple(version: str) -> tuple[int, int, int] | None: """Parse the ``X.Y.Z`` release part of a version, ignoring any PEP 440 local/pre segment (``0.1.0+g1a2b3c4`` → ``(0,1,0)``).""" m = re.match(r"\s*(\d+)\.(\d+)\.(\d+)", version or "") if not m: return None return (int(m.group(1)), int(m.group(2)), int(m.group(3))) def _cli_version_below_min(user_agent: str | None) -> bool: """True iff the request is from a ``helix-cli`` whose release version is strictly below ``CLI_MIN_SUPPORTED_VERSION``. The min-version kill-switch: a fix ships, ops bumps the floor, and every CLI older than the fix is refused at the MCP edge — they cannot keep hitting the API with a known-buggy build. Non-CLI clients (Claude Desktop, Cursor, scripts with other UAs) are never affected. ``0.0.0`` floor (default) disables enforcement entirely. """ floor = _release_tuple(settings.CLI_MIN_SUPPORTED_VERSION) if not floor or floor == (0, 0, 0): return False if not user_agent: return False m = _CLI_UA_RE.search(user_agent) if not m: return False # not a helix-cli UA → don't gate cli_ver = (int(m.group(1)), int(m.group(2)), int(m.group(3))) return cli_ver < floor def _upgrade_required(send) -> None: return _send_json(send, 426, { "success": False, "data": None, "meta": _trace_meta(), "errors": [{ "code": "CLI_UPGRADE_REQUIRED", "message": ( f"This helix CLI is older than the minimum supported " f"version ({settings.CLI_MIN_SUPPORTED_VERSION}) and has " f"been blocked because old builds may carry known bugs or " f"security issues. Download the latest from the Helix web " f"UI (Help → AI 助手集成 → helix CLI) or run " f"`helix version` to check." ), }], }) def _too_many_requests(send, retry_after: int) -> None: payload = json.dumps({ "success": False, "data": None, "meta": _trace_meta(), "errors": [{ "code": "RATE_LIMITED", "message": ( f"Too many MCP requests. Try again in " f"{retry_after}s." ), }], }).encode("utf-8") return _send_429(send, payload, retry_after) async def _send_429(send, payload: bytes, retry_after: int) -> None: await send({ "type": "http.response.start", "status": 429, "headers": [ (b"content-type", b"application/json"), (b"content-length", str(len(payload)).encode("ascii")), (b"retry-after", str(retry_after).encode("ascii")), ], }) await send({"type": "http.response.body", "body": payload}) async def _check_pat_rate_limit(pat_id) -> tuple[bool, int]: """Per-PAT fixed-window rate limit. Returns ``(allowed, retry_after)``. Fail-open on Redis errors (with a small async delay) so a Redis outage doesn't take the MCP surface down. The `apply_sensitive_rate_limit` helper in `app.core.dependencies` uses the same pattern. """ from app.core.redis_client import get_redis key = f"mcp_ratelimit:pat:{pat_id}" window = settings.MCP_RATE_LIMIT_PER_PAT_WINDOW_SECONDS cap = settings.MCP_RATE_LIMIT_PER_PAT_MAX try: redis = get_redis() # Fixed-window counter set atomically in one MULTI/EXEC: # SET key 0 EX window NX — create the window with its TTL only # when the key is first seen (no-op if # the window already exists), # INCR key — bump the request count. # The TTL is therefore anchored to the FIRST request of the window # and never extended by later requests, so the counter actually # decays after ``window`` seconds (a plain unconditional EXPIRE on # every call slides the window forward indefinitely and keeps a # PAT blocked far past the intended window under steady traffic). # Both commands run in the same atomic transaction, so there is no # crash gap that could leave the key without a TTL (which would # block the PAT until Redis restart). async with redis.pipeline(transaction=True) as pipe: await pipe.set(key, 0, ex=window, nx=True) await pipe.incr(key) results = await pipe.execute() # ``INCR`` returns int on default redis-py settings, but a downstream # config flip (``decode_responses=True``) or a connection pool # returning bytes would otherwise crash the ``count > cap`` compare # with TypeError — caught by the catch-all below and silently # turning the rate limiter into fail-open. Explicit cast keeps the # limiter honest. (R1 code review finding.) count = int(results[1]) if count > cap: ttl = await redis.ttl(key) if ttl < 0: ttl = window return False, int(ttl) return True, 0 except Exception as exc: # noqa: BLE001 logger.warning( "MCP per-PAT rate limiter Redis error (fail-open with delay): %s", type(exc).__name__, ) await asyncio.sleep(0.05) return True, 0 async def _check_auth_failure_rate_limit(client_ip: str) -> tuple[bool, int]: """Per-IP failed-auth rate limit fired BEFORE PAT hash lookup. Stops brute-force scanners from saturating the auth-DB session and from filling the audit table with unbounded ``mcp.auth_failure`` rows. Failure-audit writes themselves are gated by this check. Returns ``(allowed, retry_after_seconds)``. Fails open on Redis outage with a small delay (no DoS-by-redis-down). """ from app.core.redis_client import get_redis key = f"mcp_authfail:{client_ip}" window = settings.MCP_AUTH_FAILURE_LIMIT_PER_IP_WINDOW_SECONDS cap = settings.MCP_AUTH_FAILURE_LIMIT_PER_IP_MAX try: redis = get_redis() # Atomic fixed-window counter: same reasoning as # _check_pat_rate_limit — SET ... NX anchors the TTL to the first # failure of the window so the window can't slide forward forever. async with redis.pipeline(transaction=True) as pipe: await pipe.set(key, 0, ex=window, nx=True) await pipe.incr(key) results = await pipe.execute() # See _check_pat_rate_limit: defensive cast against decode_responses # / bytes-returning pools so the limiter never silently fails open. count = int(results[1]) if count > cap: ttl = await redis.ttl(key) if ttl < 0: ttl = window return False, int(ttl) return True, 0 except Exception as exc: # noqa: BLE001 logger.warning( "MCP auth-failure rate limiter Redis error (fail-open with " "delay): %s", type(exc).__name__, ) await asyncio.sleep(0.05) return True, 0 def _token_failure_key(token: str) -> str: """Redis key for the per-TOKEN failed-auth window. Keyed by a SHA-256 prefix of the bearer so the raw credential never lands in Redis keyspace. 32 hex chars (128 bits) is far beyond collision concern for a rate-limit bucket. """ import hashlib digest = hashlib.sha256(token.encode("utf-8")).hexdigest()[:32] return f"mcp_authfail:tok:{digest}" async def _check_token_failure_rate_limit(token: str) -> None: """Increment the per-TOKEN failed-auth window (fire-and-forget). Companion to the pre-DB peek below: each REJECTED well-formed bearer bumps its own token-keyed counter so a stuck client retrying one dead PAT gets short-circuited before the DB on its next attempts. Result is deliberately ignored — the request that trips the cap still gets its accurate 401; only subsequent attempts see 429. Reuses the per-IP window/cap settings. """ from app.core.redis_client import get_redis key = _token_failure_key(token) window = settings.MCP_AUTH_FAILURE_LIMIT_PER_IP_WINDOW_SECONDS try: redis = get_redis() async with redis.pipeline(transaction=True) as pipe: await pipe.set(key, 0, ex=window, nx=True) await pipe.incr(key) await pipe.execute() except Exception as exc: # noqa: BLE001 logger.warning( "MCP per-token failure limiter Redis error (skipped): %s", type(exc).__name__, ) async def _peek_token_failure_rate_limit(token: str) -> tuple[bool, int]: """READ-ONLY check of the per-TOKEN failed-auth window, fired BEFORE the PAT hash lookup for well-formed ``helix_pat_*`` bearers. Why per-token and not per-IP: the increment-side per-IP limiter only ran AFTER the DB round-trip for well-formed tokens — so the stuck-offboarded-client shape (valid PAT, inactive owner: full ``authenticate`` + ``last_used_*`` UPDATE + role load per retry) did one uncapped DB round-trip per attempt forever, 429 or not. A per-IP peek would close that but let one colleague's dead-PAT retry loop 429 every VALID user behind the same office NAT — an availability regression worse than the DB load it prevents (post-fix security review, 2026-06-10). Keying the peek by token hash means: valid tokens never accumulate failures and are never blocked; one dead token gets short-circuited pre-DB after the cap; rotating-garbage scanners still pay one cheap indexed SELECT per attempt but remain capped by the per-IP audit-flood gate. Fail-open on Redis errors, same as the other limiters. """ from app.core.redis_client import get_redis key = _token_failure_key(token) cap = settings.MCP_AUTH_FAILURE_LIMIT_PER_IP_MAX try: redis = get_redis() raw = await redis.get(key) if raw is None: return True, 0 if int(raw) > cap: ttl = await redis.ttl(key) if ttl < 0: ttl = settings.MCP_AUTH_FAILURE_LIMIT_PER_IP_WINDOW_SECONDS return False, int(ttl) return True, 0 except Exception as exc: # noqa: BLE001 logger.warning( "MCP auth-failure rate limiter peek Redis error (fail-open " "with delay): %s", type(exc).__name__, ) await asyncio.sleep(0.05) return True, 0 async def _audit_auth_failure( *, reason: str, client_ip: str, user_agent: str | None, ) -> None: """Best-effort audit row for an auth-rejected MCP request. Why: the `last_used_*` bookkeeping on the PAT row only updates on SUCCESS, so without this a brute-force PAT-prefix scan leaves no trail. We log + swallow on any error so the auth-fail response itself never blocks on audit-DB issues. Per-IP rate limiter upstream stops a scanner from filling the audit table. """ try: from app.audit.service import AuditService async with _db.async_session_factory() as audit_db: await AuditService(audit_db).log( trace_id=str(uuid4()), actor_type="anonymous", action="mcp.auth_failure", status="failure", resource_type="mcp_endpoint", resource_id="auth", detail={"reason": reason[:200]}, ip_address=client_ip, user_agent=user_agent, ) await audit_db.commit() except Exception: # noqa: BLE001 logger.exception("MCP auth-failure audit write failed") class PatAuthASGIMiddleware: """Plain ASGI middleware (no Starlette base class) so it can sit in front of FastMCP's Streamable HTTP app without depending on a specific framework version's middleware stack. """ def __init__(self, app) -> None: self.app = app async def __call__(self, scope, receive, send): if scope.get("type") != "http": # WebSocket / lifespan events don't carry auth — pass through. return await self.app(scope, receive, send) method = scope.get("method", "").upper() # CORS preflight requests must not require auth — browsers send # them without an Authorization header by spec. The MCP server # itself is expected to handle CORS for actual GET/POST/DELETE. if method == "OPTIONS": return await self.app(scope, receive, send) # ── Body-size cap (cheap header check, fired before any DB work) ── # Reject malicious / runaway payloads before they buffer in # memory. We trust ``Content-Length`` for the pre-DB short- # circuit; further down we wrap ``receive`` to enforce the same # cap on streamed / chunked-encoded bodies (Round-17 audit). body_len = _content_length_from_scope(scope) if ( body_len is not None and body_len > settings.MCP_REQUEST_BODY_MAX_BYTES ): return await _too_large(send) client_ip = _real_ip_from_scope(scope) user_agent = _user_agent_from_scope(scope) # WWW-Authenticate challenge pointing connectors at OAuth discovery. # Attached to every 401 so an unauthenticated probe can bootstrap # the OAuth flow (RFC 9728 §5.1). challenge = _bearer_challenge(scope) # ── CLI min-version kill-switch ── # Refuse a helix-cli older than the configured floor BEFORE auth # or any DB work: a known-buggy / vulnerable old build must not # keep operating just because its PAT is still valid. Cheap # in-memory check; no-op when the floor is 0.0.0 (default) or the # client isn't a helix-cli. if _cli_version_below_min(user_agent): return await _upgrade_required(send) # Extract bearer token auth_header: str | None = None for k, v in scope.get("headers", []): if k == b"authorization": auth_header = v.decode("latin1", errors="replace") break if not auth_header or not auth_header.lower().startswith("bearer "): allowed, retry_after = await _check_auth_failure_rate_limit(client_ip) if not allowed: return await _too_many_requests(send, retry_after) await _audit_auth_failure( reason="missing_or_malformed_authorization", client_ip=client_ip, user_agent=user_agent, ) return await _unauthorized( send, "Missing or malformed Authorization header.", www_authenticate=challenge, ) token = auth_header.split(" ", 1)[1].strip() # Two credential kinds reach the MCP endpoint: # * ``helix_pat_…`` — a personal access token (manual issue). # * an MCP OAuth 2.1 access token (JWT) minted by Helix's # co-hosted authorization server for a Claude / Cursor # connector. Both converge on the same ``user_id`` + fresh # role load; only the credential-validation step differs. is_pat = token.startswith(PAT_PREFIX) # Saturated per-TOKEN failure window → reject BEFORE any DB # work. The increment happens on actual failures below; this # read-only peek just enforces the already-tripped limit so a # stuck client retrying one dead credential can't keep buying a # PAT-hash SELECT (plus, on the valid-PAT-inactive-owner # path, a ``last_used_*`` UPDATE and a full role load) per # attempt forever. Valid tokens never accumulate failures, so # legitimate users — including everyone behind the same # office NAT — are never blocked by this gate. allowed, retry_after = await _peek_token_failure_rate_limit(token) if not allowed: return await _too_many_requests(send, retry_after) token_id = None # PAT row id when is_pat; None for OAuth. auth_method = "pat" if is_pat else "oauth" if not is_pat: # ── OAuth access-token (JWT) path ── # Decode is pure-CPU (no DB) so reject malformed/expired/ # wrong-audience tokens before touching the database. The # audience check (RFC 8707) is the OAuth boundary: a token # minted for a different resource is rejected here. from app.auth.oauth.metadata import ( acceptable_audiences, origin_from_scope, ) from app.core.security import decode_mcp_access_token origin = origin_from_scope(scope) try: payload = decode_mcp_access_token( token, audience=acceptable_audiences(origin), ) except Exception: # noqa: BLE001 — any decode/aud failure → 401 await _check_token_failure_rate_limit(token) allowed, retry_after = await _check_auth_failure_rate_limit(client_ip) if not allowed: return await _too_many_requests(send, retry_after) await _audit_auth_failure( reason="invalid_oauth_token", client_ip=client_ip, user_agent=user_agent, ) return await _unauthorized( send, "Invalid or expired token.", www_authenticate=challenge, ) try: async with _db.async_session_factory() as db: from datetime import UTC, datetime from app.core.dependencies import ( _enforce_force_logout, _load_user_with_roles, ) from app.core.exceptions import UnauthorizedError try: sub_uuid = UUID(str(payload.get("sub"))) except (ValueError, TypeError): sub_uuid = None user = ( await _load_user_with_roles(db, sub_uuid) if sub_uuid else None ) if not user or is_effectively_departed(user): await _audit_auth_failure( reason="oauth_token_owner_inactive", client_ip=client_ip, user_agent=user_agent, ) return await _unauthorized( send, "Token owner is inactive.", www_authenticate=challenge, ) # Force-logout parity: a token issued before the # owner's force_logout cutoff is dead. The OAuth # access token's ``iat`` is its issued-at. try: iat = datetime.fromtimestamp( int(payload["iat"]), tz=UTC, ) _enforce_force_logout(user, iat) except UnauthorizedError: await _audit_auth_failure( reason="oauth_force_logout_active", client_ip=client_ip, user_agent=user_agent, ) return await _unauthorized( send, "Session was terminated.", www_authenticate=challenge, ) # Grant gate: a user-revoked grant or admin-disabled # client is immediate logout — checked per request so # revoke takes effect now, not after token expiry. from app.auth.oauth import service as _oauth_service azp = payload.get("azp") or "" if not await _oauth_service.is_grant_active( db, user_id=user.id, client_id=azp, ): await _audit_auth_failure( reason="oauth_grant_revoked", client_ip=client_ip, user_agent=user_agent, ) return await _unauthorized( send, "Connection was revoked.", www_authenticate=challenge, ) user_id = user.id rate_id = payload.get("jti") or str(user_id) except Exception as exc: # noqa: BLE001 logger.exception( "MCP OAuth auth pipeline failed: %s", type(exc).__name__, ) return await _send_json(send, 500, { "success": False, "data": None, "meta": {"trace_id": None}, "errors": [{"code": "INTERNAL_ERROR", "message": "MCP authentication failed."}], }) else: try: async with _db.async_session_factory() as db: pat_service = PersonalAccessTokenService(db) pat_row = await pat_service.authenticate( token, client_ip=client_ip, user_agent=user_agent, ) if pat_row is None: await _check_token_failure_rate_limit(token) allowed, retry_after = await _check_auth_failure_rate_limit(client_ip) if not allowed: return await _too_many_requests(send, retry_after) await _audit_auth_failure( reason="invalid_or_expired_pat", client_ip=client_ip, user_agent=user_agent, ) return await _unauthorized( send, "Invalid or expired token.", www_authenticate=challenge, ) # Verify the underlying user is still active — a token # whose owner has been offboarded must stop working # even if the token row itself is unexpired. from app.core.dependencies import ( _enforce_force_logout, _load_user_with_roles, ) from app.core.exceptions import UnauthorizedError user = await _load_user_with_roles(db, pat_row.user_id) if not user or is_effectively_departed(user): await _check_token_failure_rate_limit(token) allowed, retry_after = await _check_auth_failure_rate_limit(client_ip) if not allowed: return await _too_many_requests(send, retry_after) await _audit_auth_failure( reason="token_owner_inactive", client_ip=client_ip, user_agent=user_agent, ) return await _unauthorized( send, "Token owner is inactive.", www_authenticate=challenge, ) # Force-logout parity with REST PAT auth (see # ``_enforce_force_logout`` at # ``app/core/dependencies.py:295``). When an admin # force-terminates a session (HR alarm, security # incident, suspected compromise), every credential # minted BEFORE the force_logout timestamp must stop # working — on MCP too, not just REST. Without this, # a stolen PAT survives a force-logout on the same # user, defeating the documented invariant that # "force_logout severs every credential surface". try: _enforce_force_logout(user, pat_row.created_at) except UnauthorizedError: await _check_token_failure_rate_limit(token) allowed, retry_after = await _check_auth_failure_rate_limit(client_ip) if not allowed: return await _too_many_requests(send, retry_after) await _audit_auth_failure( reason="force_logout_active", client_ip=client_ip, user_agent=user_agent, ) return await _unauthorized( send, "Session was terminated.", www_authenticate=challenge, ) user_id = user.id token_id = pat_row.id rate_id = token_id except Exception as exc: # noqa: BLE001 # Log the exception type only — never the full string. A # future PAT validator that includes the bearer in its error # message would otherwise leak the token to the application # log. The full traceback is still attached via ``exc_info`` # because ``.exception()`` captures it from the active frame. logger.exception( "MCP auth pipeline failed: %s", type(exc).__name__, ) return await _send_json(send, 500, { "success": False, "data": None, "meta": {"trace_id": None}, "errors": [{"code": "INTERNAL_ERROR", "message": "MCP authentication failed."}], }) # ── Per-credential rate limit (success path) ── # Fired AFTER successful auth so a valid credential, not a # scanner, is what gets metered. Caps the audit-write # amplification (every tool call writes ≥ 1 audit row) under # runaway clients. Keyed by PAT id (PAT path) or access-token # jti / user id (OAuth path). allowed, retry_after = await _check_pat_rate_limit(rate_id) if not allowed: return await _too_many_requests(send, retry_after) # Stash everything the tool handlers need. Capture each token so # we can ``reset`` in finally — without this, contextvars leak # past end-of-task in any future ASGI deployment that reuses # tasks across requests (uvicorn's HTTP path doesn't today, but # defense-in-depth: silent identity bleed between users is the # worst possible failure mode). # # Seed the MCP trace id from the request-level trace the outer # ``TraceIdMiddleware`` stamped (it coerces to UUID shape and # echoes it as the ``X-Trace-ID`` response header). A fresh # ``uuid4()`` here meant every ``mcp.*`` audit row carried a # trace nothing else knew — a user quoting the X-Trace-ID from # their failed call matched zero audit rows. Fallback to # uuid4() when the contextvar is unset (e.g. direct ASGI test # harness without the outer middleware). (Launch audit # 2026-06-10.) from uuid import UUID as _UUID _seed = current_request_trace_id.get() try: trace_id = _UUID(_seed) if _seed else uuid4() except (TypeError, ValueError): trace_id = uuid4() tok_user = mcp_user_id.set(user_id) tok_trace = mcp_trace_id.set(trace_id) tok_ip = mcp_client_ip.set(client_ip) tok_ua = mcp_user_agent.set(user_agent) tok_method = current_auth_method.set(auth_method) tok_pat = current_pat_token_id.set(token_id) # Mirror REST auth + trace middleware: stamp the *generic* actor # and trace contextvars too so cross-domain listeners (e.g. the # leaver-blocked ``before_flush`` trap at # ``server/app/domains/leaver_blocked/listener.py``) can attribute # unguarded writes that originate from an MCP tool call. Without # this, an MCP-initiated unguarded flip lands in the trap audit # with ``attempted_by_user_id=None`` and a synthetic ``uuid4()`` # trace_id that doesn't correlate to anything — exactly the # attribution failure mode the round-29 contextvar pair was # introduced to close. (R1 code review finding, 2026-05-28.) tok_actor = current_actor_user_id.set(user_id) tok_req_trace = current_request_trace_id.set(str(trace_id)) # Wrap ``receive`` so chunked / streaming bodies are subject to # the same hard cap as ``Content-Length``-headed requests. # Without this, an authenticated PAT holder could stream # multi-MB JSON-RPC into FastMCP and starve the worker on # memory + parse time. Track cumulative bytes across all # ``http.request`` events; when the cap is exceeded, signal # ``http.disconnect`` so the downstream handler aborts cleanly # instead of seeing a partial body. (Round-17 audit.) max_bytes = settings.MCP_REQUEST_BODY_MAX_BYTES counter = {"sent": 0, "killed": False} async def _capped_receive(): if counter["killed"]: return {"type": "http.disconnect"} event = await receive() if event.get("type") != "http.request": return event body = event.get("body") or b"" counter["sent"] += len(body) if counter["sent"] > max_bytes: counter["killed"] = True logger.warning( "MCP body-cap exceeded on streaming request " "(%d > %d bytes) — disconnecting", counter["sent"], max_bytes, ) return {"type": "http.disconnect"} return event # When the streaming cap kills the body, the downstream # FastMCP handler sees ``http.disconnect`` mid-read and its # blanket exception handler answers with a 500 — misleading # (clients treat 5xx as server fault and retry) and # inconsistent with the clean 413 the Content-Length path # returns. Rewrite the status on the way out; only a response # that STARTS after the kill flag is set is rewritten, so a # response already in flight is never corrupted. (Launch # audit 2026-06-10.) async def _capped_send(message): if ( counter["killed"] and message.get("type") == "http.response.start" and int(message.get("status", 0)) >= 500 ): message = {**message, "status": 413} await send(message) try: return await self.app(scope, _capped_receive, _capped_send) finally: mcp_user_id.reset(tok_user) mcp_trace_id.reset(tok_trace) mcp_client_ip.reset(tok_ip) mcp_user_agent.reset(tok_ua) current_auth_method.reset(tok_method) current_pat_token_id.reset(tok_pat) current_actor_user_id.reset(tok_actor) current_request_trace_id.reset(tok_req_trace) __all__ = ["PatAuthASGIMiddleware"]