spiffe / java-spiffe / 75

pending completion

Build # 75

Build Type

Pull #22

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/spiffe/java-spiffe/commit/<a class=hub.com/spiffe/java-spiffe/commit/<a class="double-link" href="https://git"><a class=hub.com/spiffe/java-spiffe/commit/<a class="double-link" href="https://git"><a class=hub.com/spiffe/java-spiffe/commit/979b55094f691aacf6ca336a52e2688f1e190a94">979b55094<a href="https://github.com/spiffe/java-spiffe/commit/979b55094f691aacf6ca336a52e2688f1e190a94">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/spiffe/java-spiffe/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/spiffe/java-spiffe/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/spiffe/java-spiffe/commit/979b55094f691aacf6ca336a52e2688f1e190a94&quot;&gt;979b55094&lt;/a&gt;&lt;a href=&quot;https://github.com/spiffe/java-spiffe/commit/979b55094f691aacf6ca336a52e2688f1e190a94&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/spiffe/java-spiffe/commit/979b55094f691aacf6ca336a52e2688f&lt;/a&gt;1e190a94">&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/spiffe/java-spiffe/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/spiffe/java-spiffe/commit/<a class="double-link" href="https://github.com/spiffe/java-spiffe/commit/f47e48f684b1d8026b09b8155c7145b9563229c4">f47e48f68</a><a href="https://github.com/spiffe/java-spiffe/commit/979b55094f691aacf6ca336a52e2688f1e190a94">&amp;quot;&amp;gt;f47e48f68&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;f47e48f68&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/spiffe/java-spiffe/commit/979b55094f691aacf6ca336a52e2688f1e190&lt;/a&gt;a94&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/spiffe/java-spiffe/commit/<a class="double-link" href="https://github.com/spiffe/java-spiffe/commit/8c3866ee8">8c3866ee8">8c3866ee8</a>

Pull Request Pull Request #22: Define and implement v2 interface.

Run Details

1157 of 1157 new or added lines in 27 files covered. (100.0%)

1012 of 1157 relevant lines covered (87.47%)

2.62 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.07

/java-spiffe-core/src/main/java/io/spiffe/workloadapi/DefaultWorkloadApiClient.java

package io.spiffe.workloadapi;

import io.grpc.Context;
import io.spiffe.bundle.jwtbundle.JwtBundleSet;
import io.spiffe.exception.JwtBundleException;
import io.spiffe.exception.JwtSvidException;
import io.spiffe.exception.SocketEndpointAddressException;
import io.spiffe.exception.X509ContextException;
import io.spiffe.exception.X509SvidException;
import io.spiffe.spiffeid.SpiffeId;
import io.spiffe.svid.jwtsvid.JwtSvid;
import io.spiffe.workloadapi.grpc.SpiffeWorkloadAPIGrpc;
import io.spiffe.workloadapi.grpc.SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIBlockingStub;
import io.spiffe.workloadapi.grpc.SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIStub;
import io.spiffe.workloadapi.grpc.Workload;
import io.spiffe.workloadapi.internal.GrpcManagedChannelFactory;
import io.spiffe.workloadapi.internal.ManagedChannelWrapper;
import io.spiffe.workloadapi.internal.SecurityHeaderInterceptor;
import io.spiffe.workloadapi.retry.ExponentialBackoffPolicy;
import io.spiffe.workloadapi.retry.RetryHandler;
import lombok.AccessLevel;
import lombok.Builder;
import lombok.Data;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.java.Log;
import lombok.val;
import org.apache.commons.lang3.StringUtils;

import java.security.KeyException;
import java.security.cert.CertificateException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.logging.Level;

import static io.spiffe.workloadapi.StreamObservers.getJwtBundleStreamObserver;
import static io.spiffe.workloadapi.StreamObservers.getX509ContextStreamObserver;

/**
 * Represents a client to interact with the Workload API.
 * <p>
 * Supports one-shot calls and watch updates for X.509 and JWT SVIDs and bundles.
 * <p>
 * The watch for updates methods support retries using an exponential backoff policy to reestablish
 * the stream connection to the Workload API.
 */
