Service Accounts - cache clearing API (#71605)

This PR adds a new Rest endpoint to clear caches used by service account
authentication.

x-pack/plugin/security/qa/service-account/src/javaRestTest/java/org/elasticsearch/xpack/security/authc/service/ServiceAccountIT.java

@@ -332,6 +332,17 @@ public class ServiceAccountIT extends ESRestTestCase {
         assertThat(responseAsMap(deleteTokenResponse2).get("found"), is(false));
     }
 
+    public void testClearCache() throws IOException {
+        final Request clearCacheRequest = new Request("POST", "_security/service/elastic/fleet-server/credential/token/"
+            + randomFrom("", "*", "api-token-1", "api-token-1,api-token2") + "/_clear_cache");
+        final Response clearCacheResponse = client().performRequest(clearCacheRequest);
+        assertOK(clearCacheResponse);
+        final Map<String, Object> clearCacheResponseMap = responseAsMap(clearCacheResponse);
+        @SuppressWarnings("unchecked")
+        final Map<String, Object> nodesMap = (Map<String, Object>) clearCacheResponseMap.get("_nodes");
+        assertThat(nodesMap.get("failed"), equalTo(0));
+    }
+
     public void testManageOwnApiKey() throws IOException {
         final String token;
         if (randomBoolean()) {

x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/xpack/security/authc/service/ServiceAccountSingleNodeTests.java

@@ -8,13 +8,18 @@
 package org.elasticsearch.xpack.security.authc.service;
 
 import org.elasticsearch.Version;
+import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.cache.Cache;
+import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.test.SecuritySingleNodeTestCase;
+import org.elasticsearch.xpack.core.security.action.ClearSecurityCacheAction;
+import org.elasticsearch.xpack.core.security.action.ClearSecurityCacheRequest;
+import org.elasticsearch.xpack.core.security.action.ClearSecurityCacheResponse;
 import org.elasticsearch.xpack.core.security.action.service.CreateServiceAccountTokenAction;
 import org.elasticsearch.xpack.core.security.action.service.CreateServiceAccountTokenRequest;
 import org.elasticsearch.xpack.core.security.action.service.CreateServiceAccountTokenResponse;
@@ -99,6 +104,40 @@ public class ServiceAccountSingleNodeTests extends SecuritySingleNodeTestCase {
         assertThat(cache.count(), equalTo(0));
     }
 
+    public void testClearCache() {
+        final IndexServiceAccountsTokenStore indexStore = node().injector().getInstance(IndexServiceAccountsTokenStore.class);
+        final Cache<String, ListenableFuture<CachingServiceAccountsTokenStore.CachedResult>> cache = indexStore.getCache();
+        final SecureString secret1 = createApiServiceToken("api-token-1");
+        final SecureString secret2 = createApiServiceToken("api-token-2");
+        assertThat(cache.count(), equalTo(0));
+
+        authenticateWithApiToken("api-token-1", secret1);
+        assertThat(cache.count(), equalTo(1));
+        authenticateWithApiToken("api-token-2", secret2);
+        assertThat(cache.count(), equalTo(2));
+
+        final ClearSecurityCacheRequest clearSecurityCacheRequest1 = new ClearSecurityCacheRequest().cacheName("service");
+        if (randomBoolean()) {
+            clearSecurityCacheRequest1.keys("elastic/fleet-server/");
+        }
+        final PlainActionFuture<ClearSecurityCacheResponse> future1 = new PlainActionFuture<>();
+        client().execute(ClearSecurityCacheAction.INSTANCE, clearSecurityCacheRequest1, future1);
+        assertThat(future1.actionGet().failures().isEmpty(), is(true));
+        assertThat(cache.count(), equalTo(0));
+
+        authenticateWithApiToken("api-token-1", secret1);
+        assertThat(cache.count(), equalTo(1));
+        authenticateWithApiToken("api-token-2", secret2);
+        assertThat(cache.count(), equalTo(2));
+
+        final ClearSecurityCacheRequest clearSecurityCacheRequest2
+            = new ClearSecurityCacheRequest().cacheName("service").keys("elastic/fleet-server/api-token-" + randomFrom("1", "2"));
+        final PlainActionFuture<ClearSecurityCacheResponse> future2 = new PlainActionFuture<>();
+        client().execute(ClearSecurityCacheAction.INSTANCE, clearSecurityCacheRequest2, future2);
+        assertThat(future2.actionGet().failures().isEmpty(), is(true));
+        assertThat(cache.count(), equalTo(1));
+    }
+
     private Client createServiceAccountClient() {
         return createServiceAccountClient(BEARER_TOKEN);
     }
@@ -116,4 +155,21 @@ public class ServiceAccountSingleNodeTests extends SecuritySingleNodeTestCase {
             null, Version.CURRENT, Authentication.AuthenticationType.TOKEN, Map.of("_token_name", tokenName)
         );
     }
+
+    private SecureString createApiServiceToken(String tokenName) {
+        final CreateServiceAccountTokenRequest createServiceAccountTokenRequest =
+            new CreateServiceAccountTokenRequest("elastic", "fleet-server", tokenName);
+        final CreateServiceAccountTokenResponse createServiceAccountTokenResponse =
+            client().execute(CreateServiceAccountTokenAction.INSTANCE, createServiceAccountTokenRequest).actionGet();
+        assertThat(createServiceAccountTokenResponse.getName(), equalTo(tokenName));
+        return createServiceAccountTokenResponse.getValue();
+    }
+
+    private void authenticateWithApiToken(String tokenName, SecureString secret) {
+        final AuthenticateRequest authenticateRequest = new AuthenticateRequest("elastic/fleet-server");
+        final AuthenticateResponse authenticateResponse =
+            createServiceAccountClient(secret.toString())
+                .execute(AuthenticateAction.INSTANCE, authenticateRequest).actionGet();
+        assertThat(authenticateResponse.authentication(), equalTo(getExpectedAuthentication(tokenName)));
+    }
 }

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

@@ -267,6 +267,7 @@ import org.elasticsearch.xpack.security.rest.action.saml.RestSamlInvalidateSessi
 import org.elasticsearch.xpack.security.rest.action.saml.RestSamlLogoutAction;
 import org.elasticsearch.xpack.security.rest.action.saml.RestSamlPrepareAuthenticationAction;
 import org.elasticsearch.xpack.security.rest.action.saml.RestSamlSpMetadataAction;
+import org.elasticsearch.xpack.security.rest.action.service.RestClearServiceAccountTokenStoreCacheAction;
 import org.elasticsearch.xpack.security.rest.action.service.RestCreateServiceAccountTokenAction;
 import org.elasticsearch.xpack.security.rest.action.service.RestDeleteServiceAccountTokenAction;
 import org.elasticsearch.xpack.security.rest.action.service.RestGetServiceAccountAction;
@@ -486,6 +487,7 @@ public class Security extends Plugin implements SystemIndexPlugin, IngestPlugin,
         securityIndex.get().addIndexStateListener(nativeRoleMappingStore::onSecurityIndexStateChange);
 
         final CacheInvalidatorRegistry cacheInvalidatorRegistry = new CacheInvalidatorRegistry();
+        cacheInvalidatorRegistry.registerAlias("service", Set.of("file_service_account_token", "index_service_account_token"));
         components.add(cacheInvalidatorRegistry);
         securityIndex.get().addIndexStateListener(cacheInvalidatorRegistry::onSecurityIndexStateChange);
 
@@ -516,7 +518,7 @@ public class Security extends Plugin implements SystemIndexPlugin, IngestPlugin,
         components.add(indexServiceAccountsTokenStore);
 
         final FileServiceAccountsTokenStore fileServiceAccountsTokenStore =
-            new FileServiceAccountsTokenStore(environment, resourceWatcherService, threadPool);
+            new FileServiceAccountsTokenStore(environment, resourceWatcherService, threadPool, cacheInvalidatorRegistry);
 
         final ServiceAccountService serviceAccountService = new ServiceAccountService(new CompositeServiceAccountsTokenStore(
             List.of(fileServiceAccountsTokenStore, indexServiceAccountsTokenStore), threadPool.getThreadContext()), httpTlsRuntimeCheck);
@@ -583,6 +585,8 @@ public class Security extends Plugin implements SystemIndexPlugin, IngestPlugin,
 
         components.add(new SecurityUsageServices(realms, allRolesStore, nativeRoleMappingStore, ipFilter.get()));
 
+        cacheInvalidatorRegistry.validate();
+
         return components;
     }
 
