Allow query_auth configuration reload

This adds support for reloading query_auth configuration.

Current implementation:

- Checks whether `auth_query` configuration is active, if so, it fetches
  (or refetches) hashes from servers and sets up pools. When it detects a
  password change, the pool is dropped and a new one is created, current
  established connections will be left connected.
- When we go from 'auth_query' configured to unconfigured the reload logic
  detects it and drops Md5 hashes stored in connection pools, so cleartext
  passwords are used instead.

Author: magec

src/auth_passthrough.rs

@@ -38,11 +38,7 @@ impl AuthPassthrough {
     pub fn from_config() -> Option<Self> {
         let config = get_config();
 
-        if config.general.auth_query_password.is_some()
-            && config.general.auth_query_user.is_some()
-            && config.general.auth_query_password.is_some()
-            && config.general.auth_query_database.is_some()
-        {
+        if config.is_auth_query_configured() {
             return Some(AuthPassthrough::new(
                 config.general.auth_query.as_ref().unwrap(),
                 config.general.auth_query_user.as_ref().unwrap(),
@@ -238,3 +234,14 @@ async fn send_auth_query(server: &mut Server, query: &str) -> Result<(), Error>
         }
     };
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_from_config() {
+        let auth_passthrough = AuthPassthrough::from_config();
+        assert!(auth_passthrough.is_none());
+    }
+}

src/config.rs

@@ -12,7 +12,7 @@ use tokio::fs::File;
 use tokio::io::AsyncReadExt;
 
 use crate::errors::Error;
-use crate::pool::{ClientServerMap, ConnectionPool};
+use crate::pool::{drop_auth_hashes, ClientServerMap, ConnectionPool};
 use crate::sharding::ShardingFunction;
 use crate::tls::{load_certs, load_keys};
 
@@ -698,6 +698,13 @@ impl Config {
         }
     }
 
+    pub fn is_auth_query_configured(&self) -> bool {
+        self.general.auth_query_password.is_some()
+            && self.general.auth_query_user.is_some()
+            && self.general.auth_query_password.is_some()
+            && self.general.auth_query_database.is_some()
+    }
+
     pub fn validate(&mut self) -> Result<(), Error> {
         // Validation for auth_query feature
         if self.general.auth_query.is_some()
@@ -815,10 +822,21 @@ pub async fn reload_config(client_server_map: ClientServerMap) -> Result<bool, E
     };
     let new_config = get_config();
 
+    if old_config.is_auth_query_configured() && !new_config.is_auth_query_configured() {
+        // We should clean up auth_hashes so new client connections
+        // dont use old hashes.
+        info!("Auth query support deactivated. Deleting old auth hashes.");
+        drop_auth_hashes()
+    }
+
     if old_config.pools != new_config.pools {
         info!("Pool configuration changed");
         ConnectionPool::from_config(client_server_map).await?;
         Ok(true)
+    } else if new_config.is_auth_query_configured() {
+        info!("Auth query configured. Reloading config also checks for password changes in pools.");
+        ConnectionPool::from_config(client_server_map).await?;
+        Ok(true)
     } else if old_config != new_config {
         Ok(true)
     } else {

src/pool.rs

@@ -147,8 +147,25 @@ pub struct ConnectionPool {
 }
 
 impl ConnectionPool {
-    async fn refresh_auth_hashes(&mut self) {
-        debug!("Auth hash refresh not implemented yet for unchanged pools.")
+    async fn auth_hash_has_changed(&mut self) -> bool {
+        if let Some(apt) = AuthPassthrough::from_config() {
+            if let Some(shard) = self.addresses.first() {
+                if let Some(address) = shard.first() {
+                    match apt.fetch_hash(address).await {
+			Ok(md5) => {
+			    if let Some(auth_hash) = self.auth_hash.as_ref() {
+				if auth_hash != &md5 {
+				    info!("Password hash changed for user: {:?}. Pool will be reloaded.", address.username);
+				    return true
+				}
+			    }
+			},
+			Err(err) => warn!("Could not obtain password hash using auth_query while trying to reload config, ignoring. Error: {:?}", err),
+		    }
+                }
+            }
+        }
+        false
     }
 
     /// Construct the connection pool from the configuration.
@@ -177,12 +194,13 @@ impl ConnectionPool {
                                 "[pool: {}][user: {}] has not changed",
                                 pool_name, user.username
                             );
-                            pool.refresh_auth_hashes().await;
-                            new_pools.insert(
-                                PoolIdentifier::new(pool_name, &user.username),
-                                pool.clone(),
-                            );
-                            continue;
+                            if !(pool.auth_hash_has_changed().await) {
+                                new_pools.insert(
+                                    PoolIdentifier::new(pool_name, &user.username),
+                                    pool.clone(),
+                                );
+                                continue;
+                            }
                         }
                         None => (),
                     }
@@ -766,6 +784,15 @@ pub fn get_pool(db: &str, user: &str) -> Option<ConnectionPool> {
         .cloned()
 }
 
+pub fn drop_auth_hashes() {
+    let mut new_pools = get_all_pools();
+
+    for (_id, pool) in new_pools.iter_mut() {
+        pool.auth_hash = None;
+    }
+    POOLS.store(Arc::new(new_pools));
+}
+
 /// Get a pointer to all configured pools.
 pub fn get_all_pools() -> HashMap<PoolIdentifier, ConnectionPool> {
     (*(*POOLS.load())).clone()