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 established only by the resolved canonical
  entrypoint path (not basename) **and** the actual native MCP transport
  lifecycle (``bind_native_mcp_transport`` before ``mcp.run``). Merely
  importing or launching the entrypoint offline does not grant mutation
  authority.
- Public caller-controlled flags (including any former
  ``allow_test_bootstrap``) never establish trusted mutation provenance.
- Offline scripts that import internals fail closed on mutations.
- Pytest remains allowed for hermetic unit tests via ``is_pytest_runtime()``.
  A separate test-only seam may establish a **test-mode** native record for
  unit tests of transport gates; that record cannot authorize production
  Gitea mutation endpoints.

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 pathlib import Path
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

# Production transport identifiers accepted by bind_native_mcp_transport.
_PRODUCTION_TRANSPORTS = frozenset({"stdio"})
_RUNTIME_MODE_PRODUCTION = "production"
_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."""


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 _package_root() -> Path:
    """Directory that contains the canonical MCP entrypoint modules."""
    return Path(__file__).resolve().parent


def canonical_entrypoint_paths() -> frozenset[str]:
    """Resolved absolute paths of official entrypoints (not basenames)."""
    root = _package_root()
    return frozenset(
        {
            str((root / "mcp_server.py").resolve()),
            str((root / "gitea_mcp_server.py").resolve()),
        }
    )


def _resolve_path(path: str | None) -> str | None:
    if not path:
        return None
    try:
        return str(Path(path).resolve())
    except (OSError, RuntimeError, ValueError):
        return None


def _caller_official_entrypoint_path() -> str | None:
    """Return the resolved canonical entrypoint path in the call stack, or None.

    Basename-only matches (e.g. an attacker file named ``mcp_server.py``
    elsewhere) are rejected. The path must equal one of
    :func:`canonical_entrypoint_paths`.
    """
    canonical = canonical_entrypoint_paths()
    for frame in inspect.stack()[1:20]:
        resolved = _resolve_path(frame.filename)
        if resolved and resolved in canonical:
            return resolved
    return None


def _caller_is_official_entrypoint() -> bool:
    """True when invoked from a resolved canonical entrypoint path (#695)."""
    return _caller_official_entrypoint_path() is not None


def _new_runtime_token() -> tuple[str, str]:
    token = secrets.token_hex(32)
    fingerprint = hashlib.sha256(token.encode()).hexdigest()[:16]
    return token, fingerprint


def mark_sanctioned_daemon() -> dict[str, Any]:
    """Claim the official entrypoint for this process (#695).

    This alone does **not** authorize mutations. Callers must subsequently
    bind the native MCP transport via :func:`bind_native_mcp_transport`.

    Only a stack frame whose **resolved absolute path** is the canonical
    ``mcp_server.py`` or ``gitea_mcp_server.py`` next to this module may
    claim the entrypoint. Basename spoofing is rejected.

    There is no public ``allow_test_bootstrap`` argument: caller-controlled
    flags must never establish trusted mutation provenance. Hermetic tests
    use :func:`install_test_native_runtime` (pytest-only, test mode).
    """
    global _NATIVE_RUNTIME
    if is_pytest_runtime():
        # Under pytest, production mark is a no-op for transport authority.
        # Tests that need a native-transport record use install_test_native_runtime.
        return native_runtime_status()

    entrypoint_path = _caller_official_entrypoint_path()
    if entrypoint_path is None:
        raise UnsanctionedRuntimeError(
            "mark_sanctioned_daemon rejected: not called from the resolved "
            "canonical MCP entrypoint path (#695). Basename-only names "
            "(e.g. a renamed runner called mcp_server.py) are insufficient. "
            "Offline import / standalone scripts cannot reconstruct native "
            "transport. Stop after native MCP failure; do not run offline "
            "mutation helpers."
        )

    token, fingerprint = _new_runtime_token()
    _NATIVE_RUNTIME = {
        "token": token,
        "token_fingerprint": fingerprint,
        "pid": os.getpid(),
        "started_at": time.time(),
        "entrypoint": "mcp_server",
        "entrypoint_path": entrypoint_path,
        "phase": _PHASE_ENTRYPOINT_CLAIMED,
        "transport": None,
        "mode": _RUNTIME_MODE_PRODUCTION,
    }
    # Legacy signal for older probes; alone does not authorize mutations.
    os.environ[SANCTIONED_DAEMON_ENV] = "1"
    return native_runtime_status()


def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]:
    """Bind the live native MCP transport lifecycle (#695).

    Must be called from the resolved canonical entrypoint immediately before
    the real MCP server transport loop (e.g. ``mcp.run(transport=\"stdio\")``).
    Requires a prior successful :func:`mark_sanctioned_daemon` claim in this
    process. Import-only or offline launch without this bind leaves
    :func:`is_native_mcp_transport` false.
    """
    global _NATIVE_RUNTIME
    transport_name = (transport or "").strip().lower()
    if transport_name not in _PRODUCTION_TRANSPORTS:
        raise UnsanctionedRuntimeError(
            f"bind_native_mcp_transport rejected: transport {transport!r} is "
            f"not a production MCP transport (#695). Allowed: "
            f"{sorted(_PRODUCTION_TRANSPORTS)}."
        )

    entrypoint_path = _caller_official_entrypoint_path()
    if entrypoint_path is None:
        raise UnsanctionedRuntimeError(
            "bind_native_mcp_transport rejected: not called from the resolved "
            "canonical MCP entrypoint path (#695)."
        )

    if _NATIVE_RUNTIME is None or int(_NATIVE_RUNTIME.get("pid") or -1) != os.getpid():
        raise UnsanctionedRuntimeError(
            "bind_native_mcp_transport rejected: no entrypoint claim in this "
            "process (#695). Call mark_sanctioned_daemon() from the official "
            "entrypoint first."
        )

    if _NATIVE_RUNTIME.get("mode") != _RUNTIME_MODE_PRODUCTION:
        raise UnsanctionedRuntimeError(
            "bind_native_mcp_transport rejected: runtime mode is not "
            "production (#695)."
        )

    claimed = (_NATIVE_RUNTIME.get("entrypoint_path") or "").strip()
    if claimed and claimed != entrypoint_path:
        raise UnsanctionedRuntimeError(
            "bind_native_mcp_transport rejected: entrypoint path mismatch "
            "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()


def install_test_native_runtime() -> dict[str, Any]:
    """Pytest-only seam for hermetic native-transport unit tests (#695).

    Establishes a **test-mode** process-local record so unit tests can exercise
    gates that require ``is_native_mcp_transport()``. This record:

    - is rejected outside pytest (including a fresh offline interpreter);
    - never uses production mode;
    - cannot authorize production Gitea mutation endpoints
      (:func:`assert_production_mutation_runtime` / production path of
      :func:`assert_sanctioned_mutation_runtime` when not under pytest).

    There is no public caller-controlled flag that forges production native
    transport.
    """
    global _NATIVE_RUNTIME
    if not is_pytest_runtime():
        raise UnsanctionedRuntimeError(
            "install_test_native_runtime rejected: test-mode native runtime "
            "is only available under pytest (#695). allow_test_bootstrap and "
            "similar caller-controlled flags do not exist and cannot authorize "
            "a fresh offline interpreter."
        )
    token, fingerprint = _new_runtime_token()
    _NATIVE_RUNTIME = {
        "token": token,
        "token_fingerprint": fingerprint,
        "pid": os.getpid(),
        "started_at": time.time(),
        "entrypoint": "test_bootstrap",
        "entrypoint_path": None,
        "phase": _PHASE_TRANSPORT_BOUND,
        "transport": "test",
        "mode": _RUNTIME_MODE_TEST,
        "bound_at": time.time(),
    }
    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 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 {
        "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
    if _NATIVE_RUNTIME.get("phase") != _PHASE_TRANSPORT_BOUND:
        return False
    if not (_NATIVE_RUNTIME.get("transport") or "").strip():
        return False
    return True


def is_production_native_mcp_transport() -> bool:
    """True only for production-mode, transport-bound native runtime."""
    if not is_native_mcp_transport():
        return False
    return (_NATIVE_RUNTIME or {}).get("mode") == _RUNTIME_MODE_PRODUCTION


def is_sanctioned_mcp_daemon() -> bool:
    """Backward-compatible name; #695 requires native transport, not env alone."""
    if is_production_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_production_mutation_runtime(context: str = "mutation") -> None:
    """Fail closed unless production native MCP transport is bound (#695).

    Test-mode bootstrap records and pytest-only hermetic allowances do **not**
    satisfy this gate. Use for production Gitea mutation endpoints that must
    never be reachable via test bootstrap.
    """
    if is_production_native_mcp_transport():
        return
    mode = (_NATIVE_RUNTIME or {}).get("mode")
    if mode == _RUNTIME_MODE_TEST:
        raise UnsanctionedRuntimeError(
            f"Test-mode native runtime cannot authorize production {context} "
            "(#695). install_test_native_runtime / former allow_test_bootstrap "
            "must never reach real Gitea mutation endpoints."
        )
    assert_sanctioned_mutation_runtime(context)


def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
    """Fail closed when mutation code runs outside native MCP transport (#695).

    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")
    if mode == _RUNTIME_MODE_TEST:
        raise UnsanctionedRuntimeError(
            f"Test-mode native runtime cannot authorize production {context} "
            "(#695). Test bootstrap cannot reach production mutation endpoints."
        )
    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 and a live "
            "transport bind."
        )
    phase = (_NATIVE_RUNTIME or {}).get("phase")
    if phase == _PHASE_ENTRYPOINT_CLAIMED:
        extra = (
            (extra + " ") if extra else " "
        ) + (
            "Entrypoint was claimed but native MCP transport was never bound "
            "(#695); offline launch/import of the real entrypoint does not "
            "grant mutation authority."
        )
    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) over "
        "native transport. "
        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}"
    )


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(),
        "production_native_mcp_transport": is_production_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"),
        "entrypoint_path": rt.get("entrypoint_path"),
        "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,
    }


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()
    transport = "native_mcp" if st["native_mcp_transport"] else "untrusted"
    if st.get("mode") == _RUNTIME_MODE_TEST and st["native_mcp_transport"]:
        transport = "test_native_mcp"
    return {
        "transport": transport,
        "native_mcp_transport": bool(st["native_mcp_transport"]),
        "production_native_mcp_transport": bool(
            st.get("production_native_mcp_transport")
        ),
        "native_runtime_pid": st.get("pid"),
        "native_token_fingerprint": st.get("token_fingerprint"),
        "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")),
    }