Merge branch 'master' into pront-changelog-types

Author: pront

Cargo.lock

@@ -0,0 +1,5 @@
+Added an optional `tls_handshake_timeout_secs` setting to the `socket` (TCP mode), `syslog` (TCP mode), `logstash`, `fluent`, and `statsd` (TCP mode) sources. When set, a TLS-enabled connection that does not complete its TLS handshake within the configured number of seconds is closed.
+
+Previously, TLS handshakes on these sources had no timeout: a client that opened a TCP connection and never completed (or never started) the TLS handshake would hold its slot against `connection_limit` indefinitely, since neither TCP keepalive nor `max_connection_duration_secs` are evaluated until after the handshake succeeds. This could let misbehaving or unresponsive clients gradually exhaust the connection limit and block legitimate traffic. The new setting is unset by default, preserving prior behavior.
+
+authors: vladimir-dd

changelog.d/tls_handshake_timeout.feature.md

@@ -53,4 +53,5 @@ ignore = [
   { id = "RUSTSEC-2026-0195", reason = "quick-xml unbounded namespace-declaration allocation in NsReader - versions needing upgrade are v0.31 (direct dep) and v0.37.4 (via opendal < 0.58)" },
   { id = "RUSTSEC-2026-0235", reason = "rkyv 0.7 is vulnerable but unmaintained; upgrading vector-buffers to rkyv 0.8 requires a breaking serialization migration" },
   { id = "RUSTSEC-2026-0253", reason = "Vector uses lru 0.18.2; aws-sdk-s3 and ratatui still require lru ^0.16.3" },
+  { id = "RUSTSEC-2026-0258", reason = "h2 0.3 via aws-smithy-http-client (hyper 0.14) and tonic 0.11 - http 1.0 / AWS SDK upgrade (https://github.com/vectordotdev/vector/issues/19179)" },
 ]

deny.toml

@@ -129,6 +129,40 @@ impl InternalEvent for TcpSocketTlsConnectionError {
     }
 }
 
+/// Emitted when a TLS handshake does not complete within the configured
+/// handshake timeout. Without this, a peer that opens a TCP connection but
+/// never completes (or never starts) the TLS handshake — a bare TCP
+/// healthcheck, a stalled client, or deliberate slow-handshake abuse — would
+/// hold its connection-limit slot indefinitely, since neither TCP keepalive
+/// nor `max_connection_duration_secs` are wired up until after the handshake
+/// succeeds.
+#[derive(Debug, NamedInternalEvent)]
+pub struct TcpSocketTlsHandshakeTimeout {
+    pub peer_addr: SocketAddr,
+    pub timeout: std::time::Duration,
+}
+
+impl InternalEvent for TcpSocketTlsHandshakeTimeout {
+    fn emit(self) {
+        warn!(
+            message = "TLS handshake timed out.",
+            peer_addr = %self.peer_addr,
+            timeout_secs = self.timeout.as_secs(),
+            error_code = "tls_handshake_timeout",
+            error_type = error_type::TIMED_OUT,
+            stage = error_stage::RECEIVING,
+        );
+        counter!(
+            CounterName::ComponentErrorsTotal,
+            "error_code" => "tls_handshake_timeout",
+            "error_type" => error_type::TIMED_OUT,
+            "stage" => error_stage::RECEIVING,
+            "mode" => "tcp",
+        )
+        .increment(1);
+    }
+}
+
 #[derive(Debug, NamedInternalEvent)]
 pub struct TcpSendAckError {
     pub error: std::io::Error,

src/internal_events/tcp.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, io, net::SocketAddr, time::Duration};
+use std::{collections::HashMap, io, net::SocketAddr, num::NonZeroU64, time::Duration};
 
 use base64::prelude::{BASE64_STANDARD, Engine as _};
 use bytes::{Buf, Bytes, BytesMut};
@@ -184,6 +184,14 @@ pub struct FluentTcpConfig {
     #[configurable(metadata(docs::examples = 65536))]
     receive_buffer_bytes: Option<usize>,
 
+    /// The timeout, in seconds, before a TLS handshake is aborted if it has not completed.
+    ///
+    /// This bounds how long a connection can hold its slot against `connection_limit`
+    /// before the TLS handshake finishes, protecting against clients that open a
+    /// connection but never complete (or never start) a handshake.
+    #[configurable(metadata(docs::type_unit = "seconds"))]
+    tls_handshake_timeout_secs: Option<NonZeroU64>,
+
     #[configurable(derived)]
     tls: Option<TlsSourceConfig>,
 
@@ -216,6 +224,7 @@ impl FluentTcpConfig {
             tls_client_metadata_key,
             self.receive_buffer_bytes,
             None,
+            self.tls_handshake_timeout_secs,
             cx,
             self.acknowledgements,
             self.connection_limit,
@@ -277,6 +286,7 @@ impl GenerateConfig for FluentConfig {
                 permit_origin: None,
                 tls: None,
                 receive_buffer_bytes: None,
+                tls_handshake_timeout_secs: None,
                 acknowledgements: Default::default(),
                 connection_limit: Some(2),
             }),
@@ -1167,6 +1177,7 @@ mod tests {
                 keepalive: None,
                 permit_origin: None,
                 receive_buffer_bytes: None,
+                tls_handshake_timeout_secs: None,
                 acknowledgements: true.into(),
                 connection_limit: None,
             }),
