style: Move PDU check functions into their own module

src/service/rooms/event_handler/fetch_auth.rs

@@ -159,13 +159,13 @@ async fn authorise_remote_auth_chain<Pdu>(
 				}
 			}
 			if !self
-				.is_event_self_authorised(
+				.auth_state_check_4(
 					&pdu,
 					room_version_rules,
 					create_event.as_pdu(),
 					&auth_events_by_key,
 				)
-				.await
+				.await?
 			{
 				self.services.pdu_metadata.mark_event_rejected(&event_id);
 			}

src/service/rooms/event_handler/handle_outlier_pdu.rs

@@ -1,13 +1,12 @@
 use std::collections::{BTreeMap, HashMap, hash_map};
 
 use conduwuit::{
-	Err, Event, EventTypeExt, PduEvent, Result, debug, debug_info, debug_warn, err, info,
-	matrix::StateKey, state_res::auth_check, trace, warn,
+	Err, Event, EventTypeExt, PduEvent, Result, debug, debug_info, debug_warn, err, info, trace,
+	warn,
 };
-use futures::future::ready;
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName,
-	canonical_json::redact, events::StateEventType, room_version_rules::RoomVersionRules,
+	canonical_json::redact,
 };
 
 use super::{check_room_id, get_room_version_rules};
@@ -213,13 +212,13 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 		}
 
 		if !self
