fix(eventing): a rejected response must not act, only be recorded

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

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,

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:

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,42 @@ 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.
+        #
+        # **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
         # 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 +113,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 +259,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:

eventing/eventbridge/kafka_in.py

@@ -16,6 +16,30 @@
 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.** `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
 
@@ -74,13 +98,25 @@ def __init__(
         self._bridge_kid = bridge_kid
         self._rejected = 0
         self._audit_failed = 0
+        self._rejected_not_stored = 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 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 +194,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 +211,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`. 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
+                        # 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 +237,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: