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

import mcp_transport_config

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.
# #931: the permitted set is defined once, in mcp_transport_config. This name
# is kept as an alias so the guard never restates a transport identifier.
_PRODUCTION_TRANSPORTS = mcp_transport_config.SUPPORTED_TRANSPORTS
_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."""


class TransportExecutionError(UnsanctionedRuntimeError):
    """Raised when a bound transport may not be served by this entrypoint (#931).

    Subclasses :class:`UnsanctionedRuntimeError` so every existing fail-closed
    handler still catches it, while letting a caller that cares distinguish
    "nothing is bound" from "something valid is bound but its listener has not
    been commissioned". Carries the structured verdict on ``.assessment``.
    """

    def __init__(self, message: str, assessment: dict[str, Any] | None = None):
        super().__init__(message)
        self.assessment = assessment or {}
        self.blocker_kind = self.assessment.get("blocker_kind")
        self.owner_issue = self.assessment.get("owner_issue")
        self.transport = self.assessment.get("transport")


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 | None = None) -> dict[str, Any]:
    """Bind the live native MCP transport lifecycle (#695 / #931).

    Must be called from the resolved canonical entrypoint immediately before
    the real MCP server transport loop (``mcp.run``). 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.

    #931: ``transport`` is now optional. Omitting it — which is what the
    production entrypoint does — resolves the identifier from deployment
    configuration via :func:`mcp_transport_config.resolve_configured_transport`,
    yielding :data:`mcp_transport_config.DEFAULT_TRANSPORT` when nothing is
    configured. An explicit argument remains supported for tests and for a
    launcher that has already resolved the value. Either way the identifier is
    validated against the single permitted set before the runtime record is
    written, so no tool can dispatch over an unregistered transport.

    The resolved value is pinned into the process-local record and is read back
    only through :func:`bound_transport`. Rebinding to a different transport is
    refused, so two guards can never observe different values in one process.
    """
    global _NATIVE_RUNTIME
    if transport is None:
        resolution = mcp_transport_config.resolve_configured_transport()
        transport_name = str(resolution["transport"])
        if not resolution["supported"]:
            raise UnsanctionedRuntimeError(
                "bind_native_mcp_transport rejected: "
                + "; ".join(resolution["reasons"])
                + " No tool is served over an unregistered transport."
            )
    else:
        transport_name = mcp_transport_config.normalize_transport(transport)
        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)."
        )

    # #931: one process binds one transport. Re-binding the same identifier is
    # idempotent (a retried launch step must not fail); re-binding a different
    # one is refused, because a guard that already read the first value would
    # otherwise disagree with a guard that reads the second.
    already_bound = (_NATIVE_RUNTIME.get("transport") or "").strip()
    if (
        already_bound
        and _NATIVE_RUNTIME.get("phase") == _PHASE_TRANSPORT_BOUND
        and already_bound != transport_name
    ):
        raise UnsanctionedRuntimeError(
            "bind_native_mcp_transport rejected: transport is already bound to "
            f"{already_bound!r} in this process; rebinding to "
            f"{transport_name!r} is forbidden (#931). Restart the server to "
            "change the deployment transport."
        )

    # 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 bound_transport() -> str | None:
    """The one authoritative bound transport identifier, or ``None`` (#931).

    This is the shared accessor every transport-aware guard reads. It reports
    the value pinned at bind time, never the environment, so changing
    ``GITEA_MCP_TRANSPORT`` after the bind cannot move what a guard observes —
    the same rule :func:`pinned_session_state_dir` applies to session state.

    ``None`` means unbound: an offline import or a launch that never reached
    the bind. Callers must treat that as fail-closed, exactly as they already
    treat :func:`is_native_mcp_transport` returning false.
    """
    if not is_native_mcp_transport():
        return None
    return (_NATIVE_RUNTIME or {}).get("transport") or None


def assert_transport_bound(context: str = "tool service") -> str:
    """Return the bound transport, or fail closed before *context* (#931).

    Called immediately before the server enters its transport loop so an
    invalid or absent bind stops the process rather than serving tools over a
    transport no guard can name.
    """
    transport = bound_transport()
    if transport:
        return transport
    raise UnsanctionedRuntimeError(
        f"No MCP transport is bound; refusing {context} (#931). "
        "bind_native_mcp_transport must succeed from the canonical entrypoint "
        "before any tool is served. Offline import and standalone launch "
        "cannot reconstruct a bind."
    )


def assess_serve_authorization() -> dict[str, Any]:
    """Structured verdict on whether this process may serve tools (#931).

    This is the decision that consumes :func:`bound_transport`. It is what stops
    the bound identifier from being reporting-only metadata: the serve path
    cannot proceed unless the value pinned at bind is one this entrypoint is
    commissioned to execute.

    Never raises; returns the verdict so callers and diagnostics can inspect it.
    """
    return mcp_transport_config.assess_transport_execution(bound_transport())


def authorize_transport_execution(context: str = "tool service") -> str:
    """Return the transport this process may serve, or fail closed (#931).

    Two distinct boundaries, in order:

    1. **Bind presence** — :func:`assert_transport_bound` enforces the
       pre-existing #695 contract, so an unbound runtime keeps its established
       failure and reason code.
    2. **Execution authorization** — the bound identifier must be one this
       entrypoint is commissioned to serve. A registered transport whose
       listener has not been commissioned is refused here, before any listener
       is created and before any tool can dispatch.

    That ordering matters: recognition, validation and durable recording all
    still happen for a remote identifier, so #931's seam is intact; only the act
    of *serving* it is withheld until its owning issue commissions it.
    """
    # Boundary 1: unbound stays exactly as fail-closed as it was under #695.
    assert_transport_bound(context)

    # Boundary 2: bound, but is this entrypoint allowed to serve it?
    assessment = assess_serve_authorization()
    if assessment.get("allowed"):
        return str(assessment["transport"])

    reasons = "; ".join(assessment.get("reasons") or []) or "not authorized"
    next_action = assessment.get("exact_next_action") or ""
    raise TransportExecutionError(
        f"Refusing {context} (#931) [{assessment.get('blocker_kind')}]: "
        f"{reasons} {next_action}".strip(),
        assessment,
    )


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"),
        # #931: the authoritative bound identifier, plus the seam that defines
        # what may be bound. ``bound_transport`` is None until a bind succeeds,
        # so an offline import is distinguishable from a stdio session.
        "bound_transport": bound_transport(),
        "transport_bound": bound_transport() is not None,
        "default_transport": mcp_transport_config.DEFAULT_TRANSPORT,
        "supported_transports": list(mcp_transport_config.supported_transports()),
        "transport_env": mcp_transport_config.TRANSPORT_ENV,
        # #931 review 635: recognition and execution authorization are distinct.
        # ``supported`` is what may be bound; ``executable`` is what this
        # entrypoint may actually serve. A recognized-but-uncommissioned
        # transport reports serve_authorized False with a named blocker.
        "executable_transports": list(mcp_transport_config.executable_transports()),
        "serve_authorized": bool(assess_serve_authorization().get("allowed")),
        "serve_authorization": assess_serve_authorization(),
        "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`` stays the trust *class* it has always been, so existing
        # durable records keep their shape. ``bound_transport`` (#931) adds the
        # bound identifier itself, which is what lets an operator tell from a
        # durable record which transport performed a mutation.
        "transport": transport,
        "bound_transport": st.get("bound_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")),
    }