Implementing

- JWT functionality in Workload API client.
- JWT Source.
- Fake Workload API.

Signed-off-by: Max Lambrecht <maxlambrecht@gmail.com>

java-spiffe-core/build.gradle

@@ -49,13 +49,14 @@ dependencies {
     implementation group: 'io.grpc', name: 'grpc-netty', version: "${grpcVersion}"
     implementation group: 'io.grpc', name: 'grpc-protobuf', version: "${grpcVersion}"
     implementation group: 'io.grpc', name: 'grpc-stub', version: "${grpcVersion}"
+    testImplementation group: 'io.grpc', name: 'grpc-testing', version: "${grpcVersion}"
 
     implementation group: 'io.netty', name: 'netty-transport-native-epoll', version: "${nettyVersion}", classifier: 'linux-x86_64'
     implementation group: 'io.netty', name: 'netty-transport-native-kqueue', version: "${nettyVersion}", classifier: 'osx-x86_64'
     compileOnly group: 'javax.annotation', name: 'javax.annotation-api', version: '1.3.2'
 
     // library for processing JWT tokens and JOSE JWK bundles
-    implementation group: 'com.nimbusds', name: 'nimbus-jose-jwt', version: '5.7'
+    implementation group: 'com.nimbusds', name: 'nimbus-jose-jwt', version: '8.17'
 
     // using bouncy castle for generating x509 certs for testing purposes
     testImplementation group: 'org.bouncycastle', name: 'bcpkix-jdk15on', version: '1.65'

java-spiffe-core/src/main/java/spiffe/exception/JwtSourceException.java

@@ -0,0 +1,19 @@
+package spiffe.exception;
+
+/**
+ * Checked thrown when there is an error creating or initializing a JWT source
+ */
+public class JwtSourceException extends Exception {
+
+    public JwtSourceException(String message) {
+        super(message);
+    }
+
+    public JwtSourceException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public JwtSourceException(Throwable cause) {
+        super(cause);
+    }
+}

java-spiffe-core/src/main/java/spiffe/svid/jwtsvid/JwtSvidSource.java

@@ -1,5 +1,6 @@
 package spiffe.svid.jwtsvid;
 
+import spiffe.exception.JwtSvidException;
 import spiffe.spiffeid.SpiffeId;
 
 /**
@@ -17,5 +18,5 @@ public interface JwtSvidSource {
      *
      * @throws //TODO: declare thrown exceptions
      */
-    JwtSvid fetchJwtSvid(SpiffeId subject, String audience, String... extraAudiences);
+    JwtSvid fetchJwtSvid(SpiffeId subject, String audience, String... extraAudiences) throws JwtSvidException;
 }

java-spiffe-core/src/main/java/spiffe/workloadapi/Address.java

