types: resolve env vars in connector cfg

Fixes: #5799 Follows the same scheme as the kubernetes secret resolver. To resolve an env var successfully, the string must start with: `${env:` and end with `}`. The env var name then has to start from a valid alphabet, or underscore. The name then may contain numbers. Example: `"table": "${env:TABLE_NAME}"` Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com>
feldera · abhizer · Apr 9, 2026 · Mar 16, 2026 · Mar 16, 2026 · Apr 8, 2026
commit e4e3467c69b9c68696a3368b359ca65cbaa7235f
diff --git a/crates/feldera-types/src/secret_ref.rs b/crates/feldera-types/src/secret_ref.rs
@@ -6,6 +6,33 @@ use thiserror::Error as ThisError;
 /// RFC 1123 specification for a DNS label, which is also used by Kubernetes.
 pub const PATTERN_RFC_1123_DNS_LABEL: &str = r"^[a-z0-9]+(-[a-z0-9]+)*$";
 
+/// POSIX pattern for an environment variable name.
+pub const PATTERN_ENV_VAR_NAME: &str = r"^[a-zA-Z_][a-zA-Z0-9_]*$";
+
+#[derive(Debug, Clone, PartialEq, Eq, ThisError)]
+pub enum EnvVarNameParseError {
+    #[error("cannot be empty")]
+    Empty,
+    #[error(
+        "must only contain alphanumeric characters and underscores (_), and start with a letter or underscore"
+    )]
+    InvalidFormat,
+}
+
+/// Validates it is a valid POSIX environment variable name.
+pub fn validate_env_var_name(name: &str) -> Result<(), EnvVarNameParseError> {
+    if name.is_empty() {
+        Err(EnvVarNameParseError::Empty)
+    } else {
+        let re = Regex::new(PATTERN_ENV_VAR_NAME).expect("valid regular expression");
+        if re.is_match(name) {
+            Ok(())
+        } else {
+            Err(EnvVarNameParseError::InvalidFormat)
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, ThisError)]
 pub enum KubernetesSecretNameParseError {
     #[error("cannot be empty")]
@@ -90,6 +117,11 @@ pub enum SecretRef {
         /// Key inside the `data:` section of the `Secret` object.
         data_key: String,
     },
+    /// Reference to a process environment variable.
+    EnvVar {
+        /// Name of the environment variable.
+        name: String,
+    },
 }
 
 impl Display for SecretRef {
@@ -98,6 +130,9 @@ impl Display for SecretRef {
             SecretRef::Kubernetes { name, data_key } => {
                 write!(f, "${{secret:kubernetes:{name}/{data_key}}}")
             }
+            SecretRef::EnvVar { name } => {
+                write!(f, "${{env:{name}}}")
+            }
         }
     }
 }
@@ -134,26 +169,45 @@ pub enum MaybeSecretRefParseError {
         data_key: String,
         e: KubernetesSecretDataKeyParseError,
     },
