Add an authentication cache for API keys (#38469)

This commit adds an authentication cache for API keys that caches the
hash of an API key with a faster hash. This will enable better
performance when API keys are used for bulk or heavy searching.

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/Security.java

@@ -438,7 +438,8 @@ public class Security extends Plugin implements ActionPlugin, IngestPlugin, Netw
             rolesProviders.addAll(extension.getRolesProviders(settings, resourceWatcherService));
         }
 
-        final ApiKeyService apiKeyService = new ApiKeyService(settings, Clock.systemUTC(), client, securityIndex.get(), clusterService);
+        final ApiKeyService apiKeyService = new ApiKeyService(settings, Clock.systemUTC(), client, securityIndex.get(), clusterService,
+            threadPool);
         components.add(apiKeyService);
         final CompositeRolesStore allRolesStore = new CompositeRolesStore(settings, fileRolesStore, nativeRolesStore, reservedRolesStore,
             privilegeStore, rolesProviders, threadPool.getThreadContext(), getLicenseState(), fieldPermissionsCache, apiKeyService);
@@ -631,6 +632,9 @@ public class Security extends Plugin implements ActionPlugin, IngestPlugin, Netw
         settingsList.add(ApiKeyService.PASSWORD_HASHING_ALGORITHM);
         settingsList.add(ApiKeyService.DELETE_TIMEOUT);
         settingsList.add(ApiKeyService.DELETE_INTERVAL);
