Merge branch 'bump-ruma' into 'next'

chore: bump ruma and axum

Closes #515

See merge request famedly/conduit!755

Cargo.toml

@@ -19,7 +19,7 @@ repository = "https://gitlab.com/famedly/conduit"
 version = "0.11.0-alpha"
 
 # See also `rust-toolchain.toml`
-rust-version = "1.83.0"
+rust-version = "1.85.0"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -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",

flake.nix

@@ -59,7 +59,7 @@
             file = ./rust-toolchain.toml;
 
             # See also `rust-toolchain.toml`
-            sha256 = "sha256-s1RPtyvDGJaX/BisLT+ifVfuhDT1nZkZ1NcK8sbwELM=";
+            sha256 = "sha256-AJ6LX/Q/Er9kS15bn9iflkUwcgYqRQxiOIL2ToVAXaU=";
           };
       });
     in

rust-toolchain.toml

@@ -9,7 +9,7 @@
 # If you're having trouble making the relevant changes, bug a maintainer.
 
 [toolchain]
-channel = "1.83.0"
+channel = "1.85.0"
 components = [
   # For rust-analyzer
   "rust-src",

src/api/appservice_server.rs

@@ -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());

src/api/client_server/capabilities.rs

@@ -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;
 

src/api/client_server/membership.rs

@@ -656,7 +656,7 @@ pub async fn joined_members_route(
     Ok(joined_members::v3::Response { joined })
 }
 
