style: Remove redundant length check for dag build

Tracking down weird soft-fails

Cargo.lock

@@ -4601,7 +4601,7 @@ dependencies = [
 [[package]]
 name = "ruma"
 version = "0.16.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "assign",
  "js_int",
@@ -4620,9 +4620,8 @@ dependencies = [
 [[package]]
 name = "ruma-appservice-api"
 version = "0.16.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
- "http",
  "js_int",
  "ruma-common",
  "ruma-events",
@@ -4633,7 +4632,7 @@ dependencies = [
 [[package]]
 name = "ruma-client-api"
 version = "0.24.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "as_variant",
  "assign",
@@ -4655,7 +4654,7 @@ dependencies = [
 [[package]]
 name = "ruma-common"
 version = "0.19.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "as_variant",
  "base64 0.22.1",
@@ -4688,7 +4687,7 @@ dependencies = [
 [[package]]
 name = "ruma-events"
 version = "0.34.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "as_variant",
  "indexmap 2.14.0",
@@ -4709,7 +4708,7 @@ dependencies = [
 [[package]]
 name = "ruma-federation-api"
 version = "0.15.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "bytes",
  "headers",
@@ -4732,7 +4731,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-validation"
 version = "0.12.1"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "js_int",
  "thiserror 2.0.18",
@@ -4741,7 +4740,7 @@ dependencies = [
 [[package]]
 name = "ruma-macros"
 version = "0.19.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "as_variant",
  "cfg-if",
@@ -4757,7 +4756,7 @@ dependencies = [
 [[package]]
 name = "ruma-push-gateway-api"
 version = "0.15.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -4769,7 +4768,7 @@ dependencies = [
 [[package]]
 name = "ruma-signatures"
 version = "0.21.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "base64 0.22.1",
  "ed25519-dalek",
@@ -4786,7 +4785,7 @@ dependencies = [
 [[package]]
 name = "ruma-state-res"
 version = "0.17.0"
-source = "git+https://github.com/gingershaped/ruwuma.git?rev=0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff#0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+source = "git+https://github.com/ruma/ruma.git?rev=fca1dbc060f730c1aaf84c9c2fb1e8a587455be8#fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 dependencies = [
  "js_int",
  "ruma-common",

Cargo.toml

@@ -349,8 +349,8 @@ version = "1.1.1"
 # Used for matrix spec type definitions and helpers
 [workspace.dependencies.ruma]
 # version = "0.14.1"
-git = "https://github.com/gingershaped/ruwuma.git"
-rev = "0713b97ed53b7cb5e639f5e2f067c0fe6ac525ff"
+git = "https://github.com/ruma/ruma.git"
+rev = "fca1dbc060f730c1aaf84c9c2fb1e8a587455be8"
 features = [
     "appservice-api-c",
     "client-api",
@@ -386,7 +386,6 @@ features = [
     "unstable-msc4406",
     "unstable-msc4439",
     "unstable-msc4466",
-    "unstable-msc4484",
     "unstable-extensible-events",
 ]
 

docker/Dockerfile

@@ -171,7 +171,6 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
     --mount=type=cache,target=/usr/local/cargo/git/db \
     --mount=type=cache,target=/app/target,id=continuwuity-cargo-target-${TARGET_CPU}-${TARGETPLATFORM}-${RUST_PROFILE} \
     bash <<'EOF'
-    set -euo pipefail
     set -o allexport
     set -o xtrace
     . /etc/environment
@@ -192,7 +191,7 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
             jq -r ".target_directory"))
     mkdir /out/sbin
     PACKAGE=conduwuit
-    CARGO_LOG=debug xx-cargo build --locked --profile ${RUST_PROFILE} \
+    xx-cargo build --locked --profile ${RUST_PROFILE} \
         --no-default-features --features ${CARGO_FEATURES} \
         -p $PACKAGE;
     BINARIES=($(cargo metadata --no-deps --format-version 1 | \
@@ -208,7 +207,6 @@ EOF
 RUN --mount=type=cache,target=/usr/local/cargo/registry \
     --mount=type=cache,target=/usr/local/cargo/git/db \
     bash <<'EOF'
-    set -euo pipefail
     set -o xtrace
     mkdir /out/sbom
     typeset -A PACKAGES

nix/rust.nix

@@ -12,7 +12,7 @@
         target:
         target.fromToolchainName {
           name = (lib.importTOML "${inputs.self}/rust-toolchain.toml").toolchain.channel;
-          sha256 = "sha256-OATSZm98Es5kIFuqaba+UvkQtFsVgJEBMmS+t6od5/U=";
+          sha256 = "sha256-h+t2xTBz5yt2YIO+1VMIIGlCU7gyp2LYOFvaV1nwOXU=";
         };
     in
     {

rust-toolchain.toml

@@ -10,7 +10,7 @@
 
 [toolchain]
 profile = "minimal"
-channel = "1.97.0"
+channel = "1.96.1"
 components = [
     # For rust-analyzer
     "rust-src",

src/admin/user/commands.rs

@@ -115,7 +115,7 @@ pub(super) async fn suspend(&self, user_id: String) -> Result {
 		// TODO: Record the actual user that sent the suspension where possible
 		self.services
 			.users
-			.suspend_account(&user_id, self.sender)
+			.suspend_account(&user_id, self.sender_or_service_user())
 			.await;
 
 		self.write_str(&format!("User {user_id} has been suspended."))
@@ -948,7 +948,7 @@ pub(super) async fn lock(&self, user_id: String) -> Result {
 
 		self.services
 			.users
-			.lock_account(&user_id, self.sender)
+			.lock_account(&user_id, self.sender_or_service_user())
 			.await;
 
 		self.write_str(&format!("User {user_id} has been locked."))

src/api/Cargo.toml

@@ -62,8 +62,6 @@ zstd_compression = [
 	"reqwest/zstd",
 ]
 
-admin_api = []
-
 [dependencies]
 async-trait.workspace = true
 axum-client-ip.workspace = true

src/api/admin/mod.rs

@@ -0,0 +1 @@
+pub mod rooms;

src/api/admin/rooms/ban.rs

@@ -6,7 +6,7 @@
 
 use crate::{Ruma, client::leave_room};
 
-/// # `PUT /_continuwuity/admin/v1/rooms/{roomID}/ban`
+/// # `PUT /_continuwuity/admin/rooms/{roomID}/ban`
 ///
 /// Bans or unbans a room.
 pub(crate) async fn ban_room(

src/api/admin/rooms/list.rs

@@ -0,0 +1,36 @@
+use axum::extract::State;
+use conduwuit::{Err, Result};
+use futures::StreamExt;
+use ruma::OwnedRoomId;
+use ruminuwuity::admin::continuwuity::rooms;
+
+use crate::Ruma;
+
+/// # `GET /_continuwuity/admin/rooms/list`
+///
+/// Lists all rooms known to this server, excluding banned ones.
+pub(crate) async fn list_rooms(
+	State(services): State<crate::State>,
+	body: Ruma<rooms::list::v1::Request>,
+) -> Result<rooms::list::v1::Response> {
+	let sender_user = body.identity.expect_sender_user()?;
+	if !services.users.is_admin(sender_user).await {
+		return Err!(Request(Forbidden("Only server administrators can use this endpoint")));
+	}
+
+	let mut rooms: Vec<OwnedRoomId> = services
+		.rooms
+		.metadata
+		.iter_ids()
+		.filter_map(|room_id| async move {
+			if !services.rooms.metadata.is_banned(&room_id).await {
+				Some(room_id.clone())
+			} else {
+				None
+			}
+		})
+		.collect()
+		.await;
+	rooms.sort();
+	Ok(rooms::list::v1::Response::new(rooms))
+}

src/api/admin/rooms/mod.rs

@@ -0,0 +1,2 @@
+pub mod ban;
+pub mod list;

src/api/client/admin/lock.rs

@@ -1,6 +1,6 @@
 use axum::extract::State;
 use conduwuit::{Err, Result};
-use futures::join;
+use futures::future::{join, join3};
 use ruma::api::client::admin::{is_user_locked, lock_user};
 
 use crate::Ruma;
@@ -12,7 +12,15 @@ pub(crate) async fn get_lock_status(
 	State(services): State<crate::State>,
 	body: Ruma<is_user_locked::v1::Request>,
 ) -> Result<is_user_locked::v1::Response> {
-	let status = services.users.status(&body.user_id).await;
+	let (admin, status) = join(
+		services.users.is_admin(body.identity.expect_sender_user()?),
+		services.users.status(&body.user_id),
+	)
+	.await;
+
+	if !admin {
+		return Err!(Request(Forbidden("Only server administrators can use this endpoint")));
+	}
 
 	status.ensure_active()?;
 
@@ -28,14 +36,22 @@ pub(crate) async fn put_lock_status(
 	State(services): State<crate::State>,
 	body: Ruma<lock_user::v1::Request>,
 ) -> Result<lock_user::v1::Response> {
-	let sender_user = body.identity.sender_user();
+	let sender_user = body.identity.expect_sender_user()?;
 
-	let (status, target_admin) =
-		join!(services.users.status(&body.user_id), services.users.is_admin(&body.user_id),);
+	let (sender_admin, status, target_admin) = join3(
+		services.users.is_admin(sender_user),
+		services.users.status(&body.user_id),
+		services.users.is_admin(&body.user_id),
+	)
+	.await;
+
+	if !sender_admin {
+		return Err!(Request(Forbidden("Only server administrators can use this endpoint")));
+	}
 
 	status.ensure_active()?;
 
-	if sender_user.is_some_and(|sender_user| body.user_id == sender_user) {
+	if body.user_id == *sender_user {
 		return Err!(Request(Forbidden("You cannot lock yourself")));
 	}
 
@@ -63,7 +79,7 @@ pub(crate) async fn put_lock_status(
 		// Notify the admin room that an account has been un/suspended
 		services
 			.admin
-			.send_text(&format!("{} has been {} by {}.", body.user_id, action, body.identity))
+			.send_text(&format!("{} has been {} by {}.", body.user_id, action, sender_user))
 			.await;
 	}
 

src/api/client/admin/mod.rs

@@ -1,5 +1,4 @@
 mod lock;
-pub(crate) mod site;
 mod suspend;
 
 pub(crate) use self::{lock::*, suspend::*};

src/api/client/admin/site/mod.rs

@@ -1,2 +0,0 @@
-pub(crate) mod rooms;
-pub(crate) mod users;

src/api/client/admin/site/rooms/list.rs

@@ -1,178 +0,0 @@
-use axum::extract::State;
-use conduwuit::{
-	Event, Result,
-	utils::stream::{BroadbandExt, WidebandExt},
-};
-use futures::StreamExt;
-use ruma::{
-	OwnedRoomId,
-	events::{
-		StateEventType,
-		room::{
-			create::RoomCreateEventContent,
-			encryption::PossiblyRedactedRoomEncryptionEventContent,
-			tombstone::PossiblyRedactedRoomTombstoneEventContent,
-		},
-	},
-};
-use ruminuwuity::admin::continuwuity::rooms;
-use tokio::join;
-
-use crate::Ruma;
-
-/// # `GET /_continuwuity/admin/rooms`
-///
-/// Lists all room IDs known to this server, excluding banned ones.
-///
-/// This is the legacy version of the endpoint, which does not support
-/// pagination or including banned rooms. It is recommended to use the
-/// `/v1/rooms` endpoint instead. This endpoint may be removed in a future
-/// release.
-pub(crate) async fn legacy_list_rooms(
-	State(services): State<crate::State>,
-	_body: Ruma<rooms::list::unstable::Request>,
-) -> Result<rooms::list::unstable::Response> {
-	let mut rooms: Vec<OwnedRoomId> = services
-		.rooms
-		.metadata
-		.iter_ids()
-		.filter_map(|room_id| async move {
-			if !services.rooms.metadata.is_banned(&room_id).await {
-				Some(room_id.clone())
-			} else {
-				None
-			}
-		})
-		.collect()
-		.await;
-	rooms.sort();
-	Ok(rooms::list::unstable::Response::new(rooms))
-}
-
-/// # `GET /_continuwuity/admin/v1/rooms`
-///
-/// Lists rooms known to this server.
-pub(crate) async fn list_rooms(
-	State(services): State<crate::State>,
-	body: Ruma<rooms::list::v1::Request>,
-) -> Result<rooms::list::v1::Response> {
-	let include_banned_rooms = body.include_banned_rooms;
-	let rooms = services
-		.rooms
-		.metadata
-		.iter_ids()
-		.wide_filter_map(|room_id| async move {
-			if include_banned_rooms || !services.rooms.metadata.is_banned(&room_id).await {
-				Some(room_id.clone())
-			} else {
-				None
-			}
-		})
-		.skip(body.offset.unwrap_or_default())
-		.take(body.limit.unwrap_or(100).min(100))
-		.broad_filter_map(|room_id| async move {
-			let (
-				banned,
-				disabled,
-				member_count,
-				local_member_count,
-				resident_server_count,
-				published,
-				create_event,
-				encryption_event,
-				name_event,
-				topic_event,
-				canonical_alias_event,
-				join_rules_event,
-				history_visibility_event,
-				tombstone_event,
-			) = join!(
-				services.rooms.metadata.is_banned(&room_id),
-				services.rooms.metadata.is_disabled(&room_id),
-				services.rooms.state_cache.room_joined_count(&room_id),
-				services
-					.rooms
-					.state_cache
-					.active_local_users_in_room(&room_id)
-					.count(),
-				services.rooms.state_cache.room_servers(&room_id).count(),
-				services.rooms.directory.is_public_room(&room_id),
-				services.rooms.state_accessor.room_state_get(
-					&room_id,
-					&StateEventType::RoomCreate,
-					""
-				),
-				services
-					.rooms
-					.state_accessor
-					.room_state_get_content::<PossiblyRedactedRoomEncryptionEventContent>(
-						&room_id,
-						&StateEventType::RoomEncryption,
-						""
-					),
-				services.rooms.state_accessor.room_state_get_content(
-					&room_id,
-					&StateEventType::RoomName,
-					""
-				),
-				services.rooms.state_accessor.room_state_get_content(
-					&room_id,
-					&StateEventType::RoomTopic,
-					""
-				),
-				services.rooms.state_accessor.room_state_get_content(
-					&room_id,
-					&StateEventType::RoomCanonicalAlias,
-					""
-				),
-				services.rooms.state_accessor.room_state_get_content(
-					&room_id,
-					&StateEventType::RoomJoinRules,
-					""
-				),
-				services.rooms.state_accessor.room_state_get_content(
-					&room_id,
-					&StateEventType::RoomHistoryVisibility,
-					""
-				),
-				services
-					.rooms
-					.state_accessor
-					.room_state_get_content::<PossiblyRedactedRoomTombstoneEventContent>(
-						&room_id,
-						&StateEventType::RoomTombstone,
-						""
-					),
-			);
-			let Ok(create_event) = create_event else {
-				return None;
-			};
-			let create_content = create_event
-				.get_content::<RoomCreateEventContent>()
-				.expect("m.room.create content must be valid");
-			Some(rooms::list::v1::MinimalRoomInfo {
-				room_id,
-				banned,
-				disabled,
-				member_count: usize::try_from(member_count.unwrap_or_default())
-					.expect("u64 should fit in usize"),
-				local_member_count,
-				resident_server_count,
-				creators: vec![create_event.sender],
-				encrypted: encryption_event.is_ok_and(|c| c.algorithm.is_some()),
-				federated: create_content.federate,
-				published,
-				version: create_content.room_version,
-				name: name_event.unwrap_or(None),
-				topic: topic_event.unwrap_or(None),
-				canonical_alias: canonical_alias_event.unwrap_or(None),
-				join_rules: join_rules_event.unwrap_or(None),
-				history_visibility: history_visibility_event.unwrap_or(None),
-				predecessor: create_content.predecessor.map(|c| c.room_id),
-				successor: tombstone_event.map_or(None, |c| c.replacement_room),
-			})
-		})
-		.collect()
-		.await;
-	Ok(rooms::list::v1::Response::new(rooms))
-}

src/api/client/admin/site/rooms/mod.rs

@@ -1,5 +0,0 @@
-mod ban;
-mod list;
-
-pub(crate) use ban::ban_room;
-pub(crate) use list::*;

src/api/client/admin/site/users/create.rs

@@ -1,142 +0,0 @@
-use axum::extract::State;
-use conduwuit::{
-	err, error, info,
-	utils::{IterStream, stream::BroadbandExt},
-	warn,
-};
-use futures::{FutureExt, StreamExt};
-use ruma::{api::client::profile::PropagateTo, profile::ProfileFieldValue};
-use ruminuwuity::admin::continuwuity::users;
-use service::users::{HashedPassword, ProfileFieldChange};
-
-use crate::router::Ruma;
-
-/// # `POST /_continuwuity/admin/v1/users/create`
-///
-/// Creates a new user.
-pub(crate) async fn create_user(
-	State(services): State<crate::State>,
-	body: Ruma<users::create::v1::Request>,
-) -> conduwuit::Result<users::create::v1::Response> {
-	let email = body
-		.email
-		.clone()
-		.map(lettre::Address::try_from)
-		.transpose()
-		.map_err(|e| err!(Request(BadJson("Invalid email address: {e}"))))?;
-
-	let ref user_id = services
-		.users
-		.determine_registration_user_id(Some(body.localpart.clone()), email.as_ref(), None)
-		.await?;
-
-	services.users.create_shadow_account(user_id).await?;
-
-	services
-		.users
-		.convert_to_local_account(user_id, HashedPassword::new(&body.password)?)
-		.await?;
-
-	if let Some(email) = &email {
-		services
-			.threepid
-			.associate_localpart_email(user_id.localpart(), email)
-			.await?;
-	}
-
-	if body.suspended {
-		services
-			.users
-			.suspend_account(user_id, body.identity.sender_user())
-			.await;
-	}
-	if body.locked {
-		services
-			.users
-			.lock_account(user_id, body.identity.sender_user())
-			.await;
-	}
-	if body.login_disabled {
-		services.users.disable_login(user_id);
-	}
-	if let Some(ref value) = body.display_name {
-		services
-			.users
-			.set_profile_field(
-				user_id,
-				ProfileFieldChange::Set(ProfileFieldValue::DisplayName(value.to_owned())),
-				PropagateTo::None,
-			)
-			.await?;
-	}
-	if let Some(ref value) = body.avatar_url {
-		services
-			.users
-			.set_profile_field(
-				user_id,
-				ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(value.to_owned())),
-				PropagateTo::None,
-			)
-			.await?;
-	}
-	if body.admin {
-		services
-			.admin
-			.make_user_admin(user_id)
-			.await
-			.inspect_err(|e| error!("failed to make new user {user_id} an admin: {e}"))
-			.ok();
-	}
-
-	body.auto_join_rooms
-		.clone()
-		.into_iter()
-		.stream()
-		.chain(
-			if body.skip_auto_join {
-				vec![]
-			} else {
-				services.config.auto_join_rooms.clone()
-			}
-			.into_iter()
-			.stream(),
-		)
-		.broad_filter_map(|room| async move {
-			services
-				.rooms
-				.alias
-				.resolve_with_servers(&room, None)
-				.await
-				.inspect_err(|e| {
-					warn!(
-						"Failed to resolve room alias to room ID when attempting to auto join \
-						 {room}: {e}"
-					);
-				})
-				.ok()
-		})
-		.for_each_concurrent(None, |(room_id, servers)| async move {
-			match services
-				.rooms
-				.membership
-				.join_room(
-					user_id,
-					&room_id,
-					Some("Automatically joining this room upon registration".to_owned()),
-					servers.as_ref(),
-				)
-				.boxed()
-				.await
-			{
-				| Err(e) => {
-					warn!("Failed to automatically join {user_id} to {room_id}: {e}");
-				},
-				| _ => {
-					info!("Automatically joined room {user_id} to {room_id}");
-				},
-			}
-		})
-		.await;
-
-	Ok(users::create::v1::Response::new(user_id.to_owned()))
-}

src/api/client/admin/site/users/list.rs

@@ -1,42 +0,0 @@
-use axum::extract::State;
-use conduwuit::utils::stream::WidebandExt;
-use futures::StreamExt;
-use ruminuwuity::admin::continuwuity::users;
-use tokio::join;
-
-use crate::router::Ruma;
-
-/// # `GET /_continuwuity/admin/v1/users`
-///
-/// Lists all users on this homeserver.
-pub(crate) async fn list_users(
-	State(services): State<crate::State>,
-	body: Ruma<users::list::v1::Request>,
-) -> conduwuit::Result<users::list::v1::Response> {
-	let users = services
-		.users
-		.stream_local_users()
-		.skip(body.offset.unwrap_or_default())
-		.take(body.limit.unwrap_or(100).min(100))
-		.wide_filter_map(|user_id| async move {
-			let (status, suspended, locked, admin, login_disabled) = join!(
-				services.users.status(&user_id),
-				services.users.is_suspended(&user_id),
-				services.users.is_locked(&user_id),
-				services.users.is_admin(&user_id),
-				services.users.is_login_disabled(&user_id),
-			);
-			Some(users::list::v1::User {
-				user_id: user_id.clone(),
-				deactivated: !status.is_active(),
-				suspended: suspended.unwrap_or_default(),
-				locked: locked.unwrap_or_default(),
-				admin,
-				login_disabled,
-			})
-		})
-		.collect()
-		.await;
-
-	Ok(users::list::v1::Response::new(users))
-}

src/api/client/admin/site/users/mod.rs

@@ -1,5 +0,0 @@
-mod create;
-mod list;
-
-pub(crate) use create::*;
-pub(crate) use list::*;

src/api/client/admin/suspend.rs

@@ -67,7 +67,7 @@ pub(crate) async fn put_suspended_status(
 	let action = if body.suspended {
 		services
 			.users
-			.suspend_account(&body.user_id, body.identity.sender_user())
+			.suspend_account(&body.user_id, sender_user)
 			.await;
 		"suspended"
 	} else {

src/api/client/membership/invite.rs

@@ -221,7 +221,7 @@ pub(crate) async fn invite_helper(
 		let pdu_id = services
 			.rooms
 			.event_handler
-			.handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, true)
+			.handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, false)
 			.boxed()
 			.await?
 			.ok_or_else(|| {

src/api/client/profile.rs

@@ -29,12 +29,6 @@ pub(crate) async fn get_profile_route(
 	State(services): State<crate::State>,
 	body: Ruma<get_profile::v3::Request>,
 ) -> Result<get_profile::v3::Response> {
-	if services.config.require_auth_for_profile_requests && body.identity.is_none() {
-		return Err!(Request(Unauthorized(
-			"This server requires authentication to view user profiles."
-		)));
-	}
-
 	let Some(profile) = fetch_full_profile(&services, &body.user_id).await else {
 		return Err!(Request(NotFound("This user's profile could not be fetched.")));
 	};
@@ -46,12 +40,6 @@ pub(crate) async fn get_profile_field_route(
 	State(services): State<crate::State>,
 	body: Ruma<get_profile_field::v3::Request>,
 ) -> Result<get_profile_field::v3::Response> {
-	if services.config.require_auth_for_profile_requests && body.identity.is_none() {
-		return Err!(Request(Unauthorized(
-			"This server requires authentication to view user profiles."
-		)));
-	}
-
 	let value = fetch_profile_field(&services, &body.user_id, body.field.clone()).await?;
 
 	Ok(assign!(get_profile_field::v3::Response::default(), { value }))

src/api/mod.rs

@@ -11,6 +11,8 @@
 pub mod router;
 pub mod server;
 
+pub mod admin;
+
 pub(crate) use self::router::{Ruma, RumaResponse, State};
 
 conduwuit::mod_ctor! {}

src/api/router.rs

@@ -16,12 +16,7 @@
 
 use self::handler::RouterExt;
 pub(super) use self::{args::Args as Ruma, auth::ClientIdentity, response::RumaResponse};
-#[cfg(feature = "admin_api")]
-use crate::client::admin::site as admin_api;
-use crate::{
-	client::{self, admin},
-	server,
-};
+use crate::{admin, client, server};
 
 pub fn build(router: Router<State>, state: State) -> Router<State> {
 	let config = &state.server.config;
@@ -196,11 +191,8 @@ pub fn build(router: Router<State>, state: State) -> Router<State> {
 		.ruma_route(&client::get_authorization_server_metadata_route)
 		.merge(client::oauth::router(state))
 		.route("/_continuwuity/server_version", get(client::continuwuity_server_version))
-		.ruma_route(&admin::site::rooms::ban_room)
-		.ruma_route(&admin::site::rooms::list_rooms)
-		.ruma_route(&admin::site::rooms::legacy_list_rooms)
-		.ruma_route(&admin::site::users::create_user)
-		.ruma_route(&admin::site::users::list_users);
+		.ruma_route(&admin::rooms::ban::ban_room)
+		.ruma_route(&admin::rooms::list::list_rooms);
 
 	if config.allow_federation {
 		router = router
@@ -283,16 +275,6 @@ pub fn build(router: Router<State>, state: State) -> Router<State> {
 			.route("/_matrix/media/r0/preview_url", any(redirect_legacy_preview));
 	}
 
-	#[cfg(feature = "admin_api")]
-	{
-		router = router
-			.ruma_route(&admin_api::users::list_users_route)
-			.ruma_route(&admin_api::users::create_user_route)
-			.ruma_route(&admin_api::rooms::ban_room)
-			.ruma_route(&admin_api::rooms::legacy_list_rooms_route)
-			.ruma_route(&admin_api::rooms::list_rooms_route);
-	};
-
 	router
 }
 

src/api/router/auth.rs

@@ -1,14 +1,11 @@
-use std::{
-	any::{Any, TypeId},
-	fmt::Display,
-};
+use std::any::{Any, TypeId};
 
 use conduwuit::{Err, Error, Result, err};
 use http::StatusCode;
 use ruma::{
 	DeviceId, OwnedDeviceId, OwnedServerName, OwnedUserId, UserId,
 	api::{
-		IncomingRequest, OAuthClientScope,
+		IncomingRequest,
 		auth_scheme::{
 			AccessToken, AccessTokenOptional, AppserviceToken, AppserviceTokenOptional,
 			AuthScheme, NoAccessToken, NoAuthentication,
@@ -80,66 +77,68 @@ pub(crate) fn appservice_info(&self) -> Option<&RegistrationInfo> {
 	pub(crate) fn is_appservice(&self) -> bool { matches!(self, Self::Appservice { .. }) }
 }
 
-impl Display for ClientIdentity {
-	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-		match self {
-			| Self::User { sender_user, sender_device } =>
-				write!(f, "{sender_user} ({sender_device})"),
-			| Self::Appservice { sender_user, appservice_info, .. } =>
-				write!(f, "appservice `{}` using {sender_user}", appservice_info.registration.id),
-		}
-	}
-}
-
 pub(crate) trait CheckAuth: AuthScheme {
 	type Identity: Send;
 
-	fn authenticate<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]> + Sync>(
+	fn authenticate<R: IncomingRequest + Any, B: AsRef<[u8]> + Sync>(
 		services: &Services,
 		incoming_request: &hyper::Request<B>,
 		query: AuthQueryParams,
 	) -> impl Future<Output = Result<Self::Identity>> + Send {
 		async move {
+			let route = TypeId::of::<R>();
+
 			let output = Self::extract_authentication(incoming_request).map_err(|err| {
 				err!(Request(Unauthorized(warn!(
-					"Failed to extract request authentication: {}",
+					"Failed to extract authorization: {}",
 					err.into()
 				))))
 			})?;
 
-			Self::verify::<R, B>(services, output, incoming_request, query).await
+			Self::verify(services, output, incoming_request, query, route).await
 		}
 	}
 
-	fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]> + Sync>(
+	fn verify<B: AsRef<[u8]> + Sync>(
 		services: &Services,
 		output: Self::Output,
 		request: &hyper::Request<B>,
 		query: AuthQueryParams,
+		route: TypeId,
 	) -> impl Future<Output = Result<Self::Identity>> + Send;
 }
 
 impl CheckAuth for ServerSignatures {
 	type Identity = OwnedServerName;
 
-	async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]> + Sync>(
+	async fn verify<B: AsRef<[u8]> + Sync>(
 		services: &Services,
 		output: Self::Output,
 		request: &hyper::Request<B>,
 		_query: AuthQueryParams,
+		_route: TypeId,
 	) -> Result<Self::Identity> {
+		let destination = services.globals.server_name();
+		if output
+			.destination
+			.as_ref()
+			.is_some_and(|supplied_destination| supplied_destination != destination)
+		{
+			return Err!(Request(Unauthorized("Destination mismatch.")));
+		}
+
 		let key = services
 			.server_keys
 			.get_verify_key(&output.origin, &output.key)
 			.await
-			.map_err(|err| {
-				err!(Request(Unauthorized(warn!("Failed to fetch signing keys: {err}"))))
+			.map_err(|e| {
+				err!(Request(Unauthorized(warn!("Failed to fetch signing keys: {e}"))))
 			})?;
 
 		let keys: PubKeys = [(output.key.to_string(), key.key)].into();
 		let keys: PubKeyMap = [(output.origin.as_str().into(), keys)].into();
 
-		match output.verify_request(request, services.globals.server_name(), &keys) {
+		match output.verify_request(request, destination, &keys) {
 			| Ok(()) => {
 				if services
 					.moderation
@@ -161,36 +160,115 @@ async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]>
 impl CheckAuth for AccessToken {
 	type Identity = ClientIdentity;
 
-	async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]> + Sync>(
+	async fn verify<B: AsRef<[u8]> + Sync>(
 		services: &Services,
 		output: Self::Output,
 		_request: &hyper::Request<B>,
 		query: AuthQueryParams,
+		route: TypeId,
 	) -> Result<Self::Identity> {
-		verify_access_token(services, output, query, TypeId::of::<R>(), R::required_scopes())
-			.await
+		if output.is_empty() {
+			return Err!(Request(Unauthorized("Missing access token.")));
+		}
+		if let Some((sender_user, sender_device, status)) =
+			services.users.find_from_token(&output).await
+		{
+			// If the token is expired we return a soft logout
+			if matches!(status, AccessTokenStatus::Expired) {
+				return Err(Error::Request(
+					ErrorKind::UnknownToken(
+						assign!(UnknownTokenErrorData::new(), { soft_logout: true }),
+					),
+					"This token has expired".into(),
+					StatusCode::UNAUTHORIZED,
+				));
+			}
+
+			// Locked users can only use /logout and /logout/all
+			if services
+				.users
+				.is_locked(&sender_user)
+				.await
+				.is_ok_and(std::convert::identity)
+			{
+				if !(route == TypeId::of::<client::session::logout::v3::Request>()
+					|| route == TypeId::of::<client::session::logout_all::v3::Request>())
+				{
+					return Err!(Request(UserLocked("Your account is locked.")));
+				}
+			}
+
+			Ok(ClientIdentity::User { sender_user, sender_device })
+		} else if let Ok(appservice_info) = services.appservice.find_from_token(&output).await {
+			let Ok(sender_user) = query.user_id.clone().map_or_else(
+				|| {
+					UserId::parse_with_server_name(
+						appservice_info.registration.sender_localpart.as_str(),
+						services.globals.server_name(),
+					)
+				},
+				UserId::parse,
+			) else {
+				return Err!(Request(InvalidUsername("Username is invalid.")));
+			};
+
+			if !appservice_info.is_user_match(&sender_user) {
+				return Err!(Request(Exclusive("User is not in namespace.")));
+			}
+
+			// MSC3202/MSC4190: Handle device_id masquerading for appservices.
+			// The device_id can be provided via `device_id` or
+			// `org.matrix.msc3202.device_id` query parameter.
+			let sender_device =
+				if let Some(device_id) = query.device_id.as_deref().map(Into::into) {
+					// Verify the device exists for this user
+					if services
+						.users
+						.get_device_metadata(&sender_user, device_id)
+						.await
+						.is_err()
+					{
+						return Err!(Request(Forbidden(
+							"Device does not exist for user or appservice cannot masquerade as \
+							 this device."
+						)));
+					}
+
+					Some(device_id.to_owned())
+				} else {
+					None
+				};
+
+			Ok(ClientIdentity::Appservice {
+				sender_user,
+				sender_device,
+				appservice_info: Box::new(appservice_info),
+			})
+		} else {
+			Err(Error::Request(
+				ErrorKind::UnknownToken(UnknownTokenErrorData::new()),
+				"Invalid token".into(),
+				StatusCode::UNAUTHORIZED,
+			))
+		}
 	}
 }
 
 impl CheckAuth for AccessTokenOptional {
 	type Identity = Option<ClientIdentity>;
 
-	async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]> + Sync>(
+	async fn verify<B: AsRef<[u8]> + Sync>(
 		services: &Services,
 		output: Self::Output,
-		_request: &hyper::Request<B>,
+		request: &hyper::Request<B>,
 		query: AuthQueryParams,
+		route: TypeId,
 	) -> Result<Self::Identity> {
 		match output {
-			| Some(token) => verify_access_token(
-				services,
-				token,
-				query,
-				TypeId::of::<R>(),
-				R::required_scopes(),
-			)
-			.await
-			.map(Some),
+			| Some(token) =>
+				<AccessToken as CheckAuth>::verify(services, token, request, query, route)
+					.await
+					.map(Some),
 			| None => Ok(None),
 		}
 	}
@@ -199,29 +277,39 @@ async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]>
 impl CheckAuth for AppserviceToken {
 	type Identity = RegistrationInfo;
 
-	async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]> + Sync>(
+	async fn verify<B: AsRef<[u8]> + Sync>(
 		services: &Services,
 		output: Self::Output,
 		_request: &hyper::Request<B>,
 		_query: AuthQueryParams,
+		_route: TypeId,
 	) -> Result<Self::Identity> {
-		verify_appservice_access_token(services, output).await
+		if output.is_empty() {
+			return Err!(Request(Unauthorized("Missing access token.")));
+		}
+		let Ok(appservice_info) = services.appservice.find_from_token(&output).await else {
+			return Err!(Request(Unauthorized("Invalid appservice token.")));
+		};
+
+		Ok(appservice_info)
 	}
 }
 
 impl CheckAuth for AppserviceTokenOptional {
 	type Identity = Option<RegistrationInfo>;
 
-	async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]> + Sync>(
+	async fn verify<B: AsRef<[u8]> + Sync>(
 		services: &Services,
 		output: Self::Output,
-		_request: &hyper::Request<B>,
-		_query: AuthQueryParams,
+		request: &hyper::Request<B>,
+		query: AuthQueryParams,
+		route: TypeId,
 	) -> Result<Self::Identity> {
 		match output {
-			| Some(token) => verify_appservice_access_token(services, token)
-				.await
-				.map(Some),
+			| Some(token) =>
+				<AppserviceToken as CheckAuth>::verify(services, token, request, query, route)
+					.await
+					.map(Some),
 			| None => Ok(None),
 		}
 	}
@@ -230,11 +318,12 @@ async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]>
 impl CheckAuth for NoAuthentication {
 	type Identity = ();
 
-	async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]> + Sync>(
+	async fn verify<B: AsRef<[u8]> + Sync>(
 		_services: &Services,
 		_output: Self::Output,
 		_request: &hyper::Request<B>,
 		_query: AuthQueryParams,
+		_route: TypeId,
 	) -> Result<Self::Identity> {
 		Ok(())
 	}
@@ -243,153 +332,31 @@ async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]>
 impl CheckAuth for NoAccessToken {
 	type Identity = Option<ClientIdentity>;
 
-	async fn verify<R: IncomingRequest<Authentication = Self> + Any, B: AsRef<[u8]> + Sync>(
+	async fn verify<B: AsRef<[u8]> + Sync>(
 		services: &Services,
 		_output: Self::Output,
 		request: &hyper::Request<B>,
 		query: AuthQueryParams,
+		route: TypeId,
 	) -> Result<Self::Identity> {
 		// We handle these the same as AccessTokenOptional
 		let token = AccessTokenOptional::extract_authentication(request).map_err(|err| {
 			err!(Request(Unauthorized(warn!("Failed to extract authorization: {}", err))))
 		})?;
 
-		match token {
-			| Some(token) => verify_access_token(
-				services,
-				token,
-				query,
-				TypeId::of::<R>(),
-				// Assume that no scopes are required for these endpoints since
-				// ostensibly they don't require authentication
-				&[],
-			)
-			.await
-			.map(Some),
-			| None => Ok(None),
+		// Check special access restrictions
+		if (route == TypeId::of::<client::profile::get_avatar_url::v3::Request>()
+			|| route == TypeId::of::<client::profile::get_display_name::v3::Request>()
+			|| route == TypeId::of::<client::profile::get_profile_field::v3::Request>()
+			|| route == TypeId::of::<client::profile::get_profile::v3::Request>())
+			&& services.config.require_auth_for_profile_requests
+			&& token.is_none()
+		{
+			return Err!(Request(Unauthorized(
+				"This server requires authentication to access user profiles."
+			)));
 		}
+
+		<AccessTokenOptional as CheckAuth>::verify(services, token, request, query, route).await
 	}
 }
-
-async fn verify_access_token(
-	services: &Services,
-	output: String,
-	query: AuthQueryParams,
-	route: TypeId,
-	required_scopes: &[OAuthClientScope],
-) -> Result<ClientIdentity> {
-	if let Some((sender_user, sender_device, status)) =
-		services.users.find_from_token(&output).await
-	{
-		// If the token is expired we return a soft logout
-		if matches!(status, AccessTokenStatus::Expired) {
-			return Err(Error::Request(
-				ErrorKind::UnknownToken(
-					assign!(UnknownTokenErrorData::new(), { soft_logout: true }),
-				),
-				"This access token has expired.".into(),
-				StatusCode::UNAUTHORIZED,
-			));
-		}
-
-		// Locked users can only use /logout and /logout/all
-		if services
-			.users
-			.is_locked(&sender_user)
-			.await
-			.is_ok_and(std::convert::identity)
-		{
-			if !(route == TypeId::of::<client::session::logout::v3::Request>()
-				|| route == TypeId::of::<client::session::logout_all::v3::Request>())
-			{
-				return Err!(Request(UserLocked("Your account is locked.")));
-			}
-		}
-
-		// If this device is bound to an OAuth session, check its scopes. This will also
-		// handle admin-only endpoints for OAuth clients.
-		if let Some(session) = services
-			.oauth
-			.get_session_info_for_device(&sender_user, &sender_device)
-			.await
-		{
-			if required_scopes
-				.iter()
-				.all(|scope| !session.scopes.contains(scope))
-			{
-				return Err!(Request(Forbidden(
-					"You don't have the necessary scopes to use this endpoint."
-				)));
-			}
-		} else {
-			// Otherwise, explicitly check if the endpoint is restricted to admins only.
-			if required_scopes.contains(&OAuthClientScope::ServerAdministration)
-				&& !services.users.is_admin(&sender_user).await
-			{
-				return Err!(Request(Forbidden(
-					"Only server administrators can use this endpoint."
-				)));
-			}
-		}
-
-		Ok(ClientIdentity::User { sender_user, sender_device })
-	} else if let Ok(appservice_info) = services.appservice.find_from_token(&output).await {
-		let Ok(sender_user) = query.user_id.clone().map_or_else(
-			|| {
-				UserId::parse_with_server_name(
-					appservice_info.registration.sender_localpart.as_str(),
-					services.globals.server_name(),
-				)
-			},
-			UserId::parse,
-		) else {
-			return Err!(Request(InvalidUsername("Username is invalid.")));
-		};
-
-		if !appservice_info.is_user_match(&sender_user) {
-			return Err!(Request(Exclusive("User is not in this appservice's namespace.")));
-		}
-
-		// MSC3202/MSC4190: Handle device_id masquerading for appservices.
-		// The device_id can be provided via `device_id` or
-		// `org.matrix.msc3202.device_id` query parameter.
-		let sender_device = if let Some(device_id) = query.device_id.as_deref().map(Into::into) {
-			// Verify the device exists for this user
-			if services
-				.users
-				.get_device_metadata(&sender_user, device_id)
-				.await
-				.is_err()
-			{
-				return Err!(Request(Forbidden("Appservice cannot masquerade as this device.")));
-			}
-
-			Some(device_id.to_owned())
-		} else {
-			None
-		};
-
-		Ok(ClientIdentity::Appservice {
-			sender_user,
-			sender_device,
-			appservice_info: Box::new(appservice_info),
-		})
-	} else {
-		Err(Error::Request(
-			ErrorKind::UnknownToken(UnknownTokenErrorData::new()),
-			"Invalid access token.".into(),
-			StatusCode::UNAUTHORIZED,
-		))
-	}
-}
-
-async fn verify_appservice_access_token(
-	services: &Services,
-	output: String,
-) -> Result<RegistrationInfo> {
-	let Ok(appservice_info) = services.appservice.find_from_token(&output).await else {
-		return Err!(Request(Unauthorized("Invalid appservice token.")));
-	};
-
-	Ok(appservice_info)
-}

src/api/server/send.rs

@@ -241,7 +241,7 @@ async fn handle(
 	pdus: impl Stream<Item = Pdu> + Send,
 	edus: impl Stream<Item = Edu> + Send,
 ) -> std::result::Result<ResolvedMap, TransactionError> {
-	// group pdus by room
+	// Group PDUs by room for parallel processing
 	let pdus = pdus
 		.collect()
 		.map(|mut pdus: Vec<_>| {
@@ -252,7 +252,7 @@ async fn handle(
 		})
 		.await;
 
-	// we can evaluate rooms concurrently
+	// Evaluate rooms in parallel
 	let results: ResolvedMap = pdus
 		.into_iter()
 		.try_stream()
@@ -266,7 +266,7 @@ async fn handle(
 		.boxed()
 		.await?;
 
-	// evaluate edus after pdus, at least for now.
+	// Evaluate EDUs after PDUs in case some of the PDUs then forbid some EDUs.
 	edus.for_each_concurrent(automatic_width(), |edu| handle_edu(services, client, origin, edu))
 		.boxed()
 		.await;
@@ -296,20 +296,12 @@ async fn handle_room(
 	// Try to sort PDUs by their dependencies, but fall back to arbitrary order on
 	// failure (e.g., cycles). This is best-effort; proper ordering is the sender's
 	// responsibility.
-	let sorted_event_ids = if pdu_map.len() >= 2 {
-		let refmap = pdu_map
-			.iter()
-			.map(|(event_id, obj)| (event_id.clone(), obj))
-			.collect();
-		build_local_dag(&refmap, DagBuilderTree::PrevEvents)
-			.await
-			.unwrap_or_else(|e| {
-				debug_warn!("Failed to build local DAG for room {room_id}: {e}");
-				pdu_map.keys().cloned().collect()
-			})
-	} else {
-		pdu_map.keys().cloned().collect()
-	};
+	let sorted_event_ids = build_local_dag(&pdu_map, DagBuilderTree::PrevEvents)
+		.await
+		.unwrap_or_else(|e| {
+			debug_warn!("Failed to build local DAG for room {room_id}: {e}");
+			pdu_map.keys().cloned().collect()
+		});
 	let mut results = Vec::with_capacity(sorted_event_ids.len());
 	for event_id in sorted_event_ids {
 		let value = pdu_map
@@ -322,7 +314,7 @@ async fn handle_room(
 		let result = services
 			.rooms
 			.event_handler
-			.handle_incoming_pdu(origin, room_id, &event_id, value.clone(), true)
+			.handle_incoming_pdu(origin, room_id, &event_id, value, false)
 			.boxed()
 			.await
 			.map(|_| ());

src/api/server/send_join.rs

@@ -200,7 +200,7 @@ async fn create_join_event(
 	let pdu_id = services
 		.rooms
 		.event_handler
-		.handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), true)
+		.handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), false)
 		.boxed()
 		.await?
 		.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;

src/api/server/send_knock.rs

@@ -166,7 +166,7 @@ pub(crate) async fn create_knock_event_v1_route(
 	let pdu_id = services
 		.rooms
 		.event_handler
-		.handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), true)
+		.handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), false)
 		.boxed()
 		.await?
 		.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;

src/api/server/send_leave.rs

@@ -153,7 +153,7 @@ async fn create_leave_event(
 	let pdu_id = services
 		.rooms
 		.event_handler
-		.handle_incoming_pdu(origin, room_id, &event_id, value, true)
+		.handle_incoming_pdu(origin, room_id, &event_id, value, false)
 		.boxed()
 		.await?
 		.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;

src/core/error/err.rs

@@ -149,6 +149,14 @@ macro_rules! err_log {
 #[macro_export]
 #[collapse_debuginfo(yes)]
 macro_rules! err_lev {
+	(debug_info) => {
+		if $crate::debug::logging() {
+			$crate::tracing::Level::INFO
+		} else {
+			$crate::tracing::Level::DEBUG
+		}
+	};
+
 	(debug_warn) => {
 		if $crate::debug::logging() {
 			$crate::tracing::Level::WARN

src/core/matrix/versions.rs

@@ -41,7 +41,5 @@ pub fn unstable_features() -> BTreeMap<String, bool> {
 		("org.matrix.msc4155".to_owned(), true),
 		// profile change propagation (https://github.com/matrix-org/matrix-spec-proposals/pull/4466)
 		("computer.gingershaped.msc4466".to_owned(), true),
-		// server admin oauth scope (https://github.com/matrix-org/matrix-spec-proposals/pull/4484)
-		("org.continuwuity.msc4484.unstable".to_owned(), true),
 	])
 }

src/main/Cargo.toml

@@ -68,7 +68,6 @@ full = [
     "jemalloc_prof",
     "perf_measurements",
     "tokio_console",
-	"conduwuit-api/admin_api",
 ]
 
 brotli_compression = [

src/ruminuwuity/admin/continuwuity/mod.rs

@@ -1,2 +1 @@
 pub mod rooms;
-pub mod users;

src/ruminuwuity/admin/continuwuity/rooms/ban.rs

@@ -1,7 +1,7 @@
 pub mod v1 {
 	use ruma::{
 		OwnedRoomAliasId, OwnedRoomId, OwnedUserId,
-		api::{OAuthClientScope, auth_scheme::AccessToken, request, response},
+		api::{auth_scheme::AccessToken, request, response},
 		metadata,
 	};
 
@@ -9,10 +9,8 @@ pub mod v1 {
 		method: PUT,
 		rate_limited: false,
 		authentication: AccessToken,
-		required_scopes: [OAuthClientScope::ServerAdministration],
 		history: {
-			unstable("org.continuwuity.admin") => "/_continuwuity/admin/rooms/{room_id}/ban",
-			1.0 => "/_continuwuity/admin/v1/rooms/{room_id}/ban",
+			1.0 => "/_continuwuity/admin/rooms/{room_id}/ban",
 		}
 	}
 
@@ -31,11 +29,8 @@ pub struct Request {
 
 	#[response]
 	pub struct Response {
-		/// Users who were successfully kicked from this room.
 		pub kicked_users: Vec<OwnedUserId>,
-		/// Users who could not be kicked from the room.
 		pub failed_kicked_users: Vec<OwnedUserId>,
-		/// Any local aliases that were removed from the room.
 		pub local_aliases: Vec<OwnedRoomAliasId>,
 	}
 

src/ruminuwuity/admin/continuwuity/rooms/list.rs

@@ -1,7 +1,7 @@
-pub mod unstable {
+pub mod v1 {
 	use ruma::{
 		OwnedRoomId,
-		api::{OAuthClientScope, auth_scheme::AccessToken, request, response},
+		api::{auth_scheme::AccessToken, request, response},
 		metadata,
 	};
 
@@ -9,9 +9,8 @@ pub mod unstable {
 		method: GET,
 		rate_limited: false,
 		authentication: AccessToken,
-		required_scopes: [OAuthClientScope::ServerAdministration],
 		history: {
-			unstable => "/_continuwuity/admin/rooms/list",
+			1.0 => "/_continuwuity/admin/rooms/list",
 		}
 	}
 
@@ -21,7 +20,6 @@ pub mod unstable {
 
 	#[response]
 	pub struct Response {
-		/// A list of room IDs known to this server.
 		pub rooms: Vec<OwnedRoomId>,
 	}
 
@@ -35,133 +33,3 @@ impl Response {
 		pub fn new(rooms: Vec<OwnedRoomId>) -> Self { Self { rooms } }
 	}
 }
-
-pub mod v1 {
-	use ruma::{
-		OwnedRoomId, OwnedUserId, RoomVersionId,
-		api::{auth_scheme::AccessToken, request, response},
-		events::room::{
-			canonical_alias::PossiblyRedactedRoomCanonicalAliasEventContent,
-			history_visibility::PossiblyRedactedRoomHistoryVisibilityEventContent,
-			join_rules::PossiblyRedactedRoomJoinRulesEventContent,
-			name::PossiblyRedactedRoomNameEventContent,
-			topic::PossiblyRedactedRoomTopicEventContent,
-		},
-		metadata,
-		serde::{default_true, is_default},
-	};
-
-	metadata! {
-		method: GET,
-		rate_limited: false,
-		authentication: AccessToken,
-		history: {
-			1.0 => "/_continuwuity/admin/v1/rooms",
-		}
-	}
-
-	#[request]
-	#[derive(Default)]
-	pub struct Request {
-		/// The maximum number of results to return in this page. Maximum (and
-		/// default) is 100.
-		#[ruma_api(query)]
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub limit: Option<usize>,
-
-		/// The number of results to skip over before returning results. Default
-		/// is 0.
-		#[ruma_api(query)]
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub offset: Option<usize>,
-
-		/// If true, includes banned rooms in the response.
-		#[ruma_api(query)]
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub include_banned_rooms: bool,
-	}
-
-	#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
-	pub struct MinimalRoomInfo {
-		/// The room's unique ID.
-		pub room_id: OwnedRoomId,
-		/// If true, this room is banned, and cannot be joined by non-admins.
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub banned: bool,
-		/// If true, this room has federation disabled, but can still be locally
-		/// used.
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub disabled: bool,
-		/// The total number of joined members in this room.
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub member_count: usize,
-		/// The total number of joined members in this room that are local to
-		/// this server.
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub local_member_count: usize,
-		/// The number of unique homeservers currently joined to this room.
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub resident_server_count: usize,
-		/// The users who created this room.
-		///
-		/// The first entry is always the sender of the `m.room.create` event.
-		/// Any entries thereafter are additional creators in v12+ rooms. An
-		/// empty vec indicates the room is not known.
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub creators: Vec<OwnedUserId>,
-		/// If true, this room has encryption enabled.
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub encrypted: bool,
-		/// If true, this room is allowed to be federated (`m.federate` is not
-		/// `false` in `m.room.create`).
-		#[serde(default = "default_true", skip_serializing_if = "is_default")]
-		pub federated: bool,
-		/// If true, this room is published to this server's room directory.
-		#[serde(default, skip_serializing_if = "is_default")]
-		pub published: bool,
-		/// The version of the room.
-		pub version: RoomVersionId,
-		/// The event content for the `m.room.name` event, if any is present.
-		/// May be redacted.
-		#[serde(default, skip_serializing_if = "Option::is_none")]
-		pub name: Option<PossiblyRedactedRoomNameEventContent>,
-		/// The event content for the `m.room.topic` event, if any is present.
-		/// May be redacted.
-		#[serde(default, skip_serializing_if = "Option::is_none")]
-		pub topic: Option<PossiblyRedactedRoomTopicEventContent>,
-		/// The event content for the `m.room.canonical_alias` event, if any is
-		/// present. May be redacted.
-		#[serde(default, skip_serializing_if = "Option::is_none")]
-		pub canonical_alias: Option<PossiblyRedactedRoomCanonicalAliasEventContent>,
-		/// The event content for the `m.room.join_rules` event, if any is
-		/// present. May be redacted.
-		#[serde(default, skip_serializing_if = "Option::is_none")]
-		pub join_rules: Option<PossiblyRedactedRoomJoinRulesEventContent>,
-		/// The event content for the `m.room.history_visibility` event, if any
-		/// is present. May be redacted.
-		#[serde(default, skip_serializing_if = "Option::is_none")]
-		pub history_visibility: Option<PossiblyRedactedRoomHistoryVisibilityEventContent>,
-		/// The ID of the room which replaces this one, if any.
-		#[serde(default, skip_serializing_if = "Option::is_none")]
-		pub successor: Option<OwnedRoomId>,
-		/// The ID of the room which preceded this one, if any.
-		#[serde(default, skip_serializing_if = "Option::is_none")]
-		pub predecessor: Option<OwnedRoomId>,
-	}
-
-	#[response]
-	pub struct Response {
-		/// A list of rooms known to this server.
-		pub rooms: Vec<MinimalRoomInfo>,
-	}
-
-	impl Request {
-		#[must_use]
-		pub fn new() -> Self { Self::default() }
-	}
-
-	impl Response {
-		#[must_use]
-		pub fn new(rooms: Vec<MinimalRoomInfo>) -> Self { Self { rooms } }
-	}
-}

src/ruminuwuity/admin/continuwuity/users/create.rs

@@ -1,105 +0,0 @@
-pub mod v1 {
-	use ruma::{
-		OwnedMxcUri, OwnedRoomOrAliasId, OwnedUserId,
-		api::{OAuthClientScope, auth_scheme::AccessToken, request, response},
-		metadata,
-	};
-
-	metadata! {
-		method: POST,
-		rate_limited: false,
-		authentication: AccessToken,
-		required_scopes: [OAuthClientScope::ServerAdministration],
-		history: {
-			1.0 => "/_continuwuity/admin/v1/users/create",
-		},
-	}
-
-	#[request]
-	pub struct Request {
-		/// The user's localpart (the identifier between `@` and `:`). Cannot be
-		/// blank.
-		pub localpart: String,
-
-		/// The user's desired password. Cannot be blank.
-		pub password: String,
-
-		/// The user's email address, if any.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub email: Option<String>,
-
-		/// The display name to set upon creation.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub display_name: Option<String>,
-
-		/// The avatar URI to set upon creation.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub avatar_url: Option<OwnedMxcUri>,
-
-		/// Suspends the user immediately upon creation. They can still log in.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub suspended: bool,
-
-		/// Locks the user immediately upon creation. They will receive
-		/// M_USER_LOCKED upon login.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub locked: bool,
-
-		/// Disables the user's login immediately upon creation.
-		///
-		/// The user can still be used if an admin generates an access token for
-		/// the account, but the user will not be able to use `POST
-		/// /_matrix/client/v3/login`.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub login_disabled: bool,
-
-		/// Promotes the user to a server administrator immediately upon
-		/// creation.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub admin: bool,
-
-		/// Skips joining rooms in the server's configured auto_join_rooms.
-		///
-		/// If this is false, all rooms in the config.toml's `auto_join_rooms`
-		/// will be automatically joined upon creation. If `auto_join_rooms`
-		/// is supplied in this request too, those rooms will be joined
-		/// afterwards.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub skip_auto_join: bool,
-
-		/// Additional rooms to auto-join the new user to. If `skip_auto_join`
-		/// is `true`, these rooms will still be joined.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub auto_join_rooms: Vec<OwnedRoomOrAliasId>,
-	}
-
-	#[response]
-	pub struct Response {
-		/// The fully qualified user ID of the newly created user.
-		pub user_id: OwnedUserId,
-	}
-
-	impl Request {
-		#[must_use]
-		pub fn new(localpart: String, password: String) -> Self {
-			Self {
-				localpart,
-				password,
-				email: None,
-				display_name: None,
-				avatar_url: None,
-				suspended: false,
-				locked: false,
-				login_disabled: false,
-				admin: false,
-				skip_auto_join: false,
-				auto_join_rooms: Vec::new(),
-			}
-		}
-	}
-
-	impl Response {
-		#[must_use]
-		pub fn new(user_id: OwnedUserId) -> Self { Self { user_id } }
-	}
-}

src/ruminuwuity/admin/continuwuity/users/list.rs

@@ -1,139 +0,0 @@
-pub mod v1 {
-	use ruma::{
-		OwnedUserId,
-		api::{OAuthClientScope, auth_scheme::AccessToken, request, response},
-		metadata,
-	};
-	use serde::Deserialize;
-
-	metadata! {
-		method: GET,
-		rate_limited: false,
-		authentication: AccessToken,
-		required_scopes: [OAuthClientScope::ServerAdministration],
-		history: {
-			1.0 => "/_continuwuity/admin/v1/users",
-		}
-	}
-
-	#[request]
-	#[derive(Default)]
-	pub struct Request {
-		/// If true, includes deactivated users in the response.
-		#[ruma_api(query)]
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub include_deactivated: bool,
-		/// If true, includes locked users in the response.
-		#[ruma_api(query)]
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub include_locked: bool,
-		/// If true, includes suspended users in the response.
-		#[ruma_api(query)]
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub include_suspended: bool,
-
-		/// The maximum number of results to return in this page. Maximum (and
-		/// default) is 100.
-		#[ruma_api(query)]
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub limit: Option<usize>,
-
-		/// The number of results to skip over before returning results. Default
-		/// is 0.
-		#[ruma_api(query)]
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub offset: Option<usize>,
-	}
-
-	#[derive(Debug, Clone, PartialEq, Eq, Deserialize, serde::Serialize)]
-	pub struct User {
-		/// The full user ID of the user.
-		pub user_id: OwnedUserId,
-
-		/// Whether this user is deactivated.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub deactivated: bool,
-
-		/// Whether this user is suspended.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub suspended: bool,
-
-		/// Whether this user is locked.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub locked: bool,
-
-		/// Whether this user is an admin.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub admin: bool,
-
-		/// Whether this user has their login disabled.
-		#[serde(default, skip_serializing_if = "ruma::serde::is_default")]
-		pub login_disabled: bool,
-	}
-
-	impl User {
-		#[must_use]
-		pub fn new(user_id: OwnedUserId) -> Self {
-			Self {
-				user_id,
-				deactivated: false,
-				suspended: false,
-				locked: false,
-				admin: false,
-				login_disabled: false,
-			}
-		}
-	}
-
-	#[response]
-	#[derive(Default)]
-	pub struct Response {
-		pub users: Vec<User>,
-	}
-
-	impl Request {
-		#[must_use]
-		pub fn new() -> Self { Self::default() }
-	}
-
-	impl Response {
-		#[must_use]
-		pub fn new(users: Vec<User>) -> Self { Self { users } }
-	}
-
-	#[cfg(test)]
-	mod tests {
-		use assign::assign;
-		use serde_json::json;
-
-		use super::*;
-
-		#[test]
-		fn request_defaults() {
-			let req = Request::new();
-			assert!(!req.include_deactivated && !req.include_locked && !req.include_suspended);
-		}
-
-		#[test]
-		fn user_serialize_omits_default_values() {
-			let user_id = OwnedUserId::try_from("@alice:example.org".to_owned()).unwrap();
-			let user = User::new(user_id.clone());
-
-			let expected = json!({ "user_id": user_id.to_string() });
-			assert_eq!(serde_json::to_value(&user).expect("failed to serialize user"), expected);
-
-			let suspended_user = assign!(user, {suspended: true});
-			let expected2 = json!({ "user_id": "@alice:example.org", "suspended": true});
-			assert_eq!(
-				serde_json::to_value(&suspended_user).expect("failed to serialize user"),
-				expected2
-			);
-		}
-
-		#[test]
-		fn response_defaults() {
-			let response = Response::default();
-			assert!(response.users.is_empty());
-		}
-	}
-}

src/ruminuwuity/admin/continuwuity/users/mod.rs

@@ -1,2 +0,0 @@
-pub mod create;
-pub mod list;

src/ruminuwuity/admin/get_suspended.rs

@@ -0,0 +1,53 @@
+//! `GET /_matrix/client/v1/admin/suspend/{userId}`
+//!
+//! Check the suspension status of a target user
+
+pub mod v1 {
+	//! `/_matrix/client/unstable/uk.timedout.msc4323/admin/suspend/{userID}`
+	//! ([msc])
+	//!
+	//! [msc]: https://github.com/matrix-org/matrix-spec-proposals/pull/4323
+
+	use ruma::{
+		OwnedUserId,
+		api::{auth_scheme::AccessToken, request, response},
+		metadata,
+	};
+
+	metadata! {
+		method: GET,
+		rate_limited: false,
+		authentication: AccessToken,
+		history: {
+			unstable => "/_matrix/client/unstable/uk.timedout.msc4323/admin/suspend/{user_id}",
+			1.18 => "/_matrix/client/v1/admin/suspend/{user_id}",
+		}
+	}
+
+	/// Request type for the get & set user suspension status endpoint.
+	#[request(error = ruma::api::error::Error)]
+	pub struct Request {
+		/// The user to look up.
+		#[ruma_api(path)]
+		pub user_id: OwnedUserId,
+	}
+
+	/// Response type for the suspension endpoints
+	#[response(error = ruma::api::error::Error)]
+	pub struct Response {
+		/// Whether the user is currently suspended.
+		pub suspended: bool,
+	}
+
+	impl Request {
+		/// Creates a new `Request` with the given user id.
+		#[must_use]
+		pub fn new(user_id: OwnedUserId) -> Self { Self { user_id } }
+	}
+
+	impl Response {
+		/// Creates a new `Response` with the given suspension status.
+		#[must_use]
+		pub fn new(suspended: bool) -> Self { Self { suspended } }
+	}
+}

src/ruminuwuity/admin/mod.rs

@@ -1 +1,3 @@
 pub mod continuwuity;
+pub mod get_suspended;
+pub mod set_suspended;

src/ruminuwuity/admin/set_suspended.rs

@@ -0,0 +1,55 @@
+//! `PUT /_matrix/client/v1/admin/suspend/{userId}`
+//!
+//! Set the suspension status of a target user
+
+pub mod v1 {
+	//! `/_matrix/client/unstable/uk.timedout.msc4323/admin/suspend/{userID}`
+	//! ([msc])
+	//!
+	//! [msc]: https://github.com/matrix-org/matrix-spec-proposals/pull/4323
+
+	use ruma::{
+		OwnedUserId,
+		api::{auth_scheme::AccessToken, request, response},
+		metadata,
+	};
+
+	metadata! {
+		method: PUT,
+		rate_limited: false,
+		authentication: AccessToken,
+		history: {
+			unstable => "/_matrix/client/unstable/uk.timedout.msc4323/admin/suspend/{user_id}",
+			1.18 => "/_matrix/client/v1/admin/suspend/{user_id}",
+		}
+	}
+
+	/// Request type for the set user suspension status endpoint.
+	#[request(error = ruma::api::error::Error)]
+	pub struct Request {
+		/// The user to look up.
+		#[ruma_api(path)]
+		pub user_id: OwnedUserId,
+
+		pub suspended: bool,
+	}
+
+	/// Response type for the suspension endpoints
+	#[response(error = ruma::api::error::Error)]
+	pub struct Response {
+		/// Whether the user is currently suspended.
+		pub suspended: bool,
+	}
+
+	impl Request {
+		/// Creates a new `Request` with the given user id.
+		#[must_use]
+		pub fn new(user_id: OwnedUserId, suspended: bool) -> Self { Self { user_id, suspended } }
+	}
+
+	impl Response {
+		/// Creates a new `Response` with the given suspension status.
+		#[must_use]
+		pub fn new(suspended: bool) -> Self { Self { suspended } }
+	}
+}

src/service/oauth/grant.rs

@@ -8,7 +8,7 @@
 };
 
 use regex::Regex;
-use ruma::{OwnedDeviceId, api::OAuthClientScope};
+use ruma::OwnedDeviceId;
 use serde::{Deserialize, Serialize};
 use url::Url;
 
@@ -55,40 +55,26 @@ pub enum Prompt {
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize, PartialOrd, Ord)]
-pub enum RequestedScope {
+pub enum Scope {
 	Device(OwnedDeviceId),
-	ApiFullAccess,
-	ServerAdministration,
+	ClientApi,
 }
 
-impl RequestedScope {
-	#[must_use]
-	pub fn as_granted_scope(&self) -> Option<OAuthClientScope> {
-		match self {
-			| Self::ApiFullAccess => Some(OAuthClientScope::ApiFullAccess),
-			| Self::ServerAdministration => Some(OAuthClientScope::ServerAdministration),
-			| Self::Device(_) => None,
-		}
-	}
-}
-
-impl PartialEq for RequestedScope {
+impl PartialEq for Scope {
 	fn eq(&self, other: &Self) -> bool { discriminant(self) == discriminant(other) }
 }
 
-impl Eq for RequestedScope {}
+impl Eq for Scope {}
 
-impl Hash for RequestedScope {
+impl Hash for Scope {
 	fn hash<H: std::hash::Hasher>(&self, state: &mut H) { discriminant(self).hash(state); }
 }
 
-impl Display for RequestedScope {
+impl Display for Scope {
 	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 		let urn = match self {
-			| Self::ApiFullAccess => "urn:matrix:client:api:*".to_owned(),
+			| Self::ClientApi => "urn:matrix:client:api:*".to_owned(),
 			| Self::Device(device_id) => format!("urn:matrix:client:device:{device_id}"),
-			| Self::ServerAdministration =>
-				"urn:matrix:client:cc.c10y.msc4484.server_administration".to_owned(),
 		};
 
 		f.write_str(&urn)
@@ -99,27 +85,22 @@ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 pub struct RawScopes(String);
 
 impl RawScopes {
-	pub fn to_scopes(&self) -> Result<BTreeSet<RequestedScope>, String> {
-		let full_access_regex =
+	pub fn to_scopes(&self) -> Result<BTreeSet<Scope>, String> {
+		let client_api_token_regex =
 			Regex::new(r"urn:matrix:(client|org.matrix.msc2967.client):api:\*").unwrap();
 		let device_token_regex = Regex::new(
 			r"urn:matrix:(client|org.matrix.msc2967.client):device:([a-zA-Z0-9-._~]{5,})",
 		)
 		.unwrap();
-		let server_administration_regex =
-			Regex::new(r"urn:matrix:client:cc.c10y.msc4484.server_administration").unwrap();
 
 		let mut scopes = BTreeSet::new();
 
 		for token in self.0.split(' ') {
 			let scope_was_new = {
-				if full_access_regex.is_match(token) {
-					scopes.insert(RequestedScope::ApiFullAccess)
+				if client_api_token_regex.is_match(token) {
+					scopes.insert(Scope::ClientApi)
 				} else if let Some(captures) = device_token_regex.captures(token) {
-					scopes
-						.insert(RequestedScope::Device(captures.get(2).unwrap().as_str().into()))
-				} else if server_administration_regex.is_match(token) {
-					scopes.insert(RequestedScope::ServerAdministration)
+					scopes.insert(Scope::Device(captures.get(2).unwrap().as_str().into()))
 				} else if token == "openid" {
 					// TODO(unspecced): Element sets this scope but doesn't use it for anything
 					true
@@ -179,15 +160,8 @@ pub enum ErrorCode {
 	InvalidClientMetadata,
 }
 
-#[derive(Serialize)]
-#[serde(untagged)]
-pub enum AuthorizationCodeResponse {
-	Success(AuthorizationCodeData),
-	Error(OAuthError),
-}
-
 #[derive(Serialize, Deserialize)]
-pub struct AuthorizationCodeData {
+pub struct AuthorizationCodeResponse {
 	pub state: String,
 	pub code: String,
 }

src/service/oauth/mod.rs

@@ -12,7 +12,7 @@
 use database::{Deserialized, Json, Map};
 use itertools::Itertools;
 use lru_cache::LruCache;
-use ruma::{DeviceId, OwnedDeviceId, OwnedUserId, UserId, api::OAuthClientScope};
+use ruma::{DeviceId, OwnedDeviceId, OwnedUserId, UserId};
 use serde::{Deserialize, Serialize};
 use url::Url;
 
@@ -21,9 +21,8 @@
 	oauth::{
 		client_metadata::{ApplicationType, ClientMetadata, ResponseType},
 		grant::{
-			AuthorizationCodeData, AuthorizationCodeQuery, AuthorizationCodeResponse,
-			CodeChallengeMethod, ErrorCode, OAuthError, RequestedScope, ResponseMode,
-			TokenRequest, TokenResponse, TokenType,
+			AuthorizationCodeQuery, AuthorizationCodeResponse, CodeChallengeMethod, ErrorCode,
+			OAuthError, ResponseMode, Scope, TokenRequest, TokenResponse, TokenType,
 		},
 	},
 	users,
@@ -52,7 +51,7 @@ struct Services {
 #[derive(Debug, Deserialize, Serialize)]
 pub struct SessionInfo {
 	pub client_id: String,
-	pub scopes: BTreeSet<OAuthClientScope>,
+	pub scopes: BTreeSet<Scope>,
 	current_refresh_token: String,
 }
 
@@ -65,7 +64,7 @@ struct RefreshTokenInfo {
 
 struct PendingCodeGrant {
 	authorizing_user: OwnedUserId,
-	requested_scopes: BTreeSet<RequestedScope>,
+	requested_scopes: BTreeSet<Scope>,
 	client_name: Option<String>,
 	expected_client_id: String,
 	expected_redirect_uri: Url,
@@ -225,59 +224,49 @@ pub async fn request_authorization_code(
 			}
 		}
 
+		let requested_scopes = query.scope.to_scopes()?;
+
 		let redirect_uri_query_separator = match query.response_mode {
 			| ResponseMode::Fragment => '#',
 			| ResponseMode::Query => '?',
 		};
 
-		let response = 'response: {
-			let requested_scopes = query.scope.to_scopes()?;
+		let code = PendingCodeGrant::generate_code();
 
-			if requested_scopes.contains(&RequestedScope::ServerAdministration) {
-				// Only server admins can request this scope
-				if !self.services.users.is_admin(&authorizing_user).await {
-					break 'response AuthorizationCodeResponse::Error(OAuthError {
-						error: ErrorCode::AccessDenied,
-						error_description: "You are not a server administrator.".into(),
-					});
-				}
-			}
-
-			let code = PendingCodeGrant::generate_code();
-
-			info!(
-				client_id = &query.client_id,
-				client_name = &client_metadata.client_name,
-				?requested_scopes,
-				?authorizing_user,
-				"Issuing oauth authorization code"
-			);
-
-			let pending_grant = PendingCodeGrant {
-				authorizing_user,
-				requested_scopes,
-				client_name: client_metadata.client_name,
-				expected_client_id: query.client_id,
-				expected_redirect_uri: query.redirect_uri.clone(),
-				code_challenge: query.code_challenge,
-				requested_at: SystemTime::now(),
-			};
-
-			self.pending_code_grants
-				.lock()
-				.await
-				.insert(code.clone(), pending_grant);
-
-			AuthorizationCodeResponse::Success(AuthorizationCodeData { state: query.state, code })
-		};
+		info!(
+			client_id = &query.client_id,
+			client_name = &client_metadata.client_name,
+			?requested_scopes,
+			?authorizing_user,
+			"Issuing oauth authorization code"
+		);
 
 		let redirect_uri = format!(
 			"{}{}{}",
 			query.redirect_uri,
 			redirect_uri_query_separator,
-			serde_urlencoded::to_string(response).unwrap(),
+			serde_urlencoded::to_string(AuthorizationCodeResponse {
+				state: query.state,
+				code: code.clone(),
+			})
+			.unwrap(),
 		);
 
+		let pending_grant = PendingCodeGrant {
+			authorizing_user,
+			requested_scopes,
+			client_name: client_metadata.client_name,
+			expected_client_id: query.client_id,
+			expected_redirect_uri: query.redirect_uri,
+			code_challenge: query.code_challenge,
+			requested_at: SystemTime::now(),
+		};
+
+		self.pending_code_grants
+			.lock()
+			.await
+			.insert(code, pending_grant);
+
 		Ok(redirect_uri)
 	}
 
@@ -350,7 +339,7 @@ pub async fn revoke_token(&self, token: String) -> Result<(), OAuthError> {
 	async fn create_session(
 		&self,
 		authorizing_user: OwnedUserId,
-		requested_scopes: BTreeSet<RequestedScope>,
+		requested_scopes: BTreeSet<Scope>,
 		client_name: Option<String>,
 		client_id: String,
 	) -> Result<TokenResponse, OAuthError> {
@@ -360,7 +349,7 @@ async fn create_session(
 		let device_id = requested_scopes
 			.iter()
 			.find_map(|scope| {
-				if let RequestedScope::Device(device_id) = scope {
+				if let Scope::Device(device_id) = scope {
 					Some(device_id)
 				} else {
 					None
@@ -402,10 +391,7 @@ async fn create_session(
 			Json(SessionInfo {
 				client_id: client_id.clone(),
 				current_refresh_token: refresh_token.clone(),
-				scopes: requested_scopes
-					.iter()
-					.filter_map(RequestedScope::as_granted_scope)
-					.collect(),
+				scopes: requested_scopes.clone(),
 			}),
 		);
 

src/service/oidc/mod.rs

@@ -33,7 +33,7 @@
 
 use crate::{
 	Dep, config, globals, media,
-	oauth::grant::AuthorizationCodeData,
+	oauth::grant::AuthorizationCodeResponse,
 	threepid,
 	users::{self, AccountStatus, ProfileFieldChange},
 };
@@ -245,7 +245,7 @@ pub async fn begin_session(&self, prompt: Option<CoreAuthPrompt>) -> (PendingSes
 	pub async fn exchange_code(
 		&self,
 		session: PendingSession,
-		response: AuthorizationCodeData,
+		response: AuthorizationCodeResponse,
 	) -> Result<Claims, &'static str> {
 		let Some(OidcClient { machine, client, .. }) = self.client.as_ref() else {
 			return Err("Delegated authentication is not enabled on this server.");

src/service/rooms/event_handler/acl_check.rs

@@ -14,6 +14,7 @@ pub async fn acl_check(&self, server_name: &ServerName, room_id: &RoomId) -> Res
 			.await
 			.map(|c: RoomServerAclEventContent| c)
 		else {
+			trace!("Room has no ACL, allowing");
 			return Ok(());
 		};
 

src/service/rooms/event_handler/fetch_and_handle_outliers.rs

@@ -1,5 +1,6 @@
 use std::{
 	collections::{HashMap, HashSet, VecDeque},
+	fmt::{Display, Formatter},
 	time::Instant,
 };
 
@@ -33,6 +34,20 @@ pub enum DagBuilderTree {
 	AuthEvents,
 }
 
+impl DagBuilderTree {
+	#[must_use]
+	pub fn as_str(&self) -> &str {
+		match self {
+			| Self::AuthEvents => "auth_events",
+			| Self::PrevEvents => "prev_events",
+		}
+	}
+}
+
+impl Display for DagBuilderTree {
+	fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.write_str(self.as_str()) }
+}
+
 /// Attempts to build a localised directed acyclic graph out of the given PDUs,
 /// returning them in a topologically sorted order.
 ///
@@ -42,23 +57,22 @@ pub enum DagBuilderTree {
 /// not account for power levels or other tie breaks.
 #[allow(clippy::implicit_hasher)]
 pub async fn build_local_dag(
-	pdu_map: &HashMap<OwnedEventId, &CanonicalJsonObject>,
+	pdu_map: &HashMap<OwnedEventId, CanonicalJsonObject>,
 	tree: DagBuilderTree,
 ) -> Result<Vec<OwnedEventId>> {
-	debug_assert!(pdu_map.len() >= 2, "needless call to build_local_dag with less than 2 PDUs");
+	if pdu_map.len() <= 1 {
+		return Ok(pdu_map.keys().cloned().collect());
+	}
+
 	let mut dag: HashMap<OwnedEventId, HashSet<OwnedEventId>> =
 		HashMap::with_capacity(pdu_map.len());
 	let mut id_origin_ts: HashMap<OwnedEventId, _> = HashMap::with_capacity(pdu_map.len());
-	let tree = match tree {
-		| DagBuilderTree::AuthEvents => "auth_events",
-		| DagBuilderTree::PrevEvents => "prev_events",
-	};
 
 	for (event_id, value) in pdu_map {
 		// Parse all prev events as event IDs - if they are missing, return an error (we
 		// can't sanely continue in this case), otherwise skip invalid prev events.
 		let prev_events = value
-			.get(tree)
+			.get(tree.as_str())
 			.and_then(CanonicalJsonValue::as_array)
 			.ok_or_else(|| err!(Request(BadJson("event JSON for {event_id} is missing {tree}"))))?
 			.iter()
@@ -370,9 +384,8 @@ async fn fetch_event_via(
 			.send_federation_request(&remote, get_event::v1::Request::new(event_id.clone()))
 			.await?;
 
-		let (calculated_event_id, value) = self
-			.parse_incoming_pdu_with_known_room(&res.pdu, room_version_rules)
-			.await?;
+		let (calculated_event_id, value) =
+			Self::parse_incoming_pdu_with_known_room(&res.pdu, room_version_rules)?;
 
 		if calculated_event_id != event_id {
 			Err!(Request(BadJson(warn!(
@@ -538,11 +551,7 @@ pub(super) async fn fetch_and_handle_auth_events<Pdu>(
 			}
 		}
 
-		let refmap: HashMap<OwnedEventId, &CanonicalJsonObject> = discovered_events
-			.iter()
-			.map(|(id, data)| (id.clone(), data))
-			.collect();
-		let seeded_ordered = build_local_dag(&refmap, DagBuilderTree::AuthEvents)
+		let seeded_ordered = build_local_dag(&discovered_events, DagBuilderTree::AuthEvents)
 			.await
 			.expect("failed to build local DAG");
 		let mut pdus = HashMap::with_capacity(seeded_ordered.len());

src/service/rooms/event_handler/fetch_auth.rs

@@ -1,9 +1,8 @@
 use std::collections::{HashMap, hash_map};
 
-use conduwuit::{Err, Event, EventTypeExt, PduEvent, Result, err};
+use conduwuit::{Err, Event, EventTypeExt, PduEvent, Result, err, warn};
 use ruma::{
-	CanonicalJsonObject, OwnedEventId, ServerName,
-	api::federation::authorization::get_event_authorization,
+	OwnedEventId, ServerName, api::federation::authorization::get_event_authorization,
 	room_version_rules::RoomVersionRules,
 };
 use tokio::join;
@@ -50,16 +49,49 @@ pub(super) async fn fetch_and_persist_event_auth<Pdu>(
 
 		for auth_pdu_json in event_auth.auth_chain {
 			let (auth_event_room_id, auth_event_id, auth_pdu_json) =
-				self.parse_incoming_pdu(&auth_pdu_json).await?;
+				match self.parse_incoming_pdu(&auth_pdu_json).await {
+					| Ok(parsed) => parsed,
+					| Err(e) => {
+						warn!(error=?e, "Dropping auth chain event as it could not be parsed");
+						continue;
+					},
+				};
+			if let Err(e) = Self::pdu_format_check_1(
+				&auth_pdu_json,
+				room_version_rules,
+				create_event.event_id(),
+			) {
+				// drop this PDU
+				warn!(%auth_event_id, error=?e, "Dropping auth chain event as it violates the room event format");
+				continue;
+			}
+			let auth_pdu_json = match self
+				.signature_hash_check_2_3(auth_pdu_json, room_version_rules)
+				.await
+			{
+				| Ok(pdu_json) => pdu_json,
+				| Err(e) => {
+					// drop this PDU
+					warn!(
+						%auth_event_id,
+						error=?e,
+						"Dropping auth chain event as it has an invalid signature"
+					);
+					continue;
+				},
+			};
+
+			// PDU check 4 is done when we've finished aggregating
 			if auth_event_room_id != incoming_event.room_id_or_hash() {
 				return Err!(Request(Forbidden(
-					"Auth event {auth_event_id} is in {auth_event_room_id}, not {}.",
+					"Auth chain event {auth_event_id} is in {auth_event_room_id}, not {}.",
 					incoming_event.room_id_or_hash()
 				)));
 			}
 			let auth_pdu = PduEvent::from_id_val(&auth_event_id, auth_pdu_json).map_err(|e| {
 				err!(Request(BadJson("Invalid PDU {auth_event_id} in auth chain: {e}")))
 			})?;
+
 			if auth_pdu.state_key().is_none() {
 				return Err!(Request(BadJson(
 					"Invalid PDU {auth_event_id} in auth_chain: not a state event"
@@ -97,17 +129,11 @@ async fn authorise_remote_auth_chain<Pdu>(
 	where
 		Pdu: Event + Send + Sync,
 	{
-		// TODO(nex): build_local_dag's signature needs changing because all callsites
-		// seem to do this refmap hack.
 		let pdu_objects = auth_chain_map
 			.iter()
-			.map(|(event_id, pdu)| (event_id, pdu.to_canonical_object()))
+			.map(|(event_id, pdu)| (event_id.clone(), pdu.to_canonical_object()))
 			.collect::<HashMap<_, _>>();
-		let refmap: HashMap<OwnedEventId, &CanonicalJsonObject> = pdu_objects
-			.iter()
-			.map(|(id, data)| ((*id).to_owned(), data))
-			.collect();
-		let auth_chain_topo = build_local_dag(&refmap, DagBuilderTree::AuthEvents)
+		let auth_chain_topo = build_local_dag(&pdu_objects, DagBuilderTree::AuthEvents)
 			.await?
 			.into_iter()
 			.map(|event_id| (event_id.clone(), auth_chain_map.get(&event_id).unwrap().to_owned()))
@@ -127,6 +153,12 @@ async fn authorise_remote_auth_chain<Pdu>(
 				continue;
 			}
 
+			// IMPORTANT: We can't use the handy dandy `handle_outlier_pdu` function here
+			// because it may then try to fetch missing auth events, resulting in deep
+			// recursion. We will do the minimum required steps to validate the PDU here.
+			// Checks 1-3 were already done before this function is called, so we only need
+			// to do check 4.
+
 			let mut auth_events_by_key: HashMap<_, _> =
 				HashMap::with_capacity(pdu.auth_events.len());
 
@@ -150,24 +182,21 @@ async fn authorise_remote_auth_chain<Pdu>(
 					},
 					| hash_map::Entry::Occupied(_) => {
 						// Duplicate auth events by key are not allowed.
-						self.services
-							.pdu_metadata
-							.mark_event_rejected(auth_event_id);
-						self.services.pdu_metadata.mark_event_rejected(&event_id);
+						self.reject_and_persist(&event_id, &pdu.to_canonical_object());
 						continue 'outer;
 					},
 				}
 			}
 			if !self
-				.is_event_self_authorised(
+				.auth_state_check_4(
 					&pdu,
 					room_version_rules,
 					create_event.as_pdu(),
 					&auth_events_by_key,
 				)
-				.await
+				.await?
 			{
-				self.services.pdu_metadata.mark_event_rejected(&event_id);
+				self.reject_and_persist(&event_id, &pdu.to_canonical_object());
 			}
 		}
 

src/service/rooms/event_handler/fetch_prev.rs

@@ -5,13 +5,13 @@
 	utils::{BoolExt, IterStream, stream::BroadbandExt},
 };
 use futures::StreamExt;
-use ruma::{CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName};
+use ruma::{MilliSecondsSinceUnixEpoch, RoomId, ServerName};
 
 use crate::rooms::event_handler::{build_local_dag, fetch_and_handle_outliers::DagBuilderTree};
 
 impl super::Service {
 	/// Fetches any missing prev_events for this event and persists them before
-	/// returning.
+	/// returning. The caller is responsible for then handling the incoming PDU.
 	pub(super) async fn fetch_prevs(
 		&self,
 		room_id: &RoomId,
@@ -84,13 +84,7 @@ pub(super) async fn fetch_prevs(
 			})
 			.collect::<HashMap<_, _>>();
 
-		let to_persist = if mapped.len() <= 1 {
-			mapped.keys().map(ToOwned::to_owned).collect()
-		} else {
-			let refmap: HashMap<OwnedEventId, &CanonicalJsonObject> =
-				mapped.iter().map(|(id, data)| (id.clone(), data)).collect();
-			build_local_dag(&refmap, DagBuilderTree::PrevEvents).await?
-		};
+		let to_persist = build_local_dag(&mapped, DagBuilderTree::PrevEvents).await?;
 
 		let job_start = Instant::now();
 		trace!("Starting to persist {} prev events", to_persist.len());

src/service/rooms/event_handler/handle_incoming_pdu.rs

@@ -4,46 +4,37 @@
 };
 
 use conduwuit::{
-	Err, Event, Result, debug, debug_error, debug_warn, defer, err, error, info,
-	matrix::PartialPdu, result::DebugInspect, trace, utils::time::jitter, warn,
+	Err, Event, Result, debug, debug_error, debug_warn, defer, matrix::PartialPdu, trace,
+	utils::time::jitter,
 };
-use futures::{
-	FutureExt, StreamExt,
-	future::{OptionFuture, try_join4},
-};
-use ruma::{CanonicalJsonValue, EventId, OwnedUserId, RoomId, ServerName, UserId};
+use futures::{FutureExt, StreamExt, future::try_join3};
+use ruma::{CanonicalJsonValue, EventId, RoomId, ServerName, UserId};
 use tokio::sync::mpsc;
 
-use crate::rooms::timeline::{RawPduId, pdu_fits};
+use crate::rooms::timeline::RawPduId;
 
 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
+	/// Handles an incoming PDU from federation.
+	///
+	/// First checks that we want to receive this PDU. If we already have it as
+	/// a timeline PDU, or we don't want to receive the PDU (e.g. origin ACL'd,
+	/// room disabled/unknown), abort.
+	///
+	/// The PDU is then handled as an outlier event, which performs [PDU checks]
+	/// 1 through 4. See: `handle_outlier_pdu`.
+	///
+	/// Once handled as an outlier, any missing prev events are fetched, and
+	/// then the PDU will be promoted/upgraded from an outlier to a timeline
+	/// event clients can see. See: `upgrade_outlier_to_timeline_pdu`. After
+	/// this finishes, the PDU is either accepted or left as an outlier.
+	///
+	/// If the PDU is successfully upgraded, the remaining extremity count of
+	/// the room is checked. If there are a potentially problematic number of
+	/// forward extremities, a squasher task is started with a debounce period,
+	/// which will eventually send a dummy event that ties up as many DAG forks
+	/// as possible.
+	///
+	/// [PDU checks]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
 	#[tracing::instrument(
 		name = "pdu",
 		skip_all,
@@ -55,51 +46,33 @@ pub async fn handle_incoming_pdu<'a>(
 		room_id: &'a RoomId,
 		event_id: &'a EventId,
 		value: BTreeMap<String, CanonicalJsonValue>,
-		is_timeline_event: bool,
+		is_backfilled_event: bool,
 	) -> Result<Option<RawPduId>> {
-		// 1. Skip the PDU if we already have it as a timeline event
+		// Skip the PDU if we already have it as a timeline event. We still re-process
+		// outliers in this scenario.
 		if let Ok(pdu_id) = self.services.timeline.get_pdu_id(event_id).await {
+			debug!("Database hit for incoming PDU, skipping processing");
 			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);
+		// If this is a membership state event for a local user, we will be interested
+		// in this event even if we aren't in the room. This allows us to process things
+		// like revoking invites over federation, without being in the room.
+		let is_interesting_member_event = value.get("type").and_then(|t| t.as_str())
+			== Some("m.room.member")
+			&& value
+				.get("state_key")
+				.and_then(|s| s.as_str())
+				.and_then(|s| UserId::parse(s).ok())
+				.is_some_and(|u| self.services.globals.user_is_local(&u));
 
-		// 1.2 Check if the room is disabled
-		let is_disabled = self.services.metadata.is_disabled(room_id).map(Ok);
-
-		// 1.3.1 Check room ACL on origin field/server
-		let origin_acl_check = self.acl_check(origin, room_id);
-
-		// 1.3.2 Check room ACL on sender's server name
-		let sender: OwnedUserId = value
-			.get("sender")
-			.and_then(|v| v.as_str())
-			.ok_or_else(|| err!("No sender in object"))
-			.and_then(|v| Ok(UserId::parse(v)?))
-			.map_err(|e| err!(Request(BadJson("PDU does not have a valid sender key: {e}"))))?;
-
-		let sender_acl_check: OptionFuture<_> = sender
-			.server_name()
-			.ne(origin)
-			.then(|| self.acl_check(sender.server_name(), room_id))
-			.into();
-
-		let (meta_exists, is_disabled, (), ()) = try_join4(
-			meta_exists,
-			is_disabled,
-			origin_acl_check,
-			sender_acl_check.map(|o| o.unwrap_or(Ok(()))),
+		let (room_exists, is_disabled, ()) = try_join3(
+			self.services.metadata.exists(room_id).map(Ok),
+			self.services.metadata.is_disabled(room_id).map(Ok),
+			self.acl_check(origin, room_id),
 		)
 		.await
 		.inspect_err(
@@ -112,21 +85,11 @@ pub async fn handle_incoming_pdu<'a>(
 			)));
 		}
 
-		if !self
-			.services
-			.state_cache
-			.server_in_room(self.services.globals.server_name(), room_id)
-			.await
-		{
-			info!(
-				%origin,
-				%room_id,
-				"Dropping inbound PDU for room we aren't participating in"
-			);
-			return Err!(Request(NotFound("This server is not participating in that room.")));
-		}
-
-		if !meta_exists {
+		if !room_exists {
+			if is_interesting_member_event {
+				// TODO: handle interesting membership events where we aren't in
+				// the room
+			}
 			return Err!(Request(NotFound("Room is unknown to this server")));
 		}
 
@@ -152,8 +115,10 @@ pub async fn handle_incoming_pdu<'a>(
 			.handle_outlier_pdu(origin, create_event, event_id, room_id, value)
 			.await?;
 
-		// 8. if not timeline event: stop
-		if !is_timeline_event {
+		// If this event is being processed as part of backfill, we don't want to end up
+		// *appending* it during the upgrade process, so we return early.
+		if is_backfilled_event {
+			debug!("Not promoting incoming event as it is being backfilled");
 			return Ok(None);
 		}
 
@@ -166,11 +131,16 @@ pub async fn handle_incoming_pdu<'a>(
 			.await?
 			.origin_server_ts();
 		if incoming_pdu.origin_server_ts() < first_ts_in_room {
+			debug_warn!(
+				"Not promoting incoming event as it is sent before we joined the room (but was \
+				 not backfilled)"
+			);
 			return Ok(None);
 		}
 
-		// 9. Fetch any missing prev events doing all checks listed here starting at 1.
-		//    These are timeline events
+		// Fetch any missing prev events doing all checks listed here starting at 1.
+		// These are timeline events.
+		// TODO: This part needs to be done in a background queue somewhere.
 
 		debug!("Fetching and persisting any missing prev events");
 		Box::pin(self.fetch_prevs(
@@ -181,13 +151,14 @@ pub async fn handle_incoming_pdu<'a>(
 			first_ts_in_room,
 		))
 		.await
-		.debug_inspect_err(|e| {
-			error!("Failed to fetch and persist incoming event's prev_events: {e:?}");
+		.inspect_err(|e| {
+			debug_error!("Failed to fetch and persist incoming event's prev_events: {e:?}");
 		})?;
 
-		let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event";
+		let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event"
+			&& incoming_pdu.state_key().is_none();
 
-		// Done with prev events, now handling the incoming event
+		// Done with prev events, now we can handle promoting the PDU
 		let pdu_id = Box::pin(self.upgrade_outlier_to_timeline_pdu(
 			incoming_pdu,
 			val,

src/service/rooms/event_handler/handle_outlier_pdu.rs

@@ -1,23 +1,22 @@
 use std::collections::{BTreeMap, HashMap, hash_map};
 
 use conduwuit::{
-	Err, Event, EventTypeExt, PduEvent, Result, debug, debug_info, debug_warn, err, info,
-	matrix::StateKey, state_res::auth_check, trace, warn,
-};
-use futures::future::ready;
-use ruma::{
-	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName,
-	canonical_json::redact, events::StateEventType, room_version_rules::RoomVersionRules,
+	Err, Event, EventTypeExt, PduEvent, Result, debug, debug_warn, err, info, trace,
 };
+use ruma::{CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName};
 
-use super::{check_room_id, get_room_version_rules};
-use crate::rooms::timeline::pdu_fits;
+use super::get_room_version_rules;
 
 impl super::Service {
 	/// Handles a PDU as an outlier, performing basic checks like signatures and
 	/// hashes, proclaimed event auth, and then adding it to the outlier tree.
+	///
+	/// This performs steps 1 through 4 of [S-S section 5.1][spec], returning
+	/// the parsed PDU and modified JSON object.
+	///
+	/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
 	#[allow(clippy::too_many_arguments)]
-	#[tracing::instrument(name="handle_outlier", skip_all, fields(%event_id))]
+	#[tracing::instrument(name = "handle_outlier", skip_all)]
 	pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 		&self,
 		origin: &'a ServerName,
@@ -29,61 +28,39 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 	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."
+		// Skip outlier handling if we already have this event as either a timeline or
+		// outlier PDU.
+		if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
+			debug!(
+				"Database hit for {event_id} (event is either an outlier or already promoted), \
+				 skipping outlier handling"
 			);
-			return Err!(Request(TooLarge("PDU is too large")));
+			value.insert(
+				"event_id".to_owned(),
+				CanonicalJsonValue::String(event_id.as_str().to_owned()),
+			);
+			return Ok((pdu_event, value));
 		}
-		// 1. Remove unsigned field
-		value.remove("unsigned");
 
-		// 2. Check signatures, otherwise drop
-		// 3. check content hash, redact if doesn't match
+		// 1. Check that the PDU follows the format for the room version
+		// (in this case, just size check)
 		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
-		{
-			| Ok(ruma::signatures::Verified::All) => {
-				if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
-					debug!(
-						"Already have event {event_id} as an outlier or timeline event, not \
-						 re-processing"
-					);
-					value.insert(
-						"event_id".to_owned(),
-						CanonicalJsonValue::String(event_id.as_str().to_owned()),
-					);
-					check_room_id(room_id, &pdu_event)?;
-					return Ok((pdu_event, value));
-				}
-				value
-			},
-			| Ok(ruma::signatures::Verified::Signatures) => {
-				if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
-					debug!(
-						"Received a redacted copy of {event_id}, but we already knew about it. \
-						 Re-using known content instead."
-					);
-					check_room_id(room_id, &pdu_event)?;
-					let obj = pdu_event.to_canonical_object();
-					return Ok((pdu_event, obj));
-				}
+		Self::pdu_format_check_1(&value, &room_version_rules, create_event.event_id())
+			.inspect_err(|e| {
+				info!(
+					err=?e,
+					"Dropping incoming PDU from {origin} because it violates the room event format"
+				);
+			})?;
 
-				debug_info!("Calculated hash does not match (redaction): {event_id}");
-				redact(value, &room_version_rules.redaction, None)
-					.map_err(|e| err!(Request(BadJson("Failed to redact {event_id}: {e}"))))?
-			},
-			| Err(e) => {
-				return Err!(Request(Forbidden(debug_error!(
-					"Signature verification failed for {event_id}: {e}"
-				))));
-			},
-		};
+		value.remove("unsigned");
+		let room_version_rules = get_room_version_rules(create_event)?;
+
+		// 2. Check signatures, otherwise drop.
+		// 3. Check content hash, redacting the event if it fails.
+		let mut incoming_pdu = self
+			.signature_hash_check_2_3(value, &room_version_rules)
+			.await?;
 
 		// Now that we have checked the signature and hashes we can add the eventID and
 		// convert to our PduEvent type
@@ -91,14 +68,11 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 			"event_id".to_owned(),
 			CanonicalJsonValue::String(event_id.as_str().to_owned()),
 		);
-
 		let pdu_event = serde_json::from_value::<PduEvent>(
 			serde_json::to_value(&incoming_pdu).expect("CanonicalJsonObj is a valid JsonValue"),
 		)
 		.map_err(|e| err!(Request(BadJson(debug_warn!("Event is not a valid PDU: {e}")))))?;
 
-		check_room_id(room_id, &pdu_event)?;
-
 		// TODO(nex): From hereon the event is not dropped, and thus always added as an
 		// outlier. However, we only do that at the end of this function, which means
 		// several duplicated calls to add_pdu_outlier. Shouldn't we just do it here
@@ -116,7 +90,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 
 		for auth_event_id in pdu_event.auth_events() {
 			if let Ok(auth_event) = self.services.timeline.get_pdu(auth_event_id).await {
-				check_room_id(room_id, &auth_event)?;
 				trace!("Found auth event {auth_event_id} for outlier event {event_id} locally");
 				auth_events.insert(auth_event_id.to_owned(), auth_event);
 			} else {
@@ -154,7 +127,7 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 		}
 
 		// Ensure none of the auth events are rejected - if they are, reject too.
-		for auth_event_id in auth_events.keys() {
+		for (auth_event_id, auth_event) in &auth_events {
 			if self
 				.services
 				.pdu_metadata
@@ -166,15 +139,27 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 					 {auth_event_id}",
 					event_id,
 				);
-				self.services.pdu_metadata.mark_event_rejected(event_id);
+				self.reject_and_persist(event_id, &incoming_pdu);
 				return Err!(Request(Forbidden(
 					"Event has rejected auth event: {auth_event_id}"
 				)));
 			}
+			if auth_event.room_id_or_hash() != room_id {
+				debug_warn!(
+					%auth_event_id,
+					auth_event_room_id=%auth_event.room_id_or_hash(),
+					expected_room_id=%room_id,
+					"Rejecting incoming event which depends on an auth event in another room.",
+				);
+				self.reject_and_persist(event_id, &incoming_pdu);
+				return Err!(Request(Forbidden(
+					"Event depends on a cross-room auth event: {auth_event_id}"
+				)));
+			}
 		}
 
-		// 6. Reject "due to auth events" if the event doesn't pass auth based on the
-		//    auth events
+		// 4. Reject "due to auth events" if the event doesn't pass auth based on the
+		//    claimed auth events
 		debug!("Checking based on auth events");
 		let mut auth_events_by_key: HashMap<_, _> = HashMap::with_capacity(auth_events.len());
 		// Build map of auth events
@@ -184,8 +169,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 				.expect("we just checked that we have all auth events")
 				.to_owned();
 
-			check_room_id(room_id, &auth_event)?;
-
 			let key = auth_event.kind().with_state_key(
 				auth_event
 					.state_key
@@ -197,98 +180,45 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 					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(Forbidden(
+					self.reject_and_persist(event_id, &incoming_pdu);
+					return Err!(Request(Forbidden(debug_warn!(
 						"Auth event's type and state_key combination exists multiple times: {}, \
 						 {}",
 						auth_event.kind,
 						auth_event.state_key().unwrap_or("")
-					)));
+					))));
 				},
 			}
 		}
 
 		if !self
-			.is_event_self_authorised(
+			.auth_state_check_4(
 				&pdu_event,
 				&room_version_rules,
 				create_event.as_pdu(),
 				&auth_events_by_key,
 			)
-			.await
+			.await?
 		{
-			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(
+			self.reject_and_persist(event_id, &incoming_pdu);
+			return Err!(Request(Forbidden(debug_warn!(
 				"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.");
+		debug!("PDU passed checks and has been persisted as an outlier");
 
 		Ok((pdu_event, incoming_pdu))
 	}
 
-	/// Helper method that turns the return value of `is_event_self_authorised`
-	/// into a `Result` depending on the value.
-	///
-	/// If the event is not authorised, a Forbidden error is returned.
-	/// Otherwise, an empty `Ok`.
-	pub(super) async fn is_event_self_authorised(
-		&self,
-		pdu: &PduEvent,
-		room_version_rules: &RoomVersionRules,
-		create_event: &PduEvent,
-		auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>,
-	) -> bool {
-		self.expect_event_is_self_authorised(
-			pdu,
-			room_version_rules,
-			create_event,
-			auth_events_by_key,
-		)
-		.await
-		.is_ok()
-	}
-
-	/// Checks PDU check 4: Passes authorisation rules based on the event's auth
-	/// events ([spec]).
-	///
-	/// If the auth check fails, false is returned, otherwise true.
-	///
-	/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
-	pub(super) async fn expect_event_is_self_authorised(
-		&self,
-		pdu: &PduEvent,
-		room_version_rules: &RoomVersionRules,
-		create_event: &PduEvent,
-		auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>,
-	) -> Result<bool> {
-		let state_fetch = |ty: &StateEventType, sk: &str| {
-			let key = (ty.to_owned(), sk.into());
-			ready(auth_events_by_key.get(&key).map(ToOwned::to_owned))
-		};
-
-		auth_check(
-			room_version_rules,
-			pdu,
-			None, // TODO: third party invite
-			state_fetch,
-			create_event,
-		)
-		.await
-		.map_err(|e| err!("Event self-authentication failed: {e:?}"))
+	/// Marks the event as rejected and then saves it as an outlier.
+	pub(super) fn reject_and_persist(&self, event_id: &EventId, pdu: &CanonicalJsonObject) {
+		self.services.pdu_metadata.mark_event_rejected(event_id);
+		self.services.outlier.add_pdu_outlier(event_id, pdu);
 	}
 }

src/service/rooms/event_handler/mod.rs

@@ -6,6 +6,7 @@
 mod handle_incoming_pdu;
 mod handle_outlier_pdu;
 mod parse_incoming_pdu;
+pub mod pdu_checks;
 mod policy_server;
 mod resolve_state;
 mod state_at_incoming;
@@ -19,7 +20,7 @@
 	DagBuilderTree, GET_MISSING_EVENTS_MAX_BATCH_SIZE, build_local_dag,
 };
 use ruma::{
-	OwnedEventId, OwnedRoomId, RoomId, events::room::create::RoomCreateEventContent,
+	OwnedEventId, OwnedRoomId, events::room::create::RoomCreateEventContent,
 	room_version_rules::RoomVersionRules,
 };
 use tokio::sync::{Notify, mpsc};
@@ -114,22 +115,6 @@ async fn event_fetch(&self, event_id: OwnedEventId) -> Option<PduEvent> {
 	}
 }
 
-fn check_room_id<Pdu: Event>(room_id: &RoomId, pdu: &Pdu) -> Result {
-	if pdu
-		.room_id()
-		.is_some_and(|claimed_room_id| claimed_room_id != room_id)
-	{
-		return Err!(Request(InvalidParam(error!(
-			pdu_event_id = %pdu.event_id(),
-			pdu_room_id = pdu.room_id().map(tracing::field::display),
-			%room_id,
-			"Found event from room in room",
-		))));
-	}
-
-	Ok(())
-}
-
 fn get_room_version_rules<Pdu: Event>(create_event: &Pdu) -> Result<RoomVersionRules> {
 	let content: RoomCreateEventContent = create_event.get_content()?;
 	let Some(room_version_rules) = content.room_version.rules() else {

src/service/rooms/event_handler/parse_incoming_pdu.rs

@@ -23,7 +23,12 @@ fn extract_room_id(event_type: &str, pdu: &CanonicalJsonObject) -> Result<OwnedR
 			.map_err(|e| err!(Request(BadJson("Invalid room_id {room_id:?} in pdu: {e}"))));
 	}
 	// If there's no room ID, and this is not a create event, it is illegal.
-	if event_type != "m.room.create" || pdu.get("state_key").is_none() {
+	let is_create = event_type == "m.room.create"
+		&& pdu
+			.get("state_key")
+			.and_then(|v| v.as_str())
+			.is_some_and(str::is_empty);
+	if !is_create {
 		return Err!(Request(BadJson("Missing room_id in pdu")));
 	}
 
@@ -42,10 +47,7 @@ fn extract_room_id(event_type: &str, pdu: &CanonicalJsonObject) -> Result<OwnedR
 		return Err!(Request(BadJson("Unknown room version in pdu")));
 	};
 
-	if !room_version_rules
-		.authorization
-		.room_create_event_id_as_room_id
-	{
+	if room_version_rules.event_format.require_room_create_room_id {
 		return Err!(Request(BadJson("Missing room_id in pdu")));
 	}
 
@@ -80,33 +82,10 @@ pub(super) fn expect_event_id_array(
 }
 
 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(())
-	}
-
-	pub async fn parse_incoming_pdu_with_known_room(
-		&self,
+	/// Parses an incoming PDU JSON object, generating an event ID for it. Does
+	/// not insert the event ID into the returned object. Does not discover the
+	/// room ID.
+	pub(super) fn parse_incoming_pdu_with_known_room(
 		pdu: &RawJsonValue,
 		room_version_rules: &RoomVersionRules,
 	) -> Result<(OwnedEventId, CanonicalJsonObject)> {
@@ -114,10 +93,13 @@ pub async fn parse_incoming_pdu_with_known_room(
 			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)?;
+		// NOTE: validation checks are now performed by `pdu_format_check_1`.
 		Ok((event_id, value))
 	}
 
+	/// Parses an incoming PDU JSON object, generating an event ID for it and
+	/// attempts to discover the associated room ID. Does not insert the event
+	/// ID into the returned object.
 	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:?}")))
@@ -142,7 +124,7 @@ pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result<Parsed> {
 			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)?;
+		// NOTE: validation checks are now performed by `pdu_format_check_1`.
 		Ok((room_id, event_id, value))
 	}
 }

src/service/rooms/event_handler/pdu_checks.rs

@@ -0,0 +1,289 @@
+use std::collections::HashMap;
+
+use conduwuit::{
+	Err, Event, EventTypeExt, PduEvent, Result, debug, debug::DebugInspect, debug_error,
+	debug_info, err, info, matrix::StateKey, state_res, trace,
+};
+use futures::future::ready;
+use ruma::{
+	CanonicalJsonObject, EventId, OwnedEventId, ServerName, api::error::ErrorKind,
+	canonical_json::redact, events::StateEventType, room_version_rules::RoomVersionRules,
+};
+
+use crate::rooms::{
+	event_handler::parse_incoming_pdu::expect_event_id_array, timeline::pdu_fits,
+};
+
+impl super::Service {
+	/// Checks that the PDU conforms to the PDU format (check 1). This is
+	/// already mostly done during deserialisation, so this function just checks
+	/// that the PDU isn't a too large.
+	pub(super) fn pdu_format_check_1(
+		pdu_json: &CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+		create_event_id: &EventId,
+	) -> Result<()> {
+		let event_format = &room_version_rules.event_format;
+		// NOTE: if we do any more validation outside of deserialisation, it has to be
+		// done here.
+
+		if !pdu_fits(pdu_json) {
+			return Err!(Request(TooLarge("PDU is too large")));
+		}
+
+		if event_format.require_room_create_room_id {
+			if pdu_json.get("room_id").is_none() {
+				return Err!(Request(BadJson("Missing required PDU field: `room_id`")));
+			}
+		}
+
+		let auth_events = expect_event_id_array(pdu_json, "auth_events")?;
+		if auth_events.len() > 10 {
+			return Err!(Request(BadJson("PDU has too many auth events")));
+		}
+
+		let create_event_in_auth_events = auth_events.iter().any(|id| id == create_event_id);
+		if !event_format.allow_room_create_in_auth_events && create_event_in_auth_events {
+			return Err!(Request(BadJson("PDU references a create event")));
+		}
+
+		let prev_events = expect_event_id_array(pdu_json, "prev_events")?;
+		if prev_events.len() > 20 {
+			return Err!(Request(BadJson("PDU has too many prev events")));
+		}
+
+		Ok(())
+	}
+
+	/// Checks that the PDU has a valid signature (check 2), and redacts it if
+	/// the content hash verification fails (check 3), returning the
+	/// potentially modified JSON. Returns an error if the PDU cannot be
+	/// redacted, or fails signature verification.
+	pub(super) async fn signature_hash_check_2_3(
+		&self,
+		pdu_json: CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<CanonicalJsonObject> {
+		match self
+			.services
+			.server_keys
+			.verify_event(&pdu_json, room_version_rules)
+			.await
+		{
+			| Ok(ruma::signatures::Verified::All) => {
+				trace!("Signatures and hashes verified successfully");
+				Ok(pdu_json)
+			},
+			| Ok(ruma::signatures::Verified::Signatures) => {
+				debug_info!("Content hash mismatch, redacting event and continuing");
+				let redacted = redact(pdu_json, &room_version_rules.redaction, None)
+					.map_err(|e| err!(Request(BadJson("Unable to redact event: {e}"))))?;
+				Ok(redacted)
+			},
+			| Err(e) => {
+				Err!(Request(Forbidden(debug_error!("Signature verification failed: {e}"))))
+			},
+		}
+	}
+
+	/// Checks PDU check 4: Passes authorisation rules based on the event's auth
+	/// events ([spec]).
+	///
+	/// If the auth check fails, false is returned, otherwise true.
+	///
+	/// [spec]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
+	pub(super) async fn auth_state_check_4(
+		&self,
+		pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+		auth_events_by_key: &HashMap<(StateEventType, StateKey), PduEvent>,
+	) -> Result<bool> {
+		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))
+		};
+
+		state_res::event_auth::auth_check(
+			room_version_rules,
+			pdu,
+			None, // TODO: third party invite
+			state_fetch,
+			create_event,
+		)
+		.await
+		.map_err(|e| err!("Event self-authentication failed: {e:?}"))
+	}
+
+	/// Checks that the event passes PDU check 5, which ensures that the event
+	/// is authorised based on the state before the event (which is the resolved
+	/// state across all prev events).
+	///
+	/// Returns a boolean indicating whether the event is authorised, and also
+	/// the resolved state before the event for later use. Returns an error if
+	/// state fetching or auth checking fails.
+	pub(super) async fn state_before_check_5(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+		origin: &ServerName,
+	) -> Result<(bool, HashMap<u64, OwnedEventId>)> {
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Resolving state at event"
+		);
+		let room_id = incoming_pdu.room_id_or_hash();
+
+		// If the incoming event only has one prev event, we can just use the state at
+		// that event, but otherwise we have to resolve across each fork. If we're
+		// missing even one of the prev events, we have to ask a remote server for help.
+		//
+		// TODO: this can be optimised by only loading auth chain events into memory,
+		// rather than the entire state.
+		let state_before = self
+			.state_before_incoming(&incoming_pdu, room_version_rules)
+			.await?;
+		let state_before = match state_before {
+			| Some(s) => s,
+			| None => {
+				trace!("Could not calculate incoming state, asking remote {origin} for it");
+				self.fetch_state(origin, create_event, &room_id, incoming_pdu.event_id())
+					.await
+					.inspect_err(|e| {
+						debug_error!("Could not fetch state from {origin}: {e}");
+					})?
+			},
+		};
+
+		if state_before.is_empty()
+			&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
+		{
+			// This can happen if the remote sends an event but cannot be reached to fetch
+			// the state at it, and all other servers in the room (which might just be the
+			// unreachable server) are unable to provide required info.
+			// returning an error here allows the upgrade to be attempted at another time.
+			return Err!(Request(Forbidden("Could not resolve incoming state before event")));
+		}
+		trace!(state_events = state_before.len(), "Calculated incoming state");
+
+		let state_fetch_state = &state_before;
+		let state_fetch = |k: StateEventType, s: StateKey| async move {
+			let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
+
+			let event_id = state_fetch_state.get(&shortstatekey)?;
+			self.services.timeline.get_pdu(event_id).await.ok()
+		};
+
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Running state-before auth check"
+		);
+
+		// PDU check: 5
+		let auth_check = state_res::event_auth::auth_check(
+			room_version_rules,
+			incoming_pdu,
+			None, // TODO: third party invite
+			|ty, sk| state_fetch(ty.clone(), sk.into()),
+			create_event.as_pdu(),
+		)
+		.await
+		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
+		Ok((auth_check, state_before))
+	}
+
+	/// Checks that the event passes PDU check 6, which ensures that the event
+	/// is authorised based on the room's current state (which is the resolved
+	/// state across all current forward extremities).
+	///
+	/// Returns a boolean indicating whether the event is authorised, or an
+	/// error if the auth check fails.
+	pub(super) async fn current_state_check_6(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		create_event: &PduEvent,
+	) -> Result<bool> {
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Gathering auth events"
+		);
+		let auth_events = self
+			.services
+			.state
+			.get_auth_events(
+				&incoming_pdu.room_id_or_hash(),
+				incoming_pdu.kind(),
+				incoming_pdu.sender(),
+				incoming_pdu.state_key(),
+				incoming_pdu.content(),
+				room_version_rules,
+			)
+			.await?;
+
+		let state_fetch = |k: &StateEventType, s: &str| {
+			let key = k.with_state_key(s);
+			ready(auth_events.get(&key).map(ToOwned::to_owned))
+		};
+
+		debug!(
+			event_id = %incoming_pdu.event_id,
+			"Running current state auth check"
+		);
+		state_res::event_auth::auth_check(
+			room_version_rules,
+			incoming_pdu,
+			None, // third-party invite
+			state_fetch,
+			create_event.as_pdu(),
+		)
+		.await
+		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))
+	}
+
+	/// Performs PDU check 7 - does the policy server allow this event.
+	///
+	/// If the policy server forbids the event, false is returned. If there is a
+	/// problem contacting the policy server, or it returns an unrecognised
+	/// response, an appropriate error is returned.
+	pub(super) async fn policy_server_check_7(
+		&self,
+		incoming_pdu: &PduEvent,
+		pdu_json: &mut CanonicalJsonObject,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<bool> {
+		let event_id = pdu_json
+			.remove("event_id")
+			.expect("event_id should be present in pdu_json at this stage");
+		if let Err(e) = self
+			.policy_server_allows_event(
+				incoming_pdu,
+				pdu_json,
+				&incoming_pdu.room_id_or_hash(),
+				room_version_rules,
+				true,
+			)
+			.await
+			.debug_inspect(|()| {
+				debug!(
+					event_id = %incoming_pdu.event_id,
+					"Event has passed policy server check."
+				);
+			}) {
+			return if matches!(e.kind(), ErrorKind::Forbidden) {
+				info!(
+					event_id = %incoming_pdu.event_id,
+					error = %e,
+					"Event has been marked as spam by policy server: {}",
+					e.message(),
+				);
+				Ok(false)
+			} else {
+				Err(e)
+			};
+		}
+		pdu_json.insert("event_id".to_owned(), event_id);
+		Ok(true)
+	}
+}

src/service/rooms/event_handler/state_at_incoming.rs

@@ -16,10 +16,31 @@
 use crate::rooms::short::ShortStateHash;
 
 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
+	/// Resolves the state before the incoming event.
+	///
+	/// If we do not know enough information to resolve the state, `Ok(None)` is
+	/// returned, and the caller will have to figure it out some other way (e.g.
+	/// by fetching the state from a remote server).
+	pub(super) async fn state_before_incoming<Pdu>(
+		&self,
+		incoming_pdu: &Pdu,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<Option<HashMap<u64, OwnedEventId>>>
+	where
+		Pdu: Event + Send + Sync,
+	{
+		if incoming_pdu.prev_events().count() == 1 {
+			self.state_before_incoming_degree_one(incoming_pdu).await
+		} else {
+			self.state_before_incoming_resolved(incoming_pdu, room_version_rules)
+				.await
+		}
+	}
+
+	/// Determines the state before the incoming pdu, when it has only one prev
+	/// event. This is a special case that does not require state resolution.
 	#[tracing::instrument(name = "state", level = "debug", skip_all)]
-	pub(super) async fn state_at_incoming_degree_one<Pdu>(
+	async fn state_before_incoming_degree_one<Pdu>(
 		&self,
 		incoming_pdu: &Pdu,
 	) -> Result<Option<HashMap<u64, OwnedEventId>>>
@@ -66,7 +87,7 @@ pub(super) async fn state_at_incoming_degree_one<Pdu>(
 				.await;
 
 			state.insert(shortstatekey, prev_event.to_owned());
-			// Now it's the state after the pdu
+			// Now it's the state at the pdu
 		}
 
 		debug_assert!(!state.is_empty(), "should be returning None for empty HashMap result");
@@ -74,11 +95,14 @@ pub(super) async fn state_at_incoming_degree_one<Pdu>(
 		Ok(Some(state))
 	}
 
+	/// Resolves the state before the incoming pdu across all of its prev
+	/// events. If we do not know enough information to resolve the state,
+	/// `Ok(None)` is returned, and the caller will have to figure it out some
+	/// other way (e.g. by fetching the state from a remote server).
 	#[tracing::instrument(name = "state", level = "debug", skip_all)]
-	pub(super) async fn state_at_incoming_resolved<Pdu>(
+	async fn state_before_incoming_resolved<Pdu>(
 		&self,
 		incoming_pdu: &Pdu,
-		room_id: &RoomId,
 		room_version_rules: &RoomVersionRules,
 	) -> Result<Option<HashMap<u64, OwnedEventId>>>
 	where
@@ -108,6 +132,7 @@ pub(super) async fn state_at_incoming_resolved<Pdu>(
 		};
 
 		trace!("Calculating fork states...");
+		let room_id = &incoming_pdu.room_id_or_hash();
 		let (fork_states, auth_chain_sets): (Vec<StateMap<_>>, Vec<HashSet<_>>) =
 			extremity_sstatehashes
 				.into_iter()
@@ -145,6 +170,8 @@ pub(super) async fn state_at_incoming_resolved<Pdu>(
 			.await
 	}
 
+	/// Determines the state at an incoming fork (aka the state at a prev
+	/// event), returning the resolved state and its associated auth chain.
 	async fn state_at_incoming_fork<Pdu>(
 		&self,
 		room_id: &RoomId,

src/service/rooms/event_handler/upgrade_outlier_pdu.rs

@@ -1,27 +1,15 @@
-use std::{borrow::Borrow, sync::Arc, time::Instant};
+use std::time::Instant;
 
 use conduwuit::{
-	Err, Result, debug, debug_error, debug_info, err, info, is_equal_to,
-	matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
-	result::DebugInspect,
+	Err, Result, debug, debug_info, debug_warn, is_true,
+	matrix::{Event, PduEvent},
 	trace,
-	utils::{
-		IterStream,
-		stream::{BroadbandExt, ReadyExt},
-	},
-	warn,
-};
-use futures::{FutureExt, StreamExt, future::ready};
-use ruma::{
-	CanonicalJsonObject, RoomId, ServerName, api::error::ErrorKind, events::StateEventType,
 };
+use ruma::{CanonicalJsonObject, RoomId, ServerName, events::StateEventType};
 use tokio::join;
 
 use super::get_room_version_rules;
-use crate::rooms::{
-	state_compressor::{CompressedState, HashSetCompressStateEvent},
-	timeline::RawPduId,
-};
+use crate::rooms::timeline::RawPduId;
 
 impl super::Service {
 	#[tracing::instrument(name="upgrade_outlier", skip_all, fields(event_id=%incoming_pdu.event_id()))]
@@ -46,16 +34,24 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 			trace!(event_id=%incoming_pdu.event_id(), "Skipping upgrade of already upgraded PDU");
 			return Ok(Some(id));
 		} else if rejected {
-			return Err!(Request(Forbidden("Event has been rejected")));
+			return Err!(Request(Forbidden(debug_info!("Event has been rejected"))));
 		} else if soft_failed {
-			return Err!(Request(Forbidden("Event has been soft-failed")));
+			// Soft-failed events cannot be promoted.
+			return Err!(Request(Forbidden(debug_info!("Event has been soft-failed"))));
 		}
 
+		// These should never happen, but they're good last-minute sanity checks to
+		// ensure we never promote totally illegal events.
 		assert_eq!(
 			*create_event.kind(),
 			StateEventType::RoomCreate.into(),
 			"tried to upgrade a PDU with a create_event that is not a room create event"
 		);
+		assert_eq!(
+			incoming_pdu.room_id_or_hash(),
+			*room_id,
+			"room ID mismatch: PDU room ID differs from parameter"
+		);
 
 		debug!(
 			event_id = %incoming_pdu.event_id,
@@ -65,326 +61,118 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 		let min_depth = self.services.metadata.get_mindepth(room_id).await;
 		let room_version_rules = get_room_version_rules(create_event)?;
 
-		// 10. Fetch missing state and auth chain events by calling /state_ids at
-		//     backwards extremities doing all the checks in this list starting at 1.
-		//     These are not timeline events.
-
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Resolving state at event"
-		);
-		let state_at_incoming_event = if incoming_pdu.prev_events().count() == 1 {
-			self.state_at_incoming_degree_one(&incoming_pdu).await?
-		} else {
-			self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_rules)
-				.await?
-		};
-		let state_at_incoming_event = match state_at_incoming_event {
-			| Some(s) => s,
-			| None => {
-				trace!("Could not calculate incoming state, asking remote {origin} for it");
-				self.fetch_state(origin, create_event, room_id, incoming_pdu.event_id())
-					.await
-					.debug_inspect_err(|e| {
-						debug_error!("Could not fetch state from {origin}: {e}");
-					})?
-			},
-		};
-
-		if state_at_incoming_event.is_empty()
-			&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
-		{
-			// This can happen if the remote sends an event but cannot be reached to fetch
-			// the state at it, and all other servers in the room (which might just be the
-			// unreachable server) are unable to provide required info.
-			// returning an error here allows the upgrade to be attempted at another time.
-			return Err!(Request(Forbidden("Could not resolve incoming state at event")));
-		}
-		trace!(state_events = state_at_incoming_event.len(), "Calculated incoming state");
-
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Performing auth check to upgrade"
-		);
-		// 11. Check the auth of the event passes based on the state of the event
-		let state_fetch_state = &state_at_incoming_event;
-		let state_fetch = |k: StateEventType, s: StateKey| async move {
-			let shortstatekey = self.services.short.get_shortstatekey(&k, &s).await.ok()?;
-
-			let event_id = state_fetch_state.get(&shortstatekey)?;
-			self.services.timeline.get_pdu(event_id).await.ok()
-		};
-
-		debug!(
-			event_id = %incoming_pdu.event_id,
-			"Running initial auth check"
-		);
-		// PDU check: 5
-		let auth_check = state_res::event_auth::auth_check(
-			&room_version_rules,
-			&incoming_pdu,
-			None, // TODO: third party invite
-			|ty, sk| state_fetch(ty.clone(), sk.into()),
-			create_event.as_pdu(),
-		)
-		.await
-		.map_err(|e| err!(Request(Forbidden("Auth check failed: {e:?}"))))?;
-
-		if !auth_check {
-			self.services
-				.pdu_metadata
-				.mark_event_rejected(incoming_pdu.event_id());
-			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,
-			)
+		// We now need to resolve the state before the event so that we can perform PDU
+		// check 5 (event auth passes based on state before the event). To do this, we
+		// either need to have all the prev events locally, or ask a remote server
+		// for the state at the event.
+		let (passes_state_before, state_before) = self
+			.state_before_check_5(&incoming_pdu, &room_version_rules, create_event, origin)
 			.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"
-			);
+		if !passes_state_before {
+			self.reject_and_persist(incoming_pdu.event_id(), &val);
+			return Err!(Request(Forbidden(debug_warn!(
+				"Event authorisation fails based on the state before the event"
+			))));
 		}
 
-		// 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
-				}
-			},
-		};
+		// Now that we know the event passes both self-authentication, and
+		// authentication based on the state before the event, we need to check that it
+		// passes based on the *current* room state (state across all forward
+		// extremities). If it doesn't, we accept it, but soft-fail it, and this
+		// prevents it being promoted.
 
-		// 13. Use state resolution to find new room state
-		// We start looking at current room state now, so lets lock the room
+		// We lock the room here to prevent the current state from changing beneath us
+		// mid-check.
 		trace!(
 			room_id = %room_id,
 			"Locking the room"
 		);
 		let state_lock = self.services.state.mutex.lock(room_id).await;
-
-		let 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
-			{
-				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(),
+		let passes_current_state = self
+			.current_state_check_6(&incoming_pdu, &room_version_rules, create_event)
+			.await
+			.inspect(|passes| {
+				if !*passes {
+					debug_warn!(
+						"Event authorisation fails based on the current room state - will be \
+						 soft-failed"
 					);
-					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);
-			}
+			})?;
 
-			// Additionally, if this is a redaction for a soft-failed event, we soft-fail it
-			// also.
+		// Determine whether this PDU should be soft-failed.
+		// If the auth check failed, invariably yes. Otherwise, only if the user isn't
+		// allowed to redact the target event (if any).
+		let mut should_soft_fail =
+			match (passes_current_state, incoming_pdu.redacts_id(&room_version_rules)) {
+				| (false, _) => true,
+				| (true, None) => false,
+				| (true, Some(redact_id)) => self
+					.services
+					.state_accessor
+					.user_can_redact(&redact_id, incoming_pdu.sender(), room_id, true)
+					.await
+					.is_ok_and(is_true!()),
+			};
 
-			// 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 !should_soft_fail {
+			// Now we can perform check 7, which is ensuring the event passes policy server
+			// checks.
+			// We explicitly only do this if we aren't already going to soft-fail the event,
+			// since the policy server refusing this event also soft-fails it.
+			debug!(event_id = %incoming_pdu.event_id, "Checking policy server for event");
+			should_soft_fail = !self
+				.policy_server_check_7(&incoming_pdu, &mut val, &room_version_rules)
+				.await
+				.inspect(|passes| {
+					if !*passes {
+						debug_warn!(
+							"Event did not pass the policy server check and will be soft-failed"
+						);
+					}
+				})?;
+
+			// TODO: this is supposed to hide redactions from policy servers and janitorial
+			// bots, however, for full efficacy it also needs to hide redactions for
+			// unknown events. This needs to be investigated at a later time.
 			if let Some(redact_id) = incoming_pdu.redacts_id(&room_version_rules) {
 				debug!(
 					redact_id = %redact_id,
-					"Checking if redaction is for a soft-failed event"
+					"Checking if redaction is for a soft-failed/rejected event"
 				);
-				if self
+				if !self
 					.services
 					.pdu_metadata
-					.is_event_soft_failed(&redact_id)
+					.is_event_accepted(&redact_id)
 					.await
 				{
-					info!(
-						redact_id = %redact_id,
-						"Redaction is for a soft-failed event"
+					debug_info!(
+						"Soft-failing valid redaction because it targets a non-accepted event"
 					);
-					soft_fail = true;
+					should_soft_fail = true;
 				}
 			}
 		}
 
+		// The PDU has now passed all checks! We can now promote it (or soft-fail it if
+		// the verdict is such).
 		trace!("Appending pdu to timeline");
-		let mut extremities: Vec<_> = self
-			.services
-			.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<_>>()
-				.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,
+				&room_version_rules,
+				state_before,
+				should_soft_fail,
 				&state_lock,
-				room_id,
 			)
 			.await?;
-		if soft_fail {
-			self.services
-				.pdu_metadata
-				.mark_event_soft_failed(incoming_pdu.event_id());
 
-			info!(
+		if should_soft_fail {
+			debug_info!(
 				elapsed = ?timer.elapsed(),
 				event_id = %incoming_pdu.event_id,
 				"Event was soft failed"

src/service/rooms/membership/mod.rs

@@ -520,7 +520,7 @@ pub async fn join_remote_room(
 						return state;
 					},
 				};
-				if !pdu_fits(&mut value.clone()) {
+				if !pdu_fits(&value.clone()) {
 					warn!(
 						"dropping incoming PDU {event_id} in room {room_id} from room join \
 						 because it exceeds 65535 bytes or is otherwise too large."

src/service/rooms/timeline/append.rs

@@ -1,69 +1,113 @@
 use std::{
-	collections::{BTreeMap, HashSet},
+	borrow::Borrow,
+	collections::{BTreeMap, HashMap, HashSet},
 	sync::Arc,
 };
 
 use conduwuit::{
-	debug_warn,
+	Result, debug, debug_warn, is_equal_to,
+	matrix::{Event, PduEvent},
 	pdu::{Count, ShortRoomId},
 	trace,
-	utils::{IterStream, TryFutureExtExt, stream::BroadbandExt},
+	utils::{
+		IterStream, TryFutureExtExt,
+		mutex_map::Guard,
+		stream::{BroadbandExt, ReadyExt},
+	},
 	warn,
 };
 use conduwuit_core::{
-	Result, err, error,
-	matrix::{
-		event::Event,
-		pdu::{PduCount, PduEvent, PduId, RawPduId},
-	},
-	utils::{self, ReadyExt},
+	err, error,
+	matrix::pdu::{PduCount, PduId, RawPduId},
+	utils::{self},
 };
-use futures::{StreamExt, TryFutureExt};
+use futures::{FutureExt, StreamExt, TryFutureExt};
 use ruma::{
-	CanonicalJsonObject, CanonicalJsonValue, EventId, RoomId, RoomVersionId, UserId,
+	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, RoomId,
+	RoomVersionId, UserId,
 	events::{
 		GlobalAccountDataEventType, TimelineEventType,
 		push_rules::PushRulesEvent,
 		room::{encrypted::Relation, redaction::RoomRedactionEventContent},
 	},
 	push::{Action, Ruleset, Tweak},
+	room_version_rules::RoomVersionRules,
 };
 
 use super::{ExtractBody, ExtractRelatesTo, ExtractRelatesToEventId, RoomMutexGuard};
-use crate::{appservice::RegistrationInfo, rooms::state_compressor::CompressedState};
+use crate::{
+	appservice::RegistrationInfo,
+	rooms::state_compressor::{CompressedState, HashSetCompressStateEvent},
+};
 
 impl super::Service {
 	/// Append the incoming event setting the state snapshot to the state from
 	/// the server that sent the event.
-	#[allow(clippy::too_many_arguments)]
-	pub async fn append_incoming_pdu<'a, Leaves>(
+	pub async fn append_incoming_pdu<'a>(
 		&'a self,
 		pdu: &'a PduEvent,
 		pdu_json: CanonicalJsonObject,
-		new_room_leaves: Leaves,
-		state_ids_compressed: Arc<CompressedState>,
-		soft_fail: bool,
+		room_version_rules: &RoomVersionRules,
+		state_before: HashMap<u64, OwnedEventId>,
+		should_soft_fail: bool,
 		state_lock: &'a RoomMutexGuard,
-		room_id: &'a RoomId,
-	) -> Result<Option<RawPduId>>
-	where
-		Leaves: Iterator<Item = &'a EventId> + Send + 'a,
-	{
+	) -> Result<Option<RawPduId>> {
+		let room_id = pdu.room_id_or_hash();
+
+		let sb = state_before.iter().map(|(ssk, eid)| (ssk, eid.as_ref()));
+		let state_ids_compressed: Arc<CompressedState> = self
+			.services
+			.state_compressor
+			.compress_state_events(sb)
+			.collect()
+			.map(Arc::new)
+			.await;
+
 		// We append to state before appending the pdu, so we don't have a moment in
 		// time with the pdu without it's state. This is okay because append_pdu can't
 		// fail.
 		self.services
 			.state
-			.set_event_state(&pdu.event_id, room_id, state_ids_compressed)
+			.set_event_state(&pdu.event_id, &room_id, state_ids_compressed)
 			.await?;
 
-		if soft_fail {
+		if should_soft_fail {
 			// Nothing else to do with a soft-failed event.
+			self.services
+				.pdu_metadata
+				.mark_event_soft_failed(pdu.event_id());
 			return Ok(None);
 		}
 
+		// Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events
+		// are not added as forward extremities.
+		// This also means we set the state here.
+		// We do this BEFORE setting the extremities so that there's never a point in
+		// time where we have fresh extremities referencing stale state.
+		let mut extremities: Vec<_> = self
+			.services
+			.state
+			.get_forward_extremities(&room_id)
+			.collect()
+			.await;
+		extremities = self
+			.progress_state_and_extremities(
+				pdu,
+				room_version_rules,
+				state_before.clone(),
+				extremities,
+				state_lock,
+			)
+			.await?;
+
 		let pdu_id = self
-			.append_pdu(pdu, pdu_json, new_room_leaves, state_lock, room_id)
+			.append_pdu(
+				pdu,
+				pdu_json,
+				extremities.iter().map(Borrow::borrow),
+				state_lock,
+				&room_id,
+			)
 			.await?;
 
 		// Process admin commands for federation events
@@ -86,11 +130,110 @@ pub async fn append_incoming_pdu<'a, Leaves>(
 			}
 		}
 
-		self.services.sync.wake_all_joined(room_id).await;
+		self.services.sync.wake_all_joined(&room_id).await;
 
 		Ok(Some(pdu_id))
 	}
 
+	/// Derives new room state from the incoming event and filters forward
+	/// extremities accordingly. Does not set forward extremities.
+	///
+	/// Only call this function if the incoming PDU is not soft-failed or
+	/// rejected.
+	async fn progress_state_and_extremities(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		state_before: HashMap<u64, OwnedEventId>,
+		forward_extremities: Vec<OwnedEventId>,
+		state_lock: &Guard<OwnedRoomId, ()>,
+	) -> Result<Vec<OwnedEventId>> {
+		if incoming_pdu.state_key().is_some() {
+			debug!("Event is a state-event. Deriving new room state");
+			self.derive_new_state(incoming_pdu, room_version_rules, state_before, state_lock)
+				.await?;
+		}
+
+		// Now we calculate the set of extremities this room has after the incoming
+		// event has been applied. We start with the previous extremities
+		trace!("Calculating extremities");
+		let mut forward_extremities = forward_extremities
+			.into_iter()
+			.stream()
+			.ready_filter(|event_id| {
+				// Remove any that are referenced by this incoming event's prev_events
+				!incoming_pdu.prev_events().any(is_equal_to!(event_id))
+			})
+			.broad_filter_map(|event_id| async move {
+				// Only keep those extremities were not referenced yet
+				self.services
+					.pdu_metadata
+					.is_event_referenced(&incoming_pdu.room_id_or_hash(), &event_id)
+					.await
+					.eq(&false)
+					.then_some(event_id)
+			})
+			.collect::<Vec<_>>()
+			.await;
+		forward_extremities.push(incoming_pdu.event_id().to_owned());
+		debug!(
+			"Retained {} extremities checked against {} prev_events",
+			forward_extremities.len(),
+			incoming_pdu.prev_events().count()
+		);
+		assert!(!forward_extremities.is_empty(), "resolved extremities cannot be empty");
+		Ok(forward_extremities)
+	}
+
+	/// Derives a new room state by adding the incoming PDU to the state before
+	/// it to create the state at, which then becomes the current room state.
+	///
+	/// The caller MUST ensure forward extremities are set appropriately,
+	/// including this incoming pdu, either before or after calling this
+	/// function. Failing to do so will result in an inconsistent current state
+	/// cache, which may affect event authentication.
+	async fn derive_new_state(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		state_before: HashMap<u64, OwnedEventId>,
+		state_lock: &Guard<OwnedRoomId, ()>,
+	) -> Result {
+		let room_id = incoming_pdu.room_id_or_hash();
+		// We also add state after incoming event to the fork states
+		let mut state_at_incoming_event = state_before;
+		let shortstatekey = self
+			.services
+			.short
+			.get_or_create_shortstatekey(
+				&incoming_pdu.kind().to_string().into(),
+				incoming_pdu.state_key().unwrap(),
+			)
+			.await;
+
+		let event_id = incoming_pdu.event_id();
+		state_at_incoming_event.insert(shortstatekey, event_id.to_owned());
+
+		debug!("Resolving new room state");
+		let new_room_state = self
+			.services
+			.event_handler
+			.resolve_state(&room_id, room_version_rules, state_at_incoming_event)
+			.await?;
+
+		debug!("Forcing new room state");
+		let HashSetCompressStateEvent { shortstatehash, added, removed } = self
+			.services
+			.state_compressor
+			.save_state(&room_id, new_room_state)
+			.await?;
+
+		self.services
+			.state
+			.force_state(&room_id, shortstatehash, added, removed, state_lock)
+			.await
+	}
+
 	/// Populates the unsigned data of a PDU
 	async fn populate_unsigned(&self, pdu: &PduEvent, pdu_json: &mut CanonicalJsonObject) {
 		let Some(state_key) = pdu.state_key() else {

src/service/rooms/timeline/backfill.rs

@@ -143,7 +143,7 @@ pub async fn get_remote_pdu(&self, room_id: &RoomId, event_id: &EventId) -> Resu
 				| Ok(value) => {
 					self.services
 						.event_handler
-						.handle_incoming_pdu(&backfill_server, room_id, event_id, value, false)
+						.handle_incoming_pdu(&backfill_server, room_id, event_id, value, true)
 						.boxed()
 						.await?;
 					debug!("Successfully backfilled {event_id} from {backfill_server}");
@@ -185,7 +185,7 @@ pub async fn backfill_pdu(&self, origin: &ServerName, pdu: Box<RawJsonValue>) ->
 
 		self.services
 			.event_handler
-			.handle_incoming_pdu(origin, &room_id, &event_id, value, false)
+			.handle_incoming_pdu(origin, &room_id, &event_id, value, true)
 			.boxed()
 			.await?;
 

src/service/rooms/timeline/create.rs

@@ -22,7 +22,9 @@
 
 use super::RoomMutexGuard;
 
-pub fn pdu_fits(owned_obj: &mut CanonicalJsonObject) -> bool {
+/// Ensures the given PDU fits inside the size limits for a PDU.
+#[must_use]
+pub fn pdu_fits(owned_obj: &CanonicalJsonObject) -> bool {
 	// room IDs, event IDs, senders, types, and state keys must all be <= 255 bytes
 	if let Some(CanonicalJsonValue::String(room_id)) = owned_obj.get("room_id") {
 		if room_id.len() > 255 {
@@ -311,7 +313,7 @@ pub async fn create_hash_and_sign_event(
 		pdu_json
 			.insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into()));
 		// Verify that the *full* PDU isn't over 64KiB.
-		if !pdu_fits(&mut pdu_json.clone()) {
+		if !pdu_fits(&pdu_json.clone()) {
 			// feckin huge PDU mate
 			return Err!(Request(TooLarge("Message/PDU is too long (exceeds 65535 bytes)")));
 		}

src/service/rooms/timeline/mod.rs

@@ -77,6 +77,7 @@ struct Services {
 	short: Dep<rooms::short::Service>,
 	state: Dep<rooms::state::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
+	state_compressor: Dep<rooms::state_compressor::Service>,
 	state_cache: Dep<rooms::state_cache::Service>,
 	sync: Dep<sync::Service>,
 	threads: Dep<rooms::threads::Service>,
@@ -110,6 +111,8 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				state: args.depend::<rooms::state::Service>("rooms::state"),
 				state_accessor: args
 					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
+				state_compressor: args
+					.depend::<rooms::state_compressor::Service>("rooms::state_compressor"),
 				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
 				sync: args.depend::<sync::Service>("sync"),
 				threads: args.depend::<rooms::threads::Service>("rooms::threads"),

src/service/sending/appservice.rs

@@ -6,7 +6,8 @@
 };
 use ruma::api::{
 	IncomingResponse, OutgoingRequest,
-	appservice::{HomeserverToken, Registration},
+	appservice::Registration,
+	auth_scheme::{AccessToken, SendAccessToken},
 	path_builder::SinglePath,
 };
 
@@ -21,9 +22,7 @@ pub async fn send_appservice_request<T>(
 		request: T,
 	) -> Result<Option<T::IncomingResponse>>
 	where
-		T: OutgoingRequest<Authentication = HomeserverToken, PathBuilder = SinglePath>
-			+ Debug
-			+ Send,
+		T: OutgoingRequest<Authentication = AccessToken, PathBuilder = SinglePath> + Debug + Send,
 	{
 		let Some(dest) = registration.url else {
 			return Ok(None);
@@ -37,7 +36,7 @@ pub async fn send_appservice_request<T>(
 
 		let hs_token = registration.hs_token.as_str();
 		let mut http_request = request
-			.try_into_http_request::<BytesMut>(&dest, hs_token, ())
+			.try_into_http_request::<BytesMut>(&dest, SendAccessToken::Appservice(hs_token), ())
 			.map_err(|e| {
 				err!(BadServerResponse(
 					warn!(appservice = %registration.id, "Failed to find destination {dest}: {e:?}")

src/service/users/account.rs

@@ -197,7 +197,7 @@ pub async fn create_local_account(
 		// register, suspend them.
 		if !was_first_user && self.services.config.suspend_on_register {
 			// Note that we can still do auto joins for suspended users
-			self.suspend_account(user_id, Some(&self.services.globals.server_user))
+			self.suspend_account(user_id, &self.services.globals.server_user)
 				.await;
 
 			// And send an @room notice to the admin room, to prompt admins to review the
@@ -389,13 +389,13 @@ pub async fn deactivate_account(&self, user_id: &UserId) -> Result<()> {
 	}
 
 	/// Suspend account, placing it in a read-only state
-	pub async fn suspend_account(&self, user_id: &UserId, suspending_user: Option<&UserId>) {
+	pub async fn suspend_account(&self, user_id: &UserId, suspending_user: &UserId) {
 		self.db.userid_suspension.raw_put(
 			user_id,
 			Json(UserSuspension {
 				suspended: true,
 				suspended_at: MilliSecondsSinceUnixEpoch::now().get().into(),
-				suspended_by: suspending_user.map(ToString::to_string),
+				suspended_by: suspending_user.to_string(),
 			}),
 		);
 	}
@@ -406,7 +406,7 @@ pub async fn unsuspend_account(&self, user_id: &UserId) {
 	}
 
 	/// Locks an account, preventing it being used until it is unlocked.
-	pub async fn lock_account(&self, user_id: &UserId, locking_user: Option<&UserId>) {
+	pub async fn lock_account(&self, user_id: &UserId, locking_user: &UserId) {
 		// NOTE: Locking is basically just suspension with a more severe effect,
 		// so we'll just re-use the suspension data structure to store the lock state.
 		let suspension = self
@@ -418,7 +418,7 @@ pub async fn lock_account(&self, user_id: &UserId, locking_user: Option<&UserId>
 			.unwrap_or_else(|_| UserSuspension {
 				suspended: true,
 				suspended_at: MilliSecondsSinceUnixEpoch::now().get().into(),
-				suspended_by: locking_user.map(ToString::to_string),
+				suspended_by: locking_user.to_string(),
 			});
 
 		self.db.userid_lock.raw_put(user_id, Json(suspension));

src/service/users/mod.rs

@@ -31,7 +31,7 @@ pub struct UserSuspension {
 	/// When the user was suspended (Unix timestamp in milliseconds)
 	pub suspended_at: u64,
 	/// User ID of who suspended this user
-	pub suspended_by: Option<String>,
+	pub suspended_by: String,
 }
 
 /// A password hash. This is only for use when setting a user's password,

src/web/pages/components/mod.rs

@@ -3,11 +3,12 @@
 use askama::{Template, filters::HtmlSafe};
 use base64::Engine;
 use conduwuit_core::{result::FlatOk, utils};
-use conduwuit_service::{Services, media::mxc::Mxc, oauth::client_metadata::ClientMetadata};
-use ruma::{
-	OwnedDeviceId, OwnedUserId, UserId,
-	api::{OAuthClientScope, client::device::Device},
+use conduwuit_service::{
+	Services,
+	media::mxc::Mxc,
+	oauth::{client_metadata::ClientMetadata, grant::Scope},
 };
+use ruma::{OwnedDeviceId, OwnedUserId, UserId, api::client::device::Device};
 
 pub(super) mod form;
 
@@ -182,7 +183,7 @@ pub(super) async fn for_device(
 #[derive(Debug, Template)]
 #[template(path = "_components/client_scopes.html.j2")]
 pub(super) struct ClientScopes {
-	pub scopes: BTreeSet<OAuthClientScope>,
+	pub scopes: BTreeSet<Scope>,
 }
 
 impl HtmlSafe for ClientScopes {}

src/web/pages/oauth/grant.rs

@@ -4,7 +4,7 @@
 	response::Redirect,
 	routing::on,
 };
-use conduwuit_service::oauth::grant::{AuthorizationCodeQuery, Prompt, RequestedScope};
+use conduwuit_service::oauth::grant::{AuthorizationCodeQuery, Prompt};
 use ruma::OwnedUserId;
 use url::Url;
 
@@ -100,13 +100,7 @@ async fn route_authorization_code(
 		return Err(WebError::BadRequest("Invalid client ID".to_owned()));
 	};
 
-	let scopes = query
-		.scope
-		.to_scopes()
-		.map_err(WebError::BadRequest)?
-		.iter()
-		.filter_map(RequestedScope::as_granted_scope)
-		.collect();
+	let scopes = query.scope.to_scopes().map_err(WebError::BadRequest)?;
 
 	let client_name = if let Some(name) = &client.client_name {
 		name

src/web/pages/oidc/complete.rs

@@ -6,7 +6,7 @@
 	response::Redirect,
 	routing::on,
 };
-use conduwuit_service::{oauth::grant::AuthorizationCodeData, oidc::SessionCompletionStatus};
+use conduwuit_service::{oauth::grant::AuthorizationCodeResponse, oidc::SessionCompletionStatus};
 use futures::FutureExt;
 use ruma::{OwnedServerName, UserId};
 use serde::{Deserialize, de::IgnoredAny};
@@ -57,7 +57,7 @@ struct LoginForm {
 async fn route_complete(
 	State(services): State<crate::State>,
 	Extension(context): Extension<TemplateContext>,
-	Expect(Query(query)): Expect<Query<AuthorizationCodeData>>,
+	Expect(Query(query)): Expect<Query<AuthorizationCodeResponse>>,
 	session_store: Session,
 	user: User<true>,
 	PostForm(form): PostForm<LoginForm>,
@@ -107,7 +107,7 @@ async fn route_complete(
 			let supplied_user_id = if let Some(form) = form {
 				if let Ok(user_id) = UserId::parse(format!(
 					"@{}:{}",
-					form.username,
+					&form.username,
 					services.globals.server_name()
 				)) && services.users.status(&user_id).await.is_active()
 				{

src/web/pages/templates/_components/client_scopes.html.j2

@@ -1,13 +1,10 @@
 <ul>
     {% for scope in scopes %}
         {% match scope %}
-            {% when OAuthClientScope::ApiFullAccess %}
+            {% when Scope::ClientApi %}
                 <li>Send messages and interact with chatrooms on your behalf</li>
-            {% when OAuthClientScope::ServerAdministration %}
-                <li>⚠️ Administrate this homeserver
-                <br><em class="negative">This is a dangerous permission. Make sure you trust this app.</em></li>
-            {% when _ %}
-                <li>missingno</li>
+            {% when Scope::Device(_) %}
+                <li>Access your Matrix account</li>
         {% endmatch %}
     {% endfor %}
 </ul>