refactor: Split remote and deactivated users into their own columns

This commit is contained in:
Ginger committed 2026-07-02 16:24:07 -04:00
1 parent d4f6c053bc
commit 69163b5391
32 files changed
+391 -278

No files matched your search

+6 -1
View File
@@ -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::<Vec<_>>().await;
let users = self
.services
.users
.stream_local_users()
.collect::<Vec<_>>()
.await;
let query_time = timer.elapsed();
let total = users.len();
+1 -1
View File
@@ -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;
+8 -3
View File
@@ -191,8 +191,13 @@ impl crate::Context<'_> {
async fn iter_users(&self) -> Result {
let timer = tokio::time::Instant::now();
let result: Vec<OwnedUserId> =
self.services.users.stream().map(Into::into).collect().await;
let result: Vec<OwnedUserId> = 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())
+48 -58
View File
@@ -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<String>,
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
+3
View File
@@ -27,6 +27,9 @@ pub enum UserCommand {
username: String,
/// New password for the user, if unspecified one is generated
password: Option<String>,
#[arg(long)]
convert_to_local_account: bool,
},
/// Get a user's associated email address.
+1 -8
View File
@@ -54,14 +54,7 @@ pub(crate) async fn parse_active_local_user_id(
user_id: &str,
) -> Result<OwnedUserId> {
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)
}
+2 -1
View File
@@ -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
+2 -2
View File
@@ -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
};
+13 -16
View File
@@ -12,20 +12,18 @@ 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 (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<lock_user::v1::Response> {
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));
+13 -16
View File
@@ -12,20 +12,18 @@ pub(crate) async fn get_suspended_status(
State(services): State<crate::State>,
body: Ruma<is_user_suspended::v1::Request>,
) -> Result<is_user_suspended::v1::Response> {
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<suspend_user::v1::Response> {
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));
+1 -1
View File
@@ -110,7 +110,7 @@ async fn fetch_full_profile(
user_id: &UserId,
) -> Option<BTreeMap<String, Value>> {
// 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);
}
+1 -1
View File
@@ -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());
}
-8
View File
@@ -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.")));
}
+30 -26
View File
@@ -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();
+2 -1
View File
@@ -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)| {
+8
View File
@@ -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
+1 -1
View File
@@ -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())
+7 -7
View File
@@ -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
+16 -9
View File
@@ -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),
+57 -4
View File
@@ -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::<OwnedUserId, String>()
.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(())
}
+1 -1
View File
@@ -248,7 +248,7 @@ impl Service {
) -> impl Stream<Item = OwnedUserId> + 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
+2 -9
View File
@@ -30,15 +30,8 @@ impl super::Service {
) -> Result {
let membership = pdu.get_content::<RoomMemberEventContent>()?;
// 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 {
+7 -1
View File
@@ -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());
}
},
+111 -61
View File
@@ -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<HashedPassword>) -> 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<Address>,
) {
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<bool> {
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<bool> {
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::<String>()
.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<Item = OwnedUserId> + Send {
/// Returns an iterator over all local users on this homeserver.
pub fn stream_local_users(&self) -> impl Stream<Item = OwnedUserId> + 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<HashedPassword>) {
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<OwnedUserId> {
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)
+1 -4
View File
@@ -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;
+12 -21
View File
@@ -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<Duration>,
initial_device_display_name: Option<String>,
client_ip: Option<String>,
) -> 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<String> {
pub async fn get_token(&self, user_id: &UserId, device_id: &DeviceId) -> Result<String> {
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<Duration>,
) -> 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<Device> {
) -> Result<Device> {
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<u64> {
pub async fn get_devicelist_version(&self, user_id: &UserId) -> Result<u64> {
self.db
.userid_devicelistversion
.get(user_id)
+6 -1
View File
@@ -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<Map>,
todeviceid_events: Arc<Map>,
token_userdeviceid: Arc<Map>,
remoteuserid_remoteuser: Arc<Map>,
userdeviceid_tokenexpires: Arc<Map>,
userdeviceid_metadata: Arc<Map>,
userdeviceid_token: Arc<Map>,
userfilterid_filter: Arc<Map>,
userid_avatarurl: Arc<Map>,
userid_deactivated: Arc<Map>,
userid_dehydrateddevice: Arc<Map>,
userid_devicelistversion: Arc<Map>,
userid_displayname: Arc<Map>,
@@ -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(),
+17
View File
@@ -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<Item = OwnedUserId> + Send {
self.db.remoteuserid_remoteuser.keys().ignore_err()
}
}
+6 -6
View File
@@ -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<Result<(), ValidationErrors>> {
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(()))
}
+1 -1
View File
@@ -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 }
},
+5 -5
View File
@@ -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<Address>,
) -> Redirect {
) -> Result<Redirect> {
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(
+2 -4
View File
@@ -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