Add an Admin API endpoint to redact all a user's events (#17506)

changelog.d/17506.feature

@@ -0,0 +1,2 @@
+Add an asynchronous Admin API endpoint [to redact all a user's events](https://element-hq.github.io/synapse/v1.116/admin_api/user_admin_api.html#redact-all-the-events-of-a-user),
+and [an endpoint to check on the status of that redaction task](https://element-hq.github.io/synapse/v1.116/admin_api/user_admin_api.html#check-the-status-of-a-redaction-process).

docs/admin_api/user_admin_api.md

@@ -1361,3 +1361,83 @@ Returns a `404` HTTP status code if no user was found, with a response body like
 ```
 
 _Added in Synapse 1.72.0._
+
+
+## Redact all the events of a user
+
+The API is 
+```
+POST /_synapse/admin/v1/user/$user_id/redact
+
+{
+  "rooms": ["!roomid1", "!roomid2"]
+}
+```
+If an empty list is provided as the key for `rooms`, all events in all the rooms the user is member of will be redacted, 
+otherwise all the events in the rooms provided in the request will be redacted. 
+
+The API starts redaction process running, and returns immediately with a JSON body with
+a redact id which can be used to query the status of the redaction process:
+
+```json
+{
+    "redact_id": "<opaque id>"
+}
+```
+
+**Parameters**
+
+The following parameters should be set in the URL:
+
+- `user_id` - The fully qualified MXID of the user: for example, `@user:server.com`.
+
+The following JSON body parameter must be provided:
+
+-  `rooms` - A list of rooms to redact the user's events in. If an empty list is provided all events in all rooms
+  the user is a member of will be redacted
+
+_Added in Synapse 1.116.0._
+
+The following JSON body parameters are optional:
+
+- `reason` - Reason the redaction is being requested, ie "spam", "abuse", etc. This will be included in each redaction event, and be visible to users.
+- `limit` - a limit on the number of the user's events to search for ones that can be redacted (events are redacted newest to oldest) in each room, defaults to 1000 if not provided
+
+
+## Check the status of a redaction process
+
+It is possible to query the status of the background task for redacting a user's events.
+The status can be queried up to 24 hours after completion of the task,
+or until Synapse is restarted (whichever happens first).
+
+The API is:
+
+```
+GET /_synapse/admin/v1/user/redact_status/$redact_id
+```
+
+A response body like the following is returned:
+
+```
+{
+  "status": "active",
+  "failed_redactions": [],
+}
+```
+
+**Parameters**
+
+The following parameters should be set in the URL:
+
+* `redact_id` - string - The ID for this redaction process, provided when the redaction was requested.
+
+
+**Response**
+
+The following fields are returned in the JSON response body:
+
+- `status` - string - one of scheduled/active/completed/failed, indicating the status of the redaction job
+- `failed_redactions` - dictionary - the keys of the dict are event ids the process was unable to redact, if any, and the values are 
+  the corresponding error that caused the redaction to fail
+
+_Added in Synapse 1.116.0._

synapse/handlers/admin.py

@@ -21,20 +21,43 @@
 
 import abc
 import logging
-from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Sequence, Set
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    List,
+    Mapping,
+    Optional,
+    Sequence,
+    Set,
+    Tuple,
+)
 
 import attr
 
-from synapse.api.constants import Direction, Membership
+from synapse.api.constants import Direction, EventTypes, Membership
+from synapse.api.errors import SynapseError
 from synapse.events import EventBase
-from synapse.types import JsonMapping, RoomStreamToken, StateMap, UserID, UserInfo
+from synapse.types import (
+    JsonMapping,
+    Requester,
+    RoomStreamToken,
+    ScheduledTask,
+    StateMap,
+    TaskStatus,
+    UserID,
+    UserInfo,
+    create_requester,
+)
 from synapse.visibility import filter_events_for_client
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
 
 logger = logging.getLogger(__name__)
 
+REDACT_ALL_EVENTS_ACTION_NAME = "redact_all_events"
+
 
 class AdminHandler:
     def __init__(self, hs: "HomeServer"):
@@ -43,6 +66,20 @@ def __init__(self, hs: "HomeServer"):
         self._storage_controllers = hs.get_storage_controllers()
         self._state_storage_controller = self._storage_controllers.state
         self._msc3866_enabled = hs.config.experimental.msc3866.enabled
+        self.event_creation_handler = hs.get_event_creation_handler()
+        self._task_scheduler = hs.get_task_scheduler()
+
+        self._task_scheduler.register_action(
+            self._redact_all_events, REDACT_ALL_EVENTS_ACTION_NAME
+        )
+
+    async def get_redact_task(self, redact_id: str) -> Optional[ScheduledTask]:
+        """Get the current status of an active redaction process
+
+        Args:
+            redact_id: redact_id returned by start_redact_events.
+        """
+        return await self._task_scheduler.get_task(redact_id)
 
     async def get_whois(self, user: UserID) -> JsonMapping:
         connections = []
@@ -313,6 +350,153 @@ async def export_user_data(self, user_id: str, writer: "ExfiltrationWriter") ->
 
         return writer.finished()
 
+    async def start_redact_events(
+        self,
+        user_id: str,
+        rooms: list,
+        requester: JsonMapping,
+        reason: Optional[str],
+        limit: Optional[int],
+    ) -> str:
+        """
+        Start a task redacting the events of the given user in the given rooms
+
+        Args:
+            user_id: the user ID of the user whose events should be redacted
+            rooms: the rooms in which to redact the user's events
+            requester: the user requesting the events
+            reason: reason for requesting the redaction, ie spam, etc
+            limit: limit on the number of events in each room to redact
+
+        Returns:
+            a unique ID which can be used to query the status of the task
+        """
+        active_tasks = await self._task_scheduler.get_tasks(
+            actions=[REDACT_ALL_EVENTS_ACTION_NAME],
+            resource_id=user_id,
+            statuses=[TaskStatus.ACTIVE],
+        )
+
+        if len(active_tasks) > 0:
+            raise SynapseError(
+                400, "Redact already in progress for user %s" % (user_id,)
+            )
+
+        if not limit:
+            limit = 1000
+
+        redact_id = await self._task_scheduler.schedule_task(
+            REDACT_ALL_EVENTS_ACTION_NAME,
+            resource_id=user_id,
+            params={
+                "rooms": rooms,
+                "requester": requester,
+                "user_id": user_id,
+                "reason": reason,
+                "limit": limit,
+            },
+        )
+
+        logger.info(
+            "starting redact events with redact_id %s",
+            redact_id,
+        )
+
+        return redact_id
+
+    async def _redact_all_events(
+        self, task: ScheduledTask
+    ) -> Tuple[TaskStatus, Optional[Mapping[str, Any]], Optional[str]]:
+        """
+        Task to redact all of a users events in the given rooms, tracking which, if any, events
+        whose redaction failed
+        """
+
+        assert task.params is not None
+        rooms = task.params.get("rooms")
+        assert rooms is not None
+
+        r = task.params.get("requester")
+        assert r is not None
+        admin = Requester.deserialize(self._store, r)
+
+        user_id = task.params.get("user_id")
+        assert user_id is not None
+
+        requester = create_requester(
+            user_id, authenticated_entity=admin.user.to_string()
+        )
+
+        reason = task.params.get("reason")
+        limit = task.params.get("limit")
+        assert limit is not None
+
+        result: Mapping[str, Any] = (
+            task.result if task.result else {"failed_redactions": {}}
+        )
+        for room in rooms:
+            room_version = await self._store.get_room_version(room)
+            event_ids = await self._store.get_events_sent_by_user_in_room(
+                user_id,
+                room,
+                limit,
+                ["m.room.member", "m.room.message"],
+            )
+            if not event_ids:
+                # there's nothing to redact
+                return TaskStatus.COMPLETE, result, None
+
+            events = await self._store.get_events_as_list(event_ids)
+            for event in events:
+                # we care about join events but not other membership events
+                if event.type == "m.room.member":
+                    content = event.content
+                    if content:
+                        if content.get("membership") == Membership.JOIN:
+                            pass
+                        else:
+                            continue
+                relations = await self._store.get_relations_for_event(
+                    room, event.event_id, event, event_type=EventTypes.Redaction
+                )
+
+                # if we've already successfully redacted this event then skip processing it
+                if relations[0]:
+                    continue
+
+                event_dict = {
+                    "type": EventTypes.Redaction,
+                    "content": {"reason": reason} if reason else {},
+                    "room_id": room,
+                    "sender": user_id,
+                }
+                if room_version.updated_redaction_rules:
+                    event_dict["content"]["redacts"] = event.event_id
+                else:
+                    event_dict["redacts"] = event.event_id
+
+                try:
+                    # set the prev event to the offending message to allow for redactions
+                    # to be processed in the case where the user has been kicked/banned before
+                    # redactions are requested
+                    (
+                        redaction,
+                        _,
+                    ) = await self.event_creation_handler.create_and_send_nonmember_event(
+                        requester,
+                        event_dict,
+                        prev_event_ids=[event.event_id],
+                        ratelimit=False,
+                    )
+                except Exception as ex:
+                    logger.info(
+                        f"Redaction of event {event.event_id} failed due to: {ex}"
+                    )
+                    result["failed_redactions"][event.event_id] = str(ex)
+                    await self._task_scheduler.update_task(task.id, result=result)
+
+        return TaskStatus.COMPLETE, result, None
+
 
 class ExfiltrationWriter(metaclass=abc.ABCMeta):
     """Interface used to specify how to write exported data."""

synapse/rest/admin/__init__.py

@@ -98,6 +98,8 @@
     DeactivateAccountRestServlet,
     PushersRestServlet,
     RateLimitRestServlet,
+    RedactUser,
+    RedactUserStatus,
     ResetPasswordRestServlet,
     SearchUsersRestServlet,
     ShadowBanRestServlet,
@@ -319,6 +321,8 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
     UserReplaceMasterCrossSigningKeyRestServlet(hs).register(http_server)
     UserByExternalId(hs).register(http_server)
     UserByThreePid(hs).register(http_server)
+    RedactUser(hs).register(http_server)
+    RedactUserStatus(hs).register(http_server)
 
     DeviceRestServlet(hs).register(http_server)
     DevicesRestServlet(hs).register(http_server)