spiffe / java-spiffe / 65

pending completion

Build # 65

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/ca27c80228ac5d0bf98aeb8f933fe70b3ee88090">ca27c8022<a href="https://github.com/spiffe/java-spiffe/commit/ca27c80228ac5d0bf98aeb8f933fe70b3ee88090">">Merge </a><a class="double-link" href="https://github.com/spiffe/java-spiffe/commit/<a class="double-link" href="https://github.com/spiffe/java-spiffe/commit/0d0f25454c8d221d5927cdcd4808937a081cf2ce">0d0f25454</a>">0d0f25454</a><a href="https://github.com/spiffe/java-spiffe/commit/ca27c80228ac5d0bf98aeb8f933fe70b3ee88090"> into 8c3866ee8">8c3866ee8</a>

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

Run Details

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

1002 of 1130 relevant lines covered (88.67%)

3.55 hits per line

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

48.08

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

package io.spiffe.workloadapi;

import io.grpc.Context;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import io.spiffe.bundle.jwtbundle.JwtBundleSet;
import io.spiffe.exception.JwtBundleException;
import io.spiffe.exception.X509ContextException;
import io.spiffe.exception.X509SvidException;
import io.spiffe.workloadapi.grpc.SpiffeWorkloadAPIGrpc;
import io.spiffe.workloadapi.grpc.Workload;
import io.spiffe.workloadapi.retry.RetryHandler;
import lombok.extern.java.Log;
import lombok.val;

import java.security.KeyException;
import java.security.cert.CertificateException;
import java.util.logging.Level;

@Log
final class StreamObservers {

    private static final String INVALID_ARGUMENT = "INVALID_ARGUMENT";

    private StreamObservers() {
    }

    static StreamObserver<Workload.X509SVIDResponse> getX509ContextStreamObserver(
            final Watcher<X509Context> watcher,
            final RetryHandler retryHandler,
            final Context.CancellableContext cancellableContext,
            final SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIStub workloadApiAsyncStub) {

        return new StreamObserver<Workload.X509SVIDResponse>() {
            @Override
            public void onNext(final Workload.X509SVIDResponse value) {
                try {
                    val x509Context = GrpcConversionUtils.toX509Context(value);
                    validateX509Context(x509Context);
                    watcher.onUpdate(x509Context);
                    retryHandler.reset();
                } catch (CertificateException | X509SvidException | X509ContextException e) {
                    watcher.onError(new X509ContextException("Error processing X.509 Context update", e));
                }
            }

            @Override
            public void onError(final Throwable t) {
                log.log(Level.SEVERE, "X.509 context observer error", t);
                handleWatchX509ContextError(t);
            }

            private void handleWatchX509ContextError(final Throwable t) {
                if (INVALID_ARGUMENT.equals(Status.fromThrowable(t).getCode().name())) {
                    watcher.onError(new X509ContextException("Canceling X.509 Context watch", t));
                } else {
                    log.log(Level.INFO, "Retrying connecting to Workload API to register X.509 context watcher");
                    retryHandler.scheduleRetry(() ->
                            cancellableContext.run(
                                    () -> workloadApiAsyncStub.fetchX509SVID(newX509SvidRequest(),
                                            this)));
                }
            }

            @Override
            public void onCompleted() {
                cancellableContext.close();
                log.info("Workload API stream is completed");
            }
        };
    }

