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 9205e53fc0e69b056ccc38a9e4a50e2ecf40f2d5 Author: Luigi De Masi <[email protected]> AuthorDate: Sat Oct 10 12:39:28 2026 +0200 CAMEL-25499: Add opt-in semantic audit input snapshots Retain the selected state per expert with bounded immutable snapshots and an optional registry redactor. Keep captured input out of observer callbacks and report omissions without affecting evaluation outcomes. Expose the policy through YAML, XML and Java with schema, persistence and isolation regressions. Co-authored-by: OpenAI Codex <[email protected]> --- .../camel/catalog/docs/semantic-language.adoc | 98 +++++- .../camel/catalog/schemas/camelYamlDsl-model.json | 16 + .../src/main/docs/semantic-language.adoc | 98 +++++- .../camel/language/semantic/SemanticLanguage.java | 3 + .../camel/semantic/SemanticAuditConfiguration.java | 25 +- .../semantic/SemanticAuditInputConfiguration.java | 65 ++++ .../camel/semantic/SemanticAuditInputRedactor.java | 27 ++ .../apache/camel/semantic/SemanticAuditRecord.java | 21 +- .../semantic/SemanticXmlRoutesBuilderLoader.java | 16 +- .../internal/SemanticAuditInputSnapshot.java | 88 +++++ .../semantic/internal/SemanticAuditService.java | 63 +++- .../yaml/SemanticDefinitionDeserializer.java | 34 +- .../camel/semantic/SemanticAuditInputTest.java | 355 +++++++++++++++++++++ design/semantic-audit.adoc | 22 +- .../resources/schema/camelYamlDsl-canonical.json | 18 ++ .../resources/schema/camelYamlDsl-model.json | 16 + .../generated/resources/schema/camelYamlDsl.json | 18 ++ .../camel/dsl/yaml/SemanticEvaluationTest.java | 16 +- 18 files changed, 963 insertions(+), 36 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 37c8f1cb7abb..34f6e30089b1 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 @@ -1138,13 +1138,97 @@ Decision records carry explicit action, operation, target, optional namespace/co and policy identifiers, reason code and evidence IDs. They never derive application correlation or tenant identity from arbitrary inbound headers. -Input, body, headers, variables, arbitrary provider metadata and exception messages are -omitted before dispatch to any backend or observer. Caller-authored score descriptions -are also omitted; the UI does not substitute today's definition for missing historical -semantics. Strings are limited to 256 code points and collections to 100 entries. -Oversized result snapshots are omitted with `resultOmitted: snapshot_limit`, preserving -the execution status without displaying altered labels or partial probabilities. These records are evidence about -an evaluation and declared action, not a copy of the submitted content. +Input is omitted by default. Headers, variables, arbitrary provider metadata and exception +messages are never copied automatically. Caller-authored score descriptions are also omitted; +static capability meanings remain available. Result and metadata strings are limited to +256 code points, and their collections to 100 entries. An oversized result is represented by +`resultOmitted: snapshot_limit` without changing the evaluation outcome. + +==== Storing evaluation input + +Enable input capture explicitly for each expert whose submitted state should be retained: + +[source,yaml] +---- +- semantic: + audit: + enabled: true + experts: + security: + enabled: true + input: + enabled: true + maxChars: 4096 + # redactor: auditInputRedactor + decisions: + enabled: false + sinks: [memory] + reader: memory +---- + +`input.enabled` defaults to `false` and does not enable auditing itself. It captures the +selected state passed to the expert (`state`, including a selected header), before provider +validation or execution. It does not take a copy of the entire exchange. Named, batch, direct +and console evaluations use the same settings. Decisions link to their evaluation records; +input is stored on the evaluation rather than duplicated on the decision. + +`maxChars` defaults to 4096 and accepts 1 through 100000. It bounds the total Unicode code +points in strings, map keys and scalar representations. Snapshots accept text, JSON-compatible +maps/lists and scalar values, with at most eight nested levels and 1000 nodes (including map +keys). They are immutable; later changes to the submitted state cannot change history. +Oversized, cyclic or unsupported input is omitted with +`inputOmitted: snapshot_limit_or_unsupported_type`. Input is never truncated, and streams or +arbitrary objects are not converted to text. `inputOmitted: unavailable` means the selected +state was not obtained, for example because the state expression failed. + +An optional `redactor` names a registry bean implementing `SemanticAuditInputRedactor`. +It receives a bounded, immutable copy on the evaluation thread and returns a sanitized value. +The return value is bounded again; a null return omits input (`inputOmitted: redacted`). +Successful redaction adds `inputRedacted: true`. Exceptions or invalid redactor output omit +input (`inputOmitted: redaction_failed`) without recording exception text or affecting the +expert result. Redactors must be thread-safe and should not block. Unknown bean references +fail startup. For example, a redactor can remove confidential fields from structured state: + +[source,java] +---- +context.getRegistry().bind("auditInputRedactor", (SemanticAuditInputRedactor) input -> { + if (input instanceof Map<?, ?> state) { + return Map.of("question", state.get("question")); + } + return null; +}); +---- + +Java uses the additional input settings map in `SemanticAuditConfiguration`: + +[source,java] +---- +new SemanticAuditConfiguration(true, Map.of("security", true, "decisions", false), + List.of("memory"), "memory", 1000, 1000, + Map.of("security", new SemanticAuditInputConfiguration(true, 4096, "auditInputRedactor"))); +---- + +The equivalent XML expert entry is: + +[source,xml] +---- +<expert name="security" enabled="true" inputEnabled="true" + inputMaxChars="4096" inputRedactor="auditInputRedactor"/> +---- + +Input settings follow the same restart requirement as the other audit settings. +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 +redactor. Input is never passed to `SemanticObserver`, even when audit input capture is enabled; +auditing and telemetry remain independent. Existing records without input remain readable. + +The TUI inspector displays *Input* (or *Input (redacted)*) in the evaluation details, including +linked evidence beside a route decision. It displays an omission reason when available, or +*not captured* for records without input. The CLI prints it with +`camel semantic audit my-app --event-id=<event-id>`; add `--json` for the stored structure. + +==== Observation hooks Register `SemanticObserver` beans to consume start/completion and explicit decision hooks independently of auditing. Beans must be registered before context startup: observers are diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camelYamlDsl-model.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camelYamlDsl-model.json index 7472732229c7..47209ce3178f 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camelYamlDsl-model.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camelYamlDsl-model.json @@ -20573,6 +20573,22 @@ "name" : "enabled", "type" : "boolean", "required" : true + }, { + "name" : "input", + "type" : "object", + "ref" : "semanticDefinitionDeserializer$AuditInputSchema" + } ] + }, + "semanticDefinitionDeserializer$AuditInputSchema" : { + "children" : [ { + "name" : "enabled", + "type" : "boolean" + }, { + "name" : "maxChars", + "type" : "integer" + }, { + "name" : "redactor", + "type" : "string" } ] }, "semanticDefinitionDeserializer$AuditSchema" : { 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 37c8f1cb7abb..34f6e30089b1 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 @@ -1138,13 +1138,97 @@ Decision records carry explicit action, operation, target, optional namespace/co and policy identifiers, reason code and evidence IDs. They never derive application correlation or tenant identity from arbitrary inbound headers. -Input, body, headers, variables, arbitrary provider metadata and exception messages are -omitted before dispatch to any backend or observer. Caller-authored score descriptions -are also omitted; the UI does not substitute today's definition for missing historical -semantics. Strings are limited to 256 code points and collections to 100 entries. -Oversized result snapshots are omitted with `resultOmitted: snapshot_limit`, preserving -the execution status without displaying altered labels or partial probabilities. These records are evidence about -an evaluation and declared action, not a copy of the submitted content. +Input is omitted by default. Headers, variables, arbitrary provider metadata and exception +messages are never copied automatically. Caller-authored score descriptions are also omitted; +static capability meanings remain available. Result and metadata strings are limited to +256 code points, and their collections to 100 entries. An oversized result is represented by +`resultOmitted: snapshot_limit` without changing the evaluation outcome. + +==== Storing evaluation input + +Enable input capture explicitly for each expert whose submitted state should be retained: + +[source,yaml] +---- +- semantic: + audit: + enabled: true + experts: + security: + enabled: true + input: + enabled: true + maxChars: 4096 + # redactor: auditInputRedactor + decisions: + enabled: false + sinks: [memory] + reader: memory +---- + +`input.enabled` defaults to `false` and does not enable auditing itself. It captures the +selected state passed to the expert (`state`, including a selected header), before provider +validation or execution. It does not take a copy of the entire exchange. Named, batch, direct +and console evaluations use the same settings. Decisions link to their evaluation records; +input is stored on the evaluation rather than duplicated on the decision. + +`maxChars` defaults to 4096 and accepts 1 through 100000. It bounds the total Unicode code +points in strings, map keys and scalar representations. Snapshots accept text, JSON-compatible +maps/lists and scalar values, with at most eight nested levels and 1000 nodes (including map +keys). They are immutable; later changes to the submitted state cannot change history. +Oversized, cyclic or unsupported input is omitted with +`inputOmitted: snapshot_limit_or_unsupported_type`. Input is never truncated, and streams or +arbitrary objects are not converted to text. `inputOmitted: unavailable` means the selected +state was not obtained, for example because the state expression failed. + +An optional `redactor` names a registry bean implementing `SemanticAuditInputRedactor`. +It receives a bounded, immutable copy on the evaluation thread and returns a sanitized value. +The return value is bounded again; a null return omits input (`inputOmitted: redacted`). +Successful redaction adds `inputRedacted: true`. Exceptions or invalid redactor output omit +input (`inputOmitted: redaction_failed`) without recording exception text or affecting the +expert result. Redactors must be thread-safe and should not block. Unknown bean references +fail startup. For example, a redactor can remove confidential fields from structured state: + +[source,java] +---- +context.getRegistry().bind("auditInputRedactor", (SemanticAuditInputRedactor) input -> { + if (input instanceof Map<?, ?> state) { + return Map.of("question", state.get("question")); + } + return null; +}); +---- + +Java uses the additional input settings map in `SemanticAuditConfiguration`: + +[source,java] +---- +new SemanticAuditConfiguration(true, Map.of("security", true, "decisions", false), + List.of("memory"), "memory", 1000, 1000, + Map.of("security", new SemanticAuditInputConfiguration(true, 4096, "auditInputRedactor"))); +---- + +The equivalent XML expert entry is: + +[source,xml] +---- +<expert name="security" enabled="true" inputEnabled="true" + inputMaxChars="4096" inputRedactor="auditInputRedactor"/> +---- + +Input settings follow the same restart requirement as the other audit settings. +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 +redactor. Input is never passed to `SemanticObserver`, even when audit input capture is enabled; +auditing and telemetry remain independent. Existing records without input remain readable. + +The TUI inspector displays *Input* (or *Input (redacted)*) in the evaluation details, including +linked evidence beside a route decision. It displays an omission reason when available, or +*not captured* for records without input. The CLI prints it with +`camel semantic audit my-app --event-id=<event-id>`; add `--json` for the stored structure. + +==== Observation hooks Register `SemanticObserver` beans to consume start/completion and explicit decision hooks independently of auditing. Beans must be registered before context startup: observers are diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/language/semantic/SemanticLanguage.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/language/semantic/SemanticLanguage.java index 546e7d17e8f4..ea64debe3cf4 100644 --- a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/language/semantic/SemanticLanguage.java +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/language/semantic/SemanticLanguage.java @@ -153,6 +153,7 @@ public class SemanticLanguage extends LanguageSupport { invocation = audit.begin(null, null, evaluation, operation, resolved.reference, resolved.capabilities.getProvider(), null, request); reason = "invalid_input"; + invocation.captureInput(input); operation.validateInput(input); resolved.provider.validateInput(evaluation, input); if (Thread.currentThread().isInterrupted()) { @@ -167,6 +168,7 @@ public class SemanticLanguage extends LanguageSupport { } catch (Exception | AssertionError failure) { if (invocation == null) { invocation = audit.begin(null, null, evaluation, null, auditExpert(evaluation), null, null, request); + invocation.captureInput(input); } invocation.complete("failed", failure instanceof InterruptedException ? "interrupted" : reason, null); if (failure instanceof InterruptedException) { @@ -720,6 +722,7 @@ public class SemanticLanguage extends LanguageSupport { active.addAll(invocations.values()); failureReason = "invalid_input"; Object state = current.state.evaluate(exchange, Object.class); + invocations.values().forEach(invocation -> invocation.captureInput(state)); // A state selector can itself evaluate another semantic expression. exchange.removeProperty(RESULT); exchange.removeProperty(RESULTS); diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditConfiguration.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditConfiguration.java index c9818d90c86d..f93f65da4d58 100644 --- a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditConfiguration.java +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditConfiguration.java @@ -24,6 +24,7 @@ import java.util.Objects; public final class SemanticAuditConfiguration { public static final SemanticAuditConfiguration DISABLED = new SemanticAuditConfiguration(false, Map.of(), List.of("memory"), "memory", 1000, 1000); + private final Map<String, SemanticAuditInputConfiguration> inputs; private final boolean enabled; private final Map<String, Boolean> experts; private final List<String> sinks; @@ -33,6 +34,16 @@ public final class SemanticAuditConfiguration { public SemanticAuditConfiguration(boolean enabled, Map<String, Boolean> experts, List<String> sinks, String reader, int capacity, int queueCapacity) { + this(enabled, experts, sinks, reader, capacity, queueCapacity, Map.of()); + } + + public SemanticAuditConfiguration(boolean enabled, Map<String, Boolean> experts, List<String> sinks, + String reader, int capacity, int queueCapacity, + Map<String, SemanticAuditInputConfiguration> inputs) { + this.inputs = Map.copyOf(inputs); + if (!experts.keySet().containsAll(inputs.keySet())) { + throw new IllegalArgumentException("Audit input settings require a declared expert"); + } this.enabled = enabled; this.experts = Map.copyOf(experts); this.sinks = List.copyOf(sinks); @@ -68,6 +79,16 @@ public final class SemanticAuditConfiguration { return experts; } + public Map<String, SemanticAuditInputConfiguration> getInputs() { + return inputs; + } + + public SemanticAuditInputConfiguration getInput(String expert) { + return expert == null + ? SemanticAuditInputConfiguration.DISABLED + : inputs.getOrDefault(expert, SemanticAuditInputConfiguration.DISABLED); + } + public List<String> getSinks() { return sinks; } @@ -88,11 +109,11 @@ public final class SemanticAuditConfiguration { public boolean equals(Object other) { return other instanceof SemanticAuditConfiguration c && enabled == c.enabled && experts.equals(c.experts) && sinks.equals(c.sinks) && reader.equals(c.reader) && capacity == c.capacity - && queueCapacity == c.queueCapacity; + && queueCapacity == c.queueCapacity && inputs.equals(c.inputs); } @Override public int hashCode() { - return Objects.hash(enabled, experts, sinks, reader, capacity, queueCapacity); + return Objects.hash(enabled, experts, sinks, reader, capacity, queueCapacity, inputs); } } diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditInputConfiguration.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditInputConfiguration.java new file mode 100644 index 000000000000..bfe0e63e1189 --- /dev/null +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditInputConfiguration.java @@ -0,0 +1,65 @@ +/* + * 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.semantic; + +import java.util.Objects; + +/** Per-expert opt-in to retain the selected evaluation input, independently of telemetry. */ +public final class SemanticAuditInputConfiguration { + public static final int DEFAULT_MAX_CHARS = 4096; + public static final int MAX_CHARS = 100000; + public static final SemanticAuditInputConfiguration DISABLED + = new SemanticAuditInputConfiguration(false, DEFAULT_MAX_CHARS, null); + private final boolean enabled; + private final int maxChars; + private final String redactor; + + public SemanticAuditInputConfiguration(boolean enabled, int maxChars, String redactor) { + if (maxChars < 1 || maxChars > MAX_CHARS || redactor != null && redactor.isBlank()) { + throw new IllegalArgumentException( + "Audit input requires maxChars between 1 and 100000 and a nonblank redactor reference"); + } + this.enabled = enabled; + this.maxChars = maxChars; + this.redactor = redactor; + } + + public boolean isEnabled() { + return enabled; + } + + /** Total code points in strings, map keys and scalar representations; oversized input is omitted, not truncated. */ + public int getMaxChars() { + return maxChars; + } + + /** Optional registry name of a {@link SemanticAuditInputRedactor}. */ + public String getRedactor() { + return redactor; + } + + @Override + public boolean equals(Object other) { + return other instanceof SemanticAuditInputConfiguration c && enabled == c.enabled && maxChars == c.maxChars + && Objects.equals(redactor, c.redactor); + } + + @Override + public int hashCode() { + return Objects.hash(enabled, maxChars, redactor); + } +} diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditInputRedactor.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditInputRedactor.java new file mode 100644 index 000000000000..ab86adbfbff6 --- /dev/null +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditInputRedactor.java @@ -0,0 +1,27 @@ +/* + * 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.semantic; + +/** Optional registry bean applied to an immutable, bounded input snapshot before it reaches any audit sink. */ +@FunctionalInterface +public interface SemanticAuditInputRedactor { + /** + * Return sanitized JSON-compatible data, or null to omit it. Called on the evaluation thread; implementations must + * be thread-safe and should not block. Throwing omits the input without changing the evaluation outcome. + */ + Object redact(Object input) throws Exception; +} diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditRecord.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditRecord.java index 9cd7aeb71950..cd996b2beaa5 100644 --- a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditRecord.java +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticAuditRecord.java @@ -23,9 +23,11 @@ import java.util.List; import java.util.Map; import java.util.Set; +import org.apache.camel.semantic.internal.SemanticAuditInputSnapshot; + /** - * Immutable, size-bounded snapshot. It never contains the exchange, input or provider exception. Schema version 1 uses - * UTC timestamps and keeps route actions separate from expert results. + * Immutable, size-bounded snapshot. It never contains the exchange or provider exception. Input is opt-in. Schema + * version 1 uses UTC timestamps and keeps route actions separate from expert results. */ public final class SemanticAuditRecord { private final Map<String, Object> fields; @@ -41,7 +43,8 @@ public final class SemanticAuditRecord { Set<String> keys = Set.of("schemaVersion", "eventId", "invocationId", "timestamp", "category", "origin", "requestId", "batchId", "definition", "target", "expert", "operation", "contextId", "exchangeId", "routeId", "semantics", "status", "reasonCode", "durationNanos", "resultOmitted", "result", "action", "namespace", "correlationId", - "policyId", "policyVersion", "rule", "evidence", "startedAt", "provider", "model", "revision"); + "policyId", "policyVersion", "rule", "evidence", "startedAt", "provider", "model", "revision", "input", + "inputOmitted", "inputRedacted"); if (!keys.containsAll(data.keySet()) || !(data.get("schemaVersion") instanceof Number version) || version.doubleValue() != 1 || !(data.get("eventId") instanceof String id) || id.isBlank() || id.length() > 256 @@ -50,7 +53,8 @@ public final class SemanticAuditRecord { || !(data.get("timestamp") instanceof String timestamp)) { throw new IllegalArgumentException("Invalid semantic audit record"); } - Set<String> structured = Set.of("schemaVersion", "durationNanos", "semantics", "result", "evidence"); + Set<String> structured + = Set.of("schemaVersion", "durationNanos", "semantics", "result", "evidence", "input", "inputRedacted"); data.forEach((key, value) -> { if (!structured.contains(key) && !(value instanceof String)) { throw new IllegalArgumentException("Invalid audit text field: " + key); @@ -100,8 +104,15 @@ public final class SemanticAuditRecord { probabilities.values().forEach(SemanticAuditRecord::validateProbability); } } + if (data.containsKey("input") && data.containsKey("inputOmitted") + || data.containsKey("inputRedacted") && (!(data.get("inputRedacted") instanceof Boolean) + || !data.containsKey("input")) + || !"evaluation".equals(category) && (data.containsKey("input") || data.containsKey("inputOmitted"))) { + throw new IllegalArgumentException("Invalid audit input fields"); + } Map<String, Object> copy = new LinkedHashMap<>(); - data.forEach((key, value) -> copy.put(key, freeze(value, 0))); + data.forEach((key, value) -> copy.put(key, "input".equals(key) + ? SemanticAuditInputSnapshot.copy(value, SemanticAuditInputConfiguration.MAX_CHARS) : freeze(value, 0))); return new SemanticAuditRecord(copy); } diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticXmlRoutesBuilderLoader.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticXmlRoutesBuilderLoader.java index 14d2015f5dda..bbc40b6f3e4d 100644 --- a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticXmlRoutesBuilderLoader.java +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticXmlRoutesBuilderLoader.java @@ -181,17 +181,27 @@ public class SemanticXmlRoutesBuilderLoader extends RoutesBuilderLoaderSupport { private SemanticAuditConfiguration audit(Element element) { attributes(element, Set.of("enabled", "reader", "capacity", "queueCapacity")); Map<String, Boolean> experts = new LinkedHashMap<>(); + Map<String, SemanticAuditInputConfiguration> inputs = new LinkedHashMap<>(); List<String> sinks = new ArrayList<>(); for (Element child : children(element)) { if ("expert".equals(child.getLocalName())) { - attributes(child, Set.of("name", "enabled")); - String name = child.getAttribute("name"); + attributes(child, Set.of("name", "enabled", "inputEnabled", "inputMaxChars", "inputRedactor")); + String name = getCamelContext().resolvePropertyPlaceholders(child.getAttribute("name")); if (!child.hasAttribute("enabled")) { throw new IllegalArgumentException("Audit expert '" + name + "' requires enabled"); } if (experts.putIfAbsent(name, auditBoolean(child.getAttribute("enabled"))) != null) { throw new IllegalArgumentException("Duplicate audit expert: " + name); } + if (child.hasAttribute("inputEnabled") || child.hasAttribute("inputMaxChars") + || child.hasAttribute("inputRedactor")) { + inputs.put(name, new SemanticAuditInputConfiguration( + child.hasAttribute("inputEnabled") && auditBoolean(child.getAttribute("inputEnabled")), + child.hasAttribute("inputMaxChars") + ? auditCapacity(child, "inputMaxChars") + : SemanticAuditInputConfiguration.DEFAULT_MAX_CHARS, + auditAttribute(child, "inputRedactor", null))); + } } else if ("sink".equals(child.getLocalName())) { attributes(child, Set.of("ref")); sinks.add(getCamelContext().resolvePropertyPlaceholders(child.getAttribute("ref"))); @@ -206,7 +216,7 @@ public class SemanticXmlRoutesBuilderLoader extends RoutesBuilderLoaderSupport { element.hasAttribute("enabled") && auditBoolean(element.getAttribute("enabled")), experts, sinks.isEmpty() ? List.of("memory") : sinks, auditAttribute(element, "reader", "memory"), auditCapacity(element, "capacity"), - auditCapacity(element, "queueCapacity")); + auditCapacity(element, "queueCapacity"), inputs); } private int auditCapacity(Element element, String key) { 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 new file mode 100644 index 000000000000..5887e68639b1 --- /dev/null +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/internal/SemanticAuditInputSnapshot.java @@ -0,0 +1,88 @@ +/* + * 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.semantic.internal; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** Bounded JSON values only: never converts arbitrary objects or consumes streams. */ +public final class SemanticAuditInputSnapshot { + private int remaining; + private int nodes = 1000; + + private SemanticAuditInputSnapshot(int maxChars) { + remaining = maxChars; + } + + public static Object copy(Object input, int maxChars) { + return new SemanticAuditInputSnapshot(maxChars).value(input, 0); + } + + private Object value(Object value, int depth) { + if (--nodes < 0 || depth > 8) { + throw new IllegalArgumentException("Input snapshot limit"); + } + if (value == null || value instanceof Boolean) { + text(String.valueOf(value)); + return value; + } + if (value instanceof String text) { + return text(text); + } + if (value instanceof Byte || value instanceof Short || value instanceof Integer || value instanceof Long + || value instanceof Float || value instanceof Double || value instanceof BigDecimal + || value instanceof BigInteger) { + if (value instanceof Double d && !Double.isFinite(d) || value instanceof Float f && !Float.isFinite(f) + || value instanceof BigInteger integer && integer.bitLength() > remaining * 4 + || value instanceof BigDecimal decimal && decimal.precision() > remaining) { + throw new IllegalArgumentException("Input snapshot limit"); + } + text(value.toString()); + return value; + } + if (value instanceof List<?> list && list.size() <= nodes) { + List<Object> copy = new ArrayList<>(); + for (Object item : list) { + copy.add(value(item, depth + 1)); + } + return Collections.unmodifiableList(copy); + } + if (value instanceof Map<?, ?> map && map.size() <= nodes / 2) { + Map<String, Object> copy = new LinkedHashMap<>(); + for (var entry : map.entrySet()) { + if (!(entry.getKey() instanceof String key)) { + throw new IllegalArgumentException("Unsupported input key"); + } + copy.put((String) value(key, depth + 1), value(entry.getValue(), depth + 1)); + } + return Collections.unmodifiableMap(copy); + } + throw new IllegalArgumentException("Unsupported input or snapshot limit"); + } + + private String text(String text) { + if (text.length() > remaining * 2 || (remaining -= text.codePointCount(0, text.length())) < 0) { + throw new IllegalArgumentException("Input snapshot limit"); + } + return text; + } +} 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 cf0577690468..097c7ff93f08 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 @@ -38,6 +38,8 @@ import org.apache.camel.builder.ThreadPoolBuilder; import org.apache.camel.semantic.LoggingSemanticAuditSink; import org.apache.camel.semantic.MemorySemanticAuditStore; import org.apache.camel.semantic.SemanticAuditConfiguration; +import org.apache.camel.semantic.SemanticAuditInputConfiguration; +import org.apache.camel.semantic.SemanticAuditInputRedactor; import org.apache.camel.semantic.SemanticAuditReader; import org.apache.camel.semantic.SemanticAuditRecord; import org.apache.camel.semantic.SemanticAuditSink; @@ -64,6 +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 List<SemanticObserver> observers = List.of(); private volatile ExecutorService executor; private MemorySemanticAuditStore memory; @@ -148,6 +151,18 @@ public final class SemanticAuditService extends ServiceSupport { if (memory == null || memory.getCapacity() != configuration.getCapacity()) { memory = new MemorySemanticAuditStore(configuration.getCapacity()); } + redactors.clear(); + for (var entry : configuration.getInputs().entrySet()) { + String name = entry.getValue().getRedactor(); + if (name != null) { + SemanticAuditInputRedactor redactor + = context.getRegistry().lookupByNameAndType(name, SemanticAuditInputRedactor.class); + if (redactor == null) { + throw new IllegalArgumentException("Unknown semantic audit input redactor: " + name); + } + redactors.put(entry.getKey(), redactor); + } + } sinkErrors.clear(); for (String name : configuration.getSinks()) { SemanticAuditSink sink = switch (name) { @@ -238,6 +253,11 @@ public final class SemanticAuditService extends ServiceSupport { status.put("enabled", configuration.isEnabled()); status.put("decisionsEnabled", configuration.isCapturing()); status.put("experts", configuration.getExperts()); + Map<String, Object> inputs = new LinkedHashMap<>(); + configuration.getInputs().forEach((expert, input) -> inputs.put(expert, + Map.of("enabled", input.isEnabled(), "maxChars", input.getMaxChars(), + "redacted", input.getRedactor() != null))); + status.put("inputs", inputs); status.put("reader", configuration.getReader()); status.put("sinks", configuration.getSinks()); status.put("capacity", configuration.getCapacity()); @@ -361,10 +381,50 @@ 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 final Map<String, Object> inputFields = new LinkedHashMap<>(); private Invocation(Map<String, Object> fields, boolean enabled) { this.fields = fields; this.enabled = enabled; + if (enabled && configuration.getInput((String) fields.get("expert")).isEnabled()) { + inputFields.put("inputOmitted", "unavailable"); + } + } + + /** 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()) { + return; + } + SemanticAuditInputConfiguration policy = configuration.getInput((String) fields.get("expert")); + SemanticAuditInputRedactor redactor = redactors.get(fields.get("expert")); + inputFields.clear(); + if (policy.getRedactor() != null && redactor == null) { + inputFields.put("inputOmitted", "redaction_failed"); + return; + } + Object snapshot; + try { + snapshot = SemanticAuditInputSnapshot.copy(input, policy.getMaxChars()); + } catch (Exception | AssertionError | LinkageError failure) { + inputFields.put("inputOmitted", "snapshot_limit_or_unsupported_type"); + return; + } + if (redactor != null) { + try { + snapshot = redactor.redact(snapshot); + if (snapshot == null) { + inputFields.put("inputOmitted", "redacted"); + return; + } + snapshot = SemanticAuditInputSnapshot.copy(snapshot, policy.getMaxChars()); + } catch (Exception | AssertionError | LinkageError failure) { + inputFields.put("inputOmitted", "redaction_failed"); + return; + } + inputFields.put("inputRedacted", true); + } + inputFields.put("input", snapshot); } public String getEventId() { @@ -404,7 +464,8 @@ public final class SemanticAuditService extends ServiceSupport { } } if (enabled) { - publish(record); + fields.putAll(inputFields); + publish(inputFields.isEmpty() ? record : SemanticAuditRecord.fromMap(fields)); } } } diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/yaml/SemanticDefinitionDeserializer.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/yaml/SemanticDefinitionDeserializer.java index d09e5be2b555..c102da9d2e6b 100644 --- a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/yaml/SemanticDefinitionDeserializer.java +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/yaml/SemanticDefinitionDeserializer.java @@ -31,6 +31,7 @@ import org.apache.camel.dsl.yaml.common.YamlDeserializerSupport; import org.apache.camel.dsl.yaml.common.exception.InvalidNodeTypeException; import org.apache.camel.dsl.yaml.common.exception.YamlDeserializationException; import org.apache.camel.semantic.SemanticAuditConfiguration; +import org.apache.camel.semantic.SemanticAuditInputConfiguration; import org.apache.camel.semantic.SemanticEvaluation; import org.apache.camel.semantic.SemanticEvaluationBuilder; import org.apache.camel.semantic.SemanticEvaluations; @@ -181,13 +182,26 @@ public class SemanticDefinitionDeserializer extends YamlDeserializerSupport impl throw new YamlDeserializationException(node, "Unknown semantic audit option"); } Map<String, Boolean> experts = new LinkedHashMap<>(); + Map<String, SemanticAuditInputConfiguration> inputs = new LinkedHashMap<>(); if (values.containsKey("experts")) { fields(values.get("experts"), "audit experts").forEach((name, expert) -> { Map<String, Node> settings = fields(expert, "audit expert"); - if (!settings.keySet().equals(Set.of("enabled"))) { - throw new YamlDeserializationException(expert, "Audit expert requires only enabled"); + if (!settings.containsKey("enabled") || !Set.of("enabled", "input").containsAll(settings.keySet())) { + throw new YamlDeserializationException(expert, "Audit expert requires enabled and optional input"); } experts.put(name, auditBoolean(context, settings.get("enabled"))); + if (settings.containsKey("input")) { + Map<String, Node> input = fields(settings.get("input"), "audit input"); + if (!Set.of("enabled", "maxChars", "redactor").containsAll(input.keySet())) { + throw new YamlDeserializationException(settings.get("input"), "Unknown audit input option"); + } + inputs.put(name, new SemanticAuditInputConfiguration( + input.containsKey("enabled") && auditBoolean(context, input.get("enabled")), + input.containsKey("maxChars") + ? auditCapacity(context, input, "maxChars") + : SemanticAuditInputConfiguration.DEFAULT_MAX_CHARS, + auditText(context, input, "redactor", null))); + } }); } List<String> sinks = values.containsKey("sinks") @@ -197,7 +211,7 @@ public class SemanticDefinitionDeserializer extends YamlDeserializerSupport impl return new SemanticAuditConfiguration( values.containsKey("enabled") && auditBoolean(context, values.get("enabled")), experts, sinks, auditText(context, values, "reader", "memory"), - auditCapacity(context, values, "capacity"), auditCapacity(context, values, "queueCapacity")); + auditCapacity(context, values, "capacity"), auditCapacity(context, values, "queueCapacity"), inputs); } private static int auditCapacity(CamelContext context, Map<String, Node> values, String key) { @@ -265,10 +279,22 @@ public class SemanticDefinitionDeserializer extends YamlDeserializerSupport impl public static class AuditSchema { } - @YamlType(properties = { @YamlProperty(name = "enabled", type = "boolean", required = true) }) + @YamlType(properties = { + @YamlProperty(name = "enabled", type = "boolean", required = true), + @YamlProperty(name = "input", + type = "object:org.apache.camel.semantic.yaml.SemanticDefinitionDeserializer$AuditInputSchema") + }) public static class AuditExpertSchema { } + @YamlType(properties = { + @YamlProperty(name = "enabled", type = "boolean"), + @YamlProperty(name = "maxChars", type = "integer"), + @YamlProperty(name = "redactor", type = "string") + }) + public static class AuditInputSchema { + } + private static SemanticEvaluation readEvaluation( CamelContext context, String name, String expert, Node definition, Map<String, Node> values) { String description = "semantic evaluation '" + name + "'"; 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 new file mode 100644 index 000000000000..1deb2ee83e0d --- /dev/null +++ b/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/SemanticAuditInputTest.java @@ -0,0 +1,355 @@ +/* + * 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.semantic; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.console.DevConsole; +import org.apache.camel.dsl.yaml.common.YamlDeserializationContext; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.language.semantic.SemanticLanguage; +import org.apache.camel.semantic.internal.SemanticAuditInputSnapshot; +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.util.json.JsonObject; +import org.apache.camel.util.json.Jsoner; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.snakeyaml.engine.v2.api.LoadSettings; +import org.snakeyaml.engine.v2.api.lowlevel.Compose; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class SemanticAuditInputTest { + @Test + void captureIsPerExpertAndNeverSentToObservers() throws Exception { + try (var context = new DefaultCamelContext()) { + var expert = new FixedSemanticExpert(); + context.getRegistry().bind("security", expert); + context.getRegistry().bind("decisions", expert); + AtomicInteger observed = new AtomicInteger(); + context.getRegistry().bind("observer", new SemanticObserver() { + public Observation started(SemanticAuditRecord record, Object parent) { + assertThat(record.toMap()).doesNotContainKeys("input", "inputOmitted", "inputRedacted"); + return completed -> { + assertThat(completed.toMap()).doesNotContainKeys("input", "inputOmitted", "inputRedacted"); + observed.incrementAndGet(); + }; + } + }); + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(true, new SemanticAuditInputConfiguration(true, 4096, null))); + context.start(); + var language = (SemanticLanguage) context.resolveLanguage("semantic"); + String input = "first line\nsecond line " + "x".repeat(300); + language.evaluate(evaluation("security"), input); + language.evaluate(evaluation("decisions"), "private"); + assertThatThrownBy(() -> audit.configure("test", configuration(true, SemanticAuditInputConfiguration.DISABLED))) + .hasMessageContaining("restart"); + audit.stop(); + assertThat(records(audit)).hasSize(2); + assertThat(records(audit).get(0).toMap()).doesNotContainKey("input"); + assertThat(records(audit).get(1).toMap()).containsEntry("input", input); + assertThat(observed).hasValue(2); + assertThat(audit.status()).containsEntry("observerErrors", 0L); + assertThat(expert.states).containsExactly(input, "private"); + } + } + + @ParameterizedTest + @ValueSource(booleans = { true, false }) + void captureDoesNotEnableAuditingAndDefaultsToDisabled(boolean auditing) throws Exception { + try (var context = new DefaultCamelContext()) { + context.getRegistry().bind("security", new FixedSemanticExpert()); + AtomicInteger redacted = new AtomicInteger(); + context.getRegistry().bind("redactor", (SemanticAuditInputRedactor) input -> { + redacted.incrementAndGet(); + return input; + }); + context.getRegistry().bind("observer", new SemanticObserver() { + public Observation started(SemanticAuditRecord record, Object parent) { + return completed -> assertThat(completed.toMap()).doesNotContainKey("input"); + } + }); + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(auditing, auditing + ? SemanticAuditInputConfiguration.DISABLED + : new SemanticAuditInputConfiguration(true, 100, "redactor"))); + context.start(); + ((SemanticLanguage) context.resolveLanguage("semantic")).evaluate(evaluation("security"), "private"); + audit.stop(); + assertThat(redacted).hasValue(0); + assertThat(records(audit)).hasSize(auditing ? 1 : 0) + .allSatisfy(record -> assertThat(record.toMap()).doesNotContainKeys("input", "inputOmitted")); + } + } + + @Test + void selectedStateIsCapturedBeforeProviderMutationAndLinkedThroughConsole() throws Exception { + try (var context = new DefaultCamelContext()) { + context.getRegistry().bind("security", new StructuredExpert()); + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(true, new SemanticAuditInputConfiguration(true, 100, null))); + SemanticEvaluations.get(context).replace("definitions", Map.of( + "a", new SemanticEvaluation("check", "security", "${header.selected}", Map.of()), + "b", new SemanticEvaluation("check", "security", "${header.selected}", Map.of()))); + context.start(); + var exchange = new DefaultExchange(context); + Map<String, Object> input = new LinkedHashMap<>(Map.of("messages", new ArrayList<>(List.of("hello")))); + exchange.getMessage().setBody("not selected"); + exchange.getMessage().setHeader("selected", input); + context.resolveLanguage("semantic").createExpression("refs:a,b").evaluate(exchange, Map.class); + SemanticAuditDecision decision = new SemanticAuditDecision(); + decision.setFields(Map.of("action", "allow")); + decision.process(exchange); + audit.stop(); + assertThat(input).containsEntry("mutated", true); + assertThat(records(audit)).hasSize(3); + assertThat(records(audit).get(0).toMap()).doesNotContainKey("input"); + for (var record : records(audit).subList(1, 3)) { + assertThat(record.toMap()).containsEntry("input", Map.of("messages", List.of("hello"))); + var restored = SemanticAuditRecord.fromMap((JsonObject) Jsoner.deserialize(Jsoner.serialize(record.toMap()))); + assertThat(restored.toMap()).containsEntry("input", record.toMap().get("input")); + assertThatThrownBy(() -> ((Map<String, Object>) restored.toMap().get("input")).put("secret", "x")) + .isInstanceOf(UnsupportedOperationException.class); + } + var console = new SemanticAuditConsole(); + console.setCamelContext(context); + var response = (JsonObject) console.call(DevConsole.MediaType.JSON, + Map.of("eventId", records(audit).get(0).getEventId())); + assertThat(Jsoner.serialize(response)).contains("hello").doesNotContain("not selected", "mutated"); + } + } + + @ParameterizedTest + @ValueSource(strings = { "replace", "omit", "throw", "oversize", "mutate" }) + void redactionIsAppliedBeforeStorageAndFailsClosed(String behavior) throws Exception { + try (var context = new DefaultCamelContext()) { + var expert = new StructuredExpert(); + context.getRegistry().bind("security", expert); + context.getRegistry().bind("redactor", (SemanticAuditInputRedactor) input -> { + return switch (behavior) { + case "replace" -> Map.of("prompt", "[redacted]"); + case "omit" -> null; + case "throw" -> throw new IllegalStateException("secret diagnostic"); + case "mutate" -> ((Map<String, Object>) input).put("prompt", "changed"); + default -> "x".repeat(101); + }; + }); + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(true, new SemanticAuditInputConfiguration(true, 100, "redactor"))); + context.start(); + Map<String, Object> input = new LinkedHashMap<>(Map.of("prompt", "secret")); + var result = ((SemanticLanguage) context.resolveLanguage("semantic")) + .evaluate(new SemanticEvaluation("check", "security", null, Map.of()), input); + assertThat(result.getValue()).isEqualTo(true); + assertThat(input).containsEntry("prompt", "secret"); + audit.stop(); + var record = records(audit).get(0); + assertThat(Jsoner.serialize(record.toMap())).doesNotContain("secret", "changed"); + if (behavior.equals("replace")) { + assertThat(record.toMap()).containsEntry("input", Map.of("prompt", "[redacted]")) + .containsEntry("inputRedacted", true); + } else { + assertThat(record.toMap()).doesNotContainKey("input").containsEntry("inputOmitted", + behavior.equals("omit") ? "redacted" : "redaction_failed"); + } + } + } + + @Test + void oversizedOrUnsupportedInputDoesNotChangeEvaluationOutcome() throws Exception { + try (var context = new DefaultCamelContext()) { + context.getRegistry().bind("security", new StructuredExpert()); + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(true, new SemanticAuditInputConfiguration(true, 4, null))); + context.start(); + var language = (SemanticLanguage) context.resolveLanguage("semantic"); + for (Object input : List.of("too long", new LinkedHashMap<>(Map.of("x", new Object())))) { + assertThat(language.evaluate(new SemanticEvaluation("check", "security", null, Map.of()), input).getValue()) + .isEqualTo(true); + } + audit.stop(); + assertThat(records(audit)).hasSize(2).allSatisfy(record -> assertThat(record.toMap()).doesNotContainKey("input") + .containsEntry("inputOmitted", "snapshot_limit_or_unsupported_type")); + } + } + + @Test + void snapshotsBoundCharactersDepthAndNodesWithoutConsumingOrStringifyingObjects() { + assertThat(SemanticAuditInputSnapshot.copy("😀\n", 2)).isEqualTo("😀\n"); + assertThatThrownBy(() -> SemanticAuditInputSnapshot.copy("😀x", 1)).isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> SemanticAuditInputSnapshot.copy(Double.NaN, 100)).isInstanceOf(IllegalArgumentException.class); + List<Object> cycle = new ArrayList<>(); + cycle.add(cycle); + assertThatThrownBy(() -> SemanticAuditInputSnapshot.copy(cycle, 100)).isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> SemanticAuditInputSnapshot.copy(Collections.nCopies(1000, ""), 100)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> SemanticAuditInputSnapshot.copy(new Object() { + public String toString() { + throw new AssertionError("must not stringify"); + } + }, 100)).isInstanceOf(IllegalArgumentException.class); + } + + @ParameterizedTest + @ValueSource(strings = { "yaml", "xml", "java" }) + void allDslsConfigureInputAndFreezeItUntilRestart(String dsl) throws Exception { + try (var context = new DefaultCamelContext()) { + context.getRegistry().bind("security", new FixedSemanticExpert()); + context.getRegistry().bind("redactor", (SemanticAuditInputRedactor) input -> "safe"); + if (dsl.equals("yaml")) { + yaml(context, "{enabled: true, maxChars: 42, redactor: redactor}"); + } else if (dsl.equals("xml")) { + context.build(); + PluginHelper.getRoutesLoader(context).loadRoutes(ResourceHelper.fromString("audit.xml", """ + <semantic xmlns="http://camel.apache.org/schema/semantic";> + <audit enabled="true"><expert name="security" enabled="true" + inputEnabled="true" inputMaxChars="42" inputRedactor="redactor"/></audit> + </semantic> + """)); + } else { + context.addRoutes(new RouteBuilder() { + public void configure() { + SemanticEvaluationsBuilder.semanticEvaluations(this) + .audit(SemanticAuditInputTest.configuration(true, + new SemanticAuditInputConfiguration(true, 42, "redactor"))) + .register(); + } + }); + } + context.start(); + var audit = SemanticAuditService.get(context); + assertThat(audit.getConfiguration().getInput("security")) + .isEqualTo(new SemanticAuditInputConfiguration(true, 42, "redactor")); + ((SemanticLanguage) context.resolveLanguage("semantic")).evaluate(evaluation("security"), "secret"); + assertThatThrownBy(() -> audit.validateConfiguration("different", + configuration(true, SemanticAuditInputConfiguration.DISABLED))) + .isInstanceOf(IllegalArgumentException.class); + audit.stop(); + assertThat(records(audit).get(0).toMap()).containsEntry("input", "safe"); + } + } + + @ParameterizedTest + @ValueSource(strings = { "{enabled: maybe}", "{enabled: true, maxChars: 0}", "{maxChars: abc}", "{unknown: true}", "true" }) + void invalidYamlInputSettingsIncludeSourcePosition(String input) throws Exception { + try (var context = new DefaultCamelContext()) { + assertThatThrownBy(() -> yaml(context, input)).hasMessageContaining("audit.yaml"); + } + } + + @Test + void persistedRecordsRejectOversizedAndContradictoryInput() { + Map<String, Object> fields = new LinkedHashMap<>( + Map.of("schemaVersion", 1, "eventId", "event", + "category", "evaluation", "timestamp", "2026-10-10T10:00:00Z")); + fields.put("input", "x".repeat(SemanticAuditInputConfiguration.MAX_CHARS + 1)); + assertThatThrownBy(() -> SemanticAuditRecord.fromMap(fields)).isInstanceOf(IllegalArgumentException.class); + fields.put("input", "hello"); + fields.put("inputOmitted", "redacted"); + assertThatThrownBy(() -> SemanticAuditRecord.fromMap(fields)).isInstanceOf(IllegalArgumentException.class); + fields.remove("inputOmitted"); + fields.put("category", "decision"); + assertThatThrownBy(() -> SemanticAuditRecord.fromMap(fields)).isInstanceOf(IllegalArgumentException.class); + fields.put("category", "evaluation"); + fields.put("inputRedacted", "yes"); + assertThatThrownBy(() -> SemanticAuditRecord.fromMap(fields)).isInstanceOf(IllegalArgumentException.class); + } + + @Test + void invocationBeforeStartupNeverBypassesAConfiguredRedactor() throws Exception { + try (var context = new DefaultCamelContext()) { + context.getRegistry().bind("redactor", (SemanticAuditInputRedactor) input -> "safe"); + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(true, new SemanticAuditInputConfiguration(true, 100, "redactor"))); + var invocation = audit.begin(null, null, evaluation("security"), null, "security", null, null, null); + invocation.captureInput("private"); + context.start(); + invocation.complete("success", "evaluation_completed", null); + audit.stop(); + assertThat(records(audit).get(0).toMap()).doesNotContainKey("input").containsEntry("inputOmitted", + "redaction_failed"); + } + } + + @Test + void unknownRedactorFailsStartup() throws Exception { + try (var context = new DefaultCamelContext()) { + var audit = SemanticAuditService.get(context); + audit.configure("test", configuration(true, new SemanticAuditInputConfiguration(true, 100, "missing"))); + audit.activate(); + assertThatThrownBy(context::start).hasMessageContaining("Unknown semantic audit input redactor: missing"); + } + } + + private static SemanticAuditConfiguration configuration(boolean enabled, SemanticAuditInputConfiguration input) { + return new SemanticAuditConfiguration( + enabled, Map.of("security", enabled, "decisions", enabled), + List.of("memory"), "memory", 100, 100, Map.of("security", input)); + } + + private static SemanticEvaluation evaluation(String expert) { + return new SemanticEvaluation("boolean", expert, null, Map.of()); + } + + private static List<SemanticAuditRecord> records(SemanticAuditService audit) throws Exception { + return audit.getReader().query(new SemanticAuditQuery(Map.of(), null, null, 200)).getRecords(); + } + + private static void yaml(DefaultCamelContext context, String input) throws Exception { + String yaml + = "- semantic:\n audit:\n enabled: true\n experts:\n security:\n enabled: true\n input: " + + input; + var settings = LoadSettings.builder().setLabel("audit.yaml").build(); + try (var deserialization = new YamlDeserializationContext(settings)) { + deserialization.setCamelContext(context); + deserialization.setResource(ResourceHelper.fromString("audit.yaml", yaml)); + deserialization.start(); + deserialization.preParse(new Compose(settings).composeString(yaml).orElseThrow()); + } + } + + @SemanticExpert(name = "structured", description = "Test", provider = "test", artifactId = "test", + operations = @SemanticOperation(name = "check", description = "Check", inputTypes = { + SemanticExpert.InputType.TEXT, SemanticExpert.InputType.STRUCTURED }, + inputRequirements = "Text or structured state", + resultType = SemanticExpert.ResultType.BOOLEAN, + resultMeaning = "Test result")) + static class StructuredExpert implements SemanticAdapter { + public void validate(SemanticEvaluation evaluation) { + SemanticCapabilities.from(getClass()).validate(evaluation); + } + + public SemanticResult evaluate(SemanticEvaluation evaluation, Object state) { + if (state instanceof Map<?, ?> map) { + ((Map<String, Object>) map).put("mutated", true); + } + return new SemanticResult(true, null, null, null, Map.of()); + } + } +} diff --git a/design/semantic-audit.adoc b/design/semantic-audit.adoc index 79a3a604127a..3ae509947434 100644 --- a/design/semantic-audit.adoc +++ b/design/semantic-audit.adoc @@ -1,7 +1,7 @@ = Semantic audit and observation proposal -Status: proposed for review in CAMEL-25499 / PR #27649. This document describes the implementation being -reviewed; it does not assert prior upstream design agreement. +Status: the base implementation was merged in CAMEL-25499 / PR #27649. The input capture extension +described below is proposed as a follow-up; it does not assert prior upstream design agreement. == Scope and related work @@ -54,3 +54,21 @@ lookup. The TUI maintains its own browsing state. Its audit MCP tool queries the changing the screen, and is added to AI sessions only when the selected integration exposes audit history. The memory reader filters a consistent snapshot outside its append lock and binds cursors to a canonical SHA-256 query fingerprint. Backend failures expose a stable error code without backend messages. + +== Opt-in input capture + +The follow-up retains the selected evaluation state only when that expert explicitly enables +`input.enabled`; auditing alone never enables input retention. Capture applies to direct, named, +batch and console evaluations. A decision refers to its evaluation evidence rather than duplicating +submitted input. Input is excluded from observer callbacks even when audit capture is enabled. + +The snapshot is taken before provider validation or execution and bounded by `maxChars` (4096 by +default), depth and node count. Unsupported or oversized state is omitted with a stable reason, +without truncation or arbitrary object conversion. An optional `SemanticAuditInputRedactor` registry +bean receives the immutable snapshot; its output is bounded again. Redaction failure omits input +without changing the evaluation result or retaining exception messages. + +Captured input follows the existing sink/reader contract and backend retention policy, including log +output when the log sink is configured. It is available through the console, CLI, TUI and MCP. There +is no separate input repository or telemetry payload path. Input settings use the same DSL authority +and restart requirement as the other audit settings; existing records without input remain readable. diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-canonical.json b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-canonical.json index 8a0cfe6ff2e1..1b75a14c0379 100644 --- a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-canonical.json +++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-canonical.json @@ -13619,10 +13619,28 @@ "properties" : { "enabled" : { "type" : "boolean" + }, + "input" : { + "$ref" : "#/items/definitions/org.apache.camel.semantic.yaml.SemanticDefinitionDeserializer$AuditInputSchema" } }, "required" : [ "enabled" ] }, + "org.apache.camel.semantic.yaml.SemanticDefinitionDeserializer$AuditInputSchema" : { + "type" : "object", + "additionalProperties" : false, + "properties" : { + "enabled" : { + "type" : "boolean" + }, + "maxChars" : { + "type" : "integer" + }, + "redactor" : { + "type" : "string" + } + } + }, "org.apache.camel.semantic.yaml.SemanticDefinitionDeserializer$AuditSchema" : { "type" : "object", "additionalProperties" : false, diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-model.json b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-model.json index 7472732229c7..47209ce3178f 100644 --- a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-model.json +++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl-model.json @@ -20573,6 +20573,22 @@ "name" : "enabled", "type" : "boolean", "required" : true + }, { + "name" : "input", + "type" : "object", + "ref" : "semanticDefinitionDeserializer$AuditInputSchema" + } ] + }, + "semanticDefinitionDeserializer$AuditInputSchema" : { + "children" : [ { + "name" : "enabled", + "type" : "boolean" + }, { + "name" : "maxChars", + "type" : "integer" + }, { + "name" : "redactor", + "type" : "string" } ] }, "semanticDefinitionDeserializer$AuditSchema" : { diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json index 2314c56c162b..8722a3fb7509 100644 --- a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json +++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json @@ -16519,10 +16519,28 @@ "properties" : { "enabled" : { "type" : "boolean" + }, + "input" : { + "$ref" : "#/items/definitions/org.apache.camel.semantic.yaml.SemanticDefinitionDeserializer$AuditInputSchema" } }, "required" : [ "enabled" ] }, + "org.apache.camel.semantic.yaml.SemanticDefinitionDeserializer$AuditInputSchema" : { + "type" : "object", + "additionalProperties" : false, + "properties" : { + "enabled" : { + "type" : "boolean" + }, + "maxChars" : { + "type" : "integer" + }, + "redactor" : { + "type" : "string" + } + } + }, "org.apache.camel.semantic.yaml.SemanticDefinitionDeserializer$AuditSchema" : { "type" : "object", "additionalProperties" : false, diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/test/java/org/apache/camel/dsl/yaml/SemanticEvaluationTest.java b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/test/java/org/apache/camel/dsl/yaml/SemanticEvaluationTest.java index 31609caf5096..8a5774ba4b4e 100644 --- a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/test/java/org/apache/camel/dsl/yaml/SemanticEvaluationTest.java +++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/test/java/org/apache/camel/dsl/yaml/SemanticEvaluationTest.java @@ -112,8 +112,9 @@ class SemanticEvaluationTest extends YamlTestSupport { } } - @Test - void auditSchemaAndDecisionBeanRecordOnlyTheEnabledExpert() throws Exception { + @ParameterizedTest + @ValueSource(booleans = { true, false }) + void auditSchemaAndDecisionBeanRecordOnlyTheEnabledExpert(boolean captureInput) throws Exception { loadRoutes(""" - beans: - name: security @@ -132,8 +133,8 @@ class SemanticEvaluationTest extends YamlTestSupport { audit: enabled: true experts: - security: {enabled: true} - decisions: {enabled: false} + security: {enabled: true, input: {enabled: %s, maxChars: 4096}} + decisions: {enabled: false, input: {enabled: true}} sinks: [memory] reader: memory capacity: 10 @@ -161,7 +162,7 @@ class SemanticEvaluationTest extends YamlTestSupport { - process: ref: recordBlocked - to: mock:blocked - """); + """.formatted(captureInput)); context.start(); context.getEndpoint("mock:blocked", MockEndpoint.class).expectedMessageCount(1); try (var template = context.createProducerTemplate()) { @@ -176,6 +177,11 @@ class SemanticEvaluationTest extends YamlTestSupport { assertThat(records.get(0).text("action")).isEqualTo("block"); assertThat(records.get(0).toMap().get("evidence")).isEqualTo(List.of(records.get(1).getEventId())); assertThat(records.get(1).getExpert()).isEqualTo("security"); + if (captureInput) { + assertThat(records.get(1).toMap()).containsEntry("input", "injection attempt"); + } else { + assertThat(records.get(1).toMap()).doesNotContainKey("input"); + } } private static String declarations(String state) {
