Add proxy.protocol.accepted.ip.range config to guard PROXY protocol data by peer IP

During v3/v4 migration on GCP and Azure, both load balancer (v3, no PROXY
protocol) and Envoy (v4, with PROXY protocol) connections coexist. Without
this guard, an attacker can send a fake PROXY header through the v3 LB to
spoof their IP and bypass IP filtering.

This adds a new config `proxy.protocol.accepted.ip.range` (CIDR notation)
that tells ProxyCustomizer to only trust PROXY protocol data from peers
whose raw TCP IP falls within the range. Connections from outside the range
get wrapped with a RawPeerRequest that overrides getRemoteAddr() back to
the raw TCP peer IP using Jetty 12's ConnectionMetaData.Wrapper, effectively
undoing the ProxyEndPoint's address override.

Companion change to ce-kafka PR #28385.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: devhritwik

core/src/main/java/io/confluent/rest/ApplicationServer.java

@@ -20,6 +20,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
+import io.confluent.rest.customizer.CidrRange;
 import io.confluent.rest.customizer.ProxyCustomizer;
 import io.confluent.rest.errorhandlers.StackTraceErrorHandler;
 import java.lang.management.ManagementFactory;
@@ -287,7 +288,8 @@ private void configureConnectors() {
           connectorConfig.getInt(RestConfig.MAX_RESPONSE_HEADER_SIZE_CONFIG));
 
       if (proxyProtocolEnabled) {
-        httpConfiguration.addCustomizer(new ProxyCustomizer());
+        httpConfiguration.addCustomizer(
+            new ProxyCustomizer(parseAcceptedIpRange(connectorConfig)));
       }
 
       // Use original IP in forwarded requests
@@ -447,6 +449,12 @@ private ConnectionFactory[] getConnectionFactories(HttpConfiguration httpConfigu
     return connectionFactories.toArray(new ConnectionFactory[0]);
   }
 
