From 48842806c7c8721347836832342306487d93ce32 Mon Sep 17 00:00:00 2001 From: Erfouni Date: Mon, 21 Sep 2026 10:22:15 +0330 Subject: [PATCH 1/3] Test the assembled server end to end, not just the gate test_secret_path_gate drives SecretPathAuth against a stub, so nothing checked that the real app still speaks MCP once FastMCP's transport sits behind the gate. An SDK upgrade that changed how the streamable-HTTP transport answers, or that began rejecting the tunnel's Host header, would only have shown up as a broken connector in ChatGPT. These build the app exactly as main() does and drive it with Starlette's TestClient (no uvicorn, no tunnel, no AutoCAD): initialize and tools/list through a tunnel-style Host header, the stateless contract (no session needed, no session id issued), JSON rather than SSE responses, the health route living under the secret path, and the bearer check sitting in front of the transport. Each assertion was checked against a deliberately broken server.py: leaving DNS-rebinding protection on, going stateful, switching to SSE, dropping the gate, and moving the health route each fail the suite. Co-Authored-By: Claude Opus 5 (1M context) --- tests/test_assembled_server.py | 133 +++++++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) create mode 100644 tests/test_assembled_server.py diff --git a/tests/test_assembled_server.py b/tests/test_assembled_server.py new file mode 100644 index 0000000..9f74c2b --- /dev/null +++ b/tests/test_assembled_server.py @@ -0,0 +1,133 @@ +"""The assembled server, end to end: FastMCP's transport behind the gate. + +test_secret_path_gate drives SecretPathAuth against a stub. Nothing checked +that the real app still answers MCP once the pieces are put together, so an +MCP SDK upgrade that changed how the streamable-HTTP transport responds -- or +started rejecting the tunnel's Host header -- would only show up as a broken +connector in ChatGPT. These build the app exactly as main() does and talk to +it through Starlette's TestClient: no uvicorn, no tunnel, and no AutoCAD, +because initialising and listing tools never touches COM. + + python -m unittest discover -s tests -t . +""" + +import os +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +# Must be in place before importing config: without it, the import writes a +# .secret file into the source tree. +os.environ.setdefault("ACAD_MCP_SECRET_PATH", "unit-test-secret") + +from starlette.testclient import TestClient # noqa: E402 + +import config # noqa: E402 +import server # noqa: E402 + +SECRET = "unit-test-secret" +MCP = f"/{SECRET}/mcp" + +# What a request looks like once ngrok has forwarded it: the Host header is the +# tunnel's public name, not 127.0.0.1. +TUNNEL = {"Host": "example.ngrok-free.app"} +MCP_HEADERS = { + **TUNNEL, + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", +} + + +def rpc(method, params=None, id_=1): + message = {"jsonrpc": "2.0", "id": id_, "method": method} + if params is not None: + message["params"] = params + return message + + +INITIALIZE = rpc("initialize", { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "unit-test", "version": "0"}, +}) + + +class AssembledServer(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls._tmp = tempfile.TemporaryDirectory() + tmp = Path(cls._tmp.name) + # build_app() creates the output and log folders; keep them out of the + # user's profile and the source tree. + cls._patches = [ + mock.patch.object(config, "SECRET_PATH", SECRET), + mock.patch.object(config, "MCP_PATH", MCP), + mock.patch.object(config, "AUTH_TOKEN", ""), + mock.patch.object(config, "SAVE_DIR", tmp / "out"), + mock.patch.object(config, "LOG_DIR", tmp / "logs"), + ] + for patch in cls._patches: + patch.start() + _mcp, cls.registrar, app = server.build_app() + # Entering the client runs the app's lifespan, which starts the + # transport's session manager -- what uvicorn does in main(). + cls.client = TestClient(app, base_url="http://127.0.0.1:8770") + cls.client.__enter__() + + @classmethod + def tearDownClass(cls): + cls.client.__exit__(None, None, None) + for patch in reversed(cls._patches): + patch.stop() + cls._tmp.cleanup() + + def post(self, message, headers=MCP_HEADERS): + return self.client.post(MCP, json=message, headers=headers) + + def test_initialize_through_the_tunnel_host(self): + response = self.post(INITIALIZE) + self.assertEqual(response.status_code, 200, response.text) + self.assertTrue( + response.headers["content-type"].startswith("application/json"), + response.headers["content-type"], + ) + result = response.json()["result"] + self.assertEqual(result["serverInfo"]["name"], "AutoCAD") + self.assertIn("draw_batch", result["instructions"]) + + def test_tools_list_needs_no_session(self): + """Stateless: a client that reconnects need not initialise again.""" + response = self.post(rpc("tools/list", id_=2)) + self.assertEqual(response.status_code, 200, response.text) + self.assertNotIn("mcp-session-id", response.headers) + names = [tool["name"] for tool in response.json()["result"]["tools"]] + self.assertEqual(len(names), self.registrar.count) + self.assertIn("draw_batch", names) + self.assertIn("capture_view", names) + + def test_health_route_sits_behind_the_secret_path(self): + response = self.client.get(f"/{SECRET}/health", headers=TUNNEL) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["tools"], self.registrar.count) + self.assertEqual(response.json()["mcp_endpoint"], MCP) + + def test_transport_is_not_reachable_without_the_secret(self): + for path in ("/mcp", "/", "/health"): + with self.subTest(path=path): + response = self.client.post(path, json=INITIALIZE, headers=MCP_HEADERS) + self.assertEqual(response.status_code, 404) + + def test_bearer_token_is_checked_in_front_of_the_transport(self): + with mock.patch.object(config, "AUTH_TOKEN", "s3cret"): + refused = self.post( + INITIALIZE, {**MCP_HEADERS, "Authorization": "Bearer wrong"}) + accepted = self.post( + INITIALIZE, {**MCP_HEADERS, "Authorization": "Bearer s3cret"}) + self.assertEqual(refused.status_code, 401) + self.assertEqual(accepted.status_code, 200, accepted.text) + + +if __name__ == "__main__": + unittest.main() From 22de42a9a53f8b2a3132786e76132ff5d300d056 Mon Sep 17 00:00:00 2001 From: Erfouni Date: Mon, 21 Sep 2026 10:23:50 +0330 Subject: [PATCH 2/3] Move to mcp 1.28.1, clearing three published advisories mcp 1.27.1 is listed as affected by GHSA-jpw9-pfvf-9f58 (HTTP transports did not bind sessions to the authenticated principal), GHSA-hvrp-rf83-w775 (experimental task handlers not scoped to their session) and GHSA-vj7q-gjh5-988w (no Host/Origin validation on the WebSocket transport). 1.28.1 is the first release that fixes all three. By each advisory's own "am I affected" test, none of them reaches this server: it runs streamable HTTP in stateless mode, never calls enable_tasks(), and does not use the WebSocket transport. The point is that anyone installing from requirements.txt, or running pip-audit over it, is not handed a version with open CVEs, and that real findings are not buried under these. The 1.x changes in between are the two fixes above, deprecation warnings for WebSocket and experimental tasks (neither used here), Python 3.14 support, and buffered per-request streamable-HTTP streams. The new end-to-end tests pass on both 1.27.1 and 1.28.1, and the real server under uvicorn answered initialize and tools/list through a tunnel-style Host header on 1.28.1. Co-Authored-By: Claude Opus 5 (1M context) --- requirements.txt | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/requirements.txt b/requirements.txt index a7743ed..0455f35 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,8 @@ # AutoCAD MCP server dependencies. # pip install -r requirements.txt # -# Both of these were already present on this machine; the pins record what the -# server was verified against on 2026-07-30. -mcp==1.27.1 +# The pins record what the server was verified against: pywin32 live with +# AutoCAD on 2026-07-30; mcp raised from 1.27.1 to 1.28.1 on 2026-09-21 to +# clear three published advisories, checked by tests/test_assembled_server.py. +mcp==1.28.1 pywin32==311 From 2aaf0c6acaa1796e46060e2b596f657f45c3436b Mon Sep 17 00:00:00 2001 From: Erfouni Date: Mon, 21 Sep 2026 10:24:36 +0330 Subject: [PATCH 3/3] Compare the bearer token in constant time SecretPathAuth compared the presented Authorization header with != which returns as soon as a character differs, so response timing can in principle reveal how much of a guessed token was right. Use hmac.compare_digest instead. Behaviour is otherwise unchanged: a missing header is still allowed, surrounding whitespace is still tolerated, and every existing gate test passes untouched. The comparison is done on UTF-8 bytes because compare_digest raises TypeError for str containing non-ASCII characters, and the header is attacker-controlled. Two new tests: one asserts the gate goes through compare_digest (it errors on the old code), the other sends a non-ASCII token and expects a 401 (it errors on a naive str comparison, which would have turned a bad token into a 500). Co-Authored-By: Claude Opus 5 (1M context) --- server.py | 10 +++++++++- tests/test_secret_path_gate.py | 20 ++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/server.py b/server.py index 7007702..3f53357 100644 --- a/server.py +++ b/server.py @@ -16,6 +16,7 @@ from __future__ import annotations import functools +import hmac import json import logging import sys @@ -135,7 +136,14 @@ async def __call__(self, scope, receive, send): if config.AUTH_TOKEN: headers = dict(scope.get("headers") or []) presented = headers.get(b"authorization", b"").decode(errors="replace") - if presented and presented.strip() != f"Bearer {config.AUTH_TOKEN}": + expected = f"Bearer {config.AUTH_TOKEN}" + # compare_digest rather than !=, so the time taken does not reveal + # how much of a guessed token was right. Compared as bytes because + # it rejects str with non-ASCII characters, which a hostile header + # can carry. + if presented and not hmac.compare_digest( + presented.strip().encode(), expected.encode() + ): log.warning("bad bearer token from %s", _client(scope)) return await _plain(send, 401, b"Unauthorized") diff --git a/tests/test_secret_path_gate.py b/tests/test_secret_path_gate.py index cfafac4..9700a1b 100644 --- a/tests/test_secret_path_gate.py +++ b/tests/test_secret_path_gate.py @@ -175,6 +175,26 @@ def test_surrounding_whitespace_is_tolerated(self): self.assertEqual(status, 200) self.assertTrue(downstream.reached) + def test_token_is_compared_in_constant_time(self): + """A plain != would let response timing leak how much of a guess matched.""" + gate, downstream = build() + with mock.patch.object(config, "AUTH_TOKEN", "s3cret"), \ + mock.patch.object(server.hmac, "compare_digest", + wraps=server.hmac.compare_digest) as digest: + status, _ = call(gate, "/unit-test-secret", + headers=[(b"authorization", b"Bearer wrong")]) + self.assertEqual(status, 401) + digest.assert_called_once() + + def test_non_ascii_token_is_refused_not_crashed(self): + """compare_digest raises on non-ASCII str; the gate must still say 401.""" + gate, downstream = build() + with mock.patch.object(config, "AUTH_TOKEN", "s3cret"): + status, _ = call(gate, "/unit-test-secret", + headers=[(b"authorization", "Bearer sëcret".encode())]) + self.assertEqual(status, 401) + self.assertFalse(downstream.reached) + def test_bad_token_on_a_bad_path_is_still_a_404(self): """The path check runs first; a wrong path must not reveal the token.""" gate, downstream = build()