From d0343db3b2fb6973e803c7a5d099da9649ab4efc Mon Sep 17 00:00:00 2001 From: rjshrjndrn Date: Mon, 20 Jul 2026 20:42:32 +0200 Subject: [PATCH 1/9] feat(socket source): add PROXY protocol v2 header parser Add a pure, I/O-free parser for the PROXY protocol v2 binary header: signature check, IPv4/IPv6 address decoding, and a TLV walk that extracts custom 0xE0-0xEF values (e.g. tenant_id) while tolerating unknown TLVs. Covered by unit tests; not yet wired into the source. --- src/sources/util/net/mod.rs | 2 + src/sources/util/net/proxy_protocol.rs | 285 +++++++++++++++++++++++++ 2 files changed, 287 insertions(+) create mode 100644 src/sources/util/net/proxy_protocol.rs diff --git a/src/sources/util/net/mod.rs b/src/sources/util/net/mod.rs index 4ae7920973b7a..782a891712109 100644 --- a/src/sources/util/net/mod.rs +++ b/src/sources/util/net/mod.rs @@ -1,4 +1,6 @@ #[cfg(feature = "sources-utils-net-tcp")] +pub mod proxy_protocol; +#[cfg(feature = "sources-utils-net-tcp")] mod tcp; #[cfg(feature = "sources-utils-net-udp")] mod udp; diff --git a/src/sources/util/net/proxy_protocol.rs b/src/sources/util/net/proxy_protocol.rs new file mode 100644 index 0000000000000..b0b0ea4e62fa4 --- /dev/null +++ b/src/sources/util/net/proxy_protocol.rs @@ -0,0 +1,285 @@ +//! Minimal PROXY protocol v2 header parser. +//! +//! Parses the binary v2 header a fronting proxy (e.g. HAProxy `send-proxy-v2`) +//! prepends to a forwarded connection, exposing the original source/destination +//! address and any TLV fields (including custom `0xE0`..=`0xEF` values such as a +//! tenant identifier). See the PROXY protocol spec, section 2.2. + +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; + +/// The 12-byte signature that begins every PROXY protocol v2 header. +pub const V2_SIGNATURE: [u8; 12] = [ + 0x0D, 0x0A, 0x0D, 0x0A, 0x00, 0x0D, 0x0A, 0x51, 0x55, 0x49, 0x54, 0x0A, +]; + +/// The fixed prefix length that must be read before the declared length is known. +pub const V2_PREFIX_LEN: usize = 16; + +/// Errors produced while parsing a PROXY protocol v2 header. +#[derive(Debug, PartialEq, Eq)] +pub enum ParseError { + /// The buffer is shorter than the bytes the header claims to contain. + Truncated, + /// The 12-byte v2 signature was not found at the start of the buffer. + BadSignature, + /// The version nibble was not `2`. + UnsupportedVersion(u8), + /// The address family/transport byte was not understood. + UnsupportedFamily(u8), +} + +/// A single Type-Length-Value entry from the v2 TLV vector. +#[derive(Debug, PartialEq, Eq, Clone)] +pub struct Tlv { + /// The TLV type byte. + pub kind: u8, + /// The raw TLV value bytes. + pub value: Vec, +} + +/// A decoded PROXY protocol v2 header. +#[derive(Debug, PartialEq, Eq, Clone, Default)] +pub struct ProxyHeader { + /// Original client source address, if the family carried one. + pub source: Option, + /// Original destination address, if the family carried one. + pub destination: Option, + /// All TLV entries found after the address block, in order. + pub tlvs: Vec, +} + +impl ProxyHeader { + /// Return the value of the first TLV matching `kind`, if present. + pub fn tlv(&self, kind: u8) -> Option<&[u8]> { + self.tlvs + .iter() + .find(|t| t.kind == kind) + .map(|t| t.value.as_slice()) + } +} + +/// Total number of bytes a complete v2 header occupies given its fixed prefix. +/// +/// Returns `None` if fewer than [`V2_PREFIX_LEN`] bytes are available or the +/// signature does not match, so the caller knows not to treat the bytes as a +/// header. +pub fn v2_total_len(prefix: &[u8]) -> Option { + if prefix.len() < V2_PREFIX_LEN || prefix[..12] != V2_SIGNATURE { + return None; + } + let declared = u16::from_be_bytes([prefix[14], prefix[15]]) as usize; + Some(V2_PREFIX_LEN + declared) +} + +/// Parse a complete PROXY protocol v2 header from the start of `buf`. +/// +/// `buf` must contain at least the full header (see [`v2_total_len`]); any +/// trailing payload bytes are ignored. Returns the decoded header and the +/// number of bytes consumed. +pub fn parse_v2(buf: &[u8]) -> Result<(usize, ProxyHeader), ParseError> { + if buf.len() < V2_PREFIX_LEN { + return Err(ParseError::Truncated); + } + if buf[..12] != V2_SIGNATURE { + return Err(ParseError::BadSignature); + } + + let ver = buf[12] >> 4; + if ver != 2 { + return Err(ParseError::UnsupportedVersion(ver)); + } + // Lower nibble of byte 12 is the command (LOCAL/PROXY); not needed for the + // minimal slice. + + let family = buf[13] >> 4; + let declared = u16::from_be_bytes([buf[14], buf[15]]) as usize; + let header_end = V2_PREFIX_LEN + declared; + if buf.len() < header_end { + return Err(ParseError::Truncated); + } + + let addr_bytes = &buf[V2_PREFIX_LEN..header_end]; + let (source, destination, addr_len) = match family { + // AF_INET + 0x1 => { + if addr_bytes.len() < 12 { + return Err(ParseError::Truncated); + } + let src_ip = Ipv4Addr::new(addr_bytes[0], addr_bytes[1], addr_bytes[2], addr_bytes[3]); + let dst_ip = Ipv4Addr::new(addr_bytes[4], addr_bytes[5], addr_bytes[6], addr_bytes[7]); + let src_port = u16::from_be_bytes([addr_bytes[8], addr_bytes[9]]); + let dst_port = u16::from_be_bytes([addr_bytes[10], addr_bytes[11]]); + ( + Some(SocketAddr::new(IpAddr::V4(src_ip), src_port)), + Some(SocketAddr::new(IpAddr::V4(dst_ip), dst_port)), + 12, + ) + } + // AF_INET6 + 0x2 => { + if addr_bytes.len() < 36 { + return Err(ParseError::Truncated); + } + let mut src = [0u8; 16]; + let mut dst = [0u8; 16]; + src.copy_from_slice(&addr_bytes[0..16]); + dst.copy_from_slice(&addr_bytes[16..32]); + let src_port = u16::from_be_bytes([addr_bytes[32], addr_bytes[33]]); + let dst_port = u16::from_be_bytes([addr_bytes[34], addr_bytes[35]]); + ( + Some(SocketAddr::new(IpAddr::V6(Ipv6Addr::from(src)), src_port)), + Some(SocketAddr::new(IpAddr::V6(Ipv6Addr::from(dst)), dst_port)), + 36, + ) + } + // AF_UNSPEC (0x0) or AF_UNIX (0x3): no IP addresses to expose. + 0x0 | 0x3 => (None, None, if family == 0x3 { 216 } else { 0 }), + other => return Err(ParseError::UnsupportedFamily(other)), + }; + + let tlvs = parse_tlvs(&addr_bytes[addr_len.min(addr_bytes.len())..]); + + Ok(( + header_end, + ProxyHeader { + source, + destination, + tlvs, + }, + )) +} + +/// Walk a TLV vector, skipping any entry whose declared length overruns the +/// buffer. Unknown types are retained as-is so callers can decide what to do. +fn parse_tlvs(mut buf: &[u8]) -> Vec { + let mut out = Vec::new(); + while buf.len() >= 3 { + let kind = buf[0]; + let len = u16::from_be_bytes([buf[1], buf[2]]) as usize; + let value_start = 3; + let value_end = value_start + len; + if value_end > buf.len() { + // Malformed length; stop rather than read out of bounds. + break; + } + out.push(Tlv { + kind, + value: buf[value_start..value_end].to_vec(), + }); + buf = &buf[value_end..]; + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Build a v2 IPv4 header with the given TLV bytes appended. + fn v4_header(tlvs: &[u8]) -> Vec { + let addr: [u8; 12] = [ + 1, 2, 3, 4, // src 1.2.3.4 + 10, 0, 0, 1, // dst 10.0.0.1 + 0xD4, 0x31, // src port 54321 + 0x01, 0xBB, // dst port 443 + ]; + let len = (addr.len() + tlvs.len()) as u16; + let mut buf = Vec::new(); + buf.extend_from_slice(&V2_SIGNATURE); + buf.push(0x21); // ver 2, cmd PROXY + buf.push(0x11); // AF_INET, STREAM + buf.extend_from_slice(&len.to_be_bytes()); + buf.extend_from_slice(&addr); + buf.extend_from_slice(tlvs); + buf + } + + #[test] + fn rejects_bad_signature() { + let buf = [0u8; 16]; + assert_eq!(parse_v2(&buf), Err(ParseError::BadSignature)); + } + + #[test] + fn rejects_truncated_prefix() { + let buf = [0x0D, 0x0A, 0x0D]; + assert_eq!(parse_v2(&buf), Err(ParseError::Truncated)); + } + + #[test] + fn decodes_ipv4_addresses_and_ports() { + let buf = v4_header(&[]); + let (consumed, header) = parse_v2(&buf).expect("valid header"); + assert_eq!(consumed, buf.len()); + assert_eq!(header.source, Some("1.2.3.4:54321".parse().unwrap())); + assert_eq!(header.destination, Some("10.0.0.1:443".parse().unwrap())); + assert!(header.tlvs.is_empty()); + } + + #[test] + fn decodes_ipv6_addresses() { + let mut addr = Vec::new(); + addr.extend_from_slice(&Ipv6Addr::LOCALHOST.octets()); + addr.extend_from_slice(&Ipv6Addr::LOCALHOST.octets()); + addr.extend_from_slice(&1234u16.to_be_bytes()); + addr.extend_from_slice(&443u16.to_be_bytes()); + let mut buf = Vec::new(); + buf.extend_from_slice(&V2_SIGNATURE); + buf.push(0x21); + buf.push(0x21); // AF_INET6, STREAM + buf.extend_from_slice(&(addr.len() as u16).to_be_bytes()); + buf.extend_from_slice(&addr); + let (_, header) = parse_v2(&buf).expect("valid header"); + assert_eq!(header.source, Some("[::1]:1234".parse().unwrap())); + } + + #[test] + fn extracts_custom_tenant_tlv() { + // TLV type 0xE0, value "rajesh.com" + let value = b"rajesh.com"; + let mut tlv = vec![0xE0]; + tlv.extend_from_slice(&(value.len() as u16).to_be_bytes()); + tlv.extend_from_slice(value); + let buf = v4_header(&tlv); + let (_, header) = parse_v2(&buf).expect("valid header"); + assert_eq!(header.tlv(0xE0), Some(&b"rajesh.com"[..])); + } + + #[test] + fn skips_unknown_ssl_container_tlv_without_failing() { + // A 0x20 SSL container TLV we do not decode; must be tolerated. + let inner = [0x01, 0x00, 0x00, 0x00, 0x00]; // client byte + verify u32 + let mut tlv = vec![0x20]; + tlv.extend_from_slice(&(inner.len() as u16).to_be_bytes()); + tlv.extend_from_slice(&inner); + // Followed by our custom tenant TLV. + let value = b"acme"; + tlv.push(0xE0); + tlv.extend_from_slice(&(value.len() as u16).to_be_bytes()); + tlv.extend_from_slice(value); + + let buf = v4_header(&tlv); + let (_, header) = parse_v2(&buf).expect("valid header"); + assert_eq!(header.tlv(0x20).map(<[u8]>::to_vec), Some(inner.to_vec())); + assert_eq!(header.tlv(0xE0), Some(&b"acme"[..])); + } + + #[test] + fn truncated_body_is_reported() { + let mut buf = v4_header(&[]); + buf.truncate(buf.len() - 2); // drop part of the address block + assert_eq!(parse_v2(&buf), Err(ParseError::Truncated)); + } + + #[test] + fn v2_total_len_reads_declared_length() { + let buf = v4_header(&[]); + assert_eq!(v2_total_len(&buf[..V2_PREFIX_LEN]), Some(buf.len())); + } + + #[test] + fn v2_total_len_rejects_non_signature() { + let buf = [0u8; 16]; + assert_eq!(v2_total_len(&buf), None); + } +} From 8d0aad42ad3c50885accc9ae282e172d8874261d Mon Sep 17 00:00:00 2001 From: rjshrjndrn Date: Mon, 20 Jul 2026 20:54:53 +0200 Subject: [PATCH 2/9] test(socket source): validate PP2 parser against real HAProxy bytes Add a fixture captured from haproxy:2.9 with send-proxy-v2 and a custom 0xE0 tenant TLV, confirming the parser's field offsets match real proxy output rather than only self-authored fixtures. --- src/sources/util/net/proxy_protocol.rs | 27 ++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/src/sources/util/net/proxy_protocol.rs b/src/sources/util/net/proxy_protocol.rs index b0b0ea4e62fa4..272071c857b72 100644 --- a/src/sources/util/net/proxy_protocol.rs +++ b/src/sources/util/net/proxy_protocol.rs @@ -282,4 +282,31 @@ mod tests { let buf = [0u8; 16]; assert_eq!(v2_total_len(&buf), None); } + + /// Real bytes captured from `haproxy:2.9` configured with + /// `send-proxy-v2 set-proxy-v2-tlv-fmt(0xE0) rajesh.com`, followed by a + /// "HELLO-PAYLOAD\n" payload. Guards against a spec misreading shared by + /// the hand-built fixtures above. + #[test] + fn parses_real_haproxy_capture() { + let wire: [u8; 55] = [ + 0x0d, 0x0a, 0x0d, 0x0a, 0x00, 0x0d, 0x0a, 0x51, 0x55, 0x49, 0x54, 0x0a, // sig + 0x21, 0x11, 0x00, 0x19, // v2/PROXY, INET/STREAM, len 25 + 0xc0, 0xa8, 0x9b, 0x01, // src 192.168.155.1 + 0xc0, 0xa8, 0x9b, 0x03, // dst 192.168.155.3 + 0xb4, 0x26, // src port 46118 + 0x1b, 0x58, // dst port 7000 + 0xe0, 0x00, 0x0a, // TLV 0xE0, len 10 + 0x72, 0x61, 0x6a, 0x65, 0x73, 0x68, 0x2e, 0x63, 0x6f, 0x6d, // "rajesh.com" + 0x48, 0x45, 0x4c, 0x4c, 0x4f, 0x2d, 0x50, 0x41, 0x59, 0x4c, 0x4f, 0x41, 0x44, + 0x0a, // "HELLO-PAYLOAD\n" payload + ]; + let (consumed, header) = parse_v2(&wire).expect("real haproxy header"); + assert_eq!(consumed, 41, "header is 16 prefix + 25 declared"); + assert_eq!(header.source, Some("192.168.155.1:46118".parse().unwrap())); + assert_eq!(header.destination, Some("192.168.155.3:7000".parse().unwrap())); + assert_eq!(header.tlv(0xE0), Some(&b"rajesh.com"[..])); + // Payload survives untouched after the header. + assert_eq!(&wire[consumed..], b"HELLO-PAYLOAD\n"); + } } From 95cbb5cf0a18e0b2e82307353c733ecfc8ef81c4 Mon Sep 17 00:00:00 2001 From: rjshrjndrn Date: Mon, 20 Jul 2026 21:02:52 +0200 Subject: [PATCH 3/9] feat(socket source): wire PROXY protocol v2 into TCP source Add an opt-in proxy_protocol flag to the socket TCP source. When enabled, the connection handler reads and strips the v2 header after the TLS handshake and replaces the recorded peer address with the original client address from the header. Gated behind a default-off TcpSource trait method so other TCP sources are unaffected. Verified end-to-end against HAProxy send-proxy-v2: with the flag on the original client IP is recovered and the payload is intact; with it off the raw header bytes leak into the stream. --- src/sources/socket/mod.rs | 31 ++++++++++++ src/sources/socket/tcp.rs | 20 ++++++++ src/sources/util/net/proxy_protocol.rs | 67 +++++++++++++++++++++++++- src/sources/util/net/tcp/mod.rs | 31 +++++++++++- 4 files changed, 147 insertions(+), 2 deletions(-) diff --git a/src/sources/socket/mod.rs b/src/sources/socket/mod.rs index 66149775d5aa8..c9757d1334873 100644 --- a/src/sources/socket/mod.rs +++ b/src/sources/socket/mod.rs @@ -462,6 +462,37 @@ mod test { crate::test_util::test_generate_config::(); } + #[test] + fn tcp_proxy_protocol_defaults_off() { + let config: SocketConfig = toml::from_str( + r#" + mode = "tcp" + address = "127.0.0.1:9000" + "#, + ) + .unwrap(); + match config.mode { + Mode::Tcp(tcp) => assert!(!tcp.proxy_protocol()), + _ => panic!("expected tcp mode"), + } + } + + #[test] + fn tcp_proxy_protocol_can_be_enabled() { + let config: SocketConfig = toml::from_str( + r#" + mode = "tcp" + address = "127.0.0.1:9000" + proxy_protocol = true + "#, + ) + .unwrap(); + match config.mode { + Mode::Tcp(tcp) => assert!(tcp.proxy_protocol()), + _ => panic!("expected tcp mode"), + } + } + //////// TCP TESTS //////// #[tokio::test] async fn tcp_it_includes_host() { diff --git a/src/sources/socket/tcp.rs b/src/sources/socket/tcp.rs index 5e1873e6f9cde..80cd03724dc75 100644 --- a/src/sources/socket/tcp.rs +++ b/src/sources/socket/tcp.rs @@ -79,6 +79,17 @@ pub struct TcpConfig { #[configurable(metadata(docs::type_unit = "connections"))] pub connection_limit: Option, + /// Whether to parse a PROXY protocol v2 header prepended by a trusted + /// upstream proxy (for example HAProxy `send-proxy-v2`) at the start of + /// each connection. + /// + /// When enabled, the original client address from the header replaces the + /// peer address recorded on each event. Enable this only when a trusted + /// proxy sits in front of this source, since the header is otherwise + /// spoofable. + #[serde(default)] + pub proxy_protocol: bool, + #[configurable(derived)] pub(super) framing: Option, @@ -115,10 +126,15 @@ impl TcpConfig { framing: None, decoding: default_decoding(), connection_limit: None, + proxy_protocol: false, log_namespace: None, } } + pub const fn proxy_protocol(&self) -> bool { + self.proxy_protocol + } + pub(super) fn host_key(&self) -> OptionalValuePath { self.host_key.clone().unwrap_or(default_host_key()) } @@ -213,6 +229,10 @@ impl TcpSource for RawTcpSource { type Decoder = Decoder; type Acker = TcpNullAcker; + fn proxy_protocol(&self) -> bool { + self.config.proxy_protocol() + } + fn decoder(&self) -> Self::Decoder { self.decoder.clone() } diff --git a/src/sources/util/net/proxy_protocol.rs b/src/sources/util/net/proxy_protocol.rs index 272071c857b72..5783f89bc3408 100644 --- a/src/sources/util/net/proxy_protocol.rs +++ b/src/sources/util/net/proxy_protocol.rs @@ -7,6 +7,8 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; +use tokio::io::{AsyncRead, AsyncReadExt}; + /// The 12-byte signature that begins every PROXY protocol v2 header. pub const V2_SIGNATURE: [u8; 12] = [ 0x0D, 0x0A, 0x0D, 0x0A, 0x00, 0x0D, 0x0A, 0x51, 0x55, 0x49, 0x54, 0x0A, @@ -149,6 +151,36 @@ pub fn parse_v2(buf: &[u8]) -> Result<(usize, ProxyHeader), ParseError> { )) } +/// Read and consume exactly one PROXY protocol v2 header from `reader`, +/// leaving the reader positioned at the first payload byte. +/// +/// Reads the fixed 16-byte prefix, derives the declared length, then reads +/// exactly that many further bytes before parsing. Because it reads only the +/// header bytes, whatever follows on the stream is untouched and available to +/// the decoder. +pub async fn read_v2_header(reader: &mut R) -> std::io::Result +where + R: AsyncRead + Unpin, +{ + let mut prefix = [0u8; V2_PREFIX_LEN]; + reader.read_exact(&mut prefix).await?; + + let total = v2_total_len(&prefix).ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + "expected PROXY protocol v2 signature", + ) + })?; + + let mut buf = vec![0u8; total]; + buf[..V2_PREFIX_LEN].copy_from_slice(&prefix); + reader.read_exact(&mut buf[V2_PREFIX_LEN..]).await?; + + let (_, header) = parse_v2(&buf) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, format!("{e:?}")))?; + Ok(header) +} + /// Walk a TLV vector, skipping any entry whose declared length overruns the /// buffer. Unknown types are retained as-is so callers can decide what to do. fn parse_tlvs(mut buf: &[u8]) -> Vec { @@ -287,6 +319,36 @@ mod tests { /// `send-proxy-v2 set-proxy-v2-tlv-fmt(0xE0) rajesh.com`, followed by a /// "HELLO-PAYLOAD\n" payload. Guards against a spec misreading shared by /// the hand-built fixtures above. + /// The exact 55 bytes captured from haproxy:2.9 (header + payload). + fn real_capture() -> [u8; 55] { + [ + 0x0d, 0x0a, 0x0d, 0x0a, 0x00, 0x0d, 0x0a, 0x51, 0x55, 0x49, 0x54, 0x0a, 0x21, 0x11, + 0x00, 0x19, 0xc0, 0xa8, 0x9b, 0x01, 0xc0, 0xa8, 0x9b, 0x03, 0xb4, 0x26, 0x1b, 0x58, + 0xe0, 0x00, 0x0a, 0x72, 0x61, 0x6a, 0x65, 0x73, 0x68, 0x2e, 0x63, 0x6f, 0x6d, 0x48, + 0x45, 0x4c, 0x4c, 0x4f, 0x2d, 0x50, 0x41, 0x59, 0x4c, 0x4f, 0x41, 0x44, 0x0a, + ] + } + + #[tokio::test] + async fn read_v2_header_consumes_only_the_header() { + let wire = real_capture(); + let mut cursor = std::io::Cursor::new(wire.to_vec()); + let header = read_v2_header(&mut cursor).await.expect("header"); + assert_eq!(header.tlv(0xE0), Some(&b"rajesh.com"[..])); + // Remaining bytes on the stream are exactly the payload. + let mut rest = Vec::new(); + tokio::io::AsyncReadExt::read_to_end(&mut cursor, &mut rest) + .await + .unwrap(); + assert_eq!(rest, b"HELLO-PAYLOAD\n"); + } + + #[tokio::test] + async fn read_v2_header_errors_on_non_pp2_stream() { + let mut cursor = std::io::Cursor::new(b"not a proxy header at all........".to_vec()); + assert!(read_v2_header(&mut cursor).await.is_err()); + } + #[test] fn parses_real_haproxy_capture() { let wire: [u8; 55] = [ @@ -304,7 +366,10 @@ mod tests { let (consumed, header) = parse_v2(&wire).expect("real haproxy header"); assert_eq!(consumed, 41, "header is 16 prefix + 25 declared"); assert_eq!(header.source, Some("192.168.155.1:46118".parse().unwrap())); - assert_eq!(header.destination, Some("192.168.155.3:7000".parse().unwrap())); + assert_eq!( + header.destination, + Some("192.168.155.3:7000".parse().unwrap()) + ); assert_eq!(header.tlv(0xE0), Some(&b"rajesh.com"[..])); // Payload survives untouched after the header. assert_eq!(&wire[consumed..], b"HELLO-PAYLOAD\n"); diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs index aebce3dc9b6ee..e73c8a77ba567 100644 --- a/src/sources/util/net/tcp/mod.rs +++ b/src/sources/util/net/tcp/mod.rs @@ -113,6 +113,13 @@ where fn handle_events(&self, _events: &mut [Event], _host: std::net::SocketAddr) {} + /// Whether to parse and strip a PROXY protocol v2 header at the start of + /// each connection. Sources that support a trusted upstream proxy override + /// this; the default is off so behavior is unchanged for everyone else. + fn proxy_protocol(&self) -> bool { + false + } + fn build_acker(&self, item: &[Self::Item]) -> Self::Acker; #[allow(clippy::too_many_arguments)] @@ -257,7 +264,7 @@ async fn handle_stream( max_connection_duration_secs: Option, source: T, mut tripwire: BoxFuture<'static, ()>, - peer_addr: SocketAddr, + mut peer_addr: SocketAddr, mut out: SourceSender, acknowledgements: bool, request_limiter: RequestLimiter, @@ -292,6 +299,28 @@ async fn handle_stream( warn!(message = "Failed configuring receive buffer size on TCP socket.", %error); } + // Parse and strip a PROXY protocol v2 header, if the source opts in. Done + // after the TLS handshake so a plaintext hop and a re-encrypted hop share + // one code path, and before wrapping the socket so only header bytes are + // consumed and the payload reaches the decoder untouched. + if source.proxy_protocol() { + match super::proxy_protocol::read_v2_header(&mut socket).await { + Ok(header) => { + if let Some(source_addr) = header.source { + peer_addr = source_addr; + } + } + Err(error) => { + warn!( + message = "Failed to read PROXY protocol v2 header; closing connection.", + %error, + %peer_addr, + ); + return; + } + } + } + let socket = socket.after_read(move |byte_size| { emit!(TcpBytesReceived { byte_size, From 2b29caceea9619e8b242864e6f4a8b9ab40f2433 Mon Sep 17 00:00:00 2001 From: rjshrjndrn Date: Mon, 20 Jul 2026 21:29:57 +0200 Subject: [PATCH 4/9] feat(socket source): expose PROXY protocol v2 TLVs on events Surface parsed v2 TLV fields as event metadata under proxy_protocol, with TLV keys as the lowercase hex type byte (e.g. 0xe0) since the wire format identifies fields by numeric type. This lets downstream transforms and sinks read arbitrary proxy-supplied values, for example routing a Kafka topic by a tenant identifier carried in a custom TLV. Verified end-to-end against HAProxy sending two custom TLVs: both appear on the event and drive a templated field value. --- src/sources/util/net/proxy_protocol.rs | 42 ++++++++++++++++++++++++++ src/sources/util/net/tcp/mod.rs | 20 ++++++++++-- 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/src/sources/util/net/proxy_protocol.rs b/src/sources/util/net/proxy_protocol.rs index 5783f89bc3408..a9efebd9281e1 100644 --- a/src/sources/util/net/proxy_protocol.rs +++ b/src/sources/util/net/proxy_protocol.rs @@ -8,6 +8,7 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use tokio::io::{AsyncRead, AsyncReadExt}; +use vrl::value::{ObjectMap, Value}; /// The 12-byte signature that begins every PROXY protocol v2 header. pub const V2_SIGNATURE: [u8; 12] = [ @@ -58,6 +59,29 @@ impl ProxyHeader { .find(|t| t.kind == kind) .map(|t| t.value.as_slice()) } + + /// Convert the parsed header into an event metadata object of the shape + /// `{ version: 2, tlvs: { "0xe0": "", .. } }`. + /// + /// TLV keys are the lowercase hex type byte (e.g. `0xe0`) since the wire + /// format identifies fields by numeric type, not name. Values are exposed + /// as UTF-8 strings when valid, otherwise as raw bytes. + pub fn into_metadata(self) -> ObjectMap { + let mut tlvs = ObjectMap::new(); + for tlv in self.tlvs { + let key = format!("0x{:02x}", tlv.kind); + let value = match String::from_utf8(tlv.value.clone()) { + Ok(s) => Value::from(s), + Err(_) => Value::from(tlv.value), + }; + tlvs.insert(key.into(), value); + } + + let mut map = ObjectMap::new(); + map.insert("version".into(), Value::from(2)); + map.insert("tlvs".into(), Value::from(tlvs)); + map + } } /// Total number of bytes a complete v2 header occupies given its fixed prefix. @@ -343,6 +367,24 @@ mod tests { assert_eq!(rest, b"HELLO-PAYLOAD\n"); } + #[test] + fn into_metadata_exposes_hex_keyed_tlvs() { + let value = b"rajesh.com"; + let mut tlv = vec![0xE0]; + tlv.extend_from_slice(&(value.len() as u16).to_be_bytes()); + tlv.extend_from_slice(value); + let buf = v4_header(&tlv); + let (_, header) = parse_v2(&buf).expect("valid header"); + + let meta = header.into_metadata(); + assert_eq!(meta.get("version"), Some(&Value::from(2))); + let tlvs = match meta.get("tlvs") { + Some(Value::Object(o)) => o, + other => panic!("expected tlvs object, got {other:?}"), + }; + assert_eq!(tlvs.get("0xe0"), Some(&Value::from("rajesh.com"))); + } + #[tokio::test] async fn read_v2_header_errors_on_non_pp2_stream() { let mut cursor = std::io::Cursor::new(b"not a proxy header at all........".to_vec()); diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs index e73c8a77ba567..16235270de1cc 100644 --- a/src/sources/util/net/tcp/mod.rs +++ b/src/sources/util/net/tcp/mod.rs @@ -303,12 +303,13 @@ async fn handle_stream( // after the TLS handshake so a plaintext hop and a re-encrypted hop share // one code path, and before wrapping the socket so only header bytes are // consumed and the payload reaches the decoder untouched. - if source.proxy_protocol() { + let proxy_metadata = if source.proxy_protocol() { match super::proxy_protocol::read_v2_header(&mut socket).await { Ok(header) => { if let Some(source_addr) = header.source { peer_addr = source_addr; } + Some(header.into_metadata()) } Err(error) => { warn!( @@ -319,7 +320,9 @@ async fn handle_stream( return; } } - } + } else { + None + }; let socket = socket.after_read(move |byte_size| { emit!(TcpBytesReceived { @@ -427,6 +430,19 @@ async fn handle_stream( } } + if let Some(proxy_metadata) = &proxy_metadata { + for event in &mut events { + let log = event.as_mut_log(); + log_namespace.insert_source_metadata( + source_name, + log, + Some(LegacyKey::Overwrite(path!("proxy_protocol"))), + path!("proxy_protocol"), + proxy_metadata.clone(), + ); + } + } + source.handle_events(&mut events, peer_addr); match out.send_batch(events).await { Ok(_) => { From b0d7cc5f43fd75e8f05a3393f7cac5904ecf26a0 Mon Sep 17 00:00:00 2001 From: rjshrjndrn Date: Mon, 20 Jul 2026 21:50:52 +0200 Subject: [PATCH 5/9] docs(socket source): document proxy_protocol option Add a changelog fragment, the generated component doc entry for the new proxy_protocol option, and an example config routing a Kafka topic by a tenant id carried in a PROXY protocol v2 custom TLV. --- .../socket_proxy_protocol_v2.feature.md | 3 ++ config/examples/haproxy_proxy_protocol.yaml | 47 +++++++++++++++++++ .../components/sources/generated/socket.cue | 15 ++++++ 3 files changed, 65 insertions(+) create mode 100644 changelog.d/socket_proxy_protocol_v2.feature.md create mode 100644 config/examples/haproxy_proxy_protocol.yaml diff --git a/changelog.d/socket_proxy_protocol_v2.feature.md b/changelog.d/socket_proxy_protocol_v2.feature.md new file mode 100644 index 0000000000000..8f18ed71163bb --- /dev/null +++ b/changelog.d/socket_proxy_protocol_v2.feature.md @@ -0,0 +1,3 @@ +Added opt-in PROXY protocol v2 support to the `socket` source in `tcp` mode via a new `proxy_protocol` option. When enabled, Vector parses and strips the v2 header prepended by a trusted upstream proxy (such as HAProxy `send-proxy-v2`), replaces the recorded peer address with the original client address, and exposes any TLV fields under the `proxy_protocol` event metadata keyed by their hex type byte (for example `proxy_protocol.tlvs."0xe0"`). This allows routing on proxy-supplied values such as a tenant identifier carried in a custom TLV. + +authors: rjshrjndrn diff --git a/config/examples/haproxy_proxy_protocol.yaml b/config/examples/haproxy_proxy_protocol.yaml new file mode 100644 index 0000000000000..b634c31f44a8b --- /dev/null +++ b/config/examples/haproxy_proxy_protocol.yaml @@ -0,0 +1,47 @@ +# Ingest logs from behind a PROXY-protocol-aware load balancer. +# +# HAProxy terminates the client connection and prepends a PROXY protocol v2 +# header carrying the original client address plus custom TLV fields. Here a +# tenant identifier is sent in custom TLV type 0xE0, which Vector uses to route +# events to a per-tenant Kafka topic. +# +# Matching HAProxy backend configuration: +# +# backend bk_vector +# mode tcp +# server vector 10.0.0.9:9000 send-proxy-v2 \ +# set-proxy-v2-tlv-fmt(0xE0) %[ssl_c_s_dn(CN)] +# +# Enable proxy_protocol only when a trusted proxy sits in front of this source; +# the header is otherwise spoofable. + +sources: + tcp_in: + type: socket + mode: tcp + address: "0.0.0.0:9000" + proxy_protocol: true + decoding: + codec: bytes + +transforms: + extract_tenant: + type: remap + inputs: + - tcp_in + source: |- + # The wire identifies TLVs by numeric type, so custom fields are keyed by + # their lowercase hex byte. Fall back to "unknown" when the TLV is absent. + .tenant_name = string(.proxy_protocol.tlvs."0xe0") ?? "unknown" + # Kafka topic names allow only [a-zA-Z0-9._-]; sanitize untrusted values. + .tenant_name = replace(.tenant_name, r'[^a-zA-Z0-9._-]', "_") + +sinks: + kafka_out: + type: kafka + inputs: + - extract_tenant + bootstrap_servers: "kafka:9092" + topic: "logs-{{ tenant_name }}" + encoding: + codec: json diff --git a/website/cue/reference/components/sources/generated/socket.cue b/website/cue/reference/components/sources/generated/socket.cue index b0b399c3ae2d0..153932e33648b 100644 --- a/website/cue/reference/components/sources/generated/socket.cue +++ b/website/cue/reference/components/sources/generated/socket.cue @@ -644,6 +644,21 @@ generated: components: sources: socket: configuration: { required: false type: string: default: "port" } + proxy_protocol: { + description: """ + Whether to parse a PROXY protocol v2 header prepended by a trusted + upstream proxy (for example HAProxy `send-proxy-v2`) at the start of + each connection. + + When enabled, the original client address from the header replaces the + peer address recorded on each event. Enable this only when a trusted + proxy sits in front of this source, since the header is otherwise + spoofable. + """ + relevant_when: "mode = \"tcp\"" + required: false + type: bool: default: false + } receive_buffer_bytes: { description: "The size of the receive buffer used for each connection." relevant_when: "mode = \"tcp\" or mode = \"udp\"" From c5f5a70ff56ce0fa66abd4a57481d3bf18382b4b Mon Sep 17 00:00:00 2001 From: rjshrjndrn Date: Tue, 21 Jul 2026 05:40:17 +0200 Subject: [PATCH 6/9] fix(socket source): harden PROXY protocol handling Address review feedback on the PROXY protocol v2 path: - Guard metadata injection with a log-event check so native metric or trace frames no longer panic the per-connection task. - Bound the header read with the shutdown signal, connection tripwire, and a timeout (max_connection_duration_secs or a short default) so a peer that connects but never sends a header cannot hold a connection-limit permit indefinitely. - Declare the proxy_protocol metadata in the socket source output schema so downstream schema and VRL consumers know the field exists. --- src/sources/socket/mod.rs | 10 ++++ src/sources/util/net/proxy_protocol.rs | 78 ++++++++++++++++++++++++++ src/sources/util/net/tcp/mod.rs | 59 ++++++++++++------- 3 files changed, 128 insertions(+), 19 deletions(-) diff --git a/src/sources/socket/mod.rs b/src/sources/socket/mod.rs index c9757d1334873..782593d14cf4e 100644 --- a/src/sources/socket/mod.rs +++ b/src/sources/socket/mod.rs @@ -243,6 +243,16 @@ impl SourceConfig for SocketConfig { .or_undefined(), None, ) + .with_source_metadata( + Self::NAME, + config + .proxy_protocol() + .then(|| LegacyKey::Overwrite(owned_value_path!("proxy_protocol"))), + &owned_value_path!("proxy_protocol"), + Kind::object(Collection::empty().with_unknown(Kind::bytes())) + .or_undefined(), + None, + ) } Mode::Udp(config) => { let legacy_host_key = config.host_key().path.map(LegacyKey::InsertIfEmpty); diff --git a/src/sources/util/net/proxy_protocol.rs b/src/sources/util/net/proxy_protocol.rs index a9efebd9281e1..52153cd646f3c 100644 --- a/src/sources/util/net/proxy_protocol.rs +++ b/src/sources/util/net/proxy_protocol.rs @@ -8,6 +8,11 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use tokio::io::{AsyncRead, AsyncReadExt}; +use vector_lib::config::LogNamespace; +use vector_lib::event::Event; +use vector_lib::lookup::PathPrefix; +use vrl::owned_value_path; +use vrl::path::OwnedValuePath; use vrl::value::{ObjectMap, Value}; /// The 12-byte signature that begins every PROXY protocol v2 header. @@ -175,6 +180,41 @@ pub fn parse_v2(buf: &[u8]) -> Result<(usize, ProxyHeader), ParseError> { )) } +/// Inject parsed PROXY protocol metadata onto each log event in `events`. +/// +/// Non-log events (metrics, traces produced by native codecs) are skipped +/// rather than coerced, since `Event::as_mut_log` panics on them. +pub fn inject_metadata( + events: &mut [Event], + metadata: &ObjectMap, + log_namespace: LogNamespace, + source_name: &'static str, +) { + use vector_lib::config::LegacyKey; + use vrl::path; + + for event in events { + if let Event::Log(log) = event { + log_namespace.insert_source_metadata( + source_name, + log, + Some(LegacyKey::Overwrite(path!("proxy_protocol"))), + path!("proxy_protocol"), + metadata.clone(), + ); + } + } +} + +/// The metadata path where PROXY protocol fields are inserted, for declaring +/// the source's output schema. +pub fn metadata_path() -> OwnedValuePath { + owned_value_path!("proxy_protocol") +} + +/// The legacy-namespace insertion prefix, exposed for schema declaration. +pub const LEGACY_PREFIX: PathPrefix = PathPrefix::Event; + /// Read and consume exactly one PROXY protocol v2 header from `reader`, /// leaving the reader positioned at the first payload byte. /// @@ -353,6 +393,44 @@ mod tests { ] } + #[test] + fn inject_metadata_skips_non_log_events() { + use vector_lib::event::{Event, Metric, MetricKind, MetricValue}; + + let value = b"rajesh.com"; + let mut tlv = vec![0xE0]; + tlv.extend_from_slice(&(value.len() as u16).to_be_bytes()); + tlv.extend_from_slice(value); + let buf = v4_header(&tlv); + let meta = parse_v2(&buf).unwrap().1.into_metadata(); + + let mut events = vec![ + Event::Log(Default::default()), + Event::Metric(Metric::new( + "m", + MetricKind::Absolute, + MetricValue::Counter { value: 1.0 }, + )), + ]; + + // Must not panic on the metric event. + inject_metadata(&mut events, &meta, LogNamespace::Legacy, "socket"); + + // Log event carries the metadata. + use vrl::event_path; + let log = events[0].as_log(); + assert_eq!( + log.get(event_path!("proxy_protocol")) + .and_then(|v| v.as_object()) + .and_then(|o| o.get("tlvs")) + .and_then(|v| v.as_object()) + .and_then(|o| o.get("0xe0")), + Some(&Value::from("rajesh.com")) + ); + // Metric event is untouched (still a metric). + assert!(matches!(events[1], Event::Metric(_))); + } + #[tokio::test] async fn read_v2_header_consumes_only_the_header() { let wire = real_capture(); diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs index 16235270de1cc..79921f1b9ed45 100644 --- a/src/sources/util/net/tcp/mod.rs +++ b/src/sources/util/net/tcp/mod.rs @@ -48,6 +48,12 @@ use crate::{ pub const MAX_IN_FLIGHT_EVENTS_TARGET: usize = 100_000; +/// Fallback deadline for reading a PROXY protocol header when no explicit +/// `max_connection_duration_secs` is configured. A valid header is sent +/// immediately after connect, so a peer that has not produced one within this +/// window is treated as a stalled/misbehaving connection and dropped. +const PROXY_HEADER_READ_TIMEOUT_SECS: u64 = 10; + pub async fn try_bind_tcp_listener( addr: SocketListenAddr, mut listenfd: ListenFd, @@ -304,21 +310,40 @@ async fn handle_stream( // one code path, and before wrapping the socket so only header bytes are // consumed and the payload reaches the decoder untouched. let proxy_metadata = if source.proxy_protocol() { - match super::proxy_protocol::read_v2_header(&mut socket).await { - Ok(header) => { - if let Some(source_addr) = header.source { - peer_addr = source_addr; + // Bound the header read so a peer that connects but stalls before + // sending the header cannot pin this task (and its connection-limit + // permit) indefinitely. Honor shutdown, the connection tripwire, and + // the configured max connection duration (falling back to a short + // default, since a valid header arrives immediately after connect). + let header_deadline = tokio::time::sleep(Duration::from_secs( + max_connection_duration_secs.unwrap_or(PROXY_HEADER_READ_TIMEOUT_SECS), + )); + tokio::select! { + result = super::proxy_protocol::read_v2_header(&mut socket) => match result { + Ok(header) => { + if let Some(source_addr) = header.source { + peer_addr = source_addr; + } + Some(header.into_metadata()) } - Some(header.into_metadata()) - } - Err(error) => { + Err(error) => { + warn!( + message = "Failed to read PROXY protocol v2 header; closing connection.", + %error, + %peer_addr, + ); + return; + } + }, + _ = header_deadline => { warn!( - message = "Failed to read PROXY protocol v2 header; closing connection.", - %error, + message = "Timed out reading PROXY protocol v2 header; closing connection.", %peer_addr, ); return; } + _ = &mut shutdown_signal => return, + _ = &mut tripwire => return, } } else { None @@ -431,16 +456,12 @@ async fn handle_stream( } if let Some(proxy_metadata) = &proxy_metadata { - for event in &mut events { - let log = event.as_mut_log(); - log_namespace.insert_source_metadata( - source_name, - log, - Some(LegacyKey::Overwrite(path!("proxy_protocol"))), - path!("proxy_protocol"), - proxy_metadata.clone(), - ); - } + super::proxy_protocol::inject_metadata( + &mut events, + proxy_metadata, + log_namespace, + source_name, + ); } source.handle_events(&mut events, peer_addr); From 47a7eb705e27eb47ea6ba9db59e732bf1f1d3c91 Mon Sep 17 00:00:00 2001 From: rjshrjndrn Date: Tue, 21 Jul 2026 05:44:45 +0200 Subject: [PATCH 7/9] fix(socket source): reject proxy_protocol with tls Proxies send the PROXY protocol v2 header unencrypted before the TLS handshake (verified against HAProxy send-proxy-v2), so a source that terminates TLS would feed the plaintext header into the acceptor and fail the handshake. Reject the combination at build time with a clear error and document that TLS must be terminated upstream. --- src/sources/socket/mod.rs | 46 +++++++++++++++++++ src/sources/socket/tcp.rs | 4 ++ .../components/sources/generated/socket.cue | 4 ++ 3 files changed, 54 insertions(+) diff --git a/src/sources/socket/mod.rs b/src/sources/socket/mod.rs index 782593d14cf4e..dbe6097dbe0f2 100644 --- a/src/sources/socket/mod.rs +++ b/src/sources/socket/mod.rs @@ -113,6 +113,18 @@ impl SourceConfig for SocketConfig { async fn build(&self, cx: SourceContext) -> crate::Result { match self.mode.clone() { Mode::Tcp(config) => { + // The PROXY protocol header is sent unencrypted before any TLS + // handshake (as HAProxy and AWS NLB do), but this source reads + // it from the post-handshake stream. Feeding the plaintext + // header into the TLS acceptor would fail the handshake, so + // reject the combination explicitly rather than break at + // runtime. + if config.proxy_protocol() && config.tls().is_some() { + return Err( + "The `proxy_protocol` option is not supported together with `tls` on the `socket` source; terminate TLS upstream and forward plaintext to this source.".into(), + ); + } + let log_namespace = cx.log_namespace(config.log_namespace); let decoding = config.decoding().clone(); @@ -504,6 +516,40 @@ mod test { } //////// TCP TESTS //////// + #[tokio::test] + async fn tcp_proxy_protocol_with_tls_is_rejected() { + use vector_lib::tls::{TlsConfig, TlsEnableableConfig, TlsSourceConfig}; + + let (guard, addr) = next_addr(); + drop(guard); + + let mut tcp = TcpConfig::from_address(addr.into()); + tcp.proxy_protocol = true; + tcp.set_tls(Some(TlsSourceConfig { + client_metadata_key: None, + tls_config: TlsEnableableConfig { + enabled: Some(true), + options: TlsConfig::default(), + }, + })); + + let (tx, _rx) = SourceSender::new_test(); + let result = SocketConfig::from(tcp) + .build(SourceContext::new_test(tx, None)) + .await; + + match result { + Ok(_) => panic!("tls + proxy_protocol must be rejected"), + Err(err) => { + let msg = err.to_string(); + assert!( + msg.contains("proxy_protocol") && msg.contains("tls"), + "unexpected error: {msg}" + ); + } + } + } + #[tokio::test] async fn tcp_it_includes_host() { assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async { diff --git a/src/sources/socket/tcp.rs b/src/sources/socket/tcp.rs index 80cd03724dc75..6db4b4cacfd85 100644 --- a/src/sources/socket/tcp.rs +++ b/src/sources/socket/tcp.rs @@ -87,6 +87,10 @@ pub struct TcpConfig { /// peer address recorded on each event. Enable this only when a trusted /// proxy sits in front of this source, since the header is otherwise /// spoofable. + /// + /// This option cannot be combined with `tls`. Proxies send the PROXY + /// protocol header unencrypted before the TLS handshake, so TLS must be + /// terminated upstream and plaintext forwarded to this source. #[serde(default)] pub proxy_protocol: bool, diff --git a/website/cue/reference/components/sources/generated/socket.cue b/website/cue/reference/components/sources/generated/socket.cue index 153932e33648b..895a701874744 100644 --- a/website/cue/reference/components/sources/generated/socket.cue +++ b/website/cue/reference/components/sources/generated/socket.cue @@ -654,6 +654,10 @@ generated: components: sources: socket: configuration: { peer address recorded on each event. Enable this only when a trusted proxy sits in front of this source, since the header is otherwise spoofable. + + This option cannot be combined with `tls`. Proxies send the PROXY + protocol header unencrypted before the TLS handshake, so TLS must be + terminated upstream and plaintext forwarded to this source. """ relevant_when: "mode = \"tcp\"" required: false From 9326d3e9898688344aecfe406a6435658fc6b1af Mon Sep 17 00:00:00 2001 From: rjshrjndrn Date: Tue, 21 Jul 2026 05:54:09 +0200 Subject: [PATCH 8/9] fix(socket source): use fixed short header read deadline The PROXY protocol header deadline was derived from max_connection_duration_secs, so a large value (long-lived connections) reopened the slowloris window. Use a fixed 2s deadline instead: the header is the first bytes after connect and is round-trip-time bound, independent of overall connection lifetime. --- src/sources/util/net/tcp/mod.rs | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs index 79921f1b9ed45..e67e2a83f4a5d 100644 --- a/src/sources/util/net/tcp/mod.rs +++ b/src/sources/util/net/tcp/mod.rs @@ -48,11 +48,11 @@ use crate::{ pub const MAX_IN_FLIGHT_EVENTS_TARGET: usize = 100_000; -/// Fallback deadline for reading a PROXY protocol header when no explicit -/// `max_connection_duration_secs` is configured. A valid header is sent -/// immediately after connect, so a peer that has not produced one within this -/// window is treated as a stalled/misbehaving connection and dropped. -const PROXY_HEADER_READ_TIMEOUT_SECS: u64 = 10; +/// Deadline for reading a PROXY protocol header. The header is the first bytes +/// a proxy writes after connect, so it is round-trip-time bound and unrelated +/// to how long the connection may subsequently live. A peer that has not sent +/// a complete header within this window is treated as stalled and dropped. +const PROXY_HEADER_READ_TIMEOUT_SECS: u64 = 2; pub async fn try_bind_tcp_listener( addr: SocketListenAddr, @@ -315,9 +315,8 @@ async fn handle_stream( // permit) indefinitely. Honor shutdown, the connection tripwire, and // the configured max connection duration (falling back to a short // default, since a valid header arrives immediately after connect). - let header_deadline = tokio::time::sleep(Duration::from_secs( - max_connection_duration_secs.unwrap_or(PROXY_HEADER_READ_TIMEOUT_SECS), - )); + let header_deadline = + tokio::time::sleep(Duration::from_secs(PROXY_HEADER_READ_TIMEOUT_SECS)); tokio::select! { result = super::proxy_protocol::read_v2_header(&mut socket) => match result { Ok(header) => { From 254a0c8cdc97d679b3d28442a47719714398ab48 Mon Sep 17 00:00:00 2001 From: rjshrjndrn Date: Tue, 21 Jul 2026 06:06:21 +0200 Subject: [PATCH 9/9] fix(socket source): honor PROXY protocol LOCAL command The parser ignored the command nibble and extracted addresses whenever the family byte looked valid. A LOCAL header (proxy's own connection, such as a health check) could therefore overwrite host/port with address-looking bytes. Parse the command: for LOCAL, ignore the header addresses and keep the real connection endpoints; for PROXY, parse as before; reject unassigned command values. --- src/sources/util/net/proxy_protocol.rs | 42 +++++++++++++++++++++++--- 1 file changed, 38 insertions(+), 4 deletions(-) diff --git a/src/sources/util/net/proxy_protocol.rs b/src/sources/util/net/proxy_protocol.rs index 52153cd646f3c..88b8288ade4d8 100644 --- a/src/sources/util/net/proxy_protocol.rs +++ b/src/sources/util/net/proxy_protocol.rs @@ -34,6 +34,8 @@ pub enum ParseError { UnsupportedVersion(u8), /// The address family/transport byte was not understood. UnsupportedFamily(u8), + /// The command nibble was neither LOCAL (`0x0`) nor PROXY (`0x1`). + UnsupportedCommand(u8), } /// A single Type-Length-Value entry from the v2 TLV vector. @@ -119,9 +121,6 @@ pub fn parse_v2(buf: &[u8]) -> Result<(usize, ProxyHeader), ParseError> { if ver != 2 { return Err(ParseError::UnsupportedVersion(ver)); } - // Lower nibble of byte 12 is the command (LOCAL/PROXY); not needed for the - // minimal slice. - let family = buf[13] >> 4; let declared = u16::from_be_bytes([buf[14], buf[15]]) as usize; let header_end = V2_PREFIX_LEN + declared; @@ -129,6 +128,19 @@ pub fn parse_v2(buf: &[u8]) -> Result<(usize, ProxyHeader), ParseError> { return Err(ParseError::Truncated); } + // Lower nibble of byte 12 is the command. + match buf[12] & 0x0f { + // LOCAL: the connection is the proxy's own (for example a health + // check), not proxied on behalf of a client. Per the spec the + // receiver must ignore the header addresses and use the real + // connection endpoints, so return an empty header while still + // consuming the declared bytes so the payload is left intact. + 0x0 => return Ok((header_end, ProxyHeader::default())), + // PROXY: addresses belong to the real client; parse them below. + 0x1 => {} + other => return Err(ParseError::UnsupportedCommand(other)), + } + let addr_bytes = &buf[V2_PREFIX_LEN..header_end]; let (source, destination, addr_len) = match family { // AF_INET @@ -273,6 +285,10 @@ mod tests { /// Build a v2 IPv4 header with the given TLV bytes appended. fn v4_header(tlvs: &[u8]) -> Vec { + v4_header_with_ver_cmd(0x21, tlvs) // ver 2, cmd PROXY + } + + fn v4_header_with_ver_cmd(ver_cmd: u8, tlvs: &[u8]) -> Vec { let addr: [u8; 12] = [ 1, 2, 3, 4, // src 1.2.3.4 10, 0, 0, 1, // dst 10.0.0.1 @@ -282,7 +298,7 @@ mod tests { let len = (addr.len() + tlvs.len()) as u16; let mut buf = Vec::new(); buf.extend_from_slice(&V2_SIGNATURE); - buf.push(0x21); // ver 2, cmd PROXY + buf.push(ver_cmd); buf.push(0x11); // AF_INET, STREAM buf.extend_from_slice(&len.to_be_bytes()); buf.extend_from_slice(&addr); @@ -360,6 +376,24 @@ mod tests { assert_eq!(header.tlv(0xE0), Some(&b"acme"[..])); } + #[test] + fn local_command_ignores_header_addresses() { + // ver 2, cmd LOCAL (0x20), but with real-looking IPv4 address bytes. + let buf = v4_header_with_ver_cmd(0x20, &[]); + let (consumed, header) = parse_v2(&buf).expect("local header parses"); + assert_eq!(consumed, buf.len(), "header bytes still consumed"); + assert_eq!(header.source, None, "LOCAL must not expose addresses"); + assert_eq!(header.destination, None); + assert!(header.tlvs.is_empty()); + } + + #[test] + fn unsupported_command_is_rejected() { + // ver 2, cmd 0x2 (unassigned). + let buf = v4_header_with_ver_cmd(0x22, &[]); + assert_eq!(parse_v2(&buf), Err(ParseError::UnsupportedCommand(2))); + } + #[test] fn truncated_body_is_reported() { let mut buf = v4_header(&[]);