From 4bcf6140aa42af299c71e84adbbc8c1f23dc1f2d Mon Sep 17 00:00:00 2001 From: Andrew Morgan Date: Tue, 13 Oct 2026 15:37:33 +0100 Subject: [PATCH] Upload and cache configured server announcement avatars --- .../handlers/server_announcement_avatar.py | 188 +++++++++ synapse/handlers/server_announcements.py | 5 +- synapse/replication/http/__init__.py | 2 + .../replication/http/server_announcements.py | 100 +++++ .../test_server_announcement_avatar.py | 387 ++++++++++++++++++ .../test_server_announcement_avatar.py | 371 +++++++++++++++++ 6 files changed, 1052 insertions(+), 1 deletion(-) create mode 100644 synapse/handlers/server_announcement_avatar.py create mode 100644 synapse/replication/http/server_announcements.py create mode 100644 tests/handlers/test_server_announcement_avatar.py create mode 100644 tests/replication/test_server_announcement_avatar.py diff --git a/synapse/handlers/server_announcement_avatar.py b/synapse/handlers/server_announcement_avatar.py new file mode 100644 index 0000000000..1438acc9d7 --- /dev/null +++ b/synapse/handlers/server_announcement_avatar.py @@ -0,0 +1,188 @@ +# +# 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: +# . +# + +import hashlib +import os +from io import BytesIO +from typing import TYPE_CHECKING + +from matrix_common.types.mxc_uri import MXCUri +from PIL import Image + +from synapse.config import ConfigError +from synapse.config.server_announcements import ServerAnnouncementRoom +from synapse.logging.context import defer_to_thread +from synapse.media._base import FileInfo +from synapse.types import UserID + +if TYPE_CHECKING: + from synapse.server import HomeServer + + +class ServerAnnouncementAvatarManager: + """Resolve avatars for main-process reconciliation, delegating uploads if needed. + + Only the configured media instance needs access to the avatar file and media + repository. The caller must persist the announcement's room mapping before + calling this helper: the uploaded avatar cache belongs to that mapping. + """ + + def __init__(self, hs: "HomeServer") -> None: + self._hs = hs + self._store = hs.get_datastores().main + + async def get_avatar_url(self, room: ServerAnnouncementRoom) -> str | None: + path = room.room_avatar_path + if path is None: + # None deliberately clears an old avatar, even if an upload is cached. + return room.room_avatar_url + + media_instance_name = self._hs.config.serverannouncements.media_instance_name + if media_instance_name != self._hs.get_instance_name(): + # Import here to avoid a cycle through replication servlet registration. + from synapse.replication.http.server_announcements import ( + ReplicationServerAnnouncementAvatarRestServlet, + ) + + upload_request = ReplicationServerAnnouncementAvatarRestServlet.make_client( + self._hs + ) + response = await upload_request( + instance_name=media_instance_name, announcement_id=room.id + ) + return response["avatar_url"] + + if not self._hs.config.media.can_load_media_repo: + raise self._avatar_error( + room, + "The configured avatar uploader must have the media repository enabled", + ) + + # Hash the contents so edits at the same path are detected on startup, + # even if the file's size and modification time are unchanged. + content, sha256, media_type = await defer_to_thread( + self._hs.get_reactor(), self._read_avatar, room, path + ) + existing_media = await self._store.get_server_announcement_avatar(room.id) + if existing_media is not None: + existing_path, existing_hash, existing_mxc = existing_media + if ( + existing_path == path + and existing_hash == sha256 + and await self._media_is_available(room, existing_mxc) + ): + return existing_mxc + + media_repo = self._hs.get_media_repository() + with BytesIO(content) as upload: + mxc = await media_repo.create_or_update_content( + media_type=media_type, + upload_name=os.path.basename(path), + content=upload, + content_length=len(content), + auth_user=UserID.from_string( + self._hs.config.serverannouncements.server_announcements_mxid + ), + ) + avatar_url = str(mxc) + # Uploads may be automatically quarantined if their hash matches other + # quarantined media. Do not publish or cache an unusable avatar. + if not await self._media_is_available(room, avatar_url): + raise self._avatar_error( + room, "Uploaded media is missing; check the media storage configuration" + ) + await self._store.store_server_announcement_avatar( + room.id, path, sha256, avatar_url + ) + return avatar_url + + async def _media_is_available( + self, room: ServerAnnouncementRoom, avatar_url: str + ) -> bool: + """Check that a locally uploaded avatar is complete and retrievable. + + A cache entry is reusable only if its local media record and content still + exist, including content held by a storage provider. Quarantined media + raises an error rather than being silently re-uploaded. + + This check is only for uploads from room_avatar_path. An administrator's + explicit room_avatar_url may be remote and is returned without fetching + or checking its availability. + """ + try: + mxc = MXCUri.from_str(avatar_url) + except ValueError: + return False + if not self._hs.is_mine_server_name(mxc.server_name): + return False + + media = await self._store.get_local_media(mxc.media_id) + if media is None: + return False + if media.quarantined_by is not None: + raise self._avatar_error( + room, + f"Media {avatar_url!r} is quarantined; unquarantine it or configure " + "a different image", + ) + if media.media_length is None: + return False + + # Consult all configured providers, not just the local filesystem, and + # close the responder even though we do not need to stream its contents. + responder = await self._hs.get_media_repository().media_storage.fetch_media( + FileInfo( + server_name=None, file_id=mxc.media_id, url_cache=bool(media.url_cache) + ) + ) + if responder is None: + return False + with responder: + return True + + def _read_avatar( + self, room: ServerAnnouncementRoom, path: str + ) -> tuple[bytes, str, str]: + """Read, validate and hash one snapshot of the file, off the reactor thread.""" + try: + with open(path, "rb") as avatar_file: + # This administrator-configured file is not subject to client upload limits. + content = avatar_file.read() + except (OSError, ValueError) as e: + raise self._avatar_error( + room, f"Cannot read file: {e}; check the path and file permissions" + ) from e + + try: + with Image.open(BytesIO(content)) as image: + media_type = Image.MIME.get(image.format or "") + if media_type is None or not media_type.startswith("image/"): + raise ValueError("image format has no supported image MIME type") + image.verify() + # verify() alone does not decode pixels for all formats (e.g. JPEG). + with Image.open(BytesIO(content)) as image: + image.load() + except (OSError, ValueError, SyntaxError, Image.DecompressionBombError) as e: + raise self._avatar_error( + room, f"Invalid or unsupported image: {e}; provide a valid image file" + ) from e + + return content, hashlib.sha256(content).hexdigest(), media_type + + @staticmethod + def _avatar_error(room: ServerAnnouncementRoom, reason: str) -> ConfigError: + return ConfigError( + f"Server announcement room {room.id!r}, room_avatar_path " + f"{room.room_avatar_path!r}: {reason}" + ) diff --git a/synapse/handlers/server_announcements.py b/synapse/handlers/server_announcements.py index d274e35f0d..51bb831c9e 100644 --- a/synapse/handlers/server_announcements.py +++ b/synapse/handlers/server_announcements.py @@ -18,6 +18,7 @@ from synapse.api.constants import EventTypes, HistoryVisibility, JoinRules, Memb from synapse.api.room_versions import RoomVersion from synapse.config import ConfigError from synapse.config.server_announcements import ServerAnnouncementRoom +from synapse.handlers.server_announcement_avatar import ServerAnnouncementAvatarManager from synapse.types import JsonDict, UserID, create_requester from synapse.types.state import StateFilter from synapse.util.duration import Duration @@ -40,6 +41,7 @@ class ServerAnnouncementsHandler: self._room_member = hs.get_room_member_handler() self._event_creation = hs.get_event_creation_handler() self._locks = hs.get_worker_locks_handler() + self._avatars = ServerAnnouncementAvatarManager(hs) self._creator = self._config.server_announcements_mxid self._requester = create_requester( self._creator, authenticated_entity=hs.hostname @@ -230,10 +232,11 @@ class ServerAnnouncementsHandler: EventTypes.Topic, {"topic": room.room_topic or ""}, ) + avatar_url = await self._avatars.get_avatar_url(room) await self._update_state( room_id, EventTypes.RoomAvatar, - {"url": room.room_avatar_url} if room.room_avatar_url is not None else {}, + {"url": avatar_url} if avatar_url is not None else {}, ) # Revoke both membership and outstanding invitations. The internal creator diff --git a/synapse/replication/http/__init__.py b/synapse/replication/http/__init__.py index d934ef8067..6e10cf4757 100644 --- a/synapse/replication/http/__init__.py +++ b/synapse/replication/http/__init__.py @@ -34,6 +34,7 @@ from synapse.replication.http import ( push, register, send_events, + server_announcements, state, streams, ) @@ -61,6 +62,7 @@ class ReplicationRestResource(JsonResource): state.register_servlets(hs, self) devices.register_servlets(hs, self) profile.register_servlets(hs, self) + server_announcements.register_servlets(hs, self) # The following can't currently be instantiated on workers. if hs.config.worker.worker_app is None: diff --git a/synapse/replication/http/server_announcements.py b/synapse/replication/http/server_announcements.py new file mode 100644 index 0000000000..4ff31ab267 --- /dev/null +++ b/synapse/replication/http/server_announcements.py @@ -0,0 +1,100 @@ +# +# 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: +# . +# + +from typing import TYPE_CHECKING + +from twisted.web.server import Request + +from synapse.api.errors import SynapseError +from synapse.config import ConfigError +from synapse.handlers.server_announcement_avatar import ServerAnnouncementAvatarManager +from synapse.http.server import HttpServer +from synapse.replication.http._base import ReplicationEndpoint +from synapse.types import JsonDict + +if TYPE_CHECKING: + from synapse.server import HomeServer + + +class ReplicationServerAnnouncementAvatarRestServlet(ReplicationEndpoint): + """Upload an announcement avatar on the designated media instance. + + POST /_synapse/replication/server_announcement_avatar// + + The caller supplies only the logical announcement ID. The target resolves the + trusted file path from its own administrator configuration, never from the + request. Authentication uses the standard replication shared secret. + """ + + NAME = "server_announcement_avatar" + PATH_ARGS = ("announcement_id",) + + # Keep the standard bounded connection retries, but do not retry timeouts + # indefinitely: the main-process reconciler owns the overall retry budget. + RETRY_ON_TIMEOUT = False + + def __init__(self, hs: "HomeServer") -> None: + super().__init__(hs) + self._hs = hs + + @staticmethod + async def _serialize_payload(announcement_id: str) -> JsonDict: # type: ignore[override] + return {} + + async def _handle_request( # type: ignore[override] + self, request: Request, content: JsonDict, announcement_id: str + ) -> tuple[int, JsonDict]: + config = self._hs.config.serverannouncements + if self._hs.get_instance_name() != config.media_instance_name: + raise SynapseError( + 400, "This instance is not the configured announcement avatar uploader" + ) + if not self._hs.config.media.can_load_media_repo: + raise SynapseError( + 400, + "The announcement avatar uploader must have the media repository enabled", + ) + + room = next((room for room in config.rooms if room.id == announcement_id), None) + if room is None: + raise SynapseError( + 404, + f"Server announcement room {announcement_id!r} is not configured on this instance", + ) + if room.room_avatar_path is None: + raise SynapseError( + 400, + f"Server announcement room {room.id!r} has no room_avatar_path configured", + ) + if ( + await self._hs.get_datastores().main.get_server_announcement_room(room.id) + is None + ): + raise SynapseError( + 400, f"Server announcement room {room.id!r} has no room mapping" + ) + + try: + avatar_url = await ServerAnnouncementAvatarManager(self._hs).get_avatar_url( + room + ) + except ConfigError as e: + # Preserve actionable file/validation/quarantine errors for the main + # process instead of returning the generic internal-server-error text. + raise SynapseError(500, str(e)) from e + return 200, {"avatar_url": avatar_url} + + +def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: + ReplicationServerAnnouncementAvatarRestServlet(hs).register(http_server) diff --git a/tests/handlers/test_server_announcement_avatar.py b/tests/handlers/test_server_announcement_avatar.py new file mode 100644 index 0000000000..2e446254c2 --- /dev/null +++ b/tests/handlers/test_server_announcement_avatar.py @@ -0,0 +1,387 @@ +# +# 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: +# . +# + +import hashlib +import os +from io import BytesIO +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +import attr +from matrix_common.types.mxc_uri import MXCUri +from parameterized import parameterized +from PIL import Image + +from twisted.internet.testing import MemoryReactor + +from synapse.api.constants import EventTypes +from synapse.config import ConfigError +from synapse.config.server_announcements import ServerAnnouncementRoom +from synapse.handlers.server_announcement_avatar import ServerAnnouncementAvatarManager +from synapse.handlers.server_announcements import ServerAnnouncementsHandler +from synapse.media.storage_provider import ( + FileStorageProviderBackend, + StorageProviderWrapper, +) +from synapse.server import HomeServer +from synapse.types import UserID +from synapse.util.clock import Clock + +from tests.unittest import HomeserverTestCase, override_config + + +class ServerAnnouncementAvatarIntegrationTestCase(HomeserverTestCase): + needs_threadpool = True + + def test_reconciliation_uploads_reuses_updates_and_clears_avatar(self) -> None: + directory = TemporaryDirectory(prefix="synapse-announcement-integration-") + self.addCleanup(directory.cleanup) + path = Path(directory.name) / "avatar.png" + self.hs.config.media.media_store_path = str(Path(directory.name) / "media") + with Image.new("RGB", (8, 8), "red") as image: + image.save(path) + config = self.hs.config.serverannouncements + config.rooms = [ + ServerAnnouncementRoom( + id="updates", + room_name="Updates", + room_topic=None, + room_avatar_url=None, + room_avatar_path=str(path), + allowed_user_ids=[], + ) + ] + handler = ServerAnnouncementsHandler(self.hs) + self.get_success(handler.reconcile_rooms()) + store = self.hs.get_datastores().main + self.assertIsNone( + self.get_success(store.get_user_by_id(config.server_announcements_mxid)) + ) + room_id = self.get_success(store.get_server_announcement_room("updates")) + assert room_id is not None + state = self.hs.get_storage_controllers().state + first = self.get_success( + state.get_current_state_event(room_id, EventTypes.RoomAvatar, "") + ) + assert first is not None + self.assertTrue(first.content["url"].startswith("mxc://test/")) + + handler = ServerAnnouncementsHandler(self.hs) + self.get_success(handler.reconcile_rooms()) + unchanged = self.get_success( + state.get_current_state_event(room_id, EventTypes.RoomAvatar, "") + ) + assert unchanged is not None + self.assertEqual(first.event_id, unchanged.event_id) + with Image.new("RGB", (8, 8), "blue") as image: + image.save(path) + self.get_success(handler.reconcile_rooms()) + changed = self.get_success( + state.get_current_state_event(room_id, EventTypes.RoomAvatar, "") + ) + assert changed is not None + self.assertNotEqual(first.content["url"], changed.content["url"]) + + config.rooms = [attr.evolve(config.rooms[0], room_avatar_path=None)] + path.unlink() + self.get_success(handler.reconcile_rooms()) + cleared = self.get_success( + state.get_current_state_event(room_id, EventTypes.RoomAvatar, "") + ) + assert cleared is not None + self.assertEqual(cleared.content, {}) + self.assertEqual( + self.get_success(store.get_server_announcement_room("updates")), room_id + ) + + +class ServerAnnouncementAvatarURLTestCase(HomeserverTestCase): + @parameterized.expand([(None,), ("mxc://elsewhere.example/avatar",)]) + @override_config({"enable_media_repo": False}) + def test_url_without_media_repository(self, avatar_url: str | None) -> None: + room = ServerAnnouncementRoom( + id="news", + room_name="News", + room_topic=None, + room_avatar_url=avatar_url, + room_avatar_path=None, + allowed_user_ids=[], + ) + # Explicit remote MXC URIs are allowed; availability checks apply only + # to locally uploaded files, not to administrator-provided URLs. + with ( + patch.object(self.hs, "get_media_repository") as get_repository, + patch.object(self.hs, "get_replication_client") as get_replication, + patch.object( + ServerAnnouncementAvatarManager, "_media_is_available" + ) as available, + ): + manager = ServerAnnouncementAvatarManager(self.hs) + self.assertEqual(self.get_success(manager.get_avatar_url(room)), avatar_url) + get_repository.assert_not_called() + get_replication.assert_not_called() + available.assert_not_called() + + +class ServerAnnouncementAvatarTestCase(HomeserverTestCase): + needs_threadpool = True + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + directory = TemporaryDirectory(prefix="synapse-announcement-avatar-") + self.addCleanup(directory.cleanup) + self.directory = Path(directory.name) + hs.config.media.media_store_path = str(self.directory / "media") + self.media_repo = hs.get_media_repository() + self.store = hs.get_datastores().main + self.uploader = hs.config.serverannouncements.server_announcements_mxid + + # Deliberately give the PNG a misleading extension. + self.avatar_path = self.directory / "avatar.jpg" + self.content = self._image_bytes() + self.avatar_path.write_bytes(self.content) + self.room = ServerAnnouncementRoom( + id="news", + room_name="News", + room_topic=None, + room_avatar_url=None, + room_avatar_path=str(self.avatar_path), + allowed_user_ids=[], + ) + self.get_success( + self.store.store_server_announcement_room(self.room.id, "!news:test") + ) + self.manager = ServerAnnouncementAvatarManager(hs) + + @staticmethod + def _image_bytes(image_format: str = "PNG", color: str = "red") -> bytes: + with BytesIO() as output, Image.new("RGB", (8, 8), color) as image: + image.save(output, format=image_format) + return output.getvalue() + + def _get_avatar(self) -> str: + avatar_url = self.get_success(self.manager.get_avatar_url(self.room)) + assert avatar_url is not None + return avatar_url + + def _assert_cached(self, avatar_url: str) -> None: + self.assertEqual( + self.get_success(self.store.get_server_announcement_avatar(self.room.id)), + ( + self.room.room_avatar_path, + hashlib.sha256(self.content).hexdigest(), + avatar_url, + ), + ) + + def _assert_stored_content(self, avatar_url: str) -> None: + mxc = MXCUri.from_str(avatar_url) + self.assertEqual(mxc.server_name, self.hs.hostname) + path = Path(self.media_repo.filepaths.local_media_filepath(mxc.media_id)) + self.assertEqual(path.read_bytes(), self.content) + + @parameterized.expand( + [("PNG", "image/png"), ("JPEG", "image/jpeg"), ("GIF", "image/gif")] + ) + def test_upload_content_type(self, image_format: str, media_type: str) -> None: + self.content = self._image_bytes(image_format) + self.avatar_path.write_bytes(self.content) + avatar_url = self._get_avatar() + mxc = MXCUri.from_str(avatar_url) + media = self.get_success(self.store.get_local_media(mxc.media_id)) + assert media is not None + self.assertEqual(media.media_type, media_type) + self.assertEqual(media.media_length, len(self.content)) + self.assertEqual(media.upload_name, self.avatar_path.name) + self.assertEqual(media.user_id, self.uploader) + self._assert_stored_content(avatar_url) + self._assert_cached(avatar_url) + + @parameterized.expand([(None,), ("mxc://elsewhere.example/new-avatar",)]) + def test_removed_path_overrides_cached_upload(self, avatar_url: str | None) -> None: + old_url = self._get_avatar() + self.avatar_path.unlink() + room = attr.evolve(self.room, room_avatar_path=None, room_avatar_url=avatar_url) + with patch.object(self.hs, "get_media_repository") as get_repository: + self.assertEqual( + self.get_success(self.manager.get_avatar_url(room)), avatar_url + ) + get_repository.assert_not_called() + self._assert_cached(old_url) + + def test_unchanged_content_reused_across_instances(self) -> None: + avatar_url = self._get_avatar() + # Rewriting identical bytes must not invalidate the content-based cache. + self.avatar_path.write_bytes(self.content) + with patch.object(self.media_repo, "create_or_update_content") as upload: + self.assertEqual(self._get_avatar(), avatar_url) + self.manager = ServerAnnouncementAvatarManager(self.hs) + self.assertEqual(self._get_avatar(), avatar_url) + upload.assert_not_called() + self._assert_cached(avatar_url) + + def test_replaced_file_at_same_path(self) -> None: + # Changing BMP pixels preserves the encoded size; preserve mtime too, to + # ensure that neither file size nor timestamps substitute for hashing. + self.content = self._image_bytes("BMP", "red") + self.avatar_path.write_bytes(self.content) + avatar_url = self._get_avatar() + old_stat = self.avatar_path.stat() + self.content = self._image_bytes("BMP", "blue") + self.assertEqual(len(self.content), old_stat.st_size) + self.avatar_path.write_bytes(self.content) + os.utime(self.avatar_path, ns=(old_stat.st_atime_ns, old_stat.st_mtime_ns)) + new_url = self._get_avatar() + self.assertNotEqual(new_url, avatar_url) + self._assert_stored_content(new_url) + self._assert_cached(new_url) + + def test_changed_path_with_same_contents(self) -> None: + avatar_url = self._get_avatar() + new_path = self.directory / "new-avatar.png" + new_path.write_bytes(self.content) + self.room = attr.evolve(self.room, room_avatar_path=str(new_path)) + new_url = self._get_avatar() + self.assertNotEqual(new_url, avatar_url) + self._assert_cached(new_url) + + def test_deleted_cached_media_is_uploaded_again(self) -> None: + avatar_url = self._get_avatar() + media_id = MXCUri.from_str(avatar_url).media_id + self.get_success(self.media_repo.delete_local_media_ids([media_id])) + self.assertIsNone(self.get_success(self.store.get_local_media(media_id))) + self.manager = ServerAnnouncementAvatarManager(self.hs) + new_url = self._get_avatar() + self.assertNotEqual(new_url, avatar_url) + self._assert_stored_content(new_url) + self._assert_cached(new_url) + + def test_missing_stored_file_is_uploaded_again(self) -> None: + avatar_url = self._get_avatar() + media_id = MXCUri.from_str(avatar_url).media_id + Path(self.media_repo.filepaths.local_media_filepath(media_id)).unlink() + self.assertIsNotNone(self.get_success(self.store.get_local_media(media_id))) + self.manager = ServerAnnouncementAvatarManager(self.hs) + new_url = self._get_avatar() + self.assertNotEqual(new_url, avatar_url) + self._assert_stored_content(new_url) + self._assert_cached(new_url) + + def test_cached_media_in_storage_provider_is_reused(self) -> None: + self.media_repo.media_storage.storage_providers = [ + StorageProviderWrapper( + FileStorageProviderBackend(self.hs, str(self.directory / "backup")), + store_local=True, + store_remote=False, + store_synchronous=True, + ), + ] + avatar_url = self._get_avatar() + media_id = MXCUri.from_str(avatar_url).media_id + Path(self.media_repo.filepaths.local_media_filepath(media_id)).unlink() + with patch.object(self.media_repo, "create_or_update_content") as upload: + self.manager = ServerAnnouncementAvatarManager(self.hs) + self.assertEqual(self._get_avatar(), avatar_url) + upload.assert_not_called() + + def test_quarantined_cached_media_is_not_reused(self) -> None: + avatar_url = self._get_avatar() + self.get_success( + self.store.quarantine_media_by_id( + self.hs.hostname, MXCUri.from_str(avatar_url).media_id, "@admin:test" + ) + ) + with patch.object(self.media_repo, "create_or_update_content") as upload: + self._assert_avatar_error("quarantined") + upload.assert_not_called() + self._assert_cached(avatar_url) + + def test_automatically_quarantined_upload_is_not_cached(self) -> None: + self.wait_for_background_updates() + mxc = self.get_success( + self.media_repo.create_or_update_content( + media_type="image/png", + upload_name="other.png", + content=BytesIO(self.content), + content_length=len(self.content), + auth_user=UserID.from_string(self.uploader), + ) + ) + self.get_success( + self.store.quarantine_media_by_id( + self.hs.hostname, mxc.media_id, "@admin:test" + ) + ) + self._assert_avatar_error("quarantined") + self.assertIsNone( + self.get_success(self.store.get_server_announcement_avatar(self.room.id)) + ) + + def test_exact_size_limit_is_allowed(self) -> None: + self.hs.config.media.max_upload_size = len(self.content) + self.manager = ServerAnnouncementAvatarManager(self.hs) + self._assert_cached(self._get_avatar()) + + def test_oversized_administrator_image_is_allowed(self) -> None: + self.hs.config.media.max_upload_size = len(self.content) - 1 + self.manager = ServerAnnouncementAvatarManager(self.hs) + avatar_url = self._get_avatar() + self._assert_cached(avatar_url) + self._assert_stored_content(avatar_url) + + @parameterized.expand( + [ + ("empty", b""), + ("text", b"not an image"), + ("truncated_png", b"\x89PNG\r\n\x1a\n"), + ] + ) + def test_invalid_image(self, name: str, content: bytes) -> None: + self.avatar_path.write_bytes(content) + with patch.object(self.media_repo, "create_or_update_content") as upload: + self._assert_avatar_error("Invalid or unsupported image") + upload.assert_not_called() + self.assertIsNone( + self.get_success(self.store.get_server_announcement_avatar(self.room.id)) + ) + + def test_truncated_image_data(self) -> None: + # The PNG header is intact, but its pixel data is incomplete. + self.avatar_path.write_bytes(self.content[:-20]) + self._assert_avatar_error("Invalid or unsupported image") + + def test_invalid_replacement_does_not_use_old_cache(self) -> None: + avatar_url = self._get_avatar() + self.avatar_path.write_bytes(b"not an image") + self._assert_avatar_error("Invalid or unsupported image") + self._assert_cached(avatar_url) + + def test_missing_avatar_file(self) -> None: + self.avatar_path.unlink() + self._assert_avatar_error("Cannot read file") + + def test_unreadable_avatar_file(self) -> None: + with patch( + "synapse.handlers.server_announcement_avatar.open", + side_effect=PermissionError("Permission denied"), + ): + self._assert_avatar_error("file permissions") + + def _assert_avatar_error(self, message: str) -> None: + failure = self.get_failure(self.manager.get_avatar_url(self.room), ConfigError) + self.assertIn(message, str(failure.value)) + self.assertTrue( + str(failure.value).startswith(f"Server announcement room {self.room.id!r},") + ) + self.assertIn(str(self.avatar_path), str(failure.value)) diff --git a/tests/replication/test_server_announcement_avatar.py b/tests/replication/test_server_announcement_avatar.py new file mode 100644 index 0000000000..5ede3ca341 --- /dev/null +++ b/tests/replication/test_server_announcement_avatar.py @@ -0,0 +1,371 @@ +# +# 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: +# . +# + +import hashlib +from io import BytesIO +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import AsyncMock, Mock, patch + +import attr +from matrix_common.types.mxc_uri import MXCUri +from parameterized import parameterized +from PIL import Image + +from twisted.internet.defer import ensureDeferred +from twisted.internet.error import ConnectError + +from synapse.api.constants import EventTypes +from synapse.api.errors import SynapseError +from synapse.config.server_announcements import ServerAnnouncementRoom +from synapse.handlers.server_announcement_avatar import ServerAnnouncementAvatarManager +from synapse.handlers.server_announcements import ServerAnnouncementsHandler +from synapse.http import RequestTimedOutError +from synapse.replication.http.server_announcements import ( + ReplicationServerAnnouncementAvatarRestServlet, +) +from synapse.server import HomeServer +from synapse.types import JsonDict + +from tests.replication._base import BaseMultiWorkerStreamTestCase +from tests.server import FakeChannel, make_request +from tests.unittest import HomeserverTestCase, override_config + + +class ServerAnnouncementAvatarWorkerTestCase(BaseMultiWorkerStreamTestCase): + needs_threadpool = True + + def setUp(self) -> None: + directory = TemporaryDirectory(prefix="synapse-announcement-worker-") + self.addCleanup(directory.cleanup) + self.directory = Path(directory.name) + self.avatar_path = self.directory / "worker-avatar.png" + self._write_image("red") + super().setUp() + + self.worker = self.make_worker_hs( + "synapse.app.generic_worker", + { + "worker_name": "media1", + "enable_media_repo": True, + "media_store_path": str(self.directory / "media"), + "server_announcements": [ + { + "id": "news", + "room_name": "News", + "room_avatar_path": str(self.avatar_path), + } + ], + }, + ) + self.store = self.hs.get_datastores().main + self.room = self.hs.config.serverannouncements.rooms[0] + self.manager = ServerAnnouncementAvatarManager(self.hs) + self.media_repo = self.worker.get_media_repository() + self.get_success( + self.store.store_server_announcement_room("news", "!news:test") + ) + # Main has neither a usable avatar path nor a media repository/storage. + self.assertFalse(self.hs.config.media.can_load_media_repo) + assert self.room.room_avatar_path is not None + self.assertFalse(Path(self.room.room_avatar_path).exists()) + self.patch( + self.hs, + "get_media_repository", + Mock(side_effect=AssertionError("main must not load media")), + ) + self.patch( + self.manager, + "_read_avatar", + Mock(side_effect=AssertionError("main must not read avatar files")), + ) + + def default_config(self) -> JsonDict: + config = super().default_config() + config.update( + { + "enable_media_repo": False, + "media_instance_running_background_jobs": "media1", + "worker_replication_secret": "announcement-test-secret", + "server_announcements": [ + { + "id": "news", + "room_name": "News", + "room_avatar_path": str( + self.directory / "not-mounted-on-main.png" + ), + } + ], + } + ) + config["instance_map"].update( + { + "media1": {"host": "testserv", "port": 9093}, + "media2": {"host": "testserv", "port": 9094}, + } + ) + return config + + def _write_image(self, color: str) -> None: + with BytesIO() as output, Image.new("RGB", (8, 8), color) as image: + image.save(output, format="PNG") + self.content = output.getvalue() + self.avatar_path.write_bytes(self.content) + + def _get_avatar(self) -> str: + avatar_url = self.get_success(self.manager.get_avatar_url(self.room)) + assert avatar_url is not None + return avatar_url + + def _assert_uploaded(self, avatar_url: str) -> None: + mxc = MXCUri.from_str(avatar_url) + self.assertEqual(mxc.server_name, self.hs.hostname) + self.assertEqual( + Path( + self.media_repo.filepaths.local_media_filepath(mxc.media_id) + ).read_bytes(), + self.content, + ) + self.assertEqual( + self.get_success(self.store.get_server_announcement_avatar("news")), + ( + str(self.avatar_path), + hashlib.sha256(self.content).hexdigest(), + avatar_url, + ), + ) + + def _request( + self, + announcement_id: str = "news", + content: JsonDict | None = None, + secret: str | None = "announcement-test-secret", + target: HomeServer | None = None, + ) -> FakeChannel: + return make_request( + self.reactor, + self._hs_to_site[target or self.worker], + "POST", + f"/_synapse/replication/server_announcement_avatar/{announcement_id}/test-txn", + content or {}, + shorthand=False, + custom_headers=[("Authorization", f"Bearer {secret}")] if secret else [], + ) + + def test_upload_and_reuse_existing_file_uri(self) -> None: + client = self.hs.get_replication_client() + with patch.object( + client, "post_json_get_json", wraps=client.post_json_get_json + ) as request: + avatar_url = self._get_avatar() + request.assert_awaited_once() + uri, payload = request.call_args.args + self.assertIn("/server_announcement_avatar/news/", uri) + # Only the logical ID (in the URI) and standard replication metadata + # cross the process boundary; never main's filesystem path. + self.assertEqual(set(payload), {"_INT_STREAM_POS"}) + self._assert_uploaded(avatar_url) + with patch.object(self.media_repo, "create_or_update_content") as upload: + self.assertEqual(self._get_avatar(), avatar_url) + # The cache survives replacing the caller, not just an in-memory object. + self.manager = ServerAnnouncementAvatarManager(self.hs) + self.assertEqual(self._get_avatar(), avatar_url) + upload.assert_not_called() + + def test_oversized_worker_avatar_is_allowed(self) -> None: + self.worker.config.media.max_upload_size = len(self.content) - 1 + self._assert_uploaded(self._get_avatar()) + + def test_changed_file_is_uploaded_again(self) -> None: + old_url = self._get_avatar() + self._write_image("blue") + new_url = self._get_avatar() + self.assertNotEqual(old_url, new_url) + self._assert_uploaded(new_url) + + def test_main_reconciliation_uses_media_worker(self) -> None: + # Let reconciliation create the actual room and its mapping before upload. + self.get_success( + self.store.db_pool.simple_delete( + table="server_announcement_rooms", + keyvalues={"id": "news"}, + desc="remove_test_mapping", + ) + ) + handler = ServerAnnouncementsHandler(self.hs) + self.get_success(handler.reconcile_rooms()) + room_id = self.get_success(self.store.get_server_announcement_room("news")) + assert room_id is not None + avatar = self.get_success( + self.hs.get_storage_controllers().state.get_current_state_event( + room_id, EventTypes.RoomAvatar, "" + ) + ) + assert avatar is not None + self._assert_uploaded(avatar.content["url"]) + self.get_success(handler.reconcile_rooms()) + unchanged = self.get_success( + self.hs.get_storage_controllers().state.get_current_state_event( + room_id, EventTypes.RoomAvatar, "" + ) + ) + assert unchanged is not None + self.assertEqual(avatar.event_id, unchanged.event_id) + + @parameterized.expand([(None,), ("mxc://remote.example/avatar",)]) + def test_direct_url_does_not_contact_media_worker( + self, avatar_url: str | None + ) -> None: + room = attr.evolve(self.room, room_avatar_path=None, room_avatar_url=avatar_url) + with patch.object(self.hs, "get_replication_client") as replication: + self.assertEqual( + self.get_success(self.manager.get_avatar_url(room)), avatar_url + ) + replication.assert_not_called() + + def test_caller_path_cannot_override_worker_configuration(self) -> None: + channel = self._request( + content={"room_avatar_path": str(self.directory / "untrusted.png")} + ) + self.assertEqual(channel.code, 200, channel.json_body) + self._assert_uploaded(channel.json_body["avatar_url"]) + + def test_unknown_announcement_is_rejected(self) -> None: + with patch.object(self.media_repo, "create_or_update_content") as upload: + channel = self._request( + "unknown", {"room_avatar_path": str(self.avatar_path)} + ) + self.assertEqual(channel.code, 404) + upload.assert_not_called() + + def test_missing_mapping_is_rejected_before_upload(self) -> None: + self.get_success( + self.store.db_pool.simple_delete( + table="server_announcement_rooms", + keyvalues={"id": "news"}, + desc="remove_test_mapping", + ) + ) + with patch.object(self.media_repo, "create_or_update_content") as upload: + channel = self._request() + self.assertEqual(channel.code, 400) + self.assertIn("no room mapping", channel.json_body["error"]) + upload.assert_not_called() + + def test_room_without_local_avatar_is_rejected(self) -> None: + room = self.worker.config.serverannouncements.rooms[0] + self.worker.config.serverannouncements.rooms = [ + attr.evolve(room, room_avatar_path=None) + ] + channel = self._request() + self.assertEqual(channel.code, 400) + self.assertIn("no room_avatar_path", channel.json_body["error"]) + + def test_unselected_media_worker_is_rejected(self) -> None: + other = self.make_worker_hs( + "synapse.app.generic_worker", + { + "worker_name": "media2", + "enable_media_repo": True, + "media_store_path": str(self.directory / "other-media"), + }, + ) + with patch.object(other, "get_media_repository") as repository: + channel = self._request(target=other) + self.assertEqual(channel.code, 400) + self.assertIn("not the configured", channel.json_body["error"]) + repository.assert_not_called() + + def test_media_disabled_on_target_is_rejected(self) -> None: + self.worker.config.media.can_load_media_repo = False + channel = self._request() + self.assertEqual(channel.code, 400) + self.assertIn("media repository enabled", channel.json_body["error"]) + + @parameterized.expand([(None,), ("wrong-secret",)]) + def test_replication_secret_is_required(self, secret: str | None) -> None: + with patch.object(self.media_repo, "create_or_update_content") as upload: + channel = self._request(secret=secret) + # Existing ReplicationEndpoint authentication failures return 500. + self.assertEqual(channel.code, 500) + upload.assert_not_called() + self.assertIsNone( + self.get_success(self.store.get_server_announcement_avatar("news")) + ) + + def test_worker_file_error_reaches_main(self) -> None: + self.avatar_path.unlink() + failure = self.get_failure(self.manager.get_avatar_url(self.room), SynapseError) + self.assertEqual(failure.value.code, 500) + self.assertTrue(failure.value.msg.startswith("Server announcement room 'news'")) + self.assertIn("Cannot read file", failure.value.msg) + self.assertIn(str(self.avatar_path), failure.value.msg) + + def test_quarantined_worker_upload_is_not_reused(self) -> None: + avatar_url = self._get_avatar() + self.get_success( + self.store.quarantine_media_by_id( + self.hs.hostname, MXCUri.from_str(avatar_url).media_id, "@admin:test" + ) + ) + with patch.object(self.media_repo, "create_or_update_content") as upload: + failure = self.get_failure( + self.manager.get_avatar_url(self.room), SynapseError + ) + self.assertIn("quarantined", failure.value.msg) + upload.assert_not_called() + + +class ServerAnnouncementAvatarRetryTestCase(HomeserverTestCase): + @parameterized.expand([(ConnectError,), (RequestTimedOutError,)]) + @override_config( + { + "enable_media_repo": False, + "media_instance_running_background_jobs": "media1", + "instance_map": {"media1": {"host": "testserv", "port": 9093}}, + } + ) + def test_network_failures_return_to_reconciler( + self, error_type: type[Exception] + ) -> None: + room = ServerAnnouncementRoom( + id="news", + room_name="News", + room_topic=None, + room_avatar_url=None, + room_avatar_path="/not-mounted-on-main/avatar.png", + allowed_user_ids=[], + ) + manager = ServerAnnouncementAvatarManager(self.hs) + with patch.object( + self.hs.get_replication_client(), + "post_json_get_json", + new_callable=AsyncMock, + ) as request: + request.side_effect = error_type("media worker unavailable") + result = ensureDeferred(manager.get_avatar_url(room)) + # Connection errors use the replication endpoint's bounded exponential + # backoff (1+2+4+8+16+32 seconds); timeouts must not loop indefinitely. + for _ in range(65): + self.reactor.advance(1) + failure = self.get_failure(result, SynapseError) + self.assertEqual(failure.value.code, 502) + if error_type is RequestTimedOutError: + request.assert_awaited_once() + else: + self.assertEqual( + request.await_count, + ReplicationServerAnnouncementAvatarRestServlet.RETRY_ON_CONNECT_ERROR_ATTEMPTS + + 2, + )