From f154ecb5428b0ab866dc3781798c02400377e0cb Mon Sep 17 00:00:00 2001 From: Erwan Leboucher Date: Sun, 13 Sep 2026 20:44:03 +0200 Subject: [PATCH] feat: Expire retained unredacted content and address review --- .../+msc2815-unredacted-content.feature.md | 2 +- conduwuit-example.toml | 12 +++- src/api/client/room/event.rs | 57 +++++++++++-------- src/core/config/mod.rs | 16 +++++- src/service/rooms/timeline/data.rs | 44 ++++++++++++-- src/service/rooms/timeline/mod.rs | 54 ++++++++++++++++-- 6 files changed, 146 insertions(+), 39 deletions(-) diff --git a/changelog.d/+msc2815-unredacted-content.feature.md b/changelog.d/+msc2815-unredacted-content.feature.md index 09cb3f3ea..f38ba7d91 100644 --- a/changelog.d/+msc2815-unredacted-content.feature.md +++ b/changelog.d/+msc2815-unredacted-content.feature.md @@ -1 +1 @@ -Added [MSC2815](https://github.com/matrix-org/matrix-spec-proposals/pull/2815), letting room moderators fetch the content of redacted events, behind the `allow_moderators_to_view_redacted_content` config option. Contributed by @eleboucher. +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. diff --git a/conduwuit-example.toml b/conduwuit-example.toml index cc2ac8d75..ec955ba76 100644 --- a/conduwuit-example.toml +++ b/conduwuit-example.toml @@ -1666,13 +1666,21 @@ # Allow room moderators to view redacted event content ([MSC2815]). # -# Redacted content is kept in the database indefinitely, and readable by -# users at or above the room's `redact` power level, and server admins. +# 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 # #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. +# +#unredacted_content_max_age_secs = 7776000 (90 days) + # Maximum time federation user can indicate typing. # # env: CONTINUWUITY_TYPING_FEDERATION_TIMEOUT_S diff --git a/src/api/client/room/event.rs b/src/api/client/room/event.rs index 91626d636..b66f8bcaf 100644 --- a/src/api/client/room/event.rs +++ b/src/api/client/room/event.rs @@ -1,6 +1,6 @@ use axum::extract::{Query, State}; use conduwuit::{Err, Error, Event, PduEvent, Result, debug_warn, err}; -use futures::{FutureExt, TryFutureExt, future::try_join}; +use futures::TryFutureExt; use http::StatusCode; use ruma::{ UserId, @@ -14,7 +14,7 @@ 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, + include_unredacted_content: Option, } /// # `GET /_matrix/client/r0/rooms/{roomId}/event/{eventId}` @@ -69,7 +69,7 @@ pub(crate) async fn get_room_event_route( } if services.config.allow_moderators_to_view_redacted_content - && params.include_unredacted_content.as_deref() == Some("true") + && params.include_unredacted_content.unwrap_or(false) && event.is_redacted() { unredact(services, sender_user, &mut event).await?; @@ -109,33 +109,40 @@ async fn unredact(services: &crate::State, sender_user: &UserId, event: &mut Pdu ))); } - let Ok(pdu_id) = services.rooms.timeline.get_pdu_id(event.event_id()).await else { - return Err(content_unavailable()); - }; - - let Ok(content) = services + let content = services .rooms .timeline - .get_unredacted_content(&pdu_id) - .await - else { - return Err(content_unavailable()); + .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::(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(()) } - -fn content_unavailable() -> Error { - let kind = serde_json::from_value::( - json!({ "errcode": "FI.MAU.MSC2815_UNREDACTED_CONTENT_DELETED" }), - ) - .expect("errcode is a valid custom error kind"); - - Error::Request( - kind, - "The unredacted content of this event is not available.".into(), - StatusCode::NOT_FOUND, - ) -} diff --git a/src/core/config/mod.rs b/src/core/config/mod.rs index 2935da3e7..548963e4c 100644 --- a/src/core/config/mod.rs +++ b/src/core/config/mod.rs @@ -1778,8 +1778,8 @@ pub struct Config { /// Allow room moderators to view redacted event content ([MSC2815]). /// - /// Redacted content is kept in the database indefinitely, and readable by - /// users at or above the room's `redact` power level, and server admins. + /// 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 /// @@ -1787,6 +1787,16 @@ pub struct Config { #[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 @@ -3254,6 +3264,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 { diff --git a/src/service/rooms/timeline/data.rs b/src/service/rooms/timeline/data.rs index 853465044..f416582fc 100644 --- a/src/service/rooms/timeline/data.rs +++ b/src/service/rooms/timeline/data.rs @@ -3,11 +3,17 @@ 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}; @@ -31,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, +} + impl Data { pub(super) fn new(args: &crate::Args<'_>) -> Self { let db = &args.db; @@ -238,19 +250,43 @@ impl Data { } pub(super) fn save_unredacted_content(&self, pdu_id: &RawPduId, content: &RawJsonValue) { - self.pduid_unredactedcontent.raw_put(pdu_id, Json(content)); + 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> { + ) -> Result { 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. diff --git a/src/service/rooms/timeline/mod.rs b/src/service/rooms/timeline/mod.rs index 1e68f21ad..d603757c6 100644 --- a/src/service/rooms/timeline/mod.rs +++ b/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}; @@ -26,6 +26,10 @@ use ruma::{ }; 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}; @@ -57,6 +61,7 @@ struct ExtractBody { } pub struct Service { + interrupt: Notify, services: Services, db: Data, pub mutex_insert: RoomMutexMap, @@ -88,6 +93,8 @@ struct Services { users: Dep, } +const PURGE_INTERVAL_SECS: u64 = 60 * 60; + type RoomMutexMap = MutexMap; pub type RoomMutexGuard = MutexMapGuard; @@ -125,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) -> 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}")?; @@ -243,9 +276,20 @@ impl Service { self.db.replace_pdu(pdu_id, pdu_json).await } - #[inline] pub async fn get_unredacted_content(&self, pdu_id: &RawPduId) -> Result> { - self.db.get_unredacted_content(pdu_id).await + 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