+    #[error(
+        "environment variable reference '{env_ref_str}' has name '{name}' which is not valid: {e}"
+    )]
+    InvalidEnvVarName {
+        env_ref_str: String,
+        name: String,
+        e: EnvVarNameParseError,
+    },
+    #[error(
+        "environment variable reference '{env_ref_str}' is not valid: name cannot be empty"
+    )]
+    EmptyEnvVarName { env_ref_str: String },
 }
 
 impl MaybeSecretRef {
-    /// Determines whether a string is just a plain string or a reference to a secret.
+    /// Determines whether a string is just a plain string, a reference to a secret,
+    /// or a reference to a process environment variable.
     ///
     /// - Secret reference: any string which starts with `${secret:` and ends with `}`
     ///   is regarded as an attempt to declare a secret reference
+    /// - Environment variable reference: any string which starts with `${env:` and ends with `}`
+    ///   is regarded as an attempt to declare an environment variable reference
     /// - Plain string: any other string
     ///
     /// A secret reference must follow the following pattern:
     /// `${secret:<provider>:<identifier>}`
     ///
-    /// An error is returned if a string is regarded as a secret reference (see above), but:
-    /// - Specifies a `<provider>` which does not exist
-    /// - Specifies a `<identifier>` which does not meet the provider-specific requirements
+    /// An environment variable reference must follow the following pattern:
+    /// `${env:<name>}`
+    ///
+    /// An error is returned if a string is regarded as a secret or env var reference (see above), but:
+    /// - Specifies a `<provider>` which does not exist (for secret refs)
+    /// - Specifies a `<name>` which does not meet the requirements
     ///
     /// Supported providers and their identifier expectations:
     /// - `${secret:kubernetes:<name>/<data key>}`
+    /// - `${env:<name>}` where `<name>` follows POSIX env var naming rules
     ///
-    /// Note that here is not checked whether the secret reference can actually be resolved.
+    /// Note that here is not checked whether the reference can actually be resolved.
     pub fn new(value: String) -> Result<MaybeSecretRef, MaybeSecretRefParseError> {
         if value.starts_with("${secret:") && value.ends_with('}') {
             // Because the pattern only has ASCII characters, they are encoded as single bytes.
@@ -191,6 +245,25 @@ impl MaybeSecretRef {
                     secret_ref_str: value,
                 })
             }
+        } else if value.starts_with("${env:") && value.ends_with('}') {
+            // Environment variable reference: `${env:<name>}`
+            // The content is extracted by slicing away the first 6 bytes ("${env:") and the last byte ("}").
+            let from_idx_incl = 6;
+            let till_idx_excl = value.len() - 1;
+            let name = value[from_idx_incl..till_idx_excl].to_string();
+            if name.is_empty() {
+                Err(MaybeSecretRefParseError::EmptyEnvVarName {
+                    env_ref_str: value,
+                })
+            } else if let Err(e) = validate_env_var_name(&name) {
+                Err(MaybeSecretRefParseError::InvalidEnvVarName {
+                    env_ref_str: value,
+                    name,
+                    e,
+                })
+            } else {
+                Ok(MaybeSecretRef::SecretRef(SecretRef::EnvVar { name }))
+            }
         } else {
             Ok(MaybeSecretRef::String(value))
         }
@@ -213,8 +286,9 @@ impl Display for MaybeSecretRef {
 #[cfg(test)]
 mod tests {
     use super::{
-        KubernetesSecretDataKeyParseError, KubernetesSecretNameParseError, MaybeSecretRef,
-        validate_kubernetes_secret_data_key, validate_kubernetes_secret_name,
+        EnvVarNameParseError, KubernetesSecretDataKeyParseError, KubernetesSecretNameParseError,
+        MaybeSecretRef, validate_env_var_name, validate_kubernetes_secret_data_key,
+        validate_kubernetes_secret_name,
     };
     use super::{MaybeSecretRefParseError, SecretRef};
 
@@ -228,6 +302,12 @@ mod tests {
             }),
             "${secret:kubernetes:example/value}"
         );
+        assert_eq!(
+            format!("{}", SecretRef::EnvVar {
+                name: "MY_VAR".to_string(),
+            }),
+            "${env:MY_VAR}"
+        );
     }
 
     #[test]
@@ -453,4 +533,58 @@ mod tests {
             assert_eq!(validate_kubernetes_secret_data_key(value), expectation);
         }
     }
