diff --git a/src/admin/context.rs b/src/admin/context.rs index 270537be..3d3cffb7 100644 --- a/src/admin/context.rs +++ b/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>>, } @@ -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()) + } } diff --git a/src/admin/processor.rs b/src/admin/processor.rs index f7b7140f..8d1fe89c 100644 --- a/src/admin/processor.rs +++ b/src/admin/processor.rs @@ -63,6 +63,7 @@ async fn process_command(services: Arc, 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(), }; diff --git a/src/admin/user/commands.rs b/src/admin/user/commands.rs index e5e481e5..d094fc5f 100644 --- a/src/admin/user/commands.rs +++ b/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) -> Result { let user_id = parse_local_user_id(self.services, &username)?; diff --git a/src/admin/user/mod.rs b/src/admin/user/mod.rs index e789376a..645d3637 100644 --- a/src/admin/user/mod.rs +++ b/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, diff --git a/src/api/client/alias.rs b/src/api/client/alias.rs index 9f1b05f8..dc7aad44 100644 --- a/src/api/client/alias.rs +++ b/src/api/client/alias.rs @@ -18,6 +18,9 @@ pub(crate) async fn create_alias_route( body: Ruma, ) -> Result { 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, ) -> Result { 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 diff --git a/src/api/client/directory.rs b/src/api/client/directory.rs index aa6ae168..2e219fd9 100644 --- a/src/api/client/directory.rs +++ b/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 diff --git a/src/api/client/media.rs b/src/api/client/media.rs index 94572413..11d5450c 100644 --- a/src/api/client/media.rs +++ b/src/api/client/media.rs @@ -52,6 +52,9 @@ pub(crate) async fn create_content_route( body: Ruma, ) -> Result { 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(); diff --git a/src/api/client/membership.rs b/src/api/client/membership.rs index 145b3cde..e6392533 100644 --- a/src/api/client/membership.rs +++ b/src/api/client/membership.rs @@ -178,6 +178,9 @@ pub(crate) async fn join_room_by_id_route( body: Ruma, ) -> Result { 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 { 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, ) -> Result { 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, body: Ruma, ) -> Result { + 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, body: Ruma, ) -> Result { + 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, ) diff --git a/src/api/client/profile.rs b/src/api/client/profile.rs index e2d1c934..bdba4078 100644 --- a/src/api/client/profile.rs +++ b/src/api/client/profile.rs @@ -36,6 +36,9 @@ pub(crate) async fn set_displayname_route( body: Ruma, ) -> Result { 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, ) -> Result { 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"))); diff --git a/src/api/client/read_marker.rs b/src/api/client/read_marker.rs index fbfc8fea..e152869c 100644 --- a/src/api/client/read_marker.rs +++ b/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([( - event.to_owned(), - BTreeMap::from_iter([( - ReceiptType::Read, - BTreeMap::from_iter([(sender_user.to_owned(), ruma::events::receipt::Receipt { - ts: Some(MilliSecondsSinceUnixEpoch::now()), - thread: ReceiptThread::Unthreaded, - })]), - )]), - )]); + if !services.users.is_suspended(sender_user).await? { + let receipt_content = BTreeMap::from_iter([( + event.to_owned(), + BTreeMap::from_iter([( + ReceiptType::Read, + BTreeMap::from_iter([( + sender_user.to_owned(), + ruma::events::receipt::Receipt { + ts: Some(MilliSecondsSinceUnixEpoch::now()), + thread: ReceiptThread::Unthreaded, + }, + )]), + )]), + )]); - services - .rooms - .read_receipt - .readreceipt_update( - sender_user, - &body.room_id, - &ruma::events::receipt::ReceiptEvent { - content: ruma::events::receipt::ReceiptEventContent(receipt_content), - room_id: body.room_id.clone(), - }, - ) - .await; + services + .rooms + .read_receipt + .readreceipt_update( + sender_user, + &body.room_id, + &ruma::events::receipt::ReceiptEvent { + content: ruma::events::receipt::ReceiptEventContent(receipt_content), + room_id: body.room_id.clone(), + }, + ) + .await; + } } if let Some(event) = &body.private_read_receipt { diff --git a/src/api/client/redact.rs b/src/api/client/redact.rs index 8dbe47a6..a8eaf91d 100644 --- a/src/api/client/redact.rs +++ b/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 { 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; diff --git a/src/api/client/room/create.rs b/src/api/client/room/create.rs index be3fd23b..d1dffc51 100644 --- a/src/api/client/room/create.rs +++ b/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), diff --git a/src/api/client/room/upgrade.rs b/src/api/client/room/upgrade.rs index da5b49fe..d8f5ea83 100644 --- a/src/api/client/room/upgrade.rs +++ b/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()); diff --git a/src/api/client/send.rs b/src/api/client/send.rs index f753fa65..b87d1822 100644 --- a/src/api/client/send.rs +++ b/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 diff --git a/src/api/client/state.rs b/src/api/client/state.rs index 2ddc8f14..07802b1b 100644 --- a/src/api/client/state.rs +++ b/src/api/client/state.rs @@ -33,6 +33,10 @@ pub(crate) async fn send_state_event_for_key_route( ) -> Result { 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, diff --git a/src/api/client/typing.rs b/src/api/client/typing.rs index 1d8d02fd..7b0df538 100644 --- a/src/api/client/typing.rs +++ b/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."))); } - - match body.state { - | Typing::Yes(duration) => { - let duration = utils::clamp( - duration.as_millis().try_into().unwrap_or(u64::MAX), + if !services.users.is_suspended(sender_user).await? { + match body.state { + | Typing::Yes(duration) => { + let duration = utils::clamp( + 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 - .config - .typing_client_timeout_min_s - .try_mul(1000)?, + .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 - .server - .config - .typing_client_timeout_max_s - .try_mul(1000)?, - ); - 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?; - }, + .rooms + .typing + .typing_remove(sender_user, &body.room_id) + .await?; + }, + } } // ping presence diff --git a/src/database/maps.rs b/src/database/maps.rs index 19f9ced4..214dbf34 100644 --- a/src/database/maps.rs +++ b/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 diff --git a/src/service/admin/mod.rs b/src/service/admin/mod.rs index 683f5400..86e12c3c 100644 --- a/src/service/admin/mod.rs +++ b/src/service/admin/mod.rs @@ -45,11 +45,13 @@ struct Services { services: StdRwLock>>, } -/// 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, + pub sender: Option>, } /// 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) -> 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, + sender: Box, + ) -> 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, ) -> ProcessorResult { - self.process_command(CommandInput { command, reply_id }) + self.process_command(CommandInput { command, reply_id, sender: None }) .await } diff --git a/src/service/rooms/timeline/mod.rs b/src/service/rooms/timeline/mod.rs index 37963246..534d8faf 100644 --- a/src/service/rooms/timeline/mod.rs +++ b/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 - .admin - .command(body, Some((*pdu.event_id).into()))?; + self.services.admin.command_with_sender( + body, + Some((*pdu.event_id).into()), + pdu.sender.clone().into(), + )?; } } }, diff --git a/src/service/users/mod.rs b/src/service/users/mod.rs index 701561a8..d2dfccd9 100644 --- a/src/service/users/mod.rs +++ b/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, userid_masterkeyid: Arc, userid_password: Arc, + userid_suspension: Arc, userid_selfsigningkeyid: Arc, userid_usersigningkeyid: Arc, useridprofilekey_value: Arc, @@ -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 { + match self + .db + .userid_suspension + .get(user_id) + .await + .deserialized::() + { + | 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)