Skip to content

Commit a523900

Browse files
xDS: fix DistributorWatcher thread safety in CertificateProvider (v1.85.x backport) (#13133)
Backport of #13087 to v1.85.x. --- A gRPC client using xDS-driven mTLS does two things concurrently: 1. Certificate provider: watches cert files on disk, periodically reloads them, and notifies subscribers when certs change. 2. Channel subscriptions: each gRPC channel that needs mTLS registers a watcher on the cert provider to get the TLS credentials. The `DistributorWatcher` sits in the middle and fans out: one cert provider pushes updates to it, and it distributes to N channel watchers. I think there are two conditions in this that can lead to a `ConcurrentModificationException`: 1. `updateSpiffeTrustMap` isn't synchronized (unlike `updateCertificate` and `updateTrustedRoots`), so it can race against `addWatcher`/`removeWatcher` from another thread. 2. if callbacks mutate their own subscription. A callchain (line numbers per 1.83.0 / 1084e97) for that is: ``` DistributorWatcher.onError() CertificateProvider.java:126 → for (watcher : downstreamWatchers) CertificateProvider.java:127 → watcher.onError(errorStatus) CertificateProvider.java:128 → DynamicSslContextProvider.onError() DynamicSslContextProvider.java:143 → callback.onException(error) DynamicSslContextProvider.java:145 → SslContextProviderSupplier.onException() SslContextProviderSupplier.java:79 → releaseSslContextProvider(toRelease) SslContextProviderSupplier.java:81 → TlsContextManagerImpl.releaseClient...() TlsContextManagerImpl.java:78 → ReferenceCountingMap.releaseInternal() ReferenceCountingMap.java:89 → if refcount == 0: value.close() ReferenceCountingMap.java:95 → CertProviderSslContextProvider.close() CertProviderSslContextProvider.java:216 → certHandle.close() CertProviderSslContextProvider.java:218 → Handle.close() CertificateProviderStore.java:62 → distWatcher.removeWatcher() CertificateProviderStore.java:64 <-- REENTRANT ``` We observed `ConcurrentModificationException` in `updateSpiffeTrustMap()` in production, consistent with the unsynchronized iteration described above. We also observed startup RPC deadlines during the rollout, but have not established the causal connection between these symptoms. These all occurred in a service that creates many (~60) channels on startup, while getting the CDS config over xDS. The service uses grpc-java / grpc-xds `1.83.0`. When we switched it to use mTLS with SPIFFE certificates, we got the following errors: ``` DEADLINE_EXCEEDED: CallOptions deadline exceeded after 4.998s. Name resolution delay 0.000000000 seconds. [closed=[], open=[[connecting_and_lb_delay=4999ms, was_still_waiting]]] ``` ``` io.grpc.StatusException: UNKNOWN at io.grpc.xds.internal.security.DynamicSslContextProvider.onError(DynamicSslContextProvider.java:145) at io.grpc.xds.internal.security.certprovider.CertificateProvider$DistributorWatcher.onError(CertificateProvider.java:128) at io.grpc.xds.internal.security.certprovider.FileWatcherCertificateProvider.checkAndReloadCertificates(FileWatcherCertificateProvider.java:149) Caused by: java.util.ConcurrentModificationException at java.base/java.util.HashMap$HashIterator.nextNode(HashMap.java:1606) at java.base/java.util.HashMap$KeyIterator.next(HashMap.java:1629) at io.grpc.xds.internal.security.certprovider.CertificateProvider$DistributorWatcher.updateSpiffeTrustMap(CertificateProvider.java:120) at io.grpc.xds.internal.security.certprovider.FileWatcherCertificateProvider.checkAndReloadCertificates(FileWatcherCertificateProvider.java:144) ``` ``` java.util.ConcurrentModificationException at java.util.HashMap$HashIterator.nextNode(HashMap.java:1606) at java.util.HashMap$KeyIterator.next(HashMap.java:1629) at CertificateProvider$DistributorWatcher.onError(CertificateProvider.java:127) at FileWatcherCertificateProvider.checkAndReloadCertificates(FileWatcherCertificateProvider.java:149) ``` Co-authored-by: Bram <bcleenders@gmail.com>
1 parent f1ba10e commit a523900

5 files changed

Lines changed: 273 additions & 73 deletions

File tree

‎xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java‎

Lines changed: 74 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -19,15 +19,19 @@
1919
import static com.google.common.base.Preconditions.checkNotNull;
2020

2121
import com.google.common.annotations.VisibleForTesting;
22+
import com.google.common.collect.ImmutableSet;
23+
import com.google.errorprone.annotations.concurrent.GuardedBy;
2224
import io.grpc.Status;
25+
import io.grpc.SynchronizationContext;
2326
import io.grpc.xds.internal.security.Closeable;
2427
import java.security.PrivateKey;
2528
import java.security.cert.X509Certificate;
26-
import java.util.Collections;
2729
import java.util.HashSet;
2830
import java.util.List;
2931
import java.util.Map;
3032
import java.util.Set;
33+
import java.util.logging.Level;
34+
import java.util.logging.Logger;
3135

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

47+
private static final Logger logger = Logger.getLogger(CertificateProvider.class.getName());
48+
4349
/** A watcher is registered to receive certificate updates. */
4450
public interface Watcher {
4551
void updateCertificate(PrivateKey key, List<X509Certificate> certChain);
@@ -53,95 +59,111 @@ public interface Watcher {
5359

5460
@VisibleForTesting
5561
public static final class DistributorWatcher implements Watcher {
56-
private PrivateKey privateKey;
57-
private List<X509Certificate> certChain;
58-
private List<X509Certificate> trustedRoots;
59-
private Map<String, List<X509Certificate>> spiffeTrustMap;
60-
61-
@VisibleForTesting
62-
final Set<Watcher> downstreamWatchers = new HashSet<>();
63-
64-
synchronized void addWatcher(Watcher watcher) {
65-
downstreamWatchers.add(watcher);
66-
if (privateKey != null && certChain != null) {
67-
sendLastCertificateUpdate(watcher);
68-
}
69-
if (trustedRoots != null) {
70-
sendLastTrustedRootsUpdate(watcher);
71-
}
72-
if (spiffeTrustMap != null) {
73-
sendLastSpiffeTrustMapUpdate(watcher);
62+
// All watcher callbacks must run in syncContext to avoid possibility of deadlock
63+
private final SynchronizationContext syncContext = new SynchronizationContext(
64+
new Thread.UncaughtExceptionHandler() {
65+
@Override
66+
public void uncaughtException(Thread t, Throwable e) {
67+
logger.log(Level.SEVERE, "Uncaught exception in DistributorWatcher callback", e);
68+
}
69+
});
70+
71+
@GuardedBy("this") private PrivateKey privateKey;
72+
@GuardedBy("this") private List<X509Certificate> certChain;
73+
@GuardedBy("this") private List<X509Certificate> trustedRoots;
74+
@GuardedBy("this") private Map<String, List<X509Certificate>> spiffeTrustMap;
75+
76+
@GuardedBy("this")
77+
private final Set<Watcher> downstreamWatchers = new HashSet<>();
78+
79+
void addWatcher(Watcher watcher) {
80+
synchronized (this) {
81+
downstreamWatchers.add(watcher);
82+
if (privateKey != null && certChain != null) {
83+
PrivateKey key = privateKey;
84+
List<X509Certificate> chain = certChain;
85+
syncContext.executeLater(() -> watcher.updateCertificate(key, chain));
86+
}
87+
if (trustedRoots != null) {
88+
List<X509Certificate> roots = trustedRoots;
89+
syncContext.executeLater(() -> watcher.updateTrustedRoots(roots));
90+
}
91+
if (spiffeTrustMap != null) {
92+
Map<String, List<X509Certificate>> map = spiffeTrustMap;
93+
syncContext.executeLater(() -> watcher.updateSpiffeTrustMap(map));
94+
}
7495
}
96+
syncContext.drain();
7597
}
7698

7799
synchronized void removeWatcher(Watcher watcher) {
78100
downstreamWatchers.remove(watcher);
79101
}
80102

81-
@VisibleForTesting public Set<Watcher> getDownstreamWatchers() {
82-
return Collections.unmodifiableSet(downstreamWatchers);
83-
}
84-
85-
private void sendLastCertificateUpdate(Watcher watcher) {
86-
watcher.updateCertificate(privateKey, certChain);
87-
}
88-
89-
private void sendLastTrustedRootsUpdate(Watcher watcher) {
90-
watcher.updateTrustedRoots(trustedRoots);
91-
}
92-
93-
private void sendLastSpiffeTrustMapUpdate(Watcher watcher) {
94-
watcher.updateSpiffeTrustMap(spiffeTrustMap);
103+
@VisibleForTesting public synchronized Set<Watcher> getDownstreamWatchers() {
104+
return ImmutableSet.copyOf(downstreamWatchers);
95105
}
96106

97107
@Override
98-
public synchronized void updateCertificate(PrivateKey key, List<X509Certificate> certChain) {
108+
public void updateCertificate(PrivateKey key, List<X509Certificate> certChain) {
99109
checkNotNull(key, "key");
100110
checkNotNull(certChain, "certChain");
101-
privateKey = key;
102-
this.certChain = certChain;
103-
for (Watcher watcher : downstreamWatchers) {
104-
sendLastCertificateUpdate(watcher);
111+
synchronized (this) {
112+
privateKey = key;
113+
this.certChain = certChain;
114+
for (Watcher watcher : downstreamWatchers) {
115+
syncContext.executeLater(() -> watcher.updateCertificate(key, certChain));
116+
}
105117
}
118+
syncContext.drain();
106119
}
107120

108121
@Override
109-
public synchronized void updateTrustedRoots(List<X509Certificate> trustedRoots) {
122+
public void updateTrustedRoots(List<X509Certificate> trustedRoots) {
110123
checkNotNull(trustedRoots, "trustedRoots");
111-
this.trustedRoots = trustedRoots;
112-
for (Watcher watcher : downstreamWatchers) {
113-
sendLastTrustedRootsUpdate(watcher);
124+
synchronized (this) {
125+
this.trustedRoots = trustedRoots;
126+
for (Watcher watcher : downstreamWatchers) {
127+
syncContext.executeLater(() -> watcher.updateTrustedRoots(trustedRoots));
128+
}
114129
}
130+
syncContext.drain();
115131
}
116132

117133
@Override
118134
public void updateSpiffeTrustMap(Map<String, List<X509Certificate>> spiffeTrustMap) {
119-
this.spiffeTrustMap = spiffeTrustMap;
120-
for (Watcher watcher : downstreamWatchers) {
121-
sendLastSpiffeTrustMapUpdate(watcher);
135+
synchronized (this) {
136+
this.spiffeTrustMap = spiffeTrustMap;
137+
for (Watcher watcher : downstreamWatchers) {
138+
syncContext.executeLater(() -> watcher.updateSpiffeTrustMap(spiffeTrustMap));
139+
}
122140
}
141+
syncContext.drain();
123142
}
124143

125144
@Override
126-
public synchronized void onError(Status errorStatus) {
127-
for (Watcher watcher : downstreamWatchers) {
128-
watcher.onError(errorStatus);
145+
public void onError(Status errorStatus) {
146+
synchronized (this) {
147+
for (Watcher watcher : downstreamWatchers) {
148+
syncContext.executeLater(() -> watcher.onError(errorStatus));
149+
}
129150
}
151+
syncContext.drain();
130152
}
131153

132-
X509Certificate getLastIdentityCert() {
154+
synchronized X509Certificate getLastIdentityCert() {
133155
if (certChain != null && !certChain.isEmpty()) {
134156
return certChain.get(0);
135157
}
136158
return null;
137159
}
138160

139-
void close() {
161+
synchronized void close() {
140162
downstreamWatchers.clear();
141163
clearValues();
142164
}
143165

144-
void clearValues() {
166+
synchronized void clearValues() {
145167
privateKey = null;
146168
certChain = null;
147169
trustedRoots = null;
@@ -152,7 +174,7 @@ void clearValues() {
152174
* Concrete subclasses will call this to register the {@link Watcher}.
153175
*
154176
* @param watcher to register
155-
* @param notifyCertUpdates if true, the provider is required to call the watcher’s
177+
* @param notifyCertUpdates if true, the provider is required to call the watcher's
156178
* updateCertificate method. Implies the Provider is capable of minting certificates.
157179
* Used by server-side and mTLS client-side. Note the Provider is always required
158180
* to call updateTrustedRoots to provide trusted-root updates.

‎xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderClientSslContextProviderTest.java‎

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -421,14 +421,7 @@ public void testProviderForClient_sslContextException_onError() throws Exception
421421

422422
TestCallback testCallback = new TestCallback(MoreExecutors.directExecutor());
423423
provider.addCallback(testCallback);
424-
try {
425-
watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE)));
426-
fail("exception expected");
427-
} catch (RuntimeException expected) {
428-
assertThat(expected)
429-
.hasMessageThat()
430-
.contains("only static certificateValidationContext expected");
431-
}
424+
watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE)));
432425
assertThat(testCallback.updatedThrowable).isNotNull();
433426
assertThat(testCallback.updatedThrowable)
434427
.hasCauseThat()

‎xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertProviderServerSslContextProviderTest.java‎

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -339,14 +339,7 @@ public void testProviderForServer_sslContextException_onError() throws Exception
339339

340340
TestCallback testCallback = new TestCallback(MoreExecutors.directExecutor());
341341
provider.addCallback(testCallback);
342-
try {
343-
watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE)));
344-
fail("exception expected");
345-
} catch (RuntimeException expected) {
346-
assertThat(expected)
347-
.hasMessageThat()
348-
.contains("only static certificateValidationContext expected");
349-
}
342+
watcherCaptor[0].updateTrustedRoots(ImmutableList.of(getCertFromResourceName(CA_PEM_FILE)));
350343
assertThat(testCallback.updatedThrowable).isNotNull();
351344
assertThat(testCallback.updatedThrowable)
352345
.hasCauseThat()

‎xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderStoreTest.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,7 @@ public void onePluginSameConfig_sameInstance() {
138138
(TestCertificateProvider) handle1.certProvider;
139139
assertThat(testCertificateProvider.startCalled).isEqualTo(1);
140140
CertificateProvider.DistributorWatcher distWatcher = testCertificateProvider.getWatcher();
141-
assertThat(distWatcher.downstreamWatchers).hasSize(2);
141+
assertThat(distWatcher.getDownstreamWatchers()).hasSize(2);
142142
PrivateKey testKey = mock(PrivateKey.class);
143143
X509Certificate cert = mock(X509Certificate.class);
144144
List<X509Certificate> testList = ImmutableList.of(cert);
@@ -154,7 +154,7 @@ public void onePluginSameConfig_sameInstance() {
154154
reset(mockWatcher2);
155155
handle1.close();
156156
assertThat(testCertificateProvider.closeCalled).isEqualTo(0);
157-
assertThat(distWatcher.downstreamWatchers).hasSize(1);
157+
assertThat(distWatcher.getDownstreamWatchers()).hasSize(1);
158158
testCertificateProvider.getWatcher().updateCertificate(testKey, testList);
159159
verify(mockWatcher1, never())
160160
.updateCertificate(any(PrivateKey.class), ArgumentMatchers.<X509Certificate>anyList());
@@ -189,7 +189,7 @@ public void onePluginSameConfig_secondWatcherAfterFirstNotify() {
189189
CertificateProvider.Watcher mockWatcher2 = mock(CertificateProvider.Watcher.class);
190190
CertificateProviderStore.Handle unused = certificateProviderStore.createOrGetProvider(
191191
"cert-name1", "plugin1", "config", mockWatcher2, true);
192-
assertThat(distWatcher.downstreamWatchers).hasSize(2);
192+
assertThat(distWatcher.getDownstreamWatchers()).hasSize(2);
193193
// updates sent to the second watcher
194194
verify(mockWatcher2, times(1)).updateCertificate(eq(testKey), eq(testList));
195195
verify(mockWatcher2, times(1)).updateTrustedRoots(eq(testList));
@@ -290,9 +290,9 @@ private static void checkDifferentInstances(
290290
assertThat(testCertificateProvider2.certProviderProvider)
291291
.isSameInstanceAs(certProviderProvider2);
292292
CertificateProvider.DistributorWatcher distWatcher1 = testCertificateProvider1.getWatcher();
293-
assertThat(distWatcher1.downstreamWatchers).hasSize(1);
293+
assertThat(distWatcher1.getDownstreamWatchers()).hasSize(1);
294294
CertificateProvider.DistributorWatcher distWatcher2 = testCertificateProvider2.getWatcher();
295-
assertThat(distWatcher2.downstreamWatchers).hasSize(1);
295+
assertThat(distWatcher2.getDownstreamWatchers()).hasSize(1);
296296
PrivateKey testKey1 = mock(PrivateKey.class);
297297
X509Certificate cert1 = mock(X509Certificate.class);
298298
List<X509Certificate> testList1 = ImmutableList.of(cert1);

0 commit comments

Comments
 (0)