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

style: Document & refactor rooms/event_handler service

Browse Source
This commit is contained in:
timedout committed 2026-06-27 18:36:44 +01:00
1 parent 0a162b7bc4
commit 19f4c24c00
13 files changed
+1884 -1846
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/service/rooms/event_handler/acl_check.rs
+30 -32
View File
@@ -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")))
}
}
}
src/service/rooms/event_handler/fetch_and_handle_outliers.rs
+202 -200
View File
@@ -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<BTreeMap<String, CanonicalJsonValue>>)>
where
Pdu: Event + Send + Sync,
Events: Iterator<Item = &'a EventId> + 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<BTreeMap<String, CanonicalJsonValue>>)>
where
Pdu: Event + Send + Sync,
Events: Iterator<Item = &'a EventId> + 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::<OwnedEventId>(
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::<OwnedEventId>(
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
}
src/service/rooms/event_handler/fetch_prev.rs
+97 -101
View File
@@ -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<OwnedEventId>,
HashMap<OwnedEventId, (PduEvent, BTreeMap<String, CanonicalJsonValue>)>,
)>
where
Pdu: Event + Send + Sync,
Events: Iterator<Item = &'a EventId> + Clone + Send,
{
let num_ids = initial_set.clone().count();
let mut eventid_info = HashMap::new();
let mut graph: HashMap<OwnedEventId, _> = HashMap::with_capacity(num_ids);
let mut todo_outlier_stack: VecDeque<OwnedEventId> =
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<OwnedEventId>,
HashMap<OwnedEventId, (PduEvent, BTreeMap<String, CanonicalJsonValue>)>,
)>
where
Pdu: Event + Send + Sync,
Events: Iterator<Item = &'a EventId> + Clone + Send,
{
let num_ids = initial_set.clone().count();
let mut eventid_info = HashMap::new();
let mut graph: HashMap<OwnedEventId, _> = HashMap::with_capacity(num_ids);
let mut todo_outlier_stack: VecDeque<OwnedEventId> =
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))
}
src/service/rooms/event_handler/fetch_state.rs
+73 -70
View File
@@ -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<Pdu>(
&self,
origin: &ServerName,
create_event: &Pdu,
room_id: &RoomId,
event_id: &EventId,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
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<ShortStateKey, OwnedEventId> = 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<Pdu>(
&self,
origin: &ServerName,
create_event: &Pdu,
room_id: &RoomId,
event_id: &EventId,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
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<ShortStateKey, OwnedEventId> =
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))
}
src/service/rooms/event_handler/handle_incoming_pdu.rs
+208 -193
View File
@@ -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<String, CanonicalJsonValue>,
is_timeline_event: bool,
) -> Result<Option<RawPduId>> {
// 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<String, CanonicalJsonValue>,
is_timeline_event: bool,
) -> Result<Option<RawPduId>> {
// 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
}
}
src/service/rooms/event_handler/handle_outlier_pdu.rs
+213 -206
View File
@@ -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<String, CanonicalJsonValue>)>
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<String, CanonicalJsonValue>)>
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::<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)?;
// Fetch all auth events
let mut auth_events: HashMap<OwnedEventId, PduEvent> = 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::<Vec<_>>()
} else {
pdu_event.auth_events().collect::<Vec<_>>()
};
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::<Vec<_>>();
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::<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)?;
// Fetch all auth events
let mut auth_events: HashMap<OwnedEventId, PduEvent> = 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::<Vec<_>>()
} else {
pdu_event.auth_events().collect::<Vec<_>>()
};
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::<Vec<_>>();
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))
}
src/service/rooms/event_handler/handle_prev_pdu.rs
+73 -67
View File
@@ -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<String, CanonicalJsonValue>)>,
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<String, CanonicalJsonValue>)>,
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(())
}
}
src/service/rooms/event_handler/mod.rs
+3
View File
@@ -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<PduEvent> {
self.services.timeline.get_pdu(&event_id).await.ok()
}
src/service/rooms/event_handler/parse_incoming_pdu.rs
+51 -49
View File
@@ -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<Vec
.try_collect()
}
/// 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.
#[implement(super::Service)]
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")));
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<Parsed> {
let value = serde_json::from_str::<CanonicalJsonObject>(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<Parsed> {
let value = serde_json::from_str::<CanonicalJsonObject>(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))
}
}
src/service/rooms/event_handler/policy_server.rs
+324 -321
View File
@@ -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::<Standard>::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::<RoomPolicyEventContent>(
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::<RoomPolicyEventContent>(
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<RawValue>,
room_id: &RoomId,
policy_server_key: &Base64<Standard, Vec<u8>>,
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<RawValue>,
room_id: &RoomId,
policy_server_key: &Base64<Standard, Vec<u8>>,
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<RawValue>,
room_id: &RoomId,
policy_server_key: &Base64<Standard, Vec<u8>>,
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<RawValue>,
room_id: &RoomId,
policy_server_key: &Base64<Standard, Vec<u8>>,
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(())
}
src/service/rooms/event_handler/resolve_state.rs
+96 -98
View File
@@ -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<u64, OwnedEventId>,
) -> Result<Arc<CompressedState>> {
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<u64, OwnedEventId>,
) -> Result<Arc<CompressedState>> {
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::<Vec<HashSet<OwnedEventId>>>();
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::<Vec<HashSet<OwnedEventId>>>();
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::<Vec<StateMap<OwnedEventId>>>();
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::<Vec<StateMap<OwnedEventId>>>();
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<OwnedEventId>],
) -> Result<StateMap<OwnedEventId>>
where
StateSets: Iterator<Item = &'a StateMap<OwnedEventId>> + 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<OwnedEventId>],
) -> Result<StateMap<OwnedEventId>>
where
StateSets: Iterator<Item = &'a StateMap<OwnedEventId>> + 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
}
}
src/service/rooms/event_handler/state_at_incoming.rs
+162 -161
View File
@@ -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<Pdu>(
&self,
incoming_pdu: &Pdu,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
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<Pdu>(
&self,
incoming_pdu: &Pdu,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
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<Pdu>(
&self,
incoming_pdu: &Pdu,
room_id: &RoomId,
room_version_rules: &RoomVersionRules,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
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::<HashMap<_, _>>()
.await
else {
return Ok(None);
};
trace!("Calculating fork states...");
let (fork_states, auth_chain_sets): (Vec<StateMap<_>>, Vec<HashSet<_>>) =
extremity_sstatehashes
.into_iter()
#[tracing::instrument(name = "state", level = "debug", skip_all)]
pub(super) async fn state_at_incoming_resolved<Pdu>(
&self,
incoming_pdu: &Pdu,
room_id: &RoomId,
room_version_rules: &RoomVersionRules,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
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::<HashMap<_, _>>()
.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<StateMap<_>>, Vec<HashSet<_>>) =
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<Pdu>(
&self,
room_id: &RoomId,
sstatehash: ShortStateHash,
prev_event: Pdu,
) -> Result<(StateMap<OwnedEventId>, HashSet<OwnedEventId>)>
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<Pdu>(
&self,
room_id: &RoomId,
sstatehash: ShortStateHash,
prev_event: Pdu,
) -> Result<(StateMap<OwnedEventId>, HashSet<OwnedEventId>)>
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
}
}
src/service/rooms/event_handler/upgrade_outlier_pdu.rs
+352 -348
View File
@@ -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<Pdu>(
&self,
incoming_pdu: PduEvent,
mut val: CanonicalJsonObject,
create_event: &Pdu,
origin: &ServerName,
room_id: &RoomId,
) -> Result<Option<RawPduId>>
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<Pdu>(
&self,
incoming_pdu: PduEvent,
mut val: CanonicalJsonObject,
create_event: &Pdu,
origin: &ServerName,
room_id: &RoomId,
) -> Result<Option<RawPduId>>
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<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
.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<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
.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::<Vec<_>>()
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::<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");
}
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)
}