feat(security): add password verifier formats (#8251)

* feat: add password verifier formats

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: harden password verifier parsing and auth config errors

- Reject pbkdf2_sha256 verifiers whose hash is not 32 bytes and bound the
  salt length, preventing short-hash verifiers from matching on a prefix.
- Verify pbkdf2_sha256 with a stack-allocated buffer.
- Report only the length, not the bytes, when a mysql native password
  verifier has an illegal length.
- Map empty frontend_auth credentials to an invalid-config error instead
  of an internal error.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* chore: update config.md

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: skip non-plain verifiers in get_one_user_pwd

Pick the first plain-text credential instead of failing when the first
user happens to hold a hashed verifier.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: format

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* chore: remove unused get_one_user_pwd

Internal flownode-to-frontend communication no longer authenticates
(see #8244), so the plain-text credential export path is dead code.
Drop get_one_user_pwd, its now-orphan as_plain_text helper, and the
related tests.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

Cargo.lock

@@ -947,9 +947,13 @@ dependencies = [
  "common-telemetry",
  "common-test-util",
  "digest 0.10.7",
+ "hex",
+ "pbkdf2",
  "sha1 0.10.6",
+ "sha2 0.10.9",
  "snafu 0.8.6",
  "sql",
+ "subtle",
  "tokio",
 ]
 

config/config.md

@@ -15,7 +15,7 @@
 | `default_timezone` | String | Unset | The default timezone of the server. |
 | `default_column_prefix` | String | Unset | The default column prefix for auto-created time index and value columns. |
 | `auto_create_table` | Bool | `true` | Server-side global switch for auto table creation on write.<br/>When `false`, a missing table is never auto-created even if the request sets the `auto_create_table` hint to `true`. Default: `true`. |
-| `user_provider` | String | Unset | The user provider for authentication.<br/>Examples: "static_user_provider:file:/path/to/users", "static_user_provider:cmd:greptime_user=greptime_pwd" |
+| `user_provider` | String | Unset | The user provider for authentication.<br/>Examples: "static_user_provider:file:/path/to/users", "static_user_provider:cmd:greptime_user=greptime_pwd"<br/>Password verifier formats: "plain:<password>", "pbkdf2_sha256:<iterations>:<hex_salt>:<hex_hash>",<br/>"mysql_native_password:<hex_sha1_sha1_password>"<br/>"pbkdf2_sha256" protects passwords at rest but is not compatible with mysql_native_password. |
 | `max_in_flight_write_bytes` | String | Unset | Maximum total memory for all concurrent write request bodies and messages (HTTP, gRPC, Flight).<br/>Set to 0 to disable the limit. Default: "0" (unlimited) |
 | `write_bytes_exhausted_policy` | String | Unset | Policy when write bytes quota is exhausted.<br/>Options: "wait" (default, 10s timeout), "wait(<duration>)" (e.g., "wait(30s)"), "fail" |
 | `init_regions_in_background` | Bool | `false` | Initialize all regions in the background during the startup.<br/>By default, it provides services after all regions have been initialized. |
@@ -233,7 +233,7 @@
 | `default_timezone` | String | Unset | The default timezone of the server. |
 | `default_column_prefix` | String | Unset | The default column prefix for auto-created time index and value columns. |
 | `auto_create_table` | Bool | `true` | Server-side global switch for auto table creation on write.<br/>When `false`, a missing table is never auto-created even if the request sets the `auto_create_table` hint to `true`. Default: `true`. |
-| `user_provider` | String | Unset | The user provider for authentication.<br/>Examples: "static_user_provider:file:/path/to/users", "static_user_provider:cmd:greptime_user=greptime_pwd" |
+| `user_provider` | String | Unset | The user provider for authentication.<br/>Examples: "static_user_provider:file:/path/to/users", "static_user_provider:cmd:greptime_user=greptime_pwd"<br/>Password verifier formats: "plain:<password>", "pbkdf2_sha256:<iterations>:<hex_salt>:<hex_hash>",<br/>"mysql_native_password:<hex_sha1_sha1_password>"<br/>"pbkdf2_sha256" protects passwords at rest but is not compatible with mysql_native_password. |
 | `max_in_flight_write_bytes` | String | Unset | Maximum total memory for all concurrent write request bodies and messages (HTTP, gRPC, Flight).<br/>Set to 0 to disable the limit. Default: "0" (unlimited) |
 | `write_bytes_exhausted_policy` | String | Unset | Policy when write bytes quota is exhausted.<br/>Options: "wait" (default, 10s timeout), "wait(<duration>)" (e.g., "wait(30s)"), "fail" |
 | `runtime` | -- | -- | The runtime options. |

config/frontend.example.toml

@@ -12,6 +12,9 @@ default_column_prefix = "greptime"
 
 ## The user provider for authentication.
 ## Examples: "static_user_provider:file:/path/to/users", "static_user_provider:cmd:greptime_user=greptime_pwd"
+## Password verifier formats: "plain:<password>", "pbkdf2_sha256:<iterations>:<hex_salt>:<hex_hash>",
+## "mysql_native_password:<hex_sha1_sha1_password>"
+## "pbkdf2_sha256" protects passwords at rest but is not compatible with mysql_native_password.
 ## @toml2docs:none-default
 #+ user_provider = "static_user_provider:file:/path/to/users"
 

config/standalone.example.toml

@@ -12,6 +12,9 @@ default_column_prefix = "greptime"
 
 ## The user provider for authentication.
 ## Examples: "static_user_provider:file:/path/to/users", "static_user_provider:cmd:greptime_user=greptime_pwd"
+## Password verifier formats: "plain:<password>", "pbkdf2_sha256:<iterations>:<hex_salt>:<hex_hash>",
+## "mysql_native_password:<hex_sha1_sha1_password>"
+## "pbkdf2_sha256" protects passwords at rest but is not compatible with mysql_native_password.
 ## @toml2docs:none-default
 #+ user_provider = "static_user_provider:file:/path/to/users"
 

src/auth/Cargo.toml

@@ -20,9 +20,13 @@ common-error.workspace = true
 common-macro.workspace = true
 common-telemetry.workspace = true
 digest = "0.10"
+hex.workspace = true
+pbkdf2 = "0.12"
 sha1 = "0.10"
+sha2 = "0.10"
 snafu.workspace = true
 sql.workspace = true
+subtle = "2.6"
 tokio.workspace = true
 
 [dev-dependencies]

src/auth/src/common.rs

@@ -108,6 +108,16 @@ pub fn auth_mysql(
     salt: Salt,
     username: &str,
     save_pwd: &[u8],
+) -> Result<()> {
+    let hash_stage_2 = mysql_native_password_hash(save_pwd);
+    auth_mysql_with_hash_stage_2(auth_data, salt, username, &hash_stage_2)
+}
+
+pub(crate) fn auth_mysql_with_hash_stage_2(
+    auth_data: HashedPassword,
+    salt: Salt,
+    username: &str,
+    hash_stage_2: &[u8],
 ) -> Result<()> {
     ensure!(
         auth_data.len() == 20,
@@ -115,9 +125,15 @@ pub fn auth_mysql(
             msg: "Illegal mysql password length"
         }
     );
+    ensure!(
+        hash_stage_2.len() == 20,
+        InvalidConfigSnafu {
+            value: hash_stage_2.len().to_string(),
+            msg: "Illegal mysql native password verifier length",
+        }
+    );
     // ref: https://github.com/mysql/mysql-server/blob/a246bad76b9271cb4333634e954040a970222e0a/sql/auth/password.cc#L62
-    let hash_stage_2 = double_sha1(save_pwd);
-    let tmp = sha1_two(salt, &hash_stage_2);
+    let tmp = sha1_two(salt, hash_stage_2);
     // xor auth_data and tmp
     let mut xor_result = [0u8; 20];
     for i in 0..20 {
@@ -134,6 +150,10 @@ pub fn auth_mysql(
     }
 }
 
+pub(crate) fn mysql_native_password_hash(save_pwd: &[u8]) -> Vec<u8> {
+    double_sha1(save_pwd)
+}
+
 fn sha1_two(input_1: &[u8], input_2: &[u8]) -> Vec<u8> {
     let mut hasher = Sha1::new();
     hasher.update(input_1);

src/auth/src/user_provider.rs

@@ -17,14 +17,17 @@ pub(crate) mod watch_file_user_provider;
 
 use std::collections::HashMap;
 use std::fs::File;
-use std::io;
 use std::io::BufRead;
 use std::path::Path;
+use std::{fmt, io};
 
 use common_base::secrets::ExposeSecret;
+use pbkdf2::pbkdf2_hmac;
+use sha2::Sha256;
 use snafu::{OptionExt, ResultExt, ensure};
+use subtle::ConstantTimeEq;
 
-use crate::common::{Identity, Password};
+use crate::common::{Identity, Password, auth_mysql_with_hash_stage_2};
 use crate::error::{
     IllegalParamSnafu, InvalidConfigSnafu, IoSnafu, Result, UnsupportedPasswordTypeSnafu,
     UserNotFoundSnafu, UserPasswordMismatchSnafu,
@@ -32,6 +35,12 @@ use crate::error::{
 use crate::user_info::{DefaultUserInfo, PermissionMode};
 use crate::{UserInfoRef, auth_mysql};
 
+const MAX_PBKDF2_SHA256_ITERATIONS: u32 = 1_000_000;
+/// PBKDF2-SHA256 derived key length, fixed to the native SHA-256 output size.
+const PBKDF2_SHA256_HASH_LEN: usize = 32;
+/// Upper bound on the salt length to reject misconfigured credentials.
+const MAX_PBKDF2_SHA256_SALT_LEN: usize = 1024;
+
 #[async_trait::async_trait]
 pub trait UserProvider: Send + Sync {
     fn name(&self) -> &str;
@@ -64,9 +73,124 @@ pub trait UserProvider: Send + Sync {
     }
 }
 
-/// Type alias for user info map
-/// Key is username, value is (password, permission_mode)
-pub type UserInfoMap = HashMap<String, (Vec<u8>, PermissionMode)>;
+#[derive(Clone, PartialEq, Eq)]
+pub(crate) enum PasswordVerifier {
+    PlainText(String),
+    Pbkdf2Sha256 {
+        iterations: u32,
+        salt: Vec<u8>,
+        hash: Vec<u8>,
+    },
+    MysqlNativePassword {
+        hash_stage_2: Vec<u8>,
+    },
+}
+
+impl fmt::Debug for PasswordVerifier {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            PasswordVerifier::PlainText(_) => {
+                f.debug_tuple("PlainText").field(&"<REDACTED>").finish()
+            }
+            PasswordVerifier::Pbkdf2Sha256 { iterations, .. } => f
+                .debug_struct("Pbkdf2Sha256")
+                .field("iterations", iterations)
+                .field("salt", &"<REDACTED>")
+                .field("hash", &"<REDACTED>")
+                .finish(),
+            PasswordVerifier::MysqlNativePassword { .. } => f
+                .debug_struct("MysqlNativePassword")
+                .field("hash_stage_2", &"<REDACTED>")
+                .finish(),
+        }
+    }
+}
+
+impl PasswordVerifier {
+    fn parse(input: &str) -> Option<Self> {
+        if let Some(password) = input.strip_prefix("plain:") {
+            return Some(Self::PlainText(password.to_string()));
+        }
+
+        if let Some(verifier) = input.strip_prefix("pbkdf2_sha256:") {
+            let mut parts = verifier.split(':');
+            let iterations = parts.next()?.parse::<u32>().ok()?;
+            let salt = hex::decode(parts.next()?).ok()?;
+            let hash = hex::decode(parts.next()?).ok()?;
+            if parts.next().is_some()
+                || iterations == 0
+                || iterations > MAX_PBKDF2_SHA256_ITERATIONS
+                || salt.is_empty()
+                || salt.len() > MAX_PBKDF2_SHA256_SALT_LEN
+                || hash.len() != PBKDF2_SHA256_HASH_LEN
+            {
+                return None;
+            }
+
+            return Some(Self::Pbkdf2Sha256 {
+                iterations,
+                salt,
+                hash,
+            });
+        }
+
+        if let Some(verifier) = input.strip_prefix("mysql_native_password:") {
+            let hash_stage_2 = hex::decode(verifier).ok()?;
+            if hash_stage_2.len() != 20 {
+                return None;
+            }
+
+            return Some(Self::MysqlNativePassword { hash_stage_2 });
+        }
+
+        Some(Self::PlainText(input.to_string()))
+    }
+
+    fn verify_plain_text(&self, password: &str) -> bool {
+        match self {
+            PasswordVerifier::PlainText(expected) => {
+                expected.as_bytes().ct_eq(password.as_bytes()).into()
+            }
+            PasswordVerifier::Pbkdf2Sha256 {
+                iterations,
+                salt,
+                hash,
+            } => {
+                if hash.len() != PBKDF2_SHA256_HASH_LEN {
+                    return false;
+                }
+                let mut actual = [0u8; PBKDF2_SHA256_HASH_LEN];
+                pbkdf2_hmac::<Sha256>(password.as_bytes(), salt, *iterations, &mut actual);
+                hash.as_slice().ct_eq(&actual[..]).into()
+            }
+            PasswordVerifier::MysqlNativePassword { .. } => false,
+        }
+    }
+
+    fn verify_mysql_native_password(
+        &self,
+        auth_data: &[u8],
+        salt: &[u8],
+        username: &str,
+    ) -> Result<()> {
+        match self {
+            PasswordVerifier::PlainText(password) => {
+                auth_mysql(auth_data, salt, username, password.as_bytes())
+            }
+            PasswordVerifier::MysqlNativePassword { hash_stage_2 } => {
+                auth_mysql_with_hash_stage_2(auth_data, salt, username, hash_stage_2)
+            }
+            PasswordVerifier::Pbkdf2Sha256 { .. } => UnsupportedPasswordTypeSnafu {
+                password_type: "mysql_native_password_with_pbkdf2_sha256_verifier",
+            }
+            .fail(),
+        }
+    }
+}
+
+/// Type alias for user info map.
+/// Key is username, value is (password verifier, permission_mode).
+pub type UserInfoMap = HashMap<String, (PasswordVerifier, PermissionMode)>;
 
 fn load_credential_from_file(filepath: &str) -> Result<UserInfoMap> {
     // check valid path
@@ -116,8 +240,15 @@ fn load_credential_from_file(filepath: &str) -> Result<UserInfoMap> {
     Ok(credential)
 }
 
-/// Parse a line of credential in the format of `username=password` or `username:permission_mode=password`
-pub(crate) fn parse_credential_line(line: &str) -> Option<(String, (Vec<u8>, PermissionMode))> {
+/// Parse a line of credential in the format of `username=password` or `username:permission_mode=password`.
+///
+/// The password part accepts legacy plain text and explicit verifier formats:
+/// - `plain:<password>`
+/// - `pbkdf2_sha256:<iterations>:<hex-encoded-salt>:<hex-encoded-hash>`
+/// - `mysql_native_password:<hex-encoded-sha1-sha1-password>`
+pub(crate) fn parse_credential_line(
+    line: &str,
+) -> Option<(String, (PasswordVerifier, PermissionMode))> {
     let parts = line.split('=').collect::<Vec<&str>>();
     if parts.len() != 2 {
         return None;
@@ -130,10 +261,9 @@ pub(crate) fn parse_credential_line(line: &str) -> Option<(String, (Vec<u8>, Per
         (username_part, PermissionMode::default())
     };
 
-    Some((
-        username.to_string(),
-        (password.as_bytes().to_vec(), permission_mode),
-    ))
+    let verifier = PasswordVerifier::parse(password)?;
+
+    Some((username.to_string(), (verifier, permission_mode)))
 }
 
 fn authenticate_with_credential(
@@ -149,7 +279,7 @@ fn authenticate_with_credential(
                     msg: "blank username"
                 }
             );
-            let (save_pwd, permission_mode) = users.get(username).context(UserNotFoundSnafu {
+            let (verifier, permission_mode) = users.get(username).context(UserNotFoundSnafu {
                 username: username.to_string(),
             })?;
 
@@ -161,7 +291,7 @@ fn authenticate_with_credential(
                             msg: "blank password"
                         }
                     );
-                    if save_pwd == pwd.expose_secret().as_bytes() {
+                    if verifier.verify_plain_text(pwd.expose_secret()) {
                         Ok(DefaultUserInfo::with_name_and_permission(
                             username,
                             *permission_mode,
@@ -173,11 +303,9 @@ fn authenticate_with_credential(
                         .fail()
                     }
                 }
-                Password::MysqlNativePassword(auth_data, salt) => {
-                    auth_mysql(auth_data, salt, username, save_pwd).map(|_| {
-                        DefaultUserInfo::with_name_and_permission(username, *permission_mode)
-                    })
-                }
+                Password::MysqlNativePassword(auth_data, salt) => verifier
+                    .verify_mysql_native_password(auth_data, salt, username)
+                    .map(|_| DefaultUserInfo::with_name_and_permission(username, *permission_mode)),
                 Password::PgMD5(_, _) => UnsupportedPasswordTypeSnafu {
                     password_type: "pg_md5",
                 }
@@ -188,7 +316,36 @@ fn authenticate_with_credential(
 }
 #[cfg(test)]
 mod tests {
+    use digest::Digest;
+    use sha1::Sha1;
+
     use super::*;
+    use crate::common::mysql_native_password_hash;
+
+    fn plain(password: &str) -> PasswordVerifier {
+        PasswordVerifier::PlainText(password.to_string())
+    }
+
+    fn sha1_one(data: &[u8]) -> Vec<u8> {
+        let mut hasher = Sha1::new();
+        hasher.update(data);
+        hasher.finalize().to_vec()
+    }
+
+    fn mysql_native_password_auth_data(password: &str, salt: &[u8]) -> Vec<u8> {
+        let hash_stage_1 = sha1_one(password.as_bytes());
+        let hash_stage_2 = mysql_native_password_hash(password.as_bytes());
+        let mut hasher = Sha1::new();
+        hasher.update(salt);
+        hasher.update(hash_stage_2);
+        let scramble = hasher.finalize();
+
+        hash_stage_1
+            .iter()
+            .zip(scramble.iter())
+            .map(|(lhs, rhs)| lhs ^ rhs)
+            .collect()
+    }
 
     #[test]
     fn test_parse_credential_line() {
@@ -198,7 +355,7 @@ mod tests {
             result,
             Some((
                 "admin".to_string(),
-                ("password123".as_bytes().to_vec(), PermissionMode::default())
+                (plain("password123"), PermissionMode::default())
             ))
         );
 
@@ -208,7 +365,7 @@ mod tests {
             result,
             Some((
                 "user".to_string(),
-                ("secret".as_bytes().to_vec(), PermissionMode::ReadOnly)
+                (plain("secret"), PermissionMode::ReadOnly)
             ))
         );
         let result = parse_credential_line("user:ro=secret");
@@ -216,7 +373,7 @@ mod tests {
             result,
             Some((
                 "user".to_string(),
-                ("secret".as_bytes().to_vec(), PermissionMode::ReadOnly)
+                (plain("secret"), PermissionMode::ReadOnly)
             ))
         );
         // Username with WriteOnly permission mode
@@ -225,7 +382,7 @@ mod tests {
             result,
             Some((
                 "writer".to_string(),
-                ("mypass".as_bytes().to_vec(), PermissionMode::WriteOnly)
+                (plain("mypass"), PermissionMode::WriteOnly)
             ))
         );
 
@@ -235,7 +392,7 @@ mod tests {
             result,
             Some((
                 "writer".to_string(),
-                ("mypass".as_bytes().to_vec(), PermissionMode::WriteOnly)
+                (plain("mypass"), PermissionMode::WriteOnly)
             ))
         );
 
@@ -245,10 +402,7 @@ mod tests {
             result,
             Some((
                 "admin".to_string(),
-                (
-                    "p@ssw0rd!123".as_bytes().to_vec(),
-                    PermissionMode::ReadWrite
-                )
+                (plain("p@ssw0rd!123"), PermissionMode::ReadWrite)
             ))
         );
 
@@ -258,10 +412,81 @@ mod tests {
             result,
             Some((
                 "user name".to_string(),
-                ("password".as_bytes().to_vec(), PermissionMode::WriteOnly)
+                (plain("password"), PermissionMode::WriteOnly)
             ))
         );
 
+        let result = parse_credential_line("user=plain:password");
+        assert_eq!(
+            result,
+            Some((
+                "user".to_string(),
+                (plain("password"), PermissionMode::default())
+            ))
+        );
+
+        let iterations = 4096;
+        let salt = b"salt";
+        let mut hash = [0u8; 32];
+        pbkdf2_hmac::<Sha256>("password".as_bytes(), salt, iterations, &mut hash);
+        let result = parse_credential_line(&format!(
+            "user=pbkdf2_sha256:{iterations}:{}:{}",
+            hex::encode(salt),
+            hex::encode(hash)
+        ));
+        assert_eq!(
+            result,
+            Some((
+                "user".to_string(),
+                (
+                    PasswordVerifier::Pbkdf2Sha256 {
+                        iterations,
+                        salt: salt.to_vec(),
+                        hash: hash.to_vec(),
+                    },
+                    PermissionMode::default()
+                )
+            ))
+        );
+
+        let result = parse_credential_line("user=pbkdf2_sha256:4096:not-hex:abcd");
+        assert_eq!(result, None);
+
+        // A well-formed but truncated hash must be rejected: a short hash would let
+        // many wrong passwords pass by matching only a few derived bytes.
+        let result = parse_credential_line(&format!(
+            "user=pbkdf2_sha256:4096:{}:abcd",
+            hex::encode(salt)
+        ));
+        assert_eq!(result, None);
+
+        let result = parse_credential_line(&format!(
+            "user=pbkdf2_sha256:{}:{}:{}",
+            MAX_PBKDF2_SHA256_ITERATIONS + 1,
+            hex::encode(salt),
+            hex::encode(hash)
+        ));
+        assert_eq!(result, None);
+
+        let hash_stage_2 = mysql_native_password_hash("password".as_bytes());
+        let result = parse_credential_line(&format!(
+            "user=mysql_native_password:{}",
+            hex::encode(&hash_stage_2)
+        ));
+        assert_eq!(
+            result,
+            Some((
+                "user".to_string(),
+                (
+                    PasswordVerifier::MysqlNativePassword { hash_stage_2 },
+                    PermissionMode::default()
+                )
+            ))
+        );
+
+        let result = parse_credential_line("user=mysql_native_password:abcd");
+        assert_eq!(result, None);
+
         // Invalid format - no equals sign
         let result = parse_credential_line("invalid_line");
         assert_eq!(result, None);
@@ -274,10 +499,7 @@ mod tests {
         let result = parse_credential_line("user=");
         assert_eq!(
             result,
-            Some((
-                "user".to_string(),
-                ("".as_bytes().to_vec(), PermissionMode::default())
-            ))
+            Some(("user".to_string(), (plain(""), PermissionMode::default())))
         );
 
         // Empty username
@@ -286,8 +508,112 @@ mod tests {
             result,
             Some((
                 "".to_string(),
-                ("password".as_bytes().to_vec(), PermissionMode::default())
+                (plain("password"), PermissionMode::default())
             ))
         );
     }
+
+    #[test]
+    fn test_authenticate_with_mysql_native_password_verifier() {
+        let password = "password";
+        let salt = b"12345678901234567890";
+        let hash_stage_2 = mysql_native_password_hash(password.as_bytes());
+        let auth_data = mysql_native_password_auth_data(password, salt);
+        let users = HashMap::from([(
+            "user".to_string(),
+            (
+                PasswordVerifier::MysqlNativePassword { hash_stage_2 },
+                PermissionMode::default(),
+            ),
+        )]);
+
+        let result = authenticate_with_credential(
+            &users,
+            Identity::UserId("user", None),
+            Password::MysqlNativePassword(&auth_data, salt),
+        );
+
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_authenticate_with_plain_text_mysql_native_password() {
+        let password = "password";
+        let salt = b"12345678901234567890";
+        let auth_data = mysql_native_password_auth_data(password, salt);
+        let users = HashMap::from([(
+            "user".to_string(),
+            (
+                PasswordVerifier::PlainText(password.to_string()),
+                PermissionMode::default(),
+            ),
+        )]);
+
+        let result = authenticate_with_credential(
+            &users,
+            Identity::UserId("user", None),
+            Password::MysqlNativePassword(&auth_data, salt),
+        );
+
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_pbkdf2_sha256_rejects_mysql_native_password() {
+        let password = "password";
+        let salt = b"salt";
+        let iterations = 4096;
+        let mut hash = [0u8; 32];
+        pbkdf2_hmac::<Sha256>(password.as_bytes(), salt, iterations, &mut hash);
+        let users = HashMap::from([(
+            "user".to_string(),
+            (
+                PasswordVerifier::Pbkdf2Sha256 {
+                    iterations,
+                    salt: salt.to_vec(),
+                    hash: hash.to_vec(),
+                },
+                PermissionMode::default(),
+            ),
+        )]);
+        let mysql_salt = b"12345678901234567890";
+        let auth_data = mysql_native_password_auth_data(password, mysql_salt);
+
+        let result = authenticate_with_credential(
+            &users,
+            Identity::UserId("user", None),
+            Password::MysqlNativePassword(&auth_data, mysql_salt),
+        );
+
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_password_verifier_debug_redacts_secrets() {
+        let debug = format!("{:?}", PasswordVerifier::PlainText("secret".to_string()));
+        assert!(debug.contains("<REDACTED>"));
+        assert!(!debug.contains("secret"));
+
+        let debug = format!(
+            "{:?}",
+            PasswordVerifier::Pbkdf2Sha256 {
+                iterations: 4096,
+                salt: b"super-secret-salt".to_vec(),
+                hash: b"super-secret-hash".to_vec(),
+            }
+        );
+        assert!(debug.contains("Pbkdf2Sha256"));
+        assert!(debug.contains("4096"));
+        assert!(!debug.contains("super-secret-salt"));
+        assert!(!debug.contains("super-secret-hash"));
+
+        let debug = format!(
+            "{:?}",
+            PasswordVerifier::MysqlNativePassword {
+                hash_stage_2: b"super-secret-hash".to_vec(),
+            }
+        );
+        assert!(debug.contains("MysqlNativePassword"));
+        assert!(!debug.contains("super-secret-hash"));
+    }
 }

src/auth/src/user_provider/static_user_provider.rs

@@ -13,9 +13,9 @@
 // limitations under the License.
 
 use async_trait::async_trait;
-use snafu::{OptionExt, ResultExt};
+use snafu::OptionExt;
 
-use crate::error::{FromUtf8Snafu, InvalidConfigSnafu, Result};
+use crate::error::{InvalidConfigSnafu, Result};
 use crate::user_provider::{
     UserInfoMap, authenticate_with_credential, load_credential_from_file, parse_credential_line,
 };
@@ -55,18 +55,6 @@ impl StaticUserProvider {
                 .fail(),
         }
     }
-
-    /// Return a random username/password pair
-    /// This is useful for invoking from other components in the cluster
-    pub fn get_one_user_pwd(&self) -> Result<(String, String)> {
-        let kv = self.users.iter().next().context(InvalidConfigSnafu {
-            value: "",
-            msg: "Expect at least one pair of username and password",
-        })?;
-        let username = kv.0;
-        let pwd = String::from_utf8(kv.1.0.clone()).context(FromUtf8Snafu)?;
-        Ok((username.clone(), pwd))
-    }
 }
 
 #[async_trait]
@@ -96,6 +84,8 @@ pub mod test {
     use std::io::{LineWriter, Write};
 
     use common_test_util::temp_dir::create_temp_dir;
+    use pbkdf2::pbkdf2_hmac;
+    use sha2::Sha256;
 
     use crate::UserProvider;
     use crate::user_info::DefaultUserInfo;
@@ -112,6 +102,28 @@ pub mod test {
         let _ = re.unwrap();
     }
 
+    async fn test_authenticate_fails(provider: &dyn UserProvider, username: &str, password: &str) {
+        let re = provider
+            .authenticate(
+                Identity::UserId(username, None),
+                Password::PlainText(password.to_string().into()),
+            )
+            .await;
+        assert!(re.is_err());
+    }
+
+    fn pbkdf2_sha256_verifier(password: &str) -> String {
+        let iterations = 4096;
+        let salt = b"salt";
+        let mut hash = [0u8; 32];
+        pbkdf2_hmac::<Sha256>(password.as_bytes(), salt, iterations, &mut hash);
+        format!(
+            "pbkdf2_sha256:{iterations}:{}:{}",
+            hex::encode(salt),
+            hex::encode(hash)
+        )
+    }
+
     #[tokio::test]
     async fn test_authorize() {
         let user_info = DefaultUserInfo::with_name("root");
@@ -129,6 +141,16 @@ pub mod test {
         test_authenticate(&provider, "admin", "654321").await;
     }
 
+    #[tokio::test]
+    async fn test_inline_provider_with_pbkdf2_sha256_verifier() {
+        let provider =
+            StaticUserProvider::new(&format!("cmd:root={}", pbkdf2_sha256_verifier("123456")))
+                .unwrap();
+
+        test_authenticate(&provider, "root", "123456").await;
+        test_authenticate_fails(&provider, "root", "654321").await;
+    }
+
     #[tokio::test]
     async fn test_file_provider() {
         let dir = create_temp_dir("test_file_provider");