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.

docs/admin_api/media_admin_api.md

@@ -247,6 +247,32 @@ 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.
+
+Request:
+
+```
+GET /_synapse/admin/v1/media/quarantine_changes?from=2
+```
+
+Where `from` is the `next_batch` value from a previous request. It is optional.
+
+Response:
+
+```json
+{
+  "next_batch": 4,
+  "rows": [
+    { "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

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"],
 }
 
 

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,47 @@ 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"""
+
+    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: int, to_token: int, 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,46 @@ async def on_POST(
         return HTTPStatus.OK, {}
 
 
+class ListQuarantineChanges(RestServlet):
+    """Lists the quarantine changes to media."""
+
+    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
+
+    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 = (
+            from_id + limit
+        )  # somewhat implied, but makes our call to the store easier
+
+        changes = await self.store.get_quarantined_media_changes(
+            from_id=from_id,
+            to_id=to_id,
+            limit=limit,
+        )
+
+        rows = [
+            {
+                "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
+        ]
+
+        # `from` is exclusive, so don't +1
+        next_batch = max(c.stream_id for c in changes) if changes else from_id
+
+        return HTTPStatus.OK, {"next_batch": next_batch, "rows": rows}
+
+
 class ProtectMediaByID(RestServlet):
     """Protect local media from being quarantined."""
 
@@ -529,6 +569,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)