Refactor CompositeRolesStore for speration of concerns

The process of role resolving is to build the Role object from an
Authentication object. The high level steps of this process is as the follows:

1. Locate the role reference for the Authentication, e.g. for regular user,
this means a collection of role names.

2. Retrieve the role descriptors for the role reference, e.g. search the
security index to get the role descriptors for the role name.

3. Build the Role object based on the role descriptors.  There are also special
cases in the above process. For example, API keys do not have role names, but
two byteReference representing the role descriptors. API keys also have a
nested Role structure for limiting the key's actual privileges based on the
owner's.

Today, this process is managed by a single CompositeRolesStore class and the
steps are not clearly separated. This PR improves the situation by introducing
a new RoleReferenceResolver class that is responsible for turning roleReference
into role descriptors. CompositeRolesStore is now only responsible for buiding
the Role from the descriptors.

The RoleReference is also a new concept introduced by this PR, along with
AuthenticationContext and Subject. They technically belong to the issue of
revisiting Authentication class (#80117). But they are needed here to faciliate
the changes. Their usage will be expanded once we start working on #80117.

A Subject knows how to return a list of RoleReference and the final Role is the
intersection of all the RoleReference. This was a concept for API keys. It is
now generalised in this PR in the light on potential expanded usage of
limitedBy roles.

Relates: #80117

Author: ywangd

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

@@ -0,0 +1,116 @@
+/*
+ * 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.core.security.authc;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.xpack.core.security.user.User;
+
+import java.util.Map;
+
+import static org.elasticsearch.xpack.core.security.authc.Authentication.AuthenticationType;
+import static org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef;
+
+public class AuthenticationContext {
+
+    private final Version version;
+    private final Subject authenticatingSubject;
+    private final Subject effectiveSubject;
+    // TODO: Rename to AuthenticationMethod
+    private final AuthenticationType type;
+
+    private AuthenticationContext(
+        Version version,
+        Subject authenticatingSubject,
+        Subject effectiveSubject,
+        AuthenticationType authenticationType
+    ) {
+        this.version = version;
+        this.authenticatingSubject = authenticatingSubject;
+        this.effectiveSubject = effectiveSubject;
+        this.type = authenticationType;
+    }
+
+    public boolean isRunAs() {
+        assert authenticatingSubject != null && effectiveSubject != null;
+        return authenticatingSubject != effectiveSubject;
+    }
+
+    public Subject getAuthenticatingSubject() {
+        return authenticatingSubject;
+    }
+
+    public Subject getEffectiveSubject() {
+        return effectiveSubject;
+    }
+
+    public Authentication toAuthentication() {
+        return new Authentication(
+            effectiveSubject.getUser(),
+            authenticatingSubject.getRealm(),
+            effectiveSubject.getRealm(),
+            version,
+            type,
+            authenticatingSubject.getMetadata()
+        );
+    }
+
+    public static AuthenticationContext fromAuthentication(Authentication authentication) {
+        final Builder builder = new Builder(authentication.getVersion());
+        builder.authenticationType(authentication.getAuthenticationType());
+        final User user = authentication.getUser();
+        if (user.isRunAs()) {
+            builder.authenticatingSubject(user.authenticatedUser(), authentication.getAuthenticatedBy(), authentication.getMetadata());
+            // The lookup user for run-as currently don't have authentication metadata associated with them because
+            // lookupUser only returns the User object. The lookup user for authorization delegation does have
+            // authentication metadata, but the realm does not expose this difference between authenticatingUser and
+            // delegateUser so effectively this is handled together with the authenticatingSubject not effectiveSubject.
+            builder.effectiveSubject(user, authentication.getLookedUpBy(), Map.of());
+        } else {
+            builder.authenticatingSubject(user, authentication.getAuthenticatedBy(), authentication.getMetadata());
+        }
+        return builder.build();
+    }
+
+    public static class Builder {
+        private final Version version;
+        private AuthenticationType authenticationType;
+        private Subject authenticatingSubject;
+        private Subject effectiveSubject;
+
+        public Builder() {
+            this(Version.CURRENT);
+        }
+
+        public Builder(Version version) {
+            this.version = version;
+        }
+
+        public Builder authenticationType(AuthenticationType authenticationType) {
+            this.authenticationType = authenticationType;
+            return this;
+        }
+
+        public Builder authenticatingSubject(User authenticatingUser, RealmRef authenticatingRealmRef, Map<String, Object> metadata) {
+            this.authenticatingSubject = new Subject(authenticatingUser, authenticatingRealmRef, version, metadata);
+            return this;
+        }
+
+        public Builder effectiveSubject(User effectiveUser, RealmRef lookupRealmRef, Map<String, Object> metadata) {
+            this.effectiveSubject = new Subject(effectiveUser, lookupRealmRef, version, metadata);
+            return this;
+        }
+
+        public AuthenticationContext build() {
+            if (effectiveSubject == null) {
+                effectiveSubject = authenticatingSubject;
+            }
+            return new AuthenticationContext(version, authenticatingSubject, effectiveSubject, authenticationType);
+        }
+    }
+
+}

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

@@ -24,5 +24,8 @@ public final class AuthenticationField {
     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";
 
+    public static final String ANONYMOUS_REALM_NAME = "__anonymous";
+    public static final String ANONYMOUS_REALM_TYPE = "__anonymous";
+
     private AuthenticationField() {}
 }

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

@@ -0,0 +1,199 @@
+/*
+ * 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.core.security.authc;
+
+import org.elasticsearch.ElasticsearchSecurityException;
+import org.elasticsearch.Version;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.util.ArrayUtils;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.xpack.core.security.authc.service.ServiceAccountSettings;
+import org.elasticsearch.xpack.core.security.authz.store.RoleReference;
+import org.elasticsearch.xpack.core.security.user.AnonymousUser;
+import org.elasticsearch.xpack.core.security.user.User;
+
+import java.util.List;
+import java.util.Map;
+
+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 subject is a more generic concept similar to user and associated to the current authentication.
+ * It is more generic than user because it can also represent API keys and service accounts.
+ * It also contains authentication level information, e.g. realm and metadata so that it can answer
+ * queries in a better encapsulated way.
+ */
+public class Subject {
+
+    public enum Type {
+        USER,
+        API_KEY,
+        SERVICE_ACCOUNT,
+    }
+
+    private final Version version;
+    private final User user;
+    private final Authentication.RealmRef realm;
+    private final Type type;
+    private final Map<String, Object> metadata;
+
+    public Subject(User user, Authentication.RealmRef realm) {
+        this(user, realm, Version.CURRENT, Map.of());
+    }
+
+    public Subject(User user, Authentication.RealmRef realm, Version version, Map<String, Object> metadata) {
+        this.version = version;
+        this.user = user;
+        this.realm = realm;
+        // Realm can be null for run-as user if it does not exist. Pretend it is a user and it will be rejected later in authorization
+        // This is to be consistent with existing behaviour.
+        if (realm == null) {
+            this.type = Type.USER;
+        } else if (AuthenticationField.API_KEY_REALM_TYPE.equals(realm.getType())) {
+            assert AuthenticationField.API_KEY_REALM_NAME.equals(realm.getName()) : "api key realm name mismatch";
+            this.type = Type.API_KEY;
+        } else if (ServiceAccountSettings.REALM_TYPE.equals(realm.getType())) {
+            assert ServiceAccountSettings.REALM_NAME.equals(realm.getName()) : "service account realm name mismatch";
+            this.type = Type.SERVICE_ACCOUNT;
+        } else {
+            this.type = Type.USER;
+        }
+        this.metadata = metadata;
+    }
+
+    public User getUser() {
+        return user;
+    }
+
+    public Authentication.RealmRef getRealm() {
+        return realm;
+    }
+
+    public Type getType() {
+        return type;
+    }
+
+    public Map<String, Object> getMetadata() {
+        return metadata;
+    }
+
+    /**
+     * Return a List of RoleReferences that represents role definitions associated to the subject.
+     * The final role of this subject should be the intersection of all role references in the list.
+     */
+    public List<RoleReference> getRoleReferences(@Nullable AnonymousUser anonymousUser) {
+        switch (type) {
+            case USER:
+                return buildRoleReferencesForUser(anonymousUser);
+            case API_KEY:
+                return buildRoleReferencesForApiKey();
+            case SERVICE_ACCOUNT:
+                return List.of(new RoleReference.ServiceAccountRoleReference(user.principal()));
+            default:
+                assert false : "unknown subject type: [" + type + "]";
+                throw new IllegalStateException("unknown subject type: [" + type + "]");
+        }
+    }
+
+    private List<RoleReference> buildRoleReferencesForUser(AnonymousUser anonymousUser) {
+        if (user.equals(anonymousUser)) {
+            return List.of(new RoleReference.NamedRoleReference(user.roles()));
+        }
+        final String[] allRoleNames;
+        if (anonymousUser == null || false == anonymousUser.enabled()) {
+            allRoleNames = user.roles();
+        } else {
+            // TODO: should we validate enable status and length of role names on instantiation time of anonymousUser?
+            if (anonymousUser.roles().length == 0) {
+                throw new IllegalStateException("anonymous is only enabled when the anonymous user has roles");
+            }
+            allRoleNames = ArrayUtils.concat(user.roles(), anonymousUser.roles());
+        }
+        return List.of(new RoleReference.NamedRoleReference(allRoleNames));
+    }
+
+    private List<RoleReference> buildRoleReferencesForApiKey() {
+        if (version.before(VERSION_API_KEY_ROLES_AS_BYTES)) {
+            return buildRolesReferenceForApiKeyBwc();
+        }
+        final String apiKeyId = (String) metadata.get(AuthenticationField.API_KEY_ID_KEY);
+        final BytesReference roleDescriptorsBytes = (BytesReference) metadata.get(API_KEY_ROLE_DESCRIPTORS_KEY);
+        final BytesReference limitedByRoleDescriptorsBytes = getLimitedByRoleDescriptorsBytes();
+        if (roleDescriptorsBytes == null && limitedByRoleDescriptorsBytes == null) {
+            throw new ElasticsearchSecurityException("no role descriptors found for API key");
+        }
+        final RoleReference.ApiKeyRoleReference limitedByRoleReference = new RoleReference.ApiKeyRoleReference(
+            apiKeyId,
+            limitedByRoleDescriptorsBytes,
+            "apikey_limited_role"
+        );
+        if (isEmptyRoleDescriptorsBytes(roleDescriptorsBytes)) {
+            return List.of(limitedByRoleReference);
+        }
+        return List.of(new RoleReference.ApiKeyRoleReference(apiKeyId, roleDescriptorsBytes, "apikey_role"), limitedByRoleReference);
+    }
+
+    private boolean isEmptyRoleDescriptorsBytes(BytesReference roleDescriptorsBytes) {
+        return roleDescriptorsBytes == null || (roleDescriptorsBytes.length() == 2 && "{}".equals(roleDescriptorsBytes.utf8ToString()));
+    }
+
+    private List<RoleReference> buildRolesReferenceForApiKeyBwc() {
+        final String apiKeyId = (String) metadata.get(AuthenticationField.API_KEY_ID_KEY);
+        final Map<String, Object> roleDescriptorsMap = getRoleDescriptorMap(API_KEY_ROLE_DESCRIPTORS_KEY);
+        final Map<String, Object> limitedByRoleDescriptorsMap = getRoleDescriptorMap(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY);
+        if (roleDescriptorsMap == null && limitedByRoleDescriptorsMap == null) {
+            throw new ElasticsearchSecurityException("no role descriptors found for API key");
+        } else {
+            final RoleReference.BwcApiKeyRoleReference limitedByRoleReference = new RoleReference.BwcApiKeyRoleReference(
+                apiKeyId,
+                limitedByRoleDescriptorsMap,
+                "_limited_role_desc"
+            );
+            if (roleDescriptorsMap == null || roleDescriptorsMap.isEmpty()) {
+                return List.of(limitedByRoleReference);
+            } else {
+                return List.of(
+                    new RoleReference.BwcApiKeyRoleReference(apiKeyId, roleDescriptorsMap, "_role_desc"),
+                    limitedByRoleReference
+                );
+            }
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private Map<String, Object> getRoleDescriptorMap(String key) {
+        return (Map<String, Object>) metadata.get(key);
+    }
+
+    // This following fixed role descriptor is for fleet-server BWC on and before 7.14.
+    // It is fixed and must NOT be updated when the fleet-server service account updates.
+    private static final BytesArray FLEET_SERVER_ROLE_DESCRIPTOR_BYTES_V_7_14 = new BytesArray(
+        "{\"elastic/fleet-server\":{\"cluster\":[\"monitor\",\"manage_own_api_key\"],"
+            + "\"indices\":[{\"names\":[\"logs-*\",\"metrics-*\",\"traces-*\",\"synthetics-*\","
+            + "\".logs-endpoint.diagnostic.collection-*\"],"
+            + "\"privileges\":[\"write\",\"create_index\",\"auto_configure\"],\"allow_restricted_indices\":false},"
+            + "{\"names\":[\".fleet-*\"],\"privileges\":[\"read\",\"write\",\"monitor\",\"create_index\",\"auto_configure\"],"
+            + "\"allow_restricted_indices\":false}],\"applications\":[],\"run_as\":[],\"metadata\":{},"
+            + "\"transient_metadata\":{\"enabled\":true}}}"
+    );
+
+    private BytesReference getLimitedByRoleDescriptorsBytes() {
+        final BytesReference bytesReference = (BytesReference) metadata.get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY);
+        // Unfortunate BWC bug fix code
+        if (bytesReference.length() == 2 && "{}".equals(bytesReference.utf8ToString())) {
+            if (ServiceAccountSettings.REALM_NAME.equals(metadata.get(AuthenticationField.API_KEY_CREATOR_REALM_NAME))
+                && "elastic/fleet-server".equals(user.principal())) {
+                return FLEET_SERVER_ROLE_DESCRIPTOR_BYTES_V_7_14;
+            }
+        }
+        return bytesReference;
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/LimitedRole.java

@@ -22,6 +22,7 @@
 import java.util.Set;
 import java.util.function.Predicate;
 
+// TODO: extract a Role interface so limitedRole can be more than 2 levels
 /**
  * A {@link Role} limited by another role.<br>
  * The effective permissions returned on {@link #authorize(String, Set, Map, FieldPermissionsCache)} call would be limited by the

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

@@ -0,0 +1,58 @@
+/*
+ * 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.core.security.authz.store;
+
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * A unique identifier that can be associated to a Role. It can be used as cache key for role caching.
+ */
+public final class RoleKey {
+
+    public static final String ROLES_STORE_SOURCE = "roles_stores";
+    public static final RoleKey ROLE_KEY_EMPTY = new RoleKey(Set.of(), "__empty_role");
+    public static final RoleKey ROLE_KEY_SUPERUSER = new RoleKey(
+        Set.of(ReservedRolesStore.SUPERUSER_ROLE_DESCRIPTOR.getName()),
+        RoleKey.ROLES_STORE_SOURCE
+    );
+
+    private final Set<String> names;
+    private final String source;
+
+    public RoleKey(Set<String> names, String source) {
+        this.names = Objects.requireNonNull(names);
+        this.source = Objects.requireNonNull(source);
+    }
+
+    public Set<String> getNames() {
+        return names;
+    }
+
+    public String getSource() {
+        return source;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        RoleKey roleKey = (RoleKey) o;
+        return names.equals(roleKey.names) && source.equals(roleKey.source);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(names, source);
+    }
+
+    @Override
+    public String toString() {
+        return "RoleKey{" + "names=" + names + ", source='" + source + '\'' + '}';
+    }
+}