Helm chart improvements including allowing user password to be pulled from K8s secret (#753)

* Make user min_pool_size configurable

* Set user server_lifetime only if specified

* Increment chart version

* Use default instea of or

* Allow enabling server_tls

* statement_timeout default value

* Allow pulling password from existing secret

---------

Co-authored-by: Mostafa Abdelraouf <mostafa.mohmmed@gmail.com>

.github/workflows/chart-lint-test.yaml

@@ -22,7 +22,7 @@ jobs:
       # Python is required because `ct lint` runs Yamale (https://github.com/23andMe/Yamale) and
       # yamllint (https://github.com/adrienverge/yamllint) which require Python
       - name: Set up Python
-        uses: actions/setup-python@v4.1.0
+        uses: actions/setup-python@v5.1.0
         with:
           python-version: 3.7
 
@@ -43,7 +43,7 @@ jobs:
         run: ct lint --config ct.yaml
 
       - name: Create kind cluster
-        uses: helm/kind-action@v1.7.0
+        uses: helm/kind-action@v1.10.0
         if: steps.list-changed.outputs.changed == 'true'
 
       - name: Run chart-testing (install)

.github/workflows/chart-release.yaml

@@ -32,7 +32,7 @@ jobs:
           version: v3.13.0
 
       - name: Run chart-releaser
-        uses: helm/chart-releaser-action@be16258da8010256c6e82849661221415f031968 # v1.5.0
+        uses: helm/chart-releaser-action@a917fd15b20e8b64b94d9158ad54cd6345335584 # v1.6.0
         with:
           charts_dir: charts
           config: cr.yaml

charts/pgcat/Chart.yaml

@@ -5,4 +5,4 @@ maintainers:
   - name: Wildcard
     email: support@w6d.io
 appVersion: "1.2.0"
-version: 0.2.0
+version: 0.2.1

charts/pgcat/templates/secret.yaml

@@ -15,6 +15,7 @@ stringData:
     connect_timeout = {{ .Values.configuration.general.connect_timeout }}
     idle_timeout = {{ .Values.configuration.general.idle_timeout | int }}
     server_lifetime = {{ .Values.configuration.general.server_lifetime | int }}
+    server_tls = {{ .Values.configuration.general.server_tls }}
     idle_client_in_transaction_timeout = {{ .Values.configuration.general.idle_client_in_transaction_timeout | int }}
     healthcheck_timeout = {{ .Values.configuration.general.healthcheck_timeout }}
     healthcheck_delay = {{ .Values.configuration.general.healthcheck_delay }}
@@ -58,11 +59,21 @@ stringData:
     ##
     [pools.{{ $pool.name | quote }}.users.{{ $index }}]
     username = {{ $user.username | quote }}
+    {{- if $user.password }}
     password = {{ $user.password | quote }}
+    {{- else if and $user.passwordSecret.name $user.passwordSecret.key }}
+    {{- $secret := (lookup "v1" "Secret" $.Release.Namespace $user.passwordSecret.name) }}
+    {{- if $secret }}
+    {{- $password := index $secret.data $user.passwordSecret.key | b64dec }}
+    password = {{ $password | quote }}
+    {{- end }}
+    {{- end }}
     pool_size = {{ $user.pool_size }}
-    statement_timeout = {{ $user.statement_timeout }}
+    statement_timeout = {{ default 0 $user.statement_timeout }}
-    min_pool_size = 3
+    min_pool_size = {{ default 3 $user.min_pool_size }}
-    server_lifetime = 60000
+    {{- if $user.server_lifetime }}
+    server_lifetime = {{ $user.server_lifetime }}
+    {{- end }}
     {{-     if and $user.server_username $user.server_password }}
     server_username = {{ $user.server_username | quote }}
     server_password = {{ $user.server_password | quote }}

charts/pgcat/values.yaml

@@ -175,6 +175,9 @@ configuration:
     # Max connection lifetime before it's closed, even if actively used.
     server_lifetime: 86400000  # 24 hours
 
+    # Whether to use TLS for server connections or not.
+    server_tls: false
+
     # How long a client is allowed to be idle while in a transaction (ms).
     idle_client_in_transaction_timeout: 0  # milliseconds
 
@@ -315,7 +318,9 @@ configuration:
     #  ## Credentials for users that may connect to this cluster
     #  ## @param users [array]
     #  ## @param users[0].username Name of the env var (required)
-    #  ## @param users[0].password Value for the env var (required)
+    #  ## @param users[0].password Value for the env var (required) leave empty to use existing secret see passwordSecret.name and passwordSecret.key
+    #  ## @param users[0].passwordSecret.name Name of the secret containing the password
+    #  ## @param users[0].passwordSecret.key Key in the secret containing the password
     #  ## @param users[0].pool_size Maximum number of server connections that can be established for this user
     #  ## @param users[0].statement_timeout Maximum query duration. Dangerous, but protects against DBs that died in a non-obvious way.
     #  users: []

src/auth_passthrough.rs

@@ -1,3 +1,4 @@
+use crate::config::AuthType;
 use crate::errors::Error;
 use crate::pool::ConnectionPool;
 use crate::server::Server;
@@ -71,6 +72,7 @@ impl AuthPassthrough {
     pub async fn fetch_hash(&self, address: &crate::config::Address) -> Result<String, Error> {
         let auth_user = crate::config::User {
             username: self.user.clone(),
+            auth_type: AuthType::MD5,
             password: Some(self.password.clone()),
             server_username: None,
             server_password: None,

src/client.rs

@@ -14,7 +14,9 @@ use tokio::sync::mpsc::Sender;
 
 use crate::admin::{generate_server_parameters_for_admin, handle_admin};
 use crate::auth_passthrough::refetch_auth_hash;
-use crate::config::{get_config, get_idle_client_in_transaction_timeout, Address, PoolMode};
+use crate::config::{
+    get_config, get_idle_client_in_transaction_timeout, Address, AuthType, PoolMode,
+};
 use crate::constants::*;
 use crate::messages::*;
 use crate::plugins::PluginOutput;
@@ -463,8 +465,8 @@ where
             .count()
             == 1;
 
-        // Kick any client that's not admin while we're in admin-only mode.
         if !admin && admin_only {
+            // Kick any client that's not admin while we're in admin-only mode.
             debug!(
                 "Rejecting non-admin connection to {} when in admin only mode",
                 pool_name
@@ -481,6 +483,105 @@ where
         let process_id: i32 = rand::random();
         let secret_key: i32 = rand::random();
 
+        let mut prepared_statements_enabled = false;
+
+        // Authenticate admin user.
+        let (transaction_mode, mut server_parameters) = if admin {
+            let config = get_config();
+            // TODO: Add SASL support.
+            // Perform MD5 authentication.
+            match config.general.admin_auth_type {
+                AuthType::Trust => (),
+                AuthType::MD5 => {
+                    let salt = md5_challenge(&mut write).await?;
+
+                    let code = match read.read_u8().await {
+                        Ok(p) => p,
+                        Err(_) => {
+                            return Err(Error::ClientSocketError(
+                                "password code".into(),
+                                client_identifier,
+                            ))
+                        }
+                    };
+
+                    // PasswordMessage
+                    if code as char != 'p' {
+                        return Err(Error::ProtocolSyncError(format!(
+                            "Expected p, got {}",
+                            code as char
+                        )));
+                    }
+
+                    let len = match read.read_i32().await {
+                        Ok(len) => len,
+                        Err(_) => {
+                            return Err(Error::ClientSocketError(
+                                "password message length".into(),
+                                client_identifier,
+                            ))
+                        }
+                    };
+
+                    let mut password_response = vec![0u8; (len - 4) as usize];
+
+                    match read.read_exact(&mut password_response).await {
+                        Ok(_) => (),
+                        Err(_) => {
+                            return Err(Error::ClientSocketError(
+                                "password message".into(),
+                                client_identifier,
+                            ))
+                        }
+                    };
+
+                    // Compare server and client hashes.
+                    let password_hash = md5_hash_password(
+                        &config.general.admin_username,
+                        &config.general.admin_password,
+                        &salt,
+                    );
+
+                    if password_hash != password_response {
+                        let error =
+                            Error::ClientGeneralError("Invalid password".into(), client_identifier);
+
+                        warn!("{}", error);
+                        wrong_password(&mut write, username).await?;
+
+                        return Err(error);
+                    }
+                }
+            }
+            (false, generate_server_parameters_for_admin())
+        }
+        // Authenticate normal user.
+        else {
+            let pool = match get_pool(pool_name, username) {
+                Some(pool) => pool,
+                None => {
+                    error_response(
+                        &mut write,
+                        &format!(
+                            "No pool configured for database: {:?}, user: {:?}",
+                            pool_name, username
+                        ),
+                    )
+                    .await?;
+
+                    return Err(Error::ClientGeneralError(
+                        "Invalid pool name".into(),
+                        client_identifier,
+                    ));
+                }
+            };
+
+            // Obtain the hash to compare, we give preference to that written in cleartext in config
+            // if there is nothing set in cleartext and auth passthrough (auth_query) is configured, we use the hash obtained
+            // when the pool was created. If there is no hash there, we try to fetch it one more time.
+            match pool.settings.user.auth_type {
+                AuthType::Trust => (),
+                AuthType::MD5 => {
                     // Perform MD5 authentication.
                     // TODO: Add SASL support.
                     let salt = md5_challenge(&mut write).await?;
@@ -525,54 +626,6 @@ where
                         }
                     };
 
-        let mut prepared_statements_enabled = false;
-
-        // Authenticate admin user.
-        let (transaction_mode, mut server_parameters) = if admin {
-            let config = get_config();
-
-            // Compare server and client hashes.
-            let password_hash = md5_hash_password(
-                &config.general.admin_username,
-                &config.general.admin_password,
-                &salt,
-            );
-
-            if password_hash != password_response {
-                let error = Error::ClientGeneralError("Invalid password".into(), client_identifier);
-
-                warn!("{}", error);
-                wrong_password(&mut write, username).await?;
-
-                return Err(error);
-            }
-
-            (false, generate_server_parameters_for_admin())
-        }
-        // Authenticate normal user.
-        else {
-            let pool = match get_pool(pool_name, username) {
-                Some(pool) => pool,
-                None => {
-                    error_response(
-                        &mut write,
-                        &format!(
-                            "No pool configured for database: {:?}, user: {:?}",
-                            pool_name, username
-                        ),
-                    )
-                    .await?;
-
-                    return Err(Error::ClientGeneralError(
-                        "Invalid pool name".into(),
-                        client_identifier,
-                    ));
-                }
-            };
-
-            // Obtain the hash to compare, we give preference to that written in cleartext in config
-            // if there is nothing set in cleartext and auth passthrough (auth_query) is configured, we use the hash obtained
-            // when the pool was created. If there is no hash there, we try to fetch it one more time.
                     let password_hash = if let Some(password) = &pool.settings.user.password {
                         Some(md5_hash_password(username, password, &salt))
                     } else {
@@ -593,7 +646,10 @@ where
 
                             match refetch_auth_hash(&pool).await {
                                 Ok(fetched_hash) => {
-                            warn!("Password for {}, obtained. Updating.", client_identifier);
+                                    warn!(
+                                        "Password for {}, obtained. Updating.",
+                                        client_identifier
+                                    );
 
                                     {
                                         let mut pool_auth_hash = pool.auth_hash.write();
@@ -658,7 +714,8 @@ where
                             ));
                         }
                     }
-
+                }
+            }
             let transaction_mode = pool.settings.pool_mode == PoolMode::Transaction;
             prepared_statements_enabled =
                 transaction_mode && pool.prepared_statement_cache.is_some();

src/config.rs

@@ -208,6 +208,9 @@ impl Address {
 pub struct User {
     pub username: String,
     pub password: Option<String>,
+
+    #[serde(default = "User::default_auth_type")]
+    pub auth_type: AuthType,
     pub server_username: Option<String>,
     pub server_password: Option<String>,
     pub pool_size: u32,
@@ -225,6 +228,7 @@ impl Default for User {
         User {
             username: String::from("postgres"),
             password: None,
+            auth_type: AuthType::MD5,
             server_username: None,
             server_password: None,
             pool_size: 15,
@@ -239,6 +243,10 @@ impl Default for User {
 }
 
 impl User {
+    pub fn default_auth_type() -> AuthType {
+        AuthType::MD5
+    }
+
     fn validate(&self) -> Result<(), Error> {
         if let Some(min_pool_size) = self.min_pool_size {
             if min_pool_size > self.pool_size {
@@ -334,6 +342,9 @@ pub struct General {
     pub admin_username: String,
     pub admin_password: String,
 
+    #[serde(default = "General::default_admin_auth_type")]
+    pub admin_auth_type: AuthType,
+
     #[serde(default = "General::default_validate_config")]
     pub validate_config: bool,
 
@@ -348,6 +359,10 @@ impl General {
         "0.0.0.0".into()
     }
 
+    pub fn default_admin_auth_type() -> AuthType {
+        AuthType::MD5
+    }
+
     pub fn default_port() -> u16 {
         5432
     }
@@ -456,6 +471,7 @@ impl Default for General {
             verify_server_certificate: false,
             admin_username: String::from("admin"),
             admin_password: String::from("admin"),
+            admin_auth_type: AuthType::MD5,
             validate_config: true,
             auth_query: None,
             auth_query_user: None,
@@ -476,6 +492,15 @@ pub enum PoolMode {
     Session,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Copy, Hash)]
+pub enum AuthType {
+    #[serde(alias = "trust", alias = "Trust")]
+    Trust,
+
+    #[serde(alias = "md5", alias = "MD5")]
+    MD5,
+}
+
 impl std::fmt::Display for PoolMode {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {

tests/python/test_auth.py

@@ -0,0 +1,71 @@
+import utils
+import signal
+
+class TestTrustAuth:
+    @classmethod
+    def setup_method(cls):
+        config= """
+        [general]
+        host = "0.0.0.0"
+        port = 6432
+        admin_username = "admin_user"
+        admin_password = ""
+        admin_auth_type = "trust"
+
+        [pools.sharded_db.users.0]
+        username = "sharding_user"
+        password = "sharding_user"
+        auth_type = "trust"
+        pool_size = 10
+        min_pool_size = 1
+        pool_mode = "transaction"
+
+        [pools.sharded_db.shards.0]
+        servers = [
+          [ "127.0.0.1", 5432, "primary" ],
+        ]
+        database = "shard0"
+        """
+        utils.pgcat_generic_start(config)
+
+    @classmethod
+    def teardown_method(self):
+        utils.pg_cat_send_signal(signal.SIGTERM)
+
+    def test_admin_trust_auth(self):
+        conn, cur = utils.connect_db_trust(admin=True)
+        cur.execute("SHOW POOLS")
+        res = cur.fetchall()
+        print(res)
+        utils.cleanup_conn(conn, cur)
+
+    def test_normal_trust_auth(self):
+        conn, cur = utils.connect_db_trust(autocommit=False)
+        cur.execute("SELECT 1")
+        res = cur.fetchall()
+        print(res)
+        utils.cleanup_conn(conn, cur)
+
+class TestMD5Auth:
+    @classmethod
+    def setup_method(cls):
+        utils.pgcat_start()
+
+    @classmethod
+    def teardown_method(self):
+        utils.pg_cat_send_signal(signal.SIGTERM)
+
+    def test_normal_db_access(self):
+        conn, cur = utils.connect_db(autocommit=False)
+        cur.execute("SELECT 1")
+        res = cur.fetchall()
+        print(res)
+        utils.cleanup_conn(conn, cur)
+
+    def test_admin_db_access(self):
+        conn, cur = utils.connect_db(admin=True)
+
+        cur.execute("SHOW POOLS")
+        res = cur.fetchall()
+        print(res)
+        utils.cleanup_conn(conn, cur)

tests/python/test_pgcat.py

@@ -1,30 +1,12 @@
-import os
+
 import signal
 import time
 
 import psycopg2
-
 import utils
 
 SHUTDOWN_TIMEOUT = 5
 
-def test_normal_db_access():
-    utils.pgcat_start()
-    conn, cur = utils.connect_db(autocommit=False)
-    cur.execute("SELECT 1")
-    res = cur.fetchall()
-    print(res)
-    utils.cleanup_conn(conn, cur)
-
-
-def test_admin_db_access():
-    conn, cur = utils.connect_db(admin=True)
-
-    cur.execute("SHOW POOLS")
-    res = cur.fetchall()
-    print(res)
-    utils.cleanup_conn(conn, cur)
-
 
 def test_shutdown_logic():
 
@@ -256,3 +238,5 @@ def test_shutdown_logic():
 
     utils.cleanup_conn(conn, cur)
     utils.pg_cat_send_signal(signal.SIGTERM)
+
+    # - - - - - - - - - - - - - - - - - -

tests/python/utils.py

@@ -1,20 +1,49 @@
-from typing import Tuple
 import os
-import psutil
 import signal
 import time
+from typing import Tuple
+import tempfile
 
+import psutil
 import psycopg2
 
 PGCAT_HOST = "127.0.0.1"
 PGCAT_PORT = "6432"
 
-def pgcat_start():
+
+def _pgcat_start(config_path: str):
     pg_cat_send_signal(signal.SIGTERM)
-    os.system("./target/debug/pgcat .circleci/pgcat.toml &")
+    os.system(f"./target/debug/pgcat {config_path} &")
     time.sleep(2)
 
 
+def pgcat_start():
+    _pgcat_start(config_path='.circleci/pgcat.toml')
+
+
+def pgcat_generic_start(config: str):
+    tmp = tempfile.NamedTemporaryFile()
+    with open(tmp.name, 'w') as f:
+        f.write(config)
+    _pgcat_start(config_path=tmp.name)
+
+
+def glauth_send_signal(signal: signal.Signals):
+    try:
+        for proc in psutil.process_iter(["pid", "name"]):
+            if proc.name() == "glauth":
+                os.kill(proc.pid, signal)
+    except Exception as e:
+        # The process can be gone when we send this signal
+        print(e)
+
+    if signal == signal.SIGTERM:
+        # Returns 0 if pgcat process exists
+        time.sleep(2)
+        if not os.system('pgrep glauth'):
+            raise Exception("glauth not closed after SIGTERM")
+
+
 def pg_cat_send_signal(signal: signal.Signals):
     try:
         for proc in psutil.process_iter(["pid", "name"]):
@@ -54,6 +83,27 @@ def connect_db(
 
     return (conn, cur)
 
+def connect_db_trust(
+    autocommit: bool = True,
+    admin: bool = False,
+) -> Tuple[psycopg2.extensions.connection, psycopg2.extensions.cursor]:
+
+    if admin:
+        user = "admin_user"
+        db = "pgcat"
+    else:
+        user = "sharding_user"
+        db = "sharded_db"
+
+    conn = psycopg2.connect(
+        f"postgres://{user}@{PGCAT_HOST}:{PGCAT_PORT}/{db}?application_name=testing_pgcat",
+        connect_timeout=2,
+    )
+    conn.autocommit = autocommit
+    cur = conn.cursor()
+
+    return (conn, cur)
+
 
 def cleanup_conn(conn: psycopg2.extensions.connection, cur: psycopg2.extensions.cursor):
     cur.close()

tests/ruby/Gemfile.lock

@@ -1,22 +1,33 @@
 GEM
   remote: https://rubygems.org/
   specs:
-    activemodel (7.0.4.1)
+    activemodel (7.1.4)
-      activesupport (= 7.0.4.1)
+      activesupport (= 7.1.4)
-    activerecord (7.0.4.1)
+    activerecord (7.1.4)
-      activemodel (= 7.0.4.1)
+      activemodel (= 7.1.4)
-      activesupport (= 7.0.4.1)
+      activesupport (= 7.1.4)
-    activesupport (7.0.4.1)
+      timeout (>= 0.4.0)
+    activesupport (7.1.4)
+      base64
+      bigdecimal
       concurrent-ruby (~> 1.0, >= 1.0.2)
+      connection_pool (>= 2.2.5)
+      drb
       i18n (>= 1.6, < 2)
       minitest (>= 5.1)
+      mutex_m
       tzinfo (~> 2.0)
     ast (2.4.2)
-    concurrent-ruby (1.1.10)
+    base64 (0.2.0)
+    bigdecimal (3.1.8)
+    concurrent-ruby (1.3.4)
+    connection_pool (2.4.1)
     diff-lcs (1.5.0)
-    i18n (1.12.0)
+    drb (2.2.1)
+    i18n (1.14.5)
       concurrent-ruby (~> 1.0)
-    minitest (5.17.0)
+    minitest (5.25.1)
+    mutex_m (0.2.0)
     parallel (1.22.1)
     parser (3.1.2.0)
       ast (~> 2.4.1)
@@ -24,7 +35,8 @@ GEM
     pg (1.3.2)
     rainbow (3.1.1)
     regexp_parser (2.3.1)
-    rexml (3.2.5)
+    rexml (3.3.6)
+      strscan
     rspec (3.11.0)
       rspec-core (~> 3.11.0)
       rspec-expectations (~> 3.11.0)
@@ -50,10 +62,12 @@ GEM
     rubocop-ast (1.17.0)
       parser (>= 3.1.1.0)
     ruby-progressbar (1.11.0)
+    strscan (3.1.0)
+    timeout (0.4.1)
     toml (0.3.0)
       parslet (>= 1.8.0, < 3.0.0)
     toxiproxy (2.0.1)
-    tzinfo (2.0.5)
+    tzinfo (2.0.6)
       concurrent-ruby (~> 1.0)
     unicode-display_width (2.1.0)