krabka_connect_worker/
config.rs

1//! Command-line and environment configuration for one connector worker.
2
3use std::{fmt::Write as _, net::SocketAddr, path::PathBuf, str::FromStr, time::Duration};
4
5use clap::Parser;
6use krabka_client_core::{
7    ClientFrameMax, ConnectionDispatchQueueCapacity,
8    security::{ClientSecurity, KeyStore, SaslCredentials, TlsConnectorConfig, TrustStore},
9};
10use krabka_connect::SecretString;
11use krabka_connect_postgres::PostgresSourceConfig;
12use krabka_replicator::config::ReplicationFactor;
13use krabka_security::{ListenerProtocol, SaslMechanism};
14use krabka_units::{ByteSize, convert::ByteSizeExt as _};
15use sha2::{Digest as _, Sha256};
16
17/// Kafka listener protocol used by the worker.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum BrokerProtocol {
20    /// Unencrypted Kafka protocol.
21    Plaintext,
22    /// Kafka over TLS.
23    Ssl,
24    /// SASL over an unencrypted connection.
25    SaslPlaintext,
26    /// SASL over TLS.
27    SaslSsl,
28}
29
30impl FromStr for BrokerProtocol {
31    type Err = String;
32
33    fn from_str(value: &str) -> Result<Self, Self::Err> {
34        match value.to_ascii_uppercase().replace('-', "_").as_str() {
35            "PLAINTEXT" => Ok(Self::Plaintext),
36            "SSL" => Ok(Self::Ssl),
37            "SASL_PLAINTEXT" => Ok(Self::SaslPlaintext),
38            "SASL_SSL" => Ok(Self::SaslSsl),
39            _ => Err(format!(
40                "unsupported broker protocol {value:?}; expected PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL"
41            )),
42        }
43    }
44}
45
46impl From<BrokerProtocol> for ListenerProtocol {
47    fn from(value: BrokerProtocol) -> Self {
48        match value {
49            BrokerProtocol::Plaintext => Self::Plaintext,
50            BrokerProtocol::Ssl => Self::Ssl,
51            BrokerProtocol::SaslPlaintext => Self::SaslPlaintext,
52            BrokerProtocol::SaslSsl => Self::SaslSsl,
53        }
54    }
55}
56
57/// Supported username/password SASL mechanisms.
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59pub enum BrokerSaslMechanism {
60    /// SASL/PLAIN.
61    Plain,
62    /// SCRAM-SHA-256.
63    ScramSha256,
64    /// SCRAM-SHA-512.
65    ScramSha512,
66}
67
68impl FromStr for BrokerSaslMechanism {
69    type Err = String;
70
71    fn from_str(value: &str) -> Result<Self, Self::Err> {
72        match value.to_ascii_uppercase().replace('_', "-").as_str() {
73            "PLAIN" => Ok(Self::Plain),
74            "SCRAM-SHA-256" => Ok(Self::ScramSha256),
75            "SCRAM-SHA-512" => Ok(Self::ScramSha512),
76            _ => Err(format!(
77                "unsupported SASL mechanism {value:?}; expected PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512"
78            )),
79        }
80    }
81}
82
83/// Complete configuration for one managed Postgres source connector.
84#[derive(Clone, Parser)]
85#[command(name = "krabka-connect-worker", version, about)]
86pub struct WorkerConfig {
87    /// Stable connector ID and checkpoint key.
88    #[arg(long, env = "KRABKA_CONNECTOR_ID", value_parser = parse_non_empty)]
89    pub connector_id: String,
90    /// Kafka bootstrap address.
91    #[arg(long, env = "KRABKA_KAFKA_BOOTSTRAP", value_parser = parse_non_empty)]
92    pub kafka_bootstrap: String,
93    /// Confluent-compatible Schema Registry base URL.
94    #[arg(long, env = "KRABKA_SCHEMA_REGISTRY_URL", value_parser = parse_non_empty)]
95    pub schema_registry_url: String,
96    /// Replicas used when the worker creates data and checkpoint topics.
97    #[arg(long, env = "KRABKA_CONNECT_REPLICATION_FACTOR", default_value = "1")]
98    pub replication_factor: ReplicationFactor,
99    /// `PostgreSQL` connection URL. Formatting is always redacted.
100    #[arg(long, env = "KRABKA_POSTGRES_URL", value_parser = parse_secret)]
101    pub postgres_url: SecretString,
102    /// `PostgreSQL` logical replication slot.
103    #[arg(long, env = "KRABKA_POSTGRES_SLOT", value_parser = parse_non_empty)]
104    pub postgres_slot: String,
105    /// `PostgreSQL` publication.
106    #[arg(
107        long,
108        env = "KRABKA_POSTGRES_PUBLICATION",
109        default_value = "krabka_connect"
110    )]
111    pub postgres_publication: String,
112    /// `PostgreSQL` schema containing the selected tables.
113    #[arg(long, env = "KRABKA_POSTGRES_SCHEMA", default_value = "public")]
114    pub postgres_schema: String,
115    /// Comma-delimited table names captured from the publication.
116    #[arg(
117        long,
118        env = "KRABKA_POSTGRES_TABLES",
119        value_delimiter = ',',
120        required = true
121    )]
122    pub postgres_tables: Vec<String>,
123    /// Prefix prepended to source topics with a dot separator; empty disables it.
124    #[arg(long, env = "KRABKA_TOPIC_PREFIX", default_value = "db")]
125    pub topic_prefix: String,
126    /// Maximum records buffered before a Kafka durability barrier.
127    #[arg(long, env = "KRABKA_CONNECT_BATCH_SIZE", default_value_t = 500, value_parser = parse_positive_usize)]
128    pub batch_size: usize,
129    /// Maximum time between Kafka durability barriers, in milliseconds.
130    #[arg(long, env = "KRABKA_CONNECT_COMMIT_INTERVAL_MS", default_value_t = 5_000, value_parser = clap::value_parser!(u64).range(1..))]
131    pub commit_interval_ms: u64,
132    /// Delay after `PostgreSQL` reports no new changes, in milliseconds.
133    #[arg(long, env = "KRABKA_CONNECT_POLL_BACKOFF_MS", default_value_t = 100, value_parser = clap::value_parser!(u64).range(1..))]
134    pub poll_backoff_ms: u64,
135    /// Address serving `/live`, `/ready`, and `/metrics`.
136    #[arg(
137        long,
138        env = "KRABKA_CONNECT_HEALTH_LISTEN",
139        default_value = "0.0.0.0:8080"
140    )]
141    pub health_listen: SocketAddr,
142    /// Capacity of each Kafka client's pending request dispatch queue.
143    #[arg(long, env = "KRABKA_CLIENT_DISPATCH_QUEUE_CAPACITY", default_value_t = 64, value_parser = parse_positive_usize)]
144    pub client_dispatch_queue_capacity: usize,
145    /// Maximum accepted Kafka response frame size in bytes.
146    #[arg(long, env = "KRABKA_CLIENT_FRAME_MAX_BYTES", default_value_t = 100 * 1024 * 1024, value_parser = clap::value_parser!(u64).range(1..))]
147    pub client_frame_max_bytes: u64,
148    /// Broker security protocol.
149    #[arg(long, env = "KRABKA_BROKER_PROTOCOL", default_value = "PLAINTEXT")]
150    pub broker_protocol: BrokerProtocol,
151    /// PEM file containing broker CA certificates.
152    #[arg(long, env = "KRABKA_BROKER_CA_PATH")]
153    pub broker_ca_path: Option<PathBuf>,
154    /// TLS SNI name used to verify the broker certificate.
155    #[arg(long, env = "KRABKA_BROKER_SERVER_NAME")]
156    pub broker_server_name: Option<String>,
157    /// PEM client certificate chain for mTLS.
158    #[arg(long, env = "KRABKA_BROKER_CERT_PATH")]
159    pub broker_cert_path: Option<PathBuf>,
160    /// PEM private key for mTLS.
161    #[arg(long, env = "KRABKA_BROKER_KEY_PATH")]
162    pub broker_key_path: Option<PathBuf>,
163    /// SASL username.
164    #[arg(long, env = "KRABKA_BROKER_SASL_USERNAME")]
165    pub broker_sasl_username: Option<String>,
166    /// SASL password. Formatting is always redacted.
167    #[arg(long, env = "KRABKA_BROKER_SASL_PASSWORD", value_parser = parse_secret)]
168    pub broker_sasl_password: Option<SecretString>,
169    /// SASL PLAIN or SCRAM mechanism.
170    #[arg(long, env = "KRABKA_BROKER_SASL_MECHANISM")]
171    pub broker_sasl_mechanism: Option<BrokerSaslMechanism>,
172}
173
174impl WorkerConfig {
175    /// Validate cross-field resource and security constraints.
176    ///
177    /// # Errors
178    ///
179    /// Returns an error for an oversized batch, invalid client limit, incomplete
180    /// mTLS identity, or security material inconsistent with the protocol.
181    pub fn validate(&self) -> Result<(), String> {
182        u32::try_from(self.batch_size)
183            .map_err(|_| "batch size must fit PostgreSQL's u32 poll limit".to_owned())?;
184        ConnectionDispatchQueueCapacity::new(self.client_dispatch_queue_capacity)?;
185        ClientFrameMax::try_from(ByteSize::from_bytes(self.client_frame_max_bytes))?;
186        if self
187            .postgres_tables
188            .iter()
189            .any(|table| table.trim().is_empty() || table.contains('.'))
190        {
191            return Err(
192                "PostgreSQL tables must be non-empty unqualified names; use --postgres-schema separately"
193                    .to_owned(),
194            );
195        }
196
197        let protocol: ListenerProtocol = self.broker_protocol.into();
198        let has_tls = self.broker_ca_path.is_some()
199            || self.broker_server_name.is_some()
200            || self.broker_cert_path.is_some()
201            || self.broker_key_path.is_some();
202        if protocol.requires_tls() {
203            if self.broker_ca_path.is_none() || self.broker_server_name.is_none() {
204                return Err("TLS requires broker CA path and server name".to_owned());
205            }
206            if self.broker_cert_path.is_some() != self.broker_key_path.is_some() {
207                return Err("mTLS requires both broker cert path and key path".to_owned());
208            }
209        } else if has_tls {
210            return Err("TLS settings require SSL or SASL_SSL broker protocol".to_owned());
211        }
212
213        let has_sasl = self.broker_sasl_username.is_some()
214            || self.broker_sasl_password.is_some()
215            || self.broker_sasl_mechanism.is_some();
216        if protocol.requires_sasl() {
217            if self.broker_sasl_username.is_none()
218                || self.broker_sasl_password.is_none()
219                || self.broker_sasl_mechanism.is_none()
220            {
221                return Err("SASL requires mechanism, username, and password".to_owned());
222            }
223        } else if has_sasl {
224            return Err(
225                "SASL settings require SASL_PLAINTEXT or SASL_SSL broker protocol".to_owned(),
226            );
227        }
228        Ok(())
229    }
230
231    /// Build the existing typed `PostgreSQL` source configuration.
232    ///
233    /// # Panics
234    ///
235    /// Panics only if called before [`Self::validate`] with a batch size that
236    /// does not fit `u32`.
237    #[must_use]
238    pub fn postgres_source(&self) -> PostgresSourceConfig {
239        PostgresSourceConfig {
240            schema_registry_url: self.schema_registry_url.clone(),
241            database_url: self.postgres_url.clone(),
242            slot_name: self.postgres_slot.clone(),
243            publication_name: self.postgres_publication.clone(),
244            schema: self.postgres_schema.clone(),
245            table_names: self.postgres_tables.clone(),
246            max_messages_per_poll: u32::try_from(self.batch_size)
247                .expect("worker config validated before source construction"),
248        }
249    }
250
251    /// Commit interval as a standard duration.
252    #[must_use]
253    pub const fn commit_interval(&self) -> Duration {
254        Duration::from_millis(self.commit_interval_ms)
255    }
256
257    /// Empty-poll backoff as a standard duration.
258    #[must_use]
259    pub const fn poll_backoff(&self) -> Duration {
260        Duration::from_millis(self.poll_backoff_ms)
261    }
262
263    /// Stable checkpoint key scoped to the connector and `PostgreSQL` source.
264    #[must_use]
265    pub fn checkpoint_key(&self) -> String {
266        let mut digest = Sha256::new();
267        digest.update(self.postgres_url.expose_secret().as_bytes());
268        digest.update([0]);
269        digest.update(self.postgres_slot.as_bytes());
270
271        let digest = digest.finalize();
272        let mut key = String::with_capacity(self.connector_id.len() + 1 + digest.len() * 2);
273        key.push_str(&self.connector_id);
274        key.push(':');
275        for byte in digest {
276            write!(&mut key, "{byte:02x}").expect("writing to a String cannot fail");
277        }
278        key
279    }
280
281    /// Validated Kafka client security, or `None` for plaintext.
282    ///
283    /// # Errors
284    ///
285    /// Returns an error when the security fields are incomplete or inconsistent.
286    pub fn client_security(&self) -> Result<Option<ClientSecurity>, String> {
287        self.validate()?;
288        let protocol: ListenerProtocol = self.broker_protocol.into();
289        if protocol == ListenerProtocol::Plaintext {
290            return Ok(None);
291        }
292        let tls = if protocol.requires_tls() {
293            let mut tls = TlsConnectorConfig::default();
294            tls.trust_store = TrustStore::PemFile(
295                self.broker_ca_path
296                    .clone()
297                    .ok_or_else(|| "TLS CA path missing after validation".to_owned())?,
298            );
299            tls.server_name = self
300                .broker_server_name
301                .clone()
302                .ok_or_else(|| "TLS server name missing after validation".to_owned())?;
303            tls.key_store = self
304                .broker_cert_path
305                .clone()
306                .zip(self.broker_key_path.clone())
307                .map(|(certificate_chain, private_key)| KeyStore::PemFiles {
308                    certificate_chain,
309                    private_key,
310                    key_password: None,
311                });
312            Some(tls)
313        } else {
314            None
315        };
316        let sasl = if protocol.requires_sasl() {
317            let username = self
318                .broker_sasl_username
319                .clone()
320                .ok_or_else(|| "SASL username missing after validation".to_owned())?;
321            let password = self
322                .broker_sasl_password
323                .as_ref()
324                .ok_or_else(|| "SASL password missing after validation".to_owned())?
325                .expose_secret()
326                .to_owned();
327            Some(
328                match self
329                    .broker_sasl_mechanism
330                    .ok_or_else(|| "SASL mechanism missing after validation".to_owned())?
331                {
332                    BrokerSaslMechanism::Plain => SaslCredentials::Plain { username, password },
333                    BrokerSaslMechanism::ScramSha256 => SaslCredentials::Scram {
334                        mechanism: SaslMechanism::ScramSha256,
335                        username,
336                        password,
337                        delegation_token: false,
338                    },
339                    BrokerSaslMechanism::ScramSha512 => SaslCredentials::Scram {
340                        mechanism: SaslMechanism::ScramSha512,
341                        username,
342                        password,
343                        delegation_token: false,
344                    },
345                },
346            )
347        } else {
348            None
349        };
350        Ok(Some(ClientSecurity {
351            protocol,
352            tls,
353            sasl,
354            sasl_host: self.broker_server_name.clone(),
355        }))
356    }
357}
358
359fn parse_non_empty(value: &str) -> Result<String, String> {
360    if value.trim().is_empty() {
361        Err("value must not be empty".to_owned())
362    } else {
363        Ok(value.to_owned())
364    }
365}
366
367fn parse_secret(value: &str) -> Result<SecretString, String> {
368    parse_non_empty(value).map(SecretString::new)
369}
370
371fn parse_positive_usize(value: &str) -> Result<usize, String> {
372    let parsed = value.parse::<usize>().map_err(|error| error.to_string())?;
373    if parsed == 0 {
374        Err("value must be greater than zero".to_owned())
375    } else {
376        Ok(parsed)
377    }
378}
379
380#[cfg(test)]
381mod tests {
382    use assert2::assert;
383
384    use super::*;
385
386    fn base_args() -> Vec<&'static str> {
387        vec![
388            "krabka-connect-worker",
389            "--connector-id=orders",
390            "--kafka-bootstrap=localhost:9092",
391            "--schema-registry-url=http://localhost:8081",
392            "--postgres-url=postgres://secret@localhost/app",
393            "--postgres-slot=orders",
394            "--postgres-tables=orders,customers",
395        ]
396    }
397
398    #[test]
399    fn parses_defaults_without_exposing_postgres_secret() {
400        let config = WorkerConfig::try_parse_from(base_args()).expect("valid CLI");
401        assert!(config.validate().is_ok());
402        assert!(config.postgres_tables == ["orders", "customers"]);
403        assert!(config.topic_prefix == "db");
404        assert!(config.batch_size == 500);
405        assert!(config.replication_factor.get() == 1);
406        assert!(config.postgres_url.to_string() == "[REDACTED]");
407    }
408
409    #[test]
410    fn builds_tls_scram_security_and_rejects_partial_mtls() {
411        let mut args = base_args();
412        args.extend([
413            "--broker-protocol=SASL_SSL",
414            "--broker-ca-path=ca.pem",
415            "--broker-server-name=broker.example",
416            "--broker-sasl-mechanism=SCRAM-SHA-512",
417            "--broker-sasl-username=user",
418            "--broker-sasl-password=secret",
419        ]);
420        let config = WorkerConfig::try_parse_from(args).expect("valid secure CLI");
421        let security = config
422            .client_security()
423            .expect("valid security")
424            .expect("secure protocol");
425        assert!(security.protocol == ListenerProtocol::SaslSsl);
426        assert!(matches!(
427            security.sasl,
428            Some(SaslCredentials::Scram {
429                mechanism: SaslMechanism::ScramSha512,
430                delegation_token: false,
431                ..
432            })
433        ));
434        let tls = security.tls.expect("TLS config");
435        assert!(tls.trust_store == TrustStore::PemFile(PathBuf::from("ca.pem")));
436        assert!(tls.server_name == "broker.example");
437        assert!(tls.hostname_verification);
438        assert!(tls.key_store.is_none());
439
440        let mut partial = base_args();
441        partial.extend([
442            "--broker-protocol=SSL",
443            "--broker-ca-path=ca.pem",
444            "--broker-server-name=broker.example",
445            "--broker-cert-path=client.pem",
446        ]);
447        let mut config = WorkerConfig::try_parse_from(partial).expect("CLI shape parses");
448        assert!(config.validate().is_err());
449        config.broker_key_path = Some(PathBuf::from("client.key"));
450        let tls = config
451            .client_security()
452            .expect("complete mTLS config")
453            .expect("secure protocol")
454            .tls
455            .expect("TLS config");
456        assert!(
457            tls.key_store
458                == Some(KeyStore::PemFiles {
459                    certificate_chain: PathBuf::from("client.pem"),
460                    private_key: PathBuf::from("client.key"),
461                    key_password: None,
462                })
463        );
464    }
465
466    #[test]
467    fn rejects_qualified_tables_and_scopes_checkpoints_to_source_identity() {
468        let base = WorkerConfig::try_parse_from(base_args()).expect("valid CLI");
469        let base_key = base.checkpoint_key();
470        assert!(base_key.starts_with("orders:"));
471        assert!(!base_key.contains("secret"));
472
473        let mut different_slot = base.clone();
474        different_slot.postgres_slot = "other".into();
475        assert!(different_slot.checkpoint_key() != base_key);
476
477        let mut different_database = base.clone();
478        different_database.postgres_url = SecretString::new("postgres://localhost/other");
479        assert!(different_database.checkpoint_key() != base_key);
480
481        let mut qualified = base;
482        qualified.postgres_tables = vec!["audit.orders".into()];
483        assert!(qualified.validate().is_err());
484    }
485}
krabka_connect_worker/config.rs

krabka_connect_worker/
config.rs
