From 967e2cc54fb1accbef7fc7bf4e644903accc7cca Mon Sep 17 00:00:00 2001 From: timedout Date: Fri, 26 Jun 2026 07:53:25 +0100 Subject: [PATCH] refactor: Split the users service into several modules Also resolvers several lints, including needless async on create_user --- src/admin/user/commands.rs | 3 +- src/api/client/account/register.rs | 2 +- src/service/admin/create.rs | 2 +- src/service/appservice/mod.rs | 5 +- src/service/rooms/state_cache/update.rs | 2 +- src/service/users/account.rs | 393 +++++++ src/service/users/device.rs | 321 ++++++ src/service/users/filters.rs | 26 + src/service/users/keys.rs | 532 +++++++++ src/service/users/mod.rs | 1323 +---------------------- src/service/users/profile.rs | 94 ++ 11 files changed, 1381 insertions(+), 1322 deletions(-) create mode 100644 src/service/users/account.rs create mode 100644 src/service/users/device.rs create mode 100644 src/service/users/filters.rs create mode 100644 src/service/users/keys.rs create mode 100644 src/service/users/profile.rs diff --git a/src/admin/user/commands.rs b/src/admin/user/commands.rs index 66589b4b6..a7b270e5a 100644 --- a/src/admin/user/commands.rs +++ b/src/admin/user/commands.rs @@ -69,8 +69,7 @@ pub(super) async fn create_user(&self, username: String, password: Option Result { // Create a user for the server let server_user = services.globals.server_user.as_ref(); - services.users.create(server_user, None).await?; + services.users.create(server_user, None)?; 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 d492f3188..1784ea021 100644 --- a/src/service/appservice/mod.rs +++ b/src/service/appservice/mod.rs @@ -109,10 +109,7 @@ async fn start_appservice(&self, id: String, registration: Registration) -> Resu )?; if !self.services.users.exists(&appservice_user_id).await { - self.services - .users - .create(&appservice_user_id, None) - .await?; + self.services.users.create(&appservice_user_id, None)?; } else if self .services .users diff --git a/src/service/rooms/state_cache/update.rs b/src/service/rooms/state_cache/update.rs index 20c6a700f..5249df182 100644 --- a/src/service/rooms/state_cache/update.rs +++ b/src/service/rooms/state_cache/update.rs @@ -38,7 +38,7 @@ pub async fn update_membership( if !self.services.globals.user_is_local(user_id) && !self.services.users.exists(user_id).await { - self.services.users.create(user_id, None).await?; + self.services.users.create(user_id, None)?; } match &membership.membership { diff --git a/src/service/users/account.rs b/src/service/users/account.rs new file mode 100644 index 000000000..7992d0967 --- /dev/null +++ b/src/service/users/account.rs @@ -0,0 +1,393 @@ +use conduwuit::{ + Err, debug_error, debug_warn, err, trace, utils, + utils::{ReadyExt, stream::TryIgnore}, +}; +use database::{Deserialized, Json}; +use futures::{Stream, StreamExt, TryFutureExt}; +use ruma::{ + MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedUserId, UserId, + events::{GlobalAccountDataEventType, ignored_user_list::IgnoredUserListEvent}, +}; +use ruminuwuity::invite_permission_config::{FilterLevel, InvitePermissionConfigEvent}; + +use crate::users::{HashedPassword, UserSuspension}; + +impl super::Service { + /// Returns true/false based on whether the recipient/receiving user has + /// ignored the sender. + pub async fn user_is_ignored(&self, sender_user: &UserId, recipient_user: &UserId) -> bool { + self.services + .account_data + .get_global(recipient_user, GlobalAccountDataEventType::IgnoredUserList) + .await + .is_ok_and(|ignored: IgnoredUserListEvent| { + ignored + .content + .ignored_users + .keys() + .any(|blocked_user| blocked_user == sender_user) + }) + } + + /// Returns the recipient's filter level for an invite from the sender. + /// + /// If the sender is ignored by the recipient, `Ignore` is returned. + /// Otherwise, the resulting value depends on their invite blocking or + /// invite filtering configuration. + pub async fn invite_filter_level( + &self, + sender_user: &UserId, + recipient_user: &UserId, + ) -> FilterLevel { + if self.user_is_ignored(sender_user, recipient_user).await { + FilterLevel::Ignore + } else { + let (stable, unstable) = tokio::join!( + self.services + .account_data + .get_global::( + recipient_user, + GlobalAccountDataEventType::InvitePermissionConfig + ), + self.services + .account_data + .get_global::( + recipient_user, + "org.matrix.msc4155.invite_permission_config".into() + ) // TODO: MSC4155 probably needs upstreaming to ruma at some point + ); + if stable.is_err() && unstable.is_err() { + return FilterLevel::Allow; + } + stable + .unwrap_or_else(|_| unstable.unwrap()) + .content + .user_filter_level(sender_user) + } + } + + /// Check if a user is an admin + #[inline] + pub async fn is_admin(&self, user_id: &UserId) -> bool { + 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, + ) -> conduwuit::Result<()> { + if !self.services.globals.user_is_local(user_id) && password.is_some() { + return Err!("Cannot create a nonlocal user with a set password"); + } + + self.set_password(user_id, password); + + Ok(()) + } + + /// Deactivates an account, removing all of their device IDs and unsetting + /// their password. + pub async fn deactivate_account(&self, user_id: &UserId) -> conduwuit::Result<()> { + // 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); + + // TODO: Unhook 3PID + 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); + } + + /// Locks an account, preventing it being used until it is unlocked. + pub async fn lock_account(&self, user_id: &UserId, locking_user: &UserId) { + // NOTE: Locking is basically just suspension with a more severe effect, + // so we'll just re-use the suspension data structure to store the lock state. + let suspension = self + .db + .userid_lock + .get(user_id) + .await + .deserialized::() + .unwrap_or_else(|_| UserSuspension { + suspended: true, + suspended_at: MilliSecondsSinceUnixEpoch::now().get().into(), + suspended_by: locking_user.to_string(), + }); + + self.db.userid_lock.raw_put(user_id, Json(suspension)); + } + + /// 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) -> conduwuit::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) -> conduwuit::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) + }, + } + } + + /// Returns true if the user is locked. Returns false if the user does not + /// exist or is not locked. + pub async fn is_locked(&self, user_id: &UserId) -> conduwuit::Result { + match self + .db + .userid_lock + .get(user_id) + .await + .deserialized::() + { + | Ok(s) => Ok(s.suspended), + | Err(e) => + if e.is_not_found() { + Ok(false) + } else { + Err(e) + }, + } + } + + /// Disables login for a user, preventing them from creating new devices, + /// but allows them to continue using their existing sessions unimpeded. + pub fn disable_login(&self, user_id: &UserId) { + self.db.userid_logindisabled.insert(user_id, ""); + } + + /// Re-enables login for a user, allowing them to create new devices again. + pub fn enable_login(&self, user_id: &UserId) { self.db.userid_logindisabled.remove(user_id); } + + /// Returns true if the target user's login is disabled. + pub async fn is_login_disabled(&self, user_id: &UserId) -> bool { + self.db + .userid_logindisabled + .exists(user_id.as_str()) + .await + .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) + } + + /// 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 + } + + /// Returns the number of users registered on this server, including + /// deactivated users. + #[inline] + pub async fn count(&self) -> usize { self.db.userid_password.count().await } + + /// Find out which user an access token belongs to. Will panic if the access + /// token is empty. + pub async fn find_from_token( + &self, + token: &str, + ) -> conduwuit::Result<(OwnedUserId, OwnedDeviceId)> { + assert!(!token.is_empty(), "Empty access token"); + self.db.token_userdeviceid.get(token).await.deserialized() + } + + /// Returns an iterator over all users on this homeserver. + pub fn stream(&self) -> impl Stream + Send { + self.db.userid_password.keys().ignore_err() + } + + /// Returns a list of active local users. + /// + /// A user account is considered `local` if the associated password is not + /// empty. + pub fn list_local_users(&self) -> impl Stream + Send + '_ { + self.db + .userid_password + .stream() + .ignore_err() + .ready_filter_map(|(u, p): (OwnedUserId, &[u8])| (!p.is_empty()).then_some(u)) + } + + /// 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""); + } + } + + /// Check a user's password. + pub async fn check_password( + &self, + user_id: &UserId, + password: &str, + ) -> conduwuit::Result { + let (hash, user_id): (String, OwnedUserId) = + if let Ok(hash) = self.db.userid_password.get(user_id).await.deserialized() { + (hash, user_id.to_owned()) + } else { + // We also check the lowercased version of the user ID to handle legacy user IDs + // better + let lowercase_user_id = UserId::parse(user_id.as_str().to_lowercase()).unwrap(); + + if let Ok(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."))); + } + }; + + if hash.is_empty() { + return Err!(Request(UserDeactivated("This user is deactivated"))); + } + + utils::hash::verify_password(password, &hash) + .inspect_err(|e| debug_error!("{e}")) + .map_err(|_| err!(Request(Forbidden("Invalid identifier or password."))))?; + + Ok(user_id) + } + + /// Creates an OpenID token, which can be used to prove that a user has + /// access to an account (primarily for integrations) + pub fn create_openid_token(&self, user_id: &UserId, token: &str) -> conduwuit::Result { + use std::num::Saturating as Sat; + + let expires_in = self.services.server.config.openid_token_ttl; + let expires_at = Sat(utils::millis_since_unix_epoch()) + Sat(expires_in) * Sat(1000); + + let mut value = expires_at.0.to_be_bytes().to_vec(); + value.extend_from_slice(user_id.as_bytes()); + + self.db + .openidtoken_expiresatuserid + .insert(token.as_bytes(), value.as_slice()); + + Ok(expires_in) + } + + /// Find out which user an OpenID access token belongs to. + pub async fn find_from_openid_token(&self, token: &str) -> conduwuit::Result { + let Ok(value) = self.db.openidtoken_expiresatuserid.get(token).await else { + return Err!(Request(Unauthorized("OpenID token is unrecognised"))); + }; + + let (expires_at_bytes, user_bytes) = value.split_at(0_u64.to_be_bytes().len()); + let expires_at = + u64::from_be_bytes(expires_at_bytes.try_into().map_err(|e| { + err!(Database("expires_at in openid_userid is invalid u64. {e}")) + })?); + + if expires_at < utils::millis_since_unix_epoch() { + debug_warn!("OpenID token is expired, removing"); + self.db.openidtoken_expiresatuserid.remove(token.as_bytes()); + + return Err!(Request(Unauthorized("OpenID token is expired"))); + } + + let user_string = utils::string_from_bytes(user_bytes) + .map_err(|e| err!(Database("User ID in openid_userid is invalid unicode. {e}")))?; + + OwnedUserId::try_from(user_string) + .map_err(|e| err!(Database("User ID in openid_userid is invalid. {e}"))) + } + + /// Creates a short-lived login token, which can be used to log in using the + /// `m.login.token` mechanism. + pub fn create_login_token(&self, user_id: &UserId, token: &str) -> u64 { + use std::num::Saturating as Sat; + + let expires_in = self.services.server.config.login_token_ttl; + let expires_at = Sat(utils::millis_since_unix_epoch()) + Sat(expires_in); + + let value = (expires_at.0, user_id); + self.db.logintoken_expiresatuserid.raw_put(token, value); + + expires_in + } + + /// Find out which user a login token belongs to. + /// Removes the token to prevent double-use attacks. + pub async fn find_from_login_token(&self, token: &str) -> conduwuit::Result { + let Ok(value) = self.db.logintoken_expiresatuserid.get(token).await else { + return Err!(Request(Forbidden("Login token is unrecognised"))); + }; + let (expires_at, user_id): (u64, OwnedUserId) = value.deserialized()?; + + if expires_at < utils::millis_since_unix_epoch() { + trace!(%user_id, ?token, "Removing expired login token"); + + self.db.logintoken_expiresatuserid.remove(token); + + return Err!(Request(Forbidden("Login token is expired"))); + } + + self.db.logintoken_expiresatuserid.remove(token); + + Ok(user_id) + } +} diff --git a/src/service/users/device.rs b/src/service/users/device.rs new file mode 100644 index 000000000..9c5ff4524 --- /dev/null +++ b/src/service/users/device.rs @@ -0,0 +1,321 @@ +use std::net::IpAddr; + +use conduwuit::{ + Err, utils, + utils::{ReadyExt, stream::TryIgnore}, +}; +use database::{Deserialized, Ignore, Interfix, Json}; +use futures::{Stream, StreamExt}; +use ruma::{ + DeviceId, MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedUserId, UserId, + api::client::device::Device, events::AnyToDeviceEvent, serde::Raw, uint, +}; +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. + pub async fn create_device( + &self, + user_id: &UserId, + device_id: &DeviceId, + token: &str, + 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}" + )))); + } + + let key = (user_id, device_id); + let mut device = Device::new(device_id.into()); + device.display_name = initial_device_display_name; + device.last_seen_ip = client_ip; + device.last_seen_ts = Some(MilliSecondsSinceUnixEpoch::now()); + + increment(&self.db.userid_devicelistversion, user_id.as_bytes()); + self.db.userdeviceid_metadata.put(key, Json(device)); + self.set_token(user_id, device_id, token).await + } + + /// 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 + .remove_dehydrated_device(user_id, Some(device_id)) + .await; + + let userdeviceid = (user_id, device_id); + + // Remove tokens + if let Ok(old_token) = self.db.userdeviceid_token.qry(&userdeviceid).await { + self.db.userdeviceid_token.del(userdeviceid); + self.db.token_userdeviceid.remove(&old_token); + } + + // Remove todevice events + let prefix = (user_id, device_id, Interfix); + self.db + .todeviceid_events + .keys_prefix_raw(&prefix) + .ignore_err() + .ready_for_each(|key| self.db.todeviceid_events.remove(key)) + .await; + + // TODO: Remove onetimekeys + + increment(&self.db.userid_devicelistversion, user_id.as_bytes()); + + self.db.userdeviceid_metadata.del(userdeviceid); + self.mark_device_key_update(user_id).await; + } + + /// Returns an iterator over all device ids of this user. + pub fn all_device_ids<'a>( + &'a self, + user_id: &'a UserId, + ) -> impl Stream + Send + 'a { + let prefix = (user_id, Interfix); + self.db + .userdeviceid_metadata + .keys_prefix(&prefix) + .ignore_err() + .map(|(_, device_id): (Ignore, OwnedDeviceId)| device_id) + } + + /// Gets the access token associated with a device. + pub async fn get_token( + &self, + user_id: &UserId, + device_id: &DeviceId, + ) -> conduwuit::Result { + let key = (user_id, device_id); + self.db.userdeviceid_token.qry(&key).await.deserialized() + } + + /// Generate a unique access token that doesn't collide with existing tokens + pub async fn generate_unique_token(&self) -> String { + loop { + let token = utils::random_string(32); + + // Check for collision with existing appservice and user tokens + let (appservice, usr) = tokio::join!( + self.services.appservice.find_from_token(&token), + self.db.token_userdeviceid.get(&token) + ); + if appservice.is_ok() || usr.is_ok() { + continue; + } + + return token; + } + } + + /// Replaces the access token of one device. + pub async fn set_token( + &self, + user_id: &UserId, + device_id: &DeviceId, + token: &str, + ) -> conduwuit::Result<()> { + let key = (user_id, device_id); + if self.db.userdeviceid_metadata.qry(&key).await.is_err() { + return Err!(Database(error!( + %user_id, + %device_id, + "User does not exist or device has no metadata." + ))); + } + + // Check for token collision with appservices + if self + .services + .appservice + .find_from_token(token) + .await + .is_ok() + { + return Err!(Request(InvalidParam( + "Token conflicts with an existing appservice token" + ))); + } + + // Remove old token + if let Ok(old_token) = self.db.userdeviceid_token.qry(&key).await { + self.db.token_userdeviceid.remove(&old_token); + // It will be removed from userdeviceid_token by the insert later + } + + // Assign token to user device combination + self.db.userdeviceid_token.put_raw(key, token); + self.db.token_userdeviceid.raw_put(token, key); + + Ok(()) + } + + /// Pushes a new to-device event into a device's inbox. + pub async fn add_to_device_event( + &self, + sender: &UserId, + target_user_id: &UserId, + target_device_id: &DeviceId, + event_type: &str, + content: serde_json::Value, + ) { + let count = self.services.globals.next_count().unwrap(); + + let key = (target_user_id, target_device_id, count); + self.db.todeviceid_events.put( + key, + Json(json!({ + "type": event_type, + "sender": sender, + "content": content, + })), + ); + } + + /// Gets all to-device events between the two counts. + pub fn get_to_device_events<'a>( + &'a self, + user_id: &'a UserId, + device_id: &'a DeviceId, + since: Option, + to: Option, + ) -> impl Stream)> + Send + 'a { + type Key = (OwnedUserId, OwnedDeviceId, u64); + + let from = (user_id, device_id, since.map_or(0, |since| since.saturating_add(1))); + + self.db + .todeviceid_events + .stream_from(&from) + .ignore_err() + .ready_take_while(move |((user_id_, device_id_, count), _): &(Key, _)| { + user_id == *user_id_ + && device_id == *device_id_ + && to.is_none_or(|to| *count <= to) + }) + .map(|((_, _, count), event)| (count, event)) + } + + /// Removes to-device events from the target device's inbox, until the given + /// count. + pub async fn remove_to_device_events( + &self, + user_id: &UserId, + device_id: &DeviceId, + until: Until, + ) where + Until: Into> + Send, + { + type Key = (OwnedUserId, OwnedDeviceId, u64); + + let until = until.into().unwrap_or(u64::MAX); + let from = (user_id, device_id, until); + self.db + .todeviceid_events + .rev_keys_from(&from) + .ignore_err() + .ready_take_while(move |(user_id_, device_id_, _): &Key| { + user_id == *user_id_ && device_id == *device_id_ + }) + .ready_for_each(|key: Key| { + self.db.todeviceid_events.del(key); + }) + .await; + } + + /// Updates device metadata and increments the device list version. + pub async fn update_device_metadata( + &self, + user_id: &UserId, + device_id: &DeviceId, + device: &Device, + ) -> conduwuit::Result<()> { + increment(&self.db.userid_devicelistversion, user_id.as_bytes()); + self.update_device_metadata_no_increment(user_id, device_id, device) + } + + /// Updates device metadata without incrementing the device list version. + /// This is namely used for updating the last_seen_ip and last_seen_ts + /// values, as those do not need a device list version bump due to them not + /// being relevant to other consumers. + fn update_device_metadata_no_increment( + &self, + user_id: &UserId, + device_id: &DeviceId, + device: &Device, + ) -> conduwuit::Result<()> { + let key = (user_id, device_id); + self.db.userdeviceid_metadata.put(key, Json(device)); + + Ok(()) + } + + /// Updates the last seen timestamp for a device. Silently does nothing if + /// the last update was less than 10 seconds ago, or the device does not + /// exist. + pub async fn update_device_last_seen( + &self, + user_id: &UserId, + device_id: Option<&DeviceId>, + ip: IpAddr, + ) { + let now = MilliSecondsSinceUnixEpoch::now(); + if let Some(device_id) = device_id { + if let Ok(mut device) = self.get_device_metadata(user_id, device_id).await { + device.last_seen_ip = Some(ip.to_string()); + // If the last update was less than 10 seconds ago, don't update the timestamp + if let Some(prev) = device.last_seen_ts { + if now.get().saturating_sub(prev.get()) < uint!(10_000) { + return; + } + } + device.last_seen_ts = Some(now); + + self.update_device_metadata_no_increment(user_id, device_id, &device) + .ok(); + } + } + } + + /// Get device metadata. + pub async fn get_device_metadata( + &self, + user_id: &UserId, + device_id: &DeviceId, + ) -> conduwuit::Result { + self.db + .userdeviceid_metadata + .qry(&(user_id, device_id)) + .await + .deserialized() + } + + /// Gets the most recent device list version for a user. + pub async fn get_devicelist_version(&self, user_id: &UserId) -> conduwuit::Result { + self.db + .userid_devicelistversion + .get(user_id) + .await + .deserialized() + } + + /// Gets metadata for all devices belonging to the target user. + pub fn all_devices_metadata<'a>( + &'a self, + user_id: &'a UserId, + ) -> impl Stream + Send + 'a { + let key = (user_id, Interfix); + self.db + .userdeviceid_metadata + .stream_prefix(&key) + .ignore_err() + .map(|(_, val): (Ignore, Device)| val) + } +} diff --git a/src/service/users/filters.rs b/src/service/users/filters.rs new file mode 100644 index 000000000..bbe7aff29 --- /dev/null +++ b/src/service/users/filters.rs @@ -0,0 +1,26 @@ +use conduwuit::utils; +use database::{Deserialized, Json}; +use ruma::{UserId, api::client::filter::FilterDefinition}; + +impl super::Service { + /// Creates a new sync filter. Returns the filter id. + pub fn create_filter(&self, user_id: &UserId, filter: &FilterDefinition) -> String { + let filter_id = utils::random_string(4); + + // TODO: filters should be de-duplicated and also not per-user + let key = (user_id, &filter_id); + self.db.userfilterid_filter.put(key, Json(filter)); + + filter_id + } + + /// Fetches a filter from a filter ID belonging to a user. + pub async fn get_filter( + &self, + user_id: &UserId, + filter_id: &str, + ) -> conduwuit::Result { + let key = (user_id, filter_id); + self.db.userfilterid_filter.qry(&key).await.deserialized() + } +} diff --git a/src/service/users/keys.rs b/src/service/users/keys.rs new file mode 100644 index 000000000..b6699f88c --- /dev/null +++ b/src/service/users/keys.rs @@ -0,0 +1,532 @@ +use std::collections::BTreeMap; + +use conduwuit::{ + Err, Error, Result, err, + utils::{ReadyExt, stream::TryIgnore, string::Unquoted}, +}; +use database::{Deserialized, Ignore, Json}; +use futures::{Stream, StreamExt, TryFutureExt}; +use ruma::{ + DeviceId, OneTimeKeyAlgorithm, OneTimeKeyId, OneTimeKeyName, OwnedKeyId, OwnedOneTimeKeyId, + OwnedUserId, RoomId, UInt, UserId, + api::error::ErrorKind, + encryption::{CrossSigningKey, DeviceKeys, OneTimeKey}, + serde::Raw, +}; + +use crate::users::{clean_signatures, parse_master_key, parse_user_signing_key}; + +impl super::Service { + /// Adds a single one-time key to a device. + pub async fn add_one_time_key( + &self, + user_id: &UserId, + device_id: &DeviceId, + one_time_key_key: &OneTimeKeyId, + one_time_key_value: &Raw, + ) -> Result { + // All devices have metadata + // Only existing devices should be able to call this, but we shouldn't assert + // either... + let key = (user_id, device_id); + if self.db.userdeviceid_metadata.qry(&key).await.is_err() { + return Err!(Database(error!( + %user_id, + %device_id, + "User does not exist or device has no metadata." + ))); + } + + let mut key = user_id.as_bytes().to_vec(); + key.push(0xFF); + key.extend_from_slice(device_id.as_bytes()); + key.push(0xFF); + // TODO: Use DeviceKeyId::to_string when it's available (and update everything, + // because there are no wrapping quotation marks anymore) + key.extend_from_slice( + serde_json::to_string(one_time_key_key) + .expect("DeviceKeyId::to_string always works") + .as_bytes(), + ); + + self.db + .onetimekeyid_onetimekeys + .raw_put(key, Json(one_time_key_value)); + + let count = self.services.globals.next_count().unwrap(); + self.db.userid_lastonetimekeyupdate.raw_put(user_id, count); + + Ok(()) + } + + /// Save a fallback key for the given user, device, and algorithm + /// This key will replace an existing fallback key. + pub async fn add_fallback_key( + &self, + user_id: &UserId, + device_id: &DeviceId, + fallback_key_id: &OneTimeKeyId, + fallback_key: &Raw, + used: bool, + ) -> Result { + // All devices have metadata + // Only existing devices should be able to call this, but we shouldn't assert + // either... + let key = (user_id, device_id); + if self.db.userdeviceid_metadata.qry(&key).await.is_err() { + return Err!(Database(error!( + %user_id, + %device_id, + "User does not exist or device has no metadata." + ))); + } + + // There is one fallback key slot per user, per device, per algorithm + // Therefore we use this as the DB key for this column + let db_key = (user_id, device_id, fallback_key_id.algorithm()); + + self.db + .fallbackkeyid_fallbackkey + .put(db_key, (used, fallback_key_id.as_str(), Json(fallback_key))); + + Ok(()) + } + + /// Returns the timestamp at when the last OTK update for the user was, or 0 + /// if the keys have never been updated. + pub async fn last_one_time_keys_update(&self, user_id: &UserId) -> u64 { + self.db + .userid_lastonetimekeyupdate + .get(user_id) + .await + .deserialized() + .unwrap_or(0) + } + + /// Consumes a one-time key belonging to the device of a given algorithm. If + /// no one-time keys are available, a fallback key is returned instead (if + /// available). If neither an OTK nor fallback key are available, NotFound + /// is returned. + pub async fn take_one_time_key( + &self, + user_id: &UserId, + device_id: &DeviceId, + key_algorithm: &OneTimeKeyAlgorithm, + ) -> Result<(OwnedKeyId, Raw)> { + let count = self.services.globals.next_count()?.to_be_bytes(); + self.db.userid_lastonetimekeyupdate.insert(user_id, count); + + let mut prefix = user_id.as_bytes().to_vec(); + prefix.push(0xFF); + prefix.extend_from_slice(device_id.as_bytes()); + prefix.push(0xFF); + prefix.push(b'"'); // Annoying quotation mark + prefix.extend_from_slice(key_algorithm.as_ref().as_bytes()); + prefix.push(b':'); + + let one_time_key = self + .db + .onetimekeyid_onetimekeys + .raw_stream_prefix(&prefix) + .ignore_err() + .next() + .await + .map(|(key, val)| { + self.db.onetimekeyid_onetimekeys.remove(key); + + let key = key + .rsplit(|&b| b == 0xFF) + .next() + .ok_or_else(|| err!(Database("OneTimeKeyId in db is invalid."))) + .unwrap(); + + let key = serde_json::from_slice(key) + .map_err(|e| err!(Database("OneTimeKeyId in db is invalid. {e}"))) + .unwrap(); + + let val = serde_json::from_slice(val) + .map_err(|e| err!(Database("OneTimeKeys in db are invalid. {e}"))) + .unwrap(); + + (key, val) + }); + + if let Some(result) = one_time_key { + return Ok(result); + } + + // No one-time key has been found. Look for a fallback key. + + let db_key = (user_id, device_id, key_algorithm); + + let fallback_key = self + .db + .fallbackkeyid_fallbackkey + .qry(&db_key) + .await + .ok() + .and_then(|handle| { + handle + .deserialized::<(bool, OwnedOneTimeKeyId, Raw)>() + .ok() + }); + + if let Some((used, fallback_key_id, fallback_key_value)) = fallback_key { + if !used { + // write the key to the database again to mark it as used + self.add_fallback_key( + user_id, + device_id, + &fallback_key_id, + &fallback_key_value, + true, + ) + .await?; + } + return Ok((fallback_key_id, fallback_key_value)); + } + + Err(err!(Request(NotFound("No one-time key or fallback key found")))) + } + + /// Returns the number of one-time keys the given device has. Does not count + /// fallback keys. + pub async fn count_one_time_keys( + &self, + user_id: &UserId, + device_id: &DeviceId, + ) -> BTreeMap { + type KeyVal<'a> = ((Ignore, Ignore, &'a Unquoted), Ignore); + + let mut algorithm_counts = BTreeMap::::new(); + let query = (user_id, device_id); + self.db + .onetimekeyid_onetimekeys + .stream_prefix(&query) + .ignore_err() + .ready_for_each(|((_, _, device_key_id), _): KeyVal<'_>| { + let one_time_key_id: &OneTimeKeyId = device_key_id + .as_str() + .try_into() + .expect("Invalid DeviceKeyID in database"); + + let count: &mut UInt = algorithm_counts + .entry(one_time_key_id.algorithm()) + .or_default(); + + *count = count.saturating_add(1_u32.into()); + }) + .await; + + algorithm_counts + } + + /// Returns a list of *unused* fallback key types. + pub async fn list_unused_fallback_key_types( + &self, + user_id: &UserId, + device_id: &DeviceId, + ) -> Vec { + type KeyVal = ((String, String, OneTimeKeyAlgorithm), (bool, String, Ignore)); + + let mut query = user_id.as_bytes().to_vec(); + query.push(0xFF); + query.extend_from_slice(device_id.as_bytes()); + query.push(0xFF); + + let mut unused_algorithms = Vec::new(); + + self.db + .fallbackkeyid_fallbackkey + .stream_prefix(&query) + .ignore_err() + .ready_for_each(|((_, _, fallback_key_algorithm), (used, ..)): KeyVal| { + if !used { + unused_algorithms.push(fallback_key_algorithm); + } + }) + .await; + + unused_algorithms + } + + /// Adds device identity keys to a device, overwriting existing ones if they + /// exist. Dispatches a device key update. + pub async fn add_device_keys( + &self, + user_id: &UserId, + device_id: &DeviceId, + device_keys: &Raw, + ) { + let key = (user_id, device_id); + + self.db.keyid_key.put(key, Json(device_keys)); + self.mark_device_key_update(user_id).await; + } + + /// Adds cross-signing keys for the given user. If notify is true, a device + /// key update is dispatched. + pub async fn add_cross_signing_keys( + &self, + user_id: &UserId, + master_key: &Option>, + self_signing_key: &Option>, + user_signing_key: &Option>, + notify: bool, + ) -> Result<()> { + let mut prefix = user_id.as_bytes().to_vec(); + prefix.push(0xFF); + + if let Some(master_key) = master_key { + let (master_key_key, _) = parse_master_key(user_id, master_key)?; + + self.db + .keyid_key + .insert(&master_key_key, master_key.json().get().as_bytes()); + + self.db + .userid_masterkeyid + .insert(user_id.as_bytes(), &master_key_key); + } + + // Self-signing key + if let Some(self_signing_key) = self_signing_key { + let mut self_signing_key_ids = self_signing_key + .deserialize() + .map_err(|e| err!(Request(InvalidParam("Invalid self signing key: {e:?}"))))? + .keys + .into_values(); + + let self_signing_key_id = self_signing_key_ids.next().ok_or(Error::BadRequest( + ErrorKind::InvalidParam, + "Self signing key contained no key.", + ))?; + + if self_signing_key_ids.next().is_some() { + return Err(Error::BadRequest( + ErrorKind::InvalidParam, + "Self signing key contained more than one key.", + )); + } + + let mut self_signing_key_key = prefix.clone(); + self_signing_key_key.extend_from_slice(self_signing_key_id.as_bytes()); + + self.db + .keyid_key + .insert(&self_signing_key_key, self_signing_key.json().get().as_bytes()); + + self.db + .userid_selfsigningkeyid + .insert(user_id.as_bytes(), &self_signing_key_key); + } + + // User-signing key + if let Some(user_signing_key) = user_signing_key { + let user_signing_key_id = parse_user_signing_key(user_signing_key)?; + + let user_signing_key_key = (user_id, &user_signing_key_id); + self.db + .keyid_key + .put_raw(user_signing_key_key, user_signing_key.json().get().as_bytes()); + + self.db + .userid_usersigningkeyid + .raw_put(user_id, user_signing_key_key); + } + + if notify { + self.mark_device_key_update(user_id).await; + } + + Ok(()) + } + + /// Appends a new signature to the given key ID. Dispatches a new device key + /// update. + pub async fn sign_key( + &self, + target_id: &UserId, + key_id: &str, + signature: (String, String), + sender_id: &UserId, + ) -> Result { + let key = (target_id, key_id); + + let mut cross_signing_key: serde_json::Value = self + .db + .keyid_key + .qry(&key) + .await + .map_err(|_| err!(Request(InvalidParam("Tried to sign nonexistent key"))))? + .deserialized() + .map_err(|e| err!(Database(debug_warn!("key in keyid_key is invalid: {e:?}"))))?; + + let signatures = cross_signing_key + .get_mut("signatures") + .ok_or_else(|| { + err!(Database(debug_warn!("key in keyid_key has no signatures field"))) + })? + .as_object_mut() + .ok_or_else(|| { + err!(Database(debug_warn!("key in keyid_key has invalid signatures field."))) + })? + .entry(sender_id.to_string()) + .or_insert_with(|| serde_json::Map::new().into()); + + signatures + .as_object_mut() + .ok_or_else(|| { + err!(Database(debug_warn!("signatures in keyid_key for a user is invalid."))) + })? + .insert(signature.0, signature.1.into()); + + let key = (target_id, key_id); + self.db.keyid_key.put(key, Json(cross_signing_key)); + + self.mark_device_key_update(target_id).await; + + Ok(()) + } + + /// Returns a stream of changed keys between the two counts. + #[inline] + pub fn keys_changed<'a>( + &'a self, + user_id: &'a UserId, + from: Option, + to: Option, + ) -> impl Stream + Send + 'a { + self.keys_changed_user_or_room(user_id.as_str(), from, to) + .map(|(user_id, ..)| user_id) + } + + /// Returns a stream of changed keys between the two counts in the given + /// room specifically. + #[inline] + pub fn room_keys_changed<'a>( + &'a self, + room_id: &'a RoomId, + from: Option, + to: Option, + ) -> impl Stream + Send + 'a { + self.keys_changed_user_or_room(room_id.as_str(), from, to) + } + + fn keys_changed_user_or_room<'a>( + &'a self, + user_or_room_id: &'a str, + from: Option, + to: Option, + ) -> impl Stream + Send + 'a { + type KeyVal<'a> = ((&'a str, u64), OwnedUserId); + + let from = from.unwrap_or(0); + let to = to.unwrap_or(u64::MAX); + let start = (user_or_room_id, from.saturating_add(1)); + self.db + .keychangeid_userid + .stream_from(&start) + .ignore_err() + .ready_take_while(move |((prefix, count), _): &KeyVal<'_>| { + *prefix == user_or_room_id && *count <= to + }) + .map(|((_, count), user_id): KeyVal<'_>| (user_id, count)) + } + + /// Marks that a user's device keys have been updated, so that other users + /// can be notified of the change. + pub async fn mark_device_key_update(&self, user_id: &UserId) { + let count = self.services.globals.next_count().unwrap(); + + self.services + .state_cache + .rooms_joined(user_id) + // Don't send key updates to unencrypted rooms + .filter_map(async |room_id| { + if self.services.state_accessor.is_encrypted_room(&room_id).await { + Some(room_id) + } else { + None + } + }) + .ready_for_each(|room_id| { + let key = (room_id, count); + self.db.keychangeid_userid.put_raw(key, user_id); + }) + .await; + + let key = (user_id, count); + self.db.keychangeid_userid.put_raw(key, user_id); + } + + /// Returns the device identity keys for a given device. + pub async fn get_device_keys<'a>( + &'a self, + user_id: &'a UserId, + device_id: &DeviceId, + ) -> Result> { + let key_id = (user_id, device_id); + self.db.keyid_key.qry(&key_id).await.deserialized() + } + + /// Gets a specific cross-signing key. + pub async fn get_key( + &self, + key_id: &[u8], + sender_user: Option<&UserId>, + user_id: &UserId, + allowed_signatures: &F, + ) -> Result> + where + F: Fn(&UserId) -> bool + Send + Sync, + { + let key: serde_json::Value = self.db.keyid_key.get(key_id).await.deserialized()?; + + let cleaned = clean_signatures(key, sender_user, user_id, allowed_signatures)?; + let raw_value = serde_json::value::to_raw_value(&cleaned)?; + Ok(Raw::from_json(raw_value)) + } + + /// Gets a user's master cross-signing key. + pub async fn get_master_key( + &self, + sender_user: Option<&UserId>, + user_id: &UserId, + allowed_signatures: &F, + ) -> Result> + where + F: Fn(&UserId) -> bool + Send + Sync, + { + let key_id = self.db.userid_masterkeyid.get(user_id).await?; + + self.get_key(&key_id, sender_user, user_id, allowed_signatures) + .await + } + + /// Gets a user's self-signing cross-signing key. + pub async fn get_self_signing_key( + &self, + sender_user: Option<&UserId>, + user_id: &UserId, + allowed_signatures: &F, + ) -> Result> + where + F: Fn(&UserId) -> bool + Send + Sync, + { + let key_id = self.db.userid_selfsigningkeyid.get(user_id).await?; + + self.get_key(&key_id, sender_user, user_id, allowed_signatures) + .await + } + + /// Gets a user's user-signing cross-signing key. + pub async fn get_user_signing_key(&self, user_id: &UserId) -> Result> { + self.db + .userid_usersigningkeyid + .get(user_id) + .and_then(|key_id| self.db.keyid_key.get(&*key_id)) + .await + .deserialized() + } +} diff --git a/src/service/users/mod.rs b/src/service/users/mod.rs index 429facfd1..58916b001 100644 --- a/src/service/users/mod.rs +++ b/src/service/users/mod.rs @@ -1,30 +1,19 @@ +pub(super) mod account; pub(super) mod dehydrated_device; +pub(super) mod device; +pub(super) mod filters; +pub(super) mod keys; +pub(super) mod profile; -use std::{collections::BTreeMap, mem, net::IpAddr, sync::Arc}; +use std::{mem, sync::Arc}; use conduwuit::{ - Err, Error, Result, Server, debug_error, debug_warn, err, trace, - utils::{self, ReadyExt, stream::TryIgnore, string::Unquoted}, + Err, Error, Result, Server, err, + utils::{self}, }; -use database::{Deserialized, Ignore, Interfix, Json, Map}; -use futures::{Stream, StreamExt, TryFutureExt}; -use ruma::{ - DeviceId, MilliSecondsSinceUnixEpoch, OneTimeKeyAlgorithm, OneTimeKeyId, OneTimeKeyName, - OwnedDeviceId, OwnedKeyId, OwnedMxcUri, OwnedOneTimeKeyId, OwnedUserId, RoomId, UInt, UserId, - api::{ - client::{device::Device, filter::FilterDefinition}, - error::ErrorKind, - }, - encryption::{CrossSigningKey, DeviceKeys, OneTimeKey}, - events::{ - AnyToDeviceEvent, GlobalAccountDataEventType, ignored_user_list::IgnoredUserListEvent, - }, - serde::Raw, - uint, -}; -use ruminuwuity::invite_permission_config::{FilterLevel, InvitePermissionConfigEvent}; +use database::Map; +use ruma::{UserId, api::error::ErrorKind, encryption::CrossSigningKey, serde::Raw}; use serde::{Deserialize, Serialize}; -use serde_json::json; use crate::{Dep, account_data, admin, appservice, globals, rooms}; @@ -138,1298 +127,6 @@ fn build(args: crate::Args<'_>) -> Result> { fn name(&self) -> &str { crate::service::make_name(std::module_path!()) } } -impl Service { - /// Returns true/false based on whether the recipient/receiving user has - /// ignored the sender. - pub async fn user_is_ignored(&self, sender_user: &UserId, recipient_user: &UserId) -> bool { - self.services - .account_data - .get_global(recipient_user, GlobalAccountDataEventType::IgnoredUserList) - .await - .is_ok_and(|ignored: IgnoredUserListEvent| { - ignored - .content - .ignored_users - .keys() - .any(|blocked_user| blocked_user == sender_user) - }) - } - - /// Returns the recipient's filter level for an invite from the sender. - pub async fn invite_filter_level( - &self, - sender_user: &UserId, - recipient_user: &UserId, - ) -> FilterLevel { - if self.user_is_ignored(sender_user, recipient_user).await { - FilterLevel::Ignore - } else { - let (stable, unstable) = tokio::join!( - self.services - .account_data - .get_global::( - recipient_user, - GlobalAccountDataEventType::InvitePermissionConfig - ), - self.services - .account_data - .get_global::( - recipient_user, - "org.matrix.msc4155.invite_permission_config".into() - ) // TODO: MSC4155 probably needs upstreaming to ruma at some point - ); - if stable.is_err() && unstable.is_err() { - return FilterLevel::Allow; - } - stable - .unwrap_or_else(|_| unstable.unwrap()) - .content - .user_filter_level(sender_user) - } - } - - /// Check if a user is an admin - #[inline] - pub async fn is_admin(&self, user_id: &UserId) -> bool { - self.services.admin.user_is_admin(user_id).await - } - - /// Create a new user account on this homeserver. - #[inline] - pub async 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"); - } - - self.set_password(user_id, password); - - Ok(()) - } - - // /// Create a new account for a local human or bot user. - // pub async fn create_local_account( - // &self, - // username: String, - // password: - // ) - - /// Deactivates an account, removing all of their device IDs and unsetting - /// their password. - pub async fn deactivate_account(&self, user_id: &UserId) -> Result<()> { - // 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); - - // TODO: Unhook 3PID - 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); - } - - /// Locks an account, preventing it being used until it is unlocked. - pub async fn lock_account(&self, user_id: &UserId, locking_user: &UserId) { - // NOTE: Locking is basically just suspension with a more severe effect, - // so we'll just re-use the suspension data structure to store the lock state. - let suspension = self - .db - .userid_lock - .get(user_id) - .await - .deserialized::() - .unwrap_or_else(|_| UserSuspension { - suspended: true, - suspended_at: MilliSecondsSinceUnixEpoch::now().get().into(), - suspended_by: locking_user.to_string(), - }); - - self.db.userid_lock.raw_put(user_id, Json(suspension)); - } - - /// 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 - .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) - }, - } - } - - /// Returns true if the user is locked. Returns false if the user does not - /// exist or is not locked. - pub async fn is_locked(&self, user_id: &UserId) -> Result { - match self - .db - .userid_lock - .get(user_id) - .await - .deserialized::() - { - | Ok(s) => Ok(s.suspended), - | Err(e) => - if e.is_not_found() { - Ok(false) - } else { - Err(e) - }, - } - } - - /// Disables login for a user, preventing them from creating new devices, - /// but allows them to continue using their existing sessions unimpeded. - pub fn disable_login(&self, user_id: &UserId) { - self.db.userid_logindisabled.insert(user_id, ""); - } - - /// Re-enables login for a user, allowing them to create new devices again. - pub fn enable_login(&self, user_id: &UserId) { self.db.userid_logindisabled.remove(user_id); } - - /// Returns true if the target user's login is disabled. - pub async fn is_login_disabled(&self, user_id: &UserId) -> bool { - self.db - .userid_logindisabled - .exists(user_id.as_str()) - .await - .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) - } - - /// 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 - } - - /// Returns the number of users registered on this server, including - /// deactivated users. - #[inline] - pub async fn count(&self) -> usize { self.db.userid_password.count().await } - - /// Find out which user an access token belongs to. Will panic if the access - /// token is empty. - pub async fn find_from_token(&self, token: &str) -> Result<(OwnedUserId, OwnedDeviceId)> { - assert!(!token.is_empty(), "Empty access token"); - self.db.token_userdeviceid.get(token).await.deserialized() - } - - /// Returns an iterator over all users on this homeserver. - pub fn stream(&self) -> impl Stream + Send { - self.db.userid_password.keys().ignore_err() - } - - /// Returns a list of active local users. - /// - /// A user account is considered `local` if the associated password is not - /// empty. - pub fn list_local_users(&self) -> impl Stream + Send + '_ { - self.db - .userid_password - .stream() - .ignore_err() - .ready_filter_map(|(u, p): (OwnedUserId, &[u8])| (!p.is_empty()).then_some(u)) - } - - /// 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""); - } - } - - /// Check a user's password. - pub async fn check_password(&self, user_id: &UserId, password: &str) -> Result { - let (hash, user_id): (String, OwnedUserId) = - if let Ok(hash) = self.db.userid_password.get(user_id).await.deserialized() { - (hash, user_id.to_owned()) - } else { - // We also check the lowercased version of the user ID to handle legacy user IDs - // better - let lowercase_user_id = UserId::parse(user_id.as_str().to_lowercase()).unwrap(); - - if let Ok(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."))); - } - }; - - if hash.is_empty() { - return Err!(Request(UserDeactivated("This user is deactivated"))); - } - - utils::hash::verify_password(password, &hash) - .inspect_err(|e| debug_error!("{e}")) - .map_err(|_| err!(Request(Forbidden("Invalid identifier or password."))))?; - - Ok(user_id) - } - - /// Returns the displayname of a user on this homeserver. - pub async fn displayname(&self, user_id: &UserId) -> Result { - self.db.userid_displayname.get(user_id).await.deserialized() - } - - /// Sets a new displayname or removes it if displayname is None. You still - /// need to notify all rooms of this change. - pub fn set_displayname(&self, user_id: &UserId, displayname: Option) { - if let Some(displayname) = displayname { - self.db.userid_displayname.insert(user_id, displayname); - } else { - self.db.userid_displayname.remove(user_id); - } - } - - /// Get the `avatar_url` of a user. - pub async fn avatar_url(&self, user_id: &UserId) -> Result { - self.db.userid_avatarurl.get(user_id).await.deserialized() - } - - /// Sets a new avatar_url or removes it if avatar_url is None. - pub fn set_avatar_url(&self, user_id: &UserId, avatar_url: Option) { - match avatar_url { - | Some(avatar_url) => { - self.db.userid_avatarurl.insert(user_id, &avatar_url); - }, - | _ => { - self.db.userid_avatarurl.remove(user_id); - }, - } - } - - /// Adds a new device to a user. The user must exist, otherwise InvalidParam - /// is returned. - pub async fn create_device( - &self, - user_id: &UserId, - device_id: &DeviceId, - token: &str, - initial_device_display_name: Option, - client_ip: Option, - ) -> Result<()> { - if !self.exists(user_id).await { - return Err!(Request(InvalidParam(error!( - "Called create_device for non-existent user {user_id}" - )))); - } - - let key = (user_id, device_id); - let mut device = Device::new(device_id.into()); - device.display_name = initial_device_display_name; - device.last_seen_ip = client_ip; - device.last_seen_ts = Some(MilliSecondsSinceUnixEpoch::now()); - - increment(&self.db.userid_devicelistversion, user_id.as_bytes()); - self.db.userdeviceid_metadata.put(key, Json(device)); - self.set_token(user_id, device_id, token).await - } - - /// 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 _: Result<_> = self - .remove_dehydrated_device(user_id, Some(device_id)) - .await; - - let userdeviceid = (user_id, device_id); - - // Remove tokens - if let Ok(old_token) = self.db.userdeviceid_token.qry(&userdeviceid).await { - self.db.userdeviceid_token.del(userdeviceid); - self.db.token_userdeviceid.remove(&old_token); - } - - // Remove todevice events - let prefix = (user_id, device_id, Interfix); - self.db - .todeviceid_events - .keys_prefix_raw(&prefix) - .ignore_err() - .ready_for_each(|key| self.db.todeviceid_events.remove(key)) - .await; - - // TODO: Remove onetimekeys - - increment(&self.db.userid_devicelistversion, user_id.as_bytes()); - - self.db.userdeviceid_metadata.del(userdeviceid); - self.mark_device_key_update(user_id).await; - } - - /// Returns an iterator over all device ids of this user. - pub fn all_device_ids<'a>( - &'a self, - user_id: &'a UserId, - ) -> impl Stream + Send + 'a { - let prefix = (user_id, Interfix); - self.db - .userdeviceid_metadata - .keys_prefix(&prefix) - .ignore_err() - .map(|(_, device_id): (Ignore, OwnedDeviceId)| device_id) - } - - /// Gets the access token associated with a device. - 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() - } - - /// Generate a unique access token that doesn't collide with existing tokens - pub async fn generate_unique_token(&self) -> String { - loop { - let token = utils::random_string(32); - - // Check for collision with existing appservice and user tokens - let (appservice, usr) = tokio::join!( - self.services.appservice.find_from_token(&token), - self.db.token_userdeviceid.get(&token) - ); - if appservice.is_ok() || usr.is_ok() { - continue; - } - - return token; - } - } - - /// Replaces the access token of one device. - pub async fn set_token( - &self, - user_id: &UserId, - device_id: &DeviceId, - token: &str, - ) -> Result<()> { - let key = (user_id, device_id); - if self.db.userdeviceid_metadata.qry(&key).await.is_err() { - return Err!(Database(error!( - %user_id, - %device_id, - "User does not exist or device has no metadata." - ))); - } - - // Check for token collision with appservices - if self - .services - .appservice - .find_from_token(token) - .await - .is_ok() - { - return Err!(Request(InvalidParam( - "Token conflicts with an existing appservice token" - ))); - } - - // Remove old token - if let Ok(old_token) = self.db.userdeviceid_token.qry(&key).await { - self.db.token_userdeviceid.remove(&old_token); - // It will be removed from userdeviceid_token by the insert later - } - - // Assign token to user device combination - self.db.userdeviceid_token.put_raw(key, token); - self.db.token_userdeviceid.raw_put(token, key); - - Ok(()) - } - - /// Adds a single one-time key to a device. - pub async fn add_one_time_key( - &self, - user_id: &UserId, - device_id: &DeviceId, - one_time_key_key: &OneTimeKeyId, - one_time_key_value: &Raw, - ) -> Result { - // All devices have metadata - // Only existing devices should be able to call this, but we shouldn't assert - // either... - let key = (user_id, device_id); - if self.db.userdeviceid_metadata.qry(&key).await.is_err() { - return Err!(Database(error!( - %user_id, - %device_id, - "User does not exist or device has no metadata." - ))); - } - - let mut key = user_id.as_bytes().to_vec(); - key.push(0xFF); - key.extend_from_slice(device_id.as_bytes()); - key.push(0xFF); - // TODO: Use DeviceKeyId::to_string when it's available (and update everything, - // because there are no wrapping quotation marks anymore) - key.extend_from_slice( - serde_json::to_string(one_time_key_key) - .expect("DeviceKeyId::to_string always works") - .as_bytes(), - ); - - self.db - .onetimekeyid_onetimekeys - .raw_put(key, Json(one_time_key_value)); - - let count = self.services.globals.next_count().unwrap(); - self.db.userid_lastonetimekeyupdate.raw_put(user_id, count); - - Ok(()) - } - - /// Save a fallback key for the given user, device, and algorithm - /// This key will replace an existing fallback key. - pub async fn add_fallback_key( - &self, - user_id: &UserId, - device_id: &DeviceId, - fallback_key_id: &OneTimeKeyId, - fallback_key: &Raw, - used: bool, - ) -> Result { - // All devices have metadata - // Only existing devices should be able to call this, but we shouldn't assert - // either... - let key = (user_id, device_id); - if self.db.userdeviceid_metadata.qry(&key).await.is_err() { - return Err!(Database(error!( - %user_id, - %device_id, - "User does not exist or device has no metadata." - ))); - } - - // There is one fallback key slot per user, per device, per algorithm - // Therefore we use this as the DB key for this column - let db_key = (user_id, device_id, fallback_key_id.algorithm()); - - self.db - .fallbackkeyid_fallbackkey - .put(db_key, (used, fallback_key_id.as_str(), Json(fallback_key))); - - Ok(()) - } - - /// Returns the timestamp at when the last OTK update for the user was, or 0 - /// if the keys have never been updated. - pub async fn last_one_time_keys_update(&self, user_id: &UserId) -> u64 { - self.db - .userid_lastonetimekeyupdate - .get(user_id) - .await - .deserialized() - .unwrap_or(0) - } - - /// Consumes a one-time key belonging to the device of a given algorithm. If - /// no one-time keys are available, a fallback key is returned instead (if - /// available). If neither an OTK nor fallback key are available, NotFound - /// is returned. - pub async fn take_one_time_key( - &self, - user_id: &UserId, - device_id: &DeviceId, - key_algorithm: &OneTimeKeyAlgorithm, - ) -> Result<(OwnedKeyId, Raw)> { - let count = self.services.globals.next_count()?.to_be_bytes(); - self.db.userid_lastonetimekeyupdate.insert(user_id, count); - - let mut prefix = user_id.as_bytes().to_vec(); - prefix.push(0xFF); - prefix.extend_from_slice(device_id.as_bytes()); - prefix.push(0xFF); - prefix.push(b'"'); // Annoying quotation mark - prefix.extend_from_slice(key_algorithm.as_ref().as_bytes()); - prefix.push(b':'); - - let one_time_key = self - .db - .onetimekeyid_onetimekeys - .raw_stream_prefix(&prefix) - .ignore_err() - .next() - .await - .map(|(key, val)| { - self.db.onetimekeyid_onetimekeys.remove(key); - - let key = key - .rsplit(|&b| b == 0xFF) - .next() - .ok_or_else(|| err!(Database("OneTimeKeyId in db is invalid."))) - .unwrap(); - - let key = serde_json::from_slice(key) - .map_err(|e| err!(Database("OneTimeKeyId in db is invalid. {e}"))) - .unwrap(); - - let val = serde_json::from_slice(val) - .map_err(|e| err!(Database("OneTimeKeys in db are invalid. {e}"))) - .unwrap(); - - (key, val) - }); - - if let Some(result) = one_time_key { - return Ok(result); - } - - // No one-time key has been found. Look for a fallback key. - - let db_key = (user_id, device_id, key_algorithm); - - let fallback_key = self - .db - .fallbackkeyid_fallbackkey - .qry(&db_key) - .await - .ok() - .and_then(|handle| { - handle - .deserialized::<(bool, OwnedOneTimeKeyId, Raw)>() - .ok() - }); - - if let Some((used, fallback_key_id, fallback_key_value)) = fallback_key { - if !used { - // write the key to the database again to mark it as used - self.add_fallback_key( - user_id, - device_id, - &fallback_key_id, - &fallback_key_value, - true, - ) - .await?; - } - return Ok((fallback_key_id, fallback_key_value)); - } - - Err(err!(Request(NotFound("No one-time key or fallback key found")))) - } - - /// Returns the number of one-time keys the given device has. Does not count - /// fallback keys. - pub async fn count_one_time_keys( - &self, - user_id: &UserId, - device_id: &DeviceId, - ) -> BTreeMap { - type KeyVal<'a> = ((Ignore, Ignore, &'a Unquoted), Ignore); - - let mut algorithm_counts = BTreeMap::::new(); - let query = (user_id, device_id); - self.db - .onetimekeyid_onetimekeys - .stream_prefix(&query) - .ignore_err() - .ready_for_each(|((Ignore, Ignore, device_key_id), Ignore): KeyVal<'_>| { - let one_time_key_id: &OneTimeKeyId = device_key_id - .as_str() - .try_into() - .expect("Invalid DeviceKeyID in database"); - - let count: &mut UInt = algorithm_counts - .entry(one_time_key_id.algorithm()) - .or_default(); - - *count = count.saturating_add(1_u32.into()); - }) - .await; - - algorithm_counts - } - - /// Returns a list of *unused* fallback key types. - pub async fn list_unused_fallback_key_types( - &self, - user_id: &UserId, - device_id: &DeviceId, - ) -> Vec { - type KeyVal = ((String, String, OneTimeKeyAlgorithm), (bool, String, Ignore)); - - let mut query = user_id.as_bytes().to_vec(); - query.push(0xFF); - query.extend_from_slice(device_id.as_bytes()); - query.push(0xFF); - - let mut unused_algorithms = Vec::new(); - - self.db - .fallbackkeyid_fallbackkey - .stream_prefix(&query) - .ignore_err() - .ready_for_each(|((_, _, fallback_key_algorithm), (used, ..)): KeyVal| { - if !used { - unused_algorithms.push(fallback_key_algorithm); - } - }) - .await; - - unused_algorithms - } - - /// Adds device identity keys to a device, overwriting existing ones if they - /// exist. Dispatches a device key update. - pub async fn add_device_keys( - &self, - user_id: &UserId, - device_id: &DeviceId, - device_keys: &Raw, - ) { - let key = (user_id, device_id); - - self.db.keyid_key.put(key, Json(device_keys)); - self.mark_device_key_update(user_id).await; - } - - /// Adds cross-signing keys for the given user. If notify is true, a device - /// key update is dispatched. - pub async fn add_cross_signing_keys( - &self, - user_id: &UserId, - master_key: &Option>, - self_signing_key: &Option>, - user_signing_key: &Option>, - notify: bool, - ) -> Result<()> { - // TODO: Check signatures - let mut prefix = user_id.as_bytes().to_vec(); - prefix.push(0xFF); - - if let Some(master_key) = master_key { - let (master_key_key, _) = parse_master_key(user_id, master_key)?; - - self.db - .keyid_key - .insert(&master_key_key, master_key.json().get().as_bytes()); - - self.db - .userid_masterkeyid - .insert(user_id.as_bytes(), &master_key_key); - } - - // Self-signing key - if let Some(self_signing_key) = self_signing_key { - let mut self_signing_key_ids = self_signing_key - .deserialize() - .map_err(|e| err!(Request(InvalidParam("Invalid self signing key: {e:?}"))))? - .keys - .into_values(); - - let self_signing_key_id = self_signing_key_ids.next().ok_or(Error::BadRequest( - ErrorKind::InvalidParam, - "Self signing key contained no key.", - ))?; - - if self_signing_key_ids.next().is_some() { - return Err(Error::BadRequest( - ErrorKind::InvalidParam, - "Self signing key contained more than one key.", - )); - } - - let mut self_signing_key_key = prefix.clone(); - self_signing_key_key.extend_from_slice(self_signing_key_id.as_bytes()); - - self.db - .keyid_key - .insert(&self_signing_key_key, self_signing_key.json().get().as_bytes()); - - self.db - .userid_selfsigningkeyid - .insert(user_id.as_bytes(), &self_signing_key_key); - } - - // User-signing key - if let Some(user_signing_key) = user_signing_key { - let user_signing_key_id = parse_user_signing_key(user_signing_key)?; - - let user_signing_key_key = (user_id, &user_signing_key_id); - self.db - .keyid_key - .put_raw(user_signing_key_key, user_signing_key.json().get().as_bytes()); - - self.db - .userid_usersigningkeyid - .raw_put(user_id, user_signing_key_key); - } - - if notify { - self.mark_device_key_update(user_id).await; - } - - Ok(()) - } - - /// Appends a new signature to the given key ID. Dispatches a new device key - /// update. - pub async fn sign_key( - &self, - target_id: &UserId, - key_id: &str, - signature: (String, String), - sender_id: &UserId, - ) -> Result { - let key = (target_id, key_id); - - let mut cross_signing_key: serde_json::Value = self - .db - .keyid_key - .qry(&key) - .await - .map_err(|_| err!(Request(InvalidParam("Tried to sign nonexistent key"))))? - .deserialized() - .map_err(|e| err!(Database(debug_warn!("key in keyid_key is invalid: {e:?}"))))?; - - let signatures = cross_signing_key - .get_mut("signatures") - .ok_or_else(|| { - err!(Database(debug_warn!("key in keyid_key has no signatures field"))) - })? - .as_object_mut() - .ok_or_else(|| { - err!(Database(debug_warn!("key in keyid_key has invalid signatures field."))) - })? - .entry(sender_id.to_string()) - .or_insert_with(|| serde_json::Map::new().into()); - - signatures - .as_object_mut() - .ok_or_else(|| { - err!(Database(debug_warn!("signatures in keyid_key for a user is invalid."))) - })? - .insert(signature.0, signature.1.into()); - - let key = (target_id, key_id); - self.db.keyid_key.put(key, Json(cross_signing_key)); - - self.mark_device_key_update(target_id).await; - - Ok(()) - } - - /// Returns a stream of changed keys between the two counts. - #[inline] - pub fn keys_changed<'a>( - &'a self, - user_id: &'a UserId, - from: Option, - to: Option, - ) -> impl Stream + Send + 'a { - self.keys_changed_user_or_room(user_id.as_str(), from, to) - .map(|(user_id, ..)| user_id) - } - - /// Returns a stream of changed keys between the two counts in the given - /// room specifically. - #[inline] - pub fn room_keys_changed<'a>( - &'a self, - room_id: &'a RoomId, - from: Option, - to: Option, - ) -> impl Stream + Send + 'a { - self.keys_changed_user_or_room(room_id.as_str(), from, to) - } - - fn keys_changed_user_or_room<'a>( - &'a self, - user_or_room_id: &'a str, - from: Option, - to: Option, - ) -> impl Stream + Send + 'a { - type KeyVal<'a> = ((&'a str, u64), OwnedUserId); - - let from = from.unwrap_or(0); - let to = to.unwrap_or(u64::MAX); - let start = (user_or_room_id, from.saturating_add(1)); - self.db - .keychangeid_userid - .stream_from(&start) - .ignore_err() - .ready_take_while(move |((prefix, count), _): &KeyVal<'_>| { - *prefix == user_or_room_id && *count <= to - }) - .map(|((_, count), user_id): KeyVal<'_>| (user_id, count)) - } - - /// Marks that a user's device keys have been updated, so that other users - /// can be notified of the change. - pub async fn mark_device_key_update(&self, user_id: &UserId) { - let count = self.services.globals.next_count().unwrap(); - - self.services - .state_cache - .rooms_joined(user_id) - // Don't send key updates to unencrypted rooms - .filter_map(async |room_id| { - if self.services.state_accessor.is_encrypted_room(&room_id).await { - Some(room_id) - } else { - None - } - }) - .ready_for_each(|room_id| { - let key = (room_id, count); - self.db.keychangeid_userid.put_raw(key, user_id); - }) - .await; - - let key = (user_id, count); - self.db.keychangeid_userid.put_raw(key, user_id); - } - - /// Returns the device identity keys for a given device. - pub async fn get_device_keys<'a>( - &'a self, - user_id: &'a UserId, - device_id: &DeviceId, - ) -> Result> { - let key_id = (user_id, device_id); - self.db.keyid_key.qry(&key_id).await.deserialized() - } - - /// Gets a specific cross-signing key. - pub async fn get_key( - &self, - key_id: &[u8], - sender_user: Option<&UserId>, - user_id: &UserId, - allowed_signatures: &F, - ) -> Result> - where - F: Fn(&UserId) -> bool + Send + Sync, - { - let key: serde_json::Value = self.db.keyid_key.get(key_id).await.deserialized()?; - - let cleaned = clean_signatures(key, sender_user, user_id, allowed_signatures)?; - let raw_value = serde_json::value::to_raw_value(&cleaned)?; - Ok(Raw::from_json(raw_value)) - } - - /// Gets a user's master cross-signing key. - pub async fn get_master_key( - &self, - sender_user: Option<&UserId>, - user_id: &UserId, - allowed_signatures: &F, - ) -> Result> - where - F: Fn(&UserId) -> bool + Send + Sync, - { - let key_id = self.db.userid_masterkeyid.get(user_id).await?; - - self.get_key(&key_id, sender_user, user_id, allowed_signatures) - .await - } - - /// Gets a user's self-signing cross-signing key. - pub async fn get_self_signing_key( - &self, - sender_user: Option<&UserId>, - user_id: &UserId, - allowed_signatures: &F, - ) -> Result> - where - F: Fn(&UserId) -> bool + Send + Sync, - { - let key_id = self.db.userid_selfsigningkeyid.get(user_id).await?; - - self.get_key(&key_id, sender_user, user_id, allowed_signatures) - .await - } - - /// Gets a user's user-signing cross-signing key. - pub async fn get_user_signing_key(&self, user_id: &UserId) -> Result> { - self.db - .userid_usersigningkeyid - .get(user_id) - .and_then(|key_id| self.db.keyid_key.get(&*key_id)) - .await - .deserialized() - } - - /// Pushes a new to-device event into a device's inbox. - pub async fn add_to_device_event( - &self, - sender: &UserId, - target_user_id: &UserId, - target_device_id: &DeviceId, - event_type: &str, - content: serde_json::Value, - ) { - let count = self.services.globals.next_count().unwrap(); - - let key = (target_user_id, target_device_id, count); - self.db.todeviceid_events.put( - key, - Json(json!({ - "type": event_type, - "sender": sender, - "content": content, - })), - ); - } - - /// Gets all to-device events between the two counts. - pub fn get_to_device_events<'a>( - &'a self, - user_id: &'a UserId, - device_id: &'a DeviceId, - since: Option, - to: Option, - ) -> impl Stream)> + Send + 'a { - type Key = (OwnedUserId, OwnedDeviceId, u64); - - let from = (user_id, device_id, since.map_or(0, |since| since.saturating_add(1))); - - self.db - .todeviceid_events - .stream_from(&from) - .ignore_err() - .ready_take_while(move |((user_id_, device_id_, count), _): &(Key, _)| { - user_id == *user_id_ - && device_id == *device_id_ - && to.is_none_or(|to| *count <= to) - }) - .map(|((_, _, count), event)| (count, event)) - } - - /// Removes to-device events from the target device's inbox, until the given - /// count. - pub async fn remove_to_device_events( - &self, - user_id: &UserId, - device_id: &DeviceId, - until: Until, - ) where - Until: Into> + Send, - { - type Key = (OwnedUserId, OwnedDeviceId, u64); - - let until = until.into().unwrap_or(u64::MAX); - let from = (user_id, device_id, until); - self.db - .todeviceid_events - .rev_keys_from(&from) - .ignore_err() - .ready_take_while(move |(user_id_, device_id_, _): &Key| { - user_id == *user_id_ && device_id == *device_id_ - }) - .ready_for_each(|key: Key| { - self.db.todeviceid_events.del(key); - }) - .await; - } - - /// Updates device metadata and increments the device list version. - pub async fn update_device_metadata( - &self, - user_id: &UserId, - device_id: &DeviceId, - device: &Device, - ) -> Result<()> { - increment(&self.db.userid_devicelistversion, user_id.as_bytes()); - self.update_device_metadata_no_increment(user_id, device_id, device) - } - - /// Updates device metadata without incrementing the device list version. - /// This is namely used for updating the last_seen_ip and last_seen_ts - /// values, as those do not need a device list version bump due to them not - /// being relevant to other consumers. - fn update_device_metadata_no_increment( - &self, - user_id: &UserId, - device_id: &DeviceId, - device: &Device, - ) -> Result<()> { - let key = (user_id, device_id); - self.db.userdeviceid_metadata.put(key, Json(device)); - - Ok(()) - } - - /// Updates the last seen timestamp for a device. Silently does nothing if - /// the last update was less than 10 seconds ago, or the device does not - /// exist. - pub async fn update_device_last_seen( - &self, - user_id: &UserId, - device_id: Option<&DeviceId>, - ip: IpAddr, - ) { - let now = MilliSecondsSinceUnixEpoch::now(); - if let Some(device_id) = device_id { - if let Ok(mut device) = self.get_device_metadata(user_id, device_id).await { - device.last_seen_ip = Some(ip.to_string()); - // If the last update was less than 10 seconds ago, don't update the timestamp - if let Some(prev) = device.last_seen_ts { - if now.get().saturating_sub(prev.get()) < uint!(10_000) { - return; - } - } - device.last_seen_ts = Some(now); - - self.update_device_metadata_no_increment(user_id, device_id, &device) - .ok(); - } - } - } - - /// Get device metadata. - pub async fn get_device_metadata( - &self, - user_id: &UserId, - device_id: &DeviceId, - ) -> Result { - self.db - .userdeviceid_metadata - .qry(&(user_id, device_id)) - .await - .deserialized() - } - - /// Gets the most recent device list version for a user. - pub async fn get_devicelist_version(&self, user_id: &UserId) -> Result { - self.db - .userid_devicelistversion - .get(user_id) - .await - .deserialized() - } - - /// Gets metadata for all devices belonging to the target user. - pub fn all_devices_metadata<'a>( - &'a self, - user_id: &'a UserId, - ) -> impl Stream + Send + 'a { - let key = (user_id, Interfix); - self.db - .userdeviceid_metadata - .stream_prefix(&key) - .ignore_err() - .map(|(_, val): (Ignore, Device)| val) - } - - /// Creates a new sync filter. Returns the filter id. - pub fn create_filter(&self, user_id: &UserId, filter: &FilterDefinition) -> String { - let filter_id = utils::random_string(4); - - let key = (user_id, &filter_id); - self.db.userfilterid_filter.put(key, Json(filter)); - - filter_id - } - - /// Fetches a filter from a filter ID belonging to a user. - pub async fn get_filter( - &self, - user_id: &UserId, - filter_id: &str, - ) -> Result { - let key = (user_id, filter_id); - self.db.userfilterid_filter.qry(&key).await.deserialized() - } - - /// Creates an OpenID token, which can be used to prove that a user has - /// access to an account (primarily for integrations) - pub fn create_openid_token(&self, user_id: &UserId, token: &str) -> Result { - use std::num::Saturating as Sat; - - let expires_in = self.services.server.config.openid_token_ttl; - let expires_at = Sat(utils::millis_since_unix_epoch()) + Sat(expires_in) * Sat(1000); - - let mut value = expires_at.0.to_be_bytes().to_vec(); - value.extend_from_slice(user_id.as_bytes()); - - self.db - .openidtoken_expiresatuserid - .insert(token.as_bytes(), value.as_slice()); - - Ok(expires_in) - } - - /// Find out which user an OpenID access token belongs to. - pub async fn find_from_openid_token(&self, token: &str) -> Result { - let Ok(value) = self.db.openidtoken_expiresatuserid.get(token).await else { - return Err!(Request(Unauthorized("OpenID token is unrecognised"))); - }; - - let (expires_at_bytes, user_bytes) = value.split_at(0_u64.to_be_bytes().len()); - let expires_at = - u64::from_be_bytes(expires_at_bytes.try_into().map_err(|e| { - err!(Database("expires_at in openid_userid is invalid u64. {e}")) - })?); - - if expires_at < utils::millis_since_unix_epoch() { - debug_warn!("OpenID token is expired, removing"); - self.db.openidtoken_expiresatuserid.remove(token.as_bytes()); - - return Err!(Request(Unauthorized("OpenID token is expired"))); - } - - let user_string = utils::string_from_bytes(user_bytes) - .map_err(|e| err!(Database("User ID in openid_userid is invalid unicode. {e}")))?; - - OwnedUserId::try_from(user_string) - .map_err(|e| err!(Database("User ID in openid_userid is invalid. {e}"))) - } - - /// Creates a short-lived login token, which can be used to log in using the - /// `m.login.token` mechanism. - pub fn create_login_token(&self, user_id: &UserId, token: &str) -> u64 { - use std::num::Saturating as Sat; - - let expires_in = self.services.server.config.login_token_ttl; - let expires_at = Sat(utils::millis_since_unix_epoch()) + Sat(expires_in); - - let value = (expires_at.0, user_id); - self.db.logintoken_expiresatuserid.raw_put(token, value); - - expires_in - } - - /// Find out which user a login token belongs to. - /// Removes the token to prevent double-use attacks. - pub async fn find_from_login_token(&self, token: &str) -> Result { - let Ok(value) = self.db.logintoken_expiresatuserid.get(token).await else { - return Err!(Request(Forbidden("Login token is unrecognised"))); - }; - let (expires_at, user_id): (u64, OwnedUserId) = value.deserialized()?; - - if expires_at < utils::millis_since_unix_epoch() { - trace!(%user_id, ?token, "Removing expired login token"); - - self.db.logintoken_expiresatuserid.remove(token); - - return Err!(Request(Forbidden("Login token is expired"))); - } - - self.db.logintoken_expiresatuserid.remove(token); - - Ok(user_id) - } - - /// Gets a specific user profile key - pub async fn profile_key( - &self, - user_id: &UserId, - profile_key: &str, - ) -> Result { - let key = (user_id, profile_key); - self.db - .useridprofilekey_value - .qry(&key) - .await - .and_then(|handle| serde_json::from_slice(&handle).map_err(Into::into)) - } - - /// Gets all the user's profile keys and values in an iterator - pub fn all_profile_keys<'a>( - &'a self, - user_id: &'a UserId, - ) -> impl Stream + 'a + Send { - type KeyVal<'a> = ((Ignore, String), &'a [u8]); - - let prefix = (user_id, Interfix); - self.db - .useridprofilekey_value - .stream_prefix(&prefix) - .ignore_err() - .map(|((_, key), value): KeyVal<'_>| Ok((key, serde_json::from_slice(value)?))) - .ignore_err() - } - - /// Sets a new profile key value, removes the key if value is None - pub fn set_profile_key( - &self, - user_id: &UserId, - profile_key: &str, - profile_key_value: Option, - ) { - let key = (user_id, profile_key); - - if let Some(value) = profile_key_value { - self.db.useridprofilekey_value.put(key, Json(value)); - } else { - self.db.useridprofilekey_value.del(key); - } - } - - /// Clears all profile data for a user, including display name and avatar - /// url. - pub async fn clear_profile(&self, user_id: &UserId) { - self.set_displayname(user_id, None); - self.set_avatar_url(user_id, None); - self.all_profile_keys(user_id) - .ready_for_each(|(key, _)| self.set_profile_key(user_id, &key, None)) - .await; - } -} - pub fn parse_master_key( user_id: &UserId, master_key: &Raw, diff --git a/src/service/users/profile.rs b/src/service/users/profile.rs new file mode 100644 index 000000000..a4ed3ee3d --- /dev/null +++ b/src/service/users/profile.rs @@ -0,0 +1,94 @@ +use conduwuit::utils::{ReadyExt, stream::TryIgnore}; +use database::{Deserialized, Ignore, Interfix, Json}; +use futures::{Stream, StreamExt}; +use ruma::{OwnedMxcUri, UserId}; + +impl super::Service { + /// Returns the displayname of a user on this homeserver. + pub async fn displayname(&self, user_id: &UserId) -> conduwuit::Result { + self.db.userid_displayname.get(user_id).await.deserialized() + } + + /// Sets a new displayname or removes it if displayname is None. You still + /// need to notify all rooms of this change. + pub fn set_displayname(&self, user_id: &UserId, displayname: Option) { + if let Some(displayname) = displayname { + self.db.userid_displayname.insert(user_id, displayname); + } else { + self.db.userid_displayname.remove(user_id); + } + } + + /// Get the `avatar_url` of a user. + pub async fn avatar_url(&self, user_id: &UserId) -> conduwuit::Result { + self.db.userid_avatarurl.get(user_id).await.deserialized() + } + + /// Sets a new avatar_url or removes it if avatar_url is None. + pub fn set_avatar_url(&self, user_id: &UserId, avatar_url: Option) { + match avatar_url { + | Some(avatar_url) => { + self.db.userid_avatarurl.insert(user_id, &avatar_url); + }, + | _ => { + self.db.userid_avatarurl.remove(user_id); + }, + } + } + + /// Gets a specific user profile key + pub async fn profile_key( + &self, + user_id: &UserId, + profile_key: &str, + ) -> conduwuit::Result { + let key = (user_id, profile_key); + self.db + .useridprofilekey_value + .qry(&key) + .await + .and_then(|handle| serde_json::from_slice(&handle).map_err(Into::into)) + } + + /// Gets all the user's profile keys and values in an iterator + pub fn all_profile_keys<'a>( + &'a self, + user_id: &'a UserId, + ) -> impl Stream + 'a + Send { + type KeyVal<'a> = ((Ignore, String), &'a [u8]); + + let prefix = (user_id, Interfix); + self.db + .useridprofilekey_value + .stream_prefix(&prefix) + .ignore_err() + .map(|((_, key), value): KeyVal<'_>| Ok((key, serde_json::from_slice(value)?))) + .ignore_err() + } + + /// Sets a new profile key value, removes the key if value is None + pub fn set_profile_key( + &self, + user_id: &UserId, + profile_key: &str, + profile_key_value: Option, + ) { + let key = (user_id, profile_key); + + if let Some(value) = profile_key_value { + self.db.useridprofilekey_value.put(key, Json(value)); + } else { + self.db.useridprofilekey_value.del(key); + } + } + + /// Clears all profile data for a user, including display name and avatar + /// url. + pub async fn clear_profile(&self, user_id: &UserId) { + self.set_displayname(user_id, None); + self.set_avatar_url(user_id, None); + self.all_profile_keys(user_id) + .ready_for_each(|(key, _)| self.set_profile_key(user_id, &key, None)) + .await; + } +}