fix: consistent SQL quoting and thread-safe TLS config

- Add quote_literal() for SQL string escaping
- Add quote_mysql_ident() for MySQL backtick quoting
- Use quote_ident() consistently for PostgreSQL/SQLite identifiers
- Replace unsafe std::env::set_var with thread-safe OnceLock
- Fix blocking std::thread::sleep in async context
- Update security test to verify credential leak fix

Author: christian-smith

src/commands/init.rs

@@ -330,7 +330,10 @@ pub async fn init(
                 .with_context(|| format!("Invalid database name: '{}'", db_info.name))?;
 
             // Try to create database atomically (avoids TOCTOU vulnerability)
-            let create_query = format!("CREATE DATABASE \"{}\"", db_info.name);
+            let create_query = format!(
+                "CREATE DATABASE {}",
+                crate::utils::quote_ident(&db_info.name)
+            );
             match target_client.execute(&create_query, &[]).await {
                 Ok(_) => {
                     tracing::info!("  Created database '{}'", db_info.name);
@@ -372,8 +375,10 @@ pub async fn init(
                                     drop_database_if_exists(&target_client, &db_info.name).await?;
 
                                     // Recreate the database
-                                    let create_query =
-                                        format!("CREATE DATABASE \"{}\"", db_info.name);
+                                    let create_query = format!(
+                                        "CREATE DATABASE {}",
+                                        crate::utils::quote_ident(&db_info.name)
+                                    );
                                     target_client
                                         .execute(&create_query, &[])
                                         .await
@@ -666,7 +671,10 @@ async fn drop_database_if_exists(target_conn: &Client, db_name: &str) -> Result<
     target_conn.execute(terminate_query, &[&db_name]).await?;
 
     // Drop the database
-    let drop_query = format!("DROP DATABASE IF EXISTS \"{}\"", db_name);
+    let drop_query = format!(
+        "DROP DATABASE IF EXISTS {}",
+        crate::utils::quote_ident(db_name)
+    );
     target_conn
         .execute(&drop_query, &[])
         .await

src/main.rs

@@ -9,8 +9,12 @@ use database_replicator::commands;
 #[command(about = "Universal database-to-PostgreSQL replication CLI", long_about = None)]
 #[command(version)]
 struct Cli {
-    /// Allow self-signed TLS certificates. Also honors SEREN_ALLOW_SELF_SIGNED_CERTS=1
-    #[arg(long = "allow-self-signed-certs", global = true, default_value_t = false)]
+    /// Allow self-signed TLS certificates (insecure - use only for testing)
+    #[arg(
+        long = "allow-self-signed-certs",
+        global = true,
+        default_value_t = false
+    )]
     allow_self_signed_certs: bool,
     #[command(subcommand)]
     command: Commands,
@@ -184,23 +188,8 @@ async fn main() -> anyhow::Result<()> {
 
     let cli = Cli::parse();
 
-    // Honor allow-self-signed flag or env var by setting env consumed in postgres::connection
-    let allow_self_signed_env = std::env::var("SEREN_ALLOW_SELF_SIGNED_CERTS")
-        .ok()
-        .map(|v| matches!(v.to_lowercase().as_str(), "1" | "true" | "yes"))
-        .unwrap_or(false)
-        // Backward compatibility with older env name
-        || std::env::var("SEREN_ALLOW_INVALID_CERTS")
-            .ok()
-            .map(|v| matches!(v.to_lowercase().as_str(), "1" | "true" | "yes"))
-            .unwrap_or(false);
-
-    if cli.allow_self_signed_certs || allow_self_signed_env {
-        // Set both names for compatibility
-        std::env::set_var("SEREN_ALLOW_SELF_SIGNED_CERTS", "1");
-        std::env::set_var("SEREN_ALLOW_INVALID_CERTS", "1");
-        tracing::warn!("Allowing self-signed/invalid TLS certificates (insecure)");
-    }
+    // Initialize TLS policy using thread-safe OnceLock
+    database_replicator::postgres::connection::init_tls_policy(cli.allow_self_signed_certs);
 
     match cli.command {
         Commands::Validate {

src/mysql/reader.rs

@@ -86,7 +86,11 @@ pub async fn get_table_row_count(
     tracing::debug!("Getting row count for table '{}.{}'", db_name, table_name);
 
     // Use backticks for identifiers to allow reserved words
-    let query = format!("SELECT COUNT(*) FROM `{}`.`{}`", db_name, table_name);
+    let query = format!(
+        "SELECT COUNT(*) FROM {}.{}",
+        crate::utils::quote_mysql_ident(db_name),
+        crate::utils::quote_mysql_ident(table_name)
+    );
 
     let count: Option<u64> = conn
         .query_first(&query)
@@ -137,7 +141,11 @@ pub async fn read_table_data(conn: &mut Conn, db_name: &str, table_name: &str) -
     tracing::info!("Reading all rows from table '{}.{}'", db_name, table_name);
 
     // Use backticks for identifiers
-    let query = format!("SELECT * FROM `{}`.`{}`", db_name, table_name);
+    let query = format!(
+        "SELECT * FROM {}.{}",
+        crate::utils::quote_mysql_ident(db_name),
+        crate::utils::quote_mysql_ident(table_name)
+    );
 
     let rows: Vec<Row> = conn
         .query(&query)

src/postgres/connection.rs

@@ -5,9 +5,28 @@ use crate::utils;
 use anyhow::{Context, Result};
 use native_tls::TlsConnector;
 use postgres_native_tls::MakeTlsConnector;
+use std::sync::OnceLock;
 use std::time::Duration;
 use tokio_postgres::Client;
 
+/// Thread-safe storage for TLS configuration set at startup
+static ALLOW_SELF_SIGNED_CERTS: OnceLock<bool> = OnceLock::new();
+
+/// Initialize the TLS certificate policy (call once at startup)
+///
+/// This must be called before any database connections are made.
+/// It is thread-safe and will only set the value once.
+///
+/// # Arguments
+///
+/// * `allow` - If true, accept self-signed/invalid TLS certificates (insecure)
+pub fn init_tls_policy(allow: bool) {
+    let _ = ALLOW_SELF_SIGNED_CERTS.set(allow);
+    if allow {
+        tracing::warn!("TLS policy: Allowing self-signed/invalid certificates (insecure)");
+    }
+}
+
 /// Add TCP keepalive parameters to a PostgreSQL connection string
 ///
 /// Automatically adds keepalive parameters to prevent idle connection timeouts
@@ -130,20 +149,11 @@ pub async fn connect(connection_string: &str) -> Result<Client> {
     )?;
 
     // Set up TLS connector for cloud connections
-    // By default, require valid certificates. Allow opt-in for self-signed/invalid certs via env.
-    let allow_self_signed = std::env::var("SEREN_ALLOW_SELF_SIGNED_CERTS")
-        .ok()
-        .map(|v| matches!(v.to_lowercase().as_str(), "1" | "true" | "yes"))
-        .unwrap_or(false)
-        // Backward compatibility with older env name
-        || std::env::var("SEREN_ALLOW_INVALID_CERTS")
-            .ok()
-            .map(|v| matches!(v.to_lowercase().as_str(), "1" | "true" | "yes"))
-            .unwrap_or(false);
+    // By default, require valid certificates. Allow opt-in via init_tls_policy() called at startup.
+    let allow_self_signed = ALLOW_SELF_SIGNED_CERTS.get().copied().unwrap_or(false);
 
     let mut tls_builder = TlsConnector::builder();
     if allow_self_signed {
-        tracing::warn!("Accepting self-signed/invalid TLS certificates");
         tls_builder.danger_accept_invalid_certs(true);
     }
 

src/replication/publication.rs

@@ -39,7 +39,10 @@ pub async fn create_publication(
     tracing::info!("Creating publication '{}'...", publication_name);
 
     if filter.is_empty() {
-        let query = format!("CREATE PUBLICATION \"{}\" FOR ALL TABLES", publication_name);
+        let query = format!(
+            "CREATE PUBLICATION {} FOR ALL TABLES",
+            crate::utils::quote_ident(publication_name)
+        );
         return execute_publication_query(client, publication_name, &query).await;
     }
 
@@ -121,8 +124,8 @@ pub async fn create_publication(
     );
 
     let query = format!(
-        "CREATE PUBLICATION \"{}\" FOR TABLE {}",
-        publication_name,
+        "CREATE PUBLICATION {} FOR TABLE {}",
+        crate::utils::quote_ident(publication_name),
         clauses.join(", ")
     );
 
@@ -218,7 +221,10 @@ pub async fn drop_publication(client: &Client, publication_name: &str) -> Result
 
     tracing::info!("Dropping publication '{}'...", publication_name);
 
-    let query = format!("DROP PUBLICATION IF EXISTS \"{}\"", publication_name);
+    let query = format!(
+        "DROP PUBLICATION IF EXISTS {}",
+        crate::utils::quote_ident(publication_name)
+    );
 
     client
         .execute(&query, &[])

src/replication/subscription.rs

@@ -50,12 +50,11 @@ pub async fn create_subscription(
         "  To avoid storing passwords, configure .pgpass on the target PostgreSQL server"
     );
 
-    // Escape single quotes in connection string to avoid breaking SQL literal
-    let escaped_conn = source_connection_string.replace('\'', "''");
-
     let query = format!(
-        "CREATE SUBSCRIPTION \"{}\" CONNECTION '{}' PUBLICATION \"{}\"",
-        subscription_name, escaped_conn, publication_name
+        "CREATE SUBSCRIPTION {} CONNECTION {} PUBLICATION {}",
+        crate::utils::quote_ident(subscription_name),
+        crate::utils::quote_literal(source_connection_string),
+        crate::utils::quote_ident(publication_name)
     );
 
     match client.execute(&query, &[]).await {
@@ -158,7 +157,10 @@ pub async fn drop_subscription(client: &Client, subscription_name: &str) -> Resu
 
     tracing::info!("Dropping subscription '{}'...", subscription_name);
 
-    let query = format!("DROP SUBSCRIPTION IF EXISTS \"{}\"", subscription_name);
+    let query = format!(
+        "DROP SUBSCRIPTION IF EXISTS {}",
+        crate::utils::quote_ident(subscription_name)
+    );
 
     client.execute(&query, &[]).await.context(format!(
         "Failed to drop subscription '{}'",

src/sqlite/reader.rs

@@ -95,7 +95,7 @@ pub fn get_table_row_count(conn: &Connection, table: &str) -> Result<usize> {
     tracing::debug!("Getting row count for table '{}'", table);
 
     // Note: table name is validated above, so it's safe to use in SQL
-    let query = format!("SELECT COUNT(*) FROM \"{}\"", table);
+    let query = format!("SELECT COUNT(*) FROM {}", crate::utils::quote_ident(table));
 
     let count: i64 = conn
         .query_row(&query, [], |row| row.get(0))
@@ -151,7 +151,7 @@ pub fn read_table_data(
     tracing::info!("Reading all data from table '{}'", table);
 
     // Note: table name is validated above
-    let query = format!("SELECT * FROM \"{}\"", table);
+    let query = format!("SELECT * FROM {}", crate::utils::quote_ident(table));
 
     let mut stmt = conn
         .prepare(&query)

src/utils.rs

@@ -261,7 +261,7 @@ where
 /// # use std::time::Duration;
 /// # use std::process::Command;
 /// # use database_replicator::utils::retry_subprocess_with_backoff;
-/// # fn example() -> Result<()> {
+/// # async fn example() -> Result<()> {
 /// retry_subprocess_with_backoff(
 ///     || {
 ///         let mut cmd = Command::new("psql");
@@ -311,7 +311,7 @@ where
                             max_retries + 1,
                             delay
                         );
-                        std::thread::sleep(delay);
+                        tokio::time::sleep(delay).await;
                         delay *= 2; // Exponential backoff
                     }
                 }
@@ -490,6 +490,57 @@ pub fn quote_ident(identifier: &str) -> String {
     quoted
 }
 
+/// Quote a SQL string literal (for use in SQL statements)
+///
+/// Escapes single quotes by doubling them and wraps the string in single quotes.
+/// Use this for string values in SQL, not for identifiers.
+///
+/// # Examples
+///
+/// ```
+/// use database_replicator::utils::quote_literal;
+/// assert_eq!(quote_literal("hello"), "'hello'");
+/// assert_eq!(quote_literal("it's"), "'it''s'");
+/// assert_eq!(quote_literal(""), "''");
+/// ```
+pub fn quote_literal(value: &str) -> String {
+    let mut quoted = String::with_capacity(value.len() + 2);
+    quoted.push('\'');
+    for ch in value.chars() {
+        if ch == '\'' {
+            quoted.push('\'');
+        }
+        quoted.push(ch);
+    }
+    quoted.push('\'');
+    quoted
+}
+
+/// Quote a MySQL identifier (database, table, column)
+///
+/// MySQL uses backticks for identifier quoting. Escapes embedded backticks
+/// by doubling them.
+///
+/// # Examples
+///
+/// ```
+/// use database_replicator::utils::quote_mysql_ident;
+/// assert_eq!(quote_mysql_ident("users"), "`users`");
+/// assert_eq!(quote_mysql_ident("user`name"), "`user``name`");
+/// ```
+pub fn quote_mysql_ident(identifier: &str) -> String {
+    let mut quoted = String::with_capacity(identifier.len() + 2);
+    quoted.push('`');
+    for ch in identifier.chars() {
+        if ch == '`' {
+            quoted.push('`');
+        }
+        quoted.push(ch);
+    }
+    quoted.push('`');
+    quoted
+}
+
 /// Validate that source and target URLs are different to prevent accidental data loss
 ///
 /// Compares two PostgreSQL connection URLs to ensure they point to different databases.

tests/integration_remote_test.rs

@@ -266,6 +266,8 @@ async fn test_remote_job_submission_with_filters() {
     // Create a job spec with database filters
     let filter = database_replicator::remote::FilterSpec {
         include_databases: Some(vec!["postgres".to_string()]),
+        exclude_databases: None,
+        include_tables: None,
         exclude_tables: None,
     };
 

tests/security_test.rs

@@ -787,33 +787,31 @@ fn test_mysql_url_with_special_chars_in_password() {
 fn test_mysql_error_messages_dont_leak_credentials() {
     use database_replicator::mysql;
 
-    // SECURITY NOTE: Current implementation includes full URL in error messages
-    // This test documents the current behavior - ideally this should be fixed
-    // to sanitize URLs before including in error messages
-
+    // SECURITY: Error messages should NOT leak passwords or full URLs
     let url_with_password = "not-mysql://admin:secretpass@host:3306/db";
 
     let result = mysql::validate_mysql_url(url_with_password);
     assert!(result.is_err(), "Invalid URL should be rejected");
 
     let error_msg = result.unwrap_err().to_string();
 
-    // KNOWN ISSUE: Error message currently contains the full URL including password
-    // This test verifies current behavior, but this should be improved
+    // Verify password is NOT leaked in error message
     assert!(
-        error_msg.contains("secretpass") || error_msg.contains("not-mysql://"),
-        "Error message currently includes full URL (known issue)"
+        !error_msg.contains("secretpass"),
+        "Error message should not contain password: {error_msg}"
+    );
+
+    // Verify full URL is NOT leaked in error message
+    assert!(
+        !error_msg.contains("not-mysql://"),
+        "Error message should not contain full malformed URL: {error_msg}"
     );
 
     // Verify it does explain the validation failure
     assert!(
         error_msg.contains("mysql://") || error_msg.contains("Invalid"),
-        "Error should explain validation requirement"
+        "Error should explain validation requirement: {error_msg}"
     );
-
-    // TODO: Enhance validate_mysql_url to sanitize URLs in error messages
-    // Expected: "Invalid MySQL connection string. Must start with 'mysql://'"
-    // (without exposing the actual malformed URL)
 }
 
 // ----------------------------------------------------------------------------