feat: allow passthrough user to change password (#842)

- fix #830 
- fix #373 
- fix: we were sending unnecessary `Terminate` message after `FATAL`
error

Author: levkk

.schema/pgdog.schema.json

@@ -1249,6 +1249,16 @@
           "description": "Enabled without TLS requirement; network traffic may expose plaintext passwords.",
           "type": "string",
           "const": "enabled_plain"
+        },
+        {
+          "description": "Enabled and allows password changes.",
+          "type": "string",
+          "const": "enabled_allow_change"
+        },
+        {
+          "description": "Enabled without TLS requirement and allows password changes.",
+          "type": "string",
+          "const": "enabled_plain_allow_change"
         }
       ]
     },

integration/ruby/auth_spec.rb

@@ -0,0 +1,66 @@
+# frozen_string_literal: true
+
+require_relative 'rspec_helper'
+
+describe 'authentication' do
+  before(:all) do
+    # Reset state so passthrough auth from previous runs is disabled.
+    adm = PG.connect('postgres://admin:pgdog@127.0.0.1:6432/admin')
+    adm.exec 'RELOAD'
+    adm.close
+  end
+
+  it 'rejects connections with bad password' do
+    expect do
+      PG.connect(dbname: 'pgdog', user: 'pgdog', password: 'wrong_password', port: 6432, host: '127.0.0.1')
+    end.to raise_error(PG::ConnectionBad, /password for user "pgdog" and database "pgdog" is wrong/)
+  end
+
+  it 'rejects connections with bad user' do
+    expect do
+      PG.connect(dbname: 'pgdog', user: 'nonexistent_user', password: 'pgdog', port: 6432, host: '127.0.0.1')
+    end.to raise_error(PG::ConnectionBad, /password for user "nonexistent_user" and database "pgdog" is wrong/)
+  end
+
+  shared_examples 'passthrough authentication' do
+    it 'authenticates pgdog_pass user via passthrough' do
+      adm = admin
+
+      # Reload to reset state, then enable passthrough auth.
+      adm.exec 'RELOAD'
+      adm.exec "SET passthrough_auth TO 'enabled_plain'"
+
+      # Verify pool starts as offline.
+      pools = adm.exec 'SHOW POOLS'
+      pass_pool = pools.select { |p| p['user'] == 'pgdog_pass' && p['database'] == 'pgdog' }.first
+      expect(pass_pool).not_to be_nil
+      expect(pass_pool['online']).to eq('f')
+
+      # Connect with passthrough user and run queries.
+      conn = PG.connect(dbname: 'pgdog', user: 'pgdog_pass', password: 'pgdog', port: 6432, host: '127.0.0.1')
+      res = conn.exec 'SELECT 1 AS one'
+      expect(res[0]['one']).to eq('1')
+      res = conn.exec 'SELECT 2 AS two'
+      expect(res[0]['two']).to eq('2')
+
+      # Verify statement_timeout is carried over from user config.
+      res = conn.exec 'SHOW statement_timeout'
+      expect(res[0]['statement_timeout']).to eq('45999ms')
+      conn.close
+
+      # Verify pool is now online.
+      pools = adm.exec 'SHOW POOLS'
+      pass_pool = pools.select { |p| p['user'] == 'pgdog_pass' && p['database'] == 'pgdog' }.first
+      expect(pass_pool).not_to be_nil
+      expect(pass_pool['online']).to eq('t')
+
+      adm.close
+    end
+  end
+
+  10.times do |i|
+    context "passthrough attempt #{i + 1}" do
+      include_examples 'passthrough authentication'
+    end
+  end
+end

integration/rust/tests/integration/auth.rs

@@ -5,25 +5,48 @@ use sqlx::{Connection, Executor, PgConnection};
 
 #[tokio::test]
 #[serial]
