feat!: add TaskAuthorizationProvider SPI for per-user task authorization

README.md

@@ -270,6 +270,82 @@ a2a.blocking.consumption.timeout.seconds=5
 
 **Note:** The reference server implementations (Quarkus-based) automatically include the MicroProfile Config integration, so properties work out of the box in `application.properties`.
 
+### 5. Task Authorization (Optional)
+
+The SDK includes an opt-in SPI for per-user task authorization. When enabled, every `RequestHandler` operation checks whether the authenticated user is allowed to access the target task before proceeding. When no provider is present, all operations are permitted (the default).
+
+> **⚠ Security note:** For multi-user deployments, a `TaskAuthorizationProvider` **must** be configured. Without one, all operations are permitted regardless of authentication — any authenticated user can read, modify, or cancel any task. Production deployments should use a fail-closed ownership policy (deny access when ownership is unknown).
+
+#### Providing an implementation
+
+Create an `@ApplicationScoped` CDI bean that implements `TaskAuthorizationProvider`:
+
+```java
+import jakarta.enterprise.context.ApplicationScoped;
+import org.a2aproject.sdk.server.auth.TaskAuthorizationProvider;
+import org.a2aproject.sdk.server.auth.TaskOperation;
+import org.a2aproject.sdk.server.ServerCallContext;
+
+@ApplicationScoped
+public class MyTaskAuthorizationProvider implements TaskAuthorizationProvider {
+
+    @Override
+    public boolean checkRead(ServerCallContext context, String taskId, TaskOperation op) {
+        // Return true to allow, false to deny.
+        // Denied reads throw TaskNotFoundError — the caller cannot distinguish
+        // "not found" from "not authorized", preventing information leakage.
+        String owner = ownershipStore.get(taskId);
+        if (owner == null) {
+            return false; // fail-closed: unknown ownership → deny
+        }
+        return owner.equals(context.getUser().getUsername());
+    }
+
+    @Override
+    public boolean checkWrite(ServerCallContext context, String taskId, TaskOperation op) {
+        return checkRead(context, taskId, op);
+    }
+
+    @Override
+    public boolean checkCreate(ServerCallContext context, TaskOperation op) {
+        return context.getUser().isAuthenticated();
+    }
+
+    @Override
+    public boolean isTaskRecorded(String taskId) {
+        return ownershipStore.contains(taskId);
+    }
+
+    @Override
+    public void recordOwnership(ServerCallContext context, String taskId, TaskOperation op) {
+        ownershipStore.put(taskId, context.getUser().getUsername());
+    }
+}
+```
+
+No additional configuration is required — the SDK automatically discovers the bean via CDI and wires it into the request pipeline. See the [`TaskAuthorizationProvider`](server-common/src/main/java/org/a2aproject/sdk/server/auth/TaskAuthorizationProvider.java) javadoc for the full contract, including thread-safety requirements and ownership-recording semantics.
+
+#### User identity in ServerCallContext
+
+Authorization decisions rely on `context.getUser()` returning the authenticated user. How the user is populated depends on the transport:
+
+- **JSON-RPC and REST**: The Quarkus route handler extracts the user from the Vert.x routing context (`rc.userContext()`) and sets it on `ServerCallContext` directly.
+- **gRPC**: The reference server includes a `QuarkusCallContextFactory` CDI bean that injects the Quarkus `SecurityIdentity` and maps it to the `ServerCallContext` `User`. This happens automatically when using the reference gRPC module. If you provide your own `CallContextFactory`, you are responsible for populating the user.
+
+> **Note:** When task authorization is required, always obtain `RequestHandler` through CDI injection. Manual instantiation via `DefaultRequestHandler.create()` bypasses the `AuthorizationRequestHandlerDecorator` and all authorization checks.
+
+#### How it works
+
+| Operation | Check |
+|-----------|-------|
+| `getTask`, `subscribeToTask`, `getTaskPushNotificationConfig`, `listTaskPushNotificationConfigs` | `checkRead` |
+| `cancelTask`, `createTaskPushNotificationConfig`, `deleteTaskPushNotificationConfig` | `checkWrite` |
+| `messageSend` / `messageSendStream` (existing task) | `checkWrite` |
+| `messageSend` / `messageSendStream` (new task) | `checkCreate`, then `recordOwnership` after creation |
+| `listTasks` | Filtering pushed to `TaskStore.list()` — calls `checkRead` per task |
+
+> Task authorization addresses task isolation for deployments that enable `TaskAuthorizationProvider` with a fail-closed ownership policy. Multi-user deployments must configure this as a required security setting, and should avoid policies that allow unknown ownership by default.
+
 ### Serving Older Protocol Versions (Backward Compatibility)
 
 The A2A Java SDK includes compatibility layers that allow your server to accept requests from clients using older protocol versions. Each compatibility layer is a separate set of modules that you add to your project as needed. **No changes to your `AgentExecutor` are needed** — the compatibility layer converts older protocol requests to v1.0 internally before delegating to your agent.

