State-reset and security mitigations.

Upgrade Ruma to present.

The following are intentionally benign for activation in a later commit:

- Hydra backports not default.
- Room version 12 not default.
- Room version 12 not listed as stable.

Do not enable them manually or you can brick your database.

Signed-off-by: Jason Volk <jason@zemos.net>

src/service/rooms/event_handler/fetch_auth.rs

@@ -140,10 +140,7 @@ async fn fetch_auth_chain(
 		let Ok(res) = self
 			.services
 			.sending
-			.send_federation_request(origin, get_event::v1::Request {
-				event_id: next_id.clone(),
-				include_unredacted_content: None,
-			})
+			.send_federation_request(origin, get_event::v1::Request { event_id: next_id.clone() })
 			.await
 			.inspect_err(|e| debug_error!("Failed to fetch event {next_id}: {e}"))
 		else {

src/service/rooms/event_handler/fetch_prev.rs

@@ -119,10 +119,10 @@ where
 		// This return value is the key used for sorting events,
 		// events are then sorted by power level, time,
 		// and lexically by event_id.
-		Ok((int!(0), MilliSecondsSinceUnixEpoch(origin_server_ts)))
+		Ok((int!(0).into(), MilliSecondsSinceUnixEpoch(origin_server_ts)))
 	};
 
-	let sorted = state_res::lexicographical_topological_sort(&graph, &event_fetch)
+	let sorted = state_res::topological_sort(&graph, &event_fetch)
 		.await
 		.map_err(|e| err!(Database(error!("Error sorting prev events: {e}"))))?;
 

src/service/rooms/event_handler/handle_incoming_pdu.rs

@@ -4,11 +4,14 @@ use futures::{
 };
 use ruma::{CanonicalJsonObject, EventId, RoomId, ServerName, UserId, events::StateEventType};
 use tuwunel_core::{
-	Err, Result, debug, debug::INFO_SPAN_LEVEL, err, implement, matrix::Event,
-	utils::stream::IterStream, warn,
+	Err, Result, debug,
+	debug::INFO_SPAN_LEVEL,
+	err, implement,
+	matrix::{Event, room_version},
+	utils::stream::IterStream,
+	warn,
 };
 
-use super::get_room_version_id;
 use crate::rooms::timeline::RawPduId;
 
 /// When receiving an event one needs to:
@@ -47,7 +50,7 @@ use crate::rooms::timeline::RawPduId;
 	ret(Debug),
 )]
 pub async fn handle_incoming_pdu<'a>(
-	&self,
+	&'a self,
 	origin: &'a ServerName,
 	room_id: &'a RoomId,
 	event_id: &'a EventId,
@@ -107,11 +110,10 @@ pub async fn handle_incoming_pdu<'a>(
 		return Err!(Request(Forbidden("Federation of this room is disabled by this server.")));
 	}
 
-	let room_version = get_room_version_id(create_event)?;
+	let room_version = room_version::from_create_event(create_event)?;
 
 	let (incoming_pdu, val) = self
 		.handle_outlier_pdu(origin, room_id, event_id, pdu, &room_version, false)
-		.boxed()
 		.await?;
 
 	// 8. if not timeline event: stop

src/service/rooms/event_handler/handle_outlier_pdu.rs

@@ -1,17 +1,16 @@
 use std::collections::{HashMap, hash_map};
 
-use futures::future::ready;
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, EventId, RoomId, RoomVersionId, ServerName,
-	events::StateEventType,
+	events::{StateEventType, TimelineEventType},
 };
 use tuwunel_core::{
 	Err, Result, debug, debug_info, err, implement,
-	matrix::{Event, PduEvent},
+	matrix::{Event, PduEvent, event::TypeExt, room_version},
 	state_res, trace, warn,
 };
 
