From d3568985d0e88e86ca1309df1d0794150c03324c Mon Sep 17 00:00:00 2001 From: NopeCharon <115794254+NopeCharon@users.noreply.github.com> Date: Fri, 30 Oct 2026 01:02:18 +0530 Subject: [PATCH] fix(workloadapi): keep X.509 context watch alive after rejected updates DefaultX509Source threw a WatcherException from the X.509 context watcher's onError. StreamObservers calls onError from inside StreamObserver.onNext when GrpcConversionUtils.toX509Context rejects an update (for example a corrupted certificate or key, or a SPIFFE ID mismatch), so the exception escaped onNext and gRPC cancelled the call. CANCELLED is non-retryable, so the watch ended permanently: the source kept serving its current SVID, never applied a later rotation, and TLS built on it started failing once that SVID expired. Apply the same handling the JWT sources got: only errors before the first update are initialization failures. The watcher records the error and init throws it, so newSource also fails when the first update is rejected asynchronously instead of returning an uninitialized source. After initialization, errors are logged and the current SVID and bundles are kept, so the stream stays open and the next valid update replaces them. Fixes #486 Signed-off-by: NopeCharon <115794254+NopeCharon@users.noreply.github.com> --- .../spiffe/workloadapi/DefaultX509Source.java | 19 ++++-- .../workloadapi/DefaultX509SourceTest.java | 67 ++++++++++++++++++- .../workloadapi/WorkloadApiClientStub.java | 3 + 3 files changed, 84 insertions(+), 5 deletions(-) diff --git a/java-spiffe-core/src/main/java/io/spiffe/workloadapi/DefaultX509Source.java b/java-spiffe-core/src/main/java/io/spiffe/workloadapi/DefaultX509Source.java index 33b22ff0..b3061a18 100644 --- a/java-spiffe-core/src/main/java/io/spiffe/workloadapi/DefaultX509Source.java +++ b/java-spiffe-core/src/main/java/io/spiffe/workloadapi/DefaultX509Source.java @@ -19,6 +19,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.logging.Level; import java.util.logging.Logger; @@ -201,7 +202,8 @@ private static WorkloadApiClient createClient(final X509SourceOptions options) private void init(final Duration timeout) throws TimeoutException { CountDownLatch done = new CountDownLatch(1); - setX509ContextWatcher(done); + AtomicReference initError = new AtomicReference<>(); + setX509ContextWatcher(done, initError); final boolean success; if (timeout.isZero()) { @@ -213,9 +215,13 @@ private void init(final Duration timeout) throws TimeoutException { if (!success) { throw new TimeoutException("Timeout waiting for X.509 Context update"); } + Throwable error = initError.get(); + if (error != null) { + throw new WatcherException("Error in X509Context watcher", error); + } } - private void setX509ContextWatcher(final CountDownLatch done) { + private void setX509ContextWatcher(final CountDownLatch done, final AtomicReference initError) { workloadApiClient.watchX509Context(new Watcher() { @Override public void onUpdate(final X509Context update) { @@ -228,8 +234,13 @@ public void onUpdate(final X509Context update) { @Override public void onError(final Throwable error) { log.log(Level.SEVERE, "Error in X509Context watcher", error); - done.countDown(); - throw new WatcherException("Error in X509Context watcher", error); + // An error before the first update fails the initialization. After that, the current SVID and + // bundles are kept: throwing here would escape the gRPC stream observer and cancel the watch, so + // later X.509 context updates (SVID rotations) would never be applied. + if (done.getCount() > 0) { + initError.compareAndSet(null, error); + done.countDown(); + } } }); } diff --git a/java-spiffe-core/src/test/java/io/spiffe/workloadapi/DefaultX509SourceTest.java b/java-spiffe-core/src/test/java/io/spiffe/workloadapi/DefaultX509SourceTest.java index 43581d6a..c84e9da8 100644 --- a/java-spiffe-core/src/test/java/io/spiffe/workloadapi/DefaultX509SourceTest.java +++ b/java-spiffe-core/src/test/java/io/spiffe/workloadapi/DefaultX509SourceTest.java @@ -1,21 +1,33 @@ package io.spiffe.workloadapi; +import com.google.protobuf.ByteString; +import io.grpc.stub.StreamObserver; +import io.grpc.testing.GrpcCleanupRule; import io.spiffe.bundle.x509bundle.X509Bundle; +import io.spiffe.bundle.x509bundle.X509BundleSet; import io.spiffe.exception.BundleNotFoundException; import io.spiffe.exception.SocketEndpointAddressException; +import io.spiffe.exception.X509ContextException; import io.spiffe.exception.X509SourceException; import io.spiffe.spiffeid.SpiffeId; import io.spiffe.spiffeid.TrustDomain; import io.spiffe.svid.x509svid.X509Svid; +import io.spiffe.workloadapi.grpc.SpiffeWorkloadAPIGrpc; +import io.spiffe.workloadapi.grpc.Workload; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import uk.org.webcompere.systemstubs.environment.EnvironmentVariables; +import java.nio.file.Files; +import java.nio.file.Paths; import java.time.Duration; +import java.util.Collections; +import static io.spiffe.utils.TestUtils.toUri; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -96,6 +108,44 @@ void testGetX509Svid_SourceIsClosed_ThrowsIllegalStateException() { } } + @Test + void testX509ContextWatcherError_afterInit_keepsX509ContextAndAppliesLaterUpdates() throws BundleNotFoundException { + TrustDomain trustDomain = TrustDomain.parse("example.org"); + X509Svid initialSvid = x509Source.getX509Svid(); + X509Bundle initialBundle = x509Source.getBundleForTrustDomain(trustDomain); + + workloadApiClient.x509ContextWatcher.onError(new X509ContextException("Error processing X.509 Context update")); + assertSame(initialSvid, x509Source.getX509Svid()); + assertSame(initialBundle, x509Source.getBundleForTrustDomain(trustDomain)); + + X509Bundle updatedBundle = new X509Bundle(trustDomain); + X509Context update = X509Context.of(Collections.singletonList(initialSvid), X509BundleSet.of(Collections.singleton(updatedBundle))); + workloadApiClient.x509ContextWatcher.onUpdate(update); + assertSame(updatedBundle, x509Source.getBundleForTrustDomain(trustDomain)); + } + + @Test + void testWatchX509Context_corruptedUpdate_laterUpdatesAreApplied() throws Exception { + byte[] svid = Files.readAllBytes(Paths.get(toUri("testdata/workloadapi/svid.der"))); + byte[] key = Files.readAllBytes(Paths.get(toUri("testdata/workloadapi/svid.key.der"))); + byte[] bundle = Files.readAllBytes(Paths.get(toUri("testdata/workloadapi/bundle.der"))); + byte[] corrupted = Files.readAllBytes(Paths.get(toUri("testdata/workloadapi/corrupted"))); + SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIImplBase fakeWorkloadApi = new SpiffeWorkloadAPIGrpc.SpiffeWorkloadAPIImplBase() { + @Override + public void fetchX509SVID(Workload.X509SVIDRequest request, StreamObserver responseObserver) { + responseObserver.onNext(newX509SvidResponse(svid, key, bundle, "initial")); + responseObserver.onNext(newX509SvidResponse(corrupted, corrupted, corrupted, "corrupted")); + responseObserver.onNext(newX509SvidResponse(svid, key, bundle, "rotated")); + } + }; + DefaultWorkloadApiClient client = WorkloadApiClientTestUtil.create(fakeWorkloadApi, new GrpcCleanupRule()); + DefaultX509Source.X509SourceOptions options = DefaultX509Source.X509SourceOptions.builder().workloadApiClient(client).build(); + + try (DefaultX509Source source = DefaultX509Source.newSource(options)) { + assertEquals("rotated", source.getX509Svid().getHint()); + } + } + @Test void newSource_success() { DefaultX509Source.X509SourceOptions options = DefaultX509Source.X509SourceOptions @@ -172,4 +222,19 @@ void newSource_noSocketAddress() throws Exception { } }); } -} \ No newline at end of file + + private static Workload.X509SVIDResponse newX509SvidResponse(byte[] svid, byte[] key, byte[] bundle, String hint) { + Workload.X509SVID x509Svid = Workload.X509SVID + .newBuilder() + .setSpiffeId("spiffe://example.org/workload-server") + .setX509Svid(ByteString.copyFrom(svid)) + .setX509SvidKey(ByteString.copyFrom(key)) + .setBundle(ByteString.copyFrom(bundle)) + .setHint(hint) + .build(); + return Workload.X509SVIDResponse + .newBuilder() + .addSvids(x509Svid) + .build(); + } +} diff --git a/java-spiffe-core/src/test/java/io/spiffe/workloadapi/WorkloadApiClientStub.java b/java-spiffe-core/src/test/java/io/spiffe/workloadapi/WorkloadApiClientStub.java index 13985c8c..6bc27a0e 100644 --- a/java-spiffe-core/src/test/java/io/spiffe/workloadapi/WorkloadApiClientStub.java +++ b/java-spiffe-core/src/test/java/io/spiffe/workloadapi/WorkloadApiClientStub.java @@ -39,6 +39,8 @@ public class WorkloadApiClientStub implements WorkloadApiClient { int fetchJwtSvidCallCount = 0; + Watcher x509ContextWatcher; + Watcher jwtBundlesWatcher; boolean closed; @@ -52,6 +54,7 @@ public X509Context fetchX509Context() { @Override public void watchX509Context(Watcher watcher) { + x509ContextWatcher = watcher; X509Context update = generateX509Context(); watcher.onUpdate(update); }