"""Master-parity staleness gate (#420). The Gitea MCP server loads its capability-gate code and workflow logic into memory when the process starts. When ``master`` advances -- for example a newly merged security gate such as the branch-delete capability gate (#408/#410) -- the running process keeps executing the *old* code until it is restarted. A stale server can therefore still perform a mutation that the updated codebase would forbid. Runtime profile/config data is already read live from disk on every call (``gitea_config.load_config`` re-reads the JSON file each time), so profile ``allowed_operations`` changes take effect immediately without a restart. The gap this module closes is *code* parity: it captures the server process's source-tree commit at startup and detects, at mutation time, when the on-disk ``master`` HEAD has advanced past it. Detected staleness fails closed with a restart-required recovery report, while read-only operations stay allowed so a stale server can still be inspected. The core assessment is pure -- callers inject the observed HEAD SHAs -- so the logic is fully unit-testable without a git checkout. """ from __future__ import annotations import os import subprocess # Environment escape hatches (ops + tests): # GITEA_MCP_DISABLE_PARITY_GATE -> disable enforcement entirely (fail open). # GITEA_TEST_CURRENT_HEAD -> force the "current" HEAD read, for tests. ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE" ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD" def read_git_head(root: str) -> str | None: """Return the current ``HEAD`` commit SHA of *root*, or ``None``. ``None`` means the SHA could not be determined (not a git checkout, git unavailable, or an error). A test override via ``GITEA_TEST_CURRENT_HEAD`` takes precedence so the gate can be exercised deterministically. """ forced = os.environ.get(ENV_TEST_CURRENT_HEAD) if forced is not None: return forced.strip() or None if not root: return None try: res = subprocess.run( ["git", "-C", root, "rev-parse", "HEAD"], capture_output=True, text=True, check=False, ) except Exception: return None if res.returncode != 0: return None return (res.stdout or "").strip() or None def capture_startup_parity(root: str, head: str | None = None) -> dict: """Capture the process source-tree baseline once at server startup. *head* may be injected (tests); otherwise it is read from *root*. The result is an opaque baseline handed back to :func:`assess_master_parity`. """ startup_head = head if head is not None else read_git_head(root) return {"root": root, "startup_head": startup_head} def _short(sha: str | None) -> str: return sha[:12] if sha else "unknown" def assess_master_parity(startup: dict | None, current_head: str | None) -> dict: """Compare the startup baseline against the current on-disk ``HEAD``. Pure: both HEADs are supplied by the caller. Returns a structured result: - ``in_parity`` -- server code matches the on-disk master (or parity could not be determined, which is not treated as stale). - ``stale`` -- the on-disk master has definitively advanced past the running process. - ``restart_required`` -- alias of ``stale``; the recovery action. - ``determinable`` -- whether both HEADs were known well enough to compare. - ``startup_head`` / ``current_head`` / ``reasons``. """ startup_head = (startup or {}).get("startup_head") reasons: list[str] = [] if startup_head is None: reasons.append( "startup commit was not captured; code parity cannot be enforced") return _result(True, False, False, startup_head, current_head, reasons) if current_head is None: reasons.append( "current workspace HEAD could not be read; code parity cannot be " "enforced") return _result(True, False, False, startup_head, current_head, reasons) if startup_head == current_head: return _result(True, False, True, startup_head, current_head, reasons) reasons.append( f"MCP server started at commit {_short(startup_head)} but the workspace " f"master is now {_short(current_head)}; restart the server to load the " f"current capability gates") return _result(False, True, True, startup_head, current_head, reasons) def _result(in_parity, stale, determinable, startup_head, current_head, reasons): return { "in_parity": in_parity, "stale": stale, "restart_required": stale, "determinable": determinable, "startup_head": startup_head, "current_head": current_head, "reasons": list(reasons), } def gate_disabled() -> bool: """Whether the parity gate is disabled by env escape hatch.""" return bool((os.environ.get(ENV_DISABLE) or "").strip()) def parity_block_reasons(assessment: dict) -> list[str]: """Block reasons for a mutation gate (empty when the mutation may proceed). A disabled gate or an in-parity / non-determinable assessment yields no reasons; only a definitively stale server blocks. """ if gate_disabled(): return [] if assessment.get("stale"): return list(assessment.get("reasons") or ["server code is stale relative to master (fail closed)"]) return [] def parity_report(assessment: dict) -> dict: """Structured stale-server report for permission-block payloads.""" return { "kind": "server_stale", "restart_required": True, "startup_head": assessment.get("startup_head"), "current_head": assessment.get("current_head"), "reasons": list(assessment.get("reasons") or []), "recovery": [ "The running MCP server is executing code older than the current " "master and may not enforce newly merged capability gates.", "Restart the Gitea MCP server so it reloads master's capability " "gates and execution profiles before retrying the mutation.", ], } def format_parity(assessment: dict) -> str: """One-line human summary for logs / runtime context.""" if assessment.get("stale"): return (f"STALE: started {_short(assessment.get('startup_head'))}, " f"master now {_short(assessment.get('current_head'))} " f"(restart required)") if not assessment.get("determinable"): return "parity indeterminate (baseline or current HEAD unknown)" return f"in parity at {_short(assessment.get('current_head'))}"