Add an API to list changes to quarantine state of media

changelog.d/19558.feature

@@ -0,0 +1 @@
+Add a ["Listing quarantined media changes" Admin API](https://element-hq.github.io/synapse/latest/admin_api/media_admin_api.html#listing-quarantined-media-changes) for retrieving a paginated record of when media became (un)quarantined.

docker/configure_workers_and_start.py

@@ -119,7 +119,7 @@
     },
     "media_repository": {
         "app": "synapse.app.generic_worker",
-        "listener_resources": ["media", "client"],
+        "listener_resources": ["media", "client", "replication"],
         "endpoint_patterns": [
             "^/_matrix/media/",
             "^/_synapse/admin/v1/purge_media_cache$",

docs/admin_api/media_admin_api.md

@@ -247,6 +247,42 @@ Response:
 {}
 ```
 
+## Listing quarantined media changes
+
+When media is quarantined or unquarantined, a change record is created in the 
+database. This API returns those change records in the order they were created.
+
+**Note**: This API should be considered *best-effort* and expected to have missing or
+duplicate records. Currently, this only captures any media explicitly (un)quarantined by
+the media quarantine admin API, and the other cases are tracked by
+https://github.com/element-hq/synapse/issues/19672. Historical media uploaded before
+Synapse 1.152.0 is backfilled in a background update on a best-effort basis.
+
+Each page has a maximum of 100 records. The first page has the oldest records, 
+paginating forwards with each `next_batch` value.
+
+Request:
+
+```
+GET /_synapse/admin/v1/media/quarantine_changes?from=2
+```
+
+Where `from` is the `next_batch` value from a previous request. It is optional
+and defaults to the first page (the value `0`).
+
+Response:
+
+```json
+{
+  "next_batch": 4,
+  "changes": [
+    { "origin": "example.org", "media_id": "abcdefg12345...", "quarantined": true },
+    { "origin": "example.org", "media_id": "abcdefg12345...", "quarantined": false },
+    { "origin": "another.example.org", "media_id": "abcdefg12345...", "quarantined": true }
+  ]
+}
+```
+
 # Delete local media
 This API deletes the *local* media from the disk of your own server.
 This includes any local thumbnails and copies of media downloaded from

docs/workers.md

@@ -576,6 +576,14 @@ configured as stream writer for the `device_lists` stream:
     ^/_matrix/client/(api/v1|r0|v3|unstable)/keys/device_signing/upload$
     ^/_matrix/client/(api/v1|r0|v3|unstable)/keys/signatures/upload$
 
+##### The `quarantined_media_changes` stream
+
+The `quarantined_media_changes` stream supports multiple writers. The following endpoints
+can be handled by any worker, but should be routed directly to one of the workers
+configured as stream writer for the `quarantined_media_changes` stream:
+
+    ^/_synapse/admin/v1/quarantine_media/.*$
+
 #### Restrict outbound federation traffic to a specific set of workers
 
 The

synapse/_scripts/synapse_port_db.py

@@ -136,6 +136,7 @@
     "users": ["shadow_banned", "approved", "locked", "suspended"],
     "un_partial_stated_event_stream": ["rejection_status_changed"],
     "users_who_share_rooms": ["share_private"],
+    "quarantined_media_changes": ["quarantined"],
 }
 
 
@@ -912,6 +913,10 @@ def alter_table(txn: LoggingTransaction) -> None:
             await self._setup_autoincrement_sequence(
                 "state_groups_pending_deletion", "sequence_number"
             )
+            await self._setup_sequence(
+                "quarantined_media_id_seq",
+                [("quarantined_media_changes", "stream_id")],
+            )
 
             # Step 3. Get tables.
             self.progress.set_state("Fetching tables")

synapse/config/workers.py

@@ -142,6 +142,8 @@ class WriterLocations:
         push_rules: The instances that write to the push stream. Currently
             can only be a single instance.
         device_lists: The instances that write to the device list stream.
+        quarantined_media_changes: The instances that write to the quarantined media
+             changes stream.
     """
 
     events: list[str] = attr.ib(
@@ -180,6 +182,10 @@ class WriterLocations:
         default=["master"],
         converter=_instance_to_list_converter,
     )
+    quarantined_media_changes: list[str] = attr.ib(
+        default=[MAIN_PROCESS_INSTANCE_NAME],
+        converter=_instance_to_list_converter,
+    )
 
 
 @attr.s(auto_attribs=True)

synapse/replication/tcp/streams/__init__.py

@@ -39,6 +39,7 @@
     PresenceStream,
     PushersStream,
     PushRulesStream,
+    QuarantinedMediaStream,
     ReceiptsStream,
     StickyEventsStream,
     Stream,
@@ -73,6 +74,7 @@
         ThreadSubscriptionsStream,
         UnPartialStatedRoomStream,
         UnPartialStatedEventStream,
+        QuarantinedMediaStream,
     )
 }
 
@@ -96,4 +98,5 @@
     "ThreadSubscriptionsStream",
     "UnPartialStatedRoomStream",
     "UnPartialStatedEventStream",
+    "QuarantinedMediaStream",
 ]

synapse/replication/tcp/streams/_base.py

@@ -808,3 +808,50 @@ async def _update_function(
             return [], to_token, False
 
         return rows, rows[-1][0], len(updates) == limit
+
+
+@attr.s(slots=True, auto_attribs=True)
+class QuarantinedMediaStreamRow:
+    """Row for QuarantinedMediaStream"""
+
+    # We store the origin and media_id as media is scoped to the origin and are uniquely
+    # identified by (origin, media_id).
+
+    origin: str
+    media_id: str
+    quarantined: bool
+
+
+class QuarantinedMediaStream(_StreamFromIdGen):
+    """Stream to track changes to (un)quarantined media."""
+
+    NAME = "quarantined_media"
+    ROW_TYPE = QuarantinedMediaStreamRow
+
+    def __init__(self, hs: "HomeServer"):
+        self.store = hs.get_datastores().main
+        super().__init__(
+            hs.get_instance_name(),
+            self._update_function,
+            self.store._quarantined_media_changes_id_gen,
+        )
+
+    async def _update_function(
+        self, instance_name: str, from_token: Token, to_token: Token, limit: int
+    ) -> StreamUpdateResult:
+        updates = await self.store.get_quarantined_media_changes(
+            from_id=from_token, to_id=to_token, limit=limit
+        )
+        rows = [
+            (
+                update.stream_id,
+                # Args to `QuarantinedMediaStreamRow`
+                (update.origin, update.media_id, update.quarantined),
+            )
+            for update in updates
+        ]
+
+        if not rows:
+            return [], to_token, False
+
+        return rows, rows[-1][0], len(updates) == limit

synapse/rest/admin/media.py

@@ -230,6 +230,70 @@ async def on_POST(
         return HTTPStatus.OK, {}
 
 
+class ListQuarantineChanges(RestServlet):
+    """Lists the quarantine changes to media.
+
+    Uses the pagination format described by https://spec.matrix.org/v1.18/appendices/#pagination
+    """
+
+    PATTERNS = admin_patterns("/media/quarantine_changes$")
+
+    def __init__(self, hs: "HomeServer"):
+        self.store = hs.get_datastores().main
+        self.auth = hs.get_auth()
+        self.server_name = hs.hostname
+        self.replication = hs.get_replication_data_handler()
+
+    async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]:
+        await assert_requester_is_admin(self.auth, request)
+
+        from_id = parse_integer(request, "from", default=0)
+        limit = 100  # arbitrary; not enough to cause problems (hopefully)
+        to_id = await self.store.get_current_quarantined_media_stream_id()
+
+        if to_id < from_id:
+            # The caller is trying to get future data, which isn't possible.
+            raise SynapseError(
+                HTTPStatus.BAD_REQUEST,
+                "The `from` position is ahead of the currently persisted position.",
+                errcode=Codes.INVALID_PARAM,
+            )
+
+        # We need to wait to ensure that our current worker is actually caught up with
+        # the stream position, otherwise we might not return what we think we're returning.
+        if not await self.store.wait_for_quarantined_media_stream_id(from_id):
+            raise SynapseError(
+                HTTPStatus.INTERNAL_SERVER_ERROR,
+                "Timed out while waiting for the worker serving this request to catch up to the given "
+                "`from` stream position. Assuming this is a valid `from` token, this indicates an issue "
+                "with Synapse or the worker deployment lagging behind the replication stream. Please try "
+                "the request again later.",
+                errcode=Codes.UNKNOWN,
+            )
+
+        changes = await self.store.get_quarantined_media_changes(
+            from_id=from_id,
+            to_id=to_id,
+            limit=limit,
+        )
+
+        serialized_changes = [
+            {
+                "origin": c.origin if c.origin is not None else self.server_name,
+                "media_id": c.media_id,
+                "quarantined": c.quarantined,
+            }
+            for c in changes
+        ]
+
+        # We know the last record will have the highest stream ID, so use that one. If
+        # there aren't any records, just return the `to_id` value because it'll be the
+        # furthest stream position possible.
+        next_batch = changes[-1].stream_id if len(changes) > 0 else to_id
+
+        return HTTPStatus.OK, {"next_batch": next_batch, "changes": serialized_changes}
+
+
 class ProtectMediaByID(RestServlet):
     """Protect local media from being quarantined."""
 
@@ -529,6 +593,7 @@ def register_servlets_for_media_repo(hs: "HomeServer", http_server: HttpServer)
     QuarantineMediaByID(hs).register(http_server)
     UnquarantineMediaByID(hs).register(http_server)
     QuarantineMediaByUser(hs).register(http_server)
+    ListQuarantineChanges(hs).register(http_server)
     ProtectMediaByID(hs).register(http_server)
     UnprotectMediaByID(hs).register(http_server)
     ListMediaInRoom(hs).register(http_server)