From 7fdf4e892e20e8f4a12b824a37da8580231c285d Mon Sep 17 00:00:00 2001 From: Jay Zhu Date: Wed, 8 Jul 2026 14:42:57 -0600 Subject: [PATCH 1/2] feat(health): add switch mTLS support Signed-off-by: Jay Zhu --- Cargo.lock | 4 + crates/health/Cargo.toml | 6 +- crates/health/example/config.example.toml | 40 +- crates/health/src/collectors/mod.rs | 2 +- crates/health/src/collectors/nmxc.rs | 66 +- crates/health/src/collectors/nmxt.rs | 144 +++- .../health/src/collectors/nvue/gnmi/client.rs | 112 ++- .../src/collectors/nvue/gnmi/subscriber.rs | 43 +- .../health/src/collectors/nvue/rest/client.rs | 124 ++- .../src/collectors/nvue/rest/collector.rs | 22 +- crates/health/src/config.rs | 335 +++++++- crates/health/src/discovery/context.rs | 45 +- crates/health/src/discovery/spawn.rs | 4 + crates/health/src/endpoint/model.rs | 29 +- crates/health/src/lib.rs | 25 +- crates/health/src/tls.rs | 722 ++++++++++++++++++ 16 files changed, 1604 insertions(+), 119 deletions(-) create mode 100644 crates/health/src/tls.rs diff --git a/Cargo.lock b/Cargo.lock index 4ab226fd88..ae1cd47dfb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2006,14 +2006,17 @@ dependencies = [ "prost", "prost-types", "rand 0.10.1", + "rcgen", "reqwest 0.13.4", "rustls", + "rustls-pemfile", "rustls-pki-types", "serde", "serde_json", "serde_with", "tempfile", "thiserror 2.0.18", + "time", "tokio", "tokio-stream", "tokio-util", @@ -2023,6 +2026,7 @@ dependencies = [ "tracing", "tracing-subscriber", "url", + "x509-parser", ] [[package]] diff --git a/crates/health/Cargo.toml b/crates/health/Cargo.toml index 164ac24f86..e6cd49b79e 100644 --- a/crates/health/Cargo.toml +++ b/crates/health/Cargo.toml @@ -43,8 +43,9 @@ hyper-rustls = { workspace = true, features = ["http2"] } hyper-util = { workspace = true } mac_address = { workspace = true } prometheus = { workspace = true } -reqwest = { workspace = true, features = ["query", "json"] } +reqwest = { workspace = true, features = ["query", "json", "rustls"] } rustls = { workspace = true } +rustls-pemfile = { workspace = true } rustls-pki-types = { workspace = true } serde = { features = ["derive"], workspace = true } serde_json = { workspace = true } @@ -80,6 +81,7 @@ nv-redfish = { workspace = true, features = [ ] } rand = { workspace = true } url = { workspace = true, features = ["serde"] } +x509-parser = { workspace = true } # [local-dependencies] carbide-health-report = { path = "../health-report" } @@ -100,7 +102,9 @@ bench-hooks = [] [dev-dependencies] carbide-test-support = { path = "../test-support" } criterion = { workspace = true } +rcgen = { workspace = true } tempfile = { workspace = true } +time = { workspace = true } [[bench]] name = "sink_pipeline" diff --git a/crates/health/example/config.example.toml b/crates/health/example/config.example.toml index 058dae1c9c..9a892909f0 100644 --- a/crates/health/example/config.example.toml +++ b/crates/health/example/config.example.toml @@ -208,18 +208,50 @@ logs_collection_interval = "5m" state_refresh_interval = "30m" logs_state_file = "/tmp/logs_collector_{machine_id}.json" +# ============================================================================== +# TLS: Shared certificate configuration for hardware-health connections +# ============================================================================== + +# Optional mTLS profile for direct switch collector connections. +# This section is shared by switch collectors, but it is not itself a collector. +# This is intentionally separate from Carbide API / SPIFFE certificate paths. +# When configured, NVUE REST, NVUE gNMI, NMX-T, and NMX-C use this CA bundle, +# client certificate, and client key for switch host connections. +# +# Switch server certificates must be valid for the TLS server name used by +# health. Collectors always open TCP connections to discovered switch IPs. If +# switch certificates contain DNS SANs instead of IP SANs, set tls_server_name +# here. The value is used only for TLS SNI and certificate verification, not +# for DNS resolution or endpoint discovery. +# For HTTP collectors, this name is also the HTTP Host header. Switch HTTP +# services must accept that shared host when tls_server_name is configured. +# +# Rotation behavior: +# - NMX-T and NVUE REST rebuild one HTTP client per switch-target collector +# on each poll iteration. This keeps reload behavior deterministic and avoids +# global client cache synchronization when switch inventory changes. Cert file +# changes are adopted on the next scrape for that switch target. +# - NMX-C and NVUE gNMI read refreshed material when a long-lived stream +# reconnects. +# - Existing HTTP clients and long-lived streams are not proactively closed on +# file changes. +# +# [tls.switch] +# ca_cert_path = "/var/run/secrets/switch-mtls/ca.crt" +# client_cert_path = "/var/run/secrets/switch-mtls/tls.crt" +# client_key_path = "/var/run/secrets/switch-mtls/tls.key" +# tls_server_name = "switches.example.forge" + # ============================================================================== # Switch Host Collectors: What data to collect from NVLink Switch Hosts # ============================================================================== # NMX-C collection connects directly to the primary switch-host gRPC endpoint # when switch discovery metadata says NMX-C is enabled for that switch. -# FabricManager (NMX-C) must be running there. Current switch endpoints use -# plaintext gRPC over HTTP/2 on 9370. +# FabricManager (NMX-C) must be running there. Without [tls.switch], +# current switch endpoints use plaintext gRPC over HTTP/2 on 9370. # NMX-C notifications currently emit log events only; metrics and health reports # are separate scope. -# -# TODO: TLS, certificate loading, and mTLS are future scope. [collectors.nmxc] # Enable the NMX-C streaming collector for eligible primary switch-host endpoints. enabled = false diff --git a/crates/health/src/collectors/mod.rs b/crates/health/src/collectors/mod.rs index bcefdf17f6..44cc5594aa 100644 --- a/crates/health/src/collectors/mod.rs +++ b/crates/health/src/collectors/mod.rs @@ -39,7 +39,7 @@ pub use logs::{ }; pub use nmxc::{NmxcCollector, NmxcCollectorConfig}; pub use nmxt::{NmxtCollector, NmxtCollectorConfig}; -pub use nvue::gnmi::subscriber::spawn_gnmi_collector; +pub(crate) use nvue::gnmi::subscriber::spawn_gnmi_collector; pub use nvue::rest::collector::{NvueRestCollector, NvueRestCollectorConfig}; pub use runtime::{ BackoffConfig, Collector, CollectorStartContext, EventStream, ExponentialBackoff, diff --git a/crates/health/src/collectors/nmxc.rs b/crates/health/src/collectors/nmxc.rs index 13effd700f..bebae764dc 100644 --- a/crates/health/src/collectors/nmxc.rs +++ b/crates/health/src/collectors/nmxc.rs @@ -16,7 +16,6 @@ */ use std::borrow::Cow; -use std::net::IpAddr; use std::sync::Arc; use std::time::Duration; @@ -35,7 +34,7 @@ use tonic::transport::{Channel, Endpoint}; use crate::HealthError; use crate::collectors::runtime::{StreamingCollector, StreamingConnectResult}; -use crate::config::NmxcCollectorConfig as NmxcCollectorOptions; +use crate::config::{MtlsProfileConfig, NmxcCollectorConfig as NmxcCollectorOptions}; use crate::endpoint::BmcEndpoint; use crate::sink::{CollectorEvent, LogRecord}; @@ -49,6 +48,9 @@ type NmxcNotificationStream = Streaming; pub struct NmxcCollectorConfig { /// User-facing collector settings from the health service configuration. pub nmxc_config: NmxcCollectorOptions, + + /// mTLS profile used for the gRPC channel when configured. + pub(crate) tls_config: Option, } /// Streaming collector for NMX-C server notifications. @@ -59,6 +61,10 @@ pub struct NmxcCollector { heartbeat_rate: u32, connect_timeout: Duration, rpc_timeout: Duration, + + // NMX-C subscriptions are long-lived. Rotated mTLS profile files are picked + // up when the stream reconnects and builds a new channel. + tls_config: Option, } #[async_trait] @@ -74,8 +80,13 @@ impl StreamingCollector for NmxcCollector { let nmxc_config = config.nmxc_config; let connect_timeout = nmxc_config.connect_timeout(); let rpc_timeout = nmxc_config.rpc_timeout(); + let switch_connect_host = endpoint.switch_connect_host_for_uri(); - let endpoint_url = nmxc_endpoint_url(&endpoint.addr.ip, nmxc_config.grpc_port); + let endpoint_url = nmxc_endpoint_url( + switch_connect_host.as_ref(), + nmxc_config.grpc_port, + config.tls_config.is_some(), + ); Ok(Self { endpoint_url, @@ -84,12 +95,19 @@ impl StreamingCollector for NmxcCollector { heartbeat_rate: nmxc_config.heartbeat_rate, connect_timeout, rpc_timeout, + tls_config: config.tls_config, }) } /// Connects to NMX-C, completes Hello and Subscribe, and maps notifications into collector events. async fn connect(&mut self) -> Result, HealthError> { - let mut client = nmxc_client(&self.endpoint_url, self.connect_timeout).await?; + let mut client = nmxc_client( + &self.endpoint_url, + self.connect_timeout, + self.tls_config.as_ref(), + ) + .await?; + send_hello(&mut client, &self.gateway_id, self.rpc_timeout).await?; let subscribe_request = SubscribeRequest { @@ -217,22 +235,21 @@ fn hello_request(gateway_id: &str) -> ClientHello { } /// Builds the switch-host gRPC endpoint URL, including IPv6 bracket formatting. -fn nmxc_endpoint_url(ip: &IpAddr, port: u16) -> String { - let host = match ip { - IpAddr::V4(v4) => v4.to_string(), - IpAddr::V6(v6) => format!("[{v6}]"), - }; - - // TODO: Replace hard-coded HTTP when NMX-C HTTPS/mTLS support is designed. - format!("http://{host}:{port}") +fn nmxc_endpoint_url(host: &str, port: u16, tls_enabled: bool) -> String { + let scheme = if tls_enabled { "https" } else { "http" }; + format!("{scheme}://{host}:{port}") } /// Creates a tonic NMX-C client with transport settings scoped to the collector. +/// +/// When an mTLS profile is configured, certificate files are read while +/// building the channel, so reconnects pick up rotated material. async fn nmxc_client( endpoint_url: &str, connect_timeout: Duration, + tls_config: Option<&MtlsProfileConfig>, ) -> Result { - let endpoint = Endpoint::from_shared(endpoint_url.to_string()) + let mut endpoint = Endpoint::from_shared(endpoint_url.to_string()) .map_err(|error| nmxc_transport_error(endpoint_url, "parse endpoint", error))? .connect_timeout(connect_timeout) // Long-lived Subscribe streams need keepalive so dead peers eventually @@ -242,6 +259,13 @@ async fn nmxc_client( .keep_alive_timeout(NMX_C_HTTP2_KEEPALIVE_TIMEOUT) .keep_alive_while_idle(true); + if let Some(config) = tls_config { + let tls_config = crate::tls::tonic_tls_config(config).await?; + endpoint = endpoint + .tls_config(tls_config) + .map_err(|error| nmxc_transport_error(endpoint_url, "configure TLS", error))?; + } + let channel = endpoint .connect() .await @@ -606,8 +630,6 @@ fn log_record( #[cfg(test)] mod tests { - use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; - use carbide_test_support::value_scenarios; use rpc::protos::nmx_c::{ ConfigKeyVal, ConfigKeyVals, FmEventPartitionChange, HealthStateChanged, ServerHeader, @@ -720,16 +742,20 @@ mod tests { .map(|(_, value)| value.as_str()) } - #[test] /// Verifies NMX-C endpoint URL formatting for IPv4 and IPv6 targets. - fn endpoint_url_brackets_ipv6() { + #[test] + fn endpoint_url_formats_ip_hosts() { value_scenarios!( - run = |ip| nmxc_endpoint_url(&ip, 9370); + run = |(host, tls_enabled)| nmxc_endpoint_url(host, 9370, tls_enabled); "endpoint URL" { - IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)) => "http://10.0.0.1:9370".to_string(), + ("10.0.0.1", false) => "http://10.0.0.1:9370".to_string(), + + ("[::1]", false) => "http://[::1]:9370".to_string(), + + ("10.0.0.1", true) => "https://10.0.0.1:9370".to_string(), - IpAddr::V6(Ipv6Addr::LOCALHOST) => "http://[::1]:9370".to_string(), + ("[::1]", true) => "https://[::1]:9370".to_string(), } ); } diff --git a/crates/health/src/collectors/nmxt.rs b/crates/health/src/collectors/nmxt.rs index 5f38280c93..0985b2f913 100644 --- a/crates/health/src/collectors/nmxt.rs +++ b/crates/health/src/collectors/nmxt.rs @@ -16,7 +16,8 @@ */ //! This module collects metrics from NMX-T telemetry endpoints on NVLink switches if the service is enabled. -//! Scrapes HTTP on 9352 (default for NMX-T) +//! Scrapes HTTP on 9352 by default. When an mTLS profile is configured, scrapes +//! HTTPS on the same port. //! //! Mapping is an EXPLICIT, catalog-row allowlist over the live NMX-T Prometheus scrape (see //! `NMXT_METRIC_MAP` and `NMXT_LABEL_MAP`). Each NMX-T source name is either: @@ -28,13 +29,14 @@ use std::borrow::Cow; use std::collections::{HashMap, HashSet}; +use std::net::SocketAddr; use std::sync::Arc; use nv_redfish::core::Bmc; use crate::HealthError; use crate::collectors::{IterationResult, PeriodicCollector}; -use crate::config::NmxtCollectorConfig as NmxtCollectorOptions; +use crate::config::{MtlsProfileConfig, NmxtCollectorConfig as NmxtCollectorOptions}; use crate::endpoint::BmcEndpoint; use crate::sink::{CollectorEvent, DataSink, EventContext, MetricSample}; @@ -424,8 +426,9 @@ fn parse_prometheus_line(line: &str) -> Option { async fn scrape_switch_nmxt_metrics( http_client: &reqwest::Client, switch_ip: &str, + tls_enabled: bool, ) -> Result, HealthError> { - let url = format!("http://{}:{}{}", switch_ip, NMXT_PORT, NMXT_ENDPOINT); + let url = nmxt_endpoint_url(switch_ip, tls_enabled); let response = http_client.get(&url).send().await.map_err(|e| { HealthError::GenericError(format!("HTTP request failed for {}: {}", switch_ip, e)) @@ -449,18 +452,42 @@ async fn scrape_switch_nmxt_metrics( Ok(parse_prometheus_metrics(&body)) } +fn nmxt_endpoint_url(switch_ip: &str, tls_enabled: bool) -> String { + let scheme = if tls_enabled { "https" } else { "http" }; + format!("{scheme}://{}:{}{}", switch_ip, NMXT_PORT, NMXT_ENDPOINT) +} + pub struct NmxtCollectorConfig { + /// User-facing NMX-T collector settings from the health service configuration. pub nmxt_config: NmxtCollectorOptions, + + /// Optional sink that receives NMX-T metric events. pub data_sink: Option>, + + /// mTLS profile used for HTTPS scrapes when configured. + pub(crate) tls_config: Option, } pub struct NmxtCollector { endpoint: Arc, - http_client: reqwest::Client, + http_client: NmxtHttpClient, + request_timeout: std::time::Duration, event_context: EventContext, data_sink: Option>, } +enum NmxtHttpClient { + Legacy(reqwest::Client), + + // Store the HTTP client prepared for the current switch-target iteration. + // This avoids a global cache that must be synchronized with switch inventory + // changes. + Tls { + config: MtlsProfileConfig, + client: Option, + }, +} + impl PeriodicCollector for NmxtCollector { type Config = NmxtCollectorConfig; @@ -472,17 +499,30 @@ impl PeriodicCollector for NmxtCollector { let event_context = EventContext::from_endpoint(endpoint.as_ref(), "nmxt"); let request_timeout = config.nmxt_config.request_timeout; - let mut http_client_builder = reqwest::Client::builder().timeout(request_timeout); - if config.nmxt_config.dangerously_skip_tls_verification { - http_client_builder = http_client_builder.danger_accept_invalid_certs(true); - } - let http_client = http_client_builder.build().map_err(|e| { - HealthError::GenericError(format!("Failed to create HTTP client: {}", e)) - })?; + let http_client = match config.tls_config { + Some(tls_config) => NmxtHttpClient::Tls { + config: tls_config, + client: None, + }, + None => { + let mut http_client_builder = reqwest::Client::builder().timeout(request_timeout); + + if config.nmxt_config.dangerously_skip_tls_verification { + http_client_builder = http_client_builder.danger_accept_invalid_certs(true); + } + + let http_client = http_client_builder.build().map_err(|e| { + HealthError::GenericError(format!("Failed to create HTTP client: {}", e)) + })?; + + NmxtHttpClient::Legacy(http_client) + } + }; Ok(Self { endpoint, http_client, + request_timeout, event_context, data_sink: config.data_sink, }) @@ -529,10 +569,27 @@ impl NmxtCollector { labels } - async fn scrape_iteration(&self) -> Result<(), HealthError> { - let switch_ip = self.endpoint.addr.ip.to_string(); + async fn scrape_iteration(&mut self) -> Result<(), HealthError> { + let tls_enabled = matches!(self.http_client, NmxtHttpClient::Tls { .. }); + + let switch_connect_host = self.endpoint.switch_connect_host_for_uri(); + + // When `tls_server_name` is configured, the URL host and HTTP Host + // header use that shared TLS identity. Reqwest resolves it to the + // discovered switch IP through the target-scoped client override below. + // This matches current switch behavior, which accepts the shared host. + let scrape_host = match &self.http_client { + NmxtHttpClient::Tls { config, .. } => config + .tls_server_name + .as_deref() + .unwrap_or_else(|| switch_connect_host.as_ref()) + .to_string(), + NmxtHttpClient::Legacy(_) => switch_connect_host.to_string(), + }; + + let http_client = self.http_client().await?; - let metrics = scrape_switch_nmxt_metrics(&self.http_client, &switch_ip).await?; + let metrics = scrape_switch_nmxt_metrics(&http_client, &scrape_host, tls_enabled).await?; self.emit_event(CollectorEvent::MetricCollectionStart); @@ -636,12 +693,65 @@ impl NmxtCollector { Ok(()) } + + async fn http_client(&mut self) -> Result { + let resolve_addr = SocketAddr::new(self.endpoint.addr.ip, NMXT_PORT); + + match &mut self.http_client { + NmxtHttpClient::Legacy(client) => Ok(client.clone()), + NmxtHttpClient::Tls { config, client } => { + // Rebuild once per poll iteration so mTLS material changes are + // picked up on the next scrape without global client cache + // synchronization. + let built_client = crate::tls::reqwest_client( + config, + self.request_timeout, + config.tls_server_name.as_ref().map(|_| resolve_addr), + ) + .await?; + + let cloned_client = built_client.clone(); + + *client = Some(built_client); + + Ok(cloned_client) + } + } + } } #[cfg(test)] mod tests { use super::*; + #[test] + fn test_nmxt_endpoint_url_switches_scheme_when_tls_enabled() { + struct TestCase { + name: &'static str, + tls_enabled: bool, + expected: &'static str, + } + + let cases = [ + TestCase { + name: "legacy HTTP", + tls_enabled: false, + expected: "http://10.0.0.9:9352/xcset/nvlink_domain_telemetry", + }, + TestCase { + name: "mTLS HTTPS", + tls_enabled: true, + expected: "https://10.0.0.9:9352/xcset/nvlink_domain_telemetry", + }, + ]; + + for case in cases { + let actual = nmxt_endpoint_url("10.0.0.9", case.tls_enabled); + + assert_eq!(actual, case.expected, "{}", case.name); + } + } + #[test] fn test_parse_prometheus_line_with_labels() { let line = r#"Effective_BER{Port_Number="2", Node_GUID="0x8e2161c8803caf64"} 1.5e-254"#; @@ -956,7 +1066,8 @@ Link_Down{Port_Number="1"} 5 }); let collector = NmxtCollector { endpoint: endpoint.clone(), - http_client: reqwest::Client::new(), + http_client: NmxtHttpClient::Legacy(reqwest::Client::new()), + request_timeout: std::time::Duration::from_secs(30), event_context: EventContext::from_endpoint(endpoint.as_ref(), "nmxt"), data_sink: Some(sink.clone()), }; @@ -1065,7 +1176,8 @@ Link_Down{Port_Number="1"} 5 }); let collector = NmxtCollector { endpoint: endpoint.clone(), - http_client: reqwest::Client::new(), + http_client: NmxtHttpClient::Legacy(reqwest::Client::new()), + request_timeout: std::time::Duration::from_secs(30), event_context: EventContext::from_endpoint(endpoint.as_ref(), "nmxt"), data_sink: Some(sink.clone()), }; diff --git a/crates/health/src/collectors/nvue/gnmi/client.rs b/crates/health/src/collectors/nvue/gnmi/client.rs index 44816caa75..0bb2ef5e54 100644 --- a/crates/health/src/collectors/nvue/gnmi/client.rs +++ b/crates/health/src/collectors/nvue/gnmi/client.rs @@ -28,7 +28,7 @@ use super::proto::{ SubscriptionMode, }; use crate::HealthError; -use crate::config::NvueGnmiPaths; +use crate::config::{MtlsProfileConfig, NvueGnmiPaths}; pub fn nvue_subscribe_paths(paths_config: &NvueGnmiPaths) -> Vec { let mut paths = Vec::with_capacity(4); @@ -106,13 +106,52 @@ pub struct GnmiClient { password: Option, request_timeout: Duration, dangerously_skip_tls_verification: bool, + tls_config: Option, } -fn configure_tls_endpoint( +/// Configuration used to build one gNMI client instance. +pub(super) struct GnmiClientConfig { + /// Switch identifier used in logs and error messages. + pub switch_id: String, + + /// Switch host or IP address used for the gNMI channel. + pub host: String, + + /// gNMI TCP port on the switch host. + pub port: u16, + + /// Optional username sent as gNMI `username` metadata. + pub username: Option, + + /// Optional password sent as gNMI `password` metadata. + pub password: Option, + + /// Timeout applied to gNMI connection and RPC operations. + pub request_timeout: Duration, + + /// Whether legacy non-mTLS connections accept invalid switch certificates. + pub dangerously_skip_tls_verification: bool, + + /// mTLS profile used when opening the gNMI channel. + pub tls_config: Option, +} + +async fn configure_tls_endpoint( endpoint: Endpoint, switch_id: &str, dangerously_skip_tls_verification: bool, + tls_config: Option<&MtlsProfileConfig>, ) -> Result { + if let Some(config) = tls_config { + // mTLS config supplies both trust roots and client identity. Use it as + // the complete TLS policy for this channel. + let tls_config = crate::tls::tonic_tls_config(config).await?; + + return endpoint.tls_config(tls_config).map_err(|e| { + HealthError::GnmiError(format!("switch {switch_id}: invalid gNMI TLS config: {e}")) + }); + } + if !dangerously_skip_tls_verification { return Ok(endpoint); } @@ -130,23 +169,16 @@ fn configure_tls_endpoint( } impl GnmiClient { - pub fn new( - switch_id: String, - host: &str, - port: u16, - username: Option, - password: Option, - request_timeout: Duration, - dangerously_skip_tls_verification: bool, - ) -> Self { + pub(super) fn new(config: GnmiClientConfig) -> Self { Self { - switch_id, - host: host.to_string(), - port, - username, - password, - request_timeout, - dangerously_skip_tls_verification, + switch_id: config.switch_id, + host: config.host, + port: config.port, + username: config.username, + password: config.password, + request_timeout: config.request_timeout, + dangerously_skip_tls_verification: config.dangerously_skip_tls_verification, + tls_config: config.tls_config, } } @@ -169,7 +201,9 @@ impl GnmiClient { Endpoint::from(uri), &self.switch_id, self.dangerously_skip_tls_verification, - )? + self.tls_config.as_ref(), + ) + .await? .connect_timeout(self.request_timeout) .timeout(self.request_timeout); @@ -498,26 +532,30 @@ mod tests { #[test] fn test_gnmi_client_stores_dangerous_tls_skip_flag() { - let strict = GnmiClient::new( - "switch-1".to_string(), - "10.0.0.9", - 9339, - None, - None, - Duration::from_secs(30), - false, - ); + let strict = GnmiClient::new(GnmiClientConfig { + switch_id: "switch-1".to_string(), + host: "10.0.0.9".to_string(), + port: 9339, + username: None, + password: None, + request_timeout: Duration::from_secs(30), + dangerously_skip_tls_verification: false, + tls_config: None, + }); + assert!(!strict.dangerously_skip_tls_verification); - let dangerous = GnmiClient::new( - "switch-1".to_string(), - "10.0.0.9", - 9339, - None, - None, - Duration::from_secs(30), - true, - ); + let dangerous = GnmiClient::new(GnmiClientConfig { + switch_id: "switch-1".to_string(), + host: "10.0.0.9".to_string(), + port: 9339, + username: None, + password: None, + request_timeout: Duration::from_secs(30), + dangerously_skip_tls_verification: true, + tls_config: None, + }); + assert!(dangerous.dangerously_skip_tls_verification); } diff --git a/crates/health/src/collectors/nvue/gnmi/subscriber.rs b/crates/health/src/collectors/nvue/gnmi/subscriber.rs index 156aabdb4f..09ed93557a 100644 --- a/crates/health/src/collectors/nvue/gnmi/subscriber.rs +++ b/crates/health/src/collectors/nvue/gnmi/subscriber.rs @@ -25,7 +25,8 @@ use tokio::sync::OnceCell; use tokio_util::sync::CancellationToken; use super::client::{ - GnmiClient, nvue_subscribe_paths, system_events_prefix, system_events_subscribe_path, + GnmiClient, GnmiClientConfig, nvue_subscribe_paths, system_events_prefix, + system_events_subscribe_path, }; use super::on_change_processor::{ GnmiOnChangeProcessor, ON_CHANGE_STREAM_ID_SYSTEM_EVENTS, OnChangeStreamMetrics, @@ -36,7 +37,7 @@ use crate::HealthError; use crate::bmc::{CREDENTIAL_REFRESH_TIMEOUT, CredentialProvider}; use crate::collectors::Collector; use crate::collectors::runtime::{BackoffConfig, ExponentialBackoff, StreamingConnectionGuard}; -use crate::config::NvueGnmiConfig; +use crate::config::{MtlsProfileConfig, NvueGnmiConfig}; use crate::endpoint::{BmcAddr, BmcCredentials, BmcEndpoint}; use crate::metrics::CollectorRegistry; use crate::sink::{CollectorEvent, DataSink, EventContext}; @@ -184,10 +185,14 @@ struct GnmiStreamConfig { #[derive(Clone)] struct GnmiClientProvider { switch_id: String, - switch_ip: String, + switch_connect_host: String, port: u16, request_timeout: Duration, dangerously_skip_tls_verification: bool, + + // Streaming subscriptions build a fresh gNMI client on reconnect, so + // rotated mTLS profile files are adopted after the stream reconnects. + tls_config: Option, credentials: Arc, } @@ -209,15 +214,16 @@ impl GnmiClientProvider { async fn new_client(&self) -> Result<(GnmiClient, u64), HealthError> { let (credentials, generation) = self.credentials.ensure().await?; Ok(( - GnmiClient::new( - self.switch_id.clone(), - &self.switch_ip, - self.port, - credentials.username, - credentials.password, - self.request_timeout, - self.dangerously_skip_tls_verification, - ), + GnmiClient::new(GnmiClientConfig { + switch_id: self.switch_id.clone(), + host: self.switch_connect_host.clone(), + port: self.port, + username: credentials.username, + password: credentials.password, + request_timeout: self.request_timeout, + dangerously_skip_tls_verification: self.dangerously_skip_tls_verification, + tls_config: self.tls_config.clone(), + }), generation, )) } @@ -441,27 +447,31 @@ async fn fetch_gnmi_username_password( )), } } -pub fn spawn_gnmi_collector( + +pub(crate) fn spawn_gnmi_collector( endpoint: &BmcEndpoint, gnmi_config: &NvueGnmiConfig, credential_provider: Arc, collector_registry: Arc, data_sink: Option>, + tls_config: Option, ) -> Result { let switch_id = endpoint .metadata .as_ref() .and_then(|m| m.serial_number().map(str::to_string)) .unwrap_or_else(|| endpoint.addr.mac.to_string()); - let switch_ip = endpoint.addr.ip.to_string(); + + let switch_connect_host = endpoint.switch_connect_host_for_uri().into_owned(); let sample_event_context = EventContext::from_endpoint(endpoint, NVUE_GNMI_SAMPLE_STREAM_ID); let client_provider = GnmiClientProvider { switch_id: switch_id.clone(), - switch_ip, + switch_connect_host, port: gnmi_config.gnmi_port, request_timeout: gnmi_config.request_timeout, dangerously_skip_tls_verification: gnmi_config.dangerously_skip_tls_verification, + tls_config, credentials: Arc::new(GnmiCredentialCache::new( credential_provider, endpoint.addr.clone(), @@ -846,10 +856,11 @@ mod tests { let addr = test_addr(); GnmiClientProvider { switch_id: "switch-1".to_string(), - switch_ip: addr.ip.to_string(), + switch_connect_host: addr.ip.to_string(), port: 9339, request_timeout: Duration::from_secs(1), dangerously_skip_tls_verification: false, + tls_config: None, credentials: Arc::new(GnmiCredentialCache::new(provider, addr)), } } diff --git a/crates/health/src/collectors/nvue/rest/client.rs b/crates/health/src/collectors/nvue/rest/client.rs index 69a62eba56..426e68b49b 100644 --- a/crates/health/src/collectors/nvue/rest/client.rs +++ b/crates/health/src/collectors/nvue/rest/client.rs @@ -16,6 +16,7 @@ */ use std::collections::HashMap; +use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use std::time::Duration; @@ -27,7 +28,7 @@ use serde::de::Error as _; use url::Url; use crate::HealthError; -use crate::config::NvueRestPaths; +use crate::config::{MtlsProfileConfig, NvueRestPaths}; const NVUE_SYSTEM_HEALTH: &str = "/nvue_v1/system/health"; const NVUE_SYSTEM_REBOOT_REASON: &str = "/nvue_v1/system/reboot/reason"; @@ -76,38 +77,97 @@ pub struct RestClient { base_url: Url, credentials: ArcSwapOption, paths: NvueRestPaths, - client: Client, + http_client: RestHttpClient, +} + +enum RestHttpClient { + Legacy(Client), + + // Store the HTTP client prepared for the current switch-target iteration. + // The resolver override is target-specific when `[tls.switch].tls_server_name` + // is set, so sharing one reqwest client across switch IPs would be incorrect. + Tls { + config: MtlsProfileConfig, + request_timeout: Duration, + + // When set, base_url uses `[tls.switch].tls_server_name` for SNI and + // certificate verification, while reqwest resolves it to this socket + // address locally. + resolve_addr: Option, + + client: Option, + }, } impl RestClient { pub fn new( switch_id: String, - host: &str, + connect_ip: IpAddr, + port: Option, request_timeout: Duration, self_signed_tls: bool, + tls_config: Option, paths: NvueRestPaths, ) -> Result { - let raw_url = format!("https://{host}"); + let tls_server_name = tls_config + .as_ref() + .and_then(|config| config.tls_server_name.clone()); + + let port = port.unwrap_or(443); + let resolve_addr = tls_server_name + .as_ref() + .map(|_| SocketAddr::new(connect_ip, port)); + + // When `tls_server_name` is configured, the URL host and HTTP Host + // header use that shared TLS identity. Reqwest resolves it to the + // discovered switch IP through the target-scoped client override below. + // This matches current switch behavior, which accepts the shared host. + let host = tls_server_name + .as_deref() + .map(ToOwned::to_owned) + .unwrap_or_else(|| match connect_ip { + IpAddr::V4(ip) => ip.to_string(), + IpAddr::V6(ip) => format!("[{ip}]"), + }); + + let raw_url = if port == 443 { + format!("https://{host}") + } else { + format!("https://{host}:{port}") + }; + let base_url = Url::parse(&raw_url) .map_err(|e| HealthError::HttpError(format!("{raw_url}: invalid base URL: {e}")))?; - let mut builder = Client::builder().timeout(request_timeout); + let http_client = match tls_config { + Some(config) => RestHttpClient::Tls { + config, + request_timeout, + resolve_addr, + client: None, + }, + None => { + let mut builder = Client::builder().timeout(request_timeout); - if self_signed_tls { - // ! dangerously accept the self-signed certificate. - builder = builder.danger_accept_invalid_certs(true); - } + if self_signed_tls { + // ! dangerously accept the self-signed certificate. + builder = builder.danger_accept_invalid_certs(true); + } - let client = builder.build().map_err(|e| { - HealthError::HttpError(format!("{base_url}: failed to create HTTP client: {e}")) - })?; + let client = builder.build().map_err(|e| { + HealthError::HttpError(format!("{base_url}: failed to create HTTP client: {e}")) + })?; + + RestHttpClient::Legacy(client) + } + }; Ok(Self { switch_id, base_url, credentials: ArcSwapOption::empty(), paths, - client, + http_client, }) } @@ -130,10 +190,30 @@ impl RestClient { base_url, credentials: ArcSwapOption::empty(), paths, - client, + http_client: RestHttpClient::Legacy(client), }) } + /// Rebuilds the target-scoped HTTP client before NVUE REST requests. + /// + /// When an mTLS profile is configured, the client is rebuilt once per poll + /// iteration. This keeps certificate reload behavior deterministic and + /// avoids a global cache that would need switch inventory synchronization. + pub async fn ensure_http_client(&mut self) -> Result<(), HealthError> { + if let RestHttpClient::Tls { + config, + request_timeout, + resolve_addr, + client, + } = &mut self.http_client + { + *client = + Some(crate::tls::reqwest_client(config, *request_timeout, *resolve_addr).await?); + } + + Ok(()) + } + pub fn set_credentials(&self, creds: UsernamePassword) { self.credentials.store(Some(Arc::new(creds))); } @@ -273,7 +353,21 @@ impl RestClient { url: Url, extra_query: &[(&str, &str)], ) -> Result { - let mut request = self.client.get(url.as_str()); + let client = match &self.http_client { + RestHttpClient::Legacy(client) => client, + RestHttpClient::Tls { + client: Some(client), + .. + } => client, + RestHttpClient::Tls { .. } => { + return Err(HealthError::HttpError(format!( + "{url}: mTLS HTTP client was not prepared for switch {}", + self.switch_id + ))); + } + }; + + let mut request = client.get(url.as_str()); // GET /interface (returning a collection) defaults to rev=applied, not operational. // There is inconsistency across the NVUE Endpoints, so we need to check each. diff --git a/crates/health/src/collectors/nvue/rest/collector.rs b/crates/health/src/collectors/nvue/rest/collector.rs index 93e209dad4..28e1243efe 100644 --- a/crates/health/src/collectors/nvue/rest/collector.rs +++ b/crates/health/src/collectors/nvue/rest/collector.rs @@ -25,7 +25,7 @@ use super::client::{ use crate::HealthError; use crate::bmc::{CREDENTIAL_REFRESH_TIMEOUT, CredentialProvider, is_auth_error}; use crate::collectors::{IterationResult, PeriodicCollector}; -use crate::config::NvueRestConfig; +use crate::config::{MtlsProfileConfig, NvueRestConfig}; use crate::endpoint::{BmcAddr, BmcCredentials, BmcEndpoint, EndpointMetadata}; use crate::sink::{ Classification, CollectorEvent, DataSink, EventContext, HealthReport, HealthReportAlert, @@ -141,9 +141,17 @@ fn fan_led_to_state(state: Option<&str>) -> Option<&'static str> { } pub struct NvueRestCollectorConfig { + /// User-facing NVUE REST collector settings from the health service configuration. pub rest_config: NvueRestConfig, + + /// Optional sink that receives NVUE REST health reports and events. pub data_sink: Option>, + + /// Credential source used to authenticate to NVUE REST. pub credential_provider: Arc, + + /// mTLS profile used for HTTPS polling when configured. + pub(crate) tls_config: Option, } pub struct NvueRestCollector { @@ -167,16 +175,18 @@ impl PeriodicCollector for NvueRestCollector { Some(EndpointMetadata::Switch(s)) => s.serial.clone(), _ => endpoint.addr.mac.to_string(), }; - let switch_ip = endpoint.addr.ip.to_string(); + let event_context = EventContext::from_endpoint(endpoint.as_ref(), COLLECTOR_NAME); let rest_cfg = &config.rest_config; // self_signed_tls is always true -- TLS cert provisioning on switches is not yet implemented let client = RestClient::new( switch_id.clone(), - &switch_ip, + endpoint.addr.ip, + endpoint.addr.port, rest_cfg.request_timeout, true, + config.tls_config, rest_cfg.paths.clone(), )?; @@ -191,6 +201,8 @@ impl PeriodicCollector for NvueRestCollector { } async fn run_iteration(&mut self) -> Result { + self.client.ensure_http_client().await?; + if !self.client.has_credentials() && let Err(error) = self.refresh_rest_credentials().await { @@ -1493,9 +1505,11 @@ mod tests { let addr = test_addr(); let client = RestClient::new( "test-switch".to_string(), - &addr.ip.to_string(), + addr.ip, + addr.port, Duration::from_millis(10), true, + None, paths_all_disabled(), ) .expect("rest client builds"); diff --git a/crates/health/src/config.rs b/crates/health/src/config.rs index 70a36fac03..44f489bae0 100644 --- a/crates/health/src/config.rs +++ b/crates/health/src/config.rs @@ -17,11 +17,12 @@ use std::fmt::Debug; use std::net::{IpAddr, SocketAddr}; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::time::Duration; use figment::Figment; use figment::providers::{Env, Format, Serialized, Toml}; +use rustls_pki_types::DnsName; use serde::{Deserialize, Deserializer, Serialize}; use url::Url; @@ -30,6 +31,8 @@ use url::Url; pub struct Config { pub endpoint_sources: EndpointSourcesConfig, + pub tls: TlsConfig, + pub sinks: SinksConfig, pub rate_limit: Configurable, @@ -61,6 +64,7 @@ impl Default for Config { fn default() -> Self { Self { endpoint_sources: EndpointSourcesConfig::default(), + tls: TlsConfig::default(), sinks: SinksConfig::default(), rate_limit: Configurable::Enabled(RateLimitConfig::default()), collectors: CollectorsConfig::default(), @@ -567,6 +571,85 @@ impl Default for CollectorsConfig { } } +/// TLS settings owned by hardware-health. +/// +/// This section is intentionally outside `[collectors]` because TLS material is +/// connection policy shared by multiple collectors, not a collector itself. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +#[serde(default)] +pub struct TlsConfig { + /// Optional mTLS profile used by direct switch collectors. + pub switch: Option, +} + +/// mTLS profile for outbound client TLS connections. +/// +/// `[tls.switch]` uses this shape for direct switch collector connections. +/// These paths are independent from the Carbide API certificate paths. The +/// files are read and validated when collectors build HTTP clients or gRPC +/// channel TLS configs. The optional TLS server name is profile-wide because +/// deployed switch certificates use the same DNS identity, and Carbide API +/// discovery does not provide switch certificate identities. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct MtlsProfileConfig { + /// Path to the CA bundle used to verify switch server certificates. + pub ca_cert_path: PathBuf, + + /// Path to the client certificate chain sent to switch services. + pub client_cert_path: PathBuf, + + /// Path to the client private key sent to switch services. + pub client_key_path: PathBuf, + + /// Optional DNS name used only for TLS SNI and server certificate checks. + /// + /// Direct switch collectors still open TCP connections to each discovered + /// switch endpoint IP. When all switch server certificates carry the same + /// DNS subjectAltName, set this field so TLS verifies that DNS identity + /// instead of requiring every switch certificate to include an IP SAN. + /// This value is never used for endpoint discovery or DNS resolution. + /// + /// For HTTP collectors, this name is also the request URL host and HTTP + /// Host header. Switch HTTP services must accept that host when this field + /// is configured. + pub tls_server_name: Option, +} + +impl MtlsProfileConfig { + fn validate(&self) -> Result<(), String> { + if self.ca_cert_path.as_os_str().is_empty() { + return Err("[tls.switch].ca_cert_path must not be empty".to_string()); + } + + if self.client_cert_path.as_os_str().is_empty() { + return Err("[tls.switch].client_cert_path must not be empty".to_string()); + } + + if self.client_key_path.as_os_str().is_empty() { + return Err("[tls.switch].client_key_path must not be empty".to_string()); + } + + if let Some(tls_server_name) = self.tls_server_name.as_deref() { + if tls_server_name.trim().is_empty() { + return Err("[tls.switch].tls_server_name must not be empty".to_string()); + } + + if tls_server_name.trim() != tls_server_name { + return Err( + "[tls.switch].tls_server_name must not contain leading or trailing whitespace" + .to_string(), + ); + } + + DnsName::try_from(tls_server_name) + .map_err(|_| "[tls.switch].tls_server_name must be a valid DNS name".to_string())?; + } + + Ok(()) + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(default)] pub struct DiscoveryConfig { @@ -1330,6 +1413,29 @@ impl Config { logs.validate()?; } + if let Some(tls_config) = &self.tls.switch { + tls_config.validate()?; + + if let Configurable::Enabled(nmxt) = &self.collectors.nmxt + && nmxt.dangerously_skip_tls_verification + { + return Err( + "[collectors.nmxt].dangerously_skip_tls_verification must be false when [tls.switch] is configured" + .to_string(), + ); + } + + if let Configurable::Enabled(nvue) = &self.collectors.nvue + && let Configurable::Enabled(gnmi) = &nvue.gnmi + && gnmi.dangerously_skip_tls_verification + { + return Err( + "[collectors.nvue.gnmi].dangerously_skip_tls_verification must be false when [tls.switch] is configured" + .to_string(), + ); + } + } + if let Configurable::Enabled(nmxc) = &self.collectors.nmxc { nmxc.validate()?; } @@ -2251,6 +2357,233 @@ dangerously_skip_tls_verification = true } } + #[test] + fn test_tls_switch_profile_absent_by_default_and_does_not_reuse_api_cert_paths() { + let config = Config::default(); + + assert!(config.tls.switch.is_none()); + + let Configurable::Enabled(carbide_api) = config.endpoint_sources.carbide_api else { + panic!("carbide api endpoint source should be enabled by default"); + }; + + assert_eq!(carbide_api.root_ca, "/var/run/secrets/spiffe.io/ca.crt"); + + assert_eq!( + carbide_api.client_cert, + "/var/run/secrets/spiffe.io/tls.crt" + ); + + assert_eq!(carbide_api.client_key, "/var/run/secrets/spiffe.io/tls.key"); + } + + #[test] + fn test_tls_switch_profile_parses_independent_paths() { + let toml = r#" +[endpoint_sources.carbide_api] +enabled = false + +[sinks.health_report] +enabled = false + +[tls.switch] +ca_cert_path = "/var/run/secrets/switch-mtls/ca.crt" +client_cert_path = "/var/run/secrets/switch-mtls/tls.crt" +client_key_path = "/var/run/secrets/switch-mtls/tls.key" +tls_server_name = "switches.example.forge" +"#; + + let config: Config = Figment::new() + .merge(Serialized::defaults(Config::default())) + .merge(Toml::string(toml)) + .extract() + .expect("failed to parse mTLS profile config"); + + config + .validate() + .expect("mTLS profile config should validate"); + + let tls_config = config + .tls + .switch + .expect("mTLS profile config should be present"); + + assert_eq!( + tls_config.ca_cert_path, + PathBuf::from("/var/run/secrets/switch-mtls/ca.crt") + ); + + assert_eq!( + tls_config.client_cert_path, + PathBuf::from("/var/run/secrets/switch-mtls/tls.crt") + ); + + assert_eq!( + tls_config.client_key_path, + PathBuf::from("/var/run/secrets/switch-mtls/tls.key") + ); + + assert_eq!( + tls_config.tls_server_name.as_deref(), + Some("switches.example.forge") + ); + } + + #[test] + fn test_tls_switch_profile_rejects_incomplete_or_unknown_fields() { + struct TestCase { + name: &'static str, + toml: &'static str, + } + + let cases = [ + TestCase { + name: "missing CA", + toml: r#" +[tls.switch] +client_cert_path = "/switch/tls.crt" +client_key_path = "/switch/tls.key" +"#, + }, + TestCase { + name: "missing client cert", + toml: r#" +[tls.switch] +ca_cert_path = "/switch/ca.crt" +client_key_path = "/switch/tls.key" +"#, + }, + TestCase { + name: "missing client key", + toml: r#" +[tls.switch] +ca_cert_path = "/switch/ca.crt" +client_cert_path = "/switch/tls.crt" +"#, + }, + TestCase { + name: "unknown field", + toml: r#" +[tls.switch] +ca_cert_path = "/switch/ca.crt" +client_cert_path = "/switch/tls.crt" +client_key_path = "/switch/tls.key" +root_ca = "/var/run/secrets/spiffe.io/ca.crt" +"#, + }, + ]; + + for case in cases { + let result = Figment::new() + .merge(Serialized::defaults(Config::default())) + .merge(Toml::string(case.toml)) + .extract::(); + + assert!(result.is_err(), "{}", case.name); + } + } + + #[test] + fn test_tls_switch_rejects_empty_paths_and_dangerous_tls_bypass() { + struct TestCase { + name: &'static str, + toml: &'static str, + expected: &'static str, + } + + let base = r#" +[endpoint_sources.carbide_api] +enabled = false + +[sinks.health_report] +enabled = false +"#; + let cases = [ + TestCase { + name: "empty CA path", + toml: r#" +[tls.switch] +ca_cert_path = "" +client_cert_path = "/switch/tls.crt" +client_key_path = "/switch/tls.key" +"#, + expected: "[tls.switch].ca_cert_path must not be empty", + }, + TestCase { + name: "empty TLS server name", + toml: r#" +[tls.switch] +ca_cert_path = "/switch/ca.crt" +client_cert_path = "/switch/tls.crt" +client_key_path = "/switch/tls.key" +tls_server_name = " " +"#, + expected: "[tls.switch].tls_server_name must not be empty", + }, + TestCase { + name: "TLS server name with surrounding whitespace", + toml: r#" +[tls.switch] +ca_cert_path = "/switch/ca.crt" +client_cert_path = "/switch/tls.crt" +client_key_path = "/switch/tls.key" +tls_server_name = " switches.example.forge " +"#, + expected: "[tls.switch].tls_server_name must not contain leading or trailing whitespace", + }, + TestCase { + name: "invalid TLS server name", + toml: r#" +[tls.switch] +ca_cert_path = "/switch/ca.crt" +client_cert_path = "/switch/tls.crt" +client_key_path = "/switch/tls.key" +tls_server_name = "not a dns name" +"#, + expected: "[tls.switch].tls_server_name must be a valid DNS name", + }, + TestCase { + name: "NMX-T dangerous skip conflict", + toml: r#" +[collectors.nmxt] +dangerously_skip_tls_verification = true + +[tls.switch] +ca_cert_path = "/switch/ca.crt" +client_cert_path = "/switch/tls.crt" +client_key_path = "/switch/tls.key" +"#, + expected: "[collectors.nmxt].dangerously_skip_tls_verification must be false when [tls.switch] is configured", + }, + TestCase { + name: "gNMI dangerous skip conflict", + toml: r#" +[collectors.nvue.gnmi] +dangerously_skip_tls_verification = true + +[tls.switch] +ca_cert_path = "/switch/ca.crt" +client_cert_path = "/switch/tls.crt" +client_key_path = "/switch/tls.key" +"#, + expected: "[collectors.nvue.gnmi].dangerously_skip_tls_verification must be false when [tls.switch] is configured", + }, + ]; + + for case in cases { + let toml = format!("{base}{}", case.toml); + let config: Config = Figment::new() + .merge(Serialized::defaults(Config::default())) + .merge(Toml::string(&toml)) + .extract() + .expect(case.name); + + let error = config.validate().expect_err(case.name); + + assert_eq!(error, case.expected, "{}", case.name); + } + } + #[test] fn test_example_config_documents_platform_environment_fan_toggle() { let toml_content = include_str!("../example/config.example.toml"); diff --git a/crates/health/src/discovery/context.rs b/crates/health/src/discovery/context.rs index 820cbeaa41..e9c7ca9129 100644 --- a/crates/health/src/discovery/context.rs +++ b/crates/health/src/discovery/context.rs @@ -29,8 +29,9 @@ use crate::config::{ Config, Configurable, DiscoveryConfig, FirmwareCollectorConfig as FirmwareCollectorOptions, LeakDetectorCollectorConfig as LeakDetectorCollectorOptions, LogsCollectorConfig as LogsCollectorOptions, MetricsCollectorConfig as MetricsCollectorOptions, - NmxcCollectorConfig as NmxcCollectorOptions, NmxtCollectorConfig as NmxtCollectorOptions, - NvueCollectorConfig as NvueCollectorOptions, SensorCollectorConfig as SensorCollectorOptions, + MtlsProfileConfig, NmxcCollectorConfig as NmxcCollectorOptions, + NmxtCollectorConfig as NmxtCollectorOptions, NvueCollectorConfig as NvueCollectorOptions, + SensorCollectorConfig as SensorCollectorOptions, }; use crate::limiter::RateLimiter; use crate::metrics::{MetricsManager, operation_duration_buckets_seconds}; @@ -231,6 +232,7 @@ pub struct DiscoveryLoopContext { pub(crate) nmxt_config: Configurable, pub(crate) nmxc_config: Configurable, pub(crate) nvue_config: Configurable, + pub(crate) tls_config: Option, /// Whether any enabled sink consumes `CollectorEvent::Log` payloads. pub(crate) log_event_sink_enabled: bool, @@ -245,6 +247,15 @@ impl DiscoveryLoopContext { limiter: Arc, metrics_manager: Arc, config: Arc, + ) -> Result { + Self::new_with_tls_config(limiter, metrics_manager, config, None) + } + + pub(crate) fn new_with_tls_config( + limiter: Arc, + metrics_manager: Arc, + config: Arc, + tls_config: Option, ) -> Result { let registry = metrics_manager.global_registry(); @@ -283,6 +294,7 @@ impl DiscoveryLoopContext { nmxt_config: config.collectors.nmxt.clone(), nmxc_config: config.collectors.nmxc.clone(), nvue_config: config.collectors.nvue.clone(), + tls_config: tls_config.or_else(|| config.tls.switch.clone()), log_event_sink_enabled: config.sinks.includes_log_events(), log_downgrade_registry: Arc::new(LogDowngradeRegistry::new()), logs_include_diagnostics: config.sinks.includes_log_diagnostics(), @@ -297,6 +309,7 @@ mod tests { use super::*; use crate::collectors::Collector; + use crate::config::MtlsProfileConfig; fn noop_collector() -> Collector { Collector::spawn_task(|_| async {}) @@ -322,4 +335,32 @@ mod tests { assert!(removed.contains(&Cow::Borrowed("removed-gNMI-endpoint"))); assert!(!removed.contains(&Cow::Borrowed("active-rest-endpoint"))); } + + #[test] + fn context_carries_tls_switch_config() { + let mut config = Config::default(); + + let tls_config = MtlsProfileConfig { + ca_cert_path: "/switch/ca.crt".into(), + client_cert_path: "/switch/tls.crt".into(), + client_key_path: "/switch/tls.key".into(), + tls_server_name: Some("switches.example.forge".to_string()), + }; + + config.tls.switch = Some(tls_config.clone()); + + let context = DiscoveryLoopContext::new( + Arc::new(crate::limiter::NoopLimiter), + Arc::new(MetricsManager::new("tls_context").expect("metrics manager")), + Arc::new(config), + ) + .expect("context should initialize"); + + let actual_tls_config = context + .tls_config + .as_ref() + .expect("[tls.switch] config should be present"); + + assert_eq!(actual_tls_config, &tls_config); + } } diff --git a/crates/health/src/discovery/spawn.rs b/crates/health/src/discovery/spawn.rs index 4b28e84532..ffa9e90e16 100644 --- a/crates/health/src/discovery/spawn.rs +++ b/crates/health/src/discovery/spawn.rs @@ -454,6 +454,7 @@ fn spawn_switch_host_collectors( NmxtCollectorConfig { nmxt_config: nmxt_cfg.clone(), data_sink: data_sink.clone(), + tls_config: ctx.tls_config.clone(), }, CollectorStartContext { limiter: ctx.limiter.clone(), @@ -501,6 +502,7 @@ fn spawn_switch_host_collectors( bmc.clone(), NmxcCollectorConfig { nmxc_config: nmxc_cfg.clone(), + tls_config: ctx.tls_config.clone(), }, data_sink, StreamingCollectorStartContext { @@ -555,6 +557,7 @@ fn spawn_switch_host_collectors( rest_config: rest_cfg.clone(), data_sink: data_sink.clone(), credential_provider, + tls_config: ctx.tls_config.clone(), }, CollectorStartContext { limiter: ctx.limiter.clone(), @@ -598,6 +601,7 @@ fn spawn_switch_host_collectors( credential_provider, collector_registry, data_sink.clone(), + ctx.tls_config.clone(), ) { Ok(handle) => { ctx.collectors diff --git a/crates/health/src/endpoint/model.rs b/crates/health/src/endpoint/model.rs index 9447ed721e..16c2bcd9d4 100644 --- a/crates/health/src/endpoint/model.rs +++ b/crates/health/src/endpoint/model.rs @@ -68,6 +68,17 @@ impl BmcEndpoint { pub fn switch_data(&self) -> Option<&SwitchData> { self.metadata.as_ref().and_then(EndpointMetadata::as_switch) } + + /// Returns the connect host direct switch collectors should place in URIs. + /// + /// Switch collectors connect to the discovered endpoint IP address. DNS + /// names used for TLS verification are handled separately. + pub fn switch_connect_host_for_uri(&self) -> Cow<'_, str> { + match self.addr.ip { + IpAddr::V4(ip) => Cow::Owned(ip.to_string()), + IpAddr::V6(ip) => Cow::Owned(format!("[{ip}]")), + } + } } #[derive(Clone, Debug)] @@ -216,7 +227,8 @@ mod tests { use mac_address::MacAddress; - use super::BmcAddr; + use super::{BmcAddr, BmcCredentials}; + use crate::endpoint::test_support::endpoint_with_creds; fn addr(ip: &str, port: Option) -> BmcAddr { BmcAddr { @@ -249,4 +261,19 @@ mod tests { assert_eq!(url.scheme(), "http"); assert_eq!(url.host_str(), Some("[2001:db8::1]")); } + + #[test] + fn switch_connect_host_for_uri_brackets_ipv6() { + let endpoint = endpoint_with_creds( + addr("2001:db8::1", Some(443)), + BmcCredentials::UsernamePassword { + username: "admin".to_string(), + password: Some("pass".to_string()), + }, + None, + None, + ); + + assert_eq!(endpoint.switch_connect_host_for_uri(), "[2001:db8::1]"); + } } diff --git a/crates/health/src/lib.rs b/crates/health/src/lib.rs index f46311fc50..73011d2e01 100644 --- a/crates/health/src/lib.rs +++ b/crates/health/src/lib.rs @@ -35,6 +35,8 @@ pub mod processor; pub mod sharding; pub mod sink; +mod tls; + pub use config::Config; pub use discovery::{DiscoveryIterationStats, DiscoveryLoopContext}; @@ -95,6 +97,10 @@ pub enum HealthError { #[error("NMX-C RPC failed: {0}")] NmxcStatus(tonic::Status), + + /// Client TLS material could not be read, validated, or applied. + #[error("mTLS profile error: {0}")] + Tls(#[source] Box), } impl From for HealthError { @@ -109,6 +115,12 @@ impl From for HealthError { } } +impl From for HealthError { + fn from(err: tls::TlsError) -> Self { + HealthError::Tls(Box::new(err)) + } +} + impl From> for HealthError { fn from(err: nv_redfish::Error) -> Self { HealthError::BmcError(Box::new(err)) @@ -257,6 +269,12 @@ fn build_data_sink( } pub async fn run_service(config: Config) -> Result<(), HealthError> { + if let Some(tls_config) = &config.tls.switch { + tls::preflight(tls_config).await?; + } + + let tls_config = config.tls.switch.clone(); + let metrics_endpoint = config.metrics_addr()?; let metrics_manager = Arc::new(MetricsManager::new(&config.metrics.prefix)?); @@ -317,7 +335,12 @@ pub async fn run_service(config: Config) -> Result<(), HealthError> { let endpoint_source = endpoint_source.clone(); let data_sink = data_sink.clone(); - let mut ctx = DiscoveryLoopContext::new(limiter, metrics_manager, config.clone())?; + let mut ctx = DiscoveryLoopContext::new_with_tls_config( + limiter, + metrics_manager, + config.clone(), + tls_config, + )?; async move { loop { diff --git a/crates/health/src/tls.rs b/crates/health/src/tls.rs new file mode 100644 index 0000000000..4d9bd3dbe7 --- /dev/null +++ b/crates/health/src/tls.rs @@ -0,0 +1,722 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +//! TLS helpers for outbound health collector connections. +//! +//! The current profile is used by switch collectors through `[tls.switch]`. +//! This module owns HTTP and gRPC TLS construction inside the health crate so +//! collector transport security does not depend on NICo API certificate +//! settings. Each client build reads and validates the configured certificate +//! files. Periodic HTTP collectors rebuild their mTLS clients once per poll +//! iteration so they adopt changed certificate files on the next scrape. +//! Streaming collectors build a new TLS config when they reconnect. + +use std::fmt; +use std::io::BufReader; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Once}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use rustls::RootCertStore; +use rustls::pki_types::{CertificateDer, PrivateKeyDer}; +use thiserror::Error; +use tokio::fs; +use tonic::transport::{ + Certificate as TonicCertificate, ClientTlsConfig, Identity as TonicIdentity, +}; +use x509_parser::prelude::*; + +use crate::config::MtlsProfileConfig; + +/// Role for one mTLS profile material file. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum TlsMaterialKind { + CaBundle, + ClientCertificate, + ClientKey, +} + +impl fmt::Display for TlsMaterialKind { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::CaBundle => f.write_str("CA bundle"), + Self::ClientCertificate => f.write_str("client certificate"), + Self::ClientKey => f.write_str("client key"), + } + } +} + +#[derive(Debug, Error)] +pub(crate) enum TlsError { + #[error("failed to read mTLS profile {kind} at {path}: {source}")] + Read { + kind: TlsMaterialKind, + path: PathBuf, + #[source] + source: std::io::Error, + }, + + #[error("mTLS profile {kind} at {path} is empty")] + Empty { + kind: TlsMaterialKind, + path: PathBuf, + }, + + #[error("failed to parse mTLS profile {kind} at {path}: {message}")] + Parse { + kind: TlsMaterialKind, + path: PathBuf, + message: String, + }, + + #[error( + "mTLS profile {kind} certificate at {path} is not valid before unix timestamp {not_before}" + )] + NotYetValid { + kind: TlsMaterialKind, + path: PathBuf, + not_before: i64, + }, + + #[error("mTLS profile {kind} certificate at {path} expired at unix timestamp {not_after}")] + Expired { + kind: TlsMaterialKind, + path: PathBuf, + not_after: i64, + }, + + #[error("mTLS profile CA bundle at {path} contains no usable trust anchors")] + NoTrustedCa { path: PathBuf }, + + #[error("mTLS profile client certificate and key are invalid or do not match: {source}")] + InvalidIdentity { + #[source] + source: rustls::Error, + }, + + #[error("failed to create mTLS profile HTTP CA bundle: {source}")] + HttpCaBundle { + #[source] + source: reqwest::Error, + }, + + #[error("failed to create mTLS profile HTTP client identity: {source}")] + HttpIdentity { + #[source] + source: reqwest::Error, + }, + + #[error("failed to create mTLS profile HTTP client: {source}")] + HttpClient { + #[source] + source: reqwest::Error, + }, +} + +struct TlsMaterial { + ca_pem: Vec, + client_cert_pem: Vec, + client_key_pem: Vec, +} + +pub(crate) async fn preflight(config: &MtlsProfileConfig) -> Result<(), TlsError> { + read_validated_material(config).await.map(|_| ()) +} + +/// Builds an HTTP client from the current mTLS profile material. +/// +/// The DNS name in `[tls.switch].tls_server_name` is used only for SNI and +/// certificate verification. When it is set, callers pass the discovered switch +/// socket address so reqwest still connects to the intended switch endpoint. +pub(crate) async fn reqwest_client( + config: &MtlsProfileConfig, + timeout: Duration, + resolve_addr: Option, +) -> Result { + let material = read_validated_material(config).await?; + let ca_certs = reqwest::Certificate::from_pem_bundle(&material.ca_pem) + .map_err(|source| TlsError::HttpCaBundle { source })?; + + let mut identity_pem = + Vec::with_capacity(material.client_cert_pem.len() + material.client_key_pem.len() + 1); + + identity_pem.extend_from_slice(&material.client_cert_pem); + identity_pem.push(b'\n'); + identity_pem.extend_from_slice(&material.client_key_pem); + + let identity = reqwest::Identity::from_pem(&identity_pem) + .map_err(|source| TlsError::HttpIdentity { source })?; + + let mut builder = reqwest::Client::builder() + .timeout(timeout) + .tls_backend_rustls() + .tls_certs_only(ca_certs) + .identity(identity); + + if let (Some(tls_server_name), Some(resolve_addr)) = + (config.tls_server_name.as_deref(), resolve_addr) + { + // The DNS name is for SNI and certificate verification only. Override + // resolution so HTTP still connects to the discovered switch IP. + builder = builder.resolve(tls_server_name, resolve_addr); + } + + builder + .build() + .map_err(|source| TlsError::HttpClient { source }) +} + +/// Builds a tonic client TLS configuration from the current mTLS profile material. +/// +/// Existing gRPC channels keep their current TLS session. Streaming collectors +/// use refreshed files when they reconnect and build a new channel. +pub(crate) async fn tonic_tls_config( + config: &MtlsProfileConfig, +) -> Result { + let material = read_validated_material(config).await?; + + let mut tls_config = ClientTlsConfig::new() + .ca_certificate(TonicCertificate::from_pem(&material.ca_pem)) + .identity(TonicIdentity::from_pem( + &material.client_cert_pem, + &material.client_key_pem, + )); + + if let Some(tls_server_name) = &config.tls_server_name { + // gRPC endpoints still use the discovered switch IP in their URI. This + // override changes only TLS SNI and certificate name verification. + tls_config = tls_config.domain_name(tls_server_name.clone()); + } + + Ok(tls_config) +} + +async fn read_validated_material(config: &MtlsProfileConfig) -> Result { + ensure_rustls_provider(); + + let (ca_pem, client_cert_pem, client_key_pem) = tokio::try_join!( + read_material(TlsMaterialKind::CaBundle, &config.ca_cert_path), + read_material(TlsMaterialKind::ClientCertificate, &config.client_cert_path,), + read_material(TlsMaterialKind::ClientKey, &config.client_key_path), + )?; + + let ca_certs = parse_certificates(TlsMaterialKind::CaBundle, &config.ca_cert_path, &ca_pem)?; + + let client_certs = parse_certificates( + TlsMaterialKind::ClientCertificate, + &config.client_cert_path, + &client_cert_pem, + )?; + + let client_key = parse_private_key(&config.client_key_path, &client_key_pem)?; + let now = unix_timestamp_now(); + + validate_certificate_times( + TlsMaterialKind::CaBundle, + &config.ca_cert_path, + &ca_certs, + now, + )?; + + validate_certificate_times( + TlsMaterialKind::ClientCertificate, + &config.client_cert_path, + &client_certs, + now, + )?; + + validate_root_store(&config.ca_cert_path, ca_certs)?; + validate_client_identity(client_certs, client_key)?; + + Ok(TlsMaterial { + ca_pem, + client_cert_pem, + client_key_pem, + }) +} + +async fn read_material(kind: TlsMaterialKind, path: &Path) -> Result, TlsError> { + let data = fs::read(path).await.map_err(|source| TlsError::Read { + kind, + path: path.to_path_buf(), + source, + })?; + + if data.iter().all(u8::is_ascii_whitespace) { + return Err(TlsError::Empty { + kind, + path: path.to_path_buf(), + }); + } + + Ok(data) +} + +fn parse_certificates( + kind: TlsMaterialKind, + path: &Path, + pem: &[u8], +) -> Result>, TlsError> { + let mut reader = BufReader::new(pem); + let certificates = rustls_pemfile::certs(&mut reader) + .collect::, _>>() + .map_err(|source| TlsError::Parse { + kind, + path: path.to_path_buf(), + message: source.to_string(), + })?; + + if certificates.is_empty() { + return Err(TlsError::Parse { + kind, + path: path.to_path_buf(), + message: "no certificate PEM blocks found".to_string(), + }); + } + + Ok(certificates) +} + +fn parse_private_key(path: &Path, pem: &[u8]) -> Result, TlsError> { + let mut reader = BufReader::new(pem); + rustls_pemfile::private_key(&mut reader) + .map_err(|source| TlsError::Parse { + kind: TlsMaterialKind::ClientKey, + path: path.to_path_buf(), + message: source.to_string(), + })? + .ok_or_else(|| TlsError::Parse { + kind: TlsMaterialKind::ClientKey, + path: path.to_path_buf(), + message: "no supported private key PEM block found".to_string(), + }) +} + +fn validate_certificate_times( + kind: TlsMaterialKind, + path: &Path, + certificates: &[CertificateDer<'static>], + now: i64, +) -> Result<(), TlsError> { + for certificate in certificates { + let (_, x509) = + X509Certificate::from_der(certificate.as_ref()).map_err(|source| TlsError::Parse { + kind, + path: path.to_path_buf(), + message: format!("invalid X.509 certificate: {source}"), + })?; + + let validity = x509.validity(); + let not_before = validity.not_before.timestamp(); + let not_after = validity.not_after.timestamp(); + + if now < not_before { + return Err(TlsError::NotYetValid { + kind, + path: path.to_path_buf(), + not_before, + }); + } + + if now > not_after { + return Err(TlsError::Expired { + kind, + path: path.to_path_buf(), + not_after, + }); + } + } + + Ok(()) +} + +fn validate_root_store( + path: &Path, + certificates: Vec>, +) -> Result<(), TlsError> { + let mut root_store = RootCertStore::empty(); + let (accepted, ignored) = root_store.add_parsable_certificates(certificates); + + if accepted == 0 { + return Err(TlsError::NoTrustedCa { + path: path.to_path_buf(), + }); + } + + if ignored != 0 { + return Err(TlsError::Parse { + kind: TlsMaterialKind::CaBundle, + path: path.to_path_buf(), + message: format!("{ignored} certificate(s) could not be used as trust anchors"), + }); + } + + Ok(()) +} + +fn validate_client_identity( + certificates: Vec>, + key: PrivateKeyDer<'static>, +) -> Result<(), TlsError> { + let root_store = RootCertStore::empty(); + + let builder = rustls::ClientConfig::builder_with_provider(Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions() + .map_err(|source| TlsError::InvalidIdentity { source })? + .with_root_certificates(root_store); + + builder + .with_client_auth_cert(certificates, key) + .map(|_| ()) + .map_err(|source| TlsError::InvalidIdentity { source }) +} + +fn ensure_rustls_provider() { + static INSTALL_PROVIDER: Once = Once::new(); + + INSTALL_PROVIDER.call_once(|| { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + }); +} + +fn unix_timestamp_now() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_secs() as i64) + .unwrap_or_default() +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use carbide_test_support::Outcome::*; + use carbide_test_support::{Case, check_cases_async}; + use rcgen::{ + BasicConstraints, Certificate, CertificateParams, DnType, ExtendedKeyUsagePurpose, IsCa, + Issuer, KeyPair, KeyUsagePurpose, date_time_ymd, + }; + use tempfile::TempDir; + + use super::*; + + struct GeneratedMaterial { + ca_pem: Vec, + client_cert_pem: Vec, + client_key_pem: Vec, + alternate_client_key_pem: Vec, + } + + #[derive(Debug, Eq, PartialEq)] + enum ExpectedTlsError { + Empty(TlsMaterialKind), + Parse(TlsMaterialKind), + InvalidIdentity, + Expired(TlsMaterialKind), + NotYetValid(TlsMaterialKind), + } + + fn valid_material() -> GeneratedMaterial { + material_with_validity( + date_time_ymd(1975, 1, 1), + date_time_ymd(4096, 1, 1), + date_time_ymd(1975, 1, 1), + date_time_ymd(4096, 1, 1), + ) + } + + fn material_with_validity( + ca_not_before: ::time::OffsetDateTime, + ca_not_after: ::time::OffsetDateTime, + client_not_before: ::time::OffsetDateTime, + client_not_after: ::time::OffsetDateTime, + ) -> GeneratedMaterial { + let (ca, issuer) = ca_with_validity(ca_not_before, ca_not_after); + let (client_cert, client_key) = + client_cert_with_validity(&issuer, client_not_before, client_not_after); + + let alternate_key = KeyPair::generate().expect("alternate client key should generate"); + + GeneratedMaterial { + ca_pem: ca.pem().into_bytes(), + client_cert_pem: client_cert.pem().into_bytes(), + client_key_pem: client_key.serialize_pem().into_bytes(), + alternate_client_key_pem: alternate_key.serialize_pem().into_bytes(), + } + } + + fn ca_with_validity( + not_before: ::time::OffsetDateTime, + not_after: ::time::OffsetDateTime, + ) -> (Certificate, Issuer<'static, KeyPair>) { + let mut params = + CertificateParams::new(Vec::new()).expect("empty subject alt names should be valid"); + + params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + params + .distinguished_name + .push(DnType::CommonName, "switch test ca"); + + params.key_usages.push(KeyUsagePurpose::DigitalSignature); + params.key_usages.push(KeyUsagePurpose::KeyCertSign); + params.key_usages.push(KeyUsagePurpose::CrlSign); + + params.not_before = not_before; + params.not_after = not_after; + + let key_pair = KeyPair::generate().expect("CA key should generate"); + let cert = params + .self_signed(&key_pair) + .expect("CA certificate should sign"); + + (cert, Issuer::new(params, key_pair)) + } + + fn client_cert_with_validity( + issuer: &Issuer<'static, KeyPair>, + not_before: ::time::OffsetDateTime, + not_after: ::time::OffsetDateTime, + ) -> (Certificate, KeyPair) { + let mut params = + CertificateParams::new(Vec::new()).expect("empty subject alt names should be valid"); + + params + .distinguished_name + .push(DnType::CommonName, "switch test client"); + + params.key_usages.push(KeyUsagePurpose::DigitalSignature); + params + .extended_key_usages + .push(ExtendedKeyUsagePurpose::ClientAuth); + + params.not_before = not_before; + params.not_after = not_after; + + let key_pair = KeyPair::generate().expect("client key should generate"); + let cert = params + .signed_by(&key_pair, issuer) + .expect("client certificate should sign"); + + (cert, key_pair) + } + + async fn write_material( + dir: &TempDir, + material: &GeneratedMaterial, + ) -> Result { + let ca_cert_path = dir.path().join("ca.crt"); + let client_cert_path = dir.path().join("tls.crt"); + let client_key_path = dir.path().join("tls.key"); + + tokio::fs::write(&ca_cert_path, &material.ca_pem).await?; + tokio::fs::write(&client_cert_path, &material.client_cert_pem).await?; + tokio::fs::write(&client_key_path, &material.client_key_pem).await?; + + Ok(MtlsProfileConfig { + ca_cert_path, + client_cert_path, + client_key_path, + tls_server_name: None, + }) + } + + fn expected_tls_error(error: TlsError) -> ExpectedTlsError { + match error { + TlsError::Empty { kind, .. } => ExpectedTlsError::Empty(kind), + TlsError::Parse { kind, .. } => ExpectedTlsError::Parse(kind), + TlsError::InvalidIdentity { .. } => ExpectedTlsError::InvalidIdentity, + TlsError::Expired { kind, .. } => ExpectedTlsError::Expired(kind), + TlsError::NotYetValid { kind, .. } => ExpectedTlsError::NotYetValid(kind), + error => panic!("unexpected mTLS profile error: {error}"), + } + } + + #[tokio::test] + async fn valid_tls_material_builds_http_and_grpc_clients() + -> Result<(), Box> { + let dir = TempDir::new()?; + let material = valid_material(); + let config = write_material(&dir, &material).await?; + + preflight(&config).await?; + let _http_client = reqwest_client(&config, Duration::from_secs(1), None).await?; + + let _grpc_tls = tonic_tls_config(&config).await?; + + Ok(()) + } + + #[tokio::test] + async fn http_client_build_reads_changed_tls_material() -> Result<(), Box> + { + let dir = TempDir::new()?; + let material = valid_material(); + let config = write_material(&dir, &material).await?; + + preflight(&config).await?; + + let rotated = valid_material(); + tokio::fs::write(&config.ca_cert_path, &rotated.ca_pem).await?; + tokio::fs::write(&config.client_cert_path, &rotated.client_cert_pem).await?; + tokio::fs::write(&config.client_key_path, &rotated.client_key_pem).await?; + + let _http_client = reqwest_client(&config, Duration::from_secs(1), None).await?; + + Ok(()) + } + + #[tokio::test] + async fn http_client_build_reports_invalid_changed_tls_material() + -> Result<(), Box> { + let dir = TempDir::new()?; + let material = valid_material(); + let config = write_material(&dir, &material).await?; + + preflight(&config).await?; + + tokio::fs::write(&config.client_key_path, &material.alternate_client_key_pem).await?; + + let result = reqwest_client(&config, Duration::from_secs(1), None).await; + + assert!(matches!(result, Err(TlsError::InvalidIdentity { .. }))); + + Ok(()) + } + + #[tokio::test] + async fn missing_tls_material_returns_path_and_role() -> Result<(), Box> + { + let dir = TempDir::new()?; + let material = valid_material(); + let config = write_material(&dir, &material).await?; + + tokio::fs::remove_file(&config.client_key_path).await?; + + let result = preflight(&config).await; + + let Err(TlsError::Read { kind, path, .. }) = result else { + panic!("expected read error for missing key"); + }; + + assert_eq!(kind, TlsMaterialKind::ClientKey); + assert_eq!(path, config.client_key_path); + + Ok(()) + } + + #[tokio::test] + async fn invalid_tls_material_returns_clear_errors() -> Result<(), Box> { + check_cases_async( + [ + Case { + scenario: "empty CA bundle", + input: { + let mut material = valid_material(); + material.ca_pem = Vec::new(); + material + }, + expect: FailsWith(ExpectedTlsError::Empty(TlsMaterialKind::CaBundle)), + }, + Case { + scenario: "malformed client certificate", + input: { + let mut material = valid_material(); + material.client_cert_pem = b"not pem".to_vec(); + material + }, + expect: FailsWith(ExpectedTlsError::Parse(TlsMaterialKind::ClientCertificate)), + }, + Case { + scenario: "malformed client key", + input: { + let mut material = valid_material(); + material.client_key_pem = b"not pem".to_vec(); + material + }, + expect: FailsWith(ExpectedTlsError::Parse(TlsMaterialKind::ClientKey)), + }, + Case { + scenario: "mismatched client key", + input: { + let mut material = valid_material(); + material.client_key_pem = material.alternate_client_key_pem.clone(); + material + }, + expect: FailsWith(ExpectedTlsError::InvalidIdentity), + }, + ], + |material| async move { + let dir = TempDir::new().expect("temp dir should create"); + + let config = write_material(&dir, &material) + .await + .expect("mTLS profile material should write"); + + preflight(&config).await.map_err(expected_tls_error) + }, + ) + .await; + + Ok(()) + } + + #[tokio::test] + async fn tls_material_rejects_expired_and_future_certificates() + -> Result<(), Box> { + check_cases_async( + [ + Case { + scenario: "expired client certificate", + input: material_with_validity( + date_time_ymd(1975, 1, 1), + date_time_ymd(4096, 1, 1), + date_time_ymd(2020, 1, 1), + date_time_ymd(2021, 1, 1), + ), + expect: FailsWith(ExpectedTlsError::Expired( + TlsMaterialKind::ClientCertificate, + )), + }, + Case { + scenario: "future CA certificate", + input: material_with_validity( + date_time_ymd(4097, 1, 1), + date_time_ymd(4098, 1, 1), + date_time_ymd(1975, 1, 1), + date_time_ymd(4096, 1, 1), + ), + expect: FailsWith(ExpectedTlsError::NotYetValid(TlsMaterialKind::CaBundle)), + }, + ], + |material| async move { + let dir = TempDir::new().expect("temp dir should create"); + + let config = write_material(&dir, &material) + .await + .expect("mTLS profile material should write"); + + preflight(&config).await.map_err(expected_tls_error) + }, + ) + .await; + + Ok(()) + } +} From 746c2acd08228fbadb941ea948fe2525ee0a3f88 Mon Sep 17 00:00:00 2001 From: Jay Zhu Date: Thu, 9 Jul 2026 01:07:23 -0600 Subject: [PATCH 2/2] perf(health): share switch mTLS HTTP client Reuse one mTLS HTTP client for the shared [tls.switch] profile instead of rebuilding reqwest clients per switch target. This keeps TLS reload bounded to the configured reload window and avoids per-target cert parsing/client construction at scale. Signed-off-by: Jay Zhu --- Cargo.lock | 2 + crates/health/Cargo.toml | 11 +- crates/health/example/config.example.toml | 14 +- crates/health/src/collectors/nmxt.rs | 109 +++--- .../health/src/collectors/nvue/rest/client.rs | 201 ++++++---- .../src/collectors/nvue/rest/collector.rs | 8 +- crates/health/src/config.rs | 5 +- crates/health/src/discovery/context.rs | 38 +- crates/health/src/discovery/spawn.rs | 4 +- crates/health/src/tls.rs | 363 +++++++++++++++--- 10 files changed, 539 insertions(+), 216 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ae1cd47dfb..f3a01e8ca4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1980,6 +1980,7 @@ dependencies = [ "arc-swap", "async-trait", "base64", + "bytes", "carbide-health-report", "carbide-instrument", "carbide-rpc", @@ -1994,6 +1995,7 @@ dependencies = [ "figment", "futures", "http", + "http-body-util", "humantime", "humantime-serde", "hyper", diff --git a/crates/health/Cargo.toml b/crates/health/Cargo.toml index e6cd49b79e..10cdc79c9d 100644 --- a/crates/health/Cargo.toml +++ b/crates/health/Cargo.toml @@ -31,16 +31,23 @@ carbide-instrument = { path = "../instrument" } arc-swap = { workspace = true } async-trait = { workspace = true } base64 = { workspace = true } +bytes = { workspace = true } chrono = { workspace = true } dashmap = { workspace = true } figment = { workspace = true, features = ["toml", "env"] } futures = { workspace = true } http = { workspace = true } +http-body-util = { workspace = true } humantime = { workspace = true } humantime-serde = { workspace = true } hyper = { workspace = true } -hyper-rustls = { workspace = true, features = ["http2"] } -hyper-util = { workspace = true } +hyper-rustls = { workspace = true, features = ["http1", "http2"] } +hyper-util = { workspace = true, features = [ + "client-legacy", + "http1", + "http2", + "tokio", +] } mac_address = { workspace = true } prometheus = { workspace = true } reqwest = { workspace = true, features = ["query", "json", "rustls"] } diff --git a/crates/health/example/config.example.toml b/crates/health/example/config.example.toml index 9a892909f0..c327f35fb6 100644 --- a/crates/health/example/config.example.toml +++ b/crates/health/example/config.example.toml @@ -209,7 +209,7 @@ state_refresh_interval = "30m" logs_state_file = "/tmp/logs_collector_{machine_id}.json" # ============================================================================== -# TLS: Shared certificate configuration for hardware-health connections +# TLS: Shared certificate configuration for switch collectors # ============================================================================== # Optional mTLS profile for direct switch collector connections. @@ -223,14 +223,14 @@ logs_state_file = "/tmp/logs_collector_{machine_id}.json" # switch certificates contain DNS SANs instead of IP SANs, set tls_server_name # here. The value is used only for TLS SNI and certificate verification, not # for DNS resolution or endpoint discovery. -# For HTTP collectors, this name is also the HTTP Host header. Switch HTTP -# services must accept that shared host when tls_server_name is configured. +# For HTTP collectors, the request URL and HTTP Host header stay on the +# discovered switch IP. Only the TLS server name changes. # # Rotation behavior: -# - NMX-T and NVUE REST rebuild one HTTP client per switch-target collector -# on each poll iteration. This keeps reload behavior deterministic and avoids -# global client cache synchronization when switch inventory changes. Cert file -# changes are adopted on the next scrape for that switch target. +# - NMX-T and NVUE REST share one cached HTTP client for this mTLS profile. +# The cache is checked on the shortest enabled HTTP collector poll interval, +# so cert file changes are adopted by the next reload window without building +# a client per switch target. # - NMX-C and NVUE gNMI read refreshed material when a long-lived stream # reconnects. # - Existing HTTP clients and long-lived streams are not proactively closed on diff --git a/crates/health/src/collectors/nmxt.rs b/crates/health/src/collectors/nmxt.rs index 0985b2f913..dbf36cca1b 100644 --- a/crates/health/src/collectors/nmxt.rs +++ b/crates/health/src/collectors/nmxt.rs @@ -29,16 +29,17 @@ use std::borrow::Cow; use std::collections::{HashMap, HashSet}; -use std::net::SocketAddr; use std::sync::Arc; use nv_redfish::core::Bmc; +use url::Url; use crate::HealthError; use crate::collectors::{IterationResult, PeriodicCollector}; -use crate::config::{MtlsProfileConfig, NmxtCollectorConfig as NmxtCollectorOptions}; +use crate::config::NmxtCollectorConfig as NmxtCollectorOptions; use crate::endpoint::BmcEndpoint; use crate::sink::{CollectorEvent, DataSink, EventContext, MetricSample}; +use crate::tls::MtlsHttpClientProvider; /// default NMX-T port const NMXT_PORT: u16 = 9352; @@ -452,6 +453,34 @@ async fn scrape_switch_nmxt_metrics( Ok(parse_prometheus_metrics(&body)) } +async fn scrape_switch_nmxt_metrics_tls( + http_client: &crate::tls::MtlsHttpClient, + switch_ip: &str, + request_timeout: std::time::Duration, +) -> Result, HealthError> { + let url = nmxt_endpoint_url(switch_ip, true); + let url = Url::parse(&url) + .map_err(|e| HealthError::GenericError(format!("{url}: invalid NMX-T URL: {e}")))?; + + let response = http_client + .get(&url, [], request_timeout) + .await + .map_err(|e| HealthError::GenericError(format!("HTTP request failed for {url}: {e}")))?; + + if !response.status.is_success() { + return Err(HealthError::GenericError(format!( + "HTTP request to {} returned status {}", + url, response.status + ))); + } + + let body = std::str::from_utf8(&response.body).map_err(|e| { + HealthError::GenericError(format!("Failed to read response body from {}: {}", url, e)) + })?; + + Ok(parse_prometheus_metrics(body)) +} + fn nmxt_endpoint_url(switch_ip: &str, tls_enabled: bool) -> String { let scheme = if tls_enabled { "https" } else { "http" }; format!("{scheme}://{}:{}{}", switch_ip, NMXT_PORT, NMXT_ENDPOINT) @@ -464,8 +493,8 @@ pub struct NmxtCollectorConfig { /// Optional sink that receives NMX-T metric events. pub data_sink: Option>, - /// mTLS profile used for HTTPS scrapes when configured. - pub(crate) tls_config: Option, + /// Shared mTLS HTTP client provider used for HTTPS scrapes when configured. + pub(crate) tls_http_client_provider: Option, } pub struct NmxtCollector { @@ -479,13 +508,10 @@ pub struct NmxtCollector { enum NmxtHttpClient { Legacy(reqwest::Client), - // Store the HTTP client prepared for the current switch-target iteration. - // This avoids a global cache that must be synchronized with switch inventory - // changes. - Tls { - config: MtlsProfileConfig, - client: Option, - }, + // The shared provider owns mTLS reload and client reuse. Target identity + // stays on the request URL, so switch inventory changes only clone or drop + // this provider. + Tls { provider: MtlsHttpClientProvider }, } impl PeriodicCollector for NmxtCollector { @@ -499,11 +525,8 @@ impl PeriodicCollector for NmxtCollector { let event_context = EventContext::from_endpoint(endpoint.as_ref(), "nmxt"); let request_timeout = config.nmxt_config.request_timeout; - let http_client = match config.tls_config { - Some(tls_config) => NmxtHttpClient::Tls { - config: tls_config, - client: None, - }, + let http_client = match config.tls_http_client_provider { + Some(provider) => NmxtHttpClient::Tls { provider }, None => { let mut http_client_builder = reqwest::Client::builder().timeout(request_timeout); @@ -570,26 +593,19 @@ impl NmxtCollector { } async fn scrape_iteration(&mut self) -> Result<(), HealthError> { - let tls_enabled = matches!(self.http_client, NmxtHttpClient::Tls { .. }); - - let switch_connect_host = self.endpoint.switch_connect_host_for_uri(); - - // When `tls_server_name` is configured, the URL host and HTTP Host - // header use that shared TLS identity. Reqwest resolves it to the - // discovered switch IP through the target-scoped client override below. - // This matches current switch behavior, which accepts the shared host. - let scrape_host = match &self.http_client { - NmxtHttpClient::Tls { config, .. } => config - .tls_server_name - .as_deref() - .unwrap_or_else(|| switch_connect_host.as_ref()) - .to_string(), - NmxtHttpClient::Legacy(_) => switch_connect_host.to_string(), - }; + let switch_connect_host = self.endpoint.switch_connect_host_for_uri().to_string(); - let http_client = self.http_client().await?; + let metrics = match &mut self.http_client { + NmxtHttpClient::Legacy(client) => { + scrape_switch_nmxt_metrics(client, &switch_connect_host, false).await? + } + NmxtHttpClient::Tls { provider } => { + let client = provider.client().await?; - let metrics = scrape_switch_nmxt_metrics(&http_client, &scrape_host, tls_enabled).await?; + scrape_switch_nmxt_metrics_tls(&client, &switch_connect_host, self.request_timeout) + .await? + } + }; self.emit_event(CollectorEvent::MetricCollectionStart); @@ -693,31 +709,6 @@ impl NmxtCollector { Ok(()) } - - async fn http_client(&mut self) -> Result { - let resolve_addr = SocketAddr::new(self.endpoint.addr.ip, NMXT_PORT); - - match &mut self.http_client { - NmxtHttpClient::Legacy(client) => Ok(client.clone()), - NmxtHttpClient::Tls { config, client } => { - // Rebuild once per poll iteration so mTLS material changes are - // picked up on the next scrape without global client cache - // synchronization. - let built_client = crate::tls::reqwest_client( - config, - self.request_timeout, - config.tls_server_name.as_ref().map(|_| resolve_addr), - ) - .await?; - - let cloned_client = built_client.clone(); - - *client = Some(built_client); - - Ok(cloned_client) - } - } - } } #[cfg(test)] diff --git a/crates/health/src/collectors/nvue/rest/client.rs b/crates/health/src/collectors/nvue/rest/client.rs index 426e68b49b..df0fe48faa 100644 --- a/crates/health/src/collectors/nvue/rest/client.rs +++ b/crates/health/src/collectors/nvue/rest/client.rs @@ -16,19 +16,22 @@ */ use std::collections::HashMap; -use std::net::{IpAddr, SocketAddr}; +use std::net::IpAddr; use std::sync::Arc; use std::time::Duration; use arc_swap::ArcSwapOption; +use base64::Engine as _; +use http::HeaderValue; +use http::header::{ACCEPT, AUTHORIZATION}; use reqwest::Client; -use reqwest::header::ACCEPT; use serde::Deserialize; use serde::de::Error as _; use url::Url; use crate::HealthError; -use crate::config::{MtlsProfileConfig, NvueRestPaths}; +use crate::config::NvueRestPaths; +use crate::tls::{MtlsHttpClient, MtlsHttpClientProvider}; const NVUE_SYSTEM_HEALTH: &str = "/nvue_v1/system/health"; const NVUE_SYSTEM_REBOOT_REASON: &str = "/nvue_v1/system/reboot/reason"; @@ -83,19 +86,19 @@ pub struct RestClient { enum RestHttpClient { Legacy(Client), - // Store the HTTP client prepared for the current switch-target iteration. - // The resolver override is target-specific when `[tls.switch].tls_server_name` - // is set, so sharing one reqwest client across switch IPs would be incorrect. + // Share the mTLS HTTP client provider across switch targets. Hyper-rustls + // applies `[tls.switch].tls_server_name` only to SNI and certificate + // verification, so the request URL and HTTP Host header remain the + // discovered switch IP. Tls { - config: MtlsProfileConfig, - request_timeout: Duration, + provider: MtlsHttpClientProvider, - // When set, base_url uses `[tls.switch].tls_server_name` for SNI and - // certificate verification, while reqwest resolves it to this socket - // address locally. - resolve_addr: Option, + // Per-iteration client clone prepared by `ensure_http_client`. It is + // cleared before refresh so an expired reload window cannot fall back + // to stale TLS material in the same collector iteration. + current_client: ArcSwapOption, - client: Option, + request_timeout: Duration, }, } @@ -106,29 +109,15 @@ impl RestClient { port: Option, request_timeout: Duration, self_signed_tls: bool, - tls_config: Option, + tls_http_client_provider: Option, paths: NvueRestPaths, ) -> Result { - let tls_server_name = tls_config - .as_ref() - .and_then(|config| config.tls_server_name.clone()); - let port = port.unwrap_or(443); - let resolve_addr = tls_server_name - .as_ref() - .map(|_| SocketAddr::new(connect_ip, port)); - - // When `tls_server_name` is configured, the URL host and HTTP Host - // header use that shared TLS identity. Reqwest resolves it to the - // discovered switch IP through the target-scoped client override below. - // This matches current switch behavior, which accepts the shared host. - let host = tls_server_name - .as_deref() - .map(ToOwned::to_owned) - .unwrap_or_else(|| match connect_ip { - IpAddr::V4(ip) => ip.to_string(), - IpAddr::V6(ip) => format!("[{ip}]"), - }); + + let host = match connect_ip { + IpAddr::V4(ip) => ip.to_string(), + IpAddr::V6(ip) => format!("[{ip}]"), + }; let raw_url = if port == 443 { format!("https://{host}") @@ -139,12 +128,11 @@ impl RestClient { let base_url = Url::parse(&raw_url) .map_err(|e| HealthError::HttpError(format!("{raw_url}: invalid base URL: {e}")))?; - let http_client = match tls_config { - Some(config) => RestHttpClient::Tls { - config, + let http_client = match tls_http_client_provider { + Some(provider) => RestHttpClient::Tls { + provider, + current_client: ArcSwapOption::empty(), request_timeout, - resolve_addr, - client: None, }, None => { let mut builder = Client::builder().timeout(request_timeout); @@ -194,21 +182,26 @@ impl RestClient { }) } - /// Rebuilds the target-scoped HTTP client before NVUE REST requests. + /// Checks the shared mTLS HTTP client cache before NVUE REST requests. /// - /// When an mTLS profile is configured, the client is rebuilt once per poll - /// iteration. This keeps certificate reload behavior deterministic and - /// avoids a global cache that would need switch inventory synchronization. + /// When an mTLS profile is configured, this asks the shared provider to + /// refresh at most once per reload window before the collector starts + /// issuing target-specific requests. pub async fn ensure_http_client(&mut self) -> Result<(), HealthError> { if let RestHttpClient::Tls { - config, - request_timeout, - resolve_addr, - client, + provider, + current_client, + .. } = &mut self.http_client { - *client = - Some(crate::tls::reqwest_client(config, *request_timeout, *resolve_addr).await?); + // Clear the old clone before refresh so a failed reload cannot be + // followed by accidental use of stale cert material in this + // collector iteration. + current_client.store(None); + + let client = provider.client().await?; + + current_client.store(Some(Arc::new(client))); } Ok(()) @@ -353,53 +346,99 @@ impl RestClient { url: Url, extra_query: &[(&str, &str)], ) -> Result { - let client = match &self.http_client { - RestHttpClient::Legacy(client) => client, - RestHttpClient::Tls { - client: Some(client), - .. - } => client, - RestHttpClient::Tls { .. } => { - return Err(HealthError::HttpError(format!( - "{url}: mTLS HTTP client was not prepared for switch {}", - self.switch_id - ))); - } - }; + let mut url = url; - let mut request = client.get(url.as_str()); + // GET /interface (returning a collection) defaults to rev=applied, not + // operational. There is inconsistency across the NVUE endpoints, so we + // need to check each. We want the actual system state (rev=operational), + // rather than defaults or what's configured (rev=applied). + url.query_pairs_mut().append_pair("rev", "operational"); - // GET /interface (returning a collection) defaults to rev=applied, not operational. - // There is inconsistency across the NVUE Endpoints, so we need to check each. - // We want the actual system state (rev=operational), rather than defaults or what's configured (rev=applied). - request = request.query(&[("rev", "operational")]); if !extra_query.is_empty() { - request = request.query(extra_query); + url.query_pairs_mut() + .extend_pairs(extra_query.iter().copied()); } - if let Some(creds) = self.credentials.load_full() { - request = request.basic_auth(&creds.username, creds.password.as_ref()); - } + let (status, body) = match &self.http_client { + RestHttpClient::Legacy(client) => { + let mut request = client + .get(url.as_str()) + .header("accept", "application/json"); - request = request.header(ACCEPT, "application/json"); + if let Some(creds) = self.credentials.load_full() { + request = request.basic_auth(&creds.username, creds.password.as_ref()); + } - let response = request.send().await.map_err(|e| { - HealthError::HttpError(format!( - "{url}: request failed for switch {}: {e}", - self.switch_id - )) - })?; + let response = request.send().await.map_err(|e| { + HealthError::HttpError(format!( + "{url}: request failed for switch {}: {e}", + self.switch_id + )) + })?; + + let status = response.status(); + + let body = response.bytes().await.map_err(|e| { + HealthError::HttpError(format!( + "{url}: failed to read response for switch {}: {e}", + self.switch_id + )) + })?; + + (status, body) + } + RestHttpClient::Tls { + provider, + current_client, + request_timeout, + } => { + let client = match current_client.load_full() { + Some(client) => client, + None => Arc::new(provider.client().await?), + }; + + let mut headers = vec![(ACCEPT, HeaderValue::from_static("application/json"))]; + + if let Some(creds) = self.credentials.load_full() { + let password = creds.password.as_deref().unwrap_or_default(); + let encoded = base64::engine::general_purpose::STANDARD + .encode(format!("{}:{password}", creds.username)); + + let value = + HeaderValue::from_str(&format!("Basic {encoded}")).map_err(|e| { + HealthError::HttpError(format!( + "{url}: failed to build authorization header for switch {}: {e}", + self.switch_id + )) + })?; + + headers.push((AUTHORIZATION, value)); + } + + let response = client + .get(&url, headers, *request_timeout) + .await + .map_err(|e| { + HealthError::HttpError(format!( + "{url}: request failed for switch {}: {e}", + self.switch_id + )) + })?; + + (response.status, response.body) + } + }; + + if !status.is_success() { + let body = String::from_utf8_lossy(&body); - if !response.status().is_success() { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); return Err(HealthError::HttpError(format!( "{url}: HTTP {status} for switch {}: {body}", self.switch_id ))); } - response.json().await.map_err(|e| { + serde_json::from_slice(&body).map_err(|e| { HealthError::HttpError(format!( "{url}: failed to parse response for switch {}: {e}", self.switch_id diff --git a/crates/health/src/collectors/nvue/rest/collector.rs b/crates/health/src/collectors/nvue/rest/collector.rs index 28e1243efe..e1b6cbd0e1 100644 --- a/crates/health/src/collectors/nvue/rest/collector.rs +++ b/crates/health/src/collectors/nvue/rest/collector.rs @@ -25,7 +25,7 @@ use super::client::{ use crate::HealthError; use crate::bmc::{CREDENTIAL_REFRESH_TIMEOUT, CredentialProvider, is_auth_error}; use crate::collectors::{IterationResult, PeriodicCollector}; -use crate::config::{MtlsProfileConfig, NvueRestConfig}; +use crate::config::NvueRestConfig; use crate::endpoint::{BmcAddr, BmcCredentials, BmcEndpoint, EndpointMetadata}; use crate::sink::{ Classification, CollectorEvent, DataSink, EventContext, HealthReport, HealthReportAlert, @@ -150,8 +150,8 @@ pub struct NvueRestCollectorConfig { /// Credential source used to authenticate to NVUE REST. pub credential_provider: Arc, - /// mTLS profile used for HTTPS polling when configured. - pub(crate) tls_config: Option, + /// Shared mTLS HTTP client provider used for HTTPS polling when configured. + pub(crate) tls_http_client_provider: Option, } pub struct NvueRestCollector { @@ -186,7 +186,7 @@ impl PeriodicCollector for NvueRestCollector { endpoint.addr.port, rest_cfg.request_timeout, true, - config.tls_config, + config.tls_http_client_provider, rest_cfg.paths.clone(), )?; diff --git a/crates/health/src/config.rs b/crates/health/src/config.rs index 44f489bae0..11ba3fc9b1 100644 --- a/crates/health/src/config.rs +++ b/crates/health/src/config.rs @@ -610,9 +610,8 @@ pub struct MtlsProfileConfig { /// instead of requiring every switch certificate to include an IP SAN. /// This value is never used for endpoint discovery or DNS resolution. /// - /// For HTTP collectors, this name is also the request URL host and HTTP - /// Host header. Switch HTTP services must accept that host when this field - /// is configured. + /// For HTTP collectors, the request URL and HTTP Host header stay on the + /// discovered switch IP. Only the TLS server name changes. pub tls_server_name: Option, } diff --git a/crates/health/src/discovery/context.rs b/crates/health/src/discovery/context.rs index e9c7ca9129..0ea6fa1c21 100644 --- a/crates/health/src/discovery/context.rs +++ b/crates/health/src/discovery/context.rs @@ -18,6 +18,7 @@ use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::sync::Arc; +use std::time::Duration; use arc_swap::ArcSwapOption; use prometheus::{Histogram, HistogramOpts}; @@ -35,6 +36,7 @@ use crate::config::{ }; use crate::limiter::RateLimiter; use crate::metrics::{MetricsManager, operation_duration_buckets_seconds}; +use crate::tls::MtlsHttpClientProvider; #[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)] pub(super) enum CollectorKind { @@ -233,6 +235,7 @@ pub struct DiscoveryLoopContext { pub(crate) nmxc_config: Configurable, pub(crate) nvue_config: Configurable, pub(crate) tls_config: Option, + pub(crate) tls_http_client_provider: Option, /// Whether any enabled sink consumes `CollectorEvent::Log` payloads. pub(crate) log_event_sink_enabled: bool, @@ -279,6 +282,17 @@ impl DiscoveryLoopContext { )?; registry.register(Box::new(discovery_endpoint_fetch_histogram.clone()))?; + let tls_config = tls_config.or_else(|| config.tls.switch.clone()); + + // Periodic HTTP switch collectors share one provider because + // `[tls.switch]` is a single profile. The shortest enabled HTTP poll + // interval bounds cert reload staleness without rebuilding a client per + // switch target. + let tls_http_client_provider = tls_config.clone().and_then(|tls_config| { + switch_http_reload_interval(&config) + .map(|reload_interval| MtlsHttpClientProvider::new(tls_config, reload_interval)) + }); + Ok(Self { collectors: CollectorState::new(), discovery_iteration_histogram, @@ -294,7 +308,8 @@ impl DiscoveryLoopContext { nmxt_config: config.collectors.nmxt.clone(), nmxc_config: config.collectors.nmxc.clone(), nvue_config: config.collectors.nvue.clone(), - tls_config: tls_config.or_else(|| config.tls.switch.clone()), + tls_config, + tls_http_client_provider, log_event_sink_enabled: config.sinks.includes_log_events(), log_downgrade_registry: Arc::new(LogDowngradeRegistry::new()), logs_include_diagnostics: config.sinks.includes_log_diagnostics(), @@ -302,6 +317,27 @@ impl DiscoveryLoopContext { } } +/// Returns the cadence at which periodic HTTP switch collectors reload mTLS material. +fn switch_http_reload_interval(config: &Config) -> Option { + let mut reload_interval = None; + + if let Configurable::Enabled(nmxt_config) = &config.collectors.nmxt { + reload_interval = Some(nmxt_config.scrape_interval); + } + + if let Configurable::Enabled(nvue_config) = &config.collectors.nvue + && let Configurable::Enabled(rest_config) = &nvue_config.rest + { + reload_interval = Some( + reload_interval + .map(|existing: Duration| existing.min(rest_config.poll_interval)) + .unwrap_or(rest_config.poll_interval), + ); + } + + reload_interval +} + #[cfg(test)] mod tests { use std::borrow::Cow; diff --git a/crates/health/src/discovery/spawn.rs b/crates/health/src/discovery/spawn.rs index ffa9e90e16..0ceb23dd2d 100644 --- a/crates/health/src/discovery/spawn.rs +++ b/crates/health/src/discovery/spawn.rs @@ -454,7 +454,7 @@ fn spawn_switch_host_collectors( NmxtCollectorConfig { nmxt_config: nmxt_cfg.clone(), data_sink: data_sink.clone(), - tls_config: ctx.tls_config.clone(), + tls_http_client_provider: ctx.tls_http_client_provider.clone(), }, CollectorStartContext { limiter: ctx.limiter.clone(), @@ -557,7 +557,7 @@ fn spawn_switch_host_collectors( rest_config: rest_cfg.clone(), data_sink: data_sink.clone(), credential_provider, - tls_config: ctx.tls_config.clone(), + tls_http_client_provider: ctx.tls_http_client_provider.clone(), }, CollectorStartContext { limiter: ctx.limiter.clone(), diff --git a/crates/health/src/tls.rs b/crates/health/src/tls.rs index 4d9bd3dbe7..a1add1aa7d 100644 --- a/crates/health/src/tls.rs +++ b/crates/health/src/tls.rs @@ -20,29 +20,39 @@ //! The current profile is used by switch collectors through `[tls.switch]`. //! This module owns HTTP and gRPC TLS construction inside the health crate so //! collector transport security does not depend on NICo API certificate -//! settings. Each client build reads and validates the configured certificate -//! files. Periodic HTTP collectors rebuild their mTLS clients once per poll -//! iteration so they adopt changed certificate files on the next scrape. -//! Streaming collectors build a new TLS config when they reconnect. +//! settings. Periodic HTTP collectors share one cached mTLS client per profile; +//! the cache refreshes on the configured reload cadence so certificate changes +//! are adopted without rebuilding a client for every switch target. Streaming +//! collectors build a new TLS config when they reconnect. use std::fmt; use std::io::BufReader; -use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::sync::{Arc, Once}; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; - +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use bytes::Bytes; +use http::{HeaderName, HeaderValue, Method, Request, StatusCode}; +use http_body_util::{BodyExt, Empty}; +use hyper_rustls::{FixedServerNameResolver, HttpsConnector}; +use hyper_util::client::legacy::Client as HyperClient; +use hyper_util::client::legacy::connect::HttpConnector; +use hyper_util::rt::TokioExecutor; use rustls::RootCertStore; -use rustls::pki_types::{CertificateDer, PrivateKeyDer}; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName}; use thiserror::Error; use tokio::fs; +use tokio::sync::RwLock; use tonic::transport::{ Certificate as TonicCertificate, ClientTlsConfig, Identity as TonicIdentity, }; +use url::Url; use x509_parser::prelude::*; use crate::config::MtlsProfileConfig; +type HyperMtlsClient = HyperClient, Empty>; + /// Role for one mTLS profile material file. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum TlsMaterialKind { @@ -108,77 +118,280 @@ pub(crate) enum TlsError { #[source] source: rustls::Error, }, +} - #[error("failed to create mTLS profile HTTP CA bundle: {source}")] - HttpCaBundle { +struct TlsMaterial { + ca_pem: Vec, + client_cert_pem: Vec, + client_key_pem: Vec, +} + +pub(crate) async fn preflight(config: &MtlsProfileConfig) -> Result<(), TlsError> { + read_validated_material(config).await.map(|_| ()) +} + +/// Cloneable HTTP client built from one validated mTLS profile. +/// +/// The client owns its connection pool and TLS configuration. Clones are cheap +/// handles to the same pool; TLS material changes require building a new +/// `MtlsHttpClient`. +#[derive(Clone)] +pub(crate) struct MtlsHttpClient { + inner: HyperMtlsClient, +} + +/// Reloading provider for the shared switch mTLS HTTP client. +/// +/// One provider is shared by all periodic HTTP switch collectors using the same +/// `[tls.switch]` profile. The provider rebuilds the underlying HTTP client +/// only after the reload interval expires, so cert file reads, parsing, and +/// connection-pool construction are not repeated per switch target. +#[derive(Clone)] +pub(crate) struct MtlsHttpClientProvider { + inner: Arc, +} + +struct MtlsHttpClientProviderInner { + config: MtlsProfileConfig, + reload_interval: Duration, + state: RwLock, +} + +struct MtlsHttpClientProviderState { + /// Cached client for the most recently accepted TLS material. + client: Option, + + /// Instant at which callers must attempt to rebuild from disk again. + next_reload: Option, +} + +/// Fully buffered response from the small mTLS HTTP client wrapper. +pub(crate) struct MtlsHttpResponse { + pub(crate) status: StatusCode, + pub(crate) body: Bytes, +} + +#[derive(Debug, Error)] +pub(crate) enum MtlsHttpError { + #[error("failed to create mTLS HTTP request: {source}")] + Request { #[source] - source: reqwest::Error, + source: http::Error, }, - #[error("failed to create mTLS profile HTTP client identity: {source}")] - HttpIdentity { + #[error("failed to execute mTLS HTTP request: {source}")] + Execute { #[source] - source: reqwest::Error, + source: hyper_util::client::legacy::Error, }, - #[error("failed to create mTLS profile HTTP client: {source}")] - HttpClient { + #[error("failed to read mTLS HTTP response body: {source}")] + Body { #[source] - source: reqwest::Error, + source: hyper::Error, + }, + + #[error("mTLS HTTP request timed out after {timeout:?}")] + Timeout { + timeout: Duration, + #[source] + source: tokio::time::error::Elapsed, }, } -struct TlsMaterial { - ca_pem: Vec, - client_cert_pem: Vec, - client_key_pem: Vec, +impl MtlsHttpClient { + /// Sends one GET request and buffers the whole response body. + /// + /// `request_timeout` covers both request execution and response body + /// collection so callers keep the same timeout shape as the previous + /// `reqwest` implementation. + pub(crate) async fn get( + &self, + url: &Url, + headers: impl IntoIterator, + request_timeout: Duration, + ) -> Result { + let mut builder = Request::builder().method(Method::GET).uri(url.as_str()); + + for (name, value) in headers { + builder = builder.header(name, value); + } + + let request = builder + .body(Empty::::new()) + .map_err(|source| MtlsHttpError::Request { source })?; + + let response = tokio::time::timeout(request_timeout, async { + let response = self + .inner + .request(request) + .await + .map_err(|source| MtlsHttpError::Execute { source })?; + + let status = response.status(); + let body = response + .into_body() + .collect() + .await + .map_err(|source| MtlsHttpError::Body { source })? + .to_bytes(); + + Ok(MtlsHttpResponse { status, body }) + }) + .await + .map_err(|source| MtlsHttpError::Timeout { + timeout: request_timeout, + source, + })??; + + Ok(response) + } } -pub(crate) async fn preflight(config: &MtlsProfileConfig) -> Result<(), TlsError> { - read_validated_material(config).await.map(|_| ()) +impl MtlsHttpClientProvider { + /// Creates a provider for one mTLS profile and reload cadence. + /// + /// The cadence is chosen by the discovery loop from the shortest enabled + /// periodic HTTP switch collector interval. + pub(crate) fn new(config: MtlsProfileConfig, reload_interval: Duration) -> Self { + Self { + inner: Arc::new(MtlsHttpClientProviderInner { + config, + reload_interval, + state: RwLock::new(MtlsHttpClientProviderState { + client: None, + next_reload: None, + }), + }), + } + } + + /// Returns the cached client or rebuilds it from current certificate files. + /// + /// Once the reload window expires, callers must see the new material or the + /// reload error. A failed reload does not advance `next_reload`, so the + /// provider does not silently keep returning stale cert material after the + /// configured reload point. + pub(crate) async fn client(&self) -> Result { + let now = Instant::now(); + + { + let state = self.inner.state.read().await; + if let (Some(client), Some(next_reload)) = (&state.client, state.next_reload) + && now < next_reload + { + return Ok(client.clone()); + } + } + + let mut state = self.inner.state.write().await; + let now = Instant::now(); + + if let (Some(client), Some(next_reload)) = (&state.client, state.next_reload) + && now < next_reload + { + return Ok(client.clone()); + } + + // Hold the Tokio write lock while rebuilding so concurrent switch + // collectors collapse into one file read, cert validation, and client + // construction. This is the expensive operation this cache exists to + // serialize. + let client = http_client(&self.inner.config).await?; + + state.next_reload = Some(now + self.inner.reload_interval); + state.client = Some(client.clone()); + + Ok(client) + } } /// Builds an HTTP client from the current mTLS profile material. /// -/// The DNS name in `[tls.switch].tls_server_name` is used only for SNI and -/// certificate verification. When it is set, callers pass the discovered switch -/// socket address so reqwest still connects to the intended switch endpoint. -pub(crate) async fn reqwest_client( - config: &MtlsProfileConfig, - timeout: Duration, - resolve_addr: Option, -) -> Result { +/// The URL host remains the discovered switch IP. When +/// `[tls.switch].tls_server_name` is set, hyper-rustls uses that value only for +/// SNI and certificate verification. +pub(crate) async fn http_client(config: &MtlsProfileConfig) -> Result { let material = read_validated_material(config).await?; - let ca_certs = reqwest::Certificate::from_pem_bundle(&material.ca_pem) - .map_err(|source| TlsError::HttpCaBundle { source })?; + let tls_config = http_tls_config(config, &material)?; + let mut http_connector = HttpConnector::new(); + + http_connector.enforce_http(false); + + let connector = match &config.tls_server_name { + Some(tls_server_name) => { + let server_name = ServerName::try_from(tls_server_name.clone()).map_err(|source| { + TlsError::Parse { + kind: TlsMaterialKind::CaBundle, + path: config.ca_cert_path.clone(), + message: format!("invalid TLS server name: {source}"), + } + })?; - let mut identity_pem = - Vec::with_capacity(material.client_cert_pem.len() + material.client_key_pem.len() + 1); + hyper_rustls::HttpsConnectorBuilder::new() + .with_tls_config(tls_config) + .https_only() + .with_server_name_resolver(FixedServerNameResolver::new(server_name)) + .enable_http1() + .enable_http2() + .wrap_connector(http_connector) + } + None => hyper_rustls::HttpsConnectorBuilder::new() + .with_tls_config(tls_config) + .https_only() + .enable_http1() + .enable_http2() + .wrap_connector(http_connector), + }; - identity_pem.extend_from_slice(&material.client_cert_pem); - identity_pem.push(b'\n'); - identity_pem.extend_from_slice(&material.client_key_pem); + Ok(MtlsHttpClient { + inner: HyperClient::builder(TokioExecutor::new()).build(connector), + }) +} - let identity = reqwest::Identity::from_pem(&identity_pem) - .map_err(|source| TlsError::HttpIdentity { source })?; +fn http_tls_config( + config: &MtlsProfileConfig, + material: &TlsMaterial, +) -> Result { + let ca_certs = parse_certificates( + TlsMaterialKind::CaBundle, + &config.ca_cert_path, + &material.ca_pem, + )?; - let mut builder = reqwest::Client::builder() - .timeout(timeout) - .tls_backend_rustls() - .tls_certs_only(ca_certs) - .identity(identity); + let mut root_store = RootCertStore::empty(); + let (accepted, ignored) = root_store.add_parsable_certificates(ca_certs); - if let (Some(tls_server_name), Some(resolve_addr)) = - (config.tls_server_name.as_deref(), resolve_addr) - { - // The DNS name is for SNI and certificate verification only. Override - // resolution so HTTP still connects to the discovered switch IP. - builder = builder.resolve(tls_server_name, resolve_addr); + if accepted == 0 { + return Err(TlsError::NoTrustedCa { + path: config.ca_cert_path.clone(), + }); } - builder - .build() - .map_err(|source| TlsError::HttpClient { source }) + if ignored != 0 { + return Err(TlsError::Parse { + kind: TlsMaterialKind::CaBundle, + path: config.ca_cert_path.clone(), + message: format!("{ignored} certificate(s) could not be used as trust anchors"), + }); + } + + let client_certs = parse_certificates( + TlsMaterialKind::ClientCertificate, + &config.client_cert_path, + &material.client_cert_pem, + )?; + + let client_key = parse_private_key(&config.client_key_path, &material.client_key_pem)?; + + rustls::ClientConfig::builder_with_provider(Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions() + .map_err(|source| TlsError::InvalidIdentity { source })? + .with_root_certificates(root_store) + .with_client_auth_cert(client_certs, client_key) + .map_err(|source| TlsError::InvalidIdentity { source }) } /// Builds a tonic client TLS configuration from the current mTLS profile material. @@ -556,7 +769,7 @@ mod tests { let config = write_material(&dir, &material).await?; preflight(&config).await?; - let _http_client = reqwest_client(&config, Duration::from_secs(1), None).await?; + let _http_client = http_client(&config).await?; let _grpc_tls = tonic_tls_config(&config).await?; @@ -577,7 +790,7 @@ mod tests { tokio::fs::write(&config.client_cert_path, &rotated.client_cert_pem).await?; tokio::fs::write(&config.client_key_path, &rotated.client_key_pem).await?; - let _http_client = reqwest_client(&config, Duration::from_secs(1), None).await?; + let _http_client = http_client(&config).await?; Ok(()) } @@ -593,7 +806,43 @@ mod tests { tokio::fs::write(&config.client_key_path, &material.alternate_client_key_pem).await?; - let result = reqwest_client(&config, Duration::from_secs(1), None).await; + let result = http_client(&config).await; + + assert!(matches!(result, Err(TlsError::InvalidIdentity { .. }))); + + Ok(()) + } + + #[tokio::test] + async fn http_client_provider_reuses_cached_client_until_reload_interval() + -> Result<(), Box> { + let dir = TempDir::new()?; + let material = valid_material(); + let config = write_material(&dir, &material).await?; + let provider = MtlsHttpClientProvider::new(config.clone(), Duration::from_secs(3600)); + + let _http_client = provider.client().await?; + + tokio::fs::write(&config.client_key_path, &material.alternate_client_key_pem).await?; + + let _cached_http_client = provider.client().await?; + + Ok(()) + } + + #[tokio::test] + async fn http_client_provider_reloads_after_reload_interval() + -> Result<(), Box> { + let dir = TempDir::new()?; + let material = valid_material(); + let config = write_material(&dir, &material).await?; + let provider = MtlsHttpClientProvider::new(config.clone(), Duration::ZERO); + + let _http_client = provider.client().await?; + + tokio::fs::write(&config.client_key_path, &material.alternate_client_key_pem).await?; + + let result = provider.client().await; assert!(matches!(result, Err(TlsError::InvalidIdentity { .. })));