Fix event-store Bandit baseline

olymp/events.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 import sqlite3
 import threading
@@ -19,8 +20,30 @@
 MAX_EVENT_LIMIT = 1000
 EventSubscriber = Callable[[dict[str, Any]], None]
 
+_LOGGER = logging.getLogger(__name__)
 _SUBSCRIBERS: list[EventSubscriber] = []
 _SUBSCRIBERS_LOCK = threading.Lock()
+_LIST_EVENTS_QUERY = """
+    SELECT *
+    FROM events
+    WHERE (? IS NULL OR event_type = ?)
+      AND (? IS NULL OR run_id = ?)
+      AND (? IS NULL OR plan_id = ?)
+      AND (? IS NULL OR node_id = ?)
+    ORDER BY event_id ASC
+    LIMIT ?
+"""
+_LIST_EVENTS_AFTER_QUERY = """
+    SELECT *
+    FROM events
+    WHERE event_id > ?
+      AND (? IS NULL OR event_type = ?)
+      AND (? IS NULL OR run_id = ?)
+      AND (? IS NULL OR plan_id = ?)
+      AND (? IS NULL OR node_id = ?)
+    ORDER BY event_id ASC
+    LIMIT ?
+"""
 
 
 def subscribe_events(handler: EventSubscriber) -> Callable[[], None]:
