Limit user to single concurrent auth per realm

[jaymode]

This commit reworks the way our realms perform caching in order to
limit each principal to a single ongoing authentication per realm. In
other words, this means that multiple requests made by the same user
will not trigger more that one authentication attempt at a time if no
entry has been stored in the cache. If an entry is present in our
cache, there is no restriction on the number of concurrent
authentications performed for this user.

This change enables us to limit the load we place on an external system
like an LDAP server and also preserve resources such as CPU on
expensive operations such as BCrypt authentication.

Closes #30355

[elasticmachine]

Pinging @elastic/es-security

[tvernum]

I like the approach, but I have a suggested change.

[tvernum]

It feels weird for this to take a tuple rather than 2 arguments (since sometimes we're constructing the Tuple just to pass as an argument and then be decomposed again)

[tvernum]

I think this needs javadoc - if only to point to the explanation in the class javadoc.

[tvernum]

The use of createdAndStartedFuture here worries me.
We need to handle the two cases (directly authenticated vs loaded from cache) differently, but this variable is a couple of steps removed from that and I fear that something like a race condition somewhere in the cache could mean that this is true, but the user was not actually authenticated with the credentials from token.

Perhaps we could set an AtomicReference to User in the listener for doAuthenticate and check whether the User being pulled from the cache matches. Then were no longer testing "did we create the future" but "did we actually authenticate this User object".

e.g.

 doAuthenticate(token, ActionListener.wrap(result -> {
                    if (result.isAuthenticated()) {
                        final User user = result.getUser();
                        authenticatedUser.set(user);

       final boolean wasAuthenticatedWithThisToken = tuple.v2() == authenticatedUser.get();
       handleResult(future, wasAuthenticatedWithThisToken, token, tuple, listener);

[jaymode]

@tvernum thanks for the suggestion. I implemented the suggestion slightly differently. Let me know what you think.

[tvernum]

LGTM

[albertzaharovits]

I have pushed e888467 and 935c6c7 to fix a minor test bug:

diff --git a/server/src/test/java/org/elasticsearch/common/cache/CacheTests.java b/server/src/test/java/org/elasticsearch/common/cache/CacheTests.java
index 3b183cce40b..a2e47e6c60f 100644
--- a/server/src/test/java/org/elasticsearch/common/cache/CacheTests.java
+++ b/server/src/test/java/org/elasticsearch/common/cache/CacheTests.java
@@ -471,7 +471,7 @@ public class CacheTests extends ESTestCase {
                     keys.add(key);
                 } else {
                     // invalidate with incorrect value
-                    cache.invalidate(key, Integer.toString(key * randomIntBetween(2, 10)));
+                    cache.invalidate(key, Integer.toString(key + randomIntBetween(2, 10)));
                 }
             }
         }
@@ -506,7 +506,7 @@ public class CacheTests extends ESTestCase {
                     invalidated.add(i);
                 } else {
                     // invalidate with incorrect value
-                    cache.invalidate(i, Integer.toString(i * randomIntBetween(2, 10)));
+                    cache.invalidate(i, Integer.toString(i + randomIntBetween(2, 10)));
                 }
             }
         }

if i is 0 the value is correct and invalidation is expected.
Sample error:

./gradlew :server:test -Dtests.seed=824E15A422DDC053 -Dtests.class=org.elasticsearch.common.cache.CacheTests -Dtests.method="testNotificationOnInvalidateWithValue { seed=[824E15A422DDC053:5A13A8665810B577]}" -Dtests.security.manager=true -Dtests.locale=es-CU -Dtests.timezone=Arctic/Longyearbyen
FAILURE 0.02s | CacheTests.testNotificationOnInvalidateWithValue { seed=[824E15A422DDC053:5A13A8665810B577]} <<< FAILURES!
   > Throwable #1: java.lang.AssertionError: expected:<[0, 5155, 4680, 2665, 3114, 4237, 5134, 1138, 2674, 179, 116, 4215, 5656, 154, 1338, 2236, 2941, 5885, 2974, 4414]> but was:<[5155, 4680, 2665, 3114, 4237, 5134, 1138, 2674, 179, 116, 4215, 5656, 154, 1338, 2236, 2941, 5885, 2974, 4414]>
   >    at __randomizedtesting.SeedInfo.seed([824E15A422DDC053:5A13A8665810B577]:0)
   >    at org.elasticsearch.common.cache.CacheTests.testNotificationOnInvalidateWithValue(CacheTests.java:513)
   >    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
   >    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
   >    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
   >    at java.base/java.lang.reflect.Method.invoke(Method.java:564)
   >    at java.base/java.lang.Thread.run(Thread.java:844)

[jaymode]

thanks @albertzaharovits

[albertzaharovits]

@jaymode I think tuple.v2().user == authenticatedUser.get() is redundant (or there's a deeper reasoning that eludes me). Can you please clarify why this check is necessary?

[tvernum]

@albertzaharovits

This was introduced in response to my feedback above

• Limit user to single concurrent auth per realm #30794 (comment)

[tvernum]

That link doesn't seem to work correctly.
It's the 3rd comment in my review above.

[albertzaharovits]

Thanks @tvernum . I understand, it's always better to test for the actual thing, if you can, instead of a surrogate.
But as it makes you feel surer on the whole it makes me feel less of. The whole seems too gnarly: handleResult has both createdAndStartedFuture and performedAuthentication as params.
Then, I think we can do without createdAndStartedFuture.
I will open a nimble refactor non-issue on this.

[albertzaharovits]

@tvernum I am going to concede on this one, for now 😭 .
I could not find a satisfactory refactoring such that the current behavior is maintained over all branches. Specifically, I was looking to minimize authenticationResult.isAuthenticated() checks, as well as to split the listener into two types, one for the thread going for the network call and the other for the stalled threads. In the end I got a huge branching monster no better than the present one.
😢