@Log
public final class DefaultWorkloadApiClient implements WorkloadApiClient {

    private final SpiffeWorkloadAPIStub workloadApiAsyncStub;
    private final SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub;
    private final ManagedChannelWrapper managedChannel;
    private final List<Context.CancellableContext> cancellableContexts;
    private final ExponentialBackoffPolicy exponentialBackoffPolicy;

    // using a scheduled executor service to be able to schedule retries
    // it is injected in each of the retryHandlers in the watch methods
    private final ScheduledExecutorService retryExecutor;

    private final ExecutorService executorService;

    private volatile boolean closed;

    private DefaultWorkloadApiClient(final SpiffeWorkloadAPIStub workloadApiAsyncStub,
                                     final SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub,
                                     final ManagedChannelWrapper managedChannel,
                                     final ExponentialBackoffPolicy exponentialBackoffPolicy,
                                     final ScheduledExecutorService retryExecutor,
                                     final ExecutorService executorService) {
        this.workloadApiAsyncStub = workloadApiAsyncStub;
        this.workloadApiBlockingStub = workloadApiBlockingStub;
        this.managedChannel = managedChannel;
        this.cancellableContexts = Collections.synchronizedList(new ArrayList<>());
        this.exponentialBackoffPolicy = exponentialBackoffPolicy;
        this.retryExecutor = retryExecutor;
        this.executorService = executorService;
    }

    DefaultWorkloadApiClient(final SpiffeWorkloadAPIStub workloadApiAsyncStub,
                                    final SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub,
                                    final ManagedChannelWrapper managedChannel) {
        this.workloadApiAsyncStub = workloadApiAsyncStub;
        this.workloadApiBlockingStub = workloadApiBlockingStub;
        this.exponentialBackoffPolicy = ExponentialBackoffPolicy.DEFAULT;
        this.executorService = Executors.newCachedThreadPool();
        this.retryExecutor = Executors.newSingleThreadScheduledExecutor();
        this.cancellableContexts = Collections.synchronizedList(new ArrayList<>());
        this.managedChannel = managedChannel;
    }

    /**
     * Creates a new Workload API client using the default socket endpoint address.
     * {@link Address#getDefaultAddress()}
     *
     * @return a {@link WorkloadApiClient}, the instance concrete type is {@link DefaultWorkloadApiClient}
     * @throws SocketEndpointAddressException if the Workload API socket endpoint address is not valid
     */
    public static WorkloadApiClient newClient() throws SocketEndpointAddressException {
        val options = ClientOptions.builder().build();
        return newClient(options);
    }

