Merge pull request 'feat: block direct MCP imports and unsanctioned keychain access (Closes #558)' (#566) from feat/issue-558-block-direct-imports into master

docs/llm-workflow-runbooks.md

@@ -413,6 +413,13 @@ 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).
+
 ## Bootstrap Review Path for self-hosted MCP fixes (#557)
 
 When a PR fixes Gitea-Tools MCP workflow code that the **live daemon** still

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.

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

gitea_mcp_server.py

@@ -9532,6 +9532,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).

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,
+    }

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:

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()