+
+    #[test]
+    #[rustfmt::skip] // Skip formatting to keep it short
+    fn env_var_name_validation() {
+        for (value, expectation) in vec![
+            ("A", Ok(())),
+            ("a", Ok(())),
+            ("_", Ok(())),
+            ("A1", Ok(())),
+            ("MY_VAR", Ok(())),
+            ("_MY_VAR", Ok(())),
+            ("MY_VAR_123", Ok(())),
+            ("", Err(EnvVarNameParseError::Empty)),
+            ("1A", Err(EnvVarNameParseError::InvalidFormat)),
+            ("MY-VAR", Err(EnvVarNameParseError::InvalidFormat)),
+            ("MY VAR", Err(EnvVarNameParseError::InvalidFormat)),
+            ("MY.VAR", Err(EnvVarNameParseError::InvalidFormat)),
+        ] {
+            assert_eq!(validate_env_var_name(value), expectation);
+        }
+    }
+
+    #[test]
+    #[rustfmt::skip] // Skip formatting to keep it short
+    fn maybe_secret_ref_parse_env_var() {
+        let values_and_expectations = vec![
+            // Valid env var references
+            ("${env:A}", Ok(MaybeSecretRef::SecretRef(SecretRef::EnvVar { name: "A".to_string() }))),
+            ("${env:MY_VAR}", Ok(MaybeSecretRef::SecretRef(SecretRef::EnvVar { name: "MY_VAR".to_string() }))),
+            ("${env:_MY_VAR}", Ok(MaybeSecretRef::SecretRef(SecretRef::EnvVar { name: "_MY_VAR".to_string() }))),
+            ("${env:MY_VAR_123}", Ok(MaybeSecretRef::SecretRef(SecretRef::EnvVar { name: "MY_VAR_123".to_string() }))),
+            // Empty name
+            ("${env:}", Err(MaybeSecretRefParseError::EmptyEnvVarName {
+                env_ref_str: "${env:}".to_string()
+            })),
+            // Invalid name: starts with digit
+            ("${env:1VAR}", Err(MaybeSecretRefParseError::InvalidEnvVarName {
+                env_ref_str: "${env:1VAR}".to_string(),
+                name: "1VAR".to_string(),
+                e: EnvVarNameParseError::InvalidFormat
+            })),
+            // Invalid name: contains hyphen
+            ("${env:MY-VAR}", Err(MaybeSecretRefParseError::InvalidEnvVarName {
+                env_ref_str: "${env:MY-VAR}".to_string(),
+                name: "MY-VAR".to_string(),
+                e: EnvVarNameParseError::InvalidFormat
+            })),
+            // Not an env var reference (no closing brace match for opening pattern)
+            ("${env:", Ok(MaybeSecretRef::String("${env:".to_string()))),
+            // Plain strings that look similar but are not env var references
+            ("$env:MY_VAR}", Ok(MaybeSecretRef::String("$env:MY_VAR}".to_string()))),
+        ];
+        test_values_and_expectations(values_and_expectations);
+    }
 }
diff --git a/crates/feldera-types/src/secret_resolver.rs b/crates/feldera-types/src/secret_resolver.rs
@@ -4,6 +4,7 @@ use serde::Serialize;
 use serde::de::DeserializeOwned;
 use serde_json::{Map, Value};
 use std::collections::BTreeSet;
+use std::env;
 use std::fmt::Debug;
 use std::fs;
 use std::io::ErrorKind;
@@ -102,6 +103,10 @@ pub enum SecretRefResolutionError {
         path: String,
         error_kind: ErrorKind,
     },
+    #[error(
+        "environment variable reference '{env_ref}' resolution failed: environment variable '{name}' is not set"
+    )]
+    EnvVarNotSet { env_ref: SecretRef, name: String },
     #[error("secret resolution led to a duplicate key in the mapping, which should not happen")]
     DuplicateKeyInMapping,
     #[error("unable to serialize connector configuration: {error}")]
@@ -171,16 +176,19 @@ fn resolve_secret_references_in_json(
     })
 }
 
