This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 3ebbf212f367 [backport camel-4.22.x] CAMEL-25458: camel-a2a - improve
how the consumer derives the caller profile (#28221)
3ebbf212f367 is described below
commit 3ebbf212f36737a7cb1b829a0e43cf20157ab976
Author: Andrea Cosentino <[email protected]>
AuthorDate: Wed Oct 21 12:34:25 2026 +0200
[backport camel-4.22.x] CAMEL-25458: camel-a2a - improve how the consumer
derives the caller profile (#28221)
CAMEL-25458: camel-a2a - improve how the consumer derives the caller
profile (#28155)
(cherry picked from commit a5e0e20ffbcd1d38c10bd219dba59e47d7737d91)
Signed-off-by: Andrea Cosentino <[email protected]>
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 +++++++++-
9 files changed, 542 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 50330c6fd2ea..8c3ac3b68e3b 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
@@ -234,7 +234,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
@@ -322,6 +322,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 50330c6fd2ea..8c3ac3b68e3b 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
@@ -234,7 +234,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
@@ -322,6 +322,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");
}
}
