krabka_admin_ui/
config.rs

1//! Runtime configuration for one admin UI instance.
2
3use std::{
4    net::SocketAddr,
5    path::PathBuf,
6    time::{Duration, Instant},
7};
8
9use clap::Parser;
10use krabka_client_core::security::{KeyStore, TlsConnectorConfig, TrustStore};
11use krabka_security::ListenerProtocol;
12use krabka_units::{parse, prelude::*};
13use thiserror::Error;
14
15/// Default maximum size of an authenticated mutation JSON body.
16pub const DEFAULT_MUTATION_JSON_BODY_LIMIT: ByteSize = mebibytes(1);
17
18/// Default server-side lifetime for an authenticated admin UI session.
19pub const DEFAULT_SESSION_TTL: Time = hours(8);
20
21/// Default Kafka request timeout for admin UI topic mutations.
22pub const DEFAULT_TOPIC_MUTATION_TIMEOUT: Time = secs(30);
23
24fn parse_body_limit(input: &str) -> Result<ByteSize, String> {
25    let value = parse::positive_byte_size(input).map_err(|error| error.to_string())?;
26    let lowered = value.bytes_usize();
27    if u64::try_from(lowered).is_ok_and(|bytes| ByteSize::from_bytes(bytes) == value) {
28        Ok(value)
29    } else {
30        Err("body limit must be a whole byte count representable by usize".to_string())
31    }
32}
33
34fn parse_session_ttl(input: &str) -> Result<Time, String> {
35    let value = parse::positive_time(input).map_err(|error| error.to_string())?;
36    let duration =
37        Duration::try_from_secs_f64(value.secs_f64()).map_err(|error| error.to_string())?;
38    Instant::now()
39        .checked_add(duration)
40        .map(|_| value)
41        .ok_or_else(|| "session TTL exceeds the platform monotonic clock".to_string())
42}
43
44fn parse_topic_mutation_timeout(input: &str) -> Result<Time, String> {
45    let value = parse::positive_time(input).map_err(|error| error.to_string())?;
46    let millis = value.secs_f64() * 1_000.0;
47    if millis.fract() == 0.0 && millis <= f64::from(i32::MAX) {
48        Ok(value)
49    } else {
50        Err("topic mutation timeout must be a whole i32 millisecond count".to_string())
51    }
52}
53
54/// Command-line and environment inputs owned by the admin UI runtime.
55#[derive(Debug, Clone, Parser)]
56#[command(name = "krabka-admin-ui")]
57pub struct AdminUiRuntimeArgs {
58    /// Maximum size of an authenticated mutation JSON body.
59    #[arg(
60        long,
61        env = "KRABKA_ADMIN_UI_MUTATION_JSON_BODY_LIMIT",
62        default_value = "1MiB",
63        value_parser = parse_body_limit
64    )]
65    pub mutation_json_body_limit: ByteSize,
66
67    /// Server-side lifetime for an authenticated session.
68    #[arg(
69        long,
70        env = "KRABKA_ADMIN_UI_SESSION_TTL",
71        default_value = "8h",
72        value_parser = parse_session_ttl
73    )]
74    pub session_ttl: Time,
75
76    /// Kafka request timeout for topic mutations.
77    #[arg(
78        long,
79        env = "KRABKA_ADMIN_UI_TOPIC_MUTATION_TIMEOUT",
80        default_value = "30s",
81        value_parser = parse_topic_mutation_timeout
82    )]
83    pub topic_mutation_timeout: Time,
84}
85
86#[derive(Debug, Clone, PartialEq, Eq)]
87pub enum BrokerSecurityConfig {
88    SaslPlaintext,
89    SaslSsl {
90        trust_roots_pem: Option<PathBuf>,
91        server_name: String,
92        client_identity: Option<(PathBuf, PathBuf)>,
93    },
94}
95
96#[derive(Debug, Clone, PartialEq)]
97pub struct AdminUiConfig {
98    pub listen_addr: SocketAddr,
99    pub cluster_name: String,
100    pub bootstrap_addrs: Vec<String>,
101    pub security: BrokerSecurityConfig,
102    pub session_ttl: Time,
103    pub mutation_json_body_limit: ByteSize,
104    pub topic_mutation_timeout: Time,
105}
106
107#[derive(Debug, Error, PartialEq, Eq)]
108pub enum ConfigError {
109    #[error("at least one KRABKA_ADMIN_UI_BOOTSTRAP address is required")]
110    MissingBootstrap,
111    #[error("KRABKA_ADMIN_UI_BOOTSTRAP contains an invalid address: {0}")]
112    InvalidBootstrapAddr(String),
113    #[error("KRABKA_ADMIN_UI_LISTEN_ADDR is invalid: {0}")]
114    InvalidListenAddr(String),
115    #[error("KRABKA_ADMIN_UI_SECURITY_PROTOCOL must be SASL_PLAINTEXT or SASL_SSL")]
116    InvalidSecurityProtocol,
117    #[error("KRABKA_ADMIN_UI_TLS_SERVER_NAME is required for SASL_SSL")]
118    MissingTlsServerName,
119    #[error(
120        "KRABKA_ADMIN_UI_TLS_CLIENT_CERT_PEM and KRABKA_ADMIN_UI_TLS_CLIENT_KEY_PEM must be set together"
121    )]
122    IncompleteTlsClientIdentity,
123}
124
125impl Default for AdminUiConfig {
126    fn default() -> Self {
127        Self {
128            listen_addr: "127.0.0.1:8088".parse().expect("static socket addr parses"),
129            cluster_name: "local".to_string(),
130            bootstrap_addrs: Vec::new(),
131            security: BrokerSecurityConfig::SaslPlaintext,
132            session_ttl: DEFAULT_SESSION_TTL,
133            mutation_json_body_limit: DEFAULT_MUTATION_JSON_BODY_LIMIT,
134            topic_mutation_timeout: DEFAULT_TOPIC_MUTATION_TIMEOUT,
135        }
136    }
137}
138
139impl AdminUiConfig {
140    /// # Errors
141    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
142    pub fn from_env() -> Result<Self, ConfigError> {
143        let mut cfg = Self::default();
144
145        if let Ok(raw) = std::env::var("KRABKA_ADMIN_UI_LISTEN_ADDR") {
146            cfg.listen_addr = raw
147                .parse()
148                .map_err(|_| ConfigError::InvalidListenAddr(raw.clone()))?;
149        }
150        if let Ok(name) = std::env::var("KRABKA_ADMIN_UI_CLUSTER_NAME") {
151            cfg.cluster_name = name;
152        }
153        if let Ok(addrs) = std::env::var("KRABKA_ADMIN_UI_BOOTSTRAP") {
154            cfg.bootstrap_addrs = addrs
155                .split(',')
156                .map(str::trim)
157                .map(ToOwned::to_owned)
158                .collect();
159        }
160        if let Ok(protocol) = std::env::var("KRABKA_ADMIN_UI_SECURITY_PROTOCOL") {
161            cfg.security = match protocol.as_str() {
162                "SASL_PLAINTEXT" => BrokerSecurityConfig::SaslPlaintext,
163                "SASL_SSL" => BrokerSecurityConfig::SaslSsl {
164                    trust_roots_pem: std::env::var_os("KRABKA_ADMIN_UI_TLS_TRUST_ROOTS_PEM")
165                        .map(PathBuf::from),
166                    server_name: std::env::var("KRABKA_ADMIN_UI_TLS_SERVER_NAME")
167                        .map_err(|_| ConfigError::MissingTlsServerName)?,
168                    client_identity: match (
169                        std::env::var_os("KRABKA_ADMIN_UI_TLS_CLIENT_CERT_PEM"),
170                        std::env::var_os("KRABKA_ADMIN_UI_TLS_CLIENT_KEY_PEM"),
171                    ) {
172                        (Some(cert), Some(key)) => Some((PathBuf::from(cert), PathBuf::from(key))),
173                        (None, None) => None,
174                        _ => return Err(ConfigError::IncompleteTlsClientIdentity),
175                    },
176                },
177                _ => return Err(ConfigError::InvalidSecurityProtocol),
178            };
179        }
180
181        cfg.validate()
182    }
183
184    /// # Errors
185    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
186    pub fn validate(self) -> Result<Self, ConfigError> {
187        if self.bootstrap_addrs.is_empty() {
188            return Err(ConfigError::MissingBootstrap);
189        }
190        for bootstrap_addr in &self.bootstrap_addrs {
191            validate_bootstrap_addr(bootstrap_addr)?;
192        }
193        if let BrokerSecurityConfig::SaslSsl { server_name, .. } = &self.security
194            && server_name.trim().is_empty()
195        {
196            return Err(ConfigError::MissingTlsServerName);
197        }
198
199        Ok(self)
200    }
201}
202
203fn validate_bootstrap_addr(addr: &str) -> Result<(), ConfigError> {
204    let Some((host, port)) = addr.rsplit_once(':') else {
205        return Err(ConfigError::InvalidBootstrapAddr(addr.to_string()));
206    };
207    if host.is_empty() || port.is_empty() {
208        return Err(ConfigError::InvalidBootstrapAddr(addr.to_string()));
209    }
210    if port.parse::<u16>().is_err() {
211        return Err(ConfigError::InvalidBootstrapAddr(addr.to_string()));
212    }
213
214    Ok(())
215}
216
217impl BrokerSecurityConfig {
218    #[must_use]
219    pub fn listener_protocol(&self) -> ListenerProtocol {
220        match self {
221            Self::SaslPlaintext => ListenerProtocol::SaslPlaintext,
222            Self::SaslSsl { .. } => ListenerProtocol::SaslSsl,
223        }
224    }
225
226    #[must_use]
227    pub fn tls(&self) -> Option<TlsConnectorConfig> {
228        match self {
229            Self::SaslPlaintext => None,
230            Self::SaslSsl {
231                trust_roots_pem,
232                server_name,
233                client_identity,
234            } => {
235                // `TlsConnectorConfig` caches its built rustls config in a
236                // private field, so it is built from its default.
237                let mut config = TlsConnectorConfig::default();
238                config.trust_store = trust_roots_pem
239                    .clone()
240                    .map_or(TrustStore::Platform, TrustStore::PemFile);
241                config.key_store =
242                    client_identity
243                        .clone()
244                        .map(|(certificate_chain, private_key)| KeyStore::PemFiles {
245                            certificate_chain,
246                            private_key,
247                            key_password: None,
248                        });
249                config.server_name.clone_from(server_name);
250                Some(config)
251            }
252        }
253    }
254}
krabka_admin_ui/config.rs

krabka_admin_ui/
config.rs