@@ -1240,6 +1251,7 @@ mod tests {
                 keepalive: None,
                 permit_origin: None,
                 receive_buffer_bytes: None,
+                tls_handshake_timeout_secs: None,
                 acknowledgements: false.into(),
                 connection_limit: None,
             }),
@@ -1298,6 +1310,7 @@ mod tests {
                 keepalive: None,
                 permit_origin: None,
                 receive_buffer_bytes: None,
+                tls_handshake_timeout_secs: None,
                 acknowledgements: false.into(),
                 connection_limit: None,
             }),
@@ -1527,6 +1540,7 @@ mod integration_tests {
                     keepalive: None,
                     permit_origin: None,
                     receive_buffer_bytes: None,
+                    tls_handshake_timeout_secs: None,
                     acknowledgements: false.into(),
                     connection_limit: None,
                 }),

src/sources/fluent/mod.rs

@@ -3,7 +3,7 @@ use std::{
     convert::TryFrom,
     io,
     net::SocketAddr,
-    num::NonZeroUsize,
+    num::{NonZeroU64, NonZeroUsize},
     time::Duration,
 };
 
@@ -62,6 +62,14 @@ pub struct LogstashConfig {
     #[configurable(metadata(docs::type_unit = "connections"))]
     connection_limit: Option<u32>,
 
+    /// The timeout, in seconds, before a TLS handshake is aborted if it has not completed.
+    ///
+    /// This bounds how long a connection can hold its slot against `connection_limit`
+    /// before the TLS handshake finishes, protecting against clients that open a
+    /// connection but never complete (or never start) a handshake.
+    #[configurable(metadata(docs::type_unit = "seconds"))]
+    tls_handshake_timeout_secs: Option<NonZeroU64>,
+
     #[configurable(derived)]
     #[serde(default, deserialize_with = "bool_or_struct")]
     acknowledgements: SourceAcknowledgementsConfig,
@@ -125,6 +133,7 @@ impl Default for LogstashConfig {
             receive_buffer_bytes: None,
             acknowledgements: Default::default(),
             connection_limit: None,
+            tls_handshake_timeout_secs: None,
             log_namespace: None,
         }
     }
