Mirrors/continuwuity
0
0 Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2026-11-13 12:16:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Reduce complexity of incoming PDU handling

Browse Source 
* Ignores any events pushed without a room ID
* Removes needless clones for PDU size checking
* Removes incorrect ACL check in incoming handler
* Fast-path handling already handled outlier events
* Remove redundant same-room checks, replace with useful ones
* Combine event rejection and persistence in a single function
* Additional safety assertions in upgrade task
* Split upgrade task into multiple subroutines for reduced cognitive complexity
* Only mutate current state and forward extremities in tandem
* Improve code documentation to better explain the logic flow
This commit is contained in:
timedout committed 2026-07-11 17:27:42 +01:00
1 parent ace0d52074
commit 99a5f2a9fb
6 files changed
+472 -409
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/api/server/send.rs
+14 -5
View File
@@ -152,7 +152,16 @@ async fn process_inbound_transaction(
.stream()
.broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu))
.inspect_err(|e| warn!("Could not parse incoming PDU: {e}"))
.ready_filter_map(Result::ok);
.ready_filter_map(Result::ok)
.ready_filter_map(|pdu| {
// Room IDs should always be present on events being pushed to us in txns.
// This also ensures incoming PDUs have the right room ID.
pdu.2
.get("room_id")
.and_then(|r| r.as_str())
.is_some_and(|v| v == pdu.0.as_str())
.then_some(pdu)
});
let edus = body
.edus
@@ -241,7 +250,7 @@ async fn handle(
pdus: impl Stream<Item = Pdu> + Send,
edus: impl Stream<Item = Edu> + Send,
) -> std::result::Result<ResolvedMap, TransactionError> {
// group pdus by room
// Group PDUs by room for parallel processing
let pdus = pdus
.collect()
.map(|mut pdus: Vec<_>| {
@@ -252,7 +261,7 @@ async fn handle(
})
.await;
// we can evaluate rooms concurrently
// Evaluate rooms in parallel
let results: ResolvedMap = pdus
.into_iter()
.try_stream()
@@ -266,7 +275,7 @@ async fn handle(
.boxed()
.await?;
// evaluate edus after pdus, at least for now.
// Evaluate EDUs after PDUs in case some of the PDUs then forbid some EDUs.
edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu))
.boxed()
.await;
@@ -322,7 +331,7 @@ async fn handle_room(
let result = services
.rooms
.event_handler
.handle_incoming_pdu(origin, room_id, &event_id, value.clone(), true)
.handle_incoming_pdu(origin, room_id, &event_id, value, true)
.boxed()
.await
.map(|_| ());
src/service/rooms/event_handler/handle_incoming_pdu.rs
+33 -83
View File
@@ -4,46 +4,16 @@ use std::{
};
use conduwuit::{
Err, Event, Result, debug, debug_error, debug_warn, defer, err, error, info,
matrix::PartialPdu, result::DebugInspect, trace, utils::time::jitter, warn,
Err, Event, Result, debug, debug_error, debug_warn, defer, error, matrix::PartialPdu,
result::DebugInspect, trace, utils::time::jitter, warn,
};
use futures::{
FutureExt, StreamExt,
future::{OptionFuture, try_join4},
};
use ruma::{CanonicalJsonValue, EventId, OwnedUserId, RoomId, ServerName, UserId};
use futures::{FutureExt, StreamExt, future::try_join3};
use ruma::{CanonicalJsonValue, EventId, RoomId, ServerName, UserId};
use tokio::sync::mpsc;
use crate::rooms::timeline::{RawPduId, pdu_fits};
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",
skip_all,
@@ -57,11 +27,12 @@ impl super::Service {
value: BTreeMap<String, CanonicalJsonValue>,
is_timeline_event: bool,
) -> Result<Option<RawPduId>> {
// 1. Skip the PDU if we already have it as a timeline event
// Skip the PDU if we already have it as a timeline event. We still re-process
// outliers in this scenario.
if let Ok(pdu_id) = self.services.timeline.get_pdu_id(event_id).await {
return Ok(Some(pdu_id));
}
if !pdu_fits(&mut value.clone()) {
if !pdu_fits(&value) {
warn!(
"dropping incoming PDU {event_id} in room {room_id} from {origin} because it \
exceeds 65535 bytes or is otherwise too large."
@@ -72,34 +43,21 @@ impl super::Service {
"processing incoming PDU from {origin} for room {room_id} with event id {event_id}"
);
// 1.1 Check we even know about the room
let meta_exists = self.services.metadata.exists(room_id).map(Ok);
// If this is a membership state event for a local user, we will be interested
// in this event even if we aren't in the room. This allows us to process things
// like revoking invites over federation, without being in the room.
let is_interesting_member_event = value.get("type").and_then(|t| t.as_str())
== Some("m.room.member")
&& value
.get("state_key")
.and_then(|s| s.as_str())
.and_then(|s| UserId::parse(s).ok())
.is_some_and(|u| self.services.globals.user_is_local(&u));
// 1.2 Check if the room is disabled
let is_disabled = self.services.metadata.is_disabled(room_id).map(Ok);
// 1.3.1 Check room ACL on origin field/server
let origin_acl_check = self.acl_check(origin, room_id);
// 1.3.2 Check room ACL on sender's server name
let sender: OwnedUserId = value
.get("sender")
.and_then(|v| v.as_str())
.ok_or_else(|| err!("No sender in object"))
.and_then(|v| Ok(UserId::parse(v)?))
.map_err(|e| err!(Request(BadJson("PDU does not have a valid sender key: {e}"))))?;
let sender_acl_check: OptionFuture<_> = sender
.server_name()
.ne(origin)
.then(|| self.acl_check(sender.server_name(), room_id))
.into();
let (meta_exists, is_disabled, (), ()) = try_join4(
meta_exists,
is_disabled,
origin_acl_check,
sender_acl_check.map(|o| o.unwrap_or(Ok(()))),
let (room_exists, is_disabled, ()) = try_join3(
self.services.metadata.exists(room_id).map(Ok),
self.services.metadata.is_disabled(room_id).map(Ok),
self.acl_check(origin, room_id),
)
.await
.inspect_err(
@@ -112,21 +70,11 @@ impl super::Service {
)));
}
if !self
.services
.state_cache
.server_in_room(self.services.globals.server_name(), room_id)
.await
{
info!(
%origin,
%room_id,
"Dropping inbound PDU for room we aren't participating in"
);
return Err!(Request(NotFound("This server is not participating in that room.")));
}
if !meta_exists {
if !room_exists && !is_interesting_member_event {
if is_interesting_member_event {
// TODO: handle interesting membership events where we aren't in
// the room
}
return Err!(Request(NotFound("Room is unknown to this server")));
}
@@ -152,7 +100,7 @@ impl super::Service {
.handle_outlier_pdu(origin, create_event, event_id, room_id, value)
.await?;
// 8. if not timeline event: stop
// If this is not a timeline event, stop now, as we don't want to de-outlier it.
if !is_timeline_event {
return Ok(None);
}
@@ -169,8 +117,9 @@ impl super::Service {
return Ok(None);
}
// 9. Fetch any missing prev events doing all checks listed here starting at 1.
// These are timeline events
// Fetch any missing prev events doing all checks listed here starting at 1.
// These are timeline events.
// TODO: This part needs to be done in a background queue somewhere.
debug!("Fetching and persisting any missing prev events");
Box::pin(self.fetch_prevs(
@@ -185,9 +134,10 @@ impl super::Service {
error!("Failed to fetch and persist incoming event's prev_events: {e:?}");
})?;
let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event";
let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event"
&& incoming_pdu.state_key().is_none();
// Done with prev events, now handling the incoming event
// Done with prev events, now we can handle promoting the PDU
let pdu_id = Box::pin(self.upgrade_outlier_to_timeline_pdu(
incoming_pdu,
val,
src/service/rooms/event_handler/handle_outlier_pdu.rs
+55 -51
View File
@@ -16,8 +16,13 @@ use crate::rooms::timeline::pdu_fits;
impl super::Service {
/// Handles a PDU as an outlier, performing basic checks like signatures and
/// hashes, proclaimed event auth, and then adding it to the outlier tree.
///
/// This performs steps 1 through 4 of [S-S section 5.1][spec], returning
/// the parsed PDU and modified JSON object.
///
/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(name="handle_outlier", skip_all, fields(%event_id))]
#[tracing::instrument(name = "handle_outlier", skip_all)]
pub(super) async fn handle_outlier_pdu<'a, Pdu>(
&self,
origin: &'a ServerName,
@@ -29,54 +34,47 @@ impl super::Service {
where
Pdu: Event + Send + Sync,
{
if !pdu_fits(&mut value.clone()) {
// Skip outlier handling if we already have this event as either a timeline or
// outlier PDU.
if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
debug!(
"Already have event {event_id} as an outlier or timeline event, not \
re-processing"
);
value.insert(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
);
check_room_id(room_id, &pdu_event)?;
return Ok((pdu_event, value));
}
// 1. Check that the PDU follows the format for the room version
// (in this case, just size check)
if !pdu_fits(&value) {
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
value.remove("unsigned");
let room_version_rules = get_room_version_rules(create_event)?;
// 2. Check signatures, otherwise drop.
// 3. Check content hash, redacting the event if it fails.
let mut incoming_pdu = match self
.services
.server_keys
.verify_event(&value, &room_version_rules)
.await
{
| Ok(ruma::signatures::Verified::All) => {
if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
debug!(
"Already have event {event_id} as an outlier or timeline event, not \
re-processing"
);
value.insert(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
);
check_room_id(room_id, &pdu_event)?;
return Ok((pdu_event, value));
}
value
},
| Ok(ruma::signatures::Verified::All) => value,
| Ok(ruma::signatures::Verified::Signatures) => {
if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
debug!(
"Received a redacted copy of {event_id}, but we already knew about it. \
Re-using known content instead."
);
check_room_id(room_id, &pdu_event)?;
let obj = pdu_event.to_canonical_object();
return Ok((pdu_event, obj));
}
debug_info!("Calculated hash does not match (redaction): {event_id}");
debug_info!("Content hash mismatch, redacting event and continuing");
redact(value, &room_version_rules.redaction, None)
.map_err(|e| err!(Request(BadJson("Failed to redact {event_id}: {e}"))))?
.map_err(|e| err!(Request(BadJson("Unable to redact {event_id}: {e}"))))?
},
| Err(e) => {
return Err!(Request(Forbidden(debug_error!(
@@ -91,14 +89,11 @@ impl super::Service {
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
);
let pdu_event = serde_json::from_value::<PduEvent>(
serde_json::to_value(&incoming_pdu).expect("CanonicalJsonObj is a valid JsonValue"),
)
.map_err(|e| err!(Request(BadJson(debug_warn!("Event is not a valid PDU: {e}")))))?;
check_room_id(room_id, &pdu_event)?;
// TODO(nex): From hereon the event is not dropped, and thus always added as an
// outlier. However, we only do that at the end of this function, which means
// several duplicated calls to add_pdu_outlier. Shouldn't we just do it here
@@ -116,7 +111,6 @@ impl super::Service {
for auth_event_id in pdu_event.auth_events() {
if let Ok(auth_event) = self.services.timeline.get_pdu(auth_event_id).await {
check_room_id(room_id, &auth_event)?;
trace!("Found auth event {auth_event_id} for outlier event {event_id} locally");
auth_events.insert(auth_event_id.to_owned(), auth_event);
} else {
@@ -154,7 +148,7 @@ impl super::Service {
}
// Ensure none of the auth events are rejected - if they are, reject too.
for auth_event_id in auth_events.keys() {
for (auth_event_id, auth_event) in &auth_events {
if self
.services
.pdu_metadata
@@ -166,15 +160,27 @@ impl super::Service {
{auth_event_id}",
event_id,
);
self.services.pdu_metadata.mark_event_rejected(event_id);
self.reject_and_persist(event_id, &incoming_pdu);
return Err!(Request(Forbidden(
"Event has rejected auth event: {auth_event_id}"
)));
}
if auth_event.room_id_or_hash() != room_id {
debug_warn!(
%auth_event_id,
auth_event_room_id=%auth_event.room_id_or_hash(),
expected_room_id=%room_id,
"Rejecting incoming event which depends on an auth event in another room.",
);
self.reject_and_persist(event_id, &incoming_pdu);
return Err!(Request(Forbidden(
"Event depends on a cross-room auth event: {auth_event_id}"
)));
}
}
// 6. Reject "due to auth events" if the event doesn't pass auth based on the
// auth events
// 4. Reject "due to auth events" if the event doesn't pass auth based on the
// claimed auth events
debug!("Checking based on auth events");
let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(auth_events.len());
// Build map of auth events
@@ -184,8 +190,6 @@ impl super::Service {
.expect("we just checked that we have all auth events")
.to_owned();
check_room_id(room_id, &auth_event)?;
let key = auth_event.kind().with_state_key(
auth_event
.state_key
@@ -197,10 +201,7 @@ impl super::Service {
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);
self.reject_and_persist(event_id, &incoming_pdu);
return Err!(Request(Forbidden(
"Auth event's type and state_key combination exists multiple times: {}, \
{}",
@@ -220,10 +221,7 @@ impl super::Service {
)
.await
{
self.services.pdu_metadata.mark_event_rejected(event_id);
self.services
.outlier
.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
self.reject_and_persist(event_id, &incoming_pdu);
return Err!(Request(Forbidden(
"Event authorisation fails based on event's claimed auth events"
)));
@@ -291,4 +289,10 @@ impl super::Service {
.await
.map_err(|e| err!("Event self-authentication failed: {e:?}"))
}
/// Marks the event as rejected and then saves it as an outlier.
pub(super) fn reject_and_persist(&self, event_id: &EventId, pdu: &CanonicalJsonObject) {
self.services.pdu_metadata.mark_event_rejected(event_id);
self.services.outlier.add_pdu_outlier(event_id, pdu);
}
}
src/service/rooms/event_handler/upgrade_outlier_pdu.rs
+365 -267
View File
@@ -1,19 +1,20 @@
use std::{borrow::Borrow, sync::Arc, time::Instant};
use std::{borrow::Borrow, collections::HashMap, sync::Arc, time::Instant};
use conduwuit::{
Err, Result, debug, debug_error, debug_info, err, info, is_equal_to,
Err, Result, debug, debug_error, debug_info, err, info, is_equal_to, is_true,
matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
result::DebugInspect,
trace,
utils::{
IterStream,
mutex_map::Guard,
stream::{BroadbandExt, ReadyExt},
},
warn,
};
use futures::{FutureExt, StreamExt, future::ready};
use ruma::{
CanonicalJsonObject, RoomId, ServerName, api::error::ErrorKind, events::StateEventType,
CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId, ServerName, api::error::ErrorKind,
events::StateEventType, room_version_rules::RoomVersionRules,
};
use tokio::join;
@@ -48,14 +49,22 @@ impl super::Service {
} else if rejected {
return Err!(Request(Forbidden("Event has been rejected")));
} else if soft_failed {
// Soft-failed events cannot be promoted.
return Err!(Request(Forbidden("Event has been soft-failed")));
}
// These should never happen, but they're good last-minute sanity checks to
// ensure we never promote totally illegal events.
assert_eq!(
*create_event.kind(),
StateEventType::RoomCreate.into(),
"tried to upgrade a PDU with a create_event that is not a room create event"
);
assert_eq!(
incoming_pdu.room_id_or_hash(),
*room_id,
"room ID mismatch: PDU room ID differs from parameter"
);
debug!(
event_id = %incoming_pdu.event_id,
@@ -65,266 +74,92 @@ impl super::Service {
let min_depth = self.services.metadata.get_mindepth(room_id).await;
let room_version_rules = get_room_version_rules(create_event)?;
// 10. Fetch missing state and auth chain events by calling /state_ids at
// backwards extremities doing all the checks in this list starting at 1.
// These are not timeline events.
// We now need to resolve the state before the event so that we can perform PDU
// check 5 (event auth passes based on state before the event). To do this, we
// either need to have all the prev events locally, or ask a remote server
// for the state at the event.
let (passes_state_before, state_before) = self
.is_authorised_by_state_before(
&incoming_pdu,
&room_version_rules,
create_event,
origin,
)
.await?;
debug!(
event_id = %incoming_pdu.event_id,
"Resolving state at event"
);
let state_at_incoming_event = if incoming_pdu.prev_events().count() == 1 {
self.state_at_incoming_degree_one(&incoming_pdu).await?
} else {
self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_rules)
.await?
};
let state_at_incoming_event = match state_at_incoming_event {
| Some(s) => s,
| None => {
trace!("Could not calculate incoming state, asking remote {origin} for it");
self.fetch_state(origin, create_event, room_id, incoming_pdu.event_id())
.await
.debug_inspect_err(|e| {
debug_error!("Could not fetch state from {origin}: {e}");
})?
},
};
if state_at_incoming_event.is_empty()
&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
{
// This can happen if the remote sends an event but cannot be reached to fetch
// the state at it, and all other servers in the room (which might just be the
// unreachable server) are unable to provide required info.
// returning an error here allows the upgrade to be attempted at another time.
return Err!(Request(Forbidden("Could not resolve incoming state at event")));
}
trace!(state_events = state_at_incoming_event.len(), "Calculated incoming state");
debug!(
event_id = %incoming_pdu.event_id,
"Performing auth check to upgrade"
);
// 11. Check the auth of the event passes based on the state of the event
let state_fetch_state = &state_at_incoming_event;
let state_fetch = |k: StateEventType, s: StateKey| async move {
let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
let event_id = state_fetch_state.get(&shortstatekey)?;
self.services.timeline.get_pdu(event_id).await.ok()
};
debug!(
event_id = %incoming_pdu.event_id,
"Running initial auth check"
);
// PDU check: 5
let auth_check = state_res::event_auth::auth_check(
&room_version_rules,
&incoming_pdu,
None, // TODO: third party invite
|ty, sk| state_fetch(ty.clone(), sk.into()),
create_event.as_pdu(),
)
.await
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
if !auth_check {
self.services
.pdu_metadata
.mark_event_rejected(incoming_pdu.event_id());
if !passes_state_before {
self.reject_and_persist(incoming_pdu.event_id(), &val);
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?;
// Now that we know the event passes both self-authentication, and
// authentication based on the state before the event, we need to check that it
// passes based on the *current* room state (state across all forward
// extremities). If it doesn't, we accept it, but soft-fail it, and this
// prevents it being promoted.
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
// We lock the room here to prevent the current state from changing beneath us
// mid-check.
trace!(
room_id = %room_id,
"Locking the room"
);
let state_lock = self.services.state.mutex.lock(room_id).await;
let passes_current_state = self
.is_authorised_by_current_state(&incoming_pdu, &room_version_rules, create_event)
.await?;
let state_ids_compressed: Arc<CompressedState> = 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
// Determine whether this PDU should be soft-failed.
// If the auth check failed, invariably yes. Otherwise, only if the user isn't
// allowed to redact the target event (if any).
let mut should_soft_fail =
match (passes_current_state, incoming_pdu.redacts_id(&room_version_rules)) {
| (false, _) => true,
| (true, None) => false,
| (true, Some(redact_id)) => self
.services
.short
.get_or_create_shortstatekey(
&incoming_pdu.kind().to_string().into(),
state_key,
)
.await;
.state_accessor
.user_can_redact(&redact_id, incoming_pdu.sender(), room_id, true)
.await
.is_ok_and(is_true!()),
};
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
if !should_soft_fail {
// Now we can perform check 7, which is ensuring the event passes policy server
// checks.
// We explicitly only do this if we aren't already going to soft-fail the event,
// since the policy server refusing this event also soft-fails it.
debug!(event_id = %incoming_pdu.event_id, "Checking policy server for event");
let tmp_evt_id = val.remove("event_id");
if let Err(e) = self
.policy_server_allows_event(
&incoming_pdu,
&mut val,
room_id,
&room_version_rules,
true,
)
.await
{
if matches!(e.kind(), ErrorKind::Forbidden) {
info!(
event_id = %incoming_pdu.event_id,
error = %e,
"Event has been marked as spam by policy server: {}",
e.message(),
);
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);
}
should_soft_fail = self
.policy_server_check(&incoming_pdu, &mut val, &room_version_rules)
.await?;
// 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.
// TODO: this is supposed to hide redactions from policy servers and janitorial
// bots, however, for full efficacy it also needs to hide redactions for
// unknown events. This needs to be investigated at a later time.
if let Some(redact_id) = incoming_pdu.redacts_id(&room_version_rules) {
debug!(
redact_id = %redact_id,
"Checking if redaction is for a soft-failed event"
"Checking if redaction is for a soft-failed/rejected event"
);
if self
if !self
.services
.pdu_metadata
.is_event_soft_failed(&redact_id)
.is_event_accepted(&redact_id)
.await
{
info!(
redact_id = %redact_id,
"Redaction is for a soft-failed event"
debug_info!(
"Soft-failing valid redaction because it targets a non-accepted event"
);
soft_fail = true;
should_soft_fail = true;
}
}
}
// The PDU has now passed all checks! We can now promote it (or soft-fail it if
// the verdict is such).
trace!("Appending pdu to timeline");
let mut extremities: Vec<_> = self
.services
@@ -332,40 +167,30 @@ impl super::Service {
.get_forward_extremities(room_id)
.collect()
.await;
if !soft_fail {
if !should_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::<Vec<_>>()
.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");
// This also means we set the state here.
// We do this BEFORE setting the extremities so that there's never a point in
// time where we have fresh extremities referencing stale state.
extremities = self
.progress_state_and_extremities(
&incoming_pdu,
&room_version_rules,
state_before.clone(),
extremities,
&state_lock,
)
.await?;
}
let state_ids_compressed: Arc<CompressedState> = self
.services
.state_compressor
.compress_state_events(state_before.iter().map(|(ssk, eid)| (ssk, eid.borrow())))
.collect()
.map(Arc::new)
.await;
let pdu_id = self
.services
.timeline
@@ -374,17 +199,18 @@ impl super::Service {
val,
extremities.iter().map(Borrow::borrow),
state_ids_compressed,
soft_fail,
should_soft_fail,
&state_lock,
room_id,
)
.await?;
if soft_fail {
if should_soft_fail {
self.services
.pdu_metadata
.mark_event_soft_failed(incoming_pdu.event_id());
info!(
debug_info!(
elapsed = ?timer.elapsed(),
event_id = %incoming_pdu.event_id,
"Event was soft failed"
@@ -407,4 +233,276 @@ impl super::Service {
Ok(pdu_id)
}
/// Checks that the event passes PDU check 5, which ensures that the event
/// is authorised based on the state before the event (which is the resolved
/// state across all prev events).
///
/// Returns a boolean indicating whether the event is authorised, and also
/// the resolved state before the event for later use. Returns an error if
/// state fetching or auth checking fails.
async fn is_authorised_by_state_before(
&self,
incoming_pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
create_event: &PduEvent,
origin: &ServerName,
) -> Result<(bool, HashMap<u64, OwnedEventId>)> {
debug!(
event_id = %incoming_pdu.event_id,
"Resolving state at event"
);
let room_id = incoming_pdu.room_id_or_hash();
// If the incoming event only has one prev event, we can just use the state at
// that event, but otherwise we have to resolve across each fork. If we're
// missing even one of the prev events, we have to ask a remote server for help.
//
// TODO: this can be optimised by only loading auth chain events into memory,
// rather than the entire state.
let state_before = if incoming_pdu.prev_events().count() == 1 {
self.state_at_incoming_degree_one(&incoming_pdu).await?
} else {
self.state_at_incoming_resolved(&incoming_pdu, &room_id, room_version_rules)
.await?
};
let state_before = match state_before {
| Some(s) => s,
| None => {
trace!("Could not calculate incoming state, asking remote {origin} for it");
self.fetch_state(origin, create_event, &room_id, incoming_pdu.event_id())
.await
.debug_inspect_err(|e| {
debug_error!("Could not fetch state from {origin}: {e}");
})?
},
};
if state_before.is_empty()
&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
{
// This can happen if the remote sends an event but cannot be reached to fetch
// the state at it, and all other servers in the room (which might just be the
// unreachable server) are unable to provide required info.
// returning an error here allows the upgrade to be attempted at another time.
return Err!(Request(Forbidden("Could not resolve incoming state before event")));
}
trace!(state_events = state_before.len(), "Calculated incoming state");
let state_fetch_state = &state_before;
let state_fetch = |k: StateEventType, s: StateKey| async move {
let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
let event_id = state_fetch_state.get(&shortstatekey)?;
self.services.timeline.get_pdu(event_id).await.ok()
};
debug!(
event_id = %incoming_pdu.event_id,
"Running state-before auth check"
);
// PDU check: 5
let auth_check = state_res::event_auth::auth_check(
room_version_rules,
incoming_pdu,
None, // TODO: third party invite
|ty, sk| state_fetch(ty.clone(), sk.into()),
create_event.as_pdu(),
)
.await
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
Ok((auth_check, state_before))
}
/// Checks that the event passes PDU check 6, which ensures that the event
/// is authorised based on the room's current state (which is the resolved
/// state across all current forward extremities).
///
/// Returns a boolean indicating whether the event is authorised, or an
/// error if the auth check fails.
async fn is_authorised_by_current_state(
&self,
incoming_pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
create_event: &PduEvent,
) -> Result<bool> {
debug!(
event_id = %incoming_pdu.event_id,
"Gathering auth events"
);
let auth_events = self
.services
.state
.get_auth_events(
&incoming_pdu.room_id_or_hash(),
incoming_pdu.kind(),
incoming_pdu.sender(),
incoming_pdu.state_key(),
incoming_pdu.content(),
room_version_rules,
)
.await?;
let state_fetch = |k: &StateEventType, s: &str| {
let key = k.with_state_key(s);
ready(auth_events.get(&key).map(ToOwned::to_owned))
};
debug!(
event_id = %incoming_pdu.event_id,
"Running current state auth check"
);
state_res::event_auth::auth_check(
room_version_rules,
incoming_pdu,
None, // third-party invite
state_fetch,
create_event.as_pdu(),
)
.await
.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))
}
/// Performs PDU check 7 - does the policy server allow this event.
///
/// If the policy server forbids the event, false is returned. If there is a
/// problem contacting the policy server, or it returns an unrecognised
/// response, an appropriate error is returned.
async fn policy_server_check(
&self,
incoming_pdu: &PduEvent,
pdu_json: &mut CanonicalJsonObject,
room_version_rules: &RoomVersionRules,
) -> Result<bool> {
let event_id = pdu_json
.remove("event_id")
.expect("event_id should be present in pdu_json at this stage");
if let Err(e) = self
.policy_server_allows_event(
incoming_pdu,
pdu_json,
&incoming_pdu.room_id_or_hash(),
room_version_rules,
true,
)
.await
.debug_inspect(|()| {
debug!(
event_id = %incoming_pdu.event_id,
"Event has passed policy server check."
);
}) {
return if matches!(e.kind(), ErrorKind::Forbidden) {
info!(
event_id = %incoming_pdu.event_id,
error = %e,
"Event has been marked as spam by policy server: {}",
e.message(),
);
Ok(false)
} else {
Err(e)
};
}
pdu_json.insert("event_id".to_owned(), event_id);
Ok(true)
}
/// Derives new room state from the incoming event and filters forward
/// extremities accordingly. Does not set forward extremities.
///
/// Only call this function if the incoming PDU is not soft-failed or
/// rejected.
async fn progress_state_and_extremities(
&self,
incoming_pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
state_before: HashMap<u64, OwnedEventId>,
forward_extremities: Vec<OwnedEventId>,
state_lock: &Guard<OwnedRoomId, ()>,
) -> Result<Vec<OwnedEventId>> {
if incoming_pdu.state_key().is_some() {
debug!("Event is a state-event. Deriving new room state");
self.derive_new_state(incoming_pdu, room_version_rules, state_before, state_lock)
.await?;
}
// Now we calculate the set of extremities this room has after the incoming
// event has been applied. We start with the previous extremities
trace!("Calculating extremities");
let mut forward_extremities = forward_extremities
.into_iter()
.stream()
.ready_filter(|event_id| {
// Remove any that are referenced by this incoming event's prev_events
!incoming_pdu.prev_events().any(is_equal_to!(event_id))
})
.broad_filter_map(|event_id| async move {
// Only keep those extremities were not referenced yet
self.services
.pdu_metadata
.is_event_referenced(&incoming_pdu.room_id_or_hash(), &event_id)
.await
.eq(&false)
.then_some(event_id)
})
.collect::<Vec<_>>()
.await;
forward_extremities.push(incoming_pdu.event_id().to_owned());
debug!(
"Retained {} extremities checked against {} prev_events",
forward_extremities.len(),
incoming_pdu.prev_events().count()
);
assert!(!forward_extremities.is_empty(), "resolved extremities cannot be empty");
Ok(forward_extremities)
}
/// Derives a new room state by adding the incoming PDU to the state before
/// it to create the state at, which then becomes the current room state.
///
/// The caller MUST ensure forward extremities are set appropriately,
/// including this incoming pdu, either before or after calling this
/// function. Failing to do so will result in an inconsistent current state
/// cache, which may affect event authentication.
async fn derive_new_state(
&self,
incoming_pdu: &PduEvent,
room_version_rules: &RoomVersionRules,
state_before: HashMap<u64, OwnedEventId>,
state_lock: &Guard<OwnedRoomId, ()>,
) -> Result {
let room_id = incoming_pdu.room_id_or_hash();
// We also add state after incoming event to the fork states
let mut state_at_incoming_event = state_before;
let shortstatekey = self
.services
.short
.get_or_create_shortstatekey(
&incoming_pdu.kind().to_string().into(),
incoming_pdu.state_key().unwrap(),
)
.await;
let event_id = incoming_pdu.event_id();
state_at_incoming_event.insert(shortstatekey, event_id.to_owned());
debug!("Resolving new room state");
let new_room_state = self
.resolve_state(&room_id, room_version_rules, state_at_incoming_event)
.await?;
debug!("Forcing new room state");
let HashSetCompressStateEvent { shortstatehash, added, removed } = self
.services
.state_compressor
.save_state(&room_id, new_room_state)
.await?;
self.services
.state
.force_state(&room_id, shortstatehash, added, removed, state_lock)
.await
}
}
src/service/rooms/membership/mod.rs
+1 -1
View File
@@ -520,7 +520,7 @@ impl Service {
return state;
},
};
if !pdu_fits(&mut value.clone()) {
if !pdu_fits(&value.clone()) {
warn!(
"dropping incoming PDU {event_id} in room {room_id} from room join \
because it exceeds 65535 bytes or is otherwise too large."
src/service/rooms/timeline/create.rs
+4 -2
View File
@@ -22,7 +22,9 @@ use serde_json::value::{RawValue, to_raw_value};
use super::RoomMutexGuard;
pub fn pdu_fits(owned_obj: &mut CanonicalJsonObject) -> bool {
/// Ensures the given PDU fits inside the size limits for a PDU.
#[must_use]
pub fn pdu_fits(owned_obj: &CanonicalJsonObject) -> bool {
// room IDs, event IDs, senders, types, and state keys must all be <= 255 bytes
if let Some(CanonicalJsonValue::String(room_id)) = owned_obj.get("room_id") {
if room_id.len() > 255 {
@@ -311,7 +313,7 @@ impl super::Service {
pdu_json
.insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into()));
// Verify that the *full* PDU isn't over 64KiB.
if !pdu_fits(&mut pdu_json.clone()) {
if !pdu_fits(&pdu_json.clone()) {
// feckin huge PDU mate
return Err!(Request(TooLarge("Message/PDU is too long (exceeds 65535 bytes)")));
}