diff --git a/src/admin/check/commands.rs b/src/admin/check/commands.rs index 14ef4f78f..db17bb109 100644 --- a/src/admin/check/commands.rs +++ b/src/admin/check/commands.rs @@ -6,7 +6,12 @@ use crate::Context; impl Context<'_> { pub(super) async fn check_all_users(&self) -> Result { let timer = tokio::time::Instant::now(); - let users = self.services.users.stream().collect::>().await; + let users = self + .services + .users + .stream_local_users() + .collect::>() + .await; let query_time = timer.elapsed(); let total = users.len(); diff --git a/src/admin/debug/commands.rs b/src/admin/debug/commands.rs index 7d792a428..de17073bd 100644 --- a/src/admin/debug/commands.rs +++ b/src/admin/debug/commands.rs @@ -612,7 +612,7 @@ impl crate::Context<'_> { // Force E2EE device list updates for all users self.services .users - .stream() + .stream_local_users() .for_each(async |user_id| self.services.users.mark_device_key_update(&user_id).await) .await; diff --git a/src/admin/query/users.rs b/src/admin/query/users.rs index 13fe9a888..7449f0689 100644 --- a/src/admin/query/users.rs +++ b/src/admin/query/users.rs @@ -191,8 +191,13 @@ impl crate::Context<'_> { async fn iter_users(&self) -> Result { let timer = tokio::time::Instant::now(); - let result: Vec = - self.services.users.stream().map(Into::into).collect().await; + let result: Vec = self + .services + .users + .stream_local_users() + .map(Into::into) + .collect() + .await; let query_time = timer.elapsed(); @@ -202,7 +207,7 @@ impl crate::Context<'_> { async fn iter_users2(&self) -> Result { let timer = tokio::time::Instant::now(); - let result: Vec<_> = self.services.users.stream().collect().await; + let result: Vec<_> = self.services.users.stream_local_users().collect().await; let result: Vec<_> = result .into_iter() .map(|user_id| String::from_utf8_lossy(user_id.as_bytes()).into_owned()) diff --git a/src/admin/user/commands.rs b/src/admin/user/commands.rs index 26dadc5b5..a9ffce7ff 100644 --- a/src/admin/user/commands.rs +++ b/src/admin/user/commands.rs @@ -20,7 +20,7 @@ use ruma::{ tag::{TagEvent, TagEventContent, TagInfo}, }, }; -use service::users::HashedPassword; +use service::users::{AccountStatus, HashedPassword}; use crate::{ get_room_info, @@ -60,7 +60,7 @@ impl crate::Context<'_> { self.services .users .create_local_account(&user_id, HashedPassword::new(password)?, None) - .await; + .await?; self.write_str(&format!("Created user {user_id} with password `{password}`")) .await @@ -103,15 +103,12 @@ impl crate::Context<'_> { pub(super) async fn suspend(&self, user_id: String) -> Result { self.bail_restricted()?; - let user_id = parse_local_user_id(self.services, &user_id)?; + let user_id = parse_active_local_user_id(self.services, &user_id).await?; 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."); } @@ -127,15 +124,12 @@ impl crate::Context<'_> { pub(super) async fn unsuspend(&self, user_id: String) -> Result { self.bail_restricted()?; - let user_id = parse_local_user_id(self.services, &user_id)?; + let user_id = parse_active_local_user_id(self.services, &user_id).await?; 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.")) @@ -147,6 +141,7 @@ impl crate::Context<'_> { logout: bool, username: String, password: Option, + convert_to_local_account: bool, ) -> Result { let user_id = parse_local_user_id(self.services, &username)?; @@ -159,15 +154,37 @@ impl crate::Context<'_> { let new_password = password.unwrap_or_else(|| utils::random_string(AUTO_GEN_PASSWORD_LENGTH)); + let new_password_hash = HashedPassword::new(&new_password)?; - self.services - .users - .set_password(&user_id, Some(HashedPassword::new(&new_password)?)); + if convert_to_local_account { + self.services + .users + .convert_to_local_account(&user_id, new_password_hash) + .await?; + } else { + match self.services.users.status(&user_id).await { + | AccountStatus::Active if !self.services.users.is_shadow(&user_id).await => { + self.services + .users + .set_password(&user_id, new_password_hash) + .await?; + }, + | AccountStatus::NotFound => { + return Err!("The provided user does not exist."); + }, + | _ => { + return Err!( + "The provided user is a shadow or deactivated account. To convert it to \ + a local account, pass the --convert-to-local-account flag." + ); + }, + } - self.write_str(&format!( - "Successfully reset the password for user {user_id}: `{new_password}`" - )) - .await?; + self.write_str(&format!( + "Successfully reset the password for user {user_id}: `{new_password}`" + )) + .await?; + } if logout { self.services @@ -919,21 +936,16 @@ impl crate::Context<'_> { pub(super) async fn lock(&self, user_id: String) -> Result { self.bail_restricted()?; - let user_id = parse_local_user_id(self.services, &user_id)?; - assert!( - self.services.globals.user_is_local(&user_id), - "Parsed user_id must be a local user" - ); + let user_id = parse_active_local_user_id(self.services, &user_id).await?; + if user_id == self.services.globals.server_user { return Err!("Not allowed to lock 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 locked."); } + self.services .users .lock_account(&user_id, self.sender_or_service_user()) @@ -945,11 +957,8 @@ impl crate::Context<'_> { pub(super) async fn unlock(&self, user_id: String) -> Result { self.bail_restricted()?; - let user_id = parse_local_user_id(self.services, &user_id)?; - assert!( - self.services.globals.user_is_local(&user_id), - "Parsed user_id must be a local user" - ); + let user_id = parse_active_local_user_id(self.services, &user_id).await?; + self.services.users.unlock_account(&user_id).await; self.write_str(&format!("User {user_id} has been unlocked.")) @@ -958,21 +967,16 @@ impl crate::Context<'_> { pub(super) async fn logout(&self, user_id: String) -> Result { self.bail_restricted()?; - let user_id = parse_local_user_id(self.services, &user_id)?; - assert!( - self.services.globals.user_is_local(&user_id), - "Parsed user_id must be a local user" - ); + let user_id = parse_active_local_user_id(self.services, &user_id).await?; + if user_id == self.services.globals.server_user { return Err!("Not allowed to log out 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!("You cannot forcefully log out admin users."); } + self.services .users .all_device_ids(&user_id) @@ -989,18 +993,12 @@ impl crate::Context<'_> { pub(super) async fn disable_login(&self, user_id: String) -> Result { self.bail_restricted()?; - let user_id = parse_local_user_id(self.services, &user_id)?; - assert!( - self.services.globals.user_is_local(&user_id), - "Parsed user_id must be a local user" - ); + let user_id = parse_active_local_user_id(self.services, &user_id).await?; + if user_id == self.services.globals.server_user { return Err!("Not allowed to disable login for 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 have their login disallowed."); } @@ -1014,14 +1012,8 @@ impl crate::Context<'_> { pub(super) async fn enable_login(&self, user_id: String) -> Result { self.bail_restricted()?; - let user_id = parse_local_user_id(self.services, &user_id)?; - assert!( - self.services.globals.user_is_local(&user_id), - "Parsed user_id must be a local user" - ); - if !self.services.users.exists(&user_id).await { - return Err!("User {user_id} does not exist."); - } + let user_id = parse_active_local_user_id(self.services, &user_id).await?; + self.services.users.enable_login(&user_id); self.write_str(&format!("{user_id} can now log in.")).await @@ -1129,10 +1121,8 @@ impl crate::Context<'_> { } pub(super) async fn reset_push_rules(&self, user_id: String) -> Result { - let user_id = parse_local_user_id(self.services, &user_id)?; - if !self.services.users.is_active(&user_id).await { - return Err!("User is not active."); - } + let user_id = parse_active_local_user_id(self.services, &user_id).await?; + recreate_push_rules_and_return(self.services, &user_id).await?; self.write_str("Reset user's push rules to the server default.") .await diff --git a/src/admin/user/mod.rs b/src/admin/user/mod.rs index b2d05d650..c5278505a 100644 --- a/src/admin/user/mod.rs +++ b/src/admin/user/mod.rs @@ -27,6 +27,9 @@ pub enum UserCommand { username: String, /// New password for the user, if unspecified one is generated password: Option, + + #[arg(long)] + convert_to_local_account: bool, }, /// Get a user's associated email address. diff --git a/src/admin/utils.rs b/src/admin/utils.rs index 3df7d844a..f39985417 100644 --- a/src/admin/utils.rs +++ b/src/admin/utils.rs @@ -54,14 +54,7 @@ pub(crate) async fn parse_active_local_user_id( user_id: &str, ) -> Result { let user_id = parse_local_user_id(services, user_id)?; - - if !services.users.exists(&user_id).await { - return Err!("User {user_id:?} does not exist on this server."); - } - - if services.users.is_deactivated(&user_id).await? { - return Err!("User {user_id:?} is deactivated."); - } + services.users.status(&user_id).await.ensure_active()?; Ok(user_id) } diff --git a/src/api/client/account/mod.rs b/src/api/client/account/mod.rs index 807ff240a..686f24fa9 100644 --- a/src/api/client/account/mod.rs +++ b/src/api/client/account/mod.rs @@ -128,7 +128,8 @@ pub(crate) async fn change_password_route( services .users - .set_password(&sender_user, Some(HashedPassword::new(&body.new_password)?)); + .set_password(&sender_user, HashedPassword::new(&body.new_password)?) + .await?; if body.logout_devices { // Logout all devices except the current one diff --git a/src/api/client/account/register.rs b/src/api/client/account/register.rs index 8f8ddf740..d41cc6a96 100644 --- a/src/api/client/account/register.rs +++ b/src/api/client/account/register.rs @@ -72,7 +72,7 @@ pub(crate) async fn register_route( .determine_registration_user_id(body.username.clone(), None, Some(appservice_info)) .await?; - services.users.create(&user_id, None)?; + services.users.create_shadow_account(&user_id).await?; user_id } else { @@ -98,7 +98,7 @@ pub(crate) async fn register_route( services .users .create_local_account(&user_id, password, identity.email) - .await; + .await?; user_id }; diff --git a/src/api/client/admin/lock.rs b/src/api/client/admin/lock.rs index f748ef1ba..81b95dd61 100644 --- a/src/api/client/admin/lock.rs +++ b/src/api/client/admin/lock.rs @@ -12,20 +12,18 @@ pub(crate) async fn get_lock_status( State(services): State, body: Ruma, ) -> Result { - let (admin, active) = join( + let (admin, status) = join( services.users.is_admin(body.identity.expect_sender_user()?), - services.users.is_active(&body.user_id), + services.users.status(&body.user_id), ) .await; + if !admin { return Err!(Request(Forbidden("Only server administrators can use this endpoint"))); } - if !services.globals.user_is_local(&body.user_id) { - return Err!(Request(InvalidParam("Can only check the lock status of local users"))); - } - if !active { - return Err!(Request(NotFound("Unknown user"))); - } + + status.ensure_active()?; + Ok(is_user_locked::v1::Response::new( services.users.is_locked(&body.user_id).await?, )) @@ -40,9 +38,9 @@ pub(crate) async fn put_lock_status( ) -> Result { let sender_user = body.identity.expect_sender_user()?; - let (sender_admin, active, target_admin) = join3( + let (sender_admin, status, target_admin) = join3( services.users.is_admin(sender_user), - services.users.is_active(&body.user_id), + services.users.status(&body.user_id), services.users.is_admin(&body.user_id), ) .await; @@ -50,18 +48,17 @@ pub(crate) async fn put_lock_status( if !sender_admin { return Err!(Request(Forbidden("Only server administrators can use this endpoint"))); } - if !services.globals.user_is_local(&body.user_id) { - return Err!(Request(InvalidParam("Can only set the lock status of local users"))); - } - if !active { - return Err!(Request(NotFound("Unknown user"))); - } + + status.ensure_active()?; + if body.user_id == *sender_user { return Err!(Request(Forbidden("You cannot lock yourself"))); } + if target_admin { return Err!(Request(Forbidden("You cannot lock another server administrator"))); } + if services.users.is_locked(&body.user_id).await? == body.locked { // No change return Ok(lock_user::v1::Response::new(body.locked)); diff --git a/src/api/client/admin/suspend.rs b/src/api/client/admin/suspend.rs index cf1beca8d..fac32f12e 100644 --- a/src/api/client/admin/suspend.rs +++ b/src/api/client/admin/suspend.rs @@ -12,20 +12,18 @@ pub(crate) async fn get_suspended_status( State(services): State, body: Ruma, ) -> Result { - let (admin, active) = join( + let (admin, status) = join( services.users.is_admin(body.identity.expect_sender_user()?), - services.users.is_active(&body.user_id), + services.users.status(&body.user_id), ) .await; + if !admin { return Err!(Request(Forbidden("Only server administrators can use this endpoint"))); } - if !services.globals.user_is_local(&body.user_id) { - return Err!(Request(InvalidParam("Can only check the suspended status of local users"))); - } - if !active { - return Err!(Request(NotFound("Unknown user"))); - } + + status.ensure_active()?; + Ok(is_user_suspended::v1::Response::new( services.users.is_suspended(&body.user_id).await?, )) @@ -40,9 +38,9 @@ pub(crate) async fn put_suspended_status( ) -> Result { let sender_user = body.identity.expect_sender_user()?; - let (sender_admin, active, target_admin) = join3( + let (sender_admin, status, target_admin) = join3( services.users.is_admin(sender_user), - services.users.is_active(&body.user_id), + services.users.status(&body.user_id), services.users.is_admin(&body.user_id), ) .await; @@ -50,18 +48,17 @@ pub(crate) async fn put_suspended_status( if !sender_admin { return Err!(Request(Forbidden("Only server administrators can use this endpoint"))); } - if !services.globals.user_is_local(&body.user_id) { - return Err!(Request(InvalidParam("Can only set the suspended status of local users"))); - } - if !active { - return Err!(Request(NotFound("Unknown user"))); - } + + status.ensure_active()?; + if body.user_id == *sender_user { return Err!(Request(Forbidden("You cannot suspend yourself"))); } + if target_admin { return Err!(Request(Forbidden("You cannot suspend another server administrator"))); } + if services.users.is_suspended(&body.user_id).await? == body.suspended { // No change return Ok(suspend_user::v1::Response::new(body.suspended)); diff --git a/src/api/client/profile.rs b/src/api/client/profile.rs index 7dd3e6371..b611e31de 100644 --- a/src/api/client/profile.rs +++ b/src/api/client/profile.rs @@ -110,7 +110,7 @@ async fn fetch_full_profile( user_id: &UserId, ) -> Option> { // If the user exists locally, fetch their local profile - if services.users.exists(user_id).await { + if services.users.status(user_id).await.is_found() { return Some(get_local_profile(services, user_id).await); } diff --git a/src/api/client/report.rs b/src/api/client/report.rs index a7567c113..a7acd9790 100644 --- a/src/api/client/report.rs +++ b/src/api/client/report.rs @@ -149,7 +149,7 @@ pub(crate) async fn report_user_route( delay_response().await; - if !services.users.is_active_local(&body.user_id).await { + if !services.users.status(&body.user_id).await.is_found() { // return 200 as to not reveal if the user exists. Recommended by spec. return Ok(report_user::v3::Response::new()); } diff --git a/src/api/client/session.rs b/src/api/client/session.rs index dd95a1c2c..c7af45fe5 100644 --- a/src/api/client/session.rs +++ b/src/api/client/session.rs @@ -88,14 +88,6 @@ pub async fn handle_login( UserId::parse_with_server_name(user_id_or_localpart, &services.config.server_name) .map_err(|_| err!(Request(InvalidUsername("User ID is malformed"))))?; - if !services.globals.user_is_local(&user_id) { - return Err!(Request(InvalidParam("User ID does not belong to this homeserver"))); - } - - if services.users.is_deactivated(&user_id).await? { - return Err!(Request(UserDeactivated("This account has been deactivated."))); - } - if services.users.is_locked(&user_id).await? { return Err!(Request(UserLocked("This account has been locked."))); } diff --git a/src/api/client/user_directory.rs b/src/api/client/user_directory.rs index 18c02163b..7d8a9a9c4 100644 --- a/src/api/client/user_directory.rs +++ b/src/api/client/user_directory.rs @@ -32,22 +32,26 @@ pub(crate) async fn search_users_route( .min(LIMIT_MAX); let search_term = body.search_term.to_lowercase(); - let mut users = services.users.stream().broad_filter_map(async |user_id| { - let display_name = services.users.displayname(&user_id).await.ok(); - let user_id_matches = user_id.as_str().to_lowercase().contains(&search_term); + let mut users = services + .users + .stream_local_users() + .chain(services.users.stream_remote_users()) + .broad_filter_map(async |user_id| { + let display_name = services.users.displayname(&user_id).await.ok(); - let display_name_matches = display_name - .as_deref() - .map(str::to_lowercase) - .is_some_and(|display_name| display_name.contains(&search_term)); + let user_id_matches = user_id.as_str().to_lowercase().contains(&search_term); - if !user_id_matches && !display_name_matches { - return None; - } + let display_name_matches = display_name + .as_deref() + .map(str::to_lowercase) + .is_some_and(|display_name| display_name.contains(&search_term)); - let user_in_public_room = - services + if !user_id_matches && !display_name_matches { + return None; + } + + let user_in_public_room = services .rooms .state_cache .rooms_joined(&user_id) @@ -60,22 +64,22 @@ pub(crate) async fn search_users_route( .await }); - let user_sees_user = services - .rooms - .state_cache - .user_sees_user(sender_user, &user_id); + let user_sees_user = services + .rooms + .state_cache + .user_sees_user(sender_user, &user_id); - pin_mut!(user_in_public_room, user_sees_user); + pin_mut!(user_in_public_room, user_sees_user); - if user_in_public_room.or(user_sees_user).await { - Some(assign!(search_users::v3::User::new(user_id.clone()), { - display_name, - avatar_url: services.users.avatar_url(&user_id).await.ok(), - })) - } else { - None - } - }); + if user_in_public_room.or(user_sees_user).await { + Some(assign!(search_users::v3::User::new(user_id.clone()), { + display_name, + avatar_url: services.users.avatar_url(&user_id).await.ok(), + })) + } else { + None + } + }); let results = users.by_ref().take(limit).collect().await; let limited = users.next().await.is_some(); diff --git a/src/api/server/send.rs b/src/api/server/send.rs index 47438ae85..ead1713c1 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -620,8 +620,9 @@ async fn handle_edu_direct_to_device( .broad_filter_map(|(target_user_id, map)| async move { services .users - .is_active_local(&target_user_id) + .status(&target_user_id) .await + .is_active() .then_some((target_user_id, map)) }) .for_each_concurrent(automatic_width(), |(target_user_id, map)| { diff --git a/src/database/maps.rs b/src/database/maps.rs index 778c098e8..dbaa71a78 100644 --- a/src/database/maps.rs +++ b/src/database/maps.rs @@ -169,6 +169,10 @@ pub(super) static MAPS: &[Descriptor] = &[ name: "registrationtoken_info", ..descriptor::RANDOM_SMALL }, + Descriptor { + name: "remoteuserid_remoteuser", + ..descriptor::RANDOM + }, Descriptor { name: "roomid_invitedcount", ..descriptor::RANDOM_SMALL @@ -398,6 +402,10 @@ pub(super) static MAPS: &[Descriptor] = &[ name: "userid_blurhash", ..descriptor::DROPPED }, + Descriptor { + name: "userid_deactivated", + ..descriptor::RANDOM_SMALL + }, Descriptor { name: "userid_dehydrateddevice", ..descriptor::RANDOM_SMALL diff --git a/src/service/admin/create.rs b/src/service/admin/create.rs index 46c838f9d..2febccaa4 100644 --- a/src/service/admin/create.rs +++ b/src/service/admin/create.rs @@ -48,7 +48,7 @@ pub async fn create_admin_room(services: &Services) -> Result { // Create a user for the server let server_user = services.globals.server_user.as_ref(); - services.users.create(server_user, None)?; + services.users.create_shadow_account(server_user).await?; let mut create_content = if room_version_rules.authorization.use_room_create_sender { RoomCreateEventContent::new_v1(server_user.into()) diff --git a/src/service/appservice/mod.rs b/src/service/appservice/mod.rs index ebd73d8af..3c3efa9bc 100644 --- a/src/service/appservice/mod.rs +++ b/src/service/appservice/mod.rs @@ -108,17 +108,17 @@ impl Service { self.services.globals.server_name(), )?; - if !self.services.users.exists(&appservice_user_id).await { - self.services.users.create(&appservice_user_id, None)?; - } else if self + if !self .services .users - .is_deactivated(&appservice_user_id) + .status(&appservice_user_id) .await - .unwrap_or(false) + .is_found() { - // Reactivate the appservice user if it was accidentally deactivated - self.services.users.set_password(&appservice_user_id, None); + self.services + .users + .create_shadow_account(&appservice_user_id) + .await?; } self.registration_info diff --git a/src/service/emergency/mod.rs b/src/service/emergency/mod.rs index 31fa8604c..eca3ccbf8 100644 --- a/src/service/emergency/mod.rs +++ b/src/service/emergency/mod.rs @@ -54,15 +54,22 @@ impl Service { async fn set_emergency_access(&self) -> Result { let server_user = &self.services.globals.server_user; - self.services.users.set_password( - server_user, - self.services - .config - .emergency_password - .as_deref() - .map(HashedPassword::new) - .transpose()?, - ); + match &self.services.config.emergency_password { + | Some(emergency_password) => { + let emergency_password = HashedPassword::new(emergency_password)?; + + self.services + .users + .convert_to_local_account(server_user, emergency_password) + .await?; + }, + | None => { + self.services + .users + .convert_to_shadow_account(server_user) + .await?; + }, + } let (ruleset, pwd_set) = match self.services.config.emergency_password { | Some(_) => (Ruleset::server_default(server_user), true), diff --git a/src/service/migrations.rs b/src/service/migrations.rs index bd087289b..f62aa21de 100644 --- a/src/service/migrations.rs +++ b/src/service/migrations.rs @@ -41,7 +41,7 @@ pub(crate) async fn migrations(services: &Services) -> Result<()> { // requires recreating the database from scratch. if users_count > 0 { let server_user = &services.globals.server_user; - if !services.users.exists(server_user).await { + if !services.users.status(server_user).await.is_found() { error!("The {server_user} server user does not exist, and the database is not new."); return Err!(Database( "Cannot reuse an existing database after changing the server name, please \ @@ -228,6 +228,18 @@ async fn migrate(services: &Services) -> Result<()> { .map_err(|e| err!("Failed to run 'fix_local_invite_state' migration': {e}"))?; } + if services.globals.db.database_version().await < 18 { + services.globals.db.bump_database_version(18); + info!("Migration: Bumped database version to 18"); + } + + if db["global"].get(SPLIT_USERID_PASSWORD).await.is_not_found() { + info!("Running migration 'split_userid_password'"); + split_userid_password(services) + .await + .map_err(|e| err!("Failed to run 'split_userid_password' migration': {e}"))?; + } + assert_eq!( services.globals.db.database_version().await, DATABASE_VERSION, @@ -242,9 +254,9 @@ async fn migrate(services: &Services) -> Result<()> { if !patterns.is_empty() { services .users - .stream() + .stream_local_users() .filter_map(async |user_id| { - if services.users.is_active_local(&user_id).await { + if services.users.status(&user_id).await.is_found() { Some(user_id) } else { None @@ -774,7 +786,7 @@ async fn fix_local_invite_state(services: &Services) -> Result { let db = &services.db; let cork = db.cork_and_sync(); - let userroomid_invitestate = services.db["userroomid_invitestate"].clone(); + let userroomid_invitestate = db["userroomid_invitestate"].clone(); // for each user invited to a room let fixed = userroomid_invitestate.stream() @@ -818,3 +830,44 @@ async fn fix_local_invite_state(services: &Services) -> Result { db.db.sort()?; Ok(()) } + +const SPLIT_USERID_PASSWORD: &str = "split_userid_password"; +async fn split_userid_password(services: &Services) -> Result { + // Split remote and deactivated users out from the `userid_password` table + + let db = &services.db; + let cork = db.cork_and_sync(); + let userid_password = db["userid_password"].clone(); + let remoteuserid_remoteuser = db["remoteuserid_remoteuser"].clone(); + let userid_deactivated = db["userid_deactivated"].clone(); + + let remote_users = userid_password + .stream::() + .ignore_err() + .fold(0_usize, async |mut remote_users, (user_id, hash)| { + if !services.globals.user_is_local(&user_id) { + assert!(hash.is_empty(), "non-empty hash {hash} for remote user {user_id}"); + + remoteuserid_remoteuser.insert(&user_id, ""); + userid_password.remove(&user_id); + remote_users = remote_users.saturating_add(1); + } else if hash.is_empty() { + if !(services.appservice.is_exclusive_user_id(&user_id).await + || user_id == services.globals.server_user) + { + info!("Marking {user_id} as deactivated"); + userid_deactivated.insert(&user_id, ""); + } + } + + remote_users + }) + .await; + + drop(cork); + info!(?remote_users, "Split userid_password."); + + db["global"].insert(FIXED_LOCAL_INVITE_STATE_MARKER, []); + db.db.sort()?; + Ok(()) +} diff --git a/src/service/rooms/state_cache/mod.rs b/src/service/rooms/state_cache/mod.rs index 46ef9a769..5c64ce04c 100644 --- a/src/service/rooms/state_cache/mod.rs +++ b/src/service/rooms/state_cache/mod.rs @@ -248,7 +248,7 @@ impl Service { ) -> impl Stream + Send + 'a { self.local_users_in_room(room_id) .filter_map(async |user_id| { - if self.services.users.is_active(&user_id).await { + if self.services.users.status(&user_id).await.is_active() { Some(user_id) } else { None diff --git a/src/service/rooms/state_cache/update.rs b/src/service/rooms/state_cache/update.rs index 5249df182..28f5098b7 100644 --- a/src/service/rooms/state_cache/update.rs +++ b/src/service/rooms/state_cache/update.rs @@ -30,15 +30,8 @@ impl super::Service { ) -> Result { let membership = pdu.get_content::()?; - // Keep track what remote users exist by adding them as "deactivated" users - // - // TODO: use futures to update remote profiles without blocking the membership - // update - #[allow(clippy::collapsible_if)] - if !self.services.globals.user_is_local(user_id) - && !self.services.users.exists(user_id).await - { - self.services.users.create(user_id, None)?; + if !self.services.globals.user_is_local(user_id) { + self.services.users.record_remote_user(user_id); } match &membership.membership { diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index ab657d4fe..132ee4f0a 100644 --- a/src/service/rooms/timeline/append.rs +++ b/src/service/rooms/timeline/append.rs @@ -255,7 +255,13 @@ impl super::Service { if let Some(state_key) = pdu.state_key() { match UserId::parse(state_key) { | Ok(target_user_id) => { - if self.services.users.is_active_local(&target_user_id).await { + if self + .services + .users + .status(&target_user_id) + .await + .is_active() + { push_targets.insert(target_user_id.clone()); } }, diff --git a/src/service/users/account.rs b/src/service/users/account.rs index d663e7940..f4ed1f564 100644 --- a/src/service/users/account.rs +++ b/src/service/users/account.rs @@ -6,7 +6,7 @@ use conduwuit::{ warn, }; use database::{Deserialized, Json}; -use futures::{FutureExt, Stream, StreamExt, TryFutureExt}; +use futures::{FutureExt, Stream, StreamExt}; use lettre::Address; use ruma::{ MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedUserId, UserId, @@ -29,6 +29,28 @@ pub enum AccessTokenStatus { Expired, } +/// The status of a user account. +#[derive(Clone, Copy)] +pub enum AccountStatus { + NotFound, + Active, + Deactivated, +} + +impl AccountStatus { + pub fn is_found(&self) -> bool { !matches!(self, Self::NotFound) } + + pub fn is_active(&self) -> bool { matches!(self, Self::Active) } + + pub fn ensure_active(&self) -> Result<()> { + match self { + | Self::Active => Ok(()), + | Self::Deactivated => Err!(Request(UserDeactivated("This account is deactivated."))), + | Self::NotFound => Err!(Request(NotFound("This account does not exist."))), + } + } +} + impl super::Service { /// Returns true/false based on whether the recipient/receiving user has /// ignored the sender. @@ -89,16 +111,16 @@ impl super::Service { self.services.admin.user_is_admin(user_id).await } - /// Create a new user account on this homeserver. Set the password to `None` - /// to create a non-local user. Non-local users with a password will return - /// an error. - #[inline] - pub fn create(&self, user_id: &UserId, password: Option) -> Result<()> { - if !self.services.globals.user_is_local(user_id) && password.is_some() { - return Err!("Cannot create a nonlocal user with a set password"); + /// Create a new shadow user account on this server. Shadow accounts + /// have no password and cannot be logged into. + pub async fn create_shadow_account(&self, user_id: &UserId) -> Result<()> { + assert!(self.services.globals.user_is_local(user_id), "user id must be local"); + + if self.status(user_id).await.is_found() { + return Err!(Request(UserInUse("An account with this user ID already exists."))); } - self.set_password(user_id, password); + self.db.userid_password.insert(user_id, ""); Ok(()) } @@ -109,9 +131,9 @@ impl super::Service { user_id: &UserId, password: HashedPassword, email: Option
, - ) { - self.create(user_id, Some(password)) - .expect("should be able to save a new local user. what happened?"); + ) -> Result<()> { + self.create_shadow_account(user_id).await?; + self.convert_to_local_account(user_id, password).await?; // Set an initial display name { @@ -230,6 +252,8 @@ impl super::Service { } info!("Created new user account for {user_id}"); + + Ok(()) } pub async fn determine_registration_user_id( @@ -297,7 +321,7 @@ impl super::Service { }, }; - if self.exists(&user_id).await { + if self.status(&user_id).await.is_found() { return Err!(Request(UserInUse("User ID is not available."))); } @@ -329,26 +353,24 @@ impl super::Service { ) .unwrap(); - if !self.exists(&user_id).await { + if !self.status(&user_id).await.is_found() { break Ok(user_id); } } } } - /// Deactivates an account, removing all of their device IDs and unsetting - /// their password. + /// Deactivates an account, removing all of their device IDs. pub async fn deactivate_account(&self, user_id: &UserId) -> Result<()> { + self.status(user_id).await.ensure_active()?; + // Remove all associated devices self.all_device_ids(user_id) .for_each(async |device_id| self.remove_device(user_id, &device_id).await) .await; - // Set the password to "" to indicate a deactivated account. Hashes will never - // result in an empty string, so the user will not be able to log in again. - // Systems like changing the password without logging in should check if the - // account is deactivated. - self.set_password(user_id, None); + // Mark user as deactivated + self.db.userid_deactivated.insert(user_id, ""); // TODO: Unhook 3PID Ok(()) @@ -393,25 +415,6 @@ impl super::Service { /// Unlocks an account, allowing the user to log in and use it again. pub async fn unlock_account(&self, user_id: &UserId) { self.db.userid_lock.remove(user_id); } - /// Check if the provided user ID belongs to an existing (possibly - /// deactivated) account on this homeserver. - #[inline] - pub async fn exists(&self, user_id: &UserId) -> bool { - self.services.globals.user_is_local(user_id) - && self.db.userid_password.get(user_id).await.is_ok() - } - - /// Check if account is deactivated (has an empty password). Returns a - /// NotFound error if the user does not exist. - pub async fn is_deactivated(&self, user_id: &UserId) -> Result { - self.db - .userid_password - .get(user_id) - .map_ok(|val| val.is_empty()) - .map_err(|_| err!(Request(NotFound("User does not exist.")))) - .await - } - /// Check if account is suspended. Returns false if the user does not exist. pub async fn is_suspended(&self, user_id: &UserId) -> Result { match self @@ -469,18 +472,33 @@ impl super::Service { .is_ok() } - /// Check if account is active (not deactivated) - pub async fn is_active(&self, user_id: &UserId) -> bool { - !self.is_deactivated(user_id).await.unwrap_or(true) + /// Checks the activation status of the provided user ID's account. + pub async fn status(&self, user_id: &UserId) -> AccountStatus { + if !self.services.globals.user_is_local(user_id) { + AccountStatus::NotFound + } else if self.db.userid_password.exists(user_id).await.is_ok() { + if self.db.userid_deactivated.exists(user_id).await.is_ok() { + AccountStatus::Deactivated + } else { + AccountStatus::Active + } + } else { + AccountStatus::NotFound + } } - /// Check if account is a local user, and is active (not deactivated) - pub async fn is_active_local(&self, user_id: &UserId) -> bool { - self.services.globals.user_is_local(user_id) && self.is_active(user_id).await + /// Checks if a user is a shadow. + pub async fn is_shadow(&self, user_id: &UserId) -> bool { + self.db + .userid_password + .get(user_id) + .await + .deserialized::() + .is_ok_and(|hash| hash.is_empty()) } /// Returns the number of users registered on this server, including - /// deactivated users. + /// deactivated and shadow users. #[inline] pub async fn count(&self) -> usize { self.db.userid_password.count().await } @@ -522,8 +540,8 @@ impl super::Service { Some((user_id, device_id, AccessTokenStatus::Valid)) } - /// Returns an iterator over all users on this homeserver. - pub fn stream(&self) -> impl Stream + Send { + /// Returns an iterator over all local users on this homeserver. + pub fn stream_local_users(&self) -> impl Stream + Send { self.db.userid_password.keys().ignore_err() } @@ -540,16 +558,49 @@ impl super::Service { } /// Set a user's password. - pub fn set_password(&self, user_id: &UserId, password: Option) { - if let Some(hash) = password { - self.db.userid_password.insert(user_id, hash.0); - } else { - self.db.userid_password.insert(user_id, b""); + pub async fn set_password(&self, user_id: &UserId, password: HashedPassword) -> Result { + self.status(user_id).await.ensure_active()?; + + self.db.userid_password.insert(user_id, password.0); + + Ok(()) + } + + /// Convert an existing user to an active local account. This will turn + /// shadow accounts into normal accounts and will clear the deactivation + /// flag on the user if it is set. + pub async fn convert_to_local_account( + &self, + user_id: &UserId, + password: HashedPassword, + ) -> Result { + if !self.status(user_id).await.is_found() { + return Err!(Request(NotFound("This account does not exist."))); } + + self.db.userid_deactivated.remove(user_id); + self.db.userid_password.insert(user_id, password.0); + + Ok(()) + } + + /// Convert an existing user to a shadow account. This will clear their + /// password and reactivate them if they are deactivated. + pub async fn convert_to_shadow_account(&self, user_id: &UserId) -> Result { + if !self.status(user_id).await.is_found() { + return Err!(Request(NotFound("This account does not exist."))); + } + + self.db.userid_deactivated.remove(user_id); + self.db.userid_password.insert(user_id, ""); + + Ok(()) } /// Check a user's password. pub async fn check_password(&self, user_id: &UserId, password: &str) -> Result { + self.status(user_id).await.ensure_active()?; + let (hash, user_id): (String, OwnedUserId) = if let Ok(hash) = self.db.userid_password.get(user_id).await.deserialized() { (hash, user_id.to_owned()) @@ -558,21 +609,20 @@ impl super::Service { // better let lowercase_user_id = UserId::parse(user_id.as_str().to_lowercase()).unwrap(); - if let Ok(hash) = self + let hash = self .db .userid_password .get(lowercase_user_id.as_str()) .await .deserialized() - { - (hash, lowercase_user_id) - } else { - return Err!(Request(Forbidden("This user cannot log in with a password."))); - } + .expect("user should exist"); + + (hash, lowercase_user_id) }; if hash.is_empty() { - return Err!(Request(UserDeactivated("This user is deactivated"))); + // Cannot log into shadow users + return Err!(Request(Forbidden("This user cannot log in with a password."))); } utils::hash::verify_password(password, &hash) diff --git a/src/service/users/dehydrated_device.rs b/src/service/users/dehydrated_device.rs index 1711e64ed..6d536f5af 100644 --- a/src/service/users/dehydrated_device.rs +++ b/src/service/users/dehydrated_device.rs @@ -21,10 +21,7 @@ pub struct DehydratedDevice { impl super::Service { /// Creates or recreates the user's dehydrated device. pub async fn set_dehydrated_device(&self, user_id: &UserId, request: Request) -> Result { - assert!( - self.exists(user_id).await, - "Tried to create dehydrated device for non-existent user" - ); + self.status(user_id).await.ensure_active()?; let existing_id = self.get_dehydrated_device_id(user_id).await; diff --git a/src/service/users/device.rs b/src/service/users/device.rs index bc704f1d8..9d436b152 100644 --- a/src/service/users/device.rs +++ b/src/service/users/device.rs @@ -4,8 +4,8 @@ use std::{ }; use conduwuit::{ - Err, utils, - utils::{ReadyExt, stream::TryIgnore}, + Err, Result, + utils::{self, ReadyExt, stream::TryIgnore}, }; use database::{Deserialized, Ignore, Interfix, Json}; use futures::{Stream, StreamExt}; @@ -18,8 +18,7 @@ use serde_json::json; use crate::users::increment; impl super::Service { - /// Adds a new device to a user. The user must exist, otherwise InvalidParam - /// is returned. + /// Adds a new device to a user. pub async fn create_device( &self, user_id: &UserId, @@ -28,12 +27,8 @@ impl super::Service { token_max_age: Option, initial_device_display_name: Option, client_ip: Option, - ) -> conduwuit::Result<()> { - if !self.exists(user_id).await { - return Err!(Request(InvalidParam(error!( - "Called create_device for non-existent user {user_id}" - )))); - } + ) -> Result<()> { + self.status(user_id).await.ensure_active()?; let key = (user_id, device_id); let mut device = Device::new(device_id.into()); @@ -50,7 +45,7 @@ impl super::Service { /// Removes a device from a user. pub async fn remove_device(&self, user_id: &UserId, device_id: &DeviceId) { // Remove dehydrated device if this is the dehydrated device - let _: conduwuit::Result<_> = self + let _ = self .remove_dehydrated_device(user_id, Some(device_id)) .await; @@ -97,11 +92,7 @@ impl super::Service { } /// Gets the access token associated with a device. - pub async fn get_token( - &self, - user_id: &UserId, - device_id: &DeviceId, - ) -> conduwuit::Result { + pub async fn get_token(&self, user_id: &UserId, device_id: &DeviceId) -> Result { let key = (user_id, device_id); self.db.userdeviceid_token.qry(&key).await.deserialized() } @@ -131,7 +122,7 @@ impl super::Service { device_id: &DeviceId, token: &str, token_max_age: Option, - ) -> conduwuit::Result<()> { + ) -> Result<()> { let key = (user_id, device_id); if self.db.userdeviceid_metadata.qry(&key).await.is_err() { return Err!(Database(error!( @@ -259,7 +250,7 @@ impl super::Service { user_id: &UserId, device_id: &DeviceId, device: &Device, - ) -> conduwuit::Result<()> { + ) -> Result<()> { increment(&self.db.userid_devicelistversion, user_id.as_bytes()); self.update_device_metadata_no_increment(user_id, device_id, device) } @@ -273,7 +264,7 @@ impl super::Service { user_id: &UserId, device_id: &DeviceId, device: &Device, - ) -> conduwuit::Result<()> { + ) -> Result<()> { let key = (user_id, device_id); self.db.userdeviceid_metadata.put(key, Json(device)); @@ -312,7 +303,7 @@ impl super::Service { &self, user_id: &UserId, device_id: &DeviceId, - ) -> conduwuit::Result { + ) -> Result { self.db .userdeviceid_metadata .qry(&(user_id, device_id)) @@ -321,7 +312,7 @@ impl super::Service { } /// Gets the most recent device list version for a user. - pub async fn get_devicelist_version(&self, user_id: &UserId) -> conduwuit::Result { + pub async fn get_devicelist_version(&self, user_id: &UserId) -> Result { self.db .userid_devicelistversion .get(user_id) diff --git a/src/service/users/mod.rs b/src/service/users/mod.rs index ae4d47ef9..734651e5a 100644 --- a/src/service/users/mod.rs +++ b/src/service/users/mod.rs @@ -4,10 +4,11 @@ pub(super) mod device; pub(super) mod filters; pub(super) mod keys; pub(super) mod profile; +pub(super) mod remote; use std::{mem, sync::Arc}; -pub use account::AccessTokenStatus; +pub use account::{AccessTokenStatus, AccountStatus}; use conduwuit::{ Err, Error, Result, err, utils::{self}, @@ -75,11 +76,13 @@ struct Data { logintoken_expiresatuserid: Arc, todeviceid_events: Arc, token_userdeviceid: Arc, + remoteuserid_remoteuser: Arc, userdeviceid_tokenexpires: Arc, userdeviceid_metadata: Arc, userdeviceid_token: Arc, userfilterid_filter: Arc, userid_avatarurl: Arc, + userid_deactivated: Arc, userid_dehydrateddevice: Arc, userid_devicelistversion: Arc, userid_displayname: Arc, @@ -121,10 +124,12 @@ impl crate::Service for Service { logintoken_expiresatuserid: args.db["logintoken_expiresatuserid"].clone(), todeviceid_events: args.db["todeviceid_events"].clone(), token_userdeviceid: args.db["token_userdeviceid"].clone(), + remoteuserid_remoteuser: args.db["remoteuserid_remoteuser"].clone(), userdeviceid_metadata: args.db["userdeviceid_metadata"].clone(), userdeviceid_token: args.db["userdeviceid_token"].clone(), userfilterid_filter: args.db["userfilterid_filter"].clone(), userid_avatarurl: args.db["userid_avatarurl"].clone(), + userid_deactivated: args.db["userid_deactivated"].clone(), userid_dehydrateddevice: args.db["userid_dehydrateddevice"].clone(), userid_devicelistversion: args.db["userid_devicelistversion"].clone(), userid_displayname: args.db["userid_displayname"].clone(), diff --git a/src/service/users/remote.rs b/src/service/users/remote.rs new file mode 100644 index 000000000..21be89e6e --- /dev/null +++ b/src/service/users/remote.rs @@ -0,0 +1,17 @@ +use conduwuit::utils::stream::TryIgnore; +use futures::Stream; +use ruma::{OwnedUserId, UserId}; + +impl super::Service { + /// Record the existence of a remote user. + pub fn record_remote_user(&self, user_id: &UserId) { + assert!(self.services.globals.user_is_local(user_id), "user is not remote"); + + self.db.remoteuserid_remoteuser.insert(user_id, ""); + } + + /// Returns a stream over all remote users this server has ever seen. + pub fn stream_remote_users(&self) -> impl Stream + Send { + self.db.remoteuserid_remoteuser.keys().ignore_err() + } +} diff --git a/src/web/pages/account/password/change.rs b/src/web/pages/account/password/change.rs index f484b72c5..a8d4adc80 100644 --- a/src/web/pages/account/password/change.rs +++ b/src/web/pages/account/password/change.rs @@ -69,7 +69,7 @@ async fn route_change_password( let user_card = UserCard::for_local_user(&services, user_id.clone()).await; let body = if let Some(form) = form { - match change_password(&services, &user_id, form).await { + match change_password(&services, &user_id, form).await? { | Ok(()) => ChangePasswordBody::Success, | Err(errors) => ChangePasswordBody::Form(ChangePasswordForm::with_errors(context.clone(), errors)), @@ -85,7 +85,7 @@ async fn change_password( services: &crate::State, user_id: &UserId, form: ChangePasswordForm, -) -> Result<(), ValidationErrors> { +) -> Result> { form.validate()?; if services @@ -100,12 +100,12 @@ async fn change_password( ValidationError::new("wrong").with_message("Incorrect password".into()), ); - return Err(errors); + return Ok(Err(errors)); } match HashedPassword::new(&form.new_password) { | Ok(hash) => { - services.users.set_password(user_id, Some(hash)); + services.users.set_password(user_id, hash).await?; }, | Err(err) => { let mut errors = ValidationErrors::new(); @@ -114,9 +114,9 @@ async fn change_password( ValidationError::new("malformed").with_message(err.message().into()), ); - return Err(errors); + return Ok(Err(errors)); }, } - Ok(()) + Ok(Ok(())) } diff --git a/src/web/pages/account/password/reset.rs b/src/web/pages/account/password/reset.rs index 7428fb4ab..b263060b7 100644 --- a/src/web/pages/account/password/reset.rs +++ b/src/web/pages/account/password/reset.rs @@ -214,7 +214,7 @@ async fn route_reset_password_validate( | Ok(hash) => { let _ = session.consume(); - services.users.set_password(&user_id, Some(hash)); + services.users.set_password(&user_id, hash).await?; ResetPasswordValidateBody::ResetSuccess { user_card } }, diff --git a/src/web/pages/account/register.rs b/src/web/pages/account/register.rs index 529901ef4..9dffbf4ed 100644 --- a/src/web/pages/account/register.rs +++ b/src/web/pages/account/register.rs @@ -310,7 +310,7 @@ async fn get_register_email_validate( response!( complete_registration(&services, session_store, completed_registration, Some(email)) - .await + .await? ) } @@ -502,7 +502,7 @@ async fn begin_registration( } else { // If email isn't required we can immediately complete registration Ok(response!( - complete_registration(services, session_store, completed_registration, None).await + complete_registration(services, session_store, completed_registration, None).await? )) } } @@ -517,11 +517,11 @@ async fn complete_registration( next, }: CompletedRegistration, email: Option
, -) -> Redirect { +) -> Result { services .users .create_local_account(&user_id, password_hash, email) - .await; + .await?; if let Some(registration_token) = registration_token { services @@ -536,7 +536,7 @@ async fn complete_registration( .await .expect("should be able to serialize user session"); - Redirect::to(&next.unwrap_or_default().target_path()) + Ok(Redirect::to(&next.unwrap_or_default().target_path())) } pub(super) async fn registration_flow_status( diff --git a/src/web/session/mod.rs b/src/web/session/mod.rs index ccc1e6b27..6bb1ba4df 100644 --- a/src/web/session/mod.rs +++ b/src/web/session/mod.rs @@ -162,10 +162,8 @@ pub(crate) async fn require_active( user_id: &UserId, allow_locked: bool, ) -> Result<(), Response> { - if !services.users.is_active(user_id).await { - return Err( - WebError::Forbidden("Your account is deactivated.".to_owned()).into_response() - ); + if let Err(err) = services.users.status(user_id).await.ensure_active() { + return Err(WebError::Forbidden(err.message()).into_response()); } if !allow_locked