@@ -122,34 +145,29 @@ def list(
         self.init()
         checked_after = _optional_positive_int(after_event_id, "after_event_id")
         checked_limit = _event_limit(limit)
-        clauses: list[str] = []
-        values: list[object] = []
-        if checked_after is not None:
-            clauses.append("event_id > ?")
-            values.append(checked_after)
-        for column, value in (
-            ("event_type", event_type),
-            ("run_id", run_id),
-            ("plan_id", plan_id),
-            ("node_id", node_id),
-        ):
-            checked = _nullable_text(value, column)
-            if checked is not None:
-                clauses.append(f"{column} = ?")
-                values.append(checked)
-        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
-        values.append(checked_limit)
+        checked_type = _nullable_text(event_type, "event_type")
+        checked_run = _nullable_text(run_id, "run_id")
+        checked_plan = _nullable_text(plan_id, "plan_id")
+        checked_node = _nullable_text(node_id, "node_id")
+        filter_values: tuple[object, ...] = (
+            checked_type,
+            checked_type,
+            checked_run,
+            checked_run,
+            checked_plan,
+            checked_plan,
+            checked_node,
+            checked_node,
+            checked_limit,
+        )
+        if checked_after is None:
+            query = _LIST_EVENTS_QUERY
+            values = filter_values
+        else:
+            query = _LIST_EVENTS_AFTER_QUERY
+            values = (checked_after, *filter_values)
         with closing(self._connect()) as db:
-            rows = db.execute(
-                f"""
-                SELECT *
-                FROM events
-                {where}
-                ORDER BY event_id ASC
-                LIMIT ?
-                """,
-                tuple(values),
-            ).fetchall()
+            rows = db.execute(query, values).fetchall()
         return [_event_record(row) for row in rows]
 
     def _connect(self) -> sqlite3.Connection:
@@ -171,11 +189,14 @@ def _notify_subscribers(event: dict[str, Any]) -> None:
     with _SUBSCRIBERS_LOCK:
         subscribers = tuple(_SUBSCRIBERS)
     for subscriber in subscribers:
-        try:
-            subscriber(dict(event))
-        except Exception:
-            # Plugin hooks must not change persisted control-plane behavior.
-            continue
+        _notify_subscriber(subscriber, event)
+
+
+def _notify_subscriber(subscriber: EventSubscriber, event: dict[str, Any]) -> None:
+    try:
+        subscriber(dict(event))
+    except Exception:
+        _LOGGER.warning("event subscriber failed after event persistence", exc_info=False)
 
 
 def _event_record(row: sqlite3.Row | None) -> dict[str, Any]:

tests/test_events.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
 import json
+import sqlite3
 import tempfile
 import unittest
 from pathlib import Path
 
+from olymp import events as events_module
 from olymp.events import EventStore, subscribe_events
 from olymp.models import ZeusNode
 from olymp.plans import lifecycle_plan
@@ -49,6 +51,92 @@ def test_event_store_persists_redacts_filters_and_notifies_plugins(self) -> None
             after = EventStore(db_path).list(after_event_id=1)
             self.assertEqual([event["event_type"] for event in after], ["plugin.other"])
 
+    def test_event_store_filters_and_paginates(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            db_path = Path(temp) / "olymp.db"
+            store = EventStore(db_path)
+            matching_ids: list[int] = []
+            for event_type, run_id, plan_id, node_id in (
+                ("plugin.test", "run-a", "plan-a", "node-a"),
+                ("plugin.other", "run-a", "plan-a", "node-a"),
+                ("plugin.test", "run-b", "plan-a", "node-a"),
+                ("plugin.test", "run-a", "plan-a", "node-a"),
+            ):
+                event = store.append(
+                    event_type,
+                    source="test",
+                    run_id=run_id,
+                    plan_id=plan_id,
+                    node_id=node_id,
+                )
+                if event_type == "plugin.test" and run_id == "run-a":
+                    matching_ids.append(int(event["event_id"]))
+
+            first_page = store.list(
+                event_type="plugin.test",
+                run_id="run-a",
+                plan_id="plan-a",
+                node_id="node-a",
+                limit=1,
+            )
+            second_page = store.list(
+                after_event_id=int(first_page[0]["event_id"]),
+                event_type="plugin.test",
+                run_id="run-a",
+                plan_id="plan-a",
+                node_id="node-a",
+                limit=1,
+            )
+            injected_filter = store.list(event_type="plugin.test' OR 1=1 --")
+            with sqlite3.connect(db_path) as db:
+                query_plan = db.execute(
+                    "EXPLAIN QUERY PLAN " + events_module._LIST_EVENTS_AFTER_QUERY,
+                    (1, None, None, None, None, None, None, None, None, 100),
+                ).fetchall()
+
+        self.assertEqual([event["event_id"] for event in first_page], matching_ids[:1])
+        self.assertEqual([event["event_id"] for event in second_page], matching_ids[1:])
+        self.assertEqual(injected_filter, [])
+        self.assertTrue(
+            any("SEARCH events USING INTEGER PRIMARY KEY" in str(row[3]) for row in query_plan),
+            query_plan,
+        )
+
+    def test_subscriber_failure_is_isolated_and_secret_safe(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            db_path = Path(temp) / "olymp.db"
+            store = EventStore(db_path)
+            observed: list[dict[str, object]] = []
+            sensitive_failure = "subscriber-secret-detail"
+
+            def failing_subscriber(event: dict[str, object]) -> None:
+                event["event_type"] = "tampered"
+                raise RuntimeError(sensitive_failure)
+
+            unsubscribe_failing = subscribe_events(failing_subscriber)
+            unsubscribe_observer = subscribe_events(lambda event: observed.append(event))
+            try:
+                with self.assertLogs("olymp.events", level="WARNING") as captured:
+                    created = store.append(
+                        "plugin.test",
+                        source="test",
+                        payload={"status": "ok"},
+                    )
+            finally:
+                unsubscribe_observer()
+                unsubscribe_failing()
+
+            persisted = store.list()
+
+        self.assertEqual(created["event_type"], "plugin.test")
+        self.assertEqual(observed, [created])
+        self.assertEqual(persisted, [created])
+        self.assertEqual(
+            captured.output,
+            ["WARNING:olymp.events:event subscriber failed after event persistence"],
+        )
+        self.assertNotIn(sensitive_failure, "\n".join(captured.output))
+
     def test_plan_run_and_mutation_methods_emit_timeline_events(self) -> None:
         with tempfile.TemporaryDirectory() as temp:
             db_path = Path(temp) / "olymp.db"