From ae202c92d03a4db4598910be95ed1b360688c368 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 26 Oct 2026 22:00:17 -0400 Subject: [PATCH 1/9] feat(nats): resolve NATS credentials from the environment so secrets stay out of walgit.toml Signed-off-by: Yordis Prieto --- Cargo.lock | 6 + Cargo.toml | 1 + crates/walgit-nats/Cargo.toml | 13 ++ crates/walgit-nats/src/auth.rs | 220 ++++++++++++++++++++++++++++++ crates/walgit-nats/src/lib.rs | 26 ++++ crates/walgit-nats/src/testing.rs | 109 +++++++++++++++ crates/walgit-nats/tests/auth.rs | 125 +++++++++++++++++ 7 files changed, 500 insertions(+) create mode 100644 crates/walgit-nats/src/auth.rs create mode 100644 crates/walgit-nats/src/testing.rs create mode 100644 crates/walgit-nats/tests/auth.rs diff --git a/Cargo.lock b/Cargo.lock index 9e174d2..d6aa3b8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6004,10 +6004,16 @@ dependencies = [ "anyhow", "async-nats", "bytes", + "nkeys", "serde_json", + "tempfile", "thiserror", + "tokio", + "tracing", + "walgit-config", "walgit-events", "walgit-git", + "walgit-nats", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index b3543f3..ca3f248 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -140,6 +140,7 @@ parking_lot = "0.12" moka = { version = "0.12", features = ["sync"] } mimalloc = "0.1" tempfile = "3" +nkeys = "0.4" flate2 = "1" async-compression = { version = "0.4", features = ["tokio", "gzip"] } pin-project-lite = "0.2" diff --git a/crates/walgit-nats/Cargo.toml b/crates/walgit-nats/Cargo.toml index 322dbf6..a69f686 100644 --- a/crates/walgit-nats/Cargo.toml +++ b/crates/walgit-nats/Cargo.toml @@ -6,12 +6,25 @@ license.workspace = true [dependencies] walgit-events.workspace = true +walgit-config.workspace = true walgit-git.workspace = true async-nats.workspace = true anyhow.workspace = true thiserror.workspace = true bytes.workspace = true +tokio.workspace = true +tracing.workspace = true serde_json.workspace = true +tempfile = { workspace = true, optional = true } + +[features] +# `testing::start_nats`: a throwaway `nats-server -js` for integration tests. +testing = ["dep:tempfile"] + +[dev-dependencies] +walgit-nats = { path = ".", features = ["testing"] } +nkeys.workspace = true +tempfile.workspace = true [lints] workspace = true diff --git a/crates/walgit-nats/src/auth.rs b/crates/walgit-nats/src/auth.rs new file mode 100644 index 0000000..3397f75 --- /dev/null +++ b/crates/walgit-nats/src/auth.rs @@ -0,0 +1,220 @@ +//! How `walgit nats` authenticates to NATS, resolved from the standard `NATS_*` environment +//! variables so secrets never sit in `walgit.toml`. + +use std::fmt; +use std::path::PathBuf; + +use anyhow::Context; + +pub const ENV_NATS_CREDS: &str = "NATS_CREDS"; +pub const ENV_NATS_NKEY: &str = "NATS_NKEY"; +pub const ENV_NATS_USER: &str = "NATS_USER"; +pub const ENV_NATS_PASSWORD: &str = "NATS_PASSWORD"; +pub const ENV_NATS_TOKEN: &str = "NATS_TOKEN"; + +/// NATS authentication method. +/// +/// When resolved from environment variables, priority order is: +/// 1. Credentials file (`NATS_CREDS`) +/// 2. `NKey` seed (`NATS_NKEY`) +/// 3. User/Password (`NATS_USER` + `NATS_PASSWORD`) +/// 4. Token (`NATS_TOKEN`) +/// 5. No auth +#[derive(Clone)] +pub enum NatsAuth { + Credentials(PathBuf), + NKey(String), + UserPassword { user: String, password: String }, + Token(String), + None, +} + +/// `NKey`, `password`, and `Token` are bearer credentials and must not be logged. +impl fmt::Debug for NatsAuth { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Credentials(path) => f.debug_tuple("Credentials").field(path).finish(), + Self::NKey(_) => f.debug_tuple("NKey").field(&"").finish(), + Self::UserPassword { user, .. } => f + .debug_struct("UserPassword") + .field("user", user) + .field("password", &"") + .finish(), + Self::Token(_) => f.debug_tuple("Token").field(&"").finish(), + Self::None => f.write_str("None"), + } + } +} + +impl NatsAuth { + pub fn description(&self) -> &'static str { + match self { + Self::Credentials(_) => "credentials file", + Self::NKey(_) => "NKey", + Self::UserPassword { .. } => "user/password", + Self::Token(_) => "token", + Self::None => "none", + } + } + + /// Resolve from the process environment. + pub fn from_process_env() -> Self { + Self::from_env(|name| std::env::var(name).ok()) + } + + /// Resolve through `var`, in the priority order of [`NatsAuth`]. + pub fn from_env(var: impl Fn(&str) -> Option) -> Self { + if let Some(path) = var(ENV_NATS_CREDS) { + return Self::Credentials(PathBuf::from(path)); + } + if let Some(seed) = var(ENV_NATS_NKEY) { + return Self::NKey(seed); + } + if let (Some(user), Some(password)) = (var(ENV_NATS_USER), var(ENV_NATS_PASSWORD)) { + return Self::UserPassword { user, password }; + } + if let Some(token) = var(ENV_NATS_TOKEN) { + return Self::Token(token); + } + Self::None + } + + pub(crate) async fn connect_options(&self) -> anyhow::Result { + Ok(match self { + Self::Credentials(path) => { + async_nats::ConnectOptions::with_credentials_file(path.clone()) + .await + .with_context(|| format!("{ENV_NATS_CREDS}: {}", path.display()))? + } + Self::NKey(seed) => async_nats::ConnectOptions::with_nkey(seed.clone()), + Self::UserPassword { user, password } => { + async_nats::ConnectOptions::with_user_and_password(user.clone(), password.clone()) + } + Self::Token(token) => async_nats::ConnectOptions::with_token(token.clone()), + Self::None => async_nats::ConnectOptions::new(), + }) + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::*; + + fn resolve(vars: &[(&str, &str)]) -> NatsAuth { + let vars: HashMap = vars + .iter() + .map(|(k, v)| ((*k).to_owned(), (*v).to_owned())) + .collect(); + NatsAuth::from_env(|name| vars.get(name).cloned()) + } + + #[test] + fn no_variables_mean_no_auth() { + assert!(matches!(resolve(&[]), NatsAuth::None)); + } + + #[test] + fn credentials_take_priority() { + let auth = resolve(&[ + ("NATS_CREDS", "/path/to/creds"), + ("NATS_NKEY", "some-nkey"), + ("NATS_TOKEN", "some-token"), + ]); + assert!( + matches!(auth, NatsAuth::Credentials(p) if p == std::path::Path::new("/path/to/creds")) + ); + } + + #[test] + fn nkey_over_user_password_and_token() { + let auth = resolve(&[ + ("NATS_NKEY", "my-nkey"), + ("NATS_USER", "user"), + ("NATS_PASSWORD", "pass"), + ("NATS_TOKEN", "tok"), + ]); + assert!(matches!(auth, NatsAuth::NKey(k) if k == "my-nkey")); + } + + #[test] + fn user_password_over_token() { + let auth = resolve(&[ + ("NATS_USER", "user"), + ("NATS_PASSWORD", "pass"), + ("NATS_TOKEN", "tok"), + ]); + assert!(matches!( + auth, + NatsAuth::UserPassword { user, password } if user == "user" && password == "pass" + )); + } + + #[test] + fn token_when_nothing_else() { + assert!( + matches!(resolve(&[("NATS_TOKEN", "my-token")]), NatsAuth::Token(t) if t == "my-token") + ); + } + + #[test] + fn user_without_password_is_no_auth() { + assert!(matches!(resolve(&[("NATS_USER", "user")]), NatsAuth::None)); + assert!(matches!( + resolve(&[("NATS_PASSWORD", "pass")]), + NatsAuth::None + )); + } + + #[test] + fn description_matches_variant() { + assert_eq!( + NatsAuth::Credentials("/a".into()).description(), + "credentials file" + ); + assert_eq!(NatsAuth::NKey("k".into()).description(), "NKey"); + assert_eq!( + NatsAuth::UserPassword { + user: "u".into(), + password: "p".into() + } + .description(), + "user/password" + ); + assert_eq!(NatsAuth::Token("t".into()).description(), "token"); + assert_eq!(NatsAuth::None.description(), "none"); + } + + #[test] + fn debug_redacts_credentials() { + let nkey = format!("{:?}", NatsAuth::NKey("super-secret-nkey".into())); + assert!(!nkey.contains("super-secret-nkey")); + assert!(nkey.contains("")); + + let user_password = format!( + "{:?}", + NatsAuth::UserPassword { + user: "alice".into(), + password: "super-secret-password".into(), + } + ); + assert!(user_password.contains("alice")); + assert!(!user_password.contains("super-secret-password")); + assert!(user_password.contains("")); + + let token = format!("{:?}", NatsAuth::Token("super-secret-token".into())); + assert!(!token.contains("super-secret-token")); + assert!(token.contains("")); + } + + #[test] + fn debug_shows_non_secret_variants() { + let creds = format!( + "{:?}", + NatsAuth::Credentials(PathBuf::from("/etc/nats/app.creds")) + ); + assert!(creds.contains("/etc/nats/app.creds")); + assert_eq!(format!("{:?}", NatsAuth::None), "None"); + } +} diff --git a/crates/walgit-nats/src/lib.rs b/crates/walgit-nats/src/lib.rs index 7c3b318..49c7027 100644 --- a/crates/walgit-nats/src/lib.rs +++ b/crates/walgit-nats/src/lib.rs @@ -1,4 +1,30 @@ //! `walgit nats`: the WAL's `ref` events (`docs/EVENTS.md`) published into NATS `JetStream`. +pub mod auth; pub mod message; pub mod subject; +#[cfg(feature = "testing")] +pub mod testing; + +use std::str::FromStr; + +use anyhow::Context; +use walgit_config::NatsEventsConfig; + +use crate::auth::NatsAuth; + +/// Connect to `[events.nats] urls` with `auth`. Only the method is logged, never the secret. +pub async fn connect(cfg: &NatsEventsConfig, auth: &NatsAuth) -> anyhow::Result { + let addrs = cfg + .urls + .iter() + .map(|u| async_nats::ServerAddr::from_str(u).with_context(|| format!("events.nats.urls: {u}"))) + .collect::>>()?; + tracing::info!(urls = ?cfg.urls, auth = auth.description(), "connecting to NATS"); + auth.connect_options() + .await? + .name("walgit") + .connect(addrs) + .await + .with_context(|| format!("connecting to NATS (auth: {})", auth.description())) +} diff --git a/crates/walgit-nats/src/testing.rs b/crates/walgit-nats/src/testing.rs new file mode 100644 index 0000000..9d03945 --- /dev/null +++ b/crates/walgit-nats/src/testing.rs @@ -0,0 +1,109 @@ +//! A throwaway `nats-server -js` for tests (feature `testing`). Shared by this crate's tests and +//! `walgit-server`'s, so neither copies the launcher. + +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +pub struct NatsServer { + child: Child, + port: u16, + args: Vec, + pub url: String, + store: tempfile::TempDir, +} + +impl NatsServer { + fn spawn(port: u16, store: &std::path::Path, args: &[String]) -> Child { + #[allow(clippy::expect_used)] + Command::new("nats-server") + .args(["-js", "-a", "127.0.0.1", "-p", &port.to_string(), "-sd"]) + .arg(store) + .args(args) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn nats-server") + } + + /// Kill the server at once (no drain), leaving its store dir for [`NatsServer::restart`]. + pub fn stop(&mut self) { + self.child.kill().ok(); + self.child.wait().ok(); + } + + /// Start the server again on the same port and store dir: streams survive, clients reconnect. + /// + /// # Panics + /// When the server does not accept connections within 15 s. + pub async fn restart(&mut self) { + self.stop(); + self.child = Self::spawn(self.port, self.store.path(), &self.args); + wait_ready(self.port).await; + } +} + +async fn wait_ready(port: u16) { + let t0 = Instant::now(); + while tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .is_err() + { + assert!( + t0.elapsed() < Duration::from_secs(15), + "nats-server did not start" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +impl Drop for NatsServer { + fn drop(&mut self) { + self.child.kill().ok(); + self.child.wait().ok(); + } +} + +/// Start `nats-server -js` on a free loopback port with a temp store dir. `None` (after saying so +/// on stderr) when the binary is not on PATH, so tests skip instead of failing. +/// +/// # Panics +/// When the server cannot be spawned or does not accept connections within 15 s. +#[allow(clippy::expect_used)] +pub async fn start_nats() -> Option { + start_nats_with(&[]).await +} + +/// [`start_nats`] with extra `nats-server` arguments (`--auth`, `--user`/`--pass`, ...). +/// +/// # Panics +/// When the server cannot be spawned or does not accept connections within 15 s. +#[allow(clippy::expect_used)] +pub async fn start_nats_with(args: &[&str]) -> Option { + if !Command::new("nats-server") + .arg("--version") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|s| s.success()) + { + eprintln!("skipping: nats-server is not available on PATH"); + return None; + } + let port = std::net::TcpListener::bind("127.0.0.1:0") + .expect("free port") + .local_addr() + .expect("local addr") + .port(); + let store = tempfile::tempdir().expect("tempdir"); + let args: Vec = args.iter().map(|a| (*a).to_owned()).collect(); + let child = NatsServer::spawn(port, store.path(), &args); + let server = NatsServer { + child, + port, + args, + url: format!("nats://127.0.0.1:{port}"), + store, + }; + wait_ready(port).await; + Some(server) +} diff --git a/crates/walgit-nats/tests/auth.rs b/crates/walgit-nats/tests/auth.rs new file mode 100644 index 0000000..a2d5a74 --- /dev/null +++ b/crates/walgit-nats/tests/auth.rs @@ -0,0 +1,125 @@ +//! `walgit nats` authenticates with whatever the `NATS_*` variables resolve to, against a real +//! `nats-server` that requires it. + +use std::time::Duration; + +use walgit_config::NatsEventsConfig; +use walgit_nats::auth::NatsAuth; +use walgit_nats::testing::{NatsServer, start_nats_with}; + +fn cfg(nats: &NatsServer) -> NatsEventsConfig { + NatsEventsConfig { + urls: vec![nats.url.clone()], + ..NatsEventsConfig::default() + } +} + +fn env(vars: &[(&str, &str)]) -> NatsAuth { + let vars: Vec<(String, String)> = vars + .iter() + .map(|(k, v)| ((*k).to_owned(), (*v).to_owned())) + .collect(); + NatsAuth::from_env(|name| vars.iter().find(|(k, _)| k == name).map(|(_, v)| v.clone())) +} + +async fn connects(nats: &NatsServer, auth: &NatsAuth) -> bool { + let attempt = tokio::time::timeout(Duration::from_secs(5), async { + let client = walgit_nats::connect(&cfg(nats), auth).await?; + client.flush().await?; + anyhow::Ok(()) + }); + matches!(attempt.await, Ok(Ok(()))) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_token_from_the_environment_authenticates() { + let Some(nats) = start_nats_with(&["--auth", "s3cret-token"]).await else { + return; + }; + assert!(connects(&nats, &env(&[("NATS_TOKEN", "s3cret-token")])).await); + assert!(!connects(&nats, &env(&[("NATS_TOKEN", "wrong")])).await); + assert!(!connects(&nats, &env(&[])).await); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_user_and_password_from_the_environment_authenticate() { + let Some(nats) = start_nats_with(&["--user", "walgit", "--pass", "s3cret-pass"]).await else { + return; + }; + assert!( + connects( + &nats, + &env(&[("NATS_USER", "walgit"), ("NATS_PASSWORD", "s3cret-pass")]) + ) + .await + ); + assert!( + !connects( + &nats, + &env(&[("NATS_USER", "walgit"), ("NATS_PASSWORD", "wrong")]) + ) + .await + ); + assert!( + !connects(&nats, &env(&[("NATS_USER", "walgit")])).await, + "a user alone is no auth" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_nkey_seed_from_the_environment_authenticates() { + let user = nkeys::KeyPair::new_user(); + let other = nkeys::KeyPair::new_user(); + let dir = tempfile::tempdir().unwrap(); + let conf = dir.path().join("nats.conf"); + std::fs::write( + &conf, + format!( + "authorization {{ users = [ {{ nkey: {} }} ] }}\n", + user.public_key() + ), + ) + .unwrap(); + let Some(nats) = start_nats_with(&["-c", conf.to_str().unwrap()]).await else { + return; + }; + assert!(connects(&nats, &env(&[("NATS_NKEY", &user.seed().unwrap())])).await); + assert!(!connects(&nats, &env(&[("NATS_NKEY", &other.seed().unwrap())])).await); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_credentials_file_wins_and_a_missing_one_fails_naming_the_variable() { + let Some(nats) = start_nats_with(&["--auth", "s3cret-token"]).await else { + return; + }; + let auth = env(&[ + ("NATS_CREDS", "/nonexistent/walgit.creds"), + ("NATS_TOKEN", "s3cret-token"), + ]); + assert!(matches!(auth, NatsAuth::Credentials(_))); + let err = walgit_nats::connect(&cfg(&nats), &auth).await.unwrap_err(); + let msg = format!("{err:#}"); + assert!( + msg.contains("NATS_CREDS") && msg.contains("/nonexistent/walgit.creds"), + "{msg}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_failed_connection_names_the_method_but_never_the_secret() { + let Some(nats) = start_nats_with(&["--auth", "s3cret-token"]).await else { + return; + }; + let auth = env(&[("NATS_TOKEN", "leaked-if-printed")]); + let err = tokio::time::timeout( + Duration::from_secs(5), + walgit_nats::connect(&cfg(&nats), &auth), + ) + .await + .expect("a rejected credential fails fast") + .unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains("auth: token"), "{msg}"); + assert!(!msg.contains("leaked-if-printed"), "{msg}"); + assert!(!format!("{auth:?}").contains("leaked-if-printed")); +} From 9e106d3ab8fc4d48ea3f30734851db8354412059 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 26 Oct 2026 22:25:44 -0400 Subject: [PATCH 2/9] fix(nats): keep a blank NATS variable from shadowing real credentials Signed-off-by: Yordis Prieto --- crates/walgit-nats/src/auth.rs | 18 +++++++++++++++++- walgit.example.toml | 2 +- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/crates/walgit-nats/src/auth.rs b/crates/walgit-nats/src/auth.rs index 3397f75..2e18151 100644 --- a/crates/walgit-nats/src/auth.rs +++ b/crates/walgit-nats/src/auth.rs @@ -62,8 +62,9 @@ impl NatsAuth { Self::from_env(|name| std::env::var(name).ok()) } - /// Resolve through `var`, in the priority order of [`NatsAuth`]. + /// Resolve through `var`, in the priority order of [`NatsAuth`]. An empty variable is unset. pub fn from_env(var: impl Fn(&str) -> Option) -> Self { + let var = |name: &str| var(name).filter(|v| !v.is_empty()); if let Some(path) = var(ENV_NATS_CREDS) { return Self::Credentials(PathBuf::from(path)); } @@ -167,6 +168,21 @@ mod tests { )); } + #[test] + fn empty_variables_are_unset() { + let auth = resolve(&[ + ("NATS_CREDS", ""), + ("NATS_NKEY", ""), + ("NATS_USER", "u"), + ("NATS_PASSWORD", "p"), + ("NATS_TOKEN", "t"), + ]); + assert!(matches!(auth, NatsAuth::UserPassword { .. }), "{auth:?}"); + let auth = resolve(&[("NATS_USER", "u"), ("NATS_PASSWORD", ""), ("NATS_TOKEN", "t")]); + assert!(matches!(auth, NatsAuth::Token(_)), "{auth:?}"); + assert!(matches!(resolve(&[("NATS_TOKEN", "")]), NatsAuth::None)); + } + #[test] fn description_matches_variant() { assert_eq!( diff --git a/walgit.example.toml b/walgit.example.toml index 7f17458..d17449f 100644 --- a/walgit.example.toml +++ b/walgit.example.toml @@ -205,7 +205,7 @@ lock_wait_warn = "1s" # WARN "lock wait" (lock, repo, wait_ms, reque [events.nats] # docs/EVENTS.md: read only by `walgit nats --config walgit.toml` # urls = ["nats://127.0.0.1:4222"] # the client connects to the first reachable server -# auth comes from the environment, first match wins: NATS_CREDS (.creds file), NATS_NKEY (seed), +# auth comes from the environment, first non-empty match wins: NATS_CREDS (.creds file), NATS_NKEY (seed), # NATS_USER + NATS_PASSWORD, NATS_TOKEN; none set = no auth; secrets are never logged # stream = "WALGIT_EVENTS" # the JetStream stream holding the events (subjects ".ref.>"); it is the resume cursor # ensure_stream = true # create the stream when missing (file storage); an existing stream is used as is From 188f53441593e99763b60dab00151cc66ca388eb Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 26 Oct 2026 22:54:44 -0400 Subject: [PATCH 3/9] fix(nats): read the NATS_* variables the way the nats CLI does so one environment drives both Signed-off-by: Yordis Prieto --- crates/walgit-nats/Cargo.toml | 2 +- crates/walgit-nats/src/auth.rs | 236 ---------- crates/walgit-nats/src/env.rs | 437 +++++++++++++++++++ crates/walgit-nats/src/lib.rs | 25 +- crates/walgit-nats/tests/{auth.rs => env.rs} | 65 ++- walgit.example.toml | 6 +- 6 files changed, 509 insertions(+), 262 deletions(-) delete mode 100644 crates/walgit-nats/src/auth.rs create mode 100644 crates/walgit-nats/src/env.rs rename crates/walgit-nats/tests/{auth.rs => env.rs} (62%) diff --git a/crates/walgit-nats/Cargo.toml b/crates/walgit-nats/Cargo.toml index a69f686..67c9e0e 100644 --- a/crates/walgit-nats/Cargo.toml +++ b/crates/walgit-nats/Cargo.toml @@ -9,6 +9,7 @@ walgit-events.workspace = true walgit-config.workspace = true walgit-git.workspace = true async-nats.workspace = true +nkeys.workspace = true anyhow.workspace = true thiserror.workspace = true bytes.workspace = true @@ -23,7 +24,6 @@ testing = ["dep:tempfile"] [dev-dependencies] walgit-nats = { path = ".", features = ["testing"] } -nkeys.workspace = true tempfile.workspace = true [lints] diff --git a/crates/walgit-nats/src/auth.rs b/crates/walgit-nats/src/auth.rs deleted file mode 100644 index 2e18151..0000000 --- a/crates/walgit-nats/src/auth.rs +++ /dev/null @@ -1,236 +0,0 @@ -//! How `walgit nats` authenticates to NATS, resolved from the standard `NATS_*` environment -//! variables so secrets never sit in `walgit.toml`. - -use std::fmt; -use std::path::PathBuf; - -use anyhow::Context; - -pub const ENV_NATS_CREDS: &str = "NATS_CREDS"; -pub const ENV_NATS_NKEY: &str = "NATS_NKEY"; -pub const ENV_NATS_USER: &str = "NATS_USER"; -pub const ENV_NATS_PASSWORD: &str = "NATS_PASSWORD"; -pub const ENV_NATS_TOKEN: &str = "NATS_TOKEN"; - -/// NATS authentication method. -/// -/// When resolved from environment variables, priority order is: -/// 1. Credentials file (`NATS_CREDS`) -/// 2. `NKey` seed (`NATS_NKEY`) -/// 3. User/Password (`NATS_USER` + `NATS_PASSWORD`) -/// 4. Token (`NATS_TOKEN`) -/// 5. No auth -#[derive(Clone)] -pub enum NatsAuth { - Credentials(PathBuf), - NKey(String), - UserPassword { user: String, password: String }, - Token(String), - None, -} - -/// `NKey`, `password`, and `Token` are bearer credentials and must not be logged. -impl fmt::Debug for NatsAuth { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::Credentials(path) => f.debug_tuple("Credentials").field(path).finish(), - Self::NKey(_) => f.debug_tuple("NKey").field(&"").finish(), - Self::UserPassword { user, .. } => f - .debug_struct("UserPassword") - .field("user", user) - .field("password", &"") - .finish(), - Self::Token(_) => f.debug_tuple("Token").field(&"").finish(), - Self::None => f.write_str("None"), - } - } -} - -impl NatsAuth { - pub fn description(&self) -> &'static str { - match self { - Self::Credentials(_) => "credentials file", - Self::NKey(_) => "NKey", - Self::UserPassword { .. } => "user/password", - Self::Token(_) => "token", - Self::None => "none", - } - } - - /// Resolve from the process environment. - pub fn from_process_env() -> Self { - Self::from_env(|name| std::env::var(name).ok()) - } - - /// Resolve through `var`, in the priority order of [`NatsAuth`]. An empty variable is unset. - pub fn from_env(var: impl Fn(&str) -> Option) -> Self { - let var = |name: &str| var(name).filter(|v| !v.is_empty()); - if let Some(path) = var(ENV_NATS_CREDS) { - return Self::Credentials(PathBuf::from(path)); - } - if let Some(seed) = var(ENV_NATS_NKEY) { - return Self::NKey(seed); - } - if let (Some(user), Some(password)) = (var(ENV_NATS_USER), var(ENV_NATS_PASSWORD)) { - return Self::UserPassword { user, password }; - } - if let Some(token) = var(ENV_NATS_TOKEN) { - return Self::Token(token); - } - Self::None - } - - pub(crate) async fn connect_options(&self) -> anyhow::Result { - Ok(match self { - Self::Credentials(path) => { - async_nats::ConnectOptions::with_credentials_file(path.clone()) - .await - .with_context(|| format!("{ENV_NATS_CREDS}: {}", path.display()))? - } - Self::NKey(seed) => async_nats::ConnectOptions::with_nkey(seed.clone()), - Self::UserPassword { user, password } => { - async_nats::ConnectOptions::with_user_and_password(user.clone(), password.clone()) - } - Self::Token(token) => async_nats::ConnectOptions::with_token(token.clone()), - Self::None => async_nats::ConnectOptions::new(), - }) - } -} - -#[cfg(test)] -mod tests { - use std::collections::HashMap; - - use super::*; - - fn resolve(vars: &[(&str, &str)]) -> NatsAuth { - let vars: HashMap = vars - .iter() - .map(|(k, v)| ((*k).to_owned(), (*v).to_owned())) - .collect(); - NatsAuth::from_env(|name| vars.get(name).cloned()) - } - - #[test] - fn no_variables_mean_no_auth() { - assert!(matches!(resolve(&[]), NatsAuth::None)); - } - - #[test] - fn credentials_take_priority() { - let auth = resolve(&[ - ("NATS_CREDS", "/path/to/creds"), - ("NATS_NKEY", "some-nkey"), - ("NATS_TOKEN", "some-token"), - ]); - assert!( - matches!(auth, NatsAuth::Credentials(p) if p == std::path::Path::new("/path/to/creds")) - ); - } - - #[test] - fn nkey_over_user_password_and_token() { - let auth = resolve(&[ - ("NATS_NKEY", "my-nkey"), - ("NATS_USER", "user"), - ("NATS_PASSWORD", "pass"), - ("NATS_TOKEN", "tok"), - ]); - assert!(matches!(auth, NatsAuth::NKey(k) if k == "my-nkey")); - } - - #[test] - fn user_password_over_token() { - let auth = resolve(&[ - ("NATS_USER", "user"), - ("NATS_PASSWORD", "pass"), - ("NATS_TOKEN", "tok"), - ]); - assert!(matches!( - auth, - NatsAuth::UserPassword { user, password } if user == "user" && password == "pass" - )); - } - - #[test] - fn token_when_nothing_else() { - assert!( - matches!(resolve(&[("NATS_TOKEN", "my-token")]), NatsAuth::Token(t) if t == "my-token") - ); - } - - #[test] - fn user_without_password_is_no_auth() { - assert!(matches!(resolve(&[("NATS_USER", "user")]), NatsAuth::None)); - assert!(matches!( - resolve(&[("NATS_PASSWORD", "pass")]), - NatsAuth::None - )); - } - - #[test] - fn empty_variables_are_unset() { - let auth = resolve(&[ - ("NATS_CREDS", ""), - ("NATS_NKEY", ""), - ("NATS_USER", "u"), - ("NATS_PASSWORD", "p"), - ("NATS_TOKEN", "t"), - ]); - assert!(matches!(auth, NatsAuth::UserPassword { .. }), "{auth:?}"); - let auth = resolve(&[("NATS_USER", "u"), ("NATS_PASSWORD", ""), ("NATS_TOKEN", "t")]); - assert!(matches!(auth, NatsAuth::Token(_)), "{auth:?}"); - assert!(matches!(resolve(&[("NATS_TOKEN", "")]), NatsAuth::None)); - } - - #[test] - fn description_matches_variant() { - assert_eq!( - NatsAuth::Credentials("/a".into()).description(), - "credentials file" - ); - assert_eq!(NatsAuth::NKey("k".into()).description(), "NKey"); - assert_eq!( - NatsAuth::UserPassword { - user: "u".into(), - password: "p".into() - } - .description(), - "user/password" - ); - assert_eq!(NatsAuth::Token("t".into()).description(), "token"); - assert_eq!(NatsAuth::None.description(), "none"); - } - - #[test] - fn debug_redacts_credentials() { - let nkey = format!("{:?}", NatsAuth::NKey("super-secret-nkey".into())); - assert!(!nkey.contains("super-secret-nkey")); - assert!(nkey.contains("")); - - let user_password = format!( - "{:?}", - NatsAuth::UserPassword { - user: "alice".into(), - password: "super-secret-password".into(), - } - ); - assert!(user_password.contains("alice")); - assert!(!user_password.contains("super-secret-password")); - assert!(user_password.contains("")); - - let token = format!("{:?}", NatsAuth::Token("super-secret-token".into())); - assert!(!token.contains("super-secret-token")); - assert!(token.contains("")); - } - - #[test] - fn debug_shows_non_secret_variants() { - let creds = format!( - "{:?}", - NatsAuth::Credentials(PathBuf::from("/etc/nats/app.creds")) - ); - assert!(creds.contains("/etc/nats/app.creds")); - assert_eq!(format!("{:?}", NatsAuth::None), "None"); - } -} diff --git a/crates/walgit-nats/src/env.rs b/crates/walgit-nats/src/env.rs new file mode 100644 index 0000000..19f2d41 --- /dev/null +++ b/crates/walgit-nats/src/env.rs @@ -0,0 +1,437 @@ +//! How `walgit nats` reaches NATS, read from the same `NATS_*` environment variables the `nats` CLI +//! reads and combined the way it combines them, so one shell environment drives both and secrets +//! never sit in `walgit.toml`. + +use std::fmt; +use std::path::PathBuf; +use std::sync::Arc; + +use anyhow::Context; + +pub const ENV_NATS_URL: &str = "NATS_URL"; +pub const ENV_NATS_USER: &str = "NATS_USER"; +pub const ENV_NATS_PASSWORD: &str = "NATS_PASSWORD"; +pub const ENV_NATS_TOKEN: &str = "NATS_TOKEN"; +pub const ENV_NATS_CREDS: &str = "NATS_CREDS"; +pub const ENV_NATS_NKEY: &str = "NATS_NKEY"; +pub const ENV_NATS_JWT: &str = "NATS_JWT"; +pub const ENV_NATS_SEED: &str = "NATS_SEED"; +pub const ENV_NATS_CERT: &str = "NATS_CERT"; +pub const ENV_NATS_KEY: &str = "NATS_KEY"; +pub const ENV_NATS_CA: &str = "NATS_CA"; +pub const ENV_NATS_CONTEXT: &str = "NATS_CONTEXT"; +pub const ENV_NATS_SOCKS_PROXY: &str = "NATS_SOCKS_PROXY"; + +/// A value that must never reach a log line. +#[derive(Clone, PartialEq, Eq)] +pub struct Secret(String); + +impl Secret { + pub fn expose(&self) -> &str { + &self.0 + } +} + +impl fmt::Debug for Secret { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("") + } +} + +/// Who the connection authenticates as. At most one applies, picked in the `nats` CLI's order. +#[derive(Clone, Debug)] +pub enum Identity { + UserPassword { user: String, password: Secret }, + Credentials(PathBuf), + NKeySeedFile(PathBuf), + JwtAndSeed { jwt: Secret, seed: Secret }, + BearerJwt(Secret), +} + +impl Identity { + fn description(&self) -> &'static str { + match self { + Self::UserPassword { .. } => "user/password", + Self::Credentials(_) => "credentials file", + Self::NKeySeedFile(_) => "nkey seed file", + Self::JwtAndSeed { .. } => "user JWT and seed", + Self::BearerJwt(_) => "bearer JWT", + } + } +} + +#[derive(Clone, Debug, Default)] +pub struct ClientTls { + pub certificate: Option<(PathBuf, PathBuf)>, + pub ca: Option, +} + +/// Everything the `NATS_*` variables say about the connection. +#[derive(Clone, Debug, Default)] +pub struct NatsEnv { + pub servers: Option>, + pub identity: Option, + pub token: Option, + pub tls: ClientTls, +} + +impl NatsEnv { + pub fn from_process_env() -> anyhow::Result { + Self::from_env(|name| std::env::var(name).ok()) + } + + /// Resolve through `var` as `nats` does: an empty variable is unset, `NATS_USER` without + /// `NATS_PASSWORD` is a token, and the identity is the first of user/password, `NATS_CREDS`, + /// `NATS_NKEY`, `NATS_JWT` with `NATS_SEED`, `NATS_JWT` alone. A token travels with any of them. + /// `NATS_CONTEXT` and `NATS_SOCKS_PROXY` are refused: ignoring them would connect somewhere or + /// somehow other than `nats` does with the same environment. + pub fn from_env(var: impl Fn(&str) -> Option) -> anyhow::Result { + let var = |name: &str| var(name).filter(|v| !v.is_empty()); + for unsupported in [ENV_NATS_CONTEXT, ENV_NATS_SOCKS_PROXY] { + anyhow::ensure!( + var(unsupported).is_none(), + "{unsupported} is set but walgit nats does not support it; unset it and use the other NATS_* variables" + ); + } + let home = var("HOME"); + let location = |name: &str| -> anyhow::Result> { + var(name) + .map(|v| file_location(name, &v, home.as_deref())) + .transpose() + }; + + let servers = var(ENV_NATS_URL).map(|urls| { + urls.split(',') + .map(str::trim) + .filter(|u| !u.is_empty()) + .map(str::to_owned) + .collect::>() + }); + let servers = servers.filter(|s| !s.is_empty()); + + let (user, password) = (var(ENV_NATS_USER), var(ENV_NATS_PASSWORD)); + let (token, user) = match (user, &password) { + (Some(user), None) => (Some(user), None), + (user, _) => (var(ENV_NATS_TOKEN), user), + }; + + let (jwt, seed) = (var(ENV_NATS_JWT), var(ENV_NATS_SEED)); + let identity = if let (Some(user), Some(password)) = (user, password) { + Some(Identity::UserPassword { + user, + password: Secret(password), + }) + } else if let Some(path) = location(ENV_NATS_CREDS)? { + Some(Identity::Credentials(path)) + } else if let Some(path) = location(ENV_NATS_NKEY)? { + Some(Identity::NKeySeedFile(path)) + } else if let Some(jwt) = jwt { + let jwt = Secret(literal(ENV_NATS_JWT, &jwt, home.as_deref())?); + match seed { + Some(seed) => Some(Identity::JwtAndSeed { + jwt, + seed: Secret(literal(ENV_NATS_SEED, &seed, home.as_deref())?), + }), + None => Some(Identity::BearerJwt(jwt)), + } + } else { + None + }; + + let certificate = match (location(ENV_NATS_CERT)?, location(ENV_NATS_KEY)?) { + (Some(cert), Some(key)) => Some((cert, key)), + _ => None, + }; + Ok(Self { + servers, + identity, + token: token.map(Secret), + tls: ClientTls { + certificate, + ca: location(ENV_NATS_CA)?, + }, + }) + } + + /// The methods in use, for logs and errors. Never contains a secret. + pub fn description(&self) -> String { + let mut parts: Vec<&str> = self.identity.iter().map(Identity::description).collect(); + if self.token.is_some() { + parts.push("token"); + } + if self.tls.certificate.is_some() { + parts.push("client certificate"); + } + if parts.is_empty() { + "none".to_owned() + } else { + parts.join(" + ") + } + } + + pub(crate) async fn connect_options(&self) -> anyhow::Result { + let mut opts = async_nats::ConnectOptions::new(); + opts = match &self.identity { + None => opts, + Some(Identity::UserPassword { user, password }) => { + opts.user_and_password(user.clone(), password.expose().to_owned()) + } + Some(Identity::Credentials(path)) => opts + .credentials_file(path) + .await + .with_context(|| format!("{ENV_NATS_CREDS}: {}", path.display()))?, + Some(Identity::NKeySeedFile(path)) => { + let contents = tokio::fs::read_to_string(path) + .await + .with_context(|| format!("{ENV_NATS_NKEY}: {}", path.display()))?; + let seed = seed_in(&contents).with_context(|| { + format!("{ENV_NATS_NKEY}: {} holds no nkey seed", path.display()) + })?; + opts.nkey(seed) + } + Some(Identity::JwtAndSeed { jwt, seed }) => { + let key = Arc::new( + nkeys::KeyPair::from_seed(seed.expose()) + .map_err(|e| anyhow::anyhow!("{ENV_NATS_SEED}: {e}"))?, + ); + opts.jwt(jwt.expose().to_owned(), move |nonce| { + let key = key.clone(); + async move { key.sign(&nonce).map_err(async_nats::AuthError::new) } + }) + } + Some(Identity::BearerJwt(jwt)) => { + opts.jwt(jwt.expose().to_owned(), |_| async { Ok(Vec::new()) }) + } + }; + if let Some(token) = &self.token { + opts = opts.token(token.expose().to_owned()); + } + if let Some((cert, key)) = &self.tls.certificate { + opts = opts.add_client_certificate(cert.clone(), key.clone()); + } + if let Some(ca) = &self.tls.ca { + opts = opts.add_root_certificates(ca.clone()); + } + Ok(opts) + } +} + +/// A URI scheme as Go's `url.Parse` sees one: at least two characters before the first `:`. +fn scheme(value: &str) -> Option<&str> { + let (scheme, _) = value.split_once(':')?; + let mut chars = scheme.chars(); + let valid = chars.next().is_some_and(|c| c.is_ascii_alphabetic()) + && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '-' | '.')); + (valid && scheme.len() >= 2).then_some(scheme) +} + +/// A `nats` LOCATION that names a file: a bare path or `file://`, with `~/` expanded. +fn file_location(name: &str, value: &str, home: Option<&str>) -> anyhow::Result { + let path = match scheme(value) { + None => value, + Some(s) if s.eq_ignore_ascii_case("file") => value.get(s.len() + 3..).unwrap_or_default(), + Some(s) => anyhow::bail!("{name}: the {s}:// scheme is not supported; give a file path"), + }; + Ok(match (path.strip_prefix("~/"), home) { + (Some(rest), Some(home)) => PathBuf::from(home).join(rest), + _ => PathBuf::from(path), + }) +} + +/// A value that is itself the secret, unless it names a file with `file://`. +fn literal(name: &str, value: &str, home: Option<&str>) -> anyhow::Result { + if scheme(value).is_none() { + return Ok(value.to_owned()); + } + let path = file_location(name, value, home)?; + let contents = + std::fs::read_to_string(&path).with_context(|| format!("{name}: {}", path.display()))?; + Ok(contents.trim_end_matches(['\r', '\n']).to_owned()) +} + +/// The seed in an nkey file, bare or inside the decorated block `nsc` writes. +fn seed_in(contents: &str) -> Option { + contents + .lines() + .map(str::trim) + .find(|line| nkeys::KeyPair::from_seed(line).is_ok()) + .map(str::to_owned) +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::path::Path; + + use super::*; + + fn resolve(vars: &[(&str, &str)]) -> anyhow::Result { + let vars: HashMap = vars + .iter() + .map(|(k, v)| ((*k).to_owned(), (*v).to_owned())) + .collect(); + NatsEnv::from_env(|name| vars.get(name).cloned()) + } + + fn identity(vars: &[(&str, &str)]) -> Option { + resolve(vars).unwrap().identity + } + + #[test] + fn no_variables_mean_no_auth_and_the_configured_servers() { + let env = resolve(&[]).unwrap(); + assert!(env.identity.is_none() && env.token.is_none() && env.servers.is_none()); + assert_eq!(env.description(), "none"); + } + + #[test] + fn nats_url_is_a_comma_separated_server_list() { + let env = resolve(&[("NATS_URL", "nats://a:4222, nats://b:4222,")]).unwrap(); + assert_eq!(env.servers.unwrap(), ["nats://a:4222", "nats://b:4222"]); + assert!(resolve(&[("NATS_URL", " , ")]).unwrap().servers.is_none()); + } + + #[test] + fn user_and_password_come_first() { + let id = identity(&[ + ("NATS_USER", "u"), + ("NATS_PASSWORD", "p"), + ("NATS_CREDS", "/c"), + ("NATS_NKEY", "/n"), + ("NATS_JWT", "j"), + ]); + assert!(matches!(id, Some(Identity::UserPassword { user, password }) + if user == "u" && password.expose() == "p")); + } + + #[test] + fn a_user_without_a_password_is_a_token_and_beats_nats_token() { + let env = resolve(&[("NATS_USER", "tok-as-user"), ("NATS_TOKEN", "tok")]).unwrap(); + assert!(env.identity.is_none()); + assert_eq!(env.token.unwrap().expose(), "tok-as-user"); + } + + #[test] + fn a_password_without_a_user_is_ignored() { + let env = resolve(&[("NATS_PASSWORD", "p"), ("NATS_TOKEN", "tok")]).unwrap(); + assert!(env.identity.is_none()); + assert_eq!(env.token.unwrap().expose(), "tok"); + } + + #[test] + fn creds_then_nkey_then_jwt() { + let all = [ + ("NATS_CREDS", "/c"), + ("NATS_NKEY", "/n"), + ("NATS_JWT", "j"), + ("NATS_SEED", "s"), + ]; + assert!(matches!(identity(&all), Some(Identity::Credentials(p)) if p == Path::new("/c"))); + assert!( + matches!(identity(&all[1..]), Some(Identity::NKeySeedFile(p)) if p == Path::new("/n")) + ); + assert!( + matches!(identity(&all[2..]), Some(Identity::JwtAndSeed { jwt, seed }) + if jwt.expose() == "j" && seed.expose() == "s") + ); + assert!(matches!(identity(&all[2..3]), Some(Identity::BearerJwt(j)) if j.expose() == "j")); + assert!(identity(&all[3..]).is_none(), "a seed alone is no identity"); + } + + #[test] + fn a_token_travels_with_the_identity() { + let env = resolve(&[("NATS_CREDS", "/c"), ("NATS_TOKEN", "tok")]).unwrap(); + assert!(matches!(env.identity, Some(Identity::Credentials(_)))); + assert_eq!(env.token.as_ref().unwrap().expose(), "tok"); + assert_eq!(env.description(), "credentials file + token"); + } + + #[test] + fn locations_are_paths_or_file_uris_with_home_expanded() { + let id = identity(&[("NATS_CREDS", "file:///etc/nats/a.creds")]); + assert!( + matches!(id, Some(Identity::Credentials(p)) if p == Path::new("/etc/nats/a.creds")) + ); + let id = identity(&[("NATS_NKEY", "~/keys/u.nk"), ("HOME", "/home/w")]); + assert!( + matches!(id, Some(Identity::NKeySeedFile(p)) if p == Path::new("/home/w/keys/u.nk")) + ); + let err = resolve(&[("NATS_CREDS", "op://vault/item")]).unwrap_err(); + assert!(format!("{err}").contains("NATS_CREDS"), "{err}"); + } + + #[test] + fn jwt_and_seed_may_name_a_file() { + let dir = tempfile::tempdir().unwrap(); + let seed = dir.path().join("seed"); + std::fs::write(&seed, "SUSEED\n").unwrap(); + let uri = format!("file://{}", seed.display()); + let id = identity(&[("NATS_JWT", "j"), ("NATS_SEED", &uri)]); + assert!(matches!(id, Some(Identity::JwtAndSeed { seed, .. }) if seed.expose() == "SUSEED")); + } + + #[test] + fn tls_needs_both_certificate_and_key_and_the_ca_stands_alone() { + let env = resolve(&[("NATS_CERT", "/c.pem"), ("NATS_CA", "/ca.pem")]).unwrap(); + assert!(env.tls.certificate.is_none()); + assert_eq!(env.tls.ca, Some(Path::new("/ca.pem").to_path_buf())); + let env = resolve(&[("NATS_CERT", "/c.pem"), ("NATS_KEY", "/k.pem")]).unwrap(); + assert_eq!( + env.tls.certificate, + Some(("/c.pem".into(), "/k.pem".into())) + ); + } + + #[test] + fn empty_variables_are_unset() { + let env = resolve(&[ + ("NATS_URL", ""), + ("NATS_CREDS", ""), + ("NATS_NKEY", ""), + ("NATS_USER", "u"), + ("NATS_PASSWORD", "p"), + ("NATS_CONTEXT", ""), + ]) + .unwrap(); + assert!(env.servers.is_none()); + assert!(matches!(env.identity, Some(Identity::UserPassword { .. }))); + let env = resolve(&[("NATS_USER", "u"), ("NATS_PASSWORD", "")]).unwrap(); + assert_eq!(env.token.unwrap().expose(), "u"); + } + + #[test] + fn variables_that_would_change_the_route_are_refused() { + for name in ["NATS_CONTEXT", "NATS_SOCKS_PROXY"] { + let err = resolve(&[(name, "x")]).unwrap_err(); + assert!(format!("{err}").contains(name), "{err}"); + } + } + + #[test] + fn the_seed_is_found_bare_or_decorated() { + let seed = nkeys::KeyPair::new_user().seed().unwrap(); + assert_eq!( + seed_in(&format!("{seed}\n")).as_deref(), + Some(seed.as_str()) + ); + let decorated = + format!("-----BEGIN USER NKEY SEED-----\n{seed}\n------END USER NKEY SEED------\n"); + assert_eq!(seed_in(&decorated).as_deref(), Some(seed.as_str())); + assert!(seed_in("not a seed").is_none()); + } + + #[test] + fn debug_never_shows_a_secret() { + let env = resolve(&[ + ("NATS_USER", "alice"), + ("NATS_PASSWORD", "super-secret-password"), + ("NATS_TOKEN", "super-secret-token"), + ]) + .unwrap(); + let shown = format!("{env:?}"); + assert!(shown.contains("alice")); + assert!(!shown.contains("super-secret"), "{shown}"); + let env = resolve(&[("NATS_JWT", "secret-jwt"), ("NATS_SEED", "secret-seed")]).unwrap(); + assert!(!format!("{env:?}").contains("secret-")); + } +} diff --git a/crates/walgit-nats/src/lib.rs b/crates/walgit-nats/src/lib.rs index 49c7027..0b35d63 100644 --- a/crates/walgit-nats/src/lib.rs +++ b/crates/walgit-nats/src/lib.rs @@ -1,6 +1,6 @@ //! `walgit nats`: the WAL's `ref` events (`docs/EVENTS.md`) published into NATS `JetStream`. -pub mod auth; +pub mod env; pub mod message; pub mod subject; #[cfg(feature = "testing")] @@ -11,20 +11,25 @@ use std::str::FromStr; use anyhow::Context; use walgit_config::NatsEventsConfig; -use crate::auth::NatsAuth; +use crate::env::NatsEnv; -/// Connect to `[events.nats] urls` with `auth`. Only the method is logged, never the secret. -pub async fn connect(cfg: &NatsEventsConfig, auth: &NatsAuth) -> anyhow::Result { - let addrs = cfg - .urls +/// Connect to `NATS_URL`, or `[events.nats] urls` when it is unset, as `env` says. Only the +/// methods are logged, never a secret. +pub async fn connect(cfg: &NatsEventsConfig, env: &NatsEnv) -> anyhow::Result { + let (urls, source) = match &env.servers { + Some(urls) => (urls, crate::env::ENV_NATS_URL), + None => (&cfg.urls, "events.nats.urls"), + }; + let addrs = urls .iter() - .map(|u| async_nats::ServerAddr::from_str(u).with_context(|| format!("events.nats.urls: {u}"))) + .map(|u| async_nats::ServerAddr::from_str(u).with_context(|| format!("{source}: {u}"))) .collect::>>()?; - tracing::info!(urls = ?cfg.urls, auth = auth.description(), "connecting to NATS"); - auth.connect_options() + let auth = env.description(); + tracing::info!(?urls, source, %auth, "connecting to NATS"); + env.connect_options() .await? .name("walgit") .connect(addrs) .await - .with_context(|| format!("connecting to NATS (auth: {})", auth.description())) + .with_context(|| format!("connecting to NATS (auth: {auth})")) } diff --git a/crates/walgit-nats/tests/auth.rs b/crates/walgit-nats/tests/env.rs similarity index 62% rename from crates/walgit-nats/tests/auth.rs rename to crates/walgit-nats/tests/env.rs index a2d5a74..2c806e8 100644 --- a/crates/walgit-nats/tests/auth.rs +++ b/crates/walgit-nats/tests/env.rs @@ -1,10 +1,10 @@ -//! `walgit nats` authenticates with whatever the `NATS_*` variables resolve to, against a real +//! `walgit nats` connects with whatever the `NATS_*` variables resolve to, against a real //! `nats-server` that requires it. use std::time::Duration; use walgit_config::NatsEventsConfig; -use walgit_nats::auth::NatsAuth; +use walgit_nats::env::NatsEnv; use walgit_nats::testing::{NatsServer, start_nats_with}; fn cfg(nats: &NatsServer) -> NatsEventsConfig { @@ -14,15 +14,17 @@ fn cfg(nats: &NatsServer) -> NatsEventsConfig { } } -fn env(vars: &[(&str, &str)]) -> NatsAuth { +#[allow(clippy::expect_used)] +fn env(vars: &[(&str, &str)]) -> NatsEnv { let vars: Vec<(String, String)> = vars .iter() .map(|(k, v)| ((*k).to_owned(), (*v).to_owned())) .collect(); - NatsAuth::from_env(|name| vars.iter().find(|(k, _)| k == name).map(|(_, v)| v.clone())) + NatsEnv::from_env(|name| vars.iter().find(|(k, _)| k == name).map(|(_, v)| v.clone())) + .expect("the variables resolve") } -async fn connects(nats: &NatsServer, auth: &NatsAuth) -> bool { +async fn connects(nats: &NatsServer, auth: &NatsEnv) -> bool { let attempt = tokio::time::timeout(Duration::from_secs(5), async { let client = walgit_nats::connect(&cfg(nats), auth).await?; client.flush().await?; @@ -60,14 +62,38 @@ async fn a_user_and_password_from_the_environment_authenticate() { ) .await ); - assert!( - !connects(&nats, &env(&[("NATS_USER", "walgit")])).await, - "a user alone is no auth" - ); + assert!(!connects(&nats, &env(&[("NATS_USER", "walgit")])).await); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_user_without_a_password_is_a_token_as_in_the_nats_cli() { + let Some(nats) = start_nats_with(&["--auth", "s3cret-token"]).await else { + return; + }; + assert!(connects(&nats, &env(&[("NATS_USER", "s3cret-token")])).await); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn nats_url_wins_over_the_configured_servers() { + let Some(nats) = start_nats_with(&[]).await else { + return; + }; + let unreachable = NatsEventsConfig { + urls: vec!["nats://127.0.0.1:1".into()], + ..NatsEventsConfig::default() + }; + let client = tokio::time::timeout( + Duration::from_secs(5), + walgit_nats::connect(&unreachable, &env(&[("NATS_URL", &nats.url)])), + ) + .await + .expect("connects without trying the configured server") + .unwrap(); + client.flush().await.unwrap(); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn an_nkey_seed_from_the_environment_authenticates() { +async fn an_nkey_seed_file_from_the_environment_authenticates() { let user = nkeys::KeyPair::new_user(); let other = nkeys::KeyPair::new_user(); let dir = tempfile::tempdir().unwrap(); @@ -83,8 +109,18 @@ async fn an_nkey_seed_from_the_environment_authenticates() { let Some(nats) = start_nats_with(&["-c", conf.to_str().unwrap()]).await else { return; }; - assert!(connects(&nats, &env(&[("NATS_NKEY", &user.seed().unwrap())])).await); - assert!(!connects(&nats, &env(&[("NATS_NKEY", &other.seed().unwrap())])).await); + let seed_file = |name: &str, key: &nkeys::KeyPair| { + let path = dir.path().join(name); + std::fs::write(&path, format!("{}\n", key.seed().unwrap())).unwrap(); + path.to_str().unwrap().to_owned() + }; + let (user_file, other_file) = (seed_file("user.nk", &user), seed_file("other.nk", &other)); + assert!(connects(&nats, &env(&[("NATS_NKEY", &user_file)])).await); + assert!(!connects(&nats, &env(&[("NATS_NKEY", &other_file)])).await); + assert!( + !connects(&nats, &env(&[("NATS_NKEY", &user.seed().unwrap())])).await, + "NATS_NKEY is a seed file, never the seed itself" + ); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] @@ -96,7 +132,10 @@ async fn the_credentials_file_wins_and_a_missing_one_fails_naming_the_variable() ("NATS_CREDS", "/nonexistent/walgit.creds"), ("NATS_TOKEN", "s3cret-token"), ]); - assert!(matches!(auth, NatsAuth::Credentials(_))); + assert!(matches!( + auth.identity, + Some(walgit_nats::env::Identity::Credentials(_)) + )); let err = walgit_nats::connect(&cfg(&nats), &auth).await.unwrap_err(); let msg = format!("{err:#}"); assert!( diff --git a/walgit.example.toml b/walgit.example.toml index d17449f..6fb0812 100644 --- a/walgit.example.toml +++ b/walgit.example.toml @@ -205,8 +205,10 @@ lock_wait_warn = "1s" # WARN "lock wait" (lock, repo, wait_ms, reque [events.nats] # docs/EVENTS.md: read only by `walgit nats --config walgit.toml` # urls = ["nats://127.0.0.1:4222"] # the client connects to the first reachable server -# auth comes from the environment, first non-empty match wins: NATS_CREDS (.creds file), NATS_NKEY (seed), -# NATS_USER + NATS_PASSWORD, NATS_TOKEN; none set = no auth; secrets are never logged +# the connection reads the `nats` CLI's variables the way it does (empty = unset): NATS_URL replaces urls; +# identity is the first of NATS_USER + NATS_PASSWORD, NATS_CREDS (file), NATS_NKEY (seed file), +# NATS_JWT + NATS_SEED, NATS_JWT; NATS_TOKEN (or NATS_USER without a password) is sent with it; +# NATS_CERT + NATS_KEY, NATS_CA for TLS; NATS_CONTEXT and NATS_SOCKS_PROXY are refused; secrets are never logged # stream = "WALGIT_EVENTS" # the JetStream stream holding the events (subjects ".ref.>"); it is the resume cursor # ensure_stream = true # create the stream when missing (file storage); an existing stream is used as is # duplicate_window = "1h" # dedup window (Nats-Msg-Id) of a stream `walgit nats` creates From 2934afcc3cd3067727082cac98bd2e98393fd4a4 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 26 Oct 2026 23:13:51 -0400 Subject: [PATCH 4/9] fix(nats): honour credentials in server URLs as the nats CLI does and keep them out of logs Signed-off-by: Yordis Prieto --- Cargo.lock | 1 + Cargo.toml | 1 + crates/walgit-nats/Cargo.toml | 1 + crates/walgit-nats/src/env.rs | 203 ++++++++++++++++++++++++++++++-- crates/walgit-nats/src/lib.rs | 20 ++-- crates/walgit-nats/tests/env.rs | 45 +++++++ walgit.example.toml | 1 + 7 files changed, 249 insertions(+), 23 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d6aa3b8..e94d41a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6005,6 +6005,7 @@ dependencies = [ "async-nats", "bytes", "nkeys", + "percent-encoding", "serde_json", "tempfile", "thiserror", diff --git a/Cargo.toml b/Cargo.toml index ca3f248..251eb62 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -141,6 +141,7 @@ moka = { version = "0.12", features = ["sync"] } mimalloc = "0.1" tempfile = "3" nkeys = "0.4" +percent-encoding = "2" flate2 = "1" async-compression = { version = "0.4", features = ["tokio", "gzip"] } pin-project-lite = "0.2" diff --git a/crates/walgit-nats/Cargo.toml b/crates/walgit-nats/Cargo.toml index 67c9e0e..2c3660e 100644 --- a/crates/walgit-nats/Cargo.toml +++ b/crates/walgit-nats/Cargo.toml @@ -10,6 +10,7 @@ walgit-config.workspace = true walgit-git.workspace = true async-nats.workspace = true nkeys.workspace = true +percent-encoding.workspace = true anyhow.workspace = true thiserror.workspace = true bytes.workspace = true diff --git a/crates/walgit-nats/src/env.rs b/crates/walgit-nats/src/env.rs index 19f2d41..ba13ccf 100644 --- a/crates/walgit-nats/src/env.rs +++ b/crates/walgit-nats/src/env.rs @@ -2,8 +2,10 @@ //! reads and combined the way it combines them, so one shell environment drives both and secrets //! never sit in `walgit.toml`. +use std::borrow::Cow; use std::fmt; use std::path::PathBuf; +use std::str::FromStr; use std::sync::Arc; use anyhow::Context; @@ -130,7 +132,7 @@ impl NatsEnv { match seed { Some(seed) => Some(Identity::JwtAndSeed { jwt, - seed: Secret(literal(ENV_NATS_SEED, &seed, home.as_deref())?), + seed: Secret(seed_value(literal(ENV_NATS_SEED, &seed, home.as_deref())?)), }), None => Some(Identity::BearerJwt(jwt)), } @@ -153,10 +155,19 @@ impl NatsEnv { }) } + /// The identity that still applies once a server URL's credentials took the user and password. + fn identity_beside(&self, url: Option<&UrlCredentials>) -> Option<&Identity> { + match (&self.identity, url) { + (Some(Identity::UserPassword { .. }), Some(_)) => None, + (identity, _) => identity.as_ref(), + } + } + /// The methods in use, for logs and errors. Never contains a secret. - pub fn description(&self) -> String { - let mut parts: Vec<&str> = self.identity.iter().map(Identity::description).collect(); - if self.token.is_some() { + pub fn description(&self, url: Option<&UrlCredentials>) -> String { + let mut parts: Vec<&str> = url.iter().map(|c| c.description()).collect(); + parts.extend(self.identity_beside(url).map(Identity::description)); + if url.is_none() && self.token.is_some() { parts.push("token"); } if self.tls.certificate.is_some() { @@ -169,9 +180,13 @@ impl NatsEnv { } } - pub(crate) async fn connect_options(&self) -> anyhow::Result { + /// As `nats` does, credentials in a server URL replace the user, password and token. + pub(crate) async fn connect_options( + &self, + url: Option<&UrlCredentials>, + ) -> anyhow::Result { let mut opts = async_nats::ConnectOptions::new(); - opts = match &self.identity { + opts = match self.identity_beside(url) { None => opts, Some(Identity::UserPassword { user, password }) => { opts.user_and_password(user.clone(), password.expose().to_owned()) @@ -203,9 +218,15 @@ impl NatsEnv { opts.jwt(jwt.expose().to_owned(), |_| async { Ok(Vec::new()) }) } }; - if let Some(token) = &self.token { - opts = opts.token(token.expose().to_owned()); - } + opts = match (url, &self.token) { + (Some(UrlCredentials::UserPassword { user, password }), _) => { + opts.user_and_password(user.clone(), password.expose().to_owned()) + } + (Some(UrlCredentials::Token(token)), _) | (None, Some(token)) => { + opts.token(token.expose().to_owned()) + } + (None, None) => opts, + }; if let Some((cert, key)) = &self.tls.certificate { opts = opts.add_client_certificate(cert.clone(), key.clone()); } @@ -216,6 +237,97 @@ impl NatsEnv { } } +/// Credentials written into a server URL as `user:password@` or `token@`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum UrlCredentials { + UserPassword { user: String, password: Secret }, + Token(Secret), +} + +impl UrlCredentials { + fn description(&self) -> &'static str { + match self { + Self::UserPassword { .. } => "URL user/password", + Self::Token(_) => "URL token", + } + } +} + +/// The servers to dial, with the credentials taken out of their URLs. +#[derive(Debug)] +pub struct Servers { + pub addrs: Vec, + pub credentials: Option, +} + +impl Servers { + /// Parse `urls` from `source`. Servers that carry credentials must carry the same ones, and + /// an error never repeats them. + pub fn parse(source: &str, urls: &[String]) -> anyhow::Result { + let mut credentials: Option = None; + let mut addrs = Vec::with_capacity(urls.len()); + for u in urls { + let (addr, found) = async_nats::ServerAddr::from_str(u) + .map_err(anyhow::Error::from) + .and_then(take_credentials) + .with_context(|| format!("{source}: {}", redacted(u)))?; + if let Some(found) = found { + match &credentials { + Some(seen) => anyhow::ensure!( + *seen == found, + "{source}: the servers carry different credentials" + ), + None => credentials = Some(found), + } + } + addrs.push(addr); + } + Ok(Self { addrs, credentials }) + } + + /// The URLs as dialled, which hold no credentials. + pub fn shown(&self) -> Vec<&str> { + self.addrs + .iter() + .map(async_nats::ServerAddr::as_url_str) + .collect() + } +} + +fn take_credentials( + addr: async_nats::ServerAddr, +) -> anyhow::Result<(async_nats::ServerAddr, Option)> { + let decode = |s: &str| { + percent_encoding::percent_decode_str(s) + .decode_utf8() + .map(Cow::into_owned) + }; + let credentials = match (addr.username(), addr.password()) { + (None, None) => return Ok((addr, None)), + (user, Some(password)) => UrlCredentials::UserPassword { + user: decode(user.unwrap_or_default())?, + password: Secret(decode(password)?), + }, + (Some(token), None) => UrlCredentials::Token(Secret(decode(token)?)), + }; + let mut url = addr.into_inner(); + url.set_password(None) + .and_then(|()| url.set_username("")) + .map_err(|()| anyhow::anyhow!("the URL cannot carry credentials"))?; + Ok((async_nats::ServerAddr::from_url(url)?, Some(credentials))) +} + +/// `url` with whatever precedes an `@` in its authority hidden, for a URL that did not parse. +fn redacted(url: &str) -> String { + let (scheme, rest) = url.split_once("://").map_or(("", url), |(s, r)| (s, r)); + let (authority, tail) = rest.split_at(rest.find(['/', '?', '#']).unwrap_or(rest.len())); + match authority.rsplit_once('@') { + Some((_, host)) if scheme.is_empty() => format!("@{host}{tail}"), + Some((_, host)) => format!("{scheme}://@{host}{tail}"), + None => url.to_owned(), + } +} + /// A URI scheme as Go's `url.Parse` sees one: at least two characters before the first `:`. fn scheme(value: &str) -> Option<&str> { let (scheme, _) = value.split_once(':')?; @@ -249,6 +361,12 @@ fn literal(name: &str, value: &str, home: Option<&str>) -> anyhow::Result String { + seed_in(&value).unwrap_or(value) +} + /// The seed in an nkey file, bare or inside the decorated block `nsc` writes. fn seed_in(contents: &str) -> Option { contents @@ -281,7 +399,7 @@ mod tests { fn no_variables_mean_no_auth_and_the_configured_servers() { let env = resolve(&[]).unwrap(); assert!(env.identity.is_none() && env.token.is_none() && env.servers.is_none()); - assert_eq!(env.description(), "none"); + assert_eq!(env.description(None), "none"); } #[test] @@ -343,7 +461,7 @@ mod tests { let env = resolve(&[("NATS_CREDS", "/c"), ("NATS_TOKEN", "tok")]).unwrap(); assert!(matches!(env.identity, Some(Identity::Credentials(_)))); assert_eq!(env.token.as_ref().unwrap().expose(), "tok"); - assert_eq!(env.description(), "credentials file + token"); + assert_eq!(env.description(None), "credentials file + token"); } #[test] @@ -370,6 +488,69 @@ mod tests { assert!(matches!(id, Some(Identity::JwtAndSeed { seed, .. }) if seed.expose() == "SUSEED")); } + #[test] + fn a_seed_file_may_hold_the_decorated_block() { + let dir = tempfile::tempdir().unwrap(); + let seed = nkeys::KeyPair::new_user().seed().unwrap(); + let path = dir.path().join("user.nk"); + std::fs::write( + &path, + format!("-----BEGIN USER NKEY SEED-----\n{seed}\n------END USER NKEY SEED------\n"), + ) + .unwrap(); + let uri = format!("file://{}", path.display()); + let id = identity(&[("NATS_JWT", "j"), ("NATS_SEED", &uri)]); + assert!(matches!(id, Some(Identity::JwtAndSeed { seed: s, .. }) if s.expose() == seed)); + } + + fn servers(urls: &[&str]) -> anyhow::Result { + let urls: Vec = urls.iter().map(|u| (*u).to_owned()).collect(); + Servers::parse("NATS_URL", &urls) + } + + #[test] + fn credentials_leave_the_url_and_are_decoded() { + let s = servers(&["nats://al%40ce:p%3Ass@a:4222", "nats://b:4222"]).unwrap(); + assert_eq!(s.shown(), ["nats://a:4222", "nats://b:4222"]); + assert!( + matches!(&s.credentials, Some(UrlCredentials::UserPassword { user, password }) + if user == "al@ce" && password.expose() == "p:ss") + ); + let s = servers(&["nats://tok@a:4222"]).unwrap(); + assert_eq!(s.shown(), ["nats://a:4222"]); + assert!(matches!(&s.credentials, Some(UrlCredentials::Token(t)) if t.expose() == "tok")); + assert!(servers(&["nats://a:4222"]).unwrap().credentials.is_none()); + } + + #[test] + fn servers_must_agree_on_their_credentials() { + let err = servers(&["nats://u:p@a:4222", "nats://u:q@b:4222"]).unwrap_err(); + assert!(!format!("{err:#}").contains(":p@"), "{err:#}"); + assert!(servers(&["nats://u:p@a:4222", "nats://b:4222", "nats://u:p@c:4222"]).is_ok()); + } + + #[test] + fn a_url_that_fails_to_parse_is_shown_without_its_credentials() { + let err = servers(&["nats://u:hidden@[bad"]).unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains("NATS_URL: nats://@[bad"), "{msg}"); + assert!(!msg.contains("hidden"), "{msg}"); + } + + #[test] + fn url_credentials_replace_the_user_password_and_token() { + let url = UrlCredentials::Token(Secret("t".into())); + let env = resolve(&[ + ("NATS_USER", "u"), + ("NATS_PASSWORD", "p"), + ("NATS_TOKEN", "x"), + ]) + .unwrap(); + assert_eq!(env.description(Some(&url)), "URL token"); + let env = resolve(&[("NATS_CREDS", "/c"), ("NATS_TOKEN", "x")]).unwrap(); + assert_eq!(env.description(Some(&url)), "URL token + credentials file"); + } + #[test] fn tls_needs_both_certificate_and_key_and_the_ca_stands_alone() { let env = resolve(&[("NATS_CERT", "/c.pem"), ("NATS_CA", "/ca.pem")]).unwrap(); diff --git a/crates/walgit-nats/src/lib.rs b/crates/walgit-nats/src/lib.rs index 0b35d63..38ada9d 100644 --- a/crates/walgit-nats/src/lib.rs +++ b/crates/walgit-nats/src/lib.rs @@ -6,30 +6,26 @@ pub mod subject; #[cfg(feature = "testing")] pub mod testing; -use std::str::FromStr; - use anyhow::Context; use walgit_config::NatsEventsConfig; -use crate::env::NatsEnv; +use crate::env::{NatsEnv, Servers}; /// Connect to `NATS_URL`, or `[events.nats] urls` when it is unset, as `env` says. Only the -/// methods are logged, never a secret. +/// methods and the URLs without their credentials are logged, never a secret. pub async fn connect(cfg: &NatsEventsConfig, env: &NatsEnv) -> anyhow::Result { let (urls, source) = match &env.servers { Some(urls) => (urls, crate::env::ENV_NATS_URL), None => (&cfg.urls, "events.nats.urls"), }; - let addrs = urls - .iter() - .map(|u| async_nats::ServerAddr::from_str(u).with_context(|| format!("{source}: {u}"))) - .collect::>>()?; - let auth = env.description(); - tracing::info!(?urls, source, %auth, "connecting to NATS"); - env.connect_options() + let servers = Servers::parse(source, urls)?; + let url_credentials = servers.credentials.as_ref(); + let auth = env.description(url_credentials); + tracing::info!(urls = ?servers.shown(), source, %auth, "connecting to NATS"); + env.connect_options(url_credentials) .await? .name("walgit") - .connect(addrs) + .connect(servers.addrs) .await .with_context(|| format!("connecting to NATS (auth: {auth})")) } diff --git a/crates/walgit-nats/tests/env.rs b/crates/walgit-nats/tests/env.rs index 2c806e8..504699a 100644 --- a/crates/walgit-nats/tests/env.rs +++ b/crates/walgit-nats/tests/env.rs @@ -162,3 +162,48 @@ async fn a_failed_connection_names_the_method_but_never_the_secret() { assert!(!msg.contains("leaked-if-printed"), "{msg}"); assert!(!format!("{auth:?}").contains("leaked-if-printed")); } + +fn with_userinfo(nats: &NatsServer, userinfo: &str) -> String { + nats.url.replacen("://", &format!("://{userinfo}@"), 1) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn credentials_in_nats_url_replace_the_user_password_and_token() { + let Some(nats) = start_nats_with(&["--user", "walgit", "--pass", "s3cret-pass"]).await else { + return; + }; + let url = with_userinfo(&nats, "walgit:s3cret-pass"); + let wrong = [("NATS_USER", "walgit"), ("NATS_PASSWORD", "wrong")]; + assert!(connects(&nats, &env(&[("NATS_URL", &url), wrong[0], wrong[1]])).await); + assert!( + !connects( + &nats, + &env(&[("NATS_URL", &with_userinfo(&nats, "walgit:wrong"))]) + ) + .await + ); + + let Some(nats) = start_nats_with(&["--auth", "s3cret-token"]).await else { + return; + }; + let url = with_userinfo(&nats, "s3cret-token"); + assert!(connects(&nats, &env(&[("NATS_URL", &url), ("NATS_TOKEN", "wrong")])).await); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_failed_connection_never_shows_the_credentials_in_the_url() { + let Some(nats) = start_nats_with(&["--auth", "s3cret-token"]).await else { + return; + }; + let auth = env(&[("NATS_URL", &with_userinfo(&nats, "leaked-if-printed"))]); + let err = tokio::time::timeout( + Duration::from_secs(5), + walgit_nats::connect(&cfg(&nats), &auth), + ) + .await + .expect("a rejected credential fails fast") + .unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains("auth: URL token"), "{msg}"); + assert!(!msg.contains("leaked-if-printed"), "{msg}"); +} diff --git a/walgit.example.toml b/walgit.example.toml index 6fb0812..a443b20 100644 --- a/walgit.example.toml +++ b/walgit.example.toml @@ -208,6 +208,7 @@ lock_wait_warn = "1s" # WARN "lock wait" (lock, repo, wait_ms, reque # the connection reads the `nats` CLI's variables the way it does (empty = unset): NATS_URL replaces urls; # identity is the first of NATS_USER + NATS_PASSWORD, NATS_CREDS (file), NATS_NKEY (seed file), # NATS_JWT + NATS_SEED, NATS_JWT; NATS_TOKEN (or NATS_USER without a password) is sent with it; +# user:password@ or token@ in a server URL replaces the user, password and token; # NATS_CERT + NATS_KEY, NATS_CA for TLS; NATS_CONTEXT and NATS_SOCKS_PROXY are refused; secrets are never logged # stream = "WALGIT_EVENTS" # the JetStream stream holding the events (subjects ".ref.>"); it is the resume cursor # ensure_stream = true # create the stream when missing (file storage); an existing stream is used as is From a7974fa9b43efbefd1739e95600e3c1975b222d8 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 26 Oct 2026 23:17:24 -0400 Subject: [PATCH 5/9] fix(nats): refuse a file: location that is not file:// instead of truncating the path Signed-off-by: Yordis Prieto --- crates/walgit-nats/src/env.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/crates/walgit-nats/src/env.rs b/crates/walgit-nats/src/env.rs index ba13ccf..f3edce9 100644 --- a/crates/walgit-nats/src/env.rs +++ b/crates/walgit-nats/src/env.rs @@ -341,7 +341,11 @@ fn scheme(value: &str) -> Option<&str> { fn file_location(name: &str, value: &str, home: Option<&str>) -> anyhow::Result { let path = match scheme(value) { None => value, - Some(s) if s.eq_ignore_ascii_case("file") => value.get(s.len() + 3..).unwrap_or_default(), + Some(s) if s.eq_ignore_ascii_case("file") => value + .get(..7) + .filter(|prefix| prefix.eq_ignore_ascii_case("file://")) + .and_then(|_| value.get(7..)) + .with_context(|| format!("{name}: write a file URI as file://"))?, Some(s) => anyhow::bail!("{name}: the {s}:// scheme is not supported; give a file path"), }; Ok(match (path.strip_prefix("~/"), home) { @@ -476,6 +480,12 @@ mod tests { ); let err = resolve(&[("NATS_CREDS", "op://vault/item")]).unwrap_err(); assert!(format!("{err}").contains("NATS_CREDS"), "{err}"); + for short in ["file:/etc/a.creds", "file:relative"] { + let err = resolve(&[("NATS_CREDS", short)]).unwrap_err(); + assert!(format!("{err}").contains("file://"), "{err}"); + } + let id = identity(&[("NATS_CREDS", "FILE:///etc/a.creds")]); + assert!(matches!(id, Some(Identity::Credentials(p)) if p == Path::new("/etc/a.creds"))); } #[test] From ed72096e624b084c58d22a70107e0daf3620f8b8 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 26 Oct 2026 23:42:54 -0400 Subject: [PATCH 6/9] fix(events): keep credentials in NATS_URL out of debug output Signed-off-by: Yordis Prieto --- crates/walgit-nats/src/env.rs | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/crates/walgit-nats/src/env.rs b/crates/walgit-nats/src/env.rs index f3edce9..facfbd4 100644 --- a/crates/walgit-nats/src/env.rs +++ b/crates/walgit-nats/src/env.rs @@ -69,7 +69,7 @@ pub struct ClientTls { } /// Everything the `NATS_*` variables say about the connection. -#[derive(Clone, Debug, Default)] +#[derive(Clone, Default)] pub struct NatsEnv { pub servers: Option>, pub identity: Option, @@ -77,6 +77,21 @@ pub struct NatsEnv { pub tls: ClientTls, } +impl fmt::Debug for NatsEnv { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let servers = self + .servers + .as_ref() + .map(|urls| urls.iter().map(|url| redacted(url)).collect::>()); + f.debug_struct("NatsEnv") + .field("servers", &servers) + .field("identity", &self.identity) + .field("token", &self.token) + .field("tls", &self.tls) + .finish() + } +} + impl NatsEnv { pub fn from_process_env() -> anyhow::Result { Self::from_env(|name| std::env::var(name).ok()) @@ -624,5 +639,9 @@ mod tests { assert!(!shown.contains("super-secret"), "{shown}"); let env = resolve(&[("NATS_JWT", "secret-jwt"), ("NATS_SEED", "secret-seed")]).unwrap(); assert!(!format!("{env:?}").contains("secret-")); + let env = resolve(&[("NATS_URL", "nats://bob:secret-pass@a:4222,nats://secret-token@b:4222")]).unwrap(); + let shown = format!("{env:?}"); + assert!(shown.contains("a:4222") && shown.contains("b:4222"), "{shown}"); + assert!(!shown.contains("secret-") && !shown.contains("bob"), "{shown}"); } } From e9f047a449ac131c5c8b3a59525457d0e1caf773 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 26 Oct 2026 23:59:04 -0400 Subject: [PATCH 7/9] fix(events): require TLS when NATS_CA or a client certificate is set Signed-off-by: Yordis Prieto --- Cargo.lock | 1 + crates/walgit-nats/Cargo.toml | 1 + crates/walgit-nats/src/env.rs | 3 +++ crates/walgit-nats/tests/env.rs | 43 +++++++++++++++++++++++++++++++++ 4 files changed, 48 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index e94d41a..f2b17ef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6006,6 +6006,7 @@ dependencies = [ "bytes", "nkeys", "percent-encoding", + "rcgen", "serde_json", "tempfile", "thiserror", diff --git a/crates/walgit-nats/Cargo.toml b/crates/walgit-nats/Cargo.toml index 2c3660e..a2583cd 100644 --- a/crates/walgit-nats/Cargo.toml +++ b/crates/walgit-nats/Cargo.toml @@ -26,6 +26,7 @@ testing = ["dep:tempfile"] [dev-dependencies] walgit-nats = { path = ".", features = ["testing"] } tempfile.workspace = true +rcgen.workspace = true [lints] workspace = true diff --git a/crates/walgit-nats/src/env.rs b/crates/walgit-nats/src/env.rs index facfbd4..ab6f420 100644 --- a/crates/walgit-nats/src/env.rs +++ b/crates/walgit-nats/src/env.rs @@ -248,6 +248,9 @@ impl NatsEnv { if let Some(ca) = &self.tls.ca { opts = opts.add_root_certificates(ca.clone()); } + if self.tls.certificate.is_some() || self.tls.ca.is_some() { + opts = opts.require_tls(true); + } Ok(opts) } } diff --git a/crates/walgit-nats/tests/env.rs b/crates/walgit-nats/tests/env.rs index 504699a..187afc4 100644 --- a/crates/walgit-nats/tests/env.rs +++ b/crates/walgit-nats/tests/env.rs @@ -207,3 +207,46 @@ async fn a_failed_connection_never_shows_the_credentials_in_the_url() { assert!(msg.contains("auth: URL token"), "{msg}"); assert!(!msg.contains("leaked-if-printed"), "{msg}"); } + +/// A CA and a server certificate for 127.0.0.1 it signed, as PEM files in `dir`. +#[allow(clippy::expect_used)] +fn tls_files(dir: &std::path::Path) -> (String, String, String) { + let ca_key = rcgen::KeyPair::generate().expect("ca key"); + let mut ca_params = rcgen::CertificateParams::new(Vec::::new()).expect("ca params"); + ca_params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained); + let ca = ca_params.self_signed(&ca_key).expect("ca"); + let issuer = rcgen::Issuer::new(ca_params, ca_key); + let key = rcgen::KeyPair::generate().expect("server key"); + let cert = rcgen::CertificateParams::new(vec!["127.0.0.1".to_owned()]) + .expect("server params") + .signed_by(&key, &issuer) + .expect("server certificate"); + let write = |name: &str, pem: String| { + let path = dir.join(name); + std::fs::write(&path, pem).expect("write pem"); + path.to_string_lossy().into_owned() + }; + ( + write("ca.pem", ca.pem()), + write("server.pem", cert.pem()), + write("server-key.pem", key.serialize_pem()), + ) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_ca_requires_tls_as_in_the_nats_cli() { + let dir = tempfile::tempdir().unwrap(); + let (ca, cert, key) = tls_files(dir.path()); + let Some(plain) = start_nats_with(&[]).await else { + return; + }; + assert!(connects(&plain, &env(&[])).await); + assert!( + !connects(&plain, &env(&[("NATS_CA", &ca)])).await, + "a plaintext server is refused, never used without TLS" + ); + let Some(tls) = start_nats_with(&["--tls", "--tlscert", &cert, "--tlskey", &key]).await else { + return; + }; + assert!(connects(&tls, &env(&[("NATS_CA", &ca)])).await, "the CA verifies the server"); +} From 5b3e732fe0ceb7a79a7384c96ee1767dd710e7da Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 27 Oct 2026 00:09:18 -0400 Subject: [PATCH 8/9] test(events): pick the rustls provider the way the walgit binary does Signed-off-by: Yordis Prieto --- Cargo.lock | 1 + crates/walgit-nats/Cargo.toml | 1 + crates/walgit-nats/tests/env.rs | 1 + 3 files changed, 3 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index f2b17ef..590465d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6007,6 +6007,7 @@ dependencies = [ "nkeys", "percent-encoding", "rcgen", + "rustls 0.23.45", "serde_json", "tempfile", "thiserror", diff --git a/crates/walgit-nats/Cargo.toml b/crates/walgit-nats/Cargo.toml index a2583cd..a7d2841 100644 --- a/crates/walgit-nats/Cargo.toml +++ b/crates/walgit-nats/Cargo.toml @@ -27,6 +27,7 @@ testing = ["dep:tempfile"] walgit-nats = { path = ".", features = ["testing"] } tempfile.workspace = true rcgen.workspace = true +rustls.workspace = true [lints] workspace = true diff --git a/crates/walgit-nats/tests/env.rs b/crates/walgit-nats/tests/env.rs index 187afc4..3f14ff4 100644 --- a/crates/walgit-nats/tests/env.rs +++ b/crates/walgit-nats/tests/env.rs @@ -235,6 +235,7 @@ fn tls_files(dir: &std::path::Path) -> (String, String, String) { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_ca_requires_tls_as_in_the_nats_cli() { + let _ = rustls::crypto::ring::default_provider().install_default(); let dir = tempfile::tempdir().unwrap(); let (ca, cert, key) = tls_files(dir.path()); let Some(plain) = start_nats_with(&[]).await else { From abe048b157360a7dc3692a888342b16ffd792655 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 27 Oct 2026 00:24:01 -0400 Subject: [PATCH 9/9] fix(events): refuse a lone NATS_CERT or NATS_KEY and hide passwords that hold URL delimiters Signed-off-by: Yordis Prieto --- crates/walgit-nats/src/env.rs | 27 ++++++++++++++++++++------- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/crates/walgit-nats/src/env.rs b/crates/walgit-nats/src/env.rs index ab6f420..8f74ddb 100644 --- a/crates/walgit-nats/src/env.rs +++ b/crates/walgit-nats/src/env.rs @@ -157,7 +157,8 @@ impl NatsEnv { let certificate = match (location(ENV_NATS_CERT)?, location(ENV_NATS_KEY)?) { (Some(cert), Some(key)) => Some((cert, key)), - _ => None, + (None, None) => None, + _ => anyhow::bail!("{ENV_NATS_CERT} and {ENV_NATS_KEY} must be set together"), }; Ok(Self { servers, @@ -335,13 +336,13 @@ fn take_credentials( Ok((async_nats::ServerAddr::from_url(url)?, Some(credentials))) } -/// `url` with whatever precedes an `@` in its authority hidden, for a URL that did not parse. +/// `url` with whatever precedes its last `@` hidden. A password may hold an unencoded `/`, `?` +/// or `#`, so the authority's end cannot be trusted to bound it. fn redacted(url: &str) -> String { let (scheme, rest) = url.split_once("://").map_or(("", url), |(s, r)| (s, r)); - let (authority, tail) = rest.split_at(rest.find(['/', '?', '#']).unwrap_or(rest.len())); - match authority.rsplit_once('@') { - Some((_, host)) if scheme.is_empty() => format!("@{host}{tail}"), - Some((_, host)) => format!("{scheme}://@{host}{tail}"), + match rest.rsplit_once('@') { + Some((_, host)) if scheme.is_empty() => format!("@{host}"), + Some((_, host)) => format!("{scheme}://@{host}"), None => url.to_owned(), } } @@ -563,6 +564,14 @@ mod tests { let msg = format!("{err:#}"); assert!(msg.contains("NATS_URL: nats://@[bad"), "{msg}"); assert!(!msg.contains("hidden"), "{msg}"); + for url in ["nats://u:hid#den@h:4222", "nats://u:hid/den@h:4222", "nats://u:hid?den@h:4222"] { + let err = servers(&[url]).unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains("nats://@h:4222"), "{msg}"); + assert!(!msg.contains("hid"), "{msg}"); + let env = resolve(&[("NATS_URL", url)]).unwrap(); + assert!(!format!("{env:?}").contains("hid"), "{env:?}"); + } } #[test] @@ -581,7 +590,11 @@ mod tests { #[test] fn tls_needs_both_certificate_and_key_and_the_ca_stands_alone() { - let env = resolve(&[("NATS_CERT", "/c.pem"), ("NATS_CA", "/ca.pem")]).unwrap(); + for half in [("NATS_CERT", "/c.pem"), ("NATS_KEY", "/k.pem")] { + let err = resolve(&[half, ("NATS_CA", "/ca.pem")]).unwrap_err(); + assert!(format!("{err:#}").contains("must be set together"), "{err:#}"); + } + let env = resolve(&[("NATS_CA", "/ca.pem")]).unwrap(); assert!(env.tls.certificate.is_none()); assert_eq!(env.tls.ca, Some(Path::new("/ca.pem").to_path_buf())); let env = resolve(&[("NATS_CERT", "/c.pem"), ("NATS_KEY", "/k.pem")]).unwrap();