diff --git a/src/api/server/send.rs b/src/api/server/send.rs index ead1713c1..9f8ab6a8f 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -41,7 +41,7 @@ use ruma::{ to_device::DeviceIdOrAllDevices, }; use service::{ - rooms::event_handler::build_local_dag, + rooms::event_handler::{DagBuilderTree, build_local_dag}, transactions::{FederationTxnState, TransactionError, TxnKey, WrappedTransactionResponse}, }; use tokio::sync::watch::{Receiver, Sender}; @@ -301,10 +301,12 @@ async fn handle_room( .iter() .map(|(event_id, obj)| (event_id.clone(), obj)) .collect(); - build_local_dag(&refmap).await.unwrap_or_else(|e| { - debug_warn!("Failed to build local DAG for room {room_id}: {e}"); - pdu_map.keys().cloned().collect() - }) + build_local_dag(&refmap, DagBuilderTree::PrevEvents) + .await + .unwrap_or_else(|e| { + debug_warn!("Failed to build local DAG for room {room_id}: {e}"); + pdu_map.keys().cloned().collect() + }) } else { pdu_map.keys().cloned().collect() }; 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 5b766f416..655b24429 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs @@ -7,7 +7,7 @@ use assign::assign; #[cfg(debug_assertions)] use conduwuit::error; use conduwuit::{ - Err, Event, PduEvent, debug, debug_error, debug_info, debug_warn, err, + Err, Event, PduEvent, Result, debug, debug_error, debug_info, debug_warn, err, result::FlatOk, state_res::lexicographical_topological_sort, trace, @@ -28,6 +28,11 @@ use crate::rooms::event_handler::parse_incoming_pdu::expect_event_id_array; pub const GET_MISSING_EVENTS_MAX_BATCH_SIZE: usize = 50; +pub enum DagBuilderTree { + PrevEvents, + AuthEvents, +} + /// Attempts to build a localised directed acyclic graph out of the given PDUs, /// returning them in a topologically sorted order. /// @@ -38,21 +43,24 @@ pub const GET_MISSING_EVENTS_MAX_BATCH_SIZE: usize = 50; #[allow(clippy::implicit_hasher)] pub async fn build_local_dag( pdu_map: &HashMap, -) -> conduwuit::Result> { + tree: DagBuilderTree, +) -> Result> { debug_assert!(pdu_map.len() >= 2, "needless call to build_local_dag with less than 2 PDUs"); let mut dag: HashMap> = HashMap::with_capacity(pdu_map.len()); let mut id_origin_ts: HashMap = HashMap::with_capacity(pdu_map.len()); + let tree = match tree { + | DagBuilderTree::AuthEvents => "auth_events", + | DagBuilderTree::PrevEvents => "prev_events", + }; for (event_id, value) in pdu_map { // Parse all prev events as event IDs - if they are missing, return an error (we // can't sanely continue in this case), otherwise skip invalid prev events. let prev_events = value - .get("prev_events") + .get(tree) .and_then(CanonicalJsonValue::as_array) - .ok_or_else(|| { - err!(Request(BadJson("event JSON for {event_id} is missing prev_events"))) - })? + .ok_or_else(|| err!(Request(BadJson("event JSON for {event_id} is missing {tree}"))))? .iter() .map(|v| v.as_str().and_then(|s| EventId::parse(s).ok())) .filter(|id| id.as_ref().is_some_and(|id| pdu_map.contains_key(id))) @@ -123,7 +131,7 @@ impl super::Service { tail: Vec, via: &ServerName, min_depth: UInt, - ) -> conduwuit::Result> { + ) -> Result> { let start = Instant::now(); #[cfg(debug_assertions)] { @@ -355,7 +363,7 @@ impl super::Service { remote: OwnedServerName, event_id: OwnedEventId, room_version_rules: &RoomVersionRules, - ) -> conduwuit::Result<(OwnedEventId, CanonicalJsonObject)> { + ) -> Result<(OwnedEventId, CanonicalJsonObject)> { let res = self .services .sending @@ -382,7 +390,7 @@ impl super::Service { candidates: impl Iterator, event_id: &EventId, room_version_rules: &RoomVersionRules, - ) -> conduwuit::Result<(OwnedEventId, CanonicalJsonObject)> { + ) -> Result<(OwnedEventId, CanonicalJsonObject)> { if let Ok(pdu_json) = self.services.timeline.get_pdu_json(event_id).await { return Ok((event_id.to_owned(), pdu_json)); } @@ -534,7 +542,7 @@ impl super::Service { .iter() .map(|(id, data)| (id.clone(), data)) .collect(); - let seeded_ordered = build_local_dag(&refmap) + let seeded_ordered = build_local_dag(&refmap, DagBuilderTree::AuthEvents) .await .expect("failed to build local DAG"); let mut pdus = HashMap::with_capacity(seeded_ordered.len()); diff --git a/src/service/rooms/event_handler/fetch_auth.rs b/src/service/rooms/event_handler/fetch_auth.rs new file mode 100644 index 000000000..6ff5d0324 --- /dev/null +++ b/src/service/rooms/event_handler/fetch_auth.rs @@ -0,0 +1,176 @@ +use std::collections::{HashMap, hash_map}; + +use conduwuit::{Err, Event, EventTypeExt, PduEvent, Result, err}; +use ruma::{ + CanonicalJsonObject, OwnedEventId, ServerName, + api::federation::authorization::get_event_authorization, + room_version_rules::RoomVersionRules, +}; +use tokio::join; + +use crate::rooms::event_handler::{build_local_dag, fetch_and_handle_outliers::DagBuilderTree}; + +impl super::Service { + /// Fetches (and persists) the incoming event's entire auth chain by asking + /// the remote server for it. The events are added to the outlier tree, but + /// no de-outliering is attempted. + /// + /// Returns a map of auth events, which includes potentially unauthorised + /// ones - it is the caller's responsibility to check rejection status of + /// each required event post factum. + #[tracing::instrument(skip_all)] + pub(super) async fn fetch_and_persist_event_auth( + &self, + incoming_event: &PduEvent, + origin: &ServerName, + room_version_rules: &RoomVersionRules, + create_event: &Pdu, + ) -> Result> + where + Pdu: Event + Send + Sync, + { + let event_auth: get_event_authorization::v1::Response = self + .services + .sending + .send_federation_request( + origin, + get_event_authorization::v1::Request::new( + incoming_event.room_id_or_hash().clone(), + incoming_event.event_id().to_owned(), + ), + ) + .await + .map_err(|e| { + err!(Request(Forbidden( + "Remote server is not divulging incoming event's auth chain: {e}" + ))) + })?; + + let mut auth_chain_map = HashMap::with_capacity(event_auth.auth_chain.len()); + + for auth_pdu_json in event_auth.auth_chain { + let (auth_event_room_id, auth_event_id, auth_pdu_json) = + self.parse_incoming_pdu(&auth_pdu_json).await?; + if auth_event_room_id != incoming_event.room_id_or_hash() { + return Err!(Request(Forbidden( + "Auth event {auth_event_id} is in {auth_event_room_id}, not {}.", + incoming_event.room_id_or_hash() + ))); + } + let auth_pdu = PduEvent::from_id_val(&auth_event_id, auth_pdu_json).map_err(|e| { + err!(Request(BadJson("Invalid PDU {auth_event_id} in auth chain: {e}"))) + })?; + if auth_pdu.state_key().is_none() { + return Err!(Request(BadJson( + "Invalid PDU {auth_event_id} in auth_chain: not a state event" + ))); + } + + auth_chain_map.insert(auth_event_id, auth_pdu); + } + + // We need to authorise each returned PDU to make sure that the caller can + // correctly detect if one of them is rejected. We don't remove the event from + // the map in case the caller incorrectly then identifies that event as missing, + // rather than rejected. + self.authorise_remote_auth_chain( + &auth_chain_map, + room_version_rules, + create_event.as_pdu(), + ) + .await?; + + Ok(auth_chain_map) + } + + /// Runs through the returned auth chain topologically and flags any events + /// that fail PDU check 4. + /// + /// An empty `Ok` is returned unless the auth chain is missing required + /// events (which means the remote server returned an invalid response). + async fn authorise_remote_auth_chain( + &self, + auth_chain_map: &HashMap, + room_version_rules: &RoomVersionRules, + create_event: &Pdu, + ) -> Result<()> + where + Pdu: Event + Send + Sync, + { + // TODO(nex): build_local_dag's signature needs changing because all callsites + // seem to do this refmap hack. + let pdu_objects = auth_chain_map + .iter() + .map(|(event_id, pdu)| (event_id, pdu.to_canonical_object())) + .collect::>(); + let refmap: HashMap = pdu_objects + .iter() + .map(|(id, data)| ((*id).to_owned(), data)) + .collect(); + let auth_chain_topo = build_local_dag(&refmap, DagBuilderTree::AuthEvents) + .await? + .into_iter() + .map(|event_id| (event_id.clone(), auth_chain_map.get(&event_id).unwrap().to_owned())) + .collect::>(); + + 'outer: for (event_id, pdu) in auth_chain_topo { + // If we know the event is rejected OR have it locally, we can skip this check. + // This is safe because if we know the event is rejected, running event auth + // will just reject it again. And, if we have it locally, and HAVEN'T flagged it + // as rejected, then we know it was at least accepted under the check we would + // otherwise be about to perform, previously. + let (is_rejected, have_locally) = join!( + self.services.pdu_metadata.is_event_rejected(&event_id), + self.services.timeline.pdu_exists(&event_id) + ); + if is_rejected || have_locally { + continue; + } + + let mut auth_events_by_key: HashMap<_, _> = + HashMap::with_capacity(pdu.auth_events.len()); + + for auth_event_id in pdu.auth_events() { + let Some(auth_event) = auth_chain_map.get(auth_event_id) else { + return Err!(Request(NotFound( + "Invalid event_auth response from remote server: {event_id} depends on \ + {auth_event_id}, but it is not present" + ))); + }; + + self.services + .outlier + .add_pdu_outlier(auth_event_id, &auth_event.to_canonical_object()); + let key = auth_event + .kind() + .with_state_key(auth_event.state_key().unwrap()); + match auth_events_by_key.entry(key) { + | hash_map::Entry::Vacant(v) => { + v.insert(auth_event.clone()); + }, + | hash_map::Entry::Occupied(_) => { + // Duplicate auth events by key are not allowed. + self.services + .pdu_metadata + .mark_event_rejected(auth_event_id); + self.services.pdu_metadata.mark_event_rejected(&event_id); + continue 'outer; + }, + } + } + if !self + .is_event_self_authorised( + &pdu, + room_version_rules, + create_event.as_pdu(), + &auth_events_by_key, + ) + .await + { + self.services.pdu_metadata.mark_event_rejected(&event_id); + } + } + + Ok(()) + } +} diff --git a/src/service/rooms/event_handler/fetch_prev.rs b/src/service/rooms/event_handler/fetch_prev.rs index 8f27d1be3..6e4f8559a 100644 --- a/src/service/rooms/event_handler/fetch_prev.rs +++ b/src/service/rooms/event_handler/fetch_prev.rs @@ -7,7 +7,7 @@ use conduwuit::{ use futures::StreamExt; use ruma::{CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName}; -use crate::rooms::event_handler::build_local_dag; +use crate::rooms::event_handler::{build_local_dag, fetch_and_handle_outliers::DagBuilderTree}; impl super::Service { /// Fetches any missing prev_events for this event and persists them before @@ -89,7 +89,7 @@ impl super::Service { } else { let refmap: HashMap = mapped.iter().map(|(id, data)| (id.clone(), data)).collect(); - build_local_dag(&refmap).await? + build_local_dag(&refmap, DagBuilderTree::PrevEvents).await? }; let job_start = Instant::now(); diff --git a/src/service/rooms/event_handler/handle_outlier_pdu.rs b/src/service/rooms/event_handler/handle_outlier_pdu.rs index 470cd6643..72e7b0629 100644 --- a/src/service/rooms/event_handler/handle_outlier_pdu.rs +++ b/src/service/rooms/event_handler/handle_outlier_pdu.rs @@ -1,14 +1,13 @@ use std::collections::{BTreeMap, HashMap, hash_map}; use conduwuit::{ - Err, Event, PduEvent, Result, debug, debug_info, debug_warn, err, info, state_res, trace, - warn, + Err, Event, EventTypeExt, PduEvent, Result, debug, debug_info, debug_warn, err, info, + matrix::StateKey, state_res::auth_check, trace, warn, }; use futures::future::ready; use ruma::{ CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName, - api::federation::authorization::get_event_authorization, canonical_json::redact, - events::StateEventType, + canonical_json::redact, events::StateEventType, room_version_rules::RoomVersionRules, }; use super::{check_room_id, get_room_version_rules}; @@ -100,10 +99,62 @@ impl super::Service { check_room_id(room_id, &pdu_event)?; + // TODO(nex): From hereon the event is not dropped, and thus always added as an + // outlier. However, we only do that at the end of this function, which means + // several duplicated calls to add_pdu_outlier. Shouldn't we just do it here + // instead, since we know it's going to be persisted as an outlier no matter + // what? the rest of this function is basically just to check PDU check 4. + + // NOTE^: Technically, persisting the event before knowing if it's rejected + // introduces a race condition in fetch_and_persist_event_auth, where we have + // the event locally, but haven't yet flagged it as rejected, which the + // fetcher perceives as "accepted". I'm not sure if that's practically possible + // though. + // Fetch all auth events let mut auth_events: HashMap = HashMap::new(); for auth_event_id in pdu_event.auth_events() { + if let Ok(auth_event) = self.services.timeline.get_pdu(auth_event_id).await { + check_room_id(room_id, &auth_event)?; + trace!("Found auth event {auth_event_id} for outlier event {event_id} locally"); + auth_events.insert(auth_event_id.to_owned(), auth_event); + } else { + debug_warn!( + "Could not find auth event {auth_event_id} for outlier event {event_id} \ + locally" + ); + } + } + + // Fetch any missing ones & reject invalid ones + if auth_events.len() != pdu_event.auth_events().count() { + info!("Missing some auth events, asking remote for auth chain"); + let auth_chain_map = self + .fetch_and_persist_event_auth( + &pdu_event, + origin, + &room_version_rules, + create_event, + ) + .await?; + for auth_event_id in pdu_event.auth_events() { + if auth_events.contains_key(auth_event_id) { + continue; + } + if let Some(auth_event) = auth_chain_map.get(auth_event_id) { + auth_events.insert(auth_event_id.to_owned(), auth_event.clone()); + } else { + return Err!(Request(Forbidden( + "Remote server is not divulging incoming event's auth events (missing: \ + {auth_event_id})" + ))); + } + } + } + + // Ensure none of the auth events are rejected - if they are, reject too. + for auth_event_id in auth_events.keys() { if self .services .pdu_metadata @@ -120,64 +171,6 @@ impl super::Service { "Event has rejected auth event: {auth_event_id}" ))); } - - if let Ok(auth_event) = self.services.timeline.get_pdu(auth_event_id).await { - check_room_id(room_id, &auth_event)?; - trace!("Found auth event {auth_event_id} for outlier event {event_id} locally"); - auth_events.insert(auth_event_id.to_owned(), auth_event); - } else { - debug_warn!( - "Could not find auth event {auth_event_id} for outlier event {event_id} \ - locally" - ); - } - } - - // Fetch any missing ones & reject invalid ones - if auth_events.len() != pdu_event.auth_events().count() { - info!("Missing some auth events, asking remote for auth chain"); - let response: get_event_authorization::v1::Response = self - .services - .sending - .send_federation_request( - origin, - get_event_authorization::v1::Request::new( - room_id.to_owned(), - event_id.to_owned(), - ), - ) - .await - .map_err(|e| { - err!(Request(Forbidden( - "Remote server is not divulging incoming event's auth chain: {e}" - ))) - })?; - let mut auth_chain_map = HashMap::with_capacity(response.auth_chain.len()); - for auth_pdu_json in response.auth_chain { - let (auth_event_room_id, auth_event_id, auth_pdu_json) = - self.parse_incoming_pdu(&auth_pdu_json).await?; - if auth_event_room_id != room_id { - return Err!(Request(Forbidden( - "Auth event {auth_event_id} is in {auth_event_room_id}, not {room_id}." - ))); - } - let auth_pdu = PduEvent::from_id_val(&auth_event_id, auth_pdu_json) - .map_err(|e| err!(Request(BadJson("Invalid PDU {auth_event_id}: {e}"))))?; - auth_chain_map.insert(auth_event_id, auth_pdu); - } - for auth_event_id in pdu_event.auth_events() { - if auth_events.contains_key(auth_event_id) { - continue; - } - if let Some(auth_event) = auth_chain_map.get(auth_event_id) { - auth_events.insert(auth_event_id.to_owned(), auth_event.clone()); - } else { - return Err!(Request(Forbidden( - "Remote server is not divulging incoming event's auth events (missing: \ - {auth_event_id})" - ))); - } - } } // 6. Reject "due to auth events" if the event doesn't pass auth based on the @@ -193,13 +186,13 @@ impl super::Service { check_room_id(room_id, &auth_event)?; - match auth_events_by_key.entry(( - auth_event.kind.to_string().into(), + let key = auth_event.kind().with_state_key( auth_event .state_key .clone() - .expect("all auth events have state keys"), - )) { + .expect("all auth events must have state keys"), + ); + match auth_events_by_key.entry(key) { | hash_map::Entry::Vacant(v) => { v.insert(auth_event); }, @@ -218,23 +211,15 @@ impl super::Service { } } - 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 { + if !self + .is_event_self_authorised( + &pdu_event, + &room_version_rules, + create_event.as_pdu(), + &auth_events_by_key, + ) + .await + { self.services.pdu_metadata.mark_event_rejected(event_id); self.services .outlier @@ -255,4 +240,55 @@ impl super::Service { Ok((pdu_event, incoming_pdu)) } + + /// Helper method that turns the return value of `is_event_self_authorised` + /// into a `Result` depending on the value. + /// + /// If the event is not authorised, a Forbidden error is returned. + /// Otherwise, an empty `Ok`. + pub(super) async fn is_event_self_authorised( + &self, + pdu: &PduEvent, + room_version_rules: &RoomVersionRules, + create_event: &PduEvent, + auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>, + ) -> bool { + self.expect_event_is_self_authorised( + pdu, + room_version_rules, + create_event, + auth_events_by_key, + ) + .await + .is_ok() + } + + /// Checks PDU check 4: Passes authorisation rules based on the event's auth + /// events ([spec]). + /// + /// If the auth check fails, false is returned, otherwise true. + /// + /// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu + pub(super) async fn expect_event_is_self_authorised( + &self, + pdu: &PduEvent, + room_version_rules: &RoomVersionRules, + create_event: &PduEvent, + auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>, + ) -> Result { + let state_fetch = |ty: &StateEventType, sk: &str| { + let key = (ty.to_owned(), sk.into()); + ready(auth_events_by_key.get(&key).map(ToOwned::to_owned)) + }; + + auth_check( + room_version_rules, + pdu, + None, // TODO: third party invite + state_fetch, + create_event, + ) + .await + .map_err(|e| err!("Event self-authentication failed: {e:?}")) + } } diff --git a/src/service/rooms/event_handler/mod.rs b/src/service/rooms/event_handler/mod.rs index 779889287..1367a2331 100644 --- a/src/service/rooms/event_handler/mod.rs +++ b/src/service/rooms/event_handler/mod.rs @@ -1,5 +1,6 @@ mod acl_check; mod fetch_and_handle_outliers; +mod fetch_auth; mod fetch_prev; mod fetch_state; mod handle_incoming_pdu; @@ -14,7 +15,9 @@ use std::{collections::HashMap, fmt::Write, sync::Arc, time::Instant}; use async_trait::async_trait; use conduwuit::{Err, Event, PduEvent, Result, Server, SyncRwLock, utils::MutexMap}; -pub use fetch_and_handle_outliers::{GET_MISSING_EVENTS_MAX_BATCH_SIZE, build_local_dag}; +pub use fetch_and_handle_outliers::{ + DagBuilderTree, GET_MISSING_EVENTS_MAX_BATCH_SIZE, build_local_dag, +}; use ruma::{ OwnedEventId, OwnedRoomId, RoomId, events::room::create::RoomCreateEventContent, room_version_rules::RoomVersionRules,