compat-0.3/reference/grpc/src/main/java/org/a2aproject/sdk/compat03/server/grpc/quarkus/QuarkusCallContextFactory_v0_3.java

@@ -0,0 +1,78 @@
+package org.a2aproject.sdk.compat03.server.grpc.quarkus;
+
+import static org.a2aproject.sdk.server.ServerCallContext.TRANSPORT_KEY;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+
+import io.grpc.Context;
+import io.grpc.Metadata;
+import io.grpc.stub.StreamObserver;
+import io.quarkus.security.identity.SecurityIdentity;
+import org.a2aproject.sdk.compat03.conversion.A2AProtocol_v0_3;
+import org.a2aproject.sdk.compat03.transport.grpc.context.GrpcContextKeys_v0_3;
+import org.a2aproject.sdk.compat03.transport.grpc.handler.CallContextFactory_v0_3;
+import org.a2aproject.sdk.server.ServerCallContext;
+import org.a2aproject.sdk.server.auth.AuthenticatedUser;
+import org.a2aproject.sdk.server.auth.UnauthenticatedUser;
+import org.a2aproject.sdk.server.auth.User;
+import org.a2aproject.sdk.spec.TransportProtocol;
+
+@ApplicationScoped
+public class QuarkusCallContextFactory_v0_3 implements CallContextFactory_v0_3 {
+
+    @Inject
+    Instance<SecurityIdentity> securityIdentityInstance;
+
+    @Override
+    public <V> ServerCallContext create(StreamObserver<V> responseObserver) {
+        User user;
+        if (securityIdentityInstance.isResolvable()) {
+            SecurityIdentity securityIdentity = securityIdentityInstance.get();
+            if (!securityIdentity.isAnonymous()) {
+                user = new AuthenticatedUser(securityIdentity.getPrincipal().getName());
+            } else {
+                user = UnauthenticatedUser.INSTANCE;
+            }
+        } else {
+            user = UnauthenticatedUser.INSTANCE;
+        }
+
+        Map<String, Object> state = new HashMap<>();
+        state.put(TRANSPORT_KEY, TransportProtocol.GRPC);
+        state.put("grpc_response_observer", responseObserver);
+
+        Context currentContext = Context.current();
+        if (currentContext != null) {
+            state.put("grpc_context", currentContext);
+            io.grpc.Metadata grpcMetadata = GrpcContextKeys_v0_3.METADATA_KEY.get(currentContext);
+            if (grpcMetadata != null) {
+                state.put("grpc_metadata", grpcMetadata);
+                Map<String, String> headers = new HashMap<>();
+                for (String key : grpcMetadata.keys()) {
+                    if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
+                        continue;
+                    }
+                    headers.put(key, grpcMetadata.get(Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER)));
+                }
+                state.put("headers", headers);
+            }
+            String methodName = GrpcContextKeys_v0_3.METHOD_NAME_KEY.get(currentContext);
+            if (methodName != null) {
+                state.put("grpc_method_name", methodName);
+            }
+            String peerInfo = GrpcContextKeys_v0_3.PEER_INFO_KEY.get(currentContext);
+            if (peerInfo != null) {
+                state.put("grpc_peer_info", peerInfo);
+            }
+        }
+
+        return new ServerCallContext(user, state, new HashSet<>(), A2AProtocol_v0_3.PROTOCOL_VERSION);
+    }
+}

compat-0.3/reference/grpc/src/test/java/org/a2aproject/sdk/compat03/server/grpc/quarkus/QuarkusA2AGrpc_v0_3_WithTaskAuthorizationTest.java

