From b2b9c79927600594360baad85aee74477b410380 Mon Sep 17 00:00:00 2001 From: timedout Date: Tue, 10 Nov 2026 20:40:28 +0000 Subject: [PATCH] fix: SEC39 Reviewed-By: timedout Co-Authored-By: Ginger --- src/service/rooms/threads/mod.rs | 151 +++++++++++++++++++++------ src/service/rooms/timeline/redact.rs | 28 ++++- 2 files changed, 142 insertions(+), 37 deletions(-) diff --git a/src/service/rooms/threads/mod.rs b/src/service/rooms/threads/mod.rs index a29ca5511..fa0956abe 100644 --- a/src/service/rooms/threads/mod.rs +++ b/src/service/rooms/threads/mod.rs @@ -3,7 +3,10 @@ use std::{collections::BTreeMap, sync::Arc}; use conduwuit::result::LogErr; use conduwuit_core::{ Event, Result, err, - matrix::pdu::{PduCount, PduEvent, PduId, RawPduId}, + matrix::{ + event::RelationTypeEqual, + pdu::{PduCount, PduEvent, PduId, RawPduId}, + }, utils::{ ReadyExt, stream::{TryIgnore, WidebandExt}, @@ -12,9 +15,14 @@ use conduwuit_core::{ use conduwuit_database::{Deserialized, Map}; use futures::{Stream, StreamExt}; use ruma::{ - CanonicalJsonValue, EventId, OwnedUserId, RoomId, UserId, - api::client::threads::get_threads::v1::IncludeThreads, events::relation::BundledThread, - serde::Raw, uint, + CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedUserId, RoomId, UInt, UserId, + api::{Direction, client::threads::get_threads::v1::IncludeThreads}, + events::{ + AnySyncMessageLikeEvent, AnySyncTimelineEvent, + relation::{BundledThread, RelationType}, + }, + serde::Raw, + uint, }; use serde_json::json; @@ -27,6 +35,7 @@ pub struct Service { struct Services { short: Dep, + pdu_metadata: Dep, timeline: Dep, } @@ -42,6 +51,7 @@ impl crate::Service for Service { }, services: Services { short: args.depend::("rooms::short"), + pdu_metadata: args.depend::("rooms::pdu_metadata"), timeline: args.depend::("rooms::timeline"), }, })) @@ -51,6 +61,88 @@ impl crate::Service for Service { } impl Service { + /// Serializes an event for the `latest_event` of a thread's aggregation, + /// stripping `unsigned` as we don't want to send e.g. txnid to clients. + fn latest_event(event: &E) -> Raw + where + E: Event, + { + let mut latest_pdu = event.clone().into_pdu(); + latest_pdu.unsigned = None; + latest_pdu + .into_format::>() + .cast_unchecked() + } + + fn set_thread(unsigned: &mut CanonicalJsonObject, thread: BundledThread) { + let content = serde_json::to_value(thread).expect("to_value always works"); + unsigned.insert( + "m.relations".to_owned(), + json!({ "m.thread": content }) + .try_into() + .expect("thread is valid json"), + ); + } + + /// Remove `event` from the aggregation of the thread matching + /// `root_event_id` + pub async fn remove_from_thread(&self, root_event_id: &EventId, event: &E) -> Result + where + E: Event + Send + Sync, + { + let Ok(root_id) = self.services.timeline.get_pdu_id(root_event_id).await else { + return Ok(()); + }; + + let mut root_pdu_json = self + .services + .timeline + .get_pdu_json_from_id(&root_id) + .await + .map_err(|e| err!(Database("Thread root pdu not found: {e:?}")))?; + + let Some(CanonicalJsonValue::Object(unsigned)) = root_pdu_json.get_mut("unsigned") else { + return Ok(()); + }; + + let thread: Vec<_> = self + .services + .pdu_metadata + .get_relations( + event.sender(), + &event.room_id_or_hash(), + root_event_id, + PduCount::max(), + usize::MAX, + 0, + Direction::Backward, + ) + .await + .into_iter() + .filter(|(_, pdu)| RelationType::Thread.relation_type_equal(pdu)) + .collect(); + + match thread.first() { + | Some((_, latest)) => Self::set_thread( + unsigned, + BundledThread::new( + Self::latest_event(latest), + thread.len().try_into().unwrap_or(UInt::MAX), + true, + ), + ), + | None => { + // No events are left in the thread, or the root is backfilled + unsigned.remove("m.relations"); + }, + } + + self.services + .timeline + .replace_pdu(&root_id, &root_pdu_json) + .await + } + pub async fn add_to_thread(&self, root_event_id: &EventId, event: &E) -> Result where E: Event + Send + Sync, @@ -78,11 +170,20 @@ impl Service { .await .map_err(|e| err!(Request(InvalidParam("Thread root pdu not found: {e:?}"))))?; - if let CanonicalJsonValue::Object(unsigned) = root_pdu_json - .entry("unsigned".to_owned()) - .or_insert_with(|| CanonicalJsonValue::Object(BTreeMap::default())) + // get_relations doesn't support backfilled targets, so redactions + // couldn't update the aggregation of a backfilled root + // + // TODO: Support backfilled threads. This needs + // `src/service/rooms/pdu_metadata/mod.rs`'s + // (TODO: Relations with backfilled pdus) to be implemented, + // and then this check can be removed. + if matches!(root_id.pdu_count(), PduCount::Normal(_)) + && let CanonicalJsonValue::Object(unsigned) = root_pdu_json + .entry("unsigned".to_owned()) + .or_insert_with(|| CanonicalJsonValue::Object(BTreeMap::default())) { - if let Some(mut relations) = unsigned + let latest_event = Self::latest_event(event); + let thread = match unsigned .get("m.relations") .and_then(|r| r.as_object()) .and_then(|r| r.get("m.thread")) @@ -90,34 +191,16 @@ impl Service { serde_json::from_value::(relations.clone().into()).ok() }) { // Thread already existed - relations.count = relations.count.saturating_add(uint!(1)); - relations.latest_event = Raw::from_json(event.content().to_owned()); - - let content = serde_json::to_value(relations).expect("to_value always works"); - - unsigned.insert( - "m.relations".to_owned(), - json!({ "m.thread": content }) - .try_into() - .expect("thread is valid json"), - ); - } else { + | Some(mut thread) => { + thread.count = thread.count.saturating_add(uint!(1)); + thread.latest_event = latest_event; + thread + }, // New thread - let relations = BundledThread::new( - Raw::from_json(event.content().to_owned()), - uint!(1), - true, - ); + | None => BundledThread::new(latest_event, uint!(1), true), + }; - let content = serde_json::to_value(relations).expect("to_value always works"); - - unsigned.insert( - "m.relations".to_owned(), - json!({ "m.thread": content }) - .try_into() - .expect("thread is valid json"), - ); - } + Self::set_thread(unsigned, thread); self.services .timeline diff --git a/src/service/rooms/timeline/redact.rs b/src/service/rooms/timeline/redact.rs index efa19dff5..92e91cffb 100644 --- a/src/service/rooms/timeline/redact.rs +++ b/src/service/rooms/timeline/redact.rs @@ -2,10 +2,11 @@ use conduwuit_core::{ Result, err, matrix::event::Event, utils::{self}, + warn, }; -use ruma::EventId; +use ruma::{EventId, events::room::encrypted::Relation}; -use super::ExtractBody; +use super::{ExtractBody, ExtractRelatesTo}; use crate::rooms::short::ShortRoomId; impl super::Service { @@ -31,6 +32,12 @@ impl super::Service { err!(Database(error!(?pdu_id, %event_id, ?e, "PDU ID points to invalid PDU."))) })?; + let thread_root = match pdu.get_content::() { + | Ok(ExtractRelatesTo { relates_to: Relation::Thread(thread) }) => + Some(thread.event_id), + | _ => None, + }; + if let Ok(content) = pdu.get_content::() { if let Some(body) = content.body { self.services @@ -51,6 +58,21 @@ impl super::Service { err!(Database(error!(%event_id, ?e, "Failed to convert PDU to canonical JSON"))) })?; - self.replace_pdu(&pdu_id, &obj).await + self.replace_pdu(&pdu_id, &obj).await?; + + // So the thread's root doesn't keep the original content in its + // `latest_event` aggregation + if let Some(thread_root) = thread_root { + self.services + .threads + .remove_from_thread(&thread_root, &pdu) + .await + .inspect_err(|e| { + warn!("Failed to remove redacted event from thread {thread_root}: {e:?}"); + }) + .ok(); + } + + Ok(()) } }