From 8ff19240474020898b6c835ff41e3fd0ef16f485 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 22:49:03 -0400 Subject: [PATCH] feat: block direct MCP imports and unsanctioned keychain access (Closes #558) Require a sanctioned MCP daemon (or pytest) before get_auth_header and keychain credential fill so raw shell imports cannot bypass preflight gates. --- docs/llm-workflow-runbooks.md | 8 ++++ docs/mcp-daemon-import-guard.md | 23 +++++++++ gitea_auth.py | 11 +++++ gitea_mcp_server.py | 5 ++ mcp_daemon_guard.py | 84 +++++++++++++++++++++++++++++++++ mcp_server.py | 11 +++++ tests/test_mcp_daemon_guard.py | 67 ++++++++++++++++++++++++++ 7 files changed, 209 insertions(+) create mode 100644 docs/mcp-daemon-import-guard.md create mode 100644 mcp_daemon_guard.py create mode 100644 tests/test_mcp_daemon_guard.py diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index d0e472d..7471b75 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -408,6 +408,14 @@ The guard blocks when the **control checkout** (repository root) is: `branches/...` directories are disposable role worktrees; the root checkout is the stable orchestration surface only. + +## No direct-import mutation path (#558) + +Never `import gitea_mcp_server` or call `gitea_auth.get_auth_header` / +keychain fill from a raw shell to bypass MCP preflight. + +Use the official MCP daemon only. See [`docs/mcp-daemon-import-guard.md`](mcp-daemon-import-guard.md). + ## Shell Spawn Hard-Stop Rule Symptom: a shell tool call returns `exit_code: -1` with empty stdout/stderr. diff --git a/docs/mcp-daemon-import-guard.md b/docs/mcp-daemon-import-guard.md new file mode 100644 index 0000000..8378dbd --- /dev/null +++ b/docs/mcp-daemon-import-guard.md @@ -0,0 +1,23 @@ +# MCP daemon import and keychain guard (#558) + +## Problem + +During deadlock debugging, agents imported `gitea_mcp_server` / ran credential +helpers from a raw shell, bypassing preflight purity and role gates. + +## Rule + +Mutation auth and keychain fill require a **sanctioned MCP daemon** process. + +| Context | Allowed | +|---------|---------| +| Official MCP entrypoint (`mcp_server.py` / `gitea_mcp_server` `__main__`) sets `GITEA_MCP_SANCTIONED_DAEMON=1` | yes | +| pytest | yes | +| `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` (operator/tests only) | yes | +| bare `python -c 'import gitea_auth; get_auth_header(...)'` | **no** | +| keychain fill without daemon | **no** unless `GITEA_ALLOW_KEYCHAIN_CLI=1` | + +## Operator note + +LLM sessions must never set the allow-direct-import or allow-keychain-cli +overrides. Those are human-only escape hatches. diff --git a/gitea_auth.py b/gitea_auth.py index 5e30213..167357b 100644 --- a/gitea_auth.py +++ b/gitea_auth.py @@ -104,6 +104,10 @@ def get_credentials(host): # 3. Optional fallback to macOS Keychain via git credential fill if not user and not password and os.environ.get("GITEA_USE_KEYCHAIN") == "1": + # #558: block raw keychain dumps outside the sanctioned MCP daemon. + import mcp_daemon_guard + + mcp_daemon_guard.assert_keychain_access_allowed() cmd_parts = ["git", "creden" + "tial", "fi" + "ll"] try: p = subprocess.Popen( @@ -116,6 +120,8 @@ def get_credentials(host): user = line.split("=", 1)[1] elif line.startswith("password="): password = line.split("=", 1)[1] + except mcp_daemon_guard.UnsanctionedRuntimeError: + raise except Exception: pass @@ -124,6 +130,11 @@ def get_credentials(host): def get_auth_header(host): """Return an ``Authorization`` header value for *host*.""" + # #558: resolving credentials for API mutation must not happen via ad-hoc + # direct imports that bypass the MCP daemon preflight wall. + import mcp_daemon_guard + + mcp_daemon_guard.assert_sanctioned_mutation_runtime("get_auth_header") host_key = host.lower().strip() # 1. Try Token-based auth from dynamic configs diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index ed650b5..9838ba5 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -9079,6 +9079,11 @@ def gitea_capability_stop_terminal_report() -> dict: # ── Entry point ─────────────────────────────────────────────────────────────── if __name__ == "__main__": + # #558: mark this process as the official MCP daemon before any tool + # dispatch so direct shell imports cannot reuse mutation/auth paths. + import mcp_daemon_guard + + mcp_daemon_guard.mark_sanctioned_daemon() # 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 new file mode 100644 index 0000000..241c68b --- /dev/null +++ b/mcp_daemon_guard.py @@ -0,0 +1,84 @@ +"""Sanctioned MCP daemon guards for imports and credential access (#558). + +Direct ``import gitea_mcp_server`` / ``import gitea_auth`` from a shell, plus +raw keychain dumps, bypass preflight purity and role gates. Mutation helpers +and keychain fallbacks therefore require an explicit sanctioned runtime. + +Sanctioned contexts (any one): +- ``GITEA_MCP_SANCTIONED_DAEMON=1`` (set by the official MCP entrypoint) +- pytest (``PYTEST_CURRENT_TEST`` present) +- explicit operator opt-in ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` (tests/tools only) + +Credential keychain fill additionally allows: +- ``GITEA_ALLOW_KEYCHAIN_CLI=1`` for operator-only non-MCP scripts that must + use git-credential (never the default for LLM shells). +""" + +from __future__ import annotations + +import os +from typing import Any + +SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON" +ALLOW_DIRECT_IMPORT_ENV = "GITEA_ALLOW_DIRECT_MCP_IMPORT" +ALLOW_KEYCHAIN_CLI_ENV = "GITEA_ALLOW_KEYCHAIN_CLI" + + +class UnsanctionedRuntimeError(RuntimeError): + """Raised when mutation/credential code runs outside a sanctioned MCP daemon.""" + + +def is_pytest_runtime() -> bool: + return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip()) + + +def is_sanctioned_mcp_daemon() -> bool: + if (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in {"1", "true", "yes"}: + return True + if (os.environ.get(ALLOW_DIRECT_IMPORT_ENV) or "").strip() in {"1", "true", "yes"}: + return True + if is_pytest_runtime(): + return True + return False + + +def mark_sanctioned_daemon() -> None: + """Call from the official MCP server entrypoint before serving tools.""" + os.environ[SANCTIONED_DAEMON_ENV] = "1" + + +def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None: + """Fail closed when server mutation code is used outside the MCP daemon.""" + if is_sanctioned_mcp_daemon(): + return + raise UnsanctionedRuntimeError( + f"Unsanctioned runtime blocked {context} (#558). " + "Do not import gitea_mcp_server / call mutation helpers from a raw " + "shell or ad-hoc script. Use the official MCP daemon entrypoint " + f"(sets {SANCTIONED_DAEMON_ENV}=1), or run under pytest. " + f"Operator-only override: {ALLOW_DIRECT_IMPORT_ENV}=1 (not for LLM sessions)." + ) + + +def assert_keychain_access_allowed() -> None: + """Fail closed for git-credential keychain fill outside sanctioned contexts.""" + if is_sanctioned_mcp_daemon(): + return + if (os.environ.get(ALLOW_KEYCHAIN_CLI_ENV) or "").strip() in {"1", "true", "yes"}: + return + raise UnsanctionedRuntimeError( + "Unsanctioned keychain/credential fill blocked (#558). " + "Token extraction via git-credential is only allowed inside the " + f"official MCP daemon ({SANCTIONED_DAEMON_ENV}=1), pytest, or with " + f"explicit operator opt-in {ALLOW_KEYCHAIN_CLI_ENV}=1." + ) + + +def runtime_status() -> dict[str, Any]: + return { + "sanctioned_daemon": is_sanctioned_mcp_daemon(), + "pytest": is_pytest_runtime(), + "sanctioned_env": SANCTIONED_DAEMON_ENV, + "allow_direct_import_env": ALLOW_DIRECT_IMPORT_ENV, + "allow_keychain_cli_env": ALLOW_KEYCHAIN_CLI_ENV, + } diff --git a/mcp_server.py b/mcp_server.py index dbe5f1b..479273b 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -37,6 +37,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). +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__. + pass + # Execute the actual server logic via exec in this namespace. impl_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "gitea_mcp_server.py") try: diff --git a/tests/test_mcp_daemon_guard.py b/tests/test_mcp_daemon_guard.py new file mode 100644 index 0000000..45007a2 --- /dev/null +++ b/tests/test_mcp_daemon_guard.py @@ -0,0 +1,67 @@ +"""Tests for sanctioned MCP daemon guards (#558).""" + +from __future__ import annotations + +import os +import unittest +from unittest.mock import patch + +import mcp_daemon_guard +import gitea_auth + + +class TestMcpDaemonGuard(unittest.TestCase): + def test_unsanctioned_blocks_mutation_runtime(self): + env = {k: v for k, v in os.environ.items() if k not in { + mcp_daemon_guard.SANCTIONED_DAEMON_ENV, + mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV, + "PYTEST_CURRENT_TEST", + }} + with patch.dict(os.environ, env, clear=True): + with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): + mcp_daemon_guard.assert_sanctioned_mutation_runtime("test") + + def test_pytest_allows_runtime(self): + # Running under pytest already sets PYTEST_CURRENT_TEST. + mcp_daemon_guard.assert_sanctioned_mutation_runtime("pytest") + + def test_mark_sanctioned_allows(self): + env = {k: v for k, v in os.environ.items() if k != "PYTEST_CURRENT_TEST"} + env[mcp_daemon_guard.SANCTIONED_DAEMON_ENV] = "1" + with patch.dict(os.environ, env, clear=True): + mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon") + + def test_keychain_blocked_without_sanction(self): + env = { + k: v + for k, v in os.environ.items() + if k + not in { + mcp_daemon_guard.SANCTIONED_DAEMON_ENV, + mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV, + mcp_daemon_guard.ALLOW_KEYCHAIN_CLI_ENV, + "PYTEST_CURRENT_TEST", + } + } + with patch.dict(os.environ, env, clear=True): + with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): + mcp_daemon_guard.assert_keychain_access_allowed() + + def test_get_auth_header_blocked_when_unsanctioned(self): + env = { + k: v + for k, v in os.environ.items() + if k + not in { + mcp_daemon_guard.SANCTIONED_DAEMON_ENV, + mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV, + "PYTEST_CURRENT_TEST", + } + } + with patch.dict(os.environ, env, clear=True): + with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError): + gitea_auth.get_auth_header("gitea.prgs.cc") + + +if __name__ == "__main__": + unittest.main()