Chaim
363485d025
הסוד נוצר ב-All Infrastructure / env main / /apps/legal-ai בשם MCP_HTTP_SHARED_SECRET, מתויג credentials, לפי הדפוס של שני טוקני-הגשר שכבר שם: COURT_FETCH_SHARED_SECRET ו-LEGAL_CHAT_SHARED_SECRET. שם זהה בקוד וב-Infisical = אין מיפוי שצריך לזכור. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
107 lines
4.5 KiB
Python
107 lines
4.5 KiB
Python
"""Bearer-token gate for the HTTP transport of the MCP server (#231.2).
|
|
|
|
Why this exists
|
|
---------------
|
|
Over ``stdio`` the protection is the pipe itself: only a process that already
|
|
runs as this user can speak to the server. ``streamable-http`` removes that
|
|
property entirely — anything that can reach the socket can call any of the 108
|
|
registered tools, and the registry includes ``case_delete``,
|
|
``precedent_library_delete``, ``document_upload`` and every block-writing tool.
|
|
An unauthenticated listener is therefore a delete-any-case endpoint.
|
|
|
|
The agent platform already speaks this dialect: it injects each runtime MCP
|
|
server as ``{type:"http", name, url, headers:[{name:"Authorization",
|
|
value:"Bearer <token>"}]}``. So a static bearer token is exactly the shape the
|
|
caller will present — no negotiation, no OAuth dance.
|
|
|
|
Design notes
|
|
------------
|
|
- We implement the SDK's own ``TokenVerifier`` protocol and let
|
|
``BearerAuthBackend`` do the enforcement, rather than adding bespoke
|
|
middleware. One auth path, the framework's (G2).
|
|
- The token is read from the environment, which the service unit populates from
|
|
Infisical. It is never defaulted, never logged, and never embedded here.
|
|
- Comparison is constant-time: a naive ``==`` leaks the token byte-by-byte to a
|
|
caller who can time responses.
|
|
- ``verify_token`` returns ``None`` (not an exception) on mismatch — that is the
|
|
protocol's "reject" signal and yields a clean 401 instead of a 500 that would
|
|
read as a server fault.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
import logging
|
|
import os
|
|
|
|
from mcp.server.auth.provider import AccessToken, TokenVerifier
|
|
|
|
logger = logging.getLogger("legal_mcp.http_auth")
|
|
|
|
#: Environment variable carrying the shared bearer token.
|
|
#:
|
|
#: Name matches the Infisical key exactly — All Infrastructure / env `main` /
|
|
#: `/apps/legal-ai` / ``MCP_HTTP_SHARED_SECRET``, tagged ``credentials``. Keeping
|
|
#: the two identical means nobody has to hold a mapping in their head, and it
|
|
#: follows the two bridge tokens already in that folder
|
|
#: (``COURT_FETCH_SHARED_SECRET``, ``LEGAL_CHAT_SHARED_SECRET``).
|
|
TOKEN_ENV = "MCP_HTTP_SHARED_SECRET"
|
|
|
|
#: Minimum acceptable token length. Short tokens are brute-forceable; refusing
|
|
#: them at startup is cheaper than discovering it from an access log.
|
|
MIN_TOKEN_LEN = 32
|
|
|
|
#: Reported as the authenticated principal. Single shared token today, so this
|
|
#: is a constant rather than a real identity — kept explicit so that audit rows
|
|
#: never imply per-agent attribution we cannot actually make.
|
|
CLIENT_ID = "mcp-http-shared"
|
|
|
|
|
|
class MissingTokenError(RuntimeError):
|
|
"""Raised when the HTTP transport is requested without a usable token.
|
|
|
|
Deliberately fatal. The tempting alternative — start anyway and log a
|
|
warning — produces a listener that looks healthy and answers every
|
|
destructive tool call. Refusing to boot is the safe failure (§6: never
|
|
swallow, never degrade silently).
|
|
"""
|
|
|
|
|
|
def load_token() -> str:
|
|
"""Return the configured bearer token, or raise if it is unusable."""
|
|
token = (os.environ.get(TOKEN_ENV) or "").strip()
|
|
if not token:
|
|
raise MissingTokenError(
|
|
f"{TOKEN_ENV} is not set. The HTTP transport exposes destructive "
|
|
f"tools and will not start without a bearer token. Set it from "
|
|
f"Infisical, or use MCP_TRANSPORT=stdio.",
|
|
)
|
|
if len(token) < MIN_TOKEN_LEN:
|
|
raise MissingTokenError(
|
|
f"{TOKEN_ENV} is shorter than {MIN_TOKEN_LEN} characters — refusing "
|
|
f"to start. Generate a long random token.",
|
|
)
|
|
return token
|
|
|
|
|
|
class StaticTokenVerifier(TokenVerifier):
|
|
"""Verifies the single shared bearer token presented by the platform.
|
|
|
|
Not an identity system: it answers "may this caller in at all", not "who is
|
|
it". Per-agent attribution would need per-agent tokens, which is a later
|
|
step once profiles exist (#231.4).
|
|
"""
|
|
|
|
def __init__(self, expected: str) -> None:
|
|
self._expected = expected
|
|
|
|
async def verify_token(self, token: str) -> AccessToken | None:
|
|
# compare_digest over bytes; it tolerates unequal lengths without
|
|
# short-circuiting, which is the whole point.
|
|
if not hmac.compare_digest(token.encode("utf-8"),
|
|
self._expected.encode("utf-8")):
|
|
# No token material in the log line — only the fact of a rejection.
|
|
logger.warning("Rejected MCP HTTP request: bearer token mismatch")
|
|
return None
|
|
return AccessToken(token=token, client_id=CLIENT_ID, scopes=[])
|