feat(socket source): add opt-in PROXY protocol v2 support

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

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

src/sources/socket/mod.rs

@@ -462,6 +462,37 @@ mod test {
         crate::test_util::test_generate_config::<SocketConfig>();
     }
 
+    #[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() {

src/sources/socket/tcp.rs

@@ -79,6 +79,17 @@ pub struct TcpConfig {
     #[configurable(metadata(docs::type_unit = "connections"))]
     pub connection_limit: Option<u32>,
 
+    /// 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<FramingConfig>,
 
@@ -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()
     }

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;