@@ -906,6 +910,7 @@ public class Security extends Plugin implements SystemIndexPlugin, IngestPlugin,
                 new RestClearRolesCacheAction(settings, getLicenseState()),
                 new RestClearPrivilegesCacheAction(settings, getLicenseState()),
                 new RestClearApiKeyCacheAction(settings, getLicenseState()),
+                new RestClearServiceAccountTokenStoreCacheAction(settings, getLicenseState()),
                 new RestGetUsersAction(settings, getLicenseState()),
                 new RestPutUserAction(settings, getLicenseState()),
                 new RestDeleteUserAction(settings, getLicenseState()),

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

@@ -19,6 +19,7 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.security.authc.support.Hasher;
+import org.elasticsearch.xpack.core.security.support.CacheIteratorHelper;
 import org.elasticsearch.xpack.security.support.CacheInvalidatorRegistry;
 
 import java.util.Collection;
@@ -40,6 +41,7 @@ public abstract class CachingServiceAccountsTokenStore implements ServiceAccount
     private final Settings settings;
     private final ThreadPool threadPool;
     private final Cache<String, ListenableFuture<CachedResult>> cache;
+    private CacheIteratorHelper<String, ListenableFuture<CachedResult>> cacheIteratorHelper;
     private final Hasher hasher;
 
     CachingServiceAccountsTokenStore(Settings settings, ThreadPool threadPool) {
@@ -51,8 +53,10 @@ public abstract class CachingServiceAccountsTokenStore implements ServiceAccount
                 .setExpireAfterWrite(ttl)
                 .setMaximumWeight(CACHE_MAX_TOKENS_SETTING.get(settings))
                 .build();
+            cacheIteratorHelper = new CacheIteratorHelper<>(cache);
         } else {
             cache = null;
+            cacheIteratorHelper = null;
         }
         hasher = Hasher.resolve(CACHE_HASH_ALGO_SETTING.get(settings));
     }
