This is an automated email from the ASF dual-hosted git repository. luigidemasi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit ac80c3fc972c733871444851aaa564aab2c6a35a Author: Luigi De Masi <[email protected]> AuthorDate: Sat Oct 10 13:57:17 2026 +0200 CAMEL-25499: Safely publish audit redactors and capture input once Publish resolved redactors as one immutable snapshot and discard them on stop so early capture cannot use partial or previous bindings. Guard input capture independently of evaluation completion and document audit failure isolation and recursive snapshot limits. Add regressions for deferred startup, changed redactors after restart, repeated capture, nested node limits, and assertion/linkage failures in snapshots and redactors. Co-authored-by: OpenAI Codex <[email protected]> --- .../camel/catalog/docs/semantic-language.adoc | 3 + .../src/main/docs/semantic-language.adoc | 3 + .../internal/SemanticAuditInputSnapshot.java | 1 + .../semantic/internal/SemanticAuditService.java | 15 +- .../camel/semantic/SemanticAuditInputTest.java | 169 +++++++++++++++++++++ 5 files changed, 187 insertions(+), 4 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/semantic-language.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/semantic-language.adoc index 34f6e30089b1..c0e7b9aad989 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/semantic-language.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/semantic-language.adoc @@ -1217,6 +1217,9 @@ The equivalent XML expert entry is: ---- Input settings follow the same restart requirement as the other audit settings. +Redactor bindings are published together at startup and discarded on stop. Capture before +configured redactors are ready omits input with `inputOmitted: redaction_failed`. Each evaluation +captures its input at most once, including when the first attempt omits it. Captured input goes to every configured audit sink, including `log` when selected, and is available through the audit reader, CLI, TUI and MCP. Existing backend retention and access controls therefore also apply to input. No automatic redaction is applied without a configured diff --git a/components/camel-ai/camel-semantic/src/main/docs/semantic-language.adoc b/components/camel-ai/camel-semantic/src/main/docs/semantic-language.adoc index 34f6e30089b1..c0e7b9aad989 100644 --- a/components/camel-ai/camel-semantic/src/main/docs/semantic-language.adoc +++ b/components/camel-ai/camel-semantic/src/main/docs/semantic-language.adoc @@ -1217,6 +1217,9 @@ The equivalent XML expert entry is: ---- Input settings follow the same restart requirement as the other audit settings. +Redactor bindings are published together at startup and discarded on stop. Capture before +configured redactors are ready omits input with `inputOmitted: redaction_failed`. Each evaluation +captures its input at most once, including when the first attempt omits it. Captured input goes to every configured audit sink, including `log` when selected, and is available through the audit reader, CLI, TUI and MCP. Existing backend retention and access controls therefore also apply to input. No automatic redaction is applied without a configured diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditInputSnapshot.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditInputSnapshot.java index 5887e68639b1..460a6bd090f2 100644 --- a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditInputSnapshot.java +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditInputSnapshot.java @@ -59,6 +59,7 @@ public final class SemanticAuditInputSnapshot { text(value.toString()); return value; } + // Collection sizes reject obvious overflows; recursive calls enforce the shared node budget. if (value instanceof List<?> list && list.size() <= nodes) { List<Object> copy = new ArrayList<>(); for (Object item : list) { diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditService.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditService.java index 097c7ff93f08..bd8c000b4001 100644 --- a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditService.java +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditService.java @@ -66,7 +66,7 @@ public final class SemanticAuditService extends ServiceSupport { private final AtomicLong observerErrors = new AtomicLong(); private final Map<String, AtomicLong> sinkErrors = new LinkedHashMap<>(); private final Map<String, SemanticAuditSink> sinks = new LinkedHashMap<>(); - private final Map<String, SemanticAuditInputRedactor> redactors = new LinkedHashMap<>(); + private volatile Map<String, SemanticAuditInputRedactor> redactors = Map.of(); private volatile List<SemanticObserver> observers = List.of(); private volatile ExecutorService executor; private MemorySemanticAuditStore memory; @@ -151,7 +151,7 @@ public final class SemanticAuditService extends ServiceSupport { if (memory == null || memory.getCapacity() != configuration.getCapacity()) { memory = new MemorySemanticAuditStore(configuration.getCapacity()); } - redactors.clear(); + Map<String, SemanticAuditInputRedactor> resolvedRedactors = new LinkedHashMap<>(); for (var entry : configuration.getInputs().entrySet()) { String name = entry.getValue().getRedactor(); if (name != null) { @@ -160,7 +160,7 @@ public final class SemanticAuditService extends ServiceSupport { if (redactor == null) { throw new IllegalArgumentException("Unknown semantic audit input redactor: " + name); } - redactors.put(entry.getKey(), redactor); + resolvedRedactors.put(entry.getKey(), redactor); } } sinkErrors.clear(); @@ -194,6 +194,8 @@ public final class SemanticAuditService extends ServiceSupport { executor = new ThreadPoolBuilder(context).poolSize(1).maxPoolSize(1) .maxQueueSize(configuration.getQueueCapacity()).rejectedPolicy(ThreadPoolRejectedPolicy.Abort) .build(this, "SemanticAuditService"); + // Publish all bindings together after startup succeeds. + redactors = Map.copyOf(resolvedRedactors); } catch (Exception e) { ServiceHelper.stopService(sinks.values()); ServiceHelper.stopService(reader); @@ -203,6 +205,7 @@ public final class SemanticAuditService extends ServiceSupport { @Override protected void doStop() throws Exception { + redactors = Map.of(); ExecutorService current = executor; executor = null; if (current != null) { @@ -381,6 +384,7 @@ public final class SemanticAuditService extends ServiceSupport { private final List<SemanticObserver.Observation> handles = new ArrayList<>(); // Confined to the invocation thread; cancellation emits a separate request record. private boolean completed; + private boolean inputCaptured; private final Map<String, Object> inputFields = new LinkedHashMap<>(); private Invocation(Map<String, Object> fields, boolean enabled) { @@ -393,9 +397,10 @@ public final class SemanticAuditService extends ServiceSupport { /** Snapshot the effective selected state before a provider can mutate it. Observers never receive this data. */ public void captureInput(Object input) { - if (fields == null || completed || inputFields.isEmpty()) { + if (fields == null || completed || inputCaptured || inputFields.isEmpty()) { return; } + inputCaptured = true; SemanticAuditInputConfiguration policy = configuration.getInput((String) fields.get("expert")); SemanticAuditInputRedactor redactor = redactors.get(fields.get("expert")); inputFields.clear(); @@ -407,6 +412,7 @@ public final class SemanticAuditService extends ServiceSupport { try { snapshot = SemanticAuditInputSnapshot.copy(input, policy.getMaxChars()); } catch (Exception | AssertionError | LinkageError failure) { + // Isolate audit snapshot failures from evaluation; fatal VM errors still propagate. inputFields.put("inputOmitted", "snapshot_limit_or_unsupported_type"); return; } @@ -419,6 +425,7 @@ public final class SemanticAuditService extends ServiceSupport { } snapshot = SemanticAuditInputSnapshot.copy(snapshot, policy.getMaxChars()); } catch (Exception | AssertionError | LinkageError failure) { + // A failing redactor omits input without affecting evaluation or retaining raw data. inputFields.put("inputOmitted", "redaction_failed"); return; } diff --git a/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/SemanticAuditInputTest.java b/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/SemanticAuditInputTest.java index 1deb2ee83e0d..ca2c67bce642 100644 --- a/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/SemanticAuditInputTest.java +++ b/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/SemanticAuditInputTest.java @@ -16,11 +16,15 @@ */ package org.apache.camel.semantic; +import java.util.AbstractList; import java.util.ArrayList; import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.apache.camel.builder.RouteBuilder; @@ -33,6 +37,7 @@ import org.apache.camel.semantic.internal.SemanticAuditService; import org.apache.camel.support.DefaultExchange; import org.apache.camel.support.PluginHelper; import org.apache.camel.support.ResourceHelper; +import org.apache.camel.support.SimpleRegistry; import org.apache.camel.util.json.JsonObject; import org.apache.camel.util.json.Jsoner; import org.junit.jupiter.api.Test; @@ -297,6 +302,170 @@ class SemanticAuditInputTest { } } + @ParameterizedTest + @ValueSource(strings = { "capture", "omit", "throw" }) + void inputIsCapturedOnceWithoutCompletingTheInvocation(String behavior) throws Exception { + try (var context = new DefaultCamelContext()) { + AtomicInteger calls = new AtomicInteger(); + context.getRegistry().bind("redactor", (SemanticAuditInputRedactor) input -> { + calls.incrementAndGet(); + return switch (behavior) { + case "omit" -> null; + case "throw" -> throw new IllegalStateException("private diagnostic"); + default -> input; + }; + }); + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(true, new SemanticAuditInputConfiguration(true, 100, "redactor"))); + context.start(); + var invocation = audit.begin(null, null, evaluation("security"), null, "security", null, null, null); + invocation.captureInput("first"); + invocation.captureInput("second"); + invocation.complete("success", "evaluation_completed", null); + audit.stop(); + assertThat(calls).hasValue(1); + assertThat(records(audit)).singleElement().satisfies(record -> { + assertThat(record.toMap()).containsEntry("status", "success"); + if (behavior.equals("capture")) { + assertThat(record.toMap()).containsEntry("input", "first"); + } else { + assertThat(record.toMap()).doesNotContainKey("input").containsEntry("inputOmitted", + behavior.equals("omit") ? "redacted" : "redaction_failed"); + } + }); + } + } + + @Test + void deferredStartupPublishesAllRedactorsTogether() throws Exception { + CountDownLatch resolving = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger lookups = new AtomicInteger(); + var registry = new SimpleRegistry() { + @Override + public <T> T lookupByNameAndType(String name, Class<T> type) { + if (type == SemanticAuditInputRedactor.class && lookups.incrementAndGet() == 2) { + resolving.countDown(); + try { + assertThat(release.await(10, TimeUnit.SECONDS)).isTrue(); + } catch (InterruptedException failure) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(failure); + } + } + return super.lookupByNameAndType(name, type); + } + }; + registry.bind("redactor", (SemanticAuditInputRedactor) input -> "safe"); + try (var context = new DefaultCamelContext(registry)) { + var audit = SemanticAuditService.get(context); + var policy = new SemanticAuditInputConfiguration(true, 100, "redactor"); + audit.configure("test", new SemanticAuditConfiguration( + true, + Map.of("security", true, "decisions", true), List.of("memory"), "memory", 100, 100, + Map.of("security", policy, "decisions", policy))); + var before = List.of( + audit.begin(null, null, evaluation("security"), null, "security", null, null, null), + audit.begin(null, null, evaluation("decisions"), null, "decisions", null, null, null)); + var startup = CompletableFuture.runAsync(context::start); + try { + assertThat(resolving.await(10, TimeUnit.SECONDS)).isTrue(); + before.forEach(invocation -> invocation.captureInput("private")); + } finally { + release.countDown(); + startup.get(10, TimeUnit.SECONDS); + } + before.forEach(invocation -> invocation.complete("success", "evaluation_completed", null)); + for (String expert : List.of("security", "decisions")) { + var invocation = audit.begin(null, null, evaluation(expert), null, expert, null, null, null); + invocation.captureInput("private"); + invocation.complete("success", "evaluation_completed", null); + } + audit.stop(); + assertThat(records(audit)).hasSize(4); + for (var invocation : before) { + var record = audit.getReader().get(invocation.getEventId()).orElseThrow(); + assertThat(record.toMap()).doesNotContainKey("input").containsEntry("inputOmitted", "redaction_failed"); + } + assertThat(records(audit).subList(0, 2)).allSatisfy(record -> assertThat(record.toMap()) + .containsEntry("input", "safe").containsEntry("inputRedacted", true)); + } finally { + release.countDown(); + } + } + + @Test + void stoppedContextDoesNotReuseThePreviousRedactor() throws Exception { + try (var context = new DefaultCamelContext()) { + context.getRegistry().bind("old", (SemanticAuditInputRedactor) input -> "old policy"); + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(true, new SemanticAuditInputConfiguration(true, 100, "old"))); + context.start(); + audit.activate(); + context.stop(); + context.getRegistry().bind("new", (SemanticAuditInputRedactor) input -> "new policy"); + audit.configure("test", configuration(true, new SemanticAuditInputConfiguration(true, 100, "new"))); + var before = audit.begin(null, null, evaluation("security"), null, "security", null, null, null); + before.captureInput("private"); + context.start(); + before.complete("success", "evaluation_completed", null); + var after = audit.begin(null, null, evaluation("security"), null, "security", null, null, null); + after.captureInput("private"); + after.complete("success", "evaluation_completed", null); + audit.stop(); + assertThat(records(audit)).hasSize(2); + assertThat(audit.getReader().get(before.getEventId()).orElseThrow().toMap()) + .doesNotContainKey("input").containsEntry("inputOmitted", "redaction_failed"); + assertThat(audit.getReader().get(after.getEventId()).orElseThrow().toMap()) + .containsEntry("input", "new policy").containsEntry("inputRedacted", true); + } + } + + @ParameterizedTest + @ValueSource(strings = { "snapshot_assertion", "snapshot_linkage", "redactor_assertion", "redactor_linkage" }) + void captureErrorsOmitInputWithoutChangingTheEvaluation(String scenario) throws Exception { + try (var context = new DefaultCamelContext()) { + Error failure = scenario.endsWith("assertion") + ? new AssertionError("private diagnostic") + : new LinkageError("private diagnostic"); + context.getRegistry().bind("security", new StructuredExpert()); + context.getRegistry().bind("redactor", (SemanticAuditInputRedactor) input -> { + throw failure; + }); + boolean redaction = scenario.startsWith("redactor"); + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(true, + new SemanticAuditInputConfiguration(true, 100, redaction ? "redactor" : null))); + context.start(); + Object input = redaction ? "private" : new AbstractList<>() { + @Override + public Object get(int index) { + throw failure; + } + + @Override + public int size() { + return 1; + } + }; + var result = ((SemanticLanguage) context.resolveLanguage("semantic")) + .evaluate(new SemanticEvaluation("check", "security", null, Map.of()), input); + assertThat(result.getValue()).isEqualTo(true); + audit.stop(); + assertThat(records(audit)).singleElement().satisfies(record -> assertThat(record.toMap()) + .containsEntry("status", "success").doesNotContainKey("input") + .containsEntry("inputOmitted", redaction ? "redaction_failed" : "snapshot_limit_or_unsupported_type")); + } + } + + @Test + void nestedSnapshotsEnforceTheSharedNodeBudget() { + var fullBudget = List.of(Collections.nCopies(998, "")); + assertThat(SemanticAuditInputSnapshot.copy(fullBudget, 100)).isEqualTo(fullBudget); + assertThatThrownBy(() -> SemanticAuditInputSnapshot.copy(List.of(Collections.nCopies(998, ""), ""), 100)) + .isInstanceOf(IllegalArgumentException.class).hasMessageContaining("snapshot limit"); + } + @Test void unknownRedactorFailsStartup() throws Exception { try (var context = new DefaultCamelContext()) {
