"""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 guard reads the literal outside the seam definition.""" default = mcp_transport_config.DEFAULT_TRANSPORT needles = (f'"{default}"', f"'{default}'") offenders: list[str] = [] for name in ( "mcp_daemon_guard.py", "gitea_mcp_server.py", "irrecoverable_provenance.py", "mcp_session_state.py", "review_quarantine.py", ): text = (REPO_ROOT / name).read_text(encoding="utf-8") for lineno, line in enumerate(text.splitlines(), start=1): if any(needle in line for needle in needles): offenders.append(f"{name}:{lineno}: {line.strip()}") 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 TestEntrypointWiring(unittest.TestCase): """The live entrypoint must use the seam, not a literal.""" 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_the_bound_transport(self): text = (REPO_ROOT / "gitea_mcp_server.py").read_text(encoding="utf-8") self.assertIn("mcp.run(transport=mcp_daemon_guard.assert_transport_bound", text) self.assertNotIn('mcp.run(transport="stdio")', text) if __name__ == "__main__": unittest.main()