From 18f5f4fbfa10ceaa7f7a7c9b33be1249db06a0c4 Mon Sep 17 00:00:00 2001 From: Jade Ellis Date: Mon, 12 Oct 2026 17:58:57 +0100 Subject: [PATCH] refactor(threepid): Use SyncMutex and decouple guard from ValidSession Change threepid sessions and send_attempts to SyncMutex and avoid holding the sessions mutex across callers of get_valid_session. --- src/api/client/account/threepid.rs | 1 - src/service/threepid/mod.rs | 146 ++++++++++++------------ src/service/uiaa/mod.rs | 37 +++--- src/web/pages/account/email.rs | 1 - src/web/pages/account/password/reset.rs | 1 - src/web/pages/account/register.rs | 1 - src/web/pages/threepid.rs | 1 - 7 files changed, 90 insertions(+), 98 deletions(-) diff --git a/src/api/client/account/threepid.rs b/src/api/client/account/threepid.rs index 93531eb02..edcd9017d 100644 --- a/src/api/client/account/threepid.rs +++ b/src/api/client/account/threepid.rs @@ -135,7 +135,6 @@ pub(crate) async fn add_3pid_route( let email = services .threepid .get_valid_session(&body.sid, &body.client_secret) - .await .map_err(|message| err!(Request(ThreepidAuthFailed("{message}"))))? .consume(); diff --git a/src/service/threepid/mod.rs b/src/service/threepid/mod.rs index 7e4ffa968..cbc1fd080 100644 --- a/src/service/threepid/mod.rs +++ b/src/service/threepid/mod.rs @@ -1,6 +1,8 @@ use std::{borrow::Cow, collections::HashMap, sync::Arc}; -use conduwuit::{Err, Error, Result, config::OidcProfileKeyImportMode, result::FlatOk}; +use conduwuit::{ + Err, Error, Result, SyncMutex, config::OidcProfileKeyImportMode, result::FlatOk, +}; use database::{Deserialized, Map}; use governor::{DefaultKeyedRateLimiter, Quota, RateLimiter}; use lettre::{Address, message::Mailbox}; @@ -9,7 +11,6 @@ use ruma::{ ClientSecret, OwnedClientSecret, OwnedSessionId, SessionId, api::error::{ErrorKind, LimitExceededErrorData}, }; -use tokio::sync::MutexGuard; pub mod session; @@ -22,8 +23,8 @@ use crate::{ pub struct Service { db: Data, services: Services, - sessions: tokio::sync::Mutex, - send_attempts: std::sync::Mutex>, + sessions: SyncMutex, + send_attempts: SyncMutex>, ratelimiter: DefaultKeyedRateLimiter
, } @@ -71,8 +72,8 @@ impl crate::Service for Service { config: args.depend("config"), mailer: args.depend("mailer"), }, - sessions: tokio::sync::Mutex::default(), - send_attempts: std::sync::Mutex::default(), + sessions: SyncMutex::default(), + send_attempts: SyncMutex::default(), ratelimiter: RateLimiter::keyed(Self::EMAIL_RATELIMIT), })) } @@ -121,65 +122,69 @@ impl Service { send_attempt: usize, ) -> Result { let mailer = self.services.mailer.expect_mailer()?; - let mut sessions = self.sessions.lock().await; + let (session_id, token_str) = { + let mut sessions = self.sessions.lock(); - let session = match sessions.get_session_by_client_secret(client_secret) { - // If a validation session already exists for this client secret, we - // can either reuse it with a new token or return early because it's - // already valid. - | Some(session) => { - if session.email != recipient.email { - return Err!(Request(InvalidParam("Wrong email for session."))); - } + let session = match sessions.get_session_by_client_secret(client_secret) { + // If a validation session already exists for this client secret, we + // can either reuse it with a new token or return early because it's + // already valid. + | Some(session) => { + if session.email != recipient.email { + return Err!(Request(InvalidParam("Wrong email for session."))); + } - match session.validation_state { - | ValidationState::Validated => { - // If the existing session is already valid, don't send - // an email. - return Ok(session.session_id.clone()); - }, - | ValidationState::Pending(ref mut token) => { - // Check ratelimiting for the target address. - if self.ratelimiter.check_key(&session.email).is_err() { - return Err(Error::BadRequest( - ErrorKind::LimitExceeded(LimitExceededErrorData::new()), - "You're sending emails too fast, try again in a few minutes.", - )); - } - - // Check the send attempt for this session. - let mut send_attempts = self.send_attempts.lock().unwrap(); - - let last_send_attempt = send_attempts - .entry((session.client_secret.clone(), session.email.clone())) - .or_default(); - - if send_attempt <= *last_send_attempt { - // If the supplied send attempt isn't higher than - // the last one, don't send an email. + match session.validation_state { + | ValidationState::Validated => { + // If the existing session is already valid, don't + // send an email. return Ok(session.session_id.clone()); - } + }, + | ValidationState::Pending(ref mut token) => { + // Check ratelimiting for the target address. + if self.ratelimiter.check_key(&session.email).is_err() { + return Err(Error::BadRequest( + ErrorKind::LimitExceeded(LimitExceededErrorData::new()), + "You're sending emails too fast, try again in a few minutes.", + )); + } - // Save this send attempt. - *last_send_attempt = send_attempt; - drop(send_attempts); + // Check the send attempt for this session. + let mut send_attempts = self.send_attempts.lock(); - // Create a new token for the existing session. - *token = ValidationToken::new_random(); + let last_send_attempt = send_attempts + .entry((session.client_secret.clone(), session.email.clone())) + .or_default(); - session - }, - } - }, - // If no session exists, create a new one. - | None => sessions.create_session(recipient.email.clone(), client_secret.to_owned()), - }; + if send_attempt <= *last_send_attempt { + // If the supplied send attempt isn't higher + // than the last + // one, don't send an email. + return Ok(session.session_id.clone()); + } - // Clone this so it can outlive the lock we're holding on `sessions` - let session_id = session.session_id.clone(); + // Save this send attempt. + *last_send_attempt = send_attempt; + drop(send_attempts); - let ValidationState::Pending(token) = &session.validation_state else { - unreachable!("session should be pending") + // Create a new token for the existing session. + *token = ValidationToken::new_random(); + + session + }, + } + }, + // If no session exists, create a new one. + | None => + sessions.create_session(recipient.email.clone(), client_secret.to_owned()), + }; + + let session_id = session.session_id.clone(); + let ValidationState::Pending(token) = &session.validation_state else { + unreachable!("session should be pending") + }; + + (session_id, token.token.clone()) }; let mut validation_url = self @@ -192,11 +197,7 @@ impl Service { validation_url .query_pairs_mut() .append_pair("session", session_id.as_str()) - .append_pair("token", &token.token); - - // Once the validation URL is built, we don't need any data borrowed - // from `sessions` anymore and can release our lock - drop(sessions); + .append_pair("token", &token_str); let message = prepare_body(validation_url.to_string()); @@ -206,12 +207,12 @@ impl Service { } /// Attempt to mark a validation session as valid using a validation token. - pub async fn try_validate_session( + pub fn try_validate_session( &self, session_id: &SessionId, supplied_token: &str, ) -> Result<(), Cow<'static, str>> { - let mut sessions = self.sessions.lock().await; + let mut sessions = self.sessions.lock(); let Some(session) = sessions.get_session(session_id) else { return Err("Validation session does not exist".into()); @@ -240,12 +241,12 @@ impl Service { } /// Get a validated validation session. - pub async fn get_valid_session( + pub fn get_valid_session( &self, session_id: &SessionId, client_secret: &ClientSecret, ) -> Result, Cow<'static, str>> { - let mut sessions = self.sessions.lock().await; + let mut sessions = self.sessions.lock(); let Some(session) = sessions.get_session(session_id) else { return Err("Validation session does not exist".into()); @@ -259,7 +260,7 @@ impl Service { Ok(ValidSession { email, session_id: session_id.to_owned(), - sessions, + service: self, }) } else { Err("This email address has not been validated. Did you use the link that was sent \ @@ -337,18 +338,21 @@ impl Service { } } -pub struct ValidSession<'lock> { +pub struct ValidSession<'a> { pub email: Address, session_id: OwnedSessionId, - sessions: MutexGuard<'lock, ValidationSessions>, + service: &'a Service, } impl ValidSession<'_> { /// Consume this session, removing it from the database and releasing the /// lock it holds. #[must_use] - pub fn consume(mut self) -> Address { - self.sessions.remove_session(&self.session_id); + pub fn consume(self) -> Address { + self.service + .sessions + .lock() + .remove_session(&self.session_id); self.email } diff --git a/src/service/uiaa/mod.rs b/src/service/uiaa/mod.rs index f78bbe923..0aebaed7e 100644 --- a/src/service/uiaa/mod.rs +++ b/src/service/uiaa/mod.rs @@ -492,31 +492,24 @@ impl Service { | AuthData::EmailIdentity(EmailIdentity { thirdparty_id_creds: ThirdpartyIdCredentials { client_secret, sid, .. }, .. - }) => { - match self - .services - .threepid - .get_valid_session(sid, client_secret) - .await - { - | Ok(session) => { - let email = session.consume(); + }) => match self.services.threepid.get_valid_session(sid, client_secret) { + | Ok(session) => { + let email = session.consume(); - if let Some(localpart) = - self.services.threepid.get_localpart_for_email(&email).await - { - identity.try_set_localpart(localpart)?; - } + if let Some(localpart) = + self.services.threepid.get_localpart_for_email(&email).await + { + identity.try_set_localpart(localpart)?; + } - identity.try_set_email(email)?; + identity.try_set_email(email)?; - Ok(AuthType::EmailIdentity) - }, - | Err(message) => Err(StandardErrorBody::new( - ErrorKind::ThreepidAuthFailed, - message.into_owned(), - )), - } + Ok(AuthType::EmailIdentity) + }, + | Err(message) => Err(StandardErrorBody::new( + ErrorKind::ThreepidAuthFailed, + message.into_owned(), + )), }, #[allow(clippy::useless_let_if_seq)] | AuthData::Password(Password { identifier, password, .. }) => { diff --git a/src/web/pages/account/email.rs b/src/web/pages/account/email.rs index 1d9ca64b2..bace631df 100644 --- a/src/web/pages/account/email.rs +++ b/src/web/pages/account/email.rs @@ -167,7 +167,6 @@ async fn get_change_email_validate( let Ok(session) = services .threepid .get_valid_session(&session_id, &client_secret) - .await else { return response!(ChangeEmailValidate::new( context, diff --git a/src/web/pages/account/password/reset.rs b/src/web/pages/account/password/reset.rs index 20b2732d9..917c86cc9 100644 --- a/src/web/pages/account/password/reset.rs +++ b/src/web/pages/account/password/reset.rs @@ -189,7 +189,6 @@ async fn route_reset_password_validate( let body = match services .threepid .get_valid_session(&query.threepid.session_id, &query.threepid.client_secret) - .await { | Ok(session) => { let Some(localpart) = services diff --git a/src/web/pages/account/register.rs b/src/web/pages/account/register.rs index 5abff481e..cdce43fea 100644 --- a/src/web/pages/account/register.rs +++ b/src/web/pages/account/register.rs @@ -289,7 +289,6 @@ async fn get_register_email_validate( let Ok(session) = services .threepid .get_valid_session(&session_id, &client_secret) - .await else { return response!(RegisterEmailValidate::new(context, session_id, client_secret, true)); }; diff --git a/src/web/pages/threepid.rs b/src/web/pages/threepid.rs index 5aefcf6f0..3d8b8b746 100644 --- a/src/web/pages/threepid.rs +++ b/src/web/pages/threepid.rs @@ -33,7 +33,6 @@ async fn threepid_validation( services .threepid .try_validate_session(&query.session, &query.token) - .await .map_err(|message| WebError::BadRequest(message.into_owned()))?; Ok(ThreepidValidation::new(context))