+  private static CidrRange parseAcceptedIpRange(RestConfig connectorConfig) {
+    String value = connectorConfig.getString(
+        RestConfig.PROXY_PROTOCOL_ACCEPTED_IP_RANGE_CONFIG).trim();
+    return value.isEmpty() ? null : CidrRange.parse(value);
+  }
+
   private void configureConnectionLimits() {
     int serverConnectionLimit = serverConfig.getServerConnectionLimit();
     if (serverConnectionLimit > 0) {

core/src/main/java/io/confluent/rest/RestConfig.java

@@ -593,6 +593,18 @@ public class RestConfig extends AbstractConfig {
           + "Default is false.";
   protected static final boolean PROXY_PROTOCOL_ENABLED_DEFAULT = false;
 
+  public static final String PROXY_PROTOCOL_ACCEPTED_IP_RANGE_CONFIG =
+      "proxy.protocol.accepted.ip.range";
+  protected static final String PROXY_PROTOCOL_ACCEPTED_IP_RANGE_DOC =
+      "If set, the server will only use PROXY protocol header information from connections "
+          + "whose peer IP address falls within the specified CIDR range (e.g., '10.240.0.0/16'). "
+          + "Connections from outside this range will have their PROXY protocol data ignored, "
+          + "and the raw peer IP will be used instead. This prevents IP spoofing during migrations "
+          + "where some connections come through a proxy (e.g., Envoy) and others do not. "
+          + "If empty (default), PROXY protocol data is used unconditionally when "
+          + "proxy.protocol.enabled is true.";
+  protected static final String PROXY_PROTOCOL_ACCEPTED_IP_RANGE_DEFAULT = "";
+
   public static final String SUPPRESS_STACK_TRACE_IN_RESPONSE = "suppress.stack.trace.response";
 
   protected static final String SUPPRESS_STACK_TRACE_IN_RESPONSE_DOC =
@@ -1249,6 +1261,12 @@ private static ConfigDef incompleteBaseConfigDef() {
             PROXY_PROTOCOL_ENABLED_DEFAULT,
             Importance.LOW,
             PROXY_PROTOCOL_ENABLED_DOC
+        ).define(
+            PROXY_PROTOCOL_ACCEPTED_IP_RANGE_CONFIG,
+            Type.STRING,
+            PROXY_PROTOCOL_ACCEPTED_IP_RANGE_DEFAULT,
+            Importance.LOW,
+            PROXY_PROTOCOL_ACCEPTED_IP_RANGE_DOC
         ).define(
             NOSNIFF_PROTECTION_ENABLED,
             Type.BOOLEAN,

core/src/main/java/io/confluent/rest/customizer/CidrRange.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2025 Confluent Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.confluent.rest.customizer;
+
+import org.apache.kafka.common.config.ConfigException;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+/**
+ * Represents a CIDR range (e.g., "10.240.0.0/16") and provides a method to check
+ * whether a given IP address falls within the range.
+ */
+public class CidrRange {
+
+  private final byte[] networkBytes;
+  private final int prefixLength;
+
+  private CidrRange(byte[] networkBytes, int prefixLength) {
+    this.networkBytes = networkBytes;
+    this.prefixLength = prefixLength;
+  }
+
+  /**
+   * Parses a CIDR notation string (e.g., "10.240.0.0/16") into a {@link CidrRange}.
+   *
+   * @param cidr the CIDR string to parse
+   * @return a {@link CidrRange} instance
+   * @throws ConfigException if the CIDR string is malformed
+   */
+  public static CidrRange parse(String cidr) {
+    String[] parts = cidr.split("/");
+    if (parts.length != 2) {
+      throw new ConfigException("Invalid CIDR notation: " + cidr);
+    }
+
+    InetAddress network;
+    try {
+      network = InetAddress.getByName(parts[0]);
+    } catch (UnknownHostException e) {
+      throw new ConfigException("Invalid network address in CIDR: " + parts[0]);
+    }
+
+    int prefixLength;
+    try {
+      prefixLength = Integer.parseInt(parts[1]);
+    } catch (NumberFormatException e) {
+      throw new ConfigException("Invalid prefix length in CIDR: " + parts[1]);
+    }
+
+    byte[] networkBytes = network.getAddress();
+    int maxPrefix = networkBytes.length * 8;
+    if (prefixLength < 0 || prefixLength > maxPrefix) {
+      throw new ConfigException(
+          "Prefix length " + prefixLength + " is out of range for "
+              + (networkBytes.length == 4 ? "IPv4" : "IPv6")
+              + " address (0-" + maxPrefix + ")");
+    }
+
+    return new CidrRange(networkBytes, prefixLength);
+  }
+
+  /**
+   * Checks whether the given address is within this CIDR range.
+   *
+   * @param address the address to check
+   * @return true if the address is within this range, false otherwise
+   */
+  public boolean contains(InetAddress address) {
+    byte[] addrBytes = address.getAddress();
+
+    // IPv4 vs IPv6 mismatch
+    if (addrBytes.length != networkBytes.length) {
+      return false;
+    }
+
+    // Compare full bytes
+    int fullBytes = prefixLength / 8;
+    for (int i = 0; i < fullBytes; i++) {
+      if (addrBytes[i] != networkBytes[i]) {
+        return false;
+      }
+    }
+
+    // Compare remaining bits
+    int remainingBits = prefixLength % 8;
+    if (remainingBits > 0) {
+      int mask = 0xFF << (8 - remainingBits);
+      if ((addrBytes[fullBytes] & mask) != (networkBytes[fullBytes] & mask)) {
+        return false;
+      }
+    }
+
+    return true;
+  }
+}

core/src/main/java/io/confluent/rest/customizer/ProxyCustomizer.java

@@ -19,20 +19,32 @@
 import org.eclipse.jetty.http.HttpFields;
 import org.eclipse.jetty.io.EndPoint;
 import org.eclipse.jetty.io.ssl.SslConnection;
+import org.eclipse.jetty.server.ConnectionMetaData;
 import org.eclipse.jetty.server.HttpConfiguration;
 import org.eclipse.jetty.server.ProxyConnectionFactory;
 import org.eclipse.jetty.server.Request;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.HashSet;
 import java.util.Set;
 
 /**
- * Similar to {@link org.eclipse.jetty.server.ProxyCustomizer} but allowing access to tlvs as well
+ * Similar to {@link org.eclipse.jetty.server.ProxyCustomizer} but allowing access to tlvs as well.
+ *
+ * <p>When an {@code acceptedIpRange} is configured, this customizer checks whether the
+ * underlying (raw) peer IP falls within the CIDR range. If it does NOT, the PROXY protocol
+ * data is ignored and the request's remote address is overridden to the raw TCP peer IP.
+ * This prevents IP spoofing during v3/v4 migrations where some connections arrive through
+ * a proxy (e.g., Envoy) and others come directly through a load balancer.</p>
  */
 public class ProxyCustomizer implements HttpConfiguration.Customizer {
 
+  private static final Logger log = LoggerFactory.getLogger(ProxyCustomizer.class);
+
   // The remote address attribute name.
   public static final String REMOTE_ADDRESS_ATTRIBUTE_NAME =
       "io.confluent.rest.proxy.remote.address";
@@ -54,6 +66,16 @@ public class ProxyCustomizer implements HttpConfiguration.Customizer {
   public static final String TLV_PROVIDER_ATTRIBUTE_NAME
       = "io.confluent.rest.proxy.tlv.provider";
 
+  private final CidrRange acceptedIpRange;
+
+  public ProxyCustomizer() {
+    this(null);
+  }
+
+  public ProxyCustomizer(CidrRange acceptedIpRange) {
+    this.acceptedIpRange = acceptedIpRange;
+  }
+
   @Override
   public Request customize(Request request, HttpFields.Mutable mutable) {
     EndPoint endPoint = request.getConnectionMetaData().getConnection().getEndPoint();
@@ -65,6 +87,23 @@ public Request customize(Request request, HttpFields.Mutable mutable) {
 
     if (endPoint instanceof ProxyConnectionFactory.ProxyEndPoint proxyEndPoint) {
       EndPoint underlyingEndpoint = proxyEndPoint.unwrap();
+
+      // Check peer IP against accepted range before using PROXY data
+      if (acceptedIpRange != null) {
+        SocketAddress rawRemote = underlyingEndpoint.getRemoteSocketAddress();
+        if (rawRemote instanceof InetSocketAddress inetRemote) {
+          InetAddress peerAddress = inetRemote.getAddress();
+          if (!acceptedIpRange.contains(peerAddress)) {
+            log.debug(
+                "Peer IP {} is not in accepted range, ignoring PROXY protocol data",
+                peerAddress.getHostAddress());
+            // Override the connection metadata to use raw TCP addresses,
+            // undoing the ProxyEndPoint's effect on getRemoteAddr()
+            return new RawPeerRequest(request, underlyingEndpoint);
+          }
+        }
+      }
+
       request = new ProxyRequest(request,
           underlyingEndpoint.getLocalSocketAddress(),
           underlyingEndpoint.getRemoteSocketAddress(),
@@ -73,6 +112,37 @@ public Request customize(Request request, HttpFields.Mutable mutable) {
     return request;
   }
 
+  /**
+   * Request wrapper that overrides the connection metadata to use raw TCP peer
+   * addresses instead of the PROXY-parsed addresses. This effectively "undoes"
+   * the {@link ProxyConnectionFactory.ProxyEndPoint}'s address override when
+   * the peer is not in the accepted IP range.
+   */
+  private static class RawPeerRequest extends Request.Wrapper {
+    private final ConnectionMetaData rawMetaData;
+
+    private RawPeerRequest(Request request, EndPoint rawEndpoint) {
+      super(request);
+      this.rawMetaData = new ConnectionMetaData.Wrapper(
+          request.getConnectionMetaData()) {
+        @Override
+        public SocketAddress getRemoteSocketAddress() {
+          return rawEndpoint.getRemoteSocketAddress();
+        }
+
+        @Override
+        public SocketAddress getLocalSocketAddress() {
+          return rawEndpoint.getLocalSocketAddress();
+        }
+      };
+    }
+
+    @Override
+    public ConnectionMetaData getConnectionMetaData() {
+      return rawMetaData;
+    }
+  }
+
   private static class ProxyRequest extends Request.Wrapper {
     private final String remoteAddress;
     private final String localAddress;