From d37122de4da44d2f84785eac277a92308225fc36 Mon Sep 17 00:00:00 2001 From: Mariusz Sabath Date: Fri, 9 Oct 2026 18:37:49 -0400 Subject: [PATCH 1/2] fix(eventing): a rejected response must not act, only be recorded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #885. With EB_REQUIRE_RESPONSE_SIGNATURE=true a response that fails verification was rewritten to phase="error" — and then acted on anyway. Enforcement changed what was recorded, not what took effect, so the sentence the demo rests on ("a forged response is refused") was false in the configuration that asks for it. Reproduced against the real functions, enforcement ON, a 2-agent batch: before: batch completed_utc set with 1 of 2 agents terminal counts {finished: 0, failed: 1} <- the agent had succeeded [groups] duplicate terminal event ... ignored <- the real answer, dropped after: batch still open, transcript holds the genuine answer, rejected=2 and both forgeries on record The forgery did not merely add a lie; the duplicate guard then discarded the truth. Three paths, one rule: a rejected envelope is evidence, not a fact — it is stored and rendered, but reaches neither on_group_event, on_member_event, nor a sequence a non-rejected row already holds. The marker is a top-level `ce.KEY_REJECTED`, not `phase` or `data`: * `phase == "error"` is a value GENUINE failures carry (runner.py emits it when an agent exits non-zero), and mark_member_finished maps it to `failed`. A guard keyed on it would stop every real failure from finishing its member and hang the batch to its deadline — silence, which §21.6 calls the worst outcome. * `data` arrives off the wire intact, so keying on it is §8.5.4 answered wrong. The `__` prefix turned out not to be enough on its own: a header named `ce___rejected` strips to exactly that key, so the marker WAS forgeable until from_kafka_binary started refusing `__`-prefixed names — and a forger able to set it could mark every genuine answer rejected, inverting the control rather than bypassing it. Found by the test written to assert the opposite. The store conflict rule is asymmetric on purpose. A genuine event keeps INSERT OR REPLACE, because at-least-once redelivery and emit.seed_seq's cold-start renumbering both depend on it — making the insert non-clobbering in general would re-open the Phase 1 per-pod sequence bug under a new name. Only a rejected event is refused a row it did not already own, and the refusal is counted (rejected_not_stored) and logged, because a silent drop is what §4.4 forbids. The group mirror is a second, independent acting path: a forged group.completed refused live was applied unconditionally on the next restart. It now verifies group.completed — and only that, and only when a keyset and bridge kid are both set. Verifying everything would be catastrophic rather than strict: the mirror replays history that predates signing, runs on every start, and the test/kind overlays keep /data on an emptyDir, so refusing unsigned records would lose every group's state on every restart. Unconditional, no new flag: `not ok` is only reachable when a keyset is loaded and enforcement is on, so the rule inherits those flags. Audit mode carries no marker and keeps acting, which is what keeps the reject-rate rollout working. test_an_approved_runner_cannot_forge_a_group_event asserted "rather than settling the group" while wiring a list's append as the callback, so nothing in it could tell whether the group settled — §8.5.1 exactly. It now drives a real GroupService and fails without the fix. 947 tests pass (924 before). final_seen still counts a rejected terminal; it is presentational and left to a follow-up with a VERIFY comment. Assisted-By: Claude (Anthropic AI) Signed-off-by: Mariusz Sabath --- eventing/agentdocs/DESIGN_PHASE2.md | 19 ++ eventing/eventbridge/__main__.py | 8 +- eventing/eventbridge/group_service.py | 18 ++ eventing/eventbridge/kafka_group_mirror.py | 41 +++- eventing/eventbridge/kafka_in.py | 64 +++++- eventing/eventbridge/store.py | 69 +++++- eventing/shared/ce.py | 52 +++++ eventing/tests/test_groups.py | 230 +++++++++++++++++++- eventing/tests/test_kafka_in.py | 235 ++++++++++++++++++++- eventing/tests/test_store.py | 82 +++++++ 10 files changed, 795 insertions(+), 23 deletions(-) diff --git a/eventing/agentdocs/DESIGN_PHASE2.md b/eventing/agentdocs/DESIGN_PHASE2.md index a21dd080..903eee1a 100644 --- a/eventing/agentdocs/DESIGN_PHASE2.md +++ b/eventing/agentdocs/DESIGN_PHASE2.md @@ -475,6 +475,25 @@ deletes the original intent loses the more useful half. ### 8.1 Promises in §4.4 that the code does not keep +**Status, 2026-10-09: the two #885 rows below are fixed; the table records what the code +did when the review found it, per §8.7.** A rejected event is now inert as well as +marked — it is stored and rendered, but it reaches neither `on_group_event`, +`on_member_event` nor a verified row's sequence, and the group mirror verifies +`group.completed` against the bridge's kid on replay. The single rule and the three +places it is read are `ce.is_rejected`; the reason it keys on a top-level marker rather +than on `phase="error"` is that genuine agent failures carry that phase, so the obvious +guard would have broken every real failure instead of the forgeries. + +Two corrections to what this table claims, found by running it: + +* The `__rejected` marker **was** forgeable from the wire. A header named + `ce___rejected` strips to exactly that key, so until `from_kafka_binary` started + refusing `__`-prefixed names a forger could mark every *genuine* answer rejected — + inverting the control rather than bypassing it. Found by the test written to assert + the opposite, which is §8.5.4's question answered the wrong way at first. +* #886 (the keyset-alone row) and #888 (the `EB_SIGNING_KID` row) were fixed earlier, in + #902 and #888's own change. + | §4.4 says | The code does | Issue | |---|---|---| | "A keyset alone verifies **and logs**" | Verifies and returns a verdict. Nothing is logged or counted, so the observable reject rate the two-flag rollout depends on does not exist. | [#886](https://github.com/rossoctl/examples/issues/886) | diff --git a/eventing/eventbridge/__main__.py b/eventing/eventbridge/__main__.py index 844402bf..14b2c102 100644 --- a/eventing/eventbridge/__main__.py +++ b/eventing/eventbridge/__main__.py @@ -204,10 +204,16 @@ def main() -> int: # commits offsets and so never re-reads, which meant a restarted pod (with an # emptyDir /data) served 404 for every earlier group even though its notifications # had already gone out. One-shot, never publishes, never notifies. + # #885 — the same keyset and kid the live consumer verifies with. The mirror uses + # them on `group.completed` only, because a forged completion refused above is + # otherwise applied unconditionally on the next restart. The half-configured case + # (a keyset with no kid) is already refused at startup above, so there is nothing + # to re-guard here. group_mirror = GroupMirror(cfg.kafka_bootstrap, cfg.response_topic, groups, topics=topics, userkeys=users.userkeys if topics.multi else (), - owners=owners if topics.multi else None) + owners=owners if topics.multi else None, + keyset=ks, bridge_kid=cfg.signing_kid or None) group_mirror.start() h = Handlers(cfg, store, producer, minter, groups=groups, diff --git a/eventing/eventbridge/group_service.py b/eventing/eventbridge/group_service.py index 1c235bb0..0c1ccadd 100644 --- a/eventing/eventbridge/group_service.py +++ b/eventing/eventbridge/group_service.py @@ -143,6 +143,17 @@ def on_member_event(self, event: dict[str, Any], *, replay: bool = False) -> Non corr = event.get("correlationid") if not groupid or not corr: return + # #885 — a rejected envelope is evidence, not a fact. Before `add_group_member`, + # because a membership row is itself a state change: it makes an unknown + # correlation a member of this batch and moves `members_seen`, which `progress` + # and `completion_reason` both read. + # + # Without this, a forged terminal wins `mark_member_finished`'s `finished_utc IS + # NULL` race and the genuine answer that follows is swallowed by the duplicate + # guard below — so the forgery does not merely add a lie, it destroys the truth. + if ce.is_rejected(event): + _log(f"rejected event for {groupid}/{corr} not counted") + return # §6.1: the store is chosen by the event's own `userkey`, which is signed (§2.6) # and therefore not rewritable in flight. Routing by the event rather than by any # ambient state is what keeps one tenant's batch progress out of another's store. @@ -177,6 +188,13 @@ def on_group_event(self, event: dict[str, Any], *, replay: bool = False) -> None groupid = event.get("groupid") if not groupid: return + # #885 — before `create_group` as well as before `complete_group`. A forged + # `started` is not harmless noise: `expected` is the denominator every progress + # figure divides by, and a forged `completed` ends a batch early and fires a + # "finished" notification for work that never ran. + if ce.is_rejected(event): + _log(f"rejected group event for {groupid} not applied") + return data = event.get("data") or {} store = self._store_for(event.get(ce.EXT_USERKEY)) if event.get("type") == ce.TYPE_GROUP_STARTED: diff --git a/eventing/eventbridge/kafka_group_mirror.py b/eventing/eventbridge/kafka_group_mirror.py index 2929d962..5e027666 100644 --- a/eventing/eventbridge/kafka_group_mirror.py +++ b/eventing/eventbridge/kafka_group_mirror.py @@ -35,7 +35,7 @@ from kafka import KafkaConsumer, TopicPartition -from shared import ce, tenancy +from shared import ce, signing, tenancy def _log(msg: str) -> None: @@ -49,7 +49,7 @@ def __init__(self, bootstrap: str, response_topic: str, groups, metadata_tries: int = 20, max_empty_polls: int = 15, topics: tenancy.TopicSet | None = None, userkeys: tuple[str, ...] = (), - owners=None) -> None: + owners=None, keyset=None, bridge_kid: str | None = None) -> None: super().__init__(daemon=True, name="kafka-group-mirror") self._bootstrap_servers = bootstrap # §3.2: unlike the live consumers this one is one-shot at startup, so it "can @@ -68,6 +68,24 @@ def __init__(self, bootstrap: str, response_topic: str, groups, # to, both for settling and for back-filling a missing `userkey`. `None` keeps # the single-tenant behaviour, where there is one store and nothing to resolve. self._owners = owners + # #885 — the mirror is a SECOND, independent acting path. A forged + # `group.completed` refused by the live consumer sits on the topic and was then + # applied unconditionally on the very next restart, so a guard in `kafka_in` + # alone closed the hole only until the pod bounced. + # + # **It verifies group.completed and nothing else, and only when both a keyset + # and a bridge kid are configured.** Verifying everything would be catastrophic + # rather than strict: the mirror exists to replay history that predates signing + # entirely, it runs on every start, and the `test`/`kind` overlays keep /data on + # an `emptyDir` — so refusing unsigned records would lose every group's state on + # every restart, which is the exact failure this module was written to prevent. + # + # `group.completed` is the only mirror path that can settle a batch: + # `on_member_event(replay=True)` returns before `maybe_complete`, and a forged + # `group.started` creates an empty group row, which is noise rather than a + # settled batch reporting work that never ran. + self._keyset = keyset + self._bridge_kid = bridge_kid # Partitions are assigned by hand and nothing is ever committed: the whole # point is to re-read from the beginning on every start, which the live # committing consumer cannot do. See run() for why there is no group id. @@ -77,6 +95,9 @@ def __init__(self, bootstrap: str, response_topic: str, groups, self.total_to_read = 0 self.rebuilt_groups = 0 self.rebuilt_members = 0 + # #885 — group.completed events skipped on replay because they could not be + # attributed to the bridge's own key. + self.refused = 0 self.settled = 0 def stop(self) -> None: @@ -220,6 +241,22 @@ def _apply(self, rec) -> str | None: d[ce.EXT_USERKEY] = owner try: if ce.is_group_event(evt): + # #885 — only the completion, and only when configured. See __init__. + if (self._keyset is not None and self._bridge_kid + and d.get("type") == ce.TYPE_GROUP_COMPLETED): + try: + ok, why, _ = signing.response_decision( + evt, self._keyset, require=True, + bridge_kid=self._bridge_kid) + except Exception as e: # noqa: BLE001 + # Fail closed: a verifier that cannot answer is not evidence + # that this completion is genuine. One skipped group is a 404 + # on its page; a wrongly-settled one reports work that never ran. + ok, why = False, f"verifier raised: {e!r}" + if not ok: + self.refused += 1 + _log(f"refusing replayed completion for {gid}: {why}") + return None self._groups.on_group_event(d, replay=True) self.rebuilt_groups += 1 else: diff --git a/eventing/eventbridge/kafka_in.py b/eventing/eventbridge/kafka_in.py index 440dea9c..97bdaf94 100644 --- a/eventing/eventbridge/kafka_in.py +++ b/eventing/eventbridge/kafka_in.py @@ -16,6 +16,24 @@ signature that can no longer be checked against anything. The original is therefore preserved under `data["rejected"]` — attrs, payload, claimed source and signature — so an incident can be verified offline rather than merely logged. + +**Stored is not the same as acted on (#885).** Marking a refused event was originally the +whole of enforcement, and it was not enough: the rewrite preserves `final` and `groupid` +so the record stays readable, which meant a rejected event went on finishing its member, +settling its batch, and overwriting the verified row at its sequence. With enforcement +on, a forged terminal could finish a batch as `failed: 1` and the genuine answer that +followed was discarded as a duplicate — so the forgery did not merely add a lie, it +destroyed the truth. A rejected envelope is therefore marked with `ce.KEY_REJECTED` and +reaches no acting path: not `on_group_event`, not `on_member_event`, and not a sequence +a non-rejected row already holds. + +**The guard reads `ok`, never `verified`, and that is load-bearing.** An unsigned +non-terminal frame and a deployment with no keyset both report `verified=False` while +being accepted *by policy* — `emit()` signs terminal events only, because a signature +costs ~222 ms and signing every stdout frame would add minutes to a chatty run. Keying +the guard on `verified` would stop every streamed frame of every genuine run from +acting, which is the §4.4 step 2 mistake: it rewrote every frame of every run to +`phase="error"` and was found against a live broker, by no unit test. """ from __future__ import annotations @@ -74,6 +92,7 @@ def __init__( self._bridge_kid = bridge_kid self._rejected = 0 self._audit_failed = 0 + self._rejected_not_stored = 0 self._stopping = threading.Event() @property @@ -81,6 +100,17 @@ def rejected(self) -> int: """Responses stored as phase=error because they did not verify.""" return self._rejected + @property + def rejected_not_stored(self) -> int: + """Rejected responses refused even the `phase=error` row, because a verified + row already held their (correlationid, sequence). #885 item 3. + + Separate from `rejected` because it is a different operational fact: the event + was refused AND it collided with a genuine answer, which is what an attempt to + overwrite a verified result looks like from the store's side. + """ + return self._rejected_not_stored + @property def audit_failed(self) -> int: """Responses accepted unchanged that enforcement *would* have rejected. @@ -158,6 +188,13 @@ def run(self) -> None: # destroy the forensic record while the docstring above still # promised it — leaving a signature whose covered attributes no # longer exist, and nothing for an operator to review. + # + # `ce.KEY_REJECTED` is the verdict itself, carried at the top + # level because that is the one thing every acting path receives. + # It is what `insert_response` and the group service read; the + # `data` keys below are for a human reading the transcript, and + # keying a control on them would be keying it on `data`, which + # arrives off the wire intact. See `ce.KEY_REJECTED`. d = dict(d, phase="error", data={ "text": f"unverified response rejected: {why}", "signature_rejected": True, @@ -168,12 +205,21 @@ def run(self) -> None: "signature": evt.get("signature"), "attrs": dict(evt.attrs), "data": evt.data}, - }) + }, **{ce.KEY_REJECTED: True}) # §21.2: route on type. A group lifecycle event carries `groupid` # but no `correlationid`, so handing it to insert_response would # violate that table's (correlationid, sequence) primary key. if ce.is_group_event(evt): - if self._on_group_event: + # #885 — `ok`, not `verified`. An unsigned non-terminal frame and + # a deployment with no keyset both report `verified=False` while + # being accepted BY POLICY, and skipping those would stop every + # streamed frame of every genuine run from reaching the group + # service. Only `not ok` is the enforcement-refused population. + # + # The group service refuses a marked envelope too (§885's single + # rule), so this is the source-side half of a rule enforced at + # both ends rather than the only guard. + if self._on_group_event and ok: try: self._on_group_event(d) except Exception as e: print(f"[kafka_in] group event: {e!r}") else: @@ -185,8 +231,18 @@ def run(self) -> None: # indistinguishable from an agent that never answered). target = (self._stores.for_event(evt) if self._stores else self._store) - target.insert_response(d) - if self._on_member_event and d.get("groupid"): + if not target.insert_response(d): + # #885 item 3 — the row at this sequence was NOT a rejection, + # so a rejected frame was refused the overwrite. Counted and + # logged rather than dropped silently: §4.4's rule is that a + # drop is indistinguishable from an agent that never + # answered. The verified row it failed to replace is the + # better record of what happened at this sequence. + self._rejected_not_stored += 1 + print(f"[kafka_in] rejected response for " + f"{d.get('correlationid')} seq {d.get('sequence')} " + f"not stored: a verified row holds that sequence") + if self._on_member_event and d.get("groupid") and ok: try: self._on_member_event(d) except Exception as e: print(f"[kafka_in] member event: {e!r}") if self._on_event: diff --git a/eventing/eventbridge/store.py b/eventing/eventbridge/store.py index 27dbff9e..36fa990b 100644 --- a/eventing/eventbridge/store.py +++ b/eventing/eventbridge/store.py @@ -7,6 +7,8 @@ import threading from typing import Any +from shared import ce + _SCHEMA_RESPONSES = """ CREATE TABLE IF NOT EXISTS responses ( correlationid TEXT NOT NULL, @@ -17,6 +19,12 @@ data_json TEXT NOT NULL, final INTEGER NOT NULL DEFAULT 0, raw_json TEXT NOT NULL, + -- #885 — this row is a refused event kept for review, not an answer. The flag is + -- `rejected` rather than `verified` on purpose: 0 is the truthful default for every + -- row written before this column existed and for every row written with verification + -- off. A `verified` column would default to "unverified" and anything keying on it + -- would start refusing ordinary traffic. + rejected INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (correlationid, sequence) ); CREATE INDEX IF NOT EXISTS responses_by_corr ON responses(correlationid); @@ -109,23 +117,63 @@ def __init__(self, root: str | pathlib.Path) -> None: self._s.execute("ALTER TABLE prompts ADD COLUMN submitter TEXT") except sqlite3.OperationalError: pass + try: + self._r.execute( + "ALTER TABLE responses ADD COLUMN rejected INTEGER NOT NULL DEFAULT 0") + except sqlite3.OperationalError: + pass self._lock = threading.Lock() self._sub_lock = threading.Lock() self._subscribers: dict[str, list[threading.Event]] = {} # ---- responses ---- - def insert_response(self, event: dict[str, Any]) -> None: + def insert_response(self, event: dict[str, Any]) -> bool: + """Store one response event. Returns False only when a rejected event was + refused because a row that is NOT a rejection already holds its sequence. + + **The conflict rule is deliberately asymmetric** (#885 item 3). + + A genuine event keeps `INSERT OR REPLACE`, because that is load-bearing: RQ-1 + accepts at-least-once delivery, so a redelivered frame must collapse onto one + row, and `emit.seed_seq` resumes numbering from the stored maximum on a cold + `/continue` turn — the fix for Phase 1's per-pod sequence restart. Making the + insert non-clobbering in general would re-open that bug under a new name. + + A rejected event may create a row, and may replace a row that is itself a + rejection so a redelivered forgery stays idempotent, but may never replace a row + that is not marked rejected. Otherwise an unsigned frame reusing a sequence + number overwrites the verified answer at that sequence, and sequence numbers are + readable without signing in. + + Both forms resolve the conflict in SQL rather than by raising: the caller runs + this outside its `try` on purpose, so a new `IntegrityError` path would end the + consume thread while the pod still reported healthy. + """ corr = event["correlationid"] sequence = int(event.get("sequence", "0")) phase = event.get("phase", "stdout") final = 1 if str(event.get("final", "false")).lower() == "true" else 0 + rejected = 1 if ce.is_rejected(event) else 0 + row = (corr, sequence, phase, event.get("id", ""), event.get("time", ""), + json.dumps(event.get("data")), final, json.dumps(event), rejected) with self._lock: - self._r.execute( - "INSERT OR REPLACE INTO responses(correlationid,sequence,phase,event_id,event_time,data_json,final,raw_json) VALUES (?,?,?,?,?,?,?,?)", - (corr, sequence, phase, event.get("id", ""), event.get("time", ""), - json.dumps(event.get("data")), final, json.dumps(event)), - ) + if rejected: + cur = self._r.execute( + "INSERT INTO responses(correlationid,sequence,phase,event_id,event_time,data_json,final,raw_json,rejected) VALUES (?,?,?,?,?,?,?,?,?) " + "ON CONFLICT(correlationid,sequence) DO UPDATE SET " + "phase=excluded.phase, event_id=excluded.event_id, " + "event_time=excluded.event_time, data_json=excluded.data_json, " + "final=excluded.final, raw_json=excluded.raw_json, rejected=1 " + "WHERE responses.rejected=1", + row) + if cur.rowcount != 1: + return False + else: + self._r.execute( + "INSERT OR REPLACE INTO responses(correlationid,sequence,phase,event_id,event_time,data_json,final,raw_json,rejected) VALUES (?,?,?,?,?,?,?,?,?)", + row) self._notify(corr) + return True def events_for(self, corr: str, since_seq: int = 0) -> list[dict[str, Any]]: rows = self._r.execute( @@ -149,6 +197,15 @@ def raw_events_for(self, corr: str) -> list[dict[str, Any]]: return [json.loads(r[0]) for r in rows] def final_seen(self, corr: str) -> bool: + # A rejected terminal still carries final=1, so this reports "answered" for a + # correlation whose only terminal event was refused. Presentational rather than + # state-advancing — it feeds the `"final"` field on two read endpoints and + # advances nothing — so it was left out of #885's scope deliberately rather + # than overlooked. + # + # VERIFY: once the follow-up lands, add `AND rejected=0` here and drop this + # comment. Keep the sentence above about why it is presentational — that is the + # part a maintainer acting on a vague note would delete. row = self._r.execute( "SELECT COUNT(*) FROM responses WHERE correlationid=? AND final=1", (corr,) ).fetchone() diff --git a/eventing/shared/ce.py b/eventing/shared/ce.py index 19a0fb5a..2c373ca8 100644 --- a/eventing/shared/ce.py +++ b/eventing/shared/ce.py @@ -85,6 +85,31 @@ CORE_ATTRS = {"specversion", "type", "source", "id", "time", "subject", "datacontenttype"} +# #885 — set by the responses consumer on an envelope whose signature was REFUSED under +# enforcement. Not a CloudEvent attribute and never on the wire: it is a local verdict, +# travelling with the one thing every acting path already receives. +# +# **Why a top-level key and not `phase` or `data`.** Both alternatives are wrong in a way +# that took reading the emit path to see: +# +# * `phase == "error"` is a GENUINE value. `eventrunner/runner.py` emits +# `phase="error", final=True` whenever an agent fails, and +# `Store.mark_member_finished` maps that to status `failed`. A guard keyed on `phase` +# would stop every real agent failure from finishing its member, so the batch would +# hang to its deadline — silence, which §21.6 calls the worst outcome for a batch. +# * `data["signature_rejected"]` is attacker-chosen. `data` arrives off the wire intact, +# so a forger could set it and turn the guard into a denial of service against genuine +# traffic. That is DESIGN_PHASE2 §8.5.4 ("can an attacker choose the input the check +# reads?") answered the wrong way. +# +# **The `__` prefix is a convention; `from_kafka_binary` is what enforces it.** The +# prefix alone is not enough and assuming it was is a mistake worth recording: a header +# named `ce___rejected` strips to exactly this key, so the wire *could* set the marker +# until the codec started refusing `__`-prefixed names. A forger able to set it could +# mark every genuine answer rejected, which is a denial of service rather than a +# bypass — the control inverted instead of removed. `test_kafka_in.py` pins the refusal. +KEY_REJECTED = "__rejected" + def is_group_event(event) -> bool: """True for a group lifecycle event. @@ -96,6 +121,22 @@ def is_group_event(event) -> bool: return event.get("type") in GROUP_TYPES +def is_rejected(event) -> bool: + """True for an envelope the responses consumer refused. #885. + + The one predicate every enforcement site reads, so the rule lives in one place: + **a rejected envelope is evidence, not a fact — it is stored and it is rendered, but + it never advances state.** + + Note what this does NOT read. The verdict is `accept`, never `verified`: an unsigned + non-terminal frame and a deployment with no keyset both come back `verified=False` + while being accepted *by policy*, and treating those as rejected would stop every + streamed frame of every genuine run from acting. That is the §4.4 step 2 mistake, + which was found against a live broker and by no unit test. + """ + return bool(event.get(KEY_REJECTED)) + + def session_uuid(correlationid: str) -> str: return str(uuid.uuid5(NAMESPACE, correlationid)) @@ -146,6 +187,17 @@ def from_kafka_binary(headers: Iterable[tuple[str, bytes | None]], value: bytes if not k or not k.startswith(CE_HEADER_PREFIX): continue name = k[len(CE_HEADER_PREFIX):] + # #885 — refuse any `__`-prefixed name. These are local verdicts the consumer + # writes onto an envelope (see KEY_REJECTED), never wire attributes, and CloudEvents + # v1.0 restricts attribute names to [a-z0-9] so nothing legitimate is lost. + # + # Without this the marker IS forgeable: a header named `ce___rejected` strips to + # `__rejected`, and a forger who can set the marker can mark every GENUINE answer + # rejected — turning the control into a denial of service. Found by the test that + # was written to assert the opposite, which is why §8.5.4 asks the question + # instead of trusting the prefix argument. + if name.startswith("__"): + continue if v is None: continue attrs[name] = v.decode("utf-8") if isinstance(v, (bytes, bytearray)) else str(v) diff --git a/eventing/tests/test_groups.py b/eventing/tests/test_groups.py index cccf656b..d73c9526 100644 --- a/eventing/tests/test_groups.py +++ b/eventing/tests/test_groups.py @@ -6,6 +6,7 @@ counter would then reach zero while agents were still running — firing "all done" early, which is the one lie a progress display must never tell. """ +import binascii import datetime as dt import io import json @@ -19,7 +20,8 @@ from eventbridge.groups import MIN_FINISHED_FOR_ETA, completion_reason, humanize, progress, summary_line from eventbridge.handlers import Handlers from eventbridge.store import Store -from shared import ce +from shared import ce, keyset +from shared import signing as S # ---- helpers ---------------------------------------------------------------- @@ -1083,11 +1085,11 @@ def _record(event_dict, offset): return _FakeRec(headers, value, offset) -def _mirror_over(records_by_partition, groups, monkeypatch, **kw): +def _mirror_over(records_by_partition, groups, monkeypatch, *, mirror_kw=None, **kw): from eventbridge import kafka_group_mirror as mod fake = _FakeConsumer(records_by_partition, **kw) monkeypatch.setattr(mod, "KafkaConsumer", lambda **_: fake) - m = mod.GroupMirror("broker:9092", "responses", groups) + m = mod.GroupMirror("broker:9092", "responses", groups, **(mirror_kw or {})) m.run() # synchronous: the scan is one-shot by design return m, fake @@ -1248,3 +1250,225 @@ def _boom(**_): monkeypatch.setattr(mod, "KafkaConsumer", _boom) mod.GroupMirror("broker:9092", "responses", fresh).run() # must not raise + + +# ---- a rejected event is evidence, not a fact (#885) ------------------------ +# +# The guard these pin keys on `ce.KEY_REJECTED` and nothing else. The first test is the +# reason: `phase="error"` is a value genuine agent failures carry, so a guard written the +# obvious way — refusing to act on `phase == "error"` — breaks every real failure instead +# of the forgeries. It is written first and it passes both before and after the fix. + +def test_a_genuine_error_phase_terminal_still_fails_its_member(svc): + """The false positive the #885 guard must not have. + + `eventrunner/runner.py` emits `phase="error", final=True` when an agent exits + non-zero or dies without a result frame, and `mark_member_finished` maps that to + status `failed`. If the guard keyed on `phase` rather than on the rejection marker, + a genuinely failed agent would never finish its member: the batch would sit at + "1 running" until the deadline sweep, which §21.6 names as the worst outcome + because it fails as silence. + """ + s, store, producer, _ = svc + gid, _ = s.create(label="one-fails", expected=1) + corr, = s.submit_members(gid, ["do a thing"]) + s.on_member_event(member_event(gid, corr, final=True, phase="error", + text="claude exited with code 1")) + members = store.group_members(gid) + assert members[0]["status"] == "failed", \ + "a genuine error terminal must still finish its member, as a failure" + assert store.group_counts(gid)["terminal"] == 1 + assert store.get_group(gid)["completed_utc"] is not None, \ + "and the batch must still complete rather than hang to its deadline" + completions = [e for e in producer.group_events + if e["type"] == ce.TYPE_GROUP_COMPLETED] + assert len(completions) == 1 + assert completions[0]["data"]["failed"] == 1 + + +def test_a_rejected_member_event_does_not_finish_a_member(svc): + """§8.5.1, followed to the outcome: the verdict has to stop the state change. + + A rejected terminal that reaches `mark_member_finished` wins the `finished_utc IS + NULL` race, so the member is finished — as a failure, because the rewrite carries + `phase="error"` — and the genuine answer arriving afterwards is discarded by the + duplicate guard. The forgery does not just add a lie; it destroys the truth. + """ + s, store, producer, _ = svc + gid, _ = s.create(label="forged", expected=1) + corr, = s.submit_members(gid, ["do a thing"]) + forged = member_event(gid, corr, final=True, phase="error", + text="unverified response rejected: no ce_signature") + forged[ce.KEY_REJECTED] = True + s.on_member_event(forged) + members = store.group_members(gid) + assert members[0]["status"] != "finished", \ + "a rejected event must not finish a member" + assert members[0]["status"] != "failed", \ + "nor fail one — it is not evidence about the agent at all" + assert store.group_counts(gid)["terminal"] == 0 + assert store.get_group(gid)["completed_utc"] is None, \ + "and must not settle the batch" + assert [e for e in producer.group_events + if e["type"] == ce.TYPE_GROUP_COMPLETED] == [] + + +def test_the_genuine_answer_still_completes_the_batch_after_a_rejected_one(svc): + """The harm in #885's item 1, end to end: the real answer must survive. + + Reproduced in the issue as `0 finished, 1 failed` followed by `duplicate terminal + event ignored` — the batch reporting a failure that did not occur while the genuine + answer was dropped on the floor. + """ + s, store, producer, _ = svc + gid, _ = s.create(label="forged-then-real", expected=1) + corr, = s.submit_members(gid, ["what is 2+2"]) + forged = member_event(gid, corr, final=True, phase="error", text="Ship the goods") + forged[ce.KEY_REJECTED] = True + s.on_member_event(forged) + s.on_member_event(member_event(gid, corr, final=True, phase="result", + text="2+2 is 4", seq=2)) + members = store.group_members(gid) + assert members[0]["status"] == "finished", \ + "the genuine answer must still be the one that finishes the member" + assert members[0]["terminal_phase"] == "result" + counts = store.group_counts(gid) + assert counts["finished"] == 1 and counts["failed"] == 0, \ + "the batch must not report a failure that never happened" + completions = [e for e in producer.group_events + if e["type"] == ce.TYPE_GROUP_COMPLETED] + assert len(completions) == 1, "and must complete exactly once" + assert completions[0]["data"]["failed"] == 0 + + +def test_a_rejected_group_completed_event_does_not_settle_a_group(svc): + """#885 item 2: `on_group_event` reads only `type` and `groupid`, both of which the + rejection rewrite deliberately preserves so the forensic record stays readable.""" + s, store, _producer, _ = svc + gid, _ = s.create(label="open", expected=2) + s.submit_members(gid, ["a", "b"]) + s.on_group_event({"type": ce.TYPE_GROUP_COMPLETED, "groupid": gid, + "data": {"reason": "all"}, ce.KEY_REJECTED: True}) + assert store.get_group(gid)["completed_utc"] is None, \ + "a forged completion must not end a batch early" + + +def test_a_rejected_group_started_event_does_not_create_a_group(svc): + """The `create_group` half, which is the state change easy to forget: a group row + with a forged label and expected count is still a state change, and `expected` is + the denominator every progress figure is computed against.""" + s, store, _producer, _ = svc + s.on_group_event({"type": ce.TYPE_GROUP_STARTED, "groupid": "forged-group-0001", + "data": {"label": "injected", "expected": 99}, + ce.KEY_REJECTED: True}) + assert store.get_group("forged-group-0001") is None, \ + "a rejected started event must not create a group" + + +# ---- the mirror is a second acting path (#885) ------------------------------- +# +# A forged `group.completed` refused by the live consumer stays on the topic, and the +# mirror replays the topic on every start. So the consumer-side guard closed the hole +# only until the pod bounced. The mirror verifies `group.completed` and nothing else, +# because it also exists to replay history published before signing did. + +_MIRROR_SEED_EB = binascii.unhexlify( + "9d61b19deffd5a60ba844af492ec2cc44449c5697b326919703bac031cae7f60") +_MIRROR_SEED_R1 = binascii.unhexlify( + "4ccd089b28ff96da9db6c346ec114e0f5b8a319f35aba624da8cf6ed4fb8a6fb") + + +def _mirror_keyset(): + return keyset.KeySet({"eb-01": S.public_key(_MIRROR_SEED_EB), + "runner-01": S.public_key(_MIRROR_SEED_R1)}) + + +def _signed_completion(gid, *, seed, kid): + evt = ce.new_event(type=ce.TYPE_GROUP_COMPLETED, source="/eb", groupid=gid, + datacontenttype="application/json", + data={"reason": "all", "finished": 2, "failed": 0}) + S.sign_into(evt, seed, kid) + headers, value = ce.to_kafka_binary(evt) + return _FakeRec(headers, value, 0) + + +def _restarted(cfg, name): + store = Store(pathlib.Path(cfg.tmpdir) / name) + prod = FakeProducer() + return GroupService(cfg, store, prod, SeqMinter()), store, prod + + +def test_the_mirror_refuses_a_group_completed_it_cannot_attribute_to_the_bridge( + svc, monkeypatch): + """#885 item 2's second half. runner-01 is approved — it is just not the bridge, + and `EB_SIGNING_KID` is documented as the only kid accepted on group lifecycle + events. Without this the forgery is refused live and applied on the next restart. + """ + _s, _store, _producer, cfg = svc + fresh, fresh_store, _fresh_prod = _restarted(cfg, "eb-refuse") + rec = _signed_completion("g-forged-0001", seed=_MIRROR_SEED_R1, kid="runner-01") + m, _ = _mirror_over({0: [rec]}, fresh, monkeypatch, + mirror_kw={"keyset": _mirror_keyset(), "bridge_kid": "eb-01"}) + assert m.refused == 1, "the refusal must be counted" + assert fresh_store.get_group("g-forged-0001") is None, \ + "a replayed completion signed by a non-bridge key must not settle anything" + + +def test_the_mirror_applies_a_group_completed_signed_by_the_bridge(svc, monkeypatch): + """The positive case: verification must not break the thing the mirror is for.""" + _s, _store, _producer, cfg = svc + fresh, fresh_store, _fresh_prod = _restarted(cfg, "eb-accept") + gid = "g-genuine-0001" + fresh_store.create_group(gid, label="b", expected=2, min_success=None, + deadline_utc=None) + rec = _signed_completion(gid, seed=_MIRROR_SEED_EB, kid="eb-01") + m, _ = _mirror_over({0: [rec]}, fresh, monkeypatch, + mirror_kw={"keyset": _mirror_keyset(), "bridge_kid": "eb-01"}) + assert m.refused == 0 + assert fresh_store.get_group(gid)["completed_utc"] is not None, \ + "a completion signed by the bridge must still settle its group" + + +def test_the_mirror_still_replays_unsigned_pre_phase_two_history(svc, monkeypatch): + """**The risk that matters more than the hole.** + + The mirror runs on every start and the test/kind overlays keep /data on an + `emptyDir`, so if enforcement made it refuse unsigned records, every restart would + lose every group's state — the exact failure this module was written to prevent, + reintroduced in the name of security. Member events and `group.started` are + therefore never verified. + """ + _s, _store, _producer, cfg = svc + fresh, fresh_store, _fresh_prod = _restarted(cfg, "eb-history") + gid = "g-legacy-0001" + started = _record({"type": ce.TYPE_GROUP_STARTED, "source": "/eb", "groupid": gid, + "datacontenttype": "application/json", + "data": {"label": "legacy", "expected": 1}}, 0) + member = _record({"type": ce.TYPE_RESPONSE, "source": "/er", + "datacontenttype": "application/json", + "correlationid": "test-agent-9001", "groupid": gid, + "sequence": "1", "phase": "result", "final": "true", + "data": {"text": "done"}}, 1) + m, _ = _mirror_over({0: [started, member]}, fresh, monkeypatch, + mirror_kw={"keyset": _mirror_keyset(), "bridge_kid": "eb-01"}) + assert m.refused == 0, "unsigned history is not a forgery" + row = fresh_store.get_group(gid) + assert row is not None and row["label"] == "legacy", \ + "an unsigned group.started must still rebuild its group" + assert fresh_store.group_counts(gid)["terminal"] == 1, \ + "and unsigned member events must still rebuild membership" + + +def test_the_mirror_verifies_nothing_when_no_keyset_is_configured(svc, monkeypatch): + """The default, per §8.5.2: with no keyset the mirror behaves exactly as before, + which is what every unsigned deployment runs.""" + _s, _store, _producer, cfg = svc + fresh, fresh_store, _fresh_prod = _restarted(cfg, "eb-nokeyset") + gid = "g-nokeyset-0001" + fresh_store.create_group(gid, label="b", expected=1, min_success=None, + deadline_utc=None) + rec = _signed_completion(gid, seed=_MIRROR_SEED_R1, kid="runner-01") + m, _ = _mirror_over({0: [rec]}, fresh, monkeypatch) + assert m.refused == 0 + assert fresh_store.get_group(gid)["completed_utc"] is not None, \ + "with verification off, replay is unchanged" diff --git a/eventing/tests/test_kafka_in.py b/eventing/tests/test_kafka_in.py index cd47708e..570cfbf4 100644 --- a/eventing/tests/test_kafka_in.py +++ b/eventing/tests/test_kafka_in.py @@ -92,6 +92,52 @@ def close(self): self.closed = True +class _FakeProducer: + """Records what the group service would publish. Deliberately a local copy of + `test_groups.py`'s: importing across test modules couples two files that otherwise + share nothing, and this one only needs the group-event half.""" + + def __init__(self): + self.requests = [] + self.group_events = [] + + def publish_request(self, **kw): + self.requests.append(kw) + return f"evt-{len(self.requests)}" + + def publish_group_event(self, *, type_, groupid, data, subject="group", # noqa: ARG002 + userkey=None): + self.group_events.append({"type": type_, "groupid": groupid, "data": data, + "userkey": userkey}) + return f"gevt-{len(self.group_events)}" + + +class _SeqMinter: + r"""Deterministic ids satisfying correlation.REGEX (^[a-z]{3,10}-[a-z]{3,12}-\d{4}$).""" + + def __init__(self): + self.n = 0 + + def mint(self): + self.n += 1 + return f"test-agent-{self.n:04d}" + + def remember(self, corr): + pass + + +def _group_service(tmp_path): + """A real GroupService over a real Store, so a test can assert the OUTCOME of a + verdict rather than the verdict. The store is shared with the consumer so both + write to one database, which is how they are wired in `__main__.py`.""" + from eventbridge.config import Cfg as EbCfg + from eventbridge.group_service import GroupService + store = Store(pathlib.Path(tmp_path) / "eb") + cfg = EbCfg(tmpdir=str(tmp_path), public_base_url="http://eb.test") + producer = _FakeProducer() + return GroupService(cfg, store, producer, _SeqMinter()), store, producer + + def _drain(tmp_path, records, monkeypatch, *, store=None, **kw): """Run the consumer over `records` to exhaustion, synchronously. @@ -294,15 +340,29 @@ def test_a_group_event_signed_by_the_bridge_is_accepted(tmp_path, monkeypatch): def test_an_approved_runner_cannot_forge_a_group_event(tmp_path, monkeypatch): """The hole a flat keyset would leave: a forged `group.completed` ends a batch early and fires a "finished" notification for work that never ran. runner-01 is - genuinely approved — it is just not the bridge.""" - groups: list[dict] = [] - c, _, _, _ = _drain(tmp_path, [_group_event(seed=SEED_R1, kid="runner-01")], - monkeypatch, keyset=_keyset(tmp_path), + genuinely approved — it is just not the bridge. + + **This test used to stop at the check.** It asserted that the dict handed to + `on_group_event` carried `phase="error"`, with the message "rather than settling the + group" — but it wired a list's `append` as the callback, so nothing in it could tell + whether the group settled. The verdict was correct and ignored, which is #885, and + asserting a verdict while claiming an outcome is DESIGN_PHASE2 §8.5.1 exactly: a + claim about an outcome has to follow the path to that outcome. It now drives a real + `GroupService` and asserts the batch. + """ + svc, store, producer = _group_service(tmp_path) + gid, _ = svc.create(label="real-work", expected=2) + svc.submit_members(gid, ["a", "b"]) + c, _, _, _ = _drain(tmp_path, [_group_event(gid=gid, seed=SEED_R1, kid="runner-01")], + monkeypatch, store=store, keyset=_keyset(tmp_path), require_signature=True, bridge_kid="eb-01", - on_group_event=groups.append) + on_group_event=svc.on_group_event) assert c.rejected == 1 - assert groups and groups[0]["phase"] == "error", \ - "the group event is marked rejected rather than settling the group" + assert store.get_group(gid)["completed_utc"] is None, \ + "a forged completion must not settle a batch whose agents never ran" + assert [e for e in producer.group_events + if e["type"] == ce.TYPE_GROUP_COMPLETED] == [], \ + "and must not publish a completion, which is what pages the operator's phone" # ---- the property this file exists for -------------------------------------- @@ -419,3 +479,164 @@ def test_an_undecodable_record_still_ends_the_loop_as_before(tmp_path, monkeypat assert "evt = ce.from_kafka_binary(rec.headers or [], rec.value)" in src assert "enable_auto_commit=True" in src, ( "the live consumer must keep committing — test_groups.py pins this too") + + +# ---- a rejected event must not act (#885) ----------------------------------- +# +# The rewrite marks a refused event; these pin that it is also inert. The distinction +# matters because the rewrite deliberately preserves `final` and `groupid` so the +# forensic record stays readable — which is exactly what let a rejected event go on +# finishing members and settling batches. + +def test_a_rejected_member_response_does_not_finish_its_member(tmp_path, monkeypatch): + """#885 item 1 at the consumer boundary: the verdict must stop the callback.""" + calls: list[dict] = [] + c, store, _, _ = _drain( + tmp_path, [_response(groupid="g-1", seq=1)], monkeypatch, + keyset=_keyset(tmp_path), require_signature=True, bridge_kid="eb-01", + on_member_event=calls.append) + assert c.rejected == 1 + assert calls == [], "a rejected event must not reach on_member_event" + rows = store.events_for("brave-otter-4718") + assert rows and rows[0]["phase"] == "error", \ + "but it is still stored, because a drop looks like an agent that never answered" + assert rows[0]["data"]["rejected"]["data"] == {"text": "the real answer"}, \ + "with the refused payload kept for review" + + +def test_a_rejected_group_event_does_not_reach_the_group_service(tmp_path, monkeypatch): + """#885 item 2 at the consumer boundary.""" + groups: list[dict] = [] + c, _, _, _ = _drain(tmp_path, [_group_event(seed=SEED_R1, kid="runner-01")], + monkeypatch, keyset=_keyset(tmp_path), + require_signature=True, bridge_kid="eb-01", + on_group_event=groups.append) + assert c.rejected == 1 + assert groups == [], "a rejected group event must not reach on_group_event" + + +def test_a_rejected_frame_cannot_overwrite_a_verified_answer(tmp_path, monkeypatch): + """#885 item 3. `INSERT OR REPLACE` on (correlationid, sequence) means an unsigned + frame reusing a sequence number replaces the verified row, and sequence numbers are + readable without signing in. Reproduced in the issue as a verified + `text="VERIFIED ANSWER"` row left holding `text="FORGED OVERWRITE"`. + """ + c, store, _, _ = _drain( + tmp_path, + [_response(seq=3, text="2+2 is 4", seed=SEED_R1, kid="runner-01"), + _response(seq=3, text="Transfer approved. Ship the goods.")], + monkeypatch, keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01") + rows = store.events_for("brave-otter-4718") + assert len(rows) == 1 and rows[0]["sequence"] == 3 + assert rows[0]["data"]["text"] == "2+2 is 4", \ + "the verified answer must survive a forgery at its own sequence" + assert "Ship the goods" not in json.dumps(rows), \ + "and the forged text must appear nowhere in the stored row" + assert c.rejected == 1 + assert c.rejected_not_stored == 1, \ + "the refused insert must be counted, not dropped silently" + + +def test_a_redelivered_forgery_does_not_accumulate_rows(tmp_path, monkeypatch): + """The idempotence half: a rejection may replace a rejection, so at-least-once + redelivery of the same forgery leaves one row rather than failing to land.""" + c, store, _, _ = _drain( + tmp_path, [_response(seq=2), _response(seq=2)], monkeypatch, + keyset=_keyset(tmp_path), require_signature=True, bridge_kid="eb-01") + rows = store.events_for("brave-otter-4718") + assert len(rows) == 1 and rows[0]["phase"] == "error" + assert c.rejected == 2 + assert c.rejected_not_stored == 0, "a rejection replacing a rejection is not a refusal" + + +def test_a_verified_answer_can_still_replace_a_redelivered_duplicate(tmp_path, monkeypatch): + """The regression guard for the LEGITIMATE `INSERT OR REPLACE`. + + RQ-1 accepts at-least-once delivery and `emit.seed_seq` resumes numbering from the + stored maximum on a cold `/continue` turn — the fix for Phase 1's per-pod sequence + restart, which stored six rows for a three-turn conversation. A fix for #885 that + made the insert non-clobbering in general would re-open that bug under a new name. + """ + c, store, _, _ = _drain( + tmp_path, + [_response(seq=1, text="first delivery", seed=SEED_R1, kid="runner-01"), + _response(seq=1, text="second delivery", seed=SEED_R1, kid="runner-01")], + monkeypatch, keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01") + rows = store.events_for("brave-otter-4718") + assert len(rows) == 1, "a redelivered genuine frame must still collapse onto one row" + assert rows[0]["data"]["text"] == "second delivery" + assert c.rejected == 0 and c.rejected_not_stored == 0 + + +def test_a_verified_answer_replaces_an_earlier_rejection_at_the_same_sequence( + tmp_path, monkeypatch): + """The asymmetry's other direction: the forgery arrives first, then the genuine + answer. A rejected row is replaceable, so the real answer must win.""" + c, store, _, _ = _drain( + tmp_path, + [_response(seq=4, text="Ship the goods"), + _response(seq=4, text="2+2 is 4", seed=SEED_R1, kid="runner-01")], + monkeypatch, keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01") + rows = store.events_for("brave-otter-4718") + assert len(rows) == 1 + assert rows[0]["phase"] == "result" and rows[0]["data"]["text"] == "2+2 is 4", \ + "a genuine answer must be able to replace a rejection at its sequence" + assert c.rejected == 1 and c.rejected_not_stored == 0 + + +def test_audit_mode_still_lets_an_unverified_event_act(tmp_path, monkeypatch): + """The guard reads `ok`, never `verified`. Audit mode exists so an operator can + watch the reject rate BEFORE enforcing, which needs the events to keep acting.""" + calls: list[dict] = [] + c, store, _, _ = _drain( + tmp_path, [_response(groupid="g-1")], monkeypatch, + keyset=_keyset(tmp_path), require_signature=False, bridge_kid="eb-01", + on_member_event=calls.append) + assert c.audit_failed == 1 and c.rejected == 0 + assert len(calls) == 1, "audit mode must not make an event inert" + assert store.events_for("brave-otter-4718")[0]["phase"] == "result", \ + "and must store it unchanged" + + +def test_unsigned_streamed_frames_still_reach_the_group_service(tmp_path, monkeypatch): + """`emit()` signs terminal events only, because a signature costs ~222 ms and + signing every stdout frame would add minutes to a chatty run. So an unsigned + non-terminal is accepted BY POLICY and reports `verified=True`. + + If a guard keyed on `verified` instead of `ok`, every streamed frame of every + genuine run would stop driving `mark_member_running`. That is the §4.4 step 2 + mistake, which rewrote every frame of every run to `phase="error"` and was found + against a live broker rather than by any unit test. + """ + calls: list[dict] = [] + c, _, _, _ = _drain( + tmp_path, + [_response(groupid="g-1", seq=1, phase="stdout", final="false", text="thinking"), + _response(groupid="g-1", seq=2, phase="result", text="done", + seed=SEED_R1, kid="runner-01")], + monkeypatch, keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01", on_member_event=calls.append) + assert c.rejected == 0, "an unsigned non-terminal frame is not a rejection" + assert len(calls) == 2, "both the streamed frame and the signed terminal must act" + + +def test_the_rejection_marker_cannot_arrive_from_the_wire(tmp_path, monkeypatch): + """The marker is only unforgeable if the wire cannot produce it. + + `from_kafka_binary` derives attribute names from `ce_`-prefixed headers, so this + would need a header literally named `ce___rejected`. Asserted rather than argued, + because the whole guard rests on it: a forger who could set the marker could mark + every genuine answer rejected, turning the control into a denial of service. + """ + rec = _response(seq=1, text="hi", seed=SEED_R1, kid="runner-01") + hdrs = list(rec.headers) + [("ce___rejected", b"true"), + (f"ce_{ce.KEY_REJECTED}", b"true")] + c, store, seen, _ = _drain(tmp_path, [_Rec(hdrs, rec.value)], monkeypatch, + keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01") + assert c.rejected == 0, "the wire must not be able to mark an event rejected" + assert seen and not ce.is_rejected(seen[0]) + assert store.events_for("brave-otter-4718")[0]["phase"] == "result" diff --git a/eventing/tests/test_store.py b/eventing/tests/test_store.py index 71b00c7c..12bc6284 100644 --- a/eventing/tests/test_store.py +++ b/eventing/tests/test_store.py @@ -3,6 +3,7 @@ import threading from eventbridge.store import Store +from shared import ce def test_insert_and_events_for(tmp_path: pathlib.Path): @@ -52,3 +53,84 @@ def test_upsert_session_increments_turns(tmp_path: pathlib.Path): s.upsert_session(corr, "uuid-x", "/tmp/work", None) session = s.get_session(corr) assert session and session["turns"] == 2 + + +# ---- the asymmetric conflict rule (#885 item 3) ----------------------------- + +def _row(corr, seq, text, *, rejected=False, phase="result"): + e = {"correlationid": corr, "sequence": seq, "phase": phase, "id": f"id-{seq}", + "time": "2026-10-09T00:00:00Z", "data": {"text": text}, "final": "true"} + if rejected: + e[ce.KEY_REJECTED] = True + return e + + +def test_a_duplicate_delivery_still_collapses_onto_one_row(tmp_path: pathlib.Path): + """The contract the asymmetry must not break, stated so the next person changing + conflict resolution can see what it protects. + + RQ-1 accepts at-least-once delivery, and `emit.seed_seq` resumes numbering from the + stored maximum on a cold `/continue` turn — the fix for the Phase 1 bug where a new + pod restarted `sequence` at 1 and stored six rows for a three-turn conversation. + """ + s = Store(tmp_path) + corr = "test-otter-0010" + assert s.insert_response(_row(corr, 1, "first delivery")) is True + assert s.insert_response(_row(corr, 1, "second delivery")) is True + rows = s.events_for(corr) + assert len(rows) == 1, "a redelivered genuine frame must collapse onto one row" + assert rows[0]["data"]["text"] == "second delivery" + + +def test_a_rejected_row_never_replaces_one_that_was_not_rejected(tmp_path: pathlib.Path): + """#885 item 3 at the store boundary, with no Kafka in the picture.""" + s = Store(tmp_path) + corr = "test-otter-0011" + s.insert_response(_row(corr, 1, "2+2 is 4")) + assert s.insert_response(_row(corr, 1, "Ship the goods", rejected=True)) is False, \ + "the refusal must be reported so the caller can count it" + rows = s.events_for(corr) + assert len(rows) == 1 and rows[0]["data"]["text"] == "2+2 is 4", \ + "the verified answer must survive" + + +def test_a_rejected_row_may_replace_another_rejection(tmp_path: pathlib.Path): + """So a redelivered forgery stays idempotent rather than accumulating rows.""" + s = Store(tmp_path) + corr = "test-otter-0012" + assert s.insert_response(_row(corr, 1, "forgery one", rejected=True)) is True + assert s.insert_response(_row(corr, 1, "forgery two", rejected=True)) is True + rows = s.events_for(corr) + assert len(rows) == 1 and rows[0]["data"]["text"] == "forgery two" + + +def test_a_genuine_row_may_replace_a_rejection(tmp_path: pathlib.Path): + """The other direction: the forgery landed first, the real answer must still win.""" + s = Store(tmp_path) + corr = "test-otter-0013" + s.insert_response(_row(corr, 1, "Ship the goods", rejected=True)) + assert s.insert_response(_row(corr, 1, "2+2 is 4")) is True + rows = s.events_for(corr) + assert len(rows) == 1 and rows[0]["data"]["text"] == "2+2 is 4" + + +def test_the_rejected_column_is_added_to_an_existing_database(tmp_path: pathlib.Path): + """The idempotent ALTER, mirroring the `prompts.submitter` precedent: a database + created before this column existed must keep working, because `CREATE TABLE IF NOT + EXISTS` is a no-op on an existing file.""" + import sqlite3 + (tmp_path / "x").mkdir() + conn = sqlite3.connect(tmp_path / "x" / "responses.sqlite", isolation_level=None) + conn.execute("CREATE TABLE responses (correlationid TEXT NOT NULL, " + "sequence INTEGER NOT NULL, phase TEXT NOT NULL, " + "event_id TEXT NOT NULL, event_time TEXT NOT NULL, " + "data_json TEXT NOT NULL, final INTEGER NOT NULL DEFAULT 0, " + "raw_json TEXT NOT NULL, PRIMARY KEY (correlationid, sequence))") + conn.close() + s = Store(tmp_path / "x") + corr = "test-otter-0014" + assert s.insert_response(_row(corr, 1, "after upgrade")) is True + assert s.insert_response(_row(corr, 1, "forged", rejected=True)) is False, \ + "the upgraded database must enforce the rule too" + assert s.events_for(corr)[0]["data"]["text"] == "after upgrade" + Store(tmp_path / "x") # reopening must not raise on the already-added column From a3882e40e630f441d531c5c893d1b42b29fdf9c0 Mon Sep 17 00:00:00 2001 From: Mariusz Sabath Date: Mon, 12 Oct 2026 09:52:11 -0400 Subject: [PATCH 2/2] fix(eventing): correct three comments the #955 review found wrong MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All three are in comments rather than in the enforcement logic, which is why the suite is unchanged apart from one new test. `verified` was described backwards in three places. `response_decision` returns `verified=True` for an unsigned non-terminal frame and for a deployment with no keyset — acceptance BY POLICY rather than a check that passed, which is what keeps `accept and not verified` meaning exactly the audit-mode population. The comments said `verified=False`. The code was right (it keys on `ok`); the comments would have sent the next reader looking for a flag that does not behave that way, and the correct argument for reading `ok` is stronger: keying on `verified` would spare nothing extra AND would make audit mode inert. The mirror paragraph answered settling and read as if it covered membership. `on_member_event(replay=True)` runs `add_group_member` and `mark_member_finished` before its `replay` check, and the marker is local rather than on the wire — so a forged terminal earlier in the topic than the genuine answer wins `finished_utc IS NULL` on every start and the batch reads back `failed: 1` for an agent that succeeded. Bounded (the batch stays open) but worst on the test/kind overlays, where /data is an emptyDir and the replay IS the state. Not fixable by widening this check: verifying member events on replay needs require=True, which refuses all pre-signing history, and "verify only if signed" is not a control because a forger omits it. Scoped with a VERIFY naming the store-level fix, on the `final_seen` model. `to_kafka_binary` had no counterpart to `from_kafka_binary`'s refusal of `__`-prefixed names. Unreachable through today's callers — the marker is set on a plain dict after `envelope_dict` — but "unreachable" is a property of the current callers, not of the codec, and the inbound filter is what the control rests on. Filtered on both sides now, with a test. Assisted-By: Claude (Anthropic AI) Signed-off-by: Mariusz Sabath --- eventing/eventbridge/kafka_group_mirror.py | 20 ++++++++++++++- eventing/eventbridge/kafka_in.py | 30 +++++++++++++--------- eventing/shared/ce.py | 23 +++++++++++++---- eventing/tests/test_kafka_in.py | 21 +++++++++++++++ 4 files changed, 76 insertions(+), 18 deletions(-) diff --git a/eventing/eventbridge/kafka_group_mirror.py b/eventing/eventbridge/kafka_group_mirror.py index 5e027666..989df300 100644 --- a/eventing/eventbridge/kafka_group_mirror.py +++ b/eventing/eventbridge/kafka_group_mirror.py @@ -80,10 +80,28 @@ def __init__(self, bootstrap: str, response_topic: str, groups, # an `emptyDir` — so refusing unsigned records would lose every group's state on # every restart, which is the exact failure this module was written to prevent. # - # `group.completed` is the only mirror path that can settle a batch: + # `group.completed` is the only mirror path that can SETTLE a batch: # `on_member_event(replay=True)` returns before `maybe_complete`, and a forged # `group.started` creates an empty group row, which is noise rather than a # settled batch reporting work that never ran. + # + # **What this does NOT cover**, stated because the guard reads as if it did: + # `on_member_event(replay=True)` runs `add_group_member` and + # `mark_member_finished` BEFORE its `replay` check, and the rejection marker is + # local rather than on the wire — so a forged terminal lying earlier in the topic + # than the genuine answer wins `finished_utc IS NULL` again on every start, and + # the batch reads back `failed: 1` for an agent that succeeded. Bounded: the + # batch stays open, because `complete_group` is guarded and `maybe_complete` is + # skipped. Worst exactly here, though — with a PVC the live run's `finished_utc` + # already blocks the replay, while on the `test`/`kind` overlays /data is an + # `emptyDir` and the replay IS the state. + # + # VERIFY: not fixable by widening this check. Verifying member events on replay + # needs the runner keyset with require=True, which refuses every pre-signing + # record, and "verify only when a signature is present" is not a control because + # a forger omits it. The fix is a replay that cannot regress a member already + # terminal in the store — a store change, not a mirror one. Keep the paragraph + # above when that lands; it is why the scope is what it is. self._keyset = keyset self._bridge_kid = bridge_kid # Partitions are assigned by hand and nothing is ever committed: the whole diff --git a/eventing/eventbridge/kafka_in.py b/eventing/eventbridge/kafka_in.py index 97bdaf94..26c613d9 100644 --- a/eventing/eventbridge/kafka_in.py +++ b/eventing/eventbridge/kafka_in.py @@ -27,13 +27,19 @@ reaches no acting path: not `on_group_event`, not `on_member_event`, and not a sequence a non-rejected row already holds. -**The guard reads `ok`, never `verified`, and that is load-bearing.** An unsigned -non-terminal frame and a deployment with no keyset both report `verified=False` while -being accepted *by policy* — `emit()` signs terminal events only, because a signature -costs ~222 ms and signing every stdout frame would add minutes to a chatty run. Keying -the guard on `verified` would stop every streamed frame of every genuine run from -acting, which is the §4.4 step 2 mistake: it rewrote every frame of every run to -`phase="error"` and was found against a live broker, by no unit test. +**The guard reads `ok`, never `verified`, and that is load-bearing.** `emit()` signs +terminal events only, because a signature costs ~222 ms and signing every stdout frame +would add minutes to a chatty run, so an unsigned non-terminal frame is accepted *by +policy*. `response_decision` reports those — and a deployment with no keyset — as +`verified=True`, acceptance by policy rather than a check that passed, which is what +keeps `accept and not verified` meaning the audit-mode population and nothing else. + +So the population to refuse is `not ok`. Keying the guard on `verified` would be wrong +in both directions: it would spare nothing extra, and it would make audit mode inert, +where the whole point is that events keep acting while the reject rate is watched. +Getting this backwards is the §4.4 step 2 mistake — verifying all-or-nothing rewrote +every streamed frame of every genuine run to `phase="error"`, and it was found against a +live broker, by no unit test. """ from __future__ import annotations @@ -210,11 +216,11 @@ def run(self) -> None: # but no `correlationid`, so handing it to insert_response would # violate that table's (correlationid, sequence) primary key. if ce.is_group_event(evt): - # #885 — `ok`, not `verified`. An unsigned non-terminal frame and - # a deployment with no keyset both report `verified=False` while - # being accepted BY POLICY, and skipping those would stop every - # streamed frame of every genuine run from reaching the group - # service. Only `not ok` is the enforcement-refused population. + # #885 — `ok`, not `verified`. Those are different questions: + # `verified` is False only for a check that FAILED, which audit + # mode accepts on purpose, while `not ok` is the + # enforcement-refused population and the only one to skip. See + # the module docstring. # # The group service refuses a marked envelope too (§885's single # rule), so this is the source-side half of a rule enforced at diff --git a/eventing/shared/ce.py b/eventing/shared/ce.py index 2c373ca8..cc178996 100644 --- a/eventing/shared/ce.py +++ b/eventing/shared/ce.py @@ -128,11 +128,13 @@ def is_rejected(event) -> bool: **a rejected envelope is evidence, not a fact — it is stored and it is rendered, but it never advances state.** - Note what this does NOT read. The verdict is `accept`, never `verified`: an unsigned - non-terminal frame and a deployment with no keyset both come back `verified=False` - while being accepted *by policy*, and treating those as rejected would stop every - streamed frame of every genuine run from acting. That is the §4.4 step 2 mistake, - which was found against a live broker and by no unit test. + Note what this does NOT read. The marker is set from `accept`, never from `verified`. + An unsigned non-terminal frame and a deployment with no keyset both come back + `verified=True` — acceptance *by policy* rather than a check that passed, which is + what keeps `accept and not verified` meaning exactly the audit-mode population. The + population to refuse is `not accept`, and nothing else: treating a failed check as + the signal would also catch audit mode, where the whole point is that events keep + acting while an operator watches the reject rate. """ return bool(event.get(KEY_REJECTED)) @@ -169,6 +171,17 @@ def to_kafka_binary(event: CloudEvent) -> tuple[list[tuple[str, bytes]], bytes]: for k, v in event.attrs.items(): if v is None or v == "": continue + # #885 — the outbound half of `from_kafka_binary`'s refusal. A `__`-prefixed name + # is a local verdict (see KEY_REJECTED), so emitting one would put a decision + # this process made onto the wire, where the next consumer cannot tell it from a + # claim the publisher is entitled to make. + # + # Not reachable today: the marker is set on a plain dict after `envelope_dict`, + # the acting paths consume dicts, and `maybe_complete` builds a fresh event + # rather than re-serialising one. Symmetric anyway, because "unreachable" is a + # property of today's callers and the inbound filter is what the control rests on. + if k.startswith("__"): + continue headers.append((f"{CE_HEADER_PREFIX}{k}", str(v).encode("utf-8"))) if event.data is None: value = b"" diff --git a/eventing/tests/test_kafka_in.py b/eventing/tests/test_kafka_in.py index 570cfbf4..e782c228 100644 --- a/eventing/tests/test_kafka_in.py +++ b/eventing/tests/test_kafka_in.py @@ -640,3 +640,24 @@ def test_the_rejection_marker_cannot_arrive_from_the_wire(tmp_path, monkeypatch) assert c.rejected == 0, "the wire must not be able to mark an event rejected" assert seen and not ce.is_rejected(seen[0]) assert store.events_for("brave-otter-4718")[0]["phase"] == "result" + + +def test_the_rejection_marker_is_never_emitted_onto_the_wire(tmp_path): + """The outbound half of the same invariant. + + `from_kafka_binary` refusing `__`-prefixed names is what makes the marker + unforgeable; `to_kafka_binary` refusing to emit them is what stops this process + publishing its own local verdict as though it were a claim the next consumer could + rely on. Unreachable through today's callers — the marker is set on a plain dict + after `envelope_dict` — which is exactly why it is pinned rather than argued. + """ + evt = ce.new_event(type=ce.TYPE_RESPONSE, source="s", + datacontenttype="application/json", + correlationid="brave-otter-4718", sequence=1, phase="result", + final="true", data={"text": "hi"}) + evt.attrs[ce.KEY_REJECTED] = "true" + headers, _ = ce.to_kafka_binary(evt) + assert not [k for k, _ in headers if "__" in k], \ + "a local verdict must not become a wire attribute" + assert not ce.is_rejected(ce.from_kafka_binary(*ce.to_kafka_binary(evt))), \ + "and must not survive a round trip"