    static StreamObserver<Workload.JWTBundlesResponse> getJwtBundleStreamObserver(
            final Watcher<JwtBundleSet> watcher,
            final RetryHandler retryHandler,
            final Context.CancellableContext cancellableContext,
            final SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIStub workloadApiAsyncStub) {
        return new StreamObserver<Workload.JWTBundlesResponse>() {

            @Override
            public void onNext(final Workload.JWTBundlesResponse value) {
                try {
                    val 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(final Throwable t) {
                log.log(Level.SEVERE, "JWT observer error", t);
                handleWatchJwtBundleError(t);
            }

            private void handleWatchJwtBundleError(final 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");
            }
        };
    }

    // validates that the X.509 context has both the SVID and the bundles
    private static void validateX509Context(final X509Context x509Context) throws X509ContextException {
        if (x509Context.getX509BundleSet() == null
                || x509Context.getX509BundleSet().getBundles() == null
                || x509Context.getX509BundleSet().getBundles().isEmpty()) {
            throw new X509ContextException("X.509 context error: no X.509 bundles found");
        }

        if (x509Context.getX509Svid() == null || x509Context.getX509Svid().isEmpty()) {
            throw new X509ContextException("X.509 context error: no X.509 SVID found");
        }
    }

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

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

1	package io.spiffe.workloadapi;
2
3	import io.grpc.Context;
4	import io.grpc.Status;
5	import io.grpc.stub.StreamObserver;
6	import io.spiffe.bundle.jwtbundle.JwtBundleSet;
7	import io.spiffe.exception.JwtBundleException;
8	import io.spiffe.exception.X509ContextException;
9	import io.spiffe.exception.X509SvidException;
10	import io.spiffe.workloadapi.grpc.SpiffeWorkloadAPIGrpc;
11	import io.spiffe.workloadapi.grpc.Workload;
12	import io.spiffe.workloadapi.retry.RetryHandler;
13	import lombok.extern.java.Log;
14	import lombok.val;
15
16	import java.security.KeyException;
17	import java.security.cert.CertificateException;
18	import java.util.logging.Level;
19
20	@Log	4✔
21	final class StreamObservers {
22
23	private static final String INVALID_ARGUMENT = "INVALID_ARGUMENT";
24
25	private StreamObservers() {
26	}
27
28	static StreamObserver<Workload.X509SVIDResponse> getX509ContextStreamObserver(
29	final Watcher<X509Context> watcher,
30	final RetryHandler retryHandler,
31	final Context.CancellableContext cancellableContext,
32	final SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIStub workloadApiAsyncStub) {
33
34	return new StreamObserver<Workload.X509SVIDResponse>() {	4✔
35	@Override
36	public void onNext(final Workload.X509SVIDResponse value) {
37	try {
38	val x509Context = GrpcConversionUtils.toX509Context(value);	4✔
39	validateX509Context(x509Context);	4✔
40	watcher.onUpdate(x509Context);	4✔
41	retryHandler.reset();	4✔
42	} catch (CertificateException \| X509SvidException \| X509ContextException e) {	×
43	watcher.onError(new X509ContextException("Error processing X.509 Context update", e));	×
44	}	4✔
45	}	4✔
46
47	@Override
48	public void onError(final Throwable t) {
49	log.log(Level.SEVERE, "X.509 context observer error", t);	×
50	handleWatchX509ContextError(t);	×
51	}	×
52
53	private void handleWatchX509ContextError(final Throwable t) {
54	if (INVALID_ARGUMENT.equals(Status.fromThrowable(t).getCode().name())) {	×
55	watcher.onError(new X509ContextException("Canceling X.509 Context watch", t));	×
56	} else {
57	log.log(Level.INFO, "Retrying connecting to Workload API to register X.509 context watcher");	×
58	retryHandler.scheduleRetry(() ->	×
59	cancellableContext.run(	×
60	() -> workloadApiAsyncStub.fetchX509SVID(newX509SvidRequest(),	×
61	this)));
62	}
63	}	×
64
65	@Override
66	public void onCompleted() {
67	cancellableContext.close();	4✔
68	log.info("Workload API stream is completed");	4✔
69	}	4✔
70	};
71	}
72
73	static StreamObserver<Workload.JWTBundlesResponse> getJwtBundleStreamObserver(
74	final Watcher<JwtBundleSet> watcher,
75	final RetryHandler retryHandler,
76	final Context.CancellableContext cancellableContext,
77	final SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIStub workloadApiAsyncStub) {
78	return new StreamObserver<Workload.JWTBundlesResponse>() {	4✔
79
80	@Override
81	public void onNext(final Workload.JWTBundlesResponse value) {
82	try {
83	val jwtBundleSet = GrpcConversionUtils.toBundleSet(value);	4✔
84	watcher.onUpdate(jwtBundleSet);	4✔
85	retryHandler.reset();	4✔
86	} catch (KeyException \| JwtBundleException e) {	×
87	watcher.onError(new JwtBundleException("Error processing JWT bundles update", e));	×
88	}	4✔
89	}	4✔
90
91	@Override
92	public void onError(final Throwable t) {
93	log.log(Level.SEVERE, "JWT observer error", t);	×
94	handleWatchJwtBundleError(t);	×
95	}	×
96
97	private void handleWatchJwtBundleError(final Throwable t) {
98	if (INVALID_ARGUMENT.equals(Status.fromThrowable(t).getCode().name())) {	×
99	watcher.onError(new JwtBundleException("Canceling JWT Bundles watch", t));	×
100	} else {
101	log.log(Level.INFO, "Retrying connecting to Workload API to register JWT Bundles watcher");	×
102	retryHandler.scheduleRetry(() ->	×
103	cancellableContext.run(() -> workloadApiAsyncStub.fetchJWTBundles(newJwtBundlesRequest(),	×
104	this)));
105	}
106	}	×
107
108	@Override
109	public void onCompleted() {
110	cancellableContext.close();	4✔
111	log.info("Workload API stream is completed");	4✔
112	}	4✔
113	};
114	}
115
116	// validates that the X.509 context has both the SVID and the bundles
117	private static void validateX509Context(final X509Context x509Context) throws X509ContextException {
118	if (x509Context.getX509BundleSet() == null	4✔
119	\|\| x509Context.getX509BundleSet().getBundles() == null	4✔
120	\|\| x509Context.getX509BundleSet().getBundles().isEmpty()) {	4✔
121	throw new X509ContextException("X.509 context error: no X.509 bundles found");	×
122	}
123
124	if (x509Context.getX509Svid() == null \|\| x509Context.getX509Svid().isEmpty()) {	4✔
125	throw new X509ContextException("X.509 context error: no X.509 SVID found");	×
126	}
127	}	4✔
128
129	private static Workload.X509SVIDRequest newX509SvidRequest() {
130	return Workload.X509SVIDRequest.newBuilder().build();	×
131	}
132
133	private static Workload.JWTBundlesRequest newJwtBundlesRequest() {
134	return Workload.JWTBundlesRequest.newBuilder().build();	×
135	}
136	}

spiffe / java-spiffe / 65

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