From 1cbbde00895dce1545290a660bc31c4ac9210b3e Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 24 Jul 2026 08:26:29 -0400 Subject: [PATCH] feat(restart): pre-restart drain proof and hard gate (#661) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `drain_proof.py`: a machine-verifiable DrainProof artifact plus a fail-closed verifier and the hard gate the sanctioned restart-apply path must consult, so a restart can never proceed on a stale or false "ready" claim (#655 umbrella, child of #658 coordinator / #659 drain / #660 checkpoints). - DrainProof: HMAC-SHA256 keyed proof-id over a canonical body using a per-process secret -> non-forgeable within the process; a proof minted in a prior daemon process will not verify after restart. Short TTL (120s). - build_drain_proof(): mints the proof from the #658 impact report + the drain-mode outcomes. Checklist: no in-flight mutations, assignments stopped, checkpoints complete, handoffs ok, leases handled, acks-or- timeout. Every check fails closed on missing/ambiguous evidence; the no-in-flight-mutations and leases-handled checks are derived from the authoritative impact report, not self-reported. - verify_drain_proof(): fail-closed — rejects missing, malformed, expired, signature-mismatched (forged/tampered/prior-process), unclean, or stale-fingerprint proofs; recomputes cleanliness from the checks rather than trusting the flag. - gate_apply_restart(): allow only on a valid clean proof; deny -> durable incident descriptor; break-glass is the only bypass and is never silent. - Checkpoint completeness is a supplied input, not a hard dependency on the (still-unmerged #660) checkpoint schema. Wire the gate into gitea_request_mcp_restart: dry_run=False now enforces the hard gate (drain_proof_json required; break-glass via request_break_glass + GITEA_BREAKGLASS_RESTART_AUTHORIZATION env). The tool still performs no actual restart — execution remains a further child. Tests: tests/test_drain_proof.py — 25 cases covering AC#1-4 (apply without proof denied, successful drain verifiable, open unsafe mutation fails, pass/fail/expired), forgery/tamper/wrong-secret/stale-fingerprint rejection, break-glass bypass, and secret hygiene. 25/25 pass (coordinator suite unaffected: 40/40 together). Links #652 #653 #655 #658 #659 #660. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01E7Fv9Bp2XWgvaWa4M1kdR7 (cherry picked from commit e7bcc952bb3e820fda95acbecefeaebfa5f8fcff) --- drain_proof.py | 726 ++++++++++++++++++++++++++++++++++++++ gitea_mcp_server.py | 66 +++- tests/test_drain_proof.py | 383 ++++++++++++++++++++ 3 files changed, 1167 insertions(+), 8 deletions(-) create mode 100644 drain_proof.py create mode 100644 tests/test_drain_proof.py diff --git a/drain_proof.py b/drain_proof.py new file mode 100644 index 0000000..d81c26e --- /dev/null +++ b/drain_proof.py @@ -0,0 +1,726 @@ +"""Pre-restart drain proof and hard gate (#661). + +A sanctioned MCP restart may only proceed after a machine-verifiable *drain +proof* attests that unsafe work is clear and checkpoints are complete. Drain +mode alone (#659) is not enough: without a proof, an apply path could still +restart on a stale or false "ready" claim, dropping mutations and orphaning +leases. This module defines the :class:`DrainProof` artifact, a fail-closed +verifier, and the hard gate the sanctioned restart-apply path must consult. + +Design rules (mirror :mod:`restart_coordinator` / :mod:`lease_lifecycle`): + +* **Pure classification.** Every function here operates on already-gathered + inputs and returns a structured result. Nothing touches the network, the + filesystem, or a live process, so multi-session fixtures drive every branch. + This module never restarts anything; the gate only *authorizes or denies*. +* **Fail closed.** A missing, expired, tampered, or unclean proof denies the + restart. Unknown checkpoint completeness is treated as *not complete*. An + incomplete impact report can never yield a clean proof. +* **Non-forgeable within the process.** The proof id is a keyed hash over the + canonical proof contents using a per-process secret. A worker session cannot + hand-craft a passing proof without that secret, and a proof minted in a prior + daemon process will not verify after a restart (the secret is regenerated). +* **No secrets leak.** The per-process secret never appears in a proof, an + ``as_dict``, an audit record, or an incident descriptor. + +Relationship to siblings (#655 umbrella): + +* **#658** ``restart_coordinator.evaluate_restart_impact`` — produces the + blast-radius impact report this proof consumes ("what would a restart + disrupt?"). ``mutations`` / ``critical_sections`` being empty is what the + no-in-flight-mutations check verifies. +* **#659** graceful drain mode — performs the drain actions and calls + :func:`build_drain_proof` to mint the artifact once its checklist passes. +* **#660** durable session checkpoints — supplies checkpoint completeness. + Because that schema may not yet be present, completeness is an *input* here, + never a hard table dependency; unknown fails closed. + +Non-goals (separate children): the emergency break-glass *workflow* (this gate +only leaves a sanctioned bypass hole authorized elsewhere), the console UI, and +the actual restart execution. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import os +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, Mapping, Sequence + +DRAIN_PROOF_VERSION = "1.0.0-issue-661" + +# Short default lifetime for a drain proof. A proof attests to a *point-in-time* +# drained state; live work can resume the moment drain mode relaxes, so the +# window in which a proof is honoured must be small (#661 security: short TTL). +DEFAULT_PROOF_TTL_SECONDS = 120 + +# Gate verdicts. +GATE_ALLOW = "allow" +GATE_DENY = "deny" +GATE_BREAK_GLASS = "break_glass" + +# The mandatory drain checklist. A proof is *clean* only when every one of these +# checks passed. Names are stable identifiers surfaced in audit + incidents. +CHECK_NO_INFLIGHT_MUTATIONS = "no_inflight_mutations" +CHECK_ASSIGNMENTS_STOPPED = "assignments_stopped" +CHECK_CHECKPOINTS_COMPLETE = "checkpoints_complete" +CHECK_HANDOFFS_OK = "handoffs_ok" +CHECK_LEASES_HANDLED = "leases_handled" +CHECK_ACKS_OR_TIMEOUT = "acks_or_timeout" + +REQUIRED_CHECKS: tuple[str, ...] = ( + CHECK_NO_INFLIGHT_MUTATIONS, + CHECK_ASSIGNMENTS_STOPPED, + CHECK_CHECKPOINTS_COMPLETE, + CHECK_HANDOFFS_OK, + CHECK_LEASES_HANDLED, + CHECK_ACKS_OR_TIMEOUT, +) + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _parse_ts(value: str | None) -> datetime | None: + if not value: + return None + text = str(value).strip() + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + dt = datetime.fromisoformat(text) + except ValueError: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def _canonical(payload: Any) -> str: + """Deterministic JSON encoding for hashing (stable key order, no spaces).""" + + return json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + + +# --------------------------------------------------------------------------- +# Per-process secret. Generated once per daemon process; regenerated on restart. +# Injectable for tests so build + verify share a secret. Never serialized. +# --------------------------------------------------------------------------- +_PROCESS_SECRET = os.urandom(32) + + +def process_secret() -> bytes: + """Return the per-process proof-signing secret (never serialized).""" + + return _PROCESS_SECRET + + +def _resolve_secret(secret: bytes | None) -> bytes: + return secret if secret is not None else _PROCESS_SECRET + + +@dataclass(frozen=True) +class DrainCheck: + """One mandatory drain checklist result.""" + + name: str + passed: bool + detail: str + + def as_dict(self) -> dict[str, Any]: + return {"name": self.name, "passed": self.passed, "detail": self.detail} + + +@dataclass(frozen=True) +class DrainProof: + """Machine-verifiable proof that a restart's blast radius has been drained. + + The ``proof_id`` is a keyed hash over the canonical proof body; it is the + tamper-evident signature verified at the gate. ``clean`` is True only when + every required check passed. The proof is honoured only until ``expires_at``. + """ + + version: str + proof_id: str + clean: bool + issued_at: str + expires_at: str + requesting_session_id: str | None + impact_fingerprint: str + checks: list[DrainCheck] + failed_checks: list[str] + + def as_dict(self) -> dict[str, Any]: + return { + "version": self.version, + "proof_id": self.proof_id, + "clean": self.clean, + "issued_at": self.issued_at, + "expires_at": self.expires_at, + "requesting_session_id": self.requesting_session_id, + "impact_fingerprint": self.impact_fingerprint, + "checks": [c.as_dict() for c in self.checks], + "failed_checks": list(self.failed_checks), + } + + +@dataclass(frozen=True) +class VerifyResult: + """Outcome of :func:`verify_drain_proof` (fail closed).""" + + valid: bool + reasons: list[str] + proof_id: str | None + clean: bool + expired: bool + tampered: bool + + def as_dict(self) -> dict[str, Any]: + return { + "valid": self.valid, + "reasons": list(self.reasons), + "proof_id": self.proof_id, + "clean": self.clean, + "expired": self.expired, + "tampered": self.tampered, + } + + +@dataclass(frozen=True) +class GateDecision: + """Outcome of :func:`gate_apply_restart`.""" + + allow: bool + verdict: str + reasons: list[str] + proof_id: str | None + break_glass: bool + incident: dict[str, Any] | None + audit_record: dict[str, Any] + + def as_dict(self) -> dict[str, Any]: + return { + "allow": self.allow, + "verdict": self.verdict, + "reasons": list(self.reasons), + "proof_id": self.proof_id, + "break_glass": self.break_glass, + "incident": self.incident, + "audit_record": dict(self.audit_record), + } + + +def impact_fingerprint(impact_report: Mapping[str, Any] | None) -> str: + """Stable fingerprint of the blast-radius state a proof was minted against. + + Binds a proof to the specific impact evaluation. If the live state changes + (a new mutation appears) between minting and gate, the caller can pass the + fresh fingerprint and the proof will be rejected as stale. + """ + + report = impact_report or {} + counts = report.get("counts") or {} + material = { + "inventory_complete": bool(report.get("inventory_complete", False)), + "verdict": report.get("verdict"), + "affected_issues": sorted(report.get("affected_issues") or []), + "affected_prs": sorted(report.get("affected_prs") or []), + "mutations": sorted( + str(m.get("lease_id")) + for m in (report.get("mutations") or []) + if isinstance(m, Mapping) + ), + "critical_sections": sorted( + str(c.get("lease_id")) + for c in (report.get("critical_sections") or []) + if isinstance(c, Mapping) + ), + "terminal_lock": bool(report.get("terminal_lock")), + "counts": { + k: counts.get(k) + for k in ( + "sessions_live_other", + "leases_disruptive", + "critical_sections", + "mutations", + ) + }, + } + return hashlib.sha256(_canonical(material).encode("utf-8")).hexdigest() + + +def _sign(body: Mapping[str, Any], secret: bytes) -> str: + """Keyed (HMAC-SHA256) signature over the canonical proof body.""" + + return hmac.new( + secret, _canonical(body).encode("utf-8"), hashlib.sha256 + ).hexdigest() + + +def _proof_body( + *, + clean: bool, + issued_at: str, + expires_at: str, + requesting_session_id: str | None, + fingerprint: str, + checks: Sequence[DrainCheck], +) -> dict[str, Any]: + """The exact fields covered by the signature. Order-independent (canonical).""" + + return { + "version": DRAIN_PROOF_VERSION, + "clean": clean, + "issued_at": issued_at, + "expires_at": expires_at, + "requesting_session_id": requesting_session_id, + "impact_fingerprint": fingerprint, + "checks": [c.as_dict() for c in checks], + } + + +def _bool_input(value: Any) -> bool: + """Strictly interpret a drain-state flag; anything not explicitly True fails.""" + + return value is True + + +def _evaluate_checks( + impact_report: Mapping[str, Any], + drain_state: Mapping[str, Any], +) -> list[DrainCheck]: + """Compute the mandatory checklist from the impact report + drain outcomes. + + The report answers "is unsafe work still in flight?"; ``drain_state`` reports + the drain-mode actions the coordinator/#659 performed. Every check fails + closed when its evidence is missing or ambiguous. + """ + + checks: list[DrainCheck] = [] + + inventory_complete = bool(impact_report.get("inventory_complete", False)) + mutations = list(impact_report.get("mutations") or []) + critical = list(impact_report.get("critical_sections") or []) + disruptive = [ + l + for l in (impact_report.get("affected_leases") or []) + if isinstance(l, Mapping) and l.get("disruptive") + ] + + # 1. No in-flight mutations. Derived from the authoritative impact report, + # not self-reported: a proof cannot claim "no mutations" while the report + # still shows mutations or unsevered critical sections. + if not inventory_complete: + checks.append( + DrainCheck( + CHECK_NO_INFLIGHT_MUTATIONS, + False, + "impact report inventory incomplete; cannot confirm mutations " + "cleared (fail closed)", + ) + ) + elif mutations or critical: + checks.append( + DrainCheck( + CHECK_NO_INFLIGHT_MUTATIONS, + False, + f"{len(mutations)} mutation(s) and {len(critical)} critical " + "section(s) still in flight", + ) + ) + else: + checks.append( + DrainCheck( + CHECK_NO_INFLIGHT_MUTATIONS, + True, + "no in-flight mutations or critical sections in impact report", + ) + ) + + # 2. New assignment halted (maintenance-drain entered). + checks.append( + DrainCheck( + CHECK_ASSIGNMENTS_STOPPED, + _bool_input(drain_state.get("assignments_stopped")), + "new work assignment halted" + if _bool_input(drain_state.get("assignments_stopped")) + else "assignments not confirmed stopped (fail closed)", + ) + ) + + # 3. Durable checkpoints complete (#660). Unknown => not complete. + cp = drain_state.get("checkpoints_complete") + checks.append( + DrainCheck( + CHECK_CHECKPOINTS_COMPLETE, + _bool_input(cp), + "all live sessions checkpointed" + if _bool_input(cp) + else "checkpoint completeness unconfirmed (fail closed)", + ) + ) + + # 4. Handoffs verified. + checks.append( + DrainCheck( + CHECK_HANDOFFS_OK, + _bool_input(drain_state.get("handoffs_verified")), + "pending handoffs verified" + if _bool_input(drain_state.get("handoffs_verified")) + else "handoffs not verified (fail closed)", + ) + ) + + # 5. Leases resolved/transferred/preserved AND none left disruptive. Requires + # both the drain-mode assertion and the report showing no disruptive lease. + leases_asserted = _bool_input(drain_state.get("leases_handled")) + if not leases_asserted: + checks.append( + DrainCheck( + CHECK_LEASES_HANDLED, + False, + "lease disposition not asserted by drain (fail closed)", + ) + ) + elif disruptive: + checks.append( + DrainCheck( + CHECK_LEASES_HANDLED, + False, + f"{len(disruptive)} disruptive lease(s) still active in report", + ) + ) + else: + checks.append( + DrainCheck( + CHECK_LEASES_HANDLED, + True, + "leases resolved/transferred/preserved; none left disruptive", + ) + ) + + # 6. Acknowledgements received, or an explicit timeout policy was applied. + acks = drain_state.get("acks") or {} + ack_values = list(acks.values()) if isinstance(acks, Mapping) else [] + all_acked = bool(ack_values) and all( + str(v).strip().lower() in {"ack", "acked", "acknowledged"} + for v in ack_values + ) + no_sessions_to_ack = isinstance(acks, Mapping) and len(ack_values) == 0 + timeout_policy = _bool_input(drain_state.get("ack_timeout_policy_applied")) + acks_ok = all_acked or no_sessions_to_ack or timeout_policy + if acks_ok: + if timeout_policy and not all_acked: + detail = "explicit ack timeout policy applied" + elif no_sessions_to_ack: + detail = "no other live sessions required to acknowledge" + else: + detail = "all affected sessions acknowledged" + else: + detail = "outstanding acks with no timeout policy (fail closed)" + checks.append(DrainCheck(CHECK_ACKS_OR_TIMEOUT, acks_ok, detail)) + + return checks + + +def build_drain_proof( + *, + impact_report: Mapping[str, Any], + drain_state: Mapping[str, Any], + requesting_session_id: str | None = None, + now: datetime | None = None, + ttl_seconds: int = DEFAULT_PROOF_TTL_SECONDS, + secret: bytes | None = None, +) -> DrainProof: + """Mint a drain proof from an impact report and the drain-mode outcomes. + + A successful drain (every checklist item passes) yields a *clean* proof with + a valid signature (AC#2). An unclean drain still yields a signed proof, but + with ``clean=False`` and the failing checks named — the gate will deny it — + so the artifact is auditable rather than silently absent. + """ + + moment = now or _utc_now() + ttl = max(1, int(ttl_seconds)) + issued_at = moment.isoformat() + expires_at = (moment + timedelta(seconds=ttl)).isoformat() + fingerprint = impact_fingerprint(impact_report) + + checks = _evaluate_checks(impact_report, drain_state) + failed = [c.name for c in checks if not c.passed] + clean = not failed + + body = _proof_body( + clean=clean, + issued_at=issued_at, + expires_at=expires_at, + requesting_session_id=requesting_session_id, + fingerprint=fingerprint, + checks=checks, + ) + proof_id = _sign(body, _resolve_secret(secret)) + + return DrainProof( + version=DRAIN_PROOF_VERSION, + proof_id=proof_id, + clean=clean, + issued_at=issued_at, + expires_at=expires_at, + requesting_session_id=requesting_session_id, + impact_fingerprint=fingerprint, + checks=checks, + failed_checks=failed, + ) + + +def verify_drain_proof( + proof: Mapping[str, Any] | None, + *, + now: datetime | None = None, + secret: bytes | None = None, + expected_impact_fingerprint: str | None = None, +) -> VerifyResult: + """Verify a drain proof, failing closed on any doubt. + + A proof is valid only when: it is present and well-formed; its signature + recomputes with the per-process secret (not forged/tampered, not minted in a + prior process); it has not expired; every required check is present and + passed; and — when ``expected_impact_fingerprint`` is supplied — it was + minted against the current blast-radius state. + """ + + moment = now or _utc_now() + reasons: list[str] = [] + + if not isinstance(proof, Mapping): + return VerifyResult( + valid=False, + reasons=["drain proof missing or not an object (fail closed)"], + proof_id=None, + clean=False, + expired=False, + tampered=False, + ) + + proof_id = proof.get("proof_id") + presented_clean = bool(proof.get("clean", False)) + + # Rebuild the signed body from the presented fields and re-sign. Any mutation + # of a covered field (including flipping ``clean`` to True) breaks the match. + raw_checks = proof.get("checks") + checks: list[DrainCheck] = [] + checks_wellformed = isinstance(raw_checks, Sequence) and not isinstance( + raw_checks, (str, bytes) + ) + if checks_wellformed: + for c in raw_checks: + if not isinstance(c, Mapping) or "name" not in c or "passed" not in c: + checks_wellformed = False + break + checks.append( + DrainCheck( + name=str(c.get("name")), + passed=bool(c.get("passed")), + detail=str(c.get("detail") or ""), + ) + ) + + tampered = False + if not checks_wellformed: + reasons.append("drain proof checks malformed (fail closed)") + tampered = True + else: + body = _proof_body( + clean=presented_clean, + issued_at=str(proof.get("issued_at") or ""), + expires_at=str(proof.get("expires_at") or ""), + requesting_session_id=proof.get("requesting_session_id"), + fingerprint=str(proof.get("impact_fingerprint") or ""), + checks=checks, + ) + expected_sig = _sign(body, _resolve_secret(secret)) + if not ( + isinstance(proof_id, str) + and hmac.compare_digest(expected_sig, proof_id) + ): + tampered = True + reasons.append( + "drain proof signature mismatch: forged, tampered, or minted " + "in a prior process (fail closed)" + ) + + expires = _parse_ts(proof.get("expires_at")) + expired = expires is None or moment >= expires + if expires is None: + reasons.append("drain proof has no valid expiry (fail closed)") + elif expired: + reasons.append(f"drain proof expired at {proof.get('expires_at')}") + + # Recompute cleanliness from the checks themselves — never trust the flag. + recomputed_failed = [c.name for c in checks if not c.passed] + present_names = {c.name for c in checks} + missing = [name for name in REQUIRED_CHECKS if name not in present_names] + recomputed_clean = checks_wellformed and not recomputed_failed and not missing + if missing: + reasons.append(f"drain proof missing required checks: {', '.join(missing)}") + if checks_wellformed and recomputed_failed: + reasons.append( + f"drain checks failed: {', '.join(sorted(set(recomputed_failed)))}" + ) + if presented_clean and not recomputed_clean: + tampered = True + reasons.append("proof claims clean but its checks do not support it") + + if expected_impact_fingerprint is not None: + if str(proof.get("impact_fingerprint") or "") != str( + expected_impact_fingerprint + ): + reasons.append( + "drain proof was minted against a different blast-radius state " + "(stale; fail closed)" + ) + + valid = (not tampered) and (not expired) and recomputed_clean and not reasons + return VerifyResult( + valid=valid, + reasons=reasons, + proof_id=proof_id if isinstance(proof_id, str) else None, + clean=recomputed_clean, + expired=expired, + tampered=tampered, + ) + + +def _incident_descriptor( + *, + reasons: Sequence[str], + requesting_session_id: str | None, + proof_id: str | None, + at: str, +) -> dict[str, Any]: + """Durable incident descriptor for a denied restart (caller creates issue). + + Kept as data (not a live Gitea call) so this module stays pure and testable; + the MCP tool layer turns it into a durable issue via the incident bridge. + """ + + return { + "kind": "restart_drain_gate_denied", + "title": "Restart denied: drain proof failed the hard gate", + "labels": ["mcp-health", "safety", "stale-runtime", "workflow-hardening"], + "reasons": list(reasons), + "requesting_session_id": requesting_session_id, + "proof_id": proof_id, + "at": at, + "drain_proof_version": DRAIN_PROOF_VERSION, + } + + +def gate_apply_restart( + *, + proof: Mapping[str, Any] | None, + now: datetime | None = None, + secret: bytes | None = None, + break_glass: bool = False, + expected_impact_fingerprint: str | None = None, + requesting_session_id: str | None = None, +) -> GateDecision: + """Hard gate for a sanctioned restart apply (#661 AC#1/#3). + + Restart apply is authorized only with a valid, unexpired, clean drain proof. + A missing, expired, tampered, or unclean proof denies the restart and emits + a durable incident descriptor. ``break_glass`` is the *only* sanctioned + bypass — authorization for it is the caller's responsibility (the emergency + workflow is a separate child); when set, the gate allows without a proof but + records the bypass so it is never silent. + """ + + moment = now or _utc_now() + at = moment.isoformat() + + if break_glass: + reasons = ["break-glass restart authorized; drain proof gate bypassed"] + audit = { + "event": "restart_gate_evaluated", + "drain_proof_version": DRAIN_PROOF_VERSION, + "evaluated_at": at, + "verdict": GATE_BREAK_GLASS, + "allow": True, + "break_glass": True, + "requesting_session_id": requesting_session_id, + "proof_id": None, + } + return GateDecision( + allow=True, + verdict=GATE_BREAK_GLASS, + reasons=reasons, + proof_id=None, + break_glass=True, + incident=None, + audit_record=audit, + ) + + result = verify_drain_proof( + proof, + now=moment, + secret=secret, + expected_impact_fingerprint=expected_impact_fingerprint, + ) + + if result.valid: + reasons = ["valid unexpired clean drain proof present; restart authorized"] + audit = { + "event": "restart_gate_evaluated", + "drain_proof_version": DRAIN_PROOF_VERSION, + "evaluated_at": at, + "verdict": GATE_ALLOW, + "allow": True, + "break_glass": False, + "requesting_session_id": requesting_session_id, + "proof_id": result.proof_id, + } + return GateDecision( + allow=True, + verdict=GATE_ALLOW, + reasons=reasons, + proof_id=result.proof_id, + break_glass=False, + incident=None, + audit_record=audit, + ) + + deny_reasons = ["restart denied: drain proof invalid (fail closed)"] + list( + result.reasons + ) + incident = _incident_descriptor( + reasons=deny_reasons, + requesting_session_id=requesting_session_id, + proof_id=result.proof_id, + at=at, + ) + audit = { + "event": "restart_gate_evaluated", + "drain_proof_version": DRAIN_PROOF_VERSION, + "evaluated_at": at, + "verdict": GATE_DENY, + "allow": False, + "break_glass": False, + "requesting_session_id": requesting_session_id, + "proof_id": result.proof_id, + "incident_kind": incident["kind"], + } + return GateDecision( + allow=False, + verdict=GATE_DENY, + reasons=deny_reasons, + proof_id=result.proof_id, + break_glass=False, + incident=incident, + audit_record=audit, + ) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 0b76dde..fa63126 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2068,6 +2068,7 @@ import lease_lifecycle # noqa: E402 import lease_policy # noqa: E402 import workflow_dashboard # noqa: E402 # #605 live queue/lease dashboard import restart_coordinator # noqa: E402 # #658 MCP restart coordinator/impact +import drain_proof # noqa: E402 # #661 pre-restart drain proof and hard gate import incident_bridge # noqa: E402 import sentry_observability # noqa: E402 (#606 optional Sentry observability) import sentry_incident_bridge # noqa: E402 (#607 Sentry→Gitea incident bridge) @@ -22342,6 +22343,8 @@ def gitea_request_mcp_restart( request_override: bool = False, session_id: str | None = None, limit: int = 200, + drain_proof_json: str | None = None, + request_break_glass: bool = False, ) -> dict: """Evaluate a proposed MCP restart and return an impact preview (#658). @@ -22351,10 +22354,15 @@ def gitea_request_mcp_restart( verdict, so the console (#642/#652) and operators can see what a restart would disrupt *before* any concurrent LLM work is destroyed. - This tool is **dry-run and never restarts anything.** The mutative apply - path is a separate child gated by a drain proof (non-goal here); calling - with ``dry_run=False`` still performs no restart and reports that apply is - not yet available. + This tool **never restarts a process.** In dry-run (the default) it returns + only the impact preview. With ``dry_run=False`` it enforces the #661 hard + gate: the apply request must present a valid, unexpired, clean drain proof + (``drain_proof_json``) or it is denied and a durable incident descriptor is + returned under ``incident``. Break-glass is the only bypass and is honoured + only when ``request_break_glass`` is set *and* the environment carries + ``GITEA_BREAKGLASS_RESTART_AUTHORIZATION``. Even an authorized gate performs + no restart here; actual execution is a further child. The gate outcome is + reported under ``apply_gate`` / ``apply_authorized``. Operator override authority is read from the process environment (``GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION``), never self-asserted by @@ -22473,12 +22481,54 @@ def gitea_request_mcp_restart( payload["requesting_session_id"] = sid payload["operator_override_requested"] = bool(request_override) payload["operator_override_authorized"] = operator_authorized + # Actual restart execution remains a further child; this tool never restarts + # a process. What #661 adds is the *hard gate*: an apply request (dry_run + # False) must present a valid, unexpired, clean drain proof, or it is denied + # and a durable incident is raised. Break-glass is the only bypass and its + # authorization is read from the environment, never self-asserted. payload["apply_supported"] = False if not dry_run: - payload["reasons"] = list(payload.get("reasons") or []) + [ - "apply requested but not supported: sanctioned restart apply is " - "gated by a drain proof (separate child); no restart performed (#658)" - ] + proof_obj: dict | None = None + proof_parse_error: str | None = None + if drain_proof_json: + try: + parsed = json.loads(drain_proof_json) + proof_obj = parsed if isinstance(parsed, dict) else None + if proof_obj is None: + proof_parse_error = "drain_proof_json is not a JSON object" + except (ValueError, TypeError) as exc: + proof_parse_error = f"invalid drain_proof_json: {_redact(str(exc))}" + + break_glass_authorized = bool( + ( + os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or "" + ).strip() + ) + break_glass = bool(request_break_glass and break_glass_authorized) + + expected_fp = drain_proof.impact_fingerprint(report.as_dict()) + gate = drain_proof.gate_apply_restart( + proof=proof_obj, + break_glass=break_glass, + expected_impact_fingerprint=expected_fp, + requesting_session_id=sid, + ) + gate_payload = gate.as_dict() + if proof_parse_error and not break_glass: + gate_payload["reasons"] = [proof_parse_error] + list( + gate_payload.get("reasons") or [] + ) + payload["apply_gate"] = gate_payload + payload["apply_authorized"] = gate.allow + payload["break_glass_requested"] = bool(request_break_glass) + payload["break_glass_authorized"] = break_glass_authorized + # Even an authorized gate performs no restart here: execution is a later + # child. The gate proves the apply path *would* be permitted. + payload["reasons"] = list(payload.get("reasons") or []) + list( + gate_payload.get("reasons") or [] + ) + if not gate.allow and gate.incident is not None: + payload["incident"] = gate.incident return payload diff --git a/tests/test_drain_proof.py b/tests/test_drain_proof.py new file mode 100644 index 0000000..eda7d54 --- /dev/null +++ b/tests/test_drain_proof.py @@ -0,0 +1,383 @@ +"""Tests for the pre-restart drain proof and hard gate (#661). + +Covers the acceptance criteria: + +1. Restart apply without a proof fails closed. +2. A successful drain produces a verifiable proof. +3. An open unsafe mutation makes the proof fail (multi-session fixture). +4. Pass / fail / expired verification paths. + +Plus the security posture: forged/tampered proofs are rejected, break-glass is +the only bypass and is never silent, a stale blast-radius fingerprint rejects a +proof, and no per-process secret ever leaks into a serialized artifact. +""" + +from __future__ import annotations + +import os +import unittest +from datetime import datetime, timedelta, timezone + +import drain_proof as dp +import restart_coordinator as rc + + +NOW = datetime(2026, 7, 24, 6, 0, 0, tzinfo=timezone.utc) +SECRET = b"unit-test-drain-proof-secret-0123456789abcdef" + + +def _live_pid() -> int: + return os.getpid() + + +def _clean_drain_state() -> dict: + """Every drain action succeeded, no sessions outstanding.""" + + return { + "assignments_stopped": True, + "checkpoints_complete": True, + "handoffs_verified": True, + "leases_handled": True, + "acks": {}, # no other live sessions to acknowledge + "ack_timeout_policy_applied": False, + } + + +def _safe_report() -> dict: + """Impact report with no other live work: a restart here is safe.""" + + report = rc.evaluate_restart_impact( + {"sessions": [], "leases": [], "inventory_complete": True}, + now=NOW, + requesting_session_id="prgs-controller-1-req", + ) + return report.as_dict() + + +def _unsafe_mutation_report() -> dict: + """Multi-session report: a second session holds a live author mutation.""" + + sessions = [ + { + "session_id": "prgs-controller-1-req", + "role": "controller", + "profile": "prgs-controller", + "pid": _live_pid(), + "status": "active", + "last_heartbeat_at": NOW.isoformat(), + }, + { + "session_id": "prgs-author-99", + "role": "author", + "profile": "prgs-author", + "pid": _live_pid(), + "status": "active", + "last_heartbeat_at": NOW.isoformat(), + }, + ] + leases = [ + { + "lease_id": "lease-mut", + "session_id": "prgs-author-99", + "role": "author", + "phase": "implementing", + "work_kind": "issue", + "work_number": 661, + "worktree_path": "branches/issue-661", + "freshness": {"freshness": "active"}, + } + ] + report = rc.evaluate_restart_impact( + {"sessions": sessions, "leases": leases, "inventory_complete": True}, + now=NOW, + requesting_session_id="prgs-controller-1-req", + ) + return report.as_dict() + + +class BuildDrainProofTests(unittest.TestCase): + def test_clean_drain_produces_verifiable_clean_proof(self): + """AC#2: a successful drain produces a verifiable proof.""" + + proof = dp.build_drain_proof( + impact_report=_safe_report(), + drain_state=_clean_drain_state(), + requesting_session_id="prgs-controller-1-req", + now=NOW, + secret=SECRET, + ) + self.assertTrue(proof.clean) + self.assertEqual(proof.failed_checks, []) + self.assertEqual( + {c.name for c in proof.checks}, set(dp.REQUIRED_CHECKS) + ) + result = dp.verify_drain_proof( + proof.as_dict(), now=NOW, secret=SECRET + ) + self.assertTrue(result.valid, result.reasons) + self.assertFalse(result.expired) + self.assertFalse(result.tampered) + + def test_open_mutation_makes_proof_unclean(self): + """AC#3: an unsafe mutation still in flight fails the proof.""" + + proof = dp.build_drain_proof( + impact_report=_unsafe_mutation_report(), + drain_state=_clean_drain_state(), + now=NOW, + secret=SECRET, + ) + self.assertFalse(proof.clean) + self.assertIn(dp.CHECK_NO_INFLIGHT_MUTATIONS, proof.failed_checks) + # Leases-handled also fails: the report still shows a disruptive lease. + self.assertIn(dp.CHECK_LEASES_HANDLED, proof.failed_checks) + result = dp.verify_drain_proof(proof.as_dict(), now=NOW, secret=SECRET) + self.assertFalse(result.valid) + + def test_incomplete_inventory_fails_no_mutations_check(self): + proof = dp.build_drain_proof( + impact_report={"inventory_complete": False}, + drain_state=_clean_drain_state(), + now=NOW, + secret=SECRET, + ) + self.assertFalse(proof.clean) + self.assertIn(dp.CHECK_NO_INFLIGHT_MUTATIONS, proof.failed_checks) + + def test_missing_checkpoint_flag_fails_closed(self): + state = _clean_drain_state() + del state["checkpoints_complete"] + proof = dp.build_drain_proof( + impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET + ) + self.assertFalse(proof.clean) + self.assertIn(dp.CHECK_CHECKPOINTS_COMPLETE, proof.failed_checks) + + def test_non_true_flags_fail_closed(self): + """A truthy-but-not-True value (e.g. the string 'yes') must not pass.""" + + state = _clean_drain_state() + state["assignments_stopped"] = "yes" + proof = dp.build_drain_proof( + impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET + ) + self.assertIn(dp.CHECK_ASSIGNMENTS_STOPPED, proof.failed_checks) + + def test_ack_timeout_policy_satisfies_ack_check(self): + state = _clean_drain_state() + state["acks"] = {"prgs-author-99": "pending"} + state["ack_timeout_policy_applied"] = True + proof = dp.build_drain_proof( + impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET + ) + names = {c.name: c.passed for c in proof.checks} + self.assertTrue(names[dp.CHECK_ACKS_OR_TIMEOUT]) + + def test_outstanding_acks_without_timeout_fail(self): + state = _clean_drain_state() + state["acks"] = {"prgs-author-99": "pending"} + state["ack_timeout_policy_applied"] = False + proof = dp.build_drain_proof( + impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET + ) + self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks) + + def test_all_acked_satisfies_ack_check(self): + state = _clean_drain_state() + state["acks"] = {"prgs-author-99": "acked", "prgs-author-2": "acknowledged"} + proof = dp.build_drain_proof( + impact_report=_safe_report(), drain_state=state, now=NOW, secret=SECRET + ) + names = {c.name: c.passed for c in proof.checks} + self.assertTrue(names[dp.CHECK_ACKS_OR_TIMEOUT]) + + +class VerifyDrainProofTests(unittest.TestCase): + def _clean_proof_dict(self) -> dict: + return dp.build_drain_proof( + impact_report=_safe_report(), + drain_state=_clean_drain_state(), + now=NOW, + secret=SECRET, + ).as_dict() + + def test_missing_proof_is_invalid(self): + result = dp.verify_drain_proof(None, now=NOW, secret=SECRET) + self.assertFalse(result.valid) + self.assertIsNone(result.proof_id) + + def test_expired_proof_is_invalid(self): + """AC#4: an expired proof fails verification.""" + + proof = self._clean_proof_dict() + later = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS + 1) + result = dp.verify_drain_proof(proof, now=later, secret=SECRET) + self.assertFalse(result.valid) + self.assertTrue(result.expired) + + def test_proof_valid_just_before_expiry(self): + proof = self._clean_proof_dict() + almost = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS - 1) + result = dp.verify_drain_proof(proof, now=almost, secret=SECRET) + self.assertTrue(result.valid, result.reasons) + + def test_wrong_secret_rejected(self): + """A proof minted in a prior process (different secret) will not verify.""" + + proof = self._clean_proof_dict() + result = dp.verify_drain_proof(proof, now=NOW, secret=b"other-secret") + self.assertFalse(result.valid) + self.assertTrue(result.tampered) + + def test_flipping_clean_flag_is_detected(self): + """Forging clean=True on an unclean proof breaks the signature.""" + + unclean = dp.build_drain_proof( + impact_report=_unsafe_mutation_report(), + drain_state=_clean_drain_state(), + now=NOW, + secret=SECRET, + ).as_dict() + self.assertFalse(unclean["clean"]) + unclean["clean"] = True # forge + result = dp.verify_drain_proof(unclean, now=NOW, secret=SECRET) + self.assertFalse(result.valid) + self.assertTrue(result.tampered) + + def test_tampering_a_check_is_detected(self): + unclean = dp.build_drain_proof( + impact_report=_unsafe_mutation_report(), + drain_state=_clean_drain_state(), + now=NOW, + secret=SECRET, + ).as_dict() + for c in unclean["checks"]: + if c["name"] == dp.CHECK_NO_INFLIGHT_MUTATIONS: + c["passed"] = True # forge the failing check to pass + result = dp.verify_drain_proof(unclean, now=NOW, secret=SECRET) + self.assertFalse(result.valid) + self.assertTrue(result.tampered) + + def test_missing_required_check_rejected(self): + proof = self._clean_proof_dict() + proof["checks"] = [ + c for c in proof["checks"] if c["name"] != dp.CHECK_HANDOFFS_OK + ] + result = dp.verify_drain_proof(proof, now=NOW, secret=SECRET) + self.assertFalse(result.valid) + + def test_stale_fingerprint_rejected(self): + proof = self._clean_proof_dict() + result = dp.verify_drain_proof( + proof, + now=NOW, + secret=SECRET, + expected_impact_fingerprint="deadbeef", + ) + self.assertFalse(result.valid) + + def test_matching_fingerprint_accepted(self): + report = _safe_report() + proof = dp.build_drain_proof( + impact_report=report, + drain_state=_clean_drain_state(), + now=NOW, + secret=SECRET, + ).as_dict() + fp = dp.impact_fingerprint(report) + result = dp.verify_drain_proof( + proof, now=NOW, secret=SECRET, expected_impact_fingerprint=fp + ) + self.assertTrue(result.valid, result.reasons) + + +class GateApplyRestartTests(unittest.TestCase): + def _clean_proof_dict(self) -> dict: + return dp.build_drain_proof( + impact_report=_safe_report(), + drain_state=_clean_drain_state(), + now=NOW, + secret=SECRET, + ).as_dict() + + def test_apply_without_proof_denied(self): + """AC#1: restart apply without a proof fails closed + raises incident.""" + + decision = dp.gate_apply_restart(proof=None, now=NOW, secret=SECRET) + self.assertFalse(decision.allow) + self.assertEqual(decision.verdict, dp.GATE_DENY) + self.assertIsNotNone(decision.incident) + self.assertEqual( + decision.incident["kind"], "restart_drain_gate_denied" + ) + + def test_apply_with_valid_proof_allowed(self): + decision = dp.gate_apply_restart( + proof=self._clean_proof_dict(), now=NOW, secret=SECRET + ) + self.assertTrue(decision.allow) + self.assertEqual(decision.verdict, dp.GATE_ALLOW) + self.assertIsNone(decision.incident) + + def test_apply_with_expired_proof_denied_with_incident(self): + later = NOW + timedelta(seconds=dp.DEFAULT_PROOF_TTL_SECONDS + 5) + decision = dp.gate_apply_restart( + proof=self._clean_proof_dict(), now=later, secret=SECRET + ) + self.assertFalse(decision.allow) + self.assertIsNotNone(decision.incident) + + def test_apply_with_unclean_proof_denied(self): + """AC#3 at the gate: an unsafe-mutation proof is denied.""" + + unclean = dp.build_drain_proof( + impact_report=_unsafe_mutation_report(), + drain_state=_clean_drain_state(), + now=NOW, + secret=SECRET, + ).as_dict() + decision = dp.gate_apply_restart(proof=unclean, now=NOW, secret=SECRET) + self.assertFalse(decision.allow) + self.assertIsNotNone(decision.incident) + + def test_break_glass_allows_without_proof_but_records_bypass(self): + decision = dp.gate_apply_restart( + proof=None, now=NOW, secret=SECRET, break_glass=True + ) + self.assertTrue(decision.allow) + self.assertEqual(decision.verdict, dp.GATE_BREAK_GLASS) + self.assertTrue(decision.break_glass) + self.assertIsNone(decision.incident) + self.assertTrue(decision.audit_record["break_glass"]) + + def test_denied_gate_carries_stale_fingerprint_reason(self): + decision = dp.gate_apply_restart( + proof=self._clean_proof_dict(), + now=NOW, + secret=SECRET, + expected_impact_fingerprint="not-the-fingerprint", + ) + self.assertFalse(decision.allow) + + +class SecretHygieneTests(unittest.TestCase): + def test_secret_never_serialized(self): + proof = dp.build_drain_proof( + impact_report=_safe_report(), + drain_state=_clean_drain_state(), + now=NOW, + secret=SECRET, + ) + blob = dp._canonical(proof.as_dict()) + self.assertNotIn(SECRET.decode(), blob) + # The signature is a hex digest, not the raw secret. + self.assertNotIn(SECRET.hex(), blob) + + def test_incident_descriptor_has_no_secret(self): + decision = dp.gate_apply_restart(proof=None, now=NOW, secret=SECRET) + blob = dp._canonical(decision.incident) + self.assertNotIn(SECRET.decode(), blob) + + +if __name__ == "__main__": + unittest.main()