diff --git a/synapse/api/auth/base.py b/synapse/api/auth/base.py
index f97a71caf7..baf458d272 100644
--- a/synapse/api/auth/base.py
+++ b/synapse/api/auth/base.py
@@ -239,6 +239,51 @@ async def check_can_change_room_list(
 
         return user_level >= send_level
 
+    async def is_moderator(self, room_id: str, requester: Requester) -> bool:
+        """Determine whether the user is moderator of the room.
+
+        Args:
+            room_id: The room_id of the room to check
+            requester: The user making the request
+        """
+        is_admin = await self.is_server_admin(requester)
+        if is_admin:
+            return True
+        await self.check_user_in_room(room_id, requester)
+
+        # We currently require the user is a "moderator" in the room. We do this
+        # by checking if they would (theoretically) be able to change the
+        # m.room.canonical_alias events
+
+        auth_events = await self._storage_controllers.state.get_current_state(
+            room_id,
+            StateFilter.from_types(
+                [
+                    POWER_KEY,
+                    CREATE_KEY,
+                ]
+            ),
+        )
+
+        send_level = event_auth.get_send_level(
+            EventTypes.CanonicalAlias,
+            "",
+            auth_events.get(POWER_KEY),
+        )
+
+        user_level = event_auth.get_user_power_level(
+            requester.user.to_string(), auth_events
+        )
+        # Check multiple moderator-level actions
+        kick_level = event_auth.get_named_level(auth_events, "kick", 50)
+        ban_level = event_auth.get_named_level(auth_events, "ban", 50)
+        redact_level = event_auth.get_named_level(auth_events, "redact", 50)
+
+        # Consider someone a moderator if they can perform key mod actions
+        moderator_threshold = min(kick_level, ban_level, redact_level)
+
+        return user_level >= send_level and user_level >= moderator_threshold
+
     @staticmethod
     def has_access_token(request: Request) -> bool:
         """Checks if the request has an access_token.
diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py
index 6af05960f5..8e35225eca 100644
--- a/synapse/config/experimental.py
+++ b/synapse/config/experimental.py
@@ -592,7 +592,12 @@ def read_config(
         # MSC3911: Linking Media to Events
         self.msc3911_enabled: bool = experimental.get("msc3911_enabled", False)
 
-        # Disable the current media create and upload endpoints
+        # MSC3911: Disable the current media create and upload endpoints
         self.msc3911_unrestricted_media_upload_disabled: bool = experimental.get(
             "msc3911_unrestricted_media_upload_disabled", False
         )
+
+        # MSC3911: Retention time for media that are attached to redacted events
+        self.msc3911_redacted_event_media_cleanup_interval: int = experimental.get(
+            "msc3911_redacted_event_media_cleanup_interval", 48 * 60 * 60 * 1000
+        )
diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index 3f46032a43..f65c03b74d 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -156,14 +156,17 @@ async def get_event(
         event_id: str,
         show_redacted: bool = False,
     ) -> Optional[EventBase]:
-        """Retrieve a single specified event.
+        """Retrieve a single specified event from the database, depending on the
+        show_redacted flag, it either hides or exposes redacted content.
 
         Args:
             user: The local user requesting the event
             room_id: The expected room id. We'll return None if the
                 event's room does not match.
             event_id: The event ID to obtain.
-            show_redacted: Should the full content of redacted events be returned?
+            show_redacted: If False (default), the returned event will have its redacted
+                content removed as users normally see it. When True, the event will be
+                returned exactly as stored, even if it’s been redacted.
         Returns:
             An event, or None if there is no event matching this ID.
         Raises:
diff --git a/synapse/media/media_repository.py b/synapse/media/media_repository.py
index 3acfd48ca3..c63aa9efdd 100644
--- a/synapse/media/media_repository.py
+++ b/synapse/media/media_repository.py
@@ -33,6 +33,7 @@
 import twisted.web.http
 from twisted.internet.defer import Deferred
 
+from synapse.api.auth.base import BaseAuth
 from synapse.api.constants import EventTypes, HistoryVisibility, Membership
 from synapse.api.errors import (
     Codes,
@@ -213,7 +214,7 @@ async def delete_old_remote_media(self, before_ts: int) -> Dict[str, int]:
             "Sorry Mario, your MediaRepository related function is in another castle"
         )
 
-    async def delete_local_media_ids(
+    async def delete_media_from_disk_by_media_ids(
         self, media_ids: List[str]
     ) -> Tuple[List[str], int]:
         raise NotImplementedError(
@@ -312,13 +313,13 @@ async def validate_media_restriction(
         return attachments
 
     async def is_media_visible(
-        self, requesting_user: UserID, media_info_object: Union[LocalMedia, RemoteMedia]
+        self, requester: Requester, media_info_object: Union[LocalMedia, RemoteMedia]
     ) -> None:
         """
         Verify that media requested for download should be visible to the user making
         the request
         """
-
+        requester_user_id_str = requester.user.to_string()
         if not self.enable_media_restriction:
             return
 
@@ -329,7 +330,7 @@ async def is_media_visible(
             # When the media has not been attached yet, only the originating user can
             # see it. But once attachments have been formed, standard other rules apply
             if isinstance(media_info_object, LocalMedia) and (
-                requesting_user.to_string() == str(media_info_object.user_id)
+                requester_user_id_str == media_info_object.user_id
             ):
                 return
 
@@ -338,7 +339,7 @@ async def is_media_visible(
                 "Media ID ('%s') as requested by '%s' was restricted but had no "
                 "attachments",
                 media_info_object.media_id,
-                requesting_user.to_string(),
+                requester_user_id_str,
             )
             raise UnauthorizedRequestAPICallError(
                 f"Media requested ('{media_info_object.media_id}') is restricted"
@@ -348,7 +349,29 @@ async def is_media_visible(
         attached_profile_user_id = media_info_object.attachments.profile_user_id
 
         if attached_event_id:
+            # Check if event is redacted or not. If it is redacted, normal user no
+            # longer have access to the media, but only moderators can see the media
+            # until it gets deleted permanently after certain period of time.
             event_base = await self.store.get_event(attached_event_id)
+            if event_base.internal_metadata.is_redacted():
+                # In the case of redacted media, check the requester's power level of the room
+                room_id = event_base.room_id
+                assert isinstance(self.auth, BaseAuth)
+                is_moderator = await self.auth.is_moderator(room_id, requester)
+                if is_moderator:
+                    # Moderator still has access to the media that is attached to redacted event
+                    return
+                else:
+                    logger.debug(
+                        "Media ID (%s) as requested by '%s' was redacted from event '%s'",
+                        media_info_object.media_id,
+                        requester_user_id_str,
+                        attached_event_id,
+                    )
+                    raise UnauthorizedRequestAPICallError(
+                        f"Media requested ('{media_info_object.media_id}') is restricted"
+                    )
+
             if event_base.is_state():
                 # The standard event visibility utility, filter_events_for_client(),
                 # does not seem to meet the needs of a good UX when restricting and
@@ -368,13 +391,13 @@ async def is_media_visible(
                     membership_now,
                     _,
                 ) = await self.store.get_local_current_membership_for_user_in_room(
-                    requesting_user.to_string(), event_base.room_id
+                    requester_user_id_str, event_base.room_id
                 )
 
                 if not membership_now:
                     membership_now = Membership.LEAVE
 
-                membership_state_key = (EventTypes.Member, requesting_user.to_string())
+                membership_state_key = (EventTypes.Member, requester_user_id_str)
                 types = (_HISTORY_VIS_KEY, membership_state_key)
 
                 # and history visibility and membership of THEN
@@ -467,7 +490,7 @@ async def is_media_visible(
                 storage_controllers = self.hs.get_storage_controllers()
                 filtered_events = await filter_events_for_client(
                     storage_controllers,
-                    requesting_user.to_string(),
+                    requester_user_id_str,
                     [event_base],
                 )
                 if len(filtered_events) > 0:
@@ -487,22 +510,22 @@ async def is_media_visible(
             if self.hs.config.server.limit_profile_requests_to_users_who_share_rooms:
                 # First take care of the case where the requesting user IS the creating
                 # user. The other function below does not handle this.
-                if requesting_user.to_string() == attached_profile_user_id.to_string():
+                if requester_user_id_str == attached_profile_user_id:
                     return
 
                 # This call returns a set() that contains which of the "other_user_ids"
                 # share a room. Since we give it only one, if bool(set()) is True, then they
                 # share some room or had at least one invite between them.
                 if not await self.store.do_users_share_a_room_joined_or_invited(
-                    requesting_user.to_string(),
-                    [attached_profile_user_id.to_string()],
+                    requester_user_id_str,
+                    [attached_profile_user_id],
                 ):
                     logger.debug(
                         "Media ID (%s) as requested by '%s' was restricted by "
                         "profile, but was not allowed(is "
                         "'limit_profile_requests_to_users_who_share_rooms' enabled?)",
                         media_info_object.media_id,
-                        requesting_user.to_string(),
+                        requester_user_id_str,
                     )
 
                     raise UnauthorizedRequestAPICallError(
@@ -521,7 +544,7 @@ async def is_media_visible(
             "Media ID (%s) as requested by '%s' was restricted, but was not "
             "allowed(media_attachments=%s)",
             media_info_object.media_id,
-            requesting_user.to_string(),
+            requester_user_id_str,
             media_info_object.attachments,
         )
         raise UnauthorizedRequestAPICallError(
@@ -609,6 +632,13 @@ def __init__(self, hs: "HomeServer"):
             self._start_update_recently_accessed, UPDATE_RECENTLY_ACCESSED_TS
         )
 
+        self.clock.looping_call(
+            self._redacted_media_cleanup,
+            float(
+                self.hs.config.experimental.msc3911_redacted_event_media_cleanup_interval
+            ),
+        )
+
         # Media retention configuration options
         self._media_retention_local_media_lifetime_ms = (
             hs.config.media.media_retention_local_media_lifetime_ms
@@ -949,7 +979,7 @@ async def get_local_media_info(
             # The file has been uploaded, so stop looping
             if media_info.media_length is not None:
                 if isinstance(request.requester, Requester):
-                    await self.is_media_visible(request.requester.user, media_info)
+                    await self.is_media_visible(request.requester, media_info)
                 return media_info
 
             # Check if the media ID has expired and still hasn't been uploaded to.
@@ -1012,7 +1042,7 @@ async def get_local_media(
             if requester is not None:
                 # Only check media visibility if this is for a local request. This will
                 # raise directly back to the client if not visible
-                await self.is_media_visible(requester.user, media_info)
+                await self.is_media_visible(requester, media_info)
             restrictions = await self.validate_media_restriction(
                 request, media_info, None, federation
             )
@@ -1242,7 +1272,7 @@ async def _get_remote_media_impl(
             # retrieved from the remote.
             if self.enable_media_restriction and requester is not None:
                 # This will raise directly back to the client if not visible
-                await self.is_media_visible(requester.user, media_info)
+                await self.is_media_visible(requester, media_info)
 
             # file_id is the ID we use to track the file locally. If we've already
             # seen the file then reuse the existing ID, otherwise generate a new
@@ -1300,7 +1330,7 @@ async def _get_remote_media_impl(
             and requester is not None
         ):
             # This will raise directly back to the client if not visible
-            await self.is_media_visible(requester.user, media_info)
+            await self.is_media_visible(requester, media_info)
 
         file_id = media_info.filesystem_id
         if not media_info.media_type:
@@ -2067,7 +2097,7 @@ async def delete_old_remote_media(self, before_ts: int) -> Dict[str, int]:
 
         return {"deleted": deleted}
 
-    async def delete_local_media_ids(
+    async def delete_media_from_disk_by_media_ids(
         self, media_ids: List[str]
     ) -> Tuple[List[str], int]:
         """
@@ -2078,7 +2108,7 @@ async def delete_local_media_ids(
         Returns:
             A tuple of (list of deleted media IDs, total deleted media IDs).
         """
-        return await self._remove_local_media_from_disk(media_ids)
+        return await self._remove_media_from_disk(media_ids)
 
     async def delete_old_local_media(
         self,
@@ -2111,9 +2141,9 @@ async def delete_old_local_media(
             include_quarantined_media=delete_quarantined_media,
             include_protected_media=delete_protected_media,
         )
-        return await self._remove_local_media_from_disk(old_media)
+        return await self._remove_media_from_disk(old_media)
 
-    async def _remove_local_media_from_disk(
+    async def _remove_media_from_disk(
         self, media_ids: List[str]
     ) -> Tuple[List[str], int]:
         """
@@ -2149,3 +2179,17 @@ async def _remove_local_media_from_disk(
             removed_media.append(media_id)
 
         return removed_media, len(removed_media)
+
+    async def _redacted_media_cleanup(self) -> None:
+        """Periodically deletes media attached to redacted events from disk."""
+        redacted_event_ids = await self.store.get_redacted_event_ids_before_interval(
+            self.hs.config.experimental.msc3911_redacted_event_media_cleanup_interval
+        )
+        media_ids_to_redact = []
+        for event_id in redacted_event_ids:
+            attached_media_ids = await self.store.get_media_ids_attached_to_event(
+                event_id
+            )
+            media_ids_to_redact.extend(attached_media_ids)
+        if media_ids_to_redact:
+            await self.delete_media_from_disk_by_media_ids(media_ids_to_redact)
diff --git a/synapse/media/thumbnailer.py b/synapse/media/thumbnailer.py
index 0b2b37d808..be38d4ae89 100644
--- a/synapse/media/thumbnailer.py
+++ b/synapse/media/thumbnailer.py
@@ -305,7 +305,7 @@ async def respond_local_thumbnail(
             if requester is not None:
                 # Only check media visibility if this is for a local request. This will
                 # raise directly back to the client if not visible
-                await self.media_repo.is_media_visible(requester.user, media_info)
+                await self.media_repo.is_media_visible(requester, media_info)
             restrictions = await self.media_repo.validate_media_restriction(
                 request, media_info, None, for_federation
             )
@@ -365,7 +365,7 @@ async def select_or_generate_local_thumbnail(
             if requester is not None:
                 # Only check media visibility if this is for a local request. This will
                 # raise directly back to the client if not visible
-                await self.media_repo.is_media_visible(requester.user, media_info)
+                await self.media_repo.is_media_visible(requester, media_info)
             restrictions = await self.media_repo.validate_media_restriction(
                 request, None, media_id, for_federation
             )
@@ -482,7 +482,7 @@ async def select_or_generate_remote_thumbnail(
         # if MSC3911 is enabled, check visibility of the media for the user
         if self.enable_media_restriction and requester is not None:
             # This will raise directly back to the client if not visible
-            await self.media_repo.is_media_visible(requester.user, media_info)
+            await self.media_repo.is_media_visible(requester, media_info)
 
         # Check if the media is cached on the client, if so return 304.
         if check_for_cached_entry_and_respond(request):
@@ -572,7 +572,7 @@ async def respond_remote_thumbnail(
         # if MSC3911 is enabled, check visibility of the media for the user
         if self.enable_media_restriction and requester is not None:
             # This will raise directly back to the client if not visible
-            await self.media_repo.is_media_visible(requester.user, media_info)
+            await self.media_repo.is_media_visible(requester, media_info)
 
         # Check if the media is cached on the client, if so return 304.
         if check_for_cached_entry_and_respond(request):
diff --git a/synapse/rest/admin/media.py b/synapse/rest/admin/media.py
index 195f22a4c2..47e4c326fc 100644
--- a/synapse/rest/admin/media.py
+++ b/synapse/rest/admin/media.py
@@ -282,9 +282,10 @@ async def on_DELETE(
 
         logger.info("Deleting local media by ID: %s", media_id)
 
-        deleted_media, total = await self.media_repository.delete_local_media_ids(
-            [media_id]
-        )
+        (
+            deleted_media,
+            total,
+        ) = await self.media_repository.delete_media_from_disk_by_media_ids([media_id])
         return HTTPStatus.OK, {"deleted_media": deleted_media, "total": total}
 
 
@@ -446,7 +447,10 @@ async def on_DELETE(
             start, limit, user_id, order_by, direction
         )
 
-        deleted_media, total = await self.media_repository.delete_local_media_ids(
+        (
+            deleted_media,
+            total,
+        ) = await self.media_repository.delete_media_from_disk_by_media_ids(
             [m.media_id for m in media]
         )
 
diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py
index d9ef93f826..adab9fe8de 100644
--- a/synapse/storage/databases/main/events_worker.py
+++ b/synapse/storage/databases/main/events_worker.py
@@ -206,8 +206,8 @@ class EventRedactBehaviour(Enum):
     What to do when retrieving a redacted event from the database.
     """
 
-    as_is = auto()
-    redact = auto()
+    as_is = auto()  # Leave redacted events untouched
+    redact = auto()  # Strip content for redacted events
     block = auto()
 
 
@@ -2672,3 +2672,39 @@ async def get_metadata_for_event(
             sender=row[0],
             received_ts=row[1],
         )
+
+    async def get_redacted_event_ids_before_interval(
+        self, interval_ms: int
+    ) -> List[str]:
+        """Get the event ids of redacted events before the given interval.
+
+        Args:
+            interval_ms: The interval in milliseconds.
+
+        Returns:
+            The list of event IDs of redacted events before the given interval,
+            or empty list if no such event exists.
+        """
+
+        def _get_redacted_event_id_before_interval_txn(
+            txn: LoggingTransaction, threshold_ts: int
+        ) -> List[str]:
+            sql = """
+                SELECT redacts
+                FROM redactions
+                WHERE received_ts < ?
+            """
+
+            txn.execute(sql, (threshold_ts,))
+            rows = txn.fetchall()
+
+            if not rows:
+                return []
+            return [row[0] for row in rows]
+
+        threshold_ts = self._clock.time_msec() - interval_ms
+        return await self.db_pool.runInteraction(
+            "_get_redacted_event_id_before_interval_txn",
+            _get_redacted_event_id_before_interval_txn,
+            threshold_ts,
+        )
diff --git a/synapse/storage/databases/main/media_repository.py b/synapse/storage/databases/main/media_repository.py
index 18ad97959e..7a692ec4c3 100644
--- a/synapse/storage/databases/main/media_repository.py
+++ b/synapse/storage/databases/main/media_repository.py
@@ -19,6 +19,7 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
+import json
 import logging
 from enum import Enum
 from http import HTTPStatus
@@ -45,6 +46,7 @@
     LoggingDatabaseConnection,
     LoggingTransaction,
 )
+from synapse.storage.engines import PostgresEngine
 from synapse.types import JsonDict, UserID
 from synapse.util import json_encoder
 
@@ -72,15 +74,15 @@ class MediaRestrictions:
     """
 
     event_id: Optional[str] = None
-    profile_user_id: Optional[UserID] = None
+    profile_user_id: Optional[str] = None
 
     def to_dict(self) -> dict:
         if self.event_id:
-            return {"org.matrix.msc3911.restrictions": {"event_id": str(self.event_id)}}
+            return {"org.matrix.msc3911.restrictions": {"event_id": self.event_id}}
         if self.profile_user_id:
             return {
                 "org.matrix.msc3911.restrictions": {
-                    "profile_user_id": str(self.profile_user_id)
+                    "profile_user_id": self.profile_user_id
                 }
             }
         return {}
@@ -370,7 +372,7 @@ def get_local_media_by_user_paginate_txn(
                     sha256,
                     restricted,
                     ma.restrictions_json->'restrictions'->>'event_id' AS event_id,
-                    ma.restrictions_json->'restrictions'->>'event_id' AS profile_user_id
+                    ma.restrictions_json->'restrictions'->>'profile_user_id' AS profile_user_id
                 FROM local_media_repository AS lmr
                 -- a LEFT JOIN allows values from the right table to be NULL if non-existent
                 LEFT JOIN media_attachments AS ma ON lmr.media_id = ma.media_id
@@ -1170,7 +1172,7 @@ async def get_media_restrictions(
         if row:
             event_id = row[0][0]
             # Because the UserID object can be None, the 'to_string()' method may not exist
-            profile_user_id = UserID.from_string(row[0][1]) if row[0][1] else None
+            profile_user_id = row[0][1] if row[0][1] else None
             return MediaRestrictions(event_id=event_id, profile_user_id=profile_user_id)
 
         return None
@@ -1308,3 +1310,34 @@ def set_media_restricted_to_user_profile_txn(
             media_id=media_id,
             json_dict=json_object,
         )
+
+    async def get_media_ids_attached_to_event(self, event_id: str) -> list[str]:
+        """
+        Get a list of media_ids that are attached to a specific event_id.
+        """
+
+        def get_media_ids_attached_to_event_txn(txn: LoggingTransaction) -> list[str]:
+            if isinstance(self.db_pool.engine, PostgresEngine):
+                # Use GIN index for Postgres
+                sql = """
+                SELECT media_id
+                FROM media_attachments
+                WHERE restrictions_json @> %s
+                """
+                json_param = json.dumps({"restrictions": {"event_id": event_id}})
+                txn.execute(sql, (json_param,))
+            else:
+                # Use cross-database compatible query for the rest
+                sql = """
+                SELECT media_id
+                FROM media_attachments
+                WHERE restrictions_json->'restrictions'->>'event_id' = ?
+                """
+                txn.execute(sql, (event_id,))
+
+            return [row[0] for row in txn.fetchall()]
+
+        return await self.db_pool.runInteraction(
+            "get_media_ids_attached_to_event",
+            get_media_ids_attached_to_event_txn,
+        )
diff --git a/tests/federation/test_federation_media.py b/tests/federation/test_federation_media.py
index b453b83115..beaca5abab 100644
--- a/tests/federation/test_federation_media.py
+++ b/tests/federation/test_federation_media.py
@@ -665,7 +665,7 @@ def test_downloading_remote_media_with_restrictions_is_in_database(self) -> None
         )
         assert isinstance(restrictions, MediaRestrictions)
         assert restrictions.profile_user_id is not None
-        assert restrictions.profile_user_id.to_string() == "@bob:example.com"
+        assert restrictions.profile_user_id == "@bob:example.com"
 
     def test_downloading_remote_media_with_no_restrictions_does_not_save_to_db(
         self,
diff --git a/tests/rest/client/test_media.py b/tests/rest/client/test_media.py
index f3f5339d4f..5bf8874d9d 100644
--- a/tests/rest/client/test_media.py
+++ b/tests/rest/client/test_media.py
@@ -69,7 +69,7 @@
 from synapse.storage.databases.main.media_repository import (
     LocalMedia,
 )
-from synapse.types import JsonDict, UserID, create_requester
+from synapse.types import JsonDict, Requester, UserID, create_requester
 from synapse.util import Clock, json_encoder
 from synapse.util.stringutils import parse_and_validate_mxc_uri, random_string
 
@@ -3748,8 +3748,18 @@ def assert_expected_result(
         else:
             maybe_assert_exception = self.assertRaises(UnauthorizedRequestAPICallError)
         with maybe_assert_exception:
+            requester = Requester(
+                user=target_user,
+                access_token_id=None,
+                is_guest=False,
+                scope={"scope"},
+                shadow_banned=False,
+                device_id=None,
+                app_service=None,
+                authenticated_entity="auth",
+            )
             self.get_success_or_raise(
-                self.media_repo.is_media_visible(target_user, media_object)
+                self.media_repo.is_media_visible(requester, media_object)
             )
 
     @parameterized.expand(
diff --git a/tests/rest/client/test_profile.py b/tests/rest/client/test_profile.py
index 52a48c22f5..c378ece275 100644
--- a/tests/rest/client/test_profile.py
+++ b/tests/rest/client/test_profile.py
@@ -1013,7 +1013,7 @@ def test_can_attach_media_to_profile_update(self) -> None:
         )
         assert restrictions is not None, str(restrictions)
         assert restrictions.event_id is None
-        assert restrictions.profile_user_id == UserID.from_string(self.user)
+        assert restrictions.profile_user_id == self.user
 
     def test_attaching_nonexistent_local_media_to_profile_fails(self) -> None:
         """
@@ -1274,7 +1274,7 @@ def test_profile_update_with_media_is_copied_and_attached_to_member_events(
         media_info = self.get_success(self.store.get_local_media(mxc_uri.media_id))
         assert media_info is not None
         assert media_info.attachments is not None
-        assert media_info.attachments.profile_user_id == UserID.from_string(self.user)
+        assert media_info.attachments.profile_user_id == self.user
 
         # Check the media was copied and attached to a member event
         events = self.get_success(
@@ -1446,7 +1446,7 @@ def test_profile_update_with_media_is_copied_and_attached_to_member_events(
         media_info = self.get_success(self.store.get_local_media(mxc_uri.media_id))
         assert media_info is not None
         assert media_info.attachments is not None
-        assert media_info.attachments.profile_user_id == UserID.from_string(self.user)
+        assert media_info.attachments.profile_user_id == self.user
 
         # Check the media was copied and attached to a member event
         events = self.get_success(
diff --git a/tests/rest/client/test_redactions.py b/tests/rest/client/test_redactions.py
index b25e184786..44828e8e70 100644
--- a/tests/rest/client/test_redactions.py
+++ b/tests/rest/client/test_redactions.py
@@ -18,22 +18,32 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
-from typing import List, Optional
+import io
+import os
+import time
+from http import HTTPStatus
+from typing import List, Optional, Tuple
 
 from parameterized import parameterized
 
 from twisted.test.proto_helpers import MemoryReactor
+from twisted.web.resource import Resource
 
+from synapse.api.auth.base import BaseAuth
 from synapse.api.constants import EventTypes, RelationTypes
 from synapse.api.room_versions import RoomVersion, RoomVersions
+from synapse.media.media_repository import (
+    MediaRepository,
+)
 from synapse.rest import admin
-from synapse.rest.client import login, room, sync
+from synapse.rest.client import login, media, room, sync
 from synapse.server import HomeServer
 from synapse.storage._base import db_to_json
 from synapse.storage.database import LoggingTransaction
-from synapse.types import JsonDict
+from synapse.types import JsonDict, Requester, UserID
 from synapse.util import Clock
 
+from tests.test_utils import SMALL_PNG
 from tests.unittest import HomeserverTestCase, override_config
 
 
@@ -656,3 +666,288 @@ def get_event(txn: LoggingTransaction) -> JsonDict:
         else:
             self.assertEqual(event_json["redacts"], event_id)
             self.assertNotIn("redacts", event_json["content"])
+
+
+class MediaDeletionOnRedactionTests(HomeserverTestCase):
+    extra_config = {
+        "experimental_features": {"msc3911_enabled": True},
+    }
+
+    servlets = [
+        media.register_servlets,
+        login.register_servlets,
+        admin.register_servlets,
+        room.register_servlets,
+    ]
+
+    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
+        config = self.default_config()
+        config.update(self.extra_config)
+        return self.setup_test_homeserver(config=config)
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.auth = hs.get_auth()
+        self.admin_handler = hs.get_admin_handler()
+        self.store = hs.get_datastores().main
+        self.media_repo = hs.get_media_repository()
+        self.bad_user = self.register_user("bad_user", "hackme")
+        self.bad_user_tok = self.login("bad_user", "hackme")
+        self.user = self.register_user("user", "pass")
+        self.user_tok = self.login("user", "pass")
+        self.admin = self.register_user("admin", "admin_pass", admin=True)
+        self.admin_tok = self.login("admin", "admin_pass")
+        self.room = self.helper.create_room_as(
+            room_creator=self.admin, tok=self.admin_tok
+        )
+
+    def create_resource_dict(self) -> dict[str, Resource]:
+        resources = super().create_resource_dict()
+        resources["/_matrix/media"] = self.hs.get_media_repository_resource()
+        return resources
+
+    def _redact_event(
+        self,
+        access_token: str,
+        room_id: str,
+        event_id: str,
+        expect_code: int = 200,
+        with_relations: Optional[List[str]] = None,
+        content: Optional[JsonDict] = None,
+    ) -> JsonDict:
+        """Helper function to send a redaction event.
+
+        Returns the json body.
+        """
+        path = "/_matrix/client/r0/rooms/%s/redact/%s" % (room_id, event_id)
+
+        request_content = content or {}
+        if with_relations:
+            request_content["org.matrix.msc3912.with_relations"] = with_relations
+
+        channel = self.make_request(
+            "POST", path, request_content, access_token=access_token
+        )
+        assert channel.code == expect_code, channel.json_body
+        return channel.json_body
+
+    def _create_test_resource(self) -> Tuple[List, List]:
+        event_ids = []
+        media_ids = []
+        self.helper.join(self.room, self.user, tok=self.user_tok)
+        self.helper.join(self.room, self.bad_user, tok=self.bad_user_tok)
+
+        for _ in range(3):
+            # Create restricted media
+            mxc_uri = self.get_success(
+                self.media_repo.create_or_update_content(
+                    "image/png",
+                    "test_png_upload",
+                    io.BytesIO(SMALL_PNG),
+                    67,
+                    UserID.from_string(self.bad_user),
+                    restricted=True,
+                )
+            )
+            # Make sure media is saved
+            assert mxc_uri is not None
+            assert isinstance(self.media_repo, MediaRepository)
+            media_path = self.media_repo.filepaths.local_media_filepath(
+                mxc_uri.media_id
+            )
+            self.assertTrue(os.path.exists(media_path))
+            assert self.get_success(self.store.get_local_media(mxc_uri.media_id))
+            media_ids.append(mxc_uri.media_id)
+
+            # Bad user create events with media attached
+            channel = self.make_request(
+                "PUT",
+                f"/rooms/{self.room}/send/m.room.message/{str(time.time())}?org.matrix.msc3911.attach_media={str(mxc_uri)}",
+                content={"msgtype": "m.text", "body": "Hi, this is a message"},
+                access_token=self.bad_user_tok,
+            )
+            assert channel.code == HTTPStatus.OK, channel.json_body
+            assert "event_id" in channel.json_body
+            event_id = channel.json_body["event_id"]
+            event_ids.append(event_id)
+
+            # Check media restrictions field has proper event_id
+            restrictions = self.get_success(
+                self.hs.get_datastores().main.get_media_restrictions(
+                    mxc_uri.server_name, mxc_uri.media_id
+                )
+            )
+            assert restrictions is not None, str(restrictions)
+            assert restrictions.event_id == event_id
+        return media_ids, event_ids
+
+    def test_is_moderator(self) -> None:
+        """Test BaseAuth function `is_moderator`"""
+        # Admin is a moderator
+        assert isinstance(self.auth, BaseAuth)
+        admin = Requester(
+            user=UserID.from_string(self.admin),
+            access_token_id=None,
+            is_guest=False,
+            scope={"scope"},
+            shadow_banned=False,
+            device_id=None,
+            app_service=None,
+            authenticated_entity="auth",
+        )
+        assert self.get_success(self.auth.is_moderator(self.room, admin))
+
+        # Normal user is not a moderator
+        self.helper.join(self.room, self.user, tok=self.user_tok)
+        user = Requester(
+            user=UserID.from_string(self.user),
+            access_token_id=None,
+            is_guest=False,
+            scope={"scope"},
+            shadow_banned=False,
+            device_id=None,
+            app_service=None,
+            authenticated_entity="auth",
+        )
+        assert not self.get_success(self.auth.is_moderator(self.room, user))
+
+        # Update power level to make normal user a moderator
+        power_levels = self.helper.get_state(
+            self.room,
+            "m.room.power_levels",
+            tok=self.user_tok,
+        )
+        power_levels["users"][self.user] = 80
+        self.helper.send_state(
+            self.room,
+            "m.room.power_levels",
+            body=power_levels,
+            tok=self.admin_tok,
+        )
+        assert self.get_success(self.auth.is_moderator(self.room, user))
+
+    def test_get_media_ids_attached_to_event(self) -> None:
+        """Test db function `get_media_ids_attached_to_event`"""
+        # Create events with media attached
+        media_ids, event_ids = self._create_test_resource()
+
+        # Check if `get_media_ids_attached_to_event` can get media ids attached to an event
+        for event_id in event_ids:
+            media_ids = self.get_success(
+                self.store.get_media_ids_attached_to_event(event_id)
+            )
+            assert len(media_ids) == 1
+
+    def test_redacting_media_deletes_attached_media(self) -> None:
+        """Test that the redacted media cleanup loop deletes media that has been
+        redacted before 48 hours.
+        """
+        # Confirm that there are no redacted events at the start
+        current_redacted_events = self.get_success(
+            self.store.get_redacted_event_ids_before_interval(0)
+        )
+        assert len(current_redacted_events) == 0
+
+        # Create events with media attached
+        media_ids, event_ids = self._create_test_resource()
+
+        # Redact the events
+        for event_id in event_ids:
+            self._redact_event(
+                self.admin_tok,
+                self.room,
+                event_id,
+                expect_code=200,
+            )
+            # Confirm the events redaction
+            event_dict = self.helper.get_event(self.room, event_id, self.admin_tok)
+            assert "redacted_because" in event_dict, event_dict
+
+        # Confirm that the redacted events are recorded in the db
+        current_redacted_events = self.get_success(
+            self.store.get_redacted_event_ids_before_interval(0)
+        )
+        # assert all(item in current_redacted_events for item in event_ids)
+        assert event_ids == current_redacted_events
+
+        # Fast forward 49 hours to make sure the redacted media cleanup loop runs
+        self.reactor.advance(49 * 60 * 60)
+
+        # Check if the media is deleted from storage.
+        for media_id in media_ids:
+            media = self.get_success(self.store.get_local_media(media_id))
+            assert media is None
+            assert isinstance(self.media_repo, MediaRepository)
+            assert not os.path.exists(
+                self.media_repo.filepaths.local_media_filepath(media_id)
+            )
+
+    def test_normal_users_lose_access_to_media_right_after_redaction(self) -> None:
+        """Test that normal users lose access to media after the event they
+        were attached to has been redacted.
+        """
+        # Create events with media attached
+        media_ids, event_ids = self._create_test_resource()
+
+        # Redact the bad user's events
+        for event_id in event_ids:
+            self._redact_event(
+                self.admin_tok,
+                self.room,
+                event_id,
+                expect_code=200,
+            )
+            event_dict = self.helper.get_event(self.room, event_id, self.admin_tok)
+            self.assertIn("redacted_because", event_dict, event_dict)
+
+        # Normal user trying to access redacted media should get 403
+        for media_id in media_ids:
+            channel = self.make_request(
+                "GET",
+                f"/_matrix/client/v1/media/download/{self.hs.hostname}/{media_id}",
+                shorthand=False,
+                access_token=self.user_tok,
+            )
+            assert channel.code == 403
+
+    def test_moderators_still_have_access_to_media_after_redaction_until_permanent_deletion(
+        self,
+    ) -> None:
+        """Test that users with moderator privileges still have access to media
+        after the event they were attached to has been redacted.
+        """
+        # Create events with media attached
+        media_ids, event_ids = self._create_test_resource()
+
+        # Redact the bad user's events
+        for event_id in event_ids:
+            self._redact_event(
+                self.admin_tok,
+                self.room,
+                event_id,
+                expect_code=200,
+            )
+            event_dict = self.helper.get_event(self.room, event_id, self.admin_tok)
+            self.assertIn("redacted_because", event_dict, event_dict)
+
+        # User with moderator privileges still have access to redacted media
+        for media_id in media_ids:
+            channel = self.make_request(
+                "GET",
+                f"/_matrix/client/v1/media/download/{self.hs.hostname}/{media_id}",
+                shorthand=False,
+                access_token=self.admin_tok,
+            )
+            assert channel.code == 200
+
+        # After 48 hours, media that are attached to the redacted events should be
+        # permanatly deleted from disk and moderators no longer have access to them
+        self.reactor.advance(49 * 60 * 60)
+
+        for media_id in media_ids:
+            channel = self.make_request(
+                "GET",
+                f"/_matrix/client/v1/media/download/{self.hs.hostname}/{media_id}",
+                shorthand=False,
+                access_token=self.admin_tok,
+            )
+            assert channel.code == 404
diff --git a/tests/storage/test_media.py b/tests/storage/test_media.py
index 6bac81c1a6..9fafb32415 100644
--- a/tests/storage/test_media.py
+++ b/tests/storage/test_media.py
@@ -2,7 +2,6 @@
 
 from synapse.api.errors import SynapseError
 from synapse.server import HomeServer
-from synapse.types import UserID
 from synapse.util import Clock
 from synapse.util.stringutils import random_string
 
@@ -41,11 +40,11 @@ def test_store_and_retrieve_media_restrictions_by_event_id(self) -> None:
         assert retrieved_restrictions.profile_user_id is None
 
     def test_store_and_retrieve_media_restrictions_by_profile_user_id(self) -> None:
-        user_id = UserID.from_string("@frank:test")
+        user_id = "@frank:test"
         media_id = random_string(24)
         self.get_success_or_raise(
             self.store.set_media_restricted_to_user_profile(
-                self.server_name, media_id, user_id.to_string()
+                self.server_name, media_id, user_id
             )
         )