From 7d93a7596c4c00e1ab41fb9ba45e5cc17941d4fe Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 9 Jul 2026 00:20:38 +0800 Subject: [PATCH] fix(targets): harden messaging backend delivery and error classification (#4506) Address a batch of correctness/security audit findings in the messaging notification backends (mqtt/nats/kafka/amqp/redis/pulsar/webhook). MQTT (backlog#971): wait for broker acknowledgement via publish_tracked + wait_completion_async (PUBACK/PUBCOMP for QoS>=1, flush for QoS0) with a 30s timeout before reporting success, instead of treating the enqueue as delivered. Replace substring error matching with typed ClientError / PublishNoticeError classification. NATS (backlog#971, #973, #983): flush() after publish to confirm the broker received the message before the durable copy is deleted; classify publish and flush failures by typed error kind; warn when credentials are sent without TLS. Kafka (backlog#973, #983): use Error::is_retriable() so transient broker states (NotLeaderForPartition, LeaderNotAvailable, RequestTimedOut, ...) are retried instead of dropped as permanent; add a bucket/object message key for per-object partition ordering; build the producer without holding the cache lock across the connect await. AMQP (backlog#973, #980): classify permanent broker protocol errors (404/403/406, NOT_ALLOWED, ...) as request-level errors rather than connectivity errors to avoid reconnect storms; bound publish and publisher-confirm waits with a timeout; check is_enabled in is_active; run full close() cleanup even when the broker close fails; warn when mandatory=false may silently drop unroutable messages. Redis (backlog#982): warn when PUBLISH reaches 0 subscribers; reuse the cached ConnectionManager for health probes instead of a fresh handshake; warn when tls_allow_insecure disables certificate verification. Pulsar (backlog#983): replace std::sync::Mutex + unwrap with parking_lot Mutex so a panic while holding the guard cannot poison later accesses. Webhook (backlog#983): drain the response body so the connection can be reused (keep-alive). The #974 redirect-follow SSRF fix already landed. Relates to rustfs/backlog#971 Relates to rustfs/backlog#973 Relates to rustfs/backlog#974 Relates to rustfs/backlog#980 Relates to rustfs/backlog#982 Relates to rustfs/backlog#983 Co-authored-by: heihutu --- crates/targets/src/target/amqp.rs | 122 +++++++++++++++--- crates/targets/src/target/kafka.rs | 81 ++++++++++-- crates/targets/src/target/mqtt.rs | 177 +++++++++++++++++++++------ crates/targets/src/target/nats.rs | 126 ++++++++++++++++++- crates/targets/src/target/pulsar.rs | 26 ++-- crates/targets/src/target/redis.rs | 38 +++++- crates/targets/src/target/webhook.rs | 4 + 7 files changed, 492 insertions(+), 82 deletions(-) diff --git a/crates/targets/src/target/amqp.rs b/crates/targets/src/target/amqp.rs index faefa3fde..2a3d096b2 100644 --- a/crates/targets/src/target/amqp.rs +++ b/crates/targets/src/target/amqp.rs @@ -51,6 +51,10 @@ use tokio::sync::Mutex as AsyncMutex; use tracing::{info, instrument, warn}; use url::Url; +/// Upper bound on how long a single publish and its publisher-confirm may block +/// before being treated as a timeout (backlog#980). +const AMQP_PUBLISH_TIMEOUT: Duration = Duration::from_secs(30); + #[derive(Clone)] pub struct AMQPArgs { pub enable: bool, @@ -238,8 +242,40 @@ fn build_publish_properties(args: &AMQPArgs) -> BasicProperties { properties } +/// Returns true for AMQP broker protocol errors that indicate a permanent, +/// non-connectivity condition (e.g. the exchange/queue does not exist, access is +/// refused, or a precondition failed). These must not be treated as transient +/// connectivity errors, otherwise a misconfigured target triggers an endless +/// reconnect storm instead of surfacing a delivery failure (backlog#973). +fn is_permanent_amqp_protocol_error(err: &lapin::Error) -> bool { + use lapin::protocol::{AMQPErrorKind, AMQPHardError, AMQPSoftError}; + if let LapinErrorKind::ProtocolError(amqp_err) = err.kind() { + return match amqp_err.kind() { + // 404 NOT_FOUND (missing exchange/queue), 403 ACCESS_REFUSED, + // 406 PRECONDITION_FAILED. + AMQPErrorKind::Soft(soft) => { + matches!( + soft, + AMQPSoftError::NOTFOUND | AMQPSoftError::ACCESSREFUSED | AMQPSoftError::PRECONDITIONFAILED + ) + } + // 530 NOT_ALLOWED, 540 NOT_IMPLEMENTED, 402 INVALID_PATH. + AMQPErrorKind::Hard(hard) => { + matches!( + hard, + AMQPHardError::NOTALLOWED | AMQPHardError::NOTIMPLEMENTED | AMQPHardError::INVALIDPATH + ) + } + }; + } + false +} + fn map_lapin_error(err: lapin::Error, context: &str) -> TargetError { let message = format!("{context}: {err}"); + if is_permanent_amqp_protocol_error(&err) { + return TargetError::Request(message); + } match err.kind() { LapinErrorKind::IOError(io_err) if io_err.kind() == std::io::ErrorKind::TimedOut => TargetError::Timeout(message), LapinErrorKind::IOError(_) @@ -332,6 +368,18 @@ where #[instrument(skip(args), fields(target_id_as_string = %id))] pub fn new(id: String, args: AMQPArgs) -> Result { args.validate()?; + if args.enable && !args.mandatory { + // With mandatory=false the broker silently discards messages that + // route to no queue, and the publish is still confirmed as success. + // Warn so operators expecting reliable delivery know unroutable + // events are dropped without a trace (backlog#980). + warn!( + target_id = %id, + exchange = %args.exchange, + routing_key = %args.routing_key, + "AMQP target has mandatory=false: messages that route to no queue are silently dropped. Set mandatory=true for reliable delivery." + ); + } let target_id = TargetID::new(id, ChannelTargetType::Amqp.as_str().to_string()); let queue_store = open_target_queue_store( &args.queue_dir, @@ -417,9 +465,11 @@ where async fn send_body(&self, body: &[u8]) -> Result<(), TargetError> { let connection = self.get_or_connect().await?; - let publish = connection - .channel - .basic_publish( + // Bound the publish and publisher-confirm waits so a stuck broker cannot + // block the send path indefinitely (backlog#980). + let publish = match tokio::time::timeout( + AMQP_PUBLISH_TIMEOUT, + connection.channel.basic_publish( self.args.exchange.clone().into(), self.args.routing_key.clone().into(), BasicPublishOptions { @@ -428,17 +478,33 @@ where }, body, build_publish_properties(&self.args), - ) - .await; + ), + ) + .await + { + Ok(publish) => publish, + Err(_) => { + self.clear_connection(); + return Err(TargetError::Timeout("AMQP publish timed out".to_string())); + } + }; - let confirm = match publish { - Ok(confirm) => confirm.await, + let confirm_future = match publish { + Ok(confirm) => confirm, Err(err) => { self.clear_connection(); return Err(map_lapin_error(err, "Failed to publish AMQP message")); } }; + let confirm = match tokio::time::timeout(AMQP_PUBLISH_TIMEOUT, confirm_future).await { + Ok(confirm) => confirm, + Err(_) => { + self.clear_connection(); + return Err(TargetError::Timeout("AMQP publisher confirm timed out".to_string())); + } + }; + match confirm { Ok(Confirmation::Ack(None) | Confirmation::NotRequested) => { self.delivery_counters.record_success(); @@ -506,6 +572,11 @@ where } async fn is_active(&self) -> Result { + // A disabled target is never active; avoid opening a connection for it + // (backlog#980). + if !self.args.enable { + return Ok(false); + } let connection = self.get_or_connect().await?; Ok(connection.connection.status().connected() && connection.channel.status().connected()) } @@ -540,12 +611,14 @@ where async fn close(&self) -> Result<(), TargetError> { let connection = self.connection.lock().take(); - if let Some(connection) = connection { - connection - .connection - .close(200, "OK".into()) - .await - .map_err(|e| map_lapin_error(e, "Failed to close AMQP connection"))?; + // Capture any close failure but still run the remaining cleanup so a + // failed broker close does not leave stale TLS/adapter state behind + // (backlog#980). + let mut close_result = Ok(()); + if let Some(connection) = connection + && let Err(e) = connection.connection.close(200, "OK".into()).await + { + close_result = Err(map_lapin_error(e, "Failed to close AMQP connection")); } self.tls_state.lock().reset(); // If a TLS reload adapter is attached, reset its error tracking @@ -554,7 +627,7 @@ where *adapter.runtime_state().last_error.write() = None; } info!(target_id = %self.id, "AMQP target closed"); - Ok(()) + close_result } fn store(&self) -> Option<&(dyn Store + Send + Sync)> { @@ -752,6 +825,27 @@ mod tests { assert_connect_failure(&err); } + #[test] + fn permanent_amqp_protocol_errors_are_not_connectivity_errors() { + use lapin::protocol::{AMQPError, AMQPErrorKind, AMQPHardError, AMQPSoftError}; + + let make = |kind: AMQPErrorKind| lapin::Error::from(LapinErrorKind::ProtocolError(AMQPError::new(kind, "boom".into()))); + + // 404 (missing exchange/queue) is permanent: it must be a request-level + // error so a misconfigured target does not reconnect-storm (backlog#973). + let not_found = make(AMQPErrorKind::Soft(AMQPSoftError::NOTFOUND)); + assert!(is_permanent_amqp_protocol_error(¬_found)); + assert!(matches!(map_lapin_error(not_found, "publish"), TargetError::Request(_))); + + assert!(is_permanent_amqp_protocol_error(&make(AMQPErrorKind::Soft(AMQPSoftError::ACCESSREFUSED)))); + assert!(is_permanent_amqp_protocol_error(&make(AMQPErrorKind::Hard(AMQPHardError::NOTALLOWED)))); + + // A transient soft error (broker resource locked) is not treated as permanent. + assert!(!is_permanent_amqp_protocol_error(&make(AMQPErrorKind::Soft( + AMQPSoftError::RESOURCELOCKED + )))); + } + #[test] fn debug_masks_secret_values() { let args = AMQPArgs { diff --git a/crates/targets/src/target/kafka.rs b/crates/targets/src/target/kafka.rs index da942e24d..928bce7cd 100644 --- a/crates/targets/src/target/kafka.rs +++ b/crates/targets/src/target/kafka.rs @@ -29,7 +29,7 @@ use crate::{ }, }; use async_trait::async_trait; -use rustfs_kafka_async::error::{ConnectionError, Error as KafkaError}; +use rustfs_kafka_async::error::{ConnectionError, Error as KafkaError, KafkaCode}; use rustfs_kafka_async::{AsyncProducer, AsyncProducerConfig, Record, RequiredAcks, SaslConfig, SecurityConfig}; use rustfs_tls_runtime::{load_cert_bundle_der_bytes, load_private_key}; use std::{fmt, marker::PhantomData, sync::Arc, time::Duration}; @@ -247,10 +247,23 @@ where E: PluginEvent, { fn map_kafka_error(err: KafkaError, context: &str) -> TargetError { - match err { - KafkaError::Connection(ConnectionError::NoHostReachable) => TargetError::NotConnected, - KafkaError::Connection(ConnectionError::Timeout(_)) => TargetError::Timeout(format!("{context}: {err}")), - KafkaError::Connection(_) => TargetError::Network(format!("{context}: {err}")), + // Prefer the client's own retriable classification so transient broker + // states (leader election / NotLeaderForPartition, coordinator load, + // network blips, RequestTimedOut) are retried via store replay instead of + // being dropped as permanent failures (backlog#973). + if err.is_retriable() { + return match &err { + KafkaError::Connection(ConnectionError::Timeout(_)) | KafkaError::Kafka(KafkaCode::RequestTimedOut) => { + TargetError::Timeout(format!("{context}: {err}")) + } + _ => TargetError::NotConnected, + }; + } + + // Non-retriable errors: configuration problems are permanent config + // errors; everything else (e.g. UnknownTopicOrPartition, authorization + // failures, oversize messages) is a permanent request-level failure. + match &err { KafkaError::Config(_) => TargetError::Configuration(format!("{context}: {err}")), _ => TargetError::Request(format!("{context}: {err}")), } @@ -332,12 +345,22 @@ where self.tls_state.lock().await.refresh(next_fingerprint); } - let mut cached = self.producer.lock().await; - if let Some(producer) = cached.as_ref() { - return Ok(Arc::clone(producer)); + { + let cached = self.producer.lock().await; + if let Some(producer) = cached.as_ref() { + return Ok(Arc::clone(producer)); + } } + // Build the producer without holding the cache lock so a slow connect + // does not block other senders (which only need to read the cache). + // Re-check the cache after building in case another task raced us + // (backlog#983). let producer = Arc::new(self.build_producer().await?); + let mut cached = self.producer.lock().await; + if let Some(existing) = cached.as_ref() { + return Ok(Arc::clone(existing)); + } *cached = Some(Arc::clone(&producer)); Ok(producer) } @@ -367,7 +390,14 @@ where let producer = self.get_or_build_producer().await?; - if let Err(err) = producer.send(&Record::from_value(&self.args.topic, body.as_slice())).await { + // Use "/" as the message key so all events for the same + // object hash to the same partition and preserve per-object ordering + // across multiple partitions (backlog#983). + let partition_key = format!("{}/{}", meta.bucket_name, meta.object_name); + if let Err(err) = producer + .send(&Record::from_key_value(&self.args.topic, partition_key, body.as_slice())) + .await + { let mapped = Self::map_kafka_error(err, "Failed to send message to Kafka"); invalidate_cache_on_connectivity_error(&mapped, || self.invalidate_cached_producer()).await; return Err(mapped); @@ -663,6 +693,39 @@ mod tests { assert_eq!(sasl.password(), "secret"); } + #[test] + fn map_kafka_error_treats_transient_broker_states_as_retriable() { + // Leader election / metadata staleness must be retried, not dropped + // as a permanent failure (backlog#973). + assert!(matches!( + KafkaTarget::::map_kafka_error(KafkaError::Kafka(KafkaCode::NotLeaderForPartition), "send"), + TargetError::NotConnected + )); + assert!(matches!( + KafkaTarget::::map_kafka_error(KafkaError::Kafka(KafkaCode::LeaderNotAvailable), "send"), + TargetError::NotConnected + )); + // RequestTimedOut is retriable and surfaced as a timeout. + assert!(matches!( + KafkaTarget::::map_kafka_error(KafkaError::Kafka(KafkaCode::RequestTimedOut), "send"), + TargetError::Timeout(_) + )); + } + + #[test] + fn map_kafka_error_treats_permanent_broker_states_as_request_error() { + // A missing topic/partition is a permanent condition; retrying would + // storm the broker. + assert!(matches!( + KafkaTarget::::map_kafka_error(KafkaError::Kafka(KafkaCode::UnknownTopicOrPartition), "send"), + TargetError::Request(_) + )); + assert!(matches!( + KafkaTarget::::map_kafka_error(KafkaError::Config("bad".to_string()), "send"), + TargetError::Configuration(_) + )); + } + #[test] fn test_debug_redacts_sasl_password_and_tls_key() { let rendered = format!( diff --git a/crates/targets/src/target/mqtt.rs b/crates/targets/src/target/mqtt.rs index 8a8560261..1aeb102fa 100644 --- a/crates/targets/src/target/mqtt.rs +++ b/crates/targets/src/target/mqtt.rs @@ -32,8 +32,8 @@ use arc_swap::ArcSwap; use async_trait::async_trait; use hyper_rustls::ConfigBuilderExt; use rumqttc::{ - AsyncClient, Broker, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, QoS, Transport, - mqttbytes::Error as MqttBytesError, + AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, PublishNoticeError, QoS, + Transport, mqttbytes::Error as MqttBytesError, }; use rustfs_config::{ EnableState, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_WS_PATH_ALLOWLIST, @@ -58,6 +58,10 @@ const DEFAULT_MQTT_TCP_PORT: u16 = 1883; const DEFAULT_MQTT_TLS_PORT: u16 = 8883; const DEFAULT_MQTT_WSS_PORT: u16 = 443; const MAX_MQTT_PACKET_SIZE_BYTES: u32 = 100 * 1024 * 1024; +/// Upper bound on how long a single publish may wait for broker acknowledgement +/// (PUBACK/PUBCOMP for QoS>=1, or network flush for QoS0) before it is treated as +/// a timeout so the durable copy is retained and replayed (backlog#971). +const MQTT_PUBLISH_CONFIRM_TIMEOUT: Duration = Duration::from_secs(30); /// Minimum delay before the supervisor rebuilds the client and event loop /// after a session exits. Also the delay used right after a session that had /// successfully connected, so a transient drop reconnects promptly. @@ -779,40 +783,78 @@ where "mqtt delivery state" ); - client - .publish(&self.args.topic, self.args.qos, false, body) - .await - .map_err(|e| { - if e.to_string().contains("Connection") || e.to_string().contains("Timeout") { - warn!( - event = EVENT_MQTT_DELIVERY_STATE, - component = LOG_COMPONENT_TARGETS, - subsystem = LOG_SUBSYSTEM_MQTT, - target_id = %self.id, - state = "publish_failed", - reason = "connectivity_error", - error = %e, - "mqtt delivery state" - ); - let err = TargetError::NotConnected; - mark_target_disconnected_on_connectivity_error(&self.connected, &err); - err - } else { - TargetError::Request(format!("Failed to publish message: {e}")) - } - })?; + // Enqueue a tracked publish so we can wait for broker acknowledgement + // (PUBACK for QoS1, PUBCOMP for QoS2, or network flush for QoS0) before + // reporting success. Previously the publish was treated as delivered as + // soon as it was queued on the event loop, so a disconnect after queueing + // silently dropped the event while its durable copy was already deleted + // (backlog#971). Error classification now matches on the typed error + // instead of substring matching on the display string. + let notice = match client.publish_tracked(&self.args.topic, self.args.qos, false, body).await { + Ok(notice) => notice, + Err(e) => { + let err = classify_mqtt_client_error(&e); + warn!( + event = EVENT_MQTT_DELIVERY_STATE, + component = LOG_COMPONENT_TARGETS, + subsystem = LOG_SUBSYSTEM_MQTT, + target_id = %self.id, + state = "publish_failed", + reason = "enqueue_error", + error = %e, + "mqtt delivery state" + ); + mark_target_disconnected_on_connectivity_error(&self.connected, &err); + return Err(err); + } + }; - debug!( - event = EVENT_MQTT_DELIVERY_STATE, - component = LOG_COMPONENT_TARGETS, - subsystem = LOG_SUBSYSTEM_MQTT, - target_id = %self.id, - topic = %self.args.topic, - state = "published", - "mqtt delivery state" - ); - self.delivery_counters.record_success(); - Ok(()) + // Release the client lock before awaiting the broker acknowledgement so a + // slow/hung broker never blocks other senders from queueing publishes. + drop(client_guard); + + match tokio::time::timeout(MQTT_PUBLISH_CONFIRM_TIMEOUT, notice.wait_completion_async()).await { + Ok(Ok(())) => { + debug!( + event = EVENT_MQTT_DELIVERY_STATE, + component = LOG_COMPONENT_TARGETS, + subsystem = LOG_SUBSYSTEM_MQTT, + target_id = %self.id, + topic = %self.args.topic, + state = "published", + "mqtt delivery state" + ); + self.delivery_counters.record_success(); + Ok(()) + } + Ok(Err(e)) => { + let err = classify_mqtt_notice_error(&e); + warn!( + event = EVENT_MQTT_DELIVERY_STATE, + component = LOG_COMPONENT_TARGETS, + subsystem = LOG_SUBSYSTEM_MQTT, + target_id = %self.id, + state = "publish_unconfirmed", + error = %e, + "mqtt delivery state" + ); + mark_target_disconnected_on_connectivity_error(&self.connected, &err); + Err(err) + } + Err(_) => { + let err = TargetError::Timeout("Timed out waiting for MQTT publish acknowledgement".to_string()); + warn!( + event = EVENT_MQTT_DELIVERY_STATE, + component = LOG_COMPONENT_TARGETS, + subsystem = LOG_SUBSYSTEM_MQTT, + target_id = %self.id, + state = "publish_confirm_timeout", + "mqtt delivery state" + ); + mark_target_disconnected_on_connectivity_error(&self.connected, &err); + Err(err) + } + } } pub fn clone_target(&self) -> Box + Send + Sync> { @@ -1190,6 +1232,33 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc TargetError { + match err { + ClientError::Request(_) | ClientError::TryRequest(_) | ClientError::TrackingUnavailable => TargetError::NotConnected, + } +} + +/// Classifies a publish acknowledgement failure returned while waiting for the +/// broker to confirm delivery. Connectivity/session problems keep the event for +/// replay; a broker rejection with a failing reason code is surfaced as a +/// request-level error (backlog#971). +fn classify_mqtt_notice_error(err: &PublishNoticeError) -> TargetError { + match err { + PublishNoticeError::Recv + | PublishNoticeError::SessionReset + | PublishNoticeError::Qos0NotFlushed + | PublishNoticeError::TopicAliasReplayUnavailable(_) => TargetError::NotConnected, + PublishNoticeError::V5PubAck(_) | PublishNoticeError::V5PubRec(_) | PublishNoticeError::V5PubComp(_) => { + TargetError::Request(format!("MQTT broker rejected publish: {err}")) + } + } +} + /// Check whether the given MQTT connection error should be considered a fatal error, /// For fatal errors, the event loop should terminate. fn is_fatal_mqtt_error(err: &ConnectionError) -> bool { @@ -1636,9 +1705,11 @@ where #[cfg(test)] mod tests { use super::{ - MQTT_RECONNECT_BACKOFF_MAX, MQTT_RECONNECT_BACKOFF_MIN, MQTTArgs, MQTTTlsConfig, QoS, next_reconnect_backoff, - reconnect_supervisor, validate_mqtt_broker_url, + ClientError, MQTT_RECONNECT_BACKOFF_MAX, MQTT_RECONNECT_BACKOFF_MIN, MQTTArgs, MQTTTlsConfig, PublishNoticeError, QoS, + classify_mqtt_client_error, classify_mqtt_notice_error, next_reconnect_backoff, reconnect_supervisor, + validate_mqtt_broker_url, }; + use crate::error::TargetError; use crate::target::{REDACTED_SECRET, TargetType}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -1646,6 +1717,38 @@ mod tests { use tokio::sync::mpsc; use url::Url; + #[test] + fn mqtt_client_error_classified_as_not_connected() { + // A publish that cannot be handed to the event loop means the client is + // not connected; the durable copy must be kept for replay (backlog#971). + assert!(matches!( + classify_mqtt_client_error(&ClientError::TrackingUnavailable), + TargetError::NotConnected + )); + } + + #[test] + fn mqtt_notice_connectivity_errors_kept_for_replay() { + for err in [ + PublishNoticeError::SessionReset, + PublishNoticeError::Qos0NotFlushed, + PublishNoticeError::Recv, + ] { + assert!( + matches!(classify_mqtt_notice_error(&err), TargetError::NotConnected), + "unconfirmed publish {err:?} should be retriable" + ); + } + } + + #[test] + fn mqtt_notice_broker_rejection_is_request_error() { + // The broker acknowledged the publish but rejected it: this is a + // request-level failure, not a transient disconnect. + let err = PublishNoticeError::V5PubAck(rumqttc::PubAckReason::NotAuthorized); + assert!(matches!(classify_mqtt_notice_error(&err), TargetError::Request(_))); + } + #[test] fn next_reconnect_backoff_doubles_until_capped() { let mut backoff = MQTT_RECONNECT_BACKOFF_MIN; diff --git a/crates/targets/src/target/nats.rs b/crates/targets/src/target/nats.rs index 569292537..7b13fe3e0 100644 --- a/crates/targets/src/target/nats.rs +++ b/crates/targets/src/target/nats.rs @@ -24,8 +24,8 @@ use crate::{ store::{Key, Store}, target::{ ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot, - TargetTlsState, TargetType, build_queued_payload_with_records, build_target_tls_fingerprint, open_target_queue_store, - persist_queued_payload_to_store, redacted_secret, + TargetTlsState, TargetType, build_queued_payload_with_records, build_target_tls_fingerprint, is_connectivity_error, + open_target_queue_store, persist_queued_payload_to_store, redacted_secret, }, }; use async_trait::async_trait; @@ -160,6 +160,21 @@ fn validate_nats_auth(args: &NATSArgs) -> Result<(), TargetError> { Ok(()) } +/// Returns true when the target is configured to send credentials (token, +/// username/password, or a credentials file) over a connection that does not +/// require TLS, which would transmit the secrets in cleartext (backlog#983). +/// +/// TLS is considered active when `tls_required` is set or the address uses the +/// `tls://` scheme. +fn nats_sends_credentials_without_tls(args: &NATSArgs) -> bool { + let has_auth = !args.token.is_empty() || !args.credentials_file.is_empty() || !args.username.is_empty(); + if !has_auth { + return false; + } + let scheme_is_tls = args.address.trim_start().to_ascii_lowercase().starts_with("tls://"); + !(args.tls_required || scheme_is_tls) +} + pub async fn connect_nats(args: &NATSArgs) -> Result { args.validate()?; @@ -228,6 +243,13 @@ where #[instrument(skip(args), fields(target_id_as_string = %id))] pub fn new(id: String, args: NATSArgs) -> Result { args.validate()?; + if args.enable && nats_sends_credentials_without_tls(&args) { + warn!( + target_id = %id, + address = %args.address, + "NATS target sends authentication credentials without TLS; secrets are transmitted in cleartext. Enable tls_required or use a tls:// address." + ); + } let target_id = TargetID::new(id, ChannelTargetType::Nats.as_str().to_string()); let queue_store = open_target_queue_store( &args.queue_dir, @@ -305,15 +327,55 @@ where async fn send_body(&self, body: Vec) -> Result<(), TargetError> { let client = self.get_or_connect().await?; - client - .publish(self.args.subject.clone(), body.into()) - .await - .map_err(|e| TargetError::Request(format!("Failed to publish NATS message: {e}")))?; + if let Err(e) = client.publish(self.args.subject.clone(), body.into()).await { + let err = classify_nats_publish_error(&e); + if is_connectivity_error(&err) { + self.invalidate_cached_client_connection().await; + self.connected.store(false, Ordering::SeqCst); + } + return Err(err); + } + + // `publish` only enqueues the message on the client's outbound channel; + // it does not guarantee the broker received it. Flush to confirm the + // message reached the server before the caller treats delivery as + // successful and deletes the durable copy (backlog#971). + if let Err(e) = client.flush().await { + let err = classify_nats_flush_error(&e); + self.invalidate_cached_client_connection().await; + self.connected.store(false, Ordering::SeqCst); + return Err(err); + } + self.delivery_counters.record_success(); Ok(()) } } +/// Classifies a NATS publish failure using the typed error kind rather than a +/// substring match. `Send` means the outbound channel is closed (the connection +/// is gone) and is retriable; payload/subject violations are permanent +/// request-level errors (backlog#971, backlog#973). +fn classify_nats_publish_error(err: &async_nats::PublishError) -> TargetError { + use async_nats::PublishErrorKind; + match err.kind() { + PublishErrorKind::Send => TargetError::NotConnected, + PublishErrorKind::MaxPayloadExceeded | PublishErrorKind::InvalidSubject => { + TargetError::Request(format!("Failed to publish NATS message: {err}")) + } + } +} + +/// Classifies a NATS flush failure. Both flush error kinds indicate the message +/// was not confirmed on the wire (a connectivity problem), so the event is kept +/// for replay (backlog#971, backlog#973). +fn classify_nats_flush_error(err: &async_nats::client::FlushError) -> TargetError { + use async_nats::client::FlushErrorKind; + match err.kind() { + FlushErrorKind::SendError | FlushErrorKind::FlushError => TargetError::NotConnected, + } +} + #[async_trait] impl Target for NATSTarget where @@ -510,4 +572,56 @@ mod tests { }; assert!(args.validate().is_err()); } + + #[test] + fn nats_credentials_without_tls_is_detected() { + // Token auth over a plaintext nats:// address without tls_required leaks + // the credential in cleartext (backlog#983). + let insecure = NATSArgs { + token: "secret-token".to_string(), + tls_required: false, + ..base_args() + }; + assert!(nats_sends_credentials_without_tls(&insecure)); + + // tls_required protects the credentials. + let with_tls = NATSArgs { + tls_required: true, + ..insecure.clone() + }; + assert!(!nats_sends_credentials_without_tls(&with_tls)); + + // A tls:// address also counts as protected. + let tls_scheme = NATSArgs { + address: "tls://127.0.0.1:4222".to_string(), + ..insecure + }; + assert!(!nats_sends_credentials_without_tls(&tls_scheme)); + + // No credentials configured: nothing to leak. + let no_auth = base_args(); + assert!(!nats_sends_credentials_without_tls(&no_auth)); + } + + #[test] + fn nats_publish_error_classification() { + use async_nats::PublishErrorKind; + // `Send` means the outbound channel is gone: retriable connectivity error. + let send_err = async_nats::PublishError::new(PublishErrorKind::Send); + assert!(matches!(classify_nats_publish_error(&send_err), TargetError::NotConnected)); + + // Payload/subject violations are permanent request-level errors. + let payload_err = async_nats::PublishError::new(PublishErrorKind::MaxPayloadExceeded); + assert!(matches!(classify_nats_publish_error(&payload_err), TargetError::Request(_))); + let subject_err = async_nats::PublishError::new(PublishErrorKind::InvalidSubject); + assert!(matches!(classify_nats_publish_error(&subject_err), TargetError::Request(_))); + } + + #[test] + fn nats_flush_error_classification() { + use async_nats::client::{FlushError, FlushErrorKind}; + for kind in [FlushErrorKind::SendError, FlushErrorKind::FlushError] { + assert!(matches!(classify_nats_flush_error(&FlushError::new(kind)), TargetError::NotConnected)); + } + } } diff --git a/crates/targets/src/target/pulsar.rs b/crates/targets/src/target/pulsar.rs index 2d6bea1ad..a9a1fc2f3 100644 --- a/crates/targets/src/target/pulsar.rs +++ b/crates/targets/src/target/pulsar.rs @@ -29,12 +29,16 @@ use crate::{ }, }; use async_trait::async_trait; +// Use parking_lot's Mutex for the synchronous client/TLS state guards: it does +// not poison on panic, so a panic while a guard is held cannot cascade into +// `.unwrap()` panics on every later access (backlog#983). +use parking_lot::Mutex; use pulsar::{Authentication, Producer, Pulsar, TokioExecutor}; use rustfs_tls_runtime::load_cert_bundle_der_bytes; use std::fmt; use std::path::Path; +use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; use tokio::sync::Mutex as AsyncMutex; use tracing::{info, instrument}; use url::Url; @@ -208,8 +212,8 @@ where Box::new(PulsarTarget:: { id: self.id.clone(), args: self.args.clone(), - client: Mutex::new(self.client.lock().unwrap().clone()), - tls_state: Mutex::new(self.tls_state.lock().unwrap().clone()), + client: Mutex::new(self.client.lock().clone()), + tls_state: Mutex::new(self.tls_state.lock().clone()), tls_adapter: self.tls_adapter.clone(), producer: AsyncMutex::new(None), store: self.store.as_ref().map(|s| s.boxed_clone()), @@ -247,12 +251,12 @@ where } fn clear_cached_client_connection(&self) { - self.client.lock().unwrap().take(); + self.client.lock().take(); } fn clear_cached_client(&self) { self.clear_cached_client_connection(); - self.tls_state.lock().unwrap().reset(); + self.tls_state.lock().reset(); } async fn get_or_connect_client(&self) -> Result, TargetError> { @@ -261,28 +265,28 @@ where if let Some(adapter) = &self.tls_adapter { let material = adapter.current_material(); { - let mut guard = self.client.lock().unwrap(); + let mut guard = self.client.lock(); *guard = Some((*material).clone()); } } else { let next_fingerprint = build_target_tls_fingerprint(&self.args.tls_ca, "", "").await?; let tls_changed = { - let tls_state_guard = self.tls_state.lock().unwrap(); + let tls_state_guard = self.tls_state.lock(); tls_state_guard.needs_update(&next_fingerprint) }; if tls_changed { self.clear_cached_client_connection(); - self.tls_state.lock().unwrap().refresh(next_fingerprint); + self.tls_state.lock().refresh(next_fingerprint); } } - if let Some(client) = self.client.lock().unwrap().clone() { + if let Some(client) = self.client.lock().clone() { return Ok(client); } let client = connect_pulsar(&self.args).await?; self.connected.store(true, Ordering::SeqCst); - let mut guard = self.client.lock().unwrap(); + let mut guard = self.client.lock(); let shared = guard.get_or_insert_with(|| client.clone()).clone(); Ok(shared) } @@ -363,7 +367,7 @@ where ) -> Result<(), TargetError> { // Pulsar client is Clone, so we clone from the Arc and store it. { - let mut guard = self.client.lock().unwrap(); + let mut guard = self.client.lock(); *guard = Some((*material).clone()); } // Producer is bound to the old client; clear it so next send rebuilds. diff --git a/crates/targets/src/target/redis.rs b/crates/targets/src/target/redis.rs index 94fa7a8eb..b62470379 100644 --- a/crates/targets/src/target/redis.rs +++ b/crates/targets/src/target/redis.rs @@ -474,8 +474,26 @@ where .publish::<_, _, i64>(self.args.channel.as_str(), body.as_slice()) .await { - Ok(_) => { - debug!(target_id = %self.id, channel = %self.args.channel, attempt, "Event published to Redis channel"); + Ok(receiver_count) => { + // PUBLISH returns the number of subscribers that received the + // message. Redis pub/sub is best-effort: with zero subscribers + // the event is delivered to no one, yet the durable copy is + // deleted. Warn so operators relying on reliable delivery are + // not silently losing events (backlog#982). + if receiver_count == 0 { + warn!( + target_id = %self.id, + channel = %self.args.channel, + "Redis PUBLISH reached 0 subscribers; the event was not received by any consumer (pub/sub is best-effort)" + ); + } + debug!( + target_id = %self.id, + channel = %self.args.channel, + attempt, + receiver_count, + "Event published to Redis channel" + ); self.delivery_counters.record_success(); return Ok(()); } @@ -528,14 +546,17 @@ where return Ok(false); } - let client = self.publisher_client.lock().clone(); - match tokio::time::timeout(Duration::from_secs(5), ping_redis_server(&client, &self.args)).await { + // Reuse the cached ConnectionManager for the health probe (via + // ensure_publisher_ready) instead of building a brand-new manager — and + // thus a fresh TCP+TLS handshake — on every health check (backlog#982). + // ensure_publisher_ready already invalidates the cached manager on a + // connectivity error so the next attempt rebuilds it. + match tokio::time::timeout(Duration::from_secs(5), self.ensure_publisher_ready()).await { Ok(Ok(())) => { self.connected.store(true, Ordering::SeqCst); Ok(true) } Ok(Err(err)) => { - invalidate_cache_on_connectivity_error(&err, || self.invalidate_cached_publisher()).await; mark_target_disconnected_on_connectivity_error(&self.connected, &err); Err(err) } @@ -650,6 +671,13 @@ pub(crate) fn build_redis_client(args: &RedisArgs) -> Result