diff --git a/src/service/rooms/event_handler/acl_check.rs b/src/service/rooms/event_handler/acl_check.rs index f847015bb..c3fcfea1c 100644 --- a/src/service/rooms/event_handler/acl_check.rs +++ b/src/service/rooms/event_handler/acl_check.rs @@ -1,42 +1,40 @@ -use conduwuit::{Err, Result, debug, implement, trace, warn}; +use conduwuit::{Err, Result, debug, debug_warn, trace}; use ruma::{ RoomId, ServerName, events::{StateEventType, room::server_acl::RoomServerAclEventContent}, }; -/// Returns Ok if the acl allows the server -#[implement(super::Service)] -#[tracing::instrument(skip_all, level = "debug")] -pub async fn acl_check(&self, server_name: &ServerName, room_id: &RoomId) -> Result { - let Ok(acl_event_content) = self - .services - .state_accessor - .room_state_get_content(room_id, &StateEventType::RoomServerAcl, "") - .await - .map(|c: RoomServerAclEventContent| c) - .inspect(|acl| trace!(%room_id, "ACL content found: {acl:?}")) - .inspect_err(|e| trace!(%room_id, "No ACL content found: {e:?}")) - else { - return Ok(()); - }; +impl super::Service { + /// Returns Ok if the acl allows the server, otherwise Forbidden. + pub async fn acl_check(&self, server_name: &ServerName, room_id: &RoomId) -> Result { + let Ok(acl_event_content) = self + .services + .state_accessor + .room_state_get_content(room_id, &StateEventType::RoomServerAcl, "") + .await + .map(|c: RoomServerAclEventContent| c) + else { + return Ok(()); + }; - if acl_event_content.allow.is_empty() { - warn!(%room_id, "Ignoring broken ACL event (allow key is empty)"); - return Ok(()); - } + if acl_event_content.allow.is_empty() { + debug_warn!(%room_id, "Ignoring broken ACL event (allow key is empty)"); + return Ok(()); + } - if acl_event_content.deny.contains(&String::from("*")) - && acl_event_content.allow.contains(&String::from("*")) - { - warn!(%room_id, "Ignoring broken ACL event (allow key and deny key both contain wildcard \"*\""); - return Ok(()); - } + if acl_event_content.deny.contains(&String::from("*")) + && acl_event_content.allow.contains(&String::from("*")) + { + debug_warn!(%room_id, "Ignoring broken ACL event (allow key and deny key both contain wildcard \"*\""); + return Ok(()); + } - if acl_event_content.is_allowed(server_name) { - trace!("server {server_name} is allowed by ACL"); - Ok(()) - } else { - debug!("Server {server_name} was denied by room ACL in {room_id}"); - Err!(Request(Forbidden("Server was denied by room ACL"))) + if acl_event_content.is_allowed(server_name) { + trace!("server {server_name} is allowed by ACL"); + Ok(()) + } else { + debug!("Server {server_name} was denied by room ACL in {room_id}"); + Err!(Request(Forbidden("Server was denied by room ACL"))) + } } } 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 e465131fb..09d819b77 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs @@ -4,8 +4,8 @@ use std::{ }; use conduwuit::{ - Event, PduEvent, debug, debug_warn, implement, matrix::event::gen_event_id_canonical_json, - trace, utils::continue_exponential_backoff_secs, warn, + Event, PduEvent, debug, debug_warn, matrix::event::gen_event_id_canonical_json, trace, + utils::continue_exponential_backoff_secs, warn, }; use ruma::{ CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName, @@ -14,220 +14,222 @@ use ruma::{ use super::get_room_version_rules; -/// Find the event and auth it. Once the event is validated (steps 1 - 8) -/// it is appended to the outliers Tree. -/// -/// Returns pdu and if we fetched it over federation the raw json. -/// -/// a. Look in the main timeline (pduid_pdu tree) -/// b. Look at outlier pdu tree -/// c. Ask origin server over federation -/// d. TODO: Ask other servers over federation? -#[implement(super::Service)] -pub(super) async fn fetch_and_handle_outliers<'a, Pdu, Events>( - &self, - origin: &'a ServerName, - events: Events, - create_event: &'a Pdu, - room_id: &'a RoomId, -) -> Vec<(PduEvent, Option>)> -where - Pdu: Event + Send + Sync, - Events: Iterator + Clone + Send, -{ - let back_off = |id| match self - .services - .globals - .bad_event_ratelimiter - .write() - .entry(id) +impl super::Service { + /// Find the event and auth it. Once the event is validated (steps 1 - 8) + /// it is appended to the outliers Tree. + /// + /// Returns pdu and if we fetched it over federation the raw json. + /// + /// a. Look in the main timeline (pduid_pdu tree) + /// b. Look at outlier pdu tree + /// c. Ask origin server over federation + /// d. TODO: Ask other servers over federation? + pub(super) async fn fetch_and_handle_outliers<'a, Pdu, Events>( + &self, + origin: &'a ServerName, + events: Events, + create_event: &'a Pdu, + room_id: &'a RoomId, + ) -> Vec<(PduEvent, Option>)> + where + Pdu: Event + Send + Sync, + Events: Iterator + Clone + Send, { - | hash_map::Entry::Vacant(e) => { - e.insert((Instant::now(), 1)); - }, - | hash_map::Entry::Occupied(mut e) => { - *e.get_mut() = (Instant::now(), e.get().1.saturating_add(1)); - }, - }; + let back_off = |id| match self + .services + .globals + .bad_event_ratelimiter + .write() + .entry(id) + { + | hash_map::Entry::Vacant(e) => { + e.insert((Instant::now(), 1)); + }, + | hash_map::Entry::Occupied(mut e) => { + *e.get_mut() = (Instant::now(), e.get().1.saturating_add(1)); + }, + }; - let mut events_with_auth_events = Vec::with_capacity(events.clone().count()); - trace!("Fetching {} outlier pdus", events.clone().count()); + let mut events_with_auth_events = Vec::with_capacity(events.clone().count()); + trace!("Fetching {} outlier pdus", events.clone().count()); - for id in events { - // a. Look in the main timeline (pduid_pdu tree) - // b. Look at outlier pdu tree - // (get_pdu_json checks both) - if let Ok(local_pdu) = self.services.timeline.get_pdu(id).await { - trace!("Found {id} in main timeline or outlier tree"); - events_with_auth_events.push((id.to_owned(), Some(local_pdu), vec![])); - continue; - } - - // c. Ask origin server over federation - // We also handle its auth chain here so we don't get a stack overflow in - // handle_outlier_pdu. - let mut todo_auth_events: VecDeque<_> = [id.to_owned()].into(); - let mut events_in_reverse_order = Vec::with_capacity(todo_auth_events.len()); - - let mut events_all = HashSet::with_capacity(todo_auth_events.len()); - while let Some(next_id) = todo_auth_events.pop_front() { - if let Some((time, tries)) = self - .services - .globals - .bad_event_ratelimiter - .read() - .get(&*next_id) - { - // Exponential backoff - const MIN_DURATION: u64 = 60 * 2; - const MAX_DURATION: u64 = 60 * 60 * 8; - if continue_exponential_backoff_secs( - MIN_DURATION, - MAX_DURATION, - time.elapsed(), - *tries, - ) { - debug_warn!( - tried = ?*tries, - elapsed = ?time.elapsed(), - "Backing off from {next_id}", - ); - continue; - } - } - - if events_all.contains(&next_id) { + for id in events { + // a. Look in the main timeline (pduid_pdu tree) + // b. Look at outlier pdu tree + // (get_pdu_json checks both) + if let Ok(local_pdu) = self.services.timeline.get_pdu(id).await { + trace!("Found {id} in main timeline or outlier tree"); + events_with_auth_events.push((id.to_owned(), Some(local_pdu), vec![])); continue; } - if self.services.timeline.pdu_exists(&next_id).await { - trace!("Found {next_id} in db"); - continue; - } + // c. Ask origin server over federation + // We also handle its auth chain here so we don't get a stack overflow in + // handle_outlier_pdu. + let mut todo_auth_events: VecDeque<_> = [id.to_owned()].into(); + let mut events_in_reverse_order = Vec::with_capacity(todo_auth_events.len()); - debug!("Fetching {next_id} over federation from {origin}."); - match self - .services - .sending - .send_federation_request( - origin, - get_event::v1::Request::new((*next_id).to_owned()), - ) - .await - { - | Ok(res) => { - debug!("Got {next_id} over federation from {origin}"); - let Ok(room_version_rules) = get_room_version_rules(create_event) else { - back_off((*next_id).to_owned()); - continue; - }; - - let Ok((calculated_event_id, value)) = - gen_event_id_canonical_json(&res.pdu, &room_version_rules) - else { - back_off((*next_id).to_owned()); - continue; - }; - - if calculated_event_id != *next_id { - warn!( - "Server didn't return event id we requested: requested: {next_id}, \ - we got {calculated_event_id}. Event: {:?}", - &res.pdu + let mut events_all = HashSet::with_capacity(todo_auth_events.len()); + while let Some(next_id) = todo_auth_events.pop_front() { + if let Some((time, tries)) = self + .services + .globals + .bad_event_ratelimiter + .read() + .get(&*next_id) + { + // Exponential backoff + const MIN_DURATION: u64 = 60 * 2; + const MAX_DURATION: u64 = 60 * 60 * 8; + if continue_exponential_backoff_secs( + MIN_DURATION, + MAX_DURATION, + time.elapsed(), + *tries, + ) { + debug_warn!( + tried = ?*tries, + elapsed = ?time.elapsed(), + "Backing off from {next_id}", ); + continue; } + } - if let Some(auth_events) = value - .get("auth_events") - .and_then(CanonicalJsonValue::as_array) - { - for auth_event in auth_events { - match serde_json::from_value::( - auth_event.clone().into(), - ) { - | Ok(auth_event) => { - trace!( - "Found auth event id {auth_event} for event {next_id}" - ); - todo_auth_events.push_back(auth_event); - }, - | _ => { - warn!("Auth event id is not valid"); - }, - } - } - } else { - warn!("Auth event list invalid"); - } - - events_in_reverse_order.push((next_id.clone(), value)); - events_all.insert(next_id); - }, - | Err(e) => { - warn!("Failed to fetch auth event {next_id} from {origin}: {e}"); - back_off((*next_id).to_owned()); - }, - } - } - - events_with_auth_events.push((id.to_owned(), None, events_in_reverse_order)); - } - - let mut pdus = Vec::with_capacity(events_with_auth_events.len()); - for (id, local_pdu, events_in_reverse_order) in events_with_auth_events { - // a. Look in the main timeline (pduid_pdu tree) - // b. Look at outlier pdu tree - // (get_pdu_json checks both) - if let Some(local_pdu) = local_pdu { - trace!("Found {id} in main timeline or outlier tree"); - pdus.push((local_pdu.clone(), None)); - } - - for (next_id, value) in events_in_reverse_order.into_iter().rev() { - if let Some((time, tries)) = self - .services - .globals - .bad_event_ratelimiter - .read() - .get(&*next_id) - { - // Exponential backoff - const MIN_DURATION: u64 = 5 * 60; - const MAX_DURATION: u64 = 60 * 60 * 24; - if continue_exponential_backoff_secs( - MIN_DURATION, - MAX_DURATION, - time.elapsed(), - *tries, - ) { - debug!("Backing off from {next_id}"); + if events_all.contains(&next_id) { continue; } + + if self.services.timeline.pdu_exists(&next_id).await { + trace!("Found {next_id} in db"); + continue; + } + + debug!("Fetching {next_id} over federation from {origin}."); + match self + .services + .sending + .send_federation_request( + origin, + get_event::v1::Request::new((*next_id).to_owned()), + ) + .await + { + | Ok(res) => { + debug!("Got {next_id} over federation from {origin}"); + let Ok(room_version_rules) = get_room_version_rules(create_event) else { + back_off((*next_id).to_owned()); + continue; + }; + + let Ok((calculated_event_id, value)) = + gen_event_id_canonical_json(&res.pdu, &room_version_rules) + else { + back_off((*next_id).to_owned()); + continue; + }; + + if calculated_event_id != *next_id { + warn!( + "Server didn't return event id we requested: requested: \ + {next_id}, we got {calculated_event_id}. Event: {:?}", + &res.pdu + ); + } + + if let Some(auth_events) = value + .get("auth_events") + .and_then(CanonicalJsonValue::as_array) + { + for auth_event in auth_events { + match serde_json::from_value::( + auth_event.clone().into(), + ) { + | Ok(auth_event) => { + trace!( + "Found auth event id {auth_event} for event \ + {next_id}" + ); + todo_auth_events.push_back(auth_event); + }, + | _ => { + warn!("Auth event id is not valid"); + }, + } + } + } else { + warn!("Auth event list invalid"); + } + + events_in_reverse_order.push((next_id.clone(), value)); + events_all.insert(next_id); + }, + | Err(e) => { + warn!("Failed to fetch auth event {next_id} from {origin}: {e}"); + back_off((*next_id).to_owned()); + }, + } } - trace!("Handling outlier {next_id}"); - match Box::pin(self.handle_outlier_pdu( - origin, - create_event, - &next_id, - room_id, - value.clone(), - true, - )) - .await - { - | Ok((pdu, json)) => - if next_id == *id { - trace!("Handled outlier {next_id} (original request)"); - pdus.push((pdu, Some(json))); + events_with_auth_events.push((id.to_owned(), None, events_in_reverse_order)); + } + + let mut pdus = Vec::with_capacity(events_with_auth_events.len()); + for (id, local_pdu, events_in_reverse_order) in events_with_auth_events { + // a. Look in the main timeline (pduid_pdu tree) + // b. Look at outlier pdu tree + // (get_pdu_json checks both) + if let Some(local_pdu) = local_pdu { + trace!("Found {id} in main timeline or outlier tree"); + pdus.push((local_pdu.clone(), None)); + } + + for (next_id, value) in events_in_reverse_order.into_iter().rev() { + if let Some((time, tries)) = self + .services + .globals + .bad_event_ratelimiter + .read() + .get(&*next_id) + { + // Exponential backoff + const MIN_DURATION: u64 = 5 * 60; + const MAX_DURATION: u64 = 60 * 60 * 24; + if continue_exponential_backoff_secs( + MIN_DURATION, + MAX_DURATION, + time.elapsed(), + *tries, + ) { + debug!("Backing off from {next_id}"); + continue; + } + } + + trace!("Handling outlier {next_id}"); + match Box::pin(self.handle_outlier_pdu( + origin, + create_event, + &next_id, + room_id, + value.clone(), + true, + )) + .await + { + | Ok((pdu, json)) => + if next_id == *id { + trace!("Handled outlier {next_id} (original request)"); + pdus.push((pdu, Some(json))); + }, + | Err(e) => { + warn!("Authentication of event {next_id} failed: {e:?}"); + back_off(next_id); }, - | Err(e) => { - warn!("Authentication of event {next_id} failed: {e:?}"); - back_off(next_id); - }, + } } } + trace!("Fetched and handled {} outlier pdus", pdus.len()); + pdus } - trace!("Fetched and handled {} outlier pdus", pdus.len()); - pdus } diff --git a/src/service/rooms/event_handler/fetch_prev.rs b/src/service/rooms/event_handler/fetch_prev.rs index efc7a434d..19c7a1089 100644 --- a/src/service/rooms/event_handler/fetch_prev.rs +++ b/src/service/rooms/event_handler/fetch_prev.rs @@ -4,7 +4,7 @@ use std::{ }; use conduwuit::{ - Event, PduEvent, Result, debug_warn, err, implement, + Event, PduEvent, Result, debug_warn, err, state_res::{self}, }; use futures::{FutureExt, future}; @@ -15,114 +15,110 @@ use ruma::{ use super::check_room_id; -#[implement(super::Service)] -#[tracing::instrument( - level = "debug", - skip_all, - fields(%origin), -)] -#[allow(clippy::type_complexity)] -pub(super) async fn fetch_prev<'a, Pdu, Events>( - &self, - origin: &ServerName, - create_event: &Pdu, - room_id: &RoomId, - first_ts_in_room: MilliSecondsSinceUnixEpoch, - initial_set: Events, -) -> Result<( - Vec, - HashMap)>, -)> -where - Pdu: Event + Send + Sync, - Events: Iterator + Clone + Send, -{ - let num_ids = initial_set.clone().count(); - let mut eventid_info = HashMap::new(); - let mut graph: HashMap = HashMap::with_capacity(num_ids); - let mut todo_outlier_stack: VecDeque = - initial_set.map(ToOwned::to_owned).collect(); +impl super::Service { + #[allow(clippy::type_complexity)] + pub(super) async fn fetch_prev<'a, Pdu, Events>( + &self, + origin: &ServerName, + create_event: &Pdu, + room_id: &RoomId, + first_ts_in_room: MilliSecondsSinceUnixEpoch, + initial_set: Events, + ) -> Result<( + Vec, + HashMap)>, + )> + where + Pdu: Event + Send + Sync, + Events: Iterator + Clone + Send, + { + let num_ids = initial_set.clone().count(); + let mut eventid_info = HashMap::new(); + let mut graph: HashMap = HashMap::with_capacity(num_ids); + let mut todo_outlier_stack: VecDeque = + initial_set.map(ToOwned::to_owned).collect(); - let mut amount = 0; + let mut amount = 0; - while let Some(prev_event_id) = todo_outlier_stack.pop_front() { - self.services.server.check_running()?; + while let Some(prev_event_id) = todo_outlier_stack.pop_front() { + self.services.server.check_running()?; - match self - .fetch_and_handle_outliers( - origin, - once(prev_event_id.as_ref()), - create_event, - room_id, - ) - .boxed() - .await - .pop() - { - | Some((pdu, mut json_opt)) => { - check_room_id(room_id, &pdu)?; + match self + .fetch_and_handle_outliers( + origin, + once(prev_event_id.as_ref()), + create_event, + room_id, + ) + .boxed() + .await + .pop() + { + | Some((pdu, mut json_opt)) => { + check_room_id(room_id, &pdu)?; - let limit = self.services.server.config.max_fetch_prev_events; - if amount > limit { - debug_warn!("Max prev event limit reached! Limit: {limit}"); - graph.insert(prev_event_id.clone(), HashSet::new()); - continue; - } - - if json_opt.is_none() { - json_opt = self - .services - .outlier - .get_outlier_pdu_json(&prev_event_id) - .await - .ok(); - } - - if let Some(json) = json_opt { - if pdu.origin_server_ts() > first_ts_in_room { - amount = amount.saturating_add(1); - for prev_prev in pdu.prev_events() { - if !graph.contains_key(prev_prev) { - todo_outlier_stack.push_back(prev_prev.to_owned()); - } - } - - graph.insert( - prev_event_id.clone(), - pdu.prev_events().map(ToOwned::to_owned).collect(), - ); - } else { - // Time based check failed + let limit = self.services.server.config.max_fetch_prev_events; + if amount > limit { + debug_warn!("Max prev event limit reached! Limit: {limit}"); graph.insert(prev_event_id.clone(), HashSet::new()); + continue; } - eventid_info.insert(prev_event_id.clone(), (pdu, json)); - } else { - // Get json failed, so this was not fetched over federation + if json_opt.is_none() { + json_opt = self + .services + .outlier + .get_outlier_pdu_json(&prev_event_id) + .await + .ok(); + } + + if let Some(json) = json_opt { + if pdu.origin_server_ts() > first_ts_in_room { + amount = amount.saturating_add(1); + for prev_prev in pdu.prev_events() { + if !graph.contains_key(prev_prev) { + todo_outlier_stack.push_back(prev_prev.to_owned()); + } + } + + graph.insert( + prev_event_id.clone(), + pdu.prev_events().map(ToOwned::to_owned).collect(), + ); + } else { + // Time based check failed + graph.insert(prev_event_id.clone(), HashSet::new()); + } + + eventid_info.insert(prev_event_id.clone(), (pdu, json)); + } else { + // Get json failed, so this was not fetched over federation + graph.insert(prev_event_id.clone(), HashSet::new()); + } + }, + | _ => { + // Fetch and handle failed graph.insert(prev_event_id.clone(), HashSet::new()); - } - }, - | _ => { - // Fetch and handle failed - graph.insert(prev_event_id.clone(), HashSet::new()); - }, + }, + } } + + let event_fetch = |event_id| { + let origin_server_ts = eventid_info + .get(&event_id) + .map_or_else(|| uint!(0), |info| info.0.origin_server_ts().get()); + + // This return value is the key used for sorting events, + // events are then sorted by power level, time, + // and lexically by event_id. + future::ok((int!(0), MilliSecondsSinceUnixEpoch(origin_server_ts))) + }; + + let sorted = state_res::lexicographical_topological_sort(&graph, &event_fetch) + .await + .map_err(|e| err!(Database(error!("Error sorting prev events: {e}"))))?; + + Ok((sorted, eventid_info)) } - - let event_fetch = |event_id| { - let origin_server_ts = eventid_info - .get(&event_id) - .map_or_else(|| uint!(0), |info| info.0.origin_server_ts().get()); - - // This return value is the key used for sorting events, - // events are then sorted by power level, time, - // and lexically by event_id. - future::ok((int!(0), MilliSecondsSinceUnixEpoch(origin_server_ts))) - }; - - let sorted = state_res::lexicographical_topological_sort(&graph, &event_fetch) - .await - .map_err(|e| err!(Database(error!("Error sorting prev events: {e}"))))?; - - Ok((sorted, eventid_info)) } diff --git a/src/service/rooms/event_handler/fetch_state.rs b/src/service/rooms/event_handler/fetch_state.rs index 790b4ec9e..d245b35e7 100644 --- a/src/service/rooms/event_handler/fetch_state.rs +++ b/src/service/rooms/event_handler/fetch_state.rs @@ -1,6 +1,6 @@ use std::collections::{HashMap, hash_map}; -use conduwuit::{Err, Event, Result, debug, debug_warn, err, implement}; +use conduwuit::{Err, Event, Result, debug, debug_warn, err}; use futures::FutureExt; use ruma::{ EventId, OwnedEventId, RoomId, ServerName, api::federation::event::get_room_state_ids, @@ -9,78 +9,81 @@ use ruma::{ use crate::rooms::short::ShortStateKey; -/// Call /state_ids to find out what the state at this pdu is. We trust the -/// server's response to some extend (sic), but we still do a lot of checks -/// on the events -#[implement(super::Service)] -#[tracing::instrument( - level = "debug", - skip_all, - fields(%origin), -)] -pub(super) async fn fetch_state( - &self, - origin: &ServerName, - create_event: &Pdu, - room_id: &RoomId, - event_id: &EventId, -) -> Result>> -where - Pdu: Event + Send + Sync, -{ - let res = self - .services - .sending - .send_federation_request( - origin, - get_room_state_ids::v1::Request::new(event_id.to_owned(), room_id.to_owned()), - ) - .await - .inspect_err(|e| debug_warn!("Fetching state for event failed: {e}"))?; - - debug!("Fetching state events"); - let state_ids = res.pdu_ids.iter().map(AsRef::as_ref); - let state_vec = self - .fetch_and_handle_outliers(origin, state_ids, create_event, room_id) - .boxed() - .await; - - let mut state: HashMap = HashMap::with_capacity(state_vec.len()); - for (pdu, _) in state_vec { - let state_key = pdu - .state_key() - .ok_or_else(|| err!(Database("Found non-state pdu in state events.")))?; - - let shortstatekey = self +impl super::Service { + /// Call /state_ids to find out what the state at this pdu is. We trust the + /// server's response to some extent (sic), but we still do a lot of checks + /// on the events + #[tracing::instrument( + level = "debug", + skip_all, + fields(%origin), + )] + pub(super) async fn fetch_state( + &self, + origin: &ServerName, + create_event: &Pdu, + room_id: &RoomId, + event_id: &EventId, + ) -> Result>> + where + Pdu: Event + Send + Sync, + { + let res = self .services - .short - .get_or_create_shortstatekey(&pdu.kind().to_string().into(), state_key) + .sending + .send_federation_request( + origin, + get_room_state_ids::v1::Request::new(event_id.to_owned(), room_id.to_owned()), + ) + .await + .inspect_err(|e| debug_warn!("Fetching state for event failed: {e}"))?; + + debug!("Fetching state events"); + let state_ids = res.pdu_ids.iter().map(AsRef::as_ref); + let state_vec = self + .fetch_and_handle_outliers(origin, state_ids, create_event, room_id) + .boxed() .await; - match state.entry(shortstatekey) { - | hash_map::Entry::Vacant(v) => { - v.insert(pdu.event_id().to_owned()); - }, - | hash_map::Entry::Occupied(_) => { - return Err!(Database( - "State event's type and state_key combination exists multiple times: {}, {}", - pdu.kind(), - state_key - )); - }, + let mut state: HashMap = + HashMap::with_capacity(state_vec.len()); + for (pdu, _) in state_vec { + let state_key = pdu + .state_key() + .ok_or_else(|| err!(Database("Found non-state pdu in state events.")))?; + + let shortstatekey = self + .services + .short + .get_or_create_shortstatekey(&pdu.kind().to_string().into(), state_key) + .await; + + match state.entry(shortstatekey) { + | hash_map::Entry::Vacant(v) => { + v.insert(pdu.event_id().to_owned()); + }, + | hash_map::Entry::Occupied(_) => { + return Err!(Database( + "State event's type and state_key combination exists multiple times: \ + {}, {}", + pdu.kind(), + state_key + )); + }, + } } + + // The original create event must still be in the state + let create_shortstatekey = self + .services + .short + .get_shortstatekey(&StateEventType::RoomCreate, "") + .await?; + + if state.get(&create_shortstatekey).map(AsRef::as_ref) != Some(create_event.event_id()) { + return Err!(Database("Incoming event refers to wrong create event.")); + } + + Ok(Some(state)) } - - // The original create event must still be in the state - let create_shortstatekey = self - .services - .short - .get_shortstatekey(&StateEventType::RoomCreate, "") - .await?; - - if state.get(&create_shortstatekey).map(AsRef::as_ref) != Some(create_event.event_id()) { - return Err!(Database("Incoming event refers to wrong create event.")); - } - - Ok(Some(state)) } diff --git a/src/service/rooms/event_handler/handle_incoming_pdu.rs b/src/service/rooms/event_handler/handle_incoming_pdu.rs index 204dfa715..d59ecf69e 100644 --- a/src/service/rooms/event_handler/handle_incoming_pdu.rs +++ b/src/service/rooms/event_handler/handle_incoming_pdu.rs @@ -5,7 +5,7 @@ use std::{ use conduwuit::{ Err, Event, PduEvent, Result, debug::INFO_SPAN_LEVEL, debug_error, debug_info, defer, err, - implement, info, trace, utils::stream::IterStream, warn, + info, trace, utils::stream::IterStream, warn, }; use futures::{ FutureExt, TryFutureExt, TryStreamExt, @@ -81,218 +81,233 @@ async fn should_rescind_invite( Ok(None) } -/// 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 -#[implement(super::Service)] -#[tracing::instrument( +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 + #[tracing::instrument( name = "pdu", level = INFO_SPAN_LEVEL, skip_all, fields(%room_id, %event_id), -)] -pub async fn handle_incoming_pdu<'a>( - &self, - origin: &'a ServerName, - room_id: &'a RoomId, - event_id: &'a EventId, - value: BTreeMap, - is_timeline_event: bool, -) -> Result> { - // 1. Skip the PDU if we already have it as a timeline event - if let Ok(pdu_id) = self.services.timeline.get_pdu_id(event_id).await { - 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); - - // 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(InvalidParam("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(()))), - ) - .await - .inspect_err(|e| debug_error!(%origin, "failed to handle incoming PDU {event_id}: {e}"))?; - - if is_disabled { - return Err!(Request(Forbidden("Federation of this room is disabled by this server."))); - } - - if !self - .services - .state_cache - .server_in_room(self.services.globals.server_name(), room_id) - .await - { - // Is this a federated invite rescind? - // copied from https://github.com/element-hq/synapse/blob/7e4588a/synapse/handlers/federation_event.py#L255-L300 - if value.get("type").and_then(|t| t.as_str()) == Some("m.room.member") { - if let Some(pdu) = - should_rescind_invite(&self.services, &mut value.clone(), &sender, room_id) - .await? - { - debug_info!( - "Invite to {room_id} appears to have been rescinded by {sender}, marking as \ - left" - ); - self.services - .state_cache - .mark_as_left(&sender, room_id, Some(pdu)) - .await; - return Ok(None); - } + )] + pub async fn handle_incoming_pdu<'a>( + &self, + origin: &'a ServerName, + room_id: &'a RoomId, + event_id: &'a EventId, + value: BTreeMap, + is_timeline_event: bool, + ) -> Result> { + // 1. Skip the PDU if we already have it as a timeline event + if let Ok(pdu_id) = self.services.timeline.get_pdu_id(event_id).await { + return Ok(Some(pdu_id)); } - info!( - %origin, - %room_id, - "Dropping inbound PDU for room we aren't participating in" + 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}" ); - return Err!(Request(NotFound("This server is not participating in that room."))); - } - if !meta_exists { - return Err!(Request(NotFound("Room is unknown to this server"))); - } + // 1.1 Check we even know about the room + let meta_exists = self.services.metadata.exists(room_id).map(Ok); - // Fetch create event - let create_event = &self - .services - .state_accessor - .get_room_create_event(room_id) - .await; + // 1.2 Check if the room is disabled + let is_disabled = self.services.metadata.is_disabled(room_id).map(Ok); - let start_time = Instant::now(); - self.federation_handletime - .write() - .insert(room_id.into(), (event_id.to_owned(), start_time)); + // 1.3.1 Check room ACL on origin field/server + let origin_acl_check = self.acl_check(origin, room_id); - defer! {{ + // 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(InvalidParam("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(()))), + ) + .await + .inspect_err( + |e| debug_error!(%origin, "failed to handle incoming PDU {event_id}: {e}"), + )?; + + if is_disabled { + return Err!(Request(Forbidden( + "Federation of this room is disabled by this server." + ))); + } + + if !self + .services + .state_cache + .server_in_room(self.services.globals.server_name(), room_id) + .await + { + // Is this a federated invite rescind? + // copied from https://github.com/element-hq/synapse/blob/7e4588a/synapse/handlers/federation_event.py#L255-L300 + if value.get("type").and_then(|t| t.as_str()) == Some("m.room.member") { + if let Some(pdu) = + should_rescind_invite(&self.services, &mut value.clone(), &sender, room_id) + .await? + { + debug_info!( + "Invite to {room_id} appears to have been rescinded by {sender}, \ + marking as left" + ); + self.services + .state_cache + .mark_as_left(&sender, room_id, Some(pdu)) + .await; + return Ok(None); + } + } + 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 { + return Err!(Request(NotFound("Room is unknown to this server"))); + } + + // Fetch create event + let create_event = &self + .services + .state_accessor + .get_room_create_event(room_id) + .await; + + let start_time = Instant::now(); self.federation_handletime .write() - .remove(room_id); - }}; + .insert(room_id.into(), (event_id.to_owned(), start_time)); - let (incoming_pdu, val) = self - .handle_outlier_pdu(origin, create_event, event_id, room_id, value, false) - .await?; + defer! {{ + self.federation_handletime + .write() + .remove(room_id); + }} - // 8. if not timeline event: stop - if !is_timeline_event { - return Ok(None); - } + let (incoming_pdu, val) = self + .handle_outlier_pdu(origin, create_event, event_id, room_id, value, false) + .await?; - // Skip old events - let first_ts_in_room = self - .services - .timeline - .first_pdu_in_room(room_id) - .await? - .origin_server_ts(); + // 8. if not timeline event: stop + if !is_timeline_event { + return Ok(None); + } - // 9. Fetch any missing prev events doing all checks listed here starting at 1. - // These are timeline events - let (sorted_prev_events, mut eventid_info) = self - .fetch_prev(origin, create_event, room_id, first_ts_in_room, incoming_pdu.prev_events()) - .await?; + // Skip old events + let first_ts_in_room = self + .services + .timeline + .first_pdu_in_room(room_id) + .await? + .origin_server_ts(); - debug!( - events = ?sorted_prev_events, - "Handling previous events" - ); - - sorted_prev_events - .iter() - .try_stream() - .map_ok(AsRef::as_ref) - .try_for_each(|prev_id| { - self.handle_prev_pdu( + // 9. Fetch any missing prev events doing all checks listed here starting at 1. + // These are timeline events + let (sorted_prev_events, mut eventid_info) = self + .fetch_prev( origin, - event_id, - room_id, - eventid_info.remove(prev_id), create_event, + room_id, first_ts_in_room, - prev_id, + incoming_pdu.prev_events(), ) - .inspect_err(move |e| { - warn!("Prev {prev_id} failed: {e}"); - match self - .services - .globals - .bad_event_ratelimiter - .write() - .entry(prev_id.into()) - { - | hash_map::Entry::Vacant(e) => { - e.insert((Instant::now(), 1)); - }, - | hash_map::Entry::Occupied(mut e) => { - let tries = e.get().1.saturating_add(1); - *e.get_mut() = (Instant::now(), tries); - }, - } - }) - .map(|_| self.services.server.check_running()) - }) - .boxed() - .await?; + .await?; - // Done with prev events, now handling the incoming event - self.upgrade_outlier_to_timeline_pdu(incoming_pdu, val, create_event, origin, room_id) - .boxed() - .await + debug!( + events = ?sorted_prev_events, + "Handling previous events" + ); + + sorted_prev_events + .iter() + .try_stream() + .map_ok(AsRef::as_ref) + .try_for_each(|prev_id| { + self.handle_prev_pdu( + origin, + event_id, + room_id, + eventid_info.remove(prev_id), + create_event, + first_ts_in_room, + prev_id, + ) + .inspect_err(move |e| { + warn!("Prev {prev_id} failed: {e}"); + match self + .services + .globals + .bad_event_ratelimiter + .write() + .entry(prev_id.into()) + { + | hash_map::Entry::Vacant(e) => { + e.insert((Instant::now(), 1)); + }, + | hash_map::Entry::Occupied(mut e) => { + let tries = e.get().1.saturating_add(1); + *e.get_mut() = (Instant::now(), tries); + }, + } + }) + .map(|_| self.services.server.check_running()) + }) + .boxed() + .await?; + + // Done with prev events, now handling the incoming event + self.upgrade_outlier_to_timeline_pdu(incoming_pdu, val, create_event, origin, room_id) + .boxed() + .await + } } diff --git a/src/service/rooms/event_handler/handle_outlier_pdu.rs b/src/service/rooms/event_handler/handle_outlier_pdu.rs index dc08ded5e..ec344433c 100644 --- a/src/service/rooms/event_handler/handle_outlier_pdu.rs +++ b/src/service/rooms/event_handler/handle_outlier_pdu.rs @@ -1,8 +1,7 @@ use std::collections::{BTreeMap, HashMap, hash_map}; use conduwuit::{ - Err, Event, PduEvent, Result, debug, debug_info, debug_warn, err, implement, state_res, - trace, warn, + Err, Event, PduEvent, Result, debug, debug_info, debug_warn, err, state_res, trace, warn, }; use futures::future::ready; use ruma::{ @@ -13,224 +12,232 @@ use ruma::{ use super::{check_room_id, get_room_version_rules}; use crate::rooms::timeline::pdu_fits; -#[implement(super::Service)] -#[allow(clippy::too_many_arguments)] -pub(super) async fn handle_outlier_pdu<'a, Pdu>( - &self, - origin: &'a ServerName, - create_event: &'a Pdu, - event_id: &'a EventId, - room_id: &'a RoomId, - mut value: CanonicalJsonObject, - auth_events_known: bool, -) -> Result<(PduEvent, BTreeMap)> -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." - ); - return Err!(Request(TooLarge("PDU is too large"))); - } - // 1. Remove unsigned field - value.remove("unsigned"); - - // 2. Check signatures, otherwise drop - // 3. check content hash, redact if doesn't match - 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 +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. + #[allow(clippy::too_many_arguments)] + pub(super) async fn handle_outlier_pdu<'a, Pdu>( + &self, + origin: &'a ServerName, + create_event: &'a Pdu, + event_id: &'a EventId, + room_id: &'a RoomId, + mut value: CanonicalJsonObject, + auth_events_known: bool, + ) -> Result<(PduEvent, BTreeMap)> + where + Pdu: Event + Send + Sync, { - | Ok(ruma::signatures::Verified::All) => value, - | Ok(ruma::signatures::Verified::Signatures) => { - // Redact - debug_info!("Calculated hash does not match (redaction): {event_id}"); - let Ok(obj) = - ruma::canonical_json::redact(value, &room_version_rules.redaction, None) - else { - return Err!(Request(InvalidParam("Redaction failed"))); - }; - - // Skip the PDU if it is redacted and we already have it as an outlier event - if self.services.timeline.pdu_exists(event_id).await { - return Err!(Request(InvalidParam( - "Event was redacted and we already knew about it" - ))); - } - - obj - }, - | Err(e) => { - return Err!(Request(InvalidParam(debug_error!( - "Signature verification failed for {event_id}: {e}" - )))); - }, - }; - - // Now that we have checked the signature and hashes we can add the eventID and - // convert to our PduEvent type - incoming_pdu - .insert("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)?; - - // Fetch all auth events - let mut auth_events: HashMap = HashMap::new(); - - for aid in pdu_event.auth_events() { - if self.services.pdu_metadata.is_event_rejected(aid).await { - debug_warn!( - "Rejecting incoming event {} which depends on rejected auth event {aid}", - event_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." ); - self.services.pdu_metadata.mark_event_rejected(event_id); - return Err!(Request(InvalidParam("Event has rejected auth event: {aid}"))); + return Err!(Request(TooLarge("PDU is too large"))); } + // 1. Remove unsigned field + value.remove("unsigned"); - if let Ok(auth_event) = self.services.timeline.get_pdu(aid).await { - check_room_id(room_id, &auth_event)?; - trace!("Found auth event {aid} for outlier event {event_id} locally"); - auth_events.insert(aid.to_owned(), auth_event); - } else { - debug_warn!("Could not find auth event {aid} for outlier event {event_id} locally"); - } - } - - // Fetch any missing ones & reject invalid ones - let missing_auth_events = if auth_events_known { - pdu_event - .auth_events() - .filter(|id| !auth_events.contains_key(*id)) - .collect::>() - } else { - pdu_event.auth_events().collect::>() - }; - if !missing_auth_events.is_empty() || !auth_events_known { - debug_info!( - "Fetching {} missing auth events for outlier event {event_id}", - missing_auth_events.len() - ); - for (pdu, _) in self - .fetch_and_handle_outliers( - origin, - missing_auth_events.iter().copied(), - create_event, - room_id, - ) + // 2. Check signatures, otherwise drop + // 3. check content hash, redact if doesn't match + 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 { - auth_events.insert(pdu.event_id().to_owned(), pdu); - } - } else { - debug!("No missing auth events for outlier event {event_id}"); - } - // reject if we are still missing some - let still_missing = pdu_event - .auth_events() - .filter(|id| !auth_events.contains_key(*id)) - .collect::>(); - if !still_missing.is_empty() { - // Don't reject: this could be a temporary condition - // TODO: use get_missing_events? - return Err!(Request(InvalidParam( - "Could not fetch all auth events for outlier event {event_id}, still missing: \ - {still_missing:?}" - ))); - } + | Ok(ruma::signatures::Verified::All) => value, + | Ok(ruma::signatures::Verified::Signatures) => { + // Redact + debug_info!("Calculated hash does not match (redaction): {event_id}"); + let Ok(obj) = + ruma::canonical_json::redact(value, &room_version_rules.redaction, None) + else { + return Err!(Request(InvalidParam("Redaction failed"))); + }; - // 6. Reject "due to auth events" if the event doesn't pass auth based on the - // auth events - debug!("Checking based on auth events"); - let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(auth_events.len()); - // Build map of auth events - for id in pdu_event.auth_events() { - let auth_event = auth_events - .get(id) - .expect("we just checked that we have all auth events") - .to_owned(); + // Skip the PDU if it is redacted and we already have it as an outlier event + if self.services.timeline.pdu_exists(event_id).await { + return Err!(Request(InvalidParam( + "Event was redacted and we already knew about it" + ))); + } - check_room_id(room_id, &auth_event)?; - - match auth_events_by_key.entry(( - auth_event.kind.to_string().into(), - auth_event - .state_key - .clone() - .expect("all auth events have state keys"), - )) { - | hash_map::Entry::Vacant(v) => { - v.insert(auth_event); + obj }, - | hash_map::Entry::Occupied(_) => { - self.services - .outlier - .add_pdu_outlier(pdu_event.event_id(), &incoming_pdu); + | Err(e) => { + return Err!(Request(InvalidParam(debug_error!( + "Signature verification failed for {event_id}: {e}" + )))); + }, + }; + + // Now that we have checked the signature and hashes we can add the eventID and + // convert to our PduEvent type + incoming_pdu.insert( + "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)?; + + // Fetch all auth events + let mut auth_events: HashMap = HashMap::new(); + + for aid in pdu_event.auth_events() { + if self.services.pdu_metadata.is_event_rejected(aid).await { + debug_warn!( + "Rejecting incoming event {} which depends on rejected auth event {aid}", + event_id, + ); self.services.pdu_metadata.mark_event_rejected(event_id); - return Err!(Request(InvalidParam( - "Auth event's type and state_key combination exists multiple times: {}, {}", - auth_event.kind, - auth_event.state_key().unwrap_or("") - ))); - }, + return Err!(Request(InvalidParam("Event has rejected auth event: {aid}"))); + } + + if let Ok(auth_event) = self.services.timeline.get_pdu(aid).await { + check_room_id(room_id, &auth_event)?; + trace!("Found auth event {aid} for outlier event {event_id} locally"); + auth_events.insert(aid.to_owned(), auth_event); + } else { + debug_warn!( + "Could not find auth event {aid} for outlier event {event_id} locally" + ); + } } - } - // The original create event must be in the auth events - if !matches!( - auth_events_by_key.get(&(StateEventType::RoomCreate, String::new().into())), - Some(_) | None - ) { - self.services.pdu_metadata.mark_event_rejected(event_id); + // Fetch any missing ones & reject invalid ones + let missing_auth_events = if auth_events_known { + pdu_event + .auth_events() + .filter(|id| !auth_events.contains_key(*id)) + .collect::>() + } else { + pdu_event.auth_events().collect::>() + }; + if !missing_auth_events.is_empty() || !auth_events_known { + debug_info!( + "Fetching {} missing auth events for outlier event {event_id}", + missing_auth_events.len() + ); + for (pdu, _) in self + .fetch_and_handle_outliers( + origin, + missing_auth_events.iter().copied(), + create_event, + room_id, + ) + .await + { + auth_events.insert(pdu.event_id().to_owned(), pdu); + } + } else { + debug!("No missing auth events for outlier event {event_id}"); + } + // reject if we are still missing some + let still_missing = pdu_event + .auth_events() + .filter(|id| !auth_events.contains_key(*id)) + .collect::>(); + if !still_missing.is_empty() { + // Don't reject: this could be a temporary condition + // TODO: use get_missing_events? + return Err!(Request(InvalidParam( + "Could not fetch all auth events for outlier event {event_id}, still missing: \ + {still_missing:?}" + ))); + } + + // 6. Reject "due to auth events" if the event doesn't pass auth based on the + // auth events + debug!("Checking based on auth events"); + let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(auth_events.len()); + // Build map of auth events + for id in pdu_event.auth_events() { + let auth_event = auth_events + .get(id) + .expect("we just checked that we have all auth events") + .to_owned(); + + check_room_id(room_id, &auth_event)?; + + match auth_events_by_key.entry(( + auth_event.kind.to_string().into(), + auth_event + .state_key + .clone() + .expect("all auth events have state keys"), + )) { + | hash_map::Entry::Vacant(v) => { + 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(InvalidParam( + "Auth event's type and state_key combination exists multiple times: {}, \ + {}", + auth_event.kind, + auth_event.state_key().unwrap_or("") + ))); + }, + } + } + + // The original create event must be in the auth events + if !matches!( + auth_events_by_key.get(&(StateEventType::RoomCreate, String::new().into())), + Some(_) | None + ) { + self.services.pdu_metadata.mark_event_rejected(event_id); + self.services + .outlier + .add_pdu_outlier(pdu_event.event_id(), &incoming_pdu); + return Err!(Request(InvalidParam("Incoming event refers to wrong create event."))); + } + + let state_fetch = |ty: &StateEventType, sk: &str| { + let key = (ty.to_owned(), sk.into()); + ready(auth_events_by_key.get(&key).map(ToOwned::to_owned)) + }; + + // PDU check: 3 + let auth_check = state_res::event_auth::auth_check( + &room_version_rules, + &pdu_event, + None, // TODO: third party invite + state_fetch, + 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(event_id); + self.services + .outlier + .add_pdu_outlier(pdu_event.event_id(), &incoming_pdu); + return Err!(Request(Forbidden( + "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); - return Err!(Request(InvalidParam("Incoming event refers to wrong create event."))); + + trace!("Added pdu as outlier."); + + Ok((pdu_event, incoming_pdu)) } - - 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)) - }; - - // PDU check: 3 - let auth_check = state_res::event_auth::auth_check( - &room_version_rules, - &pdu_event, - None, // TODO: third party invite - state_fetch, - 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(event_id); - self.services - .outlier - .add_pdu_outlier(pdu_event.event_id(), &incoming_pdu); - return Err!(Request(Forbidden( - "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."); - - Ok((pdu_event, incoming_pdu)) } diff --git a/src/service/rooms/event_handler/handle_prev_pdu.rs b/src/service/rooms/event_handler/handle_prev_pdu.rs index cb4978d94..6eb0d1383 100644 --- a/src/service/rooms/event_handler/handle_prev_pdu.rs +++ b/src/service/rooms/event_handler/handle_prev_pdu.rs @@ -1,89 +1,95 @@ use std::{collections::BTreeMap, time::Instant}; use conduwuit::{ - Err, Event, PduEvent, Result, debug::INFO_SPAN_LEVEL, defer, implement, + Err, Event, PduEvent, Result, debug::INFO_SPAN_LEVEL, defer, utils::continue_exponential_backoff_secs, }; use ruma::{CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch, RoomId, ServerName}; use tracing::debug; -#[implement(super::Service)] -#[allow(clippy::type_complexity)] -#[allow(clippy::too_many_arguments)] -#[tracing::instrument( +impl super::Service { + #[allow(clippy::type_complexity)] + #[allow(clippy::too_many_arguments)] + #[tracing::instrument( name = "prev", level = INFO_SPAN_LEVEL, skip_all, fields(%prev_id), -)] -pub(super) async fn handle_prev_pdu<'a, Pdu>( - &self, - origin: &'a ServerName, - event_id: &'a EventId, - room_id: &'a RoomId, - eventid_info: Option<(PduEvent, BTreeMap)>, - create_event: &'a Pdu, - first_ts_in_room: MilliSecondsSinceUnixEpoch, - prev_id: &'a EventId, -) -> Result -where - Pdu: Event + Send + Sync, -{ - // Check for disabled again because it might have changed - if self.services.metadata.is_disabled(room_id).await { - return Err!(Request(Forbidden(debug_warn!( - "Federaton of room {room_id} is currently disabled on this server. Request by \ - origin {origin} and event ID {event_id}" - )))); - } - - if let Some((time, tries)) = self - .services - .globals - .bad_event_ratelimiter - .read() - .get(prev_id) + )] + pub(super) async fn handle_prev_pdu<'a, Pdu>( + &self, + origin: &'a ServerName, + event_id: &'a EventId, + room_id: &'a RoomId, + eventid_info: Option<(PduEvent, BTreeMap)>, + create_event: &'a Pdu, + first_ts_in_room: MilliSecondsSinceUnixEpoch, + prev_id: &'a EventId, + ) -> Result + where + Pdu: Event + Send + Sync, { - // Exponential backoff - const MIN_DURATION: u64 = 5 * 60; - const MAX_DURATION: u64 = 60 * 60 * 24; - if continue_exponential_backoff_secs(MIN_DURATION, MAX_DURATION, time.elapsed(), *tries) { - debug!( - ?tries, - duration = ?time.elapsed(), - "Backing off from prev_event" - ); + // Check for disabled again because it might have changed + if self.services.metadata.is_disabled(room_id).await { + return Err!(Request(Forbidden(debug_warn!( + "Federaton of room {room_id} is currently disabled on this server. Request by \ + origin {origin} and event ID {event_id}" + )))); + } + + if let Some((time, tries)) = self + .services + .globals + .bad_event_ratelimiter + .read() + .get(prev_id) + { + // Exponential backoff + const MIN_DURATION: u64 = 5 * 60; + const MAX_DURATION: u64 = 60 * 60 * 24; + if continue_exponential_backoff_secs( + MIN_DURATION, + MAX_DURATION, + time.elapsed(), + *tries, + ) { + debug!( + ?tries, + duration = ?time.elapsed(), + "Backing off from prev_event" + ); + return Ok(()); + } + } + + let Some((pdu, json)) = eventid_info else { + return Ok(()); + }; + + // Skip old events + if pdu.origin_server_ts() < first_ts_in_room { return Ok(()); } - } - let Some((pdu, json)) = eventid_info else { - return Ok(()); - }; - - // Skip old events - if pdu.origin_server_ts() < first_ts_in_room { - return Ok(()); - } - - let start_time = Instant::now(); - self.federation_handletime - .write() - .insert(room_id.into(), ((*prev_id).to_owned(), start_time)); - - defer! {{ + let start_time = Instant::now(); self.federation_handletime .write() - .remove(room_id); - }}; + .insert(room_id.into(), ((*prev_id).to_owned(), start_time)); - self.upgrade_outlier_to_timeline_pdu(pdu, json, create_event, origin, room_id) - .await?; + defer! {{ + self.federation_handletime + .write() + .remove(room_id); + }}; - debug!( - elapsed = ?start_time.elapsed(), - "Handled prev_event", - ); + self.upgrade_outlier_to_timeline_pdu(pdu, json, create_event, origin, room_id) + .await?; - Ok(()) + debug!( + elapsed = ?start_time.elapsed(), + "Handled prev_event", + ); + + Ok(()) + } } diff --git a/src/service/rooms/event_handler/mod.rs b/src/service/rooms/event_handler/mod.rs index 6197bc87e..8a1922652 100644 --- a/src/service/rooms/event_handler/mod.rs +++ b/src/service/rooms/event_handler/mod.rs @@ -94,10 +94,13 @@ impl crate::Service for Service { } impl Service { + /// Checks if a single event exists. Alias for + /// `self.services.timeline.pdu_exists`. async fn event_exists(&self, event_id: OwnedEventId) -> bool { self.services.timeline.pdu_exists(&event_id).await } + /// Fetches a single PDU, returning None if there is an error. async fn event_fetch(&self, event_id: OwnedEventId) -> Option { self.services.timeline.get_pdu(&event_id).await.ok() } diff --git a/src/service/rooms/event_handler/parse_incoming_pdu.rs b/src/service/rooms/event_handler/parse_incoming_pdu.rs index b4af6b4a8..6726f4bd3 100644 --- a/src/service/rooms/event_handler/parse_incoming_pdu.rs +++ b/src/service/rooms/event_handler/parse_incoming_pdu.rs @@ -1,7 +1,7 @@ use std::str::FromStr; use conduwuit::{ - Err, Result, err, implement, + Err, Result, err, matrix::event::{gen_event_id, gen_event_id_canonical_json}, }; use itertools::Itertools; @@ -76,55 +76,57 @@ fn expect_event_id_array(value: &CanonicalJsonObject, field: &str) -> 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"))); +impl super::Service { + /// Performs some basic validation on the PDU to make sure it's not + /// obviously malformed. This is not a full validation, but guards against + /// extreme errors. + /// + /// Currently, this just validates that prev/auth events are within + /// acceptable ranges. Other servers do some additional things like + /// checking depth range, but serde will do that later when converting the + /// object to a PduEvent. + pub fn validate_pdu(&self, pdu: &CanonicalJsonObject) -> 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(()) } - 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(()) -} -#[implement(super::Service)] -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:?}"))) - })?; - let event_type = value - .get("type") - .and_then(CanonicalJsonValue::as_str) - .ok_or_else(|| err!(Request(InvalidParam("Missing or invalid type in pdu"))))?; - - let room_id = extract_room_id(event_type, &value)?; - - let room_version_rules = self - .services - .state - .get_room_version(&room_id) - .await - .unwrap_or(RoomVersionId::V1) - .rules() - .unwrap(); - - let (event_id, value) = - gen_event_id_canonical_json(pdu, &room_version_rules).map_err(|e| { - err!(Request(InvalidParam("Could not convert event to canonical json: {e}"))) + 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:?}"))) })?; - self.validate_pdu(&value)?; - Ok((room_id, event_id, value)) + let event_type = value + .get("type") + .and_then(CanonicalJsonValue::as_str) + .ok_or_else(|| err!(Request(InvalidParam("Missing or invalid type in pdu"))))?; + + let room_id = extract_room_id(event_type, &value)?; + + let room_version_rules = self + .services + .state + .get_room_version(&room_id) + .await + .unwrap_or(RoomVersionId::V1) + .rules() + .unwrap(); + + let (event_id, value) = + 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)?; + Ok((room_id, event_id, value)) + } } diff --git a/src/service/rooms/event_handler/policy_server.rs b/src/service/rooms/event_handler/policy_server.rs index cdd142277..330d48ef6 100644 --- a/src/service/rooms/event_handler/policy_server.rs +++ b/src/service/rooms/event_handler/policy_server.rs @@ -6,8 +6,8 @@ use std::{collections::BTreeMap, time::Duration}; use conduwuit::{ - Err, Error, Event, PduEvent, Result, debug, debug_error, debug_info, debug_warn, error, - implement, info, state_res::EventTypeExt, trace, utils::to_canonical_object, warn, + Err, Error, Event, PduEvent, Result, debug, debug_error, debug_info, debug_warn, error, info, + state_res::EventTypeExt, trace, utils::to_canonical_object, warn, }; use http::StatusCode; use ruma::{ @@ -93,348 +93,351 @@ pub(super) fn extract_signature( .and_then(|signature| Base64::::parse(signature).ok()) } -/// Verifies the existing policy server signature, and/or fetches a new one -/// immediately. -/// -/// If `incoming` is `true`, the event is checked for an existing signature. If -/// it has a valid one, `Ok` is returned. If it does not have a valid signature, -/// the function falls through to fetching a new one (which may be a soft-fail -/// in a future version). -/// -/// If the event is the `m.room.policy` configuration state event, -/// this check is skipped. Similarly, if there is no policy server configured in -/// the PDU's room, the configuration event is malformed, or the configured -/// server is not present in the room, the check is also skipped. -/// -/// If the policy server marks the event as spam, the relevant error is -/// returned. Otherwise, the incoming PDU JSON is mutated to include the new -/// policy server signature. Transient errors such as rate-limits are handled, -/// so any error returned by this function should be treated as final. -#[implement(super::Service)] -#[tracing::instrument(skip(self, pdu, pdu_json, room_version_rules), level = "info")] -pub async fn policy_server_allows_event( - &self, - pdu: &PduEvent, - pdu_json: &mut CanonicalJsonObject, - room_id: &RoomId, - room_version_rules: &RoomVersionRules, - incoming: bool, -) -> Result<()> { - assert!( - !pdu_json.contains_key("event_id"), - "event_id should be removed from the JSON before calling policy_server_allows_event" - ); - if pdu.event_type().with_state_key("") == (StateEventType::RoomPolicy, "".into()) { - return Ok(()); - } - let ps = match self - .services - .state_accessor - .room_state_get_content::( - room_id, - &StateEventType::RoomPolicy, - "", - ) - .await - { - | Ok(s) => s, - | Err(e) => - return if e.is_not_found() || e.kind() == ErrorKind::BadJson { - debug!(%e, "no policy server configured"); - Ok(()) - } else { - Err!("failed to load m.room.policy state event: {e}") - }, - }; - - let Some(ps_key) = ps.public_keys.get(&SigningKeyAlgorithm::Ed25519) else { - debug!( - "room has a policy server configured, but no valid public keys; skipping spam check" +impl super::Service { + /// Verifies the existing policy server signature, and/or fetches a new one + /// immediately. + /// + /// If `incoming` is `true`, the event is checked for an existing signature. + /// If it has a valid one, `Ok` is returned. If it does not have a valid + /// signature, the function falls through to fetching a new one (which may + /// be a soft-fail in a future version). + /// + /// If the event is the `m.room.policy` configuration state event, + /// this check is skipped. Similarly, if there is no policy server + /// configured in the PDU's room, the configuration event is malformed, or + /// the configured server is not present in the room, the check is also + /// skipped. + /// + /// If the policy server marks the event as spam, the relevant error is + /// returned. Otherwise, the incoming PDU JSON is mutated to include the new + /// policy server signature. Transient errors such as rate-limits are + /// handled, so any error returned by this function should be treated as + /// final. + #[tracing::instrument(skip(self, pdu, pdu_json, room_version_rules), level = "info")] + pub async fn policy_server_allows_event( + &self, + pdu: &PduEvent, + pdu_json: &mut CanonicalJsonObject, + room_id: &RoomId, + room_version_rules: &RoomVersionRules, + incoming: bool, + ) -> Result<()> { + assert!( + !pdu_json.contains_key("event_id"), + "event_id should be removed from the JSON before calling policy_server_allows_event" ); - return Ok(()); - }; + if pdu.event_type().with_state_key("") == (StateEventType::RoomPolicy, "".into()) { + return Ok(()); + } + let ps = match self + .services + .state_accessor + .room_state_get_content::( + room_id, + &StateEventType::RoomPolicy, + "", + ) + .await + { + | Ok(s) => s, + | Err(e) => + return if e.is_not_found() || e.kind() == ErrorKind::BadJson { + debug!(%e, "no policy server configured"); + Ok(()) + } else { + Err!("failed to load m.room.policy state event: {e}") + }, + }; - if !self - .services - .state_cache - .server_in_room(&ps.via, room_id) - .await - { - debug!( - via = %ps.via, - "Policy server is not in the room, skipping spam check" - ); - return Ok(()); - } + let Some(ps_key) = ps.public_keys.get(&SigningKeyAlgorithm::Ed25519) else { + debug!( + "room has a policy server configured, but no valid public keys; skipping spam \ + check" + ); + return Ok(()); + }; - if incoming { - if verify_policy_signature(&ps.via, ps_key, pdu_json, &room_version_rules.redaction) { + if !self + .services + .state_cache + .server_in_room(&ps.via, room_id) + .await + { debug!( via = %ps.via, - "Event is incoming and has a valid policy server signature" + "Policy server is not in the room, skipping spam check" ); return Ok(()); } - // N.B. In a future room version, this will be a soft failure specifically. - debug_info!( - via = %ps.via, - "Event is incoming but does not have a valid policy server signature; asking policy \ - server to sign it now" - ); - } - if ps.via == self.services.globals.server_name() - && !self.services.server.config.federation_loopback - { - error!( - %ps.via, - %room_id, - "Cannot ask ourselves for a policy signature if `federation_loopback=false`", - ); - return Ok(()); - } - - let outgoing = self - .services - .sending - .convert_to_outgoing_federation_event(pdu_json.clone()) - .await; - - debug_info!( - %ps.via, - "Asking policy server to sign event" - ); - if let Err(e) = self - .fetch_policy_server_signature(pdu, pdu_json, &ps.via, outgoing, room_id, ps_key, 0) - .await - { - if e.is_not_found() { - return Ok(()); - } - return Err(e); - } - trace!( - "Got successful response for fetching PS signature, ensuring it is signed with the \ - expected key." - ); - if verify_policy_signature(&ps.via, ps_key, pdu_json, &room_version_rules.redaction) { - Ok(()) - } else if incoming { - Err!(Request(Forbidden("Policy server signature is invalid"))) - } else { - Err(Error::Request( - ErrorKind::Unknown, - "Policy server signature is invalid".into(), - StatusCode::BAD_GATEWAY, - )) - } -} - -/// Handles an error returned by the policy server. If the error is one that -/// should be returned to the user, it is propagated, otherwise the request may -/// be retried (for example, when rate-limited). -#[allow(clippy::too_many_arguments)] -#[implement(super::Service)] -async fn handle_policy_server_error( - &self, - error: Error, - pdu: &PduEvent, - pdu_json: &mut CanonicalJsonObject, - via: &ServerName, - outgoing: Box, - room_id: &RoomId, - policy_server_key: &Base64>, - retries: u8, - timeout: Duration, -) -> Result<()> { - match error.status_code() { - | StatusCode::OK => unreachable!("ok response passed to handle_policy_server_error"), - | StatusCode::BAD_REQUEST => { - if matches!(error.kind(), ErrorKind::Forbidden) { - warn!( - via = %via, - event_id = %pdu.event_id(), - %room_id, - error = ?error, - "Policy server marked the event as spam" + if incoming { + if verify_policy_signature(&ps.via, ps_key, pdu_json, &room_version_rules.redaction) { + debug!( + via = %ps.via, + "Event is incoming and has a valid policy server signature" ); - return Err(error); + return Ok(()); } - error!( - via = %via, - event_id = %pdu.event_id(), - %room_id, - error = ?error.to_string(), - "Policy server could not understand our request", - ); - Err!(BadServerResponse("Error communicating with policy server")) - }, - | StatusCode::FORBIDDEN => { - Err!(Request(Forbidden( - "Policy server refused to sign the event due to the room ACL" - ))) - }, - | StatusCode::NOT_FOUND => { + // N.B. In a future room version, this will be a soft failure specifically. debug_info!( - via = %via, - event_id = %pdu.event_id(), - %room_id, - "Policy server is not actually a policy server or is not protecting this room: {}", - error.message() + via = %ps.via, + "Event is incoming but does not have a valid policy server signature; asking policy \ + server to sign it now" ); - Err(error) - }, - | StatusCode::TOO_MANY_REQUESTS => { - if let Some(retry_after) = error.retry_after() { - if retries >= 5 { + } + + if ps.via == self.services.globals.server_name() + && !self.services.server.config.federation_loopback + { + error!( + %ps.via, + %room_id, + "Cannot ask ourselves for a policy signature if `federation_loopback=false`", + ); + return Ok(()); + } + + let outgoing = self + .services + .sending + .convert_to_outgoing_federation_event(pdu_json.clone()) + .await; + + debug_info!( + %ps.via, + "Asking policy server to sign event" + ); + if let Err(e) = self + .fetch_policy_server_signature(pdu, pdu_json, &ps.via, outgoing, room_id, ps_key, 0) + .await + { + if e.is_not_found() { + return Ok(()); + } + return Err(e); + } + trace!( + "Got successful response for fetching PS signature, ensuring it is signed with the \ + expected key." + ); + if verify_policy_signature(&ps.via, ps_key, pdu_json, &room_version_rules.redaction) { + Ok(()) + } else if incoming { + Err!(Request(Forbidden("Policy server signature is invalid"))) + } else { + Err(Error::Request( + ErrorKind::Unknown, + "Policy server signature is invalid".into(), + StatusCode::BAD_GATEWAY, + )) + } + } + + /// Handles an error returned by the policy server. If the error is one that + /// should be returned to the user, it is propagated, otherwise the request + /// may be retried (for example, when rate-limited). + #[allow(clippy::too_many_arguments)] + async fn handle_policy_server_error( + &self, + error: Error, + pdu: &PduEvent, + pdu_json: &mut CanonicalJsonObject, + via: &ServerName, + outgoing: Box, + room_id: &RoomId, + policy_server_key: &Base64>, + retries: u8, + timeout: Duration, + ) -> Result<()> { + match error.status_code() { + | StatusCode::OK => unreachable!("ok response passed to handle_policy_server_error"), + | StatusCode::BAD_REQUEST => { + if matches!(error.kind(), ErrorKind::Forbidden) { warn!( via = %via, event_id = %pdu.event_id(), - room_id = %room_id, - retries, - "Policy server rate-limited us too many times; giving up" + %room_id, + error = ?error, + "Policy server marked the event as spam" ); - return Err(error); // Error should be passed to c2s + return Err(error); } - let saturated = retry_after.min(timeout); - // ^ don't wait more than 60 seconds - info!( + error!( + via = %via, + event_id = %pdu.event_id(), + %room_id, + error = ?error.to_string(), + "Policy server could not understand our request", + ); + Err!(BadServerResponse("Error communicating with policy server")) + }, + | StatusCode::FORBIDDEN => { + Err!(Request(Forbidden( + "Policy server refused to sign the event due to the room ACL" + ))) + }, + | StatusCode::NOT_FOUND => { + debug_info!( + via = %via, + event_id = %pdu.event_id(), + %room_id, + "Policy server is not actually a policy server or is not protecting this room: {}", + error.message() + ); + Err(error) + }, + | StatusCode::TOO_MANY_REQUESTS => { + if let Some(retry_after) = error.retry_after() { + if retries >= 5 { + warn!( + via = %via, + event_id = %pdu.event_id(), + room_id = %room_id, + retries, + "Policy server rate-limited us too many times; giving up" + ); + return Err(error); // Error should be passed to c2s + } + let saturated = retry_after.min(timeout); + // ^ don't wait more than 60 seconds + info!( + via = %via, + event_id = %pdu.event_id(), + room_id = %room_id, + retry_after = %saturated.as_secs(), + retries, + "Policy server rate-limited us; retrying after {retry_after:?}" + ); + tokio::select! { + () = self.server_shutdown.notified() => (), + () = sleep(saturated) => (), + } + if !self.services.server.running() { + return Err(error); + } + return Box::pin(self.fetch_policy_server_signature( + pdu, + pdu_json, + via, + outgoing, + room_id, + policy_server_key, + retries.saturating_add(1), + )) + .await; + } + warn!( via = %via, event_id = %pdu.event_id(), room_id = %room_id, - retry_after = %saturated.as_secs(), retries, - "Policy server rate-limited us; retrying after {retry_after:?}" + "Policy server rate-limited us without giving a retry window; giving up" ); - tokio::select! { - () = self.server_shutdown.notified() => (), - () = sleep(saturated) => (), - } - if !self.services.server.running() { - return Err(error); - } - return Box::pin(self.fetch_policy_server_signature( - pdu, - pdu_json, - via, - outgoing, - room_id, - policy_server_key, - retries.saturating_add(1), - )) - .await; - } - warn!( - via = %via, - event_id = %pdu.event_id(), - room_id = %room_id, - retries, - "Policy server rate-limited us without giving a retry window; giving up" + Err(error) + }, + | _ => Err!(BadServerResponse( + "Unexpected response from policy server: {}/{:?}", + error.status_code(), + error.kind() + )), + } + } + + /// Asks a remote policy server for a signature on this event. + /// If the policy server signs this event, the original data is mutated. + /// Otherwise, the error is handled and potentially returned. + #[allow(clippy::too_many_arguments)] + #[tracing::instrument(skip_all, fields(event_id=%pdu.event_id(), via=%via), level = "info")] + pub async fn fetch_policy_server_signature( + &self, + pdu: &PduEvent, + pdu_json: &mut CanonicalJsonObject, + via: &ServerName, + outgoing: Box, + room_id: &RoomId, + policy_server_key: &Base64>, + retries: u8, + ) -> Result<()> { + let timeout = + Duration::from_secs(self.services.server.config.policy_server_request_timeout); + debug!("Requesting policy server signature"); + let response = tokio::time::timeout( + timeout, + self.services + .sending + .send_federation_request(via, sign_event::v1::Request::new(outgoing.clone())), + ) + .await; + + let response = match response { + | Ok(Ok(response)) => response, + | Ok(Err(e)) => { + debug_error!("Error from policy server: {:?}", e); + return self + .handle_policy_server_error( + e, + pdu, + pdu_json, + via, + outgoing, + room_id, + policy_server_key, + retries, + timeout, + ) + .await; + }, + | Err(elapsed) => { + warn!( + %via, + event_id = %pdu.event_id(), + %room_id, + %elapsed, + "Policy server signature request timed out" + ); + return Err!(Request(Forbidden("Policy server did not respond in time"))); + }, + }; + + let Some(signatures) = response.signatures.get(via) else { + // NOTE: Legacy policy servers return a `200 {}` to indicate that the event was + // flagged as spam. We'll make a distinction in the error message in case + // it's unexpected. + return Err!(Request(Forbidden("Policy server did not sign the event"))); + }; + if response.signatures.len() > 1 { + warn!(?response.signatures, "Misbehaving policy server: returned signatures for extraneous servers"); + // TODO: This should return an error but doesn't because some + // servers do this. It's safe for us to not explode for now + // because we only care about the signature for `via`, but + // ideally we'd want to enforce this more strictly in the future. + } + + let Some(signature) = signatures.get( + &KeyId::parse(POLICY_SERVER_KEY_ID_ED25519).expect("policy server key ID is valid"), + ) else { + return Err!(BadServerResponse( + "Policy server did not return a signature with the expected key ID", + )); + }; + if signatures.len() > 1 { + info!(?signatures, "Misbehaving policy server: returned extraneous signatures"); + } + + pdu_json + .entry("signatures".to_owned()) + .or_insert_with(|| CanonicalJsonValue::Object(BTreeMap::default())) + .as_object_mut() + .expect("`signatures` field must be an object") + .entry(via.to_string()) + .or_insert_with(|| CanonicalJsonValue::Object(BTreeMap::default())) + .as_object_mut() + .expect("`signatures[via]` field must be an object") + .insert( + POLICY_SERVER_KEY_ID_ED25519.to_owned(), + CanonicalJsonValue::String(signature.clone()), ); - Err(error) - }, - | _ => Err!(BadServerResponse( - "Unexpected response from policy server: {}/{:?}", - error.status_code(), - error.kind() - )), + debug_info!("Policy server allowed event"); + Ok(()) } } - -/// Asks a remote policy server for a signature on this event. -/// If the policy server signs this event, the original data is mutated. -/// Otherwise, the error is handled and potentially returned. -#[allow(clippy::too_many_arguments)] -#[implement(super::Service)] -#[tracing::instrument(skip_all, fields(event_id=%pdu.event_id(), via=%via), level = "info")] -pub async fn fetch_policy_server_signature( - &self, - pdu: &PduEvent, - pdu_json: &mut CanonicalJsonObject, - via: &ServerName, - outgoing: Box, - room_id: &RoomId, - policy_server_key: &Base64>, - retries: u8, -) -> Result<()> { - let timeout = Duration::from_secs(self.services.server.config.policy_server_request_timeout); - debug!("Requesting policy server signature"); - let response = tokio::time::timeout( - timeout, - self.services - .sending - .send_federation_request(via, sign_event::v1::Request::new(outgoing.clone())), - ) - .await; - - let response = match response { - | Ok(Ok(response)) => response, - | Ok(Err(e)) => { - debug_error!("Error from policy server: {:?}", e); - return self - .handle_policy_server_error( - e, - pdu, - pdu_json, - via, - outgoing, - room_id, - policy_server_key, - retries, - timeout, - ) - .await; - }, - | Err(elapsed) => { - warn!( - %via, - event_id = %pdu.event_id(), - %room_id, - %elapsed, - "Policy server signature request timed out" - ); - return Err!(Request(Forbidden("Policy server did not respond in time"))); - }, - }; - - let Some(signatures) = response.signatures.get(via) else { - // NOTE: Legacy policy servers return a `200 {}` to indicate that the event was - // flagged as spam. We'll make a distinction in the error message in case - // it's unexpected. - return Err!(Request(Forbidden("Policy server did not sign the event"))); - }; - if response.signatures.len() > 1 { - warn!(?response.signatures, "Misbehaving policy server: returned signatures for extraneous servers"); - // TODO: This should return an error but doesn't because some servers do - // this. It's safe for us to not explode for now because we only care - // about the signature for `via`, but ideally we'd want to enforce - // this more strictly in the future. - } - - let Some(signature) = signatures - .get(&KeyId::parse(POLICY_SERVER_KEY_ID_ED25519).expect("policy server key ID is valid")) - else { - return Err!(BadServerResponse( - "Policy server did not return a signature with the expected key ID", - )); - }; - if signatures.len() > 1 { - info!(?signatures, "Misbehaving policy server: returned extraneous signatures"); - } - - pdu_json - .entry("signatures".to_owned()) - .or_insert_with(|| CanonicalJsonValue::Object(BTreeMap::default())) - .as_object_mut() - .expect("`signatures` field must be an object") - .entry(via.to_string()) - .or_insert_with(|| CanonicalJsonValue::Object(BTreeMap::default())) - .as_object_mut() - .expect("`signatures[via]` field must be an object") - .insert( - POLICY_SERVER_KEY_ID_ED25519.to_owned(), - CanonicalJsonValue::String(signature.clone()), - ); - debug_info!("Policy server allowed event"); - Ok(()) -} diff --git a/src/service/rooms/event_handler/resolve_state.rs b/src/service/rooms/event_handler/resolve_state.rs index 471876658..4f35ceecb 100644 --- a/src/service/rooms/event_handler/resolve_state.rs +++ b/src/service/rooms/event_handler/resolve_state.rs @@ -5,7 +5,7 @@ use std::{ }; use conduwuit::{ - Error, Result, err, implement, + Error, Result, err, state_res::{self, StateMap}, trace, utils::stream::{IterStream, ReadyExt, TryWidebandExt, WidebandExt}, @@ -15,110 +15,108 @@ use ruma::{OwnedEventId, RoomId, room_version_rules::RoomVersionRules}; use crate::rooms::state_compressor::CompressedState; -#[implement(super::Service)] -#[tracing::instrument(name = "resolve", level = "debug", skip_all)] -pub async fn resolve_state( - &self, - room_id: &RoomId, - room_version_rules: &RoomVersionRules, - incoming_state: HashMap, -) -> Result> { - trace!("Loading current room state ids"); - let current_sstatehash = self - .services - .state - .get_room_shortstatehash(room_id) - .map_err(|e| err!(Database(error!("No state for {room_id:?}: {e:?}")))) - .await?; +impl super::Service { + pub async fn resolve_state( + &self, + room_id: &RoomId, + room_version_rules: &RoomVersionRules, + incoming_state: HashMap, + ) -> Result> { + trace!("Loading current room state ids"); + let current_sstatehash = self + .services + .state + .get_room_shortstatehash(room_id) + .map_err(|e| err!(Database(error!("No state for {room_id:?}: {e:?}")))) + .await?; - let current_state_ids: HashMap<_, _> = self - .services - .state_accessor - .state_full_ids(current_sstatehash) - .collect() - .await; + let current_state_ids: HashMap<_, _> = self + .services + .state_accessor + .state_full_ids(current_sstatehash) + .collect() + .await; - trace!("Loading fork states"); - let fork_states = [current_state_ids, incoming_state]; - let auth_chain_sets = fork_states - .iter() - .try_stream() - .wide_and_then(|state| { - self.services - .auth_chain - .event_ids_iter(room_id, state.values().map(Borrow::borrow)) - .try_collect() - }) - .try_collect::>>(); + trace!("Loading fork states"); + let fork_states = [current_state_ids, incoming_state]; + let auth_chain_sets = fork_states + .iter() + .try_stream() + .wide_and_then(|state| { + self.services + .auth_chain + .event_ids_iter(room_id, state.values().map(Borrow::borrow)) + .try_collect() + }) + .try_collect::>>(); - let fork_states = fork_states - .iter() - .stream() - .wide_then(|fork_state| { - let shortstatekeys = fork_state.keys().copied().stream(); - let event_ids = fork_state.values().cloned().stream(); - self.services - .short - .multi_get_statekey_from_short(shortstatekeys) - .zip(event_ids) - .ready_filter_map(|(ty_sk, id)| Some((ty_sk.ok()?, id))) - .collect() - }) - .map(Ok::<_, Error>) - .try_collect::>>(); + let fork_states = fork_states + .iter() + .stream() + .wide_then(|fork_state| { + let shortstatekeys = fork_state.keys().copied().stream(); + let event_ids = fork_state.values().cloned().stream(); + self.services + .short + .multi_get_statekey_from_short(shortstatekeys) + .zip(event_ids) + .ready_filter_map(|(ty_sk, id)| Some((ty_sk.ok()?, id))) + .collect() + }) + .map(Ok::<_, Error>) + .try_collect::>>(); - let (fork_states, auth_chain_sets) = try_join(fork_states, auth_chain_sets).await?; + let (fork_states, auth_chain_sets) = try_join(fork_states, auth_chain_sets).await?; - trace!("Resolving state"); - let state = self - .state_resolution(room_version_rules, fork_states.iter(), &auth_chain_sets) - .boxed() - .await?; + trace!("Resolving state"); + let state = self + .state_resolution(room_version_rules, fork_states.iter(), &auth_chain_sets) + .boxed() + .await?; - trace!("State resolution done."); - let state_events: Vec<_> = state - .iter() - .stream() - .wide_then(|((event_type, state_key), event_id)| { - self.services - .short - .get_or_create_shortstatekey(event_type, state_key) - .map(move |shortstatekey| (shortstatekey, event_id)) - }) - .collect() - .await; + trace!("State resolution done."); + let state_events: Vec<_> = state + .iter() + .stream() + .wide_then(|((event_type, state_key), event_id)| { + self.services + .short + .get_or_create_shortstatekey(event_type, state_key) + .map(move |shortstatekey| (shortstatekey, event_id)) + }) + .collect() + .await; - trace!("Compressing state..."); - let new_room_state: CompressedState = self - .services - .state_compressor - .compress_state_events(state_events.iter().map(|(ssk, eid)| (ssk, (*eid).borrow()))) - .collect() - .await; + trace!("Compressing state..."); + let new_room_state: CompressedState = self + .services + .state_compressor + .compress_state_events(state_events.iter().map(|(ssk, eid)| (ssk, (*eid).borrow()))) + .collect() + .await; - Ok(Arc::new(new_room_state)) -} + Ok(Arc::new(new_room_state)) + } -#[implement(super::Service)] -#[tracing::instrument(name = "ruma", level = "debug", skip_all)] -pub async fn state_resolution<'a, StateSets>( - &'a self, - room_version_rules: &'a RoomVersionRules, - state_sets: StateSets, - auth_chain_sets: &'a [HashSet], -) -> Result> -where - StateSets: Iterator> + Clone + Send, -{ - let event_fetch = |event_id| self.event_fetch(event_id); - let event_exists = |event_id| self.event_exists(event_id); - state_res::resolve( - room_version_rules, - state_sets, - auth_chain_sets, - &event_fetch, - &event_exists, - ) - .map_err(|e| err!(error!("State resolution failed: {e:?}"))) - .await + pub async fn state_resolution<'a, StateSets>( + &'a self, + room_version_rules: &'a RoomVersionRules, + state_sets: StateSets, + auth_chain_sets: &'a [HashSet], + ) -> Result> + where + StateSets: Iterator> + Clone + Send, + { + let event_fetch = |event_id| self.event_fetch(event_id); + let event_exists = |event_id| self.event_exists(event_id); + state_res::resolve( + room_version_rules, + state_sets, + auth_chain_sets, + &event_fetch, + &event_exists, + ) + .map_err(|e| err!(error!("State resolution failed: {e:?}"))) + .await + } } diff --git a/src/service/rooms/event_handler/state_at_incoming.rs b/src/service/rooms/event_handler/state_at_incoming.rs index 9f11ac4b8..3dc7fc9f3 100644 --- a/src/service/rooms/event_handler/state_at_incoming.rs +++ b/src/service/rooms/event_handler/state_at_incoming.rs @@ -5,7 +5,7 @@ use std::{ }; use conduwuit::{ - Result, debug, err, error, implement, + Result, debug, err, error, matrix::{Event, StateMap}, trace, utils::stream::{BroadbandExt, IterStream, ReadyExt, TryBroadbandExt, TryWidebandExt}, @@ -15,180 +15,181 @@ use ruma::{OwnedEventId, RoomId, room_version_rules::RoomVersionRules}; use crate::rooms::short::ShortStateHash; -// TODO: if we know the prev_events of the incoming event we can avoid the -#[implement(super::Service)] -// request and build the state from a known point and resolve if > 1 prev_event -#[tracing::instrument(name = "state", level = "debug", skip_all)] -pub(super) async fn state_at_incoming_degree_one( - &self, - incoming_pdu: &Pdu, -) -> Result>> -where - Pdu: Event + Send + Sync, -{ - let prev_event = incoming_pdu - .prev_events() - .next() - .expect("at least one prev_event"); +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 + #[tracing::instrument(name = "state", level = "debug", skip_all)] + pub(super) async fn state_at_incoming_degree_one( + &self, + incoming_pdu: &Pdu, + ) -> Result>> + where + Pdu: Event + Send + Sync, + { + let prev_event = incoming_pdu + .prev_events() + .next() + .expect("at least one prev_event"); - let Ok(prev_event_sstatehash) = self - .services - .state_accessor - .pdu_shortstatehash(prev_event) - .await - else { - return Ok(None); - }; - - let mut state: HashMap<_, _> = self - .services - .state_accessor - .state_full_ids(prev_event_sstatehash) - .collect() - .await; - - debug!("Using cached state"); - let prev_pdu = self - .services - .timeline - .get_pdu(prev_event) - .await - .map_err(|e| err!(Database("Could not find prev event, but we know the state: {e:?}")))?; - - if let Some(state_key) = &prev_pdu.state_key { - let shortstatekey = self + let Ok(prev_event_sstatehash) = self .services - .short - .get_or_create_shortstatekey(&prev_pdu.kind.to_string().into(), state_key) + .state_accessor + .pdu_shortstatehash(prev_event) + .await + else { + return Ok(None); + }; + + let mut state: HashMap<_, _> = self + .services + .state_accessor + .state_full_ids(prev_event_sstatehash) + .collect() .await; - state.insert(shortstatekey, prev_event.to_owned()); - // Now it's the state after the pdu + debug!("Using cached state"); + let prev_pdu = self + .services + .timeline + .get_pdu(prev_event) + .await + .map_err(|e| { + err!(Database("Could not find prev event, but we know the state: {e:?}")) + })?; + + if let Some(state_key) = &prev_pdu.state_key { + let shortstatekey = self + .services + .short + .get_or_create_shortstatekey(&prev_pdu.kind.to_string().into(), state_key) + .await; + + state.insert(shortstatekey, prev_event.to_owned()); + // Now it's the state after the pdu + } + + debug_assert!(!state.is_empty(), "should be returning None for empty HashMap result"); + + Ok(Some(state)) } - debug_assert!(!state.is_empty(), "should be returning None for empty HashMap result"); - - Ok(Some(state)) -} - -#[implement(super::Service)] -#[tracing::instrument(name = "state", level = "debug", skip_all)] -pub(super) async fn state_at_incoming_resolved( - &self, - incoming_pdu: &Pdu, - room_id: &RoomId, - room_version_rules: &RoomVersionRules, -) -> Result>> -where - Pdu: Event + Send + Sync, -{ - trace!("Calculating extremity statehashes..."); - let Ok(extremity_sstatehashes) = incoming_pdu - .prev_events() - .try_stream() - .broad_and_then(|prev_eventid| { - self.services - .timeline - .get_pdu(prev_eventid) - .map_ok(move |prev_event| (prev_eventid, prev_event)) - }) - .broad_and_then(|(prev_eventid, prev_event)| { - self.services - .state_accessor - .pdu_shortstatehash(prev_eventid) - .map_ok(move |sstatehash| (sstatehash, prev_event)) - }) - .try_collect::>() - .await - else { - return Ok(None); - }; - - trace!("Calculating fork states..."); - let (fork_states, auth_chain_sets): (Vec>, Vec>) = - extremity_sstatehashes - .into_iter() + #[tracing::instrument(name = "state", level = "debug", skip_all)] + pub(super) async fn state_at_incoming_resolved( + &self, + incoming_pdu: &Pdu, + room_id: &RoomId, + room_version_rules: &RoomVersionRules, + ) -> Result>> + where + Pdu: Event + Send + Sync, + { + trace!("Calculating extremity statehashes..."); + let Ok(extremity_sstatehashes) = incoming_pdu + .prev_events() .try_stream() - .wide_and_then(|(sstatehash, prev_event)| { - self.state_at_incoming_fork(room_id, sstatehash, prev_event) + .broad_and_then(|prev_eventid| { + self.services + .timeline + .get_pdu(prev_eventid) + .map_ok(move |prev_event| (prev_eventid, prev_event)) }) - .try_collect() - .map_ok(Vec::into_iter) - .map_ok(Iterator::unzip) - .await?; + .broad_and_then(|(prev_eventid, prev_event)| { + self.services + .state_accessor + .pdu_shortstatehash(prev_eventid) + .map_ok(move |sstatehash| (sstatehash, prev_event)) + }) + .try_collect::>() + .await + else { + return Ok(None); + }; - let Ok(new_state) = self - .state_resolution(room_version_rules, fork_states.iter(), &auth_chain_sets) - .boxed() - .await - .inspect_err(|e| error!("State resolution failed: {e:?}")) - else { - return Ok(None); - }; + trace!("Calculating fork states..."); + let (fork_states, auth_chain_sets): (Vec>, Vec>) = + extremity_sstatehashes + .into_iter() + .try_stream() + .wide_and_then(|(sstatehash, prev_event)| { + self.state_at_incoming_fork(room_id, sstatehash, prev_event) + }) + .try_collect() + .map_ok(Vec::into_iter) + .map_ok(Iterator::unzip) + .await?; - new_state - .into_iter() - .stream() - .broad_then(|((event_type, state_key), event_id)| async move { - self.services - .short - .get_or_create_shortstatekey(&event_type, &state_key) - .map(move |shortstatekey| (shortstatekey, event_id)) - .await - }) - .collect() - .map(Some) - .map(Ok) - .await -} + let Ok(new_state) = self + .state_resolution(room_version_rules, fork_states.iter(), &auth_chain_sets) + .boxed() + .await + .inspect_err(|e| error!("State resolution failed: {e:?}")) + else { + return Ok(None); + }; -#[implement(super::Service)] -async fn state_at_incoming_fork( - &self, - room_id: &RoomId, - sstatehash: ShortStateHash, - prev_event: Pdu, -) -> Result<(StateMap, HashSet)> -where - Pdu: Event, -{ - let mut leaf_state: HashMap<_, _> = self - .services - .state_accessor - .state_full_ids(sstatehash) - .collect() - .await; - - if let Some(state_key) = prev_event.state_key() { - let shortstatekey = self - .services - .short - .get_or_create_shortstatekey(&prev_event.kind().to_string().into(), state_key) - .await; - - let event_id = prev_event.event_id(); - leaf_state.insert(shortstatekey, event_id.to_owned()); - // Now it's the state after the pdu + new_state + .into_iter() + .stream() + .broad_then(|((event_type, state_key), event_id)| async move { + self.services + .short + .get_or_create_shortstatekey(&event_type, &state_key) + .map(move |shortstatekey| (shortstatekey, event_id)) + .await + }) + .collect() + .map(Some) + .map(Ok) + .await } - let auth_chain = self - .services - .auth_chain - .event_ids_iter(room_id, leaf_state.values().map(Borrow::borrow)) - .try_collect(); + async fn state_at_incoming_fork( + &self, + room_id: &RoomId, + sstatehash: ShortStateHash, + prev_event: Pdu, + ) -> Result<(StateMap, HashSet)> + where + Pdu: Event, + { + let mut leaf_state: HashMap<_, _> = self + .services + .state_accessor + .state_full_ids(sstatehash) + .collect() + .await; - let fork_state = leaf_state - .iter() - .stream() - .broad_then(|(k, id)| { - self.services + if let Some(state_key) = prev_event.state_key() { + let shortstatekey = self + .services .short - .get_statekey_from_short(*k) - .map_ok(|(ty, sk)| ((ty, sk), id.clone())) - }) - .ready_filter_map(Result::ok) - .collect() - .map(Ok); + .get_or_create_shortstatekey(&prev_event.kind().to_string().into(), state_key) + .await; - try_join(fork_state, auth_chain).await + let event_id = prev_event.event_id(); + leaf_state.insert(shortstatekey, event_id.to_owned()); + // Now it's the state after the pdu + } + + let auth_chain = self + .services + .auth_chain + .event_ids_iter(room_id, leaf_state.values().map(Borrow::borrow)) + .try_collect(); + + let fork_state = leaf_state + .iter() + .stream() + .broad_then(|(k, id)| { + self.services + .short + .get_statekey_from_short(*k) + .map_ok(|(ty, sk)| ((ty, sk), id.clone())) + }) + .ready_filter_map(Result::ok) + .collect() + .map(Ok); + + try_join(fork_state, auth_chain).await + } } diff --git a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs index 0a711b468..523dde46b 100644 --- a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs +++ b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs @@ -1,7 +1,7 @@ use std::{borrow::Borrow, sync::Arc, time::Instant}; use conduwuit::{ - Err, Result, debug, debug_info, err, implement, info, is_equal_to, + Err, Result, debug, debug_info, err, info, is_equal_to, matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res}, trace, utils::{ @@ -22,366 +22,370 @@ use crate::rooms::{ timeline::RawPduId, }; -#[implement(super::Service)] -pub(super) async fn upgrade_outlier_to_timeline_pdu( - &self, - incoming_pdu: PduEvent, - mut val: CanonicalJsonObject, - create_event: &Pdu, - origin: &ServerName, - room_id: &RoomId, -) -> Result> -where - Pdu: Event + Send + Sync, -{ - // Skip the PDU if we already have it as a timeline event - if let Ok(pduid) = self - .services - .timeline - .get_pdu_id(incoming_pdu.event_id()) - .await +impl super::Service { + pub(super) async fn upgrade_outlier_to_timeline_pdu( + &self, + incoming_pdu: PduEvent, + mut val: CanonicalJsonObject, + create_event: &Pdu, + origin: &ServerName, + room_id: &RoomId, + ) -> Result> + where + Pdu: Event + Send + Sync, { - return Ok(Some(pduid)); - } - - let (rejected, soft_failed) = join!( - self.services - .pdu_metadata - .is_event_rejected(incoming_pdu.event_id()), - self.services - .pdu_metadata - .is_event_soft_failed(incoming_pdu.event_id()) - ); - if rejected { - return Err!(Request(InvalidParam("Event has been rejected"))); - } else if soft_failed { - return Err!(Request(InvalidParam("Event has been soft-failed"))); - } - - debug!( - event_id = %incoming_pdu.event_id, - "Upgrading PDU from outlier to timeline" - ); - let timer = Instant::now(); - 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 mut 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? - }; - - if state_at_incoming_event.is_none() { - state_at_incoming_event = self - .fetch_state(origin, create_event, room_id, incoming_pdu.event_id()) - .await?; - } - - let state_at_incoming_event = - state_at_incoming_event.expect("we always set this to some above"); - - 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, - ) - .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" - ); - } - - // 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 - } - }, - }; - - // 13. Use state resolution to find new room state - // We start looking at current room state now, so lets lock the room - 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 + // Skip the PDU if we already have it as a timeline event + if let Ok(pduid) = 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, - ) + .timeline + .get_pdu_id(incoming_pdu.event_id()) .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(), - ); - 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); + return Ok(Some(pduid)); } - // Additionally, if this is a redaction for a soft-failed event, we soft-fail it - // also. - - // 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 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" - ); - if self - .services + let (rejected, soft_failed) = join!( + self.services .pdu_metadata - .is_event_soft_failed(&redact_id) + .is_event_rejected(incoming_pdu.event_id()), + self.services + .pdu_metadata + .is_event_soft_failed(incoming_pdu.event_id()) + ); + if rejected { + return Err!(Request(InvalidParam("Event has been rejected"))); + } else if soft_failed { + return Err!(Request(InvalidParam("Event has been soft-failed"))); + } + + debug!( + event_id = %incoming_pdu.event_id, + "Upgrading PDU from outlier to timeline" + ); + let timer = Instant::now(); + 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 mut 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? + }; + + if state_at_incoming_event.is_none() { + state_at_incoming_event = self + .fetch_state(origin, create_event, room_id, incoming_pdu.event_id()) + .await?; + } + + let state_at_incoming_event = + state_at_incoming_event.expect("we always set this to some above"); + + 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, + ) + .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" + ); + } + + // 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 + } + }, + }; + + // 13. Use state resolution to find new room state + // We start looking at current room state now, so lets lock the room + 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 { - info!( - redact_id = %redact_id, - "Redaction is for a soft-failed event" + 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(), + ); + soft_fail = true; + } else { + return Err(e); + } + } else { + debug!( + event_id = %incoming_pdu.event_id, + "Event has passed policy server check." ); - soft_fail = true; + } + 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. + + // 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 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" + ); + if self + .services + .pdu_metadata + .is_event_soft_failed(&redact_id) + .await + { + info!( + redact_id = %redact_id, + "Redaction is for a soft-failed event" + ); + soft_fail = true; + } } } - } - 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::>() + trace!("Appending pdu to timeline"); + let mut extremities: Vec<_> = self + .services + .state + .get_forward_extremities(room_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"); + 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, + &state_lock, + room_id, + ) + .await?; + if soft_fail { + self.services + .pdu_metadata + .mark_event_soft_failed(incoming_pdu.event_id()); + + info!( + elapsed = ?timer.elapsed(), + event_id = %incoming_pdu.event_id, + "Event was soft failed" + ); + } else { + debug_info!( + elapsed = ?timer.elapsed(), + "Accepted", + ); + } + + // Event has passed all auth/stateres checks + drop(state_lock); + + Ok(pdu_id) } - - let pdu_id = self - .services - .timeline - .append_incoming_pdu( - &incoming_pdu, - val, - extremities.iter().map(Borrow::borrow), - state_ids_compressed, - soft_fail, - &state_lock, - room_id, - ) - .await?; - if soft_fail { - self.services - .pdu_metadata - .mark_event_soft_failed(incoming_pdu.event_id()); - - info!( - elapsed = ?timer.elapsed(), - event_id = %incoming_pdu.event_id, - "Event was soft failed" - ); - } else { - debug_info!( - elapsed = ?timer.elapsed(), - "Accepted", - ); - } - - // Event has passed all auth/stateres checks - drop(state_lock); - - Ok(pdu_id) }