From 7b8f0037ec88976c7bb089dd8396d87b713eeacf Mon Sep 17 00:00:00 2001 From: brainx <12695242+brainx@users.noreply.github.com> Date: Thu, 30 Jul 2026 20:54:36 +0200 Subject: [PATCH 1/2] Fix event-store Bandit baseline --- olymp/events.py | 58 +++++++++++++++++++-------------- tests/test_events.py | 77 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 25 deletions(-) diff --git a/olymp/events.py b/olymp/events.py index ee5a770..3ac8957 100644 --- a/olymp/events.py +++ b/olymp/events.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import logging import os import sqlite3 import threading @@ -19,6 +20,7 @@ MAX_EVENT_LIMIT = 1000 EventSubscriber = Callable[[dict[str, Any]], None] +_LOGGER = logging.getLogger(__name__) _SUBSCRIBERS: list[EventSubscriber] = [] _SUBSCRIBERS_LOCK = threading.Lock() @@ -122,33 +124,36 @@ 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") with closing(self._connect()) as db: rows = db.execute( - f""" + """ SELECT * FROM events - {where} + WHERE (? IS NULL OR 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 ? """, - tuple(values), + ( + checked_after, + checked_after, + checked_type, + checked_type, + checked_run, + checked_run, + checked_plan, + checked_plan, + checked_node, + checked_node, + checked_limit, + ), ).fetchall() return [_event_record(row) for row in rows] @@ -171,11 +176,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]: diff --git a/tests/test_events.py b/tests/test_events.py index e511c6a..361ddd4 100644 --- a/tests/test_events.py +++ b/tests/test_events.py @@ -49,6 +49,83 @@ 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 --") + + 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, []) + + 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" From 4b6f6a57077ec582eaf16750f76a7c0865d4ddba Mon Sep 17 00:00:00 2001 From: brainx <12695242+brainx@users.noreply.github.com> Date: Fri, 31 Jul 2026 21:56:20 +0200 Subject: [PATCH 2/2] Preserve event cursor rowid seek --- olymp/events.py | 65 ++++++++++++++++++++++++++------------------ tests/test_events.py | 11 ++++++++ 2 files changed, 50 insertions(+), 26 deletions(-) diff --git a/olymp/events.py b/olymp/events.py index 3ac8957..053fd6f 100644 --- a/olymp/events.py +++ b/olymp/events.py @@ -23,6 +23,27 @@ _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]: @@ -128,33 +149,25 @@ def list( 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( - """ - SELECT * - FROM events - WHERE (? IS NULL OR 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 ? - """, - ( - checked_after, - checked_after, - checked_type, - checked_type, - checked_run, - checked_run, - checked_plan, - checked_plan, - checked_node, - checked_node, - checked_limit, - ), - ).fetchall() + rows = db.execute(query, values).fetchall() return [_event_record(row) for row in rows] def _connect(self) -> sqlite3.Connection: diff --git a/tests/test_events.py b/tests/test_events.py index 361ddd4..14a3c57 100644 --- a/tests/test_events.py +++ b/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 @@ -86,10 +88,19 @@ def test_event_store_filters_and_paginates(self) -> None: 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: