Make password hash iterations configurable

Move password hash iterations to a system var.

Defaults to 600_000 and uses 600_000 for
all system users.

Author: jubrad

src/adapter/src/catalog/open.rs

@@ -324,7 +324,10 @@ impl Catalog {
             if let Some(password) = config.external_login_password_mz_system {
                 let role_auth = RoleAuth {
                     role_id: MZ_SYSTEM_ROLE_ID,
-                    password_hash: Some(scram256_hash(&password).map_err(|_| {
+                    // The external login password for mz_system should use
+                    // the default hash iterations as that is a known secure
+                    // setting.
+                    password_hash: Some(scram256_hash(&password, &None).map_err(|_| {
                         AdapterError::Internal("Failed to hash mz_system password.".to_owned())
                     })?),
                     updated_at: SYSTEM_TIME(),
@@ -1097,17 +1100,17 @@ fn add_new_remove_old_builtin_cluster_replicas_migration(
             .get_mut(&cluster.id)
             .unwrap_or(&mut empty_map);
 
-        let builtin_cluster_boostrap_config =
+        let builtin_cluster_bootstrap_config =
             builtin_cluster_config_map.get_config(builtin_replica.cluster_name)?;
         if replica_names.remove(builtin_replica.name).is_none()
             // NOTE(SangJunBak): We need to explicitly check the replication factor because
             // BUILT_IN_CLUSTER_REPLICAS is constant throughout all deployments but the replication
             // factor is configurable on bootstrap.
-            && builtin_cluster_boostrap_config.replication_factor > 0
+            && builtin_cluster_bootstrap_config.replication_factor > 0
         {
             let replica_size = match cluster.config.variant {
                 ClusterVariant::Managed(ClusterVariantManaged { ref size, .. }) => size.clone(),
-                ClusterVariant::Unmanaged => builtin_cluster_boostrap_config.size,
+                ClusterVariant::Unmanaged => builtin_cluster_bootstrap_config.size,
             };
 
             let config = builtin_cluster_replica_config(replica_size);

src/adapter/src/catalog/transact.rs

@@ -46,7 +46,7 @@ use mz_repr::{CatalogItemId, ColumnName, Diff, GlobalId, SqlColumnType, strconv}
 use mz_sql::ast::RawDataType;
 use mz_sql::catalog::{
     CatalogDatabase, CatalogError as SqlCatalogError, CatalogItem as SqlCatalogItem, CatalogRole,
-    CatalogSchema, DefaultPrivilegeAclItem, DefaultPrivilegeObject, PasswordAction,
+    CatalogSchema, DefaultPrivilegeAclItem, DefaultPrivilegeObject, PasswordAction, PasswordConfig,
     RoleAttributesRaw, RoleMembership, RoleVars,
 };
 use mz_sql::names::{
@@ -680,12 +680,17 @@ impl Catalog {
 
                 let mut existing_role = state.get_role(&id).clone();
                 let password = attributes.password.clone();
+                let non_default_password_hash_iterations =
+                    attributes.non_default_password_hash_iterations.clone();
                 existing_role.attributes = attributes.into();
                 existing_role.vars = vars;
                 let password_action = if nopassword {
                     PasswordAction::Clear
                 } else if let Some(password) = password {
-                    PasswordAction::Set(password)
+                    PasswordAction::Set(PasswordConfig {
+                        password,
+                        non_default_password_hash_iterations,
+                    })
                 } else {
                     PasswordAction::NoChange
                 };

src/adapter/src/coord/command_handler.rs

@@ -363,7 +363,15 @@ impl Coordinator {
              mock_nonce: String,
              nonce: String,
              tx: oneshot::Sender<Result<SASLChallengeResponse, AdapterError>>| {
-                let opts = mz_auth::hash::mock_sasl_challenge(&role_name, &mock_nonce);
+                let opts = mz_auth::hash::mock_sasl_challenge(
+                    &role_name,
+                    &mock_nonce,
+                    &Some(
+                        self.catalog()
+                            .system_config()
+                            .default_password_hash_iterations(),
+                    ),
+                );
                 let _ = tx.send(Ok(SASLChallengeResponse {
                     iteration_count: mz_ore::cast::u32_to_usize(opts.iterations.get()),
                     salt: BASE64_STANDARD.encode(opts.salt),

src/adapter/src/coord/sequencer/inner.rs

@@ -3530,6 +3530,11 @@ impl Coordinator {
 
                 if let Some(password) = attrs.password {
                     attributes.password = Some(password);
+                    attributes.non_default_password_hash_iterations = Some(
+                        self.catalog()
+                            .system_config()
+                            .default_password_hash_iterations(),
+                    )
                 }
 
                 if let Some(superuser) = attrs.superuser {

src/auth/src/hash.rs

@@ -18,15 +18,15 @@ use itertools::Itertools;
 
 use crate::password::Password;
 
-/// The default iteration count as suggested by
-/// <https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html>
-const DEFAULT_ITERATIONS: NonZeroU32 = NonZeroU32::new(600_000).unwrap();
-
 /// The default salt size, which isn't currently configurable.
 const DEFAULT_SALT_SIZE: usize = 32;
 
 const SHA256_OUTPUT_LEN: usize = 32;
 
+/// The default iteration count as suggested by
+/// <https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html>
+const DEFAULT_ITERATIONS: NonZeroU32 = NonZeroU32::new(600_000).unwrap();
+
 /// The options for hashing a password
 #[derive(Debug, PartialEq)]
 pub struct HashOpts {
@@ -70,21 +70,22 @@ impl Display for HashError {
 
 /// Hashes a password using PBKDF2 with SHA256
 /// and a random salt.
-pub fn hash_password(password: &Password) -> Result<PasswordHash, HashError> {
+pub fn hash_password(
+    password: &Password,
+    iterations: &Option<NonZeroU32>,
+) -> Result<PasswordHash, HashError> {
     let mut salt = [0u8; DEFAULT_SALT_SIZE];
     openssl::rand::rand_bytes(&mut salt).map_err(HashError::Openssl)?;
+    let iterations = iterations.unwrap_or(DEFAULT_ITERATIONS);
 
     let hash = hash_password_inner(
-        &HashOpts {
-            iterations: DEFAULT_ITERATIONS,
-            salt,
-        },
+        &HashOpts { iterations, salt },
         password.to_string().as_bytes(),
     )?;
 
     Ok(PasswordHash {
         salt,
-        iterations: DEFAULT_ITERATIONS,
+        iterations,
         hash,
     })
 }
@@ -115,8 +116,11 @@ pub fn hash_password_with_opts(
 /// Hashes a password using PBKDF2 with SHA256,
 /// and returns it in the SCRAM-SHA-256 format.
 /// The format is SCRAM-SHA-256$<iterations>:<salt>$<stored_key>:<server_key>
-pub fn scram256_hash(password: &Password) -> Result<String, HashError> {
-    let hashed_password = hash_password(password)?;
+pub fn scram256_hash(
+    password: &Password,
+    iterations: &Option<NonZeroU32>,
+) -> Result<String, HashError> {
+    let hashed_password = hash_password(password, iterations)?;
     Ok(scram256_hash_inner(hashed_password).to_string())
 }
 
@@ -207,14 +211,18 @@ fn generate_signature(key: &[u8], message: &str) -> Result<Vec<u8>, VerifyError>
 // Generate a mock challenge based on the username and client nonce
 // We do this so that we can present a deterministic challenge even for
 // nonexistent users, to avoid user enumeration attacks.
-pub fn mock_sasl_challenge(username: &str, mock_nonce: &str) -> HashOpts {
+pub fn mock_sasl_challenge(
+    username: &str,
+    mock_nonce: &str,
+    iterations: &Option<NonZeroU32>,
+) -> HashOpts {
     let mut buf = Vec::with_capacity(username.len() + mock_nonce.len());
     buf.extend_from_slice(username.as_bytes());
     buf.extend_from_slice(mock_nonce.as_bytes());
     let digest = openssl::sha::sha256(&buf);
 
     HashOpts {
-        iterations: DEFAULT_ITERATIONS,
+        iterations: iterations.unwrap_or(DEFAULT_ITERATIONS),
         salt: digest,
     }
 }
@@ -326,7 +334,11 @@ mod tests {
     #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `OPENSSL_init_ssl` on OS `linux`
     fn test_hash_password() {
         let password = "password".to_string();
-        let hashed_password = hash_password(&password.into()).expect("Failed to hash password");
+        let hashed_password = hash_password(
+            &password.into(),
+            &Some(NonZeroU32::new(100).expect("Trust me on this")),
+        )
+        .expect("Failed to hash password");
         assert_eq!(hashed_password.iterations, DEFAULT_ITERATIONS);
         assert_eq!(hashed_password.salt.len(), DEFAULT_SALT_SIZE);
         assert_eq!(hashed_password.hash.len(), SHA256_OUTPUT_LEN);
@@ -336,7 +348,7 @@ mod tests {
     #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `OPENSSL_init_ssl` on OS `linux`
     fn test_scram256_hash() {
         let password = "password".into();
-        let scram_hash = scram256_hash(&password).expect("Failed to hash password");
+        let scram_hash = scram256_hash(&password, &None).expect("Failed to hash password");
 
         let res = scram256_verify(&password, &scram_hash);
         assert!(res.is_ok());
@@ -363,16 +375,16 @@ mod tests {
     fn test_mock_sasl_challenge() {
         let username = "alice";
         let mock = "cnonce";
-        let opts1 = mock_sasl_challenge(username, mock);
-        let opts2 = mock_sasl_challenge(username, mock);
+        let opts1 = mock_sasl_challenge(username, mock, &None);
+        let opts2 = mock_sasl_challenge(username, mock, &None);
         assert_eq!(opts1, opts2);
     }
 
     #[mz_ore::test]
     #[cfg_attr(miri, ignore)]
     fn test_sasl_verify_success() {
         let password: Password = "password".into();
-        let hashed_password = scram256_hash(&password).expect("hash password");
+        let hashed_password = scram256_hash(&password, &None).expect("hash password");
         let auth_message = "n=user,r=clientnonce,s=somesalt"; // arbitrary auth message
 
         // Parse client_key and server_key from the SCRAM hash
@@ -426,7 +438,7 @@ mod tests {
     #[cfg_attr(miri, ignore)]
     fn test_sasl_verify_invalid_proof() {
         let password: Password = "password".into();
-        let hashed_password = scram256_hash(&password).expect("hash password");
+        let hashed_password = scram256_hash(&password, &None).expect("hash password");
         let auth_message = "n=user,r=clientnonce,s=somesalt";
         // Provide an obviously invalid base64 proof (different size / random)
         let bad_proof = BASE64_STANDARD.encode([0u8; 32]);

src/catalog/src/durable/transaction.rs

@@ -357,8 +357,11 @@ impl<'a> Transaction<'a> {
         oid: u32,
     ) -> Result<(), CatalogError> {
         if let Some(ref password) = attributes.password {
-            let hash =
-                mz_auth::hash::scram256_hash(password).expect("password hash should be valid");
+            let hash = mz_auth::hash::scram256_hash(
+                password,
+                &attributes.non_default_password_hash_iterations,
+            )
+            .expect("password hash should be valid");
             match self.role_auth.insert(
                 RoleAuthKey { role_id: id },
                 RoleAuthValue {
@@ -1489,8 +1492,11 @@ impl<'a> Transaction<'a> {
 
             match password {
                 PasswordAction::Set(new_password) => {
-                    let hash = mz_auth::hash::scram256_hash(&new_password)
-                        .expect("password hash should be valid");
+                    let hash = mz_auth::hash::scram256_hash(
+                        &new_password.password,
+                        &new_password.non_default_password_hash_iterations,
+                    )
+                    .expect("password hash should be valid");
                     let value = RoleAuthValue {
                         password_hash: Some(hash),
                         updated_at: SYSTEM_TIME(),

src/sql/src/catalog.rs

@@ -16,6 +16,7 @@ use std::collections::{BTreeMap, BTreeSet};
 use std::error::Error;
 use std::fmt;
 use std::fmt::{Debug, Display, Formatter};
+use std::num::NonZeroU32;
 use std::str::FromStr;
 use std::sync::LazyLock;
 use std::time::{Duration, Instant};
@@ -486,11 +487,20 @@ pub trait CatalogSchema {
     fn privileges(&self) -> &PrivilegeMap;
 }
 
+/// Parameters used to modify password
+#[derive(Debug, Clone, Eq, PartialEq, Arbitrary)]
+pub struct PasswordConfig {
+    /// The Password.
+    pub password: Password,
+    /// a non default iteration count for hashing the password.
+    pub non_default_password_hash_iterations: Option<NonZeroU32>,
+}
+
 /// A modification of a role password in the catalog
 #[derive(Debug, Clone, Eq, PartialEq, Arbitrary)]
 pub enum PasswordAction {
     /// Set a new password.
-    Set(Password),
+    Set(PasswordConfig),
     /// Remove the existing password.
     Clear,
     /// Leave the existing password unchanged.
@@ -499,14 +509,16 @@ pub enum PasswordAction {
 
 /// A raw representation of attributes belonging to a [`CatalogRole`] that we might
 /// get as input from the user. This includes the password.
-/// This struct explicity does not implement `Serialize` or `Deserialize` to avoid
+/// This struct explicitly does not implement `Serialize` or `Deserialize` to avoid
 /// accidentally serializing passwords.
 #[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Arbitrary)]
 pub struct RoleAttributesRaw {
     /// Indicates whether the role has inheritance of privileges.
     pub inherit: bool,
     /// The raw password of the role. This is for self managed auth, not cloud.
     pub password: Option<Password>,
+    /// An optoinal override of the hash iterations used to securely store passwords. This is for self-managed auth
+    pub non_default_password_hash_iterations: Option<NonZeroU32>,
     /// Whether or not this user is a superuser.
     pub superuser: Option<bool>,
     /// Whether this role is login
@@ -534,6 +546,7 @@ impl RoleAttributesRaw {
         RoleAttributesRaw {
             inherit: true,
             password: None,
+            non_default_password_hash_iterations: None,
             superuser: None,
             login: None,
             _private: (),
@@ -604,6 +617,7 @@ impl From<RoleAttributes> for RoleAttributesRaw {
         RoleAttributesRaw {
             inherit,
             password: None,
+            non_default_password_hash_iterations: None,
             superuser,
             login,
             _private: (),
@@ -616,6 +630,7 @@ impl From<PlannedRoleAttributes> for RoleAttributesRaw {
         PlannedRoleAttributes {
             inherit,
             password,
+            non_default_password_hash_iterations,
             superuser,
             login,
             ..
@@ -625,6 +640,7 @@ impl From<PlannedRoleAttributes> for RoleAttributesRaw {
         RoleAttributesRaw {
             inherit: inherit.unwrap_or(default_attributes.inherit),
             password,
+            non_default_password_hash_iterations,
             superuser,
             login,
             _private: (),