feat: Allow room moderators to view redacted event content (MSC2815)

chore: Clean up after Erwan

changelog.d/+msc2815-unredacted-content.feature.md

@@ -0,0 +1 @@
+Added support for [MSC2815](https://github.com/matrix-org/matrix-spec-proposals/pull/2815), allowing room moderators to fetch the original content of redacted events, behind the `allow_moderators_to_view_redacted_content` config option. Contributed by @eleboucher.

conduwuit-example.toml

@@ -1661,6 +1661,25 @@
 # env: CONTINUWUITY_ALLOW_STICKY_EVENTS
 #allow_sticky_events = false
 
+# Allow room moderators to view redacted event content ([MSC2815]).
+#
+# Redacted content is readable by users at or above the room's `redact`
+# power level, and by server admins, until it expires.
+#
+# MSC2815: https://github.com/matrix-org/matrix-spec-proposals/pull/2815
+#
+# env: CONTINUWUITY_ALLOW_MODERATORS_TO_VIEW_REDACTED_CONTENT
+#allow_moderators_to_view_redacted_content = false
+
+# How long to retain the content of redacted events, in seconds. Expired
+# content is deleted by a periodic purge and is no longer served.
+#
+# Only has an effect when allow_moderators_to_view_redacted_content is
+# enabled.
+#
+# env: CONTINUWUITY_UNREDACTED_CONTENT_MAX_AGE_SECS
+#unredacted_content_max_age_secs = 7776000 (90 days)
+
 # Maximum time federation user can indicate typing.
 #
 # env: CONTINUWUITY_TYPING_FEDERATION_TIMEOUT_S

src/api/client/room/event.rs

@@ -1,14 +1,28 @@
-use axum::extract::State;
-use conduwuit::{Err, Event, Result, debug_warn, err};
-use ruma::api::client::room::get_room_event;
+use axum::extract::{Query, State};
+use conduwuit::{Err, Error, Event, PduEvent, Result, debug_warn, err};
+use futures::TryFutureExt;
+use http::StatusCode;
+use ruma::{
+	UserId,
+	api::{client::room::get_room_event, error::ErrorKind},
+};
+use serde::Deserialize;
+use serde_json::json;
 
 use crate::{Ruma, client::is_ignored_pdu};
 
+#[derive(Deserialize)]
+pub(crate) struct Msc2815Params {
+	#[serde(rename = "fi.mau.msc2815.include_unredacted_content", default)]
+	include_unredacted_content: Option<bool>,
+}
+
 /// # `GET /_matrix/client/r0/rooms/{roomId}/event/{eventId}`
 ///
 /// Gets a single event.
 pub(crate) async fn get_room_event_route(
 	State(ref services): State<crate::State>,
+	Query(params): Query<Msc2815Params>,
 	ref body: Ruma<get_room_event::v3::Request>,
 ) -> Result<get_room_event::v3::Response> {
 	let sender_user = body.identity.expect_sender_user()?;
@@ -54,6 +68,13 @@ pub(crate) async fn get_room_event_route(
 		return Err!(Request(Forbidden("You don't have permission to view this event.")));
 	}
 
+	if services.config.allow_moderators_to_view_redacted_content
+		&& params.include_unredacted_content.unwrap_or(false)
+		&& event.is_redacted()
+	{
+		unredact(services, sender_user, &mut event).await?;
+	}
+
 	if let Err(e) = services
 		.rooms
 		.pdu_metadata
@@ -72,3 +93,56 @@ pub(crate) async fn get_room_event_route(
 
 	Ok(get_room_event::v3::Response::new(event.into_format()))
 }
+
+async fn unredact(services: &crate::State, sender_user: &UserId, event: &mut PduEvent) -> Result {
+	if !services.users.is_admin(sender_user).await {
+		let power_levels = services
+			.rooms
+			.state_accessor
+			.get_room_power_levels(&event.room_id_or_hash())
+			.await;
+
+		if power_levels.for_user(sender_user) < power_levels.redact {
+			return Err!(Request(Forbidden(
+				"You don't have permission to view redacted content in this room."
+			)));
+		}
+	}
+
+	let content = services
+		.rooms
+		.timeline
+		.get_pdu_id(event.event_id())
+		.and_then(async |pdu_id| {
+			services
+				.rooms
+				.timeline
+				.get_unredacted_content(&pdu_id)
+				.await
+		})
+		.await;
+
+	let Ok(content) = content else {
+		// CustomErrorKind has private fields; only serde can build one.
+		let keep_ms = services
+			.config
+			.unredacted_content_max_age_secs
+			.saturating_mul(1000);
+
+		let kind = serde_json::from_value::<ErrorKind>(json!({
+			"errcode": "FI.MAU.MSC2815_UNREDACTED_CONTENT_DELETED",
+			"fi.mau.msc2815.content_keep_ms": keep_ms,
+		}))
+		.expect("errcode is a valid custom error kind");
+
+		return Err(Error::Request(
+			kind,
+			"The unredacted content of this event is not available.".into(),
+			StatusCode::NOT_FOUND,
+		));
+	};
+
+	event.content = content;
+
+	Ok(())
+}

src/api/client/unversioned.rs

@@ -34,6 +34,11 @@ pub(crate) async fn get_supported_versions_route(
 		unstable_features.insert("org.matrix.msc4354".to_owned(), true);
 	}
 
+	if services.config.allow_moderators_to_view_redacted_content {
+		// moderators can view redacted content (https://github.com/matrix-org/matrix-spec-proposals/pull/2815)
+		unstable_features.insert("fi.mau.msc2815".to_owned(), true);
+	}
+
 	Ok(assign!(get_supported_versions::Response::new(versions()), {
 		unstable_features
 	}))

src/core/config/mod.rs

@@ -1773,6 +1773,27 @@ pub struct Config {
 	#[serde(default)]
 	pub allow_sticky_events: bool,
 
+	/// Allow room moderators to view redacted event content ([MSC2815]).
+	///
+	/// Redacted content is readable by users at or above the room's `redact`
+	/// power level, and by server admins, until it expires.
+	///
+	/// MSC2815: https://github.com/matrix-org/matrix-spec-proposals/pull/2815
+	///
+	/// default: false
+	#[serde(default)]
+	pub allow_moderators_to_view_redacted_content: bool,
+
+	/// How long to retain the content of redacted events, in seconds. Expired
+	/// content is deleted by a periodic purge and is no longer served.
+	///
+	/// Only has an effect when allow_moderators_to_view_redacted_content is
+	/// enabled.
+	///
+	/// default: 7776000 (90 days)
+	#[serde(default = "default_unredacted_content_max_age_secs")]
+	pub unredacted_content_max_age_secs: u64,
+
 	/// Maximum time federation user can indicate typing.
 	///
 	/// default: 30
@@ -3240,6 +3261,8 @@ fn default_remote_join_pdu_queue_capacity() -> usize { 50 }
 
 fn default_transaction_id_cache_max_age_secs() -> u64 { 60 * 60 * 2 }
 
+fn default_unredacted_content_max_age_secs() -> u64 { 60 * 60 * 24 * 90 }
+
 fn default_transaction_id_cache_max_entries() -> usize { 8192 }
 
 fn default_tracing_flame_filter() -> String {

src/database/maps.rs

@@ -153,6 +153,11 @@ pub(super) static MAPS: &[Descriptor] = &[
 		index_size: 512,
 		..descriptor::SEQUENTIAL
 	},
+	Descriptor {
+		name: "pduid_unredactedcontent",
+		key_size_hint: Some(16),
+		..descriptor::SEQUENTIAL
+	},
 	Descriptor {
 		name: "publicroomids",
 		..descriptor::RANDOM_SMALL

src/service/rooms/timeline/data.rs

@@ -3,11 +3,18 @@ use std::sync::Arc;
 use conduwuit::{
 	Err, PduCount, PduEvent, Result, at, err,
 	result::NotFound,
-	utils::{self, stream::TryReadyExt},
+	utils::{
+		self,
+		stream::{ReadyExt, TryReadyExt},
+	},
 };
 use database::{Database, Deserialized, Json, KeyVal, Map};
-use futures::{FutureExt, Stream, TryFutureExt, TryStreamExt, future::select_ok, pin_mut};
+use futures::{
+	FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt, future::select_ok, pin_mut,
+};
 use ruma::{CanonicalJsonObject, EventId, OwnedUserId, RoomId, api::Direction};
+use serde::{Deserialize, Serialize};
+use serde_json::value::RawValue as RawJsonValue;
 
 use super::{PduId, RawPduId};
 use crate::{Dep, rooms, rooms::short::ShortRoomId};
@@ -16,6 +23,7 @@ pub(super) struct Data {
 	eventid_outlierpdu: Arc<Map>,
 	eventid_pduid: Arc<Map>,
 	pduid_pdu: Arc<Map>,
+	pduid_unredactedcontent: Arc<Map>,
 	roomid_lasttimelinecount: Arc<Map>,
 	userroomid_highlightcount: Arc<Map>,
 	userroomid_notificationcount: Arc<Map>,
@@ -29,6 +37,12 @@ struct Services {
 
 pub type PdusIterItem = (PduCount, PduEvent);
 
+#[derive(Deserialize, Serialize)]
+pub(super) struct UnredactedContent {
+	pub(super) created: u64,
+	pub(super) content: Box<RawJsonValue>,
+}
+
 impl Data {
 	pub(super) fn new(args: &crate::Args<'_>) -> Self {
 		let db = &args.db;
@@ -36,6 +50,7 @@ impl Data {
 			eventid_outlierpdu: db["eventid_outlierpdu"].clone(),
 			eventid_pduid: db["eventid_pduid"].clone(),
 			pduid_pdu: db["pduid_pdu"].clone(),
+			pduid_unredactedcontent: db["pduid_unredactedcontent"].clone(),
 			roomid_lasttimelinecount: db["roomid_lasttimelinecount"].clone(),
 			userroomid_highlightcount: db["userroomid_highlightcount"].clone(),
 			userroomid_notificationcount: db["userroomid_notificationcount"].clone(),
@@ -234,6 +249,44 @@ impl Data {
 		Ok(())
 	}
 
+	pub(super) fn save_unredacted_content(&self, pdu_id: &RawPduId, content: &RawJsonValue) {
+		let entry = UnredactedContent {
+			created: utils::millis_since_unix_epoch(),
+			content: content.to_owned(),
+		};
+
+		self.pduid_unredactedcontent.raw_put(pdu_id, Json(&entry));
+	}
+
+	pub(super) async fn get_unredacted_content(
+		&self,
+		pdu_id: &RawPduId,
+	) -> Result<UnredactedContent> {
+		self.pduid_unredactedcontent
+			.get(pdu_id)
+			.await
+			.deserialized()
+	}
+
+	pub(super) async fn purge_unredacted_content(&self, cutoff: u64) -> usize {
+		let expired: Vec<_> = self
+			.pduid_unredactedcontent
+			.raw_stream()
+			.ready_filter_map(Result::ok)
+			.ready_filter_map(|(key, val)| {
+				let entry: UnredactedContent = serde_json::from_slice(val).ok()?;
+				(entry.created < cutoff).then(|| key.to_vec())
+			})
+			.collect()
+			.await;
+
+		for key in &expired {
+			self.pduid_unredactedcontent.remove(key);
+		}
+
+		expired.len()
+	}
+
 	/// Returns an iterator over all events and their tokens in a room that
 	/// happened before the event with id `until` in reverse-chronological
 	/// order.

src/service/rooms/timeline/mod.rs

@@ -6,17 +6,17 @@ mod data;
 mod helpers;
 mod redact;
 
-use std::{fmt::Write, sync::Arc};
+use std::{fmt::Write, sync::Arc, time::Duration};
 
 use async_trait::async_trait;
 pub use conduwuit_core::matrix::pdu::{PduId, RawPduId};
 use conduwuit_core::{
-	Result, Server, at, err,
+	Err, Result, Server, at, debug, err,
 	matrix::{
 		event::Event,
 		pdu::{PduCount, PduEvent},
 	},
-	utils::{MutexMap, MutexMapGuard, future::TryExtExt, stream::TryIgnore},
+	utils::{self, MutexMap, MutexMapGuard, future::TryExtExt, stream::TryIgnore},
 	warn,
 };
 use futures::{Future, Stream, TryStreamExt, pin_mut};
@@ -25,6 +25,11 @@ use ruma::{
 	events::room::encrypted::Relation,
 };
 use serde::Deserialize;
+use serde_json::value::RawValue as RawJsonValue;
+use tokio::{
+	sync::Notify,
+	time::{MissedTickBehavior, interval},
+};
 
 use self::data::Data;
 pub use self::{create::pdu_fits, data::PdusIterItem};
@@ -56,6 +61,7 @@ struct ExtractBody {
 }
 
 pub struct Service {
+	interrupt: Notify,
 	services: Services,
 	db: Data,
 	pub mutex_insert: RoomMutexMap,
@@ -87,6 +93,8 @@ struct Services {
 	users: Dep<users::Service>,
 }
 
+const PURGE_INTERVAL_SECS: u64 = 3600;
+
 type RoomMutexMap = MutexMap<OwnedRoomId, ()>;
 pub type RoomMutexGuard = MutexMapGuard<OwnedRoomId, ()>;
 
@@ -124,9 +132,35 @@ impl crate::Service for Service {
 			},
 			db: Data::new(&args),
 			mutex_insert: RoomMutexMap::new(),
+			interrupt: Notify::new(),
 		}))
 	}
 
+	#[tracing::instrument(skip_all, name = "unredacted_content_purge", level = "debug")]
+	async fn worker(self: Arc<Self>) -> Result<()> {
+		// Runs while disabled too, so old retained content still ages out.
+		let mut i = interval(Duration::from_secs(PURGE_INTERVAL_SECS));
+		i.set_missed_tick_behavior(MissedTickBehavior::Delay);
+		loop {
+			tokio::select! {
+				() = self.interrupt.notified() => break,
+				_ = i.tick() => (),
+			}
+
+			let purged = self
+				.db
+				.purge_unredacted_content(self.unredacted_content_cutoff())
+				.await;
+			if purged > 0 {
+				debug!("Purged {purged} expired unredacted event contents");
+			}
+		}
+
+		Ok(())
+	}
+
+	fn interrupt(&self) { self.interrupt.notify_waiters(); }
+
 	async fn memory_usage(&self, out: &mut (dyn Write + Send)) -> Result {
 		let mutex_insert = self.mutex_insert.len();
 		writeln!(out, "insert_mutex: {mutex_insert}")?;
@@ -242,6 +276,22 @@ impl Service {
 		self.db.replace_pdu(pdu_id, pdu_json).await
 	}
 
+	pub async fn get_unredacted_content(&self, pdu_id: &RawPduId) -> Result<Box<RawJsonValue>> {
+		let entry = self.db.get_unredacted_content(pdu_id).await?;
+
+		if entry.created < self.unredacted_content_cutoff() {
+			return Err!(Request(NotFound("Unredacted content has expired.")));
+		}
+
+		Ok(entry.content)
+	}
+
+	fn unredacted_content_cutoff(&self) -> u64 {
+		let max_age = self.services.config.unredacted_content_max_age_secs;
+
+		utils::millis_since_unix_epoch().saturating_sub(max_age.saturating_mul(1000))
+	}
+
 	/// Returns an iterator over all PDUs in a room. Unknown rooms produce no
 	/// items.
 	#[inline]

src/service/rooms/timeline/redact.rs

@@ -4,7 +4,7 @@ use conduwuit_core::{
 	utils::{self},
 	warn,
 };
-use ruma::{EventId, events::room::encrypted::Relation};
+use ruma::{EventId, events::room::encrypted::Relation, signatures::content_hash};
 
 use super::{ExtractBody, ExtractRelatesTo};
 use crate::rooms::short::ShortRoomId;
@@ -52,6 +52,20 @@ impl super::Service {
 			.get_room_version(&pdu.room_id_or_hash())
 			.await?;
 
+		if self
+			.services
+			.config
+			.allow_moderators_to_view_redacted_content
+		{
+			let mut pdu_json = self.get_pdu_json_from_id(&pdu_id).await?;
+			pdu_json.remove("event_id");
+			if content_hash(&pdu_json)
+				.is_ok_and(|hash| hash.encode() == pdu.hashes.sha256.trim_end_matches('='))
+			{
+				self.db.save_unredacted_content(&pdu_id, &pdu.content);
+			}
+		}
+
 		pdu.redact(&room_version_id, Some(reason.event_id()))?;
 
 		let obj = utils::to_canonical_object(&pdu).map_err(|e| {