diff --git a/eventing/eventbridge/kafka_in.py b/eventing/eventbridge/kafka_in.py index 599da1b9..440dea9c 100644 --- a/eventing/eventbridge/kafka_in.py +++ b/eventing/eventbridge/kafka_in.py @@ -73,6 +73,7 @@ def __init__( self._require_sig = require_signature self._bridge_kid = bridge_kid self._rejected = 0 + self._audit_failed = 0 self._stopping = threading.Event() @property @@ -80,6 +81,17 @@ 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,10 +118,10 @@ 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 @@ -117,8 +129,19 @@ def run(self) -> None: # 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 " diff --git a/eventing/shared/signing.py b/eventing/shared/signing.py index cfcf9103..44fa5f45 100644 --- a/eventing/shared/signing.py +++ b/eventing/shared/signing.py @@ -441,8 +441,8 @@ 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. @@ -450,13 +450,27 @@ def response_decision(event, ks, *, require: bool, 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 ------------------------------------------------------------ diff --git a/eventing/tests/test_kafka_in.py b/eventing/tests/test_kafka_in.py index f6a36e1a..cd47708e 100644 --- a/eventing/tests/test_kafka_in.py +++ b/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): diff --git a/eventing/tests/test_keyset.py b/eventing/tests/test_keyset.py index 95a4ff1f..b1d5eed6 100644 --- a/eventing/tests/test_keyset.py +++ b/eventing/tests/test_keyset.py @@ -402,21 +402,25 @@ 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 @@ -424,9 +428,12 @@ 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): @@ -434,7 +441,7 @@ def test_response_decision_requires_the_bridge_kid_on_group_events(tmp_path): "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