crypto: migrate reqwest from native-tls to rustls across workspace

Cargo.toml

@@ -321,24 +321,39 @@ tokio-postgres = { git = "https://github.com/MaterializeInc/rust-postgres" }
 postgres-protocol = { git = "https://github.com/MaterializeInc/rust-postgres" }
 postgres-replication = { git = "https://github.com/MaterializeInc/rust-postgres" }
 postgres-types = { git = "https://github.com/MaterializeInc/rust-postgres" }
-postgres-openssl = { git = "https://github.com/MaterializeInc/rust-postgres" }
 postgres_array = { git = "https://github.com/MaterializeInc/rust-postgres-array" }
 
 # Waiting on https://github.com/MaterializeInc/serde-value/pull/35.
 serde-value = { git = "https://github.com/MaterializeInc/serde-value.git" }
 
-# Waiting for resolution of https://github.com/launchdarkly/rust-server-sdk/issues/116
-launchdarkly-server-sdk = { git = "https://github.com/MaterializeInc/rust-server-sdk", rev = "3e0a0b98b09a2970f292577a07e1c9382b65b5da" }
+# Add enable_reqwest_rustls_no_provider feature to avoid pulling in ring,
+# which conflicts with aws-lc-fips-sys in FIPS builds.
+# See https://github.com/Azure/azure-sdk-for-rust/issues/1680
+azure_core = { git = "https://github.com/MaterializeInc/azure-sdk-for-rust.git", branch = "mz/enable-reqwest-rustls-no-provider" }
+azure_identity = { git = "https://github.com/MaterializeInc/azure-sdk-for-rust.git", branch = "mz/enable-reqwest-rustls-no-provider" }
+azure_storage = { git = "https://github.com/MaterializeInc/azure-sdk-for-rust.git", branch = "mz/enable-reqwest-rustls-no-provider" }
+azure_storage_blobs = { git = "https://github.com/MaterializeInc/azure-sdk-for-rust.git", branch = "mz/enable-reqwest-rustls-no-provider" }
+azure_svc_blobstorage = { git = "https://github.com/MaterializeInc/azure-sdk-for-rust.git", branch = "mz/enable-reqwest-rustls-no-provider" }
+
+# Upstream hardcodes hyper-rustls/ring. Switch to hyper-rustls/aws-lc-rs to
+# avoid duplicate bignum symbols with aws-lc-fips-sys in FIPS builds.
+launchdarkly-sdk-transport = { git = "https://github.com/MaterializeInc/rust-sdk-transport.git", branch = "mz/aws-lc-rs-instead-of-ring" }
 
 # Waiting on https://github.com/edenhill/librdkafka/pull/4051.
-# Also: jasonhernandez/ssl-awslc branch adds ssl-awslc feature for FIPS.
+# Branch adds ssl-awslc feature to build librdkafka against aws-lc instead of
+# vendored OpenSSL, eliminating duplicate crypto library symbols.
 rdkafka = { git = "https://github.com/MaterializeInc/rust-rdkafka.git", branch = "jasonhernandez/ssl-awslc" }
 rdkafka-sys = { git = "https://github.com/MaterializeInc/rust-rdkafka.git", branch = "jasonhernandez/ssl-awslc" }
 
+# Fix rustls backend: PKCS#8 key parsing and skip_domain_validation string match.
+# All changes should go to the `rustls-fixes` branch.
+mysql_async = { git = "https://github.com/MaterializeInc/mysql_async", branch = "rustls-fixes" }
+
 # Need to upstream a few PRs related to test builders.
 #
 # Note: All changes in our fork of tiberius should be pushed to the `mz_changes` branch.
-tiberius = { git = "https://github.com/MaterializeInc/tiberius", rev="64ca594cc22ed67d072c2d0110455da50539e1cd" }
+# SEC-248: rustls 0.23 + aws-lc-rs migration on feature branch (not yet merged to mz_changes).
+tiberius = { git = "https://github.com/MaterializeInc/tiberius", branch = "rustls-0.23-on-mz-changes" }
 
 # Allows us to use bzip2-sys rather than the rust reimpl.
 # All changes should go to the `mz_changes` branch.
