diff options
Diffstat (limited to 'synapse')
| -rw-r--r-- | synapse/federation/transport/server/__init__.py | 24 | ||||
| -rw-r--r-- | synapse/federation/transport/server/_base.py | 24 | ||||
| -rw-r--r-- | synapse/federation/transport/server/federation.py | 41 | ||||
| -rw-r--r-- | synapse/media/_base.py | 63 | ||||
| -rw-r--r-- | synapse/media/media_repository.py | 18 | ||||
| -rw-r--r-- | synapse/media/media_storage.py | 223 | ||||
| -rw-r--r-- | synapse/media/storage_provider.py | 40 |
7 files changed, 22 insertions, 411 deletions
diff --git a/synapse/federation/transport/server/__init__.py b/synapse/federation/transport/server/__init__.py index 266675c9b8..bac569e977 100644 --- a/synapse/federation/transport/server/__init__.py +++ b/synapse/federation/transport/server/__init__.py @@ -19,7 +19,6 @@ # [This file includes modifications made by New Vector Limited] # # -import inspect import logging from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Tuple, Type @@ -34,7 +33,6 @@ from synapse.federation.transport.server.federation import ( FEDERATION_SERVLET_CLASSES, FederationAccountStatusServlet, FederationUnstableClientKeysClaimServlet, - FederationUnstableMediaDownloadServlet, ) from synapse.http.server import HttpServer, JsonResource from synapse.http.servlet import ( @@ -317,28 +315,6 @@ def register_servlets( ): continue - if servletclass == FederationUnstableMediaDownloadServlet: - if ( - not hs.config.server.enable_media_repo - or not hs.config.experimental.msc3916_authenticated_media_enabled - ): - continue - - # don't load the endpoint if the storage provider is incompatible - media_repo = hs.get_media_repository() - load_download_endpoint = True - for provider in media_repo.media_storage.storage_providers: - signature = inspect.signature(provider.backend.fetch) - if "federation" not in signature.parameters: - logger.warning( - f"Federation media `/download` endpoint will not be enabled as storage provider {provider.backend} is not compatible with this endpoint." - ) - load_download_endpoint = False - break - - if not load_download_endpoint: - continue - servletclass( hs=hs, authenticator=authenticator, diff --git a/synapse/federation/transport/server/_base.py b/synapse/federation/transport/server/_base.py index 4e2717b565..db0f5076a9 100644 --- a/synapse/federation/transport/server/_base.py +++ b/synapse/federation/transport/server/_base.py @@ -360,29 +360,13 @@ class BaseFederationServlet: "request" ) return None - if ( - func.__self__.__class__.__name__ # type: ignore - == "FederationUnstableMediaDownloadServlet" - ): - response = await func( - origin, content, request, *args, **kwargs - ) - else: - response = await func( - origin, content, request.args, *args, **kwargs - ) - else: - if ( - func.__self__.__class__.__name__ # type: ignore - == "FederationUnstableMediaDownloadServlet" - ): - response = await func( - origin, content, request, *args, **kwargs - ) - else: response = await func( origin, content, request.args, *args, **kwargs ) + else: + response = await func( + origin, content, request.args, *args, **kwargs + ) finally: # if we used the origin's context as the parent, add a new span using # the servlet span as a parent, so that we have a link diff --git a/synapse/federation/transport/server/federation.py b/synapse/federation/transport/server/federation.py index 1f02451efa..a59734785f 100644 --- a/synapse/federation/transport/server/federation.py +++ b/synapse/federation/transport/server/federation.py @@ -44,13 +44,10 @@ from synapse.federation.transport.server._base import ( ) from synapse.http.servlet import ( parse_boolean_from_args, - parse_integer, parse_integer_from_args, parse_string_from_args, parse_strings_from_args, ) -from synapse.http.site import SynapseRequest -from synapse.media._base import DEFAULT_MAX_TIMEOUT_MS, MAXIMUM_ALLOWED_MAX_TIMEOUT_MS from synapse.types import JsonDict from synapse.util import SYNAPSE_VERSION from synapse.util.ratelimitutils import FederationRateLimiter @@ -790,43 +787,6 @@ class FederationAccountStatusServlet(BaseFederationServerServlet): return 200, {"account_statuses": statuses, "failures": failures} -class FederationUnstableMediaDownloadServlet(BaseFederationServerServlet): - """ - Implementation of new federation media `/download` endpoint outlined in MSC3916. Returns - a multipart/form-data response consisting of a JSON object and the requested media - item. This endpoint only returns local media. - """ - - PATH = "/media/download/(?P<media_id>[^/]*)" - PREFIX = FEDERATION_UNSTABLE_PREFIX + "/org.matrix.msc3916" - RATELIMIT = True - - def __init__( - self, - hs: "HomeServer", - ratelimiter: FederationRateLimiter, - authenticator: Authenticator, - server_name: str, - ): - super().__init__(hs, authenticator, ratelimiter, server_name) - self.media_repo = self.hs.get_media_repository() - - async def on_GET( - self, - origin: Optional[str], - content: Literal[None], - request: SynapseRequest, - media_id: str, - ) -> None: - max_timeout_ms = parse_integer( - request, "timeout_ms", default=DEFAULT_MAX_TIMEOUT_MS - ) - 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 - ) - - FEDERATION_SERVLET_CLASSES: Tuple[Type[BaseFederationServlet], ...] = ( FederationSendServlet, FederationEventServlet, @@ -858,5 +818,4 @@ FEDERATION_SERVLET_CLASSES: Tuple[Type[BaseFederationServlet], ...] = ( FederationV1SendKnockServlet, FederationMakeKnockServlet, FederationAccountStatusServlet, - FederationUnstableMediaDownloadServlet, ) diff --git a/synapse/media/_base.py b/synapse/media/_base.py index 19bca94170..3fbed6062f 100644 --- a/synapse/media/_base.py +++ b/synapse/media/_base.py @@ -25,16 +25,7 @@ import os import urllib from abc import ABC, abstractmethod from types import TracebackType -from typing import ( - TYPE_CHECKING, - Awaitable, - Dict, - Generator, - List, - Optional, - Tuple, - Type, -) +from typing import Awaitable, Dict, Generator, List, Optional, Tuple, Type import attr @@ -48,11 +39,6 @@ from synapse.http.site import SynapseRequest from synapse.logging.context import make_deferred_yieldable from synapse.util.stringutils import is_ascii -if TYPE_CHECKING: - from synapse.media.media_storage import MultipartResponder - from synapse.storage.databases.main.media_repository import LocalMedia - - logger = logging.getLogger(__name__) # list all text content types that will have the charset default to UTF-8 when @@ -274,53 +260,6 @@ def _can_encode_filename_as_token(x: str) -> bool: return True -async def respond_with_multipart_responder( - request: SynapseRequest, - responder: "Optional[MultipartResponder]", - media_info: "LocalMedia", -) -> None: - """ - Responds via a Multipart responder for the federation media `/download` requests - - Args: - request: the federation request to respond to - responder: the Multipart responder which will send the response - media_info: metadata about the media item - """ - if not responder: - respond_404(request) - return - - # If we have a responder we *must* use it as a context manager. - with responder: - if request._disconnected: - logger.warning( - "Not sending response to request %s, already disconnected.", request - ) - return - - logger.debug("Responding to media request with responder %s", responder) - if media_info.media_length is not None: - request.setHeader(b"Content-Length", b"%d" % (media_info.media_length,)) - request.setHeader( - b"Content-Type", b"multipart/mixed; boundary=%s" % responder.boundary - ) - - try: - await responder.write_to_consumer(request) - except Exception as e: - # The majority of the time this will be due to the client having gone - # away. Unfortunately, Twisted simply throws a generic exception at us - # in that case. - logger.warning("Failed to write to consumer: %s %s", type(e), e) - - # Unregister the producer, if it has one, so Twisted doesn't complain - if request.producer: - request.unregisterProducer() - - finish_request(request) - - async def respond_with_responder( request: SynapseRequest, responder: "Optional[Responder]", diff --git a/synapse/media/media_repository.py b/synapse/media/media_repository.py index c335e518a0..6ed56099ca 100644 --- a/synapse/media/media_repository.py +++ b/synapse/media/media_repository.py @@ -54,11 +54,10 @@ from synapse.media._base import ( ThumbnailInfo, get_filename_from_headers, respond_404, - respond_with_multipart_responder, respond_with_responder, ) from synapse.media.filepath import MediaFilePaths -from synapse.media.media_storage import MediaStorage, MultipartResponder +from synapse.media.media_storage import MediaStorage from synapse.media.storage_provider import StorageProviderWrapper from synapse.media.thumbnailer import Thumbnailer, ThumbnailError from synapse.media.url_previewer import UrlPreviewer @@ -430,7 +429,6 @@ class MediaRepository: media_id: str, name: Optional[str], max_timeout_ms: int, - federation: bool = False, ) -> None: """Responds to requests for local media, if exists, or returns 404. @@ -442,7 +440,6 @@ 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. - federation: whether the local media being fetched is for a federation request Returns: Resolves once a response has successfully been written to request @@ -462,17 +459,10 @@ class MediaRepository: file_info = FileInfo(None, media_id, url_cache=bool(url_cache)) - responder = await self.media_storage.fetch_media( - file_info, media_info, federation + responder = await self.media_storage.fetch_media(file_info) + await respond_with_responder( + request, responder, media_type, media_length, upload_name ) - if federation: - # this really should be a Multipart responder but just in case - assert isinstance(responder, MultipartResponder) - await respond_with_multipart_responder(request, responder, media_info) - else: - await respond_with_responder( - request, responder, media_type, media_length, upload_name - ) async def get_remote_media( self, diff --git a/synapse/media/media_storage.py b/synapse/media/media_storage.py index 2f55d12b6b..b3cd3fd8f4 100644 --- a/synapse/media/media_storage.py +++ b/synapse/media/media_storage.py @@ -19,12 +19,9 @@ # # import contextlib -import json import logging import os import shutil -from contextlib import closing -from io import BytesIO from types import TracebackType from typing import ( IO, @@ -33,19 +30,14 @@ from typing import ( AsyncIterator, BinaryIO, Callable, - List, Optional, Sequence, Tuple, Type, - Union, ) -from uuid import uuid4 import attr -from zope.interface import implementer -from twisted.internet import defer, interfaces from twisted.internet.defer import Deferred from twisted.internet.interfaces import IConsumer from twisted.protocols.basic import FileSender @@ -56,19 +48,15 @@ from synapse.logging.opentracing import start_active_span, trace, trace_with_opn from synapse.util import Clock from synapse.util.file_consumer import BackgroundFileConsumer -from ..storage.databases.main.media_repository import LocalMedia -from ..types import JsonDict from ._base import FileInfo, Responder from .filepath import MediaFilePaths if TYPE_CHECKING: - from synapse.media.storage_provider import StorageProviderWrapper + from synapse.media.storage_provider import StorageProvider from synapse.server import HomeServer logger = logging.getLogger(__name__) -CRLF = b"\r\n" - class MediaStorage: """Responsible for storing/fetching files from local sources. @@ -85,7 +73,7 @@ class MediaStorage: hs: "HomeServer", local_media_directory: str, filepaths: MediaFilePaths, - storage_providers: Sequence["StorageProviderWrapper"], + storage_providers: Sequence["StorageProvider"], ): self.hs = hs self.reactor = hs.get_reactor() @@ -181,23 +169,15 @@ class MediaStorage: raise e from None - async def fetch_media( - self, - file_info: FileInfo, - media_info: Optional[LocalMedia] = None, - federation: bool = False, - ) -> Optional[Responder]: + async def fetch_media(self, file_info: FileInfo) -> Optional[Responder]: """Attempts to fetch media described by file_info from the local cache and configured storage providers. Args: - file_info: Metadata about the media file - media_info: Metadata about the media item - federation: Whether this file is being fetched for a federation request + file_info Returns: - If the file was found returns a Responder (a Multipart Responder if the requested - file is for the federation /download endpoint), otherwise None. + Returns a Responder if the file was found, otherwise None. """ paths = [self._file_info_to_path(file_info)] @@ -217,19 +197,12 @@ class MediaStorage: local_path = os.path.join(self.local_media_directory, path) if os.path.exists(local_path): logger.debug("responding with local file %s", local_path) - if federation: - assert media_info is not None - boundary = uuid4().hex.encode("ascii") - return MultipartResponder( - open(local_path, "rb"), media_info, boundary - ) - else: - return FileResponder(open(local_path, "rb")) + return FileResponder(open(local_path, "rb")) logger.debug("local file %s did not exist", local_path) for provider in self.storage_providers: for path in paths: - res: Any = await provider.fetch(path, file_info, media_info, federation) + res: Any = await provider.fetch(path, file_info) if res: logger.debug("Streaming %s from %s", path, provider) return res @@ -343,7 +316,7 @@ class FileResponder(Responder): """Wraps an open file that can be sent to a request. Args: - open_file: A file like object to be streamed to the client, + open_file: A file like object to be streamed ot the client, is closed when finished streaming. """ @@ -364,38 +337,6 @@ class FileResponder(Responder): self.open_file.close() -class MultipartResponder(Responder): - """Wraps an open file, formats the response according to MSC3916 and sends it to a - federation request. - - Args: - open_file: A file like object to be streamed to the client, - is closed when finished streaming. - media_info: metadata about the media item - boundary: bytes to use for the multipart response boundary - """ - - def __init__(self, open_file: IO, media_info: LocalMedia, boundary: bytes) -> None: - self.open_file = open_file - self.media_info = media_info - self.boundary = boundary - - def write_to_consumer(self, consumer: IConsumer) -> Deferred: - return make_deferred_yieldable( - MultipartFileSender().beginFileTransfer( - self.open_file, consumer, self.media_info.media_type, {}, self.boundary - ) - ) - - def __exit__( - self, - exc_type: Optional[Type[BaseException]], - exc_val: Optional[BaseException], - exc_tb: Optional[TracebackType], - ) -> None: - self.open_file.close() - - class SpamMediaException(NotFoundError): """The media was blocked by a spam checker, so we simply 404 the request (in the same way as if it was quarantined). @@ -429,151 +370,3 @@ class ReadableFileWrapper: # We yield to the reactor by sleeping for 0 seconds. await self.clock.sleep(0) - - -@implementer(interfaces.IProducer) -class MultipartFileSender: - """ - A producer that sends the contents of a file to a federation request in the format - outlined in MSC3916 - a multipart/format-data response where the first field is a - JSON object and the second is the requested file. - - This is a slight re-writing of twisted.protocols.basic.FileSender to achieve the format - outlined above. - """ - - CHUNK_SIZE = 2**14 - - lastSent = "" - deferred: Optional[defer.Deferred] = None - - def beginFileTransfer( - self, - file: IO, - consumer: IConsumer, - file_content_type: str, - json_object: JsonDict, - boundary: bytes, - ) -> Deferred: - """ - Begin transferring a file - - Args: - file: The file object to read data from - consumer: The synapse request to write the data to - file_content_type: The content-type of the file - json_object: The JSON object to write to the first field of the response - boundary: bytes to be used as the multipart/form-data boundary - - Returns: A deferred whose callback will be invoked when the file has - been completely written to the consumer. The last byte written to the - consumer is passed to the callback. - """ - self.file: Optional[IO] = file - self.consumer = consumer - self.json_field = json_object - self.json_field_written = False - self.content_type_written = False - self.file_content_type = file_content_type - self.boundary = boundary - self.deferred: Deferred = defer.Deferred() - self.consumer.registerProducer(self, False) - # while it's not entirely clear why this assignment is necessary, it mirrors - # the behavior in FileSender.beginFileTransfer and thus is preserved here - deferred = self.deferred - return deferred - - def resumeProducing(self) -> None: - # write the first field, which will always be a json field - if not self.json_field_written: - self.consumer.write(CRLF + b"--" + self.boundary + CRLF) - - content_type = Header(b"Content-Type", b"application/json") - self.consumer.write(bytes(content_type) + CRLF) - - json_field = json.dumps(self.json_field) - json_bytes = json_field.encode("utf-8") - self.consumer.write(json_bytes) - self.consumer.write(CRLF + b"--" + self.boundary + CRLF) - - self.json_field_written = True - - chunk: Any = "" - if self.file: - # if we haven't written the content type yet, do so - if not self.content_type_written: - type = self.file_content_type.encode("utf-8") - content_type = Header(b"Content-Type", type) - self.consumer.write(bytes(content_type) + CRLF) - self.content_type_written = True - - chunk = self.file.read(self.CHUNK_SIZE) - - if not chunk: - # we've reached the end of the file - self.consumer.write(CRLF + b"--" + self.boundary + b"--" + CRLF) - self.file = None - self.consumer.unregisterProducer() - - if self.deferred: - self.deferred.callback(self.lastSent) - self.deferred = None - return - - self.consumer.write(chunk) - self.lastSent = chunk[-1:] - - def pauseProducing(self) -> None: - pass - - def stopProducing(self) -> None: - if self.deferred: - self.deferred.errback(Exception("Consumer asked us to stop producing")) - self.deferred = None - - -class Header: - """ - `Header` This class is a tiny wrapper that produces - request headers. We can't use standard python header - class because it encodes unicode fields using =? bla bla ?= - encoding, which is correct, but no one in HTTP world expects - that, everyone wants utf-8 raw bytes. (stolen from treq.multipart) - - """ - - def __init__( - self, - name: bytes, - value: Any, - params: Optional[List[Tuple[Any, Any]]] = None, - ): - self.name = name - self.value = value - self.params = params or [] - - def add_param(self, name: Any, value: Any) -> None: - self.params.append((name, value)) - - def __bytes__(self) -> bytes: - with closing(BytesIO()) as h: - h.write(self.name + b": " + escape(self.value).encode("us-ascii")) - if self.params: - for name, val in self.params: - h.write(b"; ") - h.write(escape(name).encode("us-ascii")) - h.write(b"=") - h.write(b'"' + escape(val).encode("utf-8") + b'"') - h.seek(0) - return h.read() - - -def escape(value: Union[str, bytes]) -> str: - """ - This function prevents header values from corrupting the request, - a newline in the file name parameter makes form-data request unreadable - for a majority of parsers. (stolen from treq.multipart) - """ - if isinstance(value, bytes): - value = value.decode("utf-8") - return value.replace("\r", "").replace("\n", "").replace('"', '\\"') diff --git a/synapse/media/storage_provider.py b/synapse/media/storage_provider.py index a2d50adf65..06e5d27a53 100644 --- a/synapse/media/storage_provider.py +++ b/synapse/media/storage_provider.py @@ -24,16 +24,14 @@ import logging import os import shutil from typing import TYPE_CHECKING, Callable, Optional -from uuid import uuid4 from synapse.config._base import Config from synapse.logging.context import defer_to_thread, run_in_background from synapse.logging.opentracing import start_active_span, trace_with_opname from synapse.util.async_helpers import maybe_awaitable -from ..storage.databases.main.media_repository import LocalMedia from ._base import FileInfo, Responder -from .media_storage import FileResponder, MultipartResponder +from .media_storage import FileResponder logger = logging.getLogger(__name__) @@ -57,21 +55,13 @@ class StorageProvider(metaclass=abc.ABCMeta): """ @abc.abstractmethod - async def fetch( - self, - path: str, - file_info: FileInfo, - media_info: Optional[LocalMedia] = None, - federation: bool = False, - ) -> Optional[Responder]: + async def fetch(self, path: str, file_info: FileInfo) -> Optional[Responder]: """Attempt to fetch the file described by file_info and stream it into writer. Args: path: Relative path of file in local cache file_info: The metadata of the file. - media_info: metadata of the media item - federation: Whether the requested media is for a federation request Returns: Returns a Responder if the provider has the file, otherwise returns None. @@ -134,13 +124,7 @@ class StorageProviderWrapper(StorageProvider): run_in_background(store) @trace_with_opname("StorageProviderWrapper.fetch") - async def fetch( - self, - path: str, - file_info: FileInfo, - media_info: Optional[LocalMedia] = None, - federation: bool = False, - ) -> Optional[Responder]: + async def fetch(self, path: str, file_info: FileInfo) -> Optional[Responder]: if file_info.url_cache: # Files in the URL preview cache definitely aren't stored here, # so avoid any potentially slow I/O or network access. @@ -148,9 +132,7 @@ class StorageProviderWrapper(StorageProvider): # store_file is supposed to return an Awaitable, but guard # against improper implementations. - return await maybe_awaitable( - self.backend.fetch(path, file_info, media_info, federation) - ) + return await maybe_awaitable(self.backend.fetch(path, file_info)) class FileStorageProviderBackend(StorageProvider): @@ -190,23 +172,11 @@ class FileStorageProviderBackend(StorageProvider): ) @trace_with_opname("FileStorageProviderBackend.fetch") - async def fetch( - self, - path: str, - file_info: FileInfo, - media_info: Optional[LocalMedia] = None, - federation: bool = False, - ) -> Optional[Responder]: + async def fetch(self, path: str, file_info: FileInfo) -> Optional[Responder]: """See StorageProvider.fetch""" backup_fname = os.path.join(self.base_directory, path) if os.path.isfile(backup_fname): - if federation: - assert media_info is not None - boundary = uuid4().hex.encode("ascii") - return MultipartResponder( - open(backup_fname, "rb"), media_info, boundary - ) return FileResponder(open(backup_fname, "rb")) return None |