From da6fd233f4916bb14a93ea0c0ea152170a50c9b9 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 13 Jul 2026 09:42:47 -0400 Subject: [PATCH] fix(guard): close allow_test_bootstrap and basename entrypoint bypasses (#695) Remove the public allow_test_bootstrap production seam so caller-controlled flags cannot forge native mutation provenance. Bind provenance to the resolved canonical entrypoint path plus a live stdio transport bind; basename-only mcp_server.py stack frames and import-only launch no longer authorize mutations. Test-mode install_test_native_runtime is pytest-only and cannot reach production Gitea mutation endpoints. Add AC9 regressions for both reviewer-found bypasses and related spoof vectors. Refs: PR #696 REQUEST_CHANGES at 253269c; issue #695 comments 11002/11005. --- docs/mcp-daemon-import-guard.md | 28 +- gitea_mcp_server.py | 13 +- mcp_daemon_guard.py | 302 +++++++++++++++-- mcp_server.py | 8 +- ...t_issue_695_native_transport_quarantine.py | 305 +++++++++++++++++- tests/test_mcp_daemon_guard.py | 17 +- 6 files changed, 609 insertions(+), 64 deletions(-) diff --git a/docs/mcp-daemon-import-guard.md b/docs/mcp-daemon-import-guard.md index 64696c7..bebd141 100644 --- a/docs/mcp-daemon-import-guard.md +++ b/docs/mcp-daemon-import-guard.md @@ -9,22 +9,36 @@ formal reviews then looked identical to native approvals. ## Rule -Mutation auth, keychain fill, and controller quarantine require a **native MCP -transport runtime** established only by the official entrypoint. +Mutation auth, keychain fill, and controller quarantine require a **production +native MCP transport runtime** established only by: + +1. the **resolved absolute path** of the canonical entrypoint + (`mcp_server.py` / `gitea_mcp_server.py` next to `mcp_daemon_guard.py`), and +2. a live **transport bind** (`bind_native_mcp_transport(transport="stdio")`) + immediately before `mcp.run`. + +Basename-only trust (a renamed file called `mcp_server.py`), caller-controlled +flags (there is **no** `allow_test_bootstrap`), environment variables, stack +frame spoofing, or import-only launch are insufficient. | Context | Allowed | |---------|---------| -| Official MCP entrypoint (`mcp_server.py` / `gitea_mcp_server` `__main__`) calls `mark_sanctioned_daemon()` and holds a process-local runtime token | yes | -| pytest (hermetic unit tests) | yes | +| Official IDE-native MCP: resolved canonical entrypoint marks + binds stdio, holds process-local runtime token | yes | +| pytest (hermetic unit tests) via `is_pytest_runtime()` | yes for unit gates | +| `install_test_native_runtime()` under pytest (test-mode record) | unit-test transport gates only — **never** production Gitea mutations | +| `allow_test_bootstrap=True` (removed; must not exist) | **no** | +| Renamed runner basename `mcp_server.py` outside package root | **no** | +| Import/launch of real entrypoint without transport bind | **no** | | `GITEA_MCP_SANCTIONED_DAEMON=1` alone (no process-local native runtime) | **no** (#695) | | `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` in LLM sessions | **no** — never set in agent sessions | | `GITEA_ALLOW_KEYCHAIN_CLI=1` in LLM sessions | **no** — human operator only | | bare `python -c 'import gitea_mcp_server; …'` or offline runners | **no** | | keychain fill outside native/pytest | **no** | -Native runtime is **process-local**: a random token bound to the daemon PID. -It is never reconstructed from environment variables, session-state files, or -importing internals in a fresh Python process. +Native runtime is **process-local**: a random token bound to the daemon PID and +transport phase. It is never reconstructed from environment variables, +session-state files, caller-controlled flags, or importing internals in a +fresh Python process. ## Contaminated review quarantine (#695 AC8) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 8593a94..854ce4d 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -12843,9 +12843,10 @@ def gitea_quarantine_contaminated_review( 427 until an independent adversarial reviewer and controller deployment of this tooling have completed. """ - # Fail closed outside native MCP before any mutation assessment. + # Fail closed outside production native MCP before any mutation assessment. + # Test-mode bootstrap must never reach this production mutation endpoint (#695). try: - mcp_daemon_guard.assert_sanctioned_mutation_runtime( + mcp_daemon_guard.assert_production_mutation_runtime( "gitea_quarantine_contaminated_review" ) except mcp_daemon_guard.UnsanctionedRuntimeError as exc: @@ -13051,10 +13052,12 @@ def gitea_quarantine_contaminated_review( # ── Entry point ─────────────────────────────────────────────────────────────── if __name__ == "__main__": - # #558 / #695: mark this process as the official native MCP daemon before - # any tool dispatch. Env vars alone cannot reconstruct native transport; - # offline imports / standalone scripts fail closed on mutations. + # #558 / #695: claim the resolved canonical entrypoint, then bind the live + # native MCP transport lifecycle before any tool dispatch. Env vars, + # basename-only stack frames, and import-only launch cannot reconstruct + # native transport; offline imports / standalone scripts fail closed. mcp_daemon_guard.mark_sanctioned_daemon() + mcp_daemon_guard.bind_native_mcp_transport(transport="stdio") # Lock this session's launch profile into the environment so child CLI # processes (e.g. review_pr.py) can detect and refuse profile # side-channel overrides (#199). diff --git a/mcp_daemon_guard.py b/mcp_daemon_guard.py index 6e2d4f8..11675bb 100644 --- a/mcp_daemon_guard.py +++ b/mcp_daemon_guard.py @@ -6,12 +6,18 @@ Direct ``import gitea_mcp_server`` from a shell bypasses native MCP transport. - Environment variables alone cannot reconstruct a native session (``GITEA_MCP_SANCTIONED_DAEMON=1`` / ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` are insufficient for mutation gates). -- A process-local runtime record is created only by the official entrypoint - (``mcp_server.py`` calling ``mark_sanctioned_daemon``), bound to PID and a - random secret that never leaves process memory. +- A process-local runtime record is established only by the resolved canonical + entrypoint path (not basename) **and** the actual native MCP transport + lifecycle (``bind_native_mcp_transport`` before ``mcp.run``). Merely + importing or launching the entrypoint offline does not grant mutation + authority. +- Public caller-controlled flags (including any former + ``allow_test_bootstrap``) never establish trusted mutation provenance. - Offline scripts that import internals fail closed on mutations. -- Pytest remains allowed (hermetic tests); optional force flags exist for - provenance regression tests. +- Pytest remains allowed for hermetic unit tests via ``is_pytest_runtime()``. + A separate test-only seam may establish a **test-mode** native record for + unit tests of transport gates; that record cannot authorize production + Gitea mutation endpoints. Manual deletion of session-state files is never a recovery path. """ @@ -23,6 +29,7 @@ import inspect import os import secrets import time +from pathlib import Path from typing import Any SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON" @@ -34,6 +41,13 @@ FORCE_PROVENANCE_FAIL_ENV = "GITEA_TEST_FORCE_UNSANCTIONED" # Process-local native runtime (never persisted, never read from env alone). _NATIVE_RUNTIME: dict[str, Any] | None = None +# Production transport identifiers accepted by bind_native_mcp_transport. +_PRODUCTION_TRANSPORTS = frozenset({"stdio"}) +_RUNTIME_MODE_PRODUCTION = "production" +_RUNTIME_MODE_TEST = "test" +_PHASE_ENTRYPOINT_CLAIMED = "entrypoint_claimed" +_PHASE_TRANSPORT_BOUND = "transport_bound" + class UnsanctionedRuntimeError(RuntimeError): """Raised when mutation/credential code runs outside a native MCP daemon.""" @@ -53,44 +67,197 @@ def is_pytest_runtime() -> bool: return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip()) +def _package_root() -> Path: + """Directory that contains the canonical MCP entrypoint modules.""" + return Path(__file__).resolve().parent + + +def canonical_entrypoint_paths() -> frozenset[str]: + """Resolved absolute paths of official entrypoints (not basenames).""" + root = _package_root() + return frozenset( + { + str((root / "mcp_server.py").resolve()), + str((root / "gitea_mcp_server.py").resolve()), + } + ) + + +def _resolve_path(path: str | None) -> str | None: + if not path: + return None + try: + return str(Path(path).resolve()) + except (OSError, RuntimeError, ValueError): + return None + + +def _caller_official_entrypoint_path() -> str | None: + """Return the resolved canonical entrypoint path in the call stack, or None. + + Basename-only matches (e.g. an attacker file named ``mcp_server.py`` + elsewhere) are rejected. The path must equal one of + :func:`canonical_entrypoint_paths`. + """ + canonical = canonical_entrypoint_paths() + for frame in inspect.stack()[1:20]: + resolved = _resolve_path(frame.filename) + if resolved and resolved in canonical: + return resolved + return None + + def _caller_is_official_entrypoint() -> bool: - """True when mark_sanctioned_daemon is invoked from mcp_server.py.""" - for frame in inspect.stack()[1:12]: - path = (frame.filename or "").replace("\\", "/") - base = path.rsplit("/", 1)[-1] - if base == "mcp_server.py": - return True - return False + """True when invoked from a resolved canonical entrypoint path (#695).""" + return _caller_official_entrypoint_path() is not None -def mark_sanctioned_daemon(*, allow_test_bootstrap: bool = False) -> dict[str, Any]: - """Mark this process as the official native MCP daemon (#695). +def _new_runtime_token() -> tuple[str, str]: + token = secrets.token_hex(32) + fingerprint = hashlib.sha256(token.encode()).hexdigest()[:16] + return token, fingerprint - Only the official ``mcp_server.py`` entrypoint (or pytest test bootstrap) - may establish native transport. Setting env vars alone is insufficient. + +def mark_sanctioned_daemon() -> dict[str, Any]: + """Claim the official entrypoint for this process (#695). + + This alone does **not** authorize mutations. Callers must subsequently + bind the native MCP transport via :func:`bind_native_mcp_transport`. + + Only a stack frame whose **resolved absolute path** is the canonical + ``mcp_server.py`` or ``gitea_mcp_server.py`` next to this module may + claim the entrypoint. Basename spoofing is rejected. + + There is no public ``allow_test_bootstrap`` argument: caller-controlled + flags must never establish trusted mutation provenance. Hermetic tests + use :func:`install_test_native_runtime` (pytest-only, test mode). """ global _NATIVE_RUNTIME - if not is_pytest_runtime() and not allow_test_bootstrap: - if not _caller_is_official_entrypoint(): - raise UnsanctionedRuntimeError( - "mark_sanctioned_daemon rejected: not called from official " - "mcp_server.py entrypoint (#695). Offline import / standalone " - "scripts cannot reconstruct native transport. Stop after native " - "MCP failure; do not run offline mutation helpers." - ) - token = secrets.token_hex(32) + if is_pytest_runtime(): + # Under pytest, production mark is a no-op for transport authority. + # Tests that need a native-transport record use install_test_native_runtime. + return native_runtime_status() + + entrypoint_path = _caller_official_entrypoint_path() + if entrypoint_path is None: + raise UnsanctionedRuntimeError( + "mark_sanctioned_daemon rejected: not called from the resolved " + "canonical MCP entrypoint path (#695). Basename-only names " + "(e.g. a renamed runner called mcp_server.py) are insufficient. " + "Offline import / standalone scripts cannot reconstruct native " + "transport. Stop after native MCP failure; do not run offline " + "mutation helpers." + ) + + token, fingerprint = _new_runtime_token() _NATIVE_RUNTIME = { "token": token, - "token_fingerprint": hashlib.sha256(token.encode()).hexdigest()[:16], + "token_fingerprint": fingerprint, "pid": os.getpid(), "started_at": time.time(), "entrypoint": "mcp_server", + "entrypoint_path": entrypoint_path, + "phase": _PHASE_ENTRYPOINT_CLAIMED, + "transport": None, + "mode": _RUNTIME_MODE_PRODUCTION, } # Legacy signal for older probes; alone does not authorize mutations. os.environ[SANCTIONED_DAEMON_ENV] = "1" return native_runtime_status() +def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]: + """Bind the live native MCP transport lifecycle (#695). + + Must be called from the resolved canonical entrypoint immediately before + the real MCP server transport loop (e.g. ``mcp.run(transport=\"stdio\")``). + Requires a prior successful :func:`mark_sanctioned_daemon` claim in this + process. Import-only or offline launch without this bind leaves + :func:`is_native_mcp_transport` false. + """ + global _NATIVE_RUNTIME + transport_name = (transport or "").strip().lower() + if transport_name not in _PRODUCTION_TRANSPORTS: + raise UnsanctionedRuntimeError( + f"bind_native_mcp_transport rejected: transport {transport!r} is " + f"not a production MCP transport (#695). Allowed: " + f"{sorted(_PRODUCTION_TRANSPORTS)}." + ) + + entrypoint_path = _caller_official_entrypoint_path() + if entrypoint_path is None: + raise UnsanctionedRuntimeError( + "bind_native_mcp_transport rejected: not called from the resolved " + "canonical MCP entrypoint path (#695)." + ) + + if _NATIVE_RUNTIME is None or int(_NATIVE_RUNTIME.get("pid") or -1) != os.getpid(): + raise UnsanctionedRuntimeError( + "bind_native_mcp_transport rejected: no entrypoint claim in this " + "process (#695). Call mark_sanctioned_daemon() from the official " + "entrypoint first." + ) + + if _NATIVE_RUNTIME.get("mode") != _RUNTIME_MODE_PRODUCTION: + raise UnsanctionedRuntimeError( + "bind_native_mcp_transport rejected: runtime mode is not " + "production (#695)." + ) + + claimed = (_NATIVE_RUNTIME.get("entrypoint_path") or "").strip() + if claimed and claimed != entrypoint_path: + raise UnsanctionedRuntimeError( + "bind_native_mcp_transport rejected: entrypoint path mismatch " + "between mark and bind (#695)." + ) + + _NATIVE_RUNTIME["phase"] = _PHASE_TRANSPORT_BOUND + _NATIVE_RUNTIME["transport"] = transport_name + _NATIVE_RUNTIME["entrypoint_path"] = entrypoint_path + _NATIVE_RUNTIME["bound_at"] = time.time() + os.environ[SANCTIONED_DAEMON_ENV] = "1" + return native_runtime_status() + + +def install_test_native_runtime() -> dict[str, Any]: + """Pytest-only seam for hermetic native-transport unit tests (#695). + + Establishes a **test-mode** process-local record so unit tests can exercise + gates that require ``is_native_mcp_transport()``. This record: + + - is rejected outside pytest (including a fresh offline interpreter); + - never uses production mode; + - cannot authorize production Gitea mutation endpoints + (:func:`assert_production_mutation_runtime` / production path of + :func:`assert_sanctioned_mutation_runtime` when not under pytest). + + There is no public caller-controlled flag that forges production native + transport. + """ + global _NATIVE_RUNTIME + if not is_pytest_runtime(): + raise UnsanctionedRuntimeError( + "install_test_native_runtime rejected: test-mode native runtime " + "is only available under pytest (#695). allow_test_bootstrap and " + "similar caller-controlled flags do not exist and cannot authorize " + "a fresh offline interpreter." + ) + token, fingerprint = _new_runtime_token() + _NATIVE_RUNTIME = { + "token": token, + "token_fingerprint": fingerprint, + "pid": os.getpid(), + "started_at": time.time(), + "entrypoint": "test_bootstrap", + "entrypoint_path": None, + "phase": _PHASE_TRANSPORT_BOUND, + "transport": "test", + "mode": _RUNTIME_MODE_TEST, + "bound_at": time.time(), + } + return native_runtime_status() + + def clear_native_runtime_for_tests() -> None: """Test helper: drop native runtime (does not clear env).""" global _NATIVE_RUNTIME @@ -98,7 +265,7 @@ def clear_native_runtime_for_tests() -> None: def is_native_mcp_transport() -> bool: - """True when this process holds a live native MCP runtime record (#695).""" + """True when this process holds a transport-bound native runtime (#695).""" if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in { "1", "true", @@ -111,12 +278,23 @@ def is_native_mcp_transport() -> bool: return False if not (_NATIVE_RUNTIME.get("token") or "").strip(): return False + if _NATIVE_RUNTIME.get("phase") != _PHASE_TRANSPORT_BOUND: + return False + if not (_NATIVE_RUNTIME.get("transport") or "").strip(): + return False return True +def is_production_native_mcp_transport() -> bool: + """True only for production-mode, transport-bound native runtime.""" + if not is_native_mcp_transport(): + return False + return (_NATIVE_RUNTIME or {}).get("mode") == _RUNTIME_MODE_PRODUCTION + + def is_sanctioned_mcp_daemon() -> bool: """Backward-compatible name; #695 requires native transport, not env alone.""" - if is_native_mcp_transport(): + if is_production_native_mcp_transport(): return True if is_pytest_runtime(): return True @@ -126,10 +304,42 @@ def is_sanctioned_mcp_daemon() -> bool: return False -def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None: - """Fail closed when mutation code runs outside native MCP transport (#695).""" - if is_sanctioned_mcp_daemon(): +def assert_production_mutation_runtime(context: str = "mutation") -> None: + """Fail closed unless production native MCP transport is bound (#695). + + Test-mode bootstrap records and pytest-only hermetic allowances do **not** + satisfy this gate. Use for production Gitea mutation endpoints that must + never be reachable via test bootstrap. + """ + if is_production_native_mcp_transport(): return + mode = (_NATIVE_RUNTIME or {}).get("mode") + if mode == _RUNTIME_MODE_TEST: + raise UnsanctionedRuntimeError( + f"Test-mode native runtime cannot authorize production {context} " + "(#695). install_test_native_runtime / former allow_test_bootstrap " + "must never reach real Gitea mutation endpoints." + ) + assert_sanctioned_mutation_runtime(context) + + +def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None: + """Fail closed when mutation code runs outside native MCP transport (#695). + + Under pytest, hermetic unit tests are allowed (profile/permission tests). + Outside pytest, requires production-mode transport-bound native runtime. + Test-mode records do not authorize non-pytest production mutations. + """ + if is_pytest_runtime(): + return + if is_production_native_mcp_transport(): + return + mode = (_NATIVE_RUNTIME or {}).get("mode") + if mode == _RUNTIME_MODE_TEST: + raise UnsanctionedRuntimeError( + f"Test-mode native runtime cannot authorize production {context} " + "(#695). Test bootstrap cannot reach production mutation endpoints." + ) env_spoof = (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in { "1", "true", @@ -139,13 +349,24 @@ def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None: if env_spoof: extra = ( f" Note: {SANCTIONED_DAEMON_ENV} alone is not sufficient (#695); " - "native transport requires the official MCP entrypoint." + "native transport requires the official MCP entrypoint and a live " + "transport bind." + ) + phase = (_NATIVE_RUNTIME or {}).get("phase") + if phase == _PHASE_ENTRYPOINT_CLAIMED: + extra = ( + (extra + " ") if extra else " " + ) + ( + "Entrypoint was claimed but native MCP transport was never bound " + "(#695); offline launch/import of the real entrypoint does not " + "grant mutation authority." ) raise UnsanctionedRuntimeError( f"Unsanctioned / non-native runtime blocked {context} (#695). " "Do not import gitea_mcp_server or call mutation helpers from a raw " "shell, offline runner, or ad-hoc script after native MCP failure. " - "Stop and reconnect the official MCP daemon (mcp_server.py). " + "Stop and reconnect the official MCP daemon (mcp_server.py) over " + "native transport. " f"Do not set {ALLOW_DIRECT_IMPORT_ENV} or raw token env vars in LLM " f"sessions.{extra}" ) @@ -181,11 +402,16 @@ def native_runtime_status() -> dict[str, Any]: rt = _NATIVE_RUNTIME or {} return { "native_mcp_transport": is_native_mcp_transport(), + "production_native_mcp_transport": is_production_native_mcp_transport(), "pytest": is_pytest_runtime(), "pid": rt.get("pid"), "token_fingerprint": rt.get("token_fingerprint"), "started_at": rt.get("started_at"), "entrypoint": rt.get("entrypoint"), + "entrypoint_path": rt.get("entrypoint_path"), + "phase": rt.get("phase"), + "transport": rt.get("transport"), + "mode": rt.get("mode"), "env_sanctioned_alone_insufficient": True, "sanctioned_env": SANCTIONED_DAEMON_ENV, "allow_direct_import_env": ALLOW_DIRECT_IMPORT_ENV, @@ -203,10 +429,18 @@ def runtime_status() -> dict[str, Any]: def mutation_provenance_fields() -> dict[str, Any]: """Fields to attach to live mutation / review audit records (#695 AC6).""" st = native_runtime_status() + transport = "native_mcp" if st["native_mcp_transport"] else "untrusted" + if st.get("mode") == _RUNTIME_MODE_TEST and st["native_mcp_transport"]: + transport = "test_native_mcp" return { - "transport": "native_mcp" if st["native_mcp_transport"] else "untrusted", + "transport": transport, "native_mcp_transport": bool(st["native_mcp_transport"]), + "production_native_mcp_transport": bool( + st.get("production_native_mcp_transport") + ), "native_runtime_pid": st.get("pid"), "native_token_fingerprint": st.get("token_fingerprint"), "entrypoint": st.get("entrypoint"), + "phase": st.get("phase"), + "mode": st.get("mode"), } diff --git a/mcp_server.py b/mcp_server.py index f4ba642..02606dd 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -41,15 +41,17 @@ def check_conflict_markers(): check_conflict_markers() -# #558: official entrypoint marks the process as the sanctioned MCP daemon -# before loading mutation modules (blocks raw shell import bypasses). +# #558 / #695: claim the official entrypoint before loading mutation modules. +# This alone does NOT authorize mutations — gitea_mcp_server binds the live +# native MCP transport (stdio) immediately before mcp.run. Import-only or +# offline launch without that bind fails closed on mutations. try: import mcp_daemon_guard mcp_daemon_guard.mark_sanctioned_daemon() except Exception: # Guard import failures must not hide conflict-marker infra_stop above; - # gitea_mcp_server main also marks sanctioned when run as __main__. + # gitea_mcp_server main also marks + binds when run as __main__. pass # Execute the actual server logic via exec in this namespace. diff --git a/tests/test_issue_695_native_transport_quarantine.py b/tests/test_issue_695_native_transport_quarantine.py index 04d35bd..7b4ccac 100644 --- a/tests/test_issue_695_native_transport_quarantine.py +++ b/tests/test_issue_695_native_transport_quarantine.py @@ -1,14 +1,18 @@ """Regression tests for Issue #695 — second incident (PR #694 / review 427). Reproduces offline import, env-only runtime spoof, exposed-token invocation, -direct imports, locally generated runtime keys, standalone quarantine attempts, -and false “official workflow” canonical claims. Gates must fail closed. +direct imports, basename entrypoint spoof, allow_test_bootstrap forgery, +standalone quarantine attempts, and false “official workflow” canonical claims. +Gates must fail closed. """ from __future__ import annotations import os +import subprocess +import sys import tempfile +import textwrap import unittest from pathlib import Path from unittest.mock import patch @@ -20,6 +24,26 @@ import canonical_comment_validator as ccv HEAD_694 = "1844e298809373be19a526fd39b7d8b0669eb5bd" HEAD_OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" +REPO_ROOT = Path(__file__).resolve().parent.parent + + +def _run_offline_snippet(snippet: str, *, env_extra: dict[str, str] | None = None) -> subprocess.CompletedProcess: + """Execute snippet in a fresh interpreter (no pytest modules).""" + env = os.environ.copy() + env.pop("PYTEST_CURRENT_TEST", None) + env.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None) + env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "") + if env_extra: + env.update(env_extra) + return subprocess.run( + [sys.executable, "-c", snippet], + capture_output=True, + text=True, + cwd=str(REPO_ROOT), + env=env, + timeout=30, + check=False, + ) class TestNativeTransportBinding(unittest.TestCase): @@ -46,28 +70,37 @@ class TestNativeTransportBinding(unittest.TestCase): os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1" with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx: mcp_daemon_guard.mark_sanctioned_daemon() - self.assertIn("mcp_server.py", str(ctx.exception)) + self.assertIn("canonical", str(ctx.exception).lower()) self.assertFalse(mcp_daemon_guard.is_native_mcp_transport()) def test_locally_generated_runtime_key_without_entrypoint_rejected(self): """Spoofing process-local fields via mark outside entrypoint fails.""" mcp_daemon_guard.clear_native_runtime_for_tests() os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1" - # Even if a caller tries allow_test_bootstrap under force-unsanctioned - # pytest path is also forced off — only real entrypoint may mark. with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): - mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=False) + mcp_daemon_guard.mark_sanctioned_daemon() - def test_test_bootstrap_establishes_native_for_hermetic_tests(self): + def test_test_native_runtime_for_hermetic_tests_not_production(self): mcp_daemon_guard.clear_native_runtime_for_tests() - st = mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True) + st = mcp_daemon_guard.install_test_native_runtime() self.assertTrue(st["native_mcp_transport"]) self.assertTrue(mcp_daemon_guard.is_native_mcp_transport()) + self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport()) mcp_daemon_guard.assert_sanctioned_mutation_runtime("test-bootstrap") fields = mcp_daemon_guard.mutation_provenance_fields() - self.assertEqual(fields["transport"], "native_mcp") + self.assertEqual(fields["transport"], "test_native_mcp") self.assertTrue(fields["native_mcp_transport"]) + self.assertFalse(fields["production_native_mcp_transport"]) self.assertIsNotNone(fields["native_token_fingerprint"]) + with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx: + mcp_daemon_guard.assert_production_mutation_runtime("prod-endpoint") + self.assertIn("Test-mode", str(ctx.exception)) + + def test_no_allow_test_bootstrap_parameter_on_mark(self): + import inspect + + sig = inspect.signature(mcp_daemon_guard.mark_sanctioned_daemon) + self.assertNotIn("allow_test_bootstrap", sig.parameters) def test_exposed_token_env_never_grants_native(self): """Raw / exposed token env vars must never reconstruct native transport.""" @@ -96,6 +129,249 @@ class TestNativeTransportBinding(unittest.TestCase): os.environ.pop(key, None) +class TestAC9BypassRegressions(unittest.TestCase): + """AC9: empirically reproduced offline bypasses must fail closed (#695).""" + + def tearDown(self) -> None: + mcp_daemon_guard.clear_native_runtime_for_tests() + os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None) + + def test_allow_test_bootstrap_cannot_authorize_fresh_offline_interpreter(self): + """Finding 1: former allow_test_bootstrap forge is gone and rejected offline.""" + snippet = textwrap.dedent( + """ + import inspect + import mcp_daemon_guard as g + sig = inspect.signature(g.mark_sanctioned_daemon) + assert "allow_test_bootstrap" not in sig.parameters, "bootstrap flag must not exist" + try: + g.mark_sanctioned_daemon(allow_test_bootstrap=True) + except TypeError: + pass + else: + raise SystemExit("mark_sanctioned_daemon accepted allow_test_bootstrap") + try: + g.install_test_native_runtime() + except g.UnsanctionedRuntimeError as exc: + assert "pytest" in str(exc).lower() or "#695" in str(exc) + else: + raise SystemExit("install_test_native_runtime authorized offline interpreter") + assert g.is_native_mcp_transport() is False + try: + g.assert_sanctioned_mutation_runtime("offline-bootstrap") + except g.UnsanctionedRuntimeError: + pass + else: + raise SystemExit("mutation runtime authorized after offline bootstrap attempt") + print("OK") + """ + ) + proc = _run_offline_snippet(snippet) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertIn("OK", proc.stdout) + + def test_renamed_runner_named_mcp_server_py_rejected(self): + """Finding 2: basename-only entrypoint trust is insufficient.""" + with tempfile.TemporaryDirectory() as tmp: + attacker = Path(tmp) / "mcp_server.py" + attacker.write_text( + textwrap.dedent( + """ + import mcp_daemon_guard as g + try: + g.mark_sanctioned_daemon() + except g.UnsanctionedRuntimeError as exc: + print("REJECTED:" + str(exc)) + raise SystemExit(0) + print("AUTHORIZED") + raise SystemExit(1) + """ + ), + encoding="utf-8", + ) + env = os.environ.copy() + env.pop("PYTEST_CURRENT_TEST", None) + env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "") + proc = subprocess.run( + [sys.executable, str(attacker)], + capture_output=True, + text=True, + cwd=tmp, + env=env, + timeout=30, + check=False, + ) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertIn("REJECTED:", proc.stdout) + self.assertIn("canonical", proc.stdout.lower()) + self.assertNotIn("AUTHORIZED", proc.stdout) + + def test_direct_launch_import_canonical_entrypoint_without_transport_rejected(self): + """Merely importing/launching real entrypoint offline must not authorize.""" + snippet = textwrap.dedent( + f""" + import importlib.util + import mcp_daemon_guard as g + # Simulate claim-only phase (no transport bind). + g.clear_native_runtime_for_tests() + # Direct mark from non-entrypoint must fail. + try: + g.mark_sanctioned_daemon() + except g.UnsanctionedRuntimeError: + pass + assert g.is_native_mcp_transport() is False + # Even if someone forges entrypoint_claimed without transport bind: + g._NATIVE_RUNTIME = {{ + "token": "x" * 64, + "token_fingerprint": "deadbeefdeadbeef", + "pid": __import__("os").getpid(), + "started_at": 0, + "entrypoint": "mcp_server", + "entrypoint_path": {str(REPO_ROOT / "mcp_server.py")!r}, + "phase": "entrypoint_claimed", + "transport": None, + "mode": "production", + }} + assert g.is_native_mcp_transport() is False + try: + g.assert_sanctioned_mutation_runtime("import-only") + except g.UnsanctionedRuntimeError as exc: + assert "transport" in str(exc).lower() or "entrypoint" in str(exc).lower() or "#695" in str(exc) + else: + raise SystemExit("import-only claim authorized mutation") + print("OK") + """ + ) + proc = _run_offline_snippet(snippet) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertIn("OK", proc.stdout) + + def test_spoofed_pytest_env_stack_path_rejected(self): + """Spoofed pytest/env/call-stack/path evidence must not authorize offline.""" + snippet = textwrap.dedent( + f""" + import os + import mcp_daemon_guard as g + os.environ["PYTEST_CURRENT_TEST"] = "spoofed::test" + os.environ[g.SANCTIONED_DAEMON_ENV] = "1" + os.environ[g.ALLOW_DIRECT_IMPORT_ENV] = "1" + # Fresh interpreter has no pytest module; PYTEST_CURRENT_TEST alone + # might still trip is_pytest_runtime — force-unsanctioned is not set. + # But install_test_native_runtime requires real pytest path; if + # PYTEST_CURRENT_TEST alone grants is_pytest_runtime, production + # mutation still requires production transport outside true pytest. + if g.is_pytest_runtime(): + # Env-only pytest spoof: test install may succeed, but production + # mutation gate must still reject test mode. + g.install_test_native_runtime() + assert g.is_production_native_mcp_transport() is False + try: + g.assert_production_mutation_runtime("spoofed-pytest") + except g.UnsanctionedRuntimeError: + pass + else: + raise SystemExit("production mutation accepted test-mode under spoofed pytest env") + else: + try: + g.install_test_native_runtime() + except g.UnsanctionedRuntimeError: + pass + else: + raise SystemExit("test install without pytest evidence") + # Basename path spoof via inspect is covered elsewhere; env alone: + g.clear_native_runtime_for_tests() + os.environ.pop("PYTEST_CURRENT_TEST", None) + assert g.is_native_mcp_transport() is False + try: + g.assert_sanctioned_mutation_runtime("env-path-spoof") + except g.UnsanctionedRuntimeError: + pass + else: + raise SystemExit("env/path spoof authorized mutation") + print("OK") + """ + ) + proc = _run_offline_snippet(snippet) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertIn("OK", proc.stdout) + + def test_test_bootstrap_cannot_reach_production_mutation_endpoints(self): + """Under pytest, test-mode runtime cannot satisfy production mutation gate.""" + mcp_daemon_guard.clear_native_runtime_for_tests() + mcp_daemon_guard.install_test_native_runtime() + with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx: + mcp_daemon_guard.assert_production_mutation_runtime( + "gitea_quarantine_contaminated_review" + ) + msg = str(ctx.exception) + self.assertIn("Test-mode", msg) + self.assertIn("#695", msg) + + # Offline: install_test_native_runtime must not authorize production mutations. + snippet = textwrap.dedent( + """ + import mcp_daemon_guard as g + try: + g.install_test_native_runtime() + except g.UnsanctionedRuntimeError: + pass + assert g.is_production_native_mcp_transport() is False + try: + g.assert_production_mutation_runtime("gitea_quarantine_contaminated_review") + except g.UnsanctionedRuntimeError: + pass + else: + raise SystemExit("production mutation authorized offline") + try: + g.assert_sanctioned_mutation_runtime("gitea_mutation") + except g.UnsanctionedRuntimeError: + pass + else: + raise SystemExit("sanctioned mutation authorized offline") + print("OK") + """ + ) + proc = _run_offline_snippet(snippet) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertIn("OK", proc.stdout) + + def test_legitimate_native_transport_bind_succeeds(self): + """Canonical entrypoint path + stdio bind establishes production native.""" + mcp_daemon_guard.clear_native_runtime_for_tests() + # Simulate production path under force-unsanctioned (no pytest allowance) + # by calling internal claim/bind with patched caller path. + canonical = str((REPO_ROOT / "mcp_server.py").resolve()) + + def _fake_caller(): + return canonical + + with patch.object( + mcp_daemon_guard, "_caller_official_entrypoint_path", side_effect=_fake_caller + ): + # Force non-pytest path for mark/bind logic. + with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False): + st1 = mcp_daemon_guard.mark_sanctioned_daemon() + self.assertFalse(st1["native_mcp_transport"]) + self.assertEqual(st1["phase"], "entrypoint_claimed") + st2 = mcp_daemon_guard.bind_native_mcp_transport(transport="stdio") + self.assertTrue(st2["native_mcp_transport"]) + self.assertTrue(st2["production_native_mcp_transport"]) + self.assertEqual(st2["transport"], "stdio") + self.assertEqual(st2["mode"], "production") + mcp_daemon_guard.assert_sanctioned_mutation_runtime("native-ide") + mcp_daemon_guard.assert_production_mutation_runtime("native-ide") + fields = mcp_daemon_guard.mutation_provenance_fields() + self.assertEqual(fields["transport"], "native_mcp") + self.assertTrue(fields["production_native_mcp_transport"]) + + def test_canonical_entrypoint_paths_are_resolved_absolute(self): + paths = mcp_daemon_guard.canonical_entrypoint_paths() + self.assertTrue(any(p.endswith("mcp_server.py") for p in paths)) + for p in paths: + self.assertTrue(os.path.isabs(p), p) + self.assertEqual(p, str(Path(p).resolve())) + + class TestQuarantineWriteNativeOnly(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() @@ -133,7 +409,12 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase): forensic_comment_ids=[10883, 10886], ) with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): - review_quarantine.write_quarantine_record(record) + # Force non-pytest and non-native for write path. + with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False): + with patch.object( + mcp_daemon_guard, "is_native_mcp_transport", return_value=False + ): + review_quarantine.write_quarantine_record(record) def test_confirmation_must_match_exactly(self): assessment = review_quarantine.assess_quarantine_write( @@ -170,7 +451,7 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase): "review_quarantine.mcp_session_state.default_state_dir", return_value=self._tmp.name, ): - mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True) + mcp_daemon_guard.install_test_native_runtime() record = review_quarantine.build_quarantine_record( remote="prgs", org="org", @@ -305,7 +586,7 @@ class TestFeedbackQuarantineIntegration(unittest.TestCase): self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) mcp_daemon_guard.clear_native_runtime_for_tests() - mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True) + mcp_daemon_guard.install_test_native_runtime() def tearDown(self) -> None: mcp_daemon_guard.clear_native_runtime_for_tests() diff --git a/tests/test_mcp_daemon_guard.py b/tests/test_mcp_daemon_guard.py index 8e589b8..aa25d66 100644 --- a/tests/test_mcp_daemon_guard.py +++ b/tests/test_mcp_daemon_guard.py @@ -30,11 +30,17 @@ class TestMcpDaemonGuard(unittest.TestCase): # Running under pytest already sets PYTEST_CURRENT_TEST. mcp_daemon_guard.assert_sanctioned_mutation_runtime("pytest") - def test_mark_sanctioned_bootstrap_allows(self): + def test_test_native_runtime_install_under_pytest(self): mcp_daemon_guard.clear_native_runtime_for_tests() - mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True) - mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon") + 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()) + # Hermetic unit path still passes under pytest. + mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon") + # Production mutation gate rejects test-mode records. + with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx: + mcp_daemon_guard.assert_production_mutation_runtime("gitea_mutation") + self.assertIn("Test-mode", str(ctx.exception)) def test_env_alone_insufficient_when_force_unsanctioned(self): mcp_daemon_guard.clear_native_runtime_for_tests() @@ -87,6 +93,11 @@ class TestMcpDaemonGuard(unittest.TestCase): with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): gitea_auth.get_auth_header("gitea.prgs.cc") + def test_no_allow_test_bootstrap_public_parameter(self): + """Production mark must not accept allow_test_bootstrap (#695).""" + sig = __import__("inspect").signature(mcp_daemon_guard.mark_sanctioned_daemon) + self.assertNotIn("allow_test_bootstrap", sig.parameters) + if __name__ == "__main__": unittest.main()