Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 101 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

fix(transport): gate transport execution behind commissioning (review 635)

Browse Source 
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]>
This commit is contained in:
jcwalker3
2026-07-28 11:15:00 -05:00
co-authored by Claude Opus 4.8
parent 2fb835a1aa
commit a143cd065b
6 changed files with 567 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_issue_931_transport_bind_seam.py
+344 -15
View File
@@ -110,21 +110,32 @@ class TestPermittedSetIsSingleSourceOfTruth(unittest.TestCase):
)
def test_no_default_transport_literal_outside_the_seam(self):
"""AC3: no guard reads the literal outside the seam definition."""
"""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 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()}")
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))
@@ -577,18 +588,336 @@ class TestStdioBehaviourUnchanged(unittest.TestCase):
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, not a literal."""
"""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_the_bound_transport(self):
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.assert_transport_bound", text)
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__":