vector/
nats.rs

1//! Shared helper functions for NATS source and sink.
2#![allow(missing_docs)]
3
4use nkeys::error::Error as NKeysError;
5use snafu::{ResultExt, Snafu};
6use vector_lib::{configurable::configurable_component, sensitive_string::SensitiveString};
7
8use crate::tls::TlsEnableableConfig;
9
10/// Errors that can occur during NATS configuration.
11#[derive(Debug, Snafu)]
12pub enum NatsConfigError {
13    #[snafu(display("NATS Auth Config Error: {}", source))]
14    AuthConfigError { source: NKeysError },
15    #[snafu(display("NATS TLS Config Error: missing key"))]
16    TlsMissingKey,
17    #[snafu(display("NATS TLS Config Error: missing cert"))]
18    TlsMissingCert,
19    #[snafu(display("NATS Credentials file error"))]
20    CredentialsFileError { source: std::io::Error },
21}
22
23/// Configuration of the authentication strategy when interacting with NATS.
24#[configurable_component]
25#[derive(Clone, Debug)]
26#[serde(rename_all = "snake_case", tag = "strategy")]
27#[configurable(metadata(
28    docs::enum_tag_description = "The strategy used to authenticate with the NATS server.
29
30More information on NATS authentication, and the various authentication strategies, can be found in the
31NATS [documentation][nats_auth_docs]. For TLS client certificate authentication specifically, see the
32`tls` settings.
33
34[nats_auth_docs]: https://docs.nats.io/running-a-nats-service/configuration/securing_nats/auth_intro"
35))]
36pub enum NatsAuthConfig {
37    /// Username/password authentication.
38    UserPassword {
39        #[configurable(derived)]
40        user_password: NatsAuthUserPassword,
41    },
42
43    /// Token authentication.
44    Token {
45        #[configurable(derived)]
46        token: NatsAuthToken,
47    },
48
49    /// Credentials file authentication. (JWT-based)
50    CredentialsFile {
51        #[configurable(derived)]
52        credentials_file: NatsAuthCredentialsFile,
53    },
54
55    /// NKey authentication.
56    Nkey {
57        #[configurable(derived)]
58        nkey: NatsAuthNKey,
59    },
60}
61
62impl std::fmt::Display for NatsAuthConfig {
63    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
64        use NatsAuthConfig::*;
65        let word = match self {
66            UserPassword { .. } => "user_password",
67            Token { .. } => "token",
68            CredentialsFile { .. } => "credentials_file",
69            Nkey { .. } => "nkey",
70        };
71        write!(f, "{word}")
72    }
73}
74
75/// Username and password configuration.
76#[configurable_component]
77#[derive(Clone, Debug)]
78#[serde(deny_unknown_fields)]
79pub struct NatsAuthUserPassword {
80    /// Username.
81    pub(crate) user: String,
82
83    /// Password.
84    pub(crate) password: SensitiveString,
85}
86
87/// Token configuration.
88#[configurable_component]
89#[derive(Clone, Debug)]
90#[serde(deny_unknown_fields)]
91pub struct NatsAuthToken {
92    /// Token.
93    pub(crate) value: SensitiveString,
94}
95
96/// Credentials file configuration.
97#[configurable_component]
98#[derive(Clone, Debug)]
99#[serde(deny_unknown_fields)]
100pub struct NatsAuthCredentialsFile {
101    /// Path to credentials file.
102    #[configurable(metadata(docs::examples = "/etc/nats/nats.creds"))]
103    pub(crate) path: String,
104}
105
106/// NKeys configuration.
107#[configurable_component]
108#[derive(Clone, Debug)]
109#[serde(deny_unknown_fields)]
110pub struct NatsAuthNKey {
111    /// User.
112    ///
113    /// Conceptually, this is equivalent to a public key.
114    pub(crate) nkey: String,
115
116    /// Seed.
117    ///
118    /// Conceptually, this is equivalent to a private key.
119    pub(crate) seed: String,
120}
121
122impl NatsAuthConfig {
123    pub(crate) fn to_nats_options(&self) -> Result<async_nats::ConnectOptions, NatsConfigError> {
124        match self {
125            NatsAuthConfig::UserPassword { user_password } => {
126                Ok(async_nats::ConnectOptions::with_user_and_password(
127                    user_password.user.clone(),
128                    user_password.password.inner().to_string(),
129                ))
130            }
131            NatsAuthConfig::CredentialsFile { credentials_file } => {
132                async_nats::ConnectOptions::with_credentials(
133                    &std::fs::read_to_string(credentials_file.path.clone())
134                        .context(CredentialsFileSnafu)?,
135                )
136                .context(CredentialsFileSnafu)
137            }
138            NatsAuthConfig::Nkey { nkey } => {
139                Ok(async_nats::ConnectOptions::with_nkey(nkey.seed.clone()))
140            }
141            NatsAuthConfig::Token { token } => Ok(async_nats::ConnectOptions::with_token(
142                token.value.inner().to_string(),
143            )),
144        }
145    }
146}
147
148/// Validate the NATS TLS cert/key pairing without touching the filesystem.
149///
150/// Mirrors the pairing check in `from_tls_auth_config`.
151pub(crate) fn validate_tls_cert_key_pair(
152    tls_config: &TlsEnableableConfig,
153) -> Result<(), NatsConfigError> {
154    if !tls_config.enabled.unwrap_or(false) {
155        return Ok(());
156    }
157    match (&tls_config.options.crt_file, &tls_config.options.key_file) {
158        (Some(_), None) => Err(NatsConfigError::TlsMissingKey),
159        (None, Some(_)) => Err(NatsConfigError::TlsMissingCert),
160        _ => Ok(()),
161    }
162}
163
164pub(crate) fn from_tls_auth_config(
165    connection_name: &str,
166    auth_config: &Option<NatsAuthConfig>,
167    tls_config: &Option<TlsEnableableConfig>,
168) -> Result<async_nats::ConnectOptions, NatsConfigError> {
169    let nats_options = match &auth_config {
170        None => async_nats::ConnectOptions::new(),
171        Some(auth) => auth.to_nats_options()?,
172    };
173
174    let nats_options = nats_options.name(connection_name);
175
176    match tls_config {
177        None => Ok(nats_options),
178        Some(tls_config) => {
179            let tls_enabled = tls_config.enabled.unwrap_or(false);
180            let nats_options = nats_options.require_tls(tls_enabled);
181            if !tls_enabled {
182                return Ok(nats_options);
183            }
184
185            validate_tls_cert_key_pair(tls_config)?;
186
187            let nats_options = match &tls_config.options.ca_file {
188                None => nats_options,
189                Some(ca_file) => nats_options.add_root_certificates(ca_file.clone()),
190            };
191
192            let nats_options = match (&tls_config.options.crt_file, &tls_config.options.key_file) {
193                (None, None) => nats_options,
194                (Some(crt_file), Some(key_file)) => {
195                    nats_options.add_client_certificate(crt_file.clone(), key_file.clone())
196                }
197                (Some(_), None) | (None, Some(_)) => {
198                    unreachable!("cert/key pairing validated by validate_tls_cert_key_pair")
199                }
200            };
201            Ok(nats_options)
202        }
203    }
204}
205
206#[cfg(test)]
207mod tests {
208    use indoc::indoc;
209
210    use super::*;
211
212    fn parse_auth(s: &str) -> Result<async_nats::ConnectOptions, crate::Error> {
213        serde_yaml::from_str(s)
214            .map_err(Into::into)
215            .and_then(|config: NatsAuthConfig| config.to_nats_options().map_err(Into::into))
216    }
217
218    #[test]
219    fn auth_user_password_ok() {
220        parse_auth(indoc! {r#"
221            strategy: user_password
222            user_password:
223              user: username
224              password: password
225        "#})
226        .unwrap();
227    }
228
229    #[test]
230    fn auth_user_password_missing_user() {
231        parse_auth(indoc! {r#"
232            strategy: user_password
233            user_password:
234              password: password
235        "#})
236        .unwrap_err();
237    }
238
239    #[test]
240    fn auth_user_password_missing_password() {
241        parse_auth(indoc! {r#"
242            strategy: user_password
243            user_password:
244              user: username
245        "#})
246        .unwrap_err();
247    }
248
249    #[test]
250    fn auth_user_password_missing_all() {
251        parse_auth(indoc! {r#"
252            strategy: user_password
253            token:
254              value: foobar
255        "#})
256        .unwrap_err();
257    }
258
259    #[test]
260    fn auth_token_ok() {
261        parse_auth(indoc! {r#"
262            strategy: token
263            token:
264              value: token
265        "#})
266        .unwrap();
267    }
268
269    #[test]
270    fn auth_token_missing() {
271        parse_auth(indoc! {r#"
272            strategy: token
273            user_password:
274              user: foobar
275        "#})
276        .unwrap_err();
277    }
278
279    #[test]
280    fn auth_credentials_file_ok() {
281        parse_auth(indoc! {r#"
282            strategy: credentials_file
283            credentials_file:
284              path: tests/integration/nats/data/nats.creds
285        "#})
286        .unwrap();
287    }
288
289    #[test]
290    fn auth_credentials_file_missing() {
291        parse_auth(indoc! {r#"
292            strategy: credentials_file
293            token:
294              value: foobar
295        "#})
296        .unwrap_err();
297    }
298
299    #[test]
300    fn auth_nkey_ok() {
301        parse_auth(indoc! {r#"
302            strategy: nkey
303            nkey:
304              nkey: UC435ZYS52HF72E2VMQF4GO6CUJOCHDUUPEBU7XDXW5AQLIC6JZ46PO5
305              seed: SUAAEZYNLTEA2MDTG7L5X7QODZXYHPOI2LT2KH5I4GD6YVP24SE766EGPA
306        "#})
307        .unwrap();
308    }
309
310    #[test]
311    fn auth_nkey_missing_nkey() {
312        parse_auth(indoc! {r#"
313            strategy: nkey
314            nkey:
315              seed: SUAAEZYNLTEA2MDTG7L5X7QODZXYHPOI2LT2KH5I4GD6YVP24SE766EGPA
316        "#})
317        .unwrap_err();
318    }
319
320    #[test]
321    fn auth_nkey_missing_seed() {
322        parse_auth(indoc! {r#"
323            strategy: nkey
324            nkey:
325              nkey: UC435ZYS52HF72E2VMQF4GO6CUJOCHDUUPEBU7XDXW5AQLIC6JZ46PO5
326        "#})
327        .unwrap_err();
328    }
329
330    #[test]
331    fn auth_nkey_missing_both() {
332        parse_auth(indoc! {r#"
333            strategy: nkey
334            user_password:
335              user: foobar
336        "#})
337        .unwrap_err();
338    }
339}
vector/nats.rs

vector/
nats.rs
