Fix the config schema not passing validation with the default config

changelog.d/18817.feature

@@ -0,0 +1 @@
+Allow serving media with a redirect to an unauthenticated, short-lived, signed URL.

docker/configure_workers_and_start.py

@@ -127,6 +127,7 @@ WORKERS_CONFIG: Dict[str, Dict[str, Any]] = {
             "^/_synapse/admin/v1/quarantine_media/.*$",
             "^/_matrix/client/v1/media/.*$",
             "^/_matrix/federation/v1/media/.*$",
+            "^/_synapse/media/",
         ],
         # The first configured media worker will run the media background jobs
         "shared_extra_conf": {

docs/usage/configuration/config_documentation.md

@@ -2182,6 +2182,39 @@ media_upload_limits:
   max_size: 500M
 ```
 ---
+### `media_redirect`
+
+*(object)* When enabled, Synapse will use HTTP redirect responses to serve media instead of directly serving the media from the media store. This can help with caching, but requires /_synapse/media/* to be routed to the media worker.
+
+This setting has the following sub-options:
+
+* `enabled` (boolean): Enables the media redirect feature. If enabled, you must specify a `media_redirect.secret` or `media_redirect.secret_path`. Defaults to `false`.
+
+* `secret` (string|null): Secret used to sign media redirect URLs. This must be set if `media_redirect.enabled` is set. Defaults to `null`.
+
+* `secret_path` (string|null): An alternative to `media_redirect.secret` that specifies a file containing the secret. Defaults to `null`.
+
+* `ttl` (duration): How long the redirect URLs should be valid for. Defaults to `"10m"`.
+
+Default configuration:
+```yaml
+media_redirect:
+  enabled: false
+```
+
+Example configurations:
+```yaml
+media_redirect:
+  enabled: true
+  secret: aiCh9gu4Zahvueveisooquu7chaiw9Ee
+```
+
+```yaml
+media_redirect:
+  enabled: true
+  secret_path: /path/to/secrets/file
+```
+---
 ### `max_image_pixels`
 
 *(byte size)* Maximum number of pixels that will be thumbnailed. Defaults to `"32M"`.

docs/workers.md

@@ -765,6 +765,7 @@ Handles the media repository. It can handle all endpoints starting with:
     /_matrix/media/
     /_matrix/client/v1/media/
     /_matrix/federation/v1/media/
+    /_synapse/media/
 
 ... and the following regular expressions matching media-specific administration APIs:
 

schema/synapse-config.schema.yaml

@@ -2433,6 +2433,44 @@ properties:
           max_size: 100M
         - time_period: 1w
           max_size: 500M
+  media_redirect:
+    type: object
+    description: >-
+      When enabled, Synapse will use HTTP redirect responses to serve media
+      instead of directly serving the media from the media store. This can help
+      with caching, but requires /_synapse/media/* to be routed to the media
+      worker.
+    properties:
+      enabled:
+        type: boolean
+        description: >-
+          Enables the media redirect feature. If enabled, you must specify a
+          `media_redirect.secret` or `media_redirect.secret_path`.
+        default: false
+      secret:
+        type: ["string", "null"]
+        description: >-
+          Secret used to sign media redirect URLs. This must be set if
+          `media_redirect.enabled` is set.
+        default: null
+      secret_path:
+        type: ["string", "null"]
+        description: >-
+          An alternative to `media_redirect.secret` that specifies a file
+          containing the secret.
+        default: null
+      ttl:
+        $ref: "#/$defs/duration"
+        description: >-
+          How long the redirect URLs should be valid for.
+        default: 10m
+    default:
+      enabled: false
+    examples:
+      - enabled: true
+        secret: aiCh9gu4Zahvueveisooquu7chaiw9Ee
+      - enabled: true
+        secret_path: /path/to/secrets/file
   max_image_pixels:
     $ref: "#/$defs/bytes"
     description: Maximum number of pixels that will be thumbnailed.

synapse/config/repository.py

@@ -21,7 +21,7 @@
 
 import logging
 import os
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 import attr
 
@@ -30,7 +30,7 @@ from synapse.types import JsonDict
 from synapse.util.check_dependencies import check_requirements
 from synapse.util.module_loader import load_module
 
-from ._base import Config, ConfigError
+from ._base import Config, ConfigError, read_file
 
 logger = logging.getLogger(__name__)
 
@@ -130,7 +130,9 @@ class MediaUploadLimit:
 class ContentRepositoryConfig(Config):
     section = "media"
 
-    def read_config(self, config: JsonDict, **kwargs: Any) -> None:
+    def read_config(
+        self, config: JsonDict, allow_secrets_in_config: bool, **kwargs: Any
+    ) -> None:
         # Only enable the media repo if either the media repo is enabled or the
         # current worker app is the media repo.
         if (
@@ -290,6 +292,55 @@ class ContentRepositoryConfig(Config):
 
         self.enable_authenticated_media = config.get("enable_authenticated_media", True)
 
+        redirect_config = config.get("media_redirect", {})
+        if redirect_config is None:
+            redirect_config = {}
+        if not isinstance(redirect_config, dict):
+            raise ConfigError(
+                "`media_redirect` must be a dictionary",
+                ("media_redirect",),
+            )
+
+        # Whether we should use a redirect to /_synapse/media/* when serving
+        # media for better caching. This requires this endpoint to be routed to
+        # the media worker.
+        self.use_redirect = redirect_config.get("enabled", False)
+
+        redirect_secret = redirect_config.get("secret")
+        if redirect_secret and not allow_secrets_in_config:
+            raise ConfigError(
+                "Config options that expect an in-line secret as value are disabled",
+                ("media_redirect", "secret"),
+            )
+        if redirect_secret is not None and not isinstance(redirect_secret, str):
+            raise ConfigError(
+                "`media_redirect.secret` must be a string.",
+                ("media_redirect", "secret"),
+            )
+
+        redirect_secret_path = redirect_config.get("secret_path")
+        if redirect_secret_path:
+            if redirect_secret:
+                raise ConfigError(
+                    "You have configured both `media_redirect.secret` and `media_redirect.secret_path`.\n"
+                    "These are mutually incompatible.",
+                    ("media_redirect", "secret_path"),
+                )
+            redirect_secret = read_file(
+                redirect_secret_path, ("media_redirect", "secret_path")
+            ).strip()
+
+        self.redirect_secret: Optional[bytes] = (
+            redirect_secret.encode("utf-8") if redirect_secret else None
+        )
+
+        if self.use_redirect and self.redirect_secret is None:
+            raise ConfigError(
+                "You have configured `media_redirect.enabled` but not set `media_redirect.secret` or `media_redirect.secret_path`."
+            )
+
+        self.redirect_ttl_ms = self.parse_duration(redirect_config.get("ttl", "10m"))
+
         self.media_upload_limits: List[MediaUploadLimit] = []
         for limit_config in config.get("media_upload_limits", []):
             time_period_ms = self.parse_duration(limit_config["time_period"])

synapse/federation/transport/server/federation.py

@@ -826,7 +826,12 @@ class FederationMediaDownloadServlet(BaseFederationServerServlet):
         )
         max_timeout_ms = min(max_timeout_ms, MAXIMUM_ALLOWED_MAX_TIMEOUT_MS)
         await self.media_repo.get_local_media(
-            request, media_id, None, max_timeout_ms, federation=True
+            request,
+            media_id,
+            None,
+            max_timeout_ms,
+            federation=True,
+            may_redirect=True,
         )
 
 
@@ -873,11 +878,27 @@ class FederationMediaThumbnailServlet(BaseFederationServerServlet):
 
         if self.dynamic_thumbnails:
             await self.thumbnail_provider.select_or_generate_local_thumbnail(
-                request, media_id, width, height, method, m_type, max_timeout_ms, True
+                request,
+                media_id,
+                width,
+                height,
+                method,
+                m_type,
+                max_timeout_ms,
+                for_federation=True,
+                may_redirect=True,
             )
         else:
             await self.thumbnail_provider.respond_local_thumbnail(
-                request, media_id, width, height, method, m_type, max_timeout_ms, True
+                request,
+                media_id,
+                width,
+                height,
+                method,
+                m_type,
+                max_timeout_ms,
+                for_federation=True,
+                may_redirect=True,
             )
         self.media_repo.mark_recently_accessed(None, media_id)
 

synapse/http/server.py

@@ -972,6 +972,25 @@ def set_corp_headers(request: Request) -> None:
     request.setHeader(b"Cross-Origin-Resource-Policy", b"cross-origin")
 
 
+def set_headers_for_media_response(request: "SynapseRequest") -> None:
+    """Set the appropriate headers when serving media responses to clients"""
+    set_cors_headers(request)
+    set_corp_headers(request)
+    request.setHeader(
+        b"Content-Security-Policy",
+        b"sandbox;"
+        b" default-src 'none';"
+        b" script-src 'none';"
+        b" plugin-types application/pdf;"
+        b" style-src 'unsafe-inline';"
+        b" media-src 'self';"
+        b" object-src 'self';",
+    )
+    # Limited non-standard form of CSP for IE11
+    request.setHeader(b"X-Content-Security-Policy", b"sandbox;")
+    request.setHeader(b"Referrer-Policy", b"no-referrer")
+
+
 def respond_with_html(request: Request, code: int, html: str) -> None:
     """
     Wraps `respond_with_html_bytes` by first encoding HTML from a str to UTF-8 bytes.

synapse/http/servlet.py

@@ -714,6 +714,17 @@ def parse_strings_from_args(
         return default
 
 
+@overload
+def parse_string_from_args(
+    args: Mapping[bytes, Sequence[bytes]],
+    name: str,
+    default: str,
+    *,
+    allowed_values: Optional[StrCollection] = None,
+    encoding: str = "ascii",
+) -> str: ...
+
+
 @overload
 def parse_string_from_args(
     args: Mapping[bytes, Sequence[bytes]],

synapse/media/_base.py

@@ -36,6 +36,7 @@ from typing import (
     Tuple,
     Type,
 )
+from uuid import uuid4
 
 import attr
 from zope.interface import implementer
@@ -63,6 +64,8 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+CRLF = b"\r\n"
+
 # list all text content types that will have the charset default to UTF-8 when
 # none is given
 TEXT_CONTENT_TYPES = [
@@ -441,6 +444,35 @@ async def respond_with_responder(
     finish_request(request)
 
 
+def respond_with_multipart_location(
+    request: SynapseRequest,
+    location: bytes,
+) -> None:
+    """Responds to a media request with a multipart/mixed response, with the
+    (empty) media metadata as a JSON object in the first part, and a `Location`
+    header as the second part
+
+    Args:
+        request: The incoming request.
+        location: The URL to give in the `Location` header.
+    """
+    boundary = uuid4().hex.encode("ascii")  # Pick a random boundary
+    request.setResponseCode(200)
+    request.setHeader(
+        b"Content-Type",
+        b"multipart/mixed; boundary=" + boundary,
+    )
+    request.write(b"--" + boundary + CRLF)
+    request.write(b"Content-Type: application/json" + CRLF + CRLF)
+    # This is an empty JSON object for now, we don't have any
+    # metadata associated with media yet
+    request.write(b"{}")
+    request.write(CRLF + b"--" + boundary + CRLF)
+    request.write(b"Location: " + location + CRLF + CRLF)
+    request.write(CRLF + b"--" + boundary + b"--" + CRLF)
+    finish_request(request)
+
+
 def respond_with_304(request: SynapseRequest) -> None:
     request.setResponseCode(304)
 

synapse/media/media_repository.py

@@ -19,12 +19,17 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
+import base64
 import errno
+import hashlib
+import hmac
 import logging
 import os
 import shutil
+from http.client import TEMPORARY_REDIRECT
 from io import BytesIO
 from typing import IO, TYPE_CHECKING, Dict, List, Optional, Set, Tuple
+from urllib.parse import urlencode
 
 import attr
 from matrix_common.types.mxc_uri import MXCUri
@@ -44,7 +49,7 @@ from synapse.api.errors import (
 )
 from synapse.api.ratelimiting import Ratelimiter
 from synapse.config.repository import ThumbnailRequirement
-from synapse.http.server import respond_with_json
+from synapse.http.server import respond_with_json, respond_with_redirect
 from synapse.http.site import SynapseRequest
 from synapse.logging.context import defer_to_thread
 from synapse.logging.opentracing import trace
@@ -55,6 +60,7 @@ from synapse.media._base import (
     check_for_cached_entry_and_respond,
     get_filename_from_headers,
     respond_404,
+    respond_with_multipart_location,
     respond_with_multipart_responder,
     respond_with_responder,
 )
@@ -69,7 +75,7 @@ from synapse.media.thumbnailer import Thumbnailer, ThumbnailError
 from synapse.media.url_previewer import UrlPreviewer
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.databases.main.media_repository import LocalMedia, RemoteMedia
-from synapse.types import UserID
+from synapse.types import StrSequence, UserID
 from synapse.util.async_helpers import Linearizer
 from synapse.util.retryutils import NotRetryingDestination
 from synapse.util.stringutils import random_string
@@ -124,6 +130,8 @@ class MediaRepository:
             cfg=hs.config.ratelimiting.remote_media_downloads,
         )
 
+        self._media_request_signature_secret = hs.config.media.redirect_secret
+
         # List of StorageProviders where we should search for media and
         # potentially upload to.
         storage_providers = []
@@ -466,6 +474,7 @@ class MediaRepository:
         max_timeout_ms: int,
         allow_authenticated: bool = True,
         federation: bool = False,
+        may_redirect: bool = False,
     ) -> None:
         """Responds to requests for local media, if exists, or returns 404.
 
@@ -477,8 +486,12 @@ class MediaRepository:
                 the filename in the Content-Disposition header of the response.
             max_timeout_ms: the maximum number of milliseconds to wait for the
                 media to be uploaded.
-            allow_authenticated: whether media marked as authenticated may be served to this request
-            federation: whether the local media being fetched is for a federation request
+            allow_authenticated: whether media marked as authenticated may be
+                served to this request
+            federation: whether the local media being fetched is for a
+                federation request
+            may_redirect: whether the request may issue a redirect instead of
+                serving the media directly
 
         Returns:
             Resolves once a response has successfully been written to request
@@ -493,6 +506,19 @@ class MediaRepository:
 
         self.mark_recently_accessed(None, media_id)
 
+        if self.hs.config.media.use_redirect and may_redirect:
+            location = self.signed_location_for_media(media_id, name)
+
+            if federation:
+                respond_with_multipart_location(request, location.encode("ascii"))
+
+            else:
+                respond_with_redirect(
+                    request, location.encode("ascii"), TEMPORARY_REDIRECT
+                )
+
+            return
+
         # Once we've checked auth we can return early if the media is cached on
         # the client
         if check_for_cached_entry_and_respond(request):
@@ -1545,3 +1571,145 @@ class MediaRepository:
             removed_media.append(media_id)
 
         return removed_media, len(removed_media)
+
+    def download_media_key(
+        self,
+        media_id: str,
+        exp: int,
+        name: Optional[str] = None,
+    ) -> StrSequence:
+        """Get the key used for the download media signature
+
+        Args:
+            media_id: The media ID of the content. (This is the same as
+                the file_id for local content.)
+            exp: The expiration time of the signature, as a unix timestamp in ms.
+            name: Optional name that, if specified, will be used as
+                the filename in the Content-Disposition header of the response.
+        """
+        if name is not None:
+            return ("download", media_id, str(exp), name)
+
+        return ("download", media_id, str(exp))
+
+    def signed_location_for_media(
+        self,
+        media_id: str,
+        name: Optional[str] = None,
+    ) -> str:
+        """Get the signed location for a media download
+
+        That URL will serve the media with no extra authentication for a limited
+        time, allowing the media to be cached by a CDN more easily.
+        """
+
+        # XXX: One potential improvement here would be to round the `exp` to
+        # the nearest 5 minutes, so that a CDN/cache can always cache the
+        # media for a little bit
+        exp = self.clock.time_msec() + self.hs.config.media.redirect_ttl_ms
+        key = self.download_media_key(
+            media_id=media_id,
+            exp=exp,
+            name=name,
+        )
+        signature = self.compute_media_request_signature(key)
+
+        # This *could* in theory be a relative redirect, but Synapse has a
+        # bug where it always treats it as absolute. Because this is used
+        # for federation request, we can't just fix the bug in Synapse and
+        # use a relative redirect, we have to wait for the fix to be rolled
+        # out across the federation.
+        name_path = f"/{name}" if name else ""
+        return f"{self.hs.config.server.public_baseurl}_synapse/media/download/{media_id}{name_path}?exp={exp}&sig={signature}"
+
+    def thumbnail_media_key(
+        self, media_id: str, parameters: str, exp: int
+    ) -> StrSequence:
+        """Get the key used for the thumbnail media signature
+
+        Args:
+            media_id: The media ID of the content. (This is the same as
+                the file_id for local content.)
+            parameters: The parameters of the thumbnail request as a string,
+                e.g. "width=100&height=100"
+            exp: The expiration time of the signature, as a unix timestamp in ms.
+        """
+        return ("thumbnail", media_id, parameters, str(exp))
+
+    def signed_location_for_thumbnail(
+        self,
+        media_id: str,
+        parameters: dict[str, str],
+    ) -> str:
+        """Get the signed location for a thumbnail media
+
+        That URL will serve the media with no extra authentication for a limited
+        time, allowing the media to be cached by a CDN more easily.
+        """
+
+        # XXX: One potential improvement here would be to round the `exp` to
+        # the nearest 5 minutes, so that a CDN/cache can always cache the
+        # media for a little bit
+        exp = self.clock.time_msec() + self.hs.config.media.redirect_ttl_ms
+        parameters_str = base64.urlsafe_b64encode(
+            urlencode(sorted(parameters.items())).encode("utf-8")
+        ).decode("ascii")
+        key = self.thumbnail_media_key(
+            media_id=media_id,
+            parameters=parameters_str,
+            exp=exp,
+        )
+        signature = self.compute_media_request_signature(key)
+
+        # This *could* in theory be a relative redirect, but Synapse has a
+        # bug where it always treats it as absolute. Because this is used
+        # for federation request, we can't just fix the bug in Synapse and
+        # use a relative redirect, we have to wait for the fix to be rolled
+        # out across the federation.
+        return f"{self.hs.config.server.public_baseurl}_synapse/media/thumbnail/{media_id}/{parameters_str}?exp={exp}&sig={signature}"
+
+    def compute_media_request_signature(self, payload: StrSequence) -> str:
+        """Compute the signature for a signed media request
+
+        This currently uses a HMAC-SHA256 signature encoded as hex, but this could
+        be swapped to an asymmetric signature.
+        """
+        assert self._media_request_signature_secret is not None, (
+            "media request signature secret not set"
+        )
+
+        # XXX: alternatively, we could do multiple rounds of HMAC with the
+        # different segments, like AWS SigV4 does
+        bytes_payload = "|".join(payload).encode("utf-8")
+
+        digest = hmac.digest(
+            key=self._media_request_signature_secret,
+            msg=bytes_payload,
+            digest=hashlib.sha256,
+        )
+        signature = digest.hex()
+        return signature
+
+    def verify_media_request_signature(
+        self, payload: StrSequence, signature: str
+    ) -> bool:
+        """Verify the signature for a signed media request
+
+        Returns True if the signature is valid, False otherwise.
+        """
+        # In case there is no secret, we can't verify the signature
+        if self._media_request_signature_secret is None:
+            return False
+
+        bytes_payload = "|".join(payload).encode("utf-8")
+        decoded_signature = bytes.fromhex(signature)
+        computed_signature = hmac.digest(
+            key=self._media_request_signature_secret,
+            msg=bytes_payload,
+            digest=hashlib.sha256,
+        )
+
+        if len(computed_signature) != len(decoded_signature):
+            return False
+
+        return hmac.compare_digest(computed_signature, decoded_signature)

synapse/media/media_storage.py

@@ -59,7 +59,7 @@ from synapse.util import Clock
 from synapse.util.file_consumer import BackgroundFileConsumer
 
 from ..types import JsonDict
-from ._base import FileInfo, Responder
+from ._base import CRLF, FileInfo, Responder
 from .filepath import MediaFilePaths
 
 if TYPE_CHECKING:
@@ -68,8 +68,6 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-CRLF = b"\r\n"
-
 
 class SHA256TransparentIOWriter:
     """Will generate a SHA256 hash from a source stream transparently.

synapse/media/thumbnailer.py

@@ -20,6 +20,7 @@
 #
 #
 import logging
+from http.client import TEMPORARY_REDIRECT
 from io import BytesIO
 from types import TracebackType
 from typing import TYPE_CHECKING, List, Optional, Tuple, Type
@@ -28,7 +29,7 @@ from PIL import Image
 
 from synapse.api.errors import Codes, NotFoundError, SynapseError, cs_error
 from synapse.config.repository import THUMBNAIL_SUPPORTED_MEDIA_FORMAT_MAP
-from synapse.http.server import respond_with_json
+from synapse.http.server import respond_with_json, respond_with_redirect
 from synapse.http.site import SynapseRequest
 from synapse.logging.opentracing import trace
 from synapse.media._base import (
@@ -37,6 +38,7 @@ from synapse.media._base import (
     check_for_cached_entry_and_respond,
     respond_404,
     respond_with_file,
+    respond_with_multipart_location,
     respond_with_multipart_responder,
     respond_with_responder,
 )
@@ -269,6 +271,7 @@ class ThumbnailProvider:
         self.media_repo = media_repo
         self.media_storage = media_storage
         self.store = hs.get_datastores().main
+        self.clock = hs.get_clock()
         self.dynamic_thumbnails = hs.config.media.dynamic_thumbnails
 
     async def respond_local_thumbnail(
@@ -282,6 +285,7 @@ class ThumbnailProvider:
         max_timeout_ms: int,
         for_federation: bool,
         allow_authenticated: bool = True,
+        may_redirect: bool = False,
     ) -> None:
         media_info = await self.media_repo.get_local_media_info(
             request, media_id, max_timeout_ms
@@ -295,6 +299,26 @@ class ThumbnailProvider:
             if media_info.authenticated:
                 raise NotFoundError()
 
+        if self.hs.config.media.use_redirect and may_redirect:
+            location = self.media_repo.signed_location_for_thumbnail(
+                media_id,
+                {
+                    "width": str(width),
+                    "height": str(height),
+                    "method": method,
+                    "type": m_type,
+                },
+            )
+
+            if for_federation:
+                respond_with_multipart_location(request, location.encode("ascii"))
+            else:
+                respond_with_redirect(
+                    request, location.encode("ascii"), TEMPORARY_REDIRECT
+                )
+
+            return
+
         # Once we've checked auth we can return early if the media is cached on
         # the client
         if check_for_cached_entry_and_respond(request):
@@ -327,6 +351,7 @@ class ThumbnailProvider:
         max_timeout_ms: int,
         for_federation: bool,
         allow_authenticated: bool = True,
+        may_redirect: bool = False,
     ) -> None:
         media_info = await self.media_repo.get_local_media_info(
             request, media_id, max_timeout_ms
@@ -340,6 +365,27 @@ class ThumbnailProvider:
             if media_info.authenticated:
                 raise NotFoundError()
 
+        if self.hs.config.media.use_redirect and may_redirect:
+            location = self.media_repo.signed_location_for_thumbnail(
+                media_id,
+                {
+                    "width": str(desired_width),
+                    "height": str(desired_height),
+                    "method": desired_method,
+                    "type": desired_type,
+                },
+            )
+
+            if for_federation:
+                respond_with_multipart_location(request, location.encode("ascii"))
+
+            else:
+                respond_with_redirect(
+                    request, location.encode("ascii"), TEMPORARY_REDIRECT
+                )
+
+            return
+
         # Once we've checked auth we can return early if the media is cached on
         # the client
         if check_for_cached_entry_and_respond(request):

synapse/rest/client/media.py

@@ -30,6 +30,7 @@ from synapse.http.server import (
     respond_with_json_bytes,
     set_corp_headers,
     set_cors_headers,
+    set_headers_for_media_response,
 )
 from synapse.http.servlet import RestServlet, parse_integer, parse_string
 from synapse.http.site import SynapseRequest
@@ -169,7 +170,8 @@ class ThumbnailResource(RestServlet):
                     method,
                     m_type,
                     max_timeout_ms,
-                    False,
+                    for_federation=False,
+                    may_redirect=True,
                 )
             else:
                 await self.thumbnailer.respond_local_thumbnail(
@@ -180,7 +182,8 @@ class ThumbnailResource(RestServlet):
                     method,
                     m_type,
                     max_timeout_ms,
-                    False,
+                    for_federation=False,
+                    may_redirect=True,
                 )
             self.media_repo.mark_recently_accessed(None, media_id)
         else:
@@ -238,21 +241,7 @@ class DownloadResource(RestServlet):
 
         await self.auth.get_user_by_req(request, allow_guest=True)
 
-        set_cors_headers(request)
-        set_corp_headers(request)
-        request.setHeader(
-            b"Content-Security-Policy",
-            b"sandbox;"
-            b" default-src 'none';"
-            b" script-src 'none';"
-            b" plugin-types application/pdf;"
-            b" style-src 'unsafe-inline';"
-            b" media-src 'self';"
-            b" object-src 'self';",
-        )
-        # Limited non-standard form of CSP for IE11
-        request.setHeader(b"X-Content-Security-Policy", b"sandbox;")
-        request.setHeader(b"Referrer-Policy", b"no-referrer")
+        set_headers_for_media_response(request)
         max_timeout_ms = parse_integer(
             request, "timeout_ms", default=DEFAULT_MAX_TIMEOUT_MS
         )
@@ -260,7 +249,7 @@ class DownloadResource(RestServlet):
 
         if self._is_mine_server_name(server_name):
             await self.media_repo.get_local_media(
-                request, media_id, file_name, max_timeout_ms
+                request, media_id, file_name, max_timeout_ms, may_redirect=True
             )
         else:
             ip_address = request.getClientAddress().host

synapse/rest/media/download_resource.py

@@ -23,7 +23,9 @@ import logging
 import re
 from typing import TYPE_CHECKING, Optional
 
-from synapse.http.server import set_corp_headers, set_cors_headers
+from synapse.http.server import (
+    set_headers_for_media_response,
+)
 from synapse.http.servlet import RestServlet, parse_boolean, parse_integer
 from synapse.http.site import SynapseRequest
 from synapse.media._base import (
@@ -62,21 +64,7 @@ class DownloadResource(RestServlet):
         # Validate the server name, raising if invalid
         parse_and_validate_server_name(server_name)
 
-        set_cors_headers(request)
-        set_corp_headers(request)
-        request.setHeader(
-            b"Content-Security-Policy",
-            b"sandbox;"
-            b" default-src 'none';"
-            b" script-src 'none';"
-            b" plugin-types application/pdf;"
-            b" style-src 'unsafe-inline';"
-            b" media-src 'self';"
-            b" object-src 'self';",
-        )
-        # Limited non-standard form of CSP for IE11
-        request.setHeader(b"X-Content-Security-Policy", b"sandbox;")
-        request.setHeader(b"Referrer-Policy", b"no-referrer")
+        set_headers_for_media_response(request)
         max_timeout_ms = parse_integer(
             request, "timeout_ms", default=DEFAULT_MAX_TIMEOUT_MS
         )

synapse/rest/synapse/client/__init__.py

@@ -31,6 +31,7 @@ from synapse.rest.synapse.client.rendezvous import MSC4108RendezvousSessionResou
 from synapse.rest.synapse.client.sso_register import SsoRegisterResource
 from synapse.rest.synapse.client.unsubscribe import UnsubscribeResource
 from synapse.rest.synapse.mas import MasResource
+from synapse.rest.synapse.media import MediaResource
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -64,6 +65,9 @@ def build_synapse_client_resource_tree(hs: "HomeServer") -> Mapping[str, Resourc
         resources["/_synapse/jwks"] = JwksResource(hs)
         resources["/_synapse/mas"] = MasResource(hs)
 
+    if hs.config.media.can_load_media_repo:
+        resources["/_synapse/media"] = MediaResource(hs)
+
     # provider-specific SSO bits. Only load these if they are enabled, since they
     # rely on optional dependencies.
     if hs.config.oidc.oidc_enabled:

synapse/rest/synapse/media/__init__.py

@@ -0,0 +1,46 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, 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 logging
+from typing import TYPE_CHECKING
+
+from twisted.web.resource import Resource
+
+from .download import DownloadResource
+from .thumbnail import ThumbnailResource
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+
+logger = logging.getLogger(__name__)
+
+
+class MediaResource(Resource):
+    """
+    Provides endpoints for signed media downloads and thumbnails.
+
+    All endpoints are mounted under the path `/_synapse/media/` and only work
+    on the media worker.
+    """
+
+    def __init__(self, hs: "HomeServer"):
+        assert hs.config.media.can_load_media_repo, (
+            "This resource should only be mounted on workers that can load the media repo"
+        )
+
+        Resource.__init__(self)
+        self.putChild(b"download", DownloadResource(hs))
+        self.putChild(b"thumbnail", ThumbnailResource(hs))

synapse/rest/synapse/media/download.py

@@ -0,0 +1,111 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, 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 logging
+from typing import TYPE_CHECKING
+
+from synapse.api.errors import NotFoundError
+from synapse.http.server import (
+    DirectServeJsonResource,
+    set_headers_for_media_response,
+)
+from synapse.http.servlet import parse_integer, parse_string
+
+if TYPE_CHECKING:
+    from synapse.http.site import SynapseRequest
+    from synapse.server import HomeServer
+
+
+logger = logging.getLogger(__name__)
+
+
+class DownloadResource(DirectServeJsonResource):
+    """
+    Serves media from the media repository, with a temporary signed URL which
+    expires after a set amount of time.
+
+        GET /_synapse/media/download/{media_id}?exp={exp}&sig={sig}
+        GET /_synapse/media/download/{media_id}/{name}?exp={exp}&sig={sig}
+
+    The intent of this resource is to allow the federation and client media APIs
+    to issue redirects to a signed URL that can then be cached by a CDN. This
+    endpoint doesn't require any extra header, and is authenticated using the
+    signature in the URL parameters.
+    """
+
+    isLeaf = True
+
+    def __init__(self, hs: "HomeServer"):
+        assert hs.config.media.can_load_media_repo, (
+            "This resource should only be mounted on workers that can load the media repo"
+        )
+
+        DirectServeJsonResource.__init__(
+            self,
+            # It is useful to have the tracing context on this endpoint as it
+            # can help debug federation issues
+            extract_context=True,
+        )
+
+        self._clock = hs.get_clock()
+        self._media_repository = hs.get_media_repository()
+
+    async def _async_render_GET(self, request: "SynapseRequest") -> None:
+        set_headers_for_media_response(request)
+
+        # Extract the media ID (and optional name) from the path
+        if request.postpath is None:
+            raise NotFoundError()
+
+        if len(request.postpath) == 1:
+            media_id = request.postpath[0].decode("utf-8")
+            name = None
+        elif len(request.postpath) == 2:
+            media_id = request.postpath[0].decode("utf-8")
+            name = request.postpath[1].decode("utf-8")
+        else:
+            raise NotFoundError()
+
+        # Get the `exp` and `sig` query parameters
+        exp = parse_integer(request=request, name="exp", required=True, negative=False)
+        sig = parse_string(request=request, name="sig", required=True)
+
+        # Check that the signature is valid
+        key = self._media_repository.download_media_key(
+            media_id=media_id, exp=exp, name=name
+        )
+        if not self._media_repository.verify_media_request_signature(key, sig):
+            logger.warning(
+                "Invalid URL signature serving media %s. key: %r, sig: %r",
+                media_id,
+                key,
+                sig,
+            )
+            raise NotFoundError()
+
+        # Check the expiry time
+        if exp < self._clock.time_msec():
+            logger.info("Expired signed URL serving media %s", media_id)
+            raise NotFoundError()
+
+        # Reply with the media
+        await self._media_repository.get_local_media(
+            request=request,
+            media_id=media_id,
+            name=name,
+            max_timeout_ms=0,  # If we got here, the media finished uploading
+            federation=False,  # This changes the response to be multipart; we explicitly don't want that
+            may_redirect=False,  # We're already on the redirected URL, we don't want to redirect again
+        )

synapse/rest/synapse/media/thumbnail.py

@@ -0,0 +1,146 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, 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 base64
+import logging
+from typing import TYPE_CHECKING
+from urllib.parse import parse_qs
+
+from synapse.api.errors import NotFoundError
+from synapse.http.server import (
+    DirectServeJsonResource,
+    set_headers_for_media_response,
+)
+from synapse.http.servlet import (
+    parse_integer,
+    parse_integer_from_args,
+    parse_string,
+    parse_string_from_args,
+)
+from synapse.media.thumbnailer import ThumbnailProvider
+
+if TYPE_CHECKING:
+    from synapse.http.site import SynapseRequest
+    from synapse.server import HomeServer
+
+
+logger = logging.getLogger(__name__)
+
+
+class ThumbnailResource(DirectServeJsonResource):
+    """
+    Serves thumbnails from the media repository, with a temporary signed URL
+    which expires after a set amount of time.
+
+        GET /_synapse/media/thumbnail/{media_id}/{parameters}?exp={exp}&sig={sig}
+
+    The intent of this resource is to allow the federation and client media APIs
+    to issue redirects to a signed URL that can then be cached by a CDN. This
+    endpoint doesn't require any extra header, and is authenticated using the
+    signature in the URL parameters.
+
+    The parameters are encoded as a form-urlencoded then base64 encoded string.
+    This avoids any automatic url decoding Twisted might do. The reason they are
+    part of the URL and not the query string is to ignore the query string when
+    caching the URL, which is possible with some CDNs.
+    """
+
+    isLeaf = True
+
+    def __init__(self, hs: "HomeServer"):
+        assert hs.config.media.can_load_media_repo, (
+            "This resource should only be mounted on workers that can load the media repo"
+        )
+
+        DirectServeJsonResource.__init__(
+            self,
+            # It is useful to have the tracing context on this endpoint as it
+            # can help debug federation issues
+            extract_context=True,
+        )
+
+        self._clock = hs.get_clock()
+        self._media_repository = hs.get_media_repository()
+        self._dynamic_thumbnails = hs.config.media.dynamic_thumbnails
+        self._thumbnailer = ThumbnailProvider(
+            hs, self._media_repository, self._media_repository.media_storage
+        )
+
+    async def _async_render_GET(self, request: "SynapseRequest") -> None:
+        set_headers_for_media_response(request)
+
+        # Extract the media ID and parameters from the path
+        if request.postpath is None or len(request.postpath) != 2:
+            raise NotFoundError()
+
+        media_id = request.postpath[0].decode("utf-8")
+        parameters = request.postpath[1]
+
+        # Get the `exp` and `sig` query parameters
+        exp = parse_integer(request=request, name="exp", required=True, negative=False)
+        sig = parse_string(request=request, name="sig", required=True)
+
+        # Check that the signature is valid
+        key = self._media_repository.thumbnail_media_key(
+            media_id=media_id,
+            parameters=parameters.decode("utf-8"),
+            exp=exp,
+        )
+        if not self._media_repository.verify_media_request_signature(key, sig):
+            logger.warning(
+                "Invalid URL signature serving media %s. key: %r, sig: %r",
+                media_id,
+                key,
+                sig,
+            )
+            raise NotFoundError()
+
+        # Check the expiry time
+        if exp < self._clock.time_msec():
+            logger.info("Expired signed URL serving media %s", media_id)
+            raise NotFoundError()
+
+        # Now parse and check the parameters
+        args = parse_qs(base64.urlsafe_b64decode(parameters))
+        width = parse_integer_from_args(args, "width", required=True)
+        height = parse_integer_from_args(args, "height", required=True)
+        method = parse_string_from_args(args, "method", "scale")
+        m_type = parse_string_from_args(args, "type", "image/png")
+
+        # Reply with the thumbnail
+        if self._dynamic_thumbnails:
+            await self._thumbnailer.select_or_generate_local_thumbnail(
+                request,
+                media_id,
+                width,
+                height,
+                method,
+                m_type,
+                max_timeout_ms=0,  # If we got here, the media finished uploading
+                for_federation=False,  # This changes the response to be multipart; we explicitly don't want that
+                may_redirect=False,  # We're already on the redirected URL, we don't want to redirect again
+            )
+        else:
+            await self._thumbnailer.respond_local_thumbnail(
+                request,
+                media_id,
+                width,
+                height,
+                method,
+                m_type,
+                max_timeout_ms=0,  # If we got here, the media finished uploading
+                for_federation=False,  # This changes the response to be multipart; we explicitly don't want that
+                may_redirect=False,  # We're already on the redirected URL, we don't want to redirect again
+            )

tests/federation/test_federation_media.py

@@ -186,6 +186,72 @@ class FederationMediaDownloadsTest(unittest.FederatingHomeserverTestCase):
         self.assertEqual(channel.is_finished(), True)
         self.assertNotIn("body", channel.result)
 
+    @unittest.override_config(
+        {"media_redirect": {"enabled": True, "secret": "supersecret"}}
+    )
+    def test_signed_redirect(self) -> None:
+        """When media redirect are enabled, we should redirect to a signed URL"""
+        content = io.BytesIO(b"file_to_stream")
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "text/plain",
+                "test_upload",
+                content,
+                46,
+                UserID.from_string("@user_id:whatever.org"),
+            )
+        )
+        # test with a text file
+        channel = self.make_signed_federation_request(
+            "GET",
+            f"/_matrix/federation/v1/media/download/{content_uri.media_id}",
+        )
+        self.pump()
+        self.assertEqual(200, channel.code)
+
+        content_type = channel.headers.getRawHeaders("content-type")
+        assert content_type is not None
+        assert "multipart/mixed" in content_type[0]
+        assert "boundary" in content_type[0]
+
+        # extract boundary
+        boundary = content_type[0].split("boundary=")[1]
+        lines = channel.text_body.split("\r\n")
+
+        # Assert the structure of the multipart body line by line.
+        # Expected structure:
+        # --boundary_value
+        # Content-Type: application/json
+        #
+        # {}
+        # --boundary_value
+        # Location: signed_url
+        #
+        #
+        # --boundary_value--
+        # (potentially a final empty line if the body ends with \r\n)
+        self.assertEqual(len(lines), 10)
+
+        # Part 1: JSON metadata
+        self.assertEqual(lines[0], f"--{boundary}")
+        self.assertEqual(lines[1], "Content-Type: application/json")
+        self.assertEqual(lines[2], "")  # Empty line separating headers from body
+        self.assertEqual(lines[3], "{}")  # JSON body
+
+        # Part 2: Redirect URL
+        self.assertEqual(lines[4], f"--{boundary}")  # Boundary for the next part
+        # The Location header contains dynamic parts (exp, sig), so use regex
+        self.assertRegex(
+            lines[5],
+            rf"^Location: https://test/_synapse/media/download/{content_uri.media_id}\?exp=\d+&sig=\w+$",
+        )
+        self.assertEqual(lines[6], "")  # First empty line after Location header
+        self.assertEqual(lines[7], "")  # Second empty line after Location header
+
+        # Final boundary
+        self.assertEqual(lines[8], f"--{boundary}--")
+        self.assertEqual(lines[9], "")
+
 
 class FederationThumbnailTest(unittest.FederatingHomeserverTestCase):
     def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
@@ -293,3 +359,69 @@ class FederationThumbnailTest(unittest.FederatingHomeserverTestCase):
             small_png.expected_cropped in field for field in stripped_bytes
         )
         self.assertTrue(found_file)
+
+    @unittest.override_config(
+        {"media_redirect": {"enabled": True, "secret": "supersecret"}}
+    )
+    def test_thumbnail_signed_redirect(self) -> None:
+        """When media redirect are enabled, we should redirect to a signed URL for thumbnails"""
+        content = io.BytesIO(small_png.data)
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "image/png",
+                "test_png_thumbnail",
+                content,
+                67,
+                UserID.from_string("@user_id:whatever.org"),
+            )
+        )
+        # test with a thumbnail request
+        channel = self.make_signed_federation_request(
+            "GET",
+            f"/_matrix/federation/v1/media/thumbnail/{content_uri.media_id}?width=32&height=32&method=scale",
+        )
+        self.pump()
+        self.assertEqual(200, channel.code)
+
+        content_type = channel.headers.getRawHeaders("content-type")
+        assert content_type is not None
+        assert "multipart/mixed" in content_type[0]
+        assert "boundary" in content_type[0]
+
+        # extract boundary
+        boundary = content_type[0].split("boundary=")[1]
+        lines = channel.text_body.split("\r\n")
+
+        # Assert the structure of the multipart body line by line.
+        # Expected structure:
+        # --boundary_value
+        # Content-Type: application/json
+        #
+        # {}
+        # --boundary_value
+        # Location: signed_url
+        #
+        #
+        # --boundary_value--
+        # (potentially a final empty line if the body ends with \r\n)
+        self.assertEqual(len(lines), 10)
+
+        # Part 1: JSON metadata
+        self.assertEqual(lines[0], f"--{boundary}")
+        self.assertEqual(lines[1], "Content-Type: application/json")
+        self.assertEqual(lines[2], "")  # Empty line separating headers from body
+        self.assertEqual(lines[3], "{}")  # JSON body
+
+        # Part 2: Redirect URL
+        self.assertEqual(lines[4], f"--{boundary}")  # Boundary for the next part
+        # The Location header contains dynamic parts (exp, sig), so use regex
+        self.assertRegex(
+            lines[5],
+            rf"^Location: https://test/_synapse/media/thumbnail/{content_uri.media_id}/[^?]+\?exp=\d+&sig=\w+$",
+        )
+        self.assertEqual(lines[6], "")  # First empty line after Location header
+        self.assertEqual(lines[7], "")  # Second empty line after Location header
+
+        # Final boundary
+        self.assertEqual(lines[8], f"--{boundary}--")
+        self.assertEqual(lines[9], "")

tests/rest/client/test_media.py

@@ -2527,6 +2527,84 @@ class DownloadAndThumbnailTestCase(unittest.HomeserverTestCase):
             )
         )
 
+    @override_config({"media_redirect": {"enabled": True, "secret": "supersecret"}})
+    def test_download_signed_redirect(self) -> None:
+        """When media redirects are enabled, we should redirect to a signed URL for downloads"""
+        # Create local media
+        content = io.BytesIO(b"file_to_stream")
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "text/plain",
+                "test_upload",
+                content,
+                14,
+                UserID.from_string(f"@user:{self.hs.hostname}"),
+            )
+        )
+
+        # Request the download
+        channel = self.make_request(
+            "GET",
+            f"/_matrix/client/v1/media/download/{self.hs.hostname}/{content_uri.media_id}",
+            shorthand=False,
+            access_token=self.tok,
+        )
+
+        # Should get a redirect response
+        self.assertEqual(channel.code, 307)
+
+        # Check the Location header for the signed URL
+        location_headers = channel.headers.getRawHeaders("Location")
+        self.assertIsNotNone(location_headers)
+        assert location_headers is not None
+        self.assertEqual(len(location_headers), 1)
+        location = location_headers[0]
+
+        # Verify the signed URL format
+        self.assertRegex(
+            location,
+            rf"^https://test/_synapse/media/download/{content_uri.media_id}\?exp=\d+&sig=\w+$",
+        )
+
+    @override_config({"media_redirect": {"enabled": True, "secret": "supersecret"}})
+    def test_thumbnail_signed_redirect(self) -> None:
+        """When media redirects are enabled, we should redirect to a signed URL for thumbnails (scaled)"""
+        # Create local media with an image
+        content = io.BytesIO(small_png.data)
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "image/png",
+                "test_png_thumbnail",
+                content,
+                67,
+                UserID.from_string(f"@user:{self.hs.hostname}"),
+            )
+        )
+
+        # Request a scaled thumbnail
+        channel = self.make_request(
+            "GET",
+            f"/_matrix/client/v1/media/thumbnail/{self.hs.hostname}/{content_uri.media_id}?width=32&height=32",
+            shorthand=False,
+            access_token=self.tok,
+        )
+
+        # Should get a redirect response
+        self.assertEqual(channel.code, 307)
+
+        # Check the Location header for the signed URL
+        location_headers = channel.headers.getRawHeaders("Location")
+        self.assertIsNotNone(location_headers)
+        assert location_headers is not None
+        self.assertEqual(len(location_headers), 1)
+        location = location_headers[0]
+
+        # Verify the signed URL format
+        self.assertRegex(
+            location,
+            rf"^https://test/_synapse/media/thumbnail/{content_uri.media_id}/[^?]+\?exp=\d+&sig=\w+$",
+        )
+
 
 configs = [
     {"extra_config": {"dynamic_thumbnails": True}},

tests/rest/synapse/media/__init__.py

@@ -0,0 +1,13 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, 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>.
+#

tests/rest/synapse/media/test_download.py

@@ -0,0 +1,282 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, 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 io
+from typing import Dict
+
+from twisted.internet.testing import MemoryReactor
+from twisted.web.resource import Resource
+
+from synapse.rest.synapse.client import build_synapse_client_resource_tree
+from synapse.server import HomeServer
+from synapse.types import UserID
+from synapse.util import Clock
+
+from tests import unittest
+
+
+class SignedDownloadTestCase(unittest.HomeserverTestCase):
+    def create_resource_dict(self) -> Dict[str, Resource]:
+        d = super().create_resource_dict()
+        d.update(build_synapse_client_resource_tree(self.hs))
+        return d
+
+    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
+        config = self.default_config()
+        config["media_redirect"] = {
+            "enabled": True,
+            "secret": "supersecret",
+        }
+
+        return self.setup_test_homeserver(config=config)
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        super().prepare(reactor, clock, hs)
+        self.media_repo = hs.get_media_repository()
+
+    def test_valid_signed_download(self) -> None:
+        """Test that a valid signed URL returns the media content"""
+        # Create test content
+        content = io.BytesIO(b"test file content")
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "text/plain",
+                "some_name.txt",
+                content,
+                17,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Generate a signed URL
+        exp = self.clock.time_msec() + 3600000  # 1 hour from now
+        key = self.media_repo.download_media_key(
+            media_id=content_uri.media_id, exp=exp, name="test_file.txt"
+        )
+        sig = self.media_repo.compute_media_request_signature(key)
+
+        # Make the request
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/download/{content_uri.media_id}/test_file.txt?exp={exp}&sig={sig}",
+            shorthand=False,
+        )
+
+        # Check the response
+        self.assertEqual(channel.code, 200)
+        self.assertEqual(channel.result["body"], b"test file content")
+
+        # Check content type
+        content_type = channel.headers.getRawHeaders("Content-Type")
+        assert content_type is not None
+        self.assertIn("text/plain", content_type[0])
+
+        # Check content disposition
+        content_disposition = channel.headers.getRawHeaders("Content-Disposition")
+        assert content_disposition is not None
+        self.assertIn("test_file.txt", content_disposition[0])
+
+    def test_valid_signed_download_without_filename(self) -> None:
+        """Test that a valid signed URL works without a filename"""
+        # Create test content
+        content = io.BytesIO(b"test file content")
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "text/plain",
+                "test_file.txt",
+                content,
+                17,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Generate a signed URL without filename
+        exp = self.clock.time_msec() + 3600000  # 1 hour from now
+        key = self.media_repo.download_media_key(
+            media_id=content_uri.media_id, exp=exp, name=None
+        )
+        sig = self.media_repo.compute_media_request_signature(key)
+
+        # Make the request
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/download/{content_uri.media_id}?exp={exp}&sig={sig}",
+            shorthand=False,
+        )
+
+        # Check the response
+        self.assertEqual(channel.code, 200)
+        self.assertEqual(channel.result["body"], b"test file content")
+
+    def test_invalid_signature(self) -> None:
+        """Test that an invalid signature returns 404"""
+        # Create test content
+        content = io.BytesIO(b"test file content")
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "text/plain",
+                "test_file.txt",
+                content,
+                17,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Use a properly formatted but invalid signature (64 hex chars like a real signature)
+        exp = self.clock.time_msec() + 3600000  # 1 hour from now
+        invalid_sig = "0" * 64  # Invalid but properly formatted signature
+
+        # Make the request
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/download/{content_uri.media_id}?exp={exp}&sig={invalid_sig}",
+            shorthand=False,
+        )
+
+        # Check the response
+        self.assertEqual(channel.code, 404)
+
+    def test_expired_url(self) -> None:
+        """Test that an expired URL returns 404"""
+        # Create test content
+        content = io.BytesIO(b"test file content")
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "text/plain",
+                "test_file.txt",
+                content,
+                17,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Generate a signed URL that will expire soon
+        exp = self.clock.time_msec() + 1000  # 1 second from now
+        key = self.media_repo.download_media_key(
+            media_id=content_uri.media_id, exp=exp, name=None
+        )
+        sig = self.media_repo.compute_media_request_signature(key)
+
+        # Advance the clock to make the URL expired
+        self.reactor.advance(2)  # Advance 2 seconds
+
+        # Make the request
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/download/{content_uri.media_id}?exp={exp}&sig={sig}",
+            shorthand=False,
+        )
+
+        # Check the response
+        self.assertEqual(channel.code, 404)
+
+    def test_missing_parameters(self) -> None:
+        """Test that missing exp or sig parameters return 404"""
+        # Create test content
+        content = io.BytesIO(b"test file content")
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "text/plain",
+                "test_file.txt",
+                content,
+                17,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Test missing exp parameter
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/download/{content_uri.media_id}?sig=somesig",
+            shorthand=False,
+        )
+        self.assertEqual(channel.code, 400)  # Bad request for missing required param
+
+        # Test missing sig parameter
+        exp = self.clock.time_msec() + 3600000
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/download/{content_uri.media_id}?exp={exp}",
+            shorthand=False,
+        )
+        self.assertEqual(channel.code, 400)  # Bad request for missing required param
+
+    def test_nonexistent_media(self) -> None:
+        """Test that requesting non-existent media returns 404"""
+        # Generate a signed URL for non-existent media
+        fake_media_id = "nonexistent"
+        exp = self.clock.time_msec() + 3600000  # 1 hour from now
+        key = self.media_repo.download_media_key(
+            media_id=fake_media_id, exp=exp, name=None
+        )
+        sig = self.media_repo.compute_media_request_signature(key)
+
+        # Make the request
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/download/{fake_media_id}?exp={exp}&sig={sig}",
+            shorthand=False,
+        )
+
+        # Check the response
+        self.assertEqual(channel.code, 404)
+
+    def test_etag_functionality(self) -> None:
+        """Test that ETag functionality works properly"""
+        # Create test content
+        content = io.BytesIO(b"test file content for etag")
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "text/plain",
+                "test_file.txt",
+                content,
+                26,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Generate a signed URL
+        exp = self.clock.time_msec() + 3600000  # 1 hour from now
+        key = self.media_repo.download_media_key(
+            media_id=content_uri.media_id, exp=exp, name=None
+        )
+        sig = self.media_repo.compute_media_request_signature(key)
+
+        # Make the first request
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/download/{content_uri.media_id}?exp={exp}&sig={sig}",
+            shorthand=False,
+        )
+
+        # Check the response has an ETag and a Cache-Control header
+        self.assertEqual(channel.code, 200)
+        etag_headers = channel.headers.getRawHeaders("ETag")
+        assert etag_headers is not None
+        etag = etag_headers[0]
+        cache_control = channel.headers.getRawHeaders("Cache-Control")
+        self.assertIsNotNone(cache_control)
+
+        # Make a second request with If-None-Match header
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/download/{content_uri.media_id}?exp={exp}&sig={sig}",
+            shorthand=False,
+            custom_headers=[("If-None-Match", etag)],
+        )
+
+        # Should get 304 Not Modified
+        self.assertEqual(channel.code, 304)
+        self.assertNotIn("body", channel.result)

tests/rest/synapse/media/test_thumbnail.py

@@ -0,0 +1,356 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2025 New Vector, 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 io
+import re
+from typing import Dict
+from urllib.parse import urlencode
+
+from twisted.internet.testing import MemoryReactor
+from twisted.web.resource import Resource
+
+from synapse.rest.synapse.client import build_synapse_client_resource_tree
+from synapse.server import HomeServer
+from synapse.types import UserID
+from synapse.util import Clock
+
+from tests import unittest
+from tests.media.test_media_storage import small_png
+
+
+class SignedThumbnailTestCase(unittest.HomeserverTestCase):
+    def create_resource_dict(self) -> Dict[str, Resource]:
+        d = super().create_resource_dict()
+        d.update(build_synapse_client_resource_tree(self.hs))
+        return d
+
+    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
+        config = self.default_config()
+        config["media_redirect"] = {
+            "enabled": True,
+            "secret": "supersecret",
+        }
+        config["dynamic_thumbnails"] = True
+
+        return self.setup_test_homeserver(config=config)
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        super().prepare(reactor, clock, hs)
+        self.media_repo = hs.get_media_repository()
+
+    def test_valid_signed_thumbnail_scaled(self) -> None:
+        """Test that a valid signed URL returns the thumbnail content (scaled)"""
+        # Create test content with an image
+        content = io.BytesIO(small_png.data)
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "image/png",
+                "test_image.png",
+                content,
+                67,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Generate a signed URL for scaled thumbnail
+        params_dict = {
+            "width": "32",
+            "height": "32",
+            "method": "scale",
+            "type": "image/png",
+        }
+        signed_url = self.media_repo.signed_location_for_thumbnail(
+            media_id=content_uri.media_id, parameters=params_dict
+        )
+        # Extract the path and query from the signed URL
+        url_path = signed_url.split("https://test", 1)[1]
+
+        # Make the request
+        channel = self.make_request(
+            "GET",
+            url_path,
+            shorthand=False,
+        )
+        self.pump()
+
+        # Check the response
+        self.assertEqual(channel.code, 200)
+
+        # Check content type
+        content_type = channel.headers.getRawHeaders("Content-Type")
+        assert content_type is not None
+        self.assertEqual(content_type[0], "image/png")
+
+        # Check that we got actual thumbnail data
+        self.assertIsNotNone(channel.result.get("body"))
+        self.assertGreater(len(channel.result["body"]), 0)
+
+    def test_valid_signed_thumbnail_cropped(self) -> None:
+        """Test that a valid signed URL returns the thumbnail content (cropped)"""
+        # Create test content with an image
+        content = io.BytesIO(small_png.data)
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "image/png",
+                "test_image.png",
+                content,
+                67,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Generate a signed URL for cropped thumbnail
+        params_dict = {
+            "width": "32",
+            "height": "32",
+            "method": "crop",
+            "type": "image/png",
+        }
+        signed_url = self.media_repo.signed_location_for_thumbnail(
+            media_id=content_uri.media_id, parameters=params_dict
+        )
+        # Extract the path and query from the signed URL
+        url_path = signed_url.split("https://test", 1)[1]
+
+        # Make the request
+        channel = self.make_request(
+            "GET",
+            url_path,
+            shorthand=False,
+        )
+        self.pump()
+
+        # Check the response
+        self.assertEqual(channel.code, 200)
+
+        # Check content type
+        content_type = channel.headers.getRawHeaders("Content-Type")
+        assert content_type is not None
+        self.assertEqual(content_type[0], "image/png")
+
+        # Check that we got actual thumbnail data
+        self.assertIsNotNone(channel.result.get("body"))
+        self.assertGreater(len(channel.result["body"]), 0)
+
+    def test_invalid_signature(self) -> None:
+        """Test that an invalid signature returns 404"""
+        # Create test content with an image
+        content = io.BytesIO(small_png.data)
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "image/png",
+                "test_image.png",
+                content,
+                67,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Generate a signed URL
+        params_dict = {
+            "width": "32",
+            "height": "32",
+        }
+        signed_url = self.media_repo.signed_location_for_thumbnail(
+            media_id=content_uri.media_id, parameters=params_dict
+        )
+
+        # Extract the path and query from the signed URL
+        url_path = signed_url.split("https://test", 1)[1]
+        invalid_sig = "0" * 64  # Invalid but properly formatted signature
+        url_path = re.sub(r"sig=\w+", "sig=" + invalid_sig, url_path)
+
+        # Make the request
+        channel = self.make_request(
+            "GET",
+            url_path,
+            shorthand=False,
+        )
+        self.pump()
+
+        # Check the response
+        self.assertEqual(channel.code, 404)
+
+    def test_expired_url(self) -> None:
+        """Test that an expired URL returns 404"""
+        # Create test content with an image
+        content = io.BytesIO(small_png.data)
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "image/png",
+                "test_image.png",
+                content,
+                67,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Generate a signed URL
+        params_dict = {
+            "width": "32",
+            "height": "32",
+        }
+        signed_url = self.media_repo.signed_location_for_thumbnail(
+            media_id=content_uri.media_id, parameters=params_dict
+        )
+        # Extract the path and query from the signed URL
+        url_path = signed_url.split("https://test", 1)[1]
+
+        # Make a first request, it should work
+        channel = self.make_request(
+            "GET",
+            url_path,
+            shorthand=False,
+        )
+        self.pump()
+
+        # Check the response
+        self.assertEqual(channel.code, 200)
+
+        # Advance the clock to make the URL expired
+        self.reactor.advance(
+            10 * 60 + 1
+        )  # Advance 10 minutes + 1 second (TTL is 10 minutes by default)
+
+        # Make a second request, it should fail
+        channel = self.make_request(
+            "GET",
+            url_path,
+            shorthand=False,
+        )
+        self.pump()
+
+        # Check the response
+        self.assertEqual(channel.code, 404)
+
+    def test_missing_parameters(self) -> None:
+        """Test that missing exp or sig parameters return 400"""
+        # Create test content with an image
+        content = io.BytesIO(small_png.data)
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "image/png",
+                "test_image.png",
+                content,
+                67,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        params_dict = {
+            "width": "32",
+            "height": "32",
+            "method": "scale",
+            "type": "image/png",
+        }
+        parameters = urlencode(params_dict)
+
+        # Test missing exp parameter
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/thumbnail/{content_uri.media_id}/{parameters}?sig=somesig",
+            shorthand=False,
+        )
+        self.pump()
+        self.assertEqual(channel.code, 400)  # Bad request for missing required param
+
+        # Test missing sig parameter
+        exp = self.clock.time_msec() + 3600000
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/media/thumbnail/{content_uri.media_id}/{parameters}?exp={exp}",
+            shorthand=False,
+        )
+        self.pump()
+        self.assertEqual(channel.code, 400)  # Bad request for missing required param
+
+    def test_nonexistent_media(self) -> None:
+        """Test that requesting non-existent media returns 404"""
+        # Generate a signed URL for non-existent media
+        fake_media_id = "nonexistent"
+        params_dict = {
+            "width": "32",
+            "height": "32",
+        }
+        signed_url = self.media_repo.signed_location_for_thumbnail(
+            media_id=fake_media_id, parameters=params_dict
+        )
+        # Extract the path and query from the signed URL
+        url_path = signed_url.split("https://test", 1)[1]
+
+        # Make the request
+        channel = self.make_request(
+            "GET",
+            url_path,
+            shorthand=False,
+        )
+        self.pump()
+
+        # Check the response
+        self.assertEqual(channel.code, 404)
+
+    def test_etag_functionality(self) -> None:
+        """Test that ETag functionality works properly"""
+        # Create test content with an image
+        content = io.BytesIO(small_png.data)
+        content_uri = self.get_success(
+            self.media_repo.create_or_update_content(
+                "image/png",
+                "test_image.png",
+                content,
+                67,
+                UserID.from_string("@user:test"),
+            )
+        )
+
+        # Generate a signed URL
+        params_dict = {
+            "width": "32",
+            "height": "32",
+        }
+        signed_url = self.media_repo.signed_location_for_thumbnail(
+            media_id=content_uri.media_id, parameters=params_dict
+        )
+        # Extract the path and query from the signed URL
+        url_path = signed_url.split("https://test", 1)[1]
+
+        # Make a first request
+        channel = self.make_request(
+            "GET",
+            url_path,
+            shorthand=False,
+        )
+        self.pump()
+
+        # Check the response has an ETag and a Cache-Control header
+        self.assertEqual(channel.code, 200)
+        etag_headers = channel.headers.getRawHeaders("ETag")
+        assert etag_headers is not None
+        etag = etag_headers[0]
+        cache_control = channel.headers.getRawHeaders("Cache-Control")
+        self.assertIsNotNone(cache_control)
+
+        # Make a second request with If-None-Match header
+        channel = self.make_request(
+            "GET",
+            url_path,
+            shorthand=False,
+            custom_headers=[("If-None-Match", etag)],
+        )
+        self.pump()
+
+        # Should get 304 Not Modified
+        self.assertEqual(channel.code, 304)
+        self.assertNotIn("body", channel.result)