    /**
     * Creates a new Workload API client configured with the given client options.
     * <p>
     * If the SPIFFE socket endpoint address is not provided in the options, it uses the default address.
     * {@link Address#getDefaultAddress()}
     *
     * @param options {@link ClientOptions}
     * @return a {@link WorkloadApiClient}, the instance concrete type is {@link DefaultWorkloadApiClient}
     * @throws SocketEndpointAddressException if the Workload API socket endpoint address is not valid
     */
    public static WorkloadApiClient newClient(@NonNull final ClientOptions options) throws SocketEndpointAddressException {

        val spiffeSocketPath = StringUtils.isNotBlank(options.spiffeSocketPath)
                ? options.spiffeSocketPath
                : Address.getDefaultAddress();

        if (options.exponentialBackoffPolicy == null) {
            options.exponentialBackoffPolicy = ExponentialBackoffPolicy.DEFAULT;
        }

        if (options.executorService == null) {
            options.executorService = Executors.newCachedThreadPool();
        }

        val socketEndpointAddress = Address.parseAddress(spiffeSocketPath);
        val managedChannel = GrpcManagedChannelFactory.newChannel(socketEndpointAddress, options.executorService);
        val securityHeaderInterceptor = new SecurityHeaderInterceptor();
        val workloadAPIAsyncStub = SpiffeWorkloadAPIGrpc
                .newStub(managedChannel.getChannel())
                .withExecutor(options.executorService)
                .withInterceptors(securityHeaderInterceptor);

        val workloadAPIBlockingStub = SpiffeWorkloadAPIGrpc
                .newBlockingStub(managedChannel.getChannel())
                .withExecutor(options.executorService)
                .withInterceptors(securityHeaderInterceptor);

        val retryExecutor = Executors.newSingleThreadScheduledExecutor();

        return new DefaultWorkloadApiClient(
                workloadAPIAsyncStub,
                workloadAPIBlockingStub,
                managedChannel,
                options.exponentialBackoffPolicy,
                retryExecutor,
                options.executorService);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public X509Context fetchX509Context() throws X509ContextException {
        try (val cancellableContext = Context.current().withCancellation()) {
            return cancellableContext.call(this::processX509Context);
        } catch (Exception e) {
            throw new X509ContextException("Error fetching X509Context", e);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void watchX509Context(@NonNull final Watcher<X509Context> watcher) {
        val retryHandler = new RetryHandler(exponentialBackoffPolicy, retryExecutor);
        val cancellableContext = Context.current().withCancellation();

        val streamObserver =
                getX509ContextStreamObserver(watcher, retryHandler, cancellableContext, workloadApiAsyncStub);

        cancellableContext.run(() -> workloadApiAsyncStub.fetchX509SVID(newX509SvidRequest(), streamObserver));
        this.cancellableContexts.add(cancellableContext);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public JwtSvid fetchJwtSvid(@NonNull String audience, String... extraAudience) throws JwtSvidException {
        final Set<String> audParam = createAudienceSet(audience, extraAudience);
        try (val cancellableContext = Context.current().withCancellation()) {
            return cancellableContext.call(() -> callFetchJwtSvid(audParam));
        } catch (Exception e) {
            throw new JwtSvidException("Error fetching JWT SVID", e);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public JwtSvid fetchJwtSvid(@NonNull final SpiffeId subject,
                                @NonNull final String audience,
                                final String... extraAudience)
            throws JwtSvidException {

        final Set<String> audParam = createAudienceSet(audience, extraAudience);

        try (val cancellableContext = Context.current().withCancellation()) {
            return cancellableContext.call(() -> callFetchJwtSvid(subject, audParam));
        } catch (Exception e) {
            throw new JwtSvidException("Error fetching JWT SVID", e);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    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);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public JwtSvid validateJwtSvid(@NonNull final String token, @NonNull final String audience)
            throws JwtSvidException {

        val request = createJwtSvidRequest(token, audience);

        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.singleton(audience));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void watchJwtBundles(@NonNull final Watcher<JwtBundleSet> watcher) {
        val retryHandler = new RetryHandler(exponentialBackoffPolicy, retryExecutor);
        val cancellableContext = Context.current().withCancellation();

        val streamObserver = getJwtBundleStreamObserver(watcher, retryHandler, cancellableContext, workloadApiAsyncStub);

        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) {
                for (val context : cancellableContexts) {
                    context.close();
                }

                if (managedChannel != null) {
                    managedChannel.close();
                }
                retryExecutor.shutdown();
                executorService.shutdown();
                closed = true;
            }
        }
        log.log(Level.INFO, "WorkloadAPI client is closed");
    }


    private X509Context processX509Context() throws X509ContextException {
        try {
            val x509SvidResponse = workloadApiBlockingStub.fetchX509SVID(newX509SvidRequest());
            if (x509SvidResponse.hasNext()) {
                return GrpcConversionUtils.toX509Context(x509SvidResponse.next());
            }
        } catch (CertificateException | X509SvidException e) {
            throw new X509ContextException("Error processing X509Context", e);
        }
        throw new X509ContextException("Error processing X509Context: x509SVIDResponse is empty");
    }

    private JwtSvid callFetchJwtSvid(final SpiffeId subject, final Set<String> audience) throws JwtSvidException {
        val jwtSvidRequest = Workload.JWTSVIDRequest.newBuilder()
                .setSpiffeId(subject.toString())
                .addAllAudience(audience)
                .build();
        val response = workloadApiBlockingStub.fetchJWTSVID(jwtSvidRequest);
        return JwtSvid.parseInsecure(response.getSvids(0).getSvid(), audience);
    }

    private JwtSvid callFetchJwtSvid(final Set<String> audience) throws JwtSvidException {
        val jwtSvidRequest = Workload.JWTSVIDRequest.newBuilder()
                .addAllAudience(audience)
                .build();
        val response = workloadApiBlockingStub.fetchJWTSVID(jwtSvidRequest);
        return JwtSvid.parseInsecure(response.getSvids(0).getSvid(), audience);
    }

    private JwtBundleSet callFetchBundles() throws JwtBundleException {
        val request = Workload.JWTBundlesRequest.newBuilder().build();
        val 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");
    }

    private Set<String> createAudienceSet(final @NonNull String audience, final String[] extraAudience) {
        final Set<String> audParam = new HashSet<>();
        audParam.add(audience);
        Collections.addAll(audParam, extraAudience);
        return audParam;
    }

    private Workload.X509SVIDRequest newX509SvidRequest() {
        return Workload.X509SVIDRequest.newBuilder().build();
    }

    private Workload.JWTBundlesRequest newJwtBundlesRequest() {
        return Workload.JWTBundlesRequest.newBuilder().build();
    }

    private Workload.ValidateJWTSVIDRequest createJwtSvidRequest(final @NonNull String token, final @NonNull String audience) {
        return Workload.ValidateJWTSVIDRequest
                .newBuilder()
                .setSvid(token)
                .setAudience(audience)
                .build();
    }

    /**
     * Options for creating a new {@link DefaultWorkloadApiClient}.
     * <p>
     * <code>spiffeSocketPath</code> Workload API Socket Endpoint address.
     * <p>
     * <code>backoffPolicy</code> A custom instance of a {@link ExponentialBackoffPolicy} to configure the retries to reconnect
     * to the Workload API.
     * <p>
     * <code>executorService</code> A custom {@link ExecutorService} to configure the Grpc stubs and channels.
     * If it is not provided, an <code>Executors.newCachedThreadPool()</code> is used by default.
     * The executorService provided will be shutdown when the WorkloadApiClient instance is closed.
     */
    @Data
    public static class ClientOptions {

        @Setter(AccessLevel.NONE)
        private String spiffeSocketPath;

        @Setter(AccessLevel.NONE)
        private ExponentialBackoffPolicy exponentialBackoffPolicy;

        @Setter(AccessLevel.NONE)
        private ExecutorService executorService;

        @Builder
        public ClientOptions(final String spiffeSocketPath,
                             final ExponentialBackoffPolicy exponentialBackoffPolicy,
                             final ExecutorService executorService) {
            this.spiffeSocketPath = spiffeSocketPath;
            this.exponentialBackoffPolicy = exponentialBackoffPolicy;
            this.executorService = executorService;
        }
    }
}

1	package io.spiffe.workloadapi;
2
3	import io.grpc.Context;
4	import io.spiffe.bundle.jwtbundle.JwtBundleSet;
5	import io.spiffe.exception.JwtBundleException;
6	import io.spiffe.exception.JwtSvidException;
7	import io.spiffe.exception.SocketEndpointAddressException;
8	import io.spiffe.exception.X509ContextException;
9	import io.spiffe.exception.X509SvidException;
10	import io.spiffe.spiffeid.SpiffeId;
11	import io.spiffe.svid.jwtsvid.JwtSvid;
12	import io.spiffe.workloadapi.grpc.SpiffeWorkloadAPIGrpc;
13	import io.spiffe.workloadapi.grpc.SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIBlockingStub;
14	import io.spiffe.workloadapi.grpc.SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIStub;
15	import io.spiffe.workloadapi.grpc.Workload;
16	import io.spiffe.workloadapi.internal.GrpcManagedChannelFactory;
17	import io.spiffe.workloadapi.internal.ManagedChannelWrapper;
18	import io.spiffe.workloadapi.internal.SecurityHeaderInterceptor;
19	import io.spiffe.workloadapi.retry.ExponentialBackoffPolicy;
20	import io.spiffe.workloadapi.retry.RetryHandler;
21	import lombok.AccessLevel;
22	import lombok.Builder;
23	import lombok.Data;
24	import lombok.NonNull;
25	import lombok.Setter;
26	import lombok.extern.java.Log;
27	import lombok.val;
28	import org.apache.commons.lang3.StringUtils;
29
30	import java.security.KeyException;
31	import java.security.cert.CertificateException;
32	import java.util.ArrayList;
33	import java.util.Collections;
34	import java.util.HashSet;
35	import java.util.List;
36	import java.util.Set;
37	import java.util.concurrent.ExecutorService;
38	import java.util.concurrent.Executors;
39	import java.util.concurrent.ScheduledExecutorService;
40	import java.util.logging.Level;
41
42	import static io.spiffe.workloadapi.StreamObservers.getJwtBundleStreamObserver;
43	import static io.spiffe.workloadapi.StreamObservers.getX509ContextStreamObserver;
44
45	/**
46	* Represents a client to interact with the Workload API.
47	* <p>
48	* Supports one-shot calls and watch updates for X.509 and JWT SVIDs and bundles.
49	* <p>
50	* The watch for updates methods support retries using an exponential backoff policy to reestablish
51	* the stream connection to the Workload API.
52	*/
53	@Log	3✔
54	public final class DefaultWorkloadApiClient implements WorkloadApiClient {
55
56	private final SpiffeWorkloadAPIStub workloadApiAsyncStub;
57	private final SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub;
58	private final ManagedChannelWrapper managedChannel;
59	private final List<Context.CancellableContext> cancellableContexts;
60	private final ExponentialBackoffPolicy exponentialBackoffPolicy;
61
62	// using a scheduled executor service to be able to schedule retries
63	// it is injected in each of the retryHandlers in the watch methods
64	private final ScheduledExecutorService retryExecutor;
65
66	private final ExecutorService executorService;
67
68	private volatile boolean closed;
69
70	private DefaultWorkloadApiClient(final SpiffeWorkloadAPIStub workloadApiAsyncStub,
71	final SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub,
72	final ManagedChannelWrapper managedChannel,
73	final ExponentialBackoffPolicy exponentialBackoffPolicy,
74	final ScheduledExecutorService retryExecutor,
75	final ExecutorService executorService) {	3✔
76	this.workloadApiAsyncStub = workloadApiAsyncStub;	3✔
77	this.workloadApiBlockingStub = workloadApiBlockingStub;	3✔
78	this.managedChannel = managedChannel;	3✔
79	this.cancellableContexts = Collections.synchronizedList(new ArrayList<>());	3✔
80	this.exponentialBackoffPolicy = exponentialBackoffPolicy;	3✔
81	this.retryExecutor = retryExecutor;	3✔
82	this.executorService = executorService;	3✔
83	}	3✔
84
85	DefaultWorkloadApiClient(final SpiffeWorkloadAPIStub workloadApiAsyncStub,
86	final SpiffeWorkloadAPIBlockingStub workloadApiBlockingStub,
87	final ManagedChannelWrapper managedChannel) {	3✔
88	this.workloadApiAsyncStub = workloadApiAsyncStub;	3✔
89	this.workloadApiBlockingStub = workloadApiBlockingStub;	3✔
90	this.exponentialBackoffPolicy = ExponentialBackoffPolicy.DEFAULT;	3✔
91	this.executorService = Executors.newCachedThreadPool();	3✔
92	this.retryExecutor = Executors.newSingleThreadScheduledExecutor();	3✔
93	this.cancellableContexts = Collections.synchronizedList(new ArrayList<>());	3✔
94	this.managedChannel = managedChannel;	3✔
95	}	3✔
96
97	/**
98	* Creates a new Workload API client using the default socket endpoint address.
99	* {@link Address#getDefaultAddress()}
100	*
101	* @return a {@link WorkloadApiClient}, the instance concrete type is {@link DefaultWorkloadApiClient}
102	* @throws SocketEndpointAddressException if the Workload API socket endpoint address is not valid
103	*/
104	public static WorkloadApiClient newClient() throws SocketEndpointAddressException {
105	val options = ClientOptions.builder().build();	3✔
106	return newClient(options);	3✔
107	}
108
109	/**
110	* Creates a new Workload API client configured with the given client options.
111	* <p>
112	* If the SPIFFE socket endpoint address is not provided in the options, it uses the default address.
113	* {@link Address#getDefaultAddress()}
114	*
115	* @param options {@link ClientOptions}
116	* @return a {@link WorkloadApiClient}, the instance concrete type is {@link DefaultWorkloadApiClient}
117	* @throws SocketEndpointAddressException if the Workload API socket endpoint address is not valid
118	*/
119	public static WorkloadApiClient newClient(@NonNull final ClientOptions options) throws SocketEndpointAddressException {	3✔
120
121	val spiffeSocketPath = StringUtils.isNotBlank(options.spiffeSocketPath)	3✔
122	? options.spiffeSocketPath	3✔
123	: Address.getDefaultAddress();	3✔
124
125	if (options.exponentialBackoffPolicy == null) {	3✔
126	options.exponentialBackoffPolicy = ExponentialBackoffPolicy.DEFAULT;	3✔
127	}
128
129	if (options.executorService == null) {	3✔
130	options.executorService = Executors.newCachedThreadPool();	3✔
131	}
132
133	val socketEndpointAddress = Address.parseAddress(spiffeSocketPath);	3✔
134	val managedChannel = GrpcManagedChannelFactory.newChannel(socketEndpointAddress, options.executorService);	3✔
135	val securityHeaderInterceptor = new SecurityHeaderInterceptor();	3✔
136	val workloadAPIAsyncStub = SpiffeWorkloadAPIGrpc	3✔
137	.newStub(managedChannel.getChannel())	3✔
138	.withExecutor(options.executorService)	3✔
139	.withInterceptors(securityHeaderInterceptor);	3✔
140
141	val workloadAPIBlockingStub = SpiffeWorkloadAPIGrpc	3✔
142	.newBlockingStub(managedChannel.getChannel())	3✔
143	.withExecutor(options.executorService)	3✔
144	.withInterceptors(securityHeaderInterceptor);	3✔
145
146	val retryExecutor = Executors.newSingleThreadScheduledExecutor();	3✔
147
148	return new DefaultWorkloadApiClient(	3✔
149	workloadAPIAsyncStub,
150	workloadAPIBlockingStub,
151	managedChannel,
152	options.exponentialBackoffPolicy,	3✔
153	retryExecutor,
154	options.executorService);	3✔
155	}
156
157	/**
158	* {@inheritDoc}
159	*/
160	@Override
161	public X509Context fetchX509Context() throws X509ContextException {
162	try (val cancellableContext = Context.current().withCancellation()) {	3✔
163	return cancellableContext.call(this::processX509Context);	3✔
164	} catch (Exception e) {	×
165	throw new X509ContextException("Error fetching X509Context", e);	×
166	}
167	}
168
169	/**
170	* {@inheritDoc}
171	*/
172	@Override
173	public void watchX509Context(@NonNull final Watcher<X509Context> watcher) {	3✔
174	val retryHandler = new RetryHandler(exponentialBackoffPolicy, retryExecutor);	3✔
175	val cancellableContext = Context.current().withCancellation();	3✔
176
177	val streamObserver =	3✔
178	getX509ContextStreamObserver(watcher, retryHandler, cancellableContext, workloadApiAsyncStub);	3✔
179
180	cancellableContext.run(() -> workloadApiAsyncStub.fetchX509SVID(newX509SvidRequest(), streamObserver));	3✔
181	this.cancellableContexts.add(cancellableContext);	3✔
182	}	3✔
183
184	/**
185	* {@inheritDoc}
186	*/
187	@Override
188	public JwtSvid fetchJwtSvid(@NonNull String audience, String... extraAudience) throws JwtSvidException {	×
189	final Set<String> audParam = createAudienceSet(audience, extraAudience);	×
190	try (val cancellableContext = Context.current().withCancellation()) {	×
191	return cancellableContext.call(() -> callFetchJwtSvid(audParam));	×
192	} catch (Exception e) {	×
193	throw new JwtSvidException("Error fetching JWT SVID", e);	×
194	}
195	}
196
197	/**
198	* {@inheritDoc}
199	*/
200	@Override
201	public JwtSvid fetchJwtSvid(@NonNull final SpiffeId subject,	3✔
202	@NonNull final String audience,	3✔
203	final String... extraAudience)
204	throws JwtSvidException {
205
206	final Set<String> audParam = createAudienceSet(audience, extraAudience);	3✔
207
208	try (val cancellableContext = Context.current().withCancellation()) {	3✔
209	return cancellableContext.call(() -> callFetchJwtSvid(subject, audParam));	3✔
210	} catch (Exception e) {	×
211	throw new JwtSvidException("Error fetching JWT SVID", e);	×
212	}
213	}
214
215	/**
216	* {@inheritDoc}
217	*/
218	@Override
219	public JwtBundleSet fetchJwtBundles() throws JwtBundleException {
220	try (val cancellableContext = Context.current().withCancellation()) {	3✔
221	return cancellableContext.call(this::callFetchBundles);	3✔
222	} catch (Exception e) {	×
223	throw new JwtBundleException("Error fetching JWT SVID", e);	×
224	}
225	}
226
227	/**
228	* {@inheritDoc}
229	*/
230	@Override
231	public JwtSvid validateJwtSvid(@NonNull final String token, @NonNull final String audience)	3✔
232	throws JwtSvidException {
233
234	val request = createJwtSvidRequest(token, audience);	3✔
235
236	try (val cancellableContext = Context.current().withCancellation()) {	3✔
237	cancellableContext.call(() -> workloadApiBlockingStub.validateJWTSVID(request));	3✔
238	} catch (Exception e) {	×
239	throw new JwtSvidException("Error validating JWT SVID", e);	×
240	}	3✔
241
242	return JwtSvid.parseInsecure(token, Collections.singleton(audience));	3✔
243	}
244
245	/**
246	* {@inheritDoc}
247	*/
248	@Override
249	public void watchJwtBundles(@NonNull final Watcher<JwtBundleSet> watcher) {	3✔
250	val retryHandler = new RetryHandler(exponentialBackoffPolicy, retryExecutor);	3✔
251	val cancellableContext = Context.current().withCancellation();	3✔
252
253	val streamObserver = getJwtBundleStreamObserver(watcher, retryHandler, cancellableContext, workloadApiAsyncStub);	3✔
254
255	cancellableContext.run(() -> workloadApiAsyncStub.fetchJWTBundles(newJwtBundlesRequest(), streamObserver));	3✔
256	this.cancellableContexts.add(cancellableContext);	3✔
257	}	3✔
258
259	/**
260	* Closes this Workload API closing the underlying channel,
261	* cancelling the contexts and shutdown the executor service.
262	*/
263	@Override
264	public void close() {
265	log.log(Level.FINE, "Closing WorkloadAPI client");	3✔
266	synchronized (this) {	3✔
267	if (!closed) {	3✔
268	for (val context : cancellableContexts) {	3✔
269	context.close();	3✔
270	}	3✔
271
272	if (managedChannel != null) {	3✔
273	managedChannel.close();	3✔
274	}
275	retryExecutor.shutdown();	3✔
276	executorService.shutdown();	3✔
277	closed = true;	3✔
278	}
279	}	3✔
280	log.log(Level.INFO, "WorkloadAPI client is closed");	3✔
281	}	3✔
282
283
284	private X509Context processX509Context() throws X509ContextException {
285	try {
286	val x509SvidResponse = workloadApiBlockingStub.fetchX509SVID(newX509SvidRequest());	3✔
287	if (x509SvidResponse.hasNext()) {	3✔
288	return GrpcConversionUtils.toX509Context(x509SvidResponse.next());	3✔
289	}
290	} catch (CertificateException \| X509SvidException e) {	×
291	throw new X509ContextException("Error processing X509Context", e);	×
292	}	×
293	throw new X509ContextException("Error processing X509Context: x509SVIDResponse is empty");	×
294	}
295
296	private JwtSvid callFetchJwtSvid(final SpiffeId subject, final Set<String> audience) throws JwtSvidException {
297	val jwtSvidRequest = Workload.JWTSVIDRequest.newBuilder()	3✔
298	.setSpiffeId(subject.toString())	3✔
299	.addAllAudience(audience)	3✔
300	.build();	3✔
301	val response = workloadApiBlockingStub.fetchJWTSVID(jwtSvidRequest);	3✔
302	return JwtSvid.parseInsecure(response.getSvids(0).getSvid(), audience);	3✔
303	}
304
305	private JwtSvid callFetchJwtSvid(final Set<String> audience) throws JwtSvidException {
306	val jwtSvidRequest = Workload.JWTSVIDRequest.newBuilder()	×
307	.addAllAudience(audience)	×
308	.build();	×
309	val response = workloadApiBlockingStub.fetchJWTSVID(jwtSvidRequest);	×
310	return JwtSvid.parseInsecure(response.getSvids(0).getSvid(), audience);	×
311	}
312
313	private JwtBundleSet callFetchBundles() throws JwtBundleException {
314	val request = Workload.JWTBundlesRequest.newBuilder().build();	3✔
315	val bundlesResponse = workloadApiBlockingStub.fetchJWTBundles(request);	3✔
316
317	if (bundlesResponse.hasNext()) {	3✔
318	try {
319	return GrpcConversionUtils.toBundleSet(bundlesResponse.next());	3✔
320	} catch (KeyException \| JwtBundleException e) {	×
321	throw new JwtBundleException("Error processing JWT Bundle response from Workload API", e);	×
322	}
323	}
324	throw new JwtBundleException("JWT Bundle response from the Workload API is empty");	×
325	}
326
327	private Set<String> createAudienceSet(final @NonNull String audience, final String[] extraAudience) {	3✔
328	final Set<String> audParam = new HashSet<>();	3✔
329	audParam.add(audience);	3✔
330	Collections.addAll(audParam, extraAudience);	3✔
331	return audParam;	3✔
332	}
333
334	private Workload.X509SVIDRequest newX509SvidRequest() {
335	return Workload.X509SVIDRequest.newBuilder().build();	3✔
336	}
337
338	private Workload.JWTBundlesRequest newJwtBundlesRequest() {
339	return Workload.JWTBundlesRequest.newBuilder().build();	3✔
340	}
341
342	private Workload.ValidateJWTSVIDRequest createJwtSvidRequest(final @NonNull String token, final @NonNull String audience) {	3✔
343	return Workload.ValidateJWTSVIDRequest	3✔
344	.newBuilder()	3✔
345	.setSvid(token)	3✔
346	.setAudience(audience)	3✔
347	.build();	3✔
348	}
349
350	/**
351	* Options for creating a new {@link DefaultWorkloadApiClient}.
352	* <p>
353	* <code>spiffeSocketPath</code> Workload API Socket Endpoint address.
354	* <p>
355	* <code>backoffPolicy</code> A custom instance of a {@link ExponentialBackoffPolicy} to configure the retries to reconnect
356	* to the Workload API.
357	* <p>
358	* <code>executorService</code> A custom {@link ExecutorService} to configure the Grpc stubs and channels.
359	* If it is not provided, an <code>Executors.newCachedThreadPool()</code> is used by default.
360	* The executorService provided will be shutdown when the WorkloadApiClient instance is closed.
361	*/
362	@Data
363	public static class ClientOptions {
364
365	@Setter(AccessLevel.NONE)
366	private String spiffeSocketPath;
367
368	@Setter(AccessLevel.NONE)
369	private ExponentialBackoffPolicy exponentialBackoffPolicy;
370
371	@Setter(AccessLevel.NONE)
372	private ExecutorService executorService;
373
374	@Builder
375	public ClientOptions(final String spiffeSocketPath,
376	final ExponentialBackoffPolicy exponentialBackoffPolicy,
377	final ExecutorService executorService) {	3✔
378	this.spiffeSocketPath = spiffeSocketPath;	3✔
379	this.exponentialBackoffPolicy = exponentialBackoffPolicy;	3✔
380	this.executorService = executorService;	3✔
381	}	3✔
382	}
383	}

spiffe / java-spiffe / 75

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous