This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24192 in repository https://gitbox.apache.org/repos/asf/camel.git
commit b773865250ced652922b861980f93e19cfe08295 Author: Claus Ibsen <[email protected]> AuthorDate: Sun Jul 19 11:18:19 2026 +0200 CAMEL-24192: camel-lra - separate compensation/completion URI allowlists and prune on route removal Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/service/lra/LRASagaRoutes.java | 43 +++--- .../apache/camel/service/lra/LRASagaService.java | 71 +++++++++- .../service/lra/LRASagaServiceAllowlistTest.java | 148 +++++++++++++++++++++ .../java/org/apache/camel/reifier/SagaReifier.java | 1 + .../java/org/apache/camel/saga/CamelSagaStep.java | 10 ++ 5 files changed, 250 insertions(+), 23 deletions(-) diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java index b2538ac31c67..57ae40880e24 100644 --- a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java +++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java @@ -18,7 +18,10 @@ package org.apache.camel.service.lra; import java.net.URLDecoder; import java.nio.charset.StandardCharsets; -import java.util.*; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; import java.util.stream.Collectors; import org.apache.camel.Exchange; @@ -44,7 +47,7 @@ public class LRASagaRoutes extends RouteBuilder { rest(sagaService.getLocalParticipantContextPath()) .put(PARTICIPANT_PATH_COMPENSATE).to("direct:lra-compensation"); from("direct:lra-compensation").routeId("lra-compensation") - .process(this::verifyRequest) + .process(this::verifyCompensationRequest) .choice() .when(header(URL_COMPENSATION_KEY).isNotNull()) .toD("${header." + URL_COMPENSATION_KEY + "}") @@ -53,7 +56,7 @@ public class LRASagaRoutes extends RouteBuilder { rest(sagaService.getLocalParticipantContextPath()) .put(PARTICIPANT_PATH_COMPLETE).to("direct:lra-completion"); from("direct:lra-completion").routeId("lra-completion") - .process(this::verifyRequest) + .process(this::verifyCompletionRequest) .choice() .when(header(URL_COMPLETION_KEY).isNotNull()) .toD("${header." + URL_COMPLETION_KEY + "}") @@ -93,47 +96,49 @@ public class LRASagaRoutes extends RouteBuilder { return URLDecoder.decode(encodedString, StandardCharsets.UTF_8); } + private void verifyCompensationRequest(Exchange exchange) { + verifyRequest(exchange, true); + } + + private void verifyCompletionRequest(Exchange exchange) { + verifyRequest(exchange, false); + } + /** - * Check if the request is pointing to an allowed URI to prevent unauthorized remote uri invocation + * Check if the request is pointing to an allowed URI to prevent unauthorized remote uri invocation. Validates only + * the URI relevant to the callback type against the corresponding allowlist. */ - private void verifyRequest(Exchange exchange) { + private void verifyRequest(Exchange exchange, boolean isCompensation) { if (exchange.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION) == null) { throw new IllegalArgumentException("Missing " + Exchange.SAGA_LONG_RUNNING_ACTION + " header in received request"); } - Set<String> usedURIs = new HashSet<>(); String compensationURI = exchange.getIn().getHeader(URL_COMPENSATION_KEY, String.class); - if (compensationURI != null) { - usedURIs.add(compensationURI); - } String completionURI = exchange.getIn().getHeader(URL_COMPLETION_KEY, String.class); - if (completionURI != null) { - usedURIs.add(completionURI); - } // CAMEL-17751: Extract URIs from the CamelHttpQuery header - if (usedURIs.isEmpty()) { + if (compensationURI == null && completionURI == null) { Map<String, String> queryParams = parseQuery(exchange.getIn().getHeader(Exchange.HTTP_QUERY, String.class)); if (!queryParams.isEmpty()) { - if (queryParams.get(URL_COMPENSATION_KEY) != null) { compensationURI = queryParams.get(URL_COMPENSATION_KEY); - usedURIs.add(compensationURI); exchange.getIn().setHeader(URL_COMPENSATION_KEY, compensationURI); } - if (queryParams.get(URL_COMPLETION_KEY) != null) { completionURI = queryParams.get(URL_COMPLETION_KEY); - usedURIs.add(completionURI); exchange.getIn().setHeader(URL_COMPLETION_KEY, completionURI); } } } - for (String uri : usedURIs) { - if (!sagaService.getRegisteredURIs().contains(uri)) { + String uri = isCompensation ? compensationURI : completionURI; + if (uri != null) { + Set<String> allowedURIs = isCompensation + ? sagaService.getRegisteredCompensationURIs() + : sagaService.getRegisteredCompletionURIs(); + if (!allowedURIs.contains(uri)) { throw new IllegalArgumentException("URI " + uri + " is not allowed"); } } diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java index 73552c994ffb..451b7bbcb809 100644 --- a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java +++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java @@ -17,9 +17,13 @@ package org.apache.camel.service.lra; import java.net.URI; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledExecutorService; import org.apache.camel.*; @@ -29,8 +33,10 @@ import org.apache.camel.saga.CamelSagaCoordinator; import org.apache.camel.saga.CamelSagaService; import org.apache.camel.saga.CamelSagaStep; import org.apache.camel.spi.Configurer; +import org.apache.camel.spi.LifecycleStrategy; import org.apache.camel.spi.Metadata; import org.apache.camel.spi.annotations.JdkService; +import org.apache.camel.support.LifecycleStrategySupport; import org.apache.camel.support.service.ServiceSupport; /** @@ -45,7 +51,10 @@ public class LRASagaService extends ServiceSupport implements StaticService, Cam private ScheduledExecutorService executorService; private LRAClient client; private LRASagaRoutes routes; - private final Set<String> sagaURIs = ConcurrentHashMap.newKeySet(); + private LifecycleStrategy lifecycleStrategy; + private final Set<String> compensationURIs = ConcurrentHashMap.newKeySet(); + private final Set<String> completionURIs = ConcurrentHashMap.newKeySet(); + private final ConcurrentHashMap<String, List<CamelSagaStep>> stepsByRouteId = new ConcurrentHashMap<>(); // we want to be able to configure these following options @Metadata @@ -79,8 +88,38 @@ public class LRASagaService extends ServiceSupport implements StaticService, Cam @Override public void registerStep(CamelSagaStep step) { - step.getCompensation().map(Endpoint::getEndpointUri).ifPresent(this.sagaURIs::add); - step.getCompletion().map(Endpoint::getEndpointUri).ifPresent(this.sagaURIs::add); + step.getCompensation().map(Endpoint::getEndpointUri).ifPresent(compensationURIs::add); + step.getCompletion().map(Endpoint::getEndpointUri).ifPresent(completionURIs::add); + if (step.getRouteId() != null) { + stepsByRouteId.computeIfAbsent(step.getRouteId(), k -> new CopyOnWriteArrayList<>()).add(step); + } + } + + void unregisterSteps(String routeId) { + List<CamelSagaStep> steps = stepsByRouteId.remove(routeId); + if (steps == null) { + return; + } + + // collect URIs still referenced by other routes + Set<String> retainedCompensation = new HashSet<>(); + Set<String> retainedCompletion = new HashSet<>(); + for (List<CamelSagaStep> remaining : stepsByRouteId.values()) { + for (CamelSagaStep s : remaining) { + s.getCompensation().map(Endpoint::getEndpointUri).ifPresent(retainedCompensation::add); + s.getCompletion().map(Endpoint::getEndpointUri).ifPresent(retainedCompletion::add); + } + } + + // remove only URIs no longer referenced + for (CamelSagaStep step : steps) { + step.getCompensation().map(Endpoint::getEndpointUri) + .filter(uri -> !retainedCompensation.contains(uri)) + .ifPresent(compensationURIs::remove); + step.getCompletion().map(Endpoint::getEndpointUri) + .filter(uri -> !retainedCompletion.contains(uri)) + .ifPresent(completionURIs::remove); + } } @Override @@ -121,6 +160,9 @@ public class LRASagaService extends ServiceSupport implements StaticService, Cam this.client.close(); this.client = null; } + compensationURIs.clear(); + completionURIs.clear(); + stepsByRouteId.clear(); } @Override @@ -136,6 +178,17 @@ public class LRASagaService extends ServiceSupport implements StaticService, Cam throw RuntimeCamelException.wrapRuntimeException(ex); } } + if (this.lifecycleStrategy == null) { + this.lifecycleStrategy = new LifecycleStrategySupport() { + @Override + public void onRoutesRemove(Collection<Route> routes) { + for (Route r : routes) { + unregisterSteps(r.getRouteId()); + } + } + }; + this.camelContext.addLifecycleStrategy(this.lifecycleStrategy); + } } @Override @@ -187,8 +240,18 @@ public class LRASagaService extends ServiceSupport implements StaticService, Cam this.localParticipantContextPath = localParticipantContextPath; } + public Set<String> getRegisteredCompensationURIs() { + return compensationURIs; + } + + public Set<String> getRegisteredCompletionURIs() { + return completionURIs; + } + public Set<String> getRegisteredURIs() { - return sagaURIs; + Set<String> all = new HashSet<>(compensationURIs); + all.addAll(completionURIs); + return all; } @Override diff --git a/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRASagaServiceAllowlistTest.java b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRASagaServiceAllowlistTest.java new file mode 100644 index 000000000000..905ed19416b6 --- /dev/null +++ b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRASagaServiceAllowlistTest.java @@ -0,0 +1,148 @@ +/* + * 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.service.lra; + +import java.util.Collections; +import java.util.Set; + +import org.apache.camel.Endpoint; +import org.apache.camel.saga.CamelSagaStep; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class LRASagaServiceAllowlistTest extends CamelTestSupport { + + private LRASagaService sagaService; + + public LRASagaServiceAllowlistTest() { + testConfiguration().withUseRouteBuilder(false); + } + + @BeforeEach + void initService() { + sagaService = new LRASagaService(); + sagaService.setCoordinatorUrl("http://localhost:8080";); + sagaService.setLocalParticipantUrl("http://localhost:8081";); + sagaService.setLocalParticipantContextPath("/lra-participant"); + sagaService.setCoordinatorContextPath("/lra-coordinator"); + sagaService.setCamelContext(context()); + } + + @DisplayName("registerStep adds compensation URI only to the compensation allowlist") + @Test + void testCompensationUriSeparation() { + Endpoint compensation = context().getEndpoint("direct:compensate1"); + CamelSagaStep step = new CamelSagaStep(compensation, null, Collections.emptyMap(), null); + step.setRouteId("route1"); + + sagaService.registerStep(step); + + assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://compensate1")); + assertFalse(sagaService.getRegisteredCompletionURIs().contains("direct://compensate1")); + } + + @DisplayName("registerStep adds completion URI only to the completion allowlist") + @Test + void testCompletionUriSeparation() { + Endpoint completion = context().getEndpoint("direct:complete1"); + CamelSagaStep step = new CamelSagaStep(null, completion, Collections.emptyMap(), null); + step.setRouteId("route1"); + + sagaService.registerStep(step); + + assertTrue(sagaService.getRegisteredCompletionURIs().contains("direct://complete1")); + assertFalse(sagaService.getRegisteredCompensationURIs().contains("direct://complete1")); + } + + @DisplayName("getRegisteredURIs returns the union of both sets") + @Test + void testGetRegisteredURIsReturnsUnion() { + Endpoint compensation = context().getEndpoint("direct:compensate1"); + Endpoint completion = context().getEndpoint("direct:complete1"); + CamelSagaStep step = new CamelSagaStep(compensation, completion, Collections.emptyMap(), null); + step.setRouteId("route1"); + + sagaService.registerStep(step); + + Set<String> all = sagaService.getRegisteredURIs(); + assertEquals(2, all.size()); + assertTrue(all.contains("direct://compensate1")); + assertTrue(all.contains("direct://complete1")); + } + + @DisplayName("unregisterSteps prunes URIs from the allowlists") + @Test + void testUnregisterStepsPrunesURIs() { + Endpoint compensation = context().getEndpoint("direct:compensate1"); + Endpoint completion = context().getEndpoint("direct:complete1"); + CamelSagaStep step = new CamelSagaStep(compensation, completion, Collections.emptyMap(), null); + step.setRouteId("testRoute"); + + sagaService.registerStep(step); + assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://compensate1")); + assertTrue(sagaService.getRegisteredCompletionURIs().contains("direct://complete1")); + + sagaService.unregisterSteps("testRoute"); + + assertFalse(sagaService.getRegisteredCompensationURIs().contains("direct://compensate1")); + assertFalse(sagaService.getRegisteredCompletionURIs().contains("direct://complete1")); + assertTrue(sagaService.getRegisteredURIs().isEmpty()); + } + + @DisplayName("Shared URIs are only pruned when all referencing routes are removed") + @Test + void testSharedUriRetainedUntilAllRoutesRemoved() { + Endpoint sharedCompensation = context().getEndpoint("direct:sharedCompensate"); + + CamelSagaStep step1 = new CamelSagaStep(sharedCompensation, null, Collections.emptyMap(), null); + step1.setRouteId("routeA"); + sagaService.registerStep(step1); + + CamelSagaStep step2 = new CamelSagaStep(sharedCompensation, null, Collections.emptyMap(), null); + step2.setRouteId("routeB"); + sagaService.registerStep(step2); + + assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://sharedCompensate")); + + // remove routeA — shared URI should remain because routeB still references it + sagaService.unregisterSteps("routeA"); + assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://sharedCompensate")); + + // remove routeB — now the URI should be pruned + sagaService.unregisterSteps("routeB"); + assertFalse(sagaService.getRegisteredCompensationURIs().contains("direct://sharedCompensate")); + } + + @DisplayName("unregisterSteps for unknown routeId is a no-op") + @Test + void testUnregisterUnknownRouteIsNoOp() { + Endpoint compensation = context().getEndpoint("direct:compensate1"); + CamelSagaStep step = new CamelSagaStep(compensation, null, Collections.emptyMap(), null); + step.setRouteId("route1"); + sagaService.registerStep(step); + + sagaService.unregisterSteps("unknownRoute"); + + assertTrue(sagaService.getRegisteredCompensationURIs().contains("direct://compensate1")); + } +} diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SagaReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SagaReifier.java index 2a3568b1ce9c..37d284c1e9ff 100644 --- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SagaReifier.java +++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SagaReifier.java @@ -95,6 +95,7 @@ public class SagaReifier extends ProcessorReifier<SagaDefinition> { CamelSagaStep step = new CamelSagaStep( compensationEndpoint, completionEndpoint, optionsMap, parseDuration(timeout)); + step.setRouteId(route.getRouteId()); SagaPropagation propagation = parse(SagaPropagation.class, definition.getPropagation()); if (propagation == null) { diff --git a/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaStep.java b/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaStep.java index 2fdd83847bd8..1b2a57091929 100644 --- a/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaStep.java +++ b/core/camel-support/src/main/java/org/apache/camel/saga/CamelSagaStep.java @@ -35,6 +35,8 @@ public class CamelSagaStep { private final Long timeoutInMilliseconds; + private String routeId; + public CamelSagaStep(Endpoint compensation, Endpoint completion, Map<String, Expression> options, Long timeoutInMilliseconds) { this.compensation = compensation; @@ -59,6 +61,14 @@ public class CamelSagaStep { return Optional.ofNullable(timeoutInMilliseconds); } + public String getRouteId() { + return routeId; + } + + public void setRouteId(String routeId) { + this.routeId = routeId; + } + public boolean isEmpty() { return compensation == null && completion == null && options.isEmpty() && timeoutInMilliseconds == null; }
