fix: SEC39

Reviewed-By: timedout <git@nexy7574.co.uk>
Co-Authored-By: Ginger <ginger@gingershaped.computer>

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<rooms::short::Service>,
+	pdu_metadata: Dep<rooms::pdu_metadata::Service>,
 	timeline: Dep<rooms::timeline::Service>,
 }
 
@@ -42,6 +51,7 @@ impl crate::Service for Service {
 			},
 			services: Services {
 				short: args.depend::<rooms::short::Service>("rooms::short"),
+				pdu_metadata: args.depend::<rooms::pdu_metadata::Service>("rooms::pdu_metadata"),
 				timeline: args.depend::<rooms::timeline::Service>("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<E>(event: &E) -> Raw<AnySyncMessageLikeEvent>
+	where
+		E: Event,
+	{
+		let mut latest_pdu = event.clone().into_pdu();
+		latest_pdu.unsigned = None;
+		latest_pdu
+			.into_format::<Raw<AnySyncTimelineEvent>>()
+			.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<E>(&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<E>(&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::<BundledThread>(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

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::<ExtractRelatesTo>() {
+			| Ok(ExtractRelatesTo { relates_to: Relation::Thread(thread) }) =>
+				Some(thread.event_id),
+			| _ => None,
+		};
+
 		if let Ok(content) = pdu.get_content::<ExtractBody>() {
 			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(())
 	}
 }