style: Remove redundant, unused functions

oops

src/admin/context.rs

@@ -7,13 +7,14 @@ use futures::{
 	io::{AsyncWriteExt, BufWriter},
 	lock::Mutex,
 };
-use ruma::EventId;
+use ruma::{EventId, UserId};
 
 pub(crate) struct Context<'a> {
 	pub(crate) services: &'a Services,
 	pub(crate) body: &'a [&'a str],
 	pub(crate) timer: SystemTime,
 	pub(crate) reply_id: Option<&'a EventId>,
+	pub(crate) sender: Option<&'a UserId>,
 	pub(crate) output: Mutex<BufWriter<Vec<u8>>>,
 }
 
@@ -36,4 +37,10 @@ impl Context<'_> {
 			output.write_all(s.as_bytes()).map_err(Into::into).await
 		})
 	}
+
+	/// Get the sender as a string, or service user ID if not available
+	pub(crate) fn sender_or_service_user(&self) -> &UserId {
+		self.sender
+			.unwrap_or_else(|| self.services.globals.server_user.as_ref())
+	}
 }

src/admin/processor.rs

@@ -63,6 +63,7 @@ async fn process_command(services: Arc<Services>, input: &CommandInput) -> Proce
 		body: &body,
 		timer: SystemTime::now(),
 		reply_id: input.reply_id.as_deref(),
+		sender: input.sender.as_deref(),
 		output: BufWriter::new(Vec::new()).into(),
 	};
 

src/admin/user/commands.rs

@@ -224,6 +224,47 @@ pub(super) async fn deactivate(&self, no_leave_rooms: bool, user_id: String) ->
 		.await
 }
 
