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

@@ -113,6 +113,18 @@ impl SourceConfig for SocketConfig {
     async fn build(&self, cx: SourceContext) -> crate::Result<super::Source> {
         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::<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_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 {

src/sources/socket/tcp.rs

@@ -79,6 +79,21 @@ 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.
+    ///
+    /// 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<FramingConfig>,
 
@@ -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()
     }

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;