oss/conduit
oss/conduit
1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2025-06-27 16:35:59 +00:00
Code Wiki Activity

chore: bump ruma and axum

Browse source 
(cherry picked from commit 5b68ce890d)
This commit is contained in:
Matthias Ahouansou 2025-06-22 00:22:55 +01:00
parent 321b7cf8c0
commit 75e0b55b16
No known key found for this signature in database
14 changed files with 492 additions and 369 deletions
Download patch file Download diff file Expand all files Collapse all files

549
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

11
Cargo.toml
View file

@ -28,18 +28,17 @@ workspace = true
[dependencies]
# Web framework
# Can't bump until https://github.com/ruma/ruma/pull/1846 is merged or superseded
axum = { version = "0.7", default-features = false, features = [
axum = { version = "0.8", default-features = false, features = [
"form",
"http1",
"http2",
"json",
"matched-path",
], optional = true }
axum-extra = { version = "0.9", features = ["typed-header"] }
axum-server = { version = "0.6", features = ["tls-rustls"] }
tower = { version = "0.4", features = ["util"] }
tower-http = { version = "0.5", features = [
axum-extra = { version = "0.10", features = ["typed-header"] }
axum-server = { version = "0.7", features = ["tls-rustls"] }
tower = { version = "0.5", features = ["util"] }
tower-http = { version = "0.6", features = [
"add-extension",
"cors",
"sensitive-headers",

4
src/api/appservice_server.rs
View file

@ -1,4 +1,4 @@
use crate::{services, utils, Error, Result, MATRIX_VERSIONS};
use crate::{services, utils, Error, Result, SUPPORTED_VERSIONS};
use bytes::BytesMut;
use ruma::api::{appservice::Registration, IncomingResponse, OutgoingRequest, SendAccessToken};
use std::{fmt::Debug, mem, time::Duration};
@ -28,7 +28,7 @@ where
.try_into_http_request::<BytesMut>(
&destination,
SendAccessToken::IfRequired(hs_token),
MATRIX_VERSIONS,
&SUPPORTED_VERSIONS,
)
.unwrap()
.map(|body| body.freeze());

3
src/api/client_server/capabilities.rs
View file

@ -1,6 +1,7 @@
use crate::{services, Result, Ruma};
use ruma::api::client::discovery::get_capabilities::{
self, Capabilities, RoomVersionStability, RoomVersionsCapability,
self,
v3::{Capabilities, RoomVersionStability, RoomVersionsCapability},
};
use std::collections::BTreeMap;

11
src/api/client_server/sync.rs
View file

@ -1757,16 +1757,7 @@ pub async fn sync_events_v5_route(
),
num_live: None, // Count events in timeline greater than global sync counter
bump_stamp,
heroes: if body
.room_subscriptions
.get(room_id)
.map(|sub| sub.include_heroes.unwrap_or_default())
.unwrap_or_default()
{
Some(heroes)
} else {
None
},
heroes: Some(heroes),
},
);
}

11
src/api/ruma_wrapper/axum.rs
View file

@ -1,7 +1,6 @@
use std::{collections::BTreeMap, iter::FromIterator, str};
use axum::{
async_trait,
body::Body,
extract::{FromRequest, Path},
response::{IntoResponse, Response},
@ -34,10 +33,10 @@ enum Token {
None,
}
#[async_trait]
impl<T, S> FromRequest<S> for Ruma<T>
where
T: IncomingRequest,
S: Sync,
{
type Rejection = Error;
@ -65,7 +64,13 @@ where
};
let metadata = T::METADATA;
let auth_header: Option<TypedHeader<Authorization<Bearer>>> = parts.extract().await?;
let auth_header: Option<TypedHeader<Authorization<Bearer>>> =
// If X-Matrix signatures are used, it causes this extraction to fail with an error
if metadata.authentication != AuthScheme::ServerSignatures {
parts.extract().await?
} else {
None
};
let path_params: Path<Vec<String>> = parts.extract().await?;
let query = parts.uri.query().unwrap_or_default();

4
src/api/server_server.rs
View file

@ -7,7 +7,7 @@ use crate::{
media::FileMeta,
pdu::{gen_event_id_canonical_json, PduBuilder},
},
services, utils, Error, PduEvent, Result, Ruma, MATRIX_VERSIONS,
services, utils, Error, PduEvent, Result, Ruma, SUPPORTED_VERSIONS,
};
use axum::{response::IntoResponse, Json};
use axum_extra::headers::{CacheControl, Header};
@ -214,7 +214,7 @@ where
.try_into_http_request::<Vec<u8>>(
&actual_destination_str,
SendAccessToken::IfRequired(""),
MATRIX_VERSIONS,
&SUPPORTED_VERSIONS,
)
.map_err(|e| {
warn!(

12
src/lib.rs
View file

@ -8,17 +8,23 @@ mod utils;
// Not async due to services() being used in many closures, and async closures are not stable as of writing
// This is the case for every other occurrence of sync Mutex/RwLock, except for database related ones, where
// the current maintainer (Timo) has asked to not modify those
use std::sync::RwLock;
use std::{
collections::BTreeSet,
sync::{LazyLock, RwLock},
};
pub use api::ruma_wrapper::{Ruma, RumaResponse};
pub use config::Config;
pub use database::KeyValueDatabase;
use ruma::api::MatrixVersion;
use ruma::api::{MatrixVersion, SupportedVersions};
pub use service::{pdu::PduEvent, Services};
pub use utils::error::{Error, Result};
pub static SERVICES: RwLock<Option<&'static Services>> = RwLock::new(None);
pub const MATRIX_VERSIONS: &[MatrixVersion] = &[MatrixVersion::V1_13];
pub static SUPPORTED_VERSIONS: LazyLock<SupportedVersions> = LazyLock::new(|| SupportedVersions {
versions: BTreeSet::from_iter([MatrixVersion::V1_13]),
features: Vec::new(),
});
pub fn services() -> &'static Services {
SERVICES

21
src/main.rs
View file

@ -401,23 +401,23 @@ fn routes(config: &Config) -> Router {
// Ruma doesn't have support for multiple paths for a single endpoint yet, and these routes
// share one Ruma request / response type pair with {get,send}_state_event_for_key_route
.route(
"/_matrix/client/r0/rooms/:room_id/state/:event_type",
"/_matrix/client/r0/rooms/{room_id}/state/{event_type}",
get(client_server::get_state_events_for_empty_key_route)
.put(client_server::send_state_event_for_empty_key_route),
)
.route(
"/_matrix/client/v3/rooms/:room_id/state/:event_type",
"/_matrix/client/v3/rooms/{room_id}/state/{event_type}",
get(client_server::get_state_events_for_empty_key_route)
.put(client_server::send_state_event_for_empty_key_route),
)
// These two endpoints allow trailing slashes
.route(
"/_matrix/client/r0/rooms/:room_id/state/:event_type/",
"/_matrix/client/r0/rooms/{room_id}/state/{event_type}/",
get(client_server::get_state_events_for_empty_key_route)
.put(client_server::send_state_event_for_empty_key_route),
)
.route(
"/_matrix/client/v3/rooms/:room_id/state/:event_type/",
"/_matrix/client/v3/rooms/{room_id}/state/{event_type}/",
get(client_server::get_state_events_for_empty_key_route)
.put(client_server::send_state_event_for_empty_key_route),
)
@ -459,11 +459,11 @@ fn routes(config: &Config) -> Router {
.ruma_route(client_server::get_hierarchy_route)
.ruma_route(client_server::well_known_client)
.route(
"/_matrix/client/r0/rooms/:room_id/initialSync",
"/_matrix/client/r0/rooms/{room_id}/initialSync",
get(initial_sync),
)
.route(
"/_matrix/client/v3/rooms/:room_id/initialSync",
"/_matrix/client/v3/rooms/{room_id}/initialSync",
get(initial_sync),
)
.route("/", get(it_works))
@ -477,7 +477,7 @@ fn routes(config: &Config) -> Router {
get(server_server::get_server_keys_route),
)
.route(
"/_matrix/key/v2/server/:key_id",
"/_matrix/key/v2/server/{key_id}",
get(server_server::get_server_keys_deprecated_route),
)
.ruma_route(server_server::get_public_rooms_route)
@ -509,8 +509,8 @@ fn routes(config: &Config) -> Router {
.ruma_route(server_server::well_known_server)
} else {
router
.route("/_matrix/federation/*path", any(federation_disabled))
.route("/_matrix/key/*path", any(federation_disabled))
.route("/_matrix/federation/{*path}", any(federation_disabled))
.route("/_matrix/key/{*path}", any(federation_disabled))
.route("/.well-known/matrix/server", any(federation_disabled))
}
}
@ -595,12 +595,11 @@ pub trait RumaHandler<T> {
macro_rules! impl_ruma_handler {
( $($ty:ident),* $(,)? ) => {
#[axum::async_trait]
#[allow(non_snake_case)]
impl<Req, E, F, Fut, $($ty,)*> RumaHandler<($($ty,)* Ruma<Req>,)> for F
where
Req: IncomingRequest + Send + 'static,
F: FnOnce($($ty,)* Ruma<Req>) -> Fut + Clone + Send + 'static,
F: FnOnce($($ty,)* Ruma<Req>) -> Fut + Clone + Send + Sync + 'static,
Fut: Future<Output = Result<Req::OutgoingResponse, E>>
+ Send,
E: IntoResponse,

7
src/service/pdu.rs
View file

@ -408,6 +408,13 @@ impl state_res::Event for PduEvent {
fn redacts(&self) -> Option<&Self::Id> {
self.redacts.as_ref()
}
// We currently don't store rejected events (see steps 6-8 of `handle_incoming_pdu`), even
// though we should according to the spec:
// https://spec.matrix.org/v1.14/rooms/v11/#rejected-events
fn rejected(&self) -> bool {
false
}
}
// These impl's allow us to dedup state snapshots when resolving state

4
src/service/pusher/mod.rs
View file

@ -2,7 +2,7 @@ mod data;
pub use data::Data;
use ruma::{events::AnySyncTimelineEvent, push::PushConditionPowerLevelsCtx};
use crate::{services, Error, PduEvent, Result, MATRIX_VERSIONS};
use crate::{services, Error, PduEvent, Result, SUPPORTED_VERSIONS};
use bytes::BytesMut;
use ruma::{
api::{
@ -58,7 +58,7 @@ impl Service {
.try_into_http_request::<BytesMut>(
&destination,
SendAccessToken::IfRequired(""),
MATRIX_VERSIONS,
&SUPPORTED_VERSIONS,
)
.map_err(|e| {
warn!("Failed to find destination {}: {}", destination, e);

123
src/service/rooms/event_handler/mod.rs
View file

@ -51,25 +51,26 @@ impl Service {
/// 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
/// 2. Check event is valid, otherwise drop
/// 3. Check signatures, otherwise drop
/// 4. Check content hash, redact if doesn't match
/// 5. 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
/// 6. 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
/// 7. Reject "due to auth events" if the event doesn't pass auth based on the auth events
/// 8. Persist this event as an outlier
/// 9. If not timeline event: stop
/// 10. 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
/// 11. 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
/// 12. Check the auth of the event passes based on the state of the event
/// 13. 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
/// 14. Use state resolution to find new room state
/// 15. Check if the event passes auth based on the "current state" of the room, if not soft fail it
// We use some AsyncRecursiveType hacks here so we can call this async function recursively
#[tracing::instrument(skip(self, value, is_timeline_event, pub_key_map))]
pub(crate) async fn handle_incoming_pdu<'a>(
@ -135,7 +136,7 @@ impl Service {
.await?;
self.check_room_id(room_id, &incoming_pdu)?;
// 8. if not timeline event: stop
// 9. if not timeline event: stop
if !is_timeline_event {
return Ok(None);
}
@ -145,7 +146,7 @@ impl Service {
return Ok(None);
}
// 9. Fetch any missing prev events doing all checks listed here starting at 1. These are timeline events
// 10. 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_unknown_prev_events(
origin,
@ -313,8 +314,9 @@ impl Service {
// 1.1. Remove unsigned field
value.remove("unsigned");
// 2. Check signatures, otherwise drop
// 3. check content hash, redact if doesn't match
// 2. Check event is valid, otherwise drop
// 3. Check signatures, otherwise drop
// 4. check content hash, redact if doesn't match
let create_event_content: RoomCreateEventContent =
serde_json::from_str(create_event.content.get()).map_err(|e| {
error!("Invalid create event: {}", e);
@ -326,6 +328,15 @@ impl Service {
.rules()
.expect("Supported room version has rules");
debug!("Checking format of join event PDU");
if let Err(e) = state_res::check_pdu_format(&value, &room_version_rules.event_format) {
warn!("Invalid PDU with event ID {event_id} received: {e}");
return Err(Error::BadRequest(
ErrorKind::InvalidParam,
"Received Invalid PDU",
));
}
// TODO: For RoomVersion6 we must check that Raw<..> is canonical do we anywhere?: https://matrix.org/docs/spec/rooms/v6#canonical-json
// We go through all the signatures we see on the value and fetch the corresponding signing
@ -421,8 +432,8 @@ impl Service {
self.check_room_id(room_id, &incoming_pdu)?;
if !auth_events_known {
// 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"
// 5. fetch any missing auth events doing all checks listed here starting at 1. These are not timeline events
// 6. 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"
// NOTE: Step 5 is not applied anymore because it failed too often
debug!(event_id = ?incoming_pdu.event_id, "Fetching auth events");
self.fetch_and_handle_outliers(
@ -440,7 +451,7 @@ impl Service {
.await;
}
// 6. Reject "due to auth events" if the event doesn't pass auth based on the auth events
// 7. Reject "due to auth events" if the event doesn't pass auth based on the auth events
debug!(
"Auth check for {} based on auth events",
incoming_pdu.event_id
@ -448,6 +459,7 @@ impl Service {
// Build map of auth events
let mut auth_events = HashMap::new();
let mut auth_events_by_event_id = HashMap::new();
for id in &incoming_pdu.auth_events {
let auth_event = match services().rooms.timeline.get_pdu(id)? {
Some(e) => e,
@ -457,46 +469,33 @@ impl Service {
}
};
self.check_room_id(room_id, &auth_event)?;
match auth_events.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(_) => {
return Err(Error::BadRequest(
ErrorKind::InvalidParam,
"Auth event's type and state_key combination exists multiple times.",
));
}
}
auth_events_by_event_id.insert(auth_event.event_id.clone(), auth_event.clone());
auth_events.insert(
(
StateEventType::from(auth_event.kind.to_string()),
auth_event
.state_key
.clone()
.expect("all auth events have state keys"),
),
auth_event,
);
}
// The original create event must be in the auth events
if !matches!(
auth_events
.get(&(StateEventType::RoomCreate, "".to_owned()))
.map(|a| a.as_ref()),
Some(_) | None
) {
return Err(Error::BadRequest(
ErrorKind::InvalidParam,
"Incoming event refers to wrong create event.",
));
}
if state_res::event_auth::auth_check(
// first time we are doing any sort of auth check, so we check state-independent
// auth rules in addition to the state-dependent ones.
if state_res::check_state_independent_auth_rules(
&room_version_rules.authorization,
&incoming_pdu,
|k, s| auth_events.get(&(k.to_string().into(), s.to_owned())),
|event_id| auth_events_by_event_id.get(event_id),
)
.is_err()
|| state_res::check_state_dependent_auth_rules(
&room_version_rules.authorization,
&incoming_pdu,
|k, s| auth_events.get(&(k.to_string().into(), s.to_owned())),
)
.is_err()
{
return Err(Error::BadRequest(
ErrorKind::InvalidParam,
@ -506,7 +505,7 @@ impl Service {
debug!("Validation successful.");
// 7. Persist the event as an outlier.
// 8. Persist the event as an outlier.
services()
.rooms
.outlier
@ -557,7 +556,7 @@ impl Service {
.rules()
.expect("Supported room version has rules");
// 10. Fetch missing state and auth chain events by calling /state_ids at backwards extremities
// 11. 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.
// TODO: if we know the prev_events of the incoming event we can avoid the request and build
@ -807,8 +806,8 @@ impl Service {
state_at_incoming_event.expect("we always set this to some above");
debug!("Starting auth check");
// 11. Check the auth of the event passes based on the state of the event
if state_res::event_auth::auth_check(
// 12. Check the auth of the event passes based on the state of the event
if state_res::check_state_dependent_auth_rules(
&room_version_rules.authorization,
&incoming_pdu,
|k, s| {
@ -840,7 +839,7 @@ impl Service {
&room_version_rules.authorization,
)?;
let soft_fail = state_res::event_auth::auth_check(
let soft_fail = state_res::check_state_dependent_auth_rules(
&room_version_rules.authorization,
&incoming_pdu,
|k, s| auth_events.get(&(k.clone(), s.to_owned())),
@ -891,7 +890,7 @@ impl Service {
}
};
// 13. Use state resolution to find new room state
// 14. Use state resolution to find new room state
// We start looking at current room state now, so lets lock the room
let mutex_state = Arc::clone(
@ -974,7 +973,7 @@ impl Service {
.await?;
}
// 14. Check if the event passes auth based on the "current state" of the room, if not soft fail it
// 15. Check if the event passes auth based on the "current state" of the room, if not soft fail it
debug!("Starting soft fail auth check");
if soft_fail {
@ -1399,7 +1398,7 @@ impl Service {
}
}
let sorted = state_res::lexicographical_topological_sort(&graph, |event_id| {
let sorted = state_res::reverse_topological_power_sort(&graph, |event_id| {
// This return value is the key used for sorting events,
// events are then sorted by power level, time,
// and lexically by event_id.

75
src/service/rooms/helpers/mod.rs
View file

@ -169,6 +169,18 @@ impl Service {
}
}
let rules = room_version_id
.rules()
.expect("Supported room version has rules");
debug!("Checking format of join event PDU");
if let Err(e) = state_res::check_pdu_format(&join_event, &rules.event_format) {
warn!(
"Invalid PDU with event ID {event_id} received from `/send_join` response: {e}"
);
return Err(Error::BadServerResponse("Received Invalid PDU"));
}
services().rooms.short.get_or_create_shortroomid(room_id)?;
info!("Parsing join event");
@ -240,28 +252,32 @@ impl Service {
}
info!("Running send_join auth check");
if let Err(e) = state_res::event_auth::auth_check(
&room_version_id
.rules()
.expect("Supported room version has rules")
.authorization,
if let Err(e) = state_res::check_state_independent_auth_rules(
&rules.authorization,
&parsed_join_pdu,
|k, s| {
services()
.rooms
.timeline
.get_pdu(
state.get(
&services()
.rooms
.short
.get_or_create_shortstatekey(&k.to_string().into(), s)
.ok()?,
)?,
)
.ok()?
},
) {
|event_id| services().rooms.timeline.get_pdu(event_id).ok().flatten(),
)
.and_then(|_| {
state_res::check_state_dependent_auth_rules(
&rules.authorization,
&parsed_join_pdu,
|k, s| {
services()
.rooms
.timeline
.get_pdu(
state.get(
&services()
.rooms
.short
.get_or_create_shortstatekey(&k.to_string().into(), s)
.ok()?,
)?,
)
.ok()?
},
)
}) {
warn!("Auth check failed: {e}");
return Err(Error::BadRequest(
ErrorKind::InvalidParam,
@ -674,6 +690,15 @@ async fn validate_and_add_event_id(
}
}
let rules = &room_version
.rules()
.expect("Supported room version has rules");
if let Err(e) = state_res::check_pdu_format(&value, &rules.event_format) {
warn!("Invalid PDU with event ID {event_id} received from `/send_join` response: {e}");
return Err(Error::BadServerResponse("Received Invalid PDU"));
}
let origin_server_ts = value.get("origin_server_ts").ok_or_else(|| {
error!("Invalid PDU, no origin_server_ts field");
Error::BadRequest(
@ -702,13 +727,7 @@ async fn validate_and_add_event_id(
.globals
.filter_keys_server_map(unfiltered_keys, origin_server_ts, room_version);
if let Err(e) = ruma::signatures::verify_event(
&keys,
&value,
&room_version
.rules()
.expect("Supported room version has rules"),
) {
if let Err(e) = ruma::signatures::verify_event(&keys, &value, rules) {
warn!("Event {} failed verification {:?} {}", event_id, pdu, e);
back_off(event_id).await;
return Err(Error::BadServerResponse("Event failed verification."));

26
src/service/rooms/timeline/mod.rs
View file

@ -713,6 +713,10 @@ impl Service {
&content,
&room_version_rules.authorization,
)?;
let mut auth_events_by_event_id = HashMap::new();
for event in auth_events.values() {
auth_events_by_event_id.insert(event.event_id.clone(), event.clone());
}
// Our depth is the maximum depth of prev_events + 1
let depth = prev_events
@ -769,10 +773,18 @@ impl Service {
signatures: None,
};
if state_res::auth_check(&room_version_rules.authorization, &pdu, |k, s| {
auth_events.get(&(k.clone(), s.to_owned()))
})
if state_res::check_state_independent_auth_rules(
&room_version_rules.authorization,
&pdu,
|event_id| auth_events_by_event_id.get(event_id),
)
.is_err()
|| state_res::check_state_dependent_auth_rules(
&room_version_rules.authorization,
&pdu,
|k, s| auth_events.get(&(k.clone(), s.to_owned())),
)
.is_err()
{
return Err(Error::BadRequest(
ErrorKind::forbidden(),
@ -814,6 +826,14 @@ impl Service {
}
}
if let Err(e) = state_res::check_pdu_format(&pdu_json, &room_version_rules.event_format) {
warn!("locally constructed event is not a valid PDU: {e}");
return Err(Error::BadRequest(
ErrorKind::InvalidParam,
"Event is invalid",
));
}
// Generate event id
pdu.event_id = EventId::parse_arc(format!(
"${}",