From 7e81918639e157bb488d679de737be1e97291920 Mon Sep 17 00:00:00 2001 From: timedout Date: Wed, 7 Oct 2026 19:58:09 +0100 Subject: [PATCH] wip --- .../fetch_and_handle_outliers.rs | 19 ++++--- src/service/rooms/event_handler/fetch_auth.rs | 2 +- .../rooms/event_handler/policy_server.rs | 6 +-- .../rooms/event_handler/resolve_state.rs | 3 +- .../rooms/event_handler/state_at_incoming.rs | 32 ++++++------ src/service/rooms/membership/mod.rs | 47 +++++++++++------- src/service/rooms/short/mod.rs | 2 +- src/service/rooms/state/mod.rs | 2 +- src/service/rooms/timeline/create.rs | 49 ++++++------------- 9 files changed, 76 insertions(+), 86 deletions(-) diff --git a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs index 24ef24f54..6d2063bac 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs @@ -10,7 +10,6 @@ use conduwuit::error; use conduwuit::{ Err, Event, PduEvent, Result, debug, debug_info, debug_warn, err, result::{FlatOk, LogErr}, - state_res::lexicographical_topological_sort, trace, utils::{IterStream, math::Expected, stream::BroadbandExt}, warn, @@ -23,7 +22,9 @@ use ruma::{ int, room_version_rules::RoomVersionRules, }; - +use ruma::events::room::power_levels::UserPowerLevel; +use ruma::state_res::utils::event_id_map::EventIdMap; +use conduwuit::debug::DebugInspect; use super::get_room_version_rules; use crate::rooms::event_handler::parse_incoming_pdu::expect_event_id_array; @@ -64,8 +65,7 @@ pub async fn build_local_dag( return Ok(pdu_map.keys().cloned().collect()); } - let mut dag: HashMap> = - HashMap::with_capacity(pdu_map.len()); + let mut dag = EventIdMap::with_capacity(pdu_map.len()); let mut id_origin_ts: HashMap = HashMap::with_capacity(pdu_map.len()); for (event_id, value) in pdu_map { @@ -94,17 +94,16 @@ pub async fn build_local_dag( } debug!(count = dag.len(), "Sorting incoming events with partial graph"); - lexicographical_topological_sort(&dag, &async |node_id| { + ruma::state_res::reverse_topological_power_sort(&dag, |node_id| { // Note: we don't bother fetching power levels because that would // massively slow this function down. This is a best-effort attempt to // order events correctly for processing, however ultimately that // should be the sender's job. - let ts = id_origin_ts.get(&node_id).copied().unwrap_or_default(); - Ok((int!(0), MilliSecondsSinceUnixEpoch(ts))) + let ts = id_origin_ts.get(node_id).copied().unwrap_or_default(); + Ok((UserPowerLevel::Infinite, MilliSecondsSinceUnixEpoch(ts))) }) - .await - .inspect(|sorted| { - debug_assert_eq!( + .debug_inspect(|sorted| { + assert_eq!( sorted.len(), pdu_map.len(), "Sorted graph was not the same size as the input graph" diff --git a/src/service/rooms/event_handler/fetch_auth.rs b/src/service/rooms/event_handler/fetch_auth.rs index 540b6b540..1a26bcf81 100644 --- a/src/service/rooms/event_handler/fetch_auth.rs +++ b/src/service/rooms/event_handler/fetch_auth.rs @@ -1,7 +1,7 @@ use std::collections::{HashMap, hash_map}; use conduwuit::{ - Err, Event, EventTypeExt, PduEvent, Result, debug_info, err, utils::TryFutureExtExt, warn, + Err, Event, EventTypeExt, PduEvent, Result, debug_info, err, warn, }; use ruma::{ OwnedEventId, ServerName, api::federation::authorization::get_event_authorization, diff --git a/src/service/rooms/event_handler/policy_server.rs b/src/service/rooms/event_handler/policy_server.rs index 0fb784d2e..87e6afe9e 100644 --- a/src/service/rooms/event_handler/policy_server.rs +++ b/src/service/rooms/event_handler/policy_server.rs @@ -5,10 +5,8 @@ use std::{collections::BTreeMap, time::Duration}; -use conduwuit::{ - Err, Error, Event, PduEvent, Result, debug, debug_error, debug_info, debug_warn, err, error, - info, state_res::EventTypeExt, trace, utils::to_canonical_object, warn, -}; +use conduwuit::{debug, debug_error, debug_info, debug_warn, err, error, + info, trace, utils::to_canonical_object, warn, Err, Error, Event, EventTypeExt, PduEvent, Result}; use http::StatusCode; use ruma::{ CanonicalJsonObject, CanonicalJsonValue, KeyId, RoomId, ServerName, SigningKeyAlgorithm, diff --git a/src/service/rooms/event_handler/resolve_state.rs b/src/service/rooms/event_handler/resolve_state.rs index 50a49a208..b24f35f9f 100644 --- a/src/service/rooms/event_handler/resolve_state.rs +++ b/src/service/rooms/event_handler/resolve_state.rs @@ -6,7 +6,6 @@ use std::{ use conduwuit::{ Error, Result, err, - state_res::{self, StateMap}, trace, utils::stream::{IterStream, ReadyExt, TryWidebandExt, WidebandExt}, warn, @@ -115,7 +114,7 @@ impl super::Service { auth_chain_sets: Vec>, ) -> Result> { let event_fetch = |event_id| self.event_fetch(event_id); - let calc_subgraph = |ss| self.calculate_conflicted_state_subgraph(ss); + let calc_subgraph = |ss| self.calculate_conflicted_state_subgraph(ss).ok(); ruma::state_res::resolve( &room_version_rules.authorization, &room_version_rules diff --git a/src/service/rooms/event_handler/state_at_incoming.rs b/src/service/rooms/event_handler/state_at_incoming.rs index 7025077c7..17473418d 100644 --- a/src/service/rooms/event_handler/state_at_incoming.rs +++ b/src/service/rooms/event_handler/state_at_incoming.rs @@ -6,13 +6,13 @@ use std::{ use conduwuit::{ Result, debug, debug_error, err, error, - matrix::{Event, StateMap}, trace, utils::stream::{BroadbandExt, IterStream, ReadyExt, TryBroadbandExt, TryWidebandExt}, }; use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, future::try_join}; -use ruma::{OwnedEventId, RoomId, room_version_rules::RoomVersionRules}; - +use ruma::{room_version_rules::RoomVersionRules, EventId, OwnedEventId, RoomId}; +use ruma::state_res::{Event, StateMap}; +use ruma::state_res::utils::event_id_set::EventIdSet; use crate::rooms::short::ShortStateHash; #[cfg(test)] @@ -30,7 +30,7 @@ impl super::Service { room_version_rules: &RoomVersionRules, ) -> Result>> where - Pdu: Event + Send + Sync, + Pdu: Event + Send + Sync, { if incoming_pdu.prev_events().count() == 1 { self.state_before_incoming_degree_one(incoming_pdu).await @@ -48,7 +48,7 @@ impl super::Service { incoming_pdu: &Pdu, ) -> Result>> where - Pdu: Event + Send + Sync, + Pdu: Event + Send + Sync, { let prev_event = incoming_pdu .prev_events() @@ -109,8 +109,11 @@ impl super::Service { room_version_rules: &RoomVersionRules, ) -> Result>> where - Pdu: Event + Send + Sync, + Pdu: Event + Send + Sync, { + if incoming_pdu.prev_events().count() == 0 { + return Ok(None); + } trace!("Calculating extremity statehashes..."); let Ok(extremity_sstatehashes) = incoming_pdu .prev_events() @@ -135,8 +138,8 @@ impl super::Service { }; trace!("Calculating fork states..."); - let room_id = &incoming_pdu.room_id_or_hash(); - let (fork_states, auth_chain_sets): (Vec>, Vec>) = + let room_id = incoming_pdu.room_id().unwrap(); + let (fork_states, auth_chain_sets): (Vec>, Vec>) = extremity_sstatehashes .into_iter() .try_stream() @@ -149,9 +152,7 @@ impl super::Service { .await?; let Ok(new_state) = self - .state_resolution(room_version_rules, fork_states.iter(), &auth_chain_sets) - .boxed() - .await + .state_resolution(room_version_rules, fork_states.as_slice(), auth_chain_sets) .inspect_err(|e| error!("State resolution failed: {e:?}")) else { return Ok(None); @@ -182,7 +183,7 @@ impl super::Service { prev_event: Pdu, ) -> Result<(StateMap, HashSet)> where - Pdu: Event, + Pdu: Event, { let mut leaf_state: HashMap<_, _> = self .services @@ -195,7 +196,7 @@ impl super::Service { let shortstatekey = self .services .short - .get_or_create_shortstatekey(&prev_event.kind().to_string().into(), state_key) + .get_or_create_shortstatekey(&prev_event.event_type().to_string().into(), state_key) .await; let event_id = prev_event.event_id(); @@ -216,11 +217,10 @@ impl super::Service { self.services .short .get_statekey_from_short(*k) - .map_ok(|(ty, sk)| ((ty, sk), id.clone())) + .map_ok(|(ty, sk)| ((ty, sk.to_string()), id.clone())) }) .ready_filter_map(Result::ok) - .collect() - .map(Ok); + .collect(); try_join(fork_state, auth_chain).await } diff --git a/src/service/rooms/membership/mod.rs b/src/service/rooms/membership/mod.rs index e7bd6037d..15a9009f7 100644 --- a/src/service/rooms/membership/mod.rs +++ b/src/service/rooms/membership/mod.rs @@ -7,7 +7,7 @@ use conduwuit::{ event::{gen_event_id, gen_event_id_canonical_json}, }, pdu::PartialPdu, - state_res, trace, + trace, utils::{self, IterStream, ReadyExt, to_canonical_object}, warn, }; @@ -25,7 +25,7 @@ use ruma::{ StateEventType, StaticEventContent, room::member::{MembershipState, RoomMemberEventContent}, }, - room::{AllowRule, JoinRule, Restricted, RoomMembership}, + room::{AllowRule, JoinRule, RoomMembership}, }; use crate::{ @@ -591,13 +591,17 @@ impl Service { debug!("Running send_join auth check"); let fetch_state = &state; - let state_fetch = |k: StateEventType, s: StateKey| async move { - let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?; + let state_fetch = |k: StateEventType, s: StateKey| { + let shortstatekey = self + .services + .short + .get_shortstatekey_blocking(&k, &s) + .ok()?; let event_id = fetch_state.get(&shortstatekey)?; - self.services.timeline.get_pdu(event_id).await.ok() + self.services.timeline.get_pdu_blocking(event_id).ok() }; - let Some(create_event) = state_fetch(StateEventType::RoomCreate, "".into()).await else { + if state_fetch(StateEventType::RoomCreate, "".into()).is_none() { let in_auth_chain = send_join_auth_chain .iter() .any(|(t, s)| t == "m.room.create" && s.is_empty()); @@ -614,20 +618,29 @@ impl Service { }; drop(send_join_auth_chain); - let auth_check = state_res::event_auth::auth_check( - &room_version.rules().unwrap(), + // TODO: These calls should be in a thread to avoid blocking the main + // event loop. + ruma::state_res::check_state_independent_auth_rules( + &room_version.rules().unwrap().authorization, + &parsed_join_pdu, + |event_id| self.services.timeline.get_pdu_blocking(event_id).ok(), + ) + .map_err(|e| { + err!(Request(Forbidden( + "Join event returned by {remote_server} is not self-authorised (try again?): {e}" + ))) + })?; + ruma::state_res::check_state_dependent_auth_rules( + &room_version.rules().unwrap().authorization, &parsed_join_pdu, |k, s| state_fetch(k.clone(), s.into()), - &create_event, ) - .await - .map_err(|e| err!(Request(Forbidden(warn!("Auth check failed: {e:?}")))))?; + .map_err(|e| { + err!(Request(Forbidden( + "Join event returned by {remote_server} is not self-authorised (try again?): {e}" + ))) + })?; - if !auth_check { - return Err!(Request(Forbidden( - "Join event returned by {remote_server} is not self-authorised (try again?)" - ))); - } let resident_before = self .services .state_cache @@ -845,7 +858,7 @@ impl Service { &self, room_id: &RoomId, entity: JoinConditionEntity<'_>, - conditions: &Restricted, + conditions: &ruma::room::Restricted, ) -> JoinConditionResult { if conditions.allow.is_empty() { // This will never be authorisable, return forbidden. diff --git a/src/service/rooms/short/mod.rs b/src/service/rooms/short/mod.rs index add982071..b5ea91cbd 100644 --- a/src/service/rooms/short/mod.rs +++ b/src/service/rooms/short/mod.rs @@ -160,7 +160,7 @@ impl Service { state_key: &str, ) -> Result { let key = (event_type, state_key); - self.db.shortstatekey_statekey.qry() + self.db.shortstatekey_statekey.qry_blocking(&key).deserialized() } /// Gets a full event ID from a short event ID. diff --git a/src/service/rooms/state/mod.rs b/src/service/rooms/state/mod.rs index 368a7216c..ca8c6651a 100644 --- a/src/service/rooms/state/mod.rs +++ b/src/service/rooms/state/mod.rs @@ -1,7 +1,7 @@ use std::{collections::HashMap, fmt::Write, sync::Arc}; use async_trait::async_trait; -use conduwuit::{debug, matrix::StateKey, smallstr::SmallString, utils::stream::WidebandExt}; +use conduwuit::{debug, utils::stream::WidebandExt}; use conduwuit_core::{ Event, PduEvent, Result, err, result::FlatOk, diff --git a/src/service/rooms/timeline/create.rs b/src/service/rooms/timeline/create.rs index d53bb14fd..d9cf873da 100644 --- a/src/service/rooms/timeline/create.rs +++ b/src/service/rooms/timeline/create.rs @@ -6,18 +6,11 @@ use conduwuit_core::{ matrix::{ event::{Event, gen_event_id}, pdu::{EventHash, PartialPdu, PduEvent}, - state_res, }, utils::{self, IterStream, ReadyExt, stream::TryIgnore}, }; use futures::{StreamExt, TryStreamExt, future, future::ready}; -use ruma::{ - CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, OwnedRoomId, RoomId, RoomVersionId, - UserId, - events::{StateEventType, TimelineEventType, room::create::RoomCreateEventContent}, - room_version_rules::RoomVersionRules, - uint, -}; +use ruma::{events::{StateEventType, TimelineEventType, room::create::RoomCreateEventContent}, room_version_rules::RoomVersionRules, state_res, uint, CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, OwnedRoomId, RoomId, RoomVersionId, UserId}; use serde_json::value::{RawValue, to_raw_value}; use super::RoomMutexGuard; @@ -142,7 +135,7 @@ impl super::Service { | None => Vec::new(), }; - let auth_events: HashMap<(StateEventType, SmallString<[u8; 48]>), PduEvent> = + let auth_events: HashMap<(StateEventType, String), PduEvent> = match room_id { | Some(room_id) => self.services @@ -231,39 +224,27 @@ impl super::Service { }, hashes: EventHash { sha256: String::new() }, signatures: None, + rejected: false, }; let auth_fetch = |k: &StateEventType, s: &str| { let key = (k.clone(), s.into()); - ready(auth_events.get(&key).map(ToOwned::to_owned)) + auth_events.get(&key).map(ToOwned::to_owned) }; let room_id_or_hash = pdu.room_id_or_hash(); - let create_pdu = match &pdu.kind { - | TimelineEventType::RoomCreate => None, - | _ => Some( - self.services - .state_accessor - .room_state_get(&room_id_or_hash, &StateEventType::RoomCreate, "") - .await - .map_err(|e| { - err!(Request(Forbidden(warn!("Failed to fetch room create event: {e}")))) - })?, - ), - }; - let create_event = match &pdu.kind { - | TimelineEventType::RoomCreate => &pdu, - | _ => create_pdu.as_ref().unwrap().as_pdu(), - }; - let auth_check = - state_res::auth_check(&room_version_rules, &pdu, auth_fetch, create_event) - .await - .map_err(|e| err!(Request(Forbidden(warn!("Auth check failed: {e:?}")))))?; - - if !auth_check { - return Err!(Request(Forbidden("Event is not authorized."))); - } + state_res::check_state_independent_auth_rules( + &room_version_rules.authorization, + &pdu, + |event_id| self.get_pdu_blocking(event_id).ok(), + ) + .map_err(|e| err!(Request(Forbidden("{e}"))))?; + state_res::check_state_dependent_auth_rules( + &room_version_rules.authorization, + &pdu, + auth_fetch + ).map_err(|e| err!(Request(Forbidden("{e}"))))?; trace!( "Event {} in room {} is authorized", pdu.event_id,