-/// Resolves a string which can potentially be a secret reference.
+/// Resolves a string which can potentially be a secret reference or an environment variable reference.
 fn resolve_potential_secret_reference_string(
     secrets_dir: &Path,
     s: String,
 ) -> Result<String, SecretRefResolutionError> {
     match MaybeSecretRef::new(s) {
         Ok(maybe_secret_ref) => match maybe_secret_ref {
             MaybeSecretRef::String(plain_str) => Ok(plain_str),
-            MaybeSecretRef::SecretRef(secret_ref) => match &secret_ref {
-                SecretRef::Kubernetes { name, data_key } => {
+            MaybeSecretRef::SecretRef(secret_ref) => match secret_ref {
+                SecretRef::Kubernetes {
+                    ref name,
+                    ref data_key,
+                } => {
                     // Secret reference: `${secret:kubernetes:<name>/<data key>}`
                     // File location: `<secrets dir>/kubernetes/<name>/<data key>`
                     let path = Path::new(secrets_dir)
@@ -224,6 +232,20 @@ fn resolve_potential_secret_reference_string(
                         }
                     }
                 }
+                SecretRef::EnvVar { ref name } => {
+                    // Environment variable reference: `${env:<name>}`
+                    // Resolved by reading the named environment variable from the process.
+                    let name = name.clone();
+                    match env::var(&name) {
+                        Ok(value) => Ok(value),
+                        Err(env::VarError::NotPresent) | Err(env::VarError::NotUnicode(_)) => {
+                            Err(SecretRefResolutionError::EnvVarNotSet {
+                                env_ref: secret_ref,
+                                name,
+                            })
+                        }
+                    }
+                }
             },
         },
         Err(e) => Err(SecretRefResolutionError::MaybeSecretRefParseFailed { e }),
@@ -565,4 +587,107 @@ mod tests {
             Some("${secret:kubernetes:e/f}".to_string())
         );
     }
+
+    #[test]
+    fn resolve_env_var_success() {
+        // Set the environment variable
+        unsafe {
+            std::env::set_var("FELDERA_TEST_ENV_VAR_ABC123", "my_value");
+        }
+
+        let dir = tempfile::tempdir().unwrap();
+        assert_eq!(
+            resolve_potential_secret_reference_string(
+                dir.path(),
+                "${env:FELDERA_TEST_ENV_VAR_ABC123}".to_string()
+            )
+            .unwrap(),
+            "my_value"
+        );
+
+        unsafe {
+            std::env::remove_var("FELDERA_TEST_ENV_VAR_ABC123");
+        }
+    }
+
+    #[test]
+    fn resolve_env_var_not_set() {
+        let dir = tempfile::tempdir().unwrap();
+        let env_ref_str = "${env:FELDERA_TEST_ENV_VAR_NOT_SET_XYZ}";
+        unsafe {
+            std::env::remove_var("FELDERA_TEST_ENV_VAR_NOT_SET_XYZ");
+        }
+
+        let MaybeSecretRef::SecretRef(expected_ref) =
+            crate::secret_ref::MaybeSecretRef::new(env_ref_str.to_string()).unwrap()
+        else {
+            unreachable!();
+        };
+
+        assert_eq!(
+            resolve_potential_secret_reference_string(dir.path(), env_ref_str.to_string())
+                .unwrap_err(),
+            SecretRefResolutionError::EnvVarNotSet {
+                env_ref: expected_ref,
+                name: "FELDERA_TEST_ENV_VAR_NOT_SET_XYZ".to_string(),
+            }
+        );
+    }
+
+    #[test]
+    fn resolve_env_var_in_connector_config() {
+        unsafe {
+            std::env::set_var("FELDERA_TEST_CONN_VAR_A", "resolved_value_a");
+            std::env::set_var("FELDERA_TEST_CONN_VAR_B", "resolved_value_b");
+        }
+
+        let connector_config_json = json!({
+            "transport": {
+              "name": "datagen",
+              "config": {
+                "plan": [{
+                    "limit": 2,
+                    "fields": {
+                        "col1": { "values": [1, 2] },
+                        "col2": { "values": ["${env:FELDERA_TEST_CONN_VAR_A}", "${env:FELDERA_TEST_CONN_VAR_B}"] }
+                    }
+                }]
+              }
+            },
+            "format": {
+              "name": "json",
+              "config": {
+                "example": "${env:FELDERA_TEST_CONN_VAR_A}"
+              }
+            }
+        });
+
+        let connector_config: ConnectorConfig =
+            serde_json::from_value(connector_config_json).unwrap();
+
+        let dir = tempfile::tempdir().unwrap();
+        let resolved =
+            resolve_secret_references_in_connector_config(dir.path(), &connector_config).unwrap();
+
+        let TransportConfig::Datagen(datagen_input_config) = resolved.transport else {
+            unreachable!();
+        };
+        assert_eq!(
+            datagen_input_config.plan[0].fields["col2"]
+                .values
+                .as_ref()
+                .unwrap(),
+            &vec![json!("resolved_value_a"), json!("resolved_value_b")]
+        );
+
+        let Some(format_config) = resolved.format else {
+            unreachable!();
+        };
+        assert_eq!(format_config.config, json!({"example": "resolved_value_a"}));
+
+        unsafe {
+            std::env::remove_var("FELDERA_TEST_CONN_VAR_A");
+            std::env::remove_var("FELDERA_TEST_CONN_VAR_B");
+        }
+    }
 }