tuwunel_service/rooms/state_res/resolve/

iterative_auth_check.rs

use futures::{Stream, StreamExt, TryFutureExt, TryStreamExt};
use ruma::{
	EventId, OwnedEventId, events::TimelineEventType, room_version_rules::RoomVersionRules,
};
use tuwunel_core::{
	Result, debug_warn, err, error,
	matrix::{Event, EventTypeExt, PduEvent, StateKey, TypeStateKey},
	smallvec::SmallVec,
	utils::stream::{IterStream, ReadyExt, TryReadyExt, TryWidebandExt},
};

use super::{
	super::{
		AuthCheckOutcome, FetchEvent, auth_types_for_event, check_state_dependent_auth_rules,
		event_auth::classify_auth_error,
	},
	StateMap,
};

type AuthEvents = SmallVec<[(TypeStateKey, PduEvent); 4]>;

/// Perform the iterative auth checks to the given list of events.
///
/// Definition in the specification:
///
/// The iterative auth checks algorithm takes as input an initial room state and
/// a sorted list of state events, and constructs a new room state by iterating
/// through the event list and applying the state event to the room state if the
/// state event is allowed by the authorization rules. If the state event is not
/// allowed by the authorization rules, then the event is ignored. If a
/// (event_type, state_key) key that is required for checking the authorization
/// rules is not present in the state, then the appropriate state event from the
/// event’s auth_events is used if the auth event is not rejected.
///
/// ## Arguments
///
/// * `rules` - The authorization rules for the current room version.
/// * `events` - The sorted state events to apply to the `partial_state`.
/// * `state` - The current state that was partially resolved for the room.
/// * `fetch_event` - Function to fetch an event in the room given its event ID.
///
/// ## Returns
///
/// Returns the partially resolved state, or an `Err(_)` if one of the state
/// events in the room has an unexpected format.
#[tracing::instrument(
	name = "iterative_auth",
	level = "debug",
	skip_all,
	fields(
		states = ?state.len(),
	)
)]
pub(super) async fn iterative_auth_check<'b, SortedPowerEvents, Fetch>(
	rules: &RoomVersionRules,
	events: SortedPowerEvents,
	state: StateMap<OwnedEventId>,
	fetch: Fetch,
) -> Result<StateMap<OwnedEventId>>
where
	SortedPowerEvents: Stream<Item = &'b EventId> + Send,
	Fetch: FetchEvent,
{
	events
		.map(Ok)
		.wide_and_then(async |event_id| {
			let event: PduEvent = fetch.get(event_id).await?;
			let state_key = event.state_key().map(StateKey::from);

			Ok(state_key.map(|state_key| (event_id, state_key, event)))
		})
		.ready_try_filter_map(Result::<_>::Ok)
		.try_fold(state, |state, (event_id, state_key, event)| {
			auth_check(rules, state, event_id, state_key, event, fetch)
		})
		.await
}

#[tracing::instrument(
	name = "check",
	level = "trace",
	skip_all,
	fields(
		%event_id,
		?state_key,
	)
)]
async fn auth_check<Fetch>(
	rules: &RoomVersionRules,
	mut state: StateMap<OwnedEventId>,
	event_id: &EventId,
	state_key: StateKey,
	event: PduEvent,
	fetch: Fetch,
) -> Result<StateMap<OwnedEventId>>
where
	Fetch: FetchEvent,
{
	let Ok(auth_types) = auth_types_for_event(
		event.event_type(),
		event.sender(),
		Some(&state_key),
		event.content(),
		&rules.authorization,
		true,
	)
	.inspect_err(|e| error!("failed to get auth types for event: {e}")) else {
		return Ok(state);
	};

	let auth_types_events = auth_types
		.stream()
		.ready_filter_map(|key| {
			state
				.get(&key)
				.map(move |auth_event_id| (auth_event_id, key))
		})
		.filter_map(async |(id, key)| {
			fetch_auth_event(id, fetch)
				.await
				.map(|event| event.map(|event| (key, event)))
		})
		.ready_try_filter_map(|(key, auth_event)| {
			Ok(auth_event
				.rejected()
				.eq(&false)
				.then_some((key, auth_event)))
		});

	// If the `m.room.create` event is not in the auth events, we need to add it,
	// because it's always part of the state and required in the auth rules.
	let also_need_create_event = *event.event_type() != TimelineEventType::RoomCreate
		&& rules
			.authorization
			.room_create_event_id_as_room_id;

	let also_create_id: Option<OwnedEventId> = also_need_create_event
		.then(|| event.room_id().as_event_id().ok())
		.flatten();

	let auth_events = event
		.auth_events()
		.chain(also_create_id.as_deref().into_iter())
		.stream()
		.filter_map(async |id| fetch_auth_event(id, fetch).await)
		.ready_try_filter_map(|auth_event| {
			let state_key = auth_event
				.state_key()
				.ok_or_else(|| err!(Request(InvalidParam("Missing state_key"))))?;

			let key_val = auth_event
				.rejected()
				.eq(&false)
				.then_some((auth_event.event_type().with_state_key(state_key), auth_event));

			Ok(key_val)
		});

	let auth_events = auth_events
		.chain(auth_types_events)
		.try_collect()
		.map_ok(|mut vec: AuthEvents| {
			vec.sort_by(|a, b| a.0.cmp(&b.0));
			vec.reverse();
			vec.dedup_by(|a, b| a.0.eq(&b.0));
			vec
		})
		.await?;

	let outcome =
		match check_state_dependent_auth_rules(rules, &event, auth_events.as_slice()).await {
			| Ok(()) => AuthCheckOutcome::Allow,
			| Err(error) => classify_auth_error(error)?,
		};

	match outcome {
		| AuthCheckOutcome::Allow => {
			let key = event.event_type().with_state_key(state_key);

			state.insert(key, event_id.to_owned());
		},
		| AuthCheckOutcome::Deny(error) => {
			debug_warn!(
				%event_id,
				sender = %event.sender(),
				event_type = ?event.event_type(),
				?state_key,
				%error,
				"event failed auth check"
			);
		},
	}

	Ok(state)
}

async fn fetch_auth_event<Fetch>(id: &EventId, fetch: Fetch) -> Option<Result<PduEvent>>
where
	Fetch: FetchEvent,
{
	match fetch.get::<PduEvent>(id).await {
		| Ok(event) => Some(Ok(event)),
		| Err(error) if error.is_not_found() => {
			debug_warn!(%id, %error, "missing auth event");
			None
		},
		| Err(error) => Some(Err(error)),
	}
}