jcwalker3andClaude Opus 4.8
a143cd065b
Addresses B1 and B2 from formal review 635 on PR #966. Both are one defect:
the seam conflated two questions that must stay separate.
recognition: "is this an identifier this system knows, and may it be
bound, pinned and recorded?"
execution: "is this entrypoint commissioned to actually serve on it?"
SUPPORTED_TRANSPORTS answered the first and was then used to answer the
second, so registering streamable-http was the same act as executing it.
With mcp.run fed the bound value, GITEA_MCP_TRANSPORT=streamable-http
reached FastMCP.run -> run_streamable_http_async and started a uvicorn
HTTP server over the whole tool surface with auth=None, no TLS and no
per-request principal. That endpoint is owned by #938 and is an explicit
non-goal of #931. B2 is why B1 was reachable: bound_transport was read
only by reporting sites, so no decision anywhere depended on it.
The fix adds an execution-authorization layer rather than removing the
identifier or mapping it to stdio, both of which would have destroyed the
seam #931 exists to create.
mcp_transport_config gains EXECUTABLE_TRANSPORTS, a strict subset of
SUPPORTED_TRANSPORTS containing only the local transport, plus
TRANSPORT_EXECUTION_OWNER recording that #938 commissions the remote
listener. assess_transport_execution returns a structured verdict:
transport, recognized, executable, allowed, blocker_kind, owner_issue,
reasons and exact_next_action.
mcp_daemon_guard gains assess_serve_authorization, which is the decision
that consumes bound_transport and is what stops it being reporting-only
metadata, and authorize_transport_execution, which enforces two ordered
boundaries: assert_transport_bound first, so an unbound runtime keeps its
pre-existing #695 failure and reason code; then execution authorization,
so a registered but uncommissioned transport is refused before any
listener exists and before any tool can dispatch. TransportExecutionError
subclasses UnsanctionedRuntimeError, so every existing fail-closed handler
still catches it while a caller that cares can distinguish the two cases.
native_runtime_status now reports executable_transports, serve_authorized
and the full serve_authorization verdict.
The entrypoint serves through authorize_transport_execution.
Net effect. stdio: unchanged, still binds and still reaches the runner.
streamable-http: still recognized, still validated, still pinned, still
recorded in decision-lock and audit provenance -- and refused at the serve
boundary with blocker transport_listener_not_commissioned naming #938.
Unregistered identifiers still fail earlier, at bind validation. Unbound
execution keeps the #695 contract. Rebind idempotence and conflict
rejection, session isolation, provenance and secret handling are untouched.
Tests: 63 in tests/test_issue_931_transport_bind_seam.py, up from 42. New
coverage proves stdio reaches the real runner; the remote identifier binds
and is recorded yet never reaches run(); no socket is bound during the
refusal, asserted with a socket.bind tripwire; the refusal names the
transport, the unmet requirement and #938; the decision genuinely consumes
bound_transport, proven by flipping only that value; no mutation is
authorized after denial; verdicts do not leak across runtimes; and the
refusal carries no credential material. Per review 635 the two file-list
scans are now globs over the production surface, and a new test asserts
there is exactly one mcp.run site and that it routes through the guard.
#956 anchors re-anchored for the two lines this change shifted,
mcp_daemon_guard.py 178->195 and 519->583. No other anchor moved.
Full suite from the branches/ worktree: 28 failed, 5864 passed, 6 skipped,
1047 subtests. Base 9b80e75c: 28 failed, 5801 passed, 6 skipped, 1047
subtests. Failing test IDs are identical sets in both directions; the +63
passed are this module.
No listener, socket, authentication, TLS or principal code is added. #938,
#957, #958 and #959 remain unimplemented.
Refs #931, #938. Addresses review 635.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
925 lines
39 KiB
Python
925 lines
39 KiB
Python
"""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()
|