+        settingsList.add(ApiKeyService.CACHE_HASH_ALGO_SETTING);
+        settingsList.add(ApiKeyService.CACHE_MAX_KEYS_SETTING);
+        settingsList.add(ApiKeyService.CACHE_TTL_SETTING);
 
         // hide settings
         settingsList.add(Setting.listSetting(SecurityField.setting("hide_settings"), Collections.emptyList(), Function.identity(),

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/ApiKeyService.java

@@ -33,12 +33,16 @@ import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.cache.Cache;
+import org.elasticsearch.common.cache.CacheBuilder;
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.concurrent.FutureUtils;
+import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.DeprecationHandler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
@@ -49,6 +53,7 @@ import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.XPackSettings;
 import org.elasticsearch.xpack.core.security.ScrollHelper;
 import org.elasticsearch.xpack.core.security.action.ApiKey;
@@ -81,6 +86,9 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -96,7 +104,6 @@ public class ApiKeyService {
     static final String API_KEY_ID_KEY = "_security_api_key_id";
     static final String API_KEY_ROLE_DESCRIPTORS_KEY = "_security_api_key_role_descriptors";
     static final String API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY = "_security_api_key_limited_by_role_descriptors";
-    static final String API_KEY_ROLE_KEY = "_security_api_key_role";
 
     public static final Setting<String> PASSWORD_HASHING_ALGORITHM = new Setting<>(
         "xpack.security.authc.api_key_hashing.algorithm", "pbkdf2", Function.identity(), v -> {
@@ -117,6 +124,12 @@ public class ApiKeyService {
             TimeValue.MINUS_ONE, Property.NodeScope);
     public static final Setting<TimeValue> DELETE_INTERVAL = Setting.timeSetting("xpack.security.authc.api_key.delete.interval",
             TimeValue.timeValueHours(24L), Property.NodeScope);
+    public static final Setting<String> CACHE_HASH_ALGO_SETTING = Setting.simpleString("xpack.security.authc.api_key.cache.hash_algo",
+        "ssha256", Setting.Property.NodeScope);
+    public static final Setting<TimeValue> CACHE_TTL_SETTING = Setting.timeSetting("xpack.security.authc.api_key.cache.ttl",
+        TimeValue.timeValueHours(24L), Property.NodeScope);
+    public static final Setting<Integer> CACHE_MAX_KEYS_SETTING = Setting.intSetting("xpack.security.authc.api_key.cache.max_keys",
+        10000, Property.NodeScope);
 
     private final Clock clock;
     private final Client client;
@@ -127,11 +140,14 @@ public class ApiKeyService {
     private final Settings settings;
     private final ExpiredApiKeysRemover expiredApiKeysRemover;
     private final TimeValue deleteInterval;
+    private final Cache<String, ListenableFuture<CachedApiKeyHashResult>> apiKeyAuthCache;
+    private final Hasher cacheHasher;
+    private final ThreadPool threadPool;
 
     private volatile long lastExpirationRunMs;
 
-    public ApiKeyService(Settings settings, Clock clock, Client client, SecurityIndexManager securityIndex,
+    public ApiKeyService(Settings settings, Clock clock, Client client, SecurityIndexManager securityIndex, ClusterService clusterService,
-                         ClusterService clusterService) {
+                         ThreadPool threadPool) {
         this.clock = clock;
         this.client = client;
         this.securityIndex = securityIndex;
@@ -141,6 +157,17 @@ public class ApiKeyService {
         this.settings = settings;
         this.deleteInterval = DELETE_INTERVAL.get(settings);
         this.expiredApiKeysRemover = new ExpiredApiKeysRemover(settings, client);
+        this.threadPool = threadPool;
+        this.cacheHasher = Hasher.resolve(CACHE_HASH_ALGO_SETTING.get(settings));
+        final TimeValue ttl = CACHE_TTL_SETTING.get(settings);
+        if (ttl.getNanos() > 0) {
+            this.apiKeyAuthCache = CacheBuilder.<String, ListenableFuture<CachedApiKeyHashResult>>builder()
+                .setExpireAfterWrite(ttl)
+                .setMaximumWeight(CACHE_MAX_KEYS_SETTING.get(settings))
+                .build();
+        } else {
+            this.apiKeyAuthCache = null;
+        }
     }
 
     /**
@@ -363,8 +390,8 @@ public class ApiKeyService {
      * @param credentials the credentials provided by the user
      * @param listener the listener to notify after verification
      */
-    static void validateApiKeyCredentials(Map<String, Object> source, ApiKeyCredentials credentials, Clock clock,
+    void validateApiKeyCredentials(Map<String, Object> source, ApiKeyCredentials credentials, Clock clock,
-                                          ActionListener<AuthenticationResult> listener) {
+                                   ActionListener<AuthenticationResult> listener) {
         final Boolean invalidated = (Boolean) source.get("api_key_invalidated");
         if (invalidated == null) {
             listener.onResponse(AuthenticationResult.terminate("api key document is missing invalidated field", null));
@@ -375,33 +402,87 @@ public class ApiKeyService {
             if (apiKeyHash == null) {
                 throw new IllegalStateException("api key hash is missing");
             }
-            final boolean verified = verifyKeyAgainstHash(apiKeyHash, credentials);
 
-            if (verified) {
+            if (apiKeyAuthCache != null) {
-                final Long expirationEpochMilli = (Long) source.get("expiration_time");
+                final AtomicBoolean valueAlreadyInCache = new AtomicBoolean(true);
-                if (expirationEpochMilli == null || Instant.ofEpochMilli(expirationEpochMilli).isAfter(clock.instant())) {
+                final ListenableFuture<CachedApiKeyHashResult> listenableCacheEntry;
-                    final Map<String, Object> creator = Objects.requireNonNull((Map<String, Object>) source.get("creator"));
+                try {
-                    final String principal = Objects.requireNonNull((String) creator.get("principal"));
+                    listenableCacheEntry = apiKeyAuthCache.computeIfAbsent(credentials.getId(),
-                    final Map<String, Object> metadata = (Map<String, Object>) creator.get("metadata");
+                        k -> {
-                    final Map<String, Object> roleDescriptors = (Map<String, Object>) source.get("role_descriptors");
+                            valueAlreadyInCache.set(false);
-                    final Map<String, Object> limitedByRoleDescriptors = (Map<String, Object>) source.get("limited_by_role_descriptors");
+                            return new ListenableFuture<>();
-                    final String[] roleNames = (roleDescriptors != null) ? roleDescriptors.keySet().toArray(Strings.EMPTY_ARRAY)
+                        });
-                            : limitedByRoleDescriptors.keySet().toArray(Strings.EMPTY_ARRAY);
+                } catch (ExecutionException e) {
-                    final User apiKeyUser = new User(principal, roleNames, null, null, metadata, true);
+                    listener.onFailure(e);
-                    final Map<String, Object> authResultMetadata = new HashMap<>();
+                    return;
-                    authResultMetadata.put(API_KEY_ROLE_DESCRIPTORS_KEY, roleDescriptors);
+                }
-                    authResultMetadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, limitedByRoleDescriptors);
+
-                    authResultMetadata.put(API_KEY_ID_KEY, credentials.getId());
+                if (valueAlreadyInCache.get()) {
-                    listener.onResponse(AuthenticationResult.success(apiKeyUser, authResultMetadata));
+                    listenableCacheEntry.addListener(ActionListener.wrap(result -> {
+                            if (result.success) {
+                                if (result.verify(credentials.getKey())) {
+                                    // move on
+                                    validateApiKeyExpiration(source, credentials, clock, listener);
+                                } else {
+                                    listener.onResponse(AuthenticationResult.unsuccessful("invalid credentials", null));
+                                }
+                            } else if (result.verify(credentials.getKey())) { // same key, pass the same result
+                                listener.onResponse(AuthenticationResult.unsuccessful("invalid credentials", null));
+                            } else {
+                                apiKeyAuthCache.invalidate(credentials.getId(), listenableCacheEntry);
+                                validateApiKeyCredentials(source, credentials, clock, listener);
+                            }
+                        }, listener::onFailure),
+                        threadPool.generic(), threadPool.getThreadContext());
                 } else {
-                    listener.onResponse(AuthenticationResult.terminate("api key is expired", null));
+                    final boolean verified = verifyKeyAgainstHash(apiKeyHash, credentials);
+                    listenableCacheEntry.onResponse(new CachedApiKeyHashResult(verified, credentials.getKey()));
+                    if (verified) {
+                        // move on
+                        validateApiKeyExpiration(source, credentials, clock, listener);
+                    } else {
+                        listener.onResponse(AuthenticationResult.unsuccessful("invalid credentials", null));
+                    }
                 }
             } else {
-                listener.onResponse(AuthenticationResult.unsuccessful("invalid credentials", null));
+                final boolean verified = verifyKeyAgainstHash(apiKeyHash, credentials);
+                if (verified) {
+                    // move on
+                    validateApiKeyExpiration(source, credentials, clock, listener);
+                } else {
+                    listener.onResponse(AuthenticationResult.unsuccessful("invalid credentials", null));
+                }
             }
         }
     }
 
+    // pkg private for testing
+    CachedApiKeyHashResult getFromCache(String id) {
+        return apiKeyAuthCache == null ? null : FutureUtils.get(apiKeyAuthCache.get(id), 0L, TimeUnit.MILLISECONDS);
+    }
+
+    private void validateApiKeyExpiration(Map<String, Object> source, ApiKeyCredentials credentials, Clock clock,
+                                  ActionListener<AuthenticationResult> listener) {
+        final Long expirationEpochMilli = (Long) source.get("expiration_time");
+        if (expirationEpochMilli == null || Instant.ofEpochMilli(expirationEpochMilli).isAfter(clock.instant())) {
+            final Map<String, Object> creator = Objects.requireNonNull((Map<String, Object>) source.get("creator"));
+            final String principal = Objects.requireNonNull((String) creator.get("principal"));
+            final Map<String, Object> metadata = (Map<String, Object>) creator.get("metadata");
+            final Map<String, Object> roleDescriptors = (Map<String, Object>) source.get("role_descriptors");
+            final Map<String, Object> limitedByRoleDescriptors = (Map<String, Object>) source.get("limited_by_role_descriptors");
+            final String[] roleNames = (roleDescriptors != null) ? roleDescriptors.keySet().toArray(Strings.EMPTY_ARRAY)
+                : limitedByRoleDescriptors.keySet().toArray(Strings.EMPTY_ARRAY);
+            final User apiKeyUser = new User(principal, roleNames, null, null, metadata, true);
+            final Map<String, Object> authResultMetadata = new HashMap<>();
+            authResultMetadata.put(API_KEY_ROLE_DESCRIPTORS_KEY, roleDescriptors);
+            authResultMetadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, limitedByRoleDescriptors);
+            authResultMetadata.put(API_KEY_ID_KEY, credentials.getId());
+            listener.onResponse(AuthenticationResult.success(apiKeyUser, authResultMetadata));
+        } else {
+            listener.onResponse(AuthenticationResult.terminate("api key is expired", null));
+        }
+    }
+
     /**
      * Gets the API Key from the <code>Authorization</code> header if the header begins with
      * <code>ApiKey </code>
@@ -851,4 +932,17 @@ public class ApiKeyService {
         }
     }
 
+    final class CachedApiKeyHashResult {
+        final boolean success;
+        final char[] hash;
+
+        CachedApiKeyHashResult(boolean success, SecureString apiKey) {
+            this.success = success;
+            this.hash = cacheHasher.hash(apiKey);
+        }
+
+        private boolean verify(SecureString password) {
+            return hash != null && cacheHasher.verify(password, hash);
+        }
+    }
 }

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/ApiKeyServiceTests.java

@@ -31,7 +31,9 @@ import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.privilege.ApplicationPrivilege;
 import org.elasticsearch.xpack.core.security.authz.store.ReservedRolesStore;
 import org.elasticsearch.xpack.core.security.user.User;
+import org.elasticsearch.xpack.security.authc.ApiKeyService.ApiKeyCredentials;
 import org.elasticsearch.xpack.security.authc.ApiKeyService.ApiKeyRoleDescriptors;
+import org.elasticsearch.xpack.security.authc.ApiKeyService.CachedApiKeyHashResult;
 import org.elasticsearch.xpack.security.authz.store.NativePrivilegeStore;
 import org.junit.After;
 import org.junit.Before;
@@ -50,6 +52,8 @@ import static org.hamcrest.Matchers.arrayContaining;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.sameInstance;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
@@ -69,7 +73,7 @@ public class ApiKeyServiceTests extends ESTestCase {
     }
 
     public void testGetCredentialsFromThreadContext() {
-        ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
+        ThreadContext threadContext = threadPool.getThreadContext();
         assertNull(ApiKeyService.getCredentialsFromHeader(threadContext));
 
         final String apiKeyAuthScheme = randomFrom("apikey", "apiKey", "ApiKey", "APikey", "APIKEY");
@@ -118,10 +122,12 @@ public class ApiKeyServiceTests extends ESTestCase {
         sourceMap.put("creator", creatorMap);
         sourceMap.put("api_key_invalidated", false);
 
+        ApiKeyService service = new ApiKeyService(Settings.EMPTY, Clock.systemUTC(), null, null,
+            ClusterServiceUtils.createClusterService(threadPool), threadPool);
         ApiKeyService.ApiKeyCredentials creds =
             new ApiKeyService.ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(apiKey.toCharArray()));
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
-        ApiKeyService.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
         AuthenticationResult result = future.get();
         assertNotNull(result);
         assertTrue(result.isAuthenticated());
@@ -134,7 +140,7 @@ public class ApiKeyServiceTests extends ESTestCase {
 
         sourceMap.put("expiration_time", Clock.systemUTC().instant().plus(1L, ChronoUnit.HOURS).toEpochMilli());
         future = new PlainActionFuture<>();
-        ApiKeyService.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
         result = future.get();
         assertNotNull(result);
         assertTrue(result.isAuthenticated());
@@ -147,7 +153,7 @@ public class ApiKeyServiceTests extends ESTestCase {
 
         sourceMap.put("expiration_time", Clock.systemUTC().instant().minus(1L, ChronoUnit.HOURS).toEpochMilli());
         future = new PlainActionFuture<>();
-        ApiKeyService.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
         result = future.get();
         assertNotNull(result);
         assertFalse(result.isAuthenticated());
@@ -155,7 +161,7 @@ public class ApiKeyServiceTests extends ESTestCase {
         sourceMap.remove("expiration_time");
         creds = new ApiKeyService.ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(randomAlphaOfLength(15).toCharArray()));
         future = new PlainActionFuture<>();
-        ApiKeyService.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
         result = future.get();
         assertNotNull(result);
         assertFalse(result.isAuthenticated());
@@ -163,7 +169,7 @@ public class ApiKeyServiceTests extends ESTestCase {
         sourceMap.put("api_key_invalidated", true);
         creds = new ApiKeyService.ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(randomAlphaOfLength(15).toCharArray()));
         future = new PlainActionFuture<>();
-        ApiKeyService.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
         result = future.get();
         assertNotNull(result);
         assertFalse(result.isAuthenticated());
@@ -188,7 +194,7 @@ public class ApiKeyServiceTests extends ESTestCase {
         final Authentication authentication = new Authentication(new User("joe"), new RealmRef("apikey", "apikey", "node"), null,
             Version.CURRENT, AuthenticationType.API_KEY, authMetadata);
         ApiKeyService service = new ApiKeyService(Settings.EMPTY, Clock.systemUTC(), null, null,
-                ClusterServiceUtils.createClusterService(threadPool));
+                ClusterServiceUtils.createClusterService(threadPool), threadPool);
 
         PlainActionFuture<ApiKeyRoleDescriptors> roleFuture = new PlainActionFuture<>();
         service.getRoleForApiKey(authentication, roleFuture);
@@ -242,7 +248,7 @@ public class ApiKeyServiceTests extends ESTestCase {
             }
         ).when(privilegesStore).getPrivileges(any(Collection.class), any(Collection.class), any(ActionListener.class));
         ApiKeyService service = new ApiKeyService(Settings.EMPTY, Clock.systemUTC(), null, null,
-                ClusterServiceUtils.createClusterService(threadPool));
+                ClusterServiceUtils.createClusterService(threadPool), threadPool);
 
         PlainActionFuture<ApiKeyRoleDescriptors> roleFuture = new PlainActionFuture<>();
         service.getRoleForApiKey(authentication, roleFuture);
@@ -258,4 +264,95 @@ public class ApiKeyServiceTests extends ESTestCase {
             assertThat(result.getLimitedByRoleDescriptors().get(0).getName(), is("limited role"));
         }
     }
+
+    public void testApiKeyCache() {
+        final String apiKey = randomAlphaOfLength(16);
+        Hasher hasher = randomFrom(Hasher.PBKDF2, Hasher.BCRYPT4, Hasher.BCRYPT);
+        final char[] hash = hasher.hash(new SecureString(apiKey.toCharArray()));
+
+        Map<String, Object> sourceMap = new HashMap<>();
+        sourceMap.put("api_key_hash", new String(hash));
+        sourceMap.put("role_descriptors", Collections.singletonMap("a role", Collections.singletonMap("cluster", "all")));
+        sourceMap.put("limited_by_role_descriptors", Collections.singletonMap("limited role", Collections.singletonMap("cluster", "all")));
+        Map<String, Object> creatorMap = new HashMap<>();
+        creatorMap.put("principal", "test_user");
+        creatorMap.put("metadata", Collections.emptyMap());
+        sourceMap.put("creator", creatorMap);
+        sourceMap.put("api_key_invalidated", false);
+
+        ApiKeyService service = new ApiKeyService(Settings.EMPTY, Clock.systemUTC(), null, null,
+            ClusterServiceUtils.createClusterService(threadPool), threadPool);
+        ApiKeyCredentials creds = new ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(apiKey.toCharArray()));
+        PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        AuthenticationResult result = future.actionGet();
+        assertThat(result.isAuthenticated(), is(true));
+        CachedApiKeyHashResult cachedApiKeyHashResult = service.getFromCache(creds.getId());
+        assertNotNull(cachedApiKeyHashResult);
+        assertThat(cachedApiKeyHashResult.success, is(true));
+
+        creds = new ApiKeyCredentials(creds.getId(), new SecureString("foobar".toCharArray()));
+        future = new PlainActionFuture<>();
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        result = future.actionGet();
+        assertThat(result.isAuthenticated(), is(false));
+        final CachedApiKeyHashResult shouldBeSame = service.getFromCache(creds.getId());
+        assertNotNull(shouldBeSame);
+        assertThat(shouldBeSame, sameInstance(cachedApiKeyHashResult));
+
+        sourceMap.put("api_key_hash", new String(hasher.hash(new SecureString("foobar".toCharArray()))));
+        creds = new ApiKeyCredentials(randomAlphaOfLength(12), new SecureString("foobar1".toCharArray()));
+        future = new PlainActionFuture<>();
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        result = future.actionGet();
+        assertThat(result.isAuthenticated(), is(false));
+        cachedApiKeyHashResult = service.getFromCache(creds.getId());
+        assertNotNull(cachedApiKeyHashResult);
+        assertThat(cachedApiKeyHashResult.success, is(false));
+
+        creds = new ApiKeyCredentials(creds.getId(), new SecureString("foobar2".toCharArray()));
+        future = new PlainActionFuture<>();
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        result = future.actionGet();
+        assertThat(result.isAuthenticated(), is(false));
+        assertThat(service.getFromCache(creds.getId()), not(sameInstance(cachedApiKeyHashResult)));
+        assertThat(service.getFromCache(creds.getId()).success, is(false));
+
+        creds = new ApiKeyCredentials(creds.getId(), new SecureString("foobar".toCharArray()));
+        future = new PlainActionFuture<>();
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        result = future.actionGet();
+        assertThat(result.isAuthenticated(), is(true));
+        assertThat(service.getFromCache(creds.getId()), not(sameInstance(cachedApiKeyHashResult)));
+        assertThat(service.getFromCache(creds.getId()).success, is(true));
+    }
+
+    public void testApiKeyCacheDisabled() {
+        final String apiKey = randomAlphaOfLength(16);
+        Hasher hasher = randomFrom(Hasher.PBKDF2, Hasher.BCRYPT4, Hasher.BCRYPT);
+        final char[] hash = hasher.hash(new SecureString(apiKey.toCharArray()));
+        final Settings settings = Settings.builder()
+            .put(ApiKeyService.CACHE_TTL_SETTING.getKey(), "0s")
+            .build();
+
+        Map<String, Object> sourceMap = new HashMap<>();
+        sourceMap.put("api_key_hash", new String(hash));
+        sourceMap.put("role_descriptors", Collections.singletonMap("a role", Collections.singletonMap("cluster", "all")));
+        sourceMap.put("limited_by_role_descriptors", Collections.singletonMap("limited role", Collections.singletonMap("cluster", "all")));
+        Map<String, Object> creatorMap = new HashMap<>();
+        creatorMap.put("principal", "test_user");
+        creatorMap.put("metadata", Collections.emptyMap());
+        sourceMap.put("creator", creatorMap);
+        sourceMap.put("api_key_invalidated", false);
+
+        ApiKeyService service = new ApiKeyService(settings, Clock.systemUTC(), null, null,
+            ClusterServiceUtils.createClusterService(threadPool), threadPool);
+        ApiKeyCredentials creds = new ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(apiKey.toCharArray()));
+        PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
+        service.validateApiKeyCredentials(sourceMap, creds, Clock.systemUTC(), future);
+        AuthenticationResult result = future.actionGet();
+        assertThat(result.isAuthenticated(), is(true));
+        CachedApiKeyHashResult cachedApiKeyHashResult = service.getFromCache(creds.getId());
+        assertNull(cachedApiKeyHashResult);
+    }
 }

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/AuthenticationServiceTests.java

@@ -210,7 +210,7 @@ public class AuthenticationServiceTests extends ESTestCase {
             return null;
         }).when(securityIndex).checkIndexVersionThenExecute(any(Consumer.class), any(Runnable.class));
         ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool);
-        apiKeyService = new ApiKeyService(settings, Clock.systemUTC(), client, securityIndex, clusterService);
+        apiKeyService = new ApiKeyService(settings, Clock.systemUTC(), client, securityIndex, clusterService, threadPool);
         tokenService = new TokenService(settings, Clock.systemUTC(), client, securityIndex, clusterService);
         service = new AuthenticationService(settings, realms, auditTrail, new DefaultAuthenticationFailureHandler(Collections.emptyMap()),
             threadPool, new AnonymousUser(settings), tokenService, apiKeyService);