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 1/3] 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() From 824c42f7e3f436eef05272276c9ced958c37a945 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 24 Jul 2026 22:36:25 -0400 Subject: [PATCH 2/3] fix(drain): fail closed on missing or unproven acknowledgement evidence (#661) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The acks_or_timeout check treated an absent `acks` key as proof that no session needed to acknowledge: `drain_state.get("acks") or {}` collapsed absent, None, and empty into the same value, and the resulting empty mapping satisfied `no_sessions_to_ack`. The impact report's counts.sessions_live_other was never consulted, so absence of evidence was read as evidence of absence. Reproduced at head 1cbbde00895dce1545290a660bc31c4ac9210b3e: with sessions_live_other = 3 and the acknowledgement key absent, acks_or_timeout passed with detail "no other live sessions required to acknowledge", the proof minted clean, and gate_apply_restart returned verdict allow — a restart authorized against three live sessions with zero acknowledgement evidence, and the resulting artifact carried a valid signature. Whether acknowledgement is required is now derived from the impact report, never from the shape of the drain state: - _live_session_count() reads counts.sessions_live_other and returns None for a missing, malformed, negative, or bool value, so an unreadable report fails closed instead of reading as "nobody was live". - Absent, None, non-mapping, empty, partially-covering, and unparseable or stale acknowledgement data all fail closed while live sessions require acknowledgement. - _is_acknowledged() no longer coerces with str(); only an explicit "ack"/"acked"/"acknowledged" string counts, so None, timestamps, and "pending"/"stale" markers are never read as an acknowledgement. - Present-but-unacknowledged entries fail closed even when the report claims zero live sessions: that contradiction is not safe to resolve in favour of the restart. - ack_timeout_policy_applied stays strict (`value is True`), so an absent, null, or non-boolean value cannot open the gate on its own. The genuine no-other-live-sessions case still passes, now justified by the report proving sessions_live_other == 0 rather than by the absence of data. Adds AcknowledgementFailClosedTests: 14 cases / 26 subtests covering missing, null, empty, malformed, stale, partial-coverage, and unproven-count inputs, the valid-acknowledgement and zero-live-session paths, timeout-policy strictness, and that a failed check blocks proof.clean, verification, and the restart gate. Restart-surface suite: 132 passed, 38 subtests (branch baseline 118 passed, 12 subtests; +14 new tests, no regressions). Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01VEaP3TohHLFWkp3Z2mmuZw --- drain_proof.py | 137 +++++++++++++++++++++--- tests/test_drain_proof.py | 213 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 334 insertions(+), 16 deletions(-) diff --git a/drain_proof.py b/drain_proof.py index d81c26e..5eec7b6 100644 --- a/drain_proof.py +++ b/drain_proof.py @@ -71,6 +71,11 @@ CHECK_HANDOFFS_OK = "handoffs_ok" CHECK_LEASES_HANDLED = "leases_handled" CHECK_ACKS_OR_TIMEOUT = "acks_or_timeout" +# The only values accepted as an explicit acknowledgement from a live session. +# Anything else — including a missing entry, a null, or a "pending"/"stale" +# marker — leaves that session unacknowledged. +_ACK_TOKENS = frozenset({"ack", "acked", "acknowledged"}) + REQUIRED_CHECKS: tuple[str, ...] = ( CHECK_NO_INFLIGHT_MUTATIONS, CHECK_ASSIGNMENTS_STOPPED, @@ -291,6 +296,38 @@ def _bool_input(value: Any) -> bool: return value is True +def _live_session_count(impact_report: Mapping[str, Any]) -> int | None: + """Other-live-session count from the report, or ``None`` when unproven. + + Only a real, non-negative integer counts. A missing ``counts`` block, a + malformed one, a non-integer, a bool (``True`` is an ``int`` in Python), or + a negative value all return ``None`` so the caller fails closed rather than + treating an unreadable report as "nobody was live". + """ + + counts = impact_report.get("counts") + if not isinstance(counts, Mapping): + return None + value = counts.get("sessions_live_other") + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + return None + return value + + +def _is_acknowledged(value: Any) -> bool: + """True only for an explicit acknowledgement token. + + Deliberately strict: the value must already be a string carrying one of the + recognised tokens. Non-strings are not coerced, so ``None``, timestamps, + objects, and states such as "pending" or "stale" are never read as an + acknowledgement. + """ + + if not isinstance(value, str): + return False + return value.strip().lower() in _ACK_TOKENS + + def _evaluate_checks( impact_report: Mapping[str, Any], drain_state: Mapping[str, Any], @@ -406,24 +443,92 @@ def _evaluate_checks( ) # 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 + # + # Whether acknowledgements are *required* is answered by the impact report, + # never inferred from the shape of the drain state. Previously an absent + # ``acks`` key collapsed to ``{}`` and was read as "no session needed to + # acknowledge", so a proof minted clean — and the restart gate allowed — + # while the report still showed other live sessions. Absence of evidence is + # not evidence of absence: missing, malformed, stale, or otherwise unproven + # acknowledgement data fails closed whenever live sessions require + # acknowledgement, and only explicitly verified acknowledgement evidence + # may permit the operation. + sessions_live_other = _live_session_count(impact_report) + live_count_known = sessions_live_other is not None + # An unknown or malformed count fails closed: it cannot prove nobody had to + # acknowledge, so acknowledgement stays required. + acks_required = (not live_count_known) or sessions_live_other > 0 + + raw_acks = drain_state.get("acks") + acks_present = isinstance(raw_acks, Mapping) + ack_values = list(raw_acks.values()) if acks_present else [] + acked_count = sum(1 for value in ack_values if _is_acknowledged(value)) + every_entry_acked = bool(ack_values) and acked_count == len(ack_values) + covers_live_sessions = live_count_known and acked_count >= sessions_live_other + # Strict by construction: only an explicit ``True`` counts, so an absent, + # null, or non-boolean ``ack_timeout_policy_applied`` can never open this + # gate on its own. 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" + + # Present-but-unacknowledged entries always fail closed, even when the + # report claims nobody was live: the drain state naming an outstanding + # session contradicts that claim, and the safe reading of a contradiction + # is that an acknowledgement is still owed. + outstanding_entries = acks_present and bool(ack_values) and not every_entry_acked + + if timeout_policy and (acks_required or outstanding_entries): + acks_ok = True + detail = "explicit ack timeout policy applied and recorded" + elif outstanding_entries: + acks_ok = False + detail = ( + f"{len(ack_values) - acked_count} of {len(ack_values)} " + "acknowledgement entries are unparseable, stale, or not " + "acknowledged (fail closed)" + ) + elif not acks_required: + acks_ok = True + detail = ( + "impact report proves no other live sessions required to " + "acknowledge (sessions_live_other=0)" + ) + elif every_entry_acked and covers_live_sessions: + acks_ok = True + detail = ( + f"{acked_count} acknowledgement entr" + f"{'y' if acked_count == 1 else 'ies'} verified; covers " + f"sessions_live_other={sessions_live_other}" + ) else: - detail = "outstanding acks with no timeout policy (fail closed)" + acks_ok = False + if not live_count_known: + detail = ( + "impact report does not prove the live-session count; " + "acknowledgement required and unproven (fail closed)" + ) + elif raw_acks is None: + detail = ( + "acknowledgement evidence absent while " + f"sessions_live_other={sessions_live_other} require " + "acknowledgement (fail closed)" + ) + elif not acks_present: + detail = ( + "acknowledgement evidence malformed: expected a mapping, got " + f"{type(raw_acks).__name__} (fail closed)" + ) + elif not ack_values: + detail = ( + "acknowledgement mapping empty while " + f"sessions_live_other={sessions_live_other} require " + "acknowledgement (fail closed)" + ) + else: + detail = ( + f"acknowledgements cover only {acked_count} session(s) but the " + f"report shows sessions_live_other={sessions_live_other} " + "(fail closed)" + ) checks.append(DrainCheck(CHECK_ACKS_OR_TIMEOUT, acks_ok, detail)) return checks diff --git a/tests/test_drain_proof.py b/tests/test_drain_proof.py index eda7d54..410af33 100644 --- a/tests/test_drain_proof.py +++ b/tests/test_drain_proof.py @@ -379,5 +379,218 @@ class SecretHygieneTests(unittest.TestCase): self.assertNotIn(SECRET.decode(), blob) +def _drained_report_with_live_sessions(count: int) -> dict: + """Report with ``count`` other live sessions but nothing in flight. + + Every other checklist item passes against this report, so a failure + isolates the acknowledgement check rather than tripping on mutations. + """ + + sessions = [ + { + "session_id": "prgs-controller-1-req", + "role": "controller", + "profile": "prgs-controller", + "pid": _live_pid(), + "status": "active", + "last_heartbeat_at": NOW.isoformat(), + } + ] + for index in range(count): + sessions.append( + { + "session_id": f"prgs-author-{index}", + "role": "author", + "profile": "prgs-author", + "pid": _live_pid(), + "status": "active", + "last_heartbeat_at": NOW.isoformat(), + } + ) + report = rc.evaluate_restart_impact( + {"sessions": sessions, "leases": [], "inventory_complete": True}, + now=NOW, + requesting_session_id="prgs-controller-1-req", + ) + return report.as_dict() + + +class AcknowledgementFailClosedTests(unittest.TestCase): + """Acknowledgement evidence must fail closed unless explicitly verified. + + Regression cover for the reviewed fail-open on PR #882: an absent ``acks`` + key collapsed to ``{}`` and was read as "no other live sessions required to + acknowledge", so a proof minted clean and the restart gate allowed while the + impact report still showed other live sessions. + """ + + def _state(self, **overrides) -> dict: + state = _clean_drain_state() + state.pop("acks", None) + state["ack_timeout_policy_applied"] = False + state.update(overrides) + return state + + def _acks_check(self, proof) -> dp.DrainCheck: + return next(c for c in proof.checks if c.name == dp.CHECK_ACKS_OR_TIMEOUT) + + def _build(self, report: dict, state: dict): + return dp.build_drain_proof( + impact_report=report, drain_state=state, now=NOW, secret=SECRET + ) + + def assertAcksFailClosed(self, report: dict, state: dict) -> None: + proof = self._build(report, state) + self.assertFalse(self._acks_check(proof).passed) + self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks) + self.assertFalse(proof.clean) + + # --- missing / null / empty / malformed ------------------------------ + + def test_missing_acks_key_with_live_sessions_fails_closed(self): + """The exact reviewed defect: absent key, three other live sessions.""" + report = _drained_report_with_live_sessions(3) + self.assertEqual(report["counts"]["sessions_live_other"], 3) + state = self._state() + self.assertNotIn("acks", state) + proof = self._build(report, state) + check = self._acks_check(proof) + self.assertFalse(check.passed) + self.assertNotIn("no other live sessions", check.detail) + self.assertIn("fail closed", check.detail) + self.assertFalse(proof.clean) + self.assertEqual(proof.failed_checks, [dp.CHECK_ACKS_OR_TIMEOUT]) + + def test_none_acks_with_live_sessions_fails_closed(self): + self.assertAcksFailClosed( + _drained_report_with_live_sessions(2), self._state(acks=None) + ) + + def test_empty_acks_with_live_sessions_fails_closed(self): + self.assertAcksFailClosed( + _drained_report_with_live_sessions(1), self._state(acks={}) + ) + + def test_malformed_acks_fail_closed(self): + for malformed in ([], "ack", 7, ("ack",), True): + with self.subTest(malformed=malformed): + self.assertAcksFailClosed( + _drained_report_with_live_sessions(1), + self._state(acks=malformed), + ) + + # --- stale / unproven values ----------------------------------------- + + def test_stale_or_unproven_ack_values_fail_closed(self): + for value in ("pending", "stale", "unknown", "", None, True, 1, NOW): + with self.subTest(value=value): + self.assertAcksFailClosed( + _drained_report_with_live_sessions(1), + self._state(acks={"prgs-author-0": value}), + ) + + def test_partial_coverage_fails_closed(self): + """Fewer acknowledgements than the report's live-session count.""" + self.assertAcksFailClosed( + _drained_report_with_live_sessions(3), + self._state(acks={"prgs-author-0": "ack"}), + ) + + def test_one_unacked_entry_among_many_fails_closed(self): + self.assertAcksFailClosed( + _drained_report_with_live_sessions(2), + self._state(acks={"prgs-author-0": "ack", "prgs-author-1": "pending"}), + ) + + def test_unproven_live_session_count_fails_closed(self): + """A missing or malformed count cannot prove nobody had to acknowledge.""" + malformed_counts = ( + None, + {}, + {"sessions_live_other": None}, + {"sessions_live_other": "3"}, + {"sessions_live_other": -1}, + {"sessions_live_other": True}, + ) + for counts in malformed_counts: + with self.subTest(counts=counts): + report = _drained_report_with_live_sessions(0) + if counts is None: + report.pop("counts", None) + else: + report["counts"] = counts + self.assertAcksFailClosed(report, self._state()) + + # --- valid evidence still passes ------------------------------------- + + def test_complete_valid_acks_pass(self): + report = _drained_report_with_live_sessions(2) + state = self._state( + acks={"prgs-author-0": "ack", "prgs-author-1": "acknowledged"} + ) + proof = self._build(report, state) + self.assertTrue(self._acks_check(proof).passed) + self.assertTrue(proof.clean) + self.assertEqual(proof.failed_checks, []) + + def test_no_other_live_sessions_still_passes(self): + """Intended behavior retained: zero live sessions needs no acks.""" + report = _drained_report_with_live_sessions(0) + self.assertEqual(report["counts"]["sessions_live_other"], 0) + proof = self._build(report, self._state()) + check = self._acks_check(proof) + self.assertTrue(check.passed) + self.assertIn("sessions_live_other=0", check.detail) + self.assertTrue(proof.clean) + + # --- timeout policy cannot become a second fail-open ------------------ + + def test_unproven_timeout_policy_cannot_open_the_gate(self): + for value in (None, "true", "yes", 1, "True", [], {}): + with self.subTest(value=value): + self.assertAcksFailClosed( + _drained_report_with_live_sessions(2), + self._state(ack_timeout_policy_applied=value), + ) + + def test_explicit_timeout_policy_permits(self): + proof = self._build( + _drained_report_with_live_sessions(2), + self._state(ack_timeout_policy_applied=True), + ) + check = self._acks_check(proof) + self.assertTrue(check.passed) + self.assertIn("timeout policy", check.detail) + self.assertTrue(proof.clean) + + # --- the gate itself must deny --------------------------------------- + + def test_failed_ack_check_denies_the_restart_gate(self): + report = _drained_report_with_live_sessions(3) + proof = self._build(report, self._state()) + self.assertFalse(proof.clean) + decision = dp.gate_apply_restart( + proof=proof.as_dict(), + now=NOW, + secret=SECRET, + expected_impact_fingerprint=dp.impact_fingerprint(report), + ) + self.assertFalse(decision.allow) + self.assertEqual(decision.verdict, dp.GATE_DENY) + self.assertIsNotNone(decision.incident) + + def test_unclean_ack_proof_fails_verification(self): + report = _drained_report_with_live_sessions(3) + proof = self._build(report, self._state()) + result = dp.verify_drain_proof( + proof.as_dict(), + now=NOW, + secret=SECRET, + expected_impact_fingerprint=dp.impact_fingerprint(report), + ) + self.assertFalse(result.valid) + self.assertFalse(result.clean) + + if __name__ == "__main__": unittest.main() From 3a9d634c17efbac767139569d13bec4b3e51038a Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 24 Jul 2026 23:55:05 -0400 Subject: [PATCH 3/3] fix(drain-proof): bind acknowledgement coverage to session identity (#661) Review 582 (REQUEST_CHANGES at 95178349) found a residual fail-open of the same class the PR set out to close. Acknowledgement coverage was decided by comparing a count against a count: covers_live_sessions = live_count_known and acked_count >= sessions_live_other Nothing bound an acknowledgement to the identity of a session that actually owed one, so acknowledgements supplied for the requesting session and for a session that does not exist satisfied the obligations of two live sessions that never answered - minting a clean, correctly signed proof and an allow verdict from the restart gate. Coverage is now derived from authoritative impact-report evidence: - New `_required_ack_sessions()` derives the required session ids from the report itself, via `ack_state` keys and/or `affected_sessions` filtered on `live and not is_requester`. The requester is excluded only on explicit `is_requester` evidence, never inferred. - When both views are present they must name the same set, and the result is reconciled against `counts.sessions_live_other`. Missing, malformed, duplicated, contradictory, or unreconcilable identity evidence fails closed and outranks every permitting path, including the timeout policy. - Coverage requires every required id to carry an explicit acknowledgement token keyed by that id. Acknowledgements for the requester, for unknown ids, or for fabricated ids never increase coverage. - Caller-supplied acknowledgement cardinality is no longer proof of anything. Failure propagates unchanged through `acks_or_timeout` -> `proof.clean` -> `failed_checks` -> `gate_apply_restart` verdict `deny` / `allow=False`. The earlier missing-acknowledgement remediation is preserved in full: absent, None, non-mapping, empty, partial, stale, and unparseable acks still fail closed, `ack_timeout_policy_applied` stays strict `value is True`, and the legitimate zero-live-sessions and explicit-timeout paths still pass. Reviewer's reproduction, before and after this commit: sessions_live_other = 2 report ack_state = {'other-0': 'pending', 'other-1': 'pending'} supplied acks = {'req': 'ack', 'totally-bogus-session': 'ack'} before: acks_or_timeout = True | proof.clean = True | gate allow after: acks_or_timeout = False | proof.clean = False | gate deny Tests: 22 new cases in `AcknowledgementIdentityBindingTests` covering the reviewer's exact exploit, wrong-ids-with-sufficient-count, partial identity match, requester-only acks, fabricated ids, unproven per-session states, missing/malformed/contradictory identity evidence, count mismatch, and the preserved success paths. Verification: - `pytest tests/test_drain_proof.py` -> 61 passed, 56 subtests (baseline at 95178349: 39 passed, 26 subtests) - Restart surface (6 modules) -> 154 passed, 68 subtests, exit 0 (baseline at 95178349: 132 passed, 38 subtests) - Full `pytest tests/` -> 5177 passed vs baseline 5155 passed; the 23 failures are identical in both runs and pre-exist at 95178349. Scope: drain_proof.py, tests/test_drain_proof.py. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01VRUZAf3Fr5n3kqhhiayN6C (cherry picked from commit 4193b63f415b066ee292386c2c89bc3d2651a0cc) --- drain_proof.py | 238 +++++++++++++++++++++++++---- tests/test_drain_proof.py | 311 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 522 insertions(+), 27 deletions(-) diff --git a/drain_proof.py b/drain_proof.py index 5eec7b6..e373abb 100644 --- a/drain_proof.py +++ b/drain_proof.py @@ -328,6 +328,160 @@ def _is_acknowledged(value: Any) -> bool: return value.strip().lower() in _ACK_TOKENS +def _session_ids_from_affected_sessions( + raw: Any, +) -> tuple[set[str] | None, str | None]: + """Live, non-requester session ids from the report's ``affected_sessions``. + + The requester is excluded only on authoritative ``is_requester`` evidence; + an entry whose ``live`` or ``is_requester`` flag is absent or not a real + bool is malformed, never assumed. Returns ``(ids, None)`` or + ``(None, detail)``. + """ + + if not isinstance(raw, (list, tuple)): + return None, ( + "impact report 'affected_sessions' is malformed: expected a list, " + f"got {type(raw).__name__} (fail closed)" + ) + collected: list[str] = [] + for index, entry in enumerate(raw): + if not isinstance(entry, Mapping): + return None, ( + f"impact report 'affected_sessions[{index}]' is malformed: " + f"expected a mapping, got {type(entry).__name__} (fail closed)" + ) + session_id = entry.get("session_id") + if not isinstance(session_id, str) or not session_id.strip(): + return None, ( + f"impact report 'affected_sessions[{index}]' has no usable " + "session_id (fail closed)" + ) + live = entry.get("live") + is_requester = entry.get("is_requester") + if not isinstance(live, bool) or not isinstance(is_requester, bool): + return None, ( + f"impact report 'affected_sessions[{index}]' " + f"({session_id.strip()}) does not prove live/is_requester with " + "explicit booleans (fail closed)" + ) + if live and not is_requester: + collected.append(session_id.strip()) + if len(set(collected)) != len(collected): + return None, ( + "impact report 'affected_sessions' names the same live session more " + "than once; required acknowledgement identities are ambiguous " + "(fail closed)" + ) + return set(collected), None + + +def _session_ids_from_ack_state(raw: Any) -> tuple[set[str] | None, str | None]: + """Session ids keyed by the report's ``ack_state``. + + ``ack_state`` is minted as ``{s.session_id: "pending" for s in + other_live_sessions}``, so its keys *are* the required set. Only the keys + are trusted here; the per-session value is the report's own placeholder and + is never read as an acknowledgement (acknowledgements come from the drain + state and must be explicit). + """ + + if not isinstance(raw, Mapping): + return None, ( + "impact report 'ack_state' is malformed: expected a mapping, got " + f"{type(raw).__name__} (fail closed)" + ) + keys: set[str] = set() + for key in raw: + if not isinstance(key, str) or not key.strip(): + return None, ( + "impact report 'ack_state' carries a non-string or empty " + "session id (fail closed)" + ) + keys.add(key.strip()) + if len(keys) != len(raw): + return None, ( + "impact report 'ack_state' names the same live session more than " + "once; required acknowledgement identities are ambiguous " + "(fail closed)" + ) + return keys, None + + +def _required_ack_sessions( + impact_report: Mapping[str, Any], +) -> tuple[frozenset[str] | None, str | None]: + """Identities of the live, non-requester sessions that owe an acknowledgement. + + Derived only from authoritative impact-report evidence, never from the + caller-supplied drain state — the same principle checks 1 and 5 already + apply. Returns ``(ids, None)`` when the report proves the required set, or + ``(None, detail)`` when the evidence is missing, malformed, contradictory, + or cannot be reconciled; every one of those fails the checklist closed. + + The report carries two independent views of the same set and both are + validated: ``affected_sessions`` filtered on ``live and not is_requester``, + and ``ack_state`` whose keys are exactly those sessions. When both are + present they must name the same set, and the result is reconciled against + ``counts.sessions_live_other`` — a report that disagrees with itself can + never authorise a restart. + """ + + from_sessions: set[str] | None = None + if "affected_sessions" in impact_report: + from_sessions, detail = _session_ids_from_affected_sessions( + impact_report.get("affected_sessions") + ) + if detail is not None: + return None, detail + + from_ack_state: set[str] | None = None + if "ack_state" in impact_report: + from_ack_state, detail = _session_ids_from_ack_state( + impact_report.get("ack_state") + ) + if detail is not None: + return None, detail + + if from_sessions is None and from_ack_state is None: + return None, ( + "impact report carries no session-identity evidence (neither " + "ack_state nor affected_sessions); required acknowledgements " + "cannot be attributed to a session (fail closed)" + ) + + if ( + from_sessions is not None + and from_ack_state is not None + and from_sessions != from_ack_state + ): + only_ack_state = sorted(from_ack_state - from_sessions) + only_sessions = sorted(from_sessions - from_ack_state) + return None, ( + "impact report contradicts itself about which sessions must " + f"acknowledge: ack_state-only={only_ack_state}, " + f"affected_sessions-only={only_sessions} (fail closed)" + ) + + required = from_ack_state if from_ack_state is not None else from_sessions + if required is None: # unreachable; guarded above, kept fail-closed + return None, "required acknowledgement identities unresolved (fail closed)" + + live_count = _live_session_count(impact_report) + if live_count is None: + return None, ( + "impact report does not prove the live-session count; required " + "acknowledgement identities cannot be reconciled (fail closed)" + ) + if len(required) != live_count: + return None, ( + f"impact report names {len(required)} live session(s) requiring " + f"acknowledgement but counts.sessions_live_other={live_count}; " + "identity evidence cannot be reconciled (fail closed)" + ) + return frozenset(required), None + + def _evaluate_checks( impact_report: Mapping[str, Any], drain_state: Mapping[str, Any], @@ -444,27 +598,55 @@ def _evaluate_checks( # 6. Acknowledgements received, or an explicit timeout policy was applied. # - # Whether acknowledgements are *required* is answered by the impact report, - # never inferred from the shape of the drain state. Previously an absent - # ``acks`` key collapsed to ``{}`` and was read as "no session needed to - # acknowledge", so a proof minted clean — and the restart gate allowed — - # while the report still showed other live sessions. Absence of evidence is - # not evidence of absence: missing, malformed, stale, or otherwise unproven - # acknowledgement data fails closed whenever live sessions require - # acknowledgement, and only explicitly verified acknowledgement evidence - # may permit the operation. + # Whether acknowledgements are *required*, and *which sessions owe them*, + # are both answered by the impact report — never inferred from the shape or + # the cardinality of the drain state. Two fail-opens of the same class were + # closed here in turn. First, an absent ``acks`` key collapsed to ``{}`` and + # was read as "no session needed to acknowledge", so a proof minted clean + # while the report still showed other live sessions. Then coverage was + # decided by comparing an acknowledgement *count* against + # ``sessions_live_other``, so acknowledgements from the requesting session + # or from ids that do not exist satisfied the obligations of the live + # sessions that never answered. + # + # Coverage is therefore bound to identity: every live, non-requester session + # the report names must itself carry an explicit acknowledgement. An + # acknowledgement from any other id — the requester, an unknown id, a + # fabricated one — is not evidence about a required session and never + # increases coverage. sessions_live_other = _live_session_count(impact_report) - live_count_known = sessions_live_other is not None - # An unknown or malformed count fails closed: it cannot prove nobody had to - # acknowledge, so acknowledgement stays required. - acks_required = (not live_count_known) or sessions_live_other > 0 + required_sessions, identity_detail = _required_ack_sessions(impact_report) + # An unknown count or unresolvable identity evidence fails closed: neither + # can prove nobody had to acknowledge, so acknowledgement stays required. + acks_required = required_sessions is None or bool(required_sessions) raw_acks = drain_state.get("acks") acks_present = isinstance(raw_acks, Mapping) ack_values = list(raw_acks.values()) if acks_present else [] acked_count = sum(1 for value in ack_values if _is_acknowledged(value)) every_entry_acked = bool(ack_values) and acked_count == len(ack_values) - covers_live_sessions = live_count_known and acked_count >= sessions_live_other + # Identity-bound coverage. Only an entry keyed by a required session id and + # carrying an explicit token counts, so neither a requester acknowledgement + # nor an unknown id can stand in for a session that stayed silent. + if acks_present and required_sessions is not None: + acked_sessions = { + session_id + for session_id in required_sessions + if any( + isinstance(key, str) + and key.strip() == session_id + and _is_acknowledged(value) + for key, value in raw_acks.items() + ) + } + else: + acked_sessions = set() + missing_sessions = ( + sorted(required_sessions - acked_sessions) + if required_sessions is not None + else [] + ) + covers_live_sessions = required_sessions is not None and not missing_sessions # Strict by construction: only an explicit ``True`` counts, so an absent, # null, or non-boolean ``ack_timeout_policy_applied`` can never open this # gate on its own. @@ -476,7 +658,13 @@ def _evaluate_checks( # is that an acknowledgement is still owed. outstanding_entries = acks_present and bool(ack_values) and not every_entry_acked - if timeout_policy and (acks_required or outstanding_entries): + if identity_detail is not None: + # Unresolvable identity evidence outranks every permitting path, + # including the timeout policy: when the report cannot say who owed an + # acknowledgement, nothing can show the obligation was discharged. + acks_ok = False + detail = identity_detail + elif timeout_policy and (acks_required or outstanding_entries): acks_ok = True detail = "explicit ack timeout policy applied and recorded" elif outstanding_entries: @@ -492,21 +680,16 @@ def _evaluate_checks( "impact report proves no other live sessions required to " "acknowledge (sessions_live_other=0)" ) - elif every_entry_acked and covers_live_sessions: + elif covers_live_sessions: acks_ok = True detail = ( - f"{acked_count} acknowledgement entr" - f"{'y' if acked_count == 1 else 'ies'} verified; covers " + f"{len(acked_sessions)} required live session(s) each acknowledged " + f"by identity: {sorted(acked_sessions)}; covers " f"sessions_live_other={sessions_live_other}" ) else: acks_ok = False - if not live_count_known: - detail = ( - "impact report does not prove the live-session count; " - "acknowledgement required and unproven (fail closed)" - ) - elif raw_acks is None: + if raw_acks is None: detail = ( "acknowledgement evidence absent while " f"sessions_live_other={sessions_live_other} require " @@ -525,9 +708,10 @@ def _evaluate_checks( ) else: detail = ( - f"acknowledgements cover only {acked_count} session(s) but the " - f"report shows sessions_live_other={sessions_live_other} " - "(fail closed)" + "no valid acknowledgement attributed to required live " + f"session(s) {missing_sessions}; the report shows " + f"sessions_live_other={sessions_live_other} and supplied " + "acknowledgements for other ids do not count (fail closed)" ) checks.append(DrainCheck(CHECK_ACKS_OR_TIMEOUT, acks_ok, detail)) diff --git a/tests/test_drain_proof.py b/tests/test_drain_proof.py index 410af33..5896ecc 100644 --- a/tests/test_drain_proof.py +++ b/tests/test_drain_proof.py @@ -592,5 +592,316 @@ class AcknowledgementFailClosedTests(unittest.TestCase): self.assertFalse(result.clean) +def _identity_report(*, requester: str, others: tuple[str, ...]) -> dict: + """Report with explicitly named requester and other live sessions. + + Unlike :func:`_drained_report_with_live_sessions`, the session ids are + chosen by the caller so a test can supply acknowledgements for the *wrong* + identities while keeping the count correct. + """ + + sessions = [ + { + "session_id": requester, + "role": "controller", + "profile": "prgs-controller", + "pid": _live_pid(), + "status": "active", + "last_heartbeat_at": NOW.isoformat(), + } + ] + for session_id in others: + sessions.append( + { + "session_id": session_id, + "role": "author", + "profile": "prgs-author", + "pid": _live_pid(), + "status": "active", + "last_heartbeat_at": NOW.isoformat(), + } + ) + report = rc.evaluate_restart_impact( + {"sessions": sessions, "leases": [], "inventory_complete": True}, + now=NOW, + requesting_session_id=requester, + ) + return report.as_dict() + + +class AcknowledgementIdentityBindingTests(unittest.TestCase): + """Acknowledgement coverage must be bound to session identity, not counted. + + Regression cover for the second reviewed fail-open on PR #882 (review 582, + blocker B1): coverage compared ``acked_count`` against + ``counts.sessions_live_other``, so acknowledgements supplied for the + requesting session and for ids that do not exist satisfied the obligations + of the live sessions that never answered. The required identities are + carried by the report itself — ``ack_state`` keys and ``affected_sessions`` + filtered on ``live and not is_requester`` — and only an acknowledgement + keyed by one of those ids may count for it. + """ + + def _state(self, **overrides) -> dict: + state = _clean_drain_state() + state.pop("acks", None) + state["ack_timeout_policy_applied"] = False + state.update(overrides) + return state + + def _acks_check(self, proof) -> dp.DrainCheck: + return next(c for c in proof.checks if c.name == dp.CHECK_ACKS_OR_TIMEOUT) + + def _build(self, report: dict, state: dict): + return dp.build_drain_proof( + impact_report=report, drain_state=state, now=NOW, secret=SECRET + ) + + def assertAcksFailClosed(self, report: dict, state: dict) -> dp.DrainCheck: + """Failure must propagate through the check, the proof, and the gate.""" + + proof = self._build(report, state) + check = self._acks_check(proof) + self.assertFalse(check.passed) + self.assertFalse(proof.clean) + self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks) + decision = dp.gate_apply_restart( + proof=proof.as_dict(), + now=NOW, + secret=SECRET, + expected_impact_fingerprint=dp.impact_fingerprint(report), + ) + self.assertEqual(decision.verdict, dp.GATE_DENY) + self.assertFalse(decision.allow) + return check + + # --- the reviewer's exact reproduction -------------------------------- + + def test_requester_plus_unknown_id_cannot_satisfy_two_live_sessions(self): + """Review 582 B1 verbatim: requester + a nonexistent session. + + ``sessions_live_other=2`` with ``ack_state`` naming ``other-0`` and + ``other-1``; the drain state supplies an acknowledgement from the + requesting session itself and from a session that does not exist. The + count matches, the identities do not. + """ + + report = _identity_report(requester="req", others=("other-0", "other-1")) + self.assertEqual(report["counts"]["sessions_live_other"], 2) + self.assertEqual( + report["ack_state"], {"other-0": "pending", "other-1": "pending"} + ) + state = self._state(acks={"req": "ack", "totally-bogus-session": "ack"}) + check = self.assertAcksFailClosed(report, state) + self.assertIn("other-0", check.detail) + self.assertIn("other-1", check.detail) + self.assertIn("fail closed", check.detail) + + # --- wrong / unknown / requester identities --------------------------- + + def test_sufficient_count_of_wrong_ids_fails_closed(self): + """Right cardinality, wrong identities: two acks, neither required.""" + + report = _identity_report(requester="req", others=("other-0", "other-1")) + state = self._state(acks={"ghost-a": "ack", "ghost-b": "ack"}) + check = self.assertAcksFailClosed(report, state) + self.assertIn("do not count", check.detail) + + def test_more_acks_than_required_still_fails_on_wrong_ids(self): + """Coverage cannot be bought with volume: five acks, none required.""" + + report = _identity_report(requester="req", others=("other-0", "other-1")) + state = self._state(acks={f"ghost-{i}": "acknowledged" for i in range(5)}) + self.assertAcksFailClosed(report, state) + + def test_partial_identity_match_fails_closed(self): + """One required id acknowledged, the rest padded with unknown ids.""" + + report = _identity_report( + requester="req", others=("other-0", "other-1", "other-2") + ) + state = self._state( + acks={"other-0": "ack", "ghost-1": "ack", "ghost-2": "ack"} + ) + check = self.assertAcksFailClosed(report, state) + self.assertIn("other-1", check.detail) + self.assertIn("other-2", check.detail) + + def test_requester_ack_never_satisfies_another_sessions_obligation(self): + """The requester is excluded from the required set and stays excluded.""" + + report = _identity_report(requester="req", others=("other-0",)) + requester_rows = [s for s in report["affected_sessions"] if s["is_requester"]] + self.assertEqual([s["session_id"] for s in requester_rows], ["req"]) + self.assertNotIn("req", report["ack_state"]) + check = self.assertAcksFailClosed(report, self._state(acks={"req": "ack"})) + self.assertIn("other-0", check.detail) + + def test_fabricated_ids_do_not_count_toward_coverage(self): + report = _identity_report(requester="req", others=("other-0",)) + for bogus in ("", " ", "other-0 extra", "OTHER-0", "other-01", "0"): + with self.subTest(bogus=bogus): + self.assertAcksFailClosed(report, self._state(acks={bogus: "ack"})) + + # --- per-session state must be explicitly valid ------------------------ + + def test_unproven_per_session_states_fail_closed(self): + """A required id present but not explicitly acknowledged fails closed.""" + + report = _identity_report(requester="req", others=("other-0", "other-1")) + for value in ("pending", "stale", "unknown", "", None, True, 1, NOW): + with self.subTest(value=value): + self.assertAcksFailClosed( + report, + self._state(acks={"other-0": "ack", "other-1": value}), + ) + + def test_report_ack_state_placeholder_is_never_read_as_an_ack(self): + """``ack_state`` values are the report's own placeholders, not evidence.""" + + report = _identity_report(requester="req", others=("other-0",)) + report["ack_state"] = {"other-0": "ack"} + self.assertAcksFailClosed(report, self._state()) + + # --- missing / malformed / contradictory identity evidence ------------- + + def test_missing_identity_evidence_fails_closed(self): + report = _identity_report(requester="req", others=("other-0",)) + report.pop("ack_state", None) + report.pop("affected_sessions", None) + check = self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"})) + self.assertIn("no session-identity evidence", check.detail) + + def test_malformed_ack_state_fails_closed(self): + for malformed in ([], "other-0", 7, None, ("other-0",)): + with self.subTest(malformed=malformed): + report = _identity_report(requester="req", others=("other-0",)) + report["ack_state"] = malformed + self.assertAcksFailClosed( + report, self._state(acks={"other-0": "ack"}) + ) + + def test_non_string_ack_state_key_fails_closed(self): + report = _identity_report(requester="req", others=("other-0",)) + report["ack_state"] = {7: "pending"} + self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"})) + + def test_malformed_affected_sessions_fails_closed(self): + for malformed in ("sessions", 7, {"session_id": "other-0"}, [None], [7]): + with self.subTest(malformed=malformed): + report = _identity_report(requester="req", others=("other-0",)) + report.pop("ack_state", None) + report["affected_sessions"] = malformed + self.assertAcksFailClosed( + report, self._state(acks={"other-0": "ack"}) + ) + + def test_affected_sessions_without_explicit_booleans_fails_closed(self): + """``live``/``is_requester`` must be real booleans, never inferred.""" + + report = _identity_report(requester="req", others=("other-0",)) + report.pop("ack_state", None) + for row in report["affected_sessions"]: + if row["session_id"] == "other-0": + row["is_requester"] = "false" + self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"})) + + def test_affected_sessions_missing_live_flag_fails_closed(self): + report = _identity_report(requester="req", others=("other-0",)) + report.pop("ack_state", None) + for row in report["affected_sessions"]: + row.pop("live", None) + self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"})) + + def test_contradictory_ack_state_and_affected_sessions_fails_closed(self): + """Both views present and disagreeing is unresolvable, not a tie-break.""" + + report = _identity_report(requester="req", others=("other-0", "other-1")) + report["ack_state"] = {"other-0": "pending", "other-9": "pending"} + check = self.assertAcksFailClosed( + report, self._state(acks={"other-0": "ack", "other-9": "ack"}) + ) + self.assertIn("contradicts itself", check.detail) + + def test_identity_count_mismatch_fails_closed(self): + """Identity evidence that cannot be reconciled with the count denies.""" + + report = _identity_report(requester="req", others=("other-0", "other-1")) + report["counts"] = dict(report["counts"], sessions_live_other=1) + check = self.assertAcksFailClosed( + report, self._state(acks={"other-0": "ack", "other-1": "ack"}) + ) + self.assertIn("cannot be reconciled", check.detail) + + def test_broken_identity_evidence_outranks_timeout_policy(self): + """The sanctioned timeout path cannot paper over an unreadable report.""" + + report = _identity_report(requester="req", others=("other-0",)) + report["ack_state"] = "not-a-mapping" + self.assertAcksFailClosed(report, self._state(ack_timeout_policy_applied=True)) + + # --- legitimate success is preserved ----------------------------------- + + def test_every_required_session_acknowledged_passes(self): + report = _identity_report( + requester="req", others=("other-0", "other-1", "other-2") + ) + state = self._state( + acks={ + "other-0": "ack", + "other-1": "acked", + "other-2": "acknowledged", + } + ) + proof = self._build(report, state) + check = self._acks_check(proof) + self.assertTrue(check.passed) + self.assertTrue(proof.clean) + self.assertEqual(proof.failed_checks, []) + self.assertIn("acknowledged by identity", check.detail) + decision = dp.gate_apply_restart( + proof=proof.as_dict(), + now=NOW, + secret=SECRET, + expected_impact_fingerprint=dp.impact_fingerprint(report), + ) + self.assertEqual(decision.verdict, dp.GATE_ALLOW) + self.assertTrue(decision.allow) + + def test_required_session_ack_tolerates_surrounding_whitespace(self): + report = _identity_report(requester="req", others=("other-0",)) + proof = self._build(report, self._state(acks={" other-0 ": " ACK "})) + self.assertTrue(self._acks_check(proof).passed) + self.assertTrue(proof.clean) + + def test_no_other_live_sessions_still_passes_with_identity_evidence(self): + report = _identity_report(requester="req", others=()) + self.assertEqual(report["counts"]["sessions_live_other"], 0) + self.assertEqual(report["ack_state"], {}) + proof = self._build(report, self._state()) + check = self._acks_check(proof) + self.assertTrue(check.passed) + self.assertIn("sessions_live_other=0", check.detail) + self.assertTrue(proof.clean) + + def test_explicit_timeout_policy_retains_intended_behavior(self): + """Valid, correctly typed timeout evidence still permits the check.""" + + report = _identity_report(requester="req", others=("other-0", "other-1")) + proof = self._build(report, self._state(ack_timeout_policy_applied=True)) + check = self._acks_check(proof) + self.assertTrue(check.passed) + self.assertIn("timeout policy", check.detail) + self.assertTrue(proof.clean) + + def test_timeout_policy_still_strictly_typed_under_identity_binding(self): + report = _identity_report(requester="req", others=("other-0",)) + for value in (None, "true", "True", 1, [], {}): + with self.subTest(value=value): + self.assertAcksFailClosed( + report, self._state(ack_timeout_policy_applied=value) + ) + + if __name__ == "__main__": unittest.main()