@@ -164,6 +173,7 @@ impl SourceConfig for LogstashConfig {
             tls_client_metadata_key,
             self.receive_buffer_bytes,
             None,
+            self.tls_handshake_timeout_secs,
             cx,
             self.acknowledgements,
             self.connection_limit,
@@ -902,6 +912,7 @@ mod test {
             receive_buffer_bytes: None,
             acknowledgements: true.into(),
             connection_limit: None,
+            tls_handshake_timeout_secs: None,
             log_namespace: None,
         }
         .build(SourceContext::new_test(sender, None))
@@ -1667,6 +1678,7 @@ mod integration_tests {
                 receive_buffer_bytes: None,
                 acknowledgements: false.into(),
                 connection_limit: None,
+                tls_handshake_timeout_secs: None,
                 log_namespace: None,
             }
             .build(SourceContext::new_test(sender, None))

src/sources/logstash.rs

@@ -143,6 +143,7 @@ impl SourceConfig for SocketConfig {
                     tls_client_metadata_key,
                     config.receive_buffer_bytes(),
                     config.max_connection_duration_secs(),
+                    config.tls_handshake_timeout_secs(),
                     cx,
                     false.into(),
                     config.connection_limit,
@@ -322,6 +323,7 @@ mod test {
     use std::{
         collections::HashMap,
         net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket},
+        num::NonZeroU64,
         sync::{
             Arc,
             atomic::{AtomicBool, Ordering},
@@ -930,6 +932,43 @@ mod test {
         }
     }
 
+    #[tokio::test]
+    async fn tcp_tls_handshake_timeout() {
+        let (tx, _) = SourceSender::new_test();
+        let (guard, addr) = next_addr();
+
+        let mut config = TcpConfig::from_address(addr.into());
+        config.set_tls(Some(TlsSourceConfig {
+            tls_config: TlsEnableableConfig::test_config(),
+            client_metadata_key: None,
+        }));
+        config.set_tls_handshake_timeout_secs(Some(NonZeroU64::new(1).unwrap()));
+
+        let source_task = SocketConfig::from(config)
+            .build(SourceContext::new_test(tx, None))
+            .await
+            .unwrap();
+
+        // Spawn the source task and wait until we're sure it's listening:
+        drop(tokio::spawn(source_task));
+        wait_for_tcp_and_release(guard, addr).await;
+
+        // Open a plain TCP connection but deliberately never send a TLS
+        // ClientHello, so the server's handshake never completes on its own.
+        let mut stream: TcpStream = TcpStream::connect(addr)
+            .await
+            .expect("stream should be able to connect");
+        let start = Instant::now();
+
+        let bytes_read = timeout(Duration::from_secs(2), stream.read(&mut [0]))
+            .await
+            .expect("timed out waiting for stream to close")
+            .expect("failed to read from stream");
+
+        assert_eq!(bytes_read, 0, "unexpectedly read data from stream");
+        assert_relative_eq!(start.elapsed().as_secs_f64(), 1.0, epsilon = 0.5);
+    }
+
     //////// UDP TESTS ////////
     async fn send_lines_udp(to: SocketAddr, lines: impl IntoIterator<Item = String>) -> UdpSocket {
         send_lines_udp_from(bind_unused_udp(), to, lines)

src/sources/socket/mod.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{num::NonZeroU64, time::Duration};
 
 use chrono::Utc;
 use serde_with::serde_as;
@@ -75,6 +75,14 @@ pub struct TcpConfig {
     #[configurable(metadata(docs::type_unit = "seconds"))]
     max_connection_duration_secs: Option<u64>,
 
+    /// The timeout, in seconds, before a TLS handshake is aborted if it has not completed.
+    ///
+    /// This bounds how long a connection can hold its slot against `connection_limit`
+    /// before the TLS handshake finishes, protecting against clients that open a
+    /// connection but never complete (or never start) a handshake.
+    #[configurable(metadata(docs::type_unit = "seconds"))]
+    tls_handshake_timeout_secs: Option<NonZeroU64>,
+
     /// The maximum number of TCP connections that are allowed at any given time.
     #[configurable(metadata(docs::type_unit = "connections"))]
     pub connection_limit: Option<u32>,
@@ -112,6 +120,7 @@ impl TcpConfig {
             tls: None,
             receive_buffer_bytes: None,
             max_connection_duration_secs: None,
+            tls_handshake_timeout_secs: None,
             framing: None,
             decoding: default_decoding(),
             connection_limit: None,
@@ -164,6 +173,15 @@ impl TcpConfig {
         self
     }
 
+    pub const fn tls_handshake_timeout_secs(&self) -> Option<NonZeroU64> {
+        self.tls_handshake_timeout_secs
+    }
+
+    pub const fn set_tls_handshake_timeout_secs(&mut self, val: Option<NonZeroU64>) -> &mut Self {
+        self.tls_handshake_timeout_secs = val;
+        self
+    }
+
     pub const fn set_shutdown_timeout_secs(&mut self, val: u64) -> &mut Self {
         self.shutdown_timeout_secs = Duration::from_secs(val);
         self

src/sources/socket/tcp.rs

@@ -1,5 +1,6 @@
 use std::{
     net::{Ipv4Addr, SocketAddr, SocketAddrV4},
+    num::NonZeroU64,
     time::Duration,
 };
 
@@ -139,6 +140,14 @@ pub struct TcpConfig {
     #[configurable(metadata(docs::type_unit = "connections"))]
     connection_limit: Option<u32>,
 
+    /// The timeout, in seconds, before a TLS handshake is aborted if it has not completed.
+    ///
+    /// This bounds how long a connection can hold its slot against `connection_limit`
+    /// before the TLS handshake finishes, protecting against clients that open a
+    /// connection but never complete (or never start) a handshake.
+    #[configurable(metadata(docs::type_unit = "seconds"))]
+    tls_handshake_timeout_secs: Option<NonZeroU64>,
+
     ///	Whether or not to sanitize incoming statsd key names. When "true", keys are sanitized by:
     /// - "/" is replaced with "-"
     /// - All whitespace is replaced with "_"
@@ -163,6 +172,7 @@ impl TcpConfig {
             shutdown_timeout_secs: default_shutdown_timeout_secs(),
             receive_buffer_bytes: None,
             connection_limit: None,
+            tls_handshake_timeout_secs: None,
             sanitize: default_sanitize(),
             convert_to: default_convert_to(),
         }
@@ -223,6 +233,7 @@ impl SourceConfig for StatsdConfig {
                     tls_client_metadata_key,
                     config.receive_buffer_bytes,
                     None,
+                    config.tls_handshake_timeout_secs,
                     cx,
                     false.into(),
                     config.connection_limit,