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/src/sources/socket/mod.rs b/src/sources/socket/mod.rs index 66149775d5aa8..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(); @@ -243,6 +255,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); @@ -462,7 +484,72 @@ 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_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 5e1873e6f9cde..6db4b4cacfd85 100644 --- a/src/sources/socket/tcp.rs +++ b/src/sources/socket/tcp.rs @@ -79,6 +79,21 @@ 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. + /// + /// 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, + #[configurable(derived)] pub(super) framing: Option, @@ -115,10 +130,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 +233,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/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..88b8288ade4d8 --- /dev/null +++ b/src/sources/util/net/proxy_protocol.rs @@ -0,0 +1,531 @@ +//! 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}; + +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. +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), + /// The command nibble was neither LOCAL (`0x0`) nor PROXY (`0x1`). + UnsupportedCommand(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()) + } + + /// 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. +/// +/// 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)); + } + 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); + } + + // 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 + 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, + }, + )) +} + +/// 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. +/// +/// 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 { + 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 { + 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 + 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(ver_cmd); + 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 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(&[]); + 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); + } + + /// 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. + /// 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, + ] + } + + #[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(); + 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"); + } + + #[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()); + assert!(read_v2_header(&mut cursor).await.is_err()); + } + + #[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"); + } +} diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs index aebce3dc9b6ee..e67e2a83f4a5d 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; +/// 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, mut listenfd: ListenFd, @@ -113,6 +119,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 +270,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 +305,49 @@ 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. + let proxy_metadata = if source.proxy_protocol() { + // 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(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()) + } + Err(error) => { + warn!( + message = "Failed to read PROXY protocol v2 header; closing connection.", + %error, + %peer_addr, + ); + return; + } + }, + _ = header_deadline => { + warn!( + message = "Timed out reading PROXY protocol v2 header; closing connection.", + %peer_addr, + ); + return; + } + _ = &mut shutdown_signal => return, + _ = &mut tripwire => return, + } + } else { + None + }; + let socket = socket.after_read(move |byte_size| { emit!(TcpBytesReceived { byte_size, @@ -398,6 +454,15 @@ async fn handle_stream( } } + if let Some(proxy_metadata) = &proxy_metadata { + super::proxy_protocol::inject_metadata( + &mut events, + proxy_metadata, + log_namespace, + source_name, + ); + } + source.handle_events(&mut events, peer_addr); match out.send_batch(events).await { Ok(_) => { diff --git a/website/cue/reference/components/sources/generated/socket.cue b/website/cue/reference/components/sources/generated/socket.cue index b0b399c3ae2d0..895a701874744 100644 --- a/website/cue/reference/components/sources/generated/socket.cue +++ b/website/cue/reference/components/sources/generated/socket.cue @@ -644,6 +644,25 @@ 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. + + 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 + 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\""