mcp-server/src/legal_mcp/server.py

@@ -52,11 +52,38 @@ async def lifespan(server: FastMCP) -> AsyncIterator[None]:
 # ignored and the server binds 8000 regardless (verified 2026-08-05; on this
 # host 8000 is already taken, so it failed loudly by luck rather than design).
 #
-# Default to loopback: this transport is unauthenticated until #231.2 lands and
-# the registry contains destructive tools.
+# Default to loopback. The bearer gate below is the real protection; the narrow
+# bind is defence in depth, not a substitute for it.
 MCP_HTTP_HOST = os.environ.get("MCP_HTTP_HOST", "127.0.0.1")
 MCP_HTTP_PORT = int(os.environ.get("MCP_HTTP_PORT", "8790"))
 
+# Bearer gate — wired only when an HTTP transport is actually selected (#231.2).
+#
+# stdio must never require a token: the pipe is the boundary there, and every
+# interactive session reaches us that way. Demanding a token on stdio would
+# break all of them for no security gain.
+#
+# Building the verifier at import time (rather than inside main()) is deliberate:
+# FastMCP takes `token_verifier` and `auth` as constructor arguments, so a
+# missing token has to fail here — before the listener exists — not after it is
+# already accepting connections.
+_http_transport = os.environ.get("MCP_TRANSPORT", "stdio").strip() in ("sse", "streamable-http")
+_auth_kwargs: dict = {}
+if _http_transport:
+    from mcp.server.auth.settings import AuthSettings
+
+    from legal_mcp.services.http_auth import StaticTokenVerifier, load_token
+
+    _base_url = f"http://{MCP_HTTP_HOST}:{MCP_HTTP_PORT}"
+    _auth_kwargs = {
+        "token_verifier": StaticTokenVerifier(load_token()),
+        # AuthSettings is what switches on the SDK's BearerAuthBackend. We are a
+        # resource server with a pre-shared token, not an OAuth client, so both
+        # URLs simply point at ourselves — they exist to satisfy the protected-
+        # resource metadata contract, and nothing issues tokens from them.
+        "auth": AuthSettings(issuer_url=_base_url, resource_server_url=_base_url),
+    }
+
 # Create MCP server
 mcp = FastMCP(
     "Ezer Mishpati - עוזר משפטי",
@@ -64,6 +91,7 @@ mcp = FastMCP(
     lifespan=lifespan,
     host=MCP_HTTP_HOST,
     port=MCP_HTTP_PORT,
+    **_auth_kwargs,
 )
 
 # ── Import and register tools ───────────────────────────────────────

mcp-server/src/legal_mcp/services/http_auth.py

@@ -0,0 +1,101 @@
+"""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. Populated from
+#: Infisical by the service unit — see docs/spec/X10-deploy-env-secrets.md.
+TOKEN_ENV = "MCP_HTTP_TOKEN"
+
+#: 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=[])