Fix passthrough auth for users without a password

integration/rust/tests/integration/auth.rs

@@ -91,3 +91,34 @@ async fn test_passthrough_auth() {
     user.execute("SELECT 1").await.unwrap();
     original.execute("SELECT 1").await.unwrap();
 }
+
+#[tokio::test]
+#[serial]
+async fn test_user_without_password_passthrough_auth() {
+    let admin = admin_sqlx().await;
+
+    admin.execute("RELOAD").await.unwrap();
+    admin.execute("SET auth_type TO 'scram'").await.unwrap();
+    assert_setting_str("auth_type", "scram").await;
+
+    let user = "postgres://pgdog2:pgdog@127.0.0.1:6432/pgdog";
+
+    let no_password_err = PgConnection::connect(user).await.err().unwrap();
+
+    assert!(
+        no_password_err
+            .to_string()
+            .contains("password for user \"pgdog2\" and database \"pgdog\" is wrong")
+    );
+
+    admin
+        .execute("SET passthrough_auth TO 'enabled_plain'")
+        .await
+        .unwrap();
+    assert_setting_str("passthrough_auth", "enabled_plain").await;
+
+    let mut user = PgConnection::connect(user).await.unwrap();
+
+    user.execute("SELECT 1").await.unwrap();
+    user.close().await.unwrap();
+}

integration/users.toml

@@ -3,6 +3,10 @@ name = "pgdog"
 database = "pgdog"
 password = "pgdog"
 
+[[users]]
+name = "pgdog2"
+database = "pgdog"
+
 [[users]]
 name = "pgdog_session"
 database = "pgdog"

pgdog/src/backend/databases.rs

@@ -18,7 +18,7 @@ use crate::frontend::router::sharding::Mapping;
 use crate::frontend::PreparedStatements;
 use crate::{
     backend::pool::PoolConfig,
-    config::{config, load, ConfigAndUsers, ManualQuery, Role},
+    config::{config, load, AuthType, ConfigAndUsers, ManualQuery, Role},
     net::{messages::BackendKeyData, tls},
 };
 
@@ -234,11 +234,24 @@ impl Databases {
     }
 
     /// Check if a cluster exists, quickly.
-    pub fn exists(&self, user: impl ToUser) -> bool {
-        if let Some(cluster) = self.databases.get(&user.to_user()) {
-            !cluster.password().is_empty()
+    pub(crate) fn exists(&self, user: impl ToUser) -> bool {
+        self.databases.contains_key(&user.to_user())
+    }
+
+    /// Check if a cluster exists, and has a non-empty password.
+    pub(crate) fn has_password(&self, user: impl ToUser) -> bool {
+        self.databases
+            .get(&user.to_user())
+            .is_some_and(|cluster| !cluster.password().is_empty())
+    }
+
+    /// Check if backend authentication can work for this user.
+    pub fn is_backend_auth_ready(&self, user: impl ToUser, authtype: &AuthType) -> bool {
+        // Trust auth doesn't need a password, so the cluster merely has to exist.
+        if authtype.trust() {
+            self.exists(user)
         } else {
-            false
+            self.has_password(user)
         }
     }
 
@@ -460,12 +473,21 @@ fn new_pool(user: &crate::config::User, config: &crate::config::Config) -> Optio
         &config.rewrite,
     );
 
+    let cluster = Cluster::new(cluster_config);
+
+    if config.general.passthrough_auth()
+        && user.password().is_empty()
+        && !config.general.auth_type.trust()
+    {
+        cluster.pause();
+    }
+
     Some((
         User {
             user: user.name.clone(),
             database: user.database.clone(),
         },
-        Cluster::new(cluster_config),
+        cluster,
     ))
 }
 
@@ -617,7 +639,7 @@ pub fn from_config(config: &ConfigAndUsers) -> Databases {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::config::{Config, ConfigAndUsers, Database, Role};
+    use crate::config::{AuthType, Config, ConfigAndUsers, Database, PassthoughAuth, Role};
 
     #[test]
     fn test_mirror_user_isolation() {
@@ -1570,4 +1592,191 @@ mod tests {
 
         assert_eq!(databases.all().len(), 1);
     }
+
+    #[test]
+    fn test_passthrough_empty_password_starts_paused() {
+        let mut config = Config::default();
+        config.general.passthrough_auth = PassthoughAuth::EnabledPlain;
+        config.databases = vec![Database {
+            name: "pgdog".to_string(),
+            host: "localhost".to_string(),
+            port: 5432,
+            role: Role::Primary,
+            ..Default::default()
+        }];
+
+        let users = crate::config::Users {
+            users: vec![crate::config::User {
+                name: "pgdog".to_string(),
+                database: "pgdog".to_string(),
+                password: None,
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        let databases = from_config(&ConfigAndUsers {
+            config,
+            users,
+            config_path: std::path::PathBuf::new(),
+            users_path: std::path::PathBuf::new(),
+        });
+
+        let key = User {
+            user: "pgdog".to_string(),
+            database: "pgdog".to_string(),
+        };
+
+        let cluster = databases.all().get(&key).expect("cluster should exist");
+
+        for shard in cluster.shards() {
+            for pool in shard.pools() {
+                assert!(pool.state().paused);
+            }
+        }
+    }
+
+    #[test]
+    fn test_passthrough_user_with_password_unpaused() {
+        let mut config = Config::default();
+        config.general.passthrough_auth = PassthoughAuth::EnabledPlain;
+        config.databases = vec![Database {
+            name: "pgdog".to_string(),
+            host: "localhost".to_string(),
+            port: 5432,
+            role: Role::Primary,
+            ..Default::default()
+        }];
+
+        let users = crate::config::Users {
+            users: vec![crate::config::User {
+                name: "pgdog".to_string(),
+                database: "pgdog".to_string(),
+                password: Some("pgdog".to_string()),
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        let databases = from_config(&ConfigAndUsers {
+            config,
+            users,
+            config_path: std::path::PathBuf::new(),
+            users_path: std::path::PathBuf::new(),
+        });
+
+        let key = User {
+            user: "pgdog".to_string(),
+            database: "pgdog".to_string(),
+        };
+
+        let cluster = databases.all().get(&key).expect("cluster should exist");
+
+        for shard in cluster.shards() {
+            for pool in shard.pools() {
+                assert!(!pool.state().paused);
+            }
+        }
+    }
+
+    #[test]
+    fn test_passthrough_empty_password_trust_starts_unpaused() {
+        let mut config = Config::default();
+        config.general.passthrough_auth = PassthoughAuth::EnabledPlain;
+        config.general.auth_type = AuthType::Trust;
+        config.databases = vec![Database {
+            name: "pgdog".to_string(),
+            host: "localhost".to_string(),
+            port: 5432,
+            role: Role::Primary,
+            ..Default::default()
+        }];
+
+        let users = crate::config::Users {
+            users: vec![crate::config::User {
+                name: "pgdog".to_string(),
+                database: "pgdog".to_string(),
+                password: None,
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        let databases = from_config(&ConfigAndUsers {
+            config,
+            users,
+            config_path: std::path::PathBuf::new(),
+            users_path: std::path::PathBuf::new(),
+        });
+
+        let key = User {
+            user: "pgdog".to_string(),
+            database: "pgdog".to_string(),
+        };
+        let cluster = databases.all().get(&key).expect("cluster should exist");
+
+        for shard in cluster.shards() {
+            for pool in shard.pools() {
+                assert!(!pool.state().paused);
+            }
+        }
+    }
+
+    #[test]
+    fn test_replace_empty_password_cluster_with_passthrough_password() {
+        let mut config = Config::default();
+        config.general.passthrough_auth = PassthoughAuth::EnabledPlain;
+        config.databases = vec![Database {
+            name: "pgdog".to_string(),
+            host: "localhost".to_string(),
+            port: 5432,
+            role: Role::Primary,
+            ..Default::default()
+        }];
+
+        let users = crate::config::Users {
+            users: vec![crate::config::User {
+                name: "pgdog".to_string(),
+                database: "pgdog".to_string(),
+                password: None,
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        let databases = from_config(&ConfigAndUsers {
+            config: config.clone(),
+            users,
+            config_path: std::path::PathBuf::new(),
+            users_path: std::path::PathBuf::new(),
+        });
+
+        let passthrough_user = crate::config::User {
+            name: "pgdog".to_string(),
+            database: "pgdog".to_string(),
+            password: Some("secret".to_string()),
+            ..Default::default()
+        };
+
+        let (user, cluster) = new_pool(&passthrough_user, &config).expect("cluster should exist");
+        let (added, databases) = databases.add(user, cluster);
+
+        assert!(added);
+        assert!(databases.has_password(("pgdog", "pgdog")));
+
+        let key = User {
+            user: "pgdog".to_string(),
+            database: "pgdog".to_string(),
+        };
+
+        let cluster = databases.all().get(&key).expect("cluster should exist");
+
+        assert_eq!(cluster.password(), "secret");
+
+        for shard in cluster.shards() {
+            for pool in shard.pools() {
+                assert!(!pool.state().paused);
+            }
+        }
+    }
 }

pgdog/src/backend/pool/cluster.rs

@@ -570,6 +570,16 @@ impl Cluster {
         self.readiness.online.store(true, Ordering::Relaxed);
     }
 
+    /// Pause all pools in this cluster.
+    pub fn pause(&self) {
+        self.shards().iter().for_each(|shard| shard.pause())
+    }
+
+    /// Resume all pools in this cluster.
+    pub fn resume(&self) {
+        self.shards().iter().for_each(|shard| shard.resume())
+    }
+
     /// Shutdown the connection pools.
     pub(crate) fn shutdown(&self) {
         for shard in self.shards() {
@@ -821,6 +831,26 @@ mod test {
         assert!(!cluster.online());
     }
 
+    #[test]
+    fn test_pause_resume_toggles_all_pools() {
+        let config = ConfigAndUsers::default();
+        let cluster = Cluster::new_test(&config);
+
+        cluster.pause();
+        for shard in cluster.shards() {
+            for pool in shard.pools() {
+                assert!(pool.state().paused);
+            }
+        }
+
+        cluster.resume();
+        for shard in cluster.shards() {
+            for pool in shard.pools() {
+                assert!(!pool.state().paused);
+            }
+        }
+    }
+
     #[tokio::test]
     async fn test_launch_schema_loading_idempotent() {
         use std::sync::atomic::Ordering;

pgdog/src/backend/pool/connection/mod.rs

@@ -325,8 +325,11 @@ impl Connection {
         match self.binding {
             Binding::Direct(_) | Binding::MultiShard(_, _) => {
                 let user = (self.user.as_str(), self.database.as_str());
+                let config = config();
                 // Check passthrough auth.
-                if config().config.general.passthrough_auth() && !databases().exists(user) {
+                if config.config.general.passthrough_auth()
+                    && !databases().is_backend_auth_ready(user, &config.config.general.auth_type)
+                {
                     if let Some(ref passthrough_password) = self.passthrough_password {
                         let new_user = User::new(&self.user, passthrough_password, &self.database);
                         databases::add(new_user);
@@ -360,6 +363,13 @@ impl Connection {
         Ok(())
     }
 
+    /// Resume pools for the currently bound cluster.
+    pub(crate) fn resume_cluster_pools(&self) {
+        if let Some(cluster) = &self.cluster {
+            cluster.resume();
+        }
+    }
+
     pub(crate) fn bind(&mut self, bind: &Bind) -> Result<(), Error> {
         match self.binding {
             Binding::MultiShard(_, ref mut state) => {

pgdog/src/backend/pool/lb/mod.rs

@@ -190,6 +190,16 @@ impl LoadBalancer {
         Monitor::spawn(self);
     }
 
+    /// Pause all target pools.
+    pub fn pause(&self) {
+        self.targets.iter().for_each(|target| target.pool.pause());
+    }
+
+    /// Resume all target pools.
+    pub fn resume(&self) {
+        self.targets.iter().for_each(|target| target.pool.resume());
+    }
+
     /// Get a live connection from the pool.
     pub async fn get(&self, request: &Request) -> Result<Guard, Error> {
         match timeout(self.checkout_timeout, self.get_internal(request)).await {