@@ -0,0 +1,65 @@
+package org.a2aproject.sdk.compat03.server.grpc.quarkus;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import org.a2aproject.sdk.compat03.client.ClientBuilder_v0_3;
+import org.a2aproject.sdk.compat03.client.transport.grpc.GrpcTransport_v0_3;
+import org.a2aproject.sdk.compat03.client.transport.grpc.GrpcTransportConfigBuilder_v0_3;
+import org.a2aproject.sdk.compat03.client.transport.spi.interceptors.auth.AuthInterceptor_v0_3;
+import org.a2aproject.sdk.compat03.conversion.AbstractA2AServerWithTaskAuthorizationTest_v0_3;
+import org.a2aproject.sdk.compat03.conversion.TaskAuthorizationTestProfile_v0_3;
+import org.a2aproject.sdk.compat03.spec.TransportProtocol_v0_3;
+import org.junit.jupiter.api.AfterAll;
+
+@QuarkusTest
+@TestProfile(TaskAuthorizationTestProfile_v0_3.class)
+public class QuarkusA2AGrpc_v0_3_WithTaskAuthorizationTest extends AbstractA2AServerWithTaskAuthorizationTest_v0_3 {
+
+    private static final Map<String, ManagedChannel> channels = new ConcurrentHashMap<>();
+
+    public QuarkusA2AGrpc_v0_3_WithTaskAuthorizationTest() {
+        super(8081);
+    }
+
+    @Override
+    protected String getTransportProtocol() {
+        return TransportProtocol_v0_3.GRPC.asString();
+    }
+
+    @Override
+    protected String getTransportUrl() {
+        return "localhost:8081";
+    }
+
+    @Override
+    protected void configureTransportWithCredentials(ClientBuilder_v0_3 builder, String username, String password) {
+        AuthInterceptor_v0_3 authInterceptor = new AuthInterceptor_v0_3(
+                (schemeName, context) -> BASIC_AUTH_SCHEME_NAME.equals(schemeName)
+                        ? getEncodedCredentials(username, password) : null);
+        builder.withTransport(GrpcTransport_v0_3.class, new GrpcTransportConfigBuilder_v0_3()
+                .channelFactory(target -> {
+                    ManagedChannel channel = ManagedChannelBuilder.forTarget(target).usePlaintext().build();
+                    channels.put(username, channel);
+                    return channel;
+                })
+                .addInterceptor(authInterceptor));
+    }
+
+    @AfterAll
+    static void closeChannels() {
+        channels.values().forEach(ch -> {
+            ch.shutdownNow();
+            try {
+                ch.awaitTermination(10, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        });
+    }
+}

compat-0.3/reference/jsonrpc/src/main/java/org/a2aproject/sdk/compat03/server/apps/quarkus/A2AServerRoutes_v0_3.java

@@ -58,6 +58,7 @@
 import org.a2aproject.sdk.compat03.util.Utils_v0_3;
 import org.a2aproject.sdk.server.PublicAgentCard;
 import org.a2aproject.sdk.server.ServerCallContext;
+import org.a2aproject.sdk.server.auth.AuthenticatedUser;
 import org.a2aproject.sdk.server.auth.UnauthenticatedUser;
 import org.a2aproject.sdk.server.auth.User;
 import org.a2aproject.sdk.server.common.quarkus.SseResponseWriter;
@@ -298,17 +299,8 @@ private ServerCallContext createCallContext(RoutingContext rc) {
             if (rc.user() == null) {
                 user = UnauthenticatedUser.INSTANCE;
             } else {
-                user = new User() {
-                    @Override
-                    public boolean isAuthenticated() {
-                        return rc.userContext().authenticated();
-                    }
-
-                    @Override
-                    public String getUsername() {
-                        return rc.user().subject();
-                    }
-                };
+                String subject = rc.user().subject();
+                user = new AuthenticatedUser(subject != null ? subject : "");
             }
             Map<String, Object> state = new HashMap<>();
 

compat-0.3/reference/jsonrpc/src/test/java/org/a2aproject/sdk/compat03/server/apps/quarkus/QuarkusA2AJSONRPC_v0_3_WithTaskAuthorizationVertxTest.java

@@ -0,0 +1,47 @@
+package org.a2aproject.sdk.compat03.server.apps.quarkus;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import io.vertx.core.Vertx;
+import jakarta.inject.Inject;
+import org.a2aproject.sdk.client.http.VertxA2AHttpClient;
+import org.a2aproject.sdk.compat03.client.ClientBuilder_v0_3;
+import org.a2aproject.sdk.compat03.client.transport.jsonrpc.JSONRPCTransport_v0_3;
+import org.a2aproject.sdk.compat03.client.transport.jsonrpc.JSONRPCTransportConfigBuilder_v0_3;
+import org.a2aproject.sdk.compat03.client.transport.spi.interceptors.auth.AuthInterceptor_v0_3;
+import org.a2aproject.sdk.compat03.conversion.AbstractA2AServerWithTaskAuthorizationTest_v0_3;
+import org.a2aproject.sdk.compat03.conversion.TaskAuthorizationTestProfile_v0_3;
+import org.a2aproject.sdk.compat03.spec.TransportProtocol_v0_3;
+
+@QuarkusTest
+@TestProfile(TaskAuthorizationTestProfile_v0_3.class)
+public class QuarkusA2AJSONRPC_v0_3_WithTaskAuthorizationVertxTest extends AbstractA2AServerWithTaskAuthorizationTest_v0_3 {
+
+    @Inject
+    Vertx vertx;
+
+    public QuarkusA2AJSONRPC_v0_3_WithTaskAuthorizationVertxTest() {
+        super(8081);
+    }
+
+    @Override
+    protected String getTransportProtocol() {
+        return TransportProtocol_v0_3.JSONRPC.asString();
+    }
+
+    @Override
+    protected String getTransportUrl() {
+        return "http://localhost:8081";
+    }
+
+    @Override
+    protected void configureTransportWithCredentials(ClientBuilder_v0_3 builder, String username, String password) {
+        AuthInterceptor_v0_3 authInterceptor = new AuthInterceptor_v0_3(
+                (schemeName, context) -> BASIC_AUTH_SCHEME_NAME.equals(schemeName)
+                        ? getEncodedCredentials(username, password) : null);
+        builder.withTransport(JSONRPCTransport_v0_3.class,
+                new JSONRPCTransportConfigBuilder_v0_3()
+                        .httpClient(new VertxA2AHttpClient(vertx))
+                        .addInterceptor(authInterceptor));
+    }
+}

compat-0.3/reference/rest/src/main/java/org/a2aproject/sdk/compat03/server/rest/quarkus/A2AServerRoutes_v0_3.java

@@ -45,6 +45,7 @@
 import org.a2aproject.sdk.compat03.transport.rest.handler.RestHandler_v0_3.HTTPRestStreamingResponse;
 import org.a2aproject.sdk.server.PublicAgentCard;
 import org.a2aproject.sdk.server.ServerCallContext;
+import org.a2aproject.sdk.server.auth.AuthenticatedUser;
 import org.a2aproject.sdk.server.auth.UnauthenticatedUser;
 import org.a2aproject.sdk.server.auth.User;
 import org.a2aproject.sdk.server.common.quarkus.SseResponseWriter;
@@ -406,24 +407,8 @@ private ServerCallContext createCallContext(RoutingContext rc, String jsonRpcMet
             if (rc.user() == null) {
                 user = UnauthenticatedUser.INSTANCE;
             } else {
-                user = new User() {
-                    @Override
-                    public boolean isAuthenticated() {
-                        if (rc.userContext() != null) {
-                            return rc.userContext().authenticated();
-                        }
-                        return false;
-                    }
-
-                    @Override
-                    public String getUsername() {
-                        if (rc.user() != null) {
-                            String subject = rc.user().subject();
-                            return subject != null ? subject : "";
-                        }
-                        return "";
-                    }
-                };
+                String subject = rc.user().subject();
+                user = new AuthenticatedUser(subject != null ? subject : "");
             }
             Map<String, Object> state = new HashMap<>();
 

compat-0.3/reference/rest/src/test/java/org/a2aproject/sdk/compat03/server/rest/quarkus/QuarkusA2ARest_v0_3_WithTaskAuthorizationVertxTest.java

@@ -0,0 +1,47 @@
+package org.a2aproject.sdk.compat03.server.rest.quarkus;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import io.vertx.core.Vertx;
+import jakarta.inject.Inject;
+import org.a2aproject.sdk.client.http.VertxA2AHttpClient;
+import org.a2aproject.sdk.compat03.client.ClientBuilder_v0_3;
+import org.a2aproject.sdk.compat03.client.transport.rest.RestTransport_v0_3;
+import org.a2aproject.sdk.compat03.client.transport.rest.RestTransportConfigBuilder_v0_3;
+import org.a2aproject.sdk.compat03.client.transport.spi.interceptors.auth.AuthInterceptor_v0_3;
+import org.a2aproject.sdk.compat03.conversion.AbstractA2AServerWithTaskAuthorizationTest_v0_3;
+import org.a2aproject.sdk.compat03.conversion.TaskAuthorizationTestProfile_v0_3;
+import org.a2aproject.sdk.compat03.spec.TransportProtocol_v0_3;
+
+@QuarkusTest
+@TestProfile(TaskAuthorizationTestProfile_v0_3.class)
+public class QuarkusA2ARest_v0_3_WithTaskAuthorizationVertxTest extends AbstractA2AServerWithTaskAuthorizationTest_v0_3 {
+
+    @Inject
+    Vertx vertx;
+
+    public QuarkusA2ARest_v0_3_WithTaskAuthorizationVertxTest() {
+        super(8081);
+    }
+
+    @Override
+    protected String getTransportProtocol() {
+        return TransportProtocol_v0_3.HTTP_JSON.asString();
+    }
+
+    @Override
+    protected String getTransportUrl() {
+        return "http://localhost:8081";
+    }
+
+    @Override
+    protected void configureTransportWithCredentials(ClientBuilder_v0_3 builder, String username, String password) {
+        AuthInterceptor_v0_3 authInterceptor = new AuthInterceptor_v0_3(
+                (schemeName, context) -> BASIC_AUTH_SCHEME_NAME.equals(schemeName)
+                        ? getEncodedCredentials(username, password) : null);
+        builder.withTransport(RestTransport_v0_3.class,
+                new RestTransportConfigBuilder_v0_3()
+                        .httpClient(new VertxA2AHttpClient(vertx))
+                        .addInterceptor(authInterceptor));
+    }
+}