liminal_server/config/
env.rs

1use std::ffi::{OsStr, OsString};
2use std::net::SocketAddr;
3use std::path::PathBuf;
4
5use crate::ServerError;
6
7use super::types::ServerConfig;
8
9const ENV_PREFIX: &str = "LIMINAL_";
10const LISTEN_ADDRESS: &str = "LIMINAL_LISTEN_ADDRESS";
11const HEALTH_LISTEN_ADDRESS: &str = "LIMINAL_HEALTH_LISTEN_ADDRESS";
12const DRAIN_TIMEOUT_MS: &str = "LIMINAL_DRAIN_TIMEOUT_MS";
13const PERSISTENCE_PATH: &str = "LIMINAL_PERSISTENCE_PATH";
14const CLUSTER_NODE_NAME: &str = "LIMINAL_CLUSTER_NODE_NAME";
15const CLUSTER_SEED_NODES: &str = "LIMINAL_CLUSTER_SEED_NODES";
16const CLUSTER_LISTEN_ADDRESS: &str = "LIMINAL_CLUSTER_LISTEN_ADDRESS";
17const CLUSTER_COOKIE: &str = "LIMINAL_CLUSTER_COOKIE";
18const AUTH_TOKEN: &str = "LIMINAL_AUTH_TOKEN";
19
20/// Applies supported `LIMINAL_` environment variable overrides to a config.
21///
22/// Absent variables leave the supplied configuration unchanged. Supported
23/// variables are `LIMINAL_LISTEN_ADDRESS`, `LIMINAL_HEALTH_LISTEN_ADDRESS`,
24/// `LIMINAL_DRAIN_TIMEOUT_MS`, `LIMINAL_PERSISTENCE_PATH`,
25/// `LIMINAL_CLUSTER_NODE_NAME`, `LIMINAL_CLUSTER_SEED_NODES`,
26/// `LIMINAL_CLUSTER_LISTEN_ADDRESS`, `LIMINAL_CLUSTER_COOKIE`, and
27/// `LIMINAL_AUTH_TOKEN`.
28///
29/// Unlike the cluster overrides — which refuse to fabricate a `[cluster]` section
30/// that the file did not declare (a partially-specified cluster is unsafe) —
31/// `LIMINAL_AUTH_TOKEN` MAY create the `[auth]` section when the file omits it.
32/// The auth section is a single scalar secret with no other required fields, so a
33/// deployment can inject the token purely from the environment (the idiomatic way
34/// to supply a secret) without also pinning it in a committed config file; an empty
35/// value still fails the later non-empty validation, so this cannot silently create
36/// a no-op gate.
37///
38/// # Errors
39///
40/// Returns [`ServerError`] when a present environment variable cannot be parsed
41/// or targets cluster configuration that was not declared in the file.
42pub fn apply_env_overrides(config: ServerConfig) -> Result<ServerConfig, ServerError> {
43    apply_env_overrides_from(config, std::env::vars_os())
44}
45
46pub(crate) fn apply_env_overrides_from<I>(
47    mut config: ServerConfig,
48    variables: I,
49) -> Result<ServerConfig, ServerError>
50where
51    I: IntoIterator<Item = (OsString, OsString)>,
52{
53    for (key, value) in variables {
54        let Some(key) = key.to_str() else {
55            continue;
56        };
57
58        if !key.starts_with(ENV_PREFIX) {
59            continue;
60        }
61
62        match key {
63            LISTEN_ADDRESS => {
64                config.listen_address = parse_socket_addr(LISTEN_ADDRESS, &value)?;
65            }
66            HEALTH_LISTEN_ADDRESS => {
67                config.health_listen_address = parse_socket_addr(HEALTH_LISTEN_ADDRESS, &value)?;
68            }
69            DRAIN_TIMEOUT_MS => {
70                config.drain_timeout_ms = parse_u64(DRAIN_TIMEOUT_MS, &value)?;
71            }
72            PERSISTENCE_PATH => {
73                config.persistence_path = Some(PathBuf::from(value));
74            }
75            CLUSTER_NODE_NAME => {
76                let node_name = env_string(CLUSTER_NODE_NAME, &value)?;
77                cluster_required(&mut config, CLUSTER_NODE_NAME)?.node_name = node_name;
78            }
79            CLUSTER_SEED_NODES => {
80                let seed_nodes = parse_seed_nodes(&value)?;
81                cluster_required(&mut config, CLUSTER_SEED_NODES)?.seed_nodes = seed_nodes;
82            }
83            CLUSTER_LISTEN_ADDRESS => {
84                let listen_address = parse_socket_addr(CLUSTER_LISTEN_ADDRESS, &value)?;
85                cluster_required(&mut config, CLUSTER_LISTEN_ADDRESS)?.listen_address =
86                    listen_address;
87            }
88            CLUSTER_COOKIE => {
89                let cookie = env_string(CLUSTER_COOKIE, &value)?;
90                cluster_required(&mut config, CLUSTER_COOKIE)?.cookie = cookie;
91            }
92            AUTH_TOKEN => {
93                // A single scalar secret is allowed to create the `[auth]` section
94                // even when the file omits it (see the module-level note); an empty
95                // value is left to fail the non-empty auth validation, not rejected
96                // here, so precedence and error reporting stay uniform.
97                let token = env_string(AUTH_TOKEN, &value)?;
98                config.auth = Some(super::types::AuthConfig { token });
99            }
100            _ => {}
101        }
102    }
103
104    Ok(config)
105}
106
107fn parse_socket_addr(name: &str, value: &OsStr) -> Result<SocketAddr, ServerError> {
108    let value = env_string(name, value)?;
109    value.parse::<SocketAddr>().map_err(|error| {
110        config_load(format!(
111            "environment variable {name} must be a socket address: {error}"
112        ))
113    })
114}
115
116fn parse_u64(name: &str, value: &OsStr) -> Result<u64, ServerError> {
117    let value = env_string(name, value)?;
118    value.parse::<u64>().map_err(|error| {
119        config_load(format!(
120            "environment variable {name} must be an unsigned integer: {error}"
121        ))
122    })
123}
124
125fn parse_seed_nodes(value: &OsStr) -> Result<Vec<SocketAddr>, ServerError> {
126    let value = env_string(CLUSTER_SEED_NODES, value)?;
127    if value.trim().is_empty() {
128        return Ok(Vec::new());
129    }
130
131    value
132        .split(',')
133        .enumerate()
134        .map(|(index, candidate)| parse_seed_node(index, candidate))
135        .collect()
136}
137
138fn parse_seed_node(index: usize, candidate: &str) -> Result<SocketAddr, ServerError> {
139    let candidate = candidate.trim();
140    if candidate.is_empty() {
141        return Err(config_load(format!(
142            "environment variable {CLUSTER_SEED_NODES} contains an empty seed node at position {}",
143            index + 1
144        )));
145    }
146
147    candidate.parse::<SocketAddr>().map_err(|error| {
148        config_load(format!(
149            "environment variable {CLUSTER_SEED_NODES} contains invalid seed node '{}' at position {}: {error}",
150            candidate,
151            index + 1
152        ))
153    })
154}
155
156fn env_string(name: &str, value: &OsStr) -> Result<String, ServerError> {
157    value.to_str().map(str::to_owned).ok_or_else(|| {
158        config_load(format!(
159            "environment variable {name} contains non-Unicode data"
160        ))
161    })
162}
163
164fn cluster_required<'a>(
165    config: &'a mut ServerConfig,
166    name: &str,
167) -> Result<&'a mut super::types::ClusterConfig, ServerError> {
168    config
169        .cluster
170        .as_mut()
171        .ok_or_else(|| ServerError::ConfigValidation {
172            message: format!(
173                "environment variable {name} requires a [cluster] section in the configuration file"
174            ),
175        })
176}
177
178const fn config_load(message: String) -> ServerError {
179    ServerError::ConfigLoad { message }
180}
181
182#[cfg(test)]
183mod tests {
184    use std::ffi::OsString;
185    use std::net::SocketAddr;
186    use std::path::{Path, PathBuf};
187
188    use crate::ServerError;
189
190    use super::apply_env_overrides_from;
191    use crate::config::types::{ChannelDef, ClusterConfig, RoutingRuleDef, ServerConfig};
192    use crate::config::{load_from_file, validate};
193
194    fn socket(address: &str) -> Result<SocketAddr, Box<dyn std::error::Error>> {
195        Ok(address.parse()?)
196    }
197
198    fn sample_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
199        Ok(ServerConfig {
200            listen_address: socket("127.0.0.1:8080")?,
201            health_listen_address: socket("127.0.0.1:8081")?,
202            drain_timeout_ms: 30_000,
203            channels: vec![ChannelDef {
204                name: "orders".to_owned(),
205                schema_ref: None,
206                durable: true,
207                loaded_schema: None,
208            }],
209            routing_rules: vec![RoutingRuleDef {
210                source_channel: "orders".to_owned(),
211                target_channel: "orders".to_owned(),
212                predicate: None,
213            }],
214            persistence_path: Some(PathBuf::from("/tmp")),
215            cluster: Some(ClusterConfig {
216                node_name: "node-a".to_owned(),
217                listen_address: socket("127.0.0.1:9000")?,
218                seed_nodes: vec![socket("127.0.0.1:9001")?],
219                cookie: "test-cookie".to_owned(),
220            }),
221            auth: None,
222        })
223    }
224
225    fn env_pair(name: &str, value: &str) -> (OsString, OsString) {
226        (OsString::from(name), OsString::from(value))
227    }
228
229    fn write_temp_config(contents: &str) -> Result<PathBuf, Box<dyn std::error::Error>> {
230        let path = std::env::temp_dir().join(format!(
231            "liminal-server-env-pipeline-{}.toml",
232            std::process::id()
233        ));
234        std::fs::write(&path, contents)?;
235        Ok(path)
236    }
237
238    fn remove_temp_file(path: &Path) -> Result<(), Box<dyn std::error::Error>> {
239        if path.exists() {
240            std::fs::remove_file(path)?;
241        }
242        Ok(())
243    }
244
245    #[test]
246    fn listen_address_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>> {
247        let config = sample_config()?;
248        let config = apply_env_overrides_from(
249            config,
250            vec![env_pair("LIMINAL_LISTEN_ADDRESS", "0.0.0.0:9090")],
251        )?;
252
253        assert_eq!(config.listen_address, socket("0.0.0.0:9090")?);
254
255        Ok(())
256    }
257
258    #[test]
259    fn health_listen_address_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>>
260    {
261        let config = sample_config()?;
262        let config = apply_env_overrides_from(
263            config,
264            vec![env_pair("LIMINAL_HEALTH_LISTEN_ADDRESS", "0.0.0.0:9191")],
265        )?;
266
267        assert_eq!(config.health_listen_address, socket("0.0.0.0:9191")?);
268
269        Ok(())
270    }
271
272    #[test]
273    fn drain_timeout_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>> {
274        let config = sample_config()?;
275        let config =
276            apply_env_overrides_from(config, vec![env_pair("LIMINAL_DRAIN_TIMEOUT_MS", "1250")])?;
277
278        assert_eq!(config.drain_timeout_ms, 1250);
279
280        Ok(())
281    }
282
283    #[test]
284    fn persistence_path_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>> {
285        let config = sample_config()?;
286        let config = apply_env_overrides_from(
287            config,
288            vec![env_pair("LIMINAL_PERSISTENCE_PATH", "/var/lib/liminal")],
289        )?;
290
291        assert_eq!(
292            config.persistence_path.as_deref(),
293            Some(Path::new("/var/lib/liminal"))
294        );
295
296        Ok(())
297    }
298
299    #[test]
300    fn cluster_overrides_replace_existing_cluster_values() -> Result<(), Box<dyn std::error::Error>>
301    {
302        let config = sample_config()?;
303        let config = apply_env_overrides_from(
304            config,
305            vec![
306                env_pair("LIMINAL_CLUSTER_NODE_NAME", "node-b"),
307                env_pair(
308                    "LIMINAL_CLUSTER_SEED_NODES",
309                    "127.0.0.1:9100, 127.0.0.1:9200",
310                ),
311            ],
312        )?;
313
314        let Some(cluster) = config.cluster else {
315            return Err("cluster config should remain present".into());
316        };
317        assert_eq!(cluster.node_name, "node-b");
318        assert_eq!(cluster.seed_nodes.len(), 2);
319        assert_eq!(cluster.seed_nodes[0], socket("127.0.0.1:9100")?);
320        assert_eq!(cluster.seed_nodes[1], socket("127.0.0.1:9200")?);
321
322        Ok(())
323    }
324
325    #[test]
326    fn cluster_listen_address_and_cookie_overrides_replace_values()
327    -> Result<(), Box<dyn std::error::Error>> {
328        let config = sample_config()?;
329        let config = apply_env_overrides_from(
330            config,
331            vec![
332                env_pair("LIMINAL_CLUSTER_LISTEN_ADDRESS", "127.0.0.1:9500"),
333                env_pair("LIMINAL_CLUSTER_COOKIE", "override-cookie"),
334            ],
335        )?;
336
337        let Some(cluster) = config.cluster else {
338            return Err("cluster config should remain present".into());
339        };
340        assert_eq!(cluster.listen_address, socket("127.0.0.1:9500")?);
341        assert_eq!(cluster.cookie, "override-cookie");
342
343        Ok(())
344    }
345
346    #[test]
347    fn cluster_listen_address_override_without_cluster_section_returns_validation_error()
348    -> Result<(), Box<dyn std::error::Error>> {
349        let mut config = sample_config()?;
350        config.cluster = None;
351        let result = apply_env_overrides_from(
352            config,
353            vec![env_pair("LIMINAL_CLUSTER_LISTEN_ADDRESS", "127.0.0.1:9500")],
354        );
355
356        assert!(matches!(result, Err(ServerError::ConfigValidation { .. })));
357
358        Ok(())
359    }
360
361    #[test]
362    fn auth_token_override_replaces_existing_token() -> Result<(), Box<dyn std::error::Error>> {
363        let mut config = sample_config()?;
364        config.auth = Some(crate::config::types::AuthConfig {
365            token: "file-token".to_owned(),
366        });
367
368        let config =
369            apply_env_overrides_from(config, vec![env_pair("LIMINAL_AUTH_TOKEN", "env-token")])?;
370
371        let auth = config.auth.ok_or("auth section should remain present")?;
372        assert_eq!(auth.token, "env-token");
373
374        Ok(())
375    }
376
377    #[test]
378    fn auth_token_override_creates_missing_auth_section() -> Result<(), Box<dyn std::error::Error>>
379    {
380        let mut config = sample_config()?;
381        config.auth = None;
382
383        let config =
384            apply_env_overrides_from(config, vec![env_pair("LIMINAL_AUTH_TOKEN", "env-token")])?;
385
386        // Unlike cluster overrides, a scalar auth secret MAY fabricate the section.
387        let auth = config.auth.ok_or("auth section should have been created")?;
388        assert_eq!(auth.token, "env-token");
389
390        Ok(())
391    }
392
393    #[test]
394    fn absent_environment_variables_leave_config_unchanged()
395    -> Result<(), Box<dyn std::error::Error>> {
396        let config = sample_config()?;
397        let original_address = config.listen_address;
398        let original_health_address = config.health_listen_address;
399        let original_drain_timeout_ms = config.drain_timeout_ms;
400        let original_path = config.persistence_path.clone();
401        let original_cluster_name = config
402            .cluster
403            .as_ref()
404            .map(|cluster| cluster.node_name.clone());
405
406        let config = apply_env_overrides_from(config, Vec::new())?;
407
408        assert_eq!(config.listen_address, original_address);
409        assert_eq!(config.health_listen_address, original_health_address);
410        assert_eq!(config.drain_timeout_ms, original_drain_timeout_ms);
411        assert_eq!(config.persistence_path, original_path);
412        assert_eq!(
413            config
414                .cluster
415                .as_ref()
416                .map(|cluster| cluster.node_name.clone()),
417            original_cluster_name
418        );
419
420        Ok(())
421    }
422
423    #[test]
424    fn invalid_listen_address_override_returns_config_load()
425    -> Result<(), Box<dyn std::error::Error>> {
426        let config = sample_config()?;
427        let result = apply_env_overrides_from(
428            config,
429            vec![env_pair("LIMINAL_LISTEN_ADDRESS", "not-a-socket")],
430        );
431
432        assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
433
434        Ok(())
435    }
436
437    #[test]
438    fn invalid_health_listen_address_override_returns_config_load()
439    -> Result<(), Box<dyn std::error::Error>> {
440        let config = sample_config()?;
441        let result = apply_env_overrides_from(
442            config,
443            vec![env_pair("LIMINAL_HEALTH_LISTEN_ADDRESS", "not-a-socket")],
444        );
445
446        assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
447
448        Ok(())
449    }
450
451    #[test]
452    fn invalid_drain_timeout_override_returns_config_load() -> Result<(), Box<dyn std::error::Error>>
453    {
454        let config = sample_config()?;
455        let result = apply_env_overrides_from(
456            config,
457            vec![env_pair("LIMINAL_DRAIN_TIMEOUT_MS", "not-a-number")],
458        );
459
460        assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
461
462        Ok(())
463    }
464
465    #[test]
466    fn cluster_override_without_cluster_section_returns_validation_error()
467    -> Result<(), Box<dyn std::error::Error>> {
468        let mut config = sample_config()?;
469        config.cluster = None;
470        let result = apply_env_overrides_from(
471            config,
472            vec![env_pair("LIMINAL_CLUSTER_NODE_NAME", "node-b")],
473        );
474
475        assert!(matches!(result, Err(ServerError::ConfigValidation { .. })));
476
477        Ok(())
478    }
479
480    #[test]
481    fn file_then_env_then_validate_pipeline_gives_env_precedence()
482    -> Result<(), Box<dyn std::error::Error>> {
483        let toml = r#"
484listen_address = "127.0.0.1:8080"
485health_listen_address = "127.0.0.1:8081"
486drain_timeout_ms = 30000
487persistence_path = "/tmp"
488
489[[channels]]
490name = "orders"
491durable = true
492
493[[routing_rules]]
494source_channel = "orders"
495target_channel = "orders"
496"#;
497        let path = write_temp_config(toml)?;
498        let config = load_from_file(&path)?;
499        let mut config = apply_env_overrides_from(
500            config,
501            vec![env_pair("LIMINAL_LISTEN_ADDRESS", "0.0.0.0:9090")],
502        )?;
503        validate(&mut config, path.parent())?;
504        remove_temp_file(&path)?;
505
506        assert_eq!(config.listen_address, socket("0.0.0.0:9090")?);
507
508        Ok(())
509    }
510}
liminal_server/config/env.rs

liminal_server/config/
env.rs
