This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a5e0e20ffbcd CAMEL-25458: camel-a2a - improve how the consumer derives
the caller profile (#28155)
a5e0e20ffbcd is described below
commit a5e0e20ffbcd1d38c10bd219dba59e47d7737d91
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Oct 20 21:00:31 2026 +0200
CAMEL-25458: camel-a2a - improve how the consumer derives the caller
profile (#28155)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/a2a-consumer.adoc | 24 +-
.../camel-a2a/src/main/docs/a2a-consumer.adoc | 24 +-
.../apache/camel/component/a2a/A2AConstants.java | 1 +
.../apache/camel/component/a2a/A2AConsumer.java | 4 +
.../camel/component/a2a/auth/A2AAuthHandler.java | 17 +-
.../camel/component/a2a/auth/A2AUserProfile.java | 48 +++-
.../component/a2a/auth/BearerTokenExtractor.java | 12 +-
.../a2a/A2AConsumerCallerProfileTest.java | 282 +++++++++++++++++++++
.../component/a2a/auth/A2AAuthHandlerTest.java | 140 +++++++++-
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 12 +
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 +
11 files changed, 566 insertions(+), 10 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/a2a-consumer.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/a2a-consumer.adoc
index 4463f4c63d8b..8021dae00172 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/a2a-consumer.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/a2a-consumer.adoc
@@ -240,7 +240,7 @@ The agent card must declare the security scheme:
}
----
-When authentication succeeds, the `CamelA2AUserProfile` exchange property is
populated with a `Map<String, Object>` containing the authenticated user's
profile (subject, issuer, scopes, claims). When using OAuth/OIDC, the full
`OAuthTokenValidationResult` is also available as the
`CamelOAuthTokenValidationResult` exchange property.
+When authentication succeeds, the `CamelA2AUserProfile` exchange property is
populated with a `Map<String, Object>` containing the authenticated user's
profile (subject, issuer, scopes, claims). When using OAuth/OIDC, the full
`OAuthTokenValidationResult` is also available as the
`CamelOAuthTokenValidationResult` exchange property. See
<<_caller_profile,Caller Profile>> for how the profile is derived.
==== API Key Authentication
@@ -332,6 +332,28 @@ The agent card must declare an `http` bearer security
scheme:
NOTE: The agent card endpoint (`GET /.well-known/agent-card.json`) is always
public — clients need it for discovery before they can authenticate.
+==== Caller Profile
+
+The consumer derives the caller profile from the security schemes it
validated, and sets it as the `CamelA2AUserProfile` exchange property of the
exchange the route processes:
+
+[cols="1,3"]
+|===
+| Scheme | Caller profile
+
+| OAuth 2.0, OpenID Connect, and HTTP bearer with `oauthProfile`
+| The validated token: `scheme`, `subject`, `issuer`, `audience`, `scopes` and
`claims`. The `OAuthTokenValidationResult` of the token is also set as the
`CamelOAuthTokenValidationResult` exchange property. A token must identify the
caller with its `sub` claim or, for a token without `sub`, with the `username`
that a token introspection response gives, or with its `client_id` or `azp`
claim, such as a client credentials token. A token that has none of these
claims is rejected.
+
+| HTTP bearer with the static `bearerToken` option
+| Only the scheme (`bearer`): every caller that sends the token is the same
caller.
+
+| API key
+| Only the scheme (`apiKey`): every caller that sends the key is the same
caller.
+|===
+
+When a security requirement combines several schemes, such as an API key and
an OAuth token, the caller profile is the one of the token, whatever the order
of the schemes in the requirement.
+
+Tasks and push notification configurations belong to the caller that created
the task: the issuer and subject of a token, or for a token without subject the
issuer and its user name, else its client. Callers that authenticate with the
same API key or static bearer token share their tasks.
+
=== SSE Streaming with `${a2a:emit()}`
Agents can emit progressive status updates during processing using the
`${a2a:emit()}` Simple language function. Clients receive these as Server-Sent
Events (SSE).
diff --git a/components/camel-ai/camel-a2a/src/main/docs/a2a-consumer.adoc
b/components/camel-ai/camel-a2a/src/main/docs/a2a-consumer.adoc
index 4463f4c63d8b..8021dae00172 100644
--- a/components/camel-ai/camel-a2a/src/main/docs/a2a-consumer.adoc
+++ b/components/camel-ai/camel-a2a/src/main/docs/a2a-consumer.adoc
@@ -240,7 +240,7 @@ The agent card must declare the security scheme:
}
----
-When authentication succeeds, the `CamelA2AUserProfile` exchange property is
populated with a `Map<String, Object>` containing the authenticated user's
profile (subject, issuer, scopes, claims). When using OAuth/OIDC, the full
`OAuthTokenValidationResult` is also available as the
`CamelOAuthTokenValidationResult` exchange property.
+When authentication succeeds, the `CamelA2AUserProfile` exchange property is
populated with a `Map<String, Object>` containing the authenticated user's
profile (subject, issuer, scopes, claims). When using OAuth/OIDC, the full
`OAuthTokenValidationResult` is also available as the
`CamelOAuthTokenValidationResult` exchange property. See
<<_caller_profile,Caller Profile>> for how the profile is derived.
==== API Key Authentication
@@ -332,6 +332,28 @@ The agent card must declare an `http` bearer security
scheme:
NOTE: The agent card endpoint (`GET /.well-known/agent-card.json`) is always
public — clients need it for discovery before they can authenticate.
+==== Caller Profile
+
+The consumer derives the caller profile from the security schemes it
validated, and sets it as the `CamelA2AUserProfile` exchange property of the
exchange the route processes:
+
+[cols="1,3"]
+|===
+| Scheme | Caller profile
+
+| OAuth 2.0, OpenID Connect, and HTTP bearer with `oauthProfile`
+| The validated token: `scheme`, `subject`, `issuer`, `audience`, `scopes` and
`claims`. The `OAuthTokenValidationResult` of the token is also set as the
`CamelOAuthTokenValidationResult` exchange property. A token must identify the
caller with its `sub` claim or, for a token without `sub`, with the `username`
that a token introspection response gives, or with its `client_id` or `azp`
claim, such as a client credentials token. A token that has none of these
claims is rejected.
+
+| HTTP bearer with the static `bearerToken` option
+| Only the scheme (`bearer`): every caller that sends the token is the same
caller.
+
+| API key
+| Only the scheme (`apiKey`): every caller that sends the key is the same
caller.
+|===
+
+When a security requirement combines several schemes, such as an API key and
an OAuth token, the caller profile is the one of the token, whatever the order
of the schemes in the requirement.
+
+Tasks and push notification configurations belong to the caller that created
the task: the issuer and subject of a token, or for a token without subject the
issuer and its user name, else its client. Callers that authenticate with the
same API key or static bearer token share their tasks.
+
=== SSE Streaming with `${a2a:emit()}`
Agents can emit progressive status updates during processing using the
`${a2a:emit()}` Simple language function. Clients receive these as Server-Sent
Events (SSE).
diff --git
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConstants.java
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConstants.java
index ad7a8c1f5059..09e7e3b45202 100644
---
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConstants.java
+++
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConstants.java
@@ -47,6 +47,7 @@ public final class A2AConstants {
javaType =
"org.apache.camel.component.a2a.streaming.A2AStreamEmitter")
public static final String STREAM_EMITTER = "CamelA2AStreamEmitter";
public static final String USER_PROFILE = "CamelA2AUserProfile";
+ public static final String OAUTH_TOKEN_VALIDATION_RESULT =
"CamelOAuthTokenValidationResult";
public static final String RESPONSE_TASK = "CamelA2AResponseTask";
@Metadata(label = "common", description = "Context ID filter for task
listing", javaType = "String")
public static final String LIST_CONTEXT_ID = "CamelA2AListContextId";
diff --git
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConsumer.java
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConsumer.java
index 9209714e7b89..1f7f35d2eda0 100644
---
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConsumer.java
+++
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AConsumer.java
@@ -1171,6 +1171,8 @@ public class A2AConsumer extends DefaultConsumer {
Exchange processorExchange = getEndpoint().createExchange();
processorExchange.setProperty(A2AConstants.USER_PROFILE,
exchange.getProperty(A2AConstants.USER_PROFILE));
+
processorExchange.setProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT,
+
exchange.getProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT));
copyNegotiatedExtensions(exchange, processorExchange);
messageSendOperation.parseRequest(processorExchange, request,
getEndpoint().getConfiguration().getDataFormat());
@@ -1484,6 +1486,8 @@ public class A2AConsumer extends DefaultConsumer {
QueueStreamEmitter emitter, Exchange requestExchange) {
Exchange processorExchange = getEndpoint().createExchange();
processorExchange.setProperty(A2AConstants.USER_PROFILE,
requestExchange.getProperty(A2AConstants.USER_PROFILE));
+
processorExchange.setProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT,
+
requestExchange.getProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT));
copyNegotiatedExtensions(requestExchange, processorExchange);
A2ADataFormat dataFormat =
getEndpoint().getConfiguration().getDataFormat();
if (dataFormat == A2ADataFormat.POJO) {
diff --git
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AAuthHandler.java
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AAuthHandler.java
index 1bb89d78b84a..53007877ab90 100644
---
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AAuthHandler.java
+++
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AAuthHandler.java
@@ -29,6 +29,7 @@ import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.a2a.A2AConfiguration;
+import org.apache.camel.component.a2a.A2AConstants;
import org.apache.camel.component.a2a.model.AgentCard;
import org.apache.camel.component.a2a.model.SecurityRequirement;
import org.apache.camel.component.a2a.model.SecurityScheme;
@@ -122,6 +123,9 @@ public class A2AAuthHandler {
try {
return validateRequirement(exchange, card, requirement);
} catch (SecurityException e) {
+ // a token validated for a requirement that is not satisfied,
such as one missing a required scope,
+ // is not the credential the caller is authenticated with
+
exchange.removeProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT);
lastFailure = e;
}
}
@@ -219,11 +223,22 @@ public class A2AAuthHandler {
}
}
+ /**
+ * The profile of a requirement with several schemes is the one that
identifies the caller, such as the profile of a
+ * validated token, whatever the order of the schemes: the profile of a
shared credential, such as an API key, only
+ * names its scheme.
+ */
private static A2AUserProfile selectProfile(A2AUserProfile current,
A2AUserProfile candidate) {
if (current == null) {
return candidate;
}
- if (candidate != null && candidate.ownerKey() != null &&
current.ownerKey() == null) {
+ if (candidate == null) {
+ return current;
+ }
+ if (candidate.identifiesCaller() && !current.identifiesCaller()) {
+ return candidate;
+ }
+ if (candidate.ownerKey() != null && current.ownerKey() == null) {
return candidate;
}
return current;
diff --git
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AUserProfile.java
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AUserProfile.java
index 0dc56db45169..193c1e82f5da 100644
---
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AUserProfile.java
+++
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/A2AUserProfile.java
@@ -29,6 +29,11 @@ import org.apache.camel.spi.OAuthTokenValidationResult;
*/
public final class A2AUserProfile {
+ // the claim that names the resource owner in a token introspection
response (RFC 7662)
+ private static final String USERNAME_CLAIM = "username";
+ // the claims that name the client a token was issued to (RFC 9068 and
OpenID Connect)
+ private static final List<String> CLIENT_CLAIMS = List.of("client_id",
"azp");
+
private final String scheme;
private final String subject;
private final String issuer;
@@ -119,14 +124,53 @@ public final class A2AUserProfile {
return Collections.unmodifiableMap(map);
}
+ /**
+ * The key of the caller that owns the tasks it creates. A validated token
is keyed by its subject or, for a token
+ * without a subject, by the user that a token introspection response
names (the {@code username} claim), else by
+ * the client the token was issued to (the {@code client_id} or {@code
azp} claim). A profile without a token, such
+ * as one of an API key, is keyed by its scheme, which every caller with
that credential shares.
+ */
public String ownerKey() {
- if (issuer != null || subject != null) {
- return String.valueOf(issuer) + "|" + String.valueOf(subject);
+ if (hasSubject()) {
+ return issuer + "|" + subject;
+ }
+ String username = claim(USERNAME_CLAIM);
+ if (username != null) {
+ return "user|" + issuer + "|" + username;
+ }
+ String client = clientId();
+ if (client != null) {
+ return "client|" + issuer + "|" + client;
}
Map<String, Object> profile = asMap();
return profile.isEmpty() ? null : profile.toString();
}
+ /**
+ * Whether this profile identifies the caller, rather than only the scheme
of a shared credential.
+ */
+ boolean identifiesCaller() {
+ return hasSubject() || claim(USERNAME_CLAIM) != null || clientId() !=
null;
+ }
+
+ private boolean hasSubject() {
+ return subject != null && !subject.isBlank();
+ }
+
+ private String clientId() {
+ for (String name : CLIENT_CLAIMS) {
+ String value = claim(name);
+ if (value != null) {
+ return value;
+ }
+ }
+ return null;
+ }
+
+ private String claim(String name) {
+ return claims.get(name) instanceof String value && !value.isBlank() ?
value : null;
+ }
+
private static void putIfPresent(Map<String, Object> map, String key,
String value) {
if (value != null) {
map.put(key, value);
diff --git
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/BearerTokenExtractor.java
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/BearerTokenExtractor.java
index 849cbe1fae89..96070222ff41 100644
---
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/BearerTokenExtractor.java
+++
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/auth/BearerTokenExtractor.java
@@ -21,6 +21,7 @@ import java.security.MessageDigest;
import org.apache.camel.Exchange;
import org.apache.camel.component.a2a.A2AConfiguration;
+import org.apache.camel.component.a2a.A2AConstants;
import org.apache.camel.spi.OAuthTokenValidationResult;
import org.apache.camel.support.OAuthHelper;
import org.slf4j.Logger;
@@ -34,7 +35,6 @@ import org.slf4j.LoggerFactory;
*/
final class BearerTokenExtractor {
- static final String OAUTH_TOKEN_VALIDATION_RESULT =
"CamelOAuthTokenValidationResult";
private static final Logger LOG =
LoggerFactory.getLogger(BearerTokenExtractor.class);
private static final String BEARER_PREFIX = "Bearer ";
@@ -84,9 +84,15 @@ final class BearerTokenExtractor {
schemeName + " token validation failed: " +
result.getError());
}
- exchange.setProperty(OAUTH_TOKEN_VALIDATION_RESULT, result);
+ A2AUserProfile profile = A2AUserProfile.fromOAuthResult(schemeName,
result);
+ if (!profile.identifiesCaller()) {
+ // the caller would own its tasks together with every other caller
of such a token
+ throw new SecurityException(
+ schemeName + " token does not identify the caller: it has
no sub, username, client_id or azp claim");
+ }
+ exchange.setProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT,
result);
- return A2AUserProfile.fromOAuthResult(schemeName, result);
+ return profile;
}
private static A2AUserProfile validateViaStaticToken(
diff --git
a/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/A2AConsumerCallerProfileTest.java
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/A2AConsumerCallerProfileTest.java
new file mode 100644
index 000000000000..2a13c89ff0e1
--- /dev/null
+++
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/A2AConsumerCallerProfileTest.java
@@ -0,0 +1,282 @@
+/*
+ * 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.component.a2a;
+
+import java.io.InputStream;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.a2a.model.AgentCapabilities;
+import org.apache.camel.component.a2a.model.AgentCard;
+import org.apache.camel.component.a2a.model.ListTasksResponse;
+import org.apache.camel.component.a2a.model.Message;
+import org.apache.camel.component.a2a.model.SecurityRequirement;
+import org.apache.camel.component.a2a.model.SecurityScheme;
+import org.apache.camel.component.a2a.model.SendMessageRequest;
+import org.apache.camel.component.a2a.model.SendMessageResponse;
+import org.apache.camel.component.a2a.model.StreamResponse;
+import org.apache.camel.component.a2a.model.Task;
+import org.apache.camel.component.a2a.model.TaskState;
+import org.apache.camel.component.a2a.model.TaskStatus;
+import org.apache.camel.component.a2a.model.TaskStatusUpdateEvent;
+import org.apache.camel.component.a2a.model.TextPart;
+import org.apache.camel.spi.OAuthTokenValidationConfig;
+import org.apache.camel.spi.OAuthTokenValidationFactory;
+import org.apache.camel.spi.OAuthTokenValidationResult;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The caller profile of a request reaches the route, and the tasks of a
caller are its own.
+ */
+class A2AConsumerCallerProfileTest extends CamelTestSupport {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
+ .registerModule(new JavaTimeModule())
+ .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
+ .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("noopRestConsumerFactory", new
NoopRestConsumerFactory());
+ context.getRestConfiguration().setComponent("noopRestConsumerFactory");
+ return context;
+ }
+
+ @Test
+ void routeReceivesTheOAuthTokenValidationResult() throws Exception {
+ AtomicReference<Object> received = new AtomicReference<>();
+ A2AConsumer consumer = createConsumer(exchange -> {
+
received.set(exchange.getProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT));
+ exchange.getMessage().setBody("ok");
+ });
+ OAuthTokenValidationResult result = OAuthTokenValidationResult.valid(
+ "user-1", "https://issuer.example";, List.of("camel-a2a"),
List.of(), Map.of(), 0);
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.setProperty(A2AConstants.USER_PROFILE, Map.of("subject",
"user-1", "issuer", "https://issuer.example";));
+ exchange.setProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT,
result);
+ sendMessage(consumer, exchange, "msg-1");
+
+ assertThat(received.get()).isSameAs(result);
+ }
+
+ @Test
+ void routeReceivesTheOAuthTokenValidationResultWhenStreaming() throws
Exception {
+ AgentCapabilities capabilities = new AgentCapabilities();
+ capabilities.setStreaming(true);
+ A2AConfiguration config = new A2AConfiguration();
+
config.setAgentCard(AgentCard.builder().setCapabilities(capabilities).build());
+
+ AtomicReference<Object> received = new AtomicReference<>();
+ AtomicReference<A2AConsumer> consumerRef = new AtomicReference<>();
+ A2AConsumer consumer = createConsumer(config, exchange -> {
+
received.set(exchange.getProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT));
+ // complete the task, which ends the stream
+ String taskId =
exchange.getMessage().getHeader(A2AConstants.TASK_ID, String.class);
+
consumerRef.get().getEndpoint().getTaskStore().notifySubscribers(taskId,
+
StreamResponse.ofStatusUpdate(TaskStatusUpdateEvent.builder().taskId(taskId)
+ .status(new
TaskStatus(TaskState.COMPLETED)).build()));
+ });
+ consumerRef.set(consumer);
+ OAuthTokenValidationResult result = OAuthTokenValidationResult.valid(
+ "user-1", "https://issuer.example";, List.of("camel-a2a"),
List.of(), Map.of(), 0);
+
+ consumer.doStart();
+ try {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.setProperty(A2AConstants.USER_PROFILE,
+ Map.of("subject", "user-1", "issuer",
"https://issuer.example";));
+ exchange.setProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT,
result);
+
exchange.getMessage().setBody(OBJECT_MAPPER.writeValueAsString(sendMessageRequest("msg-stream")));
+
+ consumer.handleMessageStream(exchange);
+ // read the stream to its end: the route has then processed the
message
+ exchange.getMessage().getBody(InputStream.class).readAllBytes();
+ } finally {
+ consumer.doStop();
+ }
+
+ assertThat(received.get()).isSameAs(result);
+ }
+
+ @Test
+ void routeDoesNotReceiveTheTokenOfARequirementThatIsNotSatisfied() throws
Exception {
+ context.getRegistry().bind(OAuthTokenValidationFactory.FACTORY, new
StubOAuthTokenValidationFactory());
+
+ Map<String, SecurityScheme> schemes = new LinkedHashMap<>();
+ schemes.put("oauth", SecurityScheme.oauth2(null));
+ schemes.put("agentKey", SecurityScheme.apiKey("header",
"X-Agent-Key"));
+ A2AConfiguration config = new A2AConfiguration();
+ config.setValidateAuth(true);
+ config.setApiKey("secret-key");
+ config.setOauthProfile("myprofile");
+ config.setAgentCard(AgentCard.builder()
+ .setSecuritySchemes(schemes)
+ .setSecurityRequirements(List.of(
+ SecurityRequirement.of("oauth", List.of("admin")),
+ SecurityRequirement.of("agentKey", List.of())))
+ .build());
+
+ AtomicReference<Object> profile = new AtomicReference<>();
+ AtomicReference<Object> result = new AtomicReference<>();
+ A2AConsumer consumer = createConsumer(config, exchange -> {
+ profile.set(exchange.getProperty(A2AConstants.USER_PROFILE));
+
result.set(exchange.getProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT));
+ exchange.getMessage().setBody("ok");
+ });
+
+ // the token is valid but lacks the admin scope, so the caller is
authenticated with the API key
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setHeader("Authorization", "Bearer valid-token");
+ exchange.getMessage().setHeader("X-Agent-Key", "secret-key");
+
exchange.getMessage().setBody(OBJECT_MAPPER.writeValueAsString(sendMessageRequest("msg-key")));
+ consumer.createDispatchProcessor(consumer::handleSendMessage,
false).process(exchange);
+
+ assertThat(profile.get()).isEqualTo(Map.of("scheme", "apiKey"));
+ assertThat(result.get()).isNull();
+ }
+
+ @Test
+ void tasksOfTheClientsOfTokensWithoutSubjectAreSeparate() throws Exception
{
+ A2AConsumer consumer = createConsumer(exchange -> {
+ exchange.getMessage().setHeader(A2AConstants.TASK_STATE,
TaskState.WORKING.getProtoName());
+ exchange.getMessage().setBody("ok");
+ });
+
+ String taskOfA = createTask(consumer, clientProfile("svc-a"));
+ String taskOfB = createTask(consumer, clientProfile("svc-b"));
+
+ assertThat(listTasks(consumer,
clientProfile("svc-a"))).containsExactly(taskOfA);
+ assertThat(listTasks(consumer,
clientProfile("svc-b"))).containsExactly(taskOfB);
+ }
+
+ @Test
+ void tasksOfTheUsersOfAClientAreSeparate() throws Exception {
+ A2AConsumer consumer = createConsumer(exchange -> {
+ exchange.getMessage().setHeader(A2AConstants.TASK_STATE,
TaskState.WORKING.getProtoName());
+ exchange.getMessage().setBody("ok");
+ });
+
+ String taskOfAlice = createTask(consumer, userProfile("alice"));
+ String taskOfBob = createTask(consumer, userProfile("bob"));
+
+ assertThat(listTasks(consumer,
userProfile("alice"))).containsExactly(taskOfAlice);
+ assertThat(listTasks(consumer,
userProfile("bob"))).containsExactly(taskOfBob);
+ }
+
+ private A2AConsumer createConsumer(Processor processor) throws Exception {
+ return createConsumer(new A2AConfiguration(), processor);
+ }
+
+ private A2AConsumer createConsumer(A2AConfiguration config, Processor
processor) throws Exception {
+ config.setName("Profile Agent");
+ config.setVersion("1.0.0");
+
+ A2AComponent component = new A2AComponent();
+ component.setCamelContext(context);
+ A2AEndpoint endpoint = new A2AEndpoint("a2a:test", component, config);
+ endpoint.setAgentCardSource("test");
+ endpoint.setCamelContext(context);
+ endpoint.doStart();
+
+ return new A2AConsumer(endpoint, processor);
+ }
+
+ // the profile of an introspected token without sub: the user and the
client it was issued to
+ private static Map<String, Object> userProfile(String username) {
+ return Map.of("scheme", "oauth2", "issuer", "https://issuer.example";,
"claims",
+ Map.of("username", username, "client_id", "svc"));
+ }
+
+ private static Map<String, Object> clientProfile(String clientId) {
+ return Map.of("scheme", "oauth2", "issuer", "https://issuer.example";,
"claims", Map.of("client_id", clientId));
+ }
+
+ private String createTask(A2AConsumer consumer, Map<String, Object>
profile) throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.setProperty(A2AConstants.USER_PROFILE, profile);
+ return sendMessage(consumer, exchange, "msg-" +
profile.hashCode()).getTask().id();
+ }
+
+ private static SendMessageRequest sendMessageRequest(String messageId) {
+ SendMessageRequest request = new SendMessageRequest();
+ request.setMessage(Message.builder()
+ .messageId(messageId)
+ .role(Message.Role.ROLE_USER)
+ .parts(List.of(new TextPart("hello")))
+ .build());
+ return request;
+ }
+
+ private SendMessageResponse sendMessage(A2AConsumer consumer, Exchange
exchange, String messageId) throws Exception {
+
exchange.getMessage().setBody(OBJECT_MAPPER.writeValueAsString(sendMessageRequest(messageId)));
+
+ consumer.handleSendMessage(exchange);
+
+ return
OBJECT_MAPPER.readValue(exchange.getMessage().getBody(byte[].class),
SendMessageResponse.class);
+ }
+
+ private List<String> listTasks(A2AConsumer consumer, Map<String, Object>
profile) throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.setProperty(A2AConstants.USER_PROFILE, profile);
+
+ consumer.handleListTasks(exchange);
+
+ ListTasksResponse response = OBJECT_MAPPER.readValue(
+ exchange.getMessage().getBody(byte[].class),
ListTasksResponse.class);
+ return response.tasks().stream().map(Task::id).toList();
+ }
+
+ private static final class StubOAuthTokenValidationFactory implements
OAuthTokenValidationFactory {
+
+ @Override
+ public OAuthTokenValidationResult
validateToken(OAuthTokenValidationConfig config, String token) {
+ return validate(token);
+ }
+
+ @Override
+ public OAuthTokenValidationResult validateToken(CamelContext context,
String profileName, String token) {
+ return validate(token);
+ }
+
+ @Override
+ public void validateConfiguration(OAuthTokenValidationConfig config) {
+ }
+
+ private static OAuthTokenValidationResult validate(String token) {
+ if ("valid-token".equals(token)) {
+ return OAuthTokenValidationResult.valid("alice",
"https://issuer.example";, List.of("camel-a2a"),
+ List.of("message:send"), Map.of(), 1234567890L);
+ }
+ return
OAuthTokenValidationResult.invalid(OAuthTokenValidationResult.ErrorCode.INVALID_TOKEN,
"invalid token");
+ }
+ }
+}
diff --git
a/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/auth/A2AAuthHandlerTest.java
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/auth/A2AAuthHandlerTest.java
index 2e9a7ee4bacd..fe85652bb68d 100644
---
a/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/auth/A2AAuthHandlerTest.java
+++
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/auth/A2AAuthHandlerTest.java
@@ -23,6 +23,7 @@ import java.util.Map;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.component.a2a.A2AConfiguration;
+import org.apache.camel.component.a2a.A2AConstants;
import org.apache.camel.component.a2a.model.AgentCard;
import org.apache.camel.component.a2a.model.SecurityRequirement;
import org.apache.camel.component.a2a.model.SecurityScheme;
@@ -253,7 +254,7 @@ class A2AAuthHandlerTest extends CamelTestSupport {
A2AUserProfile profile = handler.validateConsumerAuth(exchange,
cardWithScheme("oauth", "oauth2"));
assertOAuthProfile(profile, "oauth2");
-
assertThat(exchange.getProperty(BearerTokenExtractor.OAUTH_TOKEN_VALIDATION_RESULT))
+
assertThat(exchange.getProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT))
.isInstanceOf(OAuthTokenValidationResult.class);
}
@@ -272,7 +273,7 @@ class A2AAuthHandlerTest extends CamelTestSupport {
A2AUserProfile profile = handler.validateConsumerAuth(exchange,
cardWithScheme("oidc", "openIdConnect"));
assertOAuthProfile(profile, "openIdConnect");
-
assertThat(exchange.getProperty(BearerTokenExtractor.OAUTH_TOKEN_VALIDATION_RESULT))
+
assertThat(exchange.getProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT))
.isInstanceOf(OAuthTokenValidationResult.class);
}
@@ -366,6 +367,122 @@ class A2AAuthHandlerTest extends CamelTestSupport {
.hasMessageContaining("missing required scopes");
}
+ @Test
+ void
validateConsumerAuthKeepsTheTokenProfileWhateverTheOrderOfTheSchemes() {
+ bindOAuthValidationFactory();
+
+ A2AConfiguration config = new A2AConfiguration();
+ config.setValidateAuth(true);
+ config.setApiKey("secret-key");
+ config.setOauthProfile("myprofile");
+ A2AAuthHandler handler = new A2AAuthHandler(config, defaultHandlers());
+
+ for (List<String> order : List.of(List.of("agentKey", "oauth"),
List.of("oauth", "agentKey"))) {
+ Map<String, SecurityScheme> schemes = new LinkedHashMap<>();
+ schemes.put("agentKey", SecurityScheme.apiKey("header",
"X-Agent-Key"));
+ schemes.put("oauth", SecurityScheme.oauth2(null));
+ Map<String, List<String>> requirement = new LinkedHashMap<>();
+ order.forEach(name -> requirement.put(name, List.of()));
+ AgentCard card = new AgentCard.Builder()
+ .setName("test")
+ .setSecuritySchemes(schemes)
+
.setSecurityRequirements(List.of(SecurityRequirement.fromScopeMap(requirement)))
+ .build();
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setHeader("X-Agent-Key", "secret-key");
+ exchange.getMessage().setHeader("Authorization", "Bearer
valid-token");
+
+ assertOAuthProfile(handler.validateConsumerAuth(exchange, card),
"oauth2");
+ }
+ }
+
+ @Test
+ void validateConsumerAuthDropsTheTokenOfARequirementThatIsNotSatisfied() {
+ bindOAuthValidationFactory();
+
+ A2AConfiguration config = new A2AConfiguration();
+ config.setValidateAuth(true);
+ config.setApiKey("secret-key");
+ config.setOauthProfile("myprofile");
+ A2AAuthHandler handler = new A2AAuthHandler(config, defaultHandlers());
+
+ Map<String, SecurityScheme> schemes = new LinkedHashMap<>();
+ schemes.put("oauth", SecurityScheme.oauth2(null));
+ schemes.put("agentKey", SecurityScheme.apiKey("header",
"X-Agent-Key"));
+ AgentCard card = new AgentCard.Builder()
+ .setName("test")
+ .setSecuritySchemes(schemes)
+ .setSecurityRequirements(List.of(
+ SecurityRequirement.of("oauth", List.of("admin")),
+ SecurityRequirement.of("agentKey", List.of())))
+ .build();
+
+ // the token is valid but lacks the admin scope, so the caller is
authenticated with the API key
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setHeader("Authorization", "Bearer valid-token");
+ exchange.getMessage().setHeader("X-Agent-Key", "secret-key");
+
+ A2AUserProfile profile = handler.validateConsumerAuth(exchange, card);
+
+ assertThat(profile.getScheme()).isEqualTo("apiKey");
+
assertThat(exchange.getProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT)).isNull();
+ }
+
+ @Test
+ void validateConsumerAuthKeysATokenWithoutSubjectByItsClient() {
+ bindOAuthValidationFactory();
+
+ A2AConfiguration config = new A2AConfiguration();
+ config.setValidateAuth(true);
+ config.setOauthProfile("myprofile");
+ A2AAuthHandler handler = new A2AAuthHandler(config, defaultHandlers());
+
+ Exchange clientIdExchange = new DefaultExchange(context);
+ clientIdExchange.getMessage().setHeader("Authorization", "Bearer
client-id-token");
+ assertThat(handler.validateConsumerAuth(clientIdExchange,
cardWithScheme("oauth", "oauth2")).ownerKey())
+ .isEqualTo("client|https://issuer.example|svc-a");
+
+ Exchange azpExchange = new DefaultExchange(context);
+ azpExchange.getMessage().setHeader("Authorization", "Bearer
azp-token");
+ assertThat(handler.validateConsumerAuth(azpExchange,
cardWithScheme("oauth", "oauth2")).ownerKey())
+ .isEqualTo("client|https://issuer.example|svc-b");
+ }
+
+ @Test
+ void validateConsumerAuthKeysAnIntrospectedTokenWithoutSubjectByItsUser() {
+ bindOAuthValidationFactory();
+
+ A2AConfiguration config = new A2AConfiguration();
+ config.setValidateAuth(true);
+ config.setOauthProfile("myprofile");
+ A2AAuthHandler handler = new A2AAuthHandler(config, defaultHandlers());
+
+ // an RFC 7662 introspection response without sub names the user and
the client
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setHeader("Authorization", "Bearer
username-token");
+ assertThat(handler.validateConsumerAuth(exchange,
cardWithScheme("oauth", "oauth2")).ownerKey())
+ .isEqualTo("user|https://issuer.example|alice");
+ }
+
+ @Test
+ void validateConsumerAuthRejectsATokenThatDoesNotIdentifyTheCaller() {
+ bindOAuthValidationFactory();
+
+ A2AConfiguration config = new A2AConfiguration();
+ config.setValidateAuth(true);
+ config.setOauthProfile("myprofile");
+ A2AAuthHandler handler = new A2AAuthHandler(config, defaultHandlers());
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setHeader("Authorization", "Bearer
anonymous-token");
+
+ assertThatThrownBy(() -> handler.validateConsumerAuth(exchange,
cardWithScheme("oauth", "oauth2")))
+ .isInstanceOf(SecurityException.class)
+ .hasMessageContaining("does not identify the caller");
+
assertThat(exchange.getProperty(A2AConstants.OAUTH_TOKEN_VALIDATION_RESULT)).isNull();
+ }
+
@Test
void userProfilePreservesOAuthFieldsAndMapView() {
OAuthTokenValidationResult result = OAuthTokenValidationResult.valid(
@@ -460,6 +577,25 @@ class A2AAuthHandlerTest extends CamelTestSupport {
Map.of("tenant", "acme"),
1234567890L);
}
+ if ("client-id-token".equals(token)) {
+ return OAuthTokenValidationResult.valid(
+ null, "https://issuer.example";, List.of("camel-a2a"),
List.of(), Map.of("client_id", "svc-a"),
+ 1234567890L);
+ }
+ if ("username-token".equals(token)) {
+ return OAuthTokenValidationResult.valid(
+ null, "https://issuer.example";, List.of("camel-a2a"),
List.of(),
+ Map.of("username", "alice", "client_id", "svc-a"),
1234567890L);
+ }
+ if ("azp-token".equals(token)) {
+ return OAuthTokenValidationResult.valid(
+ null, "https://issuer.example";, List.of("camel-a2a"),
List.of(), Map.of("azp", "svc-b"),
+ 1234567890L);
+ }
+ if ("anonymous-token".equals(token)) {
+ return OAuthTokenValidationResult.valid(
+ null, "https://issuer.example";, List.of("camel-a2a"),
List.of(), Map.of(), 1234567890L);
+ }
return OAuthTokenValidationResult.invalid(ErrorCode.INVALID_TOKEN,
"invalid token");
}
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index b68963fdb497..8950a78230aa 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -120,6 +120,18 @@ such an entry: `@revoked` and `@cert-authority` entries do
not list a host.
Check that the `known_hosts` file has an entry for the host name or address,
and the port, that each
endpoint is configured with.
+=== camel-a2a - caller profile and task owner
+
+* When a security requirement of the agent card combines several schemes, such
as an API key and an OAuth token, the
+caller profile (`CamelA2AUserProfile`) is now the one of the token, whatever
the order of the schemes. Before, it was
+the profile of the first scheme, so a requirement that listed the API key
first only gave `{"scheme": "apiKey"}` to
+the route, and the tasks belonged to the API key.
+* A token without a `sub` claim is now identified by the `username` that a
token introspection response gives, else by
+its `client_id` or `azp` claim (such as a client credentials token), so the
tasks of different users and clients are
+kept apart. A valid token that has none of these claims is now rejected with
HTTP 401.
+* The `CamelOAuthTokenValidationResult` exchange property is now set on the
exchange that the route processes, as
+documented. Before, it was only set on the incoming HTTP exchange.
+
=== camel-opensearch - the credentials are only sent to the nodes of the
cluster
The `user` and `password` are now offered only to the nodes the client talks
to: the hosts of `hostAddresses` and,
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index a1673f89bf16..21a1aef9398e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -808,6 +808,18 @@ A message that the route marks rollback only every time
was removed before, and
Handle such a message in the route instead, for example by moving it aside,
rather than marking it rollback only
every time.
+=== camel-a2a - caller profile and task owner
+
+* When a security requirement of the agent card combines several schemes, such
as an API key and an OAuth token, the
+caller profile (`CamelA2AUserProfile`) is now the one of the token, whatever
the order of the schemes. Before, it was
+the profile of the first scheme, so a requirement that listed the API key
first only gave `{"scheme": "apiKey"}` to
+the route, and the tasks belonged to the API key.
+* A token without a `sub` claim is now identified by the `username` that a
token introspection response gives, else by
+its `client_id` or `azp` claim (such as a client credentials token), so the
tasks of different users and clients are
+kept apart. A valid token that has none of these claims is now rejected with
HTTP 401.
+* The `CamelOAuthTokenValidationResult` exchange property is now set on the
exchange that the route processes, as
+documented. Before, it was only set on the incoming HTTP exchange.
+
=== camel-sql, camel-mybatis, camel-jooq - an exchange marked rollback only
last is not consumed either
The SQL, MyBatis and jOOQ consumers already left a message unconsumed when the
route marked the exchange rollback
