fix(eventing): make audit mode observable

eventing/eventbridge/kafka_in.py

@@ -73,13 +73,25 @@ def __init__(
         self._require_sig = require_signature
         self._bridge_kid = bridge_kid
         self._rejected = 0
+        self._audit_failed = 0
         self._stopping = threading.Event()
 
     @property
     def rejected(self) -> int:
         """Responses stored as phase=error because they did not verify."""
         return self._rejected
 
+    @property
+    def audit_failed(self) -> int:
+        """Responses accepted unchanged that enforcement *would* have rejected.
+
+        This is the reject rate the two-flag rollout asks an operator to watch: with a
+        keyset configured and `EB_REQUIRE_RESPONSE_SIGNATURE=false`, it counts what
+        turning enforcement on would start refusing. Stays 0 once enforcement is on,
+        because then those events land in `rejected` instead.
+        """
+        return self._audit_failed
+
     def stop(self) -> None:
         self._stopping.set()
 
@@ -106,19 +118,30 @@ def run(self) -> None:
                     # here ends the for, ends the while, and the thread is gone — while
                     # the process stays up and the pod still reports healthy. The
                     # verification path must degrade, never raise.
-                    ok, why = True, "not checked"
+                    ok, why, verified = True, "not checked", True
                     if self._keyset is not None:
                         try:
-                            ok, why = signing.response_decision(
+                            ok, why, verified = signing.response_decision(
                                 evt, self._keyset, require=self._require_sig,
                                 bridge_kid=self._bridge_kid)
                         except Exception as e:  # noqa: BLE001
                             # Fail closed only where enforcement is on: if the verifier
                             # itself is broken, an unverifiable event is not evidence of
                             # anything, and silently accepting it defeats the control.
                             ok, why = (not self._require_sig), f"verifier raised: {e!r}"
+                            verified = False
                             print(f"[kafka_in] verification error: {e!r}")
                     d = ce.envelope_dict(evt)
+                    if ok and not verified:
+                        # Audit mode: the event is stored unchanged, and this line plus
+                        # the counter are the only trace that enforcement would have
+                        # refused it. §4.4's two-flag rollout ("a keyset alone verifies
+                        # and logs") is unperformable without them -- the verdict was
+                        # computed and the reason discarded, so there was no reject rate
+                        # to watch before turning enforcement on.
+                        self._audit_failed += 1
+                        print(f"[kafka_in] audit: would reject response on "
+                              f"{d.get('correlationid') or d.get('groupid')}: {why}")
                     if not ok:
                         self._rejected += 1
                         print(f"[kafka_in] unverified response on "

eventing/shared/signing.py

@@ -441,22 +441,36 @@ def _is_terminal(event) -> bool:
 
 
 def response_decision(event, ks, *, require: bool,
-                      bridge_kid: str | None = None) -> tuple[bool, str]:
-    """(accept_as_is, reason) for one event off the responses topic.
+                      bridge_kid: str | None = None) -> tuple[bool, str, bool]:
+    """(accept_as_is, reason, verified) for one event off the responses topic.
 
     Pure: no I/O, no logging, and **it does not touch `event`** — the caller owns the
     rewrite, which is what makes this testable without a Kafka consumer.
 
     Collapsing three questions into one answer is deliberate; it leaves the caller no
     policy to get wrong:
 
-    * `ks is None` — verification is not configured. Accept, as today.
-    * verified — accept.
+    * `ks is None` — verification is not configured. Accept, as today; `verified` is
+      False, because nothing was checked.
+    * verified — accept, `verified` True.
     * not verified and `require` false — **audit mode**: accept, but hand back the
       reason so the caller can log it. Enforcement rewrites persisted rows and pages
       a phone, so there has to be a way to watch the reject rate first.
     * not verified and `require` true — reject; the caller stores it as `phase=error`.
 
+    **`verified` is the third element rather than something the caller infers from
+    `reason`.** Audit mode and a clean verification both return `accept=True`, so
+    telling them apart was only possible by matching the reason prose — which exists to
+    be read by a human in a log line, not parsed. A caller that wants to count what
+    *would* have been rejected reads this flag.
+
+    So `accept and not verified` is exactly the audit-mode population: the events
+    enforcement would have refused. The unsigned non-terminal passthrough reports
+    `verified=True` because it is accepted by policy rather than in spite of a failure,
+    and so does `ks is None` — with no keyset configured there is nothing to enforce and
+    nothing that could fail it, so that population is empty and no caller-side "is a
+    keyset configured" guard is needed to keep it empty.
+
     `bridge_kid` pins group lifecycle events to EventBridge's own key. It is only
     applied when set, so a single-key deployment — where `KeySet.select(None)` returns
     the sole key and nothing needs to name a kid — keeps working untouched.
@@ -473,14 +487,22 @@ def response_decision(event, ks, *, require: bool,
     masquerade as the result.
     """
     if ks is None:
-        return True, "verification not enabled"
+        # `verified=True`: verification not being configured is an acceptance by
+        # policy (the same reading as the unsigned non-terminal passthrough), not a
+        # failed check — so `accept and not verified` stays the audit-mode
+        # population with no caller-side keyset guard to forget.
+        return True, "verification not enabled", True
     expect = bridge_kid if (bridge_kid and ce.is_group_event(event)) else None
     ok, why = verify_with_keyset(event, ks, expect_kid=expect)
     if ok:
-        return True, why
+        return True, why, True
     if not event.get("signature") and not _is_terminal(event):
-        return True, "unsigned non-terminal frame (signing covers terminal events)"
-    return (not require), why
+        # `verified=True` because this frame is accepted by the signing policy, not in
+        # spite of it: nothing was expected to be signed, so nothing failed. A caller
+        # counting audit-mode failures must not count the streamed frames of every
+        # genuine run, which is what the policy below makes unsigned on purpose.
+        return True, "unsigned non-terminal frame (signing covers terminal events)", True
+    return (not require), why, False
 
 
 # ---- key loading ------------------------------------------------------------

eventing/tests/test_kafka_in.py

@@ -343,6 +343,72 @@ def boom(*a, **k):
     assert len(seen) == 1
     assert c.rejected == 0
     assert store.events_for("brave-otter-4718")[0]["phase"] == "result"
+    assert c.audit_failed == 1, \
+        "but it is still an unverified acceptance, and audit mode exists to count those"
+
+
+# ---- audit mode: the reject rate the rollout asks you to watch ---------------
+
+def test_audit_mode_counts_and_logs_what_enforcement_would_reject(
+        tmp_path, monkeypatch, capsys):
+    """§4.4 promises "a keyset alone verifies and logs while storing events unchanged",
+    and that is the step where an operator watches the reject rate before enforcing.
+
+    The verdict was computed and the reason discarded, so nothing surfaced it: no log
+    line, no counter, and therefore no reject rate to watch. The rollout the design
+    documents could not be performed.
+    """
+    c, store, seen, _ = _drain(tmp_path, [_response()], monkeypatch,
+                               keyset=_keyset(tmp_path), require_signature=False)
+    assert c.audit_failed == 1, "the unsigned terminal response must be counted"
+    assert c.rejected == 0, "but audit mode rejects nothing"
+
+    # Stored unchanged is the other half of the promise: audit mode observes without
+    # rewriting the row a user reads.
+    rows = store.events_for("brave-otter-4718")
+    assert rows[0]["phase"] == "result"
+    assert "signature_rejected" not in (rows[0].get("data") or {})
+    assert len(seen) == 1
+
+    out = capsys.readouterr().out
+    assert "audit: would reject response on brave-otter-4718" in out
+    assert "no ce_signature" in out, "the reason is what makes the line actionable"
+
+
+def test_audit_mode_stays_quiet_for_responses_that_verify(tmp_path, monkeypatch):
+    """A counter that also counts successes is not a reject rate."""
+    c, _, _, _ = _drain(tmp_path, [_response(seed=SEED_R1, kid="runner-01")],
+                        monkeypatch, keyset=_keyset(tmp_path),
+                        require_signature=False)
+    assert c.audit_failed == 0 and c.rejected == 0
+
+
+def test_audit_mode_does_not_count_unsigned_non_terminal_frames(tmp_path, monkeypatch):
+    """`emit()` signs terminal events only, so every streamed frame of every genuine
+    run is unsigned on purpose. Counting those would bury the signal this exists for."""
+    c, _, _, _ = _drain(tmp_path,
+                        [_response(seq=1, phase="stdout", final="false", text="thinking")],
+                        monkeypatch, keyset=_keyset(tmp_path),
+                        require_signature=False)
+    assert c.audit_failed == 0, "accepted by the signing policy, not a failed check"
+
+
+def test_the_audit_counter_stays_zero_once_enforcement_is_on(tmp_path, monkeypatch):
+    """The two counters partition the population rather than double-counting it: with
+    enforcement on, a forgery lands in `rejected`, which is the observable that already
+    existed."""
+    c, _, _, _ = _drain(tmp_path, [_response()], monkeypatch,
+                        keyset=_keyset(tmp_path), require_signature=True)
+    assert c.rejected == 1 and c.audit_failed == 0
+
+
+def test_no_audit_logging_when_verification_is_not_configured(
+        tmp_path, monkeypatch, capsys):
+    """The default deployment has no keyset. Nothing was checked, so there is nothing
+    to report, and an unconfigured bridge must not log a line per event."""
+    c, _, _, _ = _drain(tmp_path, [_response()], monkeypatch)
+    assert c.audit_failed == 0
+    assert "audit:" not in capsys.readouterr().out
 
 
 def test_an_undecodable_record_still_ends_the_loop_as_before(tmp_path, monkeypatch):

eventing/tests/test_keyset.py

@@ -402,39 +402,46 @@ def test_expect_kid_pins_the_signer_to_one_identity(tmp_path):
 # ---- response_decision: enabled? verified? enforced? -------------------------
 
 def test_response_decision_accepts_everything_when_no_keyset_is_configured():
-    """Today's behaviour, which must survive untouched as the default."""
-    ok, why = S.response_decision(_event(), None, require=True)
-    assert ok and "not enabled" in why
+    """Today's behaviour, which must survive untouched as the default. `verified=True`
+    here is the same reading as the unsigned non-terminal passthrough: nothing being
+    configured to check is an acceptance by policy, not a failed check — so
+    `accept and not verified` stays the audit-mode population with no caller-side
+    keyset guard."""
+    ok, why, verified = S.response_decision(_event(), None, require=True)
+    assert ok and verified and "not enabled" in why
 
 
 def test_response_decision_accepts_a_verified_response(tmp_path):
     ks = keyset.load(_write(tmp_path, {"runner-01": PUB.hex()}))
     e = _event()
     S.sign_into(e, SEED, "runner-01")
-    assert S.response_decision(e, ks, require=True) == (True, "ok")
+    assert S.response_decision(e, ks, require=True) == (True, "ok", True)
 
 
 def test_response_decision_rejects_an_unsigned_response_when_enforcing(tmp_path):
     ks = keyset.load(_write(tmp_path, {"runner-01": PUB.hex()}))
-    ok, why = S.response_decision(_event(), ks, require=True)
+    ok, why, verified = S.response_decision(_event(), ks, require=True)
     assert not ok and "no ce_signature" in why
 
 
 def test_response_decision_reports_but_accepts_in_audit_mode(tmp_path):
     """The two-flag rollout in one assertion: a keyset alone verifies and explains,
     without yet rewriting anything a user sees."""
     ks = keyset.load(_write(tmp_path, {"runner-01": PUB.hex()}))
-    ok, why = S.response_decision(_event(), ks, require=False)
+    ok, why, verified = S.response_decision(_event(), ks, require=False)
     assert ok, "audit mode must not reject"
     assert "no ce_signature" in why, "but it must still say what was wrong"
+    assert not verified, \
+        "and it must be distinguishable from a clean pass, which is what makes the " \
+        "reject rate countable without parsing the reason prose"
 
 
 def test_response_decision_requires_the_bridge_kid_on_group_events(tmp_path):
     ks = keyset.load(_write(tmp_path, {"eventbridge": PUB.hex(),
                                        "runner-01": PUB2.hex()}))
     forged = _event(type=ce.TYPE_GROUP_COMPLETED)
     S.sign_into(forged, SEED2, "runner-01")
-    ok, why = S.response_decision(forged, ks, require=True, bridge_kid="eventbridge")
+    ok, why, verified = S.response_decision(forged, ks, require=True, bridge_kid="eventbridge")
     assert not ok and "expected a signature from kid 'eventbridge'" in why
 
     # A plain response from that same runner is still fine — the pin is per-class.
@@ -453,16 +460,19 @@ def test_an_unsigned_non_terminal_frame_is_accepted(tmp_path):
     """
     ks = keyset.load(_write(tmp_path, {"runner-01": PUB.hex()}))
     frame = _event(sequence=1, phase="stdout", final="false")
-    ok, why = S.response_decision(frame, ks, require=True)
+    ok, why, verified = S.response_decision(frame, ks, require=True)
     assert ok, why
     assert "non-terminal" in why
+    assert verified, \
+        "accepted by policy, not in spite of a failure — otherwise an audit-mode " \
+        "counter would log every streamed frame of every genuine run"
 
 
 def test_an_unsigned_terminal_response_is_still_rejected(tmp_path):
     """The line the exemption must not cross: the terminal event is the one the
     transcript presents as the answer, so refusing it unsigned is the whole control."""
     ks = keyset.load(_write(tmp_path, {"runner-01": PUB.hex()}))
-    ok, why = S.response_decision(_event(phase="result", final="true"), ks, require=True)
+    ok, why, verified = S.response_decision(_event(phase="result", final="true"), ks, require=True)
     assert not ok and "no ce_signature" in why
 
 
@@ -472,7 +482,7 @@ def test_a_non_terminal_frame_that_claims_a_signature_is_still_verified(tmp_path
     ks = keyset.load(_write(tmp_path, {"runner-01": PUB.hex()}))
     frame = _event(sequence=1, phase="stdout", final="false")
     S.sign_into(frame, SEED2, "runner-01")          # approved kid, wrong key
-    ok, why = S.response_decision(frame, ks, require=True)
+    ok, why, verified = S.response_decision(frame, ks, require=True)
     assert not ok and "does not verify" in why
 
 
@@ -484,7 +494,7 @@ def test_an_unsigned_group_event_is_rejected_despite_having_no_final_attribute(t
     grp = _event(type=ce.TYPE_GROUP_COMPLETED, groupid="g-1")
     grp.attrs.pop("final", None)
     assert "final" not in grp.attrs
-    ok, why = S.response_decision(grp, ks, require=True, bridge_kid="eventbridge")
+    ok, why, verified = S.response_decision(grp, ks, require=True, bridge_kid="eventbridge")
     assert not ok and "no ce_signature" in why