dex_sdk/client/
flow_service_tls.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Sustainable Use License 1.0.
4// You may not use this file except in compliance with the License.
5// See the LICENSE file in the repository root.
6//
7// SPDX-License-Identifier: LicenseRef-Sustainable-Use-1.0
8
9use std::error::Error;
10use std::net::{IpAddr, Ipv6Addr};
11
12use rustls::RootCertStore;
13use rustls::pki_types::{CertificateDer, pem::PemObject};
14use tonic::transport::{ClientTlsConfig, Endpoint};
15
16use crate::{SdkError, SdkResult, ServiceError};
17
18/// Enables verified TLS for a Client or Worker's outbound FlowService connection.
19///
20/// Pass independently to [`crate::ClientOptions::flow_service_tls`] and
21/// [`crate::WorkerOptions::flow_service_tls`]. Omitting these settings keeps plaintext gRPC.
22/// An empty options value enables TLS 1.2 or newer with native system roots. Additional roots
23/// extend that trust store. The endpoint must negotiate HTTP/2 through ALPN h2.
24/// The address must be `host:port`, `dns:///host:port`, or `dns://authority/host:port`.
25/// DNS authority syntax is accepted, but resolution uses the operating system resolver.
26/// Credentials remain bearer call metadata. These settings do not change an API-key token endpoint.
27///
28/// [`crate::Client::try_new`] and [`crate::Worker::try_new`] return
29/// [`SdkError::InvalidArgument`] for an invalid address, nonempty bundle without a valid root,
30/// or invalid server name. Native-root loading errors are logged; successfully loaded roots
31/// remain usable. An empty trust store constructs successfully but cannot verify a server.
32/// Connections open lazily: certificate, hostname, and handshake failures appear on the first
33/// Client call or [`crate::Worker::start`] as service errors.
34///
35/// ```no_run
36/// use std::sync::Arc;
37/// use dex_sdk::{BlobCache, BlobCacheConfig, Client, ClientOptions, FileCredential,
38///     FlowServiceTlsOptions, Registry, Worker, WorkerOptions};
39/// let cache = Arc::new(BlobCache::open(BlobCacheConfig::new(
40///     "/tmp/dex-tls-blobs", 64 * 1024 * 1024, 0,
41/// )?)?);
42/// let tls = FlowServiceTlsOptions::new()
43///     .additional_root_cas_pem(std::fs::read("/etc/dex/root-ca.pem")?);
44/// let credential = Arc::new(FileCredential::new("/var/run/secrets/dex/token")?);
45/// let registry = Registry::new();
46/// let worker = Worker::try_new(registry.clone(), cache.clone(), WorkerOptions::new()
47///     .server_address("dex.example.com:443").flow_service_tls(tls.clone())
48///     .credential(credential.clone()))?;
49/// let client = Client::try_new(registry, cache, ClientOptions::new()
50///     .server_address("dex.example.com:443").flow_service_tls(tls)
51///     .credential(credential))?;
52/// client.health_check()?;
53/// worker.start()?;
54/// # Ok::<(), Box<dyn std::error::Error>>(())
55/// ```
56#[derive(Clone, Default)]
57pub struct FlowServiceTlsOptions {
58    additional_root_cas_pem: Option<Vec<u8>>,
59    server_name: Option<String>,
60}
61
62impl FlowServiceTlsOptions {
63    /// Creates verified TLS options using native system roots and the address host.
64    ///
65    /// Equivalent to [`Self::default`]. No additional roots or server name override are set.
66    /// Pass these options to a Client or Worker as shown in the enclosing type's example.
67    pub fn new() -> Self {
68        Self::default()
69    }
70
71    /// Adds PEM-encoded CA certificates to the native system trust store.
72    ///
73    /// Defaults to no additional certificates. Empty bytes have the same effect.
74    /// Takes ownership of `value` and returns these options with that bundle. Construction reads
75    /// the bytes once; file changes are not watched. A nonempty bundle must contain at least
76    /// one parsable certificate, or construction returns [`SdkError::InvalidArgument`].
77    /// Other blocks and malformed certificates are skipped when another certificate parses.
78    /// Additional roots remain usable when the native store is empty or fails to load.
79    /// See the enclosing type's example.
80    pub fn additional_root_cas_pem(mut self, value: impl Into<Vec<u8>>) -> Self {
81        self.additional_root_cas_pem = Some(value.into());
82        self
83    }
84
85    /// Overrides the address host for certificate verification, SNI, and HTTP/2 authority.
86    ///
87    /// Defaults to verifying the address host and retaining its host and port as authority.
88    /// Takes ownership of `value` and returns these options with that override. Use a DNS name
89    /// or unbracketed IP address without a port. Empty or invalid names cause construction to
90    /// return [`SdkError::InvalidArgument`]. An IP address sends no SNI and requires a matching
91    /// IP SAN. An override becomes the authority without a port; a gateway must accept that name.
92    /// An IPv6 override is bracketed in `:authority` (for example, `[::1]`), while certificate
93    /// verification uses the bare address. See the enclosing type's example, adding
94    /// `.server_name("dex.example.com")` to the TLS options when a gateway requires that name.
95    pub fn server_name(mut self, value: impl Into<String>) -> Self {
96        self.server_name = Some(value.into());
97        self
98    }
99}
100
101impl std::fmt::Debug for FlowServiceTlsOptions {
102    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
103        formatter
104            .debug_struct("FlowServiceTlsOptions")
105            .field(
106                "additional_root_cas_pem_bytes",
107                &self.additional_root_cas_pem.as_ref().map(Vec::len),
108            )
109            .field("server_name", &self.server_name)
110            .finish()
111    }
112}
113
114pub(crate) fn flow_service_endpoint(
115    address: &str,
116    options: Option<&FlowServiceTlsOptions>,
117    operation: &'static str,
118) -> SdkResult<Endpoint> {
119    let Some(options) = options else {
120        let address = if address.contains("://") {
121            address.to_string()
122        } else {
123            format!("http://{address}")
124        };
125        return Endpoint::from_shared(address)
126            .map_err(|error| transport_error(operation, "build FlowService endpoint", error));
127    };
128    let endpoint_address = tls_endpoint_address(address)?;
129    let (host, _) = endpoint_address
130        .rsplit_once(':')
131        .expect("validated TLS endpoint has a port");
132    let host = host.trim_start_matches('[').trim_end_matches(']');
133    let mut tls = ClientTlsConfig::new().domain_name(host);
134    let mut roots = RootCertStore::empty();
135    if let Some(server_name) = &options.server_name {
136        if !is_valid_tls_name(server_name)
137            || rustls::pki_types::ServerName::try_from(server_name.as_str()).is_err()
138        {
139            return Err(invalid_tls(
140                "flow_service_tls.server_name must be a nonempty DNS name or IP address without a port",
141            ));
142        }
143        tls = tls.domain_name(server_name);
144    }
145    let mut skipped_added_count = 0;
146    if let Some(bundle) = &options.additional_root_cas_pem {
147        let (added_count, skipped_count) = add_additional_certificates(&mut roots, bundle);
148        skipped_added_count = skipped_count;
149        if !bundle.is_empty() && added_count == 0 {
150            return Err(invalid_tls(
151                "flow_service_tls.additional_root_cas_pem contains no PEM certificate",
152            ));
153        }
154    }
155    let mut endpoint = Endpoint::from_shared(format!("https://{endpoint_address}"))
156        .map_err(|error| transport_error(operation, "build FlowService endpoint", error))?;
157    if let Some(server_name) = &options.server_name {
158        let authority = if server_name.parse::<Ipv6Addr>().is_ok() {
159            format!("[{server_name}]")
160        } else {
161            server_name.clone()
162        };
163        endpoint = endpoint.origin(format!("https://{authority}").parse().map_err(|_| {
164            invalid_tls("flow_service_tls.server_name is not a valid HTTP/2 authority")
165        })?);
166    }
167    let native = rustls_native_certs::load_native_certs();
168    if !native.errors.is_empty() {
169        tracing::warn!(
170            error_count = native.errors.len(),
171            "failed to load some native root CAs"
172        );
173    }
174    let (_, skipped_count) = roots.add_parsable_certificates(native.certs);
175    if skipped_count > 0 {
176        tracing::debug!(skipped_count, "skipped invalid native root CAs");
177    }
178    if skipped_added_count > 0 {
179        tracing::warn!(
180            skipped_count = skipped_added_count,
181            "skipped invalid additional root CA PEM blocks"
182        );
183    }
184    if roots.is_empty() {
185        tracing::warn!("FlowService TLS trust store is empty; server verification will fail");
186    }
187    endpoint
188        .tls_config(tls.trust_anchors(roots.roots))
189        .map_err(|error| transport_error(operation, "build FlowService TLS connector", error))
190}
191
192fn tls_endpoint_address(address: &str) -> SdkResult<&str> {
193    let unsupported = || {
194        invalid_tls(
195            "flow_service_tls requires server_address as host:port, dns:///host:port, or dns://authority/host:port",
196        )
197    };
198    let endpoint = if let Some((scheme, target)) = address.split_once("://") {
199        if !scheme.eq_ignore_ascii_case("dns") {
200            return Err(unsupported());
201        }
202        let (authority, endpoint) = target.split_once('/').ok_or_else(unsupported)?;
203        if authority.contains([' ', '\t', '\r', '\n', '?', '#', '@']) {
204            return Err(unsupported());
205        }
206        endpoint
207    } else {
208        address
209    };
210    if endpoint.contains(['/', '?', '#']) {
211        return Err(unsupported());
212    }
213    let (host, port) = endpoint.rsplit_once(':').ok_or_else(unsupported)?;
214    let host = if let Some(bracketed_host) = host.strip_prefix('[') {
215        let ipv6 = bracketed_host.strip_suffix(']').ok_or_else(unsupported)?;
216        ipv6.parse::<Ipv6Addr>().map_err(|_| unsupported())?;
217        ipv6
218    } else {
219        if host.contains([':', '[', ']']) {
220            return Err(unsupported());
221        }
222        host
223    };
224    if rustls::pki_types::ServerName::try_from(host).is_err() {
225        return Err(invalid_tls(
226            "flow_service_tls server_address host is not a valid DNS name or IP address",
227        ));
228    }
229    if !is_valid_tls_name(host)
230        || [
231            "dns",
232            "unix",
233            "unix-abstract",
234            "passthrough",
235            "ipv4",
236            "ipv6",
237        ]
238        .contains(&host.to_ascii_lowercase().as_str())
239    {
240        return Err(unsupported());
241    }
242    if port.is_empty()
243        || !port.bytes().all(|value| value.is_ascii_digit())
244        || !port.parse::<u16>().is_ok_and(|value| value > 0)
245    {
246        return Err(invalid_tls(
247            "flow_service_tls requires server_address port 1-65535",
248        ));
249    }
250    Ok(endpoint)
251}
252
253fn is_valid_tls_name(name: &str) -> bool {
254    !name.is_empty()
255        && !name.contains([' ', '\t', '\r', '\n', '/', '#', '?', '@', '[', ']'])
256        && (!name.contains(':') || name.parse::<IpAddr>().is_ok())
257}
258
259fn add_additional_certificates(roots: &mut RootCertStore, bundle: &[u8]) -> (usize, usize) {
260    let bundle = String::from_utf8_lossy(bundle);
261    let mut added_count = 0;
262    let mut skipped_count = 0;
263    for block in bundle.split("-----BEGIN CERTIFICATE-----").skip(1) {
264        let Some((body, _)) = block.split_once("-----END CERTIFICATE-----") else {
265            skipped_count += 1;
266            continue;
267        };
268        let certificate = format!("-----BEGIN CERTIFICATE-----{body}-----END CERTIFICATE-----");
269        match CertificateDer::from_pem_slice(certificate.as_bytes()) {
270            Ok(parsed) => match roots.add(parsed.into_owned()) {
271                Ok(()) => added_count += 1,
272                Err(_) => skipped_count += 1,
273            },
274            Err(_) => skipped_count += 1,
275        }
276    }
277    (added_count, skipped_count)
278}
279
280fn invalid_tls(message: &str) -> SdkError {
281    SdkError::InvalidArgument {
282        message: message.to_string(),
283    }
284}
285
286fn transport_error(operation: &'static str, object: &str, error: impl Error + 'static) -> SdkError {
287    let mut detail = format!("{object}: {error}");
288    let mut source = error.source();
289    while let Some(cause) = source {
290        detail.push_str(": ");
291        detail.push_str(&cause.to_string());
292        source = cause.source();
293    }
294    SdkError::Service {
295        service: ServiceError::local(operation, detail),
296    }
297}
dex_sdk/client/flow_service_tls.rs

dex_sdk/client/
flow_service_tls.rs
