Add support for MSC4293 - Redact on Kick/Ban

changelog.d/18540.feature

@@ -0,0 +1 @@
+Add support for [MSC4293](https://github.com/matrix-org/matrix-spec-proposals/pull/4293) - Redact on Kick/Ban.

synapse/config/experimental.py

@@ -582,6 +582,9 @@ def read_config(
         # MSC4155: Invite filtering
         self.msc4155_enabled: bool = experimental.get("msc4155_enabled", False)
 
+        # MSC4293: Redact on Kick/Ban
+        self.msc4293_enabled: bool = experimental.get("msc4293_enabled", False)
+
         # MSC4306: Thread Subscriptions
         # (and MSC4308: sliding sync extension for thread subscriptions)
         self.msc4306_enabled: bool = experimental.get("msc4306_enabled", False)

synapse/rest/client/room.py

@@ -1100,6 +1100,7 @@ def __init__(self, hs: "HomeServer"):
         super().__init__(hs)
         self.room_member_handler = hs.get_room_member_handler()
         self.auth = hs.get_auth()
+        self.config = hs.config
 
     def register(self, http_server: HttpServer) -> None:
         # /rooms/$roomid/[join|invite|leave|ban|unban|kick]
@@ -1123,12 +1124,12 @@ async def _do(
         }:
             raise AuthError(403, "Guest access not allowed")
 
-        content = parse_json_object_from_request(request, allow_empty_body=True)
+        request_body = parse_json_object_from_request(request, allow_empty_body=True)
 
         if membership_action == "invite" and all(
-            key in content for key in ("medium", "address")
+            key in request_body for key in ("medium", "address")
         ):
-            if not all(key in content for key in ("id_server", "id_access_token")):
+            if not all(key in request_body for key in ("id_server", "id_access_token")):
                 raise SynapseError(
                     HTTPStatus.BAD_REQUEST,
                     "`id_server` and `id_access_token` are required when doing 3pid invite",
@@ -1139,12 +1140,12 @@ async def _do(
                 await self.room_member_handler.do_3pid_invite(
                     room_id,
                     requester.user,
-                    content["medium"],
-                    content["address"],
-                    content["id_server"],
+                    request_body["medium"],
+                    request_body["address"],
+                    request_body["id_server"],
                     requester,
                     txn_id,
-                    content["id_access_token"],
+                    request_body["id_access_token"],
                 )
             except ShadowBanError:
                 # Pretend the request succeeded.
@@ -1153,12 +1154,19 @@ async def _do(
 
         target = requester.user
         if membership_action in ["invite", "ban", "unban", "kick"]:
-            assert_params_in_dict(content, ["user_id"])
-            target = UserID.from_string(content["user_id"])
+            assert_params_in_dict(request_body, ["user_id"])
+            target = UserID.from_string(request_body["user_id"])
 
         event_content = None
-        if "reason" in content:
-            event_content = {"reason": content["reason"]}
+        if "reason" in request_body:
+            event_content = {"reason": request_body["reason"]}
+        if self.config.experimental.msc4293_enabled:
+            if "org.matrix.msc4293.redact_events" in request_body:
+                if event_content is None:
+                    event_content = {}
+                event_content["org.matrix.msc4293.redact_events"] = request_body[
+                    "org.matrix.msc4293.redact_events"
+                ]
 
         try:
             await self.room_member_handler.update_membership(
@@ -1167,7 +1175,7 @@ async def _do(
                 room_id=room_id,
                 action=membership_action,
                 txn_id=txn_id,
-                third_party_signed=content.get("third_party_signed", None),
+                third_party_signed=request_body.get("third_party_signed", None),
                 content=event_content,
             )
         except ShadowBanError:

synapse/storage/databases/main/events.py

@@ -377,11 +377,130 @@ async def _persist_events_and_state_updates(
 
                 event_counter.labels(event.type, origin_type, origin_entity).inc()
 
+                if (
+                    not self.hs.config.experimental.msc4293_enabled
+                    or event.type != EventTypes.Member
+                    or event.state_key is None
+                ):
+                    continue
+
+                # check if this is an unban/join that will undo a ban/kick redaction for
+                # a user in the room
+                if event.membership in [Membership.LEAVE, Membership.JOIN]:
+                    if (
+                        event.membership == Membership.LEAVE
+                        and event.sender == event.state_key
+                    ):
+                        # self-leave, ignore
+                        continue
+
+                    # if there is an existing ban/leave causing redactions for
+                    # this user/room combination update the entry with the stream
+                    # ordering when the redactions should stop - in the case of a backfilled
+                    # event where the stream ordering is negative, use the current max stream
+                    # ordering
+                    stream_ordering = event.internal_metadata.stream_ordering
+                    assert stream_ordering is not None
+                    if stream_ordering < 0:
+                        stream_ordering = self._stream_id_gen.get_current_token()
+                    await self.db_pool.simple_update(
+                        "room_ban_redactions",
+                        {"room_id": event.room_id, "user_id": event.state_key},
+                        {"redact_end_ordering": stream_ordering},
+                        desc="room_ban_redactions update redact_end_ordering",
+                    )
+
+                # check for msc4293 redact_events flag and apply if found
+                if event.membership not in [Membership.LEAVE, Membership.BAN]:
+                    continue
+                redact = event.content.get("org.matrix.msc4293.redact_events", False)
+                if not redact or not isinstance(redact, bool):
+                    continue
+                # self-bans currently are not authorized so we don't check for that
+                # case
+                if (
+                    event.membership == Membership.BAN
+                    and event.sender == event.state_key
+                ):
+                    continue
+
+                # check that sender can redact
+                redact_allowed = await self._can_sender_redact(event)
+
+                # Signal that this user's past events in this room
+                # should be redacted by adding an entry to
+                # `room_ban_redactions`.
+                if redact_allowed:
+                    await self.db_pool.simple_upsert(
+                        "room_ban_redactions",
+                        {"room_id": event.room_id, "user_id": event.state_key},
+                        {
+                            "redacting_event_id": event.event_id,
+                            "redact_end_ordering": None,
+                        },
+                        {
+                            "room_id": event.room_id,
+                            "user_id": event.state_key,
+                            "redacting_event_id": event.event_id,
+                            "redact_end_ordering": None,
+                        },
+                    )
+
+                    # normally the cache entry for a redacted event would be invalidated
+                    # by an arriving redaction event, but since we are not creating redaction
+                    # events we invalidate manually
+                    self.store._invalidate_local_get_event_cache_room_id(event.room_id)
+
+                    self.store._invalidate_async_get_event_cache_room_id(event.room_id)
+
             if new_forward_extremities:
                 self.store.get_latest_event_ids_in_room.prefill(
                     (room_id,), frozenset(new_forward_extremities)
                 )
 
+    async def _can_sender_redact(self, event: EventBase) -> bool:
+        state_filter = StateFilter.from_types(
+            [(EventTypes.PowerLevels, ""), (EventTypes.Create, "")]
+        )
+        state = await self.store.get_partial_filtered_current_state_ids(
+            event.room_id, state_filter
+        )
+        pl_id = state[(EventTypes.PowerLevels, "")]
+        pl_event = await self.store.get_event(pl_id, allow_none=True)
+
+        if pl_event is None:
+            # per the spec, if a power level event isn't in the room, grant the creator
+            # level 100 and all other users 0
+            create_id = state[(EventTypes.Create, "")]
+            create_event = await self.store.get_event(create_id, allow_none=True)
+            if create_event is None:
+                # not sure how this would happen but if it does then just deny the redaction
+                logger.warning("No create event found for room %s", event.room_id)
+                return False
+            if create_event.sender == event.sender:
+                return True
+
+        assert pl_event is not None
+        sender_level = pl_event.content.get("users", {}).get(event.sender)
+        if sender_level is None:
+            sender_level = pl_event.content.get("users_default", 0)
+
+        redact_level = pl_event.content.get("redact")
+        if redact_level is None:
+            redact_level = pl_event.content.get("events_default", 0)
+
+        room_redaction_level = pl_event.content.get("events", {}).get(
+            "m.room.redaction"
+        )
+        if room_redaction_level is not None:
+            if sender_level < room_redaction_level:
+                return False
+
+        if sender_level >= redact_level:
+            return True
+
+        return False
+
     async def _calculate_sliding_sync_table_changes(
         self,
         room_id: str,

synapse/storage/databases/main/events_worker.py

@@ -17,7 +17,7 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
-
+import json
 import logging
 import threading
 import weakref
@@ -976,6 +976,13 @@ def _invalidate_local_get_event_cache_room_id(self, room_id: str) -> None:
         self._event_ref.clear()
         self._current_event_fetches.clear()
 
+    def _invalidate_async_get_event_cache_room_id(self, room_id: str) -> None:
+        """
+        Clears the async get_event cache for a room. Currently a no-op until
+        an async get_event cache is implemented - see https://github.com/matrix-org/synapse/pull/13242
+        for preliminary work.
+        """
+
     async def _get_events_from_cache(
         self, events: Iterable[str], update_metrics: bool = True
     ) -> Dict[str, EventCacheEntry]:
@@ -1575,6 +1582,44 @@ def _fetch_event_rows(
                 if d:
                     d.redactions.append(redacter)
 
+            # check for MSC4932 redactions
+            to_check = []
+            events: List[_EventRow] = []
+            for e in evs:
+                event = event_dict.get(e)
+                if not event:
+                    continue
+                events.append(event)
+                event_json = json.loads(event.json)
+                room_id = event_json.get("room_id")
+                user_id = event_json.get("sender")
+                to_check.append((room_id, user_id))
+
+            # likely that some of these events may be for the same room/user combo, in
+            # which case we don't need to do redundant queries
+            to_check_set = set(to_check)
+            for room_and_user in to_check_set:
+                room_redactions_sql = "SELECT redacting_event_id, redact_end_ordering FROM room_ban_redactions WHERE room_id = ? and user_id = ?"
+                txn.execute(room_redactions_sql, room_and_user)
+
+                res = txn.fetchone()
+                # we have a redaction for a room, user_id combo - apply it to matching events
+                if not res:
+                    continue
+                for e_row in events:
+                    e_json = json.loads(e_row.json)
+                    room_id = e_json.get("room_id")
+                    user_id = e_json.get("sender")
+                    if room_and_user != (room_id, user_id):
+                        continue
+                    redacting_event_id, redact_end_ordering = res
+                    if redact_end_ordering:
+                        # Avoid redacting any events arriving *after* the membership event which
+                        # ends an active redaction - note that this will always redact
+                        # backfilled events, as they have a negative stream ordering
+                        if e_row.stream_ordering >= redact_end_ordering:
+                            continue
+                    e_row.redactions.append(redacting_event_id)
         return event_dict
 
     def _maybe_redact_event_row(

synapse/storage/schema/main/delta/92/08_room_ban_redactions.sql

@@ -0,0 +1,21 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2025 New Vector, Ltd
+--
+-- This program is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as
+-- published by the Free Software Foundation, either version 3 of the
+-- License, or (at your option) any later version.
+--
+-- See the GNU Affero General Public License for more details:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+CREATE TABLE room_ban_redactions(
+    room_id text NOT NULL,
+    user_id text NOT NULL,
+    redacting_event_id text NOT NULL,
+    redact_end_ordering bigint DEFAULT NULL, -- stream ordering after which redactions are not applied
+    CONSTRAINT room_ban_redaction_uniqueness UNIQUE (room_id, user_id)
+);
+