You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2026/08/24 12:57:04 UTC
(camel-kafka-connector) branch main updated: Fix #1806: apply camel.remove.headers.pattern before the sink route stages (#1815)
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 <an...@gmail.com>
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 <an...@gmail.com>
Co-authored-by: Claude Opus 5 (1M context) <no...@anthropic.com>
---
.../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;
+ }
+ }
+}