-use super::{check_room_id, to_room_version};
+use super::check_room_id;
 
 #[implement(super::Service)]
 pub(super) async fn handle_outlier_pdu(
@@ -40,7 +39,13 @@ pub(super) async fn handle_outlier_pdu(
 		| Ok(ruma::signatures::Verified::Signatures) => {
 			// Redact
 			debug_info!("Calculated hash does not match (redaction): {event_id}");
-			let Ok(obj) = ruma::canonical_json::redact(pdu_json, room_version, None) else {
+			let Some(rules) = room_version.rules() else {
+				return Err!(Request(UnsupportedRoomVersion(
+					"Cannot redact event for unknown room version {room_version:?}."
+				)));
+			};
+
+			let Ok(obj) = ruma::canonical_json::redact(pdu_json, &rules.redaction, None) else {
 				return Err!(Request(InvalidParam("Redaction failed")));
 			};
 
@@ -62,8 +67,7 @@ pub(super) async fn handle_outlier_pdu(
 
 	// Now that we have checked the signature and hashes we can add the eventID and
 	// convert to our PduEvent type
-	pdu_json
-		.insert("event_id".to_owned(), CanonicalJsonValue::String(event_id.as_str().to_owned()));
+	pdu_json.insert("event_id".to_owned(), CanonicalJsonValue::String(event_id.to_string()));
 
 	let event = serde_json::from_value::<PduEvent>(serde_json::to_value(&pdu_json)?)
 		.map_err(|e| err!(Request(BadJson(debug_warn!("Event is not a valid PDU: {e}")))))?;
@@ -83,16 +87,28 @@ pub(super) async fn handle_outlier_pdu(
 	// 6. Reject "due to auth events" if the event doesn't pass auth based on the
 	//    auth events
 	debug!("Checking based on auth events");
+
+	let room_rules = room_version::rules(room_version)?;
+	let is_create = *event.kind() == TimelineEventType::RoomCreate;
+	let is_hydra = room_rules
+		.authorization
+		.room_create_event_id_as_room_id;
+
+	let hydra_create_id = (is_hydra && !is_create).then_some(event.room_id().as_event_id()?);
+	let auth_event_ids = event
+		.auth_events()
+		.map(ToOwned::to_owned)
+		.chain(hydra_create_id.into_iter());
+
 	// Build map of auth events
-	let mut auth_events = HashMap::with_capacity(event.auth_events().count());
-	for id in event.auth_events() {
-		let Ok(auth_event) = self.services.timeline.get_pdu(id).await else {
+	let mut auth_events = HashMap::with_capacity(event.auth_events().count().saturating_add(1));
+	for id in auth_event_ids {
+		let Ok(auth_event) = self.services.timeline.get_pdu(&id).await else {
 			warn!("Could not find auth event {id}");
 			continue;
 		};
 
 		check_room_id(room_id, &auth_event)?;
-
 		match auth_events.entry((
 			auth_event.kind.to_string().into(),
 			auth_event
@@ -119,24 +135,20 @@ pub(super) async fn handle_outlier_pdu(
 		return Err!(Request(InvalidParam("Incoming event refers to wrong create event.")));
 	}
 
-	let state_fetch = |ty: &StateEventType, sk: &str| {
-		let key = (ty.to_owned(), sk.into());
-		ready(auth_events.get(&key).map(ToOwned::to_owned))
-	};
-
-	let auth_check = state_res::event_auth::auth_check(
-		&to_room_version(room_version),
+	state_res::auth_check(
+		&room_rules,
 		&event,
-		None, // TODO: third party invite
-		state_fetch,
+		&async |event_id| self.event_fetch(&event_id).await,
+		&async |event_type, state_key| {
+			auth_events
+				.get(&event_type.with_state_key(state_key.as_str()))
+				.map(ToOwned::to_owned)
+				.ok_or_else(|| err!(Request(NotFound("state not found"))))
+		},
 	)
 	.await
 	.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
 
-	if !auth_check {
-		return Err!(Request(Forbidden("Auth check failed")));
-	}
-
 	trace!("Validation successful.");
 
 	// 7. Persist the event as an outlier.

src/service/rooms/event_handler/mod.rs

@@ -19,12 +19,9 @@ use std::{
 };
 
 use async_trait::async_trait;
-use ruma::{
-	EventId, OwnedEventId, OwnedRoomId, RoomId, RoomVersionId,
-	events::room::create::RoomCreateEventContent,
-};
+use ruma::{EventId, OwnedRoomId, RoomId};
 use tuwunel_core::{
-	Err, Result, RoomVersion, Server, implement,
+	Err, Result, Server, implement,
 	matrix::{Event, PduEvent},
 	utils::{MutexMap, continue_exponential_backoff},
 };
@@ -126,17 +123,13 @@ fn is_backed_off(&self, event_id: &EventId, range: Range<Duration>) -> bool {
 }
 
 #[implement(Service)]
-async fn event_exists(&self, event_id: OwnedEventId) -> bool {
-	self.services.timeline.pdu_exists(&event_id).await
+async fn event_exists(&self, event_id: &EventId) -> bool {
+	self.services.timeline.pdu_exists(event_id).await
 }
 
 #[implement(Service)]
-async fn event_fetch(&self, event_id: OwnedEventId) -> Option<PduEvent> {
-	self.services
-		.timeline
-		.get_pdu(&event_id)
-		.await
-		.ok()
+async fn event_fetch(&self, event_id: &EventId) -> Result<PduEvent> {
+	self.services.timeline.get_pdu(event_id).await
 }
 
 fn check_room_id<Pdu: Event>(room_id: &RoomId, pdu: &Pdu) -> Result {
@@ -151,15 +144,3 @@ fn check_room_id<Pdu: Event>(room_id: &RoomId, pdu: &Pdu) -> Result {
 
 	Ok(())
 }
-
-fn get_room_version_id<Pdu: Event>(create_event: &Pdu) -> Result<RoomVersionId> {
-	let content: RoomCreateEventContent = create_event.get_content()?;
-	let room_version = content.room_version;
-
-	Ok(room_version)
-}
-
-#[inline]
-fn to_room_version(room_version_id: &RoomVersionId) -> RoomVersion {
-	RoomVersion::new(room_version_id).expect("room version is supported")
-}

src/service/rooms/event_handler/resolve_state.rs

@@ -1,14 +1,11 @@
-use std::{
-	borrow::Borrow,
-	collections::{HashMap, HashSet},
-	sync::Arc,
-};
+use std::{borrow::Borrow, collections::HashMap, sync::Arc};
 
-use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, future::try_join};
+use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
 use ruma::{OwnedEventId, RoomId, RoomVersionId};
 use tuwunel_core::{
-	Error, Result, err, implement,
-	state_res::{self, StateMap},
+	Result, err, implement,
+	matrix::room_version,
+	state_res::{self, AuthSet, StateMap},
 	trace,
 	utils::stream::{IterStream, ReadyExt, TryWidebandExt, WidebandExt},
 };
@@ -47,9 +44,9 @@ pub async fn resolve_state(
 			self.services
 				.auth_chain
 				.event_ids_iter(room_id, state.values().map(Borrow::borrow))
-				.try_collect()
+				.try_collect::<AuthSet<OwnedEventId>>()
 		})
-		.try_collect::<Vec<HashSet<OwnedEventId>>>();
+		.ready_filter_map(Result::ok);
 
 	let fork_states = fork_states
 		.iter()
@@ -62,17 +59,12 @@ pub async fn resolve_state(
 				.multi_get_statekey_from_short(shortstatekeys)
 				.zip(event_ids)
 				.ready_filter_map(|(ty_sk, id)| Some((ty_sk.ok()?, id)))
-				.collect()
-		})
-		.map(Ok::<_, Error>)
-		.try_collect::<Vec<StateMap<OwnedEventId>>>();
-
-	let (fork_states, auth_chain_sets) = try_join(fork_states, auth_chain_sets).await?;
+				.collect::<StateMap<OwnedEventId>>()
+		});
 
 	trace!("Resolving state");
 	let state = self
-		.state_resolution(room_version_id, fork_states.iter(), &auth_chain_sets)
-		.boxed()
+		.state_resolution(room_version_id, fork_states, auth_chain_sets)
 		.await?;
 
 	trace!("State resolution done.");
@@ -104,19 +96,24 @@ pub async fn resolve_state(
 }
 
 #[implement(super::Service)]
-#[tracing::instrument(name = "ruma", level = "debug", skip_all)]
-pub async fn state_resolution<'a, StateSets>(
-	&'a self,
-	room_version: &'a RoomVersionId,
+pub(super) async fn state_resolution<StateSets, AuthSets>(
+	&self,
+	room_version: &RoomVersionId,
 	state_sets: StateSets,
-	auth_chain_sets: &'a [HashSet<OwnedEventId>],
+	auth_chains: AuthSets,
 ) -> Result<StateMap<OwnedEventId>>
 where
-	StateSets: Iterator<Item = &'a StateMap<OwnedEventId>> + Clone + Send,
+	StateSets: Stream<Item = StateMap<OwnedEventId>> + Send,
+	AuthSets: Stream<Item = AuthSet<OwnedEventId>> + Send,
 {
-	let event_fetch = |event_id| self.event_fetch(event_id);
-	let event_exists = |event_id| self.event_exists(event_id);
-	state_res::resolve(room_version, state_sets, auth_chain_sets, &event_fetch, &event_exists)
-		.map_err(|e| err!(error!("State resolution failed: {e:?}")))
-		.await
+	state_res::resolve(
+		&room_version::rules(room_version)?,
+		state_sets,
+		auth_chains,
+		&async |event_id: OwnedEventId| self.event_fetch(&event_id).await,
+		&async |event_id: OwnedEventId| self.event_exists(&event_id).await,
+		self.services.server.config.hydra_backports,
+	)
+	.map_err(|e| err!(error!("State resolution failed: {e:?}")))
+	.await
 }

src/service/rooms/event_handler/state_at_incoming.rs

@@ -1,15 +1,17 @@
 use std::{
-	borrow::Borrow,
-	collections::{HashMap, HashSet},
-	iter::Iterator,
+	collections::HashMap,
+	iter::{Iterator, once},
 };
 
-use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, future::try_join};
+use futures::{
+	FutureExt, StreamExt, TryFutureExt, TryStreamExt,
+	future::{OptionFuture, try_join},
+};
 use ruma::{OwnedEventId, RoomId, RoomVersionId};
 use tuwunel_core::{
-	Result, err, implement,
-	matrix::{Event, StateMap},
-	trace,
+	Result, apply, err, implement,
+	matrix::{Event, StateMap, state_res::AuthSet},
+	ref_at, trace,
 	utils::stream::{BroadbandExt, IterStream, ReadyExt, TryBroadbandExt, TryWidebandExt},
 };
 
@@ -60,14 +62,16 @@ where
 
 	let (prev_event, mut state) = try_join(prev_event, state).await?;
 
-	if let Some(state_key) = prev_event.state_key {
+	if let Some(state_key) = prev_event.state_key() {
+		let prev_event_type = prev_event.event_type().to_cow_str().into();
+
 		let shortstatekey = self
 			.services
 			.short
-			.get_or_create_shortstatekey(&prev_event.kind.into(), &state_key)
+			.get_or_create_shortstatekey(&prev_event_type, state_key)
 			.await;
 
-		state.insert(shortstatekey, prev_event.event_id);
+		state.insert(shortstatekey, prev_event.event_id().into());
 		// Now it's the state after the pdu
 	}
 
@@ -113,26 +117,28 @@ where
 	};
 
 	trace!("Calculating fork states...");
-	let (fork_states, auth_chain_sets): (Vec<StateMap<_>>, Vec<HashSet<_>>) =
-		extremity_sstatehashes
-			.into_iter()
-			.try_stream()
-			.wide_and_then(|(sstatehash, prev_event)| {
-				self.state_at_incoming_fork(room_id, sstatehash, prev_event)
-			})
-			.try_collect()
-			.map_ok(Vec::into_iter)
-			.map_ok(Iterator::unzip)
-			.await?;
+	let (fork_states, auth_chain_sets) = extremity_sstatehashes
+		.into_iter()
+		.try_stream()
+		.wide_and_then(|(sstatehash, prev_event)| {
+			self.state_at_incoming_fork(room_id, sstatehash, prev_event)
+		})
+		.try_collect()
+		.map_ok(Vec::into_iter)
+		.map_ok(Iterator::unzip)
+		.map_ok(apply!(2, Vec::into_iter))
+		.map_ok(apply!(2, IterStream::stream))
+		.await?;
 
+	trace!("Resolving state");
 	let Ok(new_state) = self
-		.state_resolution(room_version_id, fork_states.iter(), &auth_chain_sets)
-		.boxed()
+		.state_resolution(room_version_id, fork_states, auth_chain_sets)
 		.await
 	else {
 		return Ok(None);
 	};
 
+	trace!("State resolution done.");
 	new_state
 		.into_iter()
 		.stream()
@@ -150,41 +156,55 @@ where
 }
 
 #[implement(super::Service)]
+#[tracing::instrument(
+	name = "fork",
+	level = "debug",
+	skip_all,
+	fields(
+		?sstatehash,
+		prev_event = ?prev_event.event_id(),
+	)
+)]
 async fn state_at_incoming_fork<Pdu>(
 	&self,
 	room_id: &RoomId,
 	sstatehash: ShortStateHash,
 	prev_event: Pdu,
-) -> Result<(StateMap<OwnedEventId>, HashSet<OwnedEventId>)>
+) -> Result<(StateMap<OwnedEventId>, AuthSet<OwnedEventId>)>
 where
 	Pdu: Event,
 {
-	let mut leaf_state: HashMap<_, _> = self
+	let leaf: OptionFuture<_> = prev_event
+		.state_key()
+		.map(async |state_key| {
+			self.services
+				.short
+				.get_or_create_shortstatekey(&prev_event.kind().to_cow_str().into(), state_key)
+				.map(|shortstatekey| once((shortstatekey, prev_event.event_id().to_owned())))
+				.await
+		})
+		.into();
+
+	let leaf_state_after_event: Vec<_> = self
 		.services
 		.state_accessor
 		.state_full_ids(sstatehash)
+		.chain(leaf.await.into_iter().flatten().stream())
 		.collect()
 		.await;
 
-	if let Some(state_key) = prev_event.state_key() {
-		let shortstatekey = self
-			.services
-			.short
-			.get_or_create_shortstatekey(&prev_event.kind().to_string().into(), state_key)
-			.await;
-
-		let event_id = prev_event.event_id();
-		leaf_state.insert(shortstatekey, event_id.to_owned());
-		// Now it's the state after the pdu
-	}
+	let starting_events = leaf_state_after_event
+		.iter()
+		.map(ref_at!(1))
+		.map(AsRef::as_ref);
 
 	let auth_chain = self
 		.services
 		.auth_chain
-		.event_ids_iter(room_id, leaf_state.values().map(Borrow::borrow))
+		.event_ids_iter(room_id, starting_events)
 		.try_collect();
 
-	let fork_state = leaf_state
+	let fork_state = leaf_state_after_event
 		.iter()
 		.stream()
 		.broad_then(|(k, id)| {

src/service/rooms/event_handler/upgrade_outlier_pdu.rs

@@ -1,18 +1,18 @@
 use std::{borrow::Borrow, iter::once, sync::Arc, time::Instant};
 
-use futures::{FutureExt, StreamExt, future::ready};
+use futures::{FutureExt, StreamExt};
 use ruma::{
-	CanonicalJsonObject, EventId, RoomId, RoomVersionId, ServerName, events::StateEventType,
+	CanonicalJsonObject, EventId, OwnedEventId, RoomId, RoomVersionId, ServerName,
+	events::StateEventType,
 };
 use tuwunel_core::{
 	Err, Result, debug, debug_info, err, implement, is_equal_to,
-	matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
+	matrix::{Event, EventTypeExt, PduEvent, StateKey, room_version, state_res},
 	trace,
 	utils::stream::{BroadbandExt, ReadyExt},
 	warn,
 };
 
-use super::to_room_version;
 use crate::rooms::{
 	state_compressor::{CompressedState, HashSetCompressStateEvent},
 	timeline::RawPduId,
@@ -50,6 +50,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 
 	debug!("Upgrading to timeline pdu");
 	let timer = Instant::now();
+	let room_rules = room_version::rules(room_version)?;
 
 	// 10. Fetch missing state and auth chain events by calling /state_ids at
 	//     backwards extremities doing all the checks in this list starting at 1.
@@ -77,35 +78,26 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 
 	debug!("Performing auth check");
 	// 11. Check the auth of the event passes based on the state of the event
-	let state_fetch_state = &state_at_incoming_event;
-	let state_fetch = |k: StateEventType, s: StateKey| async move {
+	let state_fetch = async |k: StateEventType, s: StateKey| {
 		let shortstatekey = self
 			.services
 			.short
-			.get_shortstatekey(&k, &s)
-			.await
-			.ok()?;
+			.get_shortstatekey(&k, s.as_str())
+			.await?;
 
-		let event_id = state_fetch_state.get(&shortstatekey)?;
-		self.services
-			.timeline
-			.get_pdu(event_id)
-			.await
-			.ok()
+		let event_id = state_at_incoming_event
+			.get(&shortstatekey)
+			.ok_or_else(|| {
+				err!(Request(NotFound(
+					"shortstatekey {shortstatekey:?} not found for ({k:?},{s:?})"
+				)))
+			})?;
+
+		self.services.timeline.get_pdu(event_id).await
 	};
 
-	let auth_check = state_res::event_auth::auth_check(
-		&to_room_version(room_version),
-		&incoming_pdu,
-		None, // TODO: third party invite
-		|ty, sk| state_fetch(ty.clone(), sk.into()),
-	)
-	.await
-	.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
-
-	if !auth_check {
-		return Err!(Request(Forbidden("Event has failed auth check with state at the event.")));
-	}
+	let event_fetch = async |event_id: OwnedEventId| self.event_fetch(&event_id).await;
+	state_res::auth_check(&room_rules, &incoming_pdu, &event_fetch, &state_fetch).await?;
 
 	debug!("Gathering auth events");
 	let auth_events = self
@@ -117,29 +109,25 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 			incoming_pdu.sender(),
 			incoming_pdu.state_key(),
 			incoming_pdu.content(),
+			&room_rules.authorization,
+			true,
 		)
 		.await?;
 
-	let state_fetch = |k: &StateEventType, s: &str| {
-		let key = k.with_state_key(s);
-		ready(auth_events.get(&key).map(ToOwned::to_owned))
+	let state_fetch = async |k: StateEventType, s: StateKey| {
+		auth_events
+			.get(&k.with_state_key(s.as_str()))
+			.map(ToOwned::to_owned)
+			.ok_or_else(|| err!(Request(NotFound("state event not found"))))
 	};
 
-	let auth_check = state_res::event_auth::auth_check(
-		&to_room_version(room_version),
-		&incoming_pdu,
-		None, // third-party invite
-		state_fetch,
-	)
-	.await
-	.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
+	state_res::auth_check(&room_rules, &incoming_pdu, &event_fetch, &state_fetch).await?;
 
 	// Soft fail check before doing state res
 	debug!("Performing soft-fail check");
-	let soft_fail = match (auth_check, incoming_pdu.redacts_id(room_version)) {
-		| (false, _) => true,
-		| (true, None) => false,
-		| (true, Some(redact_id)) =>
+	let soft_fail = match incoming_pdu.redacts_id(room_version) {
+		| None => false,
+		| Some(redact_id) =>
 			!self
 				.services
 				.state_accessor
@@ -224,11 +212,13 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 			.services
 			.state_compressor
 			.save_state(room_id, new_room_state)
+			.boxed()
 			.await?;
 
 		self.services
 			.state
 			.force_state(room_id, shortstatehash, added, removed, &state_lock)
+			.boxed()
 			.await?;
 	}