+#[admin_command]
+pub(super) async fn suspend(&self, user_id: String) -> Result {
+	let user_id = parse_local_user_id(self.services, &user_id)?;
+
+	if user_id == self.services.globals.server_user {
+		return Err!("Not allowed to suspend the server service account.",);
+	}
+
+	if !self.services.users.exists(&user_id).await {
+		return Err!("User {user_id} does not exist.");
+	}
+	if self.services.users.is_admin(&user_id).await {
+		return Err!("Admin users cannot be suspended.");
+	}
+	// TODO: Record the actual user that sent the suspension where possible
+	self.services
+		.users
+		.suspend_account(&user_id, self.sender_or_service_user())
+		.await;
+
+	self.write_str(&format!("User {user_id} has been suspended."))
+		.await
+}
+
+#[admin_command]
+pub(super) async fn unsuspend(&self, user_id: String) -> Result {
+	let user_id = parse_local_user_id(self.services, &user_id)?;
+
+	if user_id == self.services.globals.server_user {
+		return Err!("Not allowed to unsuspend the server service account.",);
+	}
+
+	if !self.services.users.exists(&user_id).await {
+		return Err!("User {user_id} does not exist.");
+	}
+	self.services.users.unsuspend_account(&user_id).await;
+
+	self.write_str(&format!("User {user_id} has been unsuspended."))
+		.await
+}
+
 #[admin_command]
 pub(super) async fn reset_password(&self, username: String, password: Option<String>) -> Result {
 	let user_id = parse_local_user_id(self.services, &username)?;

src/admin/user/mod.rs

@@ -59,6 +59,28 @@ pub(super) enum UserCommand {
 		force: bool,
 	},
 
+	/// - Suspend a user
+	///
+	/// Suspended users are able to log in, sync, and read messages, but are not
+	/// able to send events nor redact them, cannot change their profile, and
+	/// are unable to join, invite to, or knock on rooms.
+	///
+	/// Suspended users can still leave rooms and deactivate their account.
+	/// Suspending them effectively makes them read-only.
+	Suspend {
+		/// Username of the user to suspend
+		user_id: String,
+	},
+
+	/// - Unsuspend a user
+	///
+	/// Reverses the effects of the `suspend` command, allowing the user to send
+	/// messages, change their profile, create room invites, etc.
+	Unsuspend {
+		/// Username of the user to unsuspend
+		user_id: String,
+	},
+
 	/// - List local users in the database
 	#[clap(alias = "list")]
 	ListUsers,

src/api/client/alias.rs

@@ -18,6 +18,9 @@ pub(crate) async fn create_alias_route(
 	body: Ruma<create_alias::v3::Request>,
 ) -> Result<create_alias::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	services
 		.rooms
@@ -63,6 +66,9 @@ pub(crate) async fn delete_alias_route(
 	body: Ruma<delete_alias::v3::Request>,
 ) -> Result<delete_alias::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	services
 		.rooms

src/api/client/directory.rs

@@ -128,6 +128,9 @@ pub(crate) async fn set_room_visibility_route(
 		// Return 404 if the room doesn't exist
 		return Err!(Request(NotFound("Room not found")));
 	}
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	if services
 		.users

src/api/client/media.rs

@@ -52,6 +52,9 @@ pub(crate) async fn create_content_route(
 	body: Ruma<create_content::v3::Request>,
 ) -> Result<create_content::v3::Response> {
 	let user = body.sender_user.as_ref().expect("user is authenticated");
+	if services.users.is_suspended(user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	let filename = body.filename.as_deref();
 	let content_type = body.content_type.as_deref();

src/api/client/membership.rs

@@ -178,6 +178,9 @@ pub(crate) async fn join_room_by_id_route(
 	body: Ruma<join_room_by_id::v3::Request>,
 ) -> Result<join_room_by_id::v3::Response> {
 	let sender_user = body.sender_user();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	banned_room_check(
 		&services,
@@ -249,6 +252,9 @@ pub(crate) async fn join_room_by_id_or_alias_route(
 	let sender_user = body.sender_user.as_deref().expect("user is authenticated");
 	let appservice_info = &body.appservice_info;
 	let body = body.body;
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	let (servers, room_id) = match OwnedRoomId::try_from(body.room_id_or_alias) {
 		| Ok(room_id) => {
@@ -369,6 +375,9 @@ pub(crate) async fn knock_room_route(
 ) -> Result<knock_room::v3::Response> {
 	let sender_user = body.sender_user();
 	let body = &body.body;
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	let (servers, room_id) = match OwnedRoomId::try_from(body.room_id_or_alias.clone()) {
 		| Ok(room_id) => {
@@ -492,6 +501,9 @@ pub(crate) async fn invite_user_route(
 	body: Ruma<invite_user::v3::Request>,
 ) -> Result<invite_user::v3::Response> {
 	let sender_user = body.sender_user();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	if !services.users.is_admin(sender_user).await && services.config.block_non_admin_invites {
 		debug_error!(
@@ -566,6 +578,10 @@ pub(crate) async fn kick_user_route(
 	State(services): State<crate::State>,
 	body: Ruma<kick_user::v3::Request>,
 ) -> Result<kick_user::v3::Response> {
+	let sender_user = body.sender_user();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 	let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
 
 	let Ok(event) = services
@@ -601,7 +617,7 @@ pub(crate) async fn kick_user_route(
 				third_party_invite: None,
 				..event
 			}),
-			body.sender_user(),
+			sender_user,
 			&body.room_id,
 			&state_lock,
 		)
@@ -625,6 +641,10 @@ pub(crate) async fn ban_user_route(
 		return Err!(Request(Forbidden("You cannot ban yourself.")));
 	}
 
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
+
 	let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
 
 	let current_member_content = services
@@ -667,6 +687,10 @@ pub(crate) async fn unban_user_route(
 	State(services): State<crate::State>,
 	body: Ruma<unban_user::v3::Request>,
 ) -> Result<unban_user::v3::Response> {
+	let sender_user = body.sender_user();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 	let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
 
 	let current_member_content = services
@@ -695,7 +719,7 @@ pub(crate) async fn unban_user_route(
 				is_direct: None,
 				..current_member_content
 			}),
-			body.sender_user(),
+			sender_user,
 			&body.room_id,
 			&state_lock,
 		)

src/api/client/profile.rs

@@ -36,6 +36,9 @@ pub(crate) async fn set_displayname_route(
 	body: Ruma<set_display_name::v3::Request>,
 ) -> Result<set_display_name::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	if *sender_user != body.user_id && body.appservice_info.is_none() {
 		return Err!(Request(Forbidden("You cannot update the profile of another user")));
@@ -125,6 +128,9 @@ pub(crate) async fn set_avatar_url_route(
 	body: Ruma<set_avatar_url::v3::Request>,
 ) -> Result<set_avatar_url::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	if *sender_user != body.user_id && body.appservice_info.is_none() {
 		return Err!(Request(Forbidden("You cannot update the profile of another user")));

src/api/client/read_marker.rs

@@ -58,29 +58,34 @@ pub(crate) async fn set_read_marker_route(
 	}
 
 	if let Some(event) = &body.read_receipt {
-		let receipt_content = BTreeMap::from_iter([(
+		if !services.users.is_suspended(sender_user).await? {
-			event.to_owned(),
+			let receipt_content = BTreeMap::from_iter([(
-			BTreeMap::from_iter([(
+				event.to_owned(),
-				ReceiptType::Read,
+				BTreeMap::from_iter([(
-				BTreeMap::from_iter([(sender_user.to_owned(), ruma::events::receipt::Receipt {
+					ReceiptType::Read,
-					ts: Some(MilliSecondsSinceUnixEpoch::now()),
+					BTreeMap::from_iter([(
-					thread: ReceiptThread::Unthreaded,
+						sender_user.to_owned(),
-				})]),
+						ruma::events::receipt::Receipt {
-			)]),
+							ts: Some(MilliSecondsSinceUnixEpoch::now()),
-		)]);
+							thread: ReceiptThread::Unthreaded,
+						},
+					)]),
+				)]),
+			)]);
 
-		services
+			services
-			.rooms
+				.rooms
-			.read_receipt
+				.read_receipt
-			.readreceipt_update(
+				.readreceipt_update(
-				sender_user,
+					sender_user,
-				&body.room_id,
+					&body.room_id,
-				&ruma::events::receipt::ReceiptEvent {
+					&ruma::events::receipt::ReceiptEvent {
-					content: ruma::events::receipt::ReceiptEventContent(receipt_content),
+						content: ruma::events::receipt::ReceiptEventContent(receipt_content),
-					room_id: body.room_id.clone(),
+						room_id: body.room_id.clone(),
-				},
+					},
-			)
+				)
-			.await;
+				.await;
+		}
 	}
 
 	if let Some(event) = &body.private_read_receipt {

src/api/client/redact.rs

@@ -1,5 +1,5 @@
 use axum::extract::State;
-use conduwuit::{Result, matrix::pdu::PduBuilder};
+use conduwuit::{Err, Result, matrix::pdu::PduBuilder};
 use ruma::{
 	api::client::redact::redact_event, events::room::redaction::RoomRedactionEventContent,
 };
@@ -17,6 +17,10 @@ pub(crate) async fn redact_event_route(
 ) -> Result<redact_event::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 	let body = body.body;
+	if services.users.is_suspended(sender_user).await? {
+		// TODO: Users can redact their own messages while suspended
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
 

src/api/client/room/create.rs

@@ -70,6 +70,10 @@ pub(crate) async fn create_room_route(
 		));
 	}
 
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
+
 	let room_id: OwnedRoomId = match &body.room_id {
 		| Some(custom_room_id) => custom_room_id_check(&services, custom_room_id)?,
 		| _ => RoomId::new(&services.server.name),

src/api/client/room/upgrade.rs

@@ -2,7 +2,7 @@ use std::cmp::max;
 
 use axum::extract::State;
 use conduwuit::{
-	Error, Result, err, info,
+	Err, Error, Result, err, info,
 	matrix::{StateKey, pdu::PduBuilder},
 };
 use futures::StreamExt;
@@ -63,6 +63,10 @@ pub(crate) async fn upgrade_room_route(
 		));
 	}
 
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
+
 	// Create a replacement room
 	let replacement_room = RoomId::new(services.globals.server_name());
 

src/api/client/send.rs

@@ -23,6 +23,9 @@ pub(crate) async fn send_message_event_route(
 	let sender_user = body.sender_user();
 	let sender_device = body.sender_device.as_deref();
 	let appservice_info = body.appservice_info.as_ref();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	// Forbid m.room.encrypted if encryption is disabled
 	if MessageLikeEventType::RoomEncrypted == body.event_type && !services.config.allow_encryption

src/api/client/state.rs

@@ -33,6 +33,10 @@ pub(crate) async fn send_state_event_for_key_route(
 ) -> Result<send_state_event::v3::Response> {
 	let sender_user = body.sender_user();
 
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
+
 	Ok(send_state_event::v3::Response {
 		event_id: send_state_event_for_key_helper(
 			&services,

src/api/client/typing.rs

@@ -26,41 +26,42 @@ pub(crate) async fn create_typing_event_route(
 	{
 		return Err!(Request(Forbidden("You are not in this room.")));
 	}
-
+	if !services.users.is_suspended(sender_user).await? {
-	match body.state {
+		match body.state {
-		| Typing::Yes(duration) => {
+			| Typing::Yes(duration) => {
-			let duration = utils::clamp(
+				let duration = utils::clamp(
-				duration.as_millis().try_into().unwrap_or(u64::MAX),
+					duration.as_millis().try_into().unwrap_or(u64::MAX),
+					services
+						.server
+						.config
+						.typing_client_timeout_min_s
+						.try_mul(1000)?,
+					services
+						.server
+						.config
+						.typing_client_timeout_max_s
+						.try_mul(1000)?,
+				);
 				services
-					.server
+					.rooms
-					.config
+					.typing
-					.typing_client_timeout_min_s
+					.typing_add(
-					.try_mul(1000)?,
+						sender_user,
+						&body.room_id,
+						utils::millis_since_unix_epoch()
+							.checked_add(duration)
+							.expect("user typing timeout should not get this high"),
+					)
+					.await?;
+			},
+			| _ => {
 				services
-					.server
+					.rooms
-					.config
+					.typing
-					.typing_client_timeout_max_s
+					.typing_remove(sender_user, &body.room_id)
-					.try_mul(1000)?,
+					.await?;
-			);
+			},
-			services
+		}
-				.rooms
-				.typing
-				.typing_add(
-					sender_user,
-					&body.room_id,
-					utils::millis_since_unix_epoch()
-						.checked_add(duration)
-						.expect("user typing timeout should not get this high"),
-				)
-				.await?;
-		},
-		| _ => {
-			services
-				.rooms
-				.typing
-				.typing_remove(sender_user, &body.room_id)
-				.await?;
-		},
 	}
 
 	// ping presence

src/database/maps.rs

@@ -378,6 +378,10 @@ pub(super) static MAPS: &[Descriptor] = &[
 		name: "userid_password",
 		..descriptor::RANDOM
 	},
+	Descriptor {
+		name: "userid_suspension",
+		..descriptor::RANDOM_SMALL
+	},
 	Descriptor {
 		name: "userid_presenceid",
 		..descriptor::RANDOM_SMALL

src/service/admin/mod.rs

@@ -45,11 +45,13 @@ struct Services {
 	services: StdRwLock<Option<Weak<crate::Services>>>,
 }
 
-/// Inputs to a command are a multi-line string and optional reply_id.
+/// Inputs to a command are a multi-line string, optional reply_id, and optional
+/// sender.
 #[derive(Debug)]
 pub struct CommandInput {
 	pub command: String,
 	pub reply_id: Option<OwnedEventId>,
+	pub sender: Option<Box<UserId>>,
 }
 
 /// Prototype of the tab-completer. The input is buffered text when tab
@@ -162,7 +164,22 @@ impl Service {
 	pub fn command(&self, command: String, reply_id: Option<OwnedEventId>) -> Result<()> {
 		self.channel
 			.0
-			.send(CommandInput { command, reply_id })
+			.send(CommandInput { command, reply_id, sender: None })
+			.map_err(|e| err!("Failed to enqueue admin command: {e:?}"))
+	}
+
+	/// Posts a command to the command processor queue with sender information
+	/// and returns. Processing will take place on the service worker's task
+	/// asynchronously. Errors if the queue is full.
+	pub fn command_with_sender(
+		&self,
+		command: String,
+		reply_id: Option<OwnedEventId>,
+		sender: Box<UserId>,
+	) -> Result<()> {
+		self.channel
+			.0
+			.send(CommandInput { command, reply_id, sender: Some(sender) })
 			.map_err(|e| err!("Failed to enqueue admin command: {e:?}"))
 	}
 
@@ -173,7 +190,7 @@ impl Service {
 		command: String,
 		reply_id: Option<OwnedEventId>,
 	) -> ProcessorResult {
-		self.process_command(CommandInput { command, reply_id })
+		self.process_command(CommandInput { command, reply_id, sender: None })
 			.await
 	}
 

src/service/rooms/timeline/mod.rs

@@ -536,9 +536,11 @@ impl Service {
 					self.services.search.index_pdu(shortroomid, &pdu_id, &body);
 
 					if self.services.admin.is_admin_command(pdu, &body).await {
-						self.services
+						self.services.admin.command_with_sender(
-							.admin
+							body,
-							.command(body, Some((*pdu.event_id).into()))?;
+							Some((*pdu.event_id).into()),
+							pdu.sender.clone().into(),
+						)?;
 					}
 				}
 			},

src/service/users/mod.rs

@@ -16,10 +16,21 @@ use ruma::{
 	},
 	serde::Raw,
 };
+use serde::{Deserialize, Serialize};
 use serde_json::json;
 
 use crate::{Dep, account_data, admin, globals, rooms};
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct UserSuspension {
+	/// Whether the user is currently suspended
+	pub suspended: bool,
+	/// When the user was suspended (Unix timestamp in milliseconds)
+	pub suspended_at: u64,
+	/// User ID of who suspended this user
+	pub suspended_by: String,
+}
+
 pub struct Service {
 	services: Services,
 	db: Data,
@@ -52,6 +63,7 @@ struct Data {
 	userid_lastonetimekeyupdate: Arc<Map>,
 	userid_masterkeyid: Arc<Map>,
 	userid_password: Arc<Map>,
+	userid_suspension: Arc<Map>,
 	userid_selfsigningkeyid: Arc<Map>,
 	userid_usersigningkeyid: Arc<Map>,
 	useridprofilekey_value: Arc<Map>,
@@ -87,6 +99,7 @@ impl crate::Service for Service {
 				userid_lastonetimekeyupdate: args.db["userid_lastonetimekeyupdate"].clone(),
 				userid_masterkeyid: args.db["userid_masterkeyid"].clone(),
 				userid_password: args.db["userid_password"].clone(),
+				userid_suspension: args.db["userid_suspension"].clone(),
 				userid_selfsigningkeyid: args.db["userid_selfsigningkeyid"].clone(),
 				userid_usersigningkeyid: args.db["userid_usersigningkeyid"].clone(),
 				useridprofilekey_value: args.db["useridprofilekey_value"].clone(),
@@ -143,6 +156,23 @@ impl Service {
 		Ok(())
 	}
 
+	/// Suspend account, placing it in a read-only state
+	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.to_string(),
+			}),
+		);
+	}
+
+	/// Unsuspend account, placing it in a read-write state
+	pub async fn unsuspend_account(&self, user_id: &UserId) {
+		self.db.userid_suspension.remove(user_id);
+	}
+
 	/// Check if a user has an account on this homeserver.
 	#[inline]
 	pub async fn exists(&self, user_id: &UserId) -> bool {
@@ -159,6 +189,25 @@ impl Service {
 			.await
 	}
 
+	/// Check if account is suspended
+	pub async fn is_suspended(&self, user_id: &UserId) -> Result<bool> {
+		match self
+			.db
+			.userid_suspension
+			.get(user_id)
+			.await
+			.deserialized::<UserSuspension>()
+		{
+			| Ok(s) => Ok(s.suspended),
+			| Err(e) =>
+				if e.is_not_found() {
+					Ok(false)
+				} else {
+					Err(e)
+				},
+		}
+	}
+
 	/// Check if account is active, infallible
 	pub async fn is_active(&self, user_id: &UserId) -> bool {
 		!self.is_deactivated(user_id).await.unwrap_or(true)