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

fix(guard): close allow_test_bootstrap and basename entrypoint bypasses (#695)

Browse Source 
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:
sysadmin
2026-07-13 09:42:47 -04:00
parent 253269c61b
commit da6fd233f4
6 changed files with 609 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_issue_695_native_transport_quarantine.py
+293 -12
View File
@@ -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()
tests/test_mcp_daemon_guard.py
+14 -3
View File
@@ -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()