diff --git a/changelog.d/19558.feature b/changelog.d/19558.feature new file mode 100644 index 00000000000..f4c8f042bf0 --- /dev/null +++ b/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. \ No newline at end of file diff --git a/docker/configure_workers_and_start.py b/docker/configure_workers_and_start.py index 1b8d4f9989e..26c8556eff4 100755 --- a/docker/configure_workers_and_start.py +++ b/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$", diff --git a/docs/admin_api/media_admin_api.md b/docs/admin_api/media_admin_api.md index 6b96eb33564..750be85bbe9 100644 --- a/docs/admin_api/media_admin_api.md +++ b/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 diff --git a/docs/workers.md b/docs/workers.md index 201ef8f854d..8d3aad19c66 100644 --- a/docs/workers.md +++ b/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 diff --git a/synapse/_scripts/synapse_port_db.py b/synapse/_scripts/synapse_port_db.py index eedceb170e9..0b8a289d92d 100755 --- a/synapse/_scripts/synapse_port_db.py +++ b/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") diff --git a/synapse/config/workers.py b/synapse/config/workers.py index 996be88cb26..fb7378bfc81 100644 --- a/synapse/config/workers.py +++ b/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) diff --git a/synapse/replication/tcp/streams/__init__.py b/synapse/replication/tcp/streams/__init__.py index 067847617fa..e41573cf689 100644 --- a/synapse/replication/tcp/streams/__init__.py +++ b/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", ] diff --git a/synapse/replication/tcp/streams/_base.py b/synapse/replication/tcp/streams/_base.py index 1ea6b4fa857..a73f767add2 100644 --- a/synapse/replication/tcp/streams/_base.py +++ b/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 diff --git a/synapse/rest/admin/media.py b/synapse/rest/admin/media.py index d5346fe0d5c..5539b01af01 100644 --- a/synapse/rest/admin/media.py +++ b/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) diff --git a/synapse/storage/databases/main/room.py b/synapse/storage/databases/main/room.py index 7ac88e4c2aa..0a9e69e1bf5 100644 --- a/synapse/storage/databases/main/room.py +++ b/synapse/storage/databases/main/room.py @@ -21,6 +21,7 @@ # import logging +from dataclasses import dataclass from enum import Enum from typing import ( TYPE_CHECKING, @@ -44,6 +45,7 @@ from synapse.api.room_versions import RoomVersion, RoomVersions from synapse.config.homeserver import HomeServerConfig from synapse.events import EventBase +from synapse.replication.tcp.streams._base import QuarantinedMediaStream from synapse.replication.tcp.streams.partial_state import UnPartialStatedRoomStream from synapse.storage._base import ( db_to_json, @@ -58,8 +60,14 @@ from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore from synapse.storage.types import Cursor from synapse.storage.util.id_generators import IdGenerator, MultiWriterIdGenerator -from synapse.types import JsonDict, RetentionPolicy, StrCollection, ThirdPartyInstanceID +from synapse.types import ( + JsonDict, + RetentionPolicy, + StrCollection, + ThirdPartyInstanceID, +) from synapse.util.caches.descriptors import cached, cachedList +from synapse.util.duration import Duration from synapse.util.json import json_encoder from synapse.util.stringutils import MXC_REGEX @@ -102,6 +110,14 @@ class RoomStats(LargestRoomStats): public: bool +@dataclass(frozen=True) +class QuarantinedMediaUpdate: + stream_id: int # for the quarantined_media_changes stream + origin: str + media_id: str + quarantined: bool + + class RoomSortOrder(Enum): """ Enum to define the sorting method used when returning rooms with get_rooms_paginate @@ -162,11 +178,169 @@ def __init__( writers=["master"], ) + self._can_write_quarantined_media_changes = ( + self._instance_name in hs.config.worker.writers.quarantined_media_changes + ) + + self._quarantined_media_changes_id_gen: MultiWriterIdGenerator = ( + MultiWriterIdGenerator( + db_conn=db_conn, + db=database, + notifier=hs.get_replication_notifier(), + stream_name=QuarantinedMediaStream.NAME, + server_name=self.server_name, + instance_name=self._instance_name, + tables=[("quarantined_media_changes", "instance_name", "stream_id")], + sequence_name="quarantined_media_id_seq", + writers=hs.config.worker.writers.quarantined_media_changes, + ) + ) + + # Register a background update to flag already-quarantined media in the quarantine + # media changes table. This is to populate the API endpoint which consumes the + # table with initial data that callers expect (namely, a list of currently + # quarantined media). + self.db_pool.updates.register_background_update_handler( + _BackgroundUpdates.FLAG_EXISTING_QUARANTINED_MEDIA, + self._flag_existing_quarantined_media, + ) + + async def _flag_existing_quarantined_media( + self, progress: JsonDict, batch_size: int + ) -> int: + """Background update function to flag existing already-quarantined media in + the new `quarantine_media_changes` table. + + This only flags quarantined media as the API which reads the table is only + concerned with *changes* to the quarantined state - media does not start as + quarantined, so if it's already quarantined then it has changed state at + some point. Media which isn't quarantined has not changed state (as far as + this function can tell). + + We don't know when the media was originally quarantined, so this inserts as if + the media was quarantined now and are processed in an arbitrary order. + + Further, due to lack of timestamp or history, media which was quarantined then + unquarantined will not be picked up by this background task. + + Function signature is as per `register_background_update_handler` requirements. + + Args: + progress: The progress dictionary from the background update. + batch_size: The number of rows to process in each batch. + + Returns: + The number of rows inserted. + """ + # Note: we actually process 2x the batch_size per batch because we use it twice: + # once for the local_media_repository table, and once for the remote_media_cache + # table. This is fine though - we're still doing the work in batches. + + # We track the progress for the local and remote tables separately just in case + # there are media ID collisions that would make a mess of `ORDER BY media_id + # LIMIT ?`. If there are collisions towards the end of the returned set, the LIMIT + # might cut off some rows and make a future call with `WHERE media_id > ?` miss + # them too. By tracking each table separately, we avoid this kind of issue. + last_local_media_id = progress.get("last_local_media_id", "") + last_remote_media_id = progress.get("last_remote_media_id", "") + last_remote_origin = progress.get("last_remote_origin", "") + + # The `ORDER BY` here would normally miss records if the admin (un)quarantined a + # record, but that doesn't affect the background update because we also insert + # into the stream table upon quarantine status changing. Worst case is the admin + # newly quarantines some media, adding a row to the stream table, then we run + # over it again in the background update, adding a second row. Duplicate rows are + # non-issues for us. + # + # Another similar issue is if Synapse is downgraded partway through the background + # update then has more media quarantined. If Synapse is later upgraded again, the + # media that was quarantined while downgraded will only be imported if the media + # IDs are ordered higher than the last processed media ID. Media that has a lower + # ID will be skipped by the background update. + # + # Note: Already-quarantined media is indicated by the `quarantined_by` field being + # non-null. We only want quarantined media, per docstring above. + # + # This background update is considered *best effort* for the reasons above. Data + # might be missing or duplicated, and that's just going to have to be okay. This + # is further reinforced by not all changes being captured by the table anyway. + # See https://github.com/element-hq/synapse/issues/19672 for more details. + def flag_quarantined(txn: LoggingTransaction) -> int: + # It doesn't matter which order we do these in, as long as we do both of them. + txn.execute( + """ + SELECT NULL AS media_origin, media_id + FROM local_media_repository + WHERE quarantined_by IS NOT NULL + AND media_id > ? + ORDER BY media_id + LIMIT ? + """, + (last_local_media_id, batch_size), + ) + local_media_result = cast(list[tuple[str | None, str]], txn.fetchall()) + if len(local_media_result) > 0: + self._insert_quarantine_changes_txn(txn, local_media_result, True) + + # We use a >= ? on the media origin to avoid missing records when media IDs + # collide between origins (the table's unique constraint is on `(media_origin, media_id)`). + # Filtering by `(media_origin, media_id)` also makes sure we're using an index. + txn.execute( + """ + SELECT media_origin, media_id + FROM remote_media_cache + WHERE quarantined_by IS NOT NULL + AND media_origin >= ? AND media_id > ? + ORDER BY media_origin, media_id + LIMIT ? + """, + (last_remote_origin, last_remote_media_id, batch_size), + ) + remote_media_result = cast(list[tuple[str | None, str]], txn.fetchall()) + if len(remote_media_result) > 0: + self._insert_quarantine_changes_txn(txn, remote_media_result, True) + + self.db_pool.updates._background_update_progress_txn( + txn, + _BackgroundUpdates.FLAG_EXISTING_QUARANTINED_MEDIA, + { + "last_local_media_id": local_media_result[-1][1] + if len(local_media_result) > 0 + else last_local_media_id, + "last_remote_media_id": remote_media_result[-1][1] + if len(remote_media_result) > 0 + else last_remote_media_id, + "last_remote_origin": remote_media_result[-1][0] + if len(remote_media_result) > 0 + else last_remote_origin, + }, + ) + + return len(local_media_result) + len(remote_media_result) + + logger.info( + "Flagging existing quarantined media with local offset %s and remote offset %s/%s", + last_local_media_id, + last_remote_origin, + last_remote_media_id, + ) + num_flagged = await self.db_pool.runInteraction( + "_flag_existing_quarantined_media.flag_quarantined", + flag_quarantined, + ) + if num_flagged <= 0: # probably never negative, but why trust computers? + await self.db_pool.updates._end_background_update( + _BackgroundUpdates.FLAG_EXISTING_QUARANTINED_MEDIA + ) + return num_flagged + def process_replication_position( self, stream_name: str, instance_name: str, token: int ) -> None: if stream_name == UnPartialStatedRoomStream.NAME: self._un_partial_stated_rooms_stream_id_gen.advance(instance_name, token) + elif stream_name == QuarantinedMediaStream.NAME: + self._quarantined_media_changes_id_gen.advance(instance_name, token) return super().process_replication_position(stream_name, instance_name, token) async def store_room( @@ -1128,6 +1302,180 @@ def _get_media_ids_by_user_txn( return local_media_ids + async def get_current_quarantined_media_stream_id(self) -> int: + """Gets the position of the quarantined media changes stream. + + Returns: + int - the current stream ID + """ + return self._quarantined_media_changes_id_gen.get_current_token() + + async def wait_for_quarantined_media_stream_id(self, target_id: int) -> bool: + """Waits until the quarantined media changes stream reaches the given stream ID. + + See https://github.com/element-hq/synapse/pull/19644 for more details. + + TODO: Replace function and call sites with https://github.com/element-hq/synapse/pull/19644 + + Args: + target_id: The stream ID to wait for. + + Returns: + True when caught up to the target stream ID. + False when timing out while waiting. + """ + # We ideally would use something like `wait_for_stream_position` in the meantime, + # but that short circuits if the instance name matches the current instance name. + # Doing so means that if *another* writer is actually leading the to_id, then we'll + # assume that we're caught up when we aren't. + # + # NOTE: Because this is implemented to wait for stream positions by integer ID, + # we're technically waiting for *all* workers to catch up rather than just waiting + # for *our* worker to catch up. This is okay for now because the quarantined media + # stream should be pretty fast to update, and if it's not then the only thing we're + # affecting is an admin API that probably has a tool automatically retrying requests + # anyway. https://github.com/element-hq/synapse/pull/19644 does the waiting properly + # so this should be replaced by that (or similar). + + # Get the minimum shared position/ID across all workers + current_id = self._quarantined_media_changes_id_gen.get_current_token() + if current_id >= target_id: + return True # nothing to wait for: we're already caught up. + + # "This should never happen". Tokens we hand out via the API should exist. If they + # don't, then we're in a bad state and need to explode. + max_persisted_position = ( + await self._quarantined_media_changes_id_gen.get_max_allocated_token() + ) + assert max_persisted_position >= target_id, ( + f"Unable to wait for invalid future token (token={target_id} has positions " + f"ahead of our max persisted position={max_persisted_position})" + ) + + # Start waiting until we've caught up to the `stream_token` + start = self.clock.time_msec() + logged = False + while True: + # Like above, get the minimum shared ID across all workers + current_id = self._quarantined_media_changes_id_gen.get_current_token() + if current_id >= target_id: + return True + + now = self.clock.time_msec() + + # Timed out + if now - start > 10_000: + return False + + if not logged: + logger.info( + "Waiting for current token to reach %s; currently at %s", + target_id, + current_id, + ) + logged = True + + # TODO: be better + await self.clock.sleep(Duration(milliseconds=500)) + + async def get_quarantined_media_changes( + self, *, from_id: int, to_id: int, limit: int + ) -> list[QuarantinedMediaUpdate]: + """ + Get updates to quarantined media in stream ordering since `from_id`. + + Paginating forwards: from_id < x <= to_id, (ascending order) + + Args: + from_id: The starting stream ID (exclusive) + to_id: The ending stream ID (inclusive) + limit: The maximum number of rows to return + + Returns: + List of `QuarantinedMediaUpdate` update rows in stream ordering (ascending order). + + Raises: + SynapseError: If waiting for `to_id` took too long. + """ + if to_id < from_id: + # the to_id is behind the from_id, which means no results + return [] + + return await self.db_pool.runInteraction( + "get_quarantined_media_changes", + self._get_quarantined_media_changes_txn, + from_id, + to_id, + limit, + ) + + def _get_quarantined_media_changes_txn( + self, txn: LoggingTransaction, from_id: int, to_id: int, limit: int + ) -> list[QuarantinedMediaUpdate]: + txn.execute( + """ + SELECT stream_id, origin, media_id, quarantined + FROM quarantined_media_changes + WHERE ? < stream_id AND stream_id <= ? + ORDER BY stream_id ASC + LIMIT ? + """, + (from_id, to_id, limit), + ) + return [ + QuarantinedMediaUpdate( + stream_id=stream_id, + origin=origin, + media_id=media_id, + quarantined=quarantined, + ) + for stream_id, origin, media_id, quarantined in txn + ] + + def _insert_quarantine_changes_txn( + self, + txn: LoggingTransaction, + origins_and_media_ids: list[tuple[str | None, str]], + quarantined: bool, + ) -> None: + """Records media being (un)quarantined in the stream. + + Args: + txn (cursor) + origins_and_media_ids: The [origin, media_id] tuples to record. The origin + may be None if the media is local. + quarantined: Whether the media is being quarantined or unquarantined. + """ + assert self._can_write_quarantined_media_changes + medias_with_stream_ids = zip( + origins_and_media_ids, + self._quarantined_media_changes_id_gen.get_next_mult_txn( + txn, len(origins_and_media_ids) + ), + strict=True, + ) + self.db_pool.simple_insert_many_txn( + txn, + "quarantined_media_changes", + keys=( + "instance_name", + "stream_id", + "origin", + "media_id", + "quarantined", + ), + values=[ + ( + self._instance_name, + stream_id, + origin, + media_id, + quarantined, + ) + for (origin, media_id), stream_id in medias_with_stream_ids + ], + ) + def _quarantine_local_media_txn( self, txn: LoggingTransaction, @@ -1161,9 +1509,23 @@ def _quarantine_local_media_txn( if quarantined_by is not None: sql += " AND safe_from_quarantine = FALSE" + sql += " RETURNING media_id" + txn.execute(sql, [quarantined_by] + sql_many_clause_args) - # Note that a rowcount of -1 can be used to indicate no rows were affected. - total_media_quarantined += txn.rowcount if txn.rowcount > 0 else 0 + media_ids_affected = txn.fetchall() + # We use `len(media_ids_affected)` here and below because both queries may + # affect fewer or more rows than the `media_ids` input. For example, if the + # media_ids point to already-quarantined media, then nothing was updated. + # Similarly, the below query might find more media than the `media_ids` + # because it's searching for hashes instead. + total_media_quarantined += len(media_ids_affected) + if len(media_ids_affected) > 0: + # Flag media that was newly (un)quarantined in the changes table. + self._insert_quarantine_changes_txn( + txn, + [(None, media_id) for (media_id,) in media_ids_affected], + quarantined_by is not None, + ) # Update any media that was identified via hash. if hashes: @@ -1178,8 +1540,17 @@ def _quarantine_local_media_txn( if quarantined_by is not None: sql += " AND safe_from_quarantine = FALSE" + sql += " RETURNING media_id" + txn.execute(sql, [quarantined_by] + sql_many_clause_args) - total_media_quarantined += txn.rowcount if txn.rowcount > 0 else 0 + media_ids_affected = txn.fetchall() + total_media_quarantined += len(media_ids_affected) + if len(media_ids_affected) > 0: + self._insert_quarantine_changes_txn( + txn, + [(None, media_id) for (media_id,) in media_ids_affected], + quarantined_by is not None, + ) return total_media_quarantined @@ -1212,10 +1583,18 @@ def _quarantine_remote_media_txn( sql = f""" UPDATE remote_media_cache SET quarantined_by = ? - WHERE {sql_in_list_clause}""" + WHERE {sql_in_list_clause} + RETURNING media_origin, media_id""" txn.execute(sql, [quarantined_by] + sql_args) - total_media_quarantined += txn.rowcount if txn.rowcount > 0 else 0 + media_ids_affected = cast(list[tuple[str | None, str]], txn.fetchall()) + total_media_quarantined += len(media_ids_affected) + if len(media_ids_affected) > 0: + self._insert_quarantine_changes_txn( + txn, + media_ids_affected, + quarantined_by is not None, + ) if hashes: sql_many_clause_sql, sql_many_clause_args = make_in_list_sql_clause( @@ -1224,9 +1603,17 @@ def _quarantine_remote_media_txn( sql = f""" UPDATE remote_media_cache SET quarantined_by = ? - WHERE {sql_many_clause_sql}""" + WHERE {sql_many_clause_sql} + RETURNING media_origin, media_id""" txn.execute(sql, [quarantined_by] + sql_many_clause_args) - total_media_quarantined += txn.rowcount if txn.rowcount > 0 else 0 + media_ids_affected = cast(list[tuple[str | None, str]], txn.fetchall()) + total_media_quarantined += len(media_ids_affected) + if len(media_ids_affected) > 0: + self._insert_quarantine_changes_txn( + txn, + media_ids_affected, + quarantined_by is not None, + ) return total_media_quarantined @@ -2001,6 +2388,7 @@ class _BackgroundUpdates: REPLACE_ROOM_DEPTH_MIN_DEPTH = "replace_room_depth_min_depth" POPULATE_ROOMS_CREATOR_COLUMN = "populate_rooms_creator_column" ADD_ROOM_TYPE_COLUMN = "add_room_type_column" + FLAG_EXISTING_QUARANTINED_MEDIA = "flag_existing_quarantined_media" _REPLACE_ROOM_DEPTH_SQL_COMMANDS = ( diff --git a/synapse/storage/schema/main/delta/94/03_quarantined_media_tracking.sql b/synapse/storage/schema/main/delta/94/03_quarantined_media_tracking.sql new file mode 100644 index 00000000000..0eb74d85d49 --- /dev/null +++ b/synapse/storage/schema/main/delta/94/03_quarantined_media_tracking.sql @@ -0,0 +1,46 @@ +-- +-- This file is licensed under the Affero General Public License (AGPL) version 3. +-- +-- Copyright (C) 2026 Element Creations Ltd +-- +-- This program is free software: you can redistribute it and/or modify +-- it under the terms of the GNU Affero General Public License as +-- published by the Free Software Foundation, either version 3 of the +-- License, or (at your option) any later version. +-- +-- See the GNU Affero General Public License for more details: +-- . + +-- Represents a stream of when media is quarantined and unquarantined. Note that it's possible for duplicate rows to +-- exist in this table. When the background update which backfills this table is running, it orders quarantined media +-- by `media_id`. This table is also populated when an admin newly quarantines a piece of media. If the admin happens +-- to newly quarantine media that hasn't yet been processed by the background update, both the admin's API call and the +-- background update will add a row to this table. If the background update has already progressed past the media's ID, +-- then only the admin's API call will add a row to this table. +-- +-- Note also that this table might not be inserted with all possible cases of media being quarantined. For example, if +-- media is quarantined by hash upon upload or URL preview, it might not show up here. See https://github.com/element-hq/synapse/issues/19672 +-- for more details. +-- +-- Overall, this table is very much intended to be *best effort* for media that was quarantined before the table existed. +CREATE TABLE quarantined_media_changes ( + -- Position in the quarantined media stream + stream_id INTEGER NOT NULL PRIMARY KEY, + + -- Name of the worker sending this (makes us compatible with multiple writers) + instance_name TEXT NOT NULL, + + -- Media origin. NULL if local media. + -- We store the origin and media_id as media is scoped to the origin and are uniquely identified by (origin, media_id). + origin TEXT NULL, + + -- Media ID at the origin. + media_id TEXT NOT NULL, + + -- True if quarantined at this position, false otherwise. + quarantined BOOLEAN NOT NULL +); + +-- Start the background update to populate existing quarantined media in the table. See update handler for more details. +INSERT INTO background_updates (ordering, update_name, progress_json) VALUES + (9305, 'flag_existing_quarantined_media', '{}'); diff --git a/synapse/storage/schema/main/delta/94/03_quarantined_media_tracking_seq.sql.postgres b/synapse/storage/schema/main/delta/94/03_quarantined_media_tracking_seq.sql.postgres new file mode 100644 index 00000000000..85f50ba8e79 --- /dev/null +++ b/synapse/storage/schema/main/delta/94/03_quarantined_media_tracking_seq.sql.postgres @@ -0,0 +1,18 @@ +-- +-- This file is licensed under the Affero General Public License (AGPL) version 3. +-- +-- Copyright (C) 2026 Element Creations Ltd +-- +-- This program is free software: you can redistribute it and/or modify +-- it under the terms of the GNU Affero General Public License as +-- published by the Free Software Foundation, either version 3 of the +-- License, or (at your option) any later version. +-- +-- See the GNU Affero General Public License for more details: +-- . + +CREATE SEQUENCE quarantined_media_id_seq; +-- Synapse streams start at 2, because the default position is 1 +-- so any item inserted at position 1 is ignored. +-- We have to use nextval not START WITH 2, see https://github.com/element-hq/synapse/issues/18712 +SELECT nextval('quarantined_media_id_seq'); diff --git a/tests/rest/admin/test_media.py b/tests/rest/admin/test_media.py index 8cc54cc80c2..6469d305b24 100644 --- a/tests/rest/admin/test_media.py +++ b/tests/rest/admin/test_media.py @@ -756,6 +756,133 @@ def _access_media( self.assertFalse(os.path.exists(local_path)) +class ListQuarantinedMediaChangesTestCase(_AdminMediaTests): + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + self.server_name = hs.hostname + self.admin_user = self.register_user("admin", "pass", admin=True) + self.admin_user_tok = self.login("admin", "pass") + + def test_no_auth(self) -> None: + """ + Try to list quarantined media changes without authentication. + """ + + channel = self.make_request( + "GET", + "/_synapse/admin/v1/media/quarantine_changes", + ) + + self.assertEqual(401, channel.code, msg=channel.json_body) + self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"]) + + def test_requester_is_not_admin(self) -> None: + """ + If the user is not a server admin, an error is returned. + """ + self.other_user = self.register_user("user", "pass") + self.other_user_token = self.login("user", "pass") + + channel = self.make_request( + "GET", + "/_synapse/admin/v1/media/quarantine_changes", + access_token=self.other_user_token, + ) + + self.assertEqual(403, channel.code, msg=channel.json_body) + self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"]) + + def _quarantine_local_media(self, media_id: str, admin_user_tok: str) -> None: + channel = self.make_request( + "POST", + "/_synapse/admin/v1/media/quarantine/%s/%s" + % ( + self.server_name, + media_id, + ), + access_token=admin_user_tok, + ) + self.assertEqual(200, channel.code, msg=channel.json_body) + + def _local_upload(self, admin_user_tok: str) -> str: + response = self.helper.upload_media( + SMALL_PNG, tok=admin_user_tok, expect_code=200 + ) + origin_and_media_id = response["content_uri"][6:] # Cut off 'mxc://' + _origin, media_id = origin_and_media_id.split("/") + return media_id + + def test_list_quarantined_media(self) -> None: + """ + Ensure we actually get results for each page and that pagination is seamless. + """ + # Upload 105 media objects to test multiple pages + self.media_ids = [self._local_upload(self.admin_user_tok) for _ in range(105)] + + # No changes before quarantine + channel = self.make_request( + "GET", + "/_synapse/admin/v1/media/quarantine_changes", + access_token=self.admin_user_tok, + ) + self.assertEqual(200, channel.code, msg=channel.json_body) + self.assertEqual(0, len(channel.json_body["changes"])) + + # We expect to continue from the current stream position because we have no changes + self.assertEqual(1, channel.json_body["next_batch"]) + + # Quarantine by hash should kick in to get the other 104 media objects + self._quarantine_local_media(self.media_ids[0], self.admin_user_tok) + + # Page 1 (implied ?from=0) + channel = self.make_request( + "GET", + "/_synapse/admin/v1/media/quarantine_changes", + access_token=self.admin_user_tok, + ) + self.assertEqual(200, channel.code, msg=channel.json_body) + self.assertEqual(100, len(channel.json_body["changes"])) + self.assertEqual(101, channel.json_body["next_batch"]) + for row in channel.json_body["changes"]: + self.assertIn( + row["media_id"], + self.media_ids[0:100], + ) + self.assertEqual(row["origin"], self.server_name) + self.assertEqual(row["quarantined"], True) + + # Page 2 (explicit ?from, using next_batch) + channel = self.make_request( + "GET", + f"/_synapse/admin/v1/media/quarantine_changes?from={channel.json_body['next_batch']}", + access_token=self.admin_user_tok, + ) + self.assertEqual(200, channel.code, msg=channel.json_body) + self.assertEqual(5, len(channel.json_body["changes"])) + self.assertEqual(106, channel.json_body["next_batch"]) + for row in channel.json_body["changes"]: + self.assertIn( + row["media_id"], + self.media_ids[100:], + ) + self.assertEqual(row["origin"], self.server_name) + self.assertEqual(row["quarantined"], True) + + def test_list_quarantined_media_bounds_high(self) -> None: + """ + Ensure out of bounds (token stream position greater than our furthest persisted + position) requests with high `from` values are met with an appropriate error. + """ + # Page that's very much out of range + channel = self.make_request( + "GET", + "/_synapse/admin/v1/media/quarantine_changes?from=900000", + access_token=self.admin_user_tok, + ) + self.assertEqual(400, channel.code, msg=channel.json_body) + self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"]) + + class QuarantineMediaByIDTestCase(_AdminMediaTests): def upload_media_and_return_media_id(self, data: bytes) -> str: # Upload some media into the room diff --git a/tests/storage/test_room.py b/tests/storage/test_room.py index f8c5260fa2e..b4fb5d56288 100644 --- a/tests/storage/test_room.py +++ b/tests/storage/test_room.py @@ -23,9 +23,12 @@ from synapse.api.room_versions import RoomVersions from synapse.server import HomeServer +from synapse.storage.databases.main.room import _BackgroundUpdates +from synapse.storage.types import Cursor from synapse.types import RoomAlias, RoomID, UserID from synapse.util.clock import Clock +from tests.rest.admin.test_media import _AdminMediaTests from tests.unittest import HomeserverTestCase @@ -67,3 +70,89 @@ def test_get_room_with_stats_unknown_room(self) -> None: self.assertIsNone( self.get_success(self.store.get_room_with_stats("!uknown:test")) ) + + +class FlagExistingQuarantinedMediaBackgroundUpdatesTestCase(_AdminMediaTests): + """ + Test the `flag_existing_quarantined_media` background update. + """ + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + + def test_populates_quarantined_only(self) -> None: + self.register_user("admin", "pass", admin=True) + admin_user_tok = self.login("admin", "pass") + + # Upload two distinct media items so we can quarantine one. If they shared content, + # then the quarantine-by-hash code would hit both. + _unaffected_media_upload_response = self.helper.upload_media( + b"first content", tok=admin_user_tok, expect_code=200 + ) + # Upload the media we're going to quarantine + media_upload_response = self.helper.upload_media( + b"second content", tok=admin_user_tok, expect_code=200 + ) + # Extract media ID from the response + quarantined_media_origin_and_media_id = media_upload_response["content_uri"][ + 6: + ] # cut off 'mxc://' + quarantined_media_origin, quarantined_media_id = ( + quarantined_media_origin_and_media_id.split("/") + ) + + # Ideally we'd also upload remote media to ensure that `remote_media_cache` gets picked up, but that's + # a little tricky to set up in the test here. We hope that local and remote media + # are treated similarly during the background update. + + # Quarantine the media like an admin would. Because the quarantine API also inserts + # a record into the database for us, we'll clear out the `quarantined_media_changes` + # table before running the background update. This will simulate already-quarantined + # media being in the database prior to the background update. + channel = self.make_request( + "POST", + "/_synapse/admin/v1/media/quarantine/%s/%s" + % ( + quarantined_media_origin, + quarantined_media_id, + ), + access_token=admin_user_tok, + ) + self.assertEqual(200, channel.code, msg=channel.json_body) + + # Do that table clear we mentioned above + def _wipe_table(txn: Cursor) -> None: + txn.execute("DELETE FROM quarantined_media_changes") + + self.get_success( + self.store.db_pool.runInteraction( + "test_populates_quarantined_only._wipe_table", _wipe_table + ) + ) + + # Insert and run the background update + self.get_success( + self.store.db_pool.simple_insert( + table="background_updates", + values={ + "update_name": _BackgroundUpdates.FLAG_EXISTING_QUARANTINED_MEDIA, + "progress_json": "{}", + }, + ) + ) + self.store.db_pool.updates._all_done = False + self.wait_for_background_updates() + + # Check that the changes table is now populated, and has exactly 1 quarantined + # media object in it (the one we quarantined). + changes: list[tuple[str | None, str, bool]] = self.get_success( + self.store.db_pool.simple_select_list( + "quarantined_media_changes", + None, + retcols=("origin", "media_id", "quarantined"), + ) + ) + self.assertEqual(len(changes), 1) + self.assertEqual(changes[0][0], None) # origin (local media) + self.assertEqual(changes[0][1], quarantined_media_id) # media_id + self.assertEqual(changes[0][2], True) # quarantined flag