@@ -17,11 +17,13 @@ import java.util.List;
 public class Address {
 
     /**
-     * 	Environment variable holding the default Workload API address.
+     * Environment variable holding the default Workload API address.
      */
     public static final String SOCKET_ENV_VARIABLE = "SPIFFE_ENDPOINT_SOCKET";
 
-    private static final List<String> VALID_SCHEMES = Arrays.asList("unix", "tcp");
+    private static final String UNIX_SCHEME = "unix";
+    private static final String TCP_SCHEME = "tcp";
+    private static final List<String> VALID_SCHEMES = Arrays.asList(UNIX_SCHEME, TCP_SCHEME);
 
     /**
      * Returns the default Workload API address hold by the system environment variable
@@ -44,7 +46,6 @@ public class Address {
      *
      * @param address the Workload API socket address as a string
      * @return an instance of a {@link URI}
-     *
      * @throws SocketEndpointAddressException if the address could not be parsed or if it is not valid
      */
     public static URI parseAddress(String address) throws SocketEndpointAddressException {
@@ -61,15 +62,12 @@ public class Address {
         }
 
         String error = null;
-        switch (scheme) {
+        if (UNIX_SCHEME.equals(scheme)) {
-            case "unix":
+            error = validateUnixAddress(parsedAddress);
-                error = validateUnixAddress(parsedAddress);
+        }
-                break;
+
-            case "tcp":
+        if (TCP_SCHEME.equals(scheme)) {
-                error = validateTcpAddress(parsedAddress);
+            error = validateTcpAddress(parsedAddress);
-                break;
-            default:
-                error = "Workload endpoint socket URI must have a tcp:// or unix:// scheme: %s";
         }
 
         if (StringUtils.isNotBlank(error)) {
@@ -88,10 +86,6 @@ public class Address {
             return "Workload endpoint unix socket URI must not include user info: %s";
         }
 
-        if (StringUtils.isBlank(parsedAddress.getHost()) && StringUtils.isBlank(parsedAddress.getPath())) {
-            return "Workload endpoint unix socket URI must include a path: %s";
-        }
-
         if (StringUtils.isNotBlank(parsedAddress.getRawQuery())) {
             return "Workload endpoint unix socket URI must not include query values: %s";
         }
@@ -152,5 +146,6 @@ public class Address {
         }
     }
 
-    private Address() {}
+    private Address() {
+    }
 }

java-spiffe-core/src/main/java/spiffe/workloadapi/JwtSource.java

@@ -1,37 +1,239 @@
 package spiffe.workloadapi;
 
-import org.apache.commons.lang3.NotImplementedException;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NonNull;
+import lombok.extern.java.Log;
+import lombok.val;
+import org.apache.commons.lang3.exception.ExceptionUtils;
 import spiffe.bundle.jwtbundle.JwtBundle;
+import spiffe.bundle.jwtbundle.JwtBundleSet;
 import spiffe.bundle.jwtbundle.JwtBundleSource;
+import spiffe.exception.BundleNotFoundException;
+import spiffe.exception.JwtSourceException;
+import spiffe.exception.JwtSvidException;
+import spiffe.exception.SocketEndpointAddressException;
 import spiffe.spiffeid.SpiffeId;
 import spiffe.spiffeid.TrustDomain;
 import spiffe.svid.jwtsvid.JwtSvid;
 import spiffe.svid.jwtsvid.JwtSvidSource;
 
+import java.io.Closeable;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.logging.Level;
+
 /**
  * A <code>JwtSource</code> represents a source of SPIFFE JWT SVID and JWT bundles
  * maintained via the Workload API.
  */
-public class JwtSource implements JwtSvidSource, JwtBundleSource {
+@Log
+public class JwtSource implements JwtSvidSource, JwtBundleSource, Closeable {
+
+    private static final Duration DEFAULT_TIMEOUT;
+
+    static {
+        DEFAULT_TIMEOUT = Duration.ofSeconds(Long.getLong("spiffe.newJwtSource.timeout", 0));
+    }
+
+    private JwtBundleSet bundles;
+    private WorkloadApiClient workloadApiClient;
+    private volatile boolean closed;
 
     /**
-     *  Creates a new JWT source. It blocks until the initial update
+     * Creates a new JWT source. It blocks until the initial update
-     *  has been received from the Workload API.
+     * has been received from the Workload API or until the timeout configured
+     * through the system property `spiffe.newJwtSource.timeout` expires.
+     * If no timeout is configured, it blocks until it gets a JWT update from the Workload API.
+     * <p>
+     * It uses the default address socket endpoint from the environment variable to get the Workload API address.
      *
-     * @param spiffeSocketPath a path to the Workload API endpoint
+     * @return an instance of {@link JwtSource}, with the JWT bundles initialized
-     * @return an instance of a {@link JwtSource}
+     * @throws SocketEndpointAddressException if the address to the Workload API is not valid
+     * @throws JwtSourceException            if the source could not be initialized
      */
-    public static JwtSource newSource(String spiffeSocketPath) {
+    public static JwtSource newSource() throws JwtSourceException, SocketEndpointAddressException {
-        throw new NotImplementedException("Not implemented");
+        JwtSourceOptions options = JwtSourceOptions.builder().build();
+        return newSource(options, DEFAULT_TIMEOUT);
+    }
+
+    /**
+     * Creates a new JWT source. It blocks until the initial update
+     * has been received from the Workload API or until the timeout configured
+     * through the system property `spiffe.newJwtSource.timeout` expires.
+     * If no timeout is configured, it blocks until it gets a JWT update from the Workload API.
+     * <p>
+     * It uses the default address socket endpoint from the environment variable to get the Workload API address.
+     *
+     * @param timeout Time to wait for the JWT bundles update. If the timeout is Zero, it will wait indefinitely.
+     * @return an instance of {@link JwtSource}, with the JWT bundles initialized
+     * @throws SocketEndpointAddressException if the address to the Workload API is not valid
+     * @throws JwtSourceException            if the source could not be initialized
+     */
+    public static JwtSource newSource(@NonNull Duration timeout) throws JwtSourceException, SocketEndpointAddressException {
+        JwtSourceOptions options = JwtSourceOptions.builder().build();
+        return newSource(options, timeout);
+    }
+
+    /**
+     * Creates a new JWT source. It blocks until the initial update
+     * has been received from the Workload API or until the timeout configured
+     * through the system property `spiffe.newJwtSource.timeout` expires.
+     * If no timeout is configured, it blocks until it gets a JWT update from the Workload API.
+     * <p>
+     * It uses the default address socket endpoint from the environment variable to get the Workload API address.
+     *
+     * @param options {@link JwtSourceOptions}
+     * @return an instance of {@link JwtSource}, with the JWT bundles initialized
+     * @throws SocketEndpointAddressException if the address to the Workload API is not valid
+     * @throws JwtSourceException            if the source could not be initialized
+     */
+    public static JwtSource newSource(@NonNull JwtSourceOptions options) throws JwtSourceException, SocketEndpointAddressException {
+        return newSource(options, DEFAULT_TIMEOUT);
+    }
+
+
+    /**
+     * Creates a new JWT source. It blocks until the initial update
+     * has been received from the Workload API, doing retries with a backoff exponential policy,
+     * or the timeout has expired.
+     * <p>
+     * The {@link WorkloadApiClient} can be provided in the options, if it is not,
+     * a new client is created.
+     *
+     * @param timeout Time to wait for the JWT bundles update. If the timeout is Zero, it will wait indefinitely.
+     * @param options {@link JwtSourceOptions}
+     * @return an instance of {@link JwtSource}, with the JWT bundles initialized
+     * @throws SocketEndpointAddressException if the address to the Workload API is not valid
+     * @throws JwtSourceException if the source could not be initialized
+     */
+    public static JwtSource newSource(@NonNull JwtSourceOptions options, @NonNull Duration timeout) throws SocketEndpointAddressException, JwtSourceException {
+        if (options.workloadApiClient == null) {
+            options.workloadApiClient = createClient(options);
+        }
+
+        JwtSource jwtSource = new JwtSource();
+        jwtSource.workloadApiClient = options.workloadApiClient;
+
+        try {
+            jwtSource.init(timeout);
+        } catch (Exception e) {
+            jwtSource.close();
+            throw new JwtSourceException("Error creating JWT source", e);
+        }
+
+        return jwtSource;
+    }
+
+    private void init(Duration timeout) throws InterruptedException, TimeoutException {
+        CountDownLatch done = new CountDownLatch(1);
+        setJwtBundlesWatcher(done);
+
+        boolean success;
+        if (timeout.isZero()) {
+            done.await();
+            success = true;
+        } else {
+            success = done.await(timeout.getSeconds(), TimeUnit.SECONDS);
+        }
+        if (!success) {
+            throw new TimeoutException("Timeout waiting for JWT bundles update");
+        }
+    }
+
+    private void setJwtBundlesWatcher(CountDownLatch done) {
+        workloadApiClient.watchJwtBundles(new Watcher<JwtBundleSet>() {
+            @Override
+            public void onUpdate(JwtBundleSet update) {
+                log.log(Level.INFO, "Received JwtBundleSet update");
+                setJwtBundleSet(update);
+                done.countDown();
+            }
+
+            @Override
+            public void onError(Throwable error) {
+                log.log(Level.SEVERE, String.format("Error in JwtBundleSet watcher: %s", ExceptionUtils.getStackTrace(error)));
+                done.countDown();
+            }
+        });
+    }
+
+    private void setJwtBundleSet(@NonNull final JwtBundleSet update) {
+        synchronized (this) {
+            this.bundles = update;
+        }
+    }
+
+    private boolean isClosed() {
+        synchronized (this) {
+            return closed;
+        }
     }
 
     @Override
-    public JwtBundle getJwtBundleForTrustDomain(TrustDomain trustDomain) {
+    public JwtBundle getJwtBundleForTrustDomain(TrustDomain trustDomain) throws BundleNotFoundException {
-        throw new NotImplementedException("Not implemented");
+        if (isClosed()) {
+            throw new IllegalStateException("JWT bundle source is closed");
+        }
+        return bundles.getJwtBundleForTrustDomain(trustDomain);
     }
 
     @Override
-    public JwtSvid fetchJwtSvid(SpiffeId subject, String audience, String... extraAudiences) {
+    public JwtSvid fetchJwtSvid(SpiffeId subject, String audience, String... extraAudiences) throws JwtSvidException {
-        throw new NotImplementedException("Not implemented");
+        if (isClosed()) {
+            throw new IllegalStateException("JWT SVID source is closed");
+        }
+
+        return workloadApiClient.fetchJwtSvid(subject, audience, extraAudiences);
+    }
+
+    /**
+     * Closes this source, dropping the connection to the Workload API.
+     * Other source methods will return an error after close has been called.
+     */
+    @Override
+    public void close() {
+        if (!closed) {
+            synchronized (this) {
+                if (!closed) {
+                    workloadApiClient.close();
+                    closed = true;
+                }
+            }
+        }
+    }
+
+    private static WorkloadApiClient createClient(@NonNull JwtSourceOptions options) throws SocketEndpointAddressException {
+        val clientOptions = WorkloadApiClient.ClientOptions
+                .builder()
+                .spiffeSocketPath(options.spiffeSocketPath)
+                .build();
+        return WorkloadApiClient.newClient(clientOptions);
+    }
+
+    // private constructor
+    private JwtSource() {
+    }
+
+    @Data
+    public static class JwtSourceOptions {
+
+        /**
+         * Address to the Workload API, if it is not set, the default address will be used.
+         */
+        String spiffeSocketPath;
+
+        /**
+         * A custom instance of a {@link WorkloadApiClient}, if it is not set, a new instance will be created.
+         */
+        WorkloadApiClient workloadApiClient;
+
+        @Builder
+        public JwtSourceOptions(String spiffeSocketPath, WorkloadApiClient workloadApiClient) {
+            this.spiffeSocketPath = spiffeSocketPath;
+            this.workloadApiClient = workloadApiClient;
+        }
     }
 }

java-spiffe-core/src/main/java/spiffe/workloadapi/WorkloadApiClient.java

@@ -8,12 +8,9 @@ import lombok.Data;
 import lombok.NonNull;
 import lombok.extern.java.Log;
 import lombok.val;
-import org.apache.commons.lang3.NotImplementedException;
 import org.apache.commons.lang3.StringUtils;
 import spiffe.bundle.jwtbundle.JwtBundleSet;
-import spiffe.exception.SocketEndpointAddressException;
+import spiffe.exception.*;
-import spiffe.exception.X509ContextException;
-import spiffe.exception.X509SvidException;
 import spiffe.spiffeid.SpiffeId;
 import spiffe.svid.jwtsvid.JwtSvid;
 import spiffe.workloadapi.internal.*;
@@ -23,6 +20,7 @@ import spiffe.workloadapi.retry.BackoffPolicy;
 import spiffe.workloadapi.retry.RetryHandler;
 
 import java.io.Closeable;
+import java.security.KeyException;
 import java.security.cert.CertificateException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -78,6 +76,17 @@ public class WorkloadApiClient implements Closeable {
         this.executorService = executorService;
     }
 
+    // package private constructor, used to inject workloadApi stubs for testing
+    WorkloadApiClient(SpiffeWorkloadAPIStub workloadApiAsyncStub, SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub, ManagedChannelWrapper managedChannel) {
+        this.workloadApiAsyncStub = workloadApiAsyncStub;
+        this.workloadApiBlockingStub = workloadApiBlockingStub;
+        this.backoffPolicy = new BackoffPolicy();
+        this.executorService = Executors.newCachedThreadPool();
+        this.retryExecutor = Executors.newSingleThreadScheduledExecutor();
+        this.cancellableContexts = new ArrayList<>();
+        this.managedChannel = managedChannel;
+    }
+
     /**
      * Creates a new Workload API client using the default socket endpoint address.
      *
@@ -154,7 +163,7 @@ public class WorkloadApiClient implements Closeable {
      *
      * @param watcher an instance that implements a {@link Watcher}.
      */
-    public void watchX509Context(Watcher<X509Context> watcher) {
+    public void watchX509Context(@NonNull Watcher<X509Context> watcher) {
         val retryHandler = new RetryHandler(backoffPolicy, retryExecutor);
         val cancellableContext = Context.current().withCancellation();
 
@@ -164,6 +173,104 @@ public class WorkloadApiClient implements Closeable {
         this.cancellableContexts.add(cancellableContext);
     }
 
+    /**
+     * One-shot fetch call to get a SPIFFE JWT-SVID.
+     *
+     * @param subject       a SPIFFE ID
+     * @param audience      the audience of the JWT-SVID
+     * @param extraAudience the extra audience for the JWT_SVID
+     * @return an instance of a {@link JwtSvid}
+     */
+    public JwtSvid fetchJwtSvid(@NonNull SpiffeId subject, @NonNull String audience, String... extraAudience) throws JwtSvidException {
+        List<String> audParam = new ArrayList<>();
+        audParam.add(audience);
+        Collections.addAll(audParam, extraAudience);
+
+        try (val cancellableContext = Context.current().withCancellation()) {
+            return cancellableContext.call(() -> callFetchJwtSvid(subject, audParam));
+        } catch (Exception e) {
+            throw new JwtSvidException("Error fetching JWT SVID", e);
+        }
+    }
+
+    /**
+     * Fetches the JWT bundles for JWT-SVID validation, keyed by trust domain.
+     *
+     * @return an instance of a {@link JwtBundleSet}
+     * @throws JwtBundleException when there is an error getting or processing the response from the Workload API
+     */
+    public JwtBundleSet fetchJwtBundles() throws JwtBundleException {
+        try (val cancellableContext = Context.current().withCancellation()) {
+            return cancellableContext.call(this::callFetchBundles);
+        } catch (Exception e) {
+            throw new JwtBundleException("Error fetching JWT SVID", e);
+        }
+    }
+
+    /**
+     * Validates the JWT-SVID token. The parsed and validated
+     * JWT-SVID is returned.
+     *
+     * @param token    JWT token
+     * @param audience audience of the JWT
+     * @return a {@link JwtSvid} if the token and audience could be validated.
+     * @throws JwtSvidException when the token cannot be validated with the audience
+     */
+    public JwtSvid validateJwtSvid(@NonNull String token, @NonNull String audience) throws JwtSvidException {
+        Workload.ValidateJWTSVIDRequest request = Workload.ValidateJWTSVIDRequest
+                .newBuilder()
+                .setSvid(token)
+                .setAudience(audience)
+                .build();
+
+        try (val cancellableContext = Context.current().withCancellation()) {
+            cancellableContext.call(() -> workloadApiBlockingStub.validateJWTSVID(request));
+        } catch (Exception e) {
+            throw new JwtSvidException("Error validating JWT SVID", e);
+        }
+
+        return JwtSvid.parseInsecure(token, Collections.singletonList(audience));
+    }
+
+    /**
+     * Watches for JWT bundles updates.
+     *
+     * @param watcher receives the update for JwtBundles.
+     */
+    public void watchJwtBundles(@NonNull Watcher<JwtBundleSet> watcher) {
+        val retryHandler = new RetryHandler(backoffPolicy, retryExecutor);
+        val cancellableContext = Context.current().withCancellation();
+
+        val streamObserver = getJwtBundleStreamObserver(watcher, retryHandler, cancellableContext);
+
+        cancellableContext.run(() -> workloadApiAsyncStub.fetchJWTBundles(newJwtBundlesRequest(), streamObserver));
+        this.cancellableContexts.add(cancellableContext);
+    }
+
+    /**
+     * Closes this Workload API closing the underlying channel,
+     * cancelling the contexts and shutdown the executor service.
+     */
+    @Override
+    public void close() {
+        log.log(Level.FINE, "Closing WorkloadAPI client");
+        synchronized (this) {
+            if (!closed) {
+                closed = true;
+                for (val context : cancellableContexts) {
+                    context.close();
+                }
+
+                if (managedChannel != null) {
+                    managedChannel.close();
+                }
+                retryExecutor.shutdown();
+                executorService.shutdown();
+            }
+        }
+        log.log(Level.INFO, "WorkloadAPI client is closed");
+    }
+
     private StreamObserver<X509SVIDResponse> getX509ContextStreamObserver(Watcher<X509Context> watcher, RetryHandler retryHandler, Context.CancellableContext cancellableContext) {
         return new StreamObserver<X509SVIDResponse>() {
             @Override
@@ -196,7 +303,44 @@ public class WorkloadApiClient implements Closeable {
             @Override
             public void onCompleted() {
                 cancellableContext.close();
-                watcher.onError(new X509ContextException("Unexpected completed stream"));
+                log.info("Workload API stream is completed");
+            }
+        };
+    }
+
+    private StreamObserver<Workload.JWTBundlesResponse> getJwtBundleStreamObserver(Watcher<JwtBundleSet> watcher, RetryHandler retryHandler, Context.CancellableContext cancellableContext) {
+        return new StreamObserver<Workload.JWTBundlesResponse>() {
+
+            @Override
+            public void onNext(Workload.JWTBundlesResponse value) {
+                try {
+                    JwtBundleSet jwtBundleSet = GrpcConversionUtils.toBundleSet(value);
+                    watcher.onUpdate(jwtBundleSet);
+                    retryHandler.reset();
+                } catch (KeyException | JwtBundleException e) {
+                    watcher.onError(new JwtBundleException("Error processing JWT bundles update", e));
+                }
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                handleWatchJwtBundleError(t);
+            }
+
+            private void handleWatchJwtBundleError(Throwable t) {
+                if (INVALID_ARGUMENT.equals(Status.fromThrowable(t).getCode().name())) {
+                    watcher.onError(new JwtBundleException("Canceling JWT Bundles watch", t));
+                } else {
+                    log.log(Level.INFO, "Retrying connecting to Workload API to register JWT Bundles watcher");
+                    retryHandler.scheduleRetry(() ->
+                            cancellableContext.run(() -> workloadApiAsyncStub.fetchJWTBundles(newJwtBundlesRequest(), this)));
+                }
+            }
+
+            @Override
+            public void onCompleted() {
+                cancellableContext.close();
+                log.info("Workload API stream is completed");
             }
         };
     }
@@ -205,84 +349,22 @@ public class WorkloadApiClient implements Closeable {
     private void validateX509Context(X509Context x509Context) throws X509ContextException {
         if (x509Context.getX509BundleSet() == null || x509Context.getX509BundleSet().getBundles() == null ||
                 x509Context.getX509BundleSet().getBundles().isEmpty()) {
-            throw new X509ContextException("X509 context error: no X.509 bundles found");
+            throw new X509ContextException("X.509 context error: no X.509 bundles found");
         }
 
         if (x509Context.getX509Svid() == null || x509Context.getX509Svid().isEmpty()) {
-            throw new X509ContextException("X509 context error: no X.509 SVID found");
+            throw new X509ContextException("X.509 context error: no X.509 SVID found");
         }
     }
 
-    /**
-     * One-shot fetch call to get a SPIFFE JWT-SVID.
-     *
-     * @param subject       a SPIFFE ID
-     * @param audience      the audience of the JWT-SVID
-     * @param extraAudience the extra audience for the JWT_SVID
-     * @return an instance of a {@link JwtSvid}
-     * @throws //TODO: declare thrown exceptions
-     */
-    public JwtSvid fetchJwtSvid(SpiffeId subject, String audience, String... extraAudience) {
-        throw new NotImplementedException("Not implemented");
-    }
-
-    /**
-     * Fetches the JWT bundles for JWT-SVID validation, keyed by trust domain.
-     *
-     * @return an instance of a {@link JwtBundleSet}
-     * @throws //TODO: declare thrown exceptions
-     */
-    public JwtBundleSet fetchJwtBundles() {
-        throw new NotImplementedException("Not implemented");
-    }
-
-    /**
-     * Validates the JWT-SVID token. The parsed and validated
-     * JWT-SVID is returned.
-     *
-     * @param token    JWT token
-     * @param audience audience of the JWT
-     * @return the {@link JwtSvid} if the token and audience could be validated.
-     * @throws //TODO: declare thrown exceptions
-     */
-    public JwtSvid validateJwtSvid(String token, String audience) {
-        throw new NotImplementedException("Not implemented");
-    }
-
-    /**
-     * Watches for JWT bundles updates.
-     *
-     * @param jwtBundlesWatcher receives the update for JwtBundles.
-     */
-    public void watchJwtBundles(Watcher<JwtBundleSet> jwtBundlesWatcher) {
-        throw new NotImplementedException("Not implemented");
-    }
-
-    /**
-     * Closes this Workload API closing the underlying channel,
-     * cancelling the contexts and shutdown the executor service.
-     */
-    @Override
-    public void close() {
-        log.log(Level.FINE, "Closing WorkloadAPI client");
-        synchronized (this) {
-            if (!closed) {
-                closed = true;
-                for (val context : cancellableContexts) {
-                    context.close();
-                }
-                managedChannel.close();
-                retryExecutor.shutdown();
-                executorService.shutdown();
-            }
-        }
-        log.log(Level.INFO, "WorkloadAPI client is closed");
-    }
-
     private X509SVIDRequest newX509SvidRequest() {
         return X509SVIDRequest.newBuilder().build();
     }
 
+    private Workload.JWTBundlesRequest newJwtBundlesRequest() {
+        return Workload.JWTBundlesRequest.newBuilder().build();
+    }
+
     private X509Context processX509Context() throws X509ContextException {
         try {
             Iterator<X509SVIDResponse> x509SVIDResponse = workloadApiBlockingStub.fetchX509SVID(newX509SvidRequest());
@@ -295,6 +377,34 @@ public class WorkloadApiClient implements Closeable {
         throw new X509ContextException("Error processing X509Context: x509SVIDResponse is empty");
     }
 
+    private JwtSvid callFetchJwtSvid(SpiffeId subject, List<String> audience) throws JwtSvidException {
+        Workload.JWTSVIDRequest jwtsvidRequest = Workload.JWTSVIDRequest
+                .newBuilder()
+                .setSpiffeId(subject.toString())
+                .addAllAudience(audience)
+                .build();
+        Workload.JWTSVIDResponse response = workloadApiBlockingStub.fetchJWTSVID(jwtsvidRequest);
+
+        return JwtSvid.parseInsecure(response.getSvids(0).getSvid(), audience);
+    }
+
+    private JwtBundleSet callFetchBundles() throws JwtBundleException {
+        Workload.JWTBundlesRequest request = Workload.JWTBundlesRequest
+                .newBuilder()
+                .build();
+        Iterator<Workload.JWTBundlesResponse> bundlesResponse = workloadApiBlockingStub.fetchJWTBundles(request);
+
+        if (bundlesResponse.hasNext()) {
+            try {
+                return GrpcConversionUtils.toBundleSet(bundlesResponse.next());
+            } catch (KeyException | JwtBundleException e) {
+                throw new JwtBundleException("Error processing JWT Bundle response from Workload API", e);
+            }
+        }
+        throw new JwtBundleException("JWT Bundle response from the Workload API is empty");
+    }
+
+
     /**
      * Options for creating a new {@link WorkloadApiClient}. The {@link BackoffPolicy}  is used
      * to configure a {@link RetryHandler} to perform retries to reconnect to the Workload API.

java-spiffe-core/src/main/java/spiffe/workloadapi/X509Source.java

@@ -86,7 +86,7 @@ public class X509Source implements X509SvidSource, X509BundleSource, Closeable {
      * @throws SocketEndpointAddressException if the address to the Workload API is not valid
      * @throws X509SourceException            if the source could not be initialized
      */
-    public static X509Source newSource(Duration timeout) throws SocketEndpointAddressException, X509SourceException {
+    public static X509Source newSource(@NonNull Duration timeout) throws SocketEndpointAddressException, X509SourceException {
         X509SourceOptions x509SourceOptions = X509SourceOptions.builder().build();
         return newSource(x509SourceOptions, timeout);
     }
@@ -104,7 +104,7 @@ public class X509Source implements X509SvidSource, X509BundleSource, Closeable {
      * @throws SocketEndpointAddressException if the address to the Workload API is not valid
      * @throws X509SourceException            if the source could not be initialized
      */
-    public static X509Source newSource(X509SourceOptions options) throws SocketEndpointAddressException, X509SourceException {
+    public static X509Source newSource(@NonNull X509SourceOptions options) throws SocketEndpointAddressException, X509SourceException {
         return newSource(options, DEFAULT_TIMEOUT);
     }
 
@@ -122,7 +122,7 @@ public class X509Source implements X509SvidSource, X509BundleSource, Closeable {
      * @throws SocketEndpointAddressException if the address to the Workload API is not valid
      * @throws X509SourceException            if the source could not be initialized
      */
-    public static X509Source newSource(@NonNull X509SourceOptions options, Duration timeout) throws SocketEndpointAddressException, X509SourceException {
+    public static X509Source newSource(@NonNull X509SourceOptions options, @NonNull Duration timeout) throws SocketEndpointAddressException, X509SourceException {
         if (options.workloadApiClient == null) {
             options.workloadApiClient = createClient(options);
         }
@@ -185,7 +185,10 @@ public class X509Source implements X509SvidSource, X509BundleSource, Closeable {
                 }
             }
         }
+    }
 
+    // private constructor
+    private X509Source() {
     }
 
     private static WorkloadApiClient createClient(@NonNull X509SourceOptions options) throws SocketEndpointAddressException {

java-spiffe-core/src/main/java/spiffe/workloadapi/internal/GrpcConversionUtils.java

@@ -2,14 +2,18 @@ package spiffe.workloadapi.internal;
 
 import com.google.protobuf.ByteString;
 import lombok.val;
+import spiffe.bundle.jwtbundle.JwtBundle;
+import spiffe.bundle.jwtbundle.JwtBundleSet;
 import spiffe.bundle.x509bundle.X509Bundle;
 import spiffe.bundle.x509bundle.X509BundleSet;
+import spiffe.exception.JwtBundleException;
 import spiffe.exception.X509SvidException;
 import spiffe.spiffeid.SpiffeId;
 import spiffe.spiffeid.TrustDomain;
 import spiffe.svid.x509svid.X509Svid;
 import spiffe.workloadapi.X509Context;
 
+import java.security.KeyException;
 import java.security.cert.CertificateException;
 import java.util.ArrayList;
 import java.util.List;
@@ -51,14 +55,28 @@ public class GrpcConversionUtils {
 
     private static List<X509Svid> getListOfX509Svid(Workload.X509SVIDResponse x509SVIDResponse) throws X509SvidException {
         List<X509Svid> x509SvidList = new ArrayList<>();
+
         for (Workload.X509SVID x509SVID : x509SVIDResponse.getSvidsList()) {
             val svid = X509Svid.parse(
                     x509SVID.getX509Svid().toByteArray(),
                     x509SVID.getX509SvidKey().toByteArray());
             x509SvidList.add(svid);
+
+            if (!x509SVID.getSpiffeId().equals(svid.getSpiffeId().toString())) {
+                throw new X509SvidException(String.format("SPIFFE ID in X509SVIDResponse (%s) does not match SPIFFE ID in X.509 certificate (%s)", x509SVID.getSpiffeId(), svid.getSpiffeId()));
+            }
         }
         return x509SvidList;
     }
 
+    public static JwtBundleSet toBundleSet(Workload.JWTBundlesResponse bundlesResponse) throws KeyException, JwtBundleException {
+        List<JwtBundle> jwtBundles = new ArrayList<>();
+        for (Map.Entry<String, ByteString> entry : bundlesResponse.getBundlesMap().entrySet()) {
+            JwtBundle jwtBundle = JwtBundle.parse(TrustDomain.of(entry.getKey()), entry.getValue().toByteArray());
+            jwtBundles.add(jwtBundle);
+        }
+        return JwtBundleSet.of(jwtBundles);
+    }
+
     private GrpcConversionUtils() {}
 }

java-spiffe-core/src/test/java/spiffe/utils/TestUtils.java

@@ -10,10 +10,12 @@ import com.nimbusds.jose.jwk.Curve;
 import com.nimbusds.jwt.JWTClaimsSet;
 import com.nimbusds.jwt.SignedJWT;
 
+import java.lang.reflect.Field;
 import java.security.*;
 import java.security.spec.ECGenParameterSpec;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Util methods for generating KeyPairs, tokens, and other functionality used only to be used in testing.
@@ -49,16 +51,23 @@ public class TestUtils {
         }
     }
 
-    public static String generateToken(JWTClaimsSet claims, KeyPair key, String keyId) {
+    public static String generateToken(Map<String, Object> claims, KeyPair keyPair, String keyId) {
+        JWTClaimsSet jwtClaimsSet = buildJWTClaimSetFromClaimsMap(claims);
+        return generateToken(jwtClaimsSet, keyPair, keyId);
+
+
+    }
+
+    public static String generateToken(JWTClaimsSet claims, KeyPair keyPair, String keyId) {
         try {
             JWSAlgorithm algorithm;
             JWSSigner signer;
-            if ("EC".equals(key.getPublic().getAlgorithm())) {
+            if ("EC".equals(keyPair.getPublic().getAlgorithm())) {
                 algorithm = JWSAlgorithm.ES512;
-                signer = new ECDSASigner(key.getPrivate(), Curve.P_521);
+                signer = new ECDSASigner(keyPair.getPrivate(), Curve.P_521);
-            } else if ("RSA".equals(key.getPublic().getAlgorithm())) {
+            } else if ("RSA".equals(keyPair.getPublic().getAlgorithm())) {
                 algorithm = JWSAlgorithm.RS512;
-                signer = new RSASSASigner(key.getPrivate());
+                signer = new RSASSASigner(keyPair.getPrivate());
             } else {
                 throw new IllegalArgumentException("Algorithm not supported");
             }
@@ -78,4 +87,37 @@ public class TestUtils {
                 .audience(audience)
                 .build();
     }
+
+    public static JWTClaimsSet buildJWTClaimSetFromClaimsMap(Map<String, Object> claims) {
+        return new JWTClaimsSet.Builder()
+                .subject((String) claims.get("sub"))
+                .expirationTime((Date) claims.get("exp"))
+                .audience((List<String>) claims.get("aud"))
+                .build();
+    }
+
+    public static void setEnvironmentVariable(String variableName, String value) throws Exception {
+        Class<?> processEnvironment = Class.forName("java.lang.ProcessEnvironment");
+
+        Field unmodifiableMapField = getField(processEnvironment, "theUnmodifiableEnvironment");
+        Object unmodifiableMap = unmodifiableMapField.get(null);
+        injectIntoUnmodifiableMap(variableName, value, unmodifiableMap);
+
+        Field mapField = getField(processEnvironment, "theEnvironment");
+        Map<String, String> map = (Map<String, String>) mapField.get(null);
+        map.put(variableName, value);
+    }
+
+    private static Field getField(Class<?> clazz, String fieldName) throws NoSuchFieldException {
+        Field field = clazz.getDeclaredField(fieldName);
+        field.setAccessible(true);
+        return field;
+    }
+
+    private static void injectIntoUnmodifiableMap(String key, String value, Object map) throws ReflectiveOperationException {
+        Class unmodifiableMap = Class.forName("java.util.Collections$UnmodifiableMap");
+        Field field = getField(unmodifiableMap, "m");
+        Object obj = field.get(map);
+        ((Map<String, String>) obj).put(key, value);
+    }
 }

java-spiffe-core/src/test/java/spiffe/workloadapi/AddressTest.java

@@ -38,6 +38,7 @@ public class AddressTest {
                 Arguments.of("tcp://1.2.3.4:5", URI.create("tcp://1.2.3.4:5")),
                 Arguments.of("tcp:opaque", "Workload endpoint tcp socket URI must not be opaque: tcp:opaque"),
                 Arguments.of("tcp://", "Workload endpoint socket is not a valid URI: tcp://"),
+                Arguments.of("tcp:///test", "Workload endpoint tcp socket URI must include a host: tcp:///test"),
                 Arguments.of("tcp://1.2.3.4:5?whatever", "Workload endpoint tcp socket URI must not include query values: tcp://1.2.3.4:5?whatever"),
                 Arguments.of("tcp://1.2.3.4:5#whatever", "Workload endpoint tcp socket URI must not include a fragment: tcp://1.2.3.4:5#whatever"),
                 Arguments.of("tcp://john:doe@1.2.3.4:5/path", "Workload endpoint tcp socket URI must not include user info: tcp://john:doe@1.2.3.4:5/path"),

java-spiffe-core/src/test/java/spiffe/workloadapi/FakeWorkloadApi.java

@@ -0,0 +1,166 @@
+package spiffe.workloadapi;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.ProtocolStringList;
+import com.google.protobuf.Struct;
+import com.google.protobuf.Value;
+import com.nimbusds.jose.jwk.Curve;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+import org.junit.platform.commons.util.StringUtils;
+import spiffe.exception.JwtSvidException;
+import spiffe.svid.jwtsvid.JwtSvid;
+import spiffe.utils.TestUtils;
+import spiffe.workloadapi.internal.SpiffeWorkloadAPIGrpc;
+import spiffe.workloadapi.internal.Workload;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.security.KeyPair;
+import java.util.*;
+
+class FakeWorkloadApi extends SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIImplBase {
+
+    final String privateKey = "testdata/workloadapi/svid.key";
+    final String svid = "testdata/workloadapi/svid.pem";
+    final String x509Bundle = "testdata/workloadapi/bundle.pem";
+    final String jwtBundle = "testdata/workloadapi/bundle.json";
+
+
+    // Loads cert, bundle and key from files and generates a X509SVIDResponse.
+    @Override
+    public void fetchX509SVID(Workload.X509SVIDRequest request, StreamObserver<Workload.X509SVIDResponse> responseObserver) {
+        try {
+            Path pathCert = Paths.get(toUri(svid));
+            byte[] svidBytes = Files.readAllBytes(pathCert);
+
+            Path pathKey = Paths.get(toUri(privateKey));
+            byte[] keyBytes = Files.readAllBytes(pathKey);
+
+            Path pathBundle = Paths.get(toUri(x509Bundle));
+            byte[] bundleBytes = Files.readAllBytes(pathBundle);
+
+            Workload.X509SVID svid = Workload.X509SVID
+                    .newBuilder()
+                    .setSpiffeId("spiffe://example.org/workload-server")
+                    .setX509Svid(ByteString.copyFrom(svidBytes))
+                    .setX509SvidKey(ByteString.copyFrom(keyBytes))
+                    .setBundle(ByteString.copyFrom(bundleBytes))
+                    .build();
+            Workload.X509SVIDResponse response = Workload.X509SVIDResponse.newBuilder().addSvids(svid).build();
+            responseObserver.onNext(response);
+            responseObserver.onCompleted();
+        } catch (URISyntaxException | IOException e) {
+            throw new Error("Failed FakeSpiffeWorkloadApiService.fetchX509SVID", e);
+        }
+    }
+
+
+    @Override
+    public void fetchJWTSVID(Workload.JWTSVIDRequest request, StreamObserver<Workload.JWTSVIDResponse> responseObserver) {
+        Map<String, Object> claims = new HashMap<>();
+        claims.put("sub", request.getSpiffeId());
+        claims.put("aud", getAudienceList(request.getAudienceList()));
+        Date expiration = new Date(System.currentTimeMillis() + 3600000);
+        claims.put("exp", expiration);
+
+        KeyPair keyPair = TestUtils.generateECKeyPair(Curve.P_521);
+
+        String token = TestUtils.generateToken(claims, keyPair, "authority1");
+
+        Workload.JWTSVID jwtsvid = Workload.JWTSVID
+                .newBuilder()
+                .setSpiffeId("spiffe://example.org/workload-server")
+                .setSvid(token)
+                .build();
+        Workload.JWTSVIDResponse response = Workload.JWTSVIDResponse.newBuilder().addSvids(jwtsvid).build();
+        responseObserver.onNext(response);
+        responseObserver.onCompleted();
+    }
+
+    List<String> getAudienceList(ProtocolStringList audienceList) {
+        List<String> result = new ArrayList<>();
+        for (ByteString str : audienceList.asByteStringList()) {
+            result.add(str.toStringUtf8());
+        }
+        return result;
+    }
+
+    @Override
+    public void fetchJWTBundles(Workload.JWTBundlesRequest request, StreamObserver<Workload.JWTBundlesResponse> responseObserver) {
+        Path pathBundle = null;
+        try {
+            pathBundle = Paths.get(toUri(jwtBundle));
+            byte[] bundleBytes = Files.readAllBytes(pathBundle);
+
+            Workload.JWTBundlesResponse response = Workload.JWTBundlesResponse
+                    .newBuilder()
+                    .putBundles("example.org", ByteString.copyFrom(bundleBytes))
+                    .build();
+
+            responseObserver.onNext(response);
+            responseObserver.onCompleted();
+        } catch (URISyntaxException | IOException e) {
+            throw new Error("Failed FakeSpiffeWorkloadApiService.fetchJWTBundles", e);
+        }
+    }
+
+    @Override
+    public void validateJWTSVID(Workload.ValidateJWTSVIDRequest request, StreamObserver<Workload.ValidateJWTSVIDResponse> responseObserver) {
+        String audience = request.getAudience();
+        if (StringUtils.isBlank(audience)) {
+            responseObserver.onError(new StatusRuntimeException(Status.INVALID_ARGUMENT.withDescription("audience must be specified")));
+        }
+
+        String token = request.getSvid();
+        if (StringUtils.isBlank(token)) {
+            responseObserver.onError(new StatusRuntimeException(Status.INVALID_ARGUMENT.withDescription("svid must be specified")));
+        }
+
+        JwtSvid jwtSvid = null;
+        try {
+            jwtSvid = JwtSvid.parseInsecure(token, Collections.singletonList(audience));
+        } catch (JwtSvidException e) {
+            responseObserver.onError(new StatusRuntimeException(Status.INVALID_ARGUMENT.withDescription(e.getMessage())));
+        }
+
+        Struct structClaims = getClaimsStruct(jwtSvid.getClaims());
+
+        Workload.ValidateJWTSVIDResponse response = Workload.ValidateJWTSVIDResponse
+                .newBuilder()
+                .setSpiffeId(jwtSvid.getSpiffeId().toString())
+                .setClaims(structClaims)
+                .build();
+
+        responseObserver.onNext(response);
+        responseObserver.onCompleted();
+    }
+
+    private Struct getClaimsStruct(Map<String, Object> claims) {
+        Map<String, Value> valueMap = new HashMap<>();
+        Value sub = Value.newBuilder().setStringValue((String) claims.get("sub")).build();
+
+        Date expirationDate = (Date) claims.get("exp");
+        String time = String.valueOf(expirationDate.getTime());
+        Value exp = Value.newBuilder().setStringValue(time).build();
+
+        List<String> audience = (List<String>) claims.get("aud");
+        Value aud = Value.newBuilder().setStringValue(audience.get(0)).build();
+
+        valueMap.put("sub", sub);
+        valueMap.put("exp", exp);
+        valueMap.put("aud", aud);
+
+        return Struct.newBuilder().putAllFields(valueMap).build();
+    }
+
+    private URI toUri(String path) throws URISyntaxException {
+        return getClass().getClassLoader().getResource(path).toURI();
+    }
+}
+

java-spiffe-core/src/test/java/spiffe/workloadapi/JwtSourceTest.java

@@ -0,0 +1,117 @@
+package spiffe.workloadapi;
+
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.testing.GrpcCleanupRule;
+import org.junit.Rule;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import spiffe.bundle.jwtbundle.JwtBundle;
+import spiffe.exception.BundleNotFoundException;
+import spiffe.exception.JwtSourceException;
+import spiffe.exception.JwtSvidException;
+import spiffe.exception.SocketEndpointAddressException;
+import spiffe.spiffeid.SpiffeId;
+import spiffe.spiffeid.TrustDomain;
+import spiffe.svid.jwtsvid.JwtSvid;
+import spiffe.workloadapi.internal.ManagedChannelWrapper;
+import spiffe.workloadapi.internal.SecurityHeaderInterceptor;
+import spiffe.workloadapi.internal.SpiffeWorkloadAPIGrpc;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class JwtSourceTest {
+
+    @Rule
+    public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
+
+    private JwtSource jwtSource;
+
+    @BeforeEach
+    void setUp() throws IOException, JwtSourceException, SocketEndpointAddressException {
+        // Generate a unique in-process server name.
+        String serverName = InProcessServerBuilder.generateName();
+
+        // Create a server, add service, start, and register for automatic graceful shutdown.
+        FakeWorkloadApi fakeWorkloadApi = new FakeWorkloadApi();
+        Server server = InProcessServerBuilder.forName(serverName).directExecutor().addService(fakeWorkloadApi).build().start();
+        grpcCleanup.register(server);
+
+        // Create WorkloadApiClient using Stubs that will connect to the fake WorkloadApiService.
+        ManagedChannel inProcessChannel = InProcessChannelBuilder.forName(serverName).directExecutor().build();
+        grpcCleanup.register(inProcessChannel);
+
+        SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub = SpiffeWorkloadAPIGrpc
+                .newBlockingStub(inProcessChannel)
+                .withInterceptors(new SecurityHeaderInterceptor());
+
+        SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIStub workloadAPIStub = SpiffeWorkloadAPIGrpc
+                .newStub(inProcessChannel)
+                .withInterceptors(new SecurityHeaderInterceptor());
+
+        WorkloadApiClient workloadApiClient = new WorkloadApiClient(workloadAPIStub, workloadApiBlockingStub, new ManagedChannelWrapper(inProcessChannel));
+
+        JwtSource.JwtSourceOptions options = JwtSource.JwtSourceOptions.builder().workloadApiClient(workloadApiClient).build();
+        jwtSource = JwtSource.newSource(options);
+    }
+
+    @AfterEach
+    void tearDown() {
+        jwtSource.close();
+    }
+
+    @Test
+    void testGetJwtBundleForTrustDomain() {
+        try {
+            JwtBundle bundle = jwtSource.getJwtBundleForTrustDomain(TrustDomain.of("example.org"));
+            assertNotNull(bundle);
+            assertEquals(TrustDomain.of("example.org"), bundle.getTrustDomain());
+        } catch (BundleNotFoundException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void testGetJwtBundleForTrustDomain_SourceIsClosed_ThrowsIllegalStateException() {
+        jwtSource.close();
+        try {
+            jwtSource.getJwtBundleForTrustDomain(TrustDomain.of("example.org"));
+            fail("expected exception");
+        } catch (IllegalStateException e) {
+            assertEquals("JWT bundle source is closed", e.getMessage());
+        } catch (BundleNotFoundException e) {
+            fail("not expected exception", e);
+        }
+    }
+
+    @Test
+    void testFetchJwtSvid() {
+        try {
+            JwtSvid svid = jwtSource.fetchJwtSvid(SpiffeId.parse("spiffe://example.org/workload-server"), "aud1", "aud2", "aud3");
+            assertNotNull(svid);
+            assertEquals(SpiffeId.parse("spiffe://example.org/workload-server"), svid.getSpiffeId());
+            assertEquals(Arrays.asList("aud1", "aud2", "aud3"), svid.getAudience());
+        } catch (JwtSvidException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void testFetchJwtSvid_SourceIsClosed_ThrowsIllegalStateException() {
+        jwtSource.close();
+        try {
+            jwtSource.fetchJwtSvid(SpiffeId.parse("spiffe://example.org/workload-server"), "aud1", "aud2", "aud3");
+            fail("expected exception");
+        } catch (IllegalStateException e) {
+            assertEquals("JWT SVID source is closed", e.getMessage());
+        } catch (JwtSvidException e) {
+            fail(e);
+        }
+    }
+}

java-spiffe-core/src/test/java/spiffe/workloadapi/WorkloadApiClientTest.java

@@ -0,0 +1,252 @@
+package spiffe.workloadapi;
+
+import com.nimbusds.jose.jwk.Curve;
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.testing.GrpcCleanupRule;
+import org.junit.Rule;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import spiffe.bundle.jwtbundle.JwtBundle;
+import spiffe.bundle.jwtbundle.JwtBundleSet;
+import spiffe.bundle.x509bundle.X509Bundle;
+import spiffe.exception.BundleNotFoundException;
+import spiffe.exception.JwtBundleException;
+import spiffe.exception.JwtSvidException;
+import spiffe.exception.SocketEndpointAddressException;
+import spiffe.spiffeid.SpiffeId;
+import spiffe.spiffeid.TrustDomain;
+import spiffe.svid.jwtsvid.JwtSvid;
+import spiffe.utils.TestUtils;
+import spiffe.workloadapi.internal.ManagedChannelWrapper;
+import spiffe.workloadapi.internal.SecurityHeaderInterceptor;
+import spiffe.workloadapi.internal.SpiffeWorkloadAPIGrpc;
+import spiffe.workloadapi.retry.BackoffPolicy;
+
+import java.io.IOException;
+import java.security.KeyPair;
+import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class WorkloadApiClientTest {
+
+    @Rule
+    public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
+    private WorkloadApiClient workloadApiClient;
+    private ManagedChannel inProcessChannel;
+
+    @BeforeEach
+    void setUp() throws IOException {
+        // Generate a unique in-process server name.
+        String serverName = InProcessServerBuilder.generateName();
+
+        // Create a server, add service, start, and register for automatic graceful shutdown.
+        FakeWorkloadApi fakeWorkloadApi = new FakeWorkloadApi();
+        Server server = InProcessServerBuilder.forName(serverName).directExecutor().addService(fakeWorkloadApi).build().start();
+        grpcCleanup.register(server);
+
+        // Create WorkloadApiClient using Stubs that will connect to the fake WorkloadApiService.
+        inProcessChannel = InProcessChannelBuilder.forName(serverName).directExecutor().build();
+        grpcCleanup.register(inProcessChannel);
+
+        SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub = SpiffeWorkloadAPIGrpc
+                .newBlockingStub(inProcessChannel)
+                .withInterceptors(new SecurityHeaderInterceptor());
+
+        SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIStub workloadAPIStub = SpiffeWorkloadAPIGrpc
+                .newStub(inProcessChannel)
+                .withInterceptors(new SecurityHeaderInterceptor());
+
+        workloadApiClient = new WorkloadApiClient(workloadAPIStub, workloadApiBlockingStub, new ManagedChannelWrapper(inProcessChannel));
+    }
+
+    @AfterEach
+    void tearDown() {
+        workloadApiClient.close();
+    }
+
+    @Test
+    void testNewClient_defaultOptions() throws Exception {
+        try {
+            TestUtils.setEnvironmentVariable(Address.SOCKET_ENV_VARIABLE, "unix:/tmp/agent.sock" );
+            WorkloadApiClient client = WorkloadApiClient.newClient();
+            assertNotNull(client);
+        } catch (SocketEndpointAddressException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void testNewClient_customOptions() {
+        try {
+            WorkloadApiClient.ClientOptions options =
+                    WorkloadApiClient.ClientOptions
+                            .builder()
+                            .spiffeSocketPath("unix:/tmp/agent.sock")
+                            .executorService(Executors.newCachedThreadPool())
+                            .backoffPolicy(new BackoffPolicy())
+                            .build();
+
+            WorkloadApiClient client = WorkloadApiClient.newClient(options);
+            assertNotNull(client);
+        } catch (SocketEndpointAddressException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    public void testFetchX509Context() throws Exception {
+
+        X509Context x509Context = workloadApiClient.fetchX509Context();
+
+        assertEquals(SpiffeId.parse("spiffe://example.org/workload-server"), x509Context.getDefaultSvid().getSpiffeId());
+        assertNotNull(x509Context.getDefaultSvid().getChain());
+        assertNotNull(x509Context.getDefaultSvid().getPrivateKey());
+        assertNotNull(x509Context.getX509BundleSet());
+        try {
+            X509Bundle bundle = x509Context.getX509BundleSet().getX509BundleForTrustDomain(TrustDomain.of("example.org"));
+            assertNotNull(bundle);
+        } catch (BundleNotFoundException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void testWatchX509Context() throws InterruptedException {
+        CountDownLatch done = new CountDownLatch(1);
+        final X509Context[] x509Context = new X509Context[1];
+        Watcher<X509Context> contextWatcher = new Watcher<X509Context>() {
+            @Override
+            public void onUpdate(X509Context update) {
+                x509Context[0] = update;
+                done.countDown();
+
+            }
+
+            @Override
+            public void onError(Throwable e) {
+            }
+        };
+
+        workloadApiClient.watchX509Context(contextWatcher);
+        done.await();
+
+        X509Context update = x509Context[0];
+        assertNotNull(update);
+        assertEquals(SpiffeId.parse("spiffe://example.org/workload-server"), update.getDefaultSvid().getSpiffeId());
+        assertNotNull(update.getDefaultSvid().getChain());
+        assertNotNull(update.getDefaultSvid().getPrivateKey());
+        assertNotNull(update.getX509BundleSet());
+        try {
+            X509Bundle bundle = update.getX509BundleSet().getX509BundleForTrustDomain(TrustDomain.of("example.org"));
+            assertNotNull(bundle);
+        } catch (BundleNotFoundException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void testFetchJwtSvid() {
+        try {
+            JwtSvid jwtSvid = workloadApiClient.fetchJwtSvid(SpiffeId.parse("spiffe://example.org/workload-server"), "aud1", "aud2", "aud3");
+            assertNotNull(jwtSvid);
+            assertEquals(SpiffeId.parse("spiffe://example.org/workload-server"), jwtSvid.getSpiffeId());
+            assertEquals("aud1", jwtSvid.getAudience().get(0));
+            assertEquals(3, jwtSvid.getAudience().size());
+        } catch (JwtSvidException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void testValidateJwtSvid() {
+        String token = generateToken("spiffe://example.org/workload-server", Collections.singletonList("aud1"));
+        try {
+            JwtSvid jwtSvid = workloadApiClient.validateJwtSvid(token, "aud1");
+            assertNotNull(jwtSvid);
+            assertEquals(SpiffeId.parse("spiffe://example.org/workload-server"), jwtSvid.getSpiffeId());
+            assertEquals("aud1", jwtSvid.getAudience().get(0));
+            assertEquals(1, jwtSvid.getAudience().size());
+        } catch (JwtSvidException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void testFetchJwtBundles() {
+
+        JwtBundleSet jwtBundleSet = null;
+        try {
+            jwtBundleSet = workloadApiClient.fetchJwtBundles();
+        } catch (JwtBundleException e) {
+            fail(e);
+        }
+
+        assertNotNull(jwtBundleSet);
+        try {
+            JwtBundle bundle = jwtBundleSet.getJwtBundleForTrustDomain(TrustDomain.of("example.org"));
+            assertNotNull(bundle);
+            assertEquals(3, bundle.getJwtAuthorities().size());
+        } catch (BundleNotFoundException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void testWatchJwtBundles() throws InterruptedException {
+        CountDownLatch done = new CountDownLatch(1);
+
+        final JwtBundleSet[] jwtBundleSet = new JwtBundleSet[1];
+
+        Watcher<JwtBundleSet> jwtBundleSetWatcher = new Watcher<JwtBundleSet>() {
+            @Override
+            public void onUpdate(JwtBundleSet update) {
+                jwtBundleSet[0] = update;
+                done.countDown();
+
+            }
+            @Override
+            public void onError(Throwable e) {
+            }
+        };
+
+        workloadApiClient.watchJwtBundles(jwtBundleSetWatcher);
+        done.await();
+
+        JwtBundleSet update = jwtBundleSet[0];
+        assertNotNull(update);
+        try {
+            JwtBundle bundle = update.getJwtBundleForTrustDomain(TrustDomain.of("example.org"));
+            assertNotNull(bundle);
+            assertEquals(3, bundle.getJwtAuthorities().size());
+        } catch (BundleNotFoundException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void watchJwtBundles() {
+    }
+
+    @Test
+    void testClose() {
+    }
+
+    private String generateToken(String sub, List<String> aud) {
+        Map<String, Object> claims = new HashMap<>();
+        claims.put("sub", sub);
+        claims.put("aud", aud);
+        Date expiration = new Date(System.currentTimeMillis() + 3600000);
+        claims.put("exp", expiration);
+
+        KeyPair keyPair = TestUtils.generateECKeyPair(Curve.P_256);
+        return TestUtils.generateToken(claims, keyPair, "authority1");
+    }
+
+}

java-spiffe-core/src/test/java/spiffe/workloadapi/X509SourceTest.java

@@ -0,0 +1,107 @@
+package spiffe.workloadapi;
+
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.testing.GrpcCleanupRule;
+import org.junit.Rule;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import spiffe.bundle.x509bundle.X509Bundle;
+import spiffe.exception.BundleNotFoundException;
+import spiffe.exception.SocketEndpointAddressException;
+import spiffe.exception.X509SourceException;
+import spiffe.spiffeid.SpiffeId;
+import spiffe.spiffeid.TrustDomain;
+import spiffe.svid.x509svid.X509Svid;
+import spiffe.workloadapi.internal.ManagedChannelWrapper;
+import spiffe.workloadapi.internal.SecurityHeaderInterceptor;
+import spiffe.workloadapi.internal.SpiffeWorkloadAPIGrpc;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class X509SourceTest {
+
+    @Rule
+    public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
+
+    private X509Source x509Source;
+
+    @BeforeEach
+    void setUp() throws IOException, X509SourceException, SocketEndpointAddressException {
+        // Generate a unique in-process server name.
+        String serverName = InProcessServerBuilder.generateName();
+
+        // Create a server, add service, start, and register for automatic graceful shutdown.
+        FakeWorkloadApi fakeWorkloadApi = new FakeWorkloadApi();
+        Server server = InProcessServerBuilder.forName(serverName).directExecutor().addService(fakeWorkloadApi).build().start();
+        grpcCleanup.register(server);
+
+        // Create WorkloadApiClient using Stubs that will connect to the fake WorkloadApiService.
+        ManagedChannel inProcessChannel = InProcessChannelBuilder.forName(serverName).directExecutor().build();
+        grpcCleanup.register(inProcessChannel);
+
+        SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub = SpiffeWorkloadAPIGrpc
+                .newBlockingStub(inProcessChannel)
+                .withInterceptors(new SecurityHeaderInterceptor());
+
+        SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIStub workloadAPIStub = SpiffeWorkloadAPIGrpc
+                .newStub(inProcessChannel)
+                .withInterceptors(new SecurityHeaderInterceptor());
+
+        WorkloadApiClient workloadApiClient = new WorkloadApiClient(workloadAPIStub, workloadApiBlockingStub, new ManagedChannelWrapper(inProcessChannel));
+        X509Source.X509SourceOptions options = X509Source.X509SourceOptions.builder().workloadApiClient(workloadApiClient).build();
+        x509Source = X509Source.newSource(options);
+    }
+
+    @AfterEach
+    void tearDown() {
+        x509Source.close();
+    }
+
+    @Test
+    void testGetX509BundleForTrustDomain() {
+        try {
+            X509Bundle bundle = x509Source.getX509BundleForTrustDomain(TrustDomain.of("example.org"));
+            assertNotNull(bundle);
+            assertEquals(TrustDomain.of("example.org"), bundle.getTrustDomain());
+        } catch (BundleNotFoundException e) {
+            fail(e);
+        }
+    }
+
+    @Test
+    void testGetX509BundleForTrustDomain_SourceIsClosed_ThrowsIllegalStateExceptions() {
+        x509Source.close();
+        try {
+            x509Source.getX509BundleForTrustDomain(TrustDomain.of("example.org"));
+            fail("exceptions is expected");
+        } catch (IllegalStateException e) {
+            assertEquals("X509 bundle source is closed", e.getMessage());
+        } catch (BundleNotFoundException e) {
+            fail("not expected exception", e);
+        }
+    }
+
+    @Test
+    void testGetX509Svid() {
+        X509Svid x509Svid = x509Source.getX509Svid();
+        assertNotNull(x509Svid);
+        assertEquals(SpiffeId.parse("spiffe://example.org/workload-server"),x509Svid.getSpiffeId());
+    }
+
+    @Test
+    void testGetX509Svid_SourceIsClosed_ThrowsIllegalStateException() {
+        x509Source.close();
+        try {
+            x509Source.getX509Svid();
+            fail("exceptions is expected");
+        } catch (IllegalStateException e) {
+            assertEquals("X509 SVID source is closed", e.getMessage());
+        }
+    }
+}

java-spiffe-core/src/test/resources/testdata/workloadapi/bundle.json

@@ -0,0 +1,28 @@
+{
+  "keys": [
+    {
+      "use": "jwt-svid",
+      "kty": "EC",
+      "kid": "IZwJU1pKXnlj8RtsCSxTiW4p3PaS3LA0",
+      "crv": "P-256",
+      "x": "mjBLEiXTajKogbCbRrsCTO5ztSPbPbDlCEj9U7UyCik",
+      "y": "UqtfGGr8ILjW_r_XhZg2pixq1ZhsY0KeN45smdlJ4Ag"
+    },
+    {
+      "use": "jwt-svid",
+      "kty": "EC",
+      "kid": "thuC5aKXK0xEijJBImdcNE8GRneaQF8a",
+      "crv": "P-256",
+      "x": "xaOEUedrdEjVxhT5sJ7lwNgsvEfBKNgfYfyNjXyAyys",
+      "y": "p0ZWrr9ZFx3tjvBHxScidaqzeNJwwvQV-f8gi6qIGB4"
+    },
+    {
+      "use": "jwt-svid",
+      "kty": "EC",
+      "kid": "lD6u4YqXBvqZkJv9IkakyjTPMOlmYdho",
+      "crv": "P-256",
+      "x": "e1b14XBkpFRG5aALyveYz0g1Gql_zT_Mxyx_4mWIyyY",
+      "y": "K-7zKhCJReI9Mv-VqxIv9CV6cMj20qkT29H35uTtrTM"
+    }
+  ]
+}

java-spiffe-core/src/test/resources/testdata/workloadapi/bundle.pem

@@ -0,0 +1,33 @@
+-----BEGIN CERTIFICATE-----
+MIIBjjCCATSgAwIBAgIBADAKBggqhkjOPQQDAjAeMQswCQYDVQQGEwJVUzEPMA0G
+A1UEChMGU1BJRkZFMB4XDTIwMDUxNjE3MDUyNFoXDTIwMDUyMzE3MDUzNFowHjEL
+MAkGA1UEBhMCVVMxDzANBgNVBAoTBlNQSUZGRTBZMBMGByqGSM49AgEGCCqGSM49
+AwEHA0IABCQyshZ+HhyfaordIzoupU4qFd07uTRNysO6z9i0WMGzhIAA06z6JXat
+rFBsHJYXbmuXp9afh+Ivr/RcGHj08PSjYzBhMA4GA1UdDwEB/wQEAwIBhjAPBgNV
+HRMBAf8EBTADAQH/MB0GA1UdDgQWBBToJDrI0vnnlhn7EmooEWe43+hPajAfBgNV
+HREEGDAWhhRzcGlmZmU6Ly9leGFtcGxlLm9yZzAKBggqhkjOPQQDAgNIADBFAiEA
++mM9GONpM1L6QYw8c+IvWgXgr+aGoOVpmo0wWcZbc7oCIBiF1NN8p5DeU12wxoUy
+ycQCammceo4hcYLQAYGi/5Q5
+-----END CERTIFICATE-----
+-----BEGIN CERTIFICATE-----
+MIIBjjCCATSgAwIBAgIBADAKBggqhkjOPQQDAjAeMQswCQYDVQQGEwJVUzEPMA0G
+A1UEChMGU1BJRkZFMB4XDTIwMDUyMDE3MDc1N1oXDTIwMDUyNzE3MDgwN1owHjEL
+MAkGA1UEBhMCVVMxDzANBgNVBAoTBlNQSUZGRTBZMBMGByqGSM49AgEGCCqGSM49
+AwEHA0IABO3/qXKapLzDi3wgqW8Lkjm35WrJclRr8aN7IF8Px2jeJpV4KG+wdLa7
+rXSOJH8xCotu9QnQcGo4FuinMsJPlZKjYzBhMA4GA1UdDwEB/wQEAwIBhjAPBgNV
+HRMBAf8EBTADAQH/MB0GA1UdDgQWBBQEOa83CNDa8BcLL/mU3ep//rxyNjAfBgNV
+HREEGDAWhhRzcGlmZmU6Ly9leGFtcGxlLm9yZzAKBggqhkjOPQQDAgNIADBFAiBC
+RTRaKR1nphUMjFcLfopHk+VJgB97yZ8TEZRlNF8vLQIhAJchfcPmlOk9OFiAnSoU
+th2m6yJcLC3axw94n1fg0qcd
+-----END CERTIFICATE-----
+-----BEGIN CERTIFICATE-----
+MIIBjjCCATSgAwIBAgIBADAKBggqhkjOPQQDAjAeMQswCQYDVQQGEwJVUzEPMA0G
+A1UEChMGU1BJRkZFMB4XDTIwMDUyNTExNDEyMVoXDTIwMDYwMTExNDEzMVowHjEL
+MAkGA1UEBhMCVVMxDzANBgNVBAoTBlNQSUZGRTBZMBMGByqGSM49AgEGCCqGSM49
+AwEHA0IABAED6MJ6JluKjEVjKiOP8gPgcqxdJpQKI7iJLDTTd8Ums1/bXTvUxQXG
+PmMcqYAtEvTgs1ew/FDSh5L8XNvaghWjYzBhMA4GA1UdDwEB/wQEAwIBhjAPBgNV
+HRMBAf8EBTADAQH/MB0GA1UdDgQWBBQtAHWFv+CwKHD7G/VNm6oke6CTtTAfBgNV
+HREEGDAWhhRzcGlmZmU6Ly9leGFtcGxlLm9yZzAKBggqhkjOPQQDAgNIADBFAiAn
+VJkxslbz+KJMvsenGo9id3FllKxK1edi2gdyQay62gIhANK6B1ExwDYzUOB5KQUH
+XZg4m88DL41Jn2b6k+fQggVh
+-----END CERTIFICATE-----

java-spiffe-core/src/test/resources/testdata/workloadapi/svid.key

@@ -0,0 +1,5 @@
+-----BEGIN PRIVATE KEY-----
+MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQg+p4+LW7wmMYquxWg
+Z75Bwl5dA+mIrfSRbD2+gQuZkuehRANCAASFtDAPsZg187ijRjpKPv78HfshnAVx
+rgdoCkxIs3OgoPPVULfvPslALF3sWQrLUxzIk33dQ/P46o9LsweBN2Hs
+-----END PRIVATE KEY-----

java-spiffe-core/src/test/resources/testdata/workloadapi/svid.pem

@@ -0,0 +1,13 @@
+-----BEGIN CERTIFICATE-----
+MIIB6zCCAZKgAwIBAgIRANwM7+XYWJtefDH9WtDULHkwCgYIKoZIzj0EAwIwHjEL
+MAkGA1UEBhMCVVMxDzANBgNVBAoTBlNQSUZGRTAeFw0yMDA1MjYwODI1MjlaFw0y
+MDA1MjYwOTI1MzlaMB0xCzAJBgNVBAYTAlVTMQ4wDAYDVQQKEwVTUElSRTBZMBMG
+ByqGSM49AgEGCCqGSM49AwEHA0IABIW0MA+xmDXzuKNGOko+/vwd+yGcBXGuB2gK
+TEizc6Cg89VQt+8+yUAsXexZCstTHMiTfd1D8/jqj0uzB4E3YeyjgbEwga4wDgYD
+VR0PAQH/BAQDAgOoMB0GA1UdJQQWMBQGCCsGAQUFBwMBBggrBgEFBQcDAjAMBgNV
+HRMBAf8EAjAAMB0GA1UdDgQWBBQ+TzaTKiKtWiSN5vya0vck9T8OfjAfBgNVHSME
+GDAWgBQEOa83CNDa8BcLL/mU3ep//rxyNjAvBgNVHREEKDAmhiRzcGlmZmU6Ly9l
+eGFtcGxlLm9yZy93b3JrbG9hZC1zZXJ2ZXIwCgYIKoZIzj0EAwIDRwAwRAIgQEcc
+FThD3vJxp6QpOGIJaWxxSF3B2JqF4A2nc0M3Vz8CICYg750Fw8GCr0K8+Ip5dAcV
+0k1Or5t/ev63uOGy9oIz
+-----END CERTIFICATE-----