Merge branch 'main' into rewrite-resolver

src/service/Cargo.toml

@@ -121,6 +121,7 @@ yansi.workspace = true
 lettre.workspace = true
 resolvematrix.workspace = true
 serde_urlencoded.workspace = true
+openidconnect.workspace = true
 
 [target.'cfg(all(unix, target_os = "linux"))'.dependencies]
 sd-notify.workspace = true

src/service/admin/create.rs

@@ -48,7 +48,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
 
 	// Create a user for the server
 	let server_user = services.globals.server_user.as_ref();
-	services.users.create(server_user, None)?;
+	services.users.create_shadow_account(server_user).await?;
 
 	let mut create_content = if room_version_rules.authorization.use_room_create_sender {
 		RoomCreateEventContent::new_v1(server_user.into())

src/service/appservice/mod.rs

@@ -108,17 +108,17 @@ impl Service {
 			self.services.globals.server_name(),
 		)?;
 
-		if !self.services.users.exists(&appservice_user_id).await {
-			self.services.users.create(&appservice_user_id, None)?;
-		} else if self
+		if !self
 			.services
 			.users
-			.is_deactivated(&appservice_user_id)
+			.status(&appservice_user_id)
 			.await
-			.unwrap_or(false)
+			.is_found()
 		{
-			// Reactivate the appservice user if it was accidentally deactivated
-			self.services.users.set_password(&appservice_user_id, None);
+			self.services
+				.users
+				.create_shadow_account(&appservice_user_id)
+				.await?;
 		}
 
 		self.registration_info

src/service/emergency/mod.rs

@@ -54,15 +54,22 @@ impl Service {
 	async fn set_emergency_access(&self) -> Result {
 		let server_user = &self.services.globals.server_user;
 
-		self.services.users.set_password(
-			server_user,
-			self.services
-				.config
-				.emergency_password
-				.as_deref()
-				.map(HashedPassword::new)
-				.transpose()?,
-		);
+		match &self.services.config.emergency_password {
+			| Some(emergency_password) => {
+				let emergency_password = HashedPassword::new(emergency_password)?;
+
+				self.services
+					.users
+					.convert_to_local_account(server_user, emergency_password)
+					.await?;
+			},
+			| None => {
+				self.services
+					.users
+					.convert_to_shadow_account(server_user)
+					.await?;
+			},
+		}
 
 		let (ruleset, pwd_set) = match self.services.config.emergency_password {
 			| Some(_) => (Ruleset::server_default(server_user), true),

src/service/media/preview.rs

@@ -211,7 +211,6 @@ impl Service {
 		Ok(preview_data)
 	}
 
-	#[cfg(feature = "url_preview")]
 	pub async fn download_media(&self, url: &str) -> Result<(OwnedMxcUri, usize)> {
 		use conduwuit::utils::random_string;
 		use http::header::CONTENT_TYPE;
@@ -268,11 +267,6 @@ impl Service {
 		Err!(FeatureDisabled("url_preview"))
 	}
 
-	#[cfg(not(feature = "url_preview"))]
-	pub async fn download_media(&self, _url: &str) -> Result<UrlPreviewData> {
-		Err!(FeatureDisabled("url_preview"))
-	}
-
 	#[cfg(feature = "url_preview")]
 	async fn download_html(&self, url: &str) -> Result<UrlPreviewData> {
 		use webpage::HTML;

src/service/migrations.rs

@@ -41,7 +41,7 @@ pub(crate) async fn migrations(services: &Services) -> Result<()> {
 	// requires recreating the database from scratch.
 	if users_count > 0 {
 		let server_user = &services.globals.server_user;
-		if !services.users.exists(server_user).await {
+		if !services.users.status(server_user).await.is_found() {
 			error!("The {server_user} server user does not exist, and the database is not new.");
 			return Err!(Database(
 				"Cannot reuse an existing database after changing the server name, please \
@@ -228,6 +228,18 @@ async fn migrate(services: &Services) -> Result<()> {
 			.map_err(|e| err!("Failed to run 'fix_local_invite_state' migration': {e}"))?;
 	}
 
+	if services.globals.db.database_version().await < 18 {
+		services.globals.db.bump_database_version(18);
+		info!("Migration: Bumped database version to 18");
+	}
+
+	if db["global"].get(SPLIT_USERID_PASSWORD).await.is_not_found() {
+		info!("Running migration 'split_userid_password'");
+		split_userid_password(services)
+			.await
+			.map_err(|e| err!("Failed to run 'split_userid_password' migration': {e}"))?;
+	}
+
 	assert_eq!(
 		services.globals.db.database_version().await,
 		DATABASE_VERSION,
@@ -242,9 +254,9 @@ async fn migrate(services: &Services) -> Result<()> {
 		if !patterns.is_empty() {
 			services
 				.users
-				.stream()
+				.stream_local_users()
 				.filter_map(async |user_id| {
-					if services.users.is_active_local(&user_id).await {
+					if services.users.status(&user_id).await.is_found() {
 						Some(user_id)
 					} else {
 						None
@@ -774,7 +786,7 @@ async fn fix_local_invite_state(services: &Services) -> Result {
 
 	let db = &services.db;
 	let cork = db.cork_and_sync();
-	let userroomid_invitestate = services.db["userroomid_invitestate"].clone();
+	let userroomid_invitestate = db["userroomid_invitestate"].clone();
 
 	// for each user invited to a room
 	let fixed =  userroomid_invitestate.stream()
@@ -818,3 +830,44 @@ async fn fix_local_invite_state(services: &Services) -> Result {
 	db.db.sort()?;
 	Ok(())
 }
+
+const SPLIT_USERID_PASSWORD: &str = "split_userid_password";
+async fn split_userid_password(services: &Services) -> Result {
+	// Split remote and deactivated users out from the `userid_password` table
+
+	let db = &services.db;
+	let cork = db.cork_and_sync();
+	let userid_password = db["userid_password"].clone();
+	let remoteuserid_remoteuser = db["remoteuserid_remoteuser"].clone();
+	let userid_deactivated = db["userid_deactivated"].clone();
+
+	let remote_users = userid_password
+		.stream::<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(())
+}

src/service/mod.rs

@@ -28,6 +28,7 @@ pub mod mailer;
 pub mod media;
 pub mod moderation;
 pub mod oauth;
+pub mod oidc;
 pub mod presence;
 pub mod pusher;
 pub mod registration_tokens;

src/service/oauth/grant.rs

@@ -160,7 +160,7 @@ pub enum ErrorCode {
 	InvalidClientMetadata,
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, Deserialize)]
 pub struct AuthorizationCodeResponse {
 	pub state: String,
 	pub code: String,

src/service/oidc/mod.rs

@@ -0,0 +1,484 @@
+use std::{collections::HashMap, str::FromStr, sync::Arc, time::Duration};
+
+use async_trait::async_trait;
+use conduwuit::{
+	Result,
+	config::{OidcConfig, OidcProfileKeyImportMode},
+	debug, err, error, info, warn,
+};
+use database::{Deserialized, Map};
+use lettre::Address;
+use openidconnect::{
+	AdditionalClaims, AuthorizationCode, ClientSecret, CsrfToken, EmptyExtraTokenFields,
+	EndpointMaybeSet, EndpointNotSet, EndpointSet, IdTokenClaims, IdTokenFields, IssuerUrl,
+	Nonce, PkceCodeChallenge, PkceCodeVerifier, RedirectUrl, StandardErrorResponse,
+	StandardTokenResponse, TokenResponse,
+	core::{
+		CoreAuthDisplay, CoreAuthPrompt, CoreAuthenticationFlow, CoreErrorResponseType,
+		CoreGenderClaim, CoreJsonWebKey, CoreJweContentEncryptionAlgorithm,
+		CoreJwsSigningAlgorithm, CoreProviderMetadata, CoreRevocableToken,
+		CoreRevocationErrorResponse, CoreTokenIntrospectionResponse, CoreTokenType,
+	},
+	reqwest,
+};
+use ruma::{
+	OwnedUserId, UserId,
+	api::client::profile::PropagateTo,
+	profile::{ProfileFieldName, ProfileFieldValue},
+};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use tokio::{runtime, sync::SetOnce};
+use url::Url;
+
+use crate::{
+	Dep, config, globals, media,
+	oauth::grant::AuthorizationCodeResponse,
+	threepid,
+	users::{self, AccountStatus, ProfileFieldChange},
+};
+
+pub struct Service {
+	services: Services,
+	runtime: runtime::Handle,
+	db: Data,
+	client: Option<OidcClient>,
+}
+
+struct Data {
+	openidsubject_localpart: Arc<Map>,
+	openidsubject_currentpictureurl: Arc<Map>,
+}
+struct Services {
+	config: Dep<config::Service>,
+	globals: Dep<globals::Service>,
+	media: Dep<media::Service>,
+	threepid: Dep<threepid::Service>,
+	users: Dep<users::Service>,
+}
+
+struct OidcClient {
+	config: OidcConfig,
+	client_secret: ClientSecret,
+	machine: SetOnce<OidcClientMachine>,
+	client: reqwest::Client,
+}
+
+type OidcClientMachine = openidconnect::Client<
+	AllClaims,
+	CoreAuthDisplay,
+	CoreGenderClaim,
+	CoreJweContentEncryptionAlgorithm,
+	CoreJsonWebKey,
+	CoreAuthPrompt,
+	StandardErrorResponse<CoreErrorResponseType>,
+	StandardTokenResponse<
+		IdTokenFields<
+			AllClaims,
+			EmptyExtraTokenFields,
+			CoreGenderClaim,
+			CoreJweContentEncryptionAlgorithm,
+			CoreJwsSigningAlgorithm,
+		>,
+		CoreTokenType,
+	>,
+	CoreTokenIntrospectionResponse,
+	CoreRevocableToken,
+	CoreRevocationErrorResponse,
+	EndpointSet,
+	EndpointNotSet,
+	EndpointNotSet,
+	EndpointNotSet,
+	EndpointMaybeSet,
+	EndpointMaybeSet,
+>;
+
+pub type Claims = IdTokenClaims<AllClaims, CoreGenderClaim>;
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct AllClaims {
+	#[serde(flatten)]
+	pub claims: HashMap<String, Value>,
+}
+
+impl AdditionalClaims for AllClaims {}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct PendingSession {
+	pkce_verifier: PkceCodeVerifier,
+	nonce: Nonce,
+	csrf_token: CsrfToken,
+}
+
+pub enum SessionCompletionStatus {
+	NeedsUserId,
+	Complete(OwnedUserId),
+}
+
+#[async_trait]
+impl crate::Service for Service {
+	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
+		Ok(Arc::new(Self {
+			services: Services {
+				config: args.depend::<config::Service>("config"),
+                globals: args.depend::<globals::Service>("globals"),
+				media: args.depend::<media::Service>("media"),
+                threepid: args.depend::<threepid::Service>("threepid"),
+                users: args.depend::<users::Service>("users"),
+			},
+			runtime: args.server.runtime().clone(),
+            db: Data {
+                openidsubject_localpart: args.db["openidsubject_localpart"].clone(),
+				openidsubject_currentpictureurl: args.db["openidsubject_currentpictureurl"].clone(),
+            },
+            client: args.server.config.oauth.oidc.as_ref().map(|config| -> Result<OidcClient> {
+				Ok(OidcClient {
+                    config: config.clone(),
+					client_secret: if let Some(client_secret_file) = &config.client_secret_file {
+						std::fs::read_to_string(client_secret_file)
+							.map(ClientSecret::new)
+							.map_err(|err| err!("Failed to read OIDC client secret file: {err}"))?
+					} else if let Some(client_secret) = &config.client_secret {
+						client_secret.clone()
+					} else {
+						// The config check function should cause an early exit before this happens
+						panic!("neither client secret or client secret file were set");
+					},
+                    machine: SetOnce::new(),
+                    // This isn't in the client service because it has to use the `reqwest` shipped by `openidconnect`
+                    client: reqwest::ClientBuilder::new()
+                        .connect_timeout(Duration::from_secs(args.server.config.request_conn_timeout))
+                        .read_timeout(Duration::from_secs(args.server.config.request_timeout))
+                        .timeout(Duration::from_secs(args.server.config.request_total_timeout))
+                        .pool_idle_timeout(Duration::from_secs(args.server.config.request_idle_timeout))
+                        .pool_max_idle_per_host(args.server.config.request_idle_per_host.into())
+                        .user_agent(conduwuit::user_agent())
+                        .redirect(reqwest::redirect::Policy::none())
+                        .danger_accept_invalid_certs(args.server.config.allow_invalid_tls_certificates_yes_i_know_what_the_fuck_i_am_doing_with_this_and_i_know_this_is_insecure)
+                        .build()
+                        .expect("client should build")
+                })}
+			).transpose()?,
+		}))
+	}
+
+	async fn worker(self: Arc<Self>) -> Result {
+		if let Some(OidcClient { config, client_secret, machine, client }) = &self.client {
+			let redirect_url = self
+				.services
+				.config
+				.get_client_domain()
+				.join(&format!("{}/oidc/complete", conduwuit::ROUTE_PREFIX))
+				.expect("redirect url should be valid");
+
+			let provider_metadata = CoreProviderMetadata::discover_async(
+				IssuerUrl::from_url(config.discovery_url.clone()),
+				client,
+			)
+			.await
+			.map_err(|err| err!("Failed to discover OIDC provider metadata: {err}"))?;
+
+			machine
+				.set(
+					OidcClientMachine::from_provider_metadata(
+						provider_metadata,
+						config.client_id.clone(),
+						Some(client_secret.clone()),
+					)
+					.set_redirect_uri(RedirectUrl::from_url(redirect_url)),
+				)
+				.expect("machine should be empty");
+		}
+
+		Ok(())
+	}
+
+	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
+}
+
+impl Service {
+	const SERVER_MISCONFIGURED: &str =
+		"Identity server is misconfigured. Contact your homeserver's administrator.";
+
+	pub fn enabled(&self) -> bool { self.client.is_some() }
+
+	pub fn restricted_profile_fields(&self) -> Vec<ProfileFieldName> {
+		if let Some(config) = self.client.as_ref().map(|client| &client.config)
+			&& matches!(config.profile_key_import_mode, OidcProfileKeyImportMode::OnLogin)
+		{
+			config
+				.profile_key_map
+				.keys()
+				.map(|key| ProfileFieldName::from(key.as_str()))
+				.collect()
+		} else {
+			vec![]
+		}
+	}
+
+	pub async fn begin_session(&self, prompt: Option<CoreAuthPrompt>) -> (PendingSession, Url) {
+		let OidcClient { machine, config, .. } =
+			self.client.as_ref().expect("oidc should be configured");
+		let machine = machine.wait().await;
+
+		let (pkce_challenge, pkce_verifier) = PkceCodeChallenge::new_random_sha256();
+
+		let mut auth_url = machine
+			.authorize_url(
+				CoreAuthenticationFlow::AuthorizationCode,
+				CsrfToken::new_random,
+				Nonce::new_random,
+			)
+			.add_scopes(config.additional_scopes.iter().cloned())
+			.set_pkce_challenge(pkce_challenge);
+
+		if let Some(prompt) = prompt {
+			auth_url = auth_url.add_prompt(prompt);
+		}
+
+		let (auth_url, csrf_token, nonce) = auth_url.url();
+
+		(PendingSession { pkce_verifier, nonce, csrf_token }, auth_url)
+	}
+
+	pub async fn exchange_code(
+		&self,
+		session: PendingSession,
+		response: AuthorizationCodeResponse,
+	) -> Result<Claims, &'static str> {
+		let Some(OidcClient { machine, client, .. }) = self.client.as_ref() else {
+			return Err("Delegated authentication is not enabled on this server.");
+		};
+
+		let machine = machine.wait().await;
+
+		if session.csrf_token.into_secret() != response.state {
+			return Err("State mismatch.");
+		}
+
+		let token_response = machine
+			.exchange_code(AuthorizationCode::new(response.code))
+			.expect("machine should be configured correctly")
+			.set_pkce_verifier(session.pkce_verifier)
+			.request_async(client)
+			.await
+			.map_err(|err| {
+				error!("Failed to exchange OIDC authorization code: {err}");
+				"Code exchange failed."
+			})?;
+
+		let Some(id_token) = token_response.id_token() else {
+			error!("Identity server did not return an id token");
+			return Err(Self::SERVER_MISCONFIGURED);
+		};
+
+		let claims = id_token
+			.claims(&machine.id_token_verifier(), &session.nonce)
+			.map_err(|err| {
+				error!("Failed to verify id token claims: {err}");
+				Self::SERVER_MISCONFIGURED
+			})?
+			.to_owned();
+
+		info!(subject = claims.subject().as_str(), "Authenticated subject");
+
+		Ok(claims)
+	}
+
+	#[tracing::instrument(skip(self, claims), fields(subject = claims.subject().to_string()))]
+	pub async fn complete_session(
+		&self,
+		claims: &Claims,
+		supplied_user_id: Option<OwnedUserId>,
+	) -> Result<SessionCompletionStatus, &'static str> {
+		let Some(OidcClient { config, .. }) = self.client.as_ref() else {
+			return Err("Delegated authentication is not enabled on this server.");
+		};
+
+		// this is a truly awful hack but we really need all the claims in a map
+		let all_claims = serde_json::to_value(claims)
+			.expect("should be able to serialize claims")
+			.as_object()
+			.expect("claims should be an object")
+			.to_owned();
+
+		debug!(?all_claims);
+
+		let subject = claims.subject().as_str();
+
+		let user_id = if let Ok(localpart) = self
+			.db
+			.openidsubject_localpart
+			.get(subject)
+			.await
+			.deserialized::<String>()
+		{
+			UserId::parse(format!("@{localpart}:{}", self.services.globals.server_name()))
+				.expect("saved localpart should be valid")
+		} else if config.prompt_for_localpart {
+			if let Some(supplied_user_id) = supplied_user_id {
+				supplied_user_id
+			} else {
+				return Ok(SessionCompletionStatus::NeedsUserId);
+			}
+		} else if let Some(preferred_username) = all_claims
+			.get(&config.preferred_username_claim)
+			.and_then(|claim| claim.as_str())
+		{
+			self.services
+				.users
+				.determine_registration_user_id(Some(preferred_username.to_owned()), None, None)
+				.await
+				.map_err(|err| {
+					error!("Preferred username claim is not a valid localpart: {err}");
+					"Your preferred username is not a valid Matrix user ID localpart. Contact \
+					 your homeserver's administrator."
+				})?
+		} else {
+			error!("Preferred username claim was not present or was not a string");
+			return Err(Self::SERVER_MISCONFIGURED);
+		};
+
+		info!(?subject, ?user_id, "User {user_id} successfully authorized with OIDC");
+
+		// Create a shadow account for the user if necessary
+		let new_account_registered = match self.services.users.status(&user_id).await {
+			| AccountStatus::Active => {
+				// Do nothing, an account already exists
+				false
+			},
+			| AccountStatus::NotFound => {
+				// Create a new shadow user
+				self.services
+					.users
+					.create_local_account(&user_id, None, None)
+					.await
+					.map_err(|err| {
+						error!("Failed to create a shadow user for {user_id}: {err}");
+						Self::SERVER_MISCONFIGURED
+					})?;
+
+				info!(?subject, ?user_id, "Shadow user created for {user_id}");
+				true
+			},
+			| AccountStatus::Deactivated => {
+				return Err("Your account has been deactivated.");
+			},
+		};
+
+		self.link_user(&user_id, subject);
+
+		// Import profile fields
+		if matches!(config.profile_key_import_mode, OidcProfileKeyImportMode::OnLogin)
+			|| (matches!(
+				config.profile_key_import_mode,
+				OidcProfileKeyImportMode::OnRegistration
+			) && new_account_registered)
+		{
+			if let Some(email_claim) = &config.email_claim {
+				if let Some(email) = claims.email().map(|email| email.as_str())
+					&& let Ok(address) = Address::from_str(email)
+				{
+					if let Err(err) = self
+						.services
+						.threepid
+						.associate_localpart_email(user_id.localpart(), &address)
+						.await
+					{
+						warn!(?email_claim, ?address, "Failed to associate email address: {err}");
+					}
+				} else {
+					warn!(
+						?email_claim,
+						"Email claim was not present or was not a valid email address"
+					);
+				}
+			}
+
+			let user_id = user_id.clone();
+			let subject = claims.subject().to_string();
+			let profile_key_map = config.profile_key_map.clone();
+			let openidsubject_currentpictureurl = self.db.openidsubject_currentpictureurl.clone();
+			let users = self.services.users.clone();
+			let media = self.services.media.clone();
+
+			let import_task = self.runtime.spawn(async move {
+				for (field, claim) in &profile_key_map {
+					let Some(value) = all_claims.get(claim).cloned() else {
+						warn!(?field, ?claim, "IDP provided no value for this mapped claim");
+						continue;
+					};
+
+					let value = if let Some(picture_url) = value.as_str()
+						&& field == ProfileFieldName::AvatarUrl.as_str()
+						&& openidsubject_currentpictureurl
+							.get(&subject)
+							.await
+							.deserialized::<String>()
+							.ok()
+							.is_none_or(|current_picture| current_picture != picture_url)
+					{
+						match media.download_media(picture_url).await {
+							| Ok((mxc, size)) => {
+								openidsubject_currentpictureurl.insert(&subject, picture_url);
+								info!(?picture_url, ?mxc, ?size, "Downloaded profile picture");
+
+								ProfileFieldValue::AvatarUrl(mxc)
+							},
+							| Err(err) => {
+								warn!(
+									?claim,
+									?picture_url,
+									"Failed to download profile picture: {err}"
+								);
+								continue;
+							},
+						}
+					} else {
+						match ProfileFieldValue::new(field, value.clone()) {
+							| Ok(value) => value,
+							| Err(err) => {
+								warn!(
+									?field,
+									?claim,
+									?value,
+									"Failed to parse claim value for profile field: {err}"
+								);
+								continue;
+							},
+						}
+					};
+
+					if let Err(err) = users
+						.set_profile_field(
+							&user_id,
+							ProfileFieldChange::Set(value),
+							PropagateTo::Unchanged,
+						)
+						.await
+					{
+						warn!(?field, ?claim, "Error while setting profile field: {err}");
+					}
+				}
+
+				info!("Profile import complete");
+			});
+
+			// Only wait for import to complete if this is a new account,
+			// so they see the correct profile information in the account panel
+			if new_account_registered {
+				let _ = import_task.await;
+			}
+		}
+
+		Ok(SessionCompletionStatus::Complete(user_id))
+	}
+
+	pub fn link_user(&self, user_id: &UserId, subject: &str) {
+		self.db
+			.openidsubject_localpart
+			.insert(subject, user_id.localpart());
+	}
+
+	pub fn unlink_user(&self, subject: &str) { self.db.openidsubject_localpart.remove(subject); }
+}

src/service/rooms/state_cache/mod.rs

@@ -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

src/service/rooms/state_cache/update.rs

@@ -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 {

src/service/rooms/timeline/append.rs

@@ -255,7 +255,13 @@ impl super::Service {
 			if let Some(state_key) = pdu.state_key() {
 				match UserId::parse(state_key) {
 					| Ok(target_user_id) => {
-						if self.services.users.is_active_local(&target_user_id).await {
+						if self
+							.services
+							.users
+							.status(&target_user_id)
+							.await
+							.is_active()
+						{
 							push_targets.insert(target_user_id.clone());
 						}
 					},

src/service/services.rs

@@ -11,8 +11,8 @@ use crate::{
 	account_data, admin, announcements, antispam, appservice, client, config, emergency,
 	federation, firstrun, globals, key_backups, mailer,
 	manager::Manager,
-	media, moderation, oauth, presence, pusher, registration_tokens, resolver, rooms, sending,
-	server_keys,
+	media, moderation, oauth, oidc, presence, pusher, registration_tokens, resolver, rooms,
+	sending, server_keys,
 	service::{self, Args, Map, Service},
 	sync, threepid, transactions, uiaa, users,
 };
@@ -28,6 +28,7 @@ pub struct Services {
 	pub key_backups: Arc<key_backups::Service>,
 	pub media: Arc<media::Service>,
 	pub oauth: Arc<oauth::Service>,
+	pub oidc: Arc<oidc::Service>,
 	pub mailer: Arc<mailer::Service>,
 	pub presence: Arc<presence::Service>,
 	pub pusher: Arc<pusher::Service>,
@@ -85,6 +86,7 @@ impl Services {
 			key_backups: build!(key_backups::Service),
 			media: build!(media::Service),
 			oauth: build!(oauth::Service),
+			oidc: build!(oidc::Service),
 			mailer: build!(mailer::Service),
 			presence: build!(presence::Service),
 			pusher: build!(pusher::Service),

src/service/threepid/mod.rs

@@ -1,6 +1,6 @@
 use std::{borrow::Cow, collections::HashMap, sync::Arc};
 
-use conduwuit::{Err, Error, Result, result::FlatOk};
+use conduwuit::{Err, Error, Result, config::OidcProfileKeyImportMode, result::FlatOk};
 use database::{Deserialized, Map};
 use governor::{DefaultKeyedRateLimiter, Quota, RateLimiter};
 use lettre::{Address, message::Mailbox};
@@ -33,7 +33,9 @@ pub enum EmailRequirement {
 	Required,
 	/// Users may change or remove their email.
 	Optional,
-	/// Users may not change their email at all.
+	/// Users may see their email, but may not change it.
+	Locked,
+	/// Email features are disabled.
 	Unavailable,
 }
 
@@ -43,6 +45,9 @@ impl EmailRequirement {
 
 	#[must_use]
 	pub fn may_remove(&self) -> bool { matches!(self, Self::Optional) }
+
+	#[must_use]
+	pub fn may_view(&self) -> bool { !matches!(self, Self::Unavailable) }
 }
 
 struct Data {
@@ -87,7 +92,14 @@ impl Service {
 	/// Check if users are required to have an email address.
 	pub fn email_requirement(&self) -> EmailRequirement {
 		if let Some(smtp) = &self.services.config.smtp {
-			if smtp.require_email_for_registration || smtp.require_email_for_token_registration {
+			if let Some(oidc) = &self.services.config.oauth.oidc
+				&& matches!(oidc.profile_key_import_mode, OidcProfileKeyImportMode::OnLogin)
+				&& oidc.email_claim.is_some()
+			{
+				EmailRequirement::Locked
+			} else if smtp.require_email_for_registration
+				|| smtp.require_email_for_token_registration
+			{
 				EmailRequirement::Required
 			} else {
 				EmailRequirement::Optional

src/service/uiaa/mod.rs

@@ -314,7 +314,7 @@ impl Service {
 				.services
 				.config
 				.oauth
-				.compatibility_mode
+				.compatibility_mode()
 				.uiaa_available()
 		{
 			return Err!(Request(Unrecognized(

src/service/users/account.rs

@@ -6,21 +6,23 @@ 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,
+	api::client::profile::PropagateTo,
 	events::{
 		GlobalAccountDataEventType, ignored_user_list::IgnoredUserListEvent,
 		push_rules::PushRulesEvent, room::message::RoomMessageEventContent,
 	},
+	profile::ProfileFieldValue,
 	push::Ruleset,
 };
 use ruminuwuity::invite_permission_config::{FilterLevel, InvitePermissionConfigEvent};
 
 use crate::{
 	appservice::RegistrationInfo,
-	users::{HashedPassword, UserSuspension},
+	users::{HashedPassword, UserSuspension, profile::ProfileFieldChange},
 };
 
 /// The status of an access token.
@@ -29,6 +31,30 @@ pub enum AccessTokenStatus {
 	Expired,
 }
 
+/// The status of a user account.
+#[derive(Clone, Copy)]
+pub enum AccountStatus {
+	NotFound,
+	Active,
+	Deactivated,
+}
+
+impl AccountStatus {
+	#[must_use]
+	pub fn is_found(&self) -> bool { !matches!(self, Self::NotFound) }
+
+	#[must_use]
+	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,29 +115,33 @@ 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(())
 	}
 
-	/// Create a new account for a local human or bot user.
+	/// Create a new account for a local human or bot user. If `password` is
+	/// None, the account will be a shadow account.
 	pub async fn create_local_account(
 		&self,
 		user_id: &UserId,
-		password: HashedPassword,
+		password: Option<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?;
+
+		if let Some(password) = password {
+			self.convert_to_local_account(user_id, password).await?;
+		}
 
 		// Set an initial display name
 		{
@@ -123,7 +153,13 @@ impl super::Service {
 				displayname.push_str(suffix);
 			}
 
-			self.set_displayname(user_id, Some(displayname));
+			self.set_profile_field(
+				user_id,
+				ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)),
+				PropagateTo::None,
+			)
+			.await
+			.expect("should be able to set display name");
 		};
 
 		// Set default push rules
@@ -230,6 +266,8 @@ impl super::Service {
 		}
 
 		info!("Created new user account for {user_id}");
+
+		Ok(())
 	}
 
 	pub async fn determine_registration_user_id(
@@ -268,28 +306,7 @@ impl super::Service {
 				&supplied_username,
 				self.services.globals.server_name(),
 			) {
-				| Ok(user_id) => {
-					if let Err(e) = user_id.validate_strict() {
-						// Unless we are in emergency mode, we should follow synapse's behaviour
-						// on not allowing things like spaces and UTF-8 characters in
-						// usernames
-						if !emergency_mode_enabled {
-							return Err!(Request(InvalidUsername(debug_warn!(
-								"Username {supplied_username} contains disallowed characters or \
-								 spaces: {e}"
-							))));
-						}
-					}
-
-					// Don't allow registration with user IDs that aren't local
-					if !self.services.globals.user_is_local(&user_id) {
-						return Err!(Request(InvalidUsername(
-							"Username {supplied_username} is not local to this server"
-						)));
-					}
-
-					user_id
-				},
+				| Ok(user_id) => user_id,
 				| Err(e) => {
 					return Err!(Request(InvalidUsername(debug_warn!(
 						"Username {supplied_username} is not valid: {e}"
@@ -297,8 +314,27 @@ impl super::Service {
 				},
 			};
 
-			if self.exists(&user_id).await {
-				return Err!(Request(UserInUse("User ID is not available.")));
+			if let Err(e) = user_id.validate_strict() {
+				// Unless we are in emergency mode, we should follow synapse's behaviour
+				// on not allowing things like spaces and UTF-8 characters in
+				// usernames
+				if !emergency_mode_enabled {
+					return Err!(Request(InvalidUsername(debug_warn!(
+						"Username {supplied_username} contains disallowed characters or spaces: \
+						 {e}"
+					))));
+				}
+			}
+
+			// Don't allow registration with user IDs that aren't local
+			if !self.services.globals.user_is_local(&user_id) {
+				return Err!(Request(InvalidUsername(
+					"Username {supplied_username} is not local to this server"
+				)));
+			}
+
+			if self.status(&user_id).await.is_found() {
+				return Err!(Request(UserInUse("Username is not available.")));
 			}
 
 			// Check that the user ID is/is not in an appservice's namespace
@@ -329,26 +365,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 +427,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 +484,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 +552,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 +570,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 +621,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)

src/service/users/dehydrated_device.rs

@@ -21,10 +21,7 @@ pub struct DehydratedDevice {
 impl super::Service {
 	/// Creates or recreates the user's dehydrated device.
 	pub async fn set_dehydrated_device(&self, user_id: &UserId, request: Request) -> Result {
-		assert!(
-			self.exists(user_id).await,
-			"Tried to create dehydrated device for non-existent user"
-		);
+		self.status(user_id).await.ensure_active()?;
 
 		let existing_id = self.get_dehydrated_device_id(user_id).await;
 

src/service/users/device.rs

@@ -4,8 +4,8 @@ use std::{
 };
 
 use conduwuit::{
-	Err, utils,
-	utils::{ReadyExt, stream::TryIgnore},
+	Err, Result,
+	utils::{self, ReadyExt, stream::TryIgnore},
 };
 use database::{Deserialized, Ignore, Interfix, Json};
 use futures::{Stream, StreamExt};
@@ -18,8 +18,7 @@ use serde_json::json;
 use crate::users::increment;
 
 impl super::Service {
-	/// Adds a new device to a user. The user must exist, otherwise InvalidParam
-	/// is returned.
+	/// Adds a new device to a user.
 	pub async fn create_device(
 		&self,
 		user_id: &UserId,
@@ -28,12 +27,8 @@ impl super::Service {
 		token_max_age: Option<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)

src/service/users/mod.rs

@@ -4,20 +4,22 @@ 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},
 };
 use database::Map;
+pub use profile::ProfileFieldChange;
 use ruma::{UserId, api::error::ErrorKind, encryption::CrossSigningKey, serde::Raw};
 use serde::{Deserialize, Serialize};
 
 use crate::{
-	Dep, account_data, admin, appservice, config, firstrun, globals, oauth,
+	Dep, account_data, admin, appservice, config, firstrun, globals, oauth, presence,
 	rooms::{self, alias, membership},
 	threepid,
 };
@@ -61,9 +63,12 @@ struct Services {
 	globals: Dep<globals::Service>,
 	membership: Dep<membership::Service>,
 	oauth: Dep<oauth::Service>,
+	presence: Dep<presence::Service>,
+	state: Dep<rooms::state::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
 	state_cache: Dep<rooms::state_cache::Service>,
 	threepid: Dep<threepid::Service>,
+	timeline: Dep<rooms::timeline::Service>,
 }
 
 struct Data {
@@ -75,11 +80,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>,
@@ -107,10 +114,13 @@ impl crate::Service for Service {
 				globals: args.depend::<globals::Service>("globals"),
 				membership: args.depend::<membership::Service>("membership"),
 				oauth: args.depend::<oauth::Service>("oauth"),
+				presence: args.depend::<presence::Service>("presence"),
+				state: args.depend::<rooms::state::Service>("rooms::state"),
 				state_accessor: args
 					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
 				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
 				threepid: args.depend::<threepid::Service>("threepid"),
+				timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
 			},
 			db: Data {
 				keychangeid_userid: args.db["keychangeid_userid"].clone(),
@@ -121,10 +131,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(),

src/service/users/profile.rs

@@ -1,17 +1,276 @@
-use conduwuit::utils::{ReadyExt, stream::TryIgnore};
+use std::{borrow::Cow, collections::BTreeMap};
+
+use conduwuit::{
+	Err, Result,
+	pdu::PartialPdu,
+	utils::{ReadyExt, stream::TryIgnore, to_canonical_object},
+};
 use database::{Deserialized, Ignore, Interfix, Json};
 use futures::{Stream, StreamExt};
-use ruma::{OwnedMxcUri, UserId};
+use ruma::{
+	OwnedMxcUri, UserId,
+	api::client::profile::PropagateTo,
+	events::room::member::MembershipState,
+	presence::PresenceState,
+	profile::{ProfileFieldName, ProfileFieldValue},
+};
+use serde_json::{Value, to_value};
+
+pub enum ProfileFieldChange {
+	Set(ProfileFieldValue),
+	Delete(ProfileFieldName),
+}
+
+impl ProfileFieldChange {
+	fn field_name(&self) -> ProfileFieldName {
+		match self {
+			| &Self::Delete(ref name) => name.clone(),
+			| &Self::Set(ref value) => value.field_name(),
+		}
+	}
+
+	fn value(&self) -> Option<Cow<'_, Value>> {
+		if let Self::Set(value) = self {
+			Some(value.value())
+		} else {
+			None
+		}
+	}
+}
 
 impl super::Service {
+	pub async fn set_profile_field(
+		&self,
+		user_id: &UserId,
+		change: ProfileFieldChange,
+		propagate_to: PropagateTo,
+	) -> Result<()> {
+		const MAX_KEY_LENGTH_BYTES: usize = 255;
+		const MAX_PROFILE_LENGTH_BYTES: usize = 65536;
+
+		let field_name = change.field_name();
+
+		// TODO: The spec mentions special error codes (M_PROFILE_TOO_LARGE,
+		// M_KEY_TOO_LARGE) for profile field size limits, but they're not in its list
+		// of error codes and Ruma doesn't have them. Should we return those, or is
+		// M_TOO_LARGE okay?
+		if field_name.as_str().len() > MAX_KEY_LENGTH_BYTES {
+			return Err!(Request(TooLarge(
+				"Individual profile keys must not exceed {MAX_KEY_LENGTH_BYTES} bytes in length."
+			)));
+		}
+
+		// Serialize the entire profile as canonical JSON, including the new change,
+		// to check if it exceeds 64 KiB
+		{
+			let mut full_profile = self.get_local_profile(user_id).await;
+
+			match &change {
+				| ProfileFieldChange::Set(value) => {
+					full_profile.insert(
+						value.field_name().as_str().to_owned(),
+						value.value().clone().into_owned(),
+					);
+				},
+				| ProfileFieldChange::Delete(key) => {
+					full_profile.remove(key.as_str());
+				},
+			}
+
+			if let Ok(canonical_profile) = to_canonical_object(full_profile) {
+				if serde_json::to_string(&canonical_profile)
+					.expect("should be able to serialize to string")
+					.len() > MAX_PROFILE_LENGTH_BYTES
+				{
+					return Err!(
+						"Profile data must not exceed {MAX_PROFILE_LENGTH_BYTES} bytes in \
+						 length."
+					);
+				}
+			} else {
+				return Err!(Request(BadJson("Failed to canonicalize profile.")));
+			}
+		}
+
+		// Check if this change would be a no-op
+		if self
+			.get_local_profile_field(user_id, field_name.clone())
+			.await
+			.is_some_and(|value| Some(value.value()) == change.value())
+		{
+			return Ok(());
+		}
+
+		// If the user is local and changed their displayname or avatar_url, update it
+		// in all their joined rooms. This is done before updating their profile data
+		// so we can check the old value of the field if `propagate_to` is `unchanged`.
+		if matches!(field_name, ProfileFieldName::AvatarUrl | ProfileFieldName::DisplayName)
+			&& matches!(propagate_to, PropagateTo::All | PropagateTo::Unchanged)
+			&& self.services.globals.user_is_local(user_id)
+		{
+			let current_displayname = self.displayname(user_id).await.ok();
+			let current_avatar_url = self.avatar_url(user_id).await.ok();
+
+			let mut all_joined_rooms = self.services.state_cache.rooms_joined(user_id);
+
+			while let Some(room_id) = all_joined_rooms.next().await {
+				// TODO: this clobbers any custom fields on the event content
+				let mut current_membership = self
+					.services
+					.state_accessor
+					.get_member(&room_id, user_id)
+					.await
+					.expect("should be able to fetch membership event for joined room");
+
+				assert_eq!(
+					current_membership.membership,
+					MembershipState::Join,
+					"user should be joined"
+				);
+
+				// If `propagate_to` is `unchanged`, and the current value of the field we're
+				// updating was changed from its global value in this room, skip it.
+				if matches!(propagate_to, PropagateTo::Unchanged) {
+					let field_changed_from_global = match field_name {
+						| ProfileFieldName::AvatarUrl =>
+							current_membership.avatar_url.as_ref() != current_avatar_url.as_ref(),
+						| ProfileFieldName::DisplayName =>
+							current_membership.displayname.as_ref()
+								!= current_displayname.as_ref(),
+						| _ => unreachable!(),
+					};
+
+					if field_changed_from_global {
+						continue;
+					}
+				}
+
+				let state_lock = self.services.state.mutex.lock(room_id.as_str()).await;
+
+				// Preserve keys in accordance with the key copying rules
+				current_membership.reason = None;
+				current_membership.join_authorized_via_users_server = None;
+				match &change {
+					| ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(avatar_url)) => {
+						current_membership.avatar_url = Some(avatar_url.clone());
+					},
+					| ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)) => {
+						current_membership.displayname = Some(displayname.clone());
+					},
+					| ProfileFieldChange::Delete(ProfileFieldName::AvatarUrl) => {
+						current_membership.avatar_url = None;
+					},
+					| ProfileFieldChange::Delete(ProfileFieldName::DisplayName) => {
+						current_membership.displayname = None;
+					},
+					| _ => unreachable!(),
+				}
+
+				let _ = self
+					.services
+					.timeline
+					.build_and_append_pdu(
+						PartialPdu::state(user_id.to_string(), &current_membership),
+						user_id,
+						Some(&room_id),
+						&state_lock,
+					)
+					.await;
+			}
+
+			if self.services.config.allow_local_presence {
+				// Send a presence EDU to indicate the profile changed
+				let _ = self
+					.services
+					.presence
+					.ping_presence(user_id, &PresenceState::Online)
+					.await;
+			}
+		}
+
+		match change {
+			| ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)) => {
+				self.set_displayname(user_id, Some(displayname).filter(|dn| !dn.is_empty()));
+			},
+			| ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(avatar_url)) => {
+				self.set_avatar_url(user_id, Some(avatar_url).filter(|av| av.is_valid()));
+			},
+			| ProfileFieldChange::Delete(ProfileFieldName::DisplayName) => {
+				self.set_displayname(user_id, None);
+			},
+			| ProfileFieldChange::Delete(ProfileFieldName::AvatarUrl) => {
+				self.set_avatar_url(user_id, None);
+			},
+			| other => self.set_profile_key(
+				user_id,
+				other.field_name().as_str(),
+				other.value().map(Cow::into_owned),
+			),
+		}
+
+		Ok(())
+	}
+
+	pub async fn get_local_profile(&self, user_id: &UserId) -> BTreeMap<String, Value> {
+		let mut profile = BTreeMap::new();
+
+		// Get displayname and avatar_url independently because `all_profile_keys`
+		// doesn't include them
+		for field in [ProfileFieldName::AvatarUrl, ProfileFieldName::DisplayName] {
+			let key = field.as_str().to_owned();
+
+			if let Some(value) = self.get_local_profile_field(user_id, field).await {
+				profile.insert(key, value.value().into_owned());
+			}
+		}
+
+		// Insert all other profile fields
+		let mut all_fields = self.all_profile_keys(user_id);
+
+		while let Some((key, value)) = all_fields.next().await {
+			profile.insert(key, value);
+		}
+
+		profile
+	}
+
+	pub async fn get_local_profile_field(
+		&self,
+		user_id: &UserId,
+		field: ProfileFieldName,
+	) -> Option<ProfileFieldValue> {
+		let value = match field.clone() {
+			| ProfileFieldName::AvatarUrl => self
+				.avatar_url(user_id)
+				.await
+				.ok()
+				.map(to_value)
+				.transpose()
+				.expect("converting avatar url to value should succeed"),
+			| ProfileFieldName::DisplayName => self
+				.displayname(user_id)
+				.await
+				.ok()
+				.map(to_value)
+				.transpose()
+				.expect("converting displayname to value should succeed"),
+			| other => self.profile_key(user_id, other.as_str()).await.ok(),
+		}?;
+
+		Some(
+			ProfileFieldValue::new(field.as_str(), value)
+				.expect("local profile field should be valid"),
+		)
+	}
+
 	/// Returns the displayname of a user on this homeserver.
-	pub async fn displayname(&self, user_id: &UserId) -> conduwuit::Result<String> {
+	pub async fn displayname(&self, user_id: &UserId) -> Result<String> {
 		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<String>) {
+	fn set_displayname(&self, user_id: &UserId, displayname: Option<String>) {
 		if let Some(displayname) = displayname {
 			self.db.userid_displayname.insert(user_id, displayname);
 		} else {
@@ -20,12 +279,12 @@ impl super::Service {
 	}
 
 	/// Get the `avatar_url` of a user.
-	pub async fn avatar_url(&self, user_id: &UserId) -> conduwuit::Result<OwnedMxcUri> {
+	pub async fn avatar_url(&self, user_id: &UserId) -> Result<OwnedMxcUri> {
 		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<OwnedMxcUri>) {
+	fn set_avatar_url(&self, user_id: &UserId, avatar_url: Option<OwnedMxcUri>) {
 		match avatar_url {
 			| Some(avatar_url) => {
 				self.db.userid_avatarurl.insert(user_id, &avatar_url);
@@ -37,11 +296,7 @@ impl super::Service {
 	}
 
 	/// Gets a specific user profile key
-	pub async fn profile_key(
-		&self,
-		user_id: &UserId,
-		profile_key: &str,
-	) -> conduwuit::Result<serde_json::Value> {
+	pub async fn profile_key(&self, user_id: &UserId, profile_key: &str) -> Result<Value> {
 		let key = (user_id, profile_key);
 		self.db
 			.useridprofilekey_value
@@ -54,7 +309,7 @@ impl super::Service {
 	pub fn all_profile_keys<'a>(
 		&'a self,
 		user_id: &'a UserId,
-	) -> impl Stream<Item = (String, serde_json::Value)> + 'a + Send {
+	) -> impl Stream<Item = (String, Value)> + 'a + Send {
 		type KeyVal<'a> = ((Ignore, String), &'a [u8]);
 
 		let prefix = (user_id, Interfix);
@@ -67,11 +322,11 @@ impl super::Service {
 	}
 
 	/// Sets a new profile key value, removes the key if value is None
-	pub fn set_profile_key(
+	fn set_profile_key(
 		&self,
 		user_id: &UserId,
 		profile_key: &str,
-		profile_key_value: Option<serde_json::Value>,
+		profile_key_value: Option<Value>,
 	) {
 		let key = (user_id, profile_key);
 

src/service/users/remote.rs

@@ -0,0 +1,17 @@
+use conduwuit::utils::stream::TryIgnore;
+use futures::Stream;
+use ruma::{OwnedUserId, UserId};
+
+impl super::Service {
+	/// Record the existence of a remote user.
+	pub fn record_remote_user(&self, user_id: &UserId) {
+		assert!(!self.services.globals.user_is_local(user_id), "user is not remote");
+
+		self.db.remoteuserid_remoteuser.insert(user_id, "");
+	}
+
+	/// Returns a stream over all remote users this server has ever seen.
+	pub fn stream_remote_users(&self) -> impl Stream<Item = OwnedUserId> + Send {
+		self.db.remoteuserid_remoteuser.keys().ignore_err()
+	}
+}