Gitea-Tools/mcp_daemon_guard.py

"""Sanctioned MCP daemon guards for imports and credential access (#558 / #695).

Direct ``import gitea_mcp_server`` from a shell bypasses native MCP transport.
#558 introduced a daemon marker; #695 hardens it so:

- Environment variables alone cannot reconstruct a native session
  (``GITEA_MCP_SANCTIONED_DAEMON=1`` / ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` are
  insufficient for mutation gates).
- A process-local runtime record is created only by the official entrypoint
  (``mcp_server.py`` calling ``mark_sanctioned_daemon``), bound to PID and a
  random secret that never leaves process memory.
- Offline scripts that import internals fail closed on mutations.
- Pytest remains allowed (hermetic tests); optional force flags exist for
  provenance regression tests.

Manual deletion of session-state files is never a recovery path.
"""

from __future__ import annotations

import hashlib
import inspect
import os
import secrets
import time
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"
# Test-only: force provenance failure even under pytest (#695 regressions).
FORCE_PROVENANCE_FAIL_ENV = "GITEA_TEST_FORCE_UNSANCTIONED"

# Process-local native runtime (never persisted, never read from env alone).
_NATIVE_RUNTIME: dict[str, Any] | None = None


class UnsanctionedRuntimeError(RuntimeError):
    """Raised when mutation/credential code runs outside a native MCP daemon."""


def is_pytest_runtime() -> bool:
    if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
        "1",
        "true",
        "yes",
    }:
        return False
    import sys

    if "pytest" in sys.modules:
        return True
    return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip())


def _caller_is_official_entrypoint() -> bool:
    """True when mark_sanctioned_daemon is invoked from mcp_server.py."""
    for frame in inspect.stack()[1:12]:
        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]:
    """Mark this process as the official native MCP daemon (#695).

    Only the official ``mcp_server.py`` entrypoint (or pytest test bootstrap)
    may establish native transport. Setting env vars alone is insufficient.
    """
    global _NATIVE_RUNTIME
    if not is_pytest_runtime() and not allow_test_bootstrap:
        if not _caller_is_official_entrypoint():
            raise UnsanctionedRuntimeError(
                "mark_sanctioned_daemon rejected: not called from official "
                "mcp_server.py entrypoint (#695). Offline import / standalone "
                "scripts cannot reconstruct native transport. Stop after native "
                "MCP failure; do not run offline mutation helpers."
            )
    token = secrets.token_hex(32)
    _NATIVE_RUNTIME = {
        "token": token,
        "token_fingerprint": hashlib.sha256(token.encode()).hexdigest()[:16],
        "pid": os.getpid(),
        "started_at": time.time(),
        "entrypoint": "mcp_server",
    }
    # Legacy signal for older probes; alone does not authorize mutations.
    os.environ[SANCTIONED_DAEMON_ENV] = "1"
    return native_runtime_status()


def clear_native_runtime_for_tests() -> None:
    """Test helper: drop native runtime (does not clear env)."""
    global _NATIVE_RUNTIME
    _NATIVE_RUNTIME = None


def is_native_mcp_transport() -> bool:
    """True when this process holds a live native MCP runtime record (#695)."""
    if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
        "1",
        "true",
        "yes",
    }:
        return False
    if _NATIVE_RUNTIME is None:
        return False
    if int(_NATIVE_RUNTIME.get("pid") or -1) != os.getpid():
        return False
    if not (_NATIVE_RUNTIME.get("token") or "").strip():
        return False
    return True


def is_sanctioned_mcp_daemon() -> bool:
    """Backward-compatible name; #695 requires native transport, not env alone."""
    if is_native_mcp_transport():
        return True
    if is_pytest_runtime():
        return True
    # Explicit direct-import override is for non-LLM operator/test tools only.
    # It is deliberately ignored when a native runtime is expected for mutations
    # under LLM sessions (tests use pytest path). Env alone never grants native.
    return False


def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
    """Fail closed when mutation code runs outside native MCP transport (#695)."""
    if is_sanctioned_mcp_daemon():
        return
    env_spoof = (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in {
        "1",
        "true",
        "yes",
    }
    extra = ""
    if env_spoof:
        extra = (
            f" Note: {SANCTIONED_DAEMON_ENV} alone is not sufficient (#695); "
            "native transport requires the official MCP entrypoint."
        )
    raise UnsanctionedRuntimeError(
        f"Unsanctioned / non-native runtime blocked {context} (#695). "
        "Do not import gitea_mcp_server or call mutation helpers from a raw "
        "shell, offline runner, or ad-hoc script after native MCP failure. "
        "Stop and reconnect the official MCP daemon (mcp_server.py). "
        f"Do not set {ALLOW_DIRECT_IMPORT_ENV} or raw token env vars in LLM "
        f"sessions.{extra}"
    )


def assert_keychain_access_allowed() -> None:
    """Fail closed for git-credential keychain fill outside sanctioned contexts."""
    if is_sanctioned_mcp_daemon():
        return
    # Operator-only keychain CLI remains available outside LLM mutation path.
    if (os.environ.get(ALLOW_KEYCHAIN_CLI_ENV) or "").strip() in {"1", "true", "yes"}:
        if not is_pytest_runtime():
            # Still block pure env spoof of SANCTIONED_DAEMON for keychain when
            # FORCE is set for tests.
            if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
                "1",
                "true",
                "yes",
            }:
                pass
            else:
                return
    raise UnsanctionedRuntimeError(
        "Unsanctioned keychain/credential fill blocked (#558/#695). "
        "Token extraction via git-credential is only allowed inside the "
        f"official native MCP daemon or with explicit operator opt-in "
        f"{ALLOW_KEYCHAIN_CLI_ENV}=1 (never for offline mutation runners)."
    )


def native_runtime_status() -> dict[str, Any]:
    """LLM-safe native runtime status (no raw token)."""
    rt = _NATIVE_RUNTIME or {}
    return {
        "native_mcp_transport": is_native_mcp_transport(),
        "pytest": is_pytest_runtime(),
        "pid": rt.get("pid"),
        "token_fingerprint": rt.get("token_fingerprint"),
        "started_at": rt.get("started_at"),
        "entrypoint": rt.get("entrypoint"),
        "env_sanctioned_alone_insufficient": True,
        "sanctioned_env": SANCTIONED_DAEMON_ENV,
        "allow_direct_import_env": ALLOW_DIRECT_IMPORT_ENV,
        "allow_keychain_cli_env": ALLOW_KEYCHAIN_CLI_ENV,
    }


def runtime_status() -> dict[str, Any]:
    """Backward-compatible status payload."""
    status = native_runtime_status()
    status["sanctioned_daemon"] = is_sanctioned_mcp_daemon()
    return status


def mutation_provenance_fields() -> dict[str, Any]:
    """Fields to attach to live mutation / review audit records (#695 AC6)."""
    st = native_runtime_status()
    return {
        "transport": "native_mcp" if st["native_mcp_transport"] else "untrusted",
        "native_mcp_transport": bool(st["native_mcp_transport"]),
        "native_runtime_pid": st.get("pid"),
        "native_token_fingerprint": st.get("token_fingerprint"),
        "entrypoint": st.get("entrypoint"),
    }