From 361107e72b66be984e94718c39a85b36e3892e3d Mon Sep 17 00:00:00 2001 From: Jay Zhu Date: Mon, 13 Jul 2026 22:34:28 -0600 Subject: [PATCH] feat(health): add TLS and mTLS support for the OTLP sink Signed-off-by: Jay Zhu --- Cargo.lock | 1 + crates/health/Cargo.toml | 2 +- crates/health/example/config.example.toml | 26 +- crates/health/src/config.rs | 190 +++++++++++++- crates/health/src/lib.rs | 19 +- crates/health/src/otlp/drain.rs | 81 ++++-- crates/health/src/otlp/metrics_drain.rs | 81 ++++-- crates/health/src/otlp/mod.rs | 179 +++++++++++++- crates/health/src/sink/otlp.rs | 2 + crates/health/src/tls.rs | 286 +++++++++++++++++----- 10 files changed, 760 insertions(+), 107 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3e2b9a12ac..d817023818 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11695,6 +11695,7 @@ dependencies = [ "hyper-util", "percent-encoding", "pin-project", + "rustls-native-certs", "socket2", "sync_wrapper", "tokio", diff --git a/crates/health/Cargo.toml b/crates/health/Cargo.toml index 67997cada4..f709f2ac5d 100644 --- a/crates/health/Cargo.toml +++ b/crates/health/Cargo.toml @@ -67,7 +67,7 @@ tracing-subscriber = { features = [ ], workspace = true } thiserror = { workspace = true } tokio = { workspace = true } -tonic = { workspace = true } +tonic = { workspace = true, features = ["tls-native-roots"] } tonic-prost = { workspace = true } prost = { workspace = true } prost-types = { workspace = true } diff --git a/crates/health/example/config.example.toml b/crates/health/example/config.example.toml index dd7f8b428a..7aaa2a25a0 100644 --- a/crates/health/example/config.example.toml +++ b/crates/health/example/config.example.toml @@ -99,7 +99,8 @@ max_backups = 5 [sinks.otlp] enabled = false -# Each target owns its endpoint, queues, batching policy, and diagnostics policy. +# Each target owns its endpoint, queues, batching policy, diagnostics policy, +# and optional TLS policy. This active target uses plaintext HTTP. # Add more target tables for fan-out. [[sinks.otlp.targets]] endpoint = "http://localhost:4317" @@ -112,11 +113,32 @@ flush_interval = "2s" # latest-wins per endpoint while the drain is backed up. include_diagnostics = false # +# TLS-only target with an explicit CA bundle. +# The reload interval is optional and defaults to five minutes. # [[sinks.otlp.targets]] -# endpoint = "http://telemetry.example.com:4317" +# endpoint = "https://telemetry.example.com:4317" # batch_size = 1024 # flush_interval = "5s" # include_diagnostics = true +# +# [sinks.otlp.targets.tls] +# ca_cert_path = "/var/run/secrets/central-otlp/ca.crt" +# tls_server_name = "telemetry.example.com" +# reload_interval = "5m" +# +# mTLS target with an explicit CA bundle and client identity. +# [[sinks.otlp.targets]] +# endpoint = "https://telemetry.example.com:4317" +# batch_size = 1024 +# flush_interval = "5s" +# include_diagnostics = true +# +# [sinks.otlp.targets.tls] +# ca_cert_path = "/var/run/secrets/central-otlp/ca.crt" +# client_cert_path = "/var/run/secrets/central-otlp/tls.crt" +# client_key_path = "/var/run/secrets/central-otlp/tls.key" +# tls_server_name = "telemetry.example.com" +# reload_interval = "5m" [sinks.health_report] root_ca = "/var/run/secrets/spiffe.io/ca.crt" diff --git a/crates/health/src/config.rs b/crates/health/src/config.rs index 056bbfaef1..4d1c38162b 100644 --- a/crates/health/src/config.rs +++ b/crates/health/src/config.rs @@ -371,6 +371,13 @@ pub struct OtlpTargetConfig { /// Endpoint URI that receives both logs and metrics over OTLP/gRPC. pub endpoint: String, + /// Optional TLS or mTLS configuration for this endpoint. + /// + /// Omit this table for HTTPS endpoints that use platform trust roots. A + /// configured profile supplies a private CA, and supplying a client + /// certificate and key additionally enables mTLS. + pub tls: Option, + /// Maximum number of events or samples exported per request. Defaults to /// 512. #[serde(default = "OtlpTargetConfig::default_batch_size")] @@ -415,9 +422,108 @@ impl OtlpTargetConfig { return Err(format!("{path}.flush_interval must be greater than 0")); } - tonic::transport::Channel::from_shared(self.endpoint.clone()) + let endpoint = tonic::transport::Channel::from_shared(self.endpoint.clone()) .map_err(|_| format!("invalid {path}.endpoint: {}", self.endpoint))?; + if let Some(tls) = &self.tls { + if endpoint.uri().scheme_str() != Some("https") { + return Err(format!("{path}.tls requires an https endpoint")); + } + + tls.validate(&path)?; + } + + Ok(()) + } +} + +/// TLS policy for one OTLP/gRPC target. +/// +/// The CA bundle verifies the server certificate. Supplying both client paths +/// adds a client identity and enables mTLS. Each signal drain periodically +/// reloads the certificate files and adopts them only after a replacement +/// connection succeeds. A failed reload leaves the current connection active. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OtlpTlsConfig { + /// Path to the CA bundle used to verify the OTLP server certificate. + pub ca_cert_path: PathBuf, + + /// Optional path to the client certificate chain for mTLS. + pub client_cert_path: Option, + + /// Optional path to the client private key for mTLS. + pub client_key_path: Option, + + /// Optional DNS name used for TLS SNI and server certificate verification. + pub tls_server_name: Option, + + /// Interval between reloads of this target's TLS files. Defaults to five + /// minutes. + #[serde( + default = "OtlpTlsConfig::default_reload_interval", + with = "humantime_serde" + )] + pub reload_interval: Duration, +} + +impl OtlpTlsConfig { + /// Default interval between attempts to reload an OTLP target's TLS files. + pub(crate) const DEFAULT_RELOAD_INTERVAL: Duration = Duration::from_secs(5 * 60); + + fn default_reload_interval() -> Duration { + Self::DEFAULT_RELOAD_INTERVAL + } + + fn validate(&self, target_path: &str) -> Result<(), String> { + let path = format!("{target_path}.tls"); + + if self.ca_cert_path.as_os_str().is_empty() { + return Err(format!("{path}.ca_cert_path must not be empty")); + } + + match (&self.client_cert_path, &self.client_key_path) { + (Some(client_cert_path), Some(client_key_path)) => { + if client_cert_path.as_os_str().is_empty() { + return Err(format!("{path}.client_cert_path must not be empty")); + } + + if client_key_path.as_os_str().is_empty() { + return Err(format!("{path}.client_key_path must not be empty")); + } + } + (Some(_), None) => { + return Err(format!( + "{path}.client_key_path must be set when {path}.client_cert_path is set" + )); + } + (None, Some(_)) => { + return Err(format!( + "{path}.client_cert_path must be set when {path}.client_key_path is set" + )); + } + (None, None) => {} + } + + if let Some(tls_server_name) = self.tls_server_name.as_deref() { + if tls_server_name.trim().is_empty() { + return Err(format!("{path}.tls_server_name must not be empty")); + } + + if tls_server_name.trim() != tls_server_name { + return Err(format!( + "{path}.tls_server_name must not contain leading or trailing whitespace" + )); + } + + DnsName::try_from(tls_server_name) + .map_err(|_| format!("{path}.tls_server_name must be a valid DNS name"))?; + } + + if self.reload_interval.is_zero() { + return Err(format!("{path}.reload_interval must be greater than 0")); + } + Ok(()) } } @@ -1943,6 +2049,7 @@ username = "root" batch_size: 512, flush_interval: Duration::from_secs(2), include_diagnostics: false, + tls: None, }], }); @@ -1958,6 +2065,7 @@ username = "root" batch_size: 512, flush_interval: Duration::from_secs(2), include_diagnostics: false, + tls: None, }], }); @@ -2000,13 +2108,20 @@ include_diagnostics = true .merge(Toml::string( r#" [[targets]] -endpoint = "http://site.example:4317" +endpoint = "https://site.example:4317" [[targets]] endpoint = "https://central.example:4317" batch_size = 1024 flush_interval = "5s" include_diagnostics = true + +[targets.tls] +ca_cert_path = "/central/ca.crt" +client_cert_path = "/central/tls.crt" +client_key_path = "/central/tls.key" +tls_server_name = "central.example" +reload_interval = "30s" "#, )) .extract() @@ -2015,12 +2130,33 @@ include_diagnostics = true let targets = &otlp.targets; assert_eq!(targets.len(), 2); + assert!(targets[0].tls.is_none()); assert_eq!(targets[0].batch_size, 512); assert_eq!(targets[0].flush_interval, Duration::from_secs(2)); assert_eq!(targets[1].batch_size, 1024); assert_eq!(targets[1].flush_interval, Duration::from_secs(5)); assert!(targets[1].include_diagnostics); + let tls = targets[1] + .tls + .as_ref() + .expect("central target should use TLS"); + + assert_eq!(tls.ca_cert_path, PathBuf::from("/central/ca.crt")); + + assert_eq!( + tls.client_cert_path.as_deref(), + Some(Path::new("/central/tls.crt")) + ); + + assert_eq!( + tls.client_key_path.as_deref(), + Some(Path::new("/central/tls.key")) + ); + + assert_eq!(tls.tls_server_name.as_deref(), Some("central.example")); + assert_eq!(tls.reload_interval, Duration::from_secs(30)); + let mut config = Config::default(); config.sinks.otlp = Configurable::Enabled(otlp); @@ -2030,6 +2166,16 @@ include_diagnostics = true .expect("multi-target OTLP config should validate"); } + #[test] + fn otlp_tls_reload_interval_defaults_to_five_minutes() { + let tls: OtlpTlsConfig = Figment::new() + .merge(Toml::string("ca_cert_path = \"/site/ca.crt\"")) + .extract() + .expect("OTLP TLS config should parse without a reload interval"); + + assert_eq!(tls.reload_interval, OtlpTlsConfig::DEFAULT_RELOAD_INTERVAL); + } + #[test] fn otlp_target_list_rejects_invalid_target_contracts() { struct TestCase { @@ -2064,6 +2210,41 @@ endpoint = "http://site.example:4317" "#, expected: "sinks.otlp.targets[1].endpoint must be unique: http://site.example:4317", }, + TestCase { + name: "TLS with plaintext endpoint", + toml: r#" +[[targets]] +endpoint = "http://site.example:4317" + +[targets.tls] +ca_cert_path = "/site/ca.crt" +"#, + expected: "sinks.otlp.targets[0].tls requires an https endpoint", + }, + TestCase { + name: "incomplete mTLS identity", + toml: r#" +[[targets]] +endpoint = "https://site.example:4317" + +[targets.tls] +ca_cert_path = "/site/ca.crt" +client_cert_path = "/site/tls.crt" +"#, + expected: "sinks.otlp.targets[0].tls.client_key_path must be set when sinks.otlp.targets[0].tls.client_cert_path is set", + }, + TestCase { + name: "zero TLS reload interval", + toml: r#" +[[targets]] +endpoint = "https://site.example:4317" + +[targets.tls] +ca_cert_path = "/site/ca.crt" +reload_interval = "0s" +"#, + expected: "sinks.otlp.targets[0].tls.reload_interval must be greater than 0", + }, ]; for case in cases { @@ -2097,6 +2278,7 @@ endpoint = "http://site.example:4317" batch_size: 512, flush_interval: Duration::from_secs(2), include_diagnostics: false, + tls: None, }], }), ..SinksConfig::default() @@ -2133,6 +2315,7 @@ endpoint = "http://site.example:4317" batch_size: 512, flush_interval: Duration::from_secs(2), include_diagnostics: true, + tls: None, }], }), ..SinksConfig::default() @@ -2149,12 +2332,14 @@ endpoint = "http://site.example:4317" batch_size: 512, flush_interval: Duration::from_secs(2), include_diagnostics: false, + tls: None, }, OtlpTargetConfig { endpoint: "http://central.example:4317".to_string(), batch_size: 512, flush_interval: Duration::from_secs(2), include_diagnostics: true, + tls: None, }, ], }), @@ -2199,6 +2384,7 @@ endpoint = "http://site.example:4317" batch_size: 512, flush_interval: Duration::from_secs(2), include_diagnostics: false, + tls: None, }], }), ..SinksConfig::default() diff --git a/crates/health/src/lib.rs b/crates/health/src/lib.rs index c54c0d58ce..f834c253a8 100644 --- a/crates/health/src/lib.rs +++ b/crates/health/src/lib.rs @@ -101,7 +101,7 @@ pub enum HealthError { NmxcStatus(tonic::Status), /// Client TLS material could not be read, validated, or applied. - #[error("mTLS profile error: {0}")] + #[error("TLS profile error: {0}")] Tls(#[source] Box), } @@ -312,11 +312,28 @@ impl discovery::DiscoveryIteration for ServiceDiscoveryIteration { } } +/// Runs the hardware-health service after validating configured TLS profiles. +/// +/// Switch and OTLP TLS material is preflighted before listeners and background +/// tasks start, so invalid certificate configuration fails startup. +/// +/// # Errors +/// +/// Returns an error when startup validation or initialization fails, or when a +/// long-running service task exits with an error. pub async fn run_service(config: Config) -> Result<(), HealthError> { if let Some(tls_config) = &config.tls.switch { tls::preflight(tls_config).await?; } + if let Configurable::Enabled(otlp) = &config.sinks.otlp { + for target in &otlp.targets { + if let Some(tls_config) = &target.tls { + tls::otlp_preflight(tls_config).await?; + } + } + } + let tls_config = config.tls.switch.clone(); let metrics_endpoint = config.metrics_addr()?; diff --git a/crates/health/src/otlp/drain.rs b/crates/health/src/otlp/drain.rs index fb9e75d736..a191950e2d 100644 --- a/crates/health/src/otlp/drain.rs +++ b/crates/health/src/otlp/drain.rs @@ -23,9 +23,9 @@ use tonic::transport::Channel; use super::collector_logs::logs_service_client::LogsServiceClient; use super::convert::build_export_request; -use super::{OtlpExportFailed, OtlpSignal}; +use super::{OtlpExportFailed, OtlpSignal, connect_replacement_target, target_endpoint}; use crate::collectors::{BackoffConfig, ExponentialBackoff}; -use crate::config::OtlpTargetConfig; +use crate::config::{OtlpTargetConfig, OtlpTlsConfig}; use crate::sink::otlp::OtlpQueue; use crate::sink::{CollectorEvent, EventContext}; @@ -51,14 +51,30 @@ impl OtlpDrainTask { } pub async fn run(self) { - let mut client = match self.connect().await { - Some(c) => c, - None => return, - }; + let mut client = self.connect().await; let mut batch = Vec::with_capacity(self.target.batch_size); let mut interval = tokio::time::interval(self.target.flush_interval); + // Non-TLS targets use the default only to construct a dormant interval; + // the select guard below disables reloads for them. Start after one full + // period and delay missed ticks so stalled drains do not initiate a + // burst of replacement connections when they resume. + let tls_reload_period = self + .target + .tls + .as_ref() + .map_or(OtlpTlsConfig::DEFAULT_RELOAD_INTERVAL, |tls| { + tls.reload_interval + }); + + let mut tls_reload_interval = tokio::time::interval_at( + tokio::time::Instant::now() + tls_reload_period, + tls_reload_period, + ); + + tls_reload_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { tokio::select! { _ = self.queue.notified() => { @@ -75,30 +91,53 @@ impl OtlpDrainTask { self.flush(&mut client, &mut batch).await; } } + _ = tls_reload_interval.tick(), if self.target.tls.is_some() => { + match connect_replacement_target(&self.target).await { + Ok(channel) => { + client = LogsServiceClient::new(channel); + + tracing::debug!( + endpoint = %self.target.endpoint, + "refreshed otlp target TLS material" + ); + } + Err(error) => { + tracing::warn!( + ?error, + endpoint = %self.target.endpoint, + "failed to reload otlp target TLS material, keeping current client" + ); + } + } + } } } } - async fn connect(&self) -> Option> { - let endpoint = match Channel::from_shared(self.target.endpoint.clone()) { - Ok(endpoint) => endpoint, - Err(error) => { - tracing::error!( - ?error, - endpoint = %self.target.endpoint, - "invalid otlp target endpoint uri, stopping drain" - ); - - return None; - } - }; - + async fn connect(&self) -> LogsServiceClient { let mut backoff = ExponentialBackoff::new(&BackoffConfig { initial: Duration::from_secs(1), max: Duration::from_secs(30), }); loop { + let endpoint = match target_endpoint(&self.target).await { + Ok(endpoint) => endpoint, + Err(error) => { + let delay = backoff.next_delay(); + + tracing::warn!( + ?error, + endpoint = %self.target.endpoint, + retry_in = ?delay, + "failed to configure otlp target connection" + ); + + tokio::time::sleep(delay).await; + continue; + } + }; + match endpoint.connect().await { Ok(channel) => { tracing::info!( @@ -106,7 +145,7 @@ impl OtlpDrainTask { "connected to otlp target" ); - return Some(LogsServiceClient::new(channel)); + return LogsServiceClient::new(channel); } Err(error) => { let delay = backoff.next_delay(); diff --git a/crates/health/src/otlp/metrics_drain.rs b/crates/health/src/otlp/metrics_drain.rs index 9780a2a8a0..a627719a80 100644 --- a/crates/health/src/otlp/metrics_drain.rs +++ b/crates/health/src/otlp/metrics_drain.rs @@ -23,9 +23,9 @@ use tonic::transport::Channel; use super::collector_metrics::metrics_service_client::MetricsServiceClient; use super::convert::build_metrics_export_request; -use super::{OtlpExportFailed, OtlpSignal}; +use super::{OtlpExportFailed, OtlpSignal, connect_replacement_target, target_endpoint}; use crate::collectors::{BackoffConfig, ExponentialBackoff}; -use crate::config::OtlpTargetConfig; +use crate::config::{OtlpTargetConfig, OtlpTlsConfig}; use crate::sink::otlp::OtlpMetricsQueue; use crate::sink::{EventContext, MetricSample}; @@ -60,14 +60,30 @@ impl OtlpMetricsDrainTask { } pub async fn run(self) { - let mut client = match self.connect().await { - Some(c) => c, - None => return, - }; + let mut client = self.connect().await; let mut batch = Vec::with_capacity(self.target.batch_size); let mut interval = tokio::time::interval(self.target.flush_interval); + // Non-TLS targets use the default only to construct a dormant interval; + // the select guard below disables reloads for them. Start after one full + // period and delay missed ticks so stalled drains do not initiate a + // burst of replacement connections when they resume. + let tls_reload_period = self + .target + .tls + .as_ref() + .map_or(OtlpTlsConfig::DEFAULT_RELOAD_INTERVAL, |tls| { + tls.reload_interval + }); + + let mut tls_reload_interval = tokio::time::interval_at( + tokio::time::Instant::now() + tls_reload_period, + tls_reload_period, + ); + + tls_reload_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { tokio::select! { _ = self.queue.notified() => { @@ -84,30 +100,53 @@ impl OtlpMetricsDrainTask { self.flush(&mut client, &mut batch).await; } } + _ = tls_reload_interval.tick(), if self.target.tls.is_some() => { + match connect_replacement_target(&self.target).await { + Ok(channel) => { + client = MetricsServiceClient::new(channel); + + tracing::debug!( + endpoint = %self.target.endpoint, + "refreshed otlp metrics TLS material" + ); + } + Err(error) => { + tracing::warn!( + ?error, + endpoint = %self.target.endpoint, + "failed to reload otlp metrics TLS material, keeping current client" + ); + } + } + } } } } - async fn connect(&self) -> Option> { - let endpoint = match Channel::from_shared(self.target.endpoint.clone()) { - Ok(endpoint) => endpoint, - Err(error) => { - tracing::error!( - ?error, - endpoint = %self.target.endpoint, - "invalid otlp metrics endpoint uri, stopping drain" - ); - - return None; - } - }; - + async fn connect(&self) -> MetricsServiceClient { let mut backoff = ExponentialBackoff::new(&BackoffConfig { initial: Duration::from_secs(1), max: Duration::from_secs(30), }); loop { + let endpoint = match target_endpoint(&self.target).await { + Ok(endpoint) => endpoint, + Err(error) => { + let delay = backoff.next_delay(); + + tracing::warn!( + ?error, + endpoint = %self.target.endpoint, + retry_in = ?delay, + "failed to configure otlp metrics connection" + ); + + tokio::time::sleep(delay).await; + continue; + } + }; + match endpoint.connect().await { Ok(channel) => { tracing::info!( @@ -115,7 +154,7 @@ impl OtlpMetricsDrainTask { "connected to otlp metrics collector" ); - return Some(MetricsServiceClient::new(channel)); + return MetricsServiceClient::new(channel); } Err(error) => { let delay = backoff.next_delay(); diff --git a/crates/health/src/otlp/mod.rs b/crates/health/src/otlp/mod.rs index 831e8510c3..ff1077a6dd 100644 --- a/crates/health/src/otlp/mod.rs +++ b/crates/health/src/otlp/mod.rs @@ -19,7 +19,16 @@ pub mod convert; pub mod drain; pub mod metrics_drain; +use std::time::Duration; + use carbide_instrument::LabelValue; +use tonic::transport::{Channel, ClientTlsConfig, Endpoint}; + +use crate::HealthError; +use crate::config::OtlpTargetConfig; + +/// Maximum time allowed to establish a replacement OTLP channel. +const OTLP_RELOAD_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); /// Which OTLP signal a drain exports. #[derive(Debug, Clone, Copy, PartialEq, Eq, LabelValue)] @@ -28,6 +37,79 @@ pub(crate) enum OtlpSignal { Metrics, } +/// Builds an OTLP endpoint with the target's current TLS or mTLS policy. +/// +/// HTTPS targets without an explicit TLS profile use platform trust roots. An +/// explicit profile is reread on every call so reconnects can adopt rotated +/// certificate files. +/// +/// # Errors +/// +/// Returns an error when the endpoint URI is invalid or the configured TLS +/// material cannot be loaded, validated, or applied. +pub(crate) async fn target_endpoint(target: &OtlpTargetConfig) -> Result { + let endpoint = Channel::from_shared(target.endpoint.clone()).map_err(|error| { + HealthError::GenericError(format!( + "invalid OTLP target endpoint {}: {error}", + target.endpoint + )) + })?; + + let tls_config = match &target.tls { + Some(tls) => crate::tls::otlp_tonic_tls_config(tls).await?, + None if endpoint.uri().scheme_str() == Some("https") => { + ClientTlsConfig::new().with_enabled_roots() + } + None => return Ok(endpoint), + }; + + endpoint.tls_config(tls_config).map_err(|error| { + HealthError::GenericError(format!( + "invalid TLS configuration for OTLP target {}: {error}", + target.endpoint + )) + }) +} + +/// Establishes a replacement channel before a drain adopts refreshed TLS material. +/// +/// The function returns only after the TCP and TLS handshakes succeed. Reload +/// failures and attempts exceeding ten seconds return an error, allowing the +/// caller to retain its current channel. +/// +/// # Errors +/// +/// Returns an error when endpoint construction fails, the connection cannot be +/// established, or the connection attempt exceeds its deadline. +pub(crate) async fn connect_replacement_target( + target: &OtlpTargetConfig, +) -> Result { + connect_replacement_target_with_timeout(target, OTLP_RELOAD_CONNECT_TIMEOUT).await +} + +async fn connect_replacement_target_with_timeout( + target: &OtlpTargetConfig, + connect_timeout: Duration, +) -> Result { + let endpoint = target_endpoint(target).await?; + + let channel = tokio::time::timeout(connect_timeout, endpoint.connect()) + .await + .map_err(|_| { + HealthError::GenericError(format!( + "timed out connecting replacement channel for OTLP target {} after {:?}", + target.endpoint, connect_timeout + )) + })?; + + channel.map_err(|error| { + HealthError::GenericError(format!( + "failed to connect replacement channel for OTLP target {}: {error}", + target.endpoint + )) + }) +} + /// A gRPC status code as a bounded metric label: one variant per /// [`tonic::Code`], a set closed by the gRPC protocol itself. #[derive(Debug, Clone, Copy, PartialEq, Eq, LabelValue)] @@ -154,10 +236,105 @@ pub use opentelemetry::proto::resource::v1 as resource; #[cfg(test)] mod tests { + use std::time::Duration; + use carbide_instrument::emit; use carbide_instrument::testing::{MetricsCapture, capture_logs}; + use tokio::io::AsyncReadExt; + use tokio::net::TcpListener; + use tokio::sync::oneshot; + use tokio::time::timeout; + + use super::{ + OtlpExportFailed, OtlpSignal, connect_replacement_target_with_timeout, target_endpoint, + }; + use crate::HealthError; + use crate::config::OtlpTargetConfig; + + #[tokio::test] + async fn https_target_without_tls_profile_starts_tls_handshake() + -> Result<(), Box> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + + let target = OtlpTargetConfig { + endpoint: format!("https://{address}"), + tls: None, + batch_size: 1, + flush_interval: Duration::from_secs(1), + include_diagnostics: false, + }; + + let endpoint = target_endpoint(&target).await?; + + let peer = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await?; + let mut content_type = [0_u8; 1]; + + stream.read_exact(&mut content_type).await?; + + Ok::<_, std::io::Error>(content_type[0]) + }); - use super::{OtlpExportFailed, OtlpSignal}; + let _connect_result = timeout(Duration::from_secs(1), endpoint.connect()).await?; + + let content_type = timeout(Duration::from_secs(1), peer).await??; + + assert_eq!( + content_type?, 0x16, + "connection must start with a TLS record" + ); + + Ok(()) + } + + #[tokio::test] + async fn replacement_connection_times_out_without_tls_handshake() + -> Result<(), Box> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + let (accepted_tx, accepted_rx) = oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel::<()>(); + + let target = OtlpTargetConfig { + endpoint: format!("https://{address}"), + tls: None, + batch_size: 1, + flush_interval: Duration::from_secs(1), + include_diagnostics: false, + }; + + let peer = tokio::spawn(async move { + let (stream, _) = listener.accept().await?; + + accepted_tx + .send(()) + .map_err(|_| std::io::Error::other("replacement connection task stopped"))?; + + let _ = release_rx.await; + + drop(stream); + + Ok::<_, std::io::Error>(()) + }); + + let replacement = tokio::spawn(async move { + connect_replacement_target_with_timeout(&target, Duration::from_millis(100)).await + }); + + accepted_rx.await?; + + let result = replacement.await?; + + drop(release_tx); + peer.await??; + + assert!( + matches!(result, Err(HealthError::GenericError(message)) if message.contains("timed out connecting replacement channel")) + ); + + Ok(()) + } /// A dropped logs batch writes one ERROR line and ticks the counter's /// logs-signal series, labelled with the gRPC status code. diff --git a/crates/health/src/sink/otlp.rs b/crates/health/src/sink/otlp.rs index b9899fa86e..9fd37d0bbc 100644 --- a/crates/health/src/sink/otlp.rs +++ b/crates/health/src/sink/otlp.rs @@ -520,12 +520,14 @@ mod tests { batch_size: 512, flush_interval: std::time::Duration::from_secs(2), include_diagnostics: false, + tls: None, }, OtlpTargetConfig { endpoint: "http://second.example:4317".to_string(), batch_size: 512, flush_interval: std::time::Duration::from_secs(2), include_diagnostics: false, + tls: None, }, ]; diff --git a/crates/health/src/tls.rs b/crates/health/src/tls.rs index a1add1aa7d..0da6d28817 100644 --- a/crates/health/src/tls.rs +++ b/crates/health/src/tls.rs @@ -15,15 +15,17 @@ * limitations under the License. */ -//! TLS helpers for outbound health collector connections. +//! TLS helpers for outbound health 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 +//! Switch collectors use `[tls.switch]`; OTLP targets use their own `tls` +//! tables. This module owns HTTP and gRPC TLS construction inside the health +//! crate so outbound transport security does not depend on NICo API certificate //! 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. +//! collectors build a new TLS config when they reconnect. OTLP drains reload +//! their target profile periodically and adopt it only after a replacement +//! connection succeeds. use std::fmt; use std::io::BufReader; @@ -49,11 +51,11 @@ use tonic::transport::{ use url::Url; use x509_parser::prelude::*; -use crate::config::MtlsProfileConfig; +use crate::config::{MtlsProfileConfig, OtlpTlsConfig}; type HyperMtlsClient = HyperClient, Empty>; -/// Role for one mTLS profile material file. +/// Role for one TLS profile material file. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum TlsMaterialKind { CaBundle, @@ -73,7 +75,7 @@ impl fmt::Display for TlsMaterialKind { #[derive(Debug, Error)] pub(crate) enum TlsError { - #[error("failed to read mTLS profile {kind} at {path}: {source}")] + #[error("failed to read TLS profile {kind} at {path}: {source}")] Read { kind: TlsMaterialKind, path: PathBuf, @@ -81,13 +83,13 @@ pub(crate) enum TlsError { source: std::io::Error, }, - #[error("mTLS profile {kind} at {path} is empty")] + #[error("TLS profile {kind} at {path} is empty")] Empty { kind: TlsMaterialKind, path: PathBuf, }, - #[error("failed to parse mTLS profile {kind} at {path}: {message}")] + #[error("failed to parse TLS profile {kind} at {path}: {message}")] Parse { kind: TlsMaterialKind, path: PathBuf, @@ -95,7 +97,7 @@ pub(crate) enum TlsError { }, #[error( - "mTLS profile {kind} certificate at {path} is not valid before unix timestamp {not_before}" + "TLS profile {kind} certificate at {path} is not valid before unix timestamp {not_before}" )] NotYetValid { kind: TlsMaterialKind, @@ -103,31 +105,60 @@ pub(crate) enum TlsError { not_before: i64, }, - #[error("mTLS profile {kind} certificate at {path} expired at unix timestamp {not_after}")] + #[error("TLS 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")] + #[error("TLS 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}")] + #[error("TLS profile client certificate and key are invalid or do not match: {source}")] InvalidIdentity { #[source] source: rustls::Error, }, + + /// Expected client identity material was absent while constructing mTLS. + #[error("TLS profile requires a client certificate and key")] + MissingClientIdentity, + + /// An OTLP TLS profile configured only one client identity path. + #[error("TLS profile client certificate and key must be configured together")] + IncompleteClientIdentity, } struct TlsMaterial { ca_pem: Vec, + client_identity: Option, +} + +struct ClientIdentityMaterial { client_cert_pem: Vec, client_key_pem: Vec, } pub(crate) async fn preflight(config: &MtlsProfileConfig) -> Result<(), TlsError> { - read_validated_material(config).await.map(|_| ()) + read_validated_material( + &config.ca_cert_path, + Some((&config.client_cert_path, &config.client_key_path)), + ) + .await + .map(|_| ()) +} + +/// Reads and validates the CA and optional client identity for one OTLP target. +/// +/// # Errors +/// +/// Returns an error when identity paths are incomplete or configured material +/// cannot be read or validated. +pub(crate) async fn otlp_preflight(config: &OtlpTlsConfig) -> Result<(), TlsError> { + read_validated_material(&config.ca_cert_path, otlp_client_identity_paths(config)?) + .await + .map(|_| ()) } /// Cloneable HTTP client built from one validated mTLS profile. @@ -312,7 +343,12 @@ impl MtlsHttpClientProvider { /// `[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 material = read_validated_material( + &config.ca_cert_path, + Some((&config.client_cert_path, &config.client_key_path)), + ) + .await?; + let tls_config = http_tls_config(config, &material)?; let mut http_connector = HttpConnector::new(); @@ -376,13 +412,18 @@ fn http_tls_config( }); } + let client_identity = material + .client_identity + .as_ref() + .ok_or(TlsError::MissingClientIdentity)?; + let client_certs = parse_certificates( TlsMaterialKind::ClientCertificate, &config.client_cert_path, - &material.client_cert_pem, + &client_identity.client_cert_pem, )?; - let client_key = parse_private_key(&config.client_key_path, &material.client_key_pem)?; + let client_key = parse_private_key(&config.client_key_path, &client_identity.client_key_pem)?; rustls::ClientConfig::builder_with_provider(Arc::new( rustls::crypto::aws_lc_rs::default_provider(), @@ -401,65 +442,130 @@ fn http_tls_config( pub(crate) async fn tonic_tls_config( config: &MtlsProfileConfig, ) -> Result { - let material = read_validated_material(config).await?; + let material = read_validated_material( + &config.ca_cert_path, + Some((&config.client_cert_path, &config.client_key_path)), + ) + .await?; + + Ok(tonic_tls_config_from_material( + &material, + &config.tls_server_name, + )) +} + +/// Builds a Tonic TLS configuration for one OTLP target. +/// +/// # Errors +/// +/// Returns an error when identity paths are incomplete or configured material +/// cannot be read or validated. +pub(crate) async fn otlp_tonic_tls_config( + config: &OtlpTlsConfig, +) -> Result { + let material = + read_validated_material(&config.ca_cert_path, otlp_client_identity_paths(config)?).await?; + + Ok(tonic_tls_config_from_material( + &material, + &config.tls_server_name, + )) +} - 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, +fn otlp_client_identity_paths(config: &OtlpTlsConfig) -> Result, TlsError> { + match (&config.client_cert_path, &config.client_key_path) { + (Some(client_cert_path), Some(client_key_path)) => { + Ok(Some((client_cert_path, client_key_path))) + } + (None, None) => Ok(None), + _ => Err(TlsError::IncompleteClientIdentity), + } +} + +fn tonic_tls_config_from_material( + material: &TlsMaterial, + tls_server_name: &Option, +) -> ClientTlsConfig { + let mut tls_config = + ClientTlsConfig::new().ca_certificate(TonicCertificate::from_pem(&material.ca_pem)); + + if let Some(client_identity) = &material.client_identity { + tls_config = tls_config.identity(TonicIdentity::from_pem( + &client_identity.client_cert_pem, + &client_identity.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. + if let Some(tls_server_name) = tls_server_name { + // This override changes only TLS SNI and certificate verification. The + // connection still uses the configured endpoint URI. tls_config = tls_config.domain_name(tls_server_name.clone()); } - Ok(tls_config) + tls_config } -async fn read_validated_material(config: &MtlsProfileConfig) -> Result { +async fn read_validated_material( + ca_cert_path: &Path, + client_identity_paths: Option<(&Path, &Path)>, +) -> 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_pem = read_material(TlsMaterialKind::CaBundle, ca_cert_path).await?; + + let client_identity = match client_identity_paths { + Some((client_cert_path, client_key_path)) => { + let (client_cert_pem, client_key_pem) = tokio::try_join!( + read_material(TlsMaterialKind::ClientCertificate, client_cert_path), + read_material(TlsMaterialKind::ClientKey, client_key_path), + )?; + + Some(( + client_cert_path, + client_key_path, + client_cert_pem, + client_key_pem, + )) + } + None => None, + }; - let ca_certs = parse_certificates(TlsMaterialKind::CaBundle, &config.ca_cert_path, &ca_pem)?; + let ca_certs = parse_certificates(TlsMaterialKind::CaBundle, ca_cert_path, &ca_pem)?; + let now = unix_timestamp_now(); - let client_certs = parse_certificates( - TlsMaterialKind::ClientCertificate, - &config.client_cert_path, - &client_cert_pem, - )?; + validate_certificate_times(TlsMaterialKind::CaBundle, ca_cert_path, &ca_certs, now)?; + validate_root_store(ca_cert_path, ca_certs)?; - let client_key = parse_private_key(&config.client_key_path, &client_key_pem)?; - let now = unix_timestamp_now(); + let client_identity = match client_identity { + Some((client_cert_path, client_key_path, client_cert_pem, client_key_pem)) => { + let client_certs = parse_certificates( + TlsMaterialKind::ClientCertificate, + client_cert_path, + &client_cert_pem, + )?; - validate_certificate_times( - TlsMaterialKind::CaBundle, - &config.ca_cert_path, - &ca_certs, - now, - )?; + let client_key = parse_private_key(client_key_path, &client_key_pem)?; - validate_certificate_times( - TlsMaterialKind::ClientCertificate, - &config.client_cert_path, - &client_certs, - now, - )?; + validate_certificate_times( + TlsMaterialKind::ClientCertificate, + client_cert_path, + &client_certs, + now, + )?; - validate_root_store(&config.ca_cert_path, ca_certs)?; - validate_client_identity(client_certs, client_key)?; + validate_client_identity(client_certs, client_key)?; + + Some(ClientIdentityMaterial { + client_cert_pem, + client_key_pem, + }) + } + None => None, + }; Ok(TlsMaterial { ca_pem, - client_cert_pem, - client_key_pem, + client_identity, }) } @@ -776,6 +882,70 @@ mod tests { Ok(()) } + #[tokio::test] + async fn otlp_tls_material_builds_tls_and_mtls_grpc_clients() + -> Result<(), Box> { + let dir = TempDir::new()?; + let material = valid_material(); + let switch_config = write_material(&dir, &material).await?; + + let tls_config = OtlpTlsConfig { + ca_cert_path: switch_config.ca_cert_path.clone(), + client_cert_path: None, + client_key_path: None, + tls_server_name: Some("telemetry.example.com".to_string()), + reload_interval: OtlpTlsConfig::DEFAULT_RELOAD_INTERVAL, + }; + + otlp_preflight(&tls_config).await?; + let _tls = otlp_tonic_tls_config(&tls_config).await?; + + let mtls_config = OtlpTlsConfig { + ca_cert_path: switch_config.ca_cert_path, + client_cert_path: Some(switch_config.client_cert_path), + client_key_path: Some(switch_config.client_key_path), + tls_server_name: None, + reload_interval: OtlpTlsConfig::DEFAULT_RELOAD_INTERVAL, + }; + + otlp_preflight(&mtls_config).await?; + let _mtls = otlp_tonic_tls_config(&mtls_config).await?; + + Ok(()) + } + + #[tokio::test] + async fn otlp_tls_config_detects_invalid_rotated_material() + -> Result<(), Box> { + let dir = TempDir::new()?; + let material = valid_material(); + let switch_config = write_material(&dir, &material).await?; + + let config = OtlpTlsConfig { + ca_cert_path: switch_config.ca_cert_path, + client_cert_path: Some(switch_config.client_cert_path), + client_key_path: Some(switch_config.client_key_path), + tls_server_name: None, + reload_interval: OtlpTlsConfig::DEFAULT_RELOAD_INTERVAL, + }; + + otlp_preflight(&config).await?; + let _tls = otlp_tonic_tls_config(&config).await?; + + let client_key_path = config + .client_key_path + .as_ref() + .expect("test config must contain a client key path"); + + tokio::fs::write(client_key_path, &material.alternate_client_key_pem).await?; + + let result = otlp_tonic_tls_config(&config).await; + + assert!(matches!(result, Err(TlsError::InvalidIdentity { .. }))); + + Ok(()) + } + #[tokio::test] async fn http_client_build_reads_changed_tls_material() -> Result<(), Box> {