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.
This commit is contained in:
2026-07-13 09:42:47 -04:00
parent 253269c61b
commit da6fd233f4
6 changed files with 609 additions and 64 deletions
+21 -7
View File
@@ -9,22 +9,36 @@ formal reviews then looked identical to native approvals.
## Rule ## Rule
Mutation auth, keychain fill, and controller quarantine require a **native MCP Mutation auth, keychain fill, and controller quarantine require a **production
transport runtime** established only by the official entrypoint. 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 | | Context | Allowed |
|---------|---------| |---------|---------|
| Official MCP entrypoint (`mcp_server.py` / `gitea_mcp_server` `__main__`) calls `mark_sanctioned_daemon()` and holds a process-local runtime token | yes | | Official IDE-native MCP: resolved canonical entrypoint marks + binds stdio, holds process-local runtime token | yes |
| pytest (hermetic unit tests) | 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_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_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 | | `GITEA_ALLOW_KEYCHAIN_CLI=1` in LLM sessions | **no** — human operator only |
| bare `python -c 'import gitea_mcp_server; …'` or offline runners | **no** | | bare `python -c 'import gitea_mcp_server; …'` or offline runners | **no** |
| keychain fill outside native/pytest | **no** | | keychain fill outside native/pytest | **no** |
Native runtime is **process-local**: a random token bound to the daemon PID. Native runtime is **process-local**: a random token bound to the daemon PID and
It is never reconstructed from environment variables, session-state files, or transport phase. It is never reconstructed from environment variables,
importing internals in a fresh Python process. session-state files, caller-controlled flags, or importing internals in a
fresh Python process.
## Contaminated review quarantine (#695 AC8) ## Contaminated review quarantine (#695 AC8)
+8 -5
View File
@@ -12843,9 +12843,10 @@ def gitea_quarantine_contaminated_review(
427 until an independent adversarial reviewer and controller deployment 427 until an independent adversarial reviewer and controller deployment
of this tooling have completed. 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: try:
mcp_daemon_guard.assert_sanctioned_mutation_runtime( mcp_daemon_guard.assert_production_mutation_runtime(
"gitea_quarantine_contaminated_review" "gitea_quarantine_contaminated_review"
) )
except mcp_daemon_guard.UnsanctionedRuntimeError as exc: except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
@@ -13051,10 +13052,12 @@ def gitea_quarantine_contaminated_review(
# ── Entry point ─────────────────────────────────────────────────────────────── # ── Entry point ───────────────────────────────────────────────────────────────
if __name__ == "__main__": if __name__ == "__main__":
# #558 / #695: mark this process as the official native MCP daemon before # #558 / #695: claim the resolved canonical entrypoint, then bind the live
# any tool dispatch. Env vars alone cannot reconstruct native transport; # native MCP transport lifecycle before any tool dispatch. Env vars,
# offline imports / standalone scripts fail closed on mutations. # 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.mark_sanctioned_daemon()
mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
# Lock this session's launch profile into the environment so child CLI # Lock this session's launch profile into the environment so child CLI
# processes (e.g. review_pr.py) can detect and refuse profile # processes (e.g. review_pr.py) can detect and refuse profile
# side-channel overrides (#199). # side-channel overrides (#199).
+268 -34
View File
@@ -6,12 +6,18 @@ Direct ``import gitea_mcp_server`` from a shell bypasses native MCP transport.
- Environment variables alone cannot reconstruct a native session - Environment variables alone cannot reconstruct a native session
(``GITEA_MCP_SANCTIONED_DAEMON=1`` / ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` are (``GITEA_MCP_SANCTIONED_DAEMON=1`` / ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` are
insufficient for mutation gates). insufficient for mutation gates).
- A process-local runtime record is created only by the official entrypoint - A process-local runtime record is established only by the resolved canonical
(``mcp_server.py`` calling ``mark_sanctioned_daemon``), bound to PID and a entrypoint path (not basename) **and** the actual native MCP transport
random secret that never leaves process memory. 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. - Offline scripts that import internals fail closed on mutations.
- Pytest remains allowed (hermetic tests); optional force flags exist for - Pytest remains allowed for hermetic unit tests via ``is_pytest_runtime()``.
provenance regression tests. 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. Manual deletion of session-state files is never a recovery path.
""" """
@@ -23,6 +29,7 @@ import inspect
import os import os
import secrets import secrets
import time import time
from pathlib import Path
from typing import Any from typing import Any
SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON" 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). # Process-local native runtime (never persisted, never read from env alone).
_NATIVE_RUNTIME: dict[str, Any] | None = None _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): class UnsanctionedRuntimeError(RuntimeError):
"""Raised when mutation/credential code runs outside a native MCP daemon.""" """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()) 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: def _caller_is_official_entrypoint() -> bool:
"""True when mark_sanctioned_daemon is invoked from mcp_server.py.""" """True when invoked from a resolved canonical entrypoint path (#695)."""
for frame in inspect.stack()[1:12]: return _caller_official_entrypoint_path() is not None
path = (frame.filename or "").replace("\\", "/")
base = path.rsplit("/", 1)[-1]
if base == "mcp_server.py":
return True
return False
def mark_sanctioned_daemon(*, allow_test_bootstrap: bool = False) -> dict[str, Any]: def _new_runtime_token() -> tuple[str, str]:
"""Mark this process as the official native MCP daemon (#695). 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 global _NATIVE_RUNTIME
if not is_pytest_runtime() and not allow_test_bootstrap: if is_pytest_runtime():
if not _caller_is_official_entrypoint(): # Under pytest, production mark is a no-op for transport authority.
raise UnsanctionedRuntimeError( # Tests that need a native-transport record use install_test_native_runtime.
"mark_sanctioned_daemon rejected: not called from official " return native_runtime_status()
"mcp_server.py entrypoint (#695). Offline import / standalone "
"scripts cannot reconstruct native transport. Stop after native " entrypoint_path = _caller_official_entrypoint_path()
"MCP failure; do not run offline mutation helpers." if entrypoint_path is None:
) raise UnsanctionedRuntimeError(
token = secrets.token_hex(32) "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 = { _NATIVE_RUNTIME = {
"token": token, "token": token,
"token_fingerprint": hashlib.sha256(token.encode()).hexdigest()[:16], "token_fingerprint": fingerprint,
"pid": os.getpid(), "pid": os.getpid(),
"started_at": time.time(), "started_at": time.time(),
"entrypoint": "mcp_server", "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. # Legacy signal for older probes; alone does not authorize mutations.
os.environ[SANCTIONED_DAEMON_ENV] = "1" os.environ[SANCTIONED_DAEMON_ENV] = "1"
return native_runtime_status() 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: def clear_native_runtime_for_tests() -> None:
"""Test helper: drop native runtime (does not clear env).""" """Test helper: drop native runtime (does not clear env)."""
global _NATIVE_RUNTIME global _NATIVE_RUNTIME
@@ -98,7 +265,7 @@ def clear_native_runtime_for_tests() -> None:
def is_native_mcp_transport() -> bool: 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 { if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
"1", "1",
"true", "true",
@@ -111,12 +278,23 @@ def is_native_mcp_transport() -> bool:
return False return False
if not (_NATIVE_RUNTIME.get("token") or "").strip(): if not (_NATIVE_RUNTIME.get("token") or "").strip():
return False return False
if _NATIVE_RUNTIME.get("phase") != _PHASE_TRANSPORT_BOUND:
return False
if not (_NATIVE_RUNTIME.get("transport") or "").strip():
return False
return True 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: def is_sanctioned_mcp_daemon() -> bool:
"""Backward-compatible name; #695 requires native transport, not env alone.""" """Backward-compatible name; #695 requires native transport, not env alone."""
if is_native_mcp_transport(): if is_production_native_mcp_transport():
return True return True
if is_pytest_runtime(): if is_pytest_runtime():
return True return True
@@ -126,10 +304,42 @@ def is_sanctioned_mcp_daemon() -> bool:
return False return False
def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None: def assert_production_mutation_runtime(context: str = "mutation") -> None:
"""Fail closed when mutation code runs outside native MCP transport (#695).""" """Fail closed unless production native MCP transport is bound (#695).
if is_sanctioned_mcp_daemon():
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 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 { env_spoof = (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in {
"1", "1",
"true", "true",
@@ -139,13 +349,24 @@ def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
if env_spoof: if env_spoof:
extra = ( extra = (
f" Note: {SANCTIONED_DAEMON_ENV} alone is not sufficient (#695); " 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( raise UnsanctionedRuntimeError(
f"Unsanctioned / non-native runtime blocked {context} (#695). " f"Unsanctioned / non-native runtime blocked {context} (#695). "
"Do not import gitea_mcp_server or call mutation helpers from a raw " "Do not import gitea_mcp_server or call mutation helpers from a raw "
"shell, offline runner, or ad-hoc script after native MCP failure. " "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"Do not set {ALLOW_DIRECT_IMPORT_ENV} or raw token env vars in LLM "
f"sessions.{extra}" f"sessions.{extra}"
) )
@@ -181,11 +402,16 @@ def native_runtime_status() -> dict[str, Any]:
rt = _NATIVE_RUNTIME or {} rt = _NATIVE_RUNTIME or {}
return { return {
"native_mcp_transport": is_native_mcp_transport(), "native_mcp_transport": is_native_mcp_transport(),
"production_native_mcp_transport": is_production_native_mcp_transport(),
"pytest": is_pytest_runtime(), "pytest": is_pytest_runtime(),
"pid": rt.get("pid"), "pid": rt.get("pid"),
"token_fingerprint": rt.get("token_fingerprint"), "token_fingerprint": rt.get("token_fingerprint"),
"started_at": rt.get("started_at"), "started_at": rt.get("started_at"),
"entrypoint": rt.get("entrypoint"), "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, "env_sanctioned_alone_insufficient": True,
"sanctioned_env": SANCTIONED_DAEMON_ENV, "sanctioned_env": SANCTIONED_DAEMON_ENV,
"allow_direct_import_env": ALLOW_DIRECT_IMPORT_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]: def mutation_provenance_fields() -> dict[str, Any]:
"""Fields to attach to live mutation / review audit records (#695 AC6).""" """Fields to attach to live mutation / review audit records (#695 AC6)."""
st = native_runtime_status() 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 { return {
"transport": "native_mcp" if st["native_mcp_transport"] else "untrusted", "transport": transport,
"native_mcp_transport": bool(st["native_mcp_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_runtime_pid": st.get("pid"),
"native_token_fingerprint": st.get("token_fingerprint"), "native_token_fingerprint": st.get("token_fingerprint"),
"entrypoint": st.get("entrypoint"), "entrypoint": st.get("entrypoint"),
"phase": st.get("phase"),
"mode": st.get("mode"),
} }
+5 -3
View File
@@ -41,15 +41,17 @@ def check_conflict_markers():
check_conflict_markers() check_conflict_markers()
# #558: official entrypoint marks the process as the sanctioned MCP daemon # #558 / #695: claim the official entrypoint before loading mutation modules.
# before loading mutation modules (blocks raw shell import bypasses). # 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: try:
import mcp_daemon_guard import mcp_daemon_guard
mcp_daemon_guard.mark_sanctioned_daemon() mcp_daemon_guard.mark_sanctioned_daemon()
except Exception: except Exception:
# Guard import failures must not hide conflict-marker infra_stop above; # 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 pass
# Execute the actual server logic via exec in this namespace. # Execute the actual server logic via exec in this namespace.
@@ -1,14 +1,18 @@
"""Regression tests for Issue #695 — second incident (PR #694 / review 427). """Regression tests for Issue #695 — second incident (PR #694 / review 427).
Reproduces offline import, env-only runtime spoof, exposed-token invocation, Reproduces offline import, env-only runtime spoof, exposed-token invocation,
direct imports, locally generated runtime keys, standalone quarantine attempts, direct imports, basename entrypoint spoof, allow_test_bootstrap forgery,
and false “official workflow” canonical claims. Gates must fail closed. standalone quarantine attempts, and false “official workflow” canonical claims.
Gates must fail closed.
""" """
from __future__ import annotations from __future__ import annotations
import os import os
import subprocess
import sys
import tempfile import tempfile
import textwrap
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import patch from unittest.mock import patch
@@ -20,6 +24,26 @@ import canonical_comment_validator as ccv
HEAD_694 = "1844e298809373be19a526fd39b7d8b0669eb5bd" HEAD_694 = "1844e298809373be19a526fd39b7d8b0669eb5bd"
HEAD_OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" 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): class TestNativeTransportBinding(unittest.TestCase):
@@ -46,28 +70,37 @@ class TestNativeTransportBinding(unittest.TestCase):
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1" os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx: with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
mcp_daemon_guard.mark_sanctioned_daemon() 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()) self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
def test_locally_generated_runtime_key_without_entrypoint_rejected(self): def test_locally_generated_runtime_key_without_entrypoint_rejected(self):
"""Spoofing process-local fields via mark outside entrypoint fails.""" """Spoofing process-local fields via mark outside entrypoint fails."""
mcp_daemon_guard.clear_native_runtime_for_tests() mcp_daemon_guard.clear_native_runtime_for_tests()
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1" 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): 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() 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(st["native_mcp_transport"])
self.assertTrue(mcp_daemon_guard.is_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") mcp_daemon_guard.assert_sanctioned_mutation_runtime("test-bootstrap")
fields = mcp_daemon_guard.mutation_provenance_fields() 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.assertTrue(fields["native_mcp_transport"])
self.assertFalse(fields["production_native_mcp_transport"])
self.assertIsNotNone(fields["native_token_fingerprint"]) 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): def test_exposed_token_env_never_grants_native(self):
"""Raw / exposed token env vars must never reconstruct native transport.""" """Raw / exposed token env vars must never reconstruct native transport."""
@@ -96,6 +129,249 @@ class TestNativeTransportBinding(unittest.TestCase):
os.environ.pop(key, None) 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): class TestQuarantineWriteNativeOnly(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory() self._tmp = tempfile.TemporaryDirectory()
@@ -133,7 +409,12 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase):
forensic_comment_ids=[10883, 10886], forensic_comment_ids=[10883, 10886],
) )
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): 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): def test_confirmation_must_match_exactly(self):
assessment = review_quarantine.assess_quarantine_write( assessment = review_quarantine.assess_quarantine_write(
@@ -170,7 +451,7 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase):
"review_quarantine.mcp_session_state.default_state_dir", "review_quarantine.mcp_session_state.default_state_dir",
return_value=self._tmp.name, 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( record = review_quarantine.build_quarantine_record(
remote="prgs", remote="prgs",
org="org", org="org",
@@ -305,7 +586,7 @@ class TestFeedbackQuarantineIntegration(unittest.TestCase):
self._tmp = tempfile.TemporaryDirectory() self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup) self.addCleanup(self._tmp.cleanup)
mcp_daemon_guard.clear_native_runtime_for_tests() 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: def tearDown(self) -> None:
mcp_daemon_guard.clear_native_runtime_for_tests() mcp_daemon_guard.clear_native_runtime_for_tests()
+14 -3
View File
@@ -30,11 +30,17 @@ class TestMcpDaemonGuard(unittest.TestCase):
# Running under pytest already sets PYTEST_CURRENT_TEST. # Running under pytest already sets PYTEST_CURRENT_TEST.
mcp_daemon_guard.assert_sanctioned_mutation_runtime("pytest") 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.clear_native_runtime_for_tests()
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True) mcp_daemon_guard.install_test_native_runtime()
mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon")
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport()) 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): def test_env_alone_insufficient_when_force_unsanctioned(self):
mcp_daemon_guard.clear_native_runtime_for_tests() mcp_daemon_guard.clear_native_runtime_for_tests()
@@ -87,6 +93,11 @@ class TestMcpDaemonGuard(unittest.TestCase):
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
gitea_auth.get_auth_header("gitea.prgs.cc") 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__": if __name__ == "__main__":
unittest.main() unittest.main()