"""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, )