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()