Improve role cache efficiency for API key roles (#58156) (#59397)

This PR ensure that same roles are cached only once even when they are from different API keys.
API key role descriptors and limited role descriptors are now saved in Authentication#metadata
as raw bytes instead of deserialised Map<String, Object>.
Hashes of these bytes are used as keys for API key roles. Only when the required role is not found
in the cache, they will be deserialised to build the RoleDescriptors. The deserialisation is directly
from raw bytes to RoleDescriptors without going through the current detour of
"bytes -> Map -> bytes -> RoleDescriptors".

libs/x-content/src/main/java/org/elasticsearch/common/xcontent/AbstractObjectParser.java

@@ -210,6 +210,12 @@ public abstract class AbstractObjectParser<Value, Context> {
         declareField(consumer, p -> p.longValue(), field, ValueType.LONG);
     }
 
+    public void declareLongOrNull(BiConsumer<Value, Long> consumer, long nullValue, ParseField field) {
+        // Using a method reference here angers some compilers
+        declareField(consumer, p -> p.currentToken() == XContentParser.Token.VALUE_NULL ? nullValue : p.longValue(),
+            field, ValueType.LONG_OR_NULL);
+    }
+
     public void declareInt(BiConsumer<Value, Integer> consumer, ParseField field) {
         // Using a method reference here angers some compilers
         declareField(consumer, p -> p.intValue(), field, ValueType.INT);

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

@@ -10,9 +10,12 @@ import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchSecurityException;
 import org.elasticsearch.Version;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.Authentication.AuthenticationType;
@@ -23,14 +26,21 @@ import org.elasticsearch.xpack.core.security.user.User;
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Objects;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import static org.elasticsearch.xpack.core.security.authc.Authentication.VERSION_API_KEY_ROLES_AS_BYTES;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
+
 /**
  * A lightweight utility that can find the current user and authentication information for the local thread.
  */
 public class SecurityContext {
+
     private final Logger logger = LogManager.getLogger(SecurityContext.class);
 
     private final ThreadContext threadContext;
@@ -149,8 +159,27 @@ public class SecurityContext {
         final Authentication authentication = getAuthentication();
         try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
             setAuthentication(new Authentication(authentication.getUser(), authentication.getAuthenticatedBy(),
-                authentication.getLookedUpBy(), version, authentication.getAuthenticationType(), authentication.getMetadata()));
+                authentication.getLookedUpBy(), version, authentication.getAuthenticationType(),
+                rewriteMetadataForApiKeyRoleDescriptors(version, authentication)));
             consumer.accept(original);
         }
     }
+
+    private Map<String, Object> rewriteMetadataForApiKeyRoleDescriptors(Version streamVersion, Authentication authentication) {
+        Map<String, Object> metadata = authentication.getMetadata();
+        if (authentication.getAuthenticationType() == AuthenticationType.API_KEY
+            && authentication.getVersion().onOrAfter(VERSION_API_KEY_ROLES_AS_BYTES)
+            && streamVersion.before(VERSION_API_KEY_ROLES_AS_BYTES)) {
+            metadata = new HashMap<>(metadata);
+            metadata.put(
+                API_KEY_ROLE_DESCRIPTORS_KEY,
+                XContentHelper.convertToMap(
+                    (BytesReference) metadata.get(API_KEY_ROLE_DESCRIPTORS_KEY), false, XContentType.JSON).v2());
+            metadata.put(
+                API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
+                XContentHelper.convertToMap(
+                    (BytesReference) metadata.get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY), false, XContentType.JSON).v2());
+        }
+        return metadata;
+    }
 }

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

