This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/main by this push:
new a859a86a0c Fix #1806: apply camel.remove.headers.pattern before the
sink route stages (#1815)
a859a86a0c is described below
commit a859a86a0ced0e735d280271a2cbf2be1156dd29
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Aug 24 14:56:58 2026 +0200
Fix #1806: apply camel.remove.headers.pattern before the sink route stages
(#1815)
The route was assembled with ckcRemoveHeader appended last for both
directions:
from(from) -> [ckcMarshal] -> [ckcUnMarshal] -> [ckcAggregator]
-> [ckcIdempotent] -> ckcRemoveHeader -> toD(to)
That is correct for the source direction, where headers arrive from the
Camel
consumer and are mapped onto the produced record after the route has run.
It is
wrong for the sink direction, where CamelSinkTask.put maps
CamelHeader.-prefixed
record headers onto the exchange before it enters the route, so every stage
runs
before the removal does. With camel.remove.headers.pattern set, the
idempotency
expression, a configured AggregationStrategy and the data formats all still
saw
the headers the operator asked to strip.
Give the builder the direction through withRemoveHeadersFirst and set it in
CamelSinkTask, so the stage runs first on the sink path and stays last on
the
source path. The builder default keeps the previous placement, so a caller
driving CamelKafkaConnectMain.Builder directly is unaffected.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../apache/camel/kafkaconnector/CamelSinkTask.java | 1 +
.../utils/CamelKafkaConnectMain.java | 25 +++++-
.../utils/RemoveHeadersOrderTest.java | 95 ++++++++++++++++++++++
3 files changed, 120 insertions(+), 1 deletion(-)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
index b29f429267..b36a738854 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
@@ -135,6 +135,7 @@ public class CamelSinkTask extends SinkTask {
}
cms = CamelKafkaConnectMain.builder(LOCAL_URL, sinkKamelet)
+ .withRemoveHeadersFirst(true)
.withProperties(actualProps)
.withUnmarshallDataFormat(unmarshaller)
.withMarshallDataFormat(marshaller)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
b/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
index f1a4cb2bdc..9d060769f0 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
@@ -117,6 +117,7 @@ public class CamelKafkaConnectMain extends SimpleMain {
private int idempotentRepositoryKafkaMaxCacheSize;
private int idempotentRepositoryKafkaPollDuration;
private String headersExcludePattern;
+ private boolean removeHeadersFirst;
private boolean dumpRoutes = true;
public Builder(String from, String to) {
@@ -214,6 +215,23 @@ public class CamelKafkaConnectMain extends SimpleMain {
return this;
}
+ /**
+ * Controls where the header-removal stage sits in the route.
+ *
+ * On the sink path record headers are mapped onto the exchange by
{@code CamelSinkTask.put} *before* the
+ * exchange enters the route, so the stage must run first for {@code
camel.remove.headers.pattern} to keep
+ * those headers away from the marshalling, aggregation and
idempotency stages.
+ *
+ * On the source path headers arrive from the Camel consumer and are
mapped onto the produced record after
+ * the route has run, so the stage stays last and the intermediate
stages keep seeing them.
+ *
+ * @param removeHeadersFirst true for the sink direction, false (the
default) for the source direction.
+ */
+ public Builder withRemoveHeadersFirst(boolean removeHeadersFirst) {
+ this.removeHeadersFirst = removeHeadersFirst;
+ return this;
+ }
+
public Builder withDumpRoutes(boolean dumpRoutes) {
this.dumpRoutes = dumpRoutes;
return this;
@@ -375,6 +393,9 @@ public class CamelKafkaConnectMain extends SimpleMain {
//creating the actual route
ProcessorDefinition<?> rd = from(from);
+ if (removeHeadersFirst) {
+ rd = rd.kamelet("ckcRemoveHeader");
+ }
if (!ObjectHelper.isEmpty(marshallDataFormat)) {
rd = rd.kamelet("ckcMarshal");
}
@@ -387,7 +408,9 @@ public class CamelKafkaConnectMain extends SimpleMain {
if (idempotencyEnabled) {
rd = rd.kamelet("ckcIdempotent");
}
- rd = rd.kamelet("ckcRemoveHeader");
+ if (!removeHeadersFirst) {
+ rd = rd.kamelet("ckcRemoveHeader");
+ }
rd.toD(to);
}
});
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/utils/RemoveHeadersOrderTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/utils/RemoveHeadersOrderTest.java
new file mode 100644
index 0000000000..0ce66ae284
--- /dev/null
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/utils/RemoveHeadersOrderTest.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.kafkaconnector.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.kafkaconnector.CamelConnectorConfig;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * camel.remove.headers.pattern is the documented mitigation for untrusted
headers reaching the route. On the sink
+ * path the headers are mapped onto the exchange before it enters the route,
so the removal stage has to run before
+ * the stages that consume them; on the source path they arrive from the
consumer and are read off the exchange after
+ * the route, so it stays last.
+ */
+public class RemoveHeadersOrderTest {
+
+ private static final String STRIPPED_HEADER = "CamelExecCommandExecutable";
+
+ private RecordingAggregationStrategy runRouteWith(boolean
removeHeadersFirst) throws Exception {
+ Map<String, String> props = new HashMap<>();
+
props.put(CamelConnectorConfig.CAMEL_CONNECTOR_REMOVE_HEADERS_PATTERN_CONF,
STRIPPED_HEADER);
+
+ DefaultCamelContext context = new DefaultCamelContext();
+ RecordingAggregationStrategy strategy = new
RecordingAggregationStrategy();
+
context.getRegistry().bind(CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME,
strategy);
+
+ CamelKafkaConnectMain cms =
CamelKafkaConnectMain.builder("direct://start", "log://end")
+ .withProperties(props)
+ .withHeadersExcludePattern(STRIPPED_HEADER)
+ .withAggregationSize(1)
+ .withAggregationTimeout(1000L)
+ .withRemoveHeadersFirst(removeHeadersFirst)
+ .build(context);
+
+ cms.start();
+ try {
+ cms.getProducerTemplate().sendBodyAndHeader("direct://start",
"payload", STRIPPED_HEADER, "/bin/sh");
+ } finally {
+ cms.stop();
+ }
+ return strategy;
+ }
+
+ @Test
+ public void testSinkDirectionStripsHeadersBeforeTheAggregationStage()
throws Exception {
+ RecordingAggregationStrategy strategy = runRouteWith(true);
+
+ assertNull(strategy.seenHeaderValue,
+ "camel.remove.headers.pattern must strip the header before the
aggregation stage sees it on the "
+ + "sink path, but the strategy saw: " +
strategy.seenHeaderValue);
+ }
+
+ @Test
+ public void testSourceDirectionKeepsHeadersUntilTheEndOfTheRoute() throws
Exception {
+ RecordingAggregationStrategy strategy = runRouteWith(false);
+
+ assertEquals("/bin/sh", strategy.seenHeaderValue,
+ "on the source path the stripping stage stays last, so
intermediate stages still see the header");
+ }
+
+ private static final class RecordingAggregationStrategy implements
AggregationStrategy {
+
+ private volatile Object seenHeaderValue;
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ if (newExchange != null) {
+ seenHeaderValue =
newExchange.getMessage().getHeader(STRIPPED_HEADER);
+ }
+ return newExchange;
+ }
+ }
+}
