Skip to content

xDS: fix DistributorWatcher thread safety in CertificateProvider (v1.85.x backport) #13133

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kannanjgithub merged 1 commit into from
Oct 19, 2026
Merged

xDS: fix DistributorWatcher thread safety in CertificateProvider (v1.85.x backport) #13133

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
126 changes: 74 additions & 52 deletions xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,15 +19,19 @@
import static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.xds.internal.security.Closeable;
import java.security.PrivateKey;
import java.security.cert.X509Certificate;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
* A plug-in that provides certificates required by the xDS security component and created
Expand All @@ -40,6 +44,8 @@
*/
public abstract class CertificateProvider implements Closeable {

private static final Logger logger = Logger.getLogger(CertificateProvider.class.getName());

/** A watcher is registered to receive certificate updates. */
public interface Watcher {
void updateCertificate(PrivateKey key, List<X509Certificate> certChain);
Expand All @@ -53,95 +59,111 @@ public interface Watcher {

@VisibleForTesting
public static final class DistributorWatcher implements Watcher {
private PrivateKey privateKey;
private List<X509Certificate> certChain;
private List<X509Certificate> trustedRoots;
private Map<String, List<X509Certificate>> spiffeTrustMap;

@VisibleForTesting
final Set<Watcher> downstreamWatchers = new HashSet<>();

synchronized void addWatcher(Watcher watcher) {
downstreamWatchers.add(watcher);
if (privateKey != null && certChain != null) {
sendLastCertificateUpdate(watcher);
}
if (trustedRoots != null) {
sendLastTrustedRootsUpdate(watcher);
}
if (spiffeTrustMap != null) {
sendLastSpiffeTrustMapUpdate(watcher);
// All watcher callbacks must run in syncContext to avoid possibility of deadlock
private final SynchronizationContext syncContext = new SynchronizationContext(
new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
logger.log(Level.SEVERE, "Uncaught exception in DistributorWatcher callback", e);
}
});

@GuardedBy("this") private PrivateKey privateKey;
@GuardedBy("this") private List<X509Certificate> certChain;
@GuardedBy("this") private List<X509Certificate> trustedRoots;
@GuardedBy("this") private Map<String, List<X509Certificate>> spiffeTrustMap;

@GuardedBy("this")
private final Set<Watcher> downstreamWatchers = new HashSet<>();

void addWatcher(Watcher watcher) {
synchronized (this) {
downstreamWatchers.add(watcher);
if (privateKey != null && certChain != null) {
PrivateKey key = privateKey;
List<X509Certificate> chain = certChain;
syncContext.executeLater(() -> watcher.updateCertificate(key, chain));
}
if (trustedRoots != null) {
List<X509Certificate> roots = trustedRoots;
syncContext.executeLater(() -> watcher.updateTrustedRoots(roots));
}
if (spiffeTrustMap != null) {
Map<String, List<X509Certificate>> map = spiffeTrustMap;
syncContext.executeLater(() -> watcher.updateSpiffeTrustMap(map));
}
}
syncContext.drain();
}

synchronized void removeWatcher(Watcher watcher) {
downstreamWatchers.remove(watcher);
}

@VisibleForTesting public Set<Watcher> getDownstreamWatchers() {
return Collections.unmodifiableSet(downstreamWatchers);
}

private void sendLastCertificateUpdate(Watcher watcher) {
watcher.updateCertificate(privateKey, certChain);
}

private void sendLastTrustedRootsUpdate(Watcher watcher) {
watcher.updateTrustedRoots(trustedRoots);
}

private void sendLastSpiffeTrustMapUpdate(Watcher watcher) {
watcher.updateSpiffeTrustMap(spiffeTrustMap);
@VisibleForTesting public synchronized Set<Watcher> getDownstreamWatchers() {
return ImmutableSet.copyOf(downstreamWatchers);
}

@Override
public synchronized void updateCertificate(PrivateKey key, List<X509Certificate> certChain) {
public void updateCertificate(PrivateKey key, List<X509Certificate> certChain) {
checkNotNull(key, "key");
checkNotNull(certChain, "certChain");
privateKey = key;
this.certChain = certChain;
for (Watcher watcher : downstreamWatchers) {
sendLastCertificateUpdate(watcher);
synchronized (this) {
privateKey = key;
this.certChain = certChain;
for (Watcher watcher : downstreamWatchers) {
syncContext.executeLater(() -> watcher.updateCertificate(key, certChain));
}
}
syncContext.drain();
}

@Override
public synchronized void updateTrustedRoots(List<X509Certificate> trustedRoots) {
public void updateTrustedRoots(List<X509Certificate> trustedRoots) {
checkNotNull(trustedRoots, "trustedRoots");
this.trustedRoots = trustedRoots;
for (Watcher watcher : downstreamWatchers) {
sendLastTrustedRootsUpdate(watcher);
synchronized (this) {
this.trustedRoots = trustedRoots;
for (Watcher watcher : downstreamWatchers) {
syncContext.executeLater(() -> watcher.updateTrustedRoots(trustedRoots));
}
}
syncContext.drain();
}

@Override
public void updateSpiffeTrustMap(Map<String, List<X509Certificate>> spiffeTrustMap) {
this.spiffeTrustMap = spiffeTrustMap;
for (Watcher watcher : downstreamWatchers) {
sendLastSpiffeTrustMapUpdate(watcher);
synchronized (this) {
this.spiffeTrustMap = spiffeTrustMap;
for (Watcher watcher : downstreamWatchers) {
syncContext.executeLater(() -> watcher.updateSpiffeTrustMap(spiffeTrustMap));
}
}
syncContext.drain();
}

@Override
public synchronized void onError(Status errorStatus) {
for (Watcher watcher : downstreamWatchers) {
watcher.onError(errorStatus);
public void onError(Status errorStatus) {
synchronized (this) {
for (Watcher watcher : downstreamWatchers) {
syncContext.executeLater(() -> watcher.onError(errorStatus));
}
}
syncContext.drain();
}

X509Certificate getLastIdentityCert() {
synchronized X509Certificate getLastIdentityCert() {
if (certChain != null && !certChain.isEmpty()) {
return certChain.get(0);
}
return null;
}

void close() {
synchronized void close() {
downstreamWatchers.clear();
clearValues();
}

void clearValues() {
synchronized void clearValues() {
privateKey = null;
certChain = null;
trustedRoots = null;
Expand All @@ -152,7 +174,7 @@ void clearValues() {
* Concrete subclasses will call this to register the {@link Watcher}.
*
* @param watcher to register
* @param notifyCertUpdates if true, the provider is required to call the watcher’s
* @param notifyCertUpdates if true, the provider is required to call the watcher's
* updateCertificate method. Implies the Provider is capable of minting certificates.
* Used by server-side and mTLS client-side. Note the Provider is always required
* to call updateTrustedRoots to provide trusted-root updates.
Expand Down
9 changes: 1 addition & 8 deletions .../io/grpc/xds/internal/security/certprovider/CertProviderClientSslContextProviderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -421,14 +421,7 @@ public void testProviderForClient_sslContextException_onError() throws Exception

TestCallback testCallback = new TestCallback(MoreExecutors.directExecutor());
provider.addCallback(testCallback);
try {
watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE)));
fail("exception expected");
} catch (RuntimeException expected) {
assertThat(expected)
.hasMessageThat()
.contains("only static certificateValidationContext expected");
}
watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE)));
assertThat(testCallback.updatedThrowable).isNotNull();
assertThat(testCallback.updatedThrowable)
.hasCauseThat()
Expand Down
9 changes: 1 addition & 8 deletions .../io/grpc/xds/internal/security/certprovider/CertProviderServerSslContextProviderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -339,14 +339,7 @@ public void testProviderForServer_sslContextException_onError() throws Exception

TestCallback testCallback = new TestCallback(MoreExecutors.directExecutor());
provider.addCallback(testCallback);
try {
watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE)));
fail("exception expected");
} catch (RuntimeException expected) {
assertThat(expected)
.hasMessageThat()
.contains("only static certificateValidationContext expected");
}
watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE)));
assertThat(testCallback.updatedThrowable).isNotNull();
assertThat(testCallback.updatedThrowable)
.hasCauseThat()
Expand Down
10 changes: 5 additions & 5 deletions ...rc/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ public void onePluginSameConfig_sameInstance() {
(TestCertificateProvider) handle1.certProvider;
assertThat(testCertificateProvider.startCalled).isEqualTo(1);
CertificateProvider.DistributorWatcher distWatcher = testCertificateProvider.getWatcher();
assertThat(distWatcher.downstreamWatchers).hasSize(2);
assertThat(distWatcher.getDownstreamWatchers()).hasSize(2);
PrivateKey testKey = mock(PrivateKey.class);
X509Certificate cert = mock(X509Certificate.class);
List<X509Certificate> testList = ImmutableList.of(cert);
Expand All @@ -154,7 +154,7 @@ public void onePluginSameConfig_sameInstance() {
reset(mockWatcher2);
handle1.close();
assertThat(testCertificateProvider.closeCalled).isEqualTo(0);
assertThat(distWatcher.downstreamWatchers).hasSize(1);
assertThat(distWatcher.getDownstreamWatchers()).hasSize(1);
testCertificateProvider.getWatcher().updateCertificate(testKey, testList);
verify(mockWatcher1, never())
.updateCertificate(any(PrivateKey.class), ArgumentMatchers.<X509Certificate>anyList());
Expand Down Expand Up @@ -189,7 +189,7 @@ public void onePluginSameConfig_secondWatcherAfterFirstNotify() {
CertificateProvider.Watcher mockWatcher2 = mock(CertificateProvider.Watcher.class);
CertificateProviderStore.Handle unused = certificateProviderStore.createOrGetProvider(
"cert-name1", "plugin1", "config", mockWatcher2, true);
assertThat(distWatcher.downstreamWatchers).hasSize(2);
assertThat(distWatcher.getDownstreamWatchers()).hasSize(2);
// updates sent to the second watcher
verify(mockWatcher2, times(1)).updateCertificate(eq(testKey), eq(testList));
verify(mockWatcher2, times(1)).updateTrustedRoots(eq(testList));
Expand Down Expand Up @@ -290,9 +290,9 @@ private static void checkDifferentInstances(
assertThat(testCertificateProvider2.certProviderProvider)
.isSameInstanceAs(certProviderProvider2);
CertificateProvider.DistributorWatcher distWatcher1 = testCertificateProvider1.getWatcher();
assertThat(distWatcher1.downstreamWatchers).hasSize(1);
assertThat(distWatcher1.getDownstreamWatchers()).hasSize(1);
CertificateProvider.DistributorWatcher distWatcher2 = testCertificateProvider2.getWatcher();
assertThat(distWatcher2.downstreamWatchers).hasSize(1);
assertThat(distWatcher2.getDownstreamWatchers()).hasSize(1);
PrivateKey testKey1 = mock(PrivateKey.class);
X509Certificate cert1 = mock(X509Certificate.class);
List<X509Certificate> testList1 = ImmutableList.of(cert1);
Expand Down
Loading
Loading