This is an automated email from the ASF dual-hosted git repository.
kevdoran pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 7792cd9de14 NIFI-16005 Adjust proxy header validation for requests
forwarded to c… (#11317)
7792cd9de14 is described below
commit 7792cd9de14421ce82d734bdff0c30fa8526a91f
Author: Bryan Bende <[email protected]>
AuthorDate: Mon Jun 8 22:21:09 2026 -0400
NIFI-16005 Adjust proxy header validation for requests forwarded to c…
(#11317)
Signed-off-by: Kevin Doran <[email protected]>
---
.../http/replication/ReplicationHeaderUtils.java | 13 ++++
.../replication/ThreadPoolRequestReplicator.java | 12 ++++
.../replication/TestReplicationHeaderUtils.java | 43 ++++++++++++
.../TestThreadPoolRequestReplicator.java | 79 ++++++++++++++++++++++
.../coordination/http/ReplicationHeader.java | 5 +-
.../connector/ProxyHeaderValidatorCustomizer.java | 6 +-
.../web/server/StandardServerProviderTest.java | 11 +++
7 files changed, 166 insertions(+), 3 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ReplicationHeaderUtils.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ReplicationHeaderUtils.java
index 947393a54fe..c4165eb5ceb 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ReplicationHeaderUtils.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ReplicationHeaderUtils.java
@@ -109,6 +109,19 @@ public final class ReplicationHeaderUtils {
}
}
+ /**
+ * Removes the trust-bearing {@link ReplicationHeader} marker headers from
the map (case-insensitive) so that an
+ * inbound client request cannot spoof the markers that cause a receiving
node to bypass proxy host validation. The
+ * caller is expected to set the appropriate marker explicitly after
invoking this method. Other replication protocol
+ * headers, such as {@code replication-target-id}, are intentionally left
intact because they are set by the framework
+ * and must transit to the receiving node.
+ */
+ public static void stripReplicationMarkerHeaders(final Map<String, String>
headers) {
+ for (final ReplicationHeader rh : ReplicationHeader.values()) {
+ removeHeader(headers, rh.getHeader());
+ }
+ }
+
/**
* Removes hop-by-hop / transport-framing headers that should not be
forwarded in a
* replicated request (case-insensitive).
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
index de58db4f201..9d97d029cc2 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
@@ -256,6 +256,10 @@ public class ThreadPoolRequestReplicator implements
RequestReplicator, Closeable
final boolean indicateReplicated,
final boolean performVerification) {
final Map<String, String> updatedHeaders = new HashMap<>(headers);
+ // Strip any inbound replication marker headers so a client cannot
spoof them; the framework sets the
+ // appropriate marker explicitly below
+ ReplicationHeaderUtils.stripReplicationMarkerHeaders(updatedHeaders);
+
updatedHeaders.put(RequestReplicationHeader.CLUSTER_ID_GENERATION_SEED.getHeader(),
ComponentIdGenerator.generateId().toString());
if (indicateReplicated) {
updatedHeaders.put(ReplicationHeader.REQUEST_REPLICATED.getHeader(),
Boolean.TRUE.toString());
@@ -308,6 +312,14 @@ public class ThreadPoolRequestReplicator implements
RequestReplicator, Closeable
final URI uri, final Object entity, final Map<String, String>
headers) {
final Map<String, String> updatedHeaders = new HashMap<>(headers);
+ // Strip any inbound replication marker headers so a client cannot
spoof them; the forwarded marker is set
+ // explicitly below
+ ReplicationHeaderUtils.stripReplicationMarkerHeaders(updatedHeaders);
+
+ // Indicate to the Cluster Coordinator that this request was forwarded
by a node over mutual TLS, so the
+ // Coordinator can trust the proxy host headers without re-validating
them against its own allowed proxy hosts
+
updatedHeaders.put(ReplicationHeader.REQUEST_FORWARDED_TO_COORDINATOR.getHeader(),
Boolean.TRUE.toString());
+
// include the proxied entities header
updateRequestHeaders(updatedHeaders, user);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestReplicationHeaderUtils.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestReplicationHeaderUtils.java
index 09a24745ca4..1f9727de74f 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestReplicationHeaderUtils.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestReplicationHeaderUtils.java
@@ -18,6 +18,7 @@ package org.apache.nifi.cluster.coordination.http.replication;
import org.apache.nifi.authorization.user.NiFiUser;
import org.apache.nifi.authorization.user.StandardNiFiUser;
+import org.apache.nifi.cluster.coordination.http.ReplicationHeader;
import org.apache.nifi.web.security.ProxiedEntitiesUtils;
import org.junit.jupiter.api.Test;
@@ -88,6 +89,9 @@ class TestReplicationHeaderUtils {
for (final RequestReplicationHeader rh :
RequestReplicationHeader.values()) {
headers.put(rh.getHeader(), SPOOFED_VALUE);
}
+ for (final ReplicationHeader rh : ReplicationHeader.values()) {
+ headers.put(rh.getHeader(), SPOOFED_VALUE);
+ }
headers.put(CUSTOM_HEADER, SHOULD_SURVIVE_VALUE);
ReplicationHeaderUtils.stripRequestReplicationHeaders(headers);
@@ -95,9 +99,48 @@ class TestReplicationHeaderUtils {
for (final RequestReplicationHeader rh :
RequestReplicationHeader.values()) {
assertNull(headers.get(rh.getHeader()), "Replication header should
have been stripped: " + rh.getHeader());
}
+ for (final ReplicationHeader rh : ReplicationHeader.values()) {
+ assertNull(headers.get(rh.getHeader()), "Replication header should
have been stripped: " + rh.getHeader());
+ }
assertEquals(SHOULD_SURVIVE_VALUE, headers.get(CUSTOM_HEADER));
}
+ @Test
+ void testStripReplicationMarkerHeadersRemovesOnlyMarkers() {
+ final Map<String, String> headers = new HashMap<>();
+ for (final ReplicationHeader rh : ReplicationHeader.values()) {
+ headers.put(rh.getHeader(), SPOOFED_VALUE);
+ }
+ for (final RequestReplicationHeader rh :
RequestReplicationHeader.values()) {
+ headers.put(rh.getHeader(), SHOULD_SURVIVE_VALUE);
+ }
+ headers.put(CUSTOM_HEADER, SHOULD_SURVIVE_VALUE);
+
+ ReplicationHeaderUtils.stripReplicationMarkerHeaders(headers);
+
+ for (final ReplicationHeader rh : ReplicationHeader.values()) {
+ assertNull(headers.get(rh.getHeader()), "Replication marker header
should have been stripped: " + rh.getHeader());
+ }
+ for (final RequestReplicationHeader rh :
RequestReplicationHeader.values()) {
+ assertEquals(SHOULD_SURVIVE_VALUE, headers.get(rh.getHeader()),
"Request replication header should have been preserved: " + rh.getHeader());
+ }
+ assertEquals(SHOULD_SURVIVE_VALUE, headers.get(CUSTOM_HEADER));
+ }
+
+ @Test
+ void testStripReplicationMarkerHeadersCaseInsensitive() {
+ final Map<String, String> headers = new HashMap<>();
+ headers.put("Request-Replicated", "true");
+ headers.put("Request-Forwarded-To-Coordinator", "true");
+ headers.put("Replication-Target-Id", SHOULD_SURVIVE_VALUE);
+
+ ReplicationHeaderUtils.stripReplicationMarkerHeaders(headers);
+
+ assertFalse(headers.containsKey("Request-Replicated"));
+ assertFalse(headers.containsKey("Request-Forwarded-To-Coordinator"));
+ assertEquals(SHOULD_SURVIVE_VALUE,
headers.get("Replication-Target-Id"));
+ }
+
@Test
void testStripRequestReplicationHeadersCaseInsensitive() {
final Map<String, String> headers = new HashMap<>();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestThreadPoolRequestReplicator.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestThreadPoolRequestReplicator.java
index 3a7eccfb9fb..5ecfb06dea9 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestThreadPoolRequestReplicator.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestThreadPoolRequestReplicator.java
@@ -26,6 +26,7 @@ import org.apache.nifi.authorization.user.NiFiUserUtils;
import org.apache.nifi.authorization.user.StandardNiFiUser;
import org.apache.nifi.authorization.user.StandardNiFiUser.Builder;
import org.apache.nifi.cluster.coordination.ClusterCoordinator;
+import org.apache.nifi.cluster.coordination.http.ReplicationHeader;
import
org.apache.nifi.cluster.coordination.http.replication.util.MockReplicationClient;
import org.apache.nifi.cluster.coordination.node.NodeConnectionState;
import org.apache.nifi.cluster.coordination.node.NodeConnectionStatus;
@@ -56,6 +57,7 @@ import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -379,6 +381,83 @@ public class TestThreadPoolRequestReplicator {
return coordinator;
}
+ private ThreadPoolRequestReplicator createHeaderCapturingReplicator(final
Map<String, String> capturedHeaders) {
+ final ClusterCoordinator coordinator = createClusterCoordinator();
+ final NiFiProperties props =
NiFiProperties.createBasicNiFiProperties((String) null);
+ final MockReplicationClient client = new MockReplicationClient();
+ final RequestCompletionCallback callback = (uri, method, responses) ->
{ };
+
+ return new ThreadPoolRequestReplicator(5, 100, client, coordinator,
callback, EventReporter.NO_OP, props) {
+ @Override
+ protected NodeResponse replicateRequest(final PreparedRequest
request, final NodeIdentifier nodeId, final URI uri, final String requestId,
+ final StandardAsyncClusterResponse response) {
+ capturedHeaders.putAll(request.getHeaders());
+
+ final Response clientResponse = mock(Response.class);
+
when(clientResponse.getStatus()).thenReturn(Status.OK.getStatusCode());
+ return new NodeResponse(nodeId, request.getMethod(), uri,
clientResponse, -1L, requestId);
+ }
+ };
+ }
+
+ @Test
+ @Timeout(value = 15)
+ public void testForwardToCoordinatorSetsForwardedHeader() throws Exception
{
+ final NodeIdentifier coordinatorNodeId = new NodeIdentifier("1",
"localhost", 8100, "localhost", 8101, "localhost", 8102, 8103, false);
+
+ final Map<String, String> capturedHeaders = new ConcurrentHashMap<>();
+ final ThreadPoolRequestReplicator replicator =
createHeaderCapturingReplicator(capturedHeaders);
+
+ try {
+ final Authentication authentication = new
NiFiAuthenticationToken(new NiFiUserDetails(StandardNiFiUser.ANONYMOUS));
+
SecurityContextHolder.getContext().setAuthentication(authentication);
+
+ final URI uri = new
URI("http://localhost:8080/nifi-api/controller/registry-types";);
+ final Entity entity = new ProcessorEntity();
+
+ final AsyncClusterResponse response =
replicator.forwardToCoordinator(coordinatorNodeId, HttpMethod.GET, uri, entity,
new HashMap<>());
+ response.awaitMergedResponse(3, TimeUnit.SECONDS);
+
+ assertEquals(Boolean.TRUE.toString(),
capturedHeaders.get(ReplicationHeader.REQUEST_FORWARDED_TO_COORDINATOR.getHeader()));
+
assertNull(capturedHeaders.get(ReplicationHeader.REQUEST_REPLICATED.getHeader()));
+ } finally {
+ replicator.shutdown();
+ }
+ }
+
+ @Test
+ @Timeout(value = 15)
+ public void
testForwardToCoordinatorStripsSpoofedMarkersAndPreservesTargetId() throws
Exception {
+ final NodeIdentifier coordinatorNodeId = new NodeIdentifier("1",
"localhost", 8100, "localhost", 8101, "localhost", 8102, 8103, false);
+
+ final Map<String, String> capturedHeaders = new ConcurrentHashMap<>();
+ final ThreadPoolRequestReplicator replicator =
createHeaderCapturingReplicator(capturedHeaders);
+
+ try {
+ final Authentication authentication = new
NiFiAuthenticationToken(new NiFiUserDetails(StandardNiFiUser.ANONYMOUS));
+
SecurityContextHolder.getContext().setAuthentication(authentication);
+
+ final URI uri = new
URI("http://localhost:8080/nifi-api/controller/registry-types";);
+ final Entity entity = new ProcessorEntity();
+
+ final Map<String, String> inboundHeaders = new HashMap<>();
+
inboundHeaders.put(ReplicationHeader.REQUEST_REPLICATED.getHeader(), "spoofed");
+
inboundHeaders.put(ReplicationHeader.REQUEST_FORWARDED_TO_COORDINATOR.getHeader(),
"spoofed");
+
inboundHeaders.put(RequestReplicationHeader.REPLICATION_TARGET_ID.getHeader(),
"node-uuid");
+
+ final AsyncClusterResponse response =
replicator.forwardToCoordinator(coordinatorNodeId, HttpMethod.GET, uri, entity,
inboundHeaders);
+ response.awaitMergedResponse(3, TimeUnit.SECONDS);
+
+ // The spoofed replicated marker is stripped, the forwarded marker
is set by the framework, and the
+ // framework-supplied replication target id is preserved
+
assertNull(capturedHeaders.get(ReplicationHeader.REQUEST_REPLICATED.getHeader()));
+ assertEquals(Boolean.TRUE.toString(),
capturedHeaders.get(ReplicationHeader.REQUEST_FORWARDED_TO_COORDINATOR.getHeader()));
+ assertEquals("node-uuid",
capturedHeaders.get(RequestReplicationHeader.REPLICATION_TARGET_ID.getHeader()));
+ } finally {
+ replicator.shutdown();
+ }
+ }
+
@Test
@Timeout(value = 15)
public void testOneNodeRejectsTwoPhaseCommit() {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/cluster/coordination/http/ReplicationHeader.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/cluster/coordination/http/ReplicationHeader.java
index d7fddda3357..5066b811f81 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/cluster/coordination/http/ReplicationHeader.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/cluster/coordination/http/ReplicationHeader.java
@@ -21,7 +21,10 @@ package org.apache.nifi.cluster.coordination.http;
*/
public enum ReplicationHeader {
/** Boolean indicator that the Cluster Coordinator is initiating the
replicated request to other nodes */
- REQUEST_REPLICATED("request-replicated");
+ REQUEST_REPLICATED("request-replicated"),
+
+ /** Boolean indicator that a node is forwarding a request to the Cluster
Coordinator for subsequent replication */
+ REQUEST_FORWARDED_TO_COORDINATOR("request-forwarded-to-coordinator");
private final String header;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/main/java/org/apache/nifi/web/server/connector/ProxyHeaderValidatorCustomizer.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/main/java/org/apache/nifi/web/server/connector/ProxyHeaderValidatorCustomizer.java
index 80e90f08ed3..34c149c0877 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/main/java/org/apache/nifi/web/server/connector/ProxyHeaderValidatorCustomizer.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/main/java/org/apache/nifi/web/server/connector/ProxyHeaderValidatorCustomizer.java
@@ -76,10 +76,12 @@ public class ProxyHeaderValidatorCustomizer implements
HttpConfiguration.Customi
if (peerCertificate == null) {
processProxyHostHeaders(request);
} else {
- // Requests authenticated with Client Certificates but not
indicated as replicated require header validation
+ // Requests authenticated with Client Certificates that are
replicated to nodes or forwarded to the Cluster
+ // Coordinator originate from trusted nodes over mutual TLS, so
their proxy host headers are not re-validated
final HttpFields requestHeaders = request.getHeaders();
final String requestReplicated =
requestHeaders.get(ReplicationHeader.REQUEST_REPLICATED.getHeader());
- if (requestReplicated == null) {
+ final String requestForwardedToCoordinator =
requestHeaders.get(ReplicationHeader.REQUEST_FORWARDED_TO_COORDINATOR.getHeader());
+ if (requestReplicated == null && requestForwardedToCoordinator ==
null) {
processProxyHostHeaders(request);
}
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/StandardServerProviderTest.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/StandardServerProviderTest.java
index 57cd6603455..e5740b3bed8 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/StandardServerProviderTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/StandardServerProviderTest.java
@@ -255,6 +255,7 @@ class StandardServerProviderTest {
assertMisdirectedRequestsCompleted(httpClient, localhostUri);
assertReplicatedRequestCompleted(httpClient, localhostUri,
HttpStatus.MISDIRECTED_REQUEST_421);
+ assertForwardedToCoordinatorRequestCompleted(httpClient,
localhostUri, HttpStatus.MISDIRECTED_REQUEST_421);
}
}
@@ -272,6 +273,7 @@ class StandardServerProviderTest {
assertMisdirectedRequestsCompleted(httpClient, localhostUri);
assertReplicatedRequestCompleted(httpClient, localhostUri,
HttpStatus.MOVED_TEMPORARILY_302);
+ assertForwardedToCoordinatorRequestCompleted(httpClient,
localhostUri, HttpStatus.MOVED_TEMPORARILY_302);
}
}
@@ -284,6 +286,15 @@ class StandardServerProviderTest {
assertResponseStatusCode(httpClient,
proxyHostRequestReplicatedRequest, statusCodeExpected);
}
+ void assertForwardedToCoordinatorRequestCompleted(final HttpClient
httpClient, final URI localhostUri, final int statusCodeExpected) throws
IOException, InterruptedException {
+ final HttpRequest proxyHostForwardedToCoordinatorRequest =
HttpRequest.newBuilder(localhostUri)
+ .version(HttpClient.Version.HTTP_1_1)
+ .header(ProxyHeader.PROXY_HOST.getHeader(),
PUBLIC_UNKNOWN_HOST)
+
.header(ReplicationHeader.REQUEST_FORWARDED_TO_COORDINATOR.getHeader(),
Boolean.TRUE.toString())
+ .build();
+ assertResponseStatusCode(httpClient,
proxyHostForwardedToCoordinatorRequest, statusCodeExpected);
+ }
+
void assertFrontendRedirectRequestsCompleted(final HttpClient httpClient,
final URI localhostUri) throws IOException, InterruptedException {
final HttpRequest localhostRequest =
HttpRequest.newBuilder(localhostUri)
.version(HttpClient.Version.HTTP_2)
