diff --git a/synapse/media/_base.py b/synapse/media/_base.py index 7884930876..cae91cf4a1 100644 --- a/synapse/media/_base.py +++ b/synapse/media/_base.py @@ -534,6 +534,9 @@ class FileInfo: url_cache: bool = False # Whether the file is a thumbnail or not. thumbnail: ThumbnailInfo | None = None + # The SHA-256 of the original media, if known. Files are stored under this + # instead of `file_id`, with the `file_id` path kept as a fallback for older files. + sha256: str | None = None # The below properties exist to maintain compatibility with third-party modules. @property diff --git a/synapse/media/media_repository.py b/synapse/media/media_repository.py index 31bb401e65..3b0528c63d 100644 --- a/synapse/media/media_repository.py +++ b/synapse/media/media_repository.py @@ -19,6 +19,7 @@ # [This file includes modifications made by New Vector Limited] # # +import contextlib import errno import logging import os @@ -29,7 +30,6 @@ import attr from matrix_common.types.mxc_uri import MXCUri -import twisted.internet.error import twisted.web.http from twisted.internet.defer import Deferred @@ -62,7 +62,6 @@ from synapse.media.filepath import MediaFilePaths from synapse.media.media_storage import ( MediaStorage, - SHA256TransparentIOReader, SHA256TransparentIOWriter, ) from synapse.media.storage_provider import ( @@ -83,6 +82,7 @@ from synapse.util.stringutils import random_string if TYPE_CHECKING: + from synapse.handlers.worker_lock import WaitingLock from synapse.server import HomeServer logger = logging.getLogger(__name__) @@ -94,6 +94,8 @@ # that should be purged according to the configured media retention settings. MEDIA_RETENTION_CHECK_PERIOD = Duration(hours=1) +_SHA256_FILES_LOCK_NAME = "media_sha256_files" + class MediaRepository: def __init__(self, hs: "HomeServer"): @@ -117,6 +119,7 @@ def __init__(self, hs: "HomeServer"): self.thumbnail_requirements = hs.config.media.thumbnail_requirements self.remote_media_linearizer = Linearizer(name="media_remote", clock=self.clock) + self._worker_locks = hs.get_worker_locks_handler() self.recently_accessed_remotes: set[tuple[str, str]] = set() self.recently_accessed_locals: set[str] = set() @@ -346,29 +349,76 @@ async def create_or_update_content( media_id = random_string(24) file_info = FileInfo(server_name=None, file_id=media_id) - sha256reader = SHA256TransparentIOReader(content) - # This implements all of IO as it has a passthrough - fname = await self.media_storage.store_file(sha256reader.wrap(), file_info) - sha256 = sha256reader.hexdigest() - should_quarantine = await self.store.get_is_hash_quarantined(sha256) + # Hold the SHA-256 lock from just before the file is published until the + # DB row is inserted. Quarantine and quota checks run before publish, so a + # failure deletes only the temporary file. + async with contextlib.AsyncExitStack() as sha256_lock: + async with self.media_storage.store_into_file(file_info) as (f, _): + sha256writer = SHA256TransparentIOWriter.get_or_wrap(f) + # This implements all of BinaryIO as it has a passthrough + await self.media_storage.write_to_file(content, sha256writer.wrap()) + sha256 = sha256writer.hexdigest() + should_quarantine = await self.store.get_is_hash_quarantined(sha256) + + if should_quarantine: + logger.warning( + "Media has been automatically quarantined as it matched existing quarantined media" + ) - logger.info("Stored local media in file %r", fname) + await self._check_media_upload_limits(auth_user, content_length) - if should_quarantine: - logger.warning( - "Media has been automatically quarantined as it matched existing quarantined media" + # The file is published when this block exits. + await sha256_lock.enter_async_context( + self._lock_sha256_files(None, sha256) + ) + + logger.info("Stored local media %s with SHA-256 %s", media_id, sha256) + + if is_new_media: + await self.store.store_local_media( + media_id=media_id, + media_type=media_type, + time_now_ms=self.clock.time_msec(), + upload_name=upload_name, + media_length=content_length, + user_id=auth_user, + sha256=sha256, + quarantined_by="system" if should_quarantine else None, + ) + else: + await self.store.update_local_media( + media_id=media_id, + media_type=media_type, + upload_name=upload_name, + media_length=content_length, + user_id=auth_user, + sha256=sha256, + quarantined_by="system" if should_quarantine else None, + ) + + try: + await self._generate_thumbnails( + None, media_id, media_id, media_type, sha256=sha256 ) + except Exception as e: + logger.info("Failed to generate thumbnails: %s", e) + + return MXCUri(self.server_name, media_id) - # Check that the user has not exceeded any of the media upload limits. + async def _check_media_upload_limits( + self, auth_user: UserID, content_length: int + ) -> None: + """Raise if this upload would exceed the user's media quota. - # Use limits from module API if provided + Runs before the temporary upload is published, so a failure leaves no + SHA-256 file behind. + """ media_upload_limits = ( await self.media_repository_callbacks.get_media_upload_limits_for_user( auth_user.to_string() ) ) - # Otherwise use the default limits from config if media_upload_limits is None: # Note: the media upload limits are sorted so larger time periods are # first. @@ -417,35 +467,6 @@ async def create_or_update_content( can_upgrade=limit.can_upgrade, ) - if is_new_media: - await self.store.store_local_media( - media_id=media_id, - media_type=media_type, - time_now_ms=self.clock.time_msec(), - upload_name=upload_name, - media_length=content_length, - user_id=auth_user, - sha256=sha256, - quarantined_by="system" if should_quarantine else None, - ) - else: - await self.store.update_local_media( - media_id=media_id, - media_type=media_type, - upload_name=upload_name, - media_length=content_length, - user_id=auth_user, - sha256=sha256, - quarantined_by="system" if should_quarantine else None, - ) - - try: - await self._generate_thumbnails(None, media_id, media_id, media_type) - except Exception as e: - logger.info("Failed to generate thumbnails: %s", e) - - return MXCUri(self.server_name, media_id) - def respond_not_yet_uploaded(self, request: SynapseRequest) -> None: respond_with_json( request, @@ -581,7 +602,9 @@ async def get_local_media( upload_name = name if name else media_info.upload_name url_cache = media_info.url_cache - file_info = FileInfo(None, media_id, url_cache=bool(url_cache)) + file_info = FileInfo( + None, media_id, url_cache=bool(url_cache), sha256=media_info.sha256 + ) responder = await self.media_storage.fetch_media(file_info) if federation: @@ -777,7 +800,7 @@ async def _get_remote_media_impl( # If we have an entry in the DB, try and look for it if media_info: file_id = media_info.filesystem_id - file_info = FileInfo(server_name, file_id) + file_info = FileInfo(server_name, file_id, sha256=media_info.sha256) if media_info.quarantined_by and not bypass_quarantine: logger.info("Media is quarantined") @@ -832,7 +855,7 @@ async def _get_remote_media_impl( file_id = media_info.filesystem_id if not media_info.media_type: media_info = attr.evolve(media_info, media_type="application/octet-stream") - file_info = FileInfo(server_name, file_id) + file_info = FileInfo(server_name, file_id, sha256=media_info.sha256) # We generate thumbnails even if another process downloaded the media # as a) it's conceivable that the other download request dies before it @@ -841,7 +864,11 @@ async def _get_remote_media_impl( # otherwise they'll request thumbnails and get a 404 if they're not # ready yet. await self._generate_thumbnails( - server_name, media_id, file_id, media_info.media_type + server_name, + media_id, + file_id, + media_info.media_type, + sha256=media_info.sha256, ) responder = await self.media_storage.fetch_media(file_info) @@ -857,7 +884,6 @@ async def _store_remote_media_with_cleanup( media_length: int, filesystem_id: str, sha256: str, - fname: str, ) -> None: """Store remote media in database and clean up file on constraint violation.""" try: @@ -872,11 +898,9 @@ async def _store_remote_media_with_cleanup( sha256=sha256, ) except self.store.database_engine.module.IntegrityError: - # Another worker downloaded the media first. Clean up our file. - try: - os.remove(fname) - except Exception: - pass + # Another worker downloaded the media first. Clean up our file, unless + # other media shares its SHA-256 path. + await self._remove_sha256_files_if_unused(server_name, sha256) # Re-raise so the caller can handle it raise @@ -910,77 +934,87 @@ async def _download_remote_file( file_info = FileInfo(server_name=server_name, file_id=file_id) - async with self.media_storage.store_into_file(file_info) as (f, fname): - sha256writer = SHA256TransparentIOWriter(f) - try: - length, headers = await self.client.download_media( - server_name, - media_id, - # This implements all of BinaryIO as it has a passthrough - output_stream=sha256writer.wrap(), - max_size=self.max_upload_size, - max_timeout_ms=max_timeout_ms, - download_ratelimiter=download_ratelimiter, - ip_address=ip_address, - ) - except RequestSendFailed as e: - logger.warning( - "Request failed fetching remote media %s/%s: %r", - server_name, - media_id, - e, - ) - raise SynapseError(502, "Failed to fetch remote media") + async with contextlib.AsyncExitStack() as sha256_lock: + async with self.media_storage.store_into_file(file_info) as (f, fname): + sha256writer = SHA256TransparentIOWriter.get_or_wrap(f) + try: + length, headers = await self.client.download_media( + server_name, + media_id, + # This implements all of BinaryIO as it has a passthrough + output_stream=sha256writer.wrap(), + max_size=self.max_upload_size, + max_timeout_ms=max_timeout_ms, + download_ratelimiter=download_ratelimiter, + ip_address=ip_address, + ) + except RequestSendFailed as e: + logger.warning( + "Request failed fetching remote media %s/%s: %r", + server_name, + media_id, + e, + ) + raise SynapseError(502, "Failed to fetch remote media") - except HttpResponseException as e: - logger.warning( - "HTTP error fetching remote media %s/%s: %s", - server_name, - media_id, - e.response, - ) - if e.code == twisted.web.http.NOT_FOUND: - raise e.to_synapse_error() - raise SynapseError(502, "Failed to fetch remote media") + except HttpResponseException as e: + logger.warning( + "HTTP error fetching remote media %s/%s: %s", + server_name, + media_id, + e.response, + ) + if e.code == twisted.web.http.NOT_FOUND: + raise e.to_synapse_error() + raise SynapseError(502, "Failed to fetch remote media") - except SynapseError: - logger.warning( - "Failed to fetch remote media %s/%s", server_name, media_id - ) - raise - except NotRetryingDestination: - logger.warning("Not retrying destination %r", server_name) - raise SynapseError(502, "Failed to fetch remote media") - except Exception: - logger.exception( - "Failed to fetch remote media %s/%s", server_name, media_id + except SynapseError: + logger.warning( + "Failed to fetch remote media %s/%s", server_name, media_id + ) + raise + except NotRetryingDestination: + logger.warning("Not retrying destination %r", server_name) + raise SynapseError(502, "Failed to fetch remote media") + except Exception: + logger.exception( + "Failed to fetch remote media %s/%s", server_name, media_id + ) + raise SynapseError(502, "Failed to fetch remote media") + + if b"Content-Type" in headers: + media_type = headers[b"Content-Type"][0].decode("ascii") + else: + media_type = "application/octet-stream" + upload_name = get_filename_from_headers(headers) + time_now_ms = self.clock.time_msec() + + # The file is published when this block exits. + await sha256_lock.enter_async_context( + self._lock_sha256_files(server_name, sha256writer.hexdigest()) ) - raise SynapseError(502, "Failed to fetch remote media") - if b"Content-Type" in headers: - media_type = headers[b"Content-Type"][0].decode("ascii") - else: - media_type = "application/octet-stream" - upload_name = get_filename_from_headers(headers) - time_now_ms = self.clock.time_msec() - - # Multiple remote media download requests can race (when using - # multiple media repos), so this may throw a violation constraint - # exception. If it does we'll delete the newly downloaded file from - # disk. - await self._store_remote_media_with_cleanup( - server_name=server_name, - media_id=media_id, - media_type=media_type, - time_now_ms=time_now_ms, - upload_name=upload_name, - media_length=length, - filesystem_id=file_id, - sha256=sha256writer.hexdigest(), - fname=fname, - ) + # Multiple remote media download requests can race (when using + # multiple media repos), so this may throw a violation constraint + # exception. If it does we'll delete the newly downloaded file from + # disk. This already holds the SHA-256 lock. + await self._store_remote_media_with_cleanup( + server_name=server_name, + media_id=media_id, + media_type=media_type, + time_now_ms=time_now_ms, + upload_name=upload_name, + media_length=length, + filesystem_id=file_id, + sha256=sha256writer.hexdigest(), + ) - logger.info("Stored remote media in file %r", fname) + logger.info( + "Stored remote media %s/%s with SHA-256 %s", + server_name, + media_id, + sha256writer.hexdigest(), + ) if self.hs.config.media.enable_authenticated_media: authenticated = True @@ -1032,84 +1066,94 @@ async def _federation_download_remote_file( file_info = FileInfo(server_name=server_name, file_id=file_id) - async with self.media_storage.store_into_file(file_info) as (f, fname): - sha256writer = SHA256TransparentIOWriter(f) - try: - res = await self.client.federation_download_media( - server_name, - media_id, - # This implements all of BinaryIO as it has a passthrough - output_stream=sha256writer.wrap(), - max_size=self.max_upload_size, - max_timeout_ms=max_timeout_ms, - download_ratelimiter=download_ratelimiter, - ip_address=ip_address, - ) - # if we had to fall back to the _matrix/media endpoint it will only return - # the headers and length, check the length of the tuple before unpacking - if len(res) == 3: - length, headers, json = res + async with contextlib.AsyncExitStack() as sha256_lock: + async with self.media_storage.store_into_file(file_info) as (f, fname): + sha256writer = SHA256TransparentIOWriter.get_or_wrap(f) + try: + res = await self.client.federation_download_media( + server_name, + media_id, + # This implements all of BinaryIO as it has a passthrough + output_stream=sha256writer.wrap(), + max_size=self.max_upload_size, + max_timeout_ms=max_timeout_ms, + download_ratelimiter=download_ratelimiter, + ip_address=ip_address, + ) + # if we had to fall back to the _matrix/media endpoint it will only return + # the headers and length, check the length of the tuple before unpacking + if len(res) == 3: + length, headers, json = res + else: + length, headers = res + except RequestSendFailed as e: + logger.warning( + "Request failed fetching remote media %s/%s: %r", + server_name, + media_id, + e, + ) + raise SynapseError(502, "Failed to fetch remote media") + + except HttpResponseException as e: + logger.warning( + "HTTP error fetching remote media %s/%s: %s", + server_name, + media_id, + e.response, + ) + if e.code == twisted.web.http.NOT_FOUND: + raise e.to_synapse_error() + raise SynapseError(502, "Failed to fetch remote media") + + except SynapseError: + logger.warning( + "Failed to fetch remote media %s/%s", server_name, media_id + ) + raise + except NotRetryingDestination: + logger.warning("Not retrying destination %r", server_name) + raise SynapseError(502, "Failed to fetch remote media") + except Exception: + logger.exception( + "Failed to fetch remote media %s/%s", server_name, media_id + ) + raise SynapseError(502, "Failed to fetch remote media") + + if b"Content-Type" in headers: + media_type = headers[b"Content-Type"][0].decode("ascii") else: - length, headers = res - except RequestSendFailed as e: - logger.warning( - "Request failed fetching remote media %s/%s: %r", - server_name, - media_id, - e, - ) - raise SynapseError(502, "Failed to fetch remote media") + media_type = "application/octet-stream" + upload_name = get_filename_from_headers(headers) + time_now_ms = self.clock.time_msec() - except HttpResponseException as e: - logger.warning( - "HTTP error fetching remote media %s/%s: %s", - server_name, - media_id, - e.response, + # The file is published when this block exits. + await sha256_lock.enter_async_context( + self._lock_sha256_files(server_name, sha256writer.hexdigest()) ) - if e.code == twisted.web.http.NOT_FOUND: - raise e.to_synapse_error() - raise SynapseError(502, "Failed to fetch remote media") - except SynapseError: - logger.warning( - "Failed to fetch remote media %s/%s", server_name, media_id - ) - raise - except NotRetryingDestination: - logger.warning("Not retrying destination %r", server_name) - raise SynapseError(502, "Failed to fetch remote media") - except Exception: - logger.exception( - "Failed to fetch remote media %s/%s", server_name, media_id - ) - raise SynapseError(502, "Failed to fetch remote media") + # Multiple remote media download requests can race (when using + # multiple media repos), so this may throw a violation constraint + # exception. If it does we'll delete the newly downloaded file from + # disk. This already holds the SHA-256 lock. + await self._store_remote_media_with_cleanup( + server_name=server_name, + media_id=media_id, + media_type=media_type, + time_now_ms=time_now_ms, + upload_name=upload_name, + media_length=length, + filesystem_id=file_id, + sha256=sha256writer.hexdigest(), + ) - if b"Content-Type" in headers: - media_type = headers[b"Content-Type"][0].decode("ascii") - else: - media_type = "application/octet-stream" - upload_name = get_filename_from_headers(headers) - time_now_ms = self.clock.time_msec() - - # Multiple remote media download requests can race (when using - # multiple media repos), so this may throw a violation constraint - # exception. If it does we'll delete the newly downloaded file from - # disk. - await self._store_remote_media_with_cleanup( - server_name=server_name, - media_id=media_id, - media_type=media_type, - time_now_ms=time_now_ms, - upload_name=upload_name, - media_length=length, - filesystem_id=file_id, - sha256=sha256writer.hexdigest(), - fname=fname, + logger.debug( + "Stored remote media %s/%s with SHA-256 %s", + server_name, + media_id, + sha256writer.hexdigest(), ) - logger.debug("Stored remote media in file %r", fname) - if self.hs.config.media.enable_authenticated_media: authenticated = True else: @@ -1180,9 +1224,10 @@ async def generate_local_exact_thumbnail( t_type: str, url_cache: bool, animated: bool = False, + sha256: str | None = None, ) -> tuple[str, FileInfo] | None: async with self.media_storage.ensure_media_is_in_local_cache( - FileInfo(None, media_id, url_cache=url_cache) + FileInfo(None, media_id, url_cache=url_cache, sha256=sha256) ) as input_path: try: thumbnailer = Thumbnailer(input_path) @@ -1221,6 +1266,7 @@ async def generate_local_exact_thumbnail( type=t_type, length=t_byte_source.getbuffer().nbytes, ), + sha256=sha256, ) output_path = await self.media_storage.store_file( @@ -1257,9 +1303,10 @@ async def generate_remote_exact_thumbnail( t_method: str, t_type: str, animated: bool = False, + sha256: str | None = None, ) -> str | None: async with self.media_storage.ensure_media_is_in_local_cache( - FileInfo(server_name, file_id) + FileInfo(server_name, file_id, sha256=sha256) ) as input_path: try: thumbnailer = Thumbnailer(input_path) @@ -1298,6 +1345,7 @@ async def generate_remote_exact_thumbnail( type=t_type, length=t_byte_source.getbuffer().nbytes, ), + sha256=sha256, ) output_path = await self.media_storage.store_file( @@ -1334,6 +1382,7 @@ async def _generate_thumbnails( file_id: str, media_type: str, url_cache: bool = False, + sha256: str | None = None, ) -> dict | None: """Generate and store thumbnails for an image. @@ -1345,6 +1394,7 @@ async def _generate_thumbnails( media_type: The content type of the file url_cache: If we are thumbnailing images downloaded for the URL cache, used exclusively by the url previewer + sha256: The SHA-256 of the media, if known Returns: Dict with "width" and "height" keys of original image or None if the @@ -1355,7 +1405,7 @@ async def _generate_thumbnails( return None async with self.media_storage.ensure_media_is_in_local_cache( - FileInfo(server_name, file_id, url_cache=url_cache) + FileInfo(server_name, file_id, url_cache=url_cache, sha256=sha256) ) as input_path: try: thumbnailer = Thumbnailer(input_path) @@ -1454,6 +1504,7 @@ async def _generate_thumbnails( t_method, t_type, t_byte_source, + sha256, ) if ( @@ -1482,6 +1533,7 @@ async def _generate_thumbnails( t_method, ANIMATED_THUMBNAIL_TYPE, a_byte_source, + sha256, ) return {"width": m_width, "height": m_height} @@ -1497,6 +1549,7 @@ async def _store_thumbnail( t_method: str, t_type: str, t_byte_source: BytesIO, + sha256: str | None = None, ) -> None: """Store a generated thumbnail to the configured storage and database.""" file_info = FileInfo( @@ -1510,6 +1563,7 @@ async def _store_thumbnail( type=t_type, length=t_byte_source.getbuffer().nbytes, ), + sha256=sha256, ) async with self.media_storage.store_into_file(file_info) as (f, fname): @@ -1617,25 +1671,23 @@ async def delete_old_remote_media(self, before_ts: int) -> dict[str, int]: logger.info("Deleting: %r", key) - # TODO: Should we delete from the backup store - async with self.remote_media_linearizer.queue(key): - full_path = self.filepaths.remote_media_filepath(origin, file_id) - try: - os.remove(full_path) - except OSError as e: - logger.warning("Failed to remove file: %r", full_path) - if e.errno == errno.ENOENT: - pass - else: - continue - - thumbnail_dir = self.filepaths.remote_media_thumbnail_dir( - origin, file_id + media_info = await self.store.get_cached_remote_media(origin, media_id) + removed = await self._delete_media_id_files( + FileInfo(server_name=origin, file_id=file_id), + self.filepaths.remote_media_filepath(origin, file_id), + self.filepaths.remote_media_filepath_rel(origin, file_id), + self.filepaths.remote_media_thumbnail_dir(origin, file_id), ) - shutil.rmtree(thumbnail_dir, ignore_errors=True) + if not removed: + continue await self.store.delete_remote_media(origin, media_id) + if media_info and media_info.sha256: + async with self._lock_sha256_files(origin, media_info.sha256): + await self._remove_sha256_files_if_unused( + origin, media_info.sha256 + ) deleted += 1 return {"deleted": deleted} @@ -1701,24 +1753,114 @@ async def _remove_local_media_from_disk( removed_media = [] for media_id in media_ids: logger.info("Deleting media with ID '%s'", media_id) - full_path = self.filepaths.local_media_filepath(media_id) - try: - os.remove(full_path) - except OSError as e: - logger.warning("Failed to remove file: %r: %s", full_path, e) - if e.errno == errno.ENOENT: - pass - else: - continue - - thumbnail_dir = self.filepaths.local_media_thumbnail_dir(media_id) - shutil.rmtree(thumbnail_dir, ignore_errors=True) + media_info = await self.store.get_local_media(media_id) + removed = await self._delete_media_id_files( + FileInfo(server_name=None, file_id=media_id), + self.filepaths.local_media_filepath(media_id), + self.filepaths.local_media_filepath_rel(media_id), + self.filepaths.local_media_thumbnail_dir(media_id), + ) + if not removed: + continue await self.store.delete_remote_media(self.server_name, media_id) await self.store.delete_url_cache((media_id,)) await self.store.delete_url_cache_media((media_id,)) + # Files at the SHA-256 path can be shared with other media of the same + # content, so they are only removed once the DB row has been deleted + # and nothing else references them. + if media_info and media_info.sha256 and not media_info.url_cache: + async with self._lock_sha256_files(None, media_info.sha256): + await self._remove_sha256_files_if_unused(None, media_info.sha256) + removed_media.append(media_id) return removed_media, len(removed_media) + + def _lock_sha256_files(self, server_name: str | None, sha256: str) -> "WaitingLock": + """A cross-worker lock on the files stored under a SHA-256 path. + + New media must hold it from before its file is published until its DB row + is inserted, and deletion must hold it while checking references and + removing files. Otherwise deletion could remove a file whose DB row is + about to be inserted. + + Args: + server_name: The origin of remote media, or None for local media. + sha256: The SHA-256 of the media. + """ + return self._worker_locks.acquire_lock( + _SHA256_FILES_LOCK_NAME, f"{server_name or ''}/{sha256}" + ) + + async def _delete_media_id_files( + self, + file_info: FileInfo, + full_path: str, + relative_path: str, + thumbnail_dir: str, + ) -> bool: + """Remove one media ID's own files from the local disk and every provider. + + Returns: + False when the local file could not be removed. A missing file is success. + """ + try: + os.remove(full_path) + except OSError as e: + logger.warning("Failed to remove file: %r: %s", full_path, e) + if e.errno != errno.ENOENT: + return False + + await self.media_storage.delete_from_storage_providers(relative_path, file_info) + shutil.rmtree(thumbnail_dir, ignore_errors=True) + await self.media_storage.delete_directory_from_storage_providers( + os.path.relpath(thumbnail_dir, self.primary_base_path), file_info + ) + return True + + async def _remove_sha256_files_if_unused( + self, server_name: str | None, sha256: str + ) -> None: + """Remove the media file and thumbnails stored under a SHA-256 path, unless + other media with the same SHA-256 still references them. + + Must be called while holding `_lock_sha256_files` for the same path. + + Args: + server_name: The origin of remote media, or None for local media. + sha256: The SHA-256 of the media. + """ + if await self.store.is_media_sha256_in_use(sha256, server_name): + logger.debug("SHA-256 %r is in use", sha256) + return + + if server_name: + file_info = FileInfo(server_name=server_name, file_id=sha256) + full_path = self.filepaths.remote_media_filepath(server_name, sha256) + relative_path = self.filepaths.remote_media_filepath_rel( + server_name, sha256 + ) + thumbnail_dir = self.filepaths.remote_media_thumbnail_dir( + server_name, sha256 + ) + else: + file_info = FileInfo(server_name=None, file_id=sha256) + full_path = self.filepaths.local_media_filepath(sha256) + relative_path = self.filepaths.local_media_filepath_rel(sha256) + thumbnail_dir = self.filepaths.local_media_thumbnail_dir(sha256) + + try: + os.remove(full_path) + except FileNotFoundError: + pass + except OSError as e: + logger.warning("Failed to remove file: %r: %s", full_path, e) + + await self.media_storage.delete_from_storage_providers(relative_path, file_info) + shutil.rmtree(thumbnail_dir, ignore_errors=True) + await self.media_storage.delete_directory_from_storage_providers( + os.path.relpath(thumbnail_dir, self.primary_base_path), file_info + ) diff --git a/synapse/media/media_storage.py b/synapse/media/media_storage.py index 6f4b63741c..04a6c528fa 100644 --- a/synapse/media/media_storage.py +++ b/synapse/media/media_storage.py @@ -20,6 +20,7 @@ # import contextlib import hashlib +import itertools import json import logging import os @@ -79,6 +80,17 @@ def __init__(self, source: BinaryIO): self._hash = hashlib.sha256() self._source = source + @classmethod + def get_or_wrap(cls, source: BinaryIO) -> "SHA256TransparentIOWriter": + """Returns `source` if it already computes a SHA-256 (e.g. the file + yielded by `MediaStorage.store_into_file` for new media), otherwise wraps it. + + This avoids hashing the same bytes twice. + """ + if isinstance(source, cls): + return source + return cls(source) + def write(self, buffer: bytes | bytearray) -> int: """Wrapper for source.write() @@ -194,7 +206,16 @@ async def store_file(self, source: IO, file_info: FileInfo) -> str: # Write to the main media repository await self.write_to_file(source, f) - return fname + if not self.local_provider: + return fname + + # `fname` may have been a temporary file. Return the path that was kept. + if isinstance(f, SHA256TransparentIOWriter): + file_info = attr.evolve(file_info, sha256=f.hexdigest()) + return os.path.join( + self.local_media_directory, # type: ignore[arg-type] + self._candidate_paths(file_info)[0], + ) @trace_with_opname("MediaStorage.write_to_file") async def write_to_file(self, source: IO, output: IO) -> None: @@ -216,6 +237,21 @@ async def store_into_file( media_filepath can be used to read the contents from after upload, e.g. to generate thumbnails. + Where the file is written: + + * If the file is stored under a `media_id` path and there is a local media + store, it is written there directly. This is the case for URL cache files, + and thumbnails of older media without a SHA-256. + * Files stored under a SHA-256 path may be shared, so they are written to a + temporary file. After the spam check the temp file is linked into place + only when that path is absent; an existing file is kept and the temp file + is deleted. Storage providers receive the path only when it was created. + This covers new uploads, remote downloads, thumbnails of media with a + SHA-256, and media whose `file_info.sha256` is already set. + media_filepath is only valid inside the context manager. + * Without a local media store, the file is written to a temporary file + that is removed once it has been sent to the storage providers. + Args: file_info: Info about the file to store @@ -225,22 +261,64 @@ async def store_into_file( # .. write into f ... """ - path = self._file_info_to_path(file_info) - is_temp_file = False + # The relative path to store the file at, or None if it depends on the + # SHA-256 of the content, which is only known once it has been written. + path: str | None = None + if file_info.sha256 or file_info.thumbnail or file_info.url_cache: + path = self._candidate_paths(file_info)[0] - if self.local_provider: - media_filepath = os.path.join(self.local_media_directory, path) # type: ignore[arg-type] - os.makedirs(os.path.dirname(media_filepath), exist_ok=True) + # SHA-256 paths are shared. Never open one with `wb`: that truncates a + # file other media IDs may be reading, and a later failure would delete it. + is_shared_path = bool(file_info.sha256) and not file_info.url_cache + is_temp_file = path is None or is_shared_path or not self.local_provider + sha256_writer: SHA256TransparentIOWriter | None = None + media_filepath: str | None = None - with start_active_span("writing to main media repo"): - with open(media_filepath, "wb") as f: - yield f, media_filepath - else: - # No local provider, write to temp file - is_temp_file = True - with tempfile.NamedTemporaryFile(delete=False) as f: - media_filepath = f.name - yield cast(BinaryIO, f), media_filepath + try: + if not is_temp_file: + # The path is known, not shared, and there is a local media store: + # URL cache files, and thumbnails of media without a SHA-256. + media_filepath = os.path.join(self.local_media_directory, path) # type: ignore[arg-type] + f = await defer_to_thread( + self.reactor, _open_for_writing, media_filepath + ) + + with start_active_span("writing to main media repo"): + with f: + yield f, media_filepath + else: + # The path is a SHA-256 path (known or not yet), or there is no + # local media store. Keep the temp file on the same filesystem as + # the media store so it can be linked into place. + tmp = await defer_to_thread( + self.reactor, _create_temp_file, self.local_media_directory + ) + media_filepath = tmp.name + with tmp: + if path is None: + # Callers can reuse this hash with + # `SHA256TransparentIOWriter.get_or_wrap`. + sha256_writer = SHA256TransparentIOWriter(tmp) + yield sha256_writer.wrap(), media_filepath + else: + yield tmp, media_filepath + except BaseException: + # The caller failed to write the file (e.g. a failed remote download), + # so don't leave a partial temp file behind. This must not await, as + # the generator may be being closed with `GeneratorExit`. + if is_temp_file and media_filepath is not None: + _remove_quietly(media_filepath) + raise + + assert media_filepath is not None + if sha256_writer is not None: + file_info = attr.evolve(file_info, sha256=sha256_writer.hexdigest()) + path = self._file_info_to_path(file_info, file_info.sha256) + assert path is not None + + # Whether media_filepath may be removed on failure. A pre-existing SHA-256 + # file can be shared with other media, so it must not be removed. + owns_media_filepath = True # Spam check and store to other providers (runs for both local and temp file cases) try: @@ -259,22 +337,35 @@ async def store_into_file( # the spam-check API. raise SpamMediaException(errcode=spam_check[0]) - for provider in self.storage_providers: - with start_active_span(str(provider)): - await provider.store_file(path, file_info) + # Only publish once the spam check has passed. Providers such as + # `FileStorageProviderBackend` read the final local path, so a new + # hash path is linked into place first. An existing hash path is + # kept and is not uploaded again. + published = True + if is_temp_file and self.local_provider: + final_filepath = os.path.join(self.local_media_directory, path) # type: ignore[arg-type] + published = await defer_to_thread( + self.reactor, + _publish_if_absent, + media_filepath, + final_filepath, + ) + owns_media_filepath = published + media_filepath = final_filepath + is_temp_file = False + + if published: + for provider in self.storage_providers: + with start_active_span(str(provider)): + await provider.store_file(path, file_info) # If using a temp file, delete it after uploading to storage providers if is_temp_file: - try: - os.remove(media_filepath) - except Exception: - pass + await defer_to_thread(self.reactor, _remove_quietly, media_filepath) except Exception as e: - try: - os.remove(media_filepath) - except Exception: - pass + if owns_media_filepath: + await defer_to_thread(self.reactor, _remove_quietly, media_filepath) raise e from None @@ -299,7 +390,7 @@ async def fetch_media(self, file_info: FileInfo) -> Responder | None: return FileResponder(self.hs, open(local_path, "rb")) return None - paths = [self._file_info_to_path(file_info)] + paths = self._candidate_paths(file_info) # fallback for remote thumbnails with no method in the filename if file_info.thumbnail and file_info.server_name: @@ -353,12 +444,14 @@ async def ensure_media_is_in_local_cache( async with media_storage.ensure_media_is_in_local_cache(file_info) as path: # use path to read the file """ - path = self._file_info_to_path(file_info) + paths = self._candidate_paths(file_info) + if self.local_provider: - local_path = os.path.join(self.local_media_directory, path) # type: ignore[arg-type] - if os.path.exists(local_path): - yield local_path - return + for path in paths: + local_path = os.path.join(self.local_media_directory, path) # type: ignore[arg-type] + if os.path.exists(local_path): + yield local_path + return # Fallback for paths without method names # Should be removed in the future @@ -378,24 +471,25 @@ async def ensure_media_is_in_local_cache( yield legacy_local_path return - os.makedirs(os.path.dirname(local_path), exist_ok=True) - for provider in self.storage_providers: - remote_res: Any = await provider.fetch(path, file_info) - if remote_res: - with remote_res: - consumer = BackgroundFileConsumer( - open(local_path, "wb"), self.reactor - ) - await remote_res.write_to_consumer(consumer) - await consumer.wait() - yield local_path - return + for path in paths: + remote_res: Any = await provider.fetch(path, file_info) + if remote_res: + local_path = os.path.join(self.local_media_directory, path) # type: ignore[arg-type] + os.makedirs(os.path.dirname(local_path), exist_ok=True) + with remote_res: + consumer = BackgroundFileConsumer( + open(local_path, "wb"), self.reactor + ) + await remote_res.write_to_consumer(consumer) + await consumer.wait() + yield local_path + return raise NotFoundError() else: # No local provider, download to temp file and clean up after use - for provider in self.storage_providers: + for provider, path in itertools.product(self.storage_providers, paths): res: Any = await provider.fetch(path, file_info) if res: temp_path = None @@ -421,47 +515,81 @@ async def ensure_media_is_in_local_cache( raise NotFoundError() + async def delete_from_storage_providers( + self, path: str, file_info: FileInfo + ) -> None: + """Remove `path` from every configured storage provider.""" + for provider in self.storage_providers: + await provider.delete_file(path, file_info) + + async def delete_directory_from_storage_providers( + self, path: str, file_info: FileInfo + ) -> None: + """Remove a directory from every configured storage provider.""" + for provider in self.storage_providers: + await provider.delete_directory(path, file_info) + @trace - def _file_info_to_path(self, file_info: FileInfo) -> str: + def _file_info_to_path( + self, file_info: FileInfo, identifier: str | None = None + ) -> str: """Converts file_info into a relative path. The path is suitable for storing files under a directory, e.g. used to store files on local FS under the base media repository directory. + + Args: + file_info: Info about the file. + identifier: The identifier to build the path from, e.g. the SHA-256. + Defaults to `file_info.file_id`. """ + identifier = identifier or file_info.file_id if file_info.url_cache: if file_info.thumbnail: return self.filepaths.url_cache_thumbnail_rel( - media_id=file_info.file_id, + media_id=identifier, width=file_info.thumbnail.width, height=file_info.thumbnail.height, content_type=file_info.thumbnail.type, method=file_info.thumbnail.method, ) - return self.filepaths.url_cache_filepath_rel(file_info.file_id) + return self.filepaths.url_cache_filepath_rel(identifier) if file_info.server_name: if file_info.thumbnail: return self.filepaths.remote_media_thumbnail_rel( server_name=file_info.server_name, - file_id=file_info.file_id, + file_id=identifier, width=file_info.thumbnail.width, height=file_info.thumbnail.height, content_type=file_info.thumbnail.type, method=file_info.thumbnail.method, ) return self.filepaths.remote_media_filepath_rel( - file_info.server_name, file_info.file_id + file_info.server_name, identifier ) if file_info.thumbnail: return self.filepaths.local_media_thumbnail_rel( - media_id=file_info.file_id, + media_id=identifier, width=file_info.thumbnail.width, height=file_info.thumbnail.height, content_type=file_info.thumbnail.type, method=file_info.thumbnail.method, ) - return self.filepaths.local_media_filepath_rel(file_info.file_id) + return self.filepaths.local_media_filepath_rel(identifier) + + def _candidate_paths(self, file_info: FileInfo) -> list[str]: + """Returns the relative paths the file may be stored at, in lookup order: + the SHA-256 path if the hash is known, then the legacy `file_id` path. + + URL cache files always use the `file_id` path, as they are expired by date. + """ + paths = [] + if file_info.sha256 and not file_info.url_cache: + paths.append(self._file_info_to_path(file_info, file_info.sha256)) + paths.append(self._file_info_to_path(file_info)) + return paths @trace @@ -477,6 +605,51 @@ def _write_file_synchronously(source: IO, dest: IO) -> None: shutil.copyfileobj(source, dest) +def _open_for_writing(path: str) -> BinaryIO: + """Opens `path` for writing, creating its parent directories. Should be + called from a thread.""" + os.makedirs(os.path.dirname(path), exist_ok=True) + return open(path, "wb") + + +def _create_temp_file(directory: str | None) -> BinaryIO: + """Creates a temporary file in `directory`, or the system temp directory if + None. The file is not deleted on close. Should be called from a thread.""" + if directory: + os.makedirs(directory, exist_ok=True) + return cast(BinaryIO, tempfile.NamedTemporaryFile(dir=directory, delete=False)) + + +def _publish_if_absent(src: str, dest: str) -> bool: + """Link `src` onto `dest` when `dest` is absent, then remove `src`. + + Should be called from a thread. `src` and `dest` must be on the same + filesystem. The link is atomic, so a destination that appears during the + call is left untouched. + + Returns: + Whether this call created `dest`. + """ + os.makedirs(os.path.dirname(dest), exist_ok=True) + created = False + try: + os.link(src, dest) + created = True + except FileExistsError: + created = False + finally: + _remove_quietly(src) + return created + + +def _remove_quietly(path: str) -> None: + """Removes `path`, ignoring any errors.""" + try: + os.remove(path) + except Exception: + pass + + class FileResponder(Responder): """Wraps an open file that can be sent to a request. diff --git a/synapse/media/storage_provider.py b/synapse/media/storage_provider.py index c5faa25b96..4ebd64587a 100644 --- a/synapse/media/storage_provider.py +++ b/synapse/media/storage_provider.py @@ -20,10 +20,11 @@ # import abc +import errno import logging import os import shutil -from typing import TYPE_CHECKING, Callable +from typing import TYPE_CHECKING from synapse.config._base import Config from synapse.logging.context import defer_to_thread, run_in_background @@ -66,6 +67,28 @@ async def fetch(self, path: str, file_info: FileInfo) -> Responder | None: Returns a Responder if the provider has the file, otherwise returns None. """ + @abc.abstractmethod + async def delete_file(self, path: str, file_info: FileInfo) -> None: + """Remove a previously stored file. Missing files are ignored. + + Content-addressed paths are shared, so callers only invoke this when the + path has no remaining references. The default implementation does nothing + so existing third-party providers keep working. + + Args: + path: Relative path of file in local cache + file_info: The metadata of the file. + """ + + @abc.abstractmethod + async def delete_directory(self, path: str, file_info: FileInfo) -> None: + """Remove a previously stored directory, such as a thumbnail directory. + + Args: + path: Relative path of the directory in the local cache. + file_info: The metadata of the media the directory belongs to. + """ + class StorageProviderWrapper(StorageProvider): """Wraps a storage provider and provides various config options @@ -133,6 +156,27 @@ async def fetch(self, path: str, file_info: FileInfo) -> Responder | None: # against improper implementations. return await maybe_awaitable(self.backend.fetch(path, file_info)) + def _stores_file(self, file_info: FileInfo) -> bool: + if file_info.url_cache: + return False + if not file_info.server_name and not self.store_local: + return False + if file_info.server_name and not self.store_remote: + return False + return True + + @trace_with_opname("StorageProviderWrapper.delete_file") + async def delete_file(self, path: str, file_info: FileInfo) -> None: + if not self._stores_file(file_info): + return + await maybe_awaitable(self.backend.delete_file(path, file_info)) # type: ignore[arg-type] + + @trace_with_opname("StorageProviderWrapper.delete_directory") + async def delete_directory(self, path: str, file_info: FileInfo) -> None: + if not self._stores_file(file_info): + return + await maybe_awaitable(self.backend.delete_directory(path, file_info)) # type: ignore[arg-type] + class FileStorageProviderBackend(StorageProvider): """A storage provider that stores files in a directory on a filesystem. @@ -158,19 +202,25 @@ async def store_file(self, path: str, file_info: FileInfo) -> None: primary_fname = os.path.join(self.cache_directory, path) backup_fname = os.path.join(self.base_directory, path) - dirname = os.path.dirname(backup_fname) - os.makedirs(dirname, exist_ok=True) - - # mypy needs help inferring the type of the second parameter, which is generic - shutil_copyfile: Callable[[str, str], str] = shutil.copyfile - with start_active_span("shutil_copyfile"): + with start_active_span("copy if absent"): await defer_to_thread( - self.reactor, - shutil_copyfile, - primary_fname, - backup_fname, + self.reactor, _copy_if_absent, primary_fname, backup_fname ) + @trace_with_opname("FileStorageProviderBackend.delete_file") + async def delete_file(self, path: str, file_info: FileInfo) -> None: + """See StorageProvider.delete_file""" + backup_fname = os.path.join(self.base_directory, path) + await defer_to_thread(self.reactor, _remove_file, backup_fname) + + @trace_with_opname("FileStorageProviderBackend.delete_directory") + async def delete_directory(self, path: str, file_info: FileInfo) -> None: + """See StorageProvider.delete_directory""" + backup_directory = os.path.join(self.base_directory, path) + await defer_to_thread( + self.reactor, shutil.rmtree, backup_directory, ignore_errors=True + ) + @trace_with_opname("FileStorageProviderBackend.fetch") async def fetch(self, path: str, file_info: FileInfo) -> Responder | None: """See StorageProvider.fetch""" @@ -195,3 +245,32 @@ def parse_config(config: dict) -> str: just pull that out. """ return Config.ensure_directory(config["directory"]) + + +def _copy_if_absent(src: str, dest: str) -> None: + """Copy `src` to `dest` unless `dest` already exists. + + A hard link is used when both paths are on the same filesystem, so the + destination appears with its final contents. An existing destination is + left unchanged. + """ + os.makedirs(os.path.dirname(dest), exist_ok=True) + try: + os.link(src, dest) + return + except FileExistsError: + return + except OSError as e: + if e.errno != errno.EXDEV: + raise + if os.path.exists(dest): + return + shutil.copyfile(src, dest) + + +def _remove_file(path: str) -> None: + """Remove `path`, ignoring a missing file.""" + try: + os.remove(path) + except FileNotFoundError: + pass diff --git a/synapse/media/thumbnailer.py b/synapse/media/thumbnailer.py index 8e90aeb8de..71053d313e 100644 --- a/synapse/media/thumbnailer.py +++ b/synapse/media/thumbnailer.py @@ -451,6 +451,7 @@ async def respond_local_thumbnail( server_name=None, for_federation=for_federation, media_info=media_info, + sha256=media_info.sha256, ) async def select_or_generate_local_thumbnail( @@ -496,6 +497,7 @@ async def select_or_generate_local_thumbnail( file_id=media_id, url_cache=bool(media_info.url_cache), thumbnail=info, + sha256=media_info.sha256, ) responder = await self.media_storage.fetch_media(file_info) @@ -527,6 +529,7 @@ async def select_or_generate_local_thumbnail( desired_type, url_cache=bool(media_info.url_cache), animated=animated, + sha256=media_info.sha256, ) if thumbnail_result: @@ -604,6 +607,7 @@ async def select_or_generate_remote_thumbnail( server_name=server_name, file_id=file_id, thumbnail=info, + sha256=media_info.sha256, ) responder = await self.media_storage.fetch_media(file_info) @@ -625,6 +629,7 @@ async def select_or_generate_remote_thumbnail( desired_method, desired_type, animated=animated, + sha256=media_info.sha256, ) if file_path: @@ -686,6 +691,7 @@ async def respond_remote_thumbnail( url_cache=False, server_name=server_name, for_federation=False, + sha256=media_info.sha256, ) async def _select_and_respond_with_thumbnail( @@ -702,6 +708,7 @@ async def _select_and_respond_with_thumbnail( for_federation: bool, media_info: LocalMedia | None = None, server_name: str | None = None, + sha256: str | None = None, ) -> None: """ Respond to a request with an appropriate thumbnail from the previously generated thumbnails. @@ -718,6 +725,7 @@ async def _select_and_respond_with_thumbnail( server_name: The server name, if this is a remote thumbnail. for_federation: whether the request is from the federation /thumbnail request media_info: metadata about the media being requested. + sha256: The SHA-256 of the media, if known. """ logger.debug( "_select_and_respond_with_thumbnail: media_id=%s desired=%sx%s (%s) thumbnail_infos=%s", @@ -742,6 +750,7 @@ async def _select_and_respond_with_thumbnail( file_id, url_cache, server_name, + sha256, ) if not file_info: logger.info("Couldn't find a thumbnail matching the desired inputs") @@ -797,6 +806,7 @@ async def _select_and_respond_with_thumbnail( t_method=file_info.thumbnail.method, t_type=file_info.thumbnail.type, animated=regen_animated, + sha256=sha256, ) else: await self.media_repo.generate_local_exact_thumbnail( @@ -807,6 +817,7 @@ async def _select_and_respond_with_thumbnail( t_type=file_info.thumbnail.type, url_cache=url_cache, animated=regen_animated, + sha256=sha256, ) responder = await self.media_storage.fetch_media(file_info) @@ -863,6 +874,7 @@ def _select_thumbnail( file_id: str, url_cache: bool, server_name: str | None, + sha256: str | None = None, ) -> FileInfo | None: """ Choose an appropriate thumbnail from the previously generated thumbnails. @@ -876,6 +888,7 @@ def _select_thumbnail( file_id: The ID of the media that a thumbnail is being requested for. url_cache: True if this is from a URL cache. server_name: The server name, if this is a remote thumbnail. + sha256: The SHA-256 of the media, if known. Returns: The thumbnail which best matches the desired parameters. @@ -976,6 +989,7 @@ def _select_thumbnail( url_cache=url_cache, server_name=server_name, thumbnail=thumbnail_info, + sha256=sha256, ) # No matching thumbnail was found. diff --git a/synapse/storage/databases/main/media_repository.py b/synapse/storage/databases/main/media_repository.py index 50664d63e5..eaff0065a0 100644 --- a/synapse/storage/databases/main/media_repository.py +++ b/synapse/storage/databases/main/media_repository.py @@ -988,6 +988,34 @@ def _delete_url_cache_media_txn(txn: LoggingTransaction) -> None: "delete_url_cache_media", _delete_url_cache_media_txn ) + async def is_media_sha256_in_use( + self, sha256: str, origin: str | None = None + ) -> bool: + """Whether any media is still stored under the given SHA-256 path. + + Args: + sha256: The SHA-256 of the media. + origin: The origin to check remote media from, or None to check local + media. URL cache media is ignored, as it is not stored by SHA-256. + """ + if origin is None: + sql = """ + SELECT 1 FROM local_media_repository + WHERE sha256 = ? AND url_cache IS NULL + LIMIT 1 + """ + rows = await self.db_pool.execute("is_media_sha256_in_use", sql, sha256) + else: + sql = """ + SELECT 1 FROM remote_media_cache + WHERE media_origin = ? AND sha256 = ? + LIMIT 1 + """ + rows = await self.db_pool.execute( + "is_media_sha256_in_use", sql, origin, sha256 + ) + return bool(rows) + async def get_is_hash_quarantined(self, sha256: str) -> bool: """Get whether a specific sha256 hash digest matches any quarantined media. diff --git a/tests/media/test_media_storage.py b/tests/media/test_media_storage.py index da5fa8a140..ced3531b2e 100644 --- a/tests/media/test_media_storage.py +++ b/tests/media/test_media_storage.py @@ -24,7 +24,7 @@ from binascii import unhexlify from io import BytesIO from typing import Any, BinaryIO, ClassVar, Literal -from unittest.mock import MagicMock, Mock, patch +from unittest.mock import AsyncMock, MagicMock, Mock, patch from urllib import parse import attr @@ -47,7 +47,11 @@ from synapse.logging.context import make_deferred_yieldable from synapse.media._base import FileInfo, ThumbnailInfo from synapse.media.filepath import MediaFilePaths -from synapse.media.media_storage import MediaStorage, ReadableFileWrapper +from synapse.media.media_storage import ( + MediaStorage, + ReadableFileWrapper, + SHA256TransparentIOWriter, +) from synapse.media.storage_provider import ( FileStorageProviderBackend, StorageProviderWrapper, @@ -138,6 +142,127 @@ async def test_ensure_media() -> None: # actually done :/ self.get_success(test_ensure_media()) + def test_store_file_uses_sha256_path(self) -> None: + """New media is stored under its SHA-256, not its media ID.""" + self.get_success( + self.media_storage.store_file(BytesIO(SMALL_PNG), FileInfo(None, "abcdef")) + ) + + sha256_path = self.filepaths.local_media_filepath(SMALL_PNG_SHA256) + with open(sha256_path, "rb") as f: + self.assertEqual(f.read(), SMALL_PNG) + self.assertFalse(os.path.exists(self.filepaths.local_media_filepath("abcdef"))) + # The temporary file was moved into place, not left behind. + self.assertEqual(os.listdir(self.primary_base_path), ["local_content"]) + + # Storage providers receive the SHA-256 path too. + rel_path = self.filepaths.local_media_filepath_rel(SMALL_PNG_SHA256) + self.assertTrue( + os.path.exists(os.path.join(self.secondary_base_path, rel_path)) + ) + + def test_store_into_file_shares_sha256(self) -> None: + """Callers reuse the SHA-256 computed by `store_into_file`.""" + + async def store() -> str: + async with self.media_storage.store_into_file(FileInfo(None, "abcdef")) as ( + f, + _, + ): + writer = SHA256TransparentIOWriter.get_or_wrap(f) + self.assertIs(writer, f) + writer.write(SMALL_PNG) + return writer.hexdigest() + + self.assertEqual(self.get_success(store()), SMALL_PNG_SHA256) + + def test_duplicate_sha256_is_not_replaced(self) -> None: + """A second upload of the same bytes keeps the existing file and backup.""" + path = self.get_success( + self.media_storage.store_file(BytesIO(SMALL_PNG), FileInfo(None, "abcdef")) + ) + local_inode = os.stat(path).st_ino + backup = os.path.join( + self.secondary_base_path, + self.filepaths.local_media_filepath_rel(SMALL_PNG_SHA256), + ) + backup_inode = os.stat(backup).st_ino + + self.get_success( + self.media_storage.store_file(BytesIO(SMALL_PNG), FileInfo(None, "ghijkl")) + ) + + self.assertEqual(os.stat(path).st_ino, local_inode) + self.assertEqual(os.stat(backup).st_ino, backup_inode) + with open(path, "rb") as f: + self.assertEqual(f.read(), SMALL_PNG) + + def test_check_failure_before_publish_leaves_no_sha256_file(self) -> None: + """A failure before `store_into_file` exits deletes only the temp file.""" + + async def store() -> None: + async with self.media_storage.store_into_file(FileInfo(None, "abcdef")) as ( + f, + _, + ): + f.write(SMALL_PNG) + raise RuntimeError("quota") + + self.get_failure(store(), RuntimeError) + self.assertFalse( + os.path.exists(self.filepaths.local_media_filepath(SMALL_PNG_SHA256)) + ) + self.assertEqual(os.listdir(self.primary_base_path), []) + + def test_failed_thumbnail_write_keeps_shared_file(self) -> None: + """A failed write to a shared SHA-256 thumbnail neither truncates nor deletes it.""" + thumbnail = ThumbnailInfo( + width=32, height=32, method="scale", type="image/png", length=4 + ) + file_info = FileInfo( + None, "abcdef", thumbnail=thumbnail, sha256=SMALL_PNG_SHA256 + ) + path = self.get_success( + self.media_storage.store_file(BytesIO(b"good"), file_info) + ) + + with patch.object( + self.media_storage._spam_checker_module_callbacks, + "check_media_file_for_spam", + AsyncMock(return_value=(Codes.FORBIDDEN, {})), + ): + self.get_failure( + self.media_storage.store_file( + BytesIO(b"other"), attr.evolve(file_info, file_id="ghijkl") + ), + Exception, + ) + + with open(path, "rb") as f: + self.assertEqual(f.read(), b"good") + + def test_ensure_media_prefers_sha256_path(self) -> None: + """The SHA-256 path is used if it exists, else the media ID path.""" + file_info = FileInfo(None, "abcdef", sha256=SMALL_PNG_SHA256) + legacy_path = self.filepaths.local_media_filepath("abcdef") + sha256_path = self.filepaths.local_media_filepath(SMALL_PNG_SHA256) + + async def local_path() -> str: + async with self.media_storage.ensure_media_is_in_local_cache( + file_info + ) as path: + return path + + os.makedirs(os.path.dirname(legacy_path)) + with open(legacy_path, "wb") as f: + f.write(SMALL_PNG) + self.assertEqual(self.get_success(local_path()), legacy_path) + + os.makedirs(os.path.dirname(sha256_path)) + with open(sha256_path, "wb") as f: + f.write(SMALL_PNG) + self.assertEqual(self.get_success(local_path()), sha256_path) + @attr.s(auto_attribs=True, slots=True, frozen=True) class TestImage: diff --git a/tests/rest/admin/test_media.py b/tests/rest/admin/test_media.py index 6469d305b2..1a9488d495 100644 --- a/tests/rest/admin/test_media.py +++ b/tests/rest/admin/test_media.py @@ -35,7 +35,7 @@ from synapse.util.clock import Clock from tests import unittest -from tests.test_utils import SMALL_CMYK_JPEG, SMALL_PNG +from tests.test_utils import SMALL_CMYK_JPEG, SMALL_PNG, SMALL_PNG_SHA256 from tests.unittest import override_config VALID_TIMESTAMP = 1609459200000 # 2021-01-01 in milliseconds @@ -85,7 +85,7 @@ def _cache_remote_media(self, file_id: str) -> None: time_now_ms=self.clock.time_msec(), upload_name="test.png", filesystem_id=file_id, - sha256=file_id, + sha256=SMALL_PNG_SHA256, ) ) @@ -322,7 +322,7 @@ def test_delete_media(self) -> None: ) # Test if the file exists - local_path = self.filepaths.local_media_filepath(media_id) + local_path = self.filepaths.local_media_filepath(SMALL_PNG_SHA256) self.assertTrue(os.path.exists(local_path)) url = "/_synapse/admin/v1/media/%s/%s" % (self.server_name, media_id) @@ -360,6 +360,28 @@ def test_delete_media(self) -> None: # Test if the file is deleted self.assertFalse(os.path.exists(local_path)) + def test_delete_media_with_shared_sha256(self) -> None: + """ + Tests that a file shared by media with the same content is only deleted + once no media references it + """ + media_ids = [ + self.helper.upload_media( + SMALL_PNG, tok=self.admin_user_tok, expect_code=200 + )["content_uri"].split("/")[-1] + for _ in range(2) + ] + local_path = self.filepaths.local_media_filepath(SMALL_PNG_SHA256) + + for i, media_id in enumerate(media_ids): + channel = self.make_request( + "DELETE", + f"/_synapse/admin/v1/media/{self.server_name}/{media_id}", + access_token=self.admin_user_tok, + ) + self.assertEqual(200, channel.code, msg=channel.json_body) + self.assertEqual(i == 0, os.path.exists(local_path)) + class DeleteMediaByDateSizeTestCase(_AdminMediaTests): servlets = [ @@ -513,7 +535,7 @@ def test_delete_media_never_accessed(self, use_legacy_url: bool) -> None: # test that the file exists media_id = server_and_media_id.split("/")[1] - local_path = self.filepaths.local_media_filepath(media_id) + local_path = self.filepaths.local_media_filepath(SMALL_PNG_SHA256) self.assertTrue(os.path.exists(local_path)) # timestamp after upload/create @@ -722,8 +744,7 @@ def _access_media( """ Try to access a media and check the result """ - media_id = server_and_media_id.split("/")[1] - local_path = self.filepaths.local_media_filepath(media_id) + local_path = self.filepaths.local_media_filepath(SMALL_PNG_SHA256) channel = self.make_request( "GET", diff --git a/tests/rest/admin/test_user.py b/tests/rest/admin/test_user.py index 84bb7e319b..7f96ce426d 100644 --- a/tests/rest/admin/test_user.py +++ b/tests/rest/admin/test_user.py @@ -64,7 +64,7 @@ from tests import unittest from tests.replication._base import BaseMultiWorkerStreamTestCase -from tests.test_utils import SMALL_PNG +from tests.test_utils import SMALL_PNG, SMALL_PNG_SHA256 from tests.test_utils.event_injection import inject_event from tests.unittest import override_config @@ -4034,12 +4034,10 @@ def test_delete_media(self) -> None: other_user_tok = self.login("user", "pass") media_ids = self._create_media_for_user(other_user_tok, number_media) - # Test if the file exists - local_paths = [] - for media_id in media_ids: - local_path = self.filepaths.local_media_filepath(media_id) - self.assertTrue(os.path.exists(local_path)) - local_paths.append(local_path) + # Test if the file exists. All the media share the same content, and so + # the same file. + local_path = self.filepaths.local_media_filepath(SMALL_PNG_SHA256) + self.assertTrue(os.path.exists(local_path)) channel = self.make_request( "DELETE", @@ -4053,8 +4051,7 @@ def test_delete_media(self) -> None: self.assertCountEqual(channel.json_body["deleted_media"], media_ids) # Test if the file is deleted - for local_path in local_paths: - self.assertFalse(os.path.exists(local_path)) + self.assertFalse(os.path.exists(local_path)) def test_order_by(self) -> None: """ diff --git a/tests/rest/client/test_media.py b/tests/rest/client/test_media.py index 014ad18e30..2f5c34389f 100644 --- a/tests/rest/client/test_media.py +++ b/tests/rest/client/test_media.py @@ -79,7 +79,7 @@ small_png_with_transparency, ) from tests.server import FakeChannel, FakeTransport, ThreadedMemoryReactorClock -from tests.test_utils import SMALL_PNG +from tests.test_utils import SMALL_PNG, SMALL_PNG_SHA256 from tests.unittest import override_config try: @@ -145,7 +145,7 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: time_now_ms=clock.time_msec(), upload_name="test.png", filesystem_id=file_id, - sha256=file_id, + sha256=SMALL_PNG_SHA256, ) ) self.register_user("user", "password") @@ -2737,7 +2737,7 @@ def test_authenticated_media(self) -> None: time_now_ms=self.clock.time_msec(), upload_name="remote_test.png", filesystem_id=file_id, - sha256=file_id, + sha256=SMALL_PNG_SHA256, ) ) @@ -2745,7 +2745,11 @@ def test_authenticated_media(self) -> None: if self.extra_config == {"dynamic_thumbnails": False}: self.get_success( self.repo._generate_thumbnails( - "lonelyIsland", "52", file_id, "image/png" + "lonelyIsland", + "52", + file_id, + "image/png", + sha256=SMALL_PNG_SHA256, ) ) @@ -2800,6 +2804,7 @@ def test_authenticated_media(self) -> None: "user_id": "someone", "url_cache": None, "authenticated": False, + "sha256": SMALL_PNG_SHA256, }, desc="store_local_media", ) @@ -2869,7 +2874,7 @@ def test_authenticated_media_etag(self) -> None: time_now_ms=self.clock.time_msec(), upload_name="remote_test.png", filesystem_id=file_id, - sha256=file_id, + sha256=SMALL_PNG_SHA256, ) ) @@ -2877,7 +2882,11 @@ def test_authenticated_media_etag(self) -> None: if self.extra_config == {"dynamic_thumbnails": False}: self.get_success( self.repo._generate_thumbnails( - "lonelyIsland", "52", file_id, "image/png" + "lonelyIsland", + "52", + file_id, + "image/png", + sha256=SMALL_PNG_SHA256, ) ) diff --git a/tests/rest/media/test_domain_blocking.py b/tests/rest/media/test_domain_blocking.py index 0bdbaa676d..c633513252 100644 --- a/tests/rest/media/test_domain_blocking.py +++ b/tests/rest/media/test_domain_blocking.py @@ -27,7 +27,7 @@ from synapse.util.clock import Clock from tests import unittest -from tests.test_utils import SMALL_PNG +from tests.test_utils import SMALL_PNG, SMALL_PNG_SHA256 from tests.unittest import override_config @@ -60,7 +60,7 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: time_now_ms=clock.time_msec(), upload_name="test.png", filesystem_id=file_id, - sha256=file_id, + sha256=SMALL_PNG_SHA256, ) )