diff --git a/Cargo.lock b/Cargo.lock index 9e174d2..590465d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6004,10 +6004,19 @@ dependencies = [ "anyhow", "async-nats", "bytes", + "nkeys", + "percent-encoding", + "rcgen", + "rustls 0.23.45", "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..251eb62 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -140,6 +140,8 @@ parking_lot = "0.12" 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 322dbf6..a7d2841 100644 --- a/crates/walgit-nats/Cargo.toml +++ b/crates/walgit-nats/Cargo.toml @@ -6,12 +6,28 @@ license.workspace = true [dependencies] walgit-events.workspace = true +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 +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"] } +tempfile.workspace = true +rcgen.workspace = true +rustls.workspace = true [lints] workspace = true diff --git a/crates/walgit-nats/src/env.rs b/crates/walgit-nats/src/env.rs new file mode 100644 index 0000000..8f74ddb --- /dev/null +++ b/crates/walgit-nats/src/env.rs @@ -0,0 +1,663 @@ +//! 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::borrow::Cow; +use std::fmt; +use std::path::PathBuf; +use std::str::FromStr; +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, Default)] +pub struct NatsEnv { + pub servers: Option>, + pub identity: Option, + pub token: Option, + 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()) + } + + /// 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(seed_value(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, None) => None, + _ => anyhow::bail!("{ENV_NATS_CERT} and {ENV_NATS_KEY} must be set together"), + }; + Ok(Self { + servers, + identity, + token: token.map(Secret), + tls: ClientTls { + certificate, + ca: location(ENV_NATS_CA)?, + }, + }) + } + + /// 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, 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() { + parts.push("client certificate"); + } + if parts.is_empty() { + "none".to_owned() + } else { + parts.join(" + ") + } + } + + /// 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_beside(url) { + 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()) }) + } + }; + 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()); + } + 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) + } +} + +/// 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 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)); + match rest.rsplit_once('@') { + Some((_, host)) if scheme.is_empty() => format!("@{host}"), + Some((_, host)) => format!("{scheme}://@{host}"), + 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(':')?; + 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(..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) { + (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()) +} + +/// A seed given bare or in the decorated block `nsc` writes; anything else is left for the key +/// parser to reject. +fn seed_value(value: String) -> 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 + .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), "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(None), "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}"); + 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] + 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 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}"); + 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] + 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() { + 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(); + 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-")); + 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}"); + } +} diff --git a/crates/walgit-nats/src/lib.rs b/crates/walgit-nats/src/lib.rs index 7c3b318..38ada9d 100644 --- a/crates/walgit-nats/src/lib.rs +++ b/crates/walgit-nats/src/lib.rs @@ -1,4 +1,31 @@ //! `walgit nats`: the WAL's `ref` events (`docs/EVENTS.md`) published into NATS `JetStream`. +pub mod env; pub mod message; pub mod subject; +#[cfg(feature = "testing")] +pub mod testing; + +use anyhow::Context; +use walgit_config::NatsEventsConfig; + +use crate::env::{NatsEnv, Servers}; + +/// Connect to `NATS_URL`, or `[events.nats] urls` when it is unset, as `env` says. Only the +/// 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 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(servers.addrs) + .await + .with_context(|| format!("connecting to NATS (auth: {auth})")) +} 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/env.rs b/crates/walgit-nats/tests/env.rs new file mode 100644 index 0000000..3f14ff4 --- /dev/null +++ b/crates/walgit-nats/tests/env.rs @@ -0,0 +1,253 @@ +//! `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::env::NatsEnv; +use walgit_nats::testing::{NatsServer, start_nats_with}; + +fn cfg(nats: &NatsServer) -> NatsEventsConfig { + NatsEventsConfig { + urls: vec![nats.url.clone()], + ..NatsEventsConfig::default() + } +} + +#[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(); + NatsEnv::from_env(|name| vars.iter().find(|(k, _)| k == name).map(|(_, v)| v.clone())) + .expect("the variables resolve") +} + +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?; + 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); +} + +#[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_file_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; + }; + 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)] +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.identity, + Some(walgit_nats::env::Identity::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")); +} + +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}"); +} + +/// 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 _ = 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 { + 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"); +} diff --git a/walgit.example.toml b/walgit.example.toml index 7f17458..a443b20 100644 --- a/walgit.example.toml +++ b/walgit.example.toml @@ -205,8 +205,11 @@ 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), -# 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; +# 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 # duplicate_window = "1h" # dedup window (Nats-Msg-Id) of a stream `walgit nats` creates