@@ -347,15 +362,18 @@ async-compression = { git = "https://github.com/MaterializeInc/async-compression
 
 # Custom iceberg features for mz
 # All changes should go to the `mz_changes` branch.
-iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "c31a98afe789" }
-iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "c31a98afe789" }
+# SEC-239: reqwest native-tls -> rustls on feature branch (not yet merged to mz_changes).
+iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", branch = "jasonhernandez/rustls-migration" }
+iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", branch = "jasonhernandez/rustls-migration" }
 
 # Custom duckdb crate to support mz needs
 # All changes should go to the `mz_changes` branch.
 # The main change is allowing the TLS implementation to be selected via features.
 # Hopefully we can upstream these changes eventually.
 # Additionally we keep the apache crates updated to latest versions.
-duckdb = { git = "https://github.com/MaterializeInc/duckdb-rs.git", rev = "752c7efe2582" }
+# Branch mz/rustls-tls-no-provider changes libduckdb-sys default from
+# reqwest/rustls-tls (ring) to reqwest/rustls-tls-webpki-roots-no-provider (aws-lc-rs).
+duckdb = { git = "https://github.com/MaterializeInc/duckdb-rs.git", branch = "mz/rustls-tls-no-provider" }
 
 
 # BEGIN LINT CONFIG

ci/test/cargo-test/mzcompose.py