@@ -92,7 +96,12 @@ public abstract class CachingServiceAccountsTokenStore implements ServiceAccount
                 }, listener::onFailure), threadPool.generic(), threadPool.getThreadContext());
             } else {
                 doAuthenticate(token, ActionListener.wrap(success -> {
-                    logger.trace("cache service token [{}] authentication result", token.getQualifiedName());
+                    if (false == success) {
+                        // Do not cache failed attempt
+                        cache.invalidate(token.getQualifiedName(), listenableCacheEntry);
+                    } else {
+                        logger.trace("cache service token [{}] authentication result", token.getQualifiedName());
+                    }
                     listenableCacheEntry.onResponse(new CachedResult(hasher, success, token));
                     listener.onResponse(success);
                 }, e -> {
@@ -107,12 +116,25 @@ public abstract class CachingServiceAccountsTokenStore implements ServiceAccount
         }
     }
 
+    /**
+     * Invalidate cache entries with keys matching to the specified qualified token names.
+     * @param qualifiedTokenNames The list of qualified toke names. If a name has trailing
+     *                            slash, it is treated as a prefix wildcard, i.e. all keys
+     *                            with this prefix are considered matching.
+     */
     @Override
     public final void invalidate(Collection<String> qualifiedTokenNames) {
         if (cache != null) {
             logger.trace("invalidating cache for service token [{}]",
                 Strings.collectionToCommaDelimitedString(qualifiedTokenNames));
-            qualifiedTokenNames.forEach(cache::invalidate);
+            for (String qualifiedTokenName : qualifiedTokenNames) {
+                if (qualifiedTokenName.endsWith("/")) {
+                    // Wildcard case of invalidating all tokens for a service account, e.g. "elastic/fleet-server/"
+                    cacheIteratorHelper.removeKeysIf(key -> key.startsWith(qualifiedTokenName));
+                } else {
+                    cache.invalidate(qualifiedTokenName);
+                }
+            }
         }
     }
 

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

@@ -23,6 +23,7 @@ import org.elasticsearch.xpack.core.security.action.service.TokenInfo;
 import org.elasticsearch.xpack.core.security.authc.support.Hasher;
 import org.elasticsearch.xpack.core.security.support.NoOpLogger;
 import org.elasticsearch.xpack.security.authc.service.ServiceAccount.ServiceAccountId;
+import org.elasticsearch.xpack.security.support.CacheInvalidatorRegistry;
 import org.elasticsearch.xpack.security.support.FileLineParser;
 import org.elasticsearch.xpack.security.support.FileReloadListener;
 import org.elasticsearch.xpack.security.support.SecurityFiles;
@@ -47,7 +48,8 @@ public class FileServiceAccountsTokenStore extends CachingServiceAccountsTokenSt
     private final CopyOnWriteArrayList<Runnable> refreshListeners;
     private volatile Map<String, char[]> tokenHashes;
 
-    public FileServiceAccountsTokenStore(Environment env, ResourceWatcherService resourceWatcherService, ThreadPool threadPool) {
+    public FileServiceAccountsTokenStore(Environment env, ResourceWatcherService resourceWatcherService, ThreadPool threadPool,
+                                         CacheInvalidatorRegistry cacheInvalidatorRegistry) {
         super(env.settings(), threadPool);
         file = resolveFile(env);
         FileWatcher watcher = new FileWatcher(file.getParent());
@@ -63,6 +65,7 @@ public class FileServiceAccountsTokenStore extends CachingServiceAccountsTokenSt
             throw new IllegalStateException("Failed to load service_tokens file [" + file + "]", e);
         }
         refreshListeners = new CopyOnWriteArrayList<>(List.of(this::invalidateAll));
+        cacheInvalidatorRegistry.registerCacheInvalidator("file_service_account_token", this);
     }
 
     @Override
@@ -89,6 +92,11 @@ public class FileServiceAccountsTokenStore extends CachingServiceAccountsTokenSt
         refreshListeners.add(listener);
     }
 
+    @Override
+    public boolean shouldClearOnSecurityIndexStateChange() {
+        return false;
+    }
+
     private void notifyRefresh() {
         refreshListeners.forEach(Runnable::run);
     }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/rest/action/service/RestClearServiceAccountTokenStoreCacheAction.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.security.rest.action.service;
+
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestActions;
+import org.elasticsearch.xpack.core.security.action.ClearSecurityCacheAction;
+import org.elasticsearch.xpack.core.security.action.ClearSecurityCacheRequest;
+import org.elasticsearch.xpack.core.security.support.Validation;
+import org.elasticsearch.xpack.security.rest.action.SecurityBaseRestHandler;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.elasticsearch.rest.RestRequest.Method.POST;
+
+public class RestClearServiceAccountTokenStoreCacheAction extends SecurityBaseRestHandler {
+
+    public RestClearServiceAccountTokenStoreCacheAction(Settings settings, XPackLicenseState licenseState) {
+        super(settings, licenseState);
+    }
+
+    @Override
+    public List<Route> routes() {
+        return List.of(new Route(POST, "/_security/service/{namespace}/{service}/credential/token/{name}/_clear_cache"));
+    }
+
+    @Override
+    public String getName() {
+        return "xpack_security_clear_service_account_token_store_cache";
+    }
+
+    @Override
+    protected RestChannelConsumer innerPrepareRequest(RestRequest request, NodeClient client) throws IOException {
+        final String namespace = request.param("namespace");
+        final String service = request.param("service");
+        String[] tokenNames = request.paramAsStringArrayOrEmptyIfAll("name");
+
+        ClearSecurityCacheRequest req = new ClearSecurityCacheRequest().cacheName("service");
+        if (tokenNames.length == 0) {
+            // This is the wildcard case for tokenNames
+            req.keys(namespace + "/" + service + "/");
+        } else {
+            final Set<String> qualifiedTokenNames = new HashSet<>(tokenNames.length);
+            for (String name: tokenNames) {
+                if (false == Validation.isValidServiceAccountTokenName(name)) {
+                    throw new IllegalArgumentException(Validation.INVALID_SERVICE_ACCOUNT_TOKEN_NAME_MESSAGE + " got: [" + name + "]");
+                }
+                qualifiedTokenNames.add(namespace + "/" + service + "/" + name);
+            }
+            req.keys(qualifiedTokenNames.toArray(String[]::new));
+        }
+        return channel -> client.execute(ClearSecurityCacheAction.INSTANCE, req, new RestActions.NodesResponseRestListener<>(channel));
+    }
+}

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

@@ -7,9 +7,13 @@
 
 package org.elasticsearch.xpack.security.support;
 
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.util.set.Sets;
+
 import java.util.Collection;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 import static org.elasticsearch.xpack.security.support.SecurityIndexManager.isIndexDeleted;
@@ -21,6 +25,7 @@ import static org.elasticsearch.xpack.security.support.SecurityIndexManager.isMo
 public class CacheInvalidatorRegistry {
 
     private final Map<String, CacheInvalidator> cacheInvalidators = new ConcurrentHashMap<>();
+    private final Map<String, Set<String>> cacheAliases = new ConcurrentHashMap<>();
 
     public CacheInvalidatorRegistry() {
     }
@@ -32,16 +37,57 @@ public class CacheInvalidatorRegistry {
         cacheInvalidators.put(name, cacheInvalidator);
     }
 
+    public void registerAlias(String alias, Set<String> names) {
+        Objects.requireNonNull(alias, "cache alias cannot be null");
+        if (names.isEmpty()) {
+            throw new IllegalArgumentException("cache names cannot be empty for aliasing");
+        }
+        if (cacheAliases.containsKey(alias)) {
+            throw new IllegalArgumentException("cache alias already exists: [" + alias + "]");
+        }
+        cacheAliases.put(alias, names);
+    }
+
+    public void validate() {
+        for (String alias : cacheAliases.keySet()) {
+            if (cacheInvalidators.containsKey(alias)) {
+                throw new IllegalStateException("cache alias cannot clash with cache name: [" + alias + "]");
+            }
+            final Set<String> names = cacheAliases.get(alias);
+            if (false == cacheInvalidators.keySet().containsAll(names)) {
+                throw new IllegalStateException("cache names not found: ["
+                    + Strings.collectionToCommaDelimitedString(Sets.difference(names, cacheInvalidators.keySet())) + "]");
+            }
+        }
+    }
+
     public void onSecurityIndexStateChange(SecurityIndexManager.State previousState, SecurityIndexManager.State currentState) {
         if (isMoveFromRedToNonRed(previousState, currentState)
             || isIndexDeleted(previousState, currentState)
             || Objects.equals(previousState.indexUUID, currentState.indexUUID) == false
             || previousState.isIndexUpToDate != currentState.isIndexUpToDate) {
-            cacheInvalidators.values().forEach(CacheInvalidator::invalidateAll);
+            cacheInvalidators.values().stream()
+                .filter(CacheInvalidator::shouldClearOnSecurityIndexStateChange).forEach(CacheInvalidator::invalidateAll);
         }
     }
 
     public void invalidateByKey(String cacheName, Collection<String> keys) {
+        if (cacheAliases.containsKey(cacheName)) {
+            cacheAliases.get(cacheName).forEach(name -> doInvalidateByKey(name, keys));
+        } else {
+            doInvalidateByKey(cacheName, keys);
+        }
+    }
+
+    public void invalidateCache(String cacheName) {
+        if (cacheAliases.containsKey(cacheName)) {
+            cacheAliases.get(cacheName).forEach(this::doInvalidateCache);
+        } else {
+            doInvalidateCache(cacheName);
+        }
+    }
+
+    private void doInvalidateByKey(String cacheName, Collection<String> keys) {
         final CacheInvalidator cacheInvalidator = cacheInvalidators.get(cacheName);
         if (cacheInvalidator != null) {
             cacheInvalidator.invalidate(keys);
@@ -50,7 +96,7 @@ public class CacheInvalidatorRegistry {
         }
     }
 
-    public void invalidateCache(String cacheName) {
+    private void doInvalidateCache(String cacheName) {
         final CacheInvalidator cacheInvalidator = cacheInvalidators.get(cacheName);
         if (cacheInvalidator != null) {
             cacheInvalidator.invalidateAll();
@@ -63,5 +109,9 @@ public class CacheInvalidatorRegistry {
         void invalidate(Collection<String> keys);
 
         void invalidateAll();
+
+        default boolean shouldClearOnSecurityIndexStateChange() {
+            return true;
+        }
     }
 }

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

@@ -17,14 +17,18 @@ import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.security.action.service.TokenInfo;
+import org.elasticsearch.xpack.core.security.support.ValidationTests;
 import org.elasticsearch.xpack.security.authc.service.ServiceAccount.ServiceAccountId;
 import org.junit.After;
 import org.junit.Before;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
@@ -100,14 +104,16 @@ public class CachingServiceAccountsTokenStoreTests extends ESTestCase {
         store.authenticate(token2Invalid, future4);
         assertThat(future4.get(), is(false));
         assertThat(doAuthenticateInvoked.get(), is(true));
-        assertThat(cache.count(), equalTo(2));
+        assertThat(cache.count(), equalTo(1));  // invalid token not cached
         doAuthenticateInvoked.set(false); // reset
 
-        // 5th auth with the wrong token2 again should use cache
+        // 5th auth with the wrong token2 again does not use cache
         final PlainActionFuture<Boolean> future5 = new PlainActionFuture<>();
         store.authenticate(token2Invalid, future5);
         assertThat(future5.get(), is(false));
-        assertThat(doAuthenticateInvoked.get(), is(false));
+        assertThat(doAuthenticateInvoked.get(), is(true));
+        assertThat(cache.count(), equalTo(1));  // invalid token not cached
+        doAuthenticateInvoked.set(false); // reset
 
         // 6th auth with the right token2
         final PlainActionFuture<Boolean> future6 = new PlainActionFuture<>();
@@ -159,4 +165,62 @@ public class CachingServiceAccountsTokenStoreTests extends ESTestCase {
         store.authenticate(mock(ServiceAccountToken.class), future);
         assertThat(future.get(), is(success));
     }
+
+    @SuppressWarnings("unchecked")
+    public void testCacheInvalidateByKeys() {
+        final CachingServiceAccountsTokenStore store = new CachingServiceAccountsTokenStore(globalSettings, threadPool) {
+            @Override
+            void doAuthenticate(ServiceAccountToken token, ActionListener<Boolean> listener) {
+                listener.onResponse(true);
+            }
+
+            @Override
+            public void findTokensFor(ServiceAccountId accountId, ActionListener<Collection<TokenInfo>> listener) {
+                listener.onFailure(new UnsupportedOperationException());
+            }
+        };
+
+        final ServiceAccountId accountId = new ServiceAccountId(randomAlphaOfLengthBetween(3, 8), randomAlphaOfLengthBetween(3, 8));
+
+        final ArrayList<ServiceAccountToken> tokens = new ArrayList<>();
+        IntStream.range(0, randomIntBetween(3, 8)).forEach(i -> {
+            final ServiceAccountToken token = ServiceAccountToken.newToken(accountId,
+                randomValueOtherThanMany(n -> n.length() > 248, ValidationTests::randomTokenName));
+            tokens.add(token);
+            store.authenticate(token, mock(ActionListener.class));
+
+            final ServiceAccountToken tokenWithSuffix =
+                ServiceAccountToken.newToken(accountId, token.getTokenName() + randomAlphaOfLengthBetween(3, 8));
+            tokens.add(tokenWithSuffix);
+            store.authenticate(tokenWithSuffix, mock(ActionListener.class));
+        });
+        assertThat(store.getCache().count(), equalTo(tokens.size()));
+
+        // Invalidate a single entry
+        store.invalidate(List.of(randomFrom(tokens).getQualifiedName()));
+        assertThat(store.getCache().count(), equalTo(tokens.size() - 1));
+
+        // Invalidate all entries
+        store.invalidate(List.of(accountId.asPrincipal() + "/"));
+        assertThat(store.getCache().count(), equalTo(0));
+
+        // auth everything again
+        tokens.forEach(t -> store.authenticate(t, mock(ActionListener.class)));
+        assertThat(store.getCache().count(), equalTo(tokens.size()));
+
+        final int nInvalidation = randomIntBetween(1, tokens.size() - 1);
+        final List<String> tokenIdsToInvalidate = randomSubsetOf(nInvalidation, tokens).stream()
+            .map(ServiceAccountToken::getQualifiedName)
+            .collect(Collectors.toList());
+        final boolean hasPrefixWildcard = randomBoolean();
+        if (hasPrefixWildcard) {
+            tokenIdsToInvalidate.add(accountId.asPrincipal() + "/");
+        }
+        store.invalidate(tokenIdsToInvalidate);
+        if (hasPrefixWildcard) {
+            assertThat(store.getCache().count(), equalTo(0));
+        } else {
+            assertThat(store.getCache().count(), equalTo(tokens.size() - nInvalidation));
+        }
+    }
 }

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

@@ -22,6 +22,7 @@ import org.elasticsearch.xpack.core.security.action.service.TokenInfo;
 import org.elasticsearch.xpack.core.security.audit.logfile.CapturingLogger;
 import org.elasticsearch.xpack.core.security.authc.support.Hasher;
 import org.elasticsearch.xpack.security.authc.service.ServiceAccount.ServiceAccountId;
+import org.elasticsearch.xpack.security.support.CacheInvalidatorRegistry;
 import org.junit.After;
 import org.junit.Before;
 
@@ -118,7 +119,8 @@ public class FileServiceAccountsTokenStoreTests extends ESTestCase {
         try (ResourceWatcherService watcherService = new ResourceWatcherService(settings, threadPool)) {
             final CountDownLatch latch = new CountDownLatch(5);
 
-            FileServiceAccountsTokenStore store = new FileServiceAccountsTokenStore(env, watcherService, threadPool);
+            FileServiceAccountsTokenStore store = new FileServiceAccountsTokenStore(env, watcherService, threadPool,
+                mock(CacheInvalidatorRegistry.class));
             store.addListener(latch::countDown);
             //Token name shares the hashing algorithm name for convenience
             String tokenName = settings.get("xpack.security.authc.service_token_hashing.algorithm");
@@ -193,7 +195,8 @@ public class FileServiceAccountsTokenStoreTests extends ESTestCase {
         Files.createDirectories(configDir);
         Path targetFile = configDir.resolve("service_tokens");
         Files.copy(serviceTokensSourceFile, targetFile, StandardCopyOption.REPLACE_EXISTING);
-        FileServiceAccountsTokenStore store = new FileServiceAccountsTokenStore(env, mock(ResourceWatcherService.class), threadPool);
+        FileServiceAccountsTokenStore store = new FileServiceAccountsTokenStore(env, mock(ResourceWatcherService.class), threadPool,
+            mock(CacheInvalidatorRegistry.class));
 
         final ServiceAccountId accountId = new ServiceAccountId("elastic", "fleet-server");
         final PlainActionFuture<Collection<TokenInfo>> future1 = new PlainActionFuture<>();

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/rest/action/service/RestClearServiceAccountTokenStoreCacheActionTests.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.security.rest.action.service;
+
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.test.rest.FakeRestRequest;
+import org.elasticsearch.test.rest.RestActionTestCase;
+import org.elasticsearch.xpack.core.XPackSettings;
+import org.elasticsearch.xpack.core.security.action.ClearSecurityCacheRequest;
+import org.elasticsearch.xpack.core.security.action.ClearSecurityCacheResponse;
+import org.elasticsearch.xpack.core.security.support.Validation;
+import org.elasticsearch.xpack.core.security.support.ValidationTests;
+import org.junit.Before;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class RestClearServiceAccountTokenStoreCacheActionTests extends RestActionTestCase {
+
+    private Settings settings;
+    private XPackLicenseState licenseState;
+    private AtomicReference<ClearSecurityCacheRequest> requestHolder;
+
+    @Before
+    public void init() {
+        settings = Settings.builder().put(XPackSettings.SECURITY_ENABLED.getKey(), true).build();
+        licenseState = mock(XPackLicenseState.class);
+        when(licenseState.isSecurityEnabled()).thenReturn(true);
+        requestHolder = new AtomicReference<>();
+        controller().registerHandler(new RestClearServiceAccountTokenStoreCacheAction(settings, licenseState));
+        verifyingClient.setExecuteVerifier(((actionType, actionRequest) -> {
+            assertThat(actionRequest, instanceOf(ClearSecurityCacheRequest.class));
+            requestHolder.set((ClearSecurityCacheRequest) actionRequest);
+            return mock(ClearSecurityCacheResponse.class);
+        }));
+    }
+
+    public void testInnerPrepareRequestWithEmptyTokenName() {
+        final String namespace = randomAlphaOfLengthBetween(3, 8);
+        final String service = randomAlphaOfLengthBetween(3, 8);
+        final String name = randomFrom("", "*", "_all");
+        final FakeRestRequest restRequest = new FakeRestRequest.Builder(NamedXContentRegistry.EMPTY)
+            .withMethod(RestRequest.Method.POST)
+            .withPath("/_security/service/" + namespace + "/" + service + "/credential/token/" + name + "/_clear_cache")
+            .build();
+
+        dispatchRequest(restRequest);
+
+        final ClearSecurityCacheRequest clearSecurityCacheRequest = requestHolder.get();
+        assertThat(clearSecurityCacheRequest.keys(), equalTo(new String[]{ namespace + "/" + service + "/"}));
+    }
+
+    public void testInnerPrepareRequestWithValidTokenNames() {
+        final String namespace = randomAlphaOfLengthBetween(3, 8);
+        final String service = randomAlphaOfLengthBetween(3, 8);
+        final String[] names = randomArray(1, 3, String[]::new, ValidationTests::randomTokenName);
+        final FakeRestRequest restRequest = new FakeRestRequest.Builder(NamedXContentRegistry.EMPTY)
+            .withMethod(RestRequest.Method.POST)
+            .withPath("/_security/service/" + namespace + "/" + service + "/credential/token/"
+                + Strings.arrayToCommaDelimitedString(names) + "/_clear_cache")
+            .build();
+
+        dispatchRequest(restRequest);
+
+        final ClearSecurityCacheRequest clearSecurityCacheRequest = requestHolder.get();
+        assertThat(Set.of(clearSecurityCacheRequest.keys()),
+            equalTo(Arrays.stream(names).map(n -> namespace + "/" + service + "/" + n).collect(Collectors.toUnmodifiableSet())));
+    }
+
+    public void testInnerPrepareRequestWillThrowErrorOnInvalidTokenNames() {
+        final RestClearServiceAccountTokenStoreCacheAction restAction =
+            new RestClearServiceAccountTokenStoreCacheAction(Settings.EMPTY, mock(XPackLicenseState.class));
+        final String[] names = randomArray(2, 4, String[]::new,
+            () -> randomValueOtherThanMany(n -> n.contains(","), ValidationTests::randomInvalidTokenName));
+
+        final FakeRestRequest fakeRestRequest = new FakeRestRequest.Builder(NamedXContentRegistry.EMPTY)
+            .withParams(Map.of(
+                "namespace", randomAlphaOfLengthBetween(3, 8),
+                "service", randomAlphaOfLengthBetween(3, 8),
+                "name", Strings.arrayToCommaDelimitedString(names)))
+            .build();
+
+        final IllegalArgumentException e =
+            expectThrows(IllegalArgumentException.class, () -> restAction.innerPrepareRequest(fakeRestRequest, mock(NodeClient.class)));
+        assertThat(e.getMessage(), containsString(Validation.INVALID_SERVICE_ACCOUNT_TOKEN_NAME_MESSAGE));
+    }
+}

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/support/CacheInvalidatorRegistryTests.java

@@ -16,12 +16,14 @@ import org.junit.Before;
 
 import java.time.Instant;
 import java.util.List;
+import java.util.Set;
 
 import static org.hamcrest.Matchers.containsString;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class CacheInvalidatorRegistryTests extends ESTestCase {
 
@@ -42,9 +44,13 @@ public class CacheInvalidatorRegistryTests extends ESTestCase {
 
     public void testSecurityIndexStateChangeWillInvalidateAllRegisteredInvalidators() {
         final CacheInvalidator invalidator1 = mock(CacheInvalidator.class);
+        when(invalidator1.shouldClearOnSecurityIndexStateChange()).thenReturn(true);
         cacheInvalidatorRegistry.registerCacheInvalidator("service1", invalidator1);
         final CacheInvalidator invalidator2 = mock(CacheInvalidator.class);
+        when(invalidator2.shouldClearOnSecurityIndexStateChange()).thenReturn(true);
         cacheInvalidatorRegistry.registerCacheInvalidator("service2", invalidator2);
+        final CacheInvalidator invalidator3 = mock(CacheInvalidator.class);
+        cacheInvalidatorRegistry.registerCacheInvalidator("service3", invalidator3);
 
         final SecurityIndexManager.State previousState = SecurityIndexManager.State.UNRECOVERED_STATE;
         final SecurityIndexManager.State currentState = new SecurityIndexManager.State(
@@ -54,6 +60,7 @@ public class CacheInvalidatorRegistryTests extends ESTestCase {
         cacheInvalidatorRegistry.onSecurityIndexStateChange(previousState, currentState);
         verify(invalidator1).invalidateAll();
         verify(invalidator2).invalidateAll();
+        verify(invalidator3, never()).invalidateAll();
     }
 
     public void testInvalidateByKeyCallsCorrectInvalidatorObject() {
@@ -89,4 +96,46 @@ public class CacheInvalidatorRegistryTests extends ESTestCase {
                 () -> cacheInvalidatorRegistry.invalidateCache("non-exist"));
         assertThat(e.getMessage(), containsString("No cache named [non-exist] is found"));
     }
+
+    public void testRegisterAlias() {
+        final CacheInvalidator invalidator1 = mock(CacheInvalidator.class);
+        cacheInvalidatorRegistry.registerCacheInvalidator("cache1", invalidator1);
+        final CacheInvalidator invalidator2 = mock(CacheInvalidator.class);
+        cacheInvalidatorRegistry.registerCacheInvalidator("cache2", invalidator2);
+
+        final NullPointerException e1 =
+            expectThrows(NullPointerException.class, () -> cacheInvalidatorRegistry.registerAlias(null, Set.of()));
+        assertThat(e1.getMessage(), containsString("cache alias cannot be null"));
+
+        final IllegalArgumentException e2 =
+            expectThrows(IllegalArgumentException.class, () -> cacheInvalidatorRegistry.registerAlias("alias1", Set.of()));
+        assertThat(e2.getMessage(), containsString("cache names cannot be empty for aliasing"));
+
+        cacheInvalidatorRegistry.registerAlias("alias1", randomFrom(Set.of("cache1"), Set.of("cache1", "cache2")));
+
+        final IllegalArgumentException e3 =
+            expectThrows(IllegalArgumentException.class, () -> cacheInvalidatorRegistry.registerAlias("alias1", Set.of("cache1")));
+        assertThat(e3.getMessage(), containsString("cache alias already exists"));
+
+        // validation should pass
+        cacheInvalidatorRegistry.validate();
+    }
+
+    public void testValidateWillThrowForClashingAliasAndCacheNames() {
+        final CacheInvalidator invalidator1 = mock(CacheInvalidator.class);
+        cacheInvalidatorRegistry.registerCacheInvalidator("cache1", invalidator1);
+        cacheInvalidatorRegistry.registerAlias("cache1", Set.of("cache1"));
+        final IllegalStateException e =
+            expectThrows(IllegalStateException.class, () -> cacheInvalidatorRegistry.validate());
+        assertThat(e.getMessage(), containsString("cache alias cannot clash with cache name"));
+    }
+
+    public void testValidateWillThrowForNotFoundCacheNames() {
+        final CacheInvalidator invalidator1 = mock(CacheInvalidator.class);
+        cacheInvalidatorRegistry.registerCacheInvalidator("cache1", invalidator1);
+        cacheInvalidatorRegistry.registerAlias("alias1", Set.of("cache1", "cache2"));
+        final IllegalStateException e =
+            expectThrows(IllegalStateException.class, () -> cacheInvalidatorRegistry.validate());
+        assertThat(e.getMessage(), containsString("cache names not found: [cache2]"));
+    }
 }