diff --git a/docs/spec/disclosure-authorization.md b/docs/spec/disclosure-authorization.md index de5d837c..922a8178 100644 --- a/docs/spec/disclosure-authorization.md +++ b/docs/spec/disclosure-authorization.md @@ -53,10 +53,26 @@ serialization, and use the same protected replay database. Reconstructing a gate does not stop already admitted delivery on another live gate. After validation, a durable SQLite transaction consumes the request ID **before** -the delivery callback. A crash, callback exception or lost acknowledgement leaves -delivery unknown; the ID stays consumed. Never automatically retry or assign a -fresh ID to an unknown attempt. A normal callback return is an acknowledgement, -not proof of recipient installation, processing or downstream confidentiality. +the delivery callback. After the post-reservation validity recheck, the gate must +durably save a minimized disclosure-attempt record with delivery `unknown` +**before** invoking the irreversible recipient callback. If that audit write fails, +delivery is not attempted. Because that durable write can itself block, scoped +approval validity is checked again after the write and immediately before the +callback. If that final check denies dispatch, the same audit event is +best-effort corrected to `not_attempted` and the denial returns its `event_id`. +If the corrective audit write fails, the durable event remains conservatively +`unknown` and the returned denial carries that same `event_id`. In either case +the consumed request cannot be retried. A crash, `BaseException`, callback +exception or lost post-delivery acknowledgement leaves the durable outcome +`unknown`; the consumed request ID prevents automatic retry. On normal callback return the gate +best-effort upgrades that same event to `acknowledged`. A normal callback return is +an acknowledgement, not proof of recipient installation, processing or downstream +confidentiality. + +The durable audit table contains only event ID, disposition, reason and delivery. +It must not contain payload, payload digest, request ID, principal, recipient, +purpose, source scope, labels or approval. Private authorization/replay context +remains separate and access-controlled. All gates for a release authority must use the same trusted replay store. Deletion, snapshot rollback, database substitution or separate clones defeat diff --git a/src/cmcp_runtime/disclosure.py b/src/cmcp_runtime/disclosure.py index 5843c345..cc0d208a 100644 --- a/src/cmcp_runtime/disclosure.py +++ b/src/cmcp_runtime/disclosure.py @@ -136,6 +136,11 @@ def __init__(self, path: str | Path): try: connection.execute("CREATE TABLE IF NOT EXISTS disclosure_attempts " "(request_id TEXT PRIMARY KEY)") + connection.execute( + "CREATE TABLE IF NOT EXISTS disclosure_audit " + "(event_id TEXT PRIMARY KEY, disposition TEXT NOT NULL, " + "reason TEXT NOT NULL, delivery TEXT NOT NULL)" + ) connection.commit() finally: connection.close() @@ -153,6 +158,76 @@ def consume(self, request_id: str) -> bool: connection.close() return True + def record_attempt(self, observation: ReleaseObservation) -> None: + """Persist only minimized audit-facing evidence before boundary crossing.""" + connection = sqlite3.connect(self._path) + try: + connection.execute("BEGIN IMMEDIATE") + connection.execute( + "INSERT INTO disclosure_audit VALUES (?, ?, ?, ?)", + (observation.event_id, observation.disposition, + observation.reason, observation.delivery), + ) + connection.commit() + finally: + connection.close() + + def _update_audit_delivery(self, event_id: str, reason: str, delivery: Delivery) -> None: + """Durably update one existing minimized audit event.""" + connection = sqlite3.connect(self._path) + try: + connection.execute("BEGIN IMMEDIATE") + cursor = connection.execute( + "UPDATE disclosure_audit " + "SET reason = ?, delivery = ? WHERE event_id = ?", + (reason, delivery, event_id), + ) + if cursor.rowcount != 1: + raise sqlite3.IntegrityError("missing disclosure audit event") + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def acknowledge(self, event_id: str) -> None: + """Durably upgrade one pre-delivery unknown event after adapter return.""" + self._update_audit_delivery(event_id, "adapter_acknowledged", "acknowledged") + + def mark_not_attempted( + self, event_id: str, disposition: Disposition, reason: str, + ) -> None: + """Correct a prepared audit event when the final admission recheck denies dispatch.""" + connection = sqlite3.connect(self._path) + try: + connection.execute("BEGIN IMMEDIATE") + cursor = connection.execute( + "UPDATE disclosure_audit " + "SET disposition = ?, reason = ?, delivery = ? WHERE event_id = ?", + (disposition, reason, "not_attempted", event_id), + ) + if cursor.rowcount != 1: + raise sqlite3.IntegrityError("missing disclosure audit event") + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def audit_observations(self) -> tuple[ReleaseObservation, ...]: + """Return minimized durable observations; no private release context is stored.""" + connection = sqlite3.connect(self._path) + try: + rows = connection.execute( + "SELECT disposition, reason, delivery, event_id " + "FROM disclosure_audit ORDER BY rowid" + ).fetchall() + finally: + connection.close() + return tuple(ReleaseObservation(*row) for row in rows) + class DisclosureGate: """Validate and consume before invoking one operator-registered byte sink. @@ -223,12 +298,43 @@ def release( validity = self._validity(approval) if validity is not None: return validity + # Persist minimized evidence of a boundary-crossing attempt before the + # irreversible callback. No protected release context is stored here. + attempt = ReleaseObservation(disposition, "delivery_attempted", "unknown") + try: + self._store.record_attempt(attempt) + except sqlite3.Error: + return ReleaseObservation("unavailable", "audit_storage") + # The durable audit write may block long enough for a scoped approval + # to expire. Recheck immediately before the irreversible callback. + if not within and approval is not None: + validity = self._validity(approval) + if validity is not None: + try: + self._store.mark_not_attempted( + attempt.event_id, validity.disposition, validity.reason) + except sqlite3.Error: + # The corrective audit write failed. Preserve the conservative + # durable unknown state and tie the denial to that event. + return ReleaseObservation( + validity.disposition, validity.reason, "unknown", attempt.event_id) + return ReleaseObservation( + validity.disposition, validity.reason, "not_attempted", attempt.event_id) try: recipient.deliver(request.payload) except Exception: # Do not copy upstream exception text or a traceback into evidence. - return ReleaseObservation(disposition, "delivery_unknown", "unknown") - return ReleaseObservation(disposition, "adapter_acknowledged", "acknowledged") + # The durable pre-delivery record remains conservatively unknown. + return ReleaseObservation( + disposition, "delivery_unknown", "unknown", attempt.event_id) + try: + self._store.acknowledge(attempt.event_id) + except sqlite3.Error: + # Delivery happened, but durable acknowledgement did not. + return ReleaseObservation( + disposition, "delivery_unknown", "unknown", attempt.event_id) + return ReleaseObservation( + disposition, "adapter_acknowledged", "acknowledged", attempt.event_id) def _validity(self, approval: DisclosureApproval) -> ReleaseObservation | None: try: diff --git a/tests/unit/test_disclosure.py b/tests/unit/test_disclosure.py index 7c3b70c8..55acdb13 100644 --- a/tests/unit/test_disclosure.py +++ b/tests/unit/test_disclosure.py @@ -165,12 +165,122 @@ def test_replay_survives_reopen_and_failed_delivery(context, tmp_path): result = DisclosureGate(**options).release(request, approve()) assert result.delivery == "unknown" assert PAYLOAD.decode() not in json.dumps(asdict(result)) - options["replay_store"] = ReplayStore(tmp_path / "attempts.db") + reopened = ReplayStore(tmp_path / "attempts.db") + assert reopened.audit_observations() == ( + replace(result, reason="delivery_attempted"), + ) + options["replay_store"] = reopened sink.side_effect = None assert DisclosureGate(**options).release(request, approve()).reason == "replay" sink.assert_called_once_with(PAYLOAD) +def test_attempt_is_durable_before_delivery_and_ack_survives_restart(context, tmp_path): + request, sink, options, approve = context + result = DisclosureGate(**options).release(request, approve()) + assert result.delivery == "acknowledged" + reopened = ReplayStore(tmp_path / "attempts.db") + assert reopened.audit_observations() == (result,) + sink.assert_called_once_with(PAYLOAD) + + +def test_baseexception_during_delivery_leaves_durable_unknown_attempt(context, tmp_path): + request, sink, options, approve = context + sink.side_effect = KeyboardInterrupt() + with pytest.raises(KeyboardInterrupt): + DisclosureGate(**options).release(request, approve()) + observations = ReplayStore(tmp_path / "attempts.db").audit_observations() + assert len(observations) == 1 + assert observations[0].delivery == "unknown" + assert observations[0].reason == "delivery_attempted" + + +def test_expiry_rechecked_after_pre_delivery_audit_write(context, tmp_path): + request, sink, options, approve = context + # Initial validation and post-reservation validation succeed. The approval + # expires while the durable audit write is completing, before delivery. + clock = iter([100, 100, 110]) + options["now"] = lambda: next(clock) + gate = DisclosureGate(**options) + result = gate.release(request, approve()) + assert (result.disposition, result.reason, result.delivery) == ( + "denied", "approval_expired_or_early", "not_attempted") + sink.assert_not_called() + # The prepared row is corrected to the actual no-dispatch outcome and the + # returned denial is tied to the same durable event. + observations = ReplayStore(tmp_path / "attempts.db").audit_observations() + assert observations == (result,) + # The one-use request remains consumed after the failed pre-delivery recheck. + options["now"] = lambda: 100 + assert DisclosureGate(**options).release(request, approve()).reason == "replay" + + +def test_expiry_recheck_correction_failure_preserves_unknown(context, tmp_path, monkeypatch): + request, sink, options, approve = context + clock = iter([100, 100, 110]) + options["now"] = lambda: next(clock) + store = options["replay_store"] + monkeypatch.setattr( + store, "mark_not_attempted", + Mock(side_effect=__import__("sqlite3").OperationalError("disk unavailable")), + ) + result = DisclosureGate(**options).release(request, approve()) + assert (result.disposition, result.reason, result.delivery) == ( + "denied", "approval_expired_or_early", "unknown") + sink.assert_not_called() + observations = ReplayStore(tmp_path / "attempts.db").audit_observations() + assert len(observations) == 1 + assert observations[0].event_id == result.event_id + assert observations[0].reason == "delivery_attempted" + assert observations[0].delivery == "unknown" + + +def test_pre_delivery_audit_failure_fails_closed(context, monkeypatch): + request, sink, options, approve = context + monkeypatch.setattr( + options["replay_store"], "record_attempt", + Mock(side_effect=__import__("sqlite3").OperationalError("disk unavailable")), + ) + result = DisclosureGate(**options).release(request, approve()) + assert (result.disposition, result.reason, result.delivery) == ( + "unavailable", "audit_storage", "not_attempted") + sink.assert_not_called() + + +def test_post_delivery_audit_failure_retains_unknown(context, tmp_path, monkeypatch): + request, sink, options, approve = context + store = options["replay_store"] + monkeypatch.setattr( + store, "acknowledge", + Mock(side_effect=__import__("sqlite3").OperationalError("disk unavailable")), + ) + result = DisclosureGate(**options).release(request, approve()) + assert result.delivery == "unknown" + assert result.reason == "delivery_unknown" + sink.assert_called_once_with(PAYLOAD) + observations = ReplayStore(tmp_path / "attempts.db").audit_observations() + assert observations == (replace(result, reason="delivery_attempted"),) + + +def test_durable_audit_row_contains_only_minimized_fields(context, tmp_path): + request, _, options, approve = context + result = DisclosureGate(**options).release(request, approve()) + import sqlite3 + connection = sqlite3.connect(tmp_path / "attempts.db") + try: + columns = [row[1] for row in connection.execute("PRAGMA table_info(disclosure_audit)")] + row = connection.execute("SELECT * FROM disclosure_audit").fetchone() + finally: + connection.close() + assert columns == ["event_id", "disposition", "reason", "delivery"] + assert row == (result.event_id, result.disposition, result.reason, result.delivery) + encoded = json.dumps(row) + for private in (PAYLOAD.decode(), hashlib.sha256(PAYLOAD).hexdigest(), request.request_id, + request.recipient, request.workload, request.source_scope, + request.purpose, "owner"): + assert private not in encoded + + def test_shared_store_concurrent_attempts_deliver_once(context, tmp_path): request, sink, options, approve = context approval = approve()