@@ -27,6 +27,8 @@ import java.util.Objects;
 // That interface can be removed
 public class Authentication implements ToXContentObject {
 
+    public static final Version VERSION_API_KEY_ROLES_AS_BYTES = Version.V_7_9_0;
+
     private final User user;
     private final RealmRef authenticatedBy;
     private final RealmRef lookedUpBy;

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

@@ -8,6 +8,8 @@ package org.elasticsearch.xpack.core.security.authc;
 public final class AuthenticationField {
 
     public static final String AUTHENTICATION_KEY = "_xpack_security_authentication";
+    public static final String API_KEY_ROLE_DESCRIPTORS_KEY = "_security_api_key_role_descriptors";
+    public static final String API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY = "_security_api_key_limited_by_role_descriptors";
 
     private AuthenticationField() {}
 }

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

@@ -35,11 +35,13 @@ import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.CharArrays;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.ParseField;
 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.collect.Tuple;
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Setting;
@@ -51,9 +53,13 @@ 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.InstantiatingObjectParser;
+import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.common.xcontent.ObjectParserHelper;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentLocation;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
@@ -106,11 +112,16 @@ import java.util.stream.Collectors;
 import javax.crypto.SecretKeyFactory;
 
 import static org.elasticsearch.index.mapper.MapperService.SINGLE_MAPPING_NAME;
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.optionalConstructorArg;
 import static org.elasticsearch.action.bulk.TransportSingleItemBulkWriteAction.toSingleItemBulkRequest;
 import static org.elasticsearch.search.SearchService.DEFAULT_KEEPALIVE_SETTING;
 import static org.elasticsearch.xpack.core.ClientHelper.SECURITY_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
 import static org.elasticsearch.xpack.core.security.authc.Authentication.AuthenticationType;
+import static org.elasticsearch.xpack.core.security.authc.Authentication.VERSION_API_KEY_ROLES_AS_BYTES;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
 import static org.elasticsearch.xpack.core.security.index.RestrictedIndicesNames.SECURITY_MAIN_ALIAS;
 import static org.elasticsearch.xpack.security.Security.SECURITY_CRYPTO_THREAD_POOL_NAME;
 
@@ -124,9 +135,6 @@ public class ApiKeyService {
     public static final String API_KEY_REALM_TYPE = "_es_api_key";
     public static final String API_KEY_CREATOR_REALM_NAME = "_security_api_key_creator_realm_name";
     public static final String API_KEY_CREATOR_REALM_TYPE = "_security_api_key_creator_realm_type";
-    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";
-
 
     public static final Setting<String> PASSWORD_HASHING_ALGORITHM = new Setting<>(
         "xpack.security.authc.api_key.hashing.algorithm", "pbkdf2", Function.identity(), v -> {
@@ -355,8 +363,13 @@ public class ApiKeyService {
             .request();
         executeAsyncWithOrigin(ctx, SECURITY_ORIGIN, getRequest, ActionListener.<GetResponse>wrap(response -> {
                 if (response.isExists()) {
-                    final Map<String, Object> source = response.getSource();
+                    final ApiKeyDoc apiKeyDoc;
-                    validateApiKeyCredentials(docId, source, credentials, clock, ActionListener.delegateResponse(listener, (l, e) -> {
+                    try (XContentParser parser = XContentHelper.createParser(
+                        NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE,
+                        response.getSourceAsBytesRef(), XContentType.JSON)) {
+                        apiKeyDoc = ApiKeyDoc.fromXContent(parser);
+                    }
+                    validateApiKeyCredentials(docId, apiKeyDoc, credentials, clock, ActionListener.delegateResponse(listener, (l, e) -> {
                         if (ExceptionsHelper.unwrapCause(e) instanceof EsRejectedExecutionException) {
                             listener.onResponse(AuthenticationResult.terminate("server is too busy to respond", e));
                         } else {
@@ -380,6 +393,9 @@ public class ApiKeyService {
     }
 
     /**
+     * This method is kept for BWC and should only be used for authentication objects created before v7.9.0.
+     * For authentication of newer versions, use {@link #getApiKeyIdAndRoleBytes}
+     *
      * The current request has been authenticated by an API key and this method enables the
      * retrieval of role descriptors that are associated with the api key
      */
@@ -387,10 +403,11 @@ public class ApiKeyService {
         if (authentication.getAuthenticationType() != AuthenticationType.API_KEY) {
             throw new IllegalStateException("authentication type must be api key but is " + authentication.getAuthenticationType());
         }
+        assert authentication.getVersion()
+            .before(VERSION_API_KEY_ROLES_AS_BYTES) : "This method only applies to authentication objects created before v7.9.0";
 
         final Map<String, Object> metadata = authentication.getMetadata();
         final String apiKeyId = (String) metadata.get(API_KEY_ID_KEY);
-
         final Map<String, Object> roleDescriptors = (Map<String, Object>) metadata.get(API_KEY_ROLE_DESCRIPTORS_KEY);
         final Map<String, Object> authnRoleDescriptors = (Map<String, Object>) metadata.get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY);
 
@@ -406,6 +423,19 @@ public class ApiKeyService {
         }
     }
 
+    public Tuple<String, BytesReference> getApiKeyIdAndRoleBytes(Authentication authentication, boolean limitedBy) {
+        if (authentication.getAuthenticationType() != AuthenticationType.API_KEY) {
+            throw new IllegalStateException("authentication type must be api key but is " + authentication.getAuthenticationType());
+        }
+        assert authentication.getVersion()
+            .onOrAfter(VERSION_API_KEY_ROLES_AS_BYTES) : "This method only applies to authentication objects created on or after v7.9.0";
+
+        final Map<String, Object> metadata = authentication.getMetadata();
+        return new Tuple<>(
+            (String) metadata.get(API_KEY_ID_KEY),
+            (BytesReference) metadata.get(limitedBy ? API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY : API_KEY_ROLE_DESCRIPTORS_KEY));
+    }
+
     public static class ApiKeyRoleDescriptors {
 
         private final String apiKeyId;
@@ -452,27 +482,48 @@ public class ApiKeyService {
             }).collect(Collectors.toList());
     }
 
+    public List<RoleDescriptor> parseRoleDescriptors(final String apiKeyId, BytesReference bytesReference) {
+        if (bytesReference == null) {
+            return Collections.emptyList();
+        }
+
+        List<RoleDescriptor> roleDescriptors = new ArrayList<>();
+        try (
+            XContentParser parser = XContentHelper.createParser(
+                NamedXContentRegistry.EMPTY,
+                new ApiKeyLoggingDeprecationHandler(deprecationLogger, apiKeyId),
+                bytesReference,
+                XContentType.JSON)) {
+            parser.nextToken(); // skip outer start object
+            while (parser.nextToken() != XContentParser.Token.END_OBJECT) {
+                parser.nextToken(); // role name
+                String roleName = parser.currentName();
+                roleDescriptors.add(RoleDescriptor.parse(roleName, parser, false));
+            }
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+        return Collections.unmodifiableList(roleDescriptors);
+    }
+
     /**
      * Validates the ApiKey using the source map
      * @param docId the identifier of the document that was retrieved from the security index
-     * @param source the source map from a get of the ApiKey document
+     * @param apiKeyDoc the partially deserialized API key document
      * @param credentials the credentials provided by the user
      * @param listener the listener to notify after verification
      */
-    void validateApiKeyCredentials(String docId, Map<String, Object> source, ApiKeyCredentials credentials, Clock clock,
+    void validateApiKeyCredentials(String docId, ApiKeyDoc apiKeyDoc, ApiKeyCredentials credentials, Clock clock,
                                    ActionListener<AuthenticationResult> listener) {
-        final String docType = (String) source.get("doc_type");
+        if ("api_key".equals(apiKeyDoc.docType) == false) {
-        final Boolean invalidated = (Boolean) source.get("api_key_invalidated");
-        if ("api_key".equals(docType) == false) {
             listener.onResponse(
-                AuthenticationResult.unsuccessful("document [" + docId + "] is [" + docType + "] not an api key", null));
+                AuthenticationResult.unsuccessful("document [" + docId + "] is [" + apiKeyDoc.docType + "] not an api key", null));
-        } else if (invalidated == null) {
+        } else if (apiKeyDoc.invalidated == null) {
             listener.onResponse(AuthenticationResult.unsuccessful("api key document is missing invalidated field", null));
-        } else if (invalidated) {
+        } else if (apiKeyDoc.invalidated) {
             listener.onResponse(AuthenticationResult.unsuccessful("api key has been invalidated", null));
         } else {
-            final String apiKeyHash = (String) source.get("api_key_hash");
+            if (apiKeyDoc.hash == null) {
-            if (apiKeyHash == null) {
                 throw new IllegalStateException("api key hash is missing");
             }
 
@@ -495,7 +546,7 @@ public class ApiKeyService {
                             if (result.success) {
                                 if (result.verify(credentials.getKey())) {
                                     // move on
-                                    validateApiKeyExpiration(source, credentials, clock, listener);
+                                    validateApiKeyExpiration(apiKeyDoc, credentials, clock, listener);
                                 } else {
                                     listener.onResponse(AuthenticationResult.unsuccessful("invalid credentials", null));
                                 }
@@ -503,17 +554,17 @@ public class ApiKeyService {
                                 listener.onResponse(AuthenticationResult.unsuccessful("invalid credentials", null));
                             } else {
                                 apiKeyAuthCache.invalidate(credentials.getId(), listenableCacheEntry);
-                                validateApiKeyCredentials(docId, source, credentials, clock, listener);
+                                validateApiKeyCredentials(docId, apiKeyDoc, credentials, clock, listener);
                             }
                         }, listener::onFailure),
                         threadPool.generic(), threadPool.getThreadContext());
                 } else {
-                    verifyKeyAgainstHash(apiKeyHash, credentials, ActionListener.wrap(
+                    verifyKeyAgainstHash(apiKeyDoc.hash, credentials, ActionListener.wrap(
                         verified -> {
                             listenableCacheEntry.onResponse(new CachedApiKeyHashResult(verified, credentials.getKey()));
                             if (verified) {
                                 // move on
-                                validateApiKeyExpiration(source, credentials, clock, listener);
+                                validateApiKeyExpiration(apiKeyDoc, credentials, clock, listener);
                             } else {
                                 listener.onResponse(AuthenticationResult.unsuccessful("invalid credentials", null));
                             }
@@ -521,18 +572,17 @@ public class ApiKeyService {
                     ));
                 }
             } else {
-                verifyKeyAgainstHash(apiKeyHash, credentials, ActionListener.wrap(
+                verifyKeyAgainstHash(apiKeyDoc.hash, credentials, ActionListener.wrap(
                     verified -> {
                         if (verified) {
                             // move on
-                            validateApiKeyExpiration(source, credentials, clock, listener);
+                            validateApiKeyExpiration(apiKeyDoc, credentials, clock, listener);
                         } else {
                             listener.onResponse(AuthenticationResult.unsuccessful("invalid credentials", null));
                         }
                     },
                     listener::onFailure
                 ));
-
             }
         }
     }
@@ -543,23 +593,19 @@ public class ApiKeyService {
     }
 
     // package-private for testing
-    void validateApiKeyExpiration(Map<String, Object> source, ApiKeyCredentials credentials, Clock clock,
+    void validateApiKeyExpiration(ApiKeyDoc apiKeyDoc, ApiKeyCredentials credentials, Clock clock,
                                   ActionListener<AuthenticationResult> listener) {
-        final Long expirationEpochMilli = (Long) source.get("expiration_time");
+        if (apiKeyDoc.expirationTime == -1 || Instant.ofEpochMilli(apiKeyDoc.expirationTime).isAfter(clock.instant())) {
-        if (expirationEpochMilli == null || Instant.ofEpochMilli(expirationEpochMilli).isAfter(clock.instant())) {
+            final String principal = Objects.requireNonNull((String) apiKeyDoc.creator.get("principal"));
-            final Map<String, Object> creator = Objects.requireNonNull((Map<String, Object>) source.get("creator"));
+            Map<String, Object> metadata = (Map<String, Object>) apiKeyDoc.creator.get("metadata");
-            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 User apiKeyUser = new User(principal, Strings.EMPTY_ARRAY, null, null, metadata, true);
             final Map<String, Object> authResultMetadata = new HashMap<>();
-            authResultMetadata.put(API_KEY_CREATOR_REALM_NAME, creator.get("realm"));
+            authResultMetadata.put(API_KEY_CREATOR_REALM_NAME, apiKeyDoc.creator.get("realm"));
-            authResultMetadata.put(API_KEY_CREATOR_REALM_TYPE, creator.get("realm_type"));
+            authResultMetadata.put(API_KEY_CREATOR_REALM_TYPE, apiKeyDoc.creator.get("realm_type"));
-            authResultMetadata.put(API_KEY_ROLE_DESCRIPTORS_KEY, roleDescriptors);
+            authResultMetadata.put(API_KEY_ROLE_DESCRIPTORS_KEY, apiKeyDoc.roleDescriptorsBytes);
-            authResultMetadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, limitedByRoleDescriptors);
+            authResultMetadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, apiKeyDoc.limitedByRoleDescriptorsBytes);
             authResultMetadata.put(API_KEY_ID_KEY, credentials.getId());
-            authResultMetadata.put(API_KEY_NAME_KEY, source.get("name"));
+            authResultMetadata.put(API_KEY_NAME_KEY, apiKeyDoc.name);
             listener.onResponse(AuthenticationResult.success(apiKeyUser, authResultMetadata));
         } else {
             listener.onResponse(AuthenticationResult.unsuccessful("api key is expired", null));
@@ -983,4 +1029,66 @@ public class ApiKeyService {
             return hash != null && cacheHasher.verify(password, hash);
         }
     }
+
+    public static final class ApiKeyDoc {
+
+        static final InstantiatingObjectParser<ApiKeyDoc, Void> PARSER;
+        static {
+            InstantiatingObjectParser.Builder<ApiKeyDoc, Void> builder =
+                InstantiatingObjectParser.builder("api_key_doc", true, ApiKeyDoc.class);
+            builder.declareString(constructorArg(), new ParseField("doc_type"));
+            builder.declareLong(constructorArg(), new ParseField("creation_time"));
+            builder.declareLongOrNull(constructorArg(), -1, new ParseField("expiration_time"));
+            builder.declareBoolean(constructorArg(), new ParseField("api_key_invalidated"));
+            builder.declareString(optionalConstructorArg(), new ParseField("api_key_hash"));
+            builder.declareString(constructorArg(), new ParseField("name"));
+            builder.declareInt(constructorArg(), new ParseField("version"));
+            ObjectParserHelper<ApiKeyDoc, Void> parserHelper = new ObjectParserHelper<>();
+            parserHelper.declareRawObject(builder, optionalConstructorArg(), new ParseField("role_descriptors"));
+            parserHelper.declareRawObject(builder, constructorArg(), new ParseField("limited_by_role_descriptors"));
+            builder.declareObject(constructorArg(), (p, c) -> p.map(), new ParseField("creator"));
+            PARSER = builder.build();
+        }
+
+        final String docType;
+        final long creationTime;
+        final long expirationTime;
+        final Boolean invalidated;
+        @Nullable
+        final String hash;
+        final String name;
+        final int version;
+        @Nullable
+        final BytesReference roleDescriptorsBytes;
+        final BytesReference limitedByRoleDescriptorsBytes;
+        final Map<String, Object> creator;
+
+        public ApiKeyDoc(
+            String docType,
+            long creationTime,
+            long expirationTime,
+            Boolean invalidated,
+            @Nullable String hash,
+            String name,
+            int version,
+            @Nullable BytesReference roleDescriptorsBytes,
+            BytesReference limitedByRoleDescriptorsBytes,
+            Map<String, Object> creator) {
+
+            this.docType = docType;
+            this.creationTime = creationTime;
+            this.expirationTime = expirationTime;
+            this.invalidated = invalidated;
+            this.hash = hash;
+            this.name = name;
+            this.version = version;
+            this.roleDescriptorsBytes = roleDescriptorsBytes;
+            this.limitedByRoleDescriptorsBytes = limitedByRoleDescriptorsBytes;
+            this.creator = creator;
+        }
+
+        static ApiKeyDoc fromXContent(XContentParser parser) {
+            return PARSER.apply(parser, null);
+        }
+    }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/store/CompositeRolesStore.java

@@ -17,6 +17,7 @@ import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.cache.Cache;
 import org.elasticsearch.common.cache.CacheBuilder;
 import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.hash.MessageDigests;
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
@@ -71,6 +72,7 @@ import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.util.set.Sets.newHashSet;
+import static org.elasticsearch.xpack.core.security.authc.Authentication.VERSION_API_KEY_ROLES_AS_BYTES;
 import static org.elasticsearch.xpack.security.support.SecurityIndexManager.isIndexDeleted;
 import static org.elasticsearch.xpack.security.support.SecurityIndexManager.isMoveFromRedToNonRed;
 
@@ -221,20 +223,40 @@ public class CompositeRolesStore {
 
         final Authentication.AuthenticationType authType = authentication.getAuthenticationType();
         if (authType == Authentication.AuthenticationType.API_KEY) {
-            apiKeyService.getRoleForApiKey(authentication, ActionListener.wrap(apiKeyRoleDescriptors -> {
+            if (authentication.getVersion().onOrAfter(VERSION_API_KEY_ROLES_AS_BYTES)) {
-                final List<RoleDescriptor> descriptors = apiKeyRoleDescriptors.getRoleDescriptors();
+                buildAndCacheRoleForApiKey(authentication, false, ActionListener.wrap(
-                if (descriptors == null) {
+                    role -> {
-                    roleActionListener.onFailure(new IllegalStateException("missing role descriptors"));
+                        if (role == Role.EMPTY) {
-                } else if (apiKeyRoleDescriptors.getLimitedByRoleDescriptors() == null) {
+                            buildAndCacheRoleForApiKey(authentication, true, roleActionListener);
-                    buildAndCacheRoleFromDescriptors(descriptors, apiKeyRoleDescriptors.getApiKeyId() + "_role_desc", roleActionListener);
+                        } else {
-                } else {
+                            buildAndCacheRoleForApiKey(authentication, true, ActionListener.wrap(
-                    buildAndCacheRoleFromDescriptors(descriptors, apiKeyRoleDescriptors.getApiKeyId() + "_role_desc",
+                                limitedByRole -> roleActionListener.onResponse(
-                        ActionListener.wrap(role -> buildAndCacheRoleFromDescriptors(apiKeyRoleDescriptors.getLimitedByRoleDescriptors(),
+                                    limitedByRole == Role.EMPTY ? role : LimitedRole.createLimitedRole(role, limitedByRole)),
-                            apiKeyRoleDescriptors.getApiKeyId() + "_limited_role_desc", ActionListener.wrap(
+                                roleActionListener::onFailure
-                                limitedBy -> roleActionListener.onResponse(LimitedRole.createLimitedRole(role, limitedBy)),
+                            ));
-                                roleActionListener::onFailure)), roleActionListener::onFailure));
+                        }
-                }
+                    },
-            }, roleActionListener::onFailure));
+                    roleActionListener::onFailure
+                ));
+            } else {
+                apiKeyService.getRoleForApiKey(authentication, ActionListener.wrap(apiKeyRoleDescriptors -> {
+                    final List<RoleDescriptor> descriptors = apiKeyRoleDescriptors.getRoleDescriptors();
+                    if (descriptors == null) {
+                        roleActionListener.onFailure(new IllegalStateException("missing role descriptors"));
+                    } else if (apiKeyRoleDescriptors.getLimitedByRoleDescriptors() == null) {
+                        buildAndCacheRoleFromDescriptors(descriptors,
+                            apiKeyRoleDescriptors.getApiKeyId() + "_role_desc", roleActionListener);
+                    } else {
+                        buildAndCacheRoleFromDescriptors(descriptors, apiKeyRoleDescriptors.getApiKeyId() + "_role_desc",
+                            ActionListener.wrap(
+                                role -> buildAndCacheRoleFromDescriptors(apiKeyRoleDescriptors.getLimitedByRoleDescriptors(),
+                                    apiKeyRoleDescriptors.getApiKeyId() + "_limited_role_desc", ActionListener.wrap(
+                                        limitedBy -> roleActionListener.onResponse(LimitedRole.createLimitedRole(role, limitedBy)),
+                                        roleActionListener::onFailure)), roleActionListener::onFailure));
+                    }
+                }, roleActionListener::onFailure));
+            }
+
         } else {
             Set<String> roleNames = new HashSet<>(Arrays.asList(user.roles()));
             if (isAnonymousEnabled && anonymousUser.equals(user) == false) {
@@ -295,6 +317,23 @@ public class CompositeRolesStore {
         }, listener::onFailure));
     }
 
+    private void buildAndCacheRoleForApiKey(Authentication authentication, boolean limitedBy, ActionListener<Role> roleActionListener) {
+        final Tuple<String, BytesReference> apiKeyIdAndBytes = apiKeyService.getApiKeyIdAndRoleBytes(authentication, limitedBy);
+        final String roleDescriptorsHash = MessageDigests.toHexString(
+            MessageDigests.sha256().digest(BytesReference.toBytes(apiKeyIdAndBytes.v2())));
+        final RoleKey roleKey = new RoleKey(org.elasticsearch.common.collect.Set.of("apikey:" + roleDescriptorsHash),
+            limitedBy ? "apikey_limited_role" : "apikey_role");
+        final Role existing = roleCache.get(roleKey);
+        if (existing == null) {
+            final long invalidationCounter = numInvalidation.get();
+            final List<RoleDescriptor> roleDescriptors = apiKeyService.parseRoleDescriptors(apiKeyIdAndBytes.v1(), apiKeyIdAndBytes.v2());
+            buildThenMaybeCacheRole(roleKey, roleDescriptors, Collections.emptySet(),
+                true, invalidationCounter, roleActionListener);
+        } else {
+            roleActionListener.onResponse(existing);
+        }
+    }
+
     public void getRoleDescriptors(Set<String> roleNames, ActionListener<Set<RoleDescriptor>> listener) {
         roleDescriptors(roleNames, ActionListener.wrap(rolesRetrievalResult -> {
             if (rolesRetrievalResult.isSuccess()) {

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

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.security;
 
 import org.elasticsearch.Version;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext;
@@ -23,8 +24,12 @@ import org.junit.Before;
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.elasticsearch.xpack.core.security.authc.Authentication.VERSION_API_KEY_ROLES_AS_BYTES;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
 import static org.hamcrest.Matchers.instanceOf;
 
 public class SecurityContextTests extends ESTestCase {
@@ -130,4 +135,44 @@ public class SecurityContextTests extends ESTestCase {
         originalContext.restore();
         assertEquals(original, securityContext.getAuthentication());
     }
+
+    public void testExecuteAfterRewritingAuthenticationShouldRewriteApiKeyMetadataForBwc() throws IOException {
+        User user = new User("test", null, new User("authUser"));
+        RealmRef authBy = new RealmRef("_es_api_key", "_es_api_key", "node1");
+        final Map<String, Object> metadata = org.elasticsearch.common.collect.Map.of(
+            API_KEY_ROLE_DESCRIPTORS_KEY, new BytesArray("{\"a role\": {\"cluster\": [\"all\"]}}"),
+            API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, new BytesArray("{\"limitedBy role\": {\"cluster\": [\"all\"]}}")
+        );
+        final Authentication original = new Authentication(user, authBy, authBy, VERSION_API_KEY_ROLES_AS_BYTES,
+            AuthenticationType.API_KEY, metadata);
+        original.writeToContext(threadContext);
+
+        securityContext.executeAfterRewritingAuthentication(originalCtx -> {
+            Authentication authentication = securityContext.getAuthentication();
+            assertEquals(org.elasticsearch.common.collect.Map.of("a role",
+                org.elasticsearch.common.collect.Map.of("cluster", org.elasticsearch.common.collect.List.of("all"))),
+                authentication.getMetadata().get(API_KEY_ROLE_DESCRIPTORS_KEY));
+            assertEquals(org.elasticsearch.common.collect.Map.of("limitedBy role",
+                org.elasticsearch.common.collect.Map.of("cluster", org.elasticsearch.common.collect.List.of("all"))),
+                authentication.getMetadata().get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY));
+        }, Version.V_7_8_0);
+    }
+
+    public void testExecuteAfterRewritingAuthenticationShouldNotRewriteApiKeyMetadataForOldAuthenticationObject() throws IOException {
+        User user = new User("test", null, new User("authUser"));
+        RealmRef authBy = new RealmRef("_es_api_key", "_es_api_key", "node1");
+        final Map<String, Object> metadata = org.elasticsearch.common.collect.Map.of(
+            API_KEY_ROLE_DESCRIPTORS_KEY, org.elasticsearch.common.collect.Map.of(
+                "a role", org.elasticsearch.common.collect.Map.of("cluster", org.elasticsearch.common.collect.List.of("all"))),
+            API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, org.elasticsearch.common.collect.Map.of(
+                "limitedBy role", org.elasticsearch.common.collect.Map.of("cluster", org.elasticsearch.common.collect.List.of("all")))
+        );
+        final Authentication original = new Authentication(user, authBy, authBy, Version.V_7_8_0, AuthenticationType.API_KEY, metadata);
+        original.writeToContext(threadContext);
+
+        securityContext.executeAfterRewritingAuthentication(originalCtx -> {
+            Authentication authentication = securityContext.getAuthentication();
+            assertSame(metadata, authentication.getMetadata());
+        }, randomFrom(VERSION_API_KEY_ROLES_AS_BYTES, Version.V_7_8_0));
+    }
 }

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

@@ -15,13 +15,17 @@ import org.elasticsearch.action.index.IndexAction;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
@@ -31,20 +35,24 @@ import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.license.XPackLicenseState.Feature;
 import org.elasticsearch.test.ClusterServiceUtils;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.threadpool.FixedExecutorBuilder;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.XPackSettings;
+import org.elasticsearch.xpack.core.security.SecurityContext;
 import org.elasticsearch.xpack.core.security.action.CreateApiKeyRequest;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.Authentication.AuthenticationType;
 import org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.support.AuthenticationContextSerializer;
 import org.elasticsearch.xpack.core.security.authc.support.Hasher;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.privilege.ApplicationPrivilege;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.security.authc.ApiKeyService.ApiKeyCredentials;
+import org.elasticsearch.xpack.security.authc.ApiKeyService.ApiKeyDoc;
 import org.elasticsearch.xpack.security.authc.ApiKeyService.ApiKeyRoleDescriptors;
 import org.elasticsearch.xpack.security.authc.ApiKeyService.CachedApiKeyHashResult;
 import org.elasticsearch.xpack.security.authz.store.NativePrivilegeStore;
@@ -69,12 +77,16 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.test.TestMatchers.throwableWithMessage;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
 import static org.elasticsearch.xpack.core.security.authz.store.ReservedRolesStore.SUPERUSER_ROLE_DESCRIPTOR;
 import static org.elasticsearch.xpack.security.Security.SECURITY_CRYPTO_THREAD_POOL_NAME;
 import static org.hamcrest.Matchers.contains;
@@ -354,68 +366,49 @@ public class ApiKeyServiceTests extends ESTestCase {
         Hasher hasher = randomFrom(Hasher.PBKDF2, Hasher.BCRYPT4, Hasher.BCRYPT);
         final char[] hash = hasher.hash(new SecureString(apiKey.toCharArray()));
 
-        Map<String, Object> sourceMap = new HashMap<>();
+        ApiKeyDoc apiKeyDoc = buildApiKeyDoc(hash, -1, false);
-        sourceMap.put("doc_type", "api_key");
-        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("realm", "realm1");
-        creatorMap.put("realm_type", "realm_type1");
-        creatorMap.put("metadata", Collections.emptyMap());
-        sourceMap.put("creator", creatorMap);
-        sourceMap.put("api_key_invalidated", false);
 
         ApiKeyService service = createApiKeyService(Settings.EMPTY);
         ApiKeyService.ApiKeyCredentials creds =
             new ApiKeyService.ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(apiKey.toCharArray()));
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, creds, Clock.systemUTC(), future);
         AuthenticationResult result = future.get();
         assertNotNull(result);
         assertTrue(result.isAuthenticated());
         assertThat(result.getUser().principal(), is("test_user"));
         assertThat(result.getUser().roles(), is(emptyArray()));
         assertThat(result.getUser().metadata(), is(Collections.emptyMap()));
-        assertThat(result.getMetadata().get(ApiKeyService.API_KEY_ROLE_DESCRIPTORS_KEY), equalTo(sourceMap.get("role_descriptors")));
+        assertThat(result.getMetadata().get(API_KEY_ROLE_DESCRIPTORS_KEY), equalTo(apiKeyDoc.roleDescriptorsBytes));
-        assertThat(result.getMetadata().get(ApiKeyService.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY),
+        assertThat(result.getMetadata().get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY),
-            equalTo(sourceMap.get("limited_by_role_descriptors")));
+            equalTo(apiKeyDoc.limitedByRoleDescriptorsBytes));
         assertThat(result.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_NAME), is("realm1"));
 
-        sourceMap.put("expiration_time", Clock.systemUTC().instant().plus(1L, ChronoUnit.HOURS).toEpochMilli());
+        apiKeyDoc = buildApiKeyDoc(hash, Clock.systemUTC().instant().plus(1L, ChronoUnit.HOURS).toEpochMilli(), false);
         future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, creds, Clock.systemUTC(), future);
         result = future.get();
         assertNotNull(result);
         assertTrue(result.isAuthenticated());
         assertThat(result.getUser().principal(), is("test_user"));
         assertThat(result.getUser().roles(), is(emptyArray()));
         assertThat(result.getUser().metadata(), is(Collections.emptyMap()));
-        assertThat(result.getMetadata().get(ApiKeyService.API_KEY_ROLE_DESCRIPTORS_KEY), equalTo(sourceMap.get("role_descriptors")));
+        assertThat(result.getMetadata().get(API_KEY_ROLE_DESCRIPTORS_KEY), equalTo(apiKeyDoc.roleDescriptorsBytes));
-        assertThat(result.getMetadata().get(ApiKeyService.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY),
+        assertThat(result.getMetadata().get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY),
-            equalTo(sourceMap.get("limited_by_role_descriptors")));
+            equalTo(apiKeyDoc.limitedByRoleDescriptorsBytes));
         assertThat(result.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_NAME), is("realm1"));
 
-        sourceMap.put("expiration_time", Clock.systemUTC().instant().minus(1L, ChronoUnit.HOURS).toEpochMilli());
+        apiKeyDoc = buildApiKeyDoc(hash, Clock.systemUTC().instant().minus(1L, ChronoUnit.HOURS).toEpochMilli(), false);
         future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, creds, Clock.systemUTC(), future);
         result = future.get();
         assertNotNull(result);
         assertFalse(result.isAuthenticated());
 
-        sourceMap.remove("expiration_time");
+        apiKeyDoc = buildApiKeyDoc(hash, -1, true);
         creds = new ApiKeyService.ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(randomAlphaOfLength(15).toCharArray()));
         future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, creds, Clock.systemUTC(), future);
-        result = future.get();
-        assertNotNull(result);
-        assertFalse(result.isAuthenticated());
-
-        sourceMap.put("api_key_invalidated", true);
-        creds = new ApiKeyService.ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(randomAlphaOfLength(15).toCharArray()));
-        future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
         result = future.get();
         assertNotNull(result);
         assertFalse(result.isAuthenticated());
@@ -432,13 +425,14 @@ public class ApiKeyServiceTests extends ESTestCase {
         }
         Map<String, Object> authMetadata = new HashMap<>();
         authMetadata.put(ApiKeyService.API_KEY_ID_KEY, randomAlphaOfLength(12));
-        authMetadata.put(ApiKeyService.API_KEY_ROLE_DESCRIPTORS_KEY,
+        authMetadata.put(API_KEY_ROLE_DESCRIPTORS_KEY,
             Collections.singletonMap(SUPERUSER_ROLE_DESCRIPTOR.getName(), superUserRdMap));
-        authMetadata.put(ApiKeyService.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
+        authMetadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
             Collections.singletonMap(SUPERUSER_ROLE_DESCRIPTOR.getName(), superUserRdMap));
 
         final Authentication authentication = new Authentication(new User("joe"), new RealmRef("apikey", "apikey", "node"), null,
-            Version.CURRENT, AuthenticationType.API_KEY, authMetadata);
+            VersionUtils.randomVersionBetween(random(), Version.V_7_0_0, Version.V_7_8_1),
+            AuthenticationType.API_KEY, authMetadata);
         ApiKeyService service = createApiKeyService(Settings.EMPTY);
 
         PlainActionFuture<ApiKeyRoleDescriptors> roleFuture = new PlainActionFuture<>();
@@ -461,7 +455,7 @@ public class ApiKeyServiceTests extends ESTestCase {
             roleARDMap = XContentHelper.convertToMap(XContentType.JSON.xContent(),
                 BytesReference.bytes(roleARoleDescriptor.toXContent(builder, ToXContent.EMPTY_PARAMS, true)).streamInput(), false);
         }
-        authMetadata.put(ApiKeyService.API_KEY_ROLE_DESCRIPTORS_KEY,
+        authMetadata.put(API_KEY_ROLE_DESCRIPTORS_KEY,
             (emptyApiKeyRoleDescriptor) ? randomFrom(Arrays.asList(null, Collections.emptyMap()))
                 : Collections.singletonMap("a role", roleARDMap));
 
@@ -477,10 +471,11 @@ public class ApiKeyServiceTests extends ESTestCase {
                     .streamInput(),
                 false);
         }
-        authMetadata.put(ApiKeyService.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, Collections.singletonMap("limited role", limitedRdMap));
+        authMetadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, Collections.singletonMap("limited role", limitedRdMap));
 
         final Authentication authentication = new Authentication(new User("joe"), new RealmRef("apikey", "apikey", "node"), null,
-            Version.CURRENT, AuthenticationType.API_KEY, authMetadata);
+            VersionUtils.randomVersionBetween(random(), Version.V_7_0_0, Version.V_7_8_1),
+            AuthenticationType.API_KEY, authMetadata);
 
         final NativePrivilegeStore privilegesStore = mock(NativePrivilegeStore.class);
         doAnswer(i -> {
@@ -509,6 +504,54 @@ public class ApiKeyServiceTests extends ESTestCase {
         }
     }
 
+    public void testGetApiKeyIdAndRoleBytes() {
+        Map<String, Object> authMetadata = new HashMap<>();
+        final String apiKeyId = randomAlphaOfLength(12);
+        authMetadata.put(ApiKeyService.API_KEY_ID_KEY, apiKeyId);
+        final BytesReference roleBytes = new BytesArray("{\"a role\": {\"cluster\": [\"all\"]}}");
+        final BytesReference limitedByRoleBytes = new BytesArray("{\"limitedBy role\": {\"cluster\": [\"all\"]}}");
+        authMetadata.put(API_KEY_ROLE_DESCRIPTORS_KEY, roleBytes);
+        authMetadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, limitedByRoleBytes);
+
+        final Authentication authentication = new Authentication(new User("joe"), new RealmRef("apikey", "apikey", "node"), null,
+            Version.CURRENT, AuthenticationType.API_KEY, authMetadata);
+        ApiKeyService service = createApiKeyService(Settings.EMPTY);
+
+        Tuple<String, BytesReference> apiKeyIdAndRoleBytes = service.getApiKeyIdAndRoleBytes(authentication, false);
+        assertEquals(apiKeyId, apiKeyIdAndRoleBytes.v1());
+        assertEquals(roleBytes, apiKeyIdAndRoleBytes.v2());
+        apiKeyIdAndRoleBytes = service.getApiKeyIdAndRoleBytes(authentication, true);
+        assertEquals(apiKeyId, apiKeyIdAndRoleBytes.v1());
+        assertEquals(limitedByRoleBytes, apiKeyIdAndRoleBytes.v2());
+    }
+
+    public void testParseRoleDescriptors() {
+        ApiKeyService service = createApiKeyService(Settings.EMPTY);
+        final String apiKeyId = randomAlphaOfLength(12);
+        List<RoleDescriptor> roleDescriptors = service.parseRoleDescriptors(apiKeyId, null);
+        assertTrue(roleDescriptors.isEmpty());
+
+        BytesReference roleBytes = new BytesArray("{\"a role\": {\"cluster\": [\"all\"]}}");
+        roleDescriptors = service.parseRoleDescriptors(apiKeyId, roleBytes);
+        assertEquals(1, roleDescriptors.size());
+        assertEquals("a role", roleDescriptors.get(0).getName());
+        assertArrayEquals(new String[] { "all" }, roleDescriptors.get(0).getClusterPrivileges());
+        assertEquals(0, roleDescriptors.get(0).getIndicesPrivileges().length);
+        assertEquals(0, roleDescriptors.get(0).getApplicationPrivileges().length);
+
+        roleBytes = new BytesArray(
+            "{\"reporting_user\":{\"cluster\":[],\"indices\":[],\"applications\":[],\"run_as\":[],\"metadata\":{\"_reserved\":true}," +
+                "\"transient_metadata\":{\"enabled\":true}},\"superuser\":{\"cluster\":[\"all\"],\"indices\":[{\"names\":[\"*\"]," +
+                "\"privileges\":[\"all\"],\"allow_restricted_indices\":true}],\"applications\":[{\"application\":\"*\"," +
+                "\"privileges\":[\"*\"],\"resources\":[\"*\"]}],\"run_as\":[\"*\"],\"metadata\":{\"_reserved\":true}," +
+                "\"transient_metadata\":{}}}\n");
+        roleDescriptors = service.parseRoleDescriptors(apiKeyId, roleBytes);
+        assertEquals(2, roleDescriptors.size());
+        assertEquals(
+            org.elasticsearch.common.collect.Set.of("reporting_user", "superuser"),
+            roleDescriptors.stream().map(RoleDescriptor::getName).collect(Collectors.toSet()));
+    }
+
     public void testApiKeyServiceDisabled() throws Exception {
         final Settings settings = Settings.builder().put(XPackSettings.API_KEY_SERVICE_ENABLED_SETTING.getKey(), false).build();
         final ApiKeyService service = createApiKeyService(settings);
@@ -528,12 +571,12 @@ public class ApiKeyServiceTests extends ESTestCase {
         Hasher hasher = randomFrom(Hasher.PBKDF2, Hasher.BCRYPT4, Hasher.BCRYPT);
         final char[] hash = hasher.hash(new SecureString(apiKey.toCharArray()));
 
-        Map<String, Object> sourceMap = buildApiKeySourceDoc(hash);
+        ApiKeyDoc apiKeyDoc = buildApiKeyDoc(hash, -1, false);
 
         ApiKeyService service = createApiKeyService(Settings.EMPTY);
         ApiKeyCredentials creds = new ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(apiKey.toCharArray()));
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, creds, Clock.systemUTC(), future);
         AuthenticationResult result = future.actionGet();
         assertThat(result.isAuthenticated(), is(true));
         CachedApiKeyHashResult cachedApiKeyHashResult = service.getFromCache(creds.getId());
@@ -542,17 +585,17 @@ public class ApiKeyServiceTests extends ESTestCase {
 
         creds = new ApiKeyCredentials(creds.getId(), new SecureString("foobar".toCharArray()));
         future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, 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()))));
+        apiKeyDoc = buildApiKeyDoc(hasher.hash(new SecureString("foobar".toCharArray())), -1, false);
         creds = new ApiKeyCredentials(randomAlphaOfLength(12), new SecureString("foobar1".toCharArray()));
         future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, creds, Clock.systemUTC(), future);
         result = future.actionGet();
         assertThat(result.isAuthenticated(), is(false));
         cachedApiKeyHashResult = service.getFromCache(creds.getId());
@@ -561,7 +604,7 @@ public class ApiKeyServiceTests extends ESTestCase {
 
         creds = new ApiKeyCredentials(creds.getId(), new SecureString("foobar2".toCharArray()));
         future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, creds, Clock.systemUTC(), future);
         result = future.actionGet();
         assertThat(result.isAuthenticated(), is(false));
         assertThat(service.getFromCache(creds.getId()), not(sameInstance(cachedApiKeyHashResult)));
@@ -569,7 +612,7 @@ public class ApiKeyServiceTests extends ESTestCase {
 
         creds = new ApiKeyCredentials(creds.getId(), new SecureString("foobar".toCharArray()));
         future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, creds, Clock.systemUTC(), future);
         result = future.actionGet();
         assertThat(result.isAuthenticated(), is(true));
         assertThat(service.getFromCache(creds.getId()), not(sameInstance(cachedApiKeyHashResult)));
@@ -642,12 +685,12 @@ public class ApiKeyServiceTests extends ESTestCase {
             .put(ApiKeyService.CACHE_TTL_SETTING.getKey(), "0s")
             .build();
 
-        Map<String, Object> sourceMap = buildApiKeySourceDoc(hash);
+        ApiKeyDoc apiKeyDoc = buildApiKeyDoc(hash, -1, false);
 
         ApiKeyService service = createApiKeyService(settings);
         ApiKeyCredentials creds = new ApiKeyCredentials(randomAlphaOfLength(12), new SecureString(apiKey.toCharArray()));
         PlainActionFuture<AuthenticationResult> future = new PlainActionFuture<>();
-        service.validateApiKeyCredentials(creds.getId(), sourceMap, creds, Clock.systemUTC(), future);
+        service.validateApiKeyCredentials(creds.getId(), apiKeyDoc, creds, Clock.systemUTC(), future);
         AuthenticationResult result = future.actionGet();
         assertThat(result.isAuthenticated(), is(true));
         CachedApiKeyHashResult cachedApiKeyHashResult = service.getFromCache(creds.getId());
@@ -755,27 +798,79 @@ public class ApiKeyServiceTests extends ESTestCase {
         assertEquals(AuthenticationResult.Status.SUCCESS, authenticationResult3.getStatus());
     }
 
+    public void testApiKeyDocDeserialization() throws IOException {
+        final String apiKeyDocumentSource =
+            "{\"doc_type\":\"api_key\",\"creation_time\":1591919944598,\"expiration_time\":null,\"api_key_invalidated\":false," +
+                "\"api_key_hash\":\"{PBKDF2}10000$abc\",\"role_descriptors\":{\"a\":{\"cluster\":[\"all\"]}}," +
+                "\"limited_by_role_descriptors\":{\"limited_by\":{\"cluster\":[\"all\"]," +
+                "\"metadata\":{\"_reserved\":true},\"type\":\"role\"}}," +
+                "\"name\":\"key-1\",\"version\":7000099," +
+                "\"creator\":{\"principal\":\"admin\",\"metadata\":{\"foo\":\"bar\"},\"realm\":\"file1\",\"realm_type\":\"file\"}}\n";
+        final ApiKeyDoc apiKeyDoc = ApiKeyDoc.fromXContent(XContentHelper.createParser(NamedXContentRegistry.EMPTY,
+            LoggingDeprecationHandler.INSTANCE,
+            new BytesArray(apiKeyDocumentSource),
+            XContentType.JSON));
+        assertEquals("api_key", apiKeyDoc.docType);
+        assertEquals(1591919944598L, apiKeyDoc.creationTime);
+        assertEquals(-1L, apiKeyDoc.expirationTime);
+        assertFalse(apiKeyDoc.invalidated);
+        assertEquals("{PBKDF2}10000$abc", apiKeyDoc.hash);
+        assertEquals("key-1", apiKeyDoc.name);
+        assertEquals(7000099, apiKeyDoc.version);
+        assertEquals(new BytesArray("{\"a\":{\"cluster\":[\"all\"]}}"), apiKeyDoc.roleDescriptorsBytes);
+        assertEquals(new BytesArray("{\"limited_by\":{\"cluster\":[\"all\"],\"metadata\":{\"_reserved\":true},\"type\":\"role\"}}"),
+            apiKeyDoc.limitedByRoleDescriptorsBytes);
+
+        final Map<String, Object> creator = apiKeyDoc.creator;
+        assertEquals("admin", creator.get("principal"));
+        assertEquals("file1", creator.get("realm"));
+        assertEquals("file", creator.get("realm_type"));
+        assertEquals("bar", ((Map<String, Object>)creator.get("metadata")).get("foo"));
+    }
+
     public static class Utils {
+        private static final AuthenticationContextSerializer authenticationContextSerializer = new AuthenticationContextSerializer();
 
         public static Authentication createApiKeyAuthentication(ApiKeyService apiKeyService,
                                                                 Authentication authentication,
                                                                 Set<RoleDescriptor> userRoles,
-                                                                List<RoleDescriptor> keyRoles) throws Exception {
+                                                                List<RoleDescriptor> keyRoles,
+                                                                Version version) throws Exception {
             XContentBuilder keyDocSource = apiKeyService.newDocument(new SecureString("secret".toCharArray()), "test", authentication,
                     userRoles, Instant.now(), Instant.now().plus(Duration.ofSeconds(3600)), keyRoles, Version.CURRENT);
-            Map<String, Object> keyDocMap = XContentHelper.convertToMap(BytesReference.bytes(keyDocSource), true, XContentType.JSON).v2();
+            final ApiKeyDoc apiKeyDoc = ApiKeyDoc.fromXContent(
+                XContentHelper.createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE,
+                    BytesReference.bytes(keyDocSource), XContentType.JSON));
             PlainActionFuture<AuthenticationResult> authenticationResultFuture = PlainActionFuture.newFuture();
-            apiKeyService.validateApiKeyExpiration(keyDocMap, new ApiKeyService.ApiKeyCredentials("id",
+            apiKeyService.validateApiKeyExpiration(apiKeyDoc, new ApiKeyService.ApiKeyCredentials("id",
                             new SecureString("pass".toCharArray())),
                     Clock.systemUTC(), authenticationResultFuture);
-            return apiKeyService.createApiKeyAuthentication(authenticationResultFuture.get(), "node01");
+
+            final TestThreadPool threadPool = new TestThreadPool("utils");
+            try {
+                final ThreadContext threadContext = threadPool.getThreadContext();
+                final SecurityContext securityContext = new SecurityContext(Settings.EMPTY, threadContext);
+                authenticationContextSerializer.writeToContext(
+                    apiKeyService.createApiKeyAuthentication(authenticationResultFuture.get(), "node01"), threadContext);
+                final CompletableFuture<Authentication> authFuture = new CompletableFuture<>();
+                securityContext.executeAfterRewritingAuthentication((c) -> {
+                    try {
+                        authFuture.complete(authenticationContextSerializer.readFromContext(threadContext));
+                    } catch (IOException e) {
+                        throw new RuntimeException(e);
+                    }
+                }, version);
+                return authFuture.get();
+            } finally {
+                terminate(threadPool);
+            }
         }
 
         public static Authentication createApiKeyAuthentication(ApiKeyService apiKeyService,
                                                                 Authentication authentication) throws Exception {
             return createApiKeyAuthentication(apiKeyService, authentication,
                     Collections.singleton(new RoleDescriptor("user_role_" + randomAlphaOfLength(4), new String[]{"manage"}, null, null)),
-                    null);
+                    null, Version.CURRENT);
         }
     }
 
@@ -791,7 +886,11 @@ public class ApiKeyServiceTests extends ESTestCase {
     private Map<String, Object> buildApiKeySourceDoc(char[] hash) {
         Map<String, Object> sourceMap = new HashMap<>();
         sourceMap.put("doc_type", "api_key");
+        sourceMap.put("creation_time", Clock.systemUTC().instant().toEpochMilli());
+        sourceMap.put("expiration_time", -1);
         sourceMap.put("api_key_hash", new String(hash));
+        sourceMap.put("name", randomAlphaOfLength(12));
+        sourceMap.put("version", 0);
         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<>();
@@ -815,4 +914,20 @@ public class ApiKeyServiceTests extends ESTestCase {
         }
     }
 
+    private ApiKeyDoc buildApiKeyDoc(char[] hash, long expirationTime, boolean invalidated) {
+        return new ApiKeyDoc(
+            "api_key",
+            Clock.systemUTC().instant().toEpochMilli(),
+            expirationTime,
+            invalidated,
+            new String(hash),
+            randomAlphaOfLength(12),
+            0,
+            new BytesArray("{\"a role\": {\"cluster\": [\"all\"]}}"),
+            new BytesArray("{\"limited role\": {\"cluster\": [\"all\"]}}"),
+            org.elasticsearch.common.collect.Map.of(
+                "principal", "test_user", "realm", "realm1", "realm_type", "realm_type1", "metadata",
+                org.elasticsearch.common.collect.Map.of())
+        );
+    }
 }

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

@@ -1411,10 +1411,14 @@ public class AuthenticationServiceTests extends ESTestCase {
                 final Map<String, Object> source = new HashMap<>();
                 source.put("doc_type", "api_key");
                 source.put("creation_time", Instant.now().minus(5, ChronoUnit.MINUTES).toEpochMilli());
+                source.put("expiration_time", null);
                 source.put("api_key_invalidated", false);
                 source.put("api_key_hash", new String(Hasher.BCRYPT4.hash(new SecureString(key.toCharArray()))));
                 source.put("role_descriptors", Collections.singletonMap("api key role", Collections.singletonMap("cluster", "all")));
+                source.put("limited_by_role_descriptors",
+                    Collections.singletonMap("limited api key role", Collections.singletonMap("cluster", "all")));
                 source.put("name", "my api key for testApiKeyAuth");
+                source.put("version", 0);
                 Map<String, Object> creatorMap = new HashMap<>();
                 creatorMap.put("principal", "johndoe");
                 creatorMap.put("metadata", Collections.emptyMap());

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/store/CompositeRolesStoreTests.java

@@ -18,6 +18,7 @@ import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
@@ -32,6 +33,7 @@ import org.elasticsearch.license.TestUtils.UpdatableLicenseState;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.license.XPackLicenseState.Feature;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.transport.TransportRequest.Empty;
@@ -90,6 +92,9 @@ import java.util.function.Predicate;
 
 import static org.elasticsearch.mock.orig.Mockito.times;
 import static org.elasticsearch.mock.orig.Mockito.verifyNoMoreInteractions;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
+import static org.elasticsearch.xpack.security.authc.ApiKeyService.API_KEY_ID_KEY;
 import static org.elasticsearch.xpack.security.authc.ApiKeyServiceTests.Utils.createApiKeyAuthentication;
 import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -100,12 +105,14 @@ import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyBoolean;
 import static org.mockito.Matchers.anySetOf;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Matchers.isA;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -1026,9 +1033,9 @@ public class CompositeRolesStoreTests extends ESTestCase {
         }).when(nativeRolesStore).getRoleDescriptors(isA(Set.class), any(ActionListener.class));
         final ReservedRolesStore reservedRolesStore = spy(new ReservedRolesStore());
         ThreadContext threadContext = new ThreadContext(SECURITY_ENABLED_SETTINGS);
-        ApiKeyService apiKeyService = new ApiKeyService(SECURITY_ENABLED_SETTINGS, Clock.systemUTC(), mock(Client.class),
+        ApiKeyService apiKeyService = spy(new ApiKeyService(SECURITY_ENABLED_SETTINGS, Clock.systemUTC(), mock(Client.class),
                 new XPackLicenseState(SECURITY_ENABLED_SETTINGS), mock(SecurityIndexManager.class), mock(ClusterService.class),
-                mock(ThreadPool.class));
+                mock(ThreadPool.class)));
         NativePrivilegeStore nativePrivStore = mock(NativePrivilegeStore.class);
         doAnswer(invocationOnMock -> {
             ActionListener<Collection<ApplicationPrivilegeDescriptor>> listener =
@@ -1045,14 +1052,22 @@ public class CompositeRolesStoreTests extends ESTestCase {
                 new XPackLicenseState(SECURITY_ENABLED_SETTINGS), cache, apiKeyService, documentSubsetBitsetCache,
                 rds -> effectiveRoleDescriptors.set(rds));
         AuditUtil.getOrGenerateRequestId(threadContext);
-
+        final Version version = randomFrom(Version.CURRENT, VersionUtils.randomVersionBetween(random(), Version.V_7_0_0, Version.V_7_8_1));
         final Authentication authentication = createApiKeyAuthentication(apiKeyService, createAuthentication(),
-                Collections.singleton(new RoleDescriptor("user_role_" + randomAlphaOfLength(4), new String[]{"manage"}, null, null)), null);
+            Collections.singleton(new RoleDescriptor("user_role_" + randomAlphaOfLength(4), new String[]{"manage"}, null, null)),
+            null,
+            version);
 
         PlainActionFuture<Role> roleFuture = new PlainActionFuture<>();
         compositeRolesStore.getRoles(authentication.getUser(), authentication, roleFuture);
         Role role = roleFuture.actionGet();
         assertThat(effectiveRoleDescriptors.get(), is(nullValue()));
+
+        if (version == Version.CURRENT) {
+            verify(apiKeyService, times(2)).getApiKeyIdAndRoleBytes(eq(authentication), anyBoolean());
+        } else {
+            verify(apiKeyService).getRoleForApiKey(eq(authentication), any(ActionListener.class));
+        }
         assertThat(role.names().length, is(1));
         assertThat(role.names()[0], containsString("user_role_"));
     }
@@ -1071,9 +1086,9 @@ public class CompositeRolesStoreTests extends ESTestCase {
         final ReservedRolesStore reservedRolesStore = spy(new ReservedRolesStore());
         ThreadContext threadContext = new ThreadContext(SECURITY_ENABLED_SETTINGS);
 
-        ApiKeyService apiKeyService = new ApiKeyService(SECURITY_ENABLED_SETTINGS, Clock.systemUTC(), mock(Client.class),
+        ApiKeyService apiKeyService = spy(new ApiKeyService(SECURITY_ENABLED_SETTINGS, Clock.systemUTC(), mock(Client.class),
                 new XPackLicenseState(SECURITY_ENABLED_SETTINGS), mock(SecurityIndexManager.class), mock(ClusterService.class),
-                mock(ThreadPool.class));
+                mock(ThreadPool.class)));
         NativePrivilegeStore nativePrivStore = mock(NativePrivilegeStore.class);
         doAnswer(invocationOnMock -> {
             ActionListener<Collection<ApplicationPrivilegeDescriptor>> listener =
@@ -1090,16 +1105,23 @@ public class CompositeRolesStoreTests extends ESTestCase {
                 new XPackLicenseState(SECURITY_ENABLED_SETTINGS), cache, apiKeyService, documentSubsetBitsetCache,
                 rds -> effectiveRoleDescriptors.set(rds));
         AuditUtil.getOrGenerateRequestId(threadContext);
-
+        final Version version = randomFrom(Version.CURRENT, VersionUtils.randomVersionBetween(random(), Version.V_7_0_0, Version.V_7_8_1));
         final Authentication authentication = createApiKeyAuthentication(apiKeyService, createAuthentication(),
-                Collections.singleton(new RoleDescriptor("user_role_" + randomAlphaOfLength(4), new String[]{"manage"}, null, null)),
+            Collections.singleton(new RoleDescriptor("user_role_" + randomAlphaOfLength(4), new String[]{"manage"}, null, null)),
-                Collections.singletonList(new RoleDescriptor("key_role_" + randomAlphaOfLength(8), new String[]{"monitor"}, null, null)));
+            Collections.singletonList(new RoleDescriptor("key_role_" + randomAlphaOfLength(8), new String[]{"monitor"}, null, null)),
+            version);
 
         PlainActionFuture<Role> roleFuture = new PlainActionFuture<>();
         compositeRolesStore.getRoles(authentication.getUser(), authentication, roleFuture);
         Role role = roleFuture.actionGet();
         assertThat(role.checkClusterAction("cluster:admin/foo", Empty.INSTANCE, mock(Authentication.class)), is(false));
         assertThat(effectiveRoleDescriptors.get(), is(nullValue()));
+        if (version == Version.CURRENT) {
+            verify(apiKeyService).getApiKeyIdAndRoleBytes(eq(authentication), eq(false));
+            verify(apiKeyService).getApiKeyIdAndRoleBytes(eq(authentication), eq(true));
+        } else {
+            verify(apiKeyService).getRoleForApiKey(eq(authentication), any(ActionListener.class));
+        }
         assertThat(role.names().length, is(1));
         assertThat(role.names()[0], containsString("user_role_"));
     }
@@ -1184,6 +1206,108 @@ public class CompositeRolesStoreTests extends ESTestCase {
         );
     }
 
+    public void testCacheEntryIsReusedForIdenticalApiKeyRoles() {
+        final FileRolesStore fileRolesStore = mock(FileRolesStore.class);
+        doCallRealMethod().when(fileRolesStore).accept(any(Set.class), any(ActionListener.class));
+        final NativeRolesStore nativeRolesStore = mock(NativeRolesStore.class);
+        doCallRealMethod().when(nativeRolesStore).accept(any(Set.class), any(ActionListener.class));
+        when(fileRolesStore.roleDescriptors(anySetOf(String.class))).thenReturn(Collections.emptySet());
+        doAnswer((invocationOnMock) -> {
+            ActionListener<RoleRetrievalResult> callback = (ActionListener<RoleRetrievalResult>) invocationOnMock.getArguments()[1];
+            callback.onResponse(RoleRetrievalResult.failure(new RuntimeException("intentionally failed!")));
+            return null;
+        }).when(nativeRolesStore).getRoleDescriptors(isA(Set.class), any(ActionListener.class));
+        final ReservedRolesStore reservedRolesStore = spy(new ReservedRolesStore());
+        ThreadContext threadContext = new ThreadContext(SECURITY_ENABLED_SETTINGS);
+        ApiKeyService apiKeyService = mock(ApiKeyService.class);
+        NativePrivilegeStore nativePrivStore = mock(NativePrivilegeStore.class);
+        doAnswer(invocationOnMock -> {
+            ActionListener<Collection<ApplicationPrivilegeDescriptor>> listener =
+                (ActionListener<Collection<ApplicationPrivilegeDescriptor>>) invocationOnMock.getArguments()[2];
+            listener.onResponse(Collections.emptyList());
+            return Void.TYPE;
+        }).when(nativePrivStore).getPrivileges(any(Collection.class), any(Collection.class), any(ActionListener.class));
+
+        final DocumentSubsetBitsetCache documentSubsetBitsetCache = buildBitsetCache();
+        final AtomicReference<Collection<RoleDescriptor>> effectiveRoleDescriptors = new AtomicReference<Collection<RoleDescriptor>>();
+        final CompositeRolesStore compositeRolesStore = new CompositeRolesStore(SECURITY_ENABLED_SETTINGS,
+            fileRolesStore,
+            nativeRolesStore,
+            reservedRolesStore,
+            nativePrivStore,
+            Collections.emptyList(),
+            new ThreadContext(SECURITY_ENABLED_SETTINGS),
+            new XPackLicenseState(SECURITY_ENABLED_SETTINGS),
+            cache,
+            apiKeyService,
+            documentSubsetBitsetCache,
+            rds -> effectiveRoleDescriptors.set(rds));
+        AuditUtil.getOrGenerateRequestId(threadContext);
+        final BytesArray roleBytes = new BytesArray("{\"a role\": {\"cluster\": [\"all\"]}}");
+        final BytesArray limitedByRoleBytes = new BytesArray("{\"limitedBy role\": {\"cluster\": [\"all\"]}}");
+        Authentication authentication = new Authentication(new User("test api key user", "superuser"),
+            new RealmRef("_es_api_key", "_es_api_key", "node"),
+            null,
+            Version.CURRENT,
+            AuthenticationType.API_KEY,
+            org.elasticsearch.common.collect.Map.of(API_KEY_ID_KEY,
+                "key-id-1",
+                API_KEY_ROLE_DESCRIPTORS_KEY,
+                roleBytes,
+                API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
+                limitedByRoleBytes));
+        doCallRealMethod().when(apiKeyService).getApiKeyIdAndRoleBytes(eq(authentication), anyBoolean());
+
+        PlainActionFuture<Role> roleFuture = new PlainActionFuture<>();
+        compositeRolesStore.getRoles(authentication.getUser(), authentication, roleFuture);
+        roleFuture.actionGet();
+        assertThat(effectiveRoleDescriptors.get(), is(nullValue()));
+        verify(apiKeyService, times(2)).getApiKeyIdAndRoleBytes(eq(authentication), anyBoolean());
+        verify(apiKeyService).parseRoleDescriptors("key-id-1", roleBytes);
+        verify(apiKeyService).parseRoleDescriptors("key-id-1", limitedByRoleBytes);
+
+        // Different API key with the same roles should read from cache
+        authentication = new Authentication(new User("test api key user 2", "superuser"),
+            new RealmRef("_es_api_key", "_es_api_key", "node"),
+            null,
+            Version.CURRENT,
+            AuthenticationType.API_KEY,
+            org.elasticsearch.common.collect.Map.of(API_KEY_ID_KEY,
+                "key-id-2",
+                API_KEY_ROLE_DESCRIPTORS_KEY,
+                roleBytes,
+                API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
+                limitedByRoleBytes));
+        doCallRealMethod().when(apiKeyService).getApiKeyIdAndRoleBytes(eq(authentication), anyBoolean());
+        roleFuture = new PlainActionFuture<>();
+        compositeRolesStore.getRoles(authentication.getUser(), authentication, roleFuture);
+        roleFuture.actionGet();
+        assertThat(effectiveRoleDescriptors.get(), is(nullValue()));
+        verify(apiKeyService, times(2)).getApiKeyIdAndRoleBytes(eq(authentication), anyBoolean());
+        verify(apiKeyService, never()).parseRoleDescriptors(eq("key-id-2"), any(BytesReference.class));
+
+        // Different API key with the same limitedBy role should read from cache, new role should be built
+        final BytesArray anotherRoleBytes = new BytesArray("{\"b role\": {\"cluster\": [\"manage_security\"]}}");
+        authentication = new Authentication(new User("test api key user 2", "superuser"),
+            new RealmRef("_es_api_key", "_es_api_key", "node"),
+            null,
+            Version.CURRENT,
+            AuthenticationType.API_KEY,
+            org.elasticsearch.common.collect.Map.of(API_KEY_ID_KEY,
+                "key-id-3",
+                API_KEY_ROLE_DESCRIPTORS_KEY,
+                anotherRoleBytes,
+                API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
+                limitedByRoleBytes));
+        doCallRealMethod().when(apiKeyService).getApiKeyIdAndRoleBytes(eq(authentication), anyBoolean());
+        roleFuture = new PlainActionFuture<>();
+        compositeRolesStore.getRoles(authentication.getUser(), authentication, roleFuture);
+        roleFuture.actionGet();
+        assertThat(effectiveRoleDescriptors.get(), is(nullValue()));
+        verify(apiKeyService).getApiKeyIdAndRoleBytes(eq(authentication), eq(false));
+        verify(apiKeyService).parseRoleDescriptors("key-id-3", anotherRoleBytes);
+    }
+
     private Authentication createAuthentication() {
         final RealmRef lookedUpBy;
         final User user;

x-pack/qa/rolling-upgrade/build.gradle

@@ -48,6 +48,7 @@ for (Version bwcVersion : BuildParams.bwcVersions.wireCompatible) {
       setting 'xpack.security.transport.ssl.enabled', 'true'
       setting 'xpack.security.authc.token.enabled', 'true'
       setting 'xpack.security.authc.token.timeout', '60m'
+      setting 'xpack.security.authc.api_key.enabled', 'true'
       setting 'xpack.security.audit.enabled', 'true'
       setting 'xpack.security.transport.ssl.key', 'testnode.pem'
       setting 'xpack.security.transport.ssl.certificate', 'testnode.crt'

x-pack/qa/rolling-upgrade/src/test/resources/rest-api-spec/test/mixed_cluster/120_api_key_auth.yml

@@ -0,0 +1,23 @@
+---
+"Test API key authentication will work in a mixed cluster":
+
+  - skip:
+      features: headers
+
+  - do:
+      security.create_api_key:
+        body:  >
+          {
+            "name": "my-api-key"
+          }
+  - match: { name: "my-api-key" }
+  - is_true: id
+  - is_true: api_key
+  - transform_and_set: { login_creds: "#base64EncodeCredentials(id,api_key)" }
+
+  - do:
+      headers:
+        Authorization: ApiKey ${login_creds}
+      nodes.info: {}
+  - match: { _nodes.failed: 0 }
+