Merge remote-tracking branch 'origin/develop' into develop

Author: weblate

securedrop/journalist_app/api2/events.py

@@ -1,6 +1,7 @@
 from dataclasses import asdict
 
 from db import db
+from flask import current_app
 from journalist_app import utils
 from journalist_app.api2.shared import json_version, mark_source_deleted, save_reply
 from journalist_app.api2.types import (
@@ -35,13 +36,17 @@ class EventHandler:
        `journalist_api2.types.EVENT_DATA_TYPES`;
 
     3. define the handler as a static method `handle_thing_done(event: Event)`
-       in this class
+       in this class; and
 
     4. explicitly register `{"thing_done": self.handle_thing_done}` inside
       `EventHandler.process()`.
 
     This is belt-and-suspenders for ensuring that only the intended methods are
     exposed as callable event handlers.
+
+    To preserve transaction separation between events, handlers MUST return with
+    a clean SQLAlchemy session: in other words, having either successfully
+    committed or rolled back all of their changes.
     """
 
     def __init__(self, session: Session, redis: Redis) -> None:
@@ -59,7 +64,7 @@ def process(self, event: Event, minor: int) -> EventResult:
         """The per-event entry-point for handling a single event."""
 
         try:
-            if self.has_progress(event):
+            if self.is_duplicate(event):
                 return EventResult(
                     event_id=event.id,
                     status=(EventStatusCode.AlreadyReported, None),
@@ -83,35 +88,45 @@ def process(self, event: Event, minor: int) -> EventResult:
                 ),
             )
 
-        self.mark_progress(event)  # prevent races
-        result = handler(event, minor)
-        self.mark_progress(event, result.status[0])  # enforce idempotence
+        try:
+            result = handler(event, minor)
+
+            # Enforce "handlers MUST return with a clean SQLAlchemy session" above:
+            if db.session.dirty or db.session.new or db.session.deleted:
+                raise RuntimeError(f"{handler} returned with a pending database transaction")
+
+        # Catch anything not handled by the handler:
+        except Exception:
+            current_app.logger.error(f"unhandled exception in handler for {event}", exc_info=True)
+            db.session.rollback()
+            result = EventResult(
+                event.id, (EventStatusCode.InternalServerError, "failed to process event")
+            )
+
+        self.record_status(event, result.status[0])
         return result
 
     def idempotence_key(self, event: Event) -> str:
         return f"{REDIS_EVENT_PREFIX}/{self._session.user.uuid}/{event.id}"
 
-    def has_progress(self, event: Event) -> EventStatusCode:
-        return self._redis.get(self.idempotence_key(event))
-
-    def mark_progress(
-        self, event: Event, status: EventStatusCode = EventStatusCode.Processing
-    ) -> None:
-        """
-        If `status` is a non-error code, mark it as the progress of `event`, to
-        be returned later as "Already Reported".
-
-        If `status` is an error code, clear it, since `event` MAY be resubmitted
-        later.
-        """
-        if status >= EventStatusCode.BadRequest:
-            self._redis.delete(self.idempotence_key(event))
-        else:
+    def is_duplicate(self, event: Event) -> bool:
+        """Returns `True` if this event is already registered (i.e., a replay)."""
+        return (
             self._redis.set(
                 self.idempotence_key(event),
-                status,
+                EventStatusCode.Processing,
                 ex=IDEMPOTENCE_PERIOD,
+                nx=True,
             )
+            is None
+        )
+
+    def record_status(self, event: Event, status: EventStatusCode) -> None:
+        """Record the event's final status for idempotence, or clear on error to permit retry."""
+        if status >= EventStatusCode.BadRequest:
+            self._redis.delete(self.idempotence_key(event))
+        else:
+            self._redis.set(self.idempotence_key(event), status, ex=IDEMPOTENCE_PERIOD)
 
     @staticmethod
     def handle_item_deleted(event: Event, minor: int) -> EventResult:

securedrop/tests/test_journalist_api2.py

@@ -1,30 +1,41 @@
+import threading
 import uuid
 from contextlib import contextmanager
 from copy import deepcopy
 from dataclasses import asdict
 from datetime import UTC, datetime
+from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
 import journalist_app as journalist_app_module
 import pytest
 from flask import url_for
 from flask_sqlalchemy import get_debug_queries
 from journalist_app import api2
+from journalist_app.api2.events import EventHandler
 from journalist_app.api2.shared import json_version
 from journalist_app.api2.types import (
     VERSION_LEN,
     Event,
+    EventResult,
+    EventStatusCode,
     EventType,
     ItemTarget,
     SourceTarget,
 )
 from models import Reply, Source, SourceStar, Submission, db
+from redis import Redis
 from sqlalchemy.orm.exc import MultipleResultsFound
 from tests.utils import ascii_armor, decrypt_as_journalist
 from tests.utils.api_helper import get_api_headers
 from tests.utils.db_helper import init_source, submit
 
 
+@pytest.fixture
+def redis(config):
+    return Redis(decode_responses=True, **config.REDIS_KWARGS)
+
+
 def filtered_queries():
     # filter out PRAGMA, instance_config loading, etc.
     return [
@@ -832,6 +843,57 @@ def test_api2_idempotence_period(journalist_app):
     assert journalist_app.config["SESSION_LIFETIME"] <= api2.events.IDEMPOTENCE_PERIOD
 
 
+def test_api2_atomic_idempotence(redis):
+    """
+    Check that identical events received simultaneously are still only accepted
+    once.
+    """
+
+    session = MagicMock()
+    session.user.uuid = str(uuid4())
+
+    event = MagicMock()
+    event.id = str(uuid4())
+    event.type = EventType.SOURCE_STARRED
+
+    barrier = threading.Barrier(2)
+    handler_call_count = [0]
+    original_set = redis.set
+
+    def slow_set(*args, **kwargs):
+        if kwargs.get("nx"):
+            # Widen the possible TOCTOU window: hold both threads at the SETNX
+            # gate until both are about to set the not-yet-existing key, then
+            # release together so atomicity is what enforces the invariant.
+            barrier.wait(timeout=5)
+        return original_set(*args, **kwargs)
+
+    def counting_handler(ev, minor):
+        handler_call_count[0] += 1
+        return EventResult(event_id=ev.id, status=(EventStatusCode.OK, None))
+
+    with (
+        patch.object(redis, "set", slow_set),
+        patch.object(EventHandler, "handle_source_starred", staticmethod(counting_handler)),
+    ):
+        threads = [
+            threading.Thread(
+                target=EventHandler(session=session, redis=redis).process,
+                args=(event, 1),
+            )
+            for _ in range(2)
+        ]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=10)
+
+    # Should be 1: a correct implementation relies on atomic SETNX to let only
+    # one thread past the check. Fails with count == 2 if mark_progress() drops
+    # the NX flag or claim_progress() reverts to a non-atomic check-then-act.
+    assert handler_call_count[0] == 1
+
+
 def test_api2_event_ordering(journalist_app, journalist_api_token, test_files):
     """
     If two `item_deleted` events for the same item arrive out of order, the