krabka_backup/
connection.rs

1//! Kafka-compatible `--command-config` security for cluster connections.
2//!
3//! The file is parsed into the client library's [`ClientSecurity`] rather than
4//! reimplementing TLS or SASL here. Its contents can include passwords, so this
5//! module never logs or formats the parsed properties or resulting policy.
6
7use {
8    crate::BackupError,
9    clap::Args,
10    krabka_client_core::security::{
11        ClientSecurity, KeyStore, SaslCredentials, TlsConnectorConfig, TrustStore,
12    },
13    krabka_security::{ListenerProtocol, SaslMechanism},
14    std::{collections::BTreeMap, io::Error, mem, path::PathBuf},
15    tokio::fs,
16};
17
18/// Kafka client properties shared by capture and offset restoration.
19#[derive(Debug, Args, Clone, Default)]
20pub struct SecurityArgs {
21    /// Kafka client properties containing `security.protocol`, TLS and SASL.
22    #[arg(long, value_name = "FILE")]
23    pub command_config: Option<PathBuf>,
24}
25
26impl SecurityArgs {
27    /// Build the shared client security policy.
28    ///
29    /// # Errors
30    ///
31    /// Returns [`BackupError::Io`] when the file cannot be read and
32    /// [`BackupError::InvalidArgument`] when its security properties conflict.
33    pub async fn load(&self) -> Result<Option<ClientSecurity>, BackupError> {
34        let Some(path) = &self.command_config else {
35            return Ok(None);
36        };
37        let text = fs::read_to_string(path).await.map_err(|source| {
38            BackupError::Io(Error::new(
39                source.kind(),
40                format!("read command config {}: {source}", path.display()),
41            ))
42        })?;
43        let properties = parse_properties(&text)?;
44        security(&properties)
45    }
46}
47
48fn parse_properties(text: &str) -> Result<BTreeMap<String, String>, BackupError> {
49    let mut logical = Vec::new();
50    let mut pending = String::new();
51    for line in text.lines() {
52        let trailing = line
53            .chars()
54            .rev()
55            .take_while(|character| *character == '\\')
56            .count();
57        pending.push_str(line.trim_start());
58        if trailing % 2 == 1 {
59            pending.pop();
60        } else {
61            logical.push(mem::take(&mut pending));
62        }
63    }
64    if !pending.is_empty() {
65        logical.push(pending);
66    }
67
68    let mut properties = BTreeMap::new();
69    for line in logical {
70        let trimmed = line.trim_start();
71        if trimmed.is_empty() || trimmed.starts_with(['#', '!']) {
72            continue;
73        }
74        let split = property_split(trimmed);
75        let (key, value) = trimmed.split_at(split);
76        let value = value.trim_start_matches(|character: char| {
77            character.is_whitespace() || character == '=' || character == ':'
78        });
79        properties.insert(unescape(key.trim_end())?, unescape(value)?);
80    }
81    Ok(properties)
82}
83
84fn property_split(line: &str) -> usize {
85    let mut escaped = false;
86    for (index, character) in line.char_indices() {
87        if escaped {
88            escaped = false;
89        } else if character == '\\' {
90            escaped = true;
91        } else if character == '=' || character == ':' || character.is_whitespace() {
92            return index;
93        }
94    }
95    line.len()
96}
97
98fn unescape(value: &str) -> Result<String, BackupError> {
99    let mut characters = value.chars();
100    let mut unescaped = String::new();
101    while let Some(character) = characters.next() {
102        if character != '\\' {
103            unescaped.push(character);
104            continue;
105        }
106        match characters.next() {
107            Some('t') => unescaped.push('\t'),
108            Some('n') => unescaped.push('\n'),
109            Some('r') => unescaped.push('\r'),
110            Some('u') => {
111                let digits: String = characters.by_ref().take(4).collect();
112                let code = u32::from_str_radix(&digits, 16).map_err(|_| {
113                    BackupError::InvalidArgument("invalid unicode escape in command config".into())
114                })?;
115                unescaped.push(char::from_u32(code).ok_or_else(|| {
116                    BackupError::InvalidArgument(
117                        "invalid unicode code point in command config".into(),
118                    )
119                })?);
120            }
121            Some(other) => unescaped.push(other),
122            None => unescaped.push('\\'),
123        }
124    }
125    Ok(unescaped)
126}
127
128fn security(properties: &BTreeMap<String, String>) -> Result<Option<ClientSecurity>, BackupError> {
129    let protocol = match properties
130        .get("security.protocol")
131        .map_or("PLAINTEXT", String::as_str)
132    {
133        "PLAINTEXT" => ListenerProtocol::Plaintext,
134        "SSL" => ListenerProtocol::Ssl,
135        "SASL_PLAINTEXT" => ListenerProtocol::SaslPlaintext,
136        "SASL_SSL" => ListenerProtocol::SaslSsl,
137        value => return invalid(&format!("unsupported security.protocol {value}")),
138    };
139    if !protocol.requires_sasl()
140        && let Some(key) = properties.keys().find(|key| key.starts_with("sasl."))
141    {
142        return invalid(&format!("{key} requires a SASL security.protocol"));
143    }
144    if !protocol.requires_tls()
145        && let Some(key) = properties.keys().find(|key| key.starts_with("ssl."))
146    {
147        return invalid(&format!("{key} requires SSL or SASL_SSL"));
148    }
149    if protocol == ListenerProtocol::Plaintext {
150        return Ok(None);
151    }
152
153    let tls = protocol
154        .requires_tls()
155        .then(|| tls_config(properties))
156        .transpose()?;
157    let sasl = protocol
158        .requires_sasl()
159        .then(|| sasl_credentials(properties))
160        .transpose()?;
161    Ok(Some(ClientSecurity {
162        protocol,
163        tls,
164        sasl,
165        sasl_host: properties.get("sasl.kerberos.service.host").cloned(),
166    }))
167}
168
169fn tls_config(properties: &BTreeMap<String, String>) -> Result<TlsConnectorConfig, BackupError> {
170    for key in ["ssl.truststore.type", "ssl.keystore.type"] {
171        if let Some(kind) = properties.get(key)
172            && kind != "PEM"
173        {
174            return invalid(&format!("{key}={kind} is unsupported; expected PEM"));
175        }
176    }
177    let key_store = match (
178        properties.get("ssl.keystore.location"),
179        properties.get("ssl.key.location"),
180    ) {
181        (Some(cert), Some(key)) => Some(KeyStore::PemFiles {
182            certificate_chain: PathBuf::from(cert),
183            private_key: PathBuf::from(key),
184            key_password: None,
185        }),
186        (None, None) => None,
187        _ => {
188            return invalid("ssl.keystore.location and ssl.key.location must be provided together");
189        }
190    };
191    let mut tls = TlsConnectorConfig::default();
192    if let Some(location) = properties.get("ssl.truststore.location") {
193        tls.trust_store = TrustStore::PemFile(PathBuf::from(location));
194    }
195    tls.key_store = key_store;
196    tls.server_name = properties
197        .get("ssl.server.name")
198        .cloned()
199        .unwrap_or_default();
200    Ok(tls)
201}
202
203fn sasl_credentials(properties: &BTreeMap<String, String>) -> Result<SaslCredentials, BackupError> {
204    let mechanism = required(properties, "sasl.mechanism")?;
205    let options = properties
206        .get("sasl.jaas.config")
207        .map_or_default(|value| jaas_options(value));
208    match mechanism.as_str() {
209        "PLAIN" => Ok(SaslCredentials::Plain {
210            username: required(&options, "username")?,
211            password: required(&options, "password")?,
212        }),
213        "SCRAM-SHA-256" | "SCRAM-SHA-512" => Ok(SaslCredentials::Scram {
214            mechanism: if mechanism == "SCRAM-SHA-256" {
215                SaslMechanism::ScramSha256
216            } else {
217                SaslMechanism::ScramSha512
218            },
219            username: required(&options, "username")?,
220            password: required(&options, "password")?,
221            // Kafka's `ScramLoginModule` reads `tokenauth` with
222            // `Boolean.parseBoolean`: a delegation-token login.
223            delegation_token: options
224                .get("tokenauth")
225                .is_some_and(|value| value.eq_ignore_ascii_case("true")),
226        }),
227        value => invalid(&format!("unsupported sasl.mechanism {value}")),
228    }
229}
230
231fn jaas_options(value: &str) -> BTreeMap<String, String> {
232    let mut options = BTreeMap::new();
233    let mut characters = value.trim_end_matches(';').chars().peekable();
234    while let Some(character) = characters.next() {
235        if character.is_whitespace() {
236            continue;
237        }
238        let mut key = String::from(character);
239        while let Some(&character) = characters.peek() {
240            if character == '=' || character.is_whitespace() {
241                break;
242            }
243            key.push(character);
244            characters.next();
245        }
246        while characters
247            .peek()
248            .is_some_and(|character| character.is_whitespace())
249        {
250            characters.next();
251        }
252        if characters.next() != Some('=') {
253            while characters
254                .peek()
255                .is_some_and(|character| !character.is_whitespace())
256            {
257                characters.next();
258            }
259            continue;
260        }
261        while characters
262            .peek()
263            .is_some_and(|character| character.is_whitespace())
264        {
265            characters.next();
266        }
267        let quote = characters
268            .peek()
269            .copied()
270            .filter(|character| *character == '"' || *character == '\'');
271        if quote.is_some() {
272            characters.next();
273        }
274        let mut parsed = String::new();
275        let mut escaped = false;
276        for character in characters.by_ref() {
277            if escaped {
278                parsed.push(character);
279                escaped = false;
280            } else if character == '\\' {
281                escaped = true;
282            } else if quote == Some(character) || (quote.is_none() && character.is_whitespace()) {
283                break;
284            } else {
285                parsed.push(character);
286            }
287        }
288        options.insert(key, parsed);
289    }
290    options
291}
292
293fn required(values: &BTreeMap<String, String>, key: &str) -> Result<String, BackupError> {
294    values
295        .get(key)
296        .cloned()
297        .ok_or_else(|| BackupError::InvalidArgument(format!("command config is missing {key}")))
298}
299
300fn invalid<T>(message: &str) -> Result<T, BackupError> {
301    Err(BackupError::InvalidArgument(format!(
302        "invalid command config: {message}"
303    )))
304}
305
306#[cfg(test)]
307mod tests {
308    use {
309        super::*,
310        assert2::{assert, check},
311    };
312
313    #[test]
314    fn properties_support_comments_escapes_and_continuations() {
315        let parsed = parse_properties("! ignored\na\\:b = one\\\n  two\\u0033\n")
316            .expect("valid Java properties");
317        check!(parsed == BTreeMap::from([("a:b".into(), "onetwo3".into())]));
318    }
319
320    #[test]
321    fn sasl_ssl_maps_to_the_shared_client_policy() {
322        let properties = parse_properties(
323            "security.protocol=SASL_SSL\n\
324             ssl.truststore.type=PEM\n\
325             ssl.truststore.location=/etc/krabka/ca.pem\n\
326             ssl.server.name=broker.example\n\
327             sasl.mechanism=SCRAM-SHA-512\n\
328             sasl.jaas.config=x required username=backup password=not-logged;\n",
329        )
330        .expect("valid properties");
331        let policy = security(&properties)
332            .expect("valid policy")
333            .expect("secured policy");
334
335        check!(policy.protocol == ListenerProtocol::SaslSsl);
336        let tls = policy.tls.expect("TLS policy");
337        check!(tls.trust_store == TrustStore::PemFile(PathBuf::from("/etc/krabka/ca.pem")));
338        check!(tls.server_name == "broker.example");
339        assert!(let Some(SaslCredentials::Scram {
340            mechanism: SaslMechanism::ScramSha512,
341            username,
342            password,
343            delegation_token: false,
344        }) = policy.sasl);
345        check!(username == "backup");
346        check!(password == "not-logged");
347    }
348
349    /// `tokenauth` marks a delegation-token SCRAM login, read the way Kafka's
350    /// `ScramLoginModule` reads it.
351    #[yare::parameterized(
352        absent = { "x required username=id password=hmac;", false },
353        true_lowercase = { "x required username=id password=hmac tokenauth=true;", true },
354        true_uppercase = { "x required username=id password=hmac tokenauth=TRUE;", true },
355        false_value = { "x required username=id password=hmac tokenauth=false;", false },
356    )]
357    fn tokenauth_marks_a_delegation_token_login(jaas: &str, expected: bool) {
358        let properties = parse_properties(&format!(
359            "security.protocol=SASL_PLAINTEXT\n\
360             sasl.mechanism=SCRAM-SHA-256\n\
361             sasl.jaas.config={jaas}\n"
362        ))
363        .expect("valid properties");
364        let policy = security(&properties)
365            .expect("valid policy")
366            .expect("secured policy");
367        assert2::assert!(
368            let Some(SaslCredentials::Scram {
369                delegation_token, ..
370            }) = policy.sasl,
371            "{jaas}"
372        );
373        check!(delegation_token == expected, "{jaas}");
374    }
375
376    #[test]
377    fn invalid_config_errors_do_not_echo_credentials() {
378        let properties = parse_properties(
379            "security.protocol=SASL_PLAINTEXT\n\
380             sasl.mechanism=SCRAM-SHA-256\n\
381             sasl.jaas.config=x required password=hunter2;\n",
382        )
383        .expect("valid properties");
384        let error = security(&properties)
385            .expect_err("missing username must fail")
386            .to_string();
387        check!(!error.contains("hunter2"), "got: {error}");
388        check!(error.contains("username"), "got: {error}");
389    }
390
391    #[test]
392    fn tls_refuses_non_pem_stores() {
393        let properties = parse_properties(
394            "security.protocol=SSL\nssl.truststore.type=JKS\nssl.truststore.location=a.jks\n",
395        )
396        .expect("valid properties");
397        let error = security(&properties)
398            .expect_err("the shared client reads PEM, not JKS")
399            .to_string();
400        check!(error.contains("expected PEM"), "got: {error}");
401    }
402
403    #[yare::parameterized(
404        sasl_mechanism = { "sasl.mechanism", "SCRAM-SHA-512" },
405        tls_location = { "ssl.truststore.location", "/etc/krabka/ca.pem" },
406    )]
407    fn security_properties_cannot_be_silently_ignored(property: &str, value: &str) {
408        let properties = BTreeMap::from([(property.into(), value.into())]);
409        let error = security(&properties)
410            .expect_err("PLAINTEXT must reject secured-client properties")
411            .to_string();
412        check!(error.contains(property), "got: {error}");
413    }
414
415    #[test]
416    fn tls_sni_defaults_to_each_connection_target() {
417        let properties =
418            parse_properties("security.protocol=SSL\nssl.truststore.location=/etc/krabka/ca.pem\n")
419                .expect("valid properties");
420        let policy = security(&properties)
421            .expect("valid policy")
422            .expect("secured policy");
423        check!(policy.tls.expect("TLS policy").server_name.is_empty());
424    }
425}
krabka_backup/connection.rs

krabka_backup/
connection.rs