-			.is_event_self_authorised(
+			.auth_state_check_4(
 				&pdu_event,
 				&room_version_rules,
 				create_event.as_pdu(),
 				&auth_events_by_key,
 			)
-			.await
+			.await?
 		{
 			self.reject_and_persist(event_id, &incoming_pdu);
 			return Err!(Request(Forbidden(
@@ -239,57 +238,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 		Ok((pdu_event, incoming_pdu))
 	}
 
-	/// Helper method that turns the return value of `is_event_self_authorised`
-	/// into a `Result` depending on the value.
-	///
-	/// If the event is not authorised, a Forbidden error is returned.
-	/// Otherwise, an empty `Ok`.
-	pub(super) async fn is_event_self_authorised(
-		&self,
-		pdu: &PduEvent,
-		room_version_rules: &RoomVersionRules,
-		create_event: &PduEvent,
-		auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>,
-	) -> bool {
-		self.expect_event_is_self_authorised(
-			pdu,
-			room_version_rules,
-			create_event,
-			auth_events_by_key,
-		)
-		.await
-		.is_ok()
-	}
-
-	/// Checks PDU check 4: Passes authorisation rules based on the event's auth
-	/// events ([spec]).
-	///
-	/// If the auth check fails, false is returned, otherwise true.
-	///
-	/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
-	pub(super) async fn expect_event_is_self_authorised(
-		&self,
-		pdu: &PduEvent,
-		room_version_rules: &RoomVersionRules,
-		create_event: &PduEvent,
-		auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>,
-	) -> Result<bool> {
-		let state_fetch = |ty: &StateEventType, sk: &str| {
-			let key = (ty.to_owned(), sk.into());
-			ready(auth_events_by_key.get(&key).map(ToOwned::to_owned))
-		};
-
-		auth_check(
-			room_version_rules,
-			pdu,
-			None, // TODO: third party invite
-			state_fetch,
-			create_event,
-		)
-		.await
-		.map_err(|e| err!("Event self-authentication failed: {e:?}"))
-	}
-
 	/// Marks the event as rejected and then saves it as an outlier.
 	pub(super) fn reject_and_persist(&self, event_id: &EventId, pdu: &CanonicalJsonObject) {
 		self.services.pdu_metadata.mark_event_rejected(event_id);

src/service/rooms/event_handler/mod.rs

@@ -6,6 +6,7 @@
 mod handle_incoming_pdu;
 mod handle_outlier_pdu;
 mod parse_incoming_pdu;
+pub mod pdu_checks;
 mod policy_server;
 mod resolve_state;
 mod state_at_incoming;

src/service/rooms/event_handler/pdu_checks.rs

@@ -0,0 +1,217 @@
+use std::collections::HashMap;
+
+use conduwuit::{
+	Err, Event, EventTypeExt, PduEvent, Result, debug, debug::DebugInspect, debug_error, err,
+	info, matrix::StateKey, state_res, trace,
+};
+use futures::future::ready;
+use ruma::{
+	CanonicalJsonObject, OwnedEventId, ServerName, api::error::ErrorKind, events::StateEventType,
+	room_version_rules::RoomVersionRules,
+};
+
+impl super::Service {
+	/// Checks PDU check 4: Passes authorisation rules based on the event's auth
+	/// events ([spec]).
+	///
+	/// If the auth check fails, false is returned, otherwise true.
+	///
+	/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
+	pub(super) async fn auth_state_check_4(
+		&self,
+		pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+		auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>,
+	) -> Result<bool> {
+		let state_fetch = |ty: &StateEventType, sk: &str| {
+			let key = (ty.to_owned(), sk.into());
+			ready(auth_events_by_key.get(&key).map(ToOwned::to_owned))
+		};
+
+		state_res::event_auth::auth_check(
+			room_version_rules,
+			pdu,
+			None, // TODO: third party invite
+			state_fetch,
+			create_event,
+		)
+		.await
+		.map_err(|e| err!("Event self-authentication failed: {e:?}"))
+	}
+
+	/// Checks that the event passes PDU check 5, which ensures that the event
+	/// is authorised based on the state before the event (which is the resolved
+	/// state across all prev events).
+	///
+	/// Returns a boolean indicating whether the event is authorised, and also
+	/// the resolved state before the event for later use. Returns an error if
+	/// state fetching or auth checking fails.
+	pub(super) async fn state_before_check_5(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+		origin: &ServerName,
+	) -> Result<(bool, HashMap<u64, OwnedEventId>)> {
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Resolving state at event"
+		);
+		let room_id = incoming_pdu.room_id_or_hash();
+
+		// If the incoming event only has one prev event, we can just use the state at
+		// that event, but otherwise we have to resolve across each fork. If we're
+		// missing even one of the prev events, we have to ask a remote server for help.
+		//
+		// TODO: this can be optimised by only loading auth chain events into memory,
+		// rather than the entire state.
+		let state_before = if incoming_pdu.prev_events().count() == 1 {
+			self.state_at_incoming_degree_one(&incoming_pdu).await?
+		} else {
+			self.state_at_incoming_resolved(&incoming_pdu, &room_id, room_version_rules)
+				.await?
+		};
+		let state_before = match state_before {
+			| Some(s) => s,
+			| None => {
+				trace!("Could not calculate incoming state, asking remote {origin} for it");
+				self.fetch_state(origin, create_event, &room_id, incoming_pdu.event_id())
+					.await
+					.debug_inspect_err(|e| {
+						debug_error!("Could not fetch state from {origin}: {e}");
+					})?
+			},
+		};
+
+		if state_before.is_empty()
+			&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
+		{
+			// This can happen if the remote sends an event but cannot be reached to fetch
+			// the state at it, and all other servers in the room (which might just be the
+			// unreachable server) are unable to provide required info.
+			// returning an error here allows the upgrade to be attempted at another time.
+			return Err!(Request(Forbidden("Could not resolve incoming state before event")));
+		}
+		trace!(state_events = state_before.len(), "Calculated incoming state");
+
+		let state_fetch_state = &state_before;
+		let state_fetch = |k: StateEventType, s: StateKey| async move {
+			let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
+
+			let event_id = state_fetch_state.get(&shortstatekey)?;
+			self.services.timeline.get_pdu(event_id).await.ok()
+		};
+
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Running state-before auth check"
+		);
+
+		// PDU check: 5
+		let auth_check = state_res::event_auth::auth_check(
+			room_version_rules,
+			incoming_pdu,
+			None, // TODO: third party invite
+			|ty, sk| state_fetch(ty.clone(), sk.into()),
+			create_event.as_pdu(),
+		)
+		.await
+		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
+		Ok((auth_check, state_before))
+	}
+
+	/// Checks that the event passes PDU check 6, which ensures that the event
+	/// is authorised based on the room's current state (which is the resolved
+	/// state across all current forward extremities).
+	///
+	/// Returns a boolean indicating whether the event is authorised, or an
+	/// error if the auth check fails.
+	pub(super) async fn current_state_check_6(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+	) -> Result<bool> {
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Gathering auth events"
+		);
+		let auth_events = self
+			.services
+			.state
+			.get_auth_events(
+				&incoming_pdu.room_id_or_hash(),
+				incoming_pdu.kind(),
+				incoming_pdu.sender(),
+				incoming_pdu.state_key(),
+				incoming_pdu.content(),
+				room_version_rules,
+			)
+			.await?;
+
+		let state_fetch = |k: &StateEventType, s: &str| {
+			let key = k.with_state_key(s);
+			ready(auth_events.get(&key).map(ToOwned::to_owned))
+		};
+
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Running current state auth check"
+		);
+		state_res::event_auth::auth_check(
+			room_version_rules,
+			incoming_pdu,
+			None, // third-party invite
+			state_fetch,
+			create_event.as_pdu(),
+		)
+		.await
+		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))
+	}
+
+	/// Performs PDU check 7 - does the policy server allow this event.
+	///
+	/// If the policy server forbids the event, false is returned. If there is a
+	/// problem contacting the policy server, or it returns an unrecognised
+	/// response, an appropriate error is returned.
+	pub(super) async fn policy_server_check_7(
+		&self,
+		incoming_pdu: &PduEvent,
+		pdu_json: &mut CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<bool> {
+		let event_id = pdu_json
+			.remove("event_id")
+			.expect("event_id should be present in pdu_json at this stage");
+		if let Err(e) = self
+			.policy_server_allows_event(
+				incoming_pdu,
+				pdu_json,
+				&incoming_pdu.room_id_or_hash(),
+				room_version_rules,
+				true,
+			)
+			.await
+			.debug_inspect(|()| {
+				debug!(
+					event_id = %incoming_pdu.event_id,
+					"Event has passed policy server check."
+				);
+			}) {
+			return if matches!(e.kind(), ErrorKind::Forbidden) {
+				info!(
+					event_id = %incoming_pdu.event_id,
+					error = %e,
+					"Event has been marked as spam by policy server: {}",
+					e.message(),
+				);
+				Ok(false)
+			} else {
+				Err(e)
+			};
+		}
+		pdu_json.insert("event_id".to_owned(), event_id);
+		Ok(true)
+	}
+}

