fix(clickhouse): support env:VAR / vault: spec syntax for credentials

Cargo.toml

@@ -32,7 +32,7 @@ tokio = { version = ">=1.48", features = ["full"] }
 tokio-util = { version = ">=0.7", features = ["rt"] }
 
 # Shared library with transport abstraction
-hyperi-rustlib = { version = ">=2.8.14, <3", features = ["transport-kafka", "transport-grpc", "dlq-kafka", "config", "config-reload", "deployment", "version-check", "scaling", "cli-service", "top", "logger", "metrics", "metrics-dfe", "expression", "memory", "worker-batch", "governor"] }
+hyperi-rustlib = { version = ">=2.8.14, <3", features = ["transport-kafka", "transport-grpc", "dlq-kafka", "config", "config-reload", "deployment", "version-check", "scaling", "cli-service", "top", "logger", "metrics", "metrics-dfe", "expression", "memory", "worker-batch", "governor", "credential"] }
 
 # CEL expression evaluation (same crate as rustlib uses, needed for Program type)
 cel = { version = ">=0.13, <0.14", features = ["json"] }

config.example.yaml

@@ -53,7 +53,10 @@ clickhouse:
   hosts:
     - localhost:9000
   database: default
+  # username may be a literal value, env:VAR, or vault:path:key
   username: default
+  # password may be a literal value, env:VAR, or vault:path:key
+  # (vault: requires the `secrets` feature, which the loader does not enable today)
   password: ""
   # Protocol: "native" (default) - HTTP not supported in Rust version
   # Native protocol is required for performance
@@ -69,6 +72,11 @@ clickhouse:
   #   ca_cert_file: /certs/ca.pem        # Optional: custom CA cert
   #   skip_verify: false
 
+# Example using env vars (uncomment to use):
+# clickhouse:
+#   username: env:CLICKHOUSE_USERNAME
+#   password: env:CLICKHOUSE_PASSWORD
+
 # =============================================================================
 # ClickHouse Cloud Configuration
 # =============================================================================
@@ -77,7 +85,7 @@ clickhouse:
 #     - your-instance.region.clickhouse.cloud:9440
 #   database: default
 #   username: default
-#   password: "your-cloud-password"
+#   password: env:CLICKHOUSE_CLOUD_PASSWORD
 #   protocol: native
 #   tls:
 #     enabled: true

src/config/credentials.rs

@@ -0,0 +1,94 @@
+// Project:   dfe-loader
+// File:      src/config/credentials.rs
+// Purpose:   Resolve env:/vault:/literal credential specs in ClickHouseConfig
+// Language:  Rust
+//
+// License:   FSL-1.1-ALv2
+// Copyright: (c) 2026 HYPERI PTY LIMITED
+
+//! Post-load credential resolution for the loader's `ClickHouseConfig`.
+//!
+//! Run once at startup, after `Config::load()` and before
+//! `From<&loader::ClickHouseConfig>` is invoked to build the internal
+//! `clickhouse::ClickHouseConfig`.
+
+use hyperi_rustlib::config::sensitive::SensitiveString;
+use hyperi_rustlib::credential::{CredentialError, resolve};
+
+use crate::config::loader::ClickHouseConfig;
+
+/// Resolve `username` and `password` spec strings into their plaintext values.
+///
+/// `env:VAR_NAME` reads `$VAR_NAME` (hard error if unset). `vault:path:key`
+/// fetches from OpenBao (returns `VaultUnsupported` because the loader does
+/// not enable rustlib's `secrets` feature). Any other string is used literally.
+pub async fn resolve_clickhouse_credentials(
+    cfg: &mut ClickHouseConfig,
+) -> Result<(), CredentialError> {
+    cfg.username = resolve(&cfg.username).await?;
+    let pwd = resolve(cfg.password.expose()).await?;
+    cfg.password = SensitiveString::from(pwd);
+    Ok(())
+}
+
+#[cfg(test)]
+#[allow(unsafe_code, clippy::unwrap_used, clippy::expect_used)]
+mod tests {
+    use super::*;
+
+    fn base_cfg() -> ClickHouseConfig {
+        let mut c = ClickHouseConfig::default();
+        c.username = "default".to_string();
+        c.password = SensitiveString::default();
+        c
+    }
+
+    #[tokio::test]
+    async fn literal_values_pass_through_unchanged() {
+        let mut c = base_cfg();
+        c.username = "alice".to_string();
+        c.password = SensitiveString::from("plain-text".to_string());
+        resolve_clickhouse_credentials(&mut c).await.unwrap();
+        assert_eq!(c.username, "alice");
+        assert_eq!(c.password.expose(), "plain-text");
+    }
+
+    #[tokio::test]
+    async fn env_specs_get_resolved() {
+        // SAFETY: test-only; uses unique var names to avoid interference with parallel tests
+        unsafe { std::env::set_var("DFE_LOADER_TEST_CH_USER", "carol") };
+        unsafe { std::env::set_var("DFE_LOADER_TEST_CH_PASS", "s3cret!") };
+
+        let mut c = base_cfg();
+        c.username = "env:DFE_LOADER_TEST_CH_USER".to_string();
+        c.password = SensitiveString::from("env:DFE_LOADER_TEST_CH_PASS".to_string());
+
+        resolve_clickhouse_credentials(&mut c).await.unwrap();
+        assert_eq!(c.username, "carol");
+        assert_eq!(c.password.expose(), "s3cret!");
+
+        unsafe { std::env::remove_var("DFE_LOADER_TEST_CH_USER") };
+        unsafe { std::env::remove_var("DFE_LOADER_TEST_CH_PASS") };
+    }
+
+    #[tokio::test]
+    async fn missing_env_returns_clear_error() {
+        let mut c = base_cfg();
+        c.password = SensitiveString::from("env:DFE_LOADER_NONEXISTENT_VAR_XYZ".to_string());
+        let err = resolve_clickhouse_credentials(&mut c).await.unwrap_err();
+        match err {
+            CredentialError::MissingEnvVar { name } => {
+                assert_eq!(name, "DFE_LOADER_NONEXISTENT_VAR_XYZ");
+            }
+            other => panic!("expected MissingEnvVar, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn vault_spec_without_secrets_feature_errors_cleanly() {
+        let mut c = base_cfg();
+        c.password = SensitiveString::from("vault:secret/ch:password".to_string());
+        let err = resolve_clickhouse_credentials(&mut c).await.unwrap_err();
+        assert!(matches!(err, CredentialError::VaultUnsupported));
+    }
+}

src/config/mod.rs

@@ -24,6 +24,7 @@
 //! let mut rx = shared.subscribe();
 //! ```
 
+pub mod credentials;
 pub mod kafka;
 pub mod loader;
 pub mod pipeline;

src/main.rs

@@ -87,6 +87,14 @@ impl DfeApp for App {
         let config_path = self.common_args().config.as_deref().map(String::from);
 
         async move {
+            let mut config = config;
+
+            // Resolve env:/vault: credential specs on ClickHouse credentials before
+            // anyone reads them. See src/config/credentials.rs for the spec syntax.
+            dfe_loader::config::credentials::resolve_clickhouse_credentials(&mut config.clickhouse)
+                .await
+                .map_err(|e| CliError::Config(e.to_string()))?;
+
             // Fire-and-forget version check
             hyperi_rustlib::VersionCheck::new(hyperi_rustlib::VersionCheckConfig {
                 product: "dfe-loader".into(),