-pub(crate) async fn invite_helper<'a>(
+pub(crate) async fn invite_helper(
     sender_user: &UserId,
     user_id: &UserId,
     room_id: &RoomId,

src/api/client_server/session.rs

@@ -182,11 +182,11 @@ pub async fn login_route(body: Ruma<login::v3::Request>) -> Result<login::v3::Re
     let token = utils::random_string(TOKEN_LENGTH);
 
     // Determine if device_id was provided and exists in the db for this user
-    let device_exists = body.device_id.as_ref().map_or(false, |device_id| {
+    let device_exists = body.device_id.as_ref().is_some_and(|device_id| {
         services()
             .users
             .all_device_ids(&user_id)
-            .any(|x| x.as_ref().map_or(false, |v| v == device_id))
+            .any(|x| x.as_ref().is_ok_and(|v| v == device_id))
     });
 
     if device_exists {

src/api/client_server/sync.rs

@@ -818,8 +818,8 @@ async fn load_joined_room(
                         .ok()
                 });
 
-            let joined_since_last_sync = since_sender_member
-                .map_or(true, |member| member.membership != MembershipState::Join);
+            let joined_since_last_sync =
+                since_sender_member.is_none_or(|member| member.membership != MembershipState::Join);
 
             if since_shortstatehash.is_none() || joined_since_last_sync {
                 // Probably since = 0, we will do an initial sync
@@ -1402,7 +1402,7 @@ pub async fn sync_events_v5_route(
                 )?;
 
                 let joined_since_last_sync = since_sender_member
-                    .map_or(true, |member| member.membership != MembershipState::Join);
+                    .is_none_or(|member| member.membership != MembershipState::Join);
 
                 let new_encrypted_room = encrypted_room && since_encryption.is_none();
                 if encrypted_room {
@@ -1779,16 +1779,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),
             },
         );
     }

src/api/client_server/user_directory.rs

@@ -61,12 +61,11 @@ pub async fn search_users_route(
                     .rooms
                     .state_accessor
                     .room_state_get(&room, &StateEventType::RoomJoinRules, "")
-                    .map_or(false, |event| {
-                        event.map_or(false, |event| {
-                            serde_json::from_str(event.content.get())
-                                .map_or(false, |r: RoomJoinRulesEventContent| {
-                                    r.join_rule == JoinRule::Public
-                                })
+                    .is_ok_and(|event| {
+                        event.is_some_and(|event| {
+                            serde_json::from_str(event.content.get()).is_ok_and(
+                                |r: RoomJoinRulesEventContent| r.join_rule == JoinRule::Public,
+                            )
                         })
                     })
             });

src/api/ruma_wrapper/axum.rs

@@ -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();

src/api/server_server.rs

@@ -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!(

src/database/key_value/pusher.rs

@@ -22,10 +22,7 @@ impl service::pusher::Data for KeyValueDatabase {
                 let mut key = sender.as_bytes().to_vec();
                 key.push(0xff);
                 key.extend_from_slice(ids.pushkey.as_bytes());
-                self.senderkey_pusher
-                    .remove(&key)
-                    .map(|_| ())
-                    .map_err(Into::into)
+                self.senderkey_pusher.remove(&key).map(|_| ())
             }
         }
     }

src/database/key_value/rooms/state_cache.rs

@@ -221,9 +221,9 @@ impl service::rooms::state_cache::Data for KeyValueDatabase {
             .ok();
 
             let in_room = bridge_user_id
-                .map_or(false, |id| self.is_joined(&id, room_id).unwrap_or(false))
+                .is_some_and(|id| self.is_joined(&id, room_id).unwrap_or(false))
                 || self.room_members(room_id).any(|userid| {
-                    userid.map_or(false, |userid| appservice.users.is_match(userid.as_str()))
+                    userid.is_ok_and(|userid| appservice.users.is_match(userid.as_str()))
                 });
 
             self.appservice_in_room_cache
@@ -273,7 +273,7 @@ impl service::rooms::state_cache::Data for KeyValueDatabase {
     }
 
     #[tracing::instrument(skip(self))]
-    fn server_in_room<'a>(&'a self, server: &ServerName, room_id: &RoomId) -> Result<bool> {
+    fn server_in_room(&self, server: &ServerName, room_id: &RoomId) -> Result<bool> {
         let mut key = server.as_bytes().to_vec();
         key.push(0xff);
         key.extend_from_slice(room_id.as_bytes());

src/database/mod.rs

@@ -491,7 +491,7 @@ impl KeyValueDatabase {
                 for (userid, password) in db.userid_password.iter() {
                     let password = utils::string_from_bytes(&password);
 
-                    let empty_hashed_password = password.map_or(false, |password| {
+                    let empty_hashed_password = password.is_ok_and(|password| {
                         argon2::verify_encoded(&password, b"").unwrap_or(false)
                     });
 

src/lib.rs

@@ -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

src/main.rs

@@ -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,

src/service/pdu.rs

@@ -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

src/service/pusher/mod.rs

@@ -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);

src/service/rooms/event_handler/mod.rs

@@ -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,41 +469,28 @@ impl Service {
                     }
                 };
 
-                self.check_room_id(room_id, &auth_event)?;
-
-                match auth_events.entry((
-                    auth_event.kind.to_string().into(),
+                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"),
-                )) {
-                    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_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,
+                |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())),
@@ -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.

src/service/rooms/helpers/mod.rs

@@ -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,11 +252,14 @@ 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,
+                |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()
@@ -261,7 +276,8 @@ impl Service {
                             )
                             .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."));

src/service/rooms/pdu_metadata/mod.rs

@@ -87,13 +87,13 @@ impl Service {
                 let events_after: Vec<_> = relations_until // TODO: should be relations_after
                     .iter()
                     .filter(|(_, pdu)| {
-                        filter_event_type.as_ref().map_or(true, |t| &pdu.kind == t)
+                        filter_event_type.as_ref().is_none_or(|t| &pdu.kind == t)
                             && if let Ok(content) =
                                 serde_json::from_str::<ExtractRelatesToEventId>(pdu.content.get())
                             {
                                 filter_rel_type
                                     .as_ref()
-                                    .map_or(true, |r| &content.relates_to.rel_type == r)
+                                    .is_none_or(|r| &content.relates_to.rel_type == r)
                             } else {
                                 false
                             }
@@ -135,13 +135,13 @@ impl Service {
                 let events_before: Vec<_> = relations_until
                     .iter()
                     .filter(|(_, pdu)| {
-                        filter_event_type.as_ref().map_or(true, |t| &pdu.kind == t)
+                        filter_event_type.as_ref().is_none_or(|t| &pdu.kind == t)
                             && if let Ok(content) =
                                 serde_json::from_str::<ExtractRelatesToEventId>(pdu.content.get())
                             {
                                 filter_rel_type
                                     .as_ref()
-                                    .map_or(true, |r| &content.relates_to.rel_type == r)
+                                    .is_none_or(|r| &content.relates_to.rel_type == r)
                             } else {
                                 false
                             }

src/service/rooms/search/mod.rs

@@ -11,17 +11,12 @@ pub struct Service {
 
 impl Service {
     #[tracing::instrument(skip(self))]
-    pub fn index_pdu<'a>(&self, shortroomid: u64, pdu_id: &[u8], message_body: &str) -> Result<()> {
+    pub fn index_pdu(&self, shortroomid: u64, pdu_id: &[u8], message_body: &str) -> Result<()> {
         self.db.index_pdu(shortroomid, pdu_id, message_body)
     }
 
     #[tracing::instrument(skip(self))]
-    pub fn deindex_pdu<'a>(
-        &self,
-        shortroomid: u64,
-        pdu_id: &[u8],
-        message_body: &str,
-    ) -> Result<()> {
+    pub fn deindex_pdu(&self, shortroomid: u64, pdu_id: &[u8], message_body: &str) -> Result<()> {
         self.db.deindex_pdu(shortroomid, pdu_id, message_body)
     }
 

src/service/rooms/spaces/mod.rs

@@ -629,7 +629,7 @@ fn next_room_to_traverse(
     stack: &mut Vec<Vec<(OwnedRoomId, Vec<OwnedServerName>)>>,
     parents: &mut VecDeque<OwnedRoomId>,
 ) -> Option<(OwnedRoomId, Vec<OwnedServerName>)> {
-    while stack.last().map_or(false, |s| s.is_empty()) {
+    while stack.last().is_some_and(|s| s.is_empty()) {
         stack.pop();
         parents.pop_back();
     }
@@ -663,7 +663,7 @@ async fn get_stripped_space_child_events(
             if serde_json::from_str::<SpaceChildEventContent>(pdu.content.get())
                 .ok()
                 .map(|c| c.via)
-                .map_or(true, |v| v.is_empty())
+                .is_none_or(|v| v.is_empty())
             {
                 continue;
             }

src/service/rooms/state_cache/mod.rs

@@ -171,7 +171,7 @@ impl Service {
                         })
                     })
                     .transpose()?
-                    .map_or(false, |ignored| {
+                    .is_some_and(|ignored| {
                         ignored
                             .content
                             .ignored_users
@@ -236,7 +236,7 @@ impl Service {
     }
 
     #[tracing::instrument(skip(self))]
-    pub fn server_in_room<'a>(&'a self, server: &ServerName, room_id: &RoomId) -> Result<bool> {
+    pub fn server_in_room(&self, server: &ServerName, room_id: &RoomId) -> Result<bool> {
         self.db.server_in_room(server, room_id)
     }
 

src/service/rooms/timeline/mod.rs

@@ -199,7 +199,7 @@ impl Service {
     ///
     /// Returns pdu id
     #[tracing::instrument(skip(self, pdu, pdu_json, leaves))]
-    pub async fn append_pdu<'a>(
+    pub async fn append_pdu(
         &self,
         pdu: &PduEvent,
         mut pdu_json: CanonicalJsonObject,
@@ -612,8 +612,8 @@ impl Service {
                 services().globals.server_name() == pdu.sender.server_name()
                     && appservice.is_user_match(&pdu.sender)
                     || pdu.kind == TimelineEventType::RoomMember
-                        && pdu.state_key.as_ref().map_or(false, |state_key| {
-                            UserId::parse(state_key).map_or(false, |user_id| {
+                        && pdu.state_key.as_ref().is_some_and(|state_key| {
+                            UserId::parse(state_key).is_ok_and(|user_id| {
                                 services().globals.server_name() == user_id.server_name()
                                     && appservice.is_user_match(&user_id)
                             })
@@ -633,8 +633,8 @@ impl Service {
                         "",
                     ) {
                         serde_json::from_str::<RoomCanonicalAliasEventContent>(pdu.content.get())
-                            .map_or(false, |content| {
-                                content.alias.map_or(false, |alias| {
+                            .is_ok_and(|content| {
+                                content.alias.is_some_and(|alias| {
                                     appservice.aliases.is_match(alias.as_str())
                                 }) || content
                                     .alt_aliases
@@ -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,9 +773,17 @@ 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(
@@ -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!(
             "${}",
@@ -1044,7 +1064,7 @@ impl Service {
     /// Append the incoming event setting the state snapshot to the state from the
     /// server that sent the event.
     #[tracing::instrument(skip_all)]
-    pub async fn append_incoming_pdu<'a>(
+    pub async fn append_incoming_pdu(
         &self,
         pdu: &PduEvent,
         pdu_json: CanonicalJsonObject,