fix(auth): warn when credential load disables Postgres SCRAM or drops a line (#8652)

* fix(auth): warn when credential load disables Postgres SCRAM or drops a line

Static and watch user providers degraded silently in two ways:

- A single non-SCRAM verifier (mysql_native_password, or a legacy
  pbkdf2_sha256 hash that predates SCRAM) disables Postgres SCRAM for
  every user and falls back to cleartext, with no signal to the operator.
- A malformed credential line (commonly a plaintext password containing
  '=', which splits into more than two parts) was dropped without a trace.

Emit a warning at each credential load for both cases so operators don't
unknowingly serve cleartext passwords over Postgres or lose a user. This
is logging only; authentication behavior is unchanged. The SCRAM check
never logs secrets, and the malformed-line warning logs the line number
and file, never the line content.

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

* fix(auth): warn on credential file read error before truncating

A read error from lines() (I/O failure or invalid UTF-8) ends the
iterator via map_while, silently dropping every remaining credential.
Warn with the line number and file before truncating, matching the
malformed-line handling, so the drop is observable.

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

---------

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

src/auth/src/user_provider.rs

@@ -22,6 +22,7 @@ use std::path::Path;
 use std::{fmt, io};
 
 use common_base::secrets::ExposeSecret;
+use common_telemetry::warn;
 use pbkdf2::pbkdf2_hmac;
 use sha2::Sha256;
 use snafu::{OptionExt, ResultExt, ensure};
@@ -327,8 +328,24 @@ fn load_credential_from_file(filepath: &str) -> Result<UserInfoMap> {
     let file = File::open(path).context(IoSnafu)?;
     let credential = io::BufReader::new(file)
         .lines()
-        .map_while(std::result::Result::ok)
-        .filter_map(|line| {
+        .enumerate()
+        .map_while(|(idx, line)| match line {
+            Ok(line) => Some((idx, line)),
+            Err(err) => {
+                // A read error (I/O failure or invalid UTF-8) ends the iterator,
+                // so every remaining credential is dropped. Warn instead of
+                // vanishing silently, matching the malformed-line handling below.
+                warn!(
+                    "Failed to read line {} of user provider file {}: {}; \
+                     all remaining credentials are ignored",
+                    idx + 1,
+                    filepath,
+                    err
+                );
+                None
+            }
+        })
+        .filter_map(|(idx, line)| {
             // The line format is:
             // - `username=password` - Basic user with default permissions
             // - `username:permission_mode=password` - User with specific permission mode
@@ -339,7 +356,20 @@ fn load_credential_from_file(filepath: &str) -> Result<UserInfoMap> {
                 return None;
             }
 
-            parse_credential_line(line)
+            let parsed = parse_credential_line(line);
+            if parsed.is_none() {
+                // Don't log the line: it carries the password/verifier. A common
+                // cause is a plaintext password containing `=`, which splits the
+                // line into more than two parts.
+                warn!(
+                    "Ignoring malformed credential at line {} of user provider file {}: \
+                     expected `username[:permission]=verifier` with exactly one `=` \
+                     (passwords containing `=` are not supported)",
+                    idx + 1,
+                    filepath
+                );
+            }
+            parsed
         })
         .collect::<HashMap<String, _>>();
 
@@ -351,9 +381,46 @@ fn load_credential_from_file(filepath: &str) -> Result<UserInfoMap> {
         }
     );
 
+    warn_if_pg_scram_disabled(&credential);
+
     Ok(credential)
 }
 
+/// Returns the users whose verifier cannot back a Postgres SCRAM handshake.
+///
+/// Only [`PasswordVerifier::PlainText`] and [`PasswordVerifier::PgScramSha256`]
+/// support SCRAM. A `mysql_native_password` verifier is a double-SHA1 digest
+/// unrelated to PBKDF2, and a `pbkdf2_sha256` verifier was derived without
+/// SASLprep and cannot be safely reused as a SCRAM secret without the original
+/// password. Either kind forces the whole Postgres endpoint to fall back to
+/// cleartext (see [`postgres_auth_info_with_credential`]).
+fn pg_scram_unsupported_users(users: &UserInfoMap) -> Vec<&str> {
+    users
+        .iter()
+        .filter(|(_, (verifier, _))| !verifier.supports_pg_scram_sha256())
+        .map(|(username, _)| username.as_str())
+        .collect()
+}
+
+/// Warns once per credential load when the set disables Postgres SCRAM, so
+/// operators don't unknowingly serve cleartext passwords over Postgres while
+/// believing SCRAM is in effect.
+pub(crate) fn warn_if_pg_scram_disabled(users: &UserInfoMap) {
+    let unsupported = pg_scram_unsupported_users(users);
+    if !unsupported.is_empty() {
+        warn!(
+            "Postgres SCRAM authentication is disabled: {} of {} user(s) use a \
+             non-SCRAM password verifier {:?}, so all Postgres password \
+             authentication falls back to cleartext. Ensure TLS is enabled; if you \
+             rely on Postgres SCRAM, generate every user's verifier with the \
+             pg_scram_sha256 format.",
+            unsupported.len(),
+            users.len(),
+            unsupported
+        );
+    }
+}
+
 /// 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:
@@ -903,6 +970,51 @@ mod tests {
         assert!(matches!(auth_info, PgAuthInfo::Cleartext));
     }
 
+    #[test]
+    fn test_pg_scram_unsupported_users() {
+        let scram_verifier =
+            format_pg_scram_sha256_password_verifier(b"password", b"salt", 4096).unwrap();
+        let (_, scram_user) = parse_credential_line(&format!("scram={scram_verifier}")).unwrap();
+
+        let mut hash = [0u8; PBKDF2_SHA256_HASH_LEN];
+        pbkdf2_hmac::<Sha256>(b"password", b"salt", 4096, &mut hash);
+
+        let users = HashMap::from([
+            (
+                "plain".to_string(),
+                (plain("password"), PermissionMode::default()),
+            ),
+            ("scram".to_string(), scram_user),
+            (
+                "pbkdf2".to_string(),
+                (
+                    PasswordVerifier::Pbkdf2Sha256 {
+                        iterations: 4096,
+                        salt: b"salt".to_vec(),
+                        hash: hash.to_vec(),
+                    },
+                    PermissionMode::default(),
+                ),
+            ),
+            (
+                "mysql".to_string(),
+                (
+                    PasswordVerifier::MysqlNativePassword {
+                        hash_stage_2: mysql_native_password_hash(b"password"),
+                    },
+                    PermissionMode::default(),
+                ),
+            ),
+        ]);
+
+        let mut unsupported = pg_scram_unsupported_users(&users);
+        unsupported.sort();
+        // A pbkdf2_sha256 verifier is flagged alongside mysql_native_password:
+        // both force Postgres to fall back to cleartext, while plain and
+        // pg_scram back SCRAM.
+        assert_eq!(unsupported, vec!["mysql", "pbkdf2"]);
+    }
+
     #[test]
     fn test_password_verifier_debug_redacts_secrets() {
         let debug = format!(

src/auth/src/user_provider/static_user_provider.rs

@@ -18,7 +18,7 @@ use snafu::OptionExt;
 use crate::error::{InvalidConfigSnafu, Result};
 use crate::user_provider::{
     PgAuthInfo, UserInfoMap, authenticate_with_credential, load_credential_from_file,
-    parse_credential_line, postgres_auth_info_with_credential,
+    parse_credential_line, postgres_auth_info_with_credential, warn_if_pg_scram_disabled,
 };
 use crate::{Identity, Password, UserInfoRef, UserProvider};
 
@@ -48,7 +48,10 @@ impl StaticUserProvider {
                     })
                 })
                 .collect::<Result<UserInfoMap>>()
-                .map(|users| StaticUserProvider { users }),
+                .map(|users| {
+                    warn_if_pg_scram_disabled(&users);
+                    StaticUserProvider { users }
+                }),
             _ => InvalidConfigSnafu {
                 value: mode.to_string(),
                 msg: "StaticUserProviderOption must be in format `file:<path>` or `cmd:<values>`",