src/service/rooms/event_handler/upgrade_outlier_pdu.rs

@@ -1,9 +1,8 @@
 use std::{borrow::Borrow, collections::HashMap, sync::Arc, time::Instant};
 
 use conduwuit::{
-	Err, Result, debug, debug_error, debug_info, err, info, is_equal_to, is_true,
-	matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
-	result::DebugInspect,
+	Err, Result, debug, debug_info, is_equal_to, is_true,
+	matrix::{Event, PduEvent},
 	trace,
 	utils::{
 		IterStream,
@@ -11,10 +10,10 @@
 		stream::{BroadbandExt, ReadyExt},
 	},
 };
-use futures::{FutureExt, StreamExt, future::ready};
+use futures::{FutureExt, StreamExt};
 use ruma::{
-	CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId, ServerName, api::error::ErrorKind,
-	events::StateEventType, room_version_rules::RoomVersionRules,
+	CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId, ServerName, events::StateEventType,
+	room_version_rules::RoomVersionRules,
 };
 use tokio::join;
 
@@ -79,12 +78,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 		// either need to have all the prev events locally, or ask a remote server
 		// for the state at the event.
 		let (passes_state_before, state_before) = self
-			.is_authorised_by_state_before(
-				&incoming_pdu,
-				&room_version_rules,
-				create_event,
-				origin,
-			)
+			.state_before_check_5(&incoming_pdu, &room_version_rules, create_event, origin)
 			.await?;
 
 		if !passes_state_before {
@@ -108,7 +102,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 		);
 		let state_lock = self.services.state.mutex.lock(room_id).await;
 		let passes_current_state = self
-			.is_authorised_by_current_state(&incoming_pdu, &room_version_rules, create_event)
+			.current_state_check_6(&incoming_pdu, &room_version_rules, create_event)
 			.await?;
 
 		// Determine whether this PDU should be soft-failed.
@@ -133,7 +127,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 			// since the policy server refusing this event also soft-fails it.
 			debug!(event_id = %incoming_pdu.event_id, "Checking policy server for event");
 			should_soft_fail = self
-				.policy_server_check(&incoming_pdu, &mut val, &room_version_rules)
+				.policy_server_check_7(&incoming_pdu, &mut val, &room_version_rules)
 				.await?;
 
 			// TODO: this is supposed to hide redactions from policy servers and janitorial
@@ -234,181 +228,6 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 		Ok(pdu_id)
 	}
 
-	/// Checks that the event passes PDU check 5, which ensures that the event
-	/// is authorised based on the state before the event (which is the resolved
-	/// state across all prev events).
-	///
-	/// Returns a boolean indicating whether the event is authorised, and also
-	/// the resolved state before the event for later use. Returns an error if
-	/// state fetching or auth checking fails.
-	async fn is_authorised_by_state_before(
-		&self,
-		incoming_pdu: &PduEvent,
-		room_version_rules: &RoomVersionRules,
-		create_event: &PduEvent,
-		origin: &ServerName,
-	) -> Result<(bool, HashMap<u64, OwnedEventId>)> {
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Resolving state at event"
-		);
-		let room_id = incoming_pdu.room_id_or_hash();
-
-		// If the incoming event only has one prev event, we can just use the state at
-		// that event, but otherwise we have to resolve across each fork. If we're
-		// missing even one of the prev events, we have to ask a remote server for help.
-		//
-		// TODO: this can be optimised by only loading auth chain events into memory,
-		// rather than the entire state.
-		let state_before = if incoming_pdu.prev_events().count() == 1 {
-			self.state_at_incoming_degree_one(&incoming_pdu).await?
-		} else {
-			self.state_at_incoming_resolved(&incoming_pdu, &room_id, room_version_rules)
-				.await?
-		};
-		let state_before = match state_before {
-			| Some(s) => s,
-			| None => {
-				trace!("Could not calculate incoming state, asking remote {origin} for it");
-				self.fetch_state(origin, create_event, &room_id, incoming_pdu.event_id())
-					.await
-					.debug_inspect_err(|e| {
-						debug_error!("Could not fetch state from {origin}: {e}");
-					})?
-			},
-		};
-
-		if state_before.is_empty()
-			&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
-		{
-			// This can happen if the remote sends an event but cannot be reached to fetch
-			// the state at it, and all other servers in the room (which might just be the
-			// unreachable server) are unable to provide required info.
-			// returning an error here allows the upgrade to be attempted at another time.
-			return Err!(Request(Forbidden("Could not resolve incoming state before event")));
-		}
-		trace!(state_events = state_before.len(), "Calculated incoming state");
-
-		let state_fetch_state = &state_before;
-		let state_fetch = |k: StateEventType, s: StateKey| async move {
-			let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
-
-			let event_id = state_fetch_state.get(&shortstatekey)?;
-			self.services.timeline.get_pdu(event_id).await.ok()
-		};
-
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Running state-before auth check"
-		);
-
-		// PDU check: 5
-		let auth_check = state_res::event_auth::auth_check(
-			room_version_rules,
-			incoming_pdu,
-			None, // TODO: third party invite
-			|ty, sk| state_fetch(ty.clone(), sk.into()),
-			create_event.as_pdu(),
-		)
-		.await
-		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
-		Ok((auth_check, state_before))
-	}
-
-	/// Checks that the event passes PDU check 6, which ensures that the event
-	/// is authorised based on the room's current state (which is the resolved
-	/// state across all current forward extremities).
-	///
-	/// Returns a boolean indicating whether the event is authorised, or an
-	/// error if the auth check fails.
-	async fn is_authorised_by_current_state(
-		&self,
-		incoming_pdu: &PduEvent,
-		room_version_rules: &RoomVersionRules,
-		create_event: &PduEvent,
-	) -> Result<bool> {
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Gathering auth events"
-		);
-		let auth_events = self
-			.services
-			.state
-			.get_auth_events(
-				&incoming_pdu.room_id_or_hash(),
-				incoming_pdu.kind(),
-				incoming_pdu.sender(),
-				incoming_pdu.state_key(),
-				incoming_pdu.content(),
-				room_version_rules,
-			)
-			.await?;
-
-		let state_fetch = |k: &StateEventType, s: &str| {
-			let key = k.with_state_key(s);
-			ready(auth_events.get(&key).map(ToOwned::to_owned))
-		};
-
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Running current state auth check"
-		);
-		state_res::event_auth::auth_check(
-			room_version_rules,
-			incoming_pdu,
-			None, // third-party invite
-			state_fetch,
-			create_event.as_pdu(),
-		)
-		.await
-		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))
-	}
-
-	/// Performs PDU check 7 - does the policy server allow this event.
-	///
-	/// If the policy server forbids the event, false is returned. If there is a
-	/// problem contacting the policy server, or it returns an unrecognised
-	/// response, an appropriate error is returned.
-	async fn policy_server_check(
-		&self,
-		incoming_pdu: &PduEvent,
-		pdu_json: &mut CanonicalJsonObject,
-		room_version_rules: &RoomVersionRules,
-	) -> Result<bool> {
-		let event_id = pdu_json
-			.remove("event_id")
-			.expect("event_id should be present in pdu_json at this stage");
-		if let Err(e) = self
-			.policy_server_allows_event(
-				incoming_pdu,
-				pdu_json,
-				&incoming_pdu.room_id_or_hash(),
-				room_version_rules,
-				true,
-			)
-			.await
-			.debug_inspect(|()| {
-				debug!(
-					event_id = %incoming_pdu.event_id,
-					"Event has passed policy server check."
-				);
-			}) {
-			return if matches!(e.kind(), ErrorKind::Forbidden) {
-				info!(
-					event_id = %incoming_pdu.event_id,
-					error = %e,
-					"Event has been marked as spam by policy server: {}",
-					e.message(),
-				);
-				Ok(false)
-			} else {
-				Err(e)
-			};
-		}
-		pdu_json.insert("event_id".to_owned(), event_id);
-		Ok(true)
-	}
-
 	/// Derives new room state from the incoming event and filters forward
 	/// extremities accordingly. Does not set forward extremities.
 	///