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())
110    }
111
112    /// The credentials the variables `vars` give; an empty one counts as
113    /// unset.
114    pub fn from_vars(vars: impl Fn(&str) -> Option<String>) -> eyre::Result<Self> {
115        let lookup = |name: &str| vars(name).filter(|value| !value.is_empty());
116        let creds = lookup("NATS_CREDS_FILE").map(|path| NatsAuth::CredentialsFile(path.into()));
117        let nkey = lookup("NATS_NKEY_SEED").map(NatsAuth::NKeySeed);
118        let user = match (lookup("NATS_USER"), lookup("NATS_PASSWORD")) {
119            (Some(user), Some(password)) => Some(NatsAuth::UserPassword { user, password }),
120            (None, None) => None,
121            _ => eyre::bail!("NATS_USER and NATS_PASSWORD must be set together"),
122        };
123        let mut configured = [creds, nkey, user].into_iter().flatten();
124        match (configured.next(), configured.next()) {
125            (None, _) => Ok(NatsAuth::Anonymous),
126            (Some(auth), None) => Ok(auth),
127            (Some(_), Some(_)) => eyre::bail!(
128                "set only one of NATS_CREDS_FILE, NATS_NKEY_SEED or NATS_USER/NATS_PASSWORD"
129            ),
130        }
131    }
132
133    /// `options` carrying these credentials.
134    pub async fn apply(&self, options: ConnectOptions) -> eyre::Result<ConnectOptions> {
135        Ok(match self {
136            NatsAuth::Anonymous => options,
137            NatsAuth::CredentialsFile(path) => {
138                options.credentials_file(path).await.map_err(|e| {
139                    eyre::eyre!("cannot read NATS credentials file {}: {e}", path.display())
140                })?
141            }
142            NatsAuth::NKeySeed(seed) => options.nkey(seed.clone()),
143            NatsAuth::UserPassword { user, password } => {
144                options.user_and_password(user.clone(), password.clone())
145            }
146        })
147    }
148}
149
150#[cfg(test)]
151mod tests {
152    use super::*;
153    use std::collections::HashMap;
154
155    fn vars(pairs: &[(&str, &str)]) -> impl Fn(&str) -> Option<String> {
156        let map: HashMap<String, String> = pairs
157            .iter()
158            .map(|(k, v)| (k.to_string(), v.to_string()))
159            .collect();
160        move |name| map.get(name).cloned()
161    }
162
163    #[test]
164    fn single_server_url_parses() {
165        assert_eq!(parse_server_list("nats://nats:4222").unwrap().len(), 1);
166    }
167
168    #[test]
169    fn comma_separated_servers_parse_with_whitespace_and_blanks() {
170        assert_eq!(
171            parse_server_list(" nats://a:4222 , ,nats://b:4222,")
172                .unwrap()
173                .len(),
174            2
175        );
176    }
177
178    #[test]
179    fn empty_server_list_is_an_error() {
180        assert!(parse_server_list("").is_err());
181        assert!(parse_server_list(" , ,").is_err());
182    }
183
184    #[test]
185    fn redacts_credentials_only() {
186        assert_eq!(redact_urls("nats://nats:4222"), "nats://nats:4222");
187        assert_eq!(
188            redact_urls("nats://user:s3cret@nats-0:4222, tls://tok@nats-1:4222"),
189            "nats://***@nats-0:4222,tls://***@nats-1:4222"
190        );
191        assert_eq!(redact_urls("user:pw@host:4222"), "***@host:4222");
192    }
193
194    #[test]
195    fn an_empty_credential_variable_is_unset() {
196        assert_eq!(
197            NatsAuth::from_vars(vars(&[
198                ("NATS_CREDS_FILE", ""),
199                ("NATS_NKEY_SEED", ""),
200                ("NATS_USER", ""),
201                ("NATS_PASSWORD", ""),
202            ]))
203            .unwrap(),
204            NatsAuth::Anonymous
205        );
206        assert_eq!(
207            NatsAuth::from_vars(vars(&[("NATS_NKEY_SEED", "SUAB"), ("NATS_USER", "")])).unwrap(),
208            NatsAuth::NKeySeed("SUAB".into())
209        );
210        assert!(NatsAuth::from_vars(vars(&[("NATS_USER", "app"), ("NATS_PASSWORD", "")])).is_err());
211    }
212
213    #[test]
214    fn auth_is_anonymous_unless_configured() {
215        assert_eq!(NatsAuth::from_vars(vars(&[])).unwrap(), NatsAuth::Anonymous);
216        assert_eq!(
217            NatsAuth::from_vars(vars(&[("NATS_CREDS_FILE", "/etc/nats/platform.creds")])).unwrap(),
218            NatsAuth::CredentialsFile("/etc/nats/platform.creds".into())
219        );
220        assert_eq!(
221            NatsAuth::from_vars(vars(&[("NATS_USER", "platform"), ("NATS_PASSWORD", "pw")]))
222                .unwrap(),
223            NatsAuth::UserPassword {
224                user: "platform".into(),
225                password: "pw".into()
226            }
227        );
228    }
229
230    #[test]
231    fn auth_rejects_half_or_several_methods() {
232        assert!(NatsAuth::from_vars(vars(&[("NATS_USER", "platform")])).is_err());
233        assert!(NatsAuth::from_vars(vars(&[
234            ("NATS_NKEY_SEED", "SUA"),
235            ("NATS_CREDS_FILE", "/x.creds")
236        ]))
237        .is_err());
238    }
239
240    #[test]
241    fn debug_never_prints_secrets() {
242        let rendered = format!(
243            "{:?} {:?}",
244            NatsAuth::NKeySeed("SUASECRET".into()),
245            NatsAuth::UserPassword {
246                user: "u".into(),
247                password: "hunter2".into()
248            }
249        );
250        assert!(
251            !rendered.contains("SUASECRET") && !rendered.contains("hunter2"),
252            "{rendered}"
253        );
254    }
255
256    #[tokio::test]
257    async fn missing_credentials_file_is_an_error() {
258        let err = NatsAuth::CredentialsFile("/nonexistent/cufflink.creds".into())
259            .apply(ConnectOptions::new())
260            .await
261            .unwrap_err();
262        assert!(
263            err.to_string().contains("/nonexistent/cufflink.creds"),
264            "{err}"
265        );
266    }
267}
cufflink_nats/lib.rs

cufflink_nats/
lib.rs