@@ -334,7 +334,12 @@ def run_sanitizer(
 def run_cargo_nextest(
     c: Composition, args: Namespace, env: dict[str, str], metadata: Any
 ) -> None:
-    # Common args for all nextest runs
+    # The `fips` and `crypto` features on mz-ore are mutually exclusive at
+    # link time (aws-lc-fips-sys vs aws-lc-sys have duplicate symbols).
+    # We can't use --all-features because it activates both.
+    # Instead: run all packages with --all-features but exclude mz-ore,
+    # then separately test mz-ore twice — once with all non-fips features,
+    # once with fips. This ensures every feature is tested.
     nextest_common_args = [
         "--all-features",
         "--cargo-profile=ci",

deny.toml

@@ -148,8 +148,25 @@ skip = [
     { name = "toml_edit", version = "0.22.27" },
     { name = "webpki-roots", version = "0.26.11" },
     { name = "winnow", version = "0.7.15" },
+
 ]
 
+# Use rustls + aws-lc-rs instead. native-tls pulls in OpenSSL which causes
+# symbol collisions with aws-lc-rs and prevents FIPS compliance.
+[[bans.deny]]
+name = "native-tls"
+
+# Use rustls + aws-lc-rs instead. hyper-tls depends on native-tls.
+[[bans.deny]]
+name = "hyper-tls"
+
+# Use aws-lc-rs (via rdkafka ssl-awslc) instead of vendored OpenSSL.
+[[bans.deny]]
+name = "openssl-sys"
+
+[[bans.deny]]
+name = "openssl-src"
+
 [[bans.deny]]
 crate = "crossbeam-channel@0.5.14"
 reason = "memory corruption, https://github.com/MaterializeInc/database-issues/issues/9091"
@@ -199,7 +216,7 @@ wrappers = [
     "hyper-rustls",
     "launchdarkly-server-sdk",
     "launchdarkly-server-sdk-evaluation",
-    "native-tls",
+    "launchdarkly-sdk-transport",
     "opendal",
     "os_info",
     "postgres",
@@ -282,13 +299,6 @@ wrappers = [
 # Use `aws_lc_rs` instead of `ring` — ring is not FIPS-validated.
 [[bans.deny]]
 name = "ring"
-wrappers = [
-    # Third-party crate — TODO: track upstream migration.
-    "aws-config",
-    # TODO: remove tokio-postgres-rustls dep from environmentd tests
-    # and use mz-tls-util::MakeRustlsConnect instead.
-    "tokio-postgres-rustls",
-]
 
 # Use `aws_lc_rs::pbkdf2` instead.
 [[bans.deny]]

src/adapter/Cargo.toml

@@ -28,10 +28,9 @@ governor = "0.10.1"
 hex = "0.4.3"
 imbl = { version = "7.0.0", features = ["serde"] }
 http = "1.4.0"
-hyper-tls = "0.5.0"
 ipnet = "2.12.0"
 itertools = "0.14.0"
-launchdarkly-server-sdk = { version = "2.6.2", default-features = false, optional = true }
+launchdarkly-server-sdk = { version = "3.0.1", default-features = false, features = ["hyper-rustls-webpki-roots", "crypto-aws-lc-rs"], optional = true }
 maplit = "1.0.2"
 mz-adapter-types = { path = "../adapter-types" }
 mz-audit-log = { path = "../audit-log" }

src/adapter/src/config.rs

@@ -91,7 +91,10 @@ impl SystemParameterSyncConfig {
 
 #[derive(Debug, Clone)]
 pub(super) struct Metrics {
+    // TODO(SEC-259): restore metric callbacks lost in LD SDK 3.0 upgrade.
+    #[allow(dead_code)]
     pub last_cse_time_seconds: UIntGauge,
+    #[allow(dead_code)]
     pub last_sse_time_seconds: UIntGauge,
     pub params_changed: IntCounter,
 }

src/adapter/src/config/frontend.rs

@@ -10,13 +10,10 @@
 use std::collections::BTreeMap;
 use std::fs;
 use std::path::PathBuf;
-use std::sync::Arc;
 use std::time::Duration;
 
 use derivative::Derivative;
 #[cfg(feature = "telemetry")]
-use hyper_tls::HttpsConnector;
-#[cfg(feature = "telemetry")]
 use launchdarkly_server_sdk as ld;
 use mz_build_info::BuildInfo;
 use mz_cloud_provider::CloudProvider;
@@ -68,7 +65,7 @@ impl SystemParameterFrontend {
     /// Create a new [SystemParameterFrontend] initialize.
     ///
     /// This will create and initialize an [ld::Client] instance. The
-    /// [ld::Client::initialized_async] call will be attempted in a loop with an
+    /// [ld::Client::wait_for_initialization] call will be attempted in a loop with an
     /// exponential backoff with power `2s` and max duration `60s`.
     pub async fn from(sync_config: &SystemParameterSyncConfig) -> Result<Self, anyhow::Error> {
         match &sync_config.backend_config {
@@ -154,24 +151,11 @@ impl SystemParameterFrontend {
 
 #[cfg(feature = "telemetry")]
 fn ld_config(api_key: &str, metrics: &Metrics) -> ld::Config {
+    // TODO: re-add on_success callback for last_cse_time_seconds metric once
+    // we can reference hyper-rustls 0.24 types for EventProcessorBuilder<C>.
+    // The LD SDK with the `rustls` feature creates default hyper-rustls connectors.
+    let _ = metrics;
     ld::ConfigBuilder::new(api_key)
-        .event_processor(
-            ld::EventProcessorBuilder::new()
-                .https_connector(HttpsConnector::new())
-                .on_success({
-                    let last_cse_time_seconds = metrics.last_cse_time_seconds.clone();
-                    Arc::new(move |result| {
-                        if let Ok(ts) = u64::try_from(result.time_from_server / 1000) {
-                            last_cse_time_seconds.set(ts);
-                        } else {
-                            tracing::warn!(
-                                "Cannot convert time_from_server / 1000 from u128 to u64"
-                            );
-                        }
-                    })
-                }),
-        )
-        .data_source(ld::StreamingDataSourceBuilder::new().https_connector(HttpsConnector::new()))
         .build()
         .expect("valid config")
 }
@@ -187,14 +171,11 @@ async fn ld_client(
     // Start and initialize LD client for the frontend. The callback passed
     // will export the last time when an SSE event from the LD server was
     // received in a Prometheus metric.
-    ld_client.start_with_default_executor_and_callback({
-        let last_sse_time_seconds = metrics.last_sse_time_seconds.clone();
-        let now_fn = now_fn.clone();
-        Arc::new(move |_ev| {
-            let ts = now_fn() / 1000;
-            last_sse_time_seconds.set(ts);
-        })
-    });
+    // TODO: The 3.0 SDK removed start_with_default_executor_and_callback.
+    // Re-add SSE timestamp metric (last_sse_time_seconds) once the SDK
+    // exposes an event callback or observable connection state.
+    let _ = (metrics, now_fn);
+    ld_client.start_with_default_executor();
 
     let max_backoff = Duration::from_secs(60);
     let mut backoff = Duration::from_secs(5);

src/adapter/src/coord/sequencer/inner.rs

@@ -30,6 +30,7 @@ use mz_expr::{
 };
 use mz_ore::cast::CastFrom;
 use mz_ore::collections::{CollectionExt, HashSet};
+use mz_ore::future::OreFutureExt;
 use mz_ore::task::{self, JoinHandle, spawn};
 use mz_ore::tracing::OpenTelemetryContext;
 use mz_ore::{assert_none, instrument, soft_assert_or_log};
@@ -719,12 +720,20 @@ impl Coordinator {
 
             let current_storage_parameters = self.controller.storage.config().clone();
             task::spawn(|| format!("validate_connection:{conn_id}"), async move {
-                let result = match connection
-                    .validate(connection_id, &current_storage_parameters)
-                    .await
+                let result = match std::panic::AssertUnwindSafe(
+                    connection.validate(connection_id, &current_storage_parameters),
+                )
+                .ore_catch_unwind()
+                .await
                 {
-                    Ok(()) => Ok(plan),
-                    Err(err) => Err(err.into()),
+                    Ok(Ok(())) => Ok(plan),
+                    Ok(Err(err)) => Err(err.into()),
+                    Err(_panic) => {
+                        tracing::error!("connection validation panicked");
+                        Err(AdapterError::Internal(
+                            "connection validation panicked".into(),
+                        ))
+                    }
                 };
 
                 // It is not an error for validation to complete after `internal_cmd_rx` is dropped.
@@ -3668,9 +3677,20 @@ impl Coordinator {
                 async move {
                     let resolved_ids = conn.resolved_ids.clone();
                     let dependency_ids: BTreeSet<_> = resolved_ids.items().copied().collect();
-                    let result = match connection.validate(id, &current_storage_parameters).await {
-                        Ok(()) => Ok(conn),
-                        Err(err) => Err(err.into()),
+                    let result = match std::panic::AssertUnwindSafe(
+                        connection.validate(id, &current_storage_parameters),
+                    )
+                    .ore_catch_unwind()
+                    .await
+                    {
+                        Ok(Ok(())) => Ok(conn),
+                        Ok(Err(err)) => Err(err.into()),
+                        Err(_panic) => {
+                            tracing::error!("alter connection validation panicked");
+                            Err(AdapterError::Internal(
+                                "connection validation panicked".into(),
+                            ))
+                        }
                     };
 
                     // It is not an error for validation to complete after `internal_cmd_rx` is dropped.

src/authenticator/Cargo.toml

@@ -15,7 +15,7 @@ mz-auth = { path = "../auth", default-features = false }
 mz-frontegg-auth = { path = "../frontegg-auth", default-features = false }
 mz-ore = { path = "../ore", features = ["assert"] }
 mz-pgwire-common = { path = "../pgwire-common", default-features = false }
-reqwest = "0.12.24"
+reqwest = { version = "0.12.28", default-features = false, features = ["rustls-tls-webpki-roots-no-provider"] }
 tokio-postgres = { version = "0.7.15" }
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.149"

src/balancerd/Cargo.toml

@@ -23,7 +23,7 @@ humantime = "2.3.0"
 hyper = { version = "1.4.1", features = ["http1", "server"] }
 hyper-util = "0.1.20"
 jsonwebtoken = { version = "10.3.0", features = ["aws_lc_rs"] }
-launchdarkly-server-sdk = { version = "2.6.2", default-features = false, optional = true }
+launchdarkly-server-sdk = { version = "3.0.1", default-features = false, features = ["hyper-rustls-webpki-roots", "crypto-aws-lc-rs"], optional = true }
 mz-alloc = { path = "../alloc" }
 mz-alloc-default = { path = "../alloc-default", optional = true }
 mz-build-info = { path = "../build-info" }
@@ -56,7 +56,7 @@ uuid = "1.19.0"
 mz-environmentd = { path = "../environmentd", default-features = false, features = ["test"] }
 mz-frontegg-mock = { path = "../frontegg-mock" }
 postgres = "0.19.12"
-reqwest = "0.12.28"
+reqwest = { version = "0.12.28", default-features = false, features = ["rustls-tls-webpki-roots-no-provider"] }
 tempfile = "3.23.0"
 
 [features]

src/balancerd/tests/server.rs

@@ -46,6 +46,7 @@ use uuid::Uuid;
 #[mz_ore::test(tokio::test(flavor = "multi_thread", worker_threads = 1))]
 #[cfg_attr(miri, ignore)] // too slow
 async fn test_balancer() {
+    let _ = mz_ore::crypto::fips_crypto_provider();
     let ca = Ca::new_root("test ca").unwrap();
     let (server_cert, server_key) = ca
         .request_cert("server", vec![IpAddr::V4(Ipv4Addr::LOCALHOST)])

src/ccsr/Cargo.toml

@@ -12,10 +12,9 @@ workspace = true
 [dependencies]
 anyhow = "1.0.102"
 base64 = "0.22.1"
-reqwest = { version = "0.12.28", features = [
+reqwest = { version = "0.12.28", default-features = false, features = [
     "blocking",
     "json",
-    "native-tls-vendored",
     "rustls-tls-webpki-roots-no-provider",
 ] }
 mz-tls-util = { path = "../tls-util" }

src/ccsr/src/config.rs

@@ -125,7 +125,10 @@ impl ClientConfig {
             .redirect(reqwest::redirect::Policy::none())
             .timeout(timeout)
             .build()
-            .unwrap();
+            .map_err(|e| {
+                // Use {e:#} to include the full error chain (not just the top-level "builder error").
+                anyhow::anyhow!("failed to build schema registry HTTP client: {e:#}")
+            })?;
 
         Client::new(inner, self.url, self.auth, timeout)
     }

src/ccsr/src/tls.rs

@@ -28,22 +28,30 @@ impl Identity {
     pub fn from_pem(key: &[u8], cert: &[u8]) -> Result<Self, anyhow::Error> {
         let mut archive = pkcs12der_from_pem(key, cert)
             .map_err(|e| anyhow::anyhow!("failed to build PKCS#12 identity: {e}"))?;
+        // Also validate that reqwest can parse the PEM identity, since the
+        // From<Identity> conversion uses expect() and must not panic.
+        reqwest::Identity::from_pem(&archive.der)
+            .map_err(|e| anyhow::anyhow!("failed to build reqwest identity: {e}"))?;
         Ok(Identity {
             der: std::mem::take(&mut archive.der),
             pass: std::mem::take(&mut archive.pass),
         })
     }
 
-    /// Wraps [`reqwest::Identity::from_pkcs12_der`].
+    /// Constructs an identity from PEM-encoded key+cert data.
+    ///
+    /// The `der` field stores the raw PEM bytes, `pass` is unused (kept for
+    /// backward compatibility with serialized data).
     pub fn from_pkcs12_der(der: Vec<u8>, pass: String) -> Result<Self, reqwest::Error> {
-        let _ = reqwest::Identity::from_pkcs12_der(&der, &pass)?;
+        // Validate by trying to construct a reqwest Identity.
+        let _ = reqwest::Identity::from_pem(&der)?;
         Ok(Identity { der, pass })
     }
 }
 
 impl From<Identity> for reqwest::Identity {
     fn from(id: Identity) -> Self {
-        reqwest::Identity::from_pkcs12_der(&id.der, &id.pass).expect("known to be a valid identity")
+        reqwest::Identity::from_pem(&id.der).expect("known to be a valid identity")
     }
 }
 

src/cloud-api/Cargo.toml

@@ -12,7 +12,7 @@ workspace = true
 [dependencies]
 anyhow = "1.0.102"
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }
-reqwest = { version = "0.12.28", features = ["json"] }
+reqwest = { version = "0.12.28", default-features = false, features = ["json", "rustls-tls-webpki-roots-no-provider"] }
 serde = { version = "1.0.219", features = ["derive"] }
 url = "2.5.8"
 thiserror = "2.0.18"

src/cloud-resources/Cargo.toml

@@ -19,7 +19,7 @@ chrono = { version = "0.4.39", default-features = false }
 futures = "0.3.32"
 indexmap = { version = "2.10.0", default-features = false, features = ["std"] }
 k8s-openapi = { version = "0.27.0", features = ["schemars", "v1_32"] }
-kube = { version = "3.0.1", default-features = false, features = ["client", "derive", "rustls-tls", "ws", "runtime"] }
+kube = { version = "3.0.1", default-features = false, features = ["client", "derive", "rustls-tls", "aws-lc-rs", "ws", "runtime"] }
 mz-ore = { path = "../ore", default-features = false, features = ["async"] }
 mz-server-core = { path = "../server-core", default-features = false }
 rand = "0.9.2"