cufflink_nats/
lib.rs

1//! NATS connections shared by the platform, worker and web-runner: the
2//! `NATS_URL` server list, the component's credentials and the connect
3//! options every component uses.
4
5use async_nats::{Client, ConnectOptions, ServerAddr};
6use std::fmt;
7use std::path::PathBuf;
8use std::time::Duration;
9
10const CONNECTION_TIMEOUT: Duration = Duration::from_secs(2);
11const PING_INTERVAL: Duration = Duration::from_secs(10);
12
13/// The options every component connects with: its credentials and name, a
14/// first connect that happens in the background (so a NATS outage never
15/// stops a process from starting), pings that notice a dead server in
16/// ~20s, and reconnects that never give up.
17pub async fn connect_options(client_name: &str, auth: &NatsAuth) -> eyre::Result<ConnectOptions> {
18    Ok(auth
19        .apply(ConnectOptions::new())
20        .await?
21        .name(client_name)
22        .retry_on_initial_connect()
23        .connection_timeout(CONNECTION_TIMEOUT)
24        .ping_interval(PING_INTERVAL)
25        .max_reconnects(None::<usize>))
26}
27
28/// A client for `raw_urls` with [`connect_options`]. Only a malformed
29/// `NATS_URL` (or unreadable credentials) is an error; an unreachable server
30/// is not.
31pub async fn connect(raw_urls: &str, client_name: &str, auth: &NatsAuth) -> eyre::Result<Client> {
32    let servers = parse_server_list(raw_urls)?;
33    Ok(connect_options(client_name, auth)
34        .await?
35        .connect(servers)
36        .await?)
37}
38
39/// Parse `NATS_URL`: a single server or a comma-separated list.
40pub fn parse_server_list(raw: &str) -> eyre::Result<Vec<ServerAddr>> {
41    let servers = raw
42        .split(',')
43        .map(str::trim)
44        .filter(|s| !s.is_empty())
45        .map(|s| {
46            s.parse::<ServerAddr>()
47                .map_err(|e| eyre::eyre!("invalid NATS server URL '{}': {}", redact_urls(s), e))
48        })
49        .collect::<eyre::Result<Vec<_>>>()?;
50    if servers.is_empty() {
51        eyre::bail!("NATS_URL lists no servers");
52    }
53    Ok(servers)
54}
55
56/// `NATS_URL` with any `user:password@` credentials replaced, safe to log.
57pub fn redact_urls(raw: &str) -> String {
58    raw.split(',')
59        .map(|entry| {
60            let trimmed = entry.trim();
61            let (scheme, rest) = match trimmed.find("://") {
62                Some(i) => trimmed.split_at(i + 3),
63                None => ("", trimmed),
64            };
65            let authority_end = rest.find('/').unwrap_or(rest.len());
66            match rest[..authority_end].rfind('@') {
67                Some(at) => format!("{}***@{}", scheme, &rest[at + 1..]),
68                None => trimmed.to_string(),
69            }
70        })
71        .collect::<Vec<_>>()
72        .join(",")
73}
74
75/// How a component authenticates to NATS: `NATS_CREDS_FILE` (a `.creds`
76/// file), `NATS_NKEY_SEED`, or `NATS_USER` with `NATS_PASSWORD`; at most one.
77/// Without any the connection is anonymous, as before.
78#[derive(Clone, Default, PartialEq, Eq)]
79pub enum NatsAuth {
80    #[default]
81    Anonymous,
82    CredentialsFile(PathBuf),
83    NKeySeed(String),
84    UserPassword {
85        user: String,
86        password: String,
87    },
88}
89
90impl fmt::Debug for NatsAuth {
91    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
92        match self {
93            NatsAuth::Anonymous => f.write_str("Anonymous"),
94            NatsAuth::CredentialsFile(path) => {
95                f.debug_tuple("CredentialsFile").field(path).finish()
96            }
97            NatsAuth::NKeySeed(_) => f.write_str("NKeySeed(***)"),
98            NatsAuth::UserPassword { user, .. } => f
99                .debug_struct("UserPassword")
100                .field("user", user)
101                .field("password", &"***")
102                .finish(),
103        }
104    }
105}
106
107impl NatsAuth {
108    pub fn from_env() -> eyre::Result<Self> {
109        Self::from_vars(|name| std::env::var(name).ok().filter(|v| !v.is_empty()))
110    }
111
112    pub fn from_vars(lookup: impl Fn(&str) -> Option<String>) -> eyre::Result<Self> {
113        let creds = lookup("NATS_CREDS_FILE").map(|path| NatsAuth::CredentialsFile(path.into()));
114        let nkey = lookup("NATS_NKEY_SEED").map(NatsAuth::NKeySeed);
115        let user = match (lookup("NATS_USER"), lookup("NATS_PASSWORD")) {
116            (Some(user), Some(password)) => Some(NatsAuth::UserPassword { user, password }),
117            (None, None) => None,
118            _ => eyre::bail!("NATS_USER and NATS_PASSWORD must be set together"),
119        };
120        let mut configured = [creds, nkey, user].into_iter().flatten();
121        match (configured.next(), configured.next()) {
122            (None, _) => Ok(NatsAuth::Anonymous),
123            (Some(auth), None) => Ok(auth),
124            (Some(_), Some(_)) => eyre::bail!(
125                "set only one of NATS_CREDS_FILE, NATS_NKEY_SEED or NATS_USER/NATS_PASSWORD"
126            ),
127        }
128    }
129
130    /// `options` carrying these credentials.
131    pub async fn apply(&self, options: ConnectOptions) -> eyre::Result<ConnectOptions> {
132        Ok(match self {
133            NatsAuth::Anonymous => options,
134            NatsAuth::CredentialsFile(path) => {
135                options.credentials_file(path).await.map_err(|e| {
136                    eyre::eyre!("cannot read NATS credentials file {}: {e}", path.display())
137                })?
138            }
139            NatsAuth::NKeySeed(seed) => options.nkey(seed.clone()),
140            NatsAuth::UserPassword { user, password } => {
141                options.user_and_password(user.clone(), password.clone())
142            }
143        })
144    }
145}
146
147#[cfg(test)]
148mod tests {
149    use super::*;
150    use std::collections::HashMap;
151
152    fn vars(pairs: &[(&str, &str)]) -> impl Fn(&str) -> Option<String> {
153        let map: HashMap<String, String> = pairs
154            .iter()
155            .map(|(k, v)| (k.to_string(), v.to_string()))
156            .collect();
157        move |name| map.get(name).cloned()
158    }
159
160    #[test]
161    fn single_server_url_parses() {
162        assert_eq!(parse_server_list("nats://nats:4222").unwrap().len(), 1);
163    }
164
165    #[test]
166    fn comma_separated_servers_parse_with_whitespace_and_blanks() {
167        assert_eq!(
168            parse_server_list(" nats://a:4222 , ,nats://b:4222,")
169                .unwrap()
170                .len(),
171            2
172        );
173    }
174
175    #[test]
176    fn empty_server_list_is_an_error() {
177        assert!(parse_server_list("").is_err());
178        assert!(parse_server_list(" , ,").is_err());
179    }
180
181    #[test]
182    fn redacts_credentials_only() {
183        assert_eq!(redact_urls("nats://nats:4222"), "nats://nats:4222");
184        assert_eq!(
185            redact_urls("nats://user:s3cret@nats-0:4222, tls://tok@nats-1:4222"),
186            "nats://***@nats-0:4222,tls://***@nats-1:4222"
187        );
188        assert_eq!(redact_urls("user:pw@host:4222"), "***@host:4222");
189    }
190
191    #[test]
192    fn auth_is_anonymous_unless_configured() {
193        assert_eq!(NatsAuth::from_vars(vars(&[])).unwrap(), NatsAuth::Anonymous);
194        assert_eq!(
195            NatsAuth::from_vars(vars(&[("NATS_CREDS_FILE", "/etc/nats/platform.creds")])).unwrap(),
196            NatsAuth::CredentialsFile("/etc/nats/platform.creds".into())
197        );
198        assert_eq!(
199            NatsAuth::from_vars(vars(&[("NATS_USER", "platform"), ("NATS_PASSWORD", "pw")]))
200                .unwrap(),
201            NatsAuth::UserPassword {
202                user: "platform".into(),
203                password: "pw".into()
204            }
205        );
206    }
207
208    #[test]
209    fn auth_rejects_half_or_several_methods() {
210        assert!(NatsAuth::from_vars(vars(&[("NATS_USER", "platform")])).is_err());
211        assert!(NatsAuth::from_vars(vars(&[
212            ("NATS_NKEY_SEED", "SUA"),
213            ("NATS_CREDS_FILE", "/x.creds")
214        ]))
215        .is_err());
216    }
217
218    #[test]
219    fn debug_never_prints_secrets() {
220        let rendered = format!(
221            "{:?} {:?}",
222            NatsAuth::NKeySeed("SUASECRET".into()),
223            NatsAuth::UserPassword {
224                user: "u".into(),
225                password: "hunter2".into()
226            }
227        );
228        assert!(
229            !rendered.contains("SUASECRET") && !rendered.contains("hunter2"),
230            "{rendered}"
231        );
232    }
233
234    #[tokio::test]
235    async fn missing_credentials_file_is_an_error() {
236        let err = NatsAuth::CredentialsFile("/nonexistent/cufflink.creds".into())
237            .apply(ConnectOptions::new())
238            .await
239            .unwrap_err();
240        assert!(
241            err.to_string().contains("/nonexistent/cufflink.creds"),
242            "{err}"
243        );
244    }
245}
cufflink_nats/lib.rs

cufflink_nats/
lib.rs
