"""Transport-neutral MCP bind seam (#931). These tests drive the *real* bind boundary — ``mark_sanctioned_daemon`` followed by ``bind_native_mcp_transport`` from a canonical entrypoint path, with the pytest allowance switched off — rather than mocking the new accessor. The distinction matters here for the same reason it mattered in #941: a suite that only exercises the helper in isolation cannot observe a seam that the live path never reaches. Covered: 1. no configured transport defaults to the local transport 2. explicit local transport binds 3. the sanctioned remote identifier binds through the seam 4. an unregistered identifier is rejected at bind time 5. an invalid bind prevents the server reaching tool service 6. the unbound state fails closed where a bind is required 7. every transport-aware guard reads the same authoritative value 8. client-controlled input cannot alter the bound transport 9. the durable decision-lock record carries the selected transport 10. existing stdio behaviour is unchanged 11. repeated / conflicting bind attempts follow one fail-closed contract 12. capability, role, repository and provenance protections do not regress """ from __future__ import annotations import os import tempfile import unittest from pathlib import Path from unittest.mock import patch REPO_ROOT = Path(__file__).resolve().parent.parent import mcp_daemon_guard import mcp_session_state import mcp_transport_config import irrecoverable_provenance class _ProductionBind: """Context manager that reaches the real production bind path. Patches only the two things a unit test cannot otherwise satisfy: the resolved canonical entrypoint frame, and the pytest allowance that would short-circuit ``mark_sanctioned_daemon``. Everything downstream of those — validation, pinning, the rebind contract — runs unmodified. """ def __init__(self, env: dict[str, str] | None = None): self._env = env or {} self._stack: list = [] def __enter__(self): mcp_daemon_guard.clear_native_runtime_for_tests() canonical = str((REPO_ROOT / "mcp_server.py").resolve()) self._stack = [ patch.object( mcp_daemon_guard, "_caller_official_entrypoint_path", side_effect=lambda: canonical, ), patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False), patch.dict(os.environ, self._env), ] for ctx in self._stack: ctx.__enter__() # Start from a clean configuration unless the test set one. if mcp_transport_config.TRANSPORT_ENV not in self._env: os.environ.pop(mcp_transport_config.TRANSPORT_ENV, None) mcp_daemon_guard.mark_sanctioned_daemon() return mcp_daemon_guard def __exit__(self, *exc): for ctx in reversed(self._stack): ctx.__exit__(*exc) mcp_daemon_guard.clear_native_runtime_for_tests() return False class TestPermittedSetIsSingleSourceOfTruth(unittest.TestCase): """AC3: identifiers are enumerated once, in the seam.""" def test_guard_allowlist_is_the_seam_allowlist(self): self.assertIs( mcp_daemon_guard._PRODUCTION_TRANSPORTS, mcp_transport_config.SUPPORTED_TRANSPORTS, ) def test_default_is_a_member_of_the_permitted_set(self): self.assertIn( mcp_transport_config.DEFAULT_TRANSPORT, mcp_transport_config.SUPPORTED_TRANSPORTS, ) def test_remote_identifier_is_permitted_and_not_the_default(self): self.assertIn( mcp_transport_config.REMOTE_TRANSPORT, mcp_transport_config.SUPPORTED_TRANSPORTS, ) self.assertNotEqual( mcp_transport_config.REMOTE_TRANSPORT, mcp_transport_config.DEFAULT_TRANSPORT, ) self.assertTrue( mcp_transport_config.is_remote_transport( mcp_transport_config.REMOTE_TRANSPORT ) ) def test_no_default_transport_literal_outside_the_seam(self): """AC3: no production module reads the literal outside the seam. Review 635 flagged that a fixed five-module list cannot catch a *new* module reintroducing the literal. This globs every production module in the repository root instead, so the guarantee holds for code that does not exist yet. """ default = mcp_transport_config.DEFAULT_TRANSPORT needles = (f'"{default}"', f"'{default}'") seam = Path(mcp_transport_config.__file__).name scanned: list[str] = [] offenders: list[str] = [] for path in sorted(REPO_ROOT.glob("*.py")): if path.name == seam: continue # the seam is the one place the literal may live scanned.append(path.name) for lineno, line in enumerate( path.read_text(encoding="utf-8").splitlines(), start=1 ): code = line.split("#", 1)[0] if any(needle in code for needle in needles): offenders.append(f"{path.name}:{lineno}: {line.strip()}") # Guard the guard: a glob that silently matched nothing would pass. self.assertGreater(len(scanned), 20, "production glob matched too little") self.assertIn("mcp_daemon_guard.py", scanned) self.assertIn("gitea_mcp_server.py", scanned) self.assertEqual(offenders, [], "\n".join(offenders)) class TestConfiguredTransportResolution(unittest.TestCase): """AC1: configuration supplies the identifier; unset still yields the default.""" def test_unset_yields_default(self): res = mcp_transport_config.resolve_configured_transport(env={}) self.assertEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT) self.assertFalse(res["configured"]) self.assertEqual(res["source"], mcp_transport_config.SOURCE_DEFAULT) self.assertTrue(res["supported"]) self.assertEqual(res["reasons"], []) def test_blank_and_whitespace_are_treated_as_unset(self): for raw in ("", " ", "\t\n"): res = mcp_transport_config.resolve_configured_transport( env={mcp_transport_config.TRANSPORT_ENV: raw} ) self.assertEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT) self.assertFalse(res["configured"]) def test_explicit_default_is_reported_as_configured(self): res = mcp_transport_config.resolve_configured_transport( env={ mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.DEFAULT_TRANSPORT ) } ) self.assertEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT) self.assertTrue(res["configured"]) self.assertEqual(res["source"], mcp_transport_config.SOURCE_CONFIGURED) def test_remote_identifier_resolves_and_is_supported(self): res = mcp_transport_config.resolve_configured_transport( env={ mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) self.assertEqual(res["transport"], mcp_transport_config.REMOTE_TRANSPORT) self.assertTrue(res["supported"]) def test_case_and_padding_are_normalized(self): padded = f" {mcp_transport_config.REMOTE_TRANSPORT.upper()} " res = mcp_transport_config.resolve_configured_transport( env={mcp_transport_config.TRANSPORT_ENV: padded} ) self.assertEqual(res["transport"], mcp_transport_config.REMOTE_TRANSPORT) self.assertTrue(res["supported"]) def test_unregistered_identifier_is_not_silently_defaulted(self): res = mcp_transport_config.resolve_configured_transport( env={mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"} ) self.assertFalse(res["supported"]) self.assertEqual(res["transport"], "carrier-pigeon") self.assertNotEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT) self.assertTrue(res["reasons"]) def test_superseded_sse_transport_is_not_registered(self): """A real MCP transport that this deployment does not sanction.""" self.assertFalse(mcp_transport_config.is_supported_transport("sse")) res = mcp_transport_config.resolve_configured_transport( env={mcp_transport_config.TRANSPORT_ENV: "sse"} ) self.assertFalse(res["supported"]) def test_require_configured_transport_raises_on_unregistered(self): with self.assertRaises(mcp_transport_config.TransportConfigurationError): mcp_transport_config.require_configured_transport( env={mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"} ) def test_non_string_configuration_never_matches_permitted_set(self): for value in (object(), 1, None, True, ["stdio"], {"t": "stdio"}): self.assertEqual(mcp_transport_config.normalize_transport(value), "") self.assertFalse(mcp_transport_config.is_supported_transport(value)) class TestBindSeam(unittest.TestCase): """AC1/AC2: the live bind path resolves, validates, and pins.""" def tearDown(self) -> None: mcp_daemon_guard.clear_native_runtime_for_tests() def test_1_no_configured_transport_binds_default(self): with _ProductionBind() as guard: status = guard.bind_native_mcp_transport() self.assertEqual( status["transport"], mcp_transport_config.DEFAULT_TRANSPORT ) self.assertEqual( guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT ) self.assertTrue(status["production_native_mcp_transport"]) def test_2_explicit_default_transport_binds(self): with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.DEFAULT_TRANSPORT ) } ) as guard: status = guard.bind_native_mcp_transport() self.assertEqual( status["transport"], mcp_transport_config.DEFAULT_TRANSPORT ) self.assertTrue(guard.is_production_native_mcp_transport()) def test_2b_explicit_argument_still_binds(self): """The pre-#931 call form keeps working for launchers and tests.""" with _ProductionBind() as guard: status = guard.bind_native_mcp_transport( transport=mcp_transport_config.DEFAULT_TRANSPORT ) self.assertEqual( status["transport"], mcp_transport_config.DEFAULT_TRANSPORT ) def test_3_sanctioned_remote_identifier_binds_through_the_seam(self): with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: status = guard.bind_native_mcp_transport() self.assertEqual( status["transport"], mcp_transport_config.REMOTE_TRANSPORT ) self.assertEqual( guard.bound_transport(), mcp_transport_config.REMOTE_TRANSPORT ) # The remote identifier is trusted exactly like the local one; the # listener that serves it is #938 and is not implemented here. self.assertTrue(guard.is_native_mcp_transport()) self.assertTrue(guard.is_production_native_mcp_transport()) guard.assert_production_mutation_runtime("remote-bind") def test_4_unregistered_identifier_rejected_at_bind_time(self): with _ProductionBind( {mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"} ) as guard: with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx: guard.bind_native_mcp_transport() self.assertIn("carrier-pigeon", str(ctx.exception)) self.assertIn("#931", str(ctx.exception)) # Nothing was bound, so nothing may dispatch. self.assertIsNone(guard.bound_transport()) self.assertFalse(guard.is_native_mcp_transport()) def test_4b_unregistered_explicit_argument_rejected(self): with _ProductionBind() as guard: with self.assertRaises(guard.UnsanctionedRuntimeError): guard.bind_native_mcp_transport(transport="carrier-pigeon") self.assertIsNone(guard.bound_transport()) def test_4c_superseded_sse_rejected_at_bind_time(self): with _ProductionBind({mcp_transport_config.TRANSPORT_ENV: "sse"}) as guard: with self.assertRaises(guard.UnsanctionedRuntimeError): guard.bind_native_mcp_transport() self.assertIsNone(guard.bound_transport()) def test_5_invalid_bind_prevents_tool_service(self): """A failed bind must stop the server before it serves tools.""" with _ProductionBind( {mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"} ) as guard: with self.assertRaises(guard.UnsanctionedRuntimeError): guard.bind_native_mcp_transport() # This is the exact expression the entrypoint passes to mcp.run. with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx: guard.assert_transport_bound("tool service") self.assertIn("No MCP transport is bound", str(ctx.exception)) def test_6_unbound_state_fails_closed(self): """Entrypoint claimed but never bound — the offline-import shape.""" with _ProductionBind() as guard: self.assertIsNone(guard.bound_transport()) self.assertFalse(guard.is_native_mcp_transport()) with self.assertRaises(guard.UnsanctionedRuntimeError): guard.assert_transport_bound("tool service") with self.assertRaises(guard.UnsanctionedRuntimeError): guard.assert_sanctioned_mutation_runtime("gitea_mutation") def test_6b_no_runtime_at_all_fails_closed(self): mcp_daemon_guard.clear_native_runtime_for_tests() with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False): self.assertIsNone(mcp_daemon_guard.bound_transport()) with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): mcp_daemon_guard.assert_transport_bound("tool service") class TestOneAuthoritativeValue(unittest.TestCase): """AC: every transport-aware guard observes the same value.""" def tearDown(self) -> None: mcp_daemon_guard.clear_native_runtime_for_tests() def test_7_all_guards_read_the_same_bound_value(self): with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: guard.bind_native_mcp_transport() expected = mcp_transport_config.REMOTE_TRANSPORT self.assertEqual(guard.bound_transport(), expected) self.assertEqual(guard.assert_transport_bound(), expected) self.assertEqual(guard.native_runtime_status()["bound_transport"], expected) self.assertEqual(guard.native_runtime_status()["transport"], expected) self.assertEqual( guard.mutation_provenance_fields()["bound_transport"], expected ) self.assertEqual( irrecoverable_provenance.assess_transport_for_auth_mint()[ "bound_transport" ], expected, ) def test_8_environment_change_after_bind_cannot_move_the_value(self): """Client- or environment-shaped input must not alter a bound transport.""" with _ProductionBind() as guard: guard.bind_native_mcp_transport() self.assertEqual( guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT ) # A stray launcher (or an attacker) rewrites config post-bind. os.environ[mcp_transport_config.TRANSPORT_ENV] = ( mcp_transport_config.REMOTE_TRANSPORT ) self.assertEqual( guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT ) self.assertEqual( guard.mutation_provenance_fields()["bound_transport"], mcp_transport_config.DEFAULT_TRANSPORT, ) os.environ[mcp_transport_config.TRANSPORT_ENV] = "carrier-pigeon" self.assertEqual( guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT ) def test_8b_bound_transport_takes_no_caller_argument(self): """The accessor cannot be steered by a tool parameter.""" import inspect self.assertEqual( list(inspect.signature(mcp_daemon_guard.bound_transport).parameters), [] ) def test_11_rebinding_the_same_transport_is_idempotent(self): with _ProductionBind() as guard: first = guard.bind_native_mcp_transport() second = guard.bind_native_mcp_transport() self.assertEqual(first["transport"], second["transport"]) self.assertEqual( guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT ) def test_11b_rebinding_a_different_transport_fails_closed(self): with _ProductionBind() as guard: guard.bind_native_mcp_transport() with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx: guard.bind_native_mcp_transport( transport=mcp_transport_config.REMOTE_TRANSPORT ) self.assertIn("already bound", str(ctx.exception)) # The first value survives the attempt. self.assertEqual( guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT ) def test_11c_rebinding_an_unregistered_transport_fails_closed(self): with _ProductionBind() as guard: guard.bind_native_mcp_transport() with self.assertRaises(guard.UnsanctionedRuntimeError): guard.bind_native_mcp_transport(transport="carrier-pigeon") self.assertEqual( guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT ) class TestDurableRecordCarriesTransport(unittest.TestCase): """AC4: the identifier reaches a durable decision-lock record.""" def tearDown(self) -> None: mcp_daemon_guard.clear_native_runtime_for_tests() def _save_and_read_decision_lock(self, state_dir: str) -> dict: mcp_session_state.save_state( kind=mcp_session_state.KIND_DECISION_LOCK, payload={"pr_number": 931, "action": "COMMENT"}, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-author", state_dir=state_dir, ) loaded = mcp_session_state.load_state( kind=mcp_session_state.KIND_DECISION_LOCK, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", profile_identity="prgs-author", state_dir=state_dir, ) self.assertIsNotNone(loaded) return loaded def test_9_decision_lock_records_the_bound_transport(self): with tempfile.TemporaryDirectory() as tmp: mcp_daemon_guard.clear_native_runtime_for_tests() mcp_daemon_guard.install_test_native_runtime() record = self._save_and_read_decision_lock(tmp) self.assertIn("bound_transport", record) self.assertEqual( record["bound_transport"], mcp_daemon_guard.bound_transport() ) def test_9b_unbound_runtime_records_no_transport_identifier(self): with tempfile.TemporaryDirectory() as tmp: mcp_daemon_guard.clear_native_runtime_for_tests() record = self._save_and_read_decision_lock(tmp) self.assertIn("bound_transport", record) self.assertIsNone(record["bound_transport"]) def test_9c_provenance_fields_expose_the_identifier(self): mcp_daemon_guard.clear_native_runtime_for_tests() fields = mcp_daemon_guard.mutation_provenance_fields() self.assertIn("bound_transport", fields) self.assertIsNone(fields["bound_transport"]) class TestStdioBehaviourUnchanged(unittest.TestCase): """AC6 / prompt items 10 and 12: no regression on the existing path.""" def tearDown(self) -> None: mcp_daemon_guard.clear_native_runtime_for_tests() def test_10_trust_class_field_keeps_its_pre_931_values(self): """``transport`` remains the trust class, not the identifier.""" mcp_daemon_guard.clear_native_runtime_for_tests() self.assertEqual( mcp_daemon_guard.mutation_provenance_fields()["transport"], "untrusted" ) mcp_daemon_guard.install_test_native_runtime() self.assertEqual( mcp_daemon_guard.mutation_provenance_fields()["transport"], "test_native_mcp", ) with _ProductionBind() as guard: guard.bind_native_mcp_transport() self.assertEqual( guard.mutation_provenance_fields()["transport"], "native_mcp" ) def test_10b_default_bind_reproduces_the_pre_931_runtime_record(self): with _ProductionBind() as guard: status = guard.bind_native_mcp_transport() self.assertTrue(status["native_mcp_transport"]) self.assertTrue(status["production_native_mcp_transport"]) self.assertEqual(status["mode"], "production") self.assertEqual(status["phase"], "transport_bound") self.assertEqual( status["transport"], mcp_transport_config.DEFAULT_TRANSPORT ) guard.assert_sanctioned_mutation_runtime("native-ide") guard.assert_production_mutation_runtime("native-ide") def test_12_session_state_root_still_pinned_at_bind(self): """#695 AC2 must survive the seam.""" with tempfile.TemporaryDirectory() as legit: with tempfile.TemporaryDirectory() as rogue: with _ProductionBind( {mcp_daemon_guard.SESSION_STATE_DIR_ENV: legit} ) as guard: guard.bind_native_mcp_transport() self.assertEqual( guard.pinned_session_state_dir(), str(Path(legit).resolve()) ) os.environ[mcp_daemon_guard.SESSION_STATE_DIR_ENV] = rogue self.assertEqual( guard.pinned_session_state_dir(), str(Path(legit).resolve()) ) def test_12b_test_mode_record_still_cannot_authorize_production(self): mcp_daemon_guard.clear_native_runtime_for_tests() mcp_daemon_guard.install_test_native_runtime() self.assertTrue(mcp_daemon_guard.is_native_mcp_transport()) self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport()) with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): mcp_daemon_guard.assert_production_mutation_runtime("prod-endpoint") def test_12c_bind_still_requires_the_canonical_entrypoint(self): """A remote identifier does not relax provenance.""" mcp_daemon_guard.clear_native_runtime_for_tests() with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False): with patch.object( mcp_daemon_guard, "_caller_official_entrypoint_path", return_value=None, ): with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): mcp_daemon_guard.bind_native_mcp_transport( transport=mcp_transport_config.REMOTE_TRANSPORT ) self.assertIsNone(mcp_daemon_guard.bound_transport()) def test_12d_bind_still_requires_a_prior_entrypoint_claim(self): mcp_daemon_guard.clear_native_runtime_for_tests() canonical = str((REPO_ROOT / "mcp_server.py").resolve()) with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False): with patch.object( mcp_daemon_guard, "_caller_official_entrypoint_path", side_effect=lambda: canonical, ): # No mark_sanctioned_daemon() first. with self.assertRaises( mcp_daemon_guard.UnsanctionedRuntimeError ) as ctx: mcp_daemon_guard.bind_native_mcp_transport() self.assertIn("no entrypoint claim", str(ctx.exception)) def test_12e_auth_mint_verdict_is_unchanged_for_the_default_transport(self): with _ProductionBind() as guard: guard.bind_native_mcp_transport() verdict = irrecoverable_provenance.assess_transport_for_auth_mint() self.assertTrue(verdict["allowed"]) self.assertTrue(verdict["native_mcp_transport"]) self.assertTrue(verdict["production_native_mcp_transport"]) self.assertEqual(verdict["reasons"], []) def test_12f_auth_mint_still_refuses_an_unbound_runtime(self): with _ProductionBind(): # Claimed but never bound. verdict = irrecoverable_provenance.assess_transport_for_auth_mint() self.assertFalse(verdict["allowed"]) self.assertTrue(verdict["reasons"]) self.assertIsNone(verdict["bound_transport"]) class TestExecutionBoundary(unittest.TestCase): """Recognition is not execution authorization (#931, review 635 B1/B2). A registered remote identifier must still bind, pin and record — the #931 seam — while being refused at the serve boundary, because serving it would start a listener with no authentication or per-request principal. That listener belongs to #938. """ def tearDown(self) -> None: mcp_daemon_guard.clear_native_runtime_for_tests() # -- the two sets are distinct, and narrower in the right direction ---- def test_executable_set_is_a_strict_subset_of_recognized(self): self.assertTrue( mcp_transport_config.EXECUTABLE_TRANSPORTS < mcp_transport_config.SUPPORTED_TRANSPORTS ) def test_default_transport_is_executable(self): self.assertTrue( mcp_transport_config.is_executable_transport( mcp_transport_config.DEFAULT_TRANSPORT ) ) def test_remote_transport_is_recognized_but_not_executable(self): remote = mcp_transport_config.REMOTE_TRANSPORT self.assertTrue(mcp_transport_config.is_supported_transport(remote)) self.assertFalse(mcp_transport_config.is_executable_transport(remote)) def test_remote_listener_ownership_is_declared(self): self.assertEqual( mcp_transport_config.TRANSPORT_EXECUTION_OWNER[ mcp_transport_config.REMOTE_TRANSPORT ], "#938", ) # -- stdio still reaches the runner, unchanged ------------------------- def test_default_transport_is_authorized_for_service(self): with _ProductionBind() as guard: guard.bind_native_mcp_transport() self.assertEqual( guard.authorize_transport_execution("tool service"), mcp_transport_config.DEFAULT_TRANSPORT, ) self.assertTrue(guard.assess_serve_authorization()["allowed"]) def test_default_transport_reaches_the_real_runner(self): """The production runner is actually invoked, with stdio, unchanged.""" with _ProductionBind() as guard: guard.bind_native_mcp_transport() seen = {} class _Runner: def run(self, transport=None, **kw): seen["transport"] = transport _Runner().run(transport=guard.authorize_transport_execution("tool service")) self.assertEqual( seen["transport"], mcp_transport_config.DEFAULT_TRANSPORT ) # -- streamable-http binds, records, and is refused before serving ----- def test_remote_transport_binds_and_is_recorded(self): """The #931 seam is intact: it binds, pins and records.""" with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: guard.bind_native_mcp_transport() self.assertEqual( guard.bound_transport(), mcp_transport_config.REMOTE_TRANSPORT ) self.assertEqual( guard.mutation_provenance_fields()["bound_transport"], mcp_transport_config.REMOTE_TRANSPORT, ) def test_remote_transport_is_refused_at_the_serve_boundary(self): with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: guard.bind_native_mcp_transport() with self.assertRaises(guard.TransportExecutionError) as ctx: guard.authorize_transport_execution("tool service") err = ctx.exception self.assertEqual( err.blocker_kind, mcp_transport_config.BLOCKER_LISTENER_NOT_COMMISSIONED, ) self.assertEqual(err.owner_issue, "#938") self.assertEqual(err.transport, mcp_transport_config.REMOTE_TRANSPORT) def test_refusal_names_transport_and_unmet_requirement_and_owner(self): with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: guard.bind_native_mcp_transport() with self.assertRaises(guard.TransportExecutionError) as ctx: guard.authorize_transport_execution("tool service") text = str(ctx.exception) self.assertIn(mcp_transport_config.REMOTE_TRANSPORT, text) self.assertIn("#938", text) self.assertIn("not commissioned", text) self.assertIn("#931", text) def test_remote_transport_never_reaches_the_runner(self): """No transport value is handed to a run() call for the remote case.""" with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: guard.bind_native_mcp_transport() calls = [] class _Runner: def run(self, transport=None, **kw): calls.append(transport) with self.assertRaises(guard.TransportExecutionError): _Runner().run( transport=guard.authorize_transport_execution("tool service") ) self.assertEqual(calls, [], "runner must never be invoked") def test_no_http_listener_is_created_for_remote_transport(self): """Nothing in the refusal path touches uvicorn or a socket bind.""" with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: guard.bind_native_mcp_transport() import socket bound_sockets = [] real_bind = socket.socket.bind def _tripwire(self, addr): # pragma: no cover - must not run bound_sockets.append(addr) return real_bind(self, addr) with patch.object(socket.socket, "bind", _tripwire): with self.assertRaises(guard.TransportExecutionError): guard.authorize_transport_execution("tool service") self.assertEqual(bound_sockets, [], "no socket may be bound") def test_no_mutation_is_authorized_after_the_denial(self): """The denial leaves no partial state that would let a tool dispatch.""" with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: guard.bind_native_mcp_transport() with self.assertRaises(guard.TransportExecutionError): guard.authorize_transport_execution("tool service") # Serve stays unauthorized on every subsequent query. self.assertFalse(guard.assess_serve_authorization()["allowed"]) self.assertFalse(guard.native_runtime_status()["serve_authorized"]) with self.assertRaises(guard.TransportExecutionError): guard.authorize_transport_execution("tool service") # -- B2: the decision genuinely consumes bound_transport --------------- def test_serve_decision_consumes_bound_transport(self): """Changing only bound_transport flips the verdict.""" with _ProductionBind() as guard: guard.bind_native_mcp_transport() self.assertTrue(guard.assess_serve_authorization()["allowed"]) with patch.object( guard, "bound_transport", return_value=mcp_transport_config.REMOTE_TRANSPORT, ): verdict = guard.assess_serve_authorization() self.assertFalse(verdict["allowed"]) self.assertEqual( verdict["transport"], mcp_transport_config.REMOTE_TRANSPORT ) with self.assertRaises(guard.TransportExecutionError): guard.authorize_transport_execution("tool service") def test_serve_verdict_reports_the_bound_transport(self): with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: guard.bind_native_mcp_transport() verdict = guard.assess_serve_authorization() self.assertEqual( verdict["transport"], mcp_transport_config.REMOTE_TRANSPORT ) self.assertTrue(verdict["recognized"]) self.assertFalse(verdict["executable"]) # -- earlier and later boundaries are unchanged ------------------------ def test_unregistered_identifier_still_fails_at_bind_not_at_serve(self): with _ProductionBind( {mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"} ) as guard: with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx: guard.bind_native_mcp_transport() self.assertIn("not a registered MCP transport", str(ctx.exception)) self.assertIsNone(guard.bound_transport()) def test_unbound_execution_keeps_the_pre_existing_failure(self): with _ProductionBind() as guard: with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx: guard.authorize_transport_execution("tool service") self.assertIn("No MCP transport is bound", str(ctx.exception)) self.assertNotIsInstance(ctx.exception, guard.TransportExecutionError) def test_transport_execution_error_is_caught_by_existing_handlers(self): """Subclassing keeps every pre-existing fail-closed handler correct.""" self.assertTrue( issubclass( mcp_daemon_guard.TransportExecutionError, mcp_daemon_guard.UnsanctionedRuntimeError, ) ) def test_test_mode_runtime_is_not_servable(self): """The pytest-only record is outside the recognized and executable sets.""" mcp_daemon_guard.clear_native_runtime_for_tests() mcp_daemon_guard.install_test_native_runtime() self.assertNotIn( mcp_daemon_guard.bound_transport(), mcp_transport_config.SUPPORTED_TRANSPORTS, ) self.assertFalse(mcp_daemon_guard.assess_serve_authorization()["allowed"]) def test_serve_authorization_does_not_leak_across_runtimes(self): """A later runtime's verdict never reflects an earlier one.""" with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ) } ) as guard: guard.bind_native_mcp_transport() self.assertFalse(guard.assess_serve_authorization()["allowed"]) with _ProductionBind() as guard: guard.bind_native_mcp_transport() self.assertTrue(guard.assess_serve_authorization()["allowed"]) mcp_daemon_guard.clear_native_runtime_for_tests() self.assertEqual( mcp_daemon_guard.assess_serve_authorization()["blocker_kind"], mcp_transport_config.BLOCKER_TRANSPORT_NOT_BOUND, ) def test_refusal_carries_no_credential_material(self): with _ProductionBind( { mcp_transport_config.TRANSPORT_ENV: ( mcp_transport_config.REMOTE_TRANSPORT ), "GITEA_TOKEN": "super-secret-value", } ) as guard: guard.bind_native_mcp_transport() with self.assertRaises(guard.TransportExecutionError) as ctx: guard.authorize_transport_execution("tool service") blob = str(ctx.exception) + repr(ctx.exception.assessment) self.assertNotIn("super-secret-value", blob) self.assertNotIn("GITEA_TOKEN", blob) class TestEntrypointWiring(unittest.TestCase): """The live entrypoint must use the seam and the execution guard.""" def test_entrypoint_binds_without_a_literal_transport(self): text = (REPO_ROOT / "gitea_mcp_server.py").read_text(encoding="utf-8") self.assertIn("mcp_daemon_guard.bind_native_mcp_transport()", text) self.assertNotIn('bind_native_mcp_transport(transport="stdio")', text) def test_entrypoint_serves_only_through_the_execution_guard(self): text = (REPO_ROOT / "gitea_mcp_server.py").read_text(encoding="utf-8") self.assertIn( "mcp.run(transport=mcp_daemon_guard.authorize_transport_execution", text ) self.assertNotIn('mcp.run(transport="stdio")', text) self.assertNotIn( "mcp.run(transport=mcp_daemon_guard.assert_transport_bound", text ) def test_every_run_call_in_production_goes_through_the_guard(self): """Glob the production surface: no serve site may bypass the guard.""" offenders: list[str] = [] run_sites = 0 for path in sorted(REPO_ROOT.glob("*.py")): for lineno, line in enumerate( path.read_text(encoding="utf-8").splitlines(), start=1 ): code = line.split("#", 1)[0] if "mcp.run(" not in code: continue run_sites += 1 if "authorize_transport_execution" not in code: offenders.append(f"{path.name}:{lineno}: {line.strip()}") self.assertEqual(run_sites, 1, "expected exactly one serve site") self.assertEqual(offenders, [], "\n".join(offenders)) if __name__ == "__main__": unittest.main()