-async fn test_auth() {
+async fn test_auth_types() {
     let admin = admin_sqlx().await;
-    let bad_password = "postgres://pgdog:skjfhjk23h4234@127.0.0.1:6432/pgdog";
-
-    admin.execute("SET auth_type TO 'trust'").await.unwrap();
-    assert_setting_str("auth_type", "trust").await;
-
-    let mut any_password = PgConnection::connect(bad_password).await.unwrap();
-    any_password.execute("SELECT 1").await.unwrap();
-
-    let mut empty_password = PgConnection::connect("postgres://pgdog@127.0.0.1:6432/pgdog")
-        .await
-        .unwrap();
-    empty_password.execute("SELECT 1").await.unwrap();
-
-    admin.execute("SET auth_type TO 'scram'").await.unwrap();
-    assert_setting_str("auth_type", "scram").await;
+    let good = "postgres://pgdog:pgdog@127.0.0.1:6432/pgdog";
+    let bad = "postgres://pgdog:wrong@127.0.0.1:6432/pgdog";
+    let none = "postgres://pgdog@127.0.0.1:6432/pgdog";
+
+    for auth_type in ["md5", "scram", "plain", "trust"] {
+        admin
+            .execute(format!("SET auth_type TO '{auth_type}'").as_str())
+            .await
+            .unwrap();
+        assert_setting_str("auth_type", auth_type).await;
+
+        let mut conn = PgConnection::connect(good).await.unwrap();
+        conn.execute("SELECT 1").await.unwrap();
+
+        if auth_type == "trust" {
+            let mut conn = PgConnection::connect(bad).await.unwrap();
+            conn.execute("SELECT 1").await.unwrap();
+
+            let mut conn = PgConnection::connect(none).await.unwrap();
+            conn.execute("SELECT 1").await.unwrap();
+        } else {
+            let bad_err = PgConnection::connect(bad).await.err().unwrap();
+            assert!(
+                bad_err
+                    .to_string()
+                    .contains("password for user \"pgdog\" and database \"pgdog\" is wrong"),
+                "{auth_type}: bad password error: {bad_err}"
+            );
+            let none_err = PgConnection::connect(none).await.err().unwrap();
+            assert!(
+                none_err
+                    .to_string()
+                    .contains("password for user \"pgdog\" and database \"pgdog\" is wrong"),
+                "{auth_type}: no password error: {none_err}"
+            );
+        }
+    }
 
-    assert!(PgConnection::connect(bad_password).await.is_err());
+    // Reset config.
+    admin.execute("RELOAD").await.unwrap();
 }
 
 #[tokio::test]
@@ -91,3 +114,55 @@ async fn test_passthrough_auth() {
     user.execute("SELECT 1").await.unwrap();
     original.execute("SELECT 1").await.unwrap();
 }
+
+#[tokio::test]
+async fn test_passthrough_password_change() {
+    let admin = admin_sqlx().await;
+    let mut direct =
+        PgConnection::connect("postgres://pgdog:pgdog@127.0.0.1:5432/pgdog?sslmode=disable")
+            .await
+            .unwrap();
+
+    // Ensure clean state.
+    admin.execute("RELOAD").await.unwrap();
+    admin
+        .execute("SET passthrough_auth TO 'enabled_plain_allow_change'")
+        .await
+        .unwrap();
+    assert_setting_str("passthrough_auth", "enabled_plain_allow_change").await;
+
+    // Make sure pgdog1 has the original password.
+    direct
+        .execute("ALTER USER pgdog1 PASSWORD 'pgdog'")
+        .await
+        .unwrap();
+
+    // Connect with original password and keep connection alive.
+    let mut existing = PgConnection::connect("postgres://pgdog1:pgdog@127.0.0.1:6432/pgdog")
+        .await
+        .unwrap();
+    existing.execute("SELECT 1").await.unwrap();
+
+    // Change password in PostgreSQL directly.
+    direct
+        .execute("ALTER USER pgdog1 PASSWORD 'new_password'")
+        .await
+        .unwrap();
+
+    // New connection with new password should work.
+    let mut new_conn = PgConnection::connect("postgres://pgdog1:new_password@127.0.0.1:6432/pgdog")
+        .await
+        .unwrap();
+    new_conn.execute("SELECT 1").await.unwrap();
+
+    // Existing connection should still work.
+    existing.execute("SELECT 1").await.unwrap();
+
+    // Cleanup: restore original password.
+    direct
+        .execute("ALTER USER pgdog1 PASSWORD 'pgdog'")
+        .await
+        .unwrap();
+
+    admin.execute("RELOAD").await.unwrap();
+}

integration/rust/tests/integration/reload.rs

@@ -1,8 +1,8 @@
 use std::time::Duration;
 
-use rust::setup::{admin_tokio, backends, connection_failover};
+use rust::setup::{admin_tokio, backends, connection_failover, connection_sqlx_direct};
 use serial_test::serial;
-use sqlx::Executor;
+use sqlx::{Executor, Row, postgres::PgPoolOptions};
 use tokio::time::sleep;
 
 #[tokio::test]
@@ -34,6 +34,103 @@ async fn test_reload() {
     conn.close().await;
 }
 
+#[tokio::test]
+#[serial]
+async fn test_reload_connection_count_stable() {
+    sleep(Duration::from_secs(1)).await;
+    let admin = admin_tokio().await;
+    let direct = connection_sqlx_direct().await;
+
+    // Count PgDog connections on Postgres before reload.
+    let before: i64 = direct
+        .fetch_one("SELECT COUNT(*)::BIGINT FROM pg_stat_activity WHERE application_name = 'PgDog'")
+        .await
+        .unwrap()
+        .get(0);
+
+    assert!(before > 0, "expected pgdog connections before reload");
+    eprintln!("connections before RELOAD: {before}");
+
+    admin.simple_query("RELOAD").await.unwrap();
+    sleep(Duration::from_millis(500)).await;
+
+    let after: i64 = direct
+        .fetch_one("SELECT COUNT(*)::BIGINT FROM pg_stat_activity WHERE application_name = 'PgDog'")
+        .await
+        .unwrap()
+        .get(0);
+
+    eprintln!("connections after RELOAD: {after}");
+
+    assert_eq!(
+        before, after,
+        "connection count changed after RELOAD: before={before}, after={after}"
+    );
+
+    direct.close().await;
+}
+
+#[tokio::test]
+#[serial]
+async fn test_reload_pool_size_not_exceeded() {
+    let pool_size: i64 = 50; // default_pool_size
+    let num_clients = 20;
+    let app_name = "test_reload_pool_limit";
+
+    sleep(Duration::from_secs(1)).await;
+    let admin = admin_tokio().await;
+    let direct = connection_sqlx_direct().await;
+
+    // Spin up clients that continuously run queries.
+    let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let mut handles = vec![];
+
+    for _ in 0..num_clients {
+        let stop = stop.clone();
+        handles.push(tokio::spawn(async move {
+            let pool = PgPoolOptions::new()
+                .max_connections(1)
+                .connect(&format!(
+                    "postgres://pgdog:pgdog@127.0.0.1:6432/pgdog?application_name={app_name}"
+                ))
+                .await
+                .unwrap();
+            while !stop.load(std::sync::atomic::Ordering::Relaxed) {
+                let _ = pool.execute("SELECT 1").await;
+                sleep(Duration::from_millis(10)).await;
+            }
+            pool.close().await;
+        }));
+    }
+
+    // Let clients warm up.
+    sleep(Duration::from_millis(500)).await;
+
+    // Issue multiple reloads while clients are active and check pool size each time.
+    for i in 0..10 {
+        admin.simple_query("RELOAD").await.unwrap();
+        sleep(Duration::from_millis(100)).await;
+
+        let query = format!(
+            "SELECT COUNT(*)::BIGINT FROM pg_stat_activity WHERE application_name = '{app_name}'"
+        );
+        let count: i64 = direct.fetch_one(query.as_str()).await.unwrap().get(0);
+
+        eprintln!("reload {i}: {count} connections (pool_size={pool_size})");
+        assert!(
+            count <= pool_size,
+            "pool size exceeded after reload {i}: {count} > {pool_size}"
+        );
+    }
+
+    stop.store(true, std::sync::atomic::Ordering::Relaxed);
+    for h in handles {
+        h.await.unwrap();
+    }
+
+    direct.close().await;
+}
+
 #[tokio::test]
 #[serial]
 async fn test_reconnect() {

integration/toxi/toxi_spec.rb

@@ -57,7 +57,7 @@ def warm_up
         end
         25.times do
           c.exec 'SELECT 1'
-        rescue PG::SystemError
+        rescue PG::SystemError, PG::ConnectionBad
           c = conn # reconnect
           errors.increment
         end
@@ -84,11 +84,12 @@ def warm_up
       25.times do
         Sharded.where(id: 1).first
         ok += 1
-      rescue StandardError
+      rescue StandardError => e
+        puts "Error: #{e.class}: #{e.message}"
         errors += 1
       end
     end
-    expect(errors).to be <= 1
+    expect(errors).to be <= 2
     expect(25 - ok).to eq(errors)
   end
 end

integration/users.toml

@@ -83,3 +83,11 @@ database = "pgdog_schema_no_cross"
 server_user = "pgdog"
 cross_shard_disabled = true
 pooler_mode = "session"
+
+[[users]]
+name = "pgdog_pass"
+server_user = "pgdog"
+pool_size = 5
+database = "pgdog"
+min_pool_size = 1
+statement_timeout = 45_999

pgdog-config/src/auth.rs

@@ -17,6 +17,19 @@ pub enum PassthroughAuth {
     Enabled,
     /// Enabled without TLS requirement; network traffic may expose plaintext passwords.
     EnabledPlain,
+    /// Enabled and allows password changes.
+    EnabledAllowChange,
+    /// Enabled without TLS requirement and allows password changes.
+    EnabledPlainAllowChange,
+}
+
+impl PassthroughAuth {
+    pub fn allows_change(&self) -> bool {
+        matches!(
+            self,
+            Self::EnabledPlainAllowChange | Self::EnabledAllowChange
+        )
+    }
 }
 
 /// authentication mechanism for client connections.

pgdog-config/src/general.rs

@@ -1129,8 +1129,15 @@ impl General {
     }
 
     pub fn passthrough_auth(&self) -> bool {
-        self.tls().is_some() && self.passthrough_auth == PassthroughAuth::Enabled
-            || self.passthrough_auth == PassthroughAuth::EnabledPlain
+        self.tls().is_some()
+            && matches!(
+                self.passthrough_auth,
+                PassthroughAuth::Enabled | PassthroughAuth::EnabledAllowChange
+            )
+            || matches!(
+                self.passthrough_auth,
+                PassthroughAuth::EnabledPlain | PassthroughAuth::EnabledPlainAllowChange
+            )
     }
 
     /// Support for LISTEN/NOTIFY.