Upload and cache configured server announcement avatars

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+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}"
+        )

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

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:

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+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/<announcement_id>/<txn_id>
+
+    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)

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+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))

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+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,
+                )