From 883b7acb399538574eef17d4e7fad60745ec9fa1 Mon Sep 17 00:00:00 2001 From: Jacob Taylor Date: Sat, 11 Jul 2026 06:34:54 -0700 Subject: [PATCH] fix: Merge nex/fix/pdu-handler-refresh --- src/api/client/membership/invite.rs | 2 +- src/api/server/send.rs | 28 +- src/api/server/send_join.rs | 2 +- src/api/server/send_knock.rs | 2 +- src/api/server/send_leave.rs | 2 +- src/core/error/err.rs | 8 + src/service/rooms/event_handler/acl_check.rs | 1 + .../fetch_and_handle_outliers.rs | 39 +- src/service/rooms/event_handler/fetch_auth.rs | 69 +++- src/service/rooms/event_handler/fetch_prev.rs | 12 +- .../event_handler/handle_incoming_pdu.rs | 159 +++---- .../rooms/event_handler/handle_outlier_pdu.rs | 202 +++------ src/service/rooms/event_handler/mod.rs | 19 +- .../rooms/event_handler/parse_incoming_pdu.rs | 50 +-- src/service/rooms/event_handler/pdu_checks.rs | 289 +++++++++++++ .../rooms/event_handler/state_at_incoming.rs | 39 +- .../event_handler/upgrade_outlier_pdu.rs | 390 ++++-------------- src/service/rooms/membership/mod.rs | 2 +- src/service/rooms/timeline/append.rs | 195 +++++++-- src/service/rooms/timeline/backfill.rs | 4 +- src/service/rooms/timeline/create.rs | 6 +- src/service/rooms/timeline/mod.rs | 3 + 22 files changed, 838 insertions(+), 685 deletions(-) create mode 100644 src/service/rooms/event_handler/pdu_checks.rs diff --git a/src/api/client/membership/invite.rs b/src/api/client/membership/invite.rs index 788f431ae..f4d310ea6 100644 --- a/src/api/client/membership/invite.rs +++ b/src/api/client/membership/invite.rs @@ -221,7 +221,7 @@ pub(crate) async fn invite_helper( let pdu_id = services .rooms .event_handler - .handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, true) + .handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, false) .boxed() .await? .ok_or_else(|| { diff --git a/src/api/server/send.rs b/src/api/server/send.rs index 9f8ab6a8f..cb3a8eeea 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -241,7 +241,7 @@ async fn handle( pdus: impl Stream + Send, edus: impl Stream + Send, ) -> std::result::Result { - // group pdus by room + // Group PDUs by room for parallel processing let pdus = pdus .collect() .map(|mut pdus: Vec<_>| { @@ -252,7 +252,7 @@ async fn handle( }) .await; - // we can evaluate rooms concurrently + // Evaluate rooms in parallel let results: ResolvedMap = pdus .into_iter() .try_stream() @@ -266,7 +266,7 @@ async fn handle( .boxed() .await?; - // evaluate edus after pdus, at least for now. + // Evaluate EDUs after PDUs in case some of the PDUs then forbid some EDUs. edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu)) .boxed() .await; @@ -296,20 +296,12 @@ async fn handle_room( // Try to sort PDUs by their dependencies, but fall back to arbitrary order on // failure (e.g., cycles). This is best-effort; proper ordering is the sender's // responsibility. - let sorted_event_ids = if pdu_map.len() >= 2 { - let refmap = pdu_map - .iter() - .map(|(event_id, obj)| (event_id.clone(), obj)) - .collect(); - build_local_dag(&refmap, DagBuilderTree::PrevEvents) - .await - .unwrap_or_else(|e| { - debug_warn!("Failed to build local DAG for room {room_id}: {e}"); - pdu_map.keys().cloned().collect() - }) - } else { - pdu_map.keys().cloned().collect() - }; + let sorted_event_ids = build_local_dag(&pdu_map, DagBuilderTree::PrevEvents) + .await + .unwrap_or_else(|e| { + debug_warn!("Failed to build local DAG for room {room_id}: {e}"); + pdu_map.keys().cloned().collect() + }); let mut results = Vec::with_capacity(sorted_event_ids.len()); for event_id in sorted_event_ids { let value = pdu_map @@ -322,7 +314,7 @@ async fn handle_room( let result = services .rooms .event_handler - .handle_incoming_pdu(origin, room_id, &event_id, value.clone(), true) + .handle_incoming_pdu(origin, room_id, &event_id, value, false) .boxed() .await .map(|_| ()); diff --git a/src/api/server/send_join.rs b/src/api/server/send_join.rs index 5498d03d3..b72c4c18a 100644 --- a/src/api/server/send_join.rs +++ b/src/api/server/send_join.rs @@ -200,7 +200,7 @@ async fn create_join_event( let pdu_id = services .rooms .event_handler - .handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), true) + .handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), false) .boxed() .await? .ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?; diff --git a/src/api/server/send_knock.rs b/src/api/server/send_knock.rs index 76e8f363f..377a8792d 100644 --- a/src/api/server/send_knock.rs +++ b/src/api/server/send_knock.rs @@ -166,7 +166,7 @@ pub(crate) async fn create_knock_event_v1_route( let pdu_id = services .rooms .event_handler - .handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), true) + .handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), false) .boxed() .await? .ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?; diff --git a/src/api/server/send_leave.rs b/src/api/server/send_leave.rs index 7b1ec9c03..121683224 100644 --- a/src/api/server/send_leave.rs +++ b/src/api/server/send_leave.rs @@ -153,7 +153,7 @@ async fn create_leave_event( let pdu_id = services .rooms .event_handler - .handle_incoming_pdu(origin, room_id, &event_id, value, true) + .handle_incoming_pdu(origin, room_id, &event_id, value, false) .boxed() .await? .ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?; diff --git a/src/core/error/err.rs b/src/core/error/err.rs index 131eea2ef..c1ae95654 100644 --- a/src/core/error/err.rs +++ b/src/core/error/err.rs @@ -149,6 +149,14 @@ macro_rules! err_log { #[macro_export] #[collapse_debuginfo(yes)] macro_rules! err_lev { + (debug_info) => { + if $crate::debug::logging() { + $crate::tracing::Level::INFO + } else { + $crate::tracing::Level::DEBUG + } + }; + (debug_warn) => { if $crate::debug::logging() { $crate::tracing::Level::WARN diff --git a/src/service/rooms/event_handler/acl_check.rs b/src/service/rooms/event_handler/acl_check.rs index c3fcfea1c..5492c9ba1 100644 --- a/src/service/rooms/event_handler/acl_check.rs +++ b/src/service/rooms/event_handler/acl_check.rs @@ -14,6 +14,7 @@ pub async fn acl_check(&self, server_name: &ServerName, room_id: &RoomId) -> Res .await .map(|c: RoomServerAclEventContent| c) else { + trace!("Room has no ACL, allowing"); return Ok(()); }; 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 655b24429..67c5a4a16 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs @@ -1,5 +1,6 @@ use std::{ collections::{HashMap, HashSet, VecDeque}, + fmt::{Display, Formatter}, time::Instant, }; @@ -33,6 +34,20 @@ pub enum DagBuilderTree { AuthEvents, } +impl DagBuilderTree { + #[must_use] + pub fn as_str(&self) -> &str { + match self { + | Self::AuthEvents => "auth_events", + | Self::PrevEvents => "prev_events", + } + } +} + +impl Display for DagBuilderTree { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.write_str(self.as_str()) } +} + /// Attempts to build a localised directed acyclic graph out of the given PDUs, /// returning them in a topologically sorted order. /// @@ -42,23 +57,22 @@ pub enum DagBuilderTree { /// not account for power levels or other tie breaks. #[allow(clippy::implicit_hasher)] pub async fn build_local_dag( - pdu_map: &HashMap, + pdu_map: &HashMap, tree: DagBuilderTree, ) -> Result> { - debug_assert!(pdu_map.len() >= 2, "needless call to build_local_dag with less than 2 PDUs"); + if pdu_map.len() <= 1 { + return Ok(pdu_map.keys().cloned().collect()); + } + let mut dag: HashMap> = HashMap::with_capacity(pdu_map.len()); let mut id_origin_ts: HashMap = HashMap::with_capacity(pdu_map.len()); - let tree = match tree { - | DagBuilderTree::AuthEvents => "auth_events", - | DagBuilderTree::PrevEvents => "prev_events", - }; for (event_id, value) in pdu_map { // Parse all prev events as event IDs - if they are missing, return an error (we // can't sanely continue in this case), otherwise skip invalid prev events. let prev_events = value - .get(tree) + .get(tree.as_str()) .and_then(CanonicalJsonValue::as_array) .ok_or_else(|| err!(Request(BadJson("event JSON for {event_id} is missing {tree}"))))? .iter() @@ -370,9 +384,8 @@ async fn fetch_event_via( .send_federation_request(&remote, get_event::v1::Request::new(event_id.clone())) .await?; - let (calculated_event_id, value) = self - .parse_incoming_pdu_with_known_room(&res.pdu, room_version_rules) - .await?; + let (calculated_event_id, value) = + Self::parse_incoming_pdu_with_known_room(&res.pdu, room_version_rules)?; if calculated_event_id != event_id { Err!(Request(BadJson(warn!( @@ -538,11 +551,7 @@ pub(super) async fn fetch_and_handle_auth_events( } } - let refmap: HashMap = discovered_events - .iter() - .map(|(id, data)| (id.clone(), data)) - .collect(); - let seeded_ordered = build_local_dag(&refmap, DagBuilderTree::AuthEvents) + let seeded_ordered = build_local_dag(&discovered_events, DagBuilderTree::AuthEvents) .await .expect("failed to build local DAG"); let mut pdus = HashMap::with_capacity(seeded_ordered.len()); diff --git a/src/service/rooms/event_handler/fetch_auth.rs b/src/service/rooms/event_handler/fetch_auth.rs index 6ff5d0324..cf4d95d49 100644 --- a/src/service/rooms/event_handler/fetch_auth.rs +++ b/src/service/rooms/event_handler/fetch_auth.rs @@ -1,9 +1,8 @@ use std::collections::{HashMap, hash_map}; -use conduwuit::{Err, Event, EventTypeExt, PduEvent, Result, err}; +use conduwuit::{Err, Event, EventTypeExt, PduEvent, Result, err, warn}; use ruma::{ - CanonicalJsonObject, OwnedEventId, ServerName, - api::federation::authorization::get_event_authorization, + OwnedEventId, ServerName, api::federation::authorization::get_event_authorization, room_version_rules::RoomVersionRules, }; use tokio::join; @@ -50,16 +49,49 @@ pub(super) async fn fetch_and_persist_event_auth( for auth_pdu_json in event_auth.auth_chain { let (auth_event_room_id, auth_event_id, auth_pdu_json) = - self.parse_incoming_pdu(&auth_pdu_json).await?; + match self.parse_incoming_pdu(&auth_pdu_json).await { + | Ok(parsed) => parsed, + | Err(e) => { + warn!(error=?e, "Dropping auth chain event as it could not be parsed"); + continue; + }, + }; + if let Err(e) = Self::pdu_format_check_1( + &auth_pdu_json, + room_version_rules, + create_event.event_id(), + ) { + // drop this PDU + warn!(%auth_event_id, error=?e, "Dropping auth chain event as it violates the room event format"); + continue; + } + let auth_pdu_json = match self + .signature_hash_check_2_3(auth_pdu_json, room_version_rules) + .await + { + | Ok(pdu_json) => pdu_json, + | Err(e) => { + // drop this PDU + warn!( + %auth_event_id, + error=?e, + "Dropping auth chain event as it has an invalid signature" + ); + continue; + }, + }; + + // PDU check 4 is done when we've finished aggregating if auth_event_room_id != incoming_event.room_id_or_hash() { return Err!(Request(Forbidden( - "Auth event {auth_event_id} is in {auth_event_room_id}, not {}.", + "Auth chain event {auth_event_id} is in {auth_event_room_id}, not {}.", incoming_event.room_id_or_hash() ))); } let auth_pdu = PduEvent::from_id_val(&auth_event_id, auth_pdu_json).map_err(|e| { err!(Request(BadJson("Invalid PDU {auth_event_id} in auth chain: {e}"))) })?; + if auth_pdu.state_key().is_none() { return Err!(Request(BadJson( "Invalid PDU {auth_event_id} in auth_chain: not a state event" @@ -97,17 +129,11 @@ async fn authorise_remote_auth_chain( where Pdu: Event + Send + Sync, { - // TODO(nex): build_local_dag's signature needs changing because all callsites - // seem to do this refmap hack. let pdu_objects = auth_chain_map .iter() - .map(|(event_id, pdu)| (event_id, pdu.to_canonical_object())) + .map(|(event_id, pdu)| (event_id.clone(), pdu.to_canonical_object())) .collect::>(); - let refmap: HashMap = pdu_objects - .iter() - .map(|(id, data)| ((*id).to_owned(), data)) - .collect(); - let auth_chain_topo = build_local_dag(&refmap, DagBuilderTree::AuthEvents) + let auth_chain_topo = build_local_dag(&pdu_objects, DagBuilderTree::AuthEvents) .await? .into_iter() .map(|event_id| (event_id.clone(), auth_chain_map.get(&event_id).unwrap().to_owned())) @@ -127,6 +153,12 @@ async fn authorise_remote_auth_chain( continue; } + // IMPORTANT: We can't use the handy dandy `handle_outlier_pdu` function here + // because it may then try to fetch missing auth events, resulting in deep + // recursion. We will do the minimum required steps to validate the PDU here. + // Checks 1-3 were already done before this function is called, so we only need + // to do check 4. + let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(pdu.auth_events.len()); @@ -150,24 +182,21 @@ async fn authorise_remote_auth_chain( }, | hash_map::Entry::Occupied(_) => { // Duplicate auth events by key are not allowed. - self.services - .pdu_metadata - .mark_event_rejected(auth_event_id); - self.services.pdu_metadata.mark_event_rejected(&event_id); + self.reject_and_persist(&event_id, &pdu.to_canonical_object()); continue 'outer; }, } } 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); + self.reject_and_persist(&event_id, &pdu.to_canonical_object()); } } diff --git a/src/service/rooms/event_handler/fetch_prev.rs b/src/service/rooms/event_handler/fetch_prev.rs index 6e4f8559a..cf85f10f4 100644 --- a/src/service/rooms/event_handler/fetch_prev.rs +++ b/src/service/rooms/event_handler/fetch_prev.rs @@ -5,13 +5,13 @@ utils::{BoolExt, IterStream, stream::BroadbandExt}, }; use futures::StreamExt; -use ruma::{CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName}; +use ruma::{MilliSecondsSinceUnixEpoch, RoomId, ServerName}; use crate::rooms::event_handler::{build_local_dag, fetch_and_handle_outliers::DagBuilderTree}; impl super::Service { /// Fetches any missing prev_events for this event and persists them before - /// returning. + /// returning. The caller is responsible for then handling the incoming PDU. pub(super) async fn fetch_prevs( &self, room_id: &RoomId, @@ -84,13 +84,7 @@ pub(super) async fn fetch_prevs( }) .collect::>(); - let to_persist = if mapped.len() <= 1 { - mapped.keys().map(ToOwned::to_owned).collect() - } else { - let refmap: HashMap = - mapped.iter().map(|(id, data)| (id.clone(), data)).collect(); - build_local_dag(&refmap, DagBuilderTree::PrevEvents).await? - }; + let to_persist = build_local_dag(&mapped, DagBuilderTree::PrevEvents).await?; let job_start = Instant::now(); trace!("Starting to persist {} prev events", to_persist.len()); diff --git a/src/service/rooms/event_handler/handle_incoming_pdu.rs b/src/service/rooms/event_handler/handle_incoming_pdu.rs index b5343d825..aa4ce5d24 100644 --- a/src/service/rooms/event_handler/handle_incoming_pdu.rs +++ b/src/service/rooms/event_handler/handle_incoming_pdu.rs @@ -4,46 +4,37 @@ }; use conduwuit::{ - Err, Event, Result, debug, debug_error, debug_warn, defer, err, error, info, - matrix::PartialPdu, result::DebugInspect, trace, utils::time::jitter, warn, + Err, Event, Result, debug, debug_error, debug_warn, defer, matrix::PartialPdu, trace, + utils::time::jitter, }; -use futures::{ - FutureExt, StreamExt, - future::{OptionFuture, try_join4}, -}; -use ruma::{CanonicalJsonValue, EventId, OwnedUserId, RoomId, ServerName, UserId}; +use futures::{FutureExt, StreamExt, future::try_join3}; +use ruma::{CanonicalJsonValue, EventId, RoomId, ServerName, UserId}; use tokio::sync::mpsc; -use crate::rooms::timeline::{RawPduId, pdu_fits}; +use crate::rooms::timeline::RawPduId; impl super::Service { - /// When receiving an event one needs to: - /// 0. Check the server is in the room - /// 1. Skip the PDU if we already know about it - /// 1.1. Remove unsigned field - /// 2. Check signatures, otherwise drop - /// 3. Check content hash, redact if doesn't match - /// 4. Fetch any missing auth events doing all checks listed here starting - /// at 1. These are not timeline events - /// 5. Reject "due to auth events" if can't get all the auth events or some - /// of the auth events are also rejected "due to auth events" - /// 6. Reject "due to auth events" if the event doesn't pass auth based on - /// the auth events - /// 7. Persist this event as an outlier - /// 8. If not timeline event: stop - /// 9. Fetch any missing prev events doing all checks listed here starting - /// at 1. These are timeline events - /// 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. These are not timeline events - /// 11. Check the auth of the event passes based on the state of the event - /// 12. Ensure that the state is derived from the previous current state - /// (i.e. we calculated by doing state res where one of the inputs was a - /// previously trusted set of state, don't just trust a set of state we - /// got from a remote) - /// 13. Use state resolution to find new room state - /// 14. Check if the event passes auth based on the "current state" of the - /// room, if not soft fail it + /// Handles an incoming PDU from federation. + /// + /// First checks that we want to receive this PDU. If we already have it as + /// a timeline PDU, or we don't want to receive the PDU (e.g. origin ACL'd, + /// room disabled/unknown), abort. + /// + /// The PDU is then handled as an outlier event, which performs [PDU checks] + /// 1 through 4. See: `handle_outlier_pdu`. + /// + /// Once handled as an outlier, any missing prev events are fetched, and + /// then the PDU will be promoted/upgraded from an outlier to a timeline + /// event clients can see. See: `upgrade_outlier_to_timeline_pdu`. After + /// this finishes, the PDU is either accepted or left as an outlier. + /// + /// If the PDU is successfully upgraded, the remaining extremity count of + /// the room is checked. If there are a potentially problematic number of + /// forward extremities, a squasher task is started with a debounce period, + /// which will eventually send a dummy event that ties up as many DAG forks + /// as possible. + /// + /// [PDU checks]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu #[tracing::instrument( name = "pdu", skip_all, @@ -55,51 +46,33 @@ pub async fn handle_incoming_pdu<'a>( room_id: &'a RoomId, event_id: &'a EventId, value: BTreeMap, - is_timeline_event: bool, + is_backfilled_event: bool, ) -> Result> { - // 1. Skip the PDU if we already have it as a timeline event + // Skip the PDU if we already have it as a timeline event. We still re-process + // outliers in this scenario. if let Ok(pdu_id) = self.services.timeline.get_pdu_id(event_id).await { + debug!("Database hit for incoming PDU, skipping processing"); return Ok(Some(pdu_id)); } - if !pdu_fits(&mut value.clone()) { - warn!( - "dropping incoming PDU {event_id} in room {room_id} from {origin} because it \ - exceeds 65535 bytes or is otherwise too large." - ); - return Err!(Request(TooLarge("PDU is too large"))); - } trace!( "processing incoming PDU from {origin} for room {room_id} with event id {event_id}" ); - // 1.1 Check we even know about the room - let meta_exists = self.services.metadata.exists(room_id).map(Ok); + // If this is a membership state event for a local user, we will be interested + // in this event even if we aren't in the room. This allows us to process things + // like revoking invites over federation, without being in the room. + let is_interesting_member_event = value.get("type").and_then(|t| t.as_str()) + == Some("m.room.member") + && value + .get("state_key") + .and_then(|s| s.as_str()) + .and_then(|s| UserId::parse(s).ok()) + .is_some_and(|u| self.services.globals.user_is_local(&u)); - // 1.2 Check if the room is disabled - let is_disabled = self.services.metadata.is_disabled(room_id).map(Ok); - - // 1.3.1 Check room ACL on origin field/server - let origin_acl_check = self.acl_check(origin, room_id); - - // 1.3.2 Check room ACL on sender's server name - let sender: OwnedUserId = value - .get("sender") - .and_then(|v| v.as_str()) - .ok_or_else(|| err!("No sender in object")) - .and_then(|v| Ok(UserId::parse(v)?)) - .map_err(|e| err!(Request(BadJson("PDU does not have a valid sender key: {e}"))))?; - - let sender_acl_check: OptionFuture<_> = sender - .server_name() - .ne(origin) - .then(|| self.acl_check(sender.server_name(), room_id)) - .into(); - - let (meta_exists, is_disabled, (), ()) = try_join4( - meta_exists, - is_disabled, - origin_acl_check, - sender_acl_check.map(|o| o.unwrap_or(Ok(()))), + let (room_exists, is_disabled, ()) = try_join3( + self.services.metadata.exists(room_id).map(Ok), + self.services.metadata.is_disabled(room_id).map(Ok), + self.acl_check(origin, room_id), ) .await .inspect_err( @@ -112,21 +85,11 @@ pub async fn handle_incoming_pdu<'a>( ))); } - if !self - .services - .state_cache - .server_in_room(self.services.globals.server_name(), room_id) - .await - { - info!( - %origin, - %room_id, - "Dropping inbound PDU for room we aren't participating in" - ); - return Err!(Request(NotFound("This server is not participating in that room."))); - } - - if !meta_exists { + if !room_exists { + if is_interesting_member_event { + // TODO: handle interesting membership events where we aren't in + // the room + } return Err!(Request(NotFound("Room is unknown to this server"))); } @@ -152,8 +115,10 @@ pub async fn handle_incoming_pdu<'a>( .handle_outlier_pdu(origin, create_event, event_id, room_id, value) .await?; - // 8. if not timeline event: stop - if !is_timeline_event { + // If this event is being processed as part of backfill, we don't want to end up + // *appending* it during the upgrade process, so we return early. + if is_backfilled_event { + debug!("Not promoting incoming event as it is being backfilled"); return Ok(None); } @@ -166,11 +131,16 @@ pub async fn handle_incoming_pdu<'a>( .await? .origin_server_ts(); if incoming_pdu.origin_server_ts() < first_ts_in_room { + debug_warn!( + "Not promoting incoming event as it is sent before we joined the room (but was \ + not backfilled)" + ); return Ok(None); } - // 9. Fetch any missing prev events doing all checks listed here starting at 1. - // These are timeline events + // Fetch any missing prev events doing all checks listed here starting at 1. + // These are timeline events. + // TODO: This part needs to be done in a background queue somewhere. debug!("Fetching and persisting any missing prev events"); Box::pin(self.fetch_prevs( @@ -181,13 +151,14 @@ pub async fn handle_incoming_pdu<'a>( first_ts_in_room, )) .await - .debug_inspect_err(|e| { - error!("Failed to fetch and persist incoming event's prev_events: {e:?}"); + .inspect_err(|e| { + debug_error!("Failed to fetch and persist incoming event's prev_events: {e:?}"); })?; - let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event"; + let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event" + && incoming_pdu.state_key().is_none(); - // Done with prev events, now handling the incoming event + // Done with prev events, now we can handle promoting the PDU let pdu_id = Box::pin(self.upgrade_outlier_to_timeline_pdu( incoming_pdu, val, diff --git a/src/service/rooms/event_handler/handle_outlier_pdu.rs b/src/service/rooms/event_handler/handle_outlier_pdu.rs index 72e7b0629..72285406d 100644 --- a/src/service/rooms/event_handler/handle_outlier_pdu.rs +++ b/src/service/rooms/event_handler/handle_outlier_pdu.rs @@ -1,23 +1,22 @@ 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, -}; -use futures::future::ready; -use ruma::{ - CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName, - canonical_json::redact, events::StateEventType, room_version_rules::RoomVersionRules, + Err, Event, EventTypeExt, PduEvent, Result, debug, debug_warn, err, info, trace, }; +use ruma::{CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName}; -use super::{check_room_id, get_room_version_rules}; -use crate::rooms::timeline::pdu_fits; +use super::get_room_version_rules; impl super::Service { /// Handles a PDU as an outlier, performing basic checks like signatures and /// hashes, proclaimed event auth, and then adding it to the outlier tree. + /// + /// This performs steps 1 through 4 of [S-S section 5.1][spec], returning + /// the parsed PDU and modified JSON object. + /// + /// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu #[allow(clippy::too_many_arguments)] - #[tracing::instrument(name="handle_outlier", skip_all, fields(%event_id))] + #[tracing::instrument(name = "handle_outlier", skip_all)] pub(super) async fn handle_outlier_pdu<'a, Pdu>( &self, origin: &'a ServerName, @@ -29,61 +28,39 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>( where Pdu: Event + Send + Sync, { - if !pdu_fits(&mut value.clone()) { - warn!( - "dropping incoming PDU {event_id} in room {room_id} from {origin} because it \ - exceeds 65535 bytes or is otherwise too large." + // Skip outlier handling if we already have this event as either a timeline or + // outlier PDU. + if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await { + debug!( + "Database hit for {event_id} (event is either an outlier or already promoted), \ + skipping outlier handling" ); - return Err!(Request(TooLarge("PDU is too large"))); + value.insert( + "event_id".to_owned(), + CanonicalJsonValue::String(event_id.as_str().to_owned()), + ); + return Ok((pdu_event, value)); } - // 1. Remove unsigned field - value.remove("unsigned"); - // 2. Check signatures, otherwise drop - // 3. check content hash, redact if doesn't match + // 1. Check that the PDU follows the format for the room version + // (in this case, just size check) let room_version_rules = get_room_version_rules(create_event)?; - let mut incoming_pdu = match self - .services - .server_keys - .verify_event(&value, &room_version_rules) - .await - { - | Ok(ruma::signatures::Verified::All) => { - if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await { - debug!( - "Already have event {event_id} as an outlier or timeline event, not \ - re-processing" - ); - value.insert( - "event_id".to_owned(), - CanonicalJsonValue::String(event_id.as_str().to_owned()), - ); - check_room_id(room_id, &pdu_event)?; - return Ok((pdu_event, value)); - } - value - }, - | Ok(ruma::signatures::Verified::Signatures) => { - if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await { - debug!( - "Received a redacted copy of {event_id}, but we already knew about it. \ - Re-using known content instead." - ); - check_room_id(room_id, &pdu_event)?; - let obj = pdu_event.to_canonical_object(); - return Ok((pdu_event, obj)); - } + Self::pdu_format_check_1(&value, &room_version_rules, create_event.event_id()) + .inspect_err(|e| { + info!( + err=?e, + "Dropping incoming PDU from {origin} because it violates the room event format" + ); + })?; - debug_info!("Calculated hash does not match (redaction): {event_id}"); - redact(value, &room_version_rules.redaction, None) - .map_err(|e| err!(Request(BadJson("Failed to redact {event_id}: {e}"))))? - }, - | Err(e) => { - return Err!(Request(Forbidden(debug_error!( - "Signature verification failed for {event_id}: {e}" - )))); - }, - }; + value.remove("unsigned"); + let room_version_rules = get_room_version_rules(create_event)?; + + // 2. Check signatures, otherwise drop. + // 3. Check content hash, redacting the event if it fails. + let mut incoming_pdu = self + .signature_hash_check_2_3(value, &room_version_rules) + .await?; // Now that we have checked the signature and hashes we can add the eventID and // convert to our PduEvent type @@ -91,14 +68,11 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>( "event_id".to_owned(), CanonicalJsonValue::String(event_id.as_str().to_owned()), ); - let pdu_event = serde_json::from_value::( serde_json::to_value(&incoming_pdu).expect("CanonicalJsonObj is a valid JsonValue"), ) .map_err(|e| err!(Request(BadJson(debug_warn!("Event is not a valid PDU: {e}")))))?; - check_room_id(room_id, &pdu_event)?; - // TODO(nex): From hereon the event is not dropped, and thus always added as an // outlier. However, we only do that at the end of this function, which means // several duplicated calls to add_pdu_outlier. Shouldn't we just do it here @@ -116,7 +90,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>( for auth_event_id in pdu_event.auth_events() { if let Ok(auth_event) = self.services.timeline.get_pdu(auth_event_id).await { - check_room_id(room_id, &auth_event)?; trace!("Found auth event {auth_event_id} for outlier event {event_id} locally"); auth_events.insert(auth_event_id.to_owned(), auth_event); } else { @@ -154,7 +127,7 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>( } // Ensure none of the auth events are rejected - if they are, reject too. - for auth_event_id in auth_events.keys() { + for (auth_event_id, auth_event) in &auth_events { if self .services .pdu_metadata @@ -166,15 +139,27 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>( {auth_event_id}", event_id, ); - self.services.pdu_metadata.mark_event_rejected(event_id); + self.reject_and_persist(event_id, &incoming_pdu); return Err!(Request(Forbidden( "Event has rejected auth event: {auth_event_id}" ))); } + if auth_event.room_id_or_hash() != room_id { + debug_warn!( + %auth_event_id, + auth_event_room_id=%auth_event.room_id_or_hash(), + expected_room_id=%room_id, + "Rejecting incoming event which depends on an auth event in another room.", + ); + self.reject_and_persist(event_id, &incoming_pdu); + return Err!(Request(Forbidden( + "Event depends on a cross-room auth event: {auth_event_id}" + ))); + } } - // 6. Reject "due to auth events" if the event doesn't pass auth based on the - // auth events + // 4. Reject "due to auth events" if the event doesn't pass auth based on the + // claimed auth events debug!("Checking based on auth events"); let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(auth_events.len()); // Build map of auth events @@ -184,8 +169,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>( .expect("we just checked that we have all auth events") .to_owned(); - check_room_id(room_id, &auth_event)?; - let key = auth_event.kind().with_state_key( auth_event .state_key @@ -197,98 +180,45 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>( v.insert(auth_event); }, | hash_map::Entry::Occupied(_) => { - self.services - .outlier - .add_pdu_outlier(pdu_event.event_id(), &incoming_pdu); - self.services.pdu_metadata.mark_event_rejected(event_id); - return Err!(Request(Forbidden( + self.reject_and_persist(event_id, &incoming_pdu); + return Err!(Request(Forbidden(debug_warn!( "Auth event's type and state_key combination exists multiple times: {}, \ {}", auth_event.kind, auth_event.state_key().unwrap_or("") - ))); + )))); }, } } 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.services.pdu_metadata.mark_event_rejected(event_id); - self.services - .outlier - .add_pdu_outlier(pdu_event.event_id(), &incoming_pdu); - return Err!(Request(Forbidden( + self.reject_and_persist(event_id, &incoming_pdu); + return Err!(Request(Forbidden(debug_warn!( "Event authorisation fails based on event's claimed auth events" - ))); + )))); } - trace!("Validation successful."); - // 7. Persist the event as an outlier. self.services .outlier .add_pdu_outlier(pdu_event.event_id(), &incoming_pdu); - trace!("Added pdu as outlier."); + debug!("PDU passed checks and has been persisted as an outlier"); 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 { - 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); + self.services.outlier.add_pdu_outlier(event_id, pdu); } } diff --git a/src/service/rooms/event_handler/mod.rs b/src/service/rooms/event_handler/mod.rs index 9275d1c89..3788a961b 100644 --- a/src/service/rooms/event_handler/mod.rs +++ b/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; @@ -19,7 +20,7 @@ DagBuilderTree, GET_MISSING_EVENTS_MAX_BATCH_SIZE, build_local_dag, }; use ruma::{ - OwnedEventId, OwnedRoomId, RoomId, events::room::create::RoomCreateEventContent, + OwnedEventId, OwnedRoomId, events::room::create::RoomCreateEventContent, room_version_rules::RoomVersionRules, }; use tokio::sync::{Notify, mpsc}; @@ -114,22 +115,6 @@ async fn event_fetch(&self, event_id: OwnedEventId) -> Option { } } -fn check_room_id(room_id: &RoomId, pdu: &Pdu) -> Result { - if pdu - .room_id() - .is_some_and(|claimed_room_id| claimed_room_id != room_id) - { - return Err!(Request(InvalidParam(error!( - pdu_event_id = %pdu.event_id(), - pdu_room_id = pdu.room_id().map(tracing::field::display), - %room_id, - "Found event from room in room", - )))); - } - - Ok(()) -} - fn get_room_version_rules(create_event: &Pdu) -> Result { let content: RoomCreateEventContent = create_event.get_content()?; let Some(room_version_rules) = content.room_version.rules() else { diff --git a/src/service/rooms/event_handler/parse_incoming_pdu.rs b/src/service/rooms/event_handler/parse_incoming_pdu.rs index 067851515..66122bf96 100644 --- a/src/service/rooms/event_handler/parse_incoming_pdu.rs +++ b/src/service/rooms/event_handler/parse_incoming_pdu.rs @@ -23,7 +23,12 @@ fn extract_room_id(event_type: &str, pdu: &CanonicalJsonObject) -> Result Result Result { - // Since v3: - // `event_id` should not be present on the PDU. - // NOTE: The above is ignored since technically it's still allowed to be - // included, but should be ignored instead. - // `auth_events` and `prev_events` must be an array of event IDs - let auth_events = expect_event_id_array(pdu, "auth_events")?; - if auth_events.len() > 10 { - return Err!(Request(BadJson("PDU has too many auth events"))); - } - let prev_events = expect_event_id_array(pdu, "prev_events")?; - if prev_events.len() > 20 { - return Err!(Request(BadJson("PDU has too many prev events"))); - } - Ok(()) - } - - pub async fn parse_incoming_pdu_with_known_room( - &self, + /// Parses an incoming PDU JSON object, generating an event ID for it. Does + /// not insert the event ID into the returned object. Does not discover the + /// room ID. + pub(super) fn parse_incoming_pdu_with_known_room( pdu: &RawJsonValue, room_version_rules: &RoomVersionRules, ) -> Result<(OwnedEventId, CanonicalJsonObject)> { @@ -114,10 +93,13 @@ pub async fn parse_incoming_pdu_with_known_room( gen_event_id_canonical_json(pdu, room_version_rules).map_err(|e| { err!(Request(InvalidParam("Could not convert event to canonical json: {e}"))) })?; - self.validate_pdu(&value)?; + // NOTE: validation checks are now performed by `pdu_format_check_1`. Ok((event_id, value)) } + /// Parses an incoming PDU JSON object, generating an event ID for it and + /// attempts to discover the associated room ID. Does not insert the event + /// ID into the returned object. pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result { let value = serde_json::from_str::(pdu.get()).map_err(|e| { err!(BadServerResponse(debug_warn!("Error parsing incoming event {e:?}"))) @@ -142,7 +124,7 @@ pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result { gen_event_id_canonical_json(pdu, &room_version_rules).map_err(|e| { err!(Request(InvalidParam("Could not convert event to canonical json: {e}"))) })?; - self.validate_pdu(&value)?; + // NOTE: validation checks are now performed by `pdu_format_check_1`. Ok((room_id, event_id, value)) } } diff --git a/src/service/rooms/event_handler/pdu_checks.rs b/src/service/rooms/event_handler/pdu_checks.rs new file mode 100644 index 000000000..1288b97a9 --- /dev/null +++ b/src/service/rooms/event_handler/pdu_checks.rs @@ -0,0 +1,289 @@ +use std::collections::HashMap; + +use conduwuit::{ + Err, Event, EventTypeExt, PduEvent, Result, debug, debug::DebugInspect, debug_error, + debug_info, err, info, matrix::StateKey, state_res, trace, +}; +use futures::future::ready; +use ruma::{ + CanonicalJsonObject, EventId, OwnedEventId, ServerName, api::error::ErrorKind, + canonical_json::redact, events::StateEventType, room_version_rules::RoomVersionRules, +}; + +use crate::rooms::{ + event_handler::parse_incoming_pdu::expect_event_id_array, timeline::pdu_fits, +}; + +impl super::Service { + /// Checks that the PDU conforms to the PDU format (check 1). This is + /// already mostly done during deserialisation, so this function just checks + /// that the PDU isn't a too large. + pub(super) fn pdu_format_check_1( + pdu_json: &CanonicalJsonObject, + room_version_rules: &RoomVersionRules, + create_event_id: &EventId, + ) -> Result<()> { + let event_format = &room_version_rules.event_format; + // NOTE: if we do any more validation outside of deserialisation, it has to be + // done here. + + if !pdu_fits(pdu_json) { + return Err!(Request(TooLarge("PDU is too large"))); + } + + if event_format.require_room_create_room_id { + if pdu_json.get("room_id").is_none() { + return Err!(Request(BadJson("Missing required PDU field: `room_id`"))); + } + } + + let auth_events = expect_event_id_array(pdu_json, "auth_events")?; + if auth_events.len() > 10 { + return Err!(Request(BadJson("PDU has too many auth events"))); + } + + let create_event_in_auth_events = auth_events.iter().any(|id| id == create_event_id); + if !event_format.allow_room_create_in_auth_events && create_event_in_auth_events { + return Err!(Request(BadJson("PDU references a create event"))); + } + + let prev_events = expect_event_id_array(pdu_json, "prev_events")?; + if prev_events.len() > 20 { + return Err!(Request(BadJson("PDU has too many prev events"))); + } + + Ok(()) + } + + /// Checks that the PDU has a valid signature (check 2), and redacts it if + /// the content hash verification fails (check 3), returning the + /// potentially modified JSON. Returns an error if the PDU cannot be + /// redacted, or fails signature verification. + pub(super) async fn signature_hash_check_2_3( + &self, + pdu_json: CanonicalJsonObject, + room_version_rules: &RoomVersionRules, + ) -> Result { + match self + .services + .server_keys + .verify_event(&pdu_json, room_version_rules) + .await + { + | Ok(ruma::signatures::Verified::All) => { + trace!("Signatures and hashes verified successfully"); + Ok(pdu_json) + }, + | Ok(ruma::signatures::Verified::Signatures) => { + debug_info!("Content hash mismatch, redacting event and continuing"); + let redacted = redact(pdu_json, &room_version_rules.redaction, None) + .map_err(|e| err!(Request(BadJson("Unable to redact event: {e}"))))?; + Ok(redacted) + }, + | Err(e) => { + Err!(Request(Forbidden(debug_error!("Signature verification failed: {e}")))) + }, + } + } + + /// 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 { + 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)> { + 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 = self + .state_before_incoming(&incoming_pdu, 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 + .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 { + 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 { + 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) + } +} diff --git a/src/service/rooms/event_handler/state_at_incoming.rs b/src/service/rooms/event_handler/state_at_incoming.rs index 08b4357ed..e232df7bc 100644 --- a/src/service/rooms/event_handler/state_at_incoming.rs +++ b/src/service/rooms/event_handler/state_at_incoming.rs @@ -16,10 +16,31 @@ use crate::rooms::short::ShortStateHash; impl super::Service { - // TODO: if we know the prev_events of the incoming event we can avoid the - // request and build the state from a known point and resolve if > 1 prev_event + /// Resolves the state before the incoming event. + /// + /// If we do not know enough information to resolve the state, `Ok(None)` is + /// returned, and the caller will have to figure it out some other way (e.g. + /// by fetching the state from a remote server). + pub(super) async fn state_before_incoming( + &self, + incoming_pdu: &Pdu, + room_version_rules: &RoomVersionRules, + ) -> Result>> + where + Pdu: Event + Send + Sync, + { + if incoming_pdu.prev_events().count() == 1 { + self.state_before_incoming_degree_one(incoming_pdu).await + } else { + self.state_before_incoming_resolved(incoming_pdu, room_version_rules) + .await + } + } + + /// Determines the state before the incoming pdu, when it has only one prev + /// event. This is a special case that does not require state resolution. #[tracing::instrument(name = "state", level = "debug", skip_all)] - pub(super) async fn state_at_incoming_degree_one( + async fn state_before_incoming_degree_one( &self, incoming_pdu: &Pdu, ) -> Result>> @@ -66,7 +87,7 @@ pub(super) async fn state_at_incoming_degree_one( .await; state.insert(shortstatekey, prev_event.to_owned()); - // Now it's the state after the pdu + // Now it's the state at the pdu } debug_assert!(!state.is_empty(), "should be returning None for empty HashMap result"); @@ -74,11 +95,14 @@ pub(super) async fn state_at_incoming_degree_one( Ok(Some(state)) } + /// Resolves the state before the incoming pdu across all of its prev + /// events. If we do not know enough information to resolve the state, + /// `Ok(None)` is returned, and the caller will have to figure it out some + /// other way (e.g. by fetching the state from a remote server). #[tracing::instrument(name = "state", level = "debug", skip_all)] - pub(super) async fn state_at_incoming_resolved( + async fn state_before_incoming_resolved( &self, incoming_pdu: &Pdu, - room_id: &RoomId, room_version_rules: &RoomVersionRules, ) -> Result>> where @@ -108,6 +132,7 @@ pub(super) async fn state_at_incoming_resolved( }; trace!("Calculating fork states..."); + let room_id = &incoming_pdu.room_id_or_hash(); let (fork_states, auth_chain_sets): (Vec>, Vec>) = extremity_sstatehashes .into_iter() @@ -145,6 +170,8 @@ pub(super) async fn state_at_incoming_resolved( .await } + /// Determines the state at an incoming fork (aka the state at a prev + /// event), returning the resolved state and its associated auth chain. async fn state_at_incoming_fork( &self, room_id: &RoomId, diff --git a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs index c6099ccf8..09aa9e41e 100644 --- a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs +++ b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs @@ -1,27 +1,15 @@ -use std::{borrow::Borrow, sync::Arc, time::Instant}; +use std::time::Instant; use conduwuit::{ - Err, Result, debug, debug_error, debug_info, err, info, is_equal_to, - matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res}, - result::DebugInspect, + Err, Result, debug, debug_info, debug_warn, is_true, + matrix::{Event, PduEvent}, trace, - utils::{ - IterStream, - stream::{BroadbandExt, ReadyExt}, - }, - warn, -}; -use futures::{FutureExt, StreamExt, future::ready}; -use ruma::{ - CanonicalJsonObject, RoomId, ServerName, api::error::ErrorKind, events::StateEventType, }; +use ruma::{CanonicalJsonObject, RoomId, ServerName, events::StateEventType}; use tokio::join; use super::get_room_version_rules; -use crate::rooms::{ - state_compressor::{CompressedState, HashSetCompressStateEvent}, - timeline::RawPduId, -}; +use crate::rooms::timeline::RawPduId; impl super::Service { #[tracing::instrument(name="upgrade_outlier", skip_all, fields(event_id=%incoming_pdu.event_id()))] @@ -46,16 +34,24 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu( trace!(event_id=%incoming_pdu.event_id(), "Skipping upgrade of already upgraded PDU"); return Ok(Some(id)); } else if rejected { - return Err!(Request(Forbidden("Event has been rejected"))); + return Err!(Request(Forbidden(debug_info!("Event has been rejected")))); } else if soft_failed { - return Err!(Request(Forbidden("Event has been soft-failed"))); + // Soft-failed events cannot be promoted. + return Err!(Request(Forbidden(debug_info!("Event has been soft-failed")))); } + // These should never happen, but they're good last-minute sanity checks to + // ensure we never promote totally illegal events. assert_eq!( *create_event.kind(), StateEventType::RoomCreate.into(), "tried to upgrade a PDU with a create_event that is not a room create event" ); + assert_eq!( + incoming_pdu.room_id_or_hash(), + *room_id, + "room ID mismatch: PDU room ID differs from parameter" + ); debug!( event_id = %incoming_pdu.event_id, @@ -65,326 +61,118 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu( let min_depth = self.services.metadata.get_mindepth(room_id).await; let room_version_rules = get_room_version_rules(create_event)?; - // 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. - // These are not timeline events. - - debug!( - event_id = %incoming_pdu.event_id, - "Resolving state at event" - ); - let state_at_incoming_event = 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_at_incoming_event = match state_at_incoming_event { - | 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_at_incoming_event.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 at event"))); - } - trace!(state_events = state_at_incoming_event.len(), "Calculated incoming state"); - - debug!( - event_id = %incoming_pdu.event_id, - "Performing auth check to upgrade" - ); - // 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 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 initial 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:?}"))))?; - - if !auth_check { - self.services - .pdu_metadata - .mark_event_rejected(incoming_pdu.event_id()); - return Err!(Request(Forbidden( - "Event authorisation fails based on the state before the event" - ))); - } - - debug!( - event_id = %incoming_pdu.event_id, - "Gathering auth events" - ); - let auth_events = self - .services - .state - .get_auth_events( - room_id, - incoming_pdu.kind(), - incoming_pdu.sender(), - incoming_pdu.state_key(), - incoming_pdu.content(), - &room_version_rules, - ) + // We now need to resolve the state before the event so that we can perform PDU + // check 5 (event auth passes based on state before the event). To do this, we + // 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 + .state_before_check_5(&incoming_pdu, &room_version_rules, create_event, origin) .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 auth check with claimed state auth" - ); - // PDU check: 6 - let 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:?}"))))?; - if !auth_check { - warn!( - event_id = %incoming_pdu.event_id, - "Event authorization fails based on the current state of the room" - ); + if !passes_state_before { + self.reject_and_persist(incoming_pdu.event_id(), &val); + return Err!(Request(Forbidden(debug_warn!( + "Event authorisation fails based on the state before the event" + )))); } - // Soft fail check before doing state res - debug!( - event_id = %incoming_pdu.event_id, - "Performing soft-fail check" - ); - let mut soft_fail = match (auth_check, incoming_pdu.redacts_id(&room_version_rules)) { - | (false, _) => true, - | (true, None) => false, - | (true, Some(redact_id)) => { - if !self - .services - .state_accessor - .user_can_redact(&redact_id, incoming_pdu.sender(), room_id, true) - .await? - { - warn!(redacts = %redact_id, "User is not allowed to redact event"); - true - } else { - false - } - }, - }; + // Now that we know the event passes both self-authentication, and + // authentication based on the state before the event, we need to check that it + // passes based on the *current* room state (state across all forward + // extremities). If it doesn't, we accept it, but soft-fail it, and this + // prevents it being promoted. - // 13. Use state resolution to find new room state - // We start looking at current room state now, so lets lock the room + // We lock the room here to prevent the current state from changing beneath us + // mid-check. trace!( room_id = %room_id, "Locking the room" ); let state_lock = self.services.state.mutex.lock(room_id).await; - - let state_ids_compressed: Arc = self - .services - .state_compressor - .compress_state_events( - state_at_incoming_event - .iter() - .map(|(ssk, eid)| (ssk, eid.borrow())), - ) - .collect() - .map(Arc::new) - .await; - - if incoming_pdu.state_key().is_some() { - debug!("Event is a state-event. Deriving new room state"); - - // We also add state after incoming event to the fork states - let mut state_after = state_at_incoming_event.clone(); - if let Some(state_key) = incoming_pdu.state_key() { - let shortstatekey = self - .services - .short - .get_or_create_shortstatekey( - &incoming_pdu.kind().to_string().into(), - state_key, - ) - .await; - - let event_id = incoming_pdu.event_id(); - state_after.insert(shortstatekey, event_id.to_owned()); - } - - let new_room_state = self - .resolve_state(room_id, &room_version_rules, state_after) - .await?; - - // Set the new room state to the resolved state - debug!("Forcing new room state"); - let HashSetCompressStateEvent { shortstatehash, added, removed } = self - .services - .state_compressor - .save_state(room_id, new_room_state) - .await?; - - self.services - .state - .force_state(room_id, shortstatehash, added, removed, &state_lock) - .await?; - } - - if !soft_fail { - // Don't call the below checks on events that have already soft-failed, there's - // no reason to re-calculate that. - // 14-pre. ask the policy server to sign the event, if possible - debug!(event_id = %incoming_pdu.event_id, "Checking policy server for event"); - let tmp_evt_id = val.remove("event_id"); - if let Err(e) = self - .policy_server_allows_event( - &incoming_pdu, - &mut val, - room_id, - &room_version_rules, - true, - ) - .await - { - 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(), + let passes_current_state = self + .current_state_check_6(&incoming_pdu, &room_version_rules, create_event) + .await + .inspect(|passes| { + if !*passes { + debug_warn!( + "Event authorisation fails based on the current room state - will be \ + soft-failed" ); - soft_fail = true; - } else { - return Err(e); } - } else { - debug!( - event_id = %incoming_pdu.event_id, - "Event has passed policy server check." - ); - } - if let Some(id) = tmp_evt_id { - val.insert("event_id".to_owned(), id); - } + })?; - // Additionally, if this is a redaction for a soft-failed event, we soft-fail it - // also. + // Determine whether this PDU should be soft-failed. + // If the auth check failed, invariably yes. Otherwise, only if the user isn't + // allowed to redact the target event (if any). + let mut should_soft_fail = + match (passes_current_state, incoming_pdu.redacts_id(&room_version_rules)) { + | (false, _) => true, + | (true, None) => false, + | (true, Some(redact_id)) => self + .services + .state_accessor + .user_can_redact(&redact_id, incoming_pdu.sender(), room_id, true) + .await + .is_ok_and(is_true!()), + }; - // TODO: this is supposed to hide redactions from policy servers, however, for - // full efficacy it also needs to hide redactions for unknown events. This - // needs to be investigated at a later time. + if !should_soft_fail { + // Now we can perform check 7, which is ensuring the event passes policy server + // checks. + // We explicitly only do this if we aren't already going to soft-fail the event, + // 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_7(&incoming_pdu, &mut val, &room_version_rules) + .await + .inspect(|passes| { + if !*passes { + debug_warn!( + "Event did not pass the policy server check and will be soft-failed" + ); + } + })?; + + // TODO: this is supposed to hide redactions from policy servers and janitorial + // bots, however, for full efficacy it also needs to hide redactions for + // unknown events. This needs to be investigated at a later time. if let Some(redact_id) = incoming_pdu.redacts_id(&room_version_rules) { debug!( redact_id = %redact_id, - "Checking if redaction is for a soft-failed event" + "Checking if redaction is for a soft-failed/rejected event" ); - if self + if !self .services .pdu_metadata - .is_event_soft_failed(&redact_id) + .is_event_accepted(&redact_id) .await { - info!( - redact_id = %redact_id, - "Redaction is for a soft-failed event" + debug_info!( + "Soft-failing valid redaction because it targets a non-accepted event" ); - soft_fail = true; + should_soft_fail = true; } } } + // The PDU has now passed all checks! We can now promote it (or soft-fail it if + // the verdict is such). trace!("Appending pdu to timeline"); - let mut extremities: Vec<_> = self - .services - .state - .get_forward_extremities(room_id) - .collect() - .await; - if !soft_fail { - // Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events - // are not added as forward extremities. - - // Now we calculate the set of extremities this room has after the incoming - // event has been applied. We start with the previous extremities (aka leaves) - trace!("Calculating extremities"); - extremities = extremities - .into_iter() - .stream() - .ready_filter(|event_id| { - // Remove any that are referenced by this incoming event's prev_events - !incoming_pdu.prev_events().any(is_equal_to!(event_id)) - }) - .broad_filter_map(|event_id| async move { - // Only keep those extremities were not referenced yet - self.services - .pdu_metadata - .is_event_referenced(room_id, &event_id) - .await - .eq(&false) - .then_some(event_id) - }) - .collect::>() - .await; - extremities.push(incoming_pdu.event_id().to_owned()); - debug!( - "Retained {} extremities checked against {} prev_events", - extremities.len(), - incoming_pdu.prev_events().count() - ); - assert!(!extremities.is_empty(), "extremities must not empty"); - } - let pdu_id = self .services .timeline .append_incoming_pdu( &incoming_pdu, val, - extremities.iter().map(Borrow::borrow), - state_ids_compressed, - soft_fail, + &room_version_rules, + state_before, + should_soft_fail, &state_lock, - room_id, ) .await?; - if soft_fail { - self.services - .pdu_metadata - .mark_event_soft_failed(incoming_pdu.event_id()); - info!( + if should_soft_fail { + debug_info!( elapsed = ?timer.elapsed(), event_id = %incoming_pdu.event_id, "Event was soft failed" diff --git a/src/service/rooms/membership/mod.rs b/src/service/rooms/membership/mod.rs index 02e0a8d26..22758ad1d 100644 --- a/src/service/rooms/membership/mod.rs +++ b/src/service/rooms/membership/mod.rs @@ -520,7 +520,7 @@ pub async fn join_remote_room( return state; }, }; - if !pdu_fits(&mut value.clone()) { + if !pdu_fits(&value.clone()) { warn!( "dropping incoming PDU {event_id} in room {room_id} from room join \ because it exceeds 65535 bytes or is otherwise too large." diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index 3d7dcbcc1..c1959c94d 100644 --- a/src/service/rooms/timeline/append.rs +++ b/src/service/rooms/timeline/append.rs @@ -1,69 +1,113 @@ use std::{ - collections::{BTreeMap, HashSet}, + borrow::Borrow, + collections::{BTreeMap, HashMap, HashSet}, sync::Arc, }; use conduwuit::{ - debug_warn, + Result, debug, debug_warn, is_equal_to, + matrix::{Event, PduEvent}, pdu::{Count, ShortRoomId}, trace, - utils::{IterStream, TryFutureExtExt, stream::BroadbandExt}, + utils::{ + IterStream, TryFutureExtExt, + mutex_map::Guard, + stream::{BroadbandExt, ReadyExt}, + }, warn, }; use conduwuit_core::{ - Result, err, error, - matrix::{ - event::Event, - pdu::{PduCount, PduEvent, PduId, RawPduId}, - }, - utils::{self, ReadyExt}, + err, error, + matrix::pdu::{PduCount, PduId, RawPduId}, + utils::{self}, }; -use futures::{StreamExt, TryFutureExt}; +use futures::{FutureExt, StreamExt, TryFutureExt}; use ruma::{ - CanonicalJsonObject, CanonicalJsonValue, EventId, RoomId, RoomVersionId, UserId, + CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, RoomId, + RoomVersionId, UserId, events::{ GlobalAccountDataEventType, TimelineEventType, push_rules::PushRulesEvent, room::{encrypted::Relation, redaction::RoomRedactionEventContent}, }, push::{Action, Ruleset, Tweak}, + room_version_rules::RoomVersionRules, }; use super::{ExtractBody, ExtractRelatesTo, ExtractRelatesToEventId, RoomMutexGuard}; -use crate::{appservice::RegistrationInfo, rooms::state_compressor::CompressedState}; +use crate::{ + appservice::RegistrationInfo, + rooms::state_compressor::{CompressedState, HashSetCompressStateEvent}, +}; impl super::Service { /// Append the incoming event setting the state snapshot to the state from /// the server that sent the event. - #[allow(clippy::too_many_arguments)] - pub async fn append_incoming_pdu<'a, Leaves>( + pub async fn append_incoming_pdu<'a>( &'a self, pdu: &'a PduEvent, pdu_json: CanonicalJsonObject, - new_room_leaves: Leaves, - state_ids_compressed: Arc, - soft_fail: bool, + room_version_rules: &RoomVersionRules, + state_before: HashMap, + should_soft_fail: bool, state_lock: &'a RoomMutexGuard, - room_id: &'a RoomId, - ) -> Result> - where - Leaves: Iterator + Send + 'a, - { + ) -> Result> { + let room_id = pdu.room_id_or_hash(); + + let sb = state_before.iter().map(|(ssk, eid)| (ssk, eid.as_ref())); + let state_ids_compressed: Arc = self + .services + .state_compressor + .compress_state_events(sb) + .collect() + .map(Arc::new) + .await; + // We append to state before appending the pdu, so we don't have a moment in // time with the pdu without it's state. This is okay because append_pdu can't // fail. self.services .state - .set_event_state(&pdu.event_id, room_id, state_ids_compressed) + .set_event_state(&pdu.event_id, &room_id, state_ids_compressed) .await?; - if soft_fail { + if should_soft_fail { // Nothing else to do with a soft-failed event. + self.services + .pdu_metadata + .mark_event_soft_failed(pdu.event_id()); return Ok(None); } + // Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events + // are not added as forward extremities. + // This also means we set the state here. + // We do this BEFORE setting the extremities so that there's never a point in + // time where we have fresh extremities referencing stale state. + let mut extremities: Vec<_> = self + .services + .state + .get_forward_extremities(&room_id) + .collect() + .await; + extremities = self + .progress_state_and_extremities( + pdu, + room_version_rules, + state_before.clone(), + extremities, + state_lock, + ) + .await?; + let pdu_id = self - .append_pdu(pdu, pdu_json, new_room_leaves, state_lock, room_id) + .append_pdu( + pdu, + pdu_json, + extremities.iter().map(Borrow::borrow), + state_lock, + &room_id, + ) .await?; // Process admin commands for federation events @@ -86,11 +130,110 @@ pub async fn append_incoming_pdu<'a, Leaves>( } } - self.services.sync.wake_all_joined(room_id).await; + self.services.sync.wake_all_joined(&room_id).await; Ok(Some(pdu_id)) } + /// Derives new room state from the incoming event and filters forward + /// extremities accordingly. Does not set forward extremities. + /// + /// Only call this function if the incoming PDU is not soft-failed or + /// rejected. + async fn progress_state_and_extremities( + &self, + incoming_pdu: &PduEvent, + room_version_rules: &RoomVersionRules, + state_before: HashMap, + forward_extremities: Vec, + state_lock: &Guard, + ) -> Result> { + if incoming_pdu.state_key().is_some() { + debug!("Event is a state-event. Deriving new room state"); + self.derive_new_state(incoming_pdu, room_version_rules, state_before, state_lock) + .await?; + } + + // Now we calculate the set of extremities this room has after the incoming + // event has been applied. We start with the previous extremities + trace!("Calculating extremities"); + let mut forward_extremities = forward_extremities + .into_iter() + .stream() + .ready_filter(|event_id| { + // Remove any that are referenced by this incoming event's prev_events + !incoming_pdu.prev_events().any(is_equal_to!(event_id)) + }) + .broad_filter_map(|event_id| async move { + // Only keep those extremities were not referenced yet + self.services + .pdu_metadata + .is_event_referenced(&incoming_pdu.room_id_or_hash(), &event_id) + .await + .eq(&false) + .then_some(event_id) + }) + .collect::>() + .await; + forward_extremities.push(incoming_pdu.event_id().to_owned()); + debug!( + "Retained {} extremities checked against {} prev_events", + forward_extremities.len(), + incoming_pdu.prev_events().count() + ); + assert!(!forward_extremities.is_empty(), "resolved extremities cannot be empty"); + Ok(forward_extremities) + } + + /// Derives a new room state by adding the incoming PDU to the state before + /// it to create the state at, which then becomes the current room state. + /// + /// The caller MUST ensure forward extremities are set appropriately, + /// including this incoming pdu, either before or after calling this + /// function. Failing to do so will result in an inconsistent current state + /// cache, which may affect event authentication. + async fn derive_new_state( + &self, + incoming_pdu: &PduEvent, + room_version_rules: &RoomVersionRules, + state_before: HashMap, + state_lock: &Guard, + ) -> Result { + let room_id = incoming_pdu.room_id_or_hash(); + // We also add state after incoming event to the fork states + let mut state_at_incoming_event = state_before; + let shortstatekey = self + .services + .short + .get_or_create_shortstatekey( + &incoming_pdu.kind().to_string().into(), + incoming_pdu.state_key().unwrap(), + ) + .await; + + let event_id = incoming_pdu.event_id(); + state_at_incoming_event.insert(shortstatekey, event_id.to_owned()); + + debug!("Resolving new room state"); + let new_room_state = self + .services + .event_handler + .resolve_state(&room_id, room_version_rules, state_at_incoming_event) + .await?; + + debug!("Forcing new room state"); + let HashSetCompressStateEvent { shortstatehash, added, removed } = self + .services + .state_compressor + .save_state(&room_id, new_room_state) + .await?; + + self.services + .state + .force_state(&room_id, shortstatehash, added, removed, state_lock) + .await + } + /// Populates the unsigned data of a PDU async fn populate_unsigned(&self, pdu: &PduEvent, pdu_json: &mut CanonicalJsonObject) { let Some(state_key) = pdu.state_key() else { diff --git a/src/service/rooms/timeline/backfill.rs b/src/service/rooms/timeline/backfill.rs index 36b01c8e8..e40cf7c5e 100644 --- a/src/service/rooms/timeline/backfill.rs +++ b/src/service/rooms/timeline/backfill.rs @@ -143,7 +143,7 @@ pub async fn get_remote_pdu(&self, room_id: &RoomId, event_id: &EventId) -> Resu | Ok(value) => { self.services .event_handler - .handle_incoming_pdu(&backfill_server, room_id, event_id, value, false) + .handle_incoming_pdu(&backfill_server, room_id, event_id, value, true) .boxed() .await?; debug!("Successfully backfilled {event_id} from {backfill_server}"); @@ -185,7 +185,7 @@ pub async fn backfill_pdu(&self, origin: &ServerName, pdu: Box) -> self.services .event_handler - .handle_incoming_pdu(origin, &room_id, &event_id, value, false) + .handle_incoming_pdu(origin, &room_id, &event_id, value, true) .boxed() .await?; diff --git a/src/service/rooms/timeline/create.rs b/src/service/rooms/timeline/create.rs index 9cdd3ea34..2f05ca8cc 100644 --- a/src/service/rooms/timeline/create.rs +++ b/src/service/rooms/timeline/create.rs @@ -22,7 +22,9 @@ use super::RoomMutexGuard; -pub fn pdu_fits(owned_obj: &mut CanonicalJsonObject) -> bool { +/// Ensures the given PDU fits inside the size limits for a PDU. +#[must_use] +pub fn pdu_fits(owned_obj: &CanonicalJsonObject) -> bool { // room IDs, event IDs, senders, types, and state keys must all be <= 255 bytes if let Some(CanonicalJsonValue::String(room_id)) = owned_obj.get("room_id") { if room_id.len() > 255 { @@ -311,7 +313,7 @@ pub async fn create_hash_and_sign_event( pdu_json .insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into())); // Verify that the *full* PDU isn't over 64KiB. - if !pdu_fits(&mut pdu_json.clone()) { + if !pdu_fits(&pdu_json.clone()) { // feckin huge PDU mate return Err!(Request(TooLarge("Message/PDU is too long (exceeds 65535 bytes)"))); } diff --git a/src/service/rooms/timeline/mod.rs b/src/service/rooms/timeline/mod.rs index c96707b43..a3d0fbd7c 100644 --- a/src/service/rooms/timeline/mod.rs +++ b/src/service/rooms/timeline/mod.rs @@ -77,6 +77,7 @@ struct Services { short: Dep, state: Dep, state_accessor: Dep, + state_compressor: Dep, state_cache: Dep, sync: Dep, threads: Dep, @@ -110,6 +111,8 @@ fn build(args: crate::Args<'_>) -> Result> { state: args.depend::("rooms::state"), state_accessor: args .depend::("rooms::state_accessor"), + state_compressor: args + .depend::("rooms::state_compressor"), state_cache: args.depend::("rooms::state_cache"), sync: args.depend::("sync"), threads: args.depend::("rooms::threads"),