Pass the rng and clock around

crates/storage/Cargo.toml

@@ -19,6 +19,7 @@ tracing = "0.1.37"
 argon2 = { version = "0.4.1", features = ["password-hash"] }
 password-hash = { version = "0.4.2", features = ["std"] }
 rand = "0.8.5"
+rand_chacha = "0.3.1"
 url = { version = "2.3.1", features = ["serde"] }
 uuid = "1.2.1"
 ulid = { version = "1.0.0", features = ["uuid", "serde"] }

crates/storage/src/compat.rs

@@ -19,6 +19,7 @@ use mas_data_model::{
     CompatAccessToken, CompatRefreshToken, CompatSession, CompatSsoLogin, CompatSsoLoginState,
     Device, User, UserEmail,
 };
+use rand::Rng;
 use sqlx::{Acquire, PgExecutor, Postgres};
 use thiserror::Error;
 use tokio::task;
@@ -27,7 +28,7 @@ use ulid::Ulid;
 use url::Url;
 use uuid::Uuid;
 
-use crate::{user::lookup_user_by_username, DatabaseInconsistencyError, PostgresqlBackend};
+use crate::{user::lookup_user_by_username, Clock, DatabaseInconsistencyError, PostgresqlBackend};
 
 struct CompatAccessTokenLookup {
     compat_access_token_id: Uuid,
@@ -67,6 +68,7 @@ impl CompatAccessTokenLookupError {
 #[tracing::instrument(skip_all, err)]
 pub async fn lookup_active_compat_access_token(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     token: &str,
 ) -> Result<
     (
@@ -112,7 +114,7 @@ pub async fn lookup_active_compat_access_token(
 
     // Check for token expiration
     if let Some(expires_at) = res.compat_access_token_expires_at {
-        if expires_at < Utc::now() {
+        if expires_at < clock.now() {
             return Err(CompatAccessTokenLookupError::Expired { when: expires_at });
         }
     }
@@ -311,7 +313,9 @@ pub async fn lookup_active_compat_refresh_token(
     err(Display),
 )]
 pub async fn compat_login(
-    conn: impl Acquire<'_, Database = Postgres>,
+    conn: impl Acquire<'_, Database = Postgres> + Send,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     username: &str,
     password: &str,
     device: Device,
@@ -348,8 +352,8 @@ pub async fn compat_login(
     .instrument(tracing::info_span!("Verify hashed password"))
     .await??;
 
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("compat_session.id", tracing::field::display(id));
 
     sqlx::query!(
@@ -392,12 +396,14 @@ pub async fn compat_login(
 )]
 pub async fn add_compat_access_token(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     session: &CompatSession<PostgresqlBackend>,
     token: String,
     expires_after: Option<Duration>,
 ) -> Result<CompatAccessToken<PostgresqlBackend>, anyhow::Error> {
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("compat_access_token.id", tracing::field::display(id));
 
     let expires_at = expires_after.map(|expires_after| created_at + expires_after);
@@ -436,9 +442,10 @@ pub async fn add_compat_access_token(
 )]
 pub async fn expire_compat_access_token(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     access_token: CompatAccessToken<PostgresqlBackend>,
 ) -> Result<(), anyhow::Error> {
-    let expires_at = Utc::now();
+    let expires_at = clock.now();
     let res = sqlx::query!(
         r#"
             UPDATE compat_access_tokens
@@ -474,12 +481,14 @@ pub async fn expire_compat_access_token(
 )]
 pub async fn add_compat_refresh_token(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     session: &CompatSession<PostgresqlBackend>,
     access_token: &CompatAccessToken<PostgresqlBackend>,
     token: String,
 ) -> Result<CompatRefreshToken<PostgresqlBackend>, anyhow::Error> {
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("compat_refresh_token.id", tracing::field::display(id));
 
     sqlx::query!(
@@ -514,9 +523,10 @@ pub async fn add_compat_refresh_token(
 )]
 pub async fn compat_logout(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     token: &str,
 ) -> Result<(), anyhow::Error> {
-    let finished_at = Utc::now();
+    let finished_at = clock.now();
     // TODO: this does not check for token expiration
     let compat_session_id = sqlx::query_scalar!(
         r#"
@@ -552,9 +562,10 @@ pub async fn compat_logout(
 )]
 pub async fn consume_compat_refresh_token(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     refresh_token: CompatRefreshToken<PostgresqlBackend>,
 ) -> Result<(), anyhow::Error> {
-    let consumed_at = Utc::now();
+    let consumed_at = clock.now();
     let res = sqlx::query!(
         r#"
             UPDATE compat_refresh_tokens
@@ -587,11 +598,13 @@ pub async fn consume_compat_refresh_token(
 )]
 pub async fn insert_compat_sso_login(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     login_token: String,
     redirect_uri: Url,
 ) -> Result<CompatSsoLogin<PostgresqlBackend>, anyhow::Error> {
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("compat_sso_login.id", tracing::field::display(id));
 
     sqlx::query!(
@@ -845,7 +858,9 @@ pub async fn get_compat_sso_login_by_token(
     err(Display),
 )]
 pub async fn fullfill_compat_sso_login(
-    conn: impl Acquire<'_, Database = Postgres>,
+    conn: impl Acquire<'_, Database = Postgres> + Send,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     user: User<PostgresqlBackend>,
     mut login: CompatSsoLogin<PostgresqlBackend>,
     device: Device,
@@ -856,8 +871,8 @@ pub async fn fullfill_compat_sso_login(
 
     let mut txn = conn.begin().await.context("could not start transaction")?;
 
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("user.id", tracing::field::display(user.data));
 
     sqlx::query!(
@@ -883,7 +898,7 @@ pub async fn fullfill_compat_sso_login(
         finished_at: None,
     };
 
-    let fulfilled_at = Utc::now();
+    let fulfilled_at = clock.now();
     sqlx::query!(
         r#"
             UPDATE compat_sso_logins
@@ -924,6 +939,7 @@ pub async fn fullfill_compat_sso_login(
 )]
 pub async fn mark_compat_sso_login_as_exchanged(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     mut login: CompatSsoLogin<PostgresqlBackend>,
 ) -> Result<CompatSsoLogin<PostgresqlBackend>, anyhow::Error> {
     let (fulfilled_at, session) = match login.state {
@@ -934,7 +950,7 @@ pub async fn mark_compat_sso_login_as_exchanged(
         _ => bail!("sso login in wrong state"),
     };
 
-    let exchanged_at = Utc::now();
+    let exchanged_at = clock.now();
     sqlx::query!(
         r#"
             UPDATE compat_sso_logins

crates/storage/src/lib.rs

@@ -15,7 +15,12 @@
 //! Interactions with the database
 
 #![forbid(unsafe_code)]
-#![deny(clippy::all, clippy::str_to_string, rustdoc::broken_intra_doc_links)]
+#![deny(
+    clippy::all,
+    clippy::str_to_string,
+    clippy::future_not_send,
+    rustdoc::broken_intra_doc_links
+)]
 #![warn(clippy::pedantic)]
 #![allow(
     clippy::missing_errors_doc,
@@ -23,12 +28,27 @@
     clippy::module_name_repetitions
 )]
 
+use chrono::{DateTime, Utc};
 use mas_data_model::{StorageBackend, StorageBackendMarker};
 use serde::Serialize;
 use sqlx::migrate::Migrator;
 use thiserror::Error;
 use ulid::Ulid;
 
+#[derive(Default, Debug, Clone, Copy)]
+pub struct Clock {
+    _private: (),
+}
+
+impl Clock {
+    #[must_use]
+    pub fn now(&self) -> DateTime<Utc> {
+        // This is the clock used elsewhere, it's fine to call Utc::now here
+        #[allow(clippy::disallowed_methods)]
+        Utc::now()
+    }
+}
+
 #[derive(Debug, Error)]
 #[error("database query returned an inconsistent state")]
 pub struct DatabaseInconsistencyError;

crates/storage/src/oauth2/access_token.rs

@@ -15,13 +15,14 @@
 use anyhow::Context;
 use chrono::{DateTime, Duration, Utc};
 use mas_data_model::{AccessToken, Authentication, BrowserSession, Session, User, UserEmail};
+use rand::Rng;
 use sqlx::{Acquire, PgExecutor, Postgres};
 use thiserror::Error;
 use ulid::Ulid;
 use uuid::Uuid;
 
 use super::client::{lookup_client, ClientFetchError};
-use crate::{DatabaseInconsistencyError, PostgresqlBackend};
+use crate::{Clock, DatabaseInconsistencyError, PostgresqlBackend};
 
 #[tracing::instrument(
     skip_all,
@@ -35,13 +36,15 @@ use crate::{DatabaseInconsistencyError, PostgresqlBackend};
 )]
 pub async fn add_access_token(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     session: &Session<PostgresqlBackend>,
     access_token: String,
     expires_after: Duration,
 ) -> Result<AccessToken<PostgresqlBackend>, anyhow::Error> {
-    let created_at = Utc::now();
+    let created_at = clock.now();
     let expires_at = created_at + expires_after;
-    let id = Ulid::from_datetime(created_at.into());
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
 
     tracing::Span::current().record("access_token.id", tracing::field::display(id));
 
@@ -243,9 +246,10 @@ where
 )]
 pub async fn revoke_access_token(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     access_token: AccessToken<PostgresqlBackend>,
 ) -> anyhow::Result<()> {
-    let revoked_at = Utc::now();
+    let revoked_at = clock.now();
     let res = sqlx::query!(
         r#"
             UPDATE oauth2_access_tokens
@@ -266,9 +270,9 @@ pub async fn revoke_access_token(
     }
 }
 
-pub async fn cleanup_expired(executor: impl PgExecutor<'_>) -> anyhow::Result<u64> {
+pub async fn cleanup_expired(executor: impl PgExecutor<'_>, clock: &Clock) -> anyhow::Result<u64> {
     // Cleanup token which expired more than 15 minutes ago
-    let threshold = Utc::now() - Duration::minutes(15);
+    let threshold = clock.now() - Duration::minutes(15);
     let res = sqlx::query!(
         r#"
             DELETE FROM oauth2_access_tokens

crates/storage/src/oauth2/authorization_grant.rs

@@ -24,13 +24,14 @@ use mas_data_model::{
 };
 use mas_iana::oauth::PkceCodeChallengeMethod;
 use oauth2_types::{requests::ResponseMode, scope::Scope};
+use rand::Rng;
 use sqlx::{PgConnection, PgExecutor};
 use ulid::Ulid;
 use url::Url;
 use uuid::Uuid;
 
 use super::client::lookup_client;
-use crate::{DatabaseInconsistencyError, PostgresqlBackend};
+use crate::{Clock, DatabaseInconsistencyError, PostgresqlBackend};
 
 #[tracing::instrument(
     skip_all,
@@ -43,6 +44,8 @@ use crate::{DatabaseInconsistencyError, PostgresqlBackend};
 #[allow(clippy::too_many_arguments)]
 pub async fn new_authorization_grant(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     client: Client<PostgresqlBackend>,
     redirect_uri: Url,
     scope: Scope,
@@ -67,8 +70,8 @@ pub async fn new_authorization_grant(
     let max_age_i32 = max_age.map(|x| i32::try_from(u32::from(x)).unwrap_or(i32::MAX));
     let code_str = code.as_ref().map(|c| &c.code);
 
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("grant.id", tracing::field::display(id));
 
     sqlx::query!(
@@ -504,11 +507,13 @@ pub async fn lookup_grant_by_code(
 )]
 pub async fn derive_session(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     grant: &AuthorizationGrant<PostgresqlBackend>,
     browser_session: BrowserSession<PostgresqlBackend>,
 ) -> Result<Session<PostgresqlBackend>, anyhow::Error> {
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("session.id", tracing::field::display(id));
 
     sqlx::query!(
@@ -623,9 +628,10 @@ pub async fn give_consent_to_grant(
 )]
 pub async fn exchange_grant(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     mut grant: AuthorizationGrant<PostgresqlBackend>,
 ) -> Result<AuthorizationGrant<PostgresqlBackend>, anyhow::Error> {
-    let exchanged_at = Utc::now();
+    let exchanged_at = clock.now();
     sqlx::query!(
         r#"
             UPDATE oauth2_authorization_grants

crates/storage/src/oauth2/consent.rs

@@ -14,14 +14,14 @@
 
 use std::str::FromStr;
 
-use chrono::Utc;
 use mas_data_model::{Client, User};
 use oauth2_types::scope::{Scope, ScopeToken};
+use rand::Rng;
 use sqlx::PgExecutor;
 use ulid::Ulid;
 use uuid::Uuid;
 
-use crate::PostgresqlBackend;
+use crate::{Clock, PostgresqlBackend};
 
 #[tracing::instrument(
     skip_all,
@@ -67,17 +67,19 @@ pub async fn fetch_client_consent(
 )]
 pub async fn insert_client_consent(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     user: &User<PostgresqlBackend>,
     client: &Client<PostgresqlBackend>,
     scope: &Scope,
 ) -> Result<(), anyhow::Error> {
-    let now = Utc::now();
+    let now = clock.now();
     let (tokens, ids): (Vec<String>, Vec<Uuid>) = scope
         .iter()
         .map(|token| {
             (
                 token.to_string(),
-                Uuid::from(Ulid::from_datetime(now.into())),
+                Uuid::from(Ulid::from_datetime_with_source(now.into(), &mut rng)),
             )
         })
         .unzip();

crates/storage/src/oauth2/mod.rs

@@ -12,12 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use chrono::Utc;
 use mas_data_model::Session;
 use sqlx::PgExecutor;
 use uuid::Uuid;
 
-use crate::PostgresqlBackend;
+use crate::{Clock, PostgresqlBackend};
 
 pub mod access_token;
 pub mod authorization_grant;
@@ -37,9 +36,10 @@ pub mod refresh_token;
 )]
 pub async fn end_oauth_session(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     session: Session<PostgresqlBackend>,
 ) -> Result<(), anyhow::Error> {
-    let finished_at = Utc::now();
+    let finished_at = clock.now();
     let res = sqlx::query!(
         r#"
             UPDATE oauth2_sessions

crates/storage/src/oauth2/refresh_token.rs

@@ -17,13 +17,14 @@ use chrono::{DateTime, Utc};
 use mas_data_model::{
     AccessToken, Authentication, BrowserSession, RefreshToken, Session, User, UserEmail,
 };
+use rand::Rng;
 use sqlx::{PgConnection, PgExecutor};
 use thiserror::Error;
 use ulid::Ulid;
 use uuid::Uuid;
 
 use super::client::{lookup_client, ClientFetchError};
-use crate::{DatabaseInconsistencyError, PostgresqlBackend};
+use crate::{Clock, DatabaseInconsistencyError, PostgresqlBackend};
 
 #[tracing::instrument(
     skip_all,
@@ -38,12 +39,14 @@ use crate::{DatabaseInconsistencyError, PostgresqlBackend};
 )]
 pub async fn add_refresh_token(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     session: &Session<PostgresqlBackend>,
     access_token: AccessToken<PostgresqlBackend>,
     refresh_token: String,
 ) -> anyhow::Result<RefreshToken<PostgresqlBackend>> {
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("refresh_token.id", tracing::field::display(id));
 
     sqlx::query!(
@@ -263,9 +266,10 @@ pub async fn lookup_active_refresh_token(
 )]
 pub async fn consume_refresh_token(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     refresh_token: &RefreshToken<PostgresqlBackend>,
 ) -> Result<(), anyhow::Error> {
-    let consumed_at = Utc::now();
+    let consumed_at = clock.now();
     let res = sqlx::query!(
         r#"
             UPDATE oauth2_refresh_tokens

crates/storage/src/user.rs

@@ -22,7 +22,7 @@ use mas_data_model::{
     UserEmailVerificationState,
 };
 use password_hash::{PasswordHash, PasswordHasher, SaltString};
-use rand::thread_rng;
+use rand::{CryptoRng, Rng};
 use sqlx::{Acquire, PgExecutor, Postgres, Transaction};
 use thiserror::Error;
 use tokio::task;
@@ -31,6 +31,7 @@ use ulid::Ulid;
 use uuid::Uuid;
 
 use super::{DatabaseInconsistencyError, PostgresqlBackend};
+use crate::Clock;
 
 #[derive(Debug, Clone)]
 struct UserLookup {
@@ -68,7 +69,9 @@ pub enum LoginError {
     err,
 )]
 pub async fn login(
-    conn: impl Acquire<'_, Database = Postgres>,
+    conn: impl Acquire<'_, Database = Postgres> + Send,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     username: &str,
     password: &str,
 ) -> Result<BrowserSession<PostgresqlBackend>, LoginError> {
@@ -86,8 +89,8 @@ pub async fn login(
             }
         })?;
 
-    let mut session = start_session(&mut txn, user).await?;
-    authenticate_session(&mut txn, &mut session, password)
+    let mut session = start_session(&mut txn, &mut rng, clock, user).await?;
+    authenticate_session(&mut txn, &mut rng, clock, &mut session, password)
         .await
         .map_err(|source| {
             if matches!(source, AuthenticationError::Password { .. }) {
@@ -230,10 +233,12 @@ pub async fn lookup_active_session(
 )]
 pub async fn start_session(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     user: User<PostgresqlBackend>,
 ) -> Result<BrowserSession<PostgresqlBackend>, anyhow::Error> {
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("user_session.id", tracing::field::display(id));
 
     sqlx::query!(
@@ -301,13 +306,16 @@ pub enum AuthenticationError {
 #[tracing::instrument(
     skip_all,
     fields(
-        session.id = %session.data,
-        user.id = %session.user.data
+        user.id = %session.user.data,
+        user_session.id = %session.data,
+        user_session_authentication.id,
     ),
     err,
 )]
 pub async fn authenticate_session(
     txn: &mut Transaction<'_, Postgres>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     session: &mut BrowserSession<PostgresqlBackend>,
     password: &str,
 ) -> Result<(), AuthenticationError> {
@@ -341,8 +349,13 @@ pub async fn authenticate_session(
     .await??;
 
     // That went well, let's insert the auth info
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
+    tracing::Span::current().record(
+        "user_session_authentication.id",
+        tracing::field::display(id),
+    );
+
     sqlx::query!(
         r#"
             INSERT INTO user_session_authentications
@@ -376,12 +389,14 @@ pub async fn authenticate_session(
 )]
 pub async fn register_user(
     txn: &mut Transaction<'_, Postgres>,
-    phf: impl PasswordHasher,
+    mut rng: impl CryptoRng + Rng + Send,
+    clock: &Clock,
+    phf: impl PasswordHasher + Send,
     username: &str,
     password: &str,
 ) -> Result<User<PostgresqlBackend>, anyhow::Error> {
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("user.id", tracing::field::display(id));
 
     sqlx::query!(
@@ -405,7 +420,7 @@ pub async fn register_user(
         primary_email: None,
     };
 
-    set_password(txn.borrow_mut(), phf, &user, password).await?;
+    set_password(txn.borrow_mut(), &mut rng, clock, phf, &user, password).await?;
 
     Ok(user)
 }
@@ -420,15 +435,17 @@ pub async fn register_user(
 )]
 pub async fn set_password(
     executor: impl PgExecutor<'_>,
-    phf: impl PasswordHasher,
+    mut rng: impl CryptoRng + Rng + Send,
+    clock: &Clock,
+    phf: impl PasswordHasher + Send,
     user: &User<PostgresqlBackend>,
     password: &str,
 ) -> Result<(), anyhow::Error> {
-    let created_at = Utc::now();
+    let created_at = clock.now();
     let id = Ulid::from_datetime(created_at.into());
     tracing::Span::current().record("user_password.id", tracing::field::display(id));
 
-    let salt = SaltString::generate(thread_rng());
+    let salt = SaltString::generate(&mut rng);
     let hashed_password = PasswordHash::generate(phf, password, salt.as_str())?;
 
     sqlx::query_scalar!(
@@ -456,9 +473,10 @@ pub async fn set_password(
 )]
 pub async fn end_session(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     session: &BrowserSession<PostgresqlBackend>,
 ) -> Result<(), anyhow::Error> {
-    let now = Utc::now();
+    let now = clock.now();
     let res = sqlx::query!(
         r#"
             UPDATE user_sessions
@@ -672,11 +690,13 @@ pub async fn get_user_email(
 )]
 pub async fn add_user_email(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     user: &User<PostgresqlBackend>,
     email: String,
 ) -> Result<UserEmail<PostgresqlBackend>, anyhow::Error> {
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("user_email.id", tracing::field::display(id));
 
     sqlx::query!(
@@ -842,9 +862,10 @@ pub async fn lookup_user_email_by_id(
 )]
 pub async fn mark_user_email_as_verified(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     mut email: UserEmail<PostgresqlBackend>,
 ) -> Result<UserEmail<PostgresqlBackend>, anyhow::Error> {
-    let confirmed_at = Utc::now();
+    let confirmed_at = clock.now();
     sqlx::query!(
         r#"
             UPDATE user_emails
@@ -881,10 +902,11 @@ struct UserEmailConfirmationCodeLookup {
 )]
 pub async fn lookup_user_email_verification_code(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     email: UserEmail<PostgresqlBackend>,
     code: &str,
 ) -> Result<UserEmailVerification<PostgresqlBackend>, anyhow::Error> {
-    let now = Utc::now();
+    let now = clock.now();
 
     let res = sqlx::query_as!(
         UserEmailConfirmationCodeLookup,
@@ -935,13 +957,14 @@ pub async fn lookup_user_email_verification_code(
 )]
 pub async fn consume_email_verification(
     executor: impl PgExecutor<'_>,
+    clock: &Clock,
     mut verification: UserEmailVerification<PostgresqlBackend>,
 ) -> Result<UserEmailVerification<PostgresqlBackend>, anyhow::Error> {
     if !matches!(verification.state, UserEmailVerificationState::Valid) {
         bail!("user email verification in wrong state");
     }
 
-    let consumed_at = Utc::now();
+    let consumed_at = clock.now();
 
     sqlx::query!(
         r#"
@@ -974,12 +997,14 @@ pub async fn consume_email_verification(
 )]
 pub async fn add_user_email_verification_code(
     executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
     email: UserEmail<PostgresqlBackend>,
     max_age: chrono::Duration,
     code: String,
 ) -> Result<UserEmailVerification<PostgresqlBackend>, anyhow::Error> {
-    let created_at = Utc::now();
-    let id = Ulid::from_datetime(created_at.into());
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
     tracing::Span::current().record("user_email_confirmation.id", tracing::field::display(id));
     let expires_at = created_at + max_age;
 
@@ -1013,23 +1038,27 @@ pub async fn add_user_email_verification_code(
 
 #[cfg(test)]
 mod tests {
+    use rand::SeedableRng;
+
     use super::*;
 
     #[sqlx::test(migrator = "crate::MIGRATOR")]
     async fn test_user_registration_and_login(pool: sqlx::PgPool) -> anyhow::Result<()> {
+        let clock = Clock::default();
+        let mut rng = rand_chacha::ChaChaRng::seed_from_u64(42);
         let mut txn = pool.begin().await?;
 
         let exists = username_exists(&mut txn, "john").await?;
         assert!(!exists);
 
         let hasher = Argon2::default();
-        let user = register_user(&mut txn, hasher, "john", "hunter2").await?;
+        let user = register_user(&mut txn, &mut rng, &clock, hasher, "john", "hunter2").await?;
         assert_eq!(user.username, "john");
 
         let exists = username_exists(&mut txn, "john").await?;
         assert!(exists);
 
-        let session = login(&mut txn, "john", "hunter2").await?;
+        let session = login(&mut txn, &mut rng, &clock, "john", "hunter2").await?;
         assert_eq!(session.user.data, user.data);
 
         let user2 = lookup_user_by_username(&mut txn, "john").await?;