Skip to content

Commit 1514bbc

Browse files
jaymodejaymode
authored
Security: propagate auth result to listeners (#36900)
After #30794, our caching realms limit each principal to a single auth attempt at a time. This prevents hammering of external servers but can cause a significant performance hit when requests need to go through a realm that takes a long time to attempt to authenticate in order to get to the realm that actually authenticates. In order to address this, this change will propagate failed results to listeners if they use the same set of credentials that the authentication attempt used. This does prevent these stalled requests from retrying the authentication attempt but the implementation does allow for new requests to retry the attempt.
1 parent dd69553 commit 1514bbc

File tree

2 files changed

+131
-84
lines changed

2 files changed

+131
-84
lines changed

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/support/CachingUsernamePasswordRealm.java‎

Lines changed: 61 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
package org.elasticsearch.xpack.security.authc.support;
77

88
import org.elasticsearch.action.ActionListener;
9+
import org.elasticsearch.common.Nullable;
910
import org.elasticsearch.common.cache.Cache;
1011
import org.elasticsearch.common.cache.CacheBuilder;
1112
import org.elasticsearch.common.settings.SecureString;
@@ -29,7 +30,7 @@
2930

3031
public abstract class CachingUsernamePasswordRealm extends UsernamePasswordRealm implements CachingRealm {
3132

32-
private final Cache<String, ListenableFuture<UserWithHash>> cache;
33+
private final Cache<String, ListenableFuture<CachedResult>> cache;
3334
private final ThreadPool threadPool;
3435
private final boolean authenticationEnabled;
3536
final Hasher cacheHasher;
@@ -40,7 +41,7 @@ protected CachingUsernamePasswordRealm(RealmConfig config, ThreadPool threadPool
4041
this.threadPool = threadPool;
4142
final TimeValue ttl = this.config.getSetting(CachingUsernamePasswordRealmSettings.CACHE_TTL_SETTING);
4243
if (ttl.getNanos() > 0) {
43-
cache = CacheBuilder.<String, ListenableFuture<UserWithHash>>builder()
44+
cache = CacheBuilder.<String, ListenableFuture<CachedResult>>builder()
4445
.setExpireAfterWrite(ttl)
4546
.setMaximumWeight(this.config.getSetting(CachingUsernamePasswordRealmSettings.CACHE_MAX_USERS_SETTING))
4647
.build();
@@ -122,58 +123,61 @@ private void authenticateWithCache(UsernamePasswordToken token, ActionListener<A
122123
assert cache != null;
123124
try {
124125
final AtomicBoolean authenticationInCache = new AtomicBoolean(true);
125-
final ListenableFuture<UserWithHash> listenableCacheEntry = cache.computeIfAbsent(token.principal(), k -> {
126+
final ListenableFuture<CachedResult> listenableCacheEntry = cache.computeIfAbsent(token.principal(), k -> {
126127
authenticationInCache.set(false);
127128
return new ListenableFuture<>();
128129
});
129130
if (authenticationInCache.get()) {
130131
// there is a cached or an inflight authenticate request
131-
listenableCacheEntry.addListener(ActionListener.wrap(authenticatedUserWithHash -> {
132-
if (authenticatedUserWithHash != null && authenticatedUserWithHash.verify(token.credentials())) {
133-
// cached credential hash matches the credential hash for this forestalled request
134-
handleCachedAuthentication(authenticatedUserWithHash.user, ActionListener.wrap(cacheResult -> {
135-
if (cacheResult.isAuthenticated()) {
136-
logger.debug("realm [{}] authenticated user [{}], with roles [{}]",
137-
name(), token.principal(), cacheResult.getUser().roles());
138-
} else {
139-
logger.debug("realm [{}] authenticated user [{}] from cache, but then failed [{}]",
140-
name(), token.principal(), cacheResult.getMessage());
141-
}
142-
listener.onResponse(cacheResult);
143-
}, listener::onFailure));
132+
listenableCacheEntry.addListener(ActionListener.wrap(cachedResult -> {
133+
final boolean credsMatch = cachedResult.verify(token.credentials());
134+
if (cachedResult.authenticationResult.isAuthenticated()) {
135+
if (credsMatch) {
136+
// cached credential hash matches the credential hash for this forestalled request
137+
handleCachedAuthentication(cachedResult.user, ActionListener.wrap(cacheResult -> {
138+
if (cacheResult.isAuthenticated()) {
139+
logger.debug("realm [{}] authenticated user [{}], with roles [{}]",
140+
name(), token.principal(), cacheResult.getUser().roles());
141+
} else {
142+
logger.debug("realm [{}] authenticated user [{}] from cache, but then failed [{}]",
143+
name(), token.principal(), cacheResult.getMessage());
144+
}
145+
listener.onResponse(cacheResult);
146+
}, listener::onFailure));
147+
} else {
148+
// its credential hash does not match the
149+
// hash of the credential for this forestalled request.
150+
// clear cache and try to reach the authentication source again because password
151+
// might have changed there and the local cached hash got stale
152+
cache.invalidate(token.principal(), listenableCacheEntry);
153+
authenticateWithCache(token, listener);
154+
}
155+
} else if (credsMatch) {
156+
// not authenticated but instead of hammering reuse the result. a new
157+
// request will trigger a retried auth
158+
listener.onResponse(cachedResult.authenticationResult);
144159
} else {
145-
// The inflight request has failed or its credential hash does not match the
146-
// hash of the credential for this forestalled request.
147-
// clear cache and try to reach the authentication source again because password
148-
// might have changed there and the local cached hash got stale
149160
cache.invalidate(token.principal(), listenableCacheEntry);
150161
authenticateWithCache(token, listener);
151162
}
152-
}, e -> {
153-
// the inflight request failed, so try again, but first (always) make sure cache
154-
// is cleared of the failed authentication
155-
cache.invalidate(token.principal(), listenableCacheEntry);
156-
authenticateWithCache(token, listener);
157-
}), threadPool.executor(ThreadPool.Names.GENERIC), threadPool.getThreadContext());
163+
}, listener::onFailure), threadPool.executor(ThreadPool.Names.GENERIC), threadPool.getThreadContext());
158164
} else {
159165
// attempt authentication against the authentication source
160166
doAuthenticate(token, ActionListener.wrap(authResult -> {
161-
if (authResult.isAuthenticated() && authResult.getUser().enabled()) {
162-
// compute the credential hash of this successful authentication request
163-
final UserWithHash userWithHash = new UserWithHash(authResult.getUser(), token.credentials(), cacheHasher);
164-
// notify any forestalled request listeners; they will not reach to the
165-
// authentication request and instead will use this hash for comparison
166-
listenableCacheEntry.onResponse(userWithHash);
167-
} else {
168-
// notify any forestalled request listeners; they will retry the request
169-
listenableCacheEntry.onResponse(null);
167+
if (authResult.isAuthenticated() == false || authResult.getUser().enabled() == false) {
168+
// a new request should trigger a new authentication
169+
cache.invalidate(token.principal(), listenableCacheEntry);
170170
}
171-
// notify the listener of the inflight authentication request; this request is not retried
171+
// notify any forestalled request listeners; they will not reach to the
172+
// authentication request and instead will use this result if they contain
173+
// the same credentials
174+
listenableCacheEntry.onResponse(new CachedResult(authResult, cacheHasher, authResult.getUser(), token.credentials()));
172175
listener.onResponse(authResult);
173176
}, e -> {
174-
// notify any staved off listeners; they will retry the request
177+
cache.invalidate(token.principal(), listenableCacheEntry);
178+
// notify any staved off listeners; they will propagate this error
175179
listenableCacheEntry.onFailure(e);
176-
// notify the listener of the inflight authentication request; this request is not retried
180+
// notify the listener of the inflight authentication request
177181
listener.onFailure(e);
178182
}));
179183
}
@@ -225,35 +229,31 @@ private void lookupWithCache(String username, ActionListener<User> listener) {
225229
assert cache != null;
226230
try {
227231
final AtomicBoolean lookupInCache = new AtomicBoolean(true);
228-
final ListenableFuture<UserWithHash> listenableCacheEntry = cache.computeIfAbsent(username, key -> {
232+
final ListenableFuture<CachedResult> listenableCacheEntry = cache.computeIfAbsent(username, key -> {
229233
lookupInCache.set(false);
230234
return new ListenableFuture<>();
231235
});
232236
if (false == lookupInCache.get()) {
233237
// attempt lookup against the user directory
234238
doLookupUser(username, ActionListener.wrap(user -> {
235-
if (user != null) {
236-
// user found
237-
final UserWithHash userWithHash = new UserWithHash(user, null, null);
238-
// notify forestalled request listeners
239-
listenableCacheEntry.onResponse(userWithHash);
240-
} else {
239+
final CachedResult result = new CachedResult(AuthenticationResult.notHandled(), cacheHasher, user, null);
240+
if (user == null) {
241241
// user not found, invalidate cache so that subsequent requests are forwarded to
242242
// the user directory
243243
cache.invalidate(username, listenableCacheEntry);
244-
// notify forestalled request listeners
245-
listenableCacheEntry.onResponse(null);
246244
}
245+
// notify forestalled request listeners
246+
listenableCacheEntry.onResponse(result);
247247
}, e -> {
248248
// the next request should be forwarded, not halted by a failed lookup attempt
249249
cache.invalidate(username, listenableCacheEntry);
250250
// notify forestalled listeners
251251
listenableCacheEntry.onFailure(e);
252252
}));
253253
}
254-
listenableCacheEntry.addListener(ActionListener.wrap(userWithHash -> {
255-
if (userWithHash != null) {
256-
listener.onResponse(userWithHash.user);
254+
listenableCacheEntry.addListener(ActionListener.wrap(cachedResult -> {
255+
if (cachedResult.user != null) {
256+
listener.onResponse(cachedResult.user);
257257
} else {
258258
listener.onResponse(null);
259259
}
@@ -265,16 +265,21 @@ private void lookupWithCache(String username, ActionListener<User> listener) {
265265

266266
protected abstract void doLookupUser(String username, ActionListener<User> listener);
267267

268-
private static class UserWithHash {
269-
final User user;
270-
final char[] hash;
268+
private static class CachedResult {
269+
private final AuthenticationResult authenticationResult;
270+
private final User user;
271+
private final char[] hash;
271272

272-
UserWithHash(User user, SecureString password, Hasher hasher) {
273-
this.user = Objects.requireNonNull(user);
273+
private CachedResult(AuthenticationResult result, Hasher hasher, @Nullable User user, @Nullable SecureString password) {
274+
this.authenticationResult = Objects.requireNonNull(result);
275+
if (authenticationResult.isAuthenticated() && user == null) {
276+
throw new IllegalArgumentException("authentication cannot be successful with a null user");
277+
}
278+
this.user = user;
274279
this.hash = password == null ? null : hasher.hash(password);
275280
}
276281

277-
boolean verify(SecureString password) {
282+
private boolean verify(SecureString password) {
278283
return hash != null && Hasher.verifyHash(password, hash);
279284
}
280285
}

‎x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/support/CachingUsernamePasswordRealmTests.java‎

Lines changed: 70 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -58,13 +58,13 @@ public void setup() {
5858
}
5959

6060
@After
61-
public void stop() throws InterruptedException {
61+
public void stop() {
6262
if (threadPool != null) {
6363
terminate(threadPool);
6464
}
6565
}
6666

67-
public void testCacheSettings() throws Exception {
67+
public void testCacheSettings() {
6868
String cachingHashAlgo = Hasher.values()[randomIntBetween(0, Hasher.values().length - 1)].name().toLowerCase(Locale.ROOT);
6969
int maxUsers = randomIntBetween(10, 100);
7070
TimeValue ttl = TimeValue.timeValueMinutes(randomIntBetween(10, 20));
@@ -352,7 +352,7 @@ private void sleepUntil(long until) throws InterruptedException {
352352
}
353353
}
354354

355-
public void testAuthenticateContract() throws Exception {
355+
public void testAuthenticateContract() {
356356
Realm realm = new FailingAuthenticationRealm(globalSettings, threadPool);
357357
PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
358358
realm.authenticate(new UsernamePasswordToken("user", new SecureString("pass")), future);
@@ -366,7 +366,7 @@ public void testAuthenticateContract() throws Exception {
366366
assertThat(e.getMessage(), containsString("whatever exception"));
367367
}
368368

369-
public void testLookupContract() throws Exception {
369+
public void testLookupContract() {
370370
Realm realm = new FailingAuthenticationRealm(globalSettings, threadPool);
371371
PlainActionFuture<User> future = new PlainActionFuture<>();
372372
realm.lookupUser("user", future);
@@ -380,7 +380,7 @@ public void testLookupContract() throws Exception {
380380
assertThat(e.getMessage(), containsString("lookup exception"));
381381
}
382382

383-
public void testReturnDifferentObjectFromCache() throws Exception {
383+
public void testReturnDifferentObjectFromCache() {
384384
final AtomicReference<User> userArg = new AtomicReference<>();
385385
final AtomicReference<AuthenticationResult> result = new AtomicReference<>();
386386
Realm realm = new AlwaysAuthenticateCachingRealm(globalSettings, threadPool) {
@@ -473,6 +473,71 @@ protected void doLookupUser(String username, ActionListener<User> listener) {
473473
assertEquals(1, authCounter.get());
474474
}
475475

476+
public void testUnauthenticatedResultPropagatesWithSameCreds() throws Exception {
477+
final String username = "username";
478+
final SecureString password = SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING;
479+
final AtomicInteger authCounter = new AtomicInteger(0);
480+
final Hasher pwdHasher = Hasher.resolve(randomFrom("pbkdf2", "pbkdf2_1000", "bcrypt", "bcrypt9"));
481+
final String passwordHash = new String(pwdHasher.hash(password));
482+
RealmConfig config = new RealmConfig(new RealmConfig.RealmIdentifier("caching", "test_realm"), globalSettings,
483+
TestEnvironment.newEnvironment(globalSettings), new ThreadContext(Settings.EMPTY));
484+
485+
final int numberOfProcessors = Runtime.getRuntime().availableProcessors();
486+
final int numberOfThreads = scaledRandomIntBetween((numberOfProcessors + 1) / 2, numberOfProcessors * 3);
487+
final CountDownLatch latch = new CountDownLatch(1 + numberOfThreads);
488+
List<Thread> threads = new ArrayList<>(numberOfThreads);
489+
final SecureString credsToUse = new SecureString(randomAlphaOfLength(12).toCharArray());
490+
final CachingUsernamePasswordRealm realm = new CachingUsernamePasswordRealm(config, threadPool) {
491+
@Override
492+
protected void doAuthenticate(UsernamePasswordToken token, ActionListener<AuthenticationResult> listener) {
493+
authCounter.incrementAndGet();
494+
// do something slow
495+
if (pwdHasher.verify(token.credentials(), passwordHash.toCharArray())) {
496+
listener.onFailure(new IllegalStateException("password auth should never succeed"));
497+
} else {
498+
listener.onResponse(AuthenticationResult.unsuccessful("password verification failed", null));
499+
}
500+
}
501+
502+
@Override
503+
protected void doLookupUser(String username, ActionListener<User> listener) {
504+
listener.onFailure(new UnsupportedOperationException("this method should not be called"));
505+
}
506+
};
507+
for (int i = 0; i < numberOfThreads; i++) {
508+
threads.add(new Thread(() -> {
509+
try {
510+
latch.countDown();
511+
latch.await();
512+
final UsernamePasswordToken token = new UsernamePasswordToken(username, credsToUse);
513+
514+
realm.authenticate(token, ActionListener.wrap((result) -> {
515+
if (result.isAuthenticated()) {
516+
throw new IllegalStateException("invalid password led to an authenticated result: " + result);
517+
}
518+
assertThat(result.getMessage(), containsString("password verification failed"));
519+
}, (e) -> {
520+
logger.error("caught exception", e);
521+
fail("unexpected exception - " + e);
522+
}));
523+
524+
} catch (InterruptedException e) {
525+
logger.error("thread was interrupted", e);
526+
Thread.currentThread().interrupt();
527+
}
528+
}));
529+
}
530+
531+
for (Thread thread : threads) {
532+
thread.start();
533+
}
534+
latch.countDown();
535+
for (Thread thread : threads) {
536+
thread.join();
537+
}
538+
assertEquals(1, authCounter.get());
539+
}
540+
476541
public void testCacheConcurrency() throws Exception {
477542
final String username = "username";
478543
final SecureString password = SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING;
@@ -704,27 +769,4 @@ protected void doLookupUser(String username, ActionListener<User> listener) {
704769
listener.onResponse(new User(username, new String[]{"lookupRole1", "lookupRole2"}));
705770
}
706771
}
707-
708-
static class LookupNotSupportedRealm extends CachingUsernamePasswordRealm {
709-
710-
public final AtomicInteger authInvocationCounter = new AtomicInteger(0);
711-
public final AtomicInteger lookupInvocationCounter = new AtomicInteger(0);
712-
713-
LookupNotSupportedRealm(Settings globalSettings, ThreadPool threadPool) {
714-
super(new RealmConfig(new RealmConfig.RealmIdentifier("caching", "lookup-notsupported-test"), globalSettings,
715-
TestEnvironment.newEnvironment(globalSettings), threadPool.getThreadContext()), threadPool);
716-
}
717-
718-
@Override
719-
protected void doAuthenticate(UsernamePasswordToken token, ActionListener<AuthenticationResult> listener) {
720-
authInvocationCounter.incrementAndGet();
721-
listener.onResponse(AuthenticationResult.success(new User(token.principal(), new String[]{"testRole1", "testRole2"})));
722-
}
723-
724-
@Override
725-
protected void doLookupUser(String username, ActionListener<User> listener) {
726-
lookupInvocationCounter.incrementAndGet();
727-
listener.onFailure(new UnsupportedOperationException("don't call lookup if lookup isn't supported!!!"));
728-
}
729-
}
730772
}

0 commit comments

Comments
 (0)