From 2e17a5c329da80e4a110c8033c910d5790ec51cd Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 13 Jul 2026 21:48:11 -0400 Subject: [PATCH] fix(guard): pin session-state authority and reject direct-import mutation (#695) Close remaining AC1/AC2/AC6-8 gaps after REQUEST_CHANGES and the PR #701 recurrence: GITEA_ALLOW_DIRECT_MCP_IMPORT never authorizes mutations; production transport bind pins GITEA_MCP_SESSION_STATE_DIR so redirected dirs cannot forge independent decision locks; mark/submit fail closed offline; quarantine continues to void contaminated merge eligibility. Regression tests reproduce the PR #701 direct-import + state-dir override sequence. Full suite: 2665 passed, 6 skipped. Closes #695 (remediation on PR #696) --- docs/mcp-daemon-import-guard.md | 3 +- gitea_mcp_server.py | 31 ++++ mcp_daemon_guard.py | 73 +++++++- mcp_session_state.py | 50 ++++++ tests/conftest.py | 11 ++ ...t_issue_695_native_transport_quarantine.py | 169 +++++++++++++++++- 6 files changed, 334 insertions(+), 3 deletions(-) diff --git a/docs/mcp-daemon-import-guard.md b/docs/mcp-daemon-import-guard.md index bebd141..88e3c2f 100644 --- a/docs/mcp-daemon-import-guard.md +++ b/docs/mcp-daemon-import-guard.md @@ -30,7 +30,8 @@ frame spoofing, or import-only launch are insufficient. | 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_DIRECT_MCP_IMPORT=1` in LLM sessions | **no** — never set; never authorizes mutations (#695 AC1 / PR #701) | +| Override `GITEA_MCP_SESSION_STATE_DIR` mid-session | **no** — production bind pins state root; redirect cannot forge independent decision locks (#695 AC2 / PR #701) | | `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** | diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 854ce4d..c433a1d 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -3305,6 +3305,18 @@ def _save_review_decision_lock(data): payload["profile_identity"] = binding["profile_identity"] if binding.get("remote") and not payload.get("remote"): payload["remote"] = binding["remote"] + # #695 AC6: stamp native transport provenance on durable decision locks. + try: + payload.update( + { + k: v + for k, v in mcp_daemon_guard.mutation_provenance_fields().items() + if v is not None + } + ) + except Exception: + payload.setdefault("transport", "untrusted") + payload.setdefault("native_mcp_transport", False) persisted = mcp_session_state.save_state( kind=mcp_session_state.KIND_DECISION_LOCK, payload=payload, @@ -4041,6 +4053,15 @@ def _evaluate_pr_review_submission( reasons.extend(review_workflow_load.recovery_handoff_without_replay()) return result if live: + # #695 AC1/AC2: offline direct-import submit (PR #701 run_submit.py) fails closed. + try: + mcp_daemon_guard.assert_sanctioned_mutation_runtime( + "gitea_submit_pr_review" + ) + mcp_daemon_guard.assert_no_direct_import_bypass("gitea_submit_pr_review") + except mcp_daemon_guard.UnsanctionedRuntimeError as exc: + reasons.append(str(exc)) + return result ns_gate = _live_namespace_health_gate("review_pr") if ns_gate: reasons.extend(ns_gate) @@ -4228,6 +4249,16 @@ def gitea_mark_final_review_decision( repo: str | None = None, ) -> dict: """Mark validation complete; the final review decision is ready to submit.""" + # #695 AC1/AC2: direct import / redirected session state cannot mark final. + try: + mcp_daemon_guard.assert_sanctioned_mutation_runtime( + "gitea_mark_final_review_decision" + ) + mcp_daemon_guard.assert_no_direct_import_bypass( + "gitea_mark_final_review_decision" + ) + except mcp_daemon_guard.UnsanctionedRuntimeError as exc: + return {"marked_ready": False, "reasons": [str(exc)]} action = (action or "").strip().lower() lock = _load_review_decision_lock() if lock is None: diff --git a/mcp_daemon_guard.py b/mcp_daemon_guard.py index 11675bb..b3d0ee8 100644 --- a/mcp_daemon_guard.py +++ b/mcp_daemon_guard.py @@ -48,6 +48,11 @@ _RUNTIME_MODE_TEST = "test" _PHASE_ENTRYPOINT_CLAIMED = "entrypoint_claimed" _PHASE_TRANSPORT_BOUND = "transport_bound" +# Default session-state root (mirrors mcp_session_state; kept local to avoid +# import cycles). Used only to pin authority at transport bind (#695 AC2). +_DEFAULT_SESSION_STATE_DIR = os.path.expanduser("~/.cache/gitea-tools/session-state") +SESSION_STATE_DIR_ENV = "GITEA_MCP_SESSION_STATE_DIR" + class UnsanctionedRuntimeError(RuntimeError): """Raised when mutation/credential code runs outside a native MCP daemon.""" @@ -211,10 +216,22 @@ def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]: "between mark and bind (#695)." ) + # Pin session-state root for this server lifetime (#695 AC2 / PR #701). + # Changing GITEA_MCP_SESSION_STATE_DIR after bind must not manufacture a + # second authority domain for decision locks / workflow proofs. + raw_state = (os.environ.get(SESSION_STATE_DIR_ENV) or "").strip() + if not raw_state: + raw_state = _DEFAULT_SESSION_STATE_DIR + try: + pinned_state = str(Path(raw_state).resolve()) + except (OSError, RuntimeError, ValueError): + pinned_state = raw_state + _NATIVE_RUNTIME["phase"] = _PHASE_TRANSPORT_BOUND _NATIVE_RUNTIME["transport"] = transport_name _NATIVE_RUNTIME["entrypoint_path"] = entrypoint_path _NATIVE_RUNTIME["bound_at"] = time.time() + _NATIVE_RUNTIME["session_state_dir"] = pinned_state os.environ[SANCTIONED_DAEMON_ENV] = "1" return native_runtime_status() @@ -264,6 +281,50 @@ def clear_native_runtime_for_tests() -> None: _NATIVE_RUNTIME = None +def pinned_session_state_dir() -> str | None: + """Session-state root pinned for this production transport lifetime (#695 AC2). + + When production native transport is bound, durable session proofs must use + this directory only. Env overrides of ``GITEA_MCP_SESSION_STATE_DIR`` after + bind are ignored so redirected dirs (e.g. ``.mcp_session_701``) cannot + manufacture independent decision-lock authority (PR #701 recurrence). + """ + if not is_production_native_mcp_transport(): + return None + pinned = (_NATIVE_RUNTIME or {}).get("session_state_dir") + text = (str(pinned) if pinned is not None else "").strip() + return text or None + + +def direct_import_env_enabled() -> bool: + """True when the legacy direct-import opt-in env is set (never authorizes).""" + return (os.environ.get(ALLOW_DIRECT_IMPORT_ENV) or "").strip().lower() in { + "1", + "true", + "yes", + } + + +def assert_no_direct_import_bypass(context: str = "mutation") -> None: + """Fail closed when GITEA_ALLOW_DIRECT_MCP_IMPORT is used for mutations (#695 AC1). + + The env flag is never a sanctioned recovery path for LLM/agent sessions. + Under pytest hermetic tests this is a no-op so unit tests can set the flag + to prove it does not grant authority. + """ + if is_pytest_runtime(): + return + if not direct_import_env_enabled(): + return + raise UnsanctionedRuntimeError( + f"{ALLOW_DIRECT_IMPORT_ENV} does not authorize {context} (#695 AC1). " + "Direct import of gitea_mcp_server mutation tools is forbidden. " + "Stop after native MCP failure; reconnect the official MCP daemon. " + "Do not set direct-import flags, offline runners, or redirected " + f"{SESSION_STATE_DIR_ENV} directories to reconstruct gates." + ) + + def is_native_mcp_transport() -> bool: """True when this process holds a transport-bound native runtime (#695).""" if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in { @@ -329,9 +390,12 @@ def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None: 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. + ``GITEA_ALLOW_DIRECT_MCP_IMPORT`` never authorizes mutations (#695 AC1). """ if is_pytest_runtime(): return + # AC1: direct-import env is never a mutation recovery path (PR #701). + assert_no_direct_import_bypass(context) if is_production_native_mcp_transport(): return mode = (_NATIVE_RUNTIME or {}).get("mode") @@ -367,7 +431,8 @@ def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None: "shell, offline runner, or ad-hoc script after native MCP failure. " "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}, override " + f"{SESSION_STATE_DIR_ENV}, or use raw token env vars in LLM " f"sessions.{extra}" ) @@ -412,9 +477,13 @@ def native_runtime_status() -> dict[str, Any]: "phase": rt.get("phase"), "transport": rt.get("transport"), "mode": rt.get("mode"), + "session_state_dir": pinned_session_state_dir() or rt.get("session_state_dir"), + "session_state_dir_pinned": pinned_session_state_dir() is not None, + "direct_import_env_set": direct_import_env_enabled(), "env_sanctioned_alone_insufficient": True, "sanctioned_env": SANCTIONED_DAEMON_ENV, "allow_direct_import_env": ALLOW_DIRECT_IMPORT_ENV, + "session_state_dir_env": SESSION_STATE_DIR_ENV, "allow_keychain_cli_env": ALLOW_KEYCHAIN_CLI_ENV, } @@ -443,4 +512,6 @@ def mutation_provenance_fields() -> dict[str, Any]: "entrypoint": st.get("entrypoint"), "phase": st.get("phase"), "mode": st.get("mode"), + "session_state_dir": st.get("session_state_dir"), + "session_state_dir_pinned": bool(st.get("session_state_dir_pinned")), } diff --git a/mcp_session_state.py b/mcp_session_state.py index d7192ea..b94b90a 100644 --- a/mcp_session_state.py +++ b/mcp_session_state.py @@ -44,6 +44,34 @@ SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK" def default_state_dir() -> str: + """Resolve the durable session-state root. + + When production native MCP transport is bound (#695 AC2), the directory + pinned at transport bind is authoritative: later overrides of + ``GITEA_MCP_SESSION_STATE_DIR`` cannot manufacture a second authority + domain (PR #701 recurrence: ``.mcp_session_701`` evasion of cross-PR + decision locks). + """ + try: + import mcp_daemon_guard + + pinned = mcp_daemon_guard.pinned_session_state_dir() + if pinned: + return pinned + except Exception: + # Fail open to env/default only when guard is unavailable (e.g. partial + # import during bootstrap). Mutation gates still fail closed separately. + pass + raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip() + return raw or DEFAULT_STATE_DIR + + +def env_session_state_dir_unpinned() -> str: + """Raw env/default session-state dir ignoring production transport pin. + + Intended for diagnostics and tests that assert pin behavior — not for + mutation-sensitive durable proofs under a bound native daemon. + """ raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip() return raw or DEFAULT_STATE_DIR @@ -362,6 +390,26 @@ def save_state( body["org"] = key_org if key_repo is not None: body["repo"] = key_repo + # Stamp session-state authority used for this write (#695 AC2 / AC6). + body.setdefault("session_state_dir", root) + try: + import mcp_daemon_guard + + prov = mcp_daemon_guard.mutation_provenance_fields() + body.setdefault( + "native_token_fingerprint", prov.get("native_token_fingerprint") + ) + body.setdefault( + "native_mcp_transport", bool(prov.get("native_mcp_transport")) + ) + body.setdefault( + "production_native_mcp_transport", + bool(prov.get("production_native_mcp_transport")), + ) + body.setdefault("transport", prov.get("transport")) + except Exception: + body.setdefault("native_mcp_transport", False) + body.setdefault("transport", "untrusted") envelope = { "kind": kind, @@ -373,6 +421,8 @@ def save_state( "recorded_at": body["recorded_at"], "updated_at": body["updated_at"], "writer_pid": body["writer_pid"], + "session_state_dir": body.get("session_state_dir"), + "transport": body.get("transport"), "payload": body, } _write_json(path, envelope) diff --git a/tests/conftest.py b/tests/conftest.py index 40ece4d..447f74a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -58,6 +58,17 @@ def _reset_mutation_authority(monkeypatch): _fallback: str = state_dir, _env_key: str = mcp_session_state.STATE_DIR_ENV, ) -> str: + # #695 AC2: when production native transport has pinned a session + # state root, that pin is authoritative even under test isolation + # (PR #701 redirected-state regression). + try: + import mcp_daemon_guard + + pinned = mcp_daemon_guard.pinned_session_state_dir() + if pinned: + return pinned + except Exception: + pass raw = (os.environ.get(_env_key) or "").strip() return raw or _fallback diff --git a/tests/test_issue_695_native_transport_quarantine.py b/tests/test_issue_695_native_transport_quarantine.py index 7b4ccac..25d17ce 100644 --- a/tests/test_issue_695_native_transport_quarantine.py +++ b/tests/test_issue_695_native_transport_quarantine.py @@ -63,7 +63,16 @@ class TestNativeTransportBinding(unittest.TestCase): mcp_daemon_guard.assert_sanctioned_mutation_runtime("offline_import") msg = str(ctx.exception) self.assertIn("#695", msg) - self.assertIn("not sufficient", msg.lower() + " " + msg) + # FORCE disables pytest allowance; direct-import env is rejected first + # (AC1). Without ALLOW_DIRECT, env-alone also yields "not sufficient". + lowered = msg.lower() + self.assertTrue( + "direct" in lowered + or "not sufficient" in lowered + or "allow_direct" in lowered + or "gitea_allow_direct" in lowered, + msg, + ) def test_direct_import_mark_rejected_outside_entrypoint(self): mcp_daemon_guard.clear_native_runtime_for_tests() @@ -677,5 +686,163 @@ class TestDocsStopAfterNativeFailure(unittest.TestCase): self.assertIn("GITEA_ALLOW_DIRECT_MCP_IMPORT", doc) +class TestPR701DirectImportSessionOverrideSequence(unittest.TestCase): + """AC9 regression: exact PR #701 contaminated-approval sequence must fail closed. + + Observed attack: + - GITEA_ALLOW_DIRECT_MCP_IMPORT=1 + - import mutation tools from gitea_mcp_server + - GITEA_MCP_SESSION_STATE_DIR=.mcp_session_701 (cross-PR lock evasion) + - mark_final + submit_pr_review + """ + + def tearDown(self) -> None: + mcp_daemon_guard.clear_native_runtime_for_tests() + os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None) + os.environ.pop(mcp_daemon_guard.SANCTIONED_DAEMON_ENV, None) + os.environ.pop(mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV, None) + os.environ.pop(mcp_daemon_guard.SESSION_STATE_DIR_ENV, None) + + def test_offline_run_submit_sequence_fails_closed(self): + """Fresh interpreter: direct import + state-dir override cannot mark/submit.""" + with tempfile.TemporaryDirectory() as tmp: + redirect = str(Path(tmp) / ".mcp_session_701") + snippet = textwrap.dedent( + f""" + import os + import sys + os.environ["GITEA_ALLOW_DIRECT_MCP_IMPORT"] = "1" + os.environ["GITEA_MCP_SESSION_STATE_DIR"] = {redirect!r} + os.environ["GITEA_MCP_PROFILE"] = "prgs-reviewer" + # No pytest modules in this subprocess. + import mcp_daemon_guard as g + assert g.is_native_mcp_transport() is False + assert g.is_production_native_mcp_transport() is False + try: + g.assert_sanctioned_mutation_runtime("run_submit_mark") + except g.UnsanctionedRuntimeError as exc: + msg = str(exc) + assert "GITEA_ALLOW_DIRECT_MCP_IMPORT" in msg or "#695" in msg + else: + raise SystemExit("direct-import env authorized mutation runtime") + try: + g.mark_sanctioned_daemon() + except g.UnsanctionedRuntimeError: + pass + else: + raise SystemExit("mark_sanctioned_daemon authorized offline import") + # Simulate decision-lock write into redirected dir only — must not + # establish native authority. + import mcp_session_state as ss + wrote = ss.save_state( + kind=ss.KIND_DECISION_LOCK, + payload={{ + "final_review_decision_ready": True, + "ready_pr_number": 701, + "ready_action": "approve", + "ready_expected_head_sha": "6b675f5c834b41f9d74e8a54294ff44dddf28ae4", + "session_profile": "prgs-reviewer", + "session_profile_lock": "prgs-reviewer", + "remote": "prgs", + }}, + profile_identity="prgs-reviewer", + state_dir={redirect!r}, + ) + assert wrote is not None + assert g.is_native_mcp_transport() is False + try: + g.assert_no_direct_import_bypass("gitea_submit_pr_review") + except g.UnsanctionedRuntimeError: + pass + else: + raise SystemExit("direct-import bypass accepted for submit") + print("OK") + """ + ) + proc = _run_offline_snippet(snippet) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertIn("OK", proc.stdout) + + def test_session_state_dir_pin_ignores_post_bind_redirect(self): + """AC2: after production bind, env STATE_DIR override is ignored.""" + mcp_daemon_guard.clear_native_runtime_for_tests() + import mcp_session_state + + with tempfile.TemporaryDirectory() as tmp: + legitimate = str(Path(tmp) / "legitimate-state") + rogue = str(Path(tmp) / ".mcp_session_701") + os.makedirs(legitimate, mode=0o700, exist_ok=True) + os.makedirs(rogue, mode=0o700, exist_ok=True) + os.environ[mcp_daemon_guard.SESSION_STATE_DIR_ENV] = legitimate + 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, + ): + with patch.object( + mcp_daemon_guard, "is_pytest_runtime", return_value=False + ): + mcp_daemon_guard.mark_sanctioned_daemon() + mcp_daemon_guard.bind_native_mcp_transport(transport="stdio") + pinned = mcp_daemon_guard.pinned_session_state_dir() + self.assertEqual(pinned, str(Path(legitimate).resolve())) + # Attacker redirects env after bind (PR #701). + os.environ[mcp_daemon_guard.SESSION_STATE_DIR_ENV] = rogue + self.assertEqual( + mcp_daemon_guard.pinned_session_state_dir(), + str(Path(legitimate).resolve()), + ) + self.assertEqual( + mcp_session_state.default_state_dir(), + str(Path(legitimate).resolve()), + ) + self.assertNotEqual( + mcp_session_state.default_state_dir(), + str(Path(rogue).resolve()), + ) + # Unpinned env view still sees rogue (diagnostics only). + unpinned = mcp_session_state.env_session_state_dir_unpinned() + self.assertTrue( + unpinned == rogue + or Path(unpinned).resolve() == Path(rogue).resolve(), + unpinned, + ) + + def test_direct_import_env_does_not_authorize_under_force_unsanctioned(self): + mcp_daemon_guard.clear_native_runtime_for_tests() + os.environ[mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV] = "1" + os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1" + self.assertTrue(mcp_daemon_guard.direct_import_env_enabled()) + # Under pytest, assert_no_direct_import_bypass is a no-op; FORCE path + # still blocks is_native / assert_sanctioned via force-unsanctioned. + self.assertFalse(mcp_daemon_guard.is_native_mcp_transport()) + with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): + mcp_daemon_guard.assert_sanctioned_mutation_runtime("direct-import") + + def test_quarantine_voids_merge_approval_for_contaminated_review(self): + """AC6–AC8: quarantined APPROVED does not satisfy merge approval head.""" + entry = { + "verdict": "APPROVED", + "dismissed": False, + "reviewed_head_sha": "6b675f5c834b41f9d74e8a54294ff44dddf28ae4", + "review_id": 431, + "submitted_at": "2026-07-13T23:52:34Z", + "quarantined": True, + } + result = merge_approval_gate.assess_merge_approval_head( + current_head_sha="6b675f5c834b41f9d74e8a54294ff44dddf28ae4", + latest_by_reviewer={"sysadmin": entry}, + quarantined_review_ids={431}, + ) + self.assertFalse(result["approval_at_current_head"]) + self.assertEqual(result["quarantined_approvals_at_current_head"], 1) + self.assertIn("quarantined", (result["stale_approval_block_reason"] or "")) + + if __name__ == "__main__": unittest.main()