krabka_client_core/
security.rs

1//! Client-side TLS/SASL security surface for [`crate::Client`].
2//!
3//! This module mirrors the TLS settings of Kafka's `SslConfigs` and the
4//! trust and key store handling of `DefaultSslEngineFactory`. The public
5//! clients and the inter-broker dialer negotiate the same way.
6
7mod stores;
8
9#[cfg(not(target_family = "wasm"))]
10use std::sync::OnceLock;
11use std::{
12    fmt, io,
13    path::PathBuf,
14    sync::{Arc, Mutex, PoisonError},
15};
16
17use krabka_security::ListenerProtocol;
18use rustls::{
19    CertificateError, DigitallySignedStruct, SignatureScheme, SupportedProtocolVersion,
20    client::{
21        ResolvesClientCert, WebPkiServerVerifier,
22        danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier},
23    },
24    crypto::{CryptoProvider, KeyProvider, ring::default_provider},
25    pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime},
26    sign::CertifiedKey,
27    version,
28};
29use thiserror::Error;
30use tokio_rustls::TlsConnector;
31
32pub use crate::sasl::SaslCredentials;
33
34/// Return the hostname from one Kafka `host:port` address.
35#[must_use]
36pub fn connection_target_host(address: &str) -> &str {
37    address.strip_prefix('[').map_or_else(
38        || address.rsplit_once(':').map_or(address, |(host, _)| host),
39        |bracketed| {
40            bracketed
41                .split_once(']')
42                .map_or(bracketed, |(host, _)| host)
43        },
44    )
45}
46
47/// A secret configuration value, such as Kafka's `ssl.key.password`.
48///
49/// `Debug` prints `[hidden]`, as Kafka's `Password.toString` does, so a
50/// logged configuration does not show the secret.
51#[derive(Clone, PartialEq, Eq)]
52pub struct Password(String);
53
54impl Password {
55    /// Wrap a secret value.
56    #[must_use]
57    pub fn new(value: impl Into<String>) -> Self {
58        Self(value.into())
59    }
60
61    /// Return the secret value.
62    #[must_use]
63    pub fn value(&self) -> &str {
64        &self.0
65    }
66}
67
68impl fmt::Debug for Password {
69    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
70        formatter.write_str("[hidden]")
71    }
72}
73
74impl From<&str> for Password {
75    fn from(value: &str) -> Self {
76        Self::new(value)
77    }
78}
79
80impl From<String> for Password {
81    fn from(value: String) -> Self {
82        Self(value)
83    }
84}
85
86/// The certificates that verify the broker certificate (Kafka
87/// `ssl.truststore.*`).
88#[derive(Clone, Debug, Default, PartialEq, Eq)]
89pub enum TrustStore {
90    /// The platform trust store. Kafka uses the JVM default trust store when
91    /// `ssl.truststore.location` and `ssl.truststore.certificates` are not
92    /// set (`DefaultSslEngineFactory.getTrustManagers` with a null
93    /// `KeyStore`). WebAssembly has no platform trust store, so there this
94    /// store fails to build with [`TlsConfigError::PlatformTrustStore`].
95    #[default]
96    Platform,
97    /// A PEM file of CA certificates (`ssl.truststore.type=PEM` with
98    /// `ssl.truststore.location`).
99    PemFile(PathBuf),
100    /// PEM CA certificates in the configuration
101    /// (`ssl.truststore.certificates`).
102    Pem(String),
103    /// A PKCS#12 trust store file (`ssl.truststore.type=PKCS12`). The client
104    /// trusts each certificate entry that has no private key.
105    Pkcs12 { path: PathBuf, password: Password },
106    /// A JKS trust store file (`ssl.truststore.type=JKS`). Without a password
107    /// the client does not check the store digest, as Java's
108    /// `KeyStore.load(stream, null)` does.
109    Jks {
110        path: PathBuf,
111        password: Option<Password>,
112    },
113}
114
115/// The client certificate chain and private key for mutual TLS (Kafka
116/// `ssl.keystore.*`).
117///
118/// A private key can be PKCS#8 (`PRIVATE KEY`) or PKCS#8 encrypted with
119/// PBES2 (`ENCRYPTED PRIVATE KEY`). An encrypted key needs `key_password`
120/// (`ssl.key.password`), as Kafka's `PemStore.privateKey` does.
121#[derive(Clone, Debug, PartialEq, Eq)]
122pub enum KeyStore {
123    /// One PEM file with the private key and the certificate chain
124    /// (`ssl.keystore.type=PEM` with `ssl.keystore.location`).
125    PemFile {
126        path: PathBuf,
127        key_password: Option<Password>,
128    },
129    /// A PEM certificate chain file and a separate PEM private key file.
130    PemFiles {
131        certificate_chain: PathBuf,
132        private_key: PathBuf,
133        key_password: Option<Password>,
134    },
135    /// A PEM certificate chain and private key in the configuration
136    /// (`ssl.keystore.certificate.chain` and `ssl.keystore.key`).
137    Pem {
138        certificate_chain: String,
139        private_key: String,
140        key_password: Option<Password>,
141    },
142    /// A PKCS#12 key store file (`ssl.keystore.type=PKCS12`). The store
143    /// password also decrypts the key.
144    Pkcs12 { path: PathBuf, password: Password },
145    /// A JKS key store file (`ssl.keystore.type=JKS`). `key_password`
146    /// decrypts the key. Without it, the store password decrypts the key, as
147    /// Kafka's `FileBasedStore` does when `ssl.key.password` is not set.
148    Jks {
149        path: PathBuf,
150        password: Password,
151        key_password: Option<Password>,
152    },
153}
154
155/// A TLS protocol version name of `ssl.protocol` and `ssl.enabled.protocols`.
156#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
157pub enum TlsVersion {
158    /// `TLSv1.2`.
159    Tls12,
160    /// `TLSv1.3`.
161    Tls13,
162}
163
164impl TlsVersion {
165    /// Parse a Kafka protocol name, `TLSv1.2` or `TLSv1.3`.
166    ///
167    /// # Errors
168    /// Returns [`TlsConfigError::UnsupportedProtocol`] for any other name,
169    /// such as `TLSv1.1`, which this client does not implement.
170    pub fn parse(name: &str) -> Result<Self, TlsConfigError> {
171        match name {
172            "TLSv1.2" => Ok(Self::Tls12),
173            "TLSv1.3" => Ok(Self::Tls13),
174            other => Err(TlsConfigError::UnsupportedProtocol(other.to_owned())),
175        }
176    }
177
178    const fn rustls(self) -> &'static SupportedProtocolVersion {
179        match self {
180            Self::Tls12 => &version::TLS12,
181            Self::Tls13 => &version::TLS13,
182        }
183    }
184}
185
186/// An invalid TLS configuration. Kafka raises these as an
187/// `InvalidConfigurationException` or a `KafkaException` when it builds the
188/// SSL engine factory.
189#[derive(Debug, Error)]
190#[non_exhaustive]
191pub enum TlsConfigError {
192    /// A store file could not be read.
193    #[error("cannot read {path}: {source}")]
194    Read {
195        path: PathBuf,
196        #[source]
197        source: io::Error,
198    },
199    /// A PEM value holds no certificate or no usable private key.
200    #[error("invalid PEM: {0}")]
201    Pem(String),
202    /// A private key could not be decrypted or decoded.
203    #[error("invalid private key: {0}")]
204    PrivateKey(String),
205    /// A PKCS#12 store could not be read.
206    #[error("invalid PKCS#12 store: {0}")]
207    Pkcs12(String),
208    /// A JKS store could not be read.
209    #[error("invalid JKS store: {0}")]
210    Jks(String),
211    /// The platform trust store holds no certificate.
212    #[error("the platform trust store holds no certificate: {0}")]
213    PlatformTrustStore(String),
214    /// A trust store holds no certificate that can be a trust anchor.
215    #[error("the trust store holds no usable CA certificate")]
216    EmptyTrustStore,
217    /// A protocol name is not `TLSv1.2` or `TLSv1.3`.
218    #[error("unsupported TLS protocol {0:?}")]
219    UnsupportedProtocol(String),
220    /// No enabled protocol is at or below `ssl.protocol`.
221    #[error("no enabled TLS protocol is at or below {0:?}")]
222    NoProtocol(TlsVersion),
223    /// A cipher suite name is not a TLS cipher suite name.
224    #[error("unknown cipher suite {0:?}")]
225    UnknownCipherSuite(String),
226    /// None of the configured cipher suites is available.
227    #[error("no configured cipher suite is supported: {0:?}")]
228    NoCipherSuite(Vec<String>),
229    /// rustls rejected the configuration.
230    #[error("TLS configuration: {0}")]
231    Rustls(String),
232}
233
234/// Client-side TLS settings, as Kafka's `SslConfigs` defines them.
235///
236/// `Default` gives Kafka's defaults: the platform trust store, no client
237/// certificate, hostname verification on, `ssl.protocol=TLSv1.3`,
238/// `ssl.enabled.protocols=TLSv1.2,TLSv1.3`, and the default cipher suites.
239#[derive(Clone, Debug, PartialEq, Eq)]
240pub struct TlsConnectorConfig {
241    /// The CA certificates that verify the broker certificate.
242    pub trust_store: TrustStore,
243    /// The client certificate and key for mutual TLS. `None` presents no
244    /// client certificate.
245    pub key_store: Option<KeyStore>,
246    /// Explicit SNI / server-name override. An empty string uses each
247    /// connection's target hostname, including brokers learned from metadata.
248    pub server_name: String,
249    /// Whether the client checks that the broker certificate names the host
250    /// it connects to. Kafka's `ssl.endpoint.identification.algorithm=https`
251    /// (the default) turns it on, and an empty value turns it off. With the
252    /// check off, the client still verifies the certificate chain.
253    pub hostname_verification: bool,
254    /// The highest protocol version that the client uses (`ssl.protocol`).
255    pub protocol: TlsVersion,
256    /// The protocol versions that the client offers
257    /// (`ssl.enabled.protocols`).
258    pub enabled_protocols: Vec<TlsVersion>,
259    /// The cipher suites that the client offers, by IANA name
260    /// (`ssl.cipher.suites`). `None` offers the default suites. The client
261    /// skips a known suite name that it does not implement, such as a CBC
262    /// suite.
263    pub cipher_suites: Option<Vec<String>>,
264    /// The configuration that [`Self::connector`] built last, shared by the
265    /// clones of this value.
266    built: BuiltConfig,
267}
268
269/// The last built client configuration and the settings that built it.
270///
271/// Kafka builds the SSL context once per client. The pool clones the
272/// settings for each broker, so a clone reuses the configuration while its
273/// settings, other than the server name, stay the same.
274#[derive(Clone, Default)]
275struct BuiltConfig(Arc<Mutex<Option<Built>>>);
276
277/// The settings of a build and the configuration that they gave.
278type Built = (TlsConnectorConfig, Arc<rustls::ClientConfig>);
279
280impl PartialEq for BuiltConfig {
281    fn eq(&self, _other: &Self) -> bool {
282        true
283    }
284}
285
286impl Eq for BuiltConfig {}
287
288impl fmt::Debug for BuiltConfig {
289    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
290        formatter.write_str("BuiltConfig")
291    }
292}
293
294impl Default for TlsConnectorConfig {
295    fn default() -> Self {
296        Self {
297            trust_store: TrustStore::Platform,
298            key_store: None,
299            server_name: String::new(),
300            hostname_verification: true,
301            protocol: TlsVersion::Tls13,
302            enabled_protocols: vec![TlsVersion::Tls12, TlsVersion::Tls13],
303            cipher_suites: None,
304            built: BuiltConfig::default(),
305        }
306    }
307}
308
309/// The platform trust store, loaded once for the process as the JVM loads its
310/// default trust store once.
311#[cfg(not(target_family = "wasm"))]
312fn platform_roots() -> Result<Arc<rustls::RootCertStore>, TlsConfigError> {
313    static ROOTS: OnceLock<Result<Arc<rustls::RootCertStore>, String>> = OnceLock::new();
314    ROOTS
315        .get_or_init(|| {
316            let loaded = rustls_native_certs::load_native_certs();
317            let mut roots = rustls::RootCertStore::empty();
318            let (_, ignored) = roots.add_parsable_certificates(loaded.certs);
319            if roots.is_empty() {
320                let errors = loaded
321                    .errors
322                    .iter()
323                    .map(ToString::to_string)
324                    .collect::<Vec<_>>();
325                return Err(format!(
326                    "{ignored} unparsable certificates, errors {errors:?}"
327                ));
328            }
329            Ok(Arc::new(roots))
330        })
331        .clone()
332        .map_err(TlsConfigError::PlatformTrustStore)
333}
334
335/// WebAssembly has no platform trust store, so a TLS connection there needs a
336/// configured trust store.
337#[cfg(target_family = "wasm")]
338fn platform_roots() -> Result<Arc<rustls::RootCertStore>, TlsConfigError> {
339    Err(TlsConfigError::PlatformTrustStore(
340        "WebAssembly has no platform trust store, so set ssl.truststore.location".to_owned(),
341    ))
342}
343
344impl TlsConnectorConfig {
345    /// Build a `rustls::ClientConfig`.
346    ///
347    /// # Errors
348    /// Returns [`TlsConfigError`] when a store does not load, when a protocol
349    /// or cipher suite setting leaves nothing to offer, or when the platform
350    /// trust store is empty or, on WebAssembly, absent.
351    pub fn build(&self) -> Result<Arc<rustls::ClientConfig>, TlsConfigError> {
352        self.build_with_platform_roots(platform_roots)
353    }
354
355    /// Build the client configuration. `platform` supplies the trust store
356    /// for [`TrustStore::Platform`], which lets a test use its own CA as the
357    /// platform store.
358    fn build_with_platform_roots(
359        &self,
360        platform: impl FnOnce() -> Result<Arc<rustls::RootCertStore>, TlsConfigError>,
361    ) -> Result<Arc<rustls::ClientConfig>, TlsConfigError> {
362        let roots = match &self.trust_store {
363            TrustStore::Platform => platform()?,
364            store => Arc::new(stores::trust_anchors(store)?),
365        };
366        let provider = Arc::new(self.crypto_provider()?);
367        let versions = self.protocol_versions()?;
368        let webpki = WebPkiServerVerifier::builder_with_provider(roots, Arc::clone(&provider))
369            .build()
370            .map_err(|error| TlsConfigError::Rustls(error.to_string()))?;
371        let builder = rustls::ClientConfig::builder_with_provider(provider)
372            .with_protocol_versions(&versions)
373            .map_err(|error| TlsConfigError::Rustls(error.to_string()))?;
374        let builder = if self.hostname_verification {
375            builder.with_webpki_verifier(webpki)
376        } else {
377            builder
378                .dangerous()
379                .with_custom_certificate_verifier(Arc::new(ChainOnlyVerifier { inner: webpki }))
380        };
381        let config = match &self.key_store {
382            Some(store) => {
383                let key_provider = builder.crypto_provider().key_provider;
384                let identities = stores::key_pairs(store)?
385                    .into_iter()
386                    .map(|(chain, key)| ClientIdentity::new(chain, key, key_provider))
387                    .collect::<Result<Vec<_>, _>>()?;
388                builder.with_client_cert_resolver(Arc::new(IssuerResolver { identities }))
389            }
390            None => builder.with_no_client_auth(),
391        };
392        Ok(Arc::new(config))
393    }
394
395    /// Build a ready `TlsConnector`.
396    ///
397    /// The first call builds the client configuration, and later calls on
398    /// this value or its clones reuse it while the settings stay the same. A
399    /// store file that changes after the first build therefore does not
400    /// change the connections, as a Kafka client reads its stores once.
401    ///
402    /// # Errors
403    /// Propagates [`Self::build`] failures.
404    pub fn connector(&self) -> Result<TlsConnector, TlsConfigError> {
405        let settings = Self {
406            server_name: String::new(),
407            built: BuiltConfig::default(),
408            ..self.clone()
409        };
410        let mut built = self.built.0.lock().unwrap_or_else(PoisonError::into_inner);
411        if let Some((built_settings, config)) = built.as_ref()
412            && *built_settings == settings
413        {
414            return Ok(TlsConnector::from(Arc::clone(config)));
415        }
416        let config = self.build()?;
417        *built = Some((settings, Arc::clone(&config)));
418        drop(built);
419        Ok(TlsConnector::from(config))
420    }
421
422    /// The enabled protocol versions at or below [`Self::protocol`].
423    fn protocol_versions(&self) -> Result<Vec<&'static SupportedProtocolVersion>, TlsConfigError> {
424        let mut enabled = self
425            .enabled_protocols
426            .iter()
427            .copied()
428            .filter(|version| *version <= self.protocol)
429            .collect::<Vec<_>>();
430        enabled.sort_unstable();
431        enabled.dedup();
432        if enabled.is_empty() {
433            return Err(TlsConfigError::NoProtocol(self.protocol));
434        }
435        Ok(enabled.into_iter().map(TlsVersion::rustls).collect())
436    }
437
438    /// The process crypto provider, limited to [`Self::cipher_suites`].
439    fn crypto_provider(&self) -> Result<CryptoProvider, TlsConfigError> {
440        let mut provider = CryptoProvider::get_default()
441            .map_or_else(default_provider, |provider| provider.as_ref().clone());
442        let Some(names) = &self.cipher_suites else {
443            return Ok(provider);
444        };
445        let mut wanted = Vec::with_capacity(names.len());
446        for name in names {
447            wanted.push(cipher_suite(name)?);
448        }
449        provider
450            .cipher_suites
451            .retain(|suite| wanted.contains(&suite.suite()));
452        if provider.cipher_suites.is_empty() {
453            return Err(TlsConfigError::NoCipherSuite(names.clone()));
454        }
455        Ok(provider)
456    }
457}
458
459/// Map an IANA cipher suite name, as Kafka's `ssl.cipher.suites` holds it, to
460/// the rustls suite. rustls names the TLS 1.3 suites `TLS13_*` where IANA
461/// names them `TLS_*`.
462fn cipher_suite(name: &str) -> Result<rustls::CipherSuite, TlsConfigError> {
463    let tls13 = name
464        .strip_prefix("TLS_")
465        .map(|rest| format!("TLS13_{rest}"));
466    (0..=u16::MAX)
467        .map(rustls::CipherSuite::from)
468        .find(|suite| {
469            suite
470                .as_str()
471                .is_some_and(|known| known == name || tls13.as_deref() == Some(known))
472        })
473        .ok_or_else(|| TlsConfigError::UnknownCipherSuite(name.to_owned()))
474}
475
476/// One client certificate chain and its signing key.
477#[derive(Debug)]
478struct ClientIdentity {
479    key: Arc<CertifiedKey>,
480    /// The DER issuer names of the certificates of the chain.
481    issuers: Vec<Vec<u8>>,
482}
483
484impl ClientIdentity {
485    fn new(
486        chain: Vec<CertificateDer<'static>>,
487        key: PrivateKeyDer<'static>,
488        key_provider: &dyn KeyProvider,
489    ) -> Result<Self, TlsConfigError> {
490        use x509_cert::der::{Decode as _, Encode as _};
491
492        let issuers = chain
493            .iter()
494            .map(|certificate| {
495                x509_cert::Certificate::from_der(certificate)
496                    .and_then(|certificate| certificate.tbs_certificate().issuer().to_der())
497                    .map_err(|error| TlsConfigError::Pem(format!("invalid certificate: {error}")))
498            })
499            .collect::<Result<Vec<_>, _>>()?;
500        let signing_key = key_provider
501            .load_private_key(key)
502            .map_err(|error| TlsConfigError::PrivateKey(error.to_string()))?;
503        Ok(Self {
504            key: Arc::new(CertifiedKey::new(chain, signing_key)),
505            issuers,
506        })
507    }
508}
509
510/// Pick the client identity that the broker can accept, as the `SunX509`
511/// key manager of Kafka's default `ssl.keymanager.algorithm` does in
512/// `chooseClientAlias`: the first key entry whose key type the broker allows
513/// and, when the broker names certificate authorities, whose chain has a
514/// certificate issued by one of them. With no match the client sends no
515/// certificate.
516#[derive(Debug)]
517struct IssuerResolver {
518    identities: Vec<ClientIdentity>,
519}
520
521impl ResolvesClientCert for IssuerResolver {
522    fn resolve(
523        &self,
524        root_hint_subjects: &[&[u8]],
525        sigschemes: &[SignatureScheme],
526    ) -> Option<Arc<CertifiedKey>> {
527        self.identities
528            .iter()
529            .find(|identity| {
530                identity.key.key.choose_scheme(sigschemes).is_some()
531                    && (root_hint_subjects.is_empty()
532                        || identity
533                            .issuers
534                            .iter()
535                            .any(|issuer| root_hint_subjects.contains(&issuer.as_slice())))
536            })
537            .map(|identity| Arc::clone(&identity.key))
538    }
539
540    fn has_certs(&self) -> bool {
541        !self.identities.is_empty()
542    }
543}
544
545/// A server certificate verifier that checks the chain and skips the host
546/// name, as Kafka's empty `ssl.endpoint.identification.algorithm` does.
547#[derive(Debug)]
548struct ChainOnlyVerifier {
549    inner: Arc<WebPkiServerVerifier>,
550}
551
552impl ServerCertVerifier for ChainOnlyVerifier {
553    fn verify_server_cert(
554        &self,
555        end_entity: &CertificateDer<'_>,
556        intermediates: &[CertificateDer<'_>],
557        server_name: &ServerName<'_>,
558        ocsp_response: &[u8],
559        now: UnixTime,
560    ) -> Result<ServerCertVerified, rustls::Error> {
561        // rustls-webpki checks the chain before the name, so a name error
562        // means that the chain is valid.
563        match self.inner.verify_server_cert(
564            end_entity,
565            intermediates,
566            server_name,
567            ocsp_response,
568            now,
569        ) {
570            Err(rustls::Error::InvalidCertificate(
571                CertificateError::NotValidForName | CertificateError::NotValidForNameContext { .. },
572            )) => Ok(ServerCertVerified::assertion()),
573            result => result,
574        }
575    }
576
577    fn verify_tls12_signature(
578        &self,
579        message: &[u8],
580        cert: &CertificateDer<'_>,
581        dss: &DigitallySignedStruct,
582    ) -> Result<HandshakeSignatureValid, rustls::Error> {
583        self.inner.verify_tls12_signature(message, cert, dss)
584    }
585
586    fn verify_tls13_signature(
587        &self,
588        message: &[u8],
589        cert: &CertificateDer<'_>,
590        dss: &DigitallySignedStruct,
591    ) -> Result<HandshakeSignatureValid, rustls::Error> {
592        self.inner.verify_tls13_signature(message, cert, dss)
593    }
594
595    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
596        self.inner.supported_verify_schemes()
597    }
598}
599
600/// Full client security policy: the listener protocol to speak, plus the TLS
601/// and SASL material it implies.
602///
603/// The `None` fields must match `protocol`. A `SaslSsl` policy needs both
604/// `tls` and `sasl`.
605#[derive(Debug, Clone)]
606pub struct ClientSecurity {
607    pub protocol: ListenerProtocol,
608    pub tls: Option<TlsConnectorConfig>,
609    pub sasl: Option<SaslCredentials>,
610    /// Canonical hostname for the SASL handshake. This is the GSSAPI service
611    /// principal host (`service_name/<sasl_host>`). It is meaningful whenever
612    /// the protocol [`requires_sasl`], independent of TLS: a
613    /// `SASL_PLAINTEXT` listener has no `tls` to source the host from, so
614    /// without this GSSAPI would fall back to `localhost` and Kerberos
615    /// would reject the principal. `None` falls back to a non-empty
616    /// `tls.server_name`, then the connection's target host. PLAIN/SCRAM
617    /// ignore it.
618    ///
619    /// [`requires_sasl`]: ListenerProtocol::requires_sasl
620    pub sasl_host: Option<String>,
621}
622
623impl ClientSecurity {
624    /// Clone this policy and fill its TLS server name from a connection target.
625    #[must_use]
626    pub fn for_target_host(&self, target_host: &str) -> Self {
627        let mut security = self.clone();
628        if let Some(tls) = &mut security.tls
629            && tls.server_name.is_empty()
630        {
631            target_host.clone_into(&mut tls.server_name);
632        }
633        security
634    }
635
636    /// Resolve the hostname handed to the SASL handshake, the GSSAPI SPN host.
637    ///
638    /// This method prefers the explicit [`Self::sasl_host`], then the TLS SNI
639    /// ([`TlsConnectorConfig::server_name`]), then the connection's target
640    /// `host` if known. If it knows none of them, it returns `"localhost"`.
641    #[must_use]
642    pub fn sasl_handshake_host<'a>(&'a self, target_host: Option<&'a str>) -> &'a str {
643        self.sasl_host
644            .as_deref()
645            .or_else(|| {
646                self.tls
647                    .as_ref()
648                    .map(|tls| tls.server_name.as_str())
649                    .filter(|name| !name.is_empty())
650            })
651            .or(target_host)
652            .unwrap_or("localhost")
653    }
654}
655
656#[cfg(test)]
657mod tests;
krabka_client_core/security.rs

krabka_client_core/
security.rs
