Add manage roles privilege (#110633)

This PR adds functionality to limit the resources and privileges an
Elasticsearch user can grant permissions to when creating a role. This
is achieved using a new
[global](https://www.elastic.co/guide/en/elasticsearch/reference/current/defining-roles.html)
(configurable/request aware) cluster privilege , named `role`, with a
sub-key called `manage/indices` which is an array where each entry is a
pair of  [index
patterns](https://docs.google.com/document/d/1VN73C2KpmvvOW85-XGUqMmnMwXrfK4aoxRtG8tPqk7Y/edit#heading=h.z74zwo30t0pf)
and [index
privileges](https://www.elastic.co/guide/en/elasticsearch/reference/current/security-privileges.html#privileges-list-indices).

## Definition - Using a role with this privilege to create, update or
delete roles with privileges on indices outside of the indices matched
by the [index
pattern](https://docs.google.com/document/d/1VN73C2KpmvvOW85-XGUqMmnMwXrfK4aoxRtG8tPqk7Y/edit#heading=h.z74zwo30t0pf)
in the indices array, will fail.  - Using a role with this privilege to
try to create, update or delete roles with cluster, run_as, etc.
privileges will fail.  - Using a role with this privilege with
restricted indices will fail. - Other broader privileges (such as
manage_security) will nullify this privilege.

## Example Create `test-manage` role:

```
POST _security/role/test-manage
{
    "global": {
        "role": {
            "manage": {
                "indices": [
                    {
                        "names": ["allowed-index-prefix-*"],
                        "privileges":["read"]
                    }
                ]
            }
        }
    }
}
```

And then a user with that role creates a role:

```
POST _security/role/a-test-role
{
    "indices": [
        {
            "names": [
                "allowed-index-prefix-some-index"
            ],
            "privileges": [
                "read"
            ]}]
}
```

But this would fail for:

```
POST _security/role/a-test-role
{
    "indices": [
        {
            "names": [
                "not-allowed-index-prefix-some-index"
            ],
            "privileges": [
                "read"
            ]}]
}
```

## Backwards compatibility and mixed cluster concerns - A new mapping
version has been added to the security index to store the new privilege.
- If the new mapping version is not applied and a role descriptor with
the new global privilege is written, the write will fail causing an
exception. - When sending role descriptors over the transport layer in a
mixed cluster, the new global privilege needs to be excluded for older
versions. This is hanled with a new transport version.  - If a role
descriptor is serialized for API keys on one node in a mixed cluster and
read from another, an older node might not be able to deserialize it, so
it needs to be removed before being written in mixed cluster with old
nodes. This is handled in the API key service.  - If a role descriptor
containing a global privilege is in a put role request in a mixed
cluster where it's not supported on all nodes, fail request to create
role. - RCS is not applicable here since RCS only considers cluster
privileges and index privileges (not global cluster privileges). - This
doesn't include remote privileges, since the current use case with
connectors doesn't need roles to be created on a cluster separate from
the cluster where the search data resides.

## Follow up work - Create a docs PR  - Error handling for actions that
use manage roles. Should configurable cluster privileges that grant
restricted usage of actions be listed in error authorization error
messages?

Author: jfreden

docs/changelog/110633.yaml

@@ -0,0 +1,5 @@
+pr: 110633
+summary: Add manage roles privilege
+area: Authorization
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -198,6 +198,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ALLOW_PARTIAL_SEARCH_RESULTS_IN_PIT = def(8_728_00_0);
     public static final TransportVersion RANK_DOCS_RETRIEVER = def(8_729_00_0);
     public static final TransportVersion ESQL_ES_FIELD_CACHED_SERIALIZATION = def(8_730_00_0);
+    public static final TransportVersion ADD_MANAGE_ROLES_PRIVILEGE = def(8_731_00_0);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

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

@@ -149,7 +149,7 @@ public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
             new NamedWriteableRegistry.Entry(ClusterState.Custom.class, TokenMetadata.TYPE, TokenMetadata::new),
             new NamedWriteableRegistry.Entry(NamedDiff.class, TokenMetadata.TYPE, TokenMetadata::readDiffFrom),
             new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.SECURITY, SecurityFeatureSetUsage::new),
-            // security : conditional privileges
+            // security : configurable cluster privileges
             new NamedWriteableRegistry.Entry(
                 ConfigurableClusterPrivilege.class,
                 ConfigurableClusterPrivileges.ManageApplicationPrivileges.WRITEABLE_NAME,
@@ -160,6 +160,11 @@ public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
                 ConfigurableClusterPrivileges.WriteProfileDataPrivileges.WRITEABLE_NAME,
                 ConfigurableClusterPrivileges.WriteProfileDataPrivileges::createFrom
             ),
+            new NamedWriteableRegistry.Entry(
+                ConfigurableClusterPrivilege.class,
+                ConfigurableClusterPrivileges.ManageRolesPrivilege.WRITEABLE_NAME,
+                ConfigurableClusterPrivileges.ManageRolesPrivilege::createFrom
+            ),
             // security : role-mappings
             new NamedWriteableRegistry.Entry(Metadata.Custom.class, RoleMappingMetadata.TYPE, RoleMappingMetadata::new),
             new NamedWriteableRegistry.Entry(NamedDiff.class, RoleMappingMetadata.TYPE, RoleMappingMetadata::readDiffFrom),

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

@@ -10,13 +10,15 @@
 import org.apache.lucene.util.automaton.Operations;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
+import org.elasticsearch.xpack.core.security.authz.RestrictedIndices;
 import org.elasticsearch.xpack.core.security.authz.privilege.ClusterPrivilege;
 import org.elasticsearch.xpack.core.security.support.Automatons;
 
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.function.Predicate;
 
 /**
@@ -84,6 +86,16 @@ public static class Builder {
         private final List<Automaton> actionAutomatons = new ArrayList<>();
         private final List<PermissionCheck> permissionChecks = new ArrayList<>();
 
+        private final RestrictedIndices restrictedIndices;
+
+        public Builder(RestrictedIndices restrictedIndices) {
+            this.restrictedIndices = restrictedIndices;
+        }
+
+        public Builder() {
+            this.restrictedIndices = null;
+        }
+
         public Builder add(
             final ClusterPrivilege clusterPrivilege,
             final Set<String> allowedActionPatterns,
@@ -110,6 +122,16 @@ public Builder add(final ClusterPrivilege clusterPrivilege, final PermissionChec
             return this;
         }
 
+        public Builder addWithPredicateSupplier(
+            final ClusterPrivilege clusterPrivilege,
+            final Set<String> allowedActionPatterns,
+            final Function<RestrictedIndices, Predicate<TransportRequest>> requestPredicateSupplier
+        ) {
+            final Automaton actionAutomaton = createAutomaton(allowedActionPatterns, Set.of());
+            Predicate<TransportRequest> requestPredicate = requestPredicateSupplier.apply(restrictedIndices);
+            return add(clusterPrivilege, new ActionRequestBasedPermissionCheck(clusterPrivilege, actionAutomaton, requestPredicate));
+        }
+
         public ClusterPermission build() {
             if (clusterPrivileges.isEmpty()) {
                 return NONE;

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

@@ -20,6 +20,7 @@
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.xpack.core.security.authz.RestrictedIndices;
 import org.elasticsearch.xpack.core.security.authz.accesscontrol.IndicesAccessControl;
@@ -86,6 +87,7 @@ public Builder addGroup(
         public IndicesPermission build() {
             return new IndicesPermission(restrictedIndices, groups.toArray(Group.EMPTY_ARRAY));
         }
+
     }
 
     private IndicesPermission(RestrictedIndices restrictedIndices, Group[] groups) {
@@ -238,6 +240,21 @@ public boolean check(String action) {
         return false;
     }
 
+    public boolean checkResourcePrivileges(
+        Set<String> checkForIndexPatterns,
+        boolean allowRestrictedIndices,
+        Set<String> checkForPrivileges,
+        @Nullable ResourcePrivilegesMap.Builder resourcePrivilegesMapBuilder
+    ) {
+        return checkResourcePrivileges(
+            checkForIndexPatterns,
+            allowRestrictedIndices,
+            checkForPrivileges,
+            false,
+            resourcePrivilegesMapBuilder
+        );
+    }
+
     /**
      * For given index patterns and index privileges determines allowed privileges and creates an instance of {@link ResourcePrivilegesMap}
      * holding a map of resource to {@link ResourcePrivileges} where resource is index pattern and the map of index privilege to whether it
@@ -246,6 +263,7 @@ public boolean check(String action) {
      * @param checkForIndexPatterns check permission grants for the set of index patterns
      * @param allowRestrictedIndices if {@code true} then checks permission grants even for restricted indices by index matching
      * @param checkForPrivileges check permission grants for the set of index privileges
+     * @param combineIndexGroups combine index groups to enable checking against regular expressions
      * @param resourcePrivilegesMapBuilder out-parameter for returning the details on which privilege over which resource is granted or not.
      *                                     Can be {@code null} when no such details are needed so the method can return early, after
      *                                     encountering the first privilege that is not granted over some resource.
@@ -255,26 +273,28 @@ public boolean checkResourcePrivileges(
         Set<String> checkForIndexPatterns,
         boolean allowRestrictedIndices,
         Set<String> checkForPrivileges,
+        boolean combineIndexGroups,
         @Nullable ResourcePrivilegesMap.Builder resourcePrivilegesMapBuilder
     ) {
-        final Map<IndicesPermission.Group, Automaton> predicateCache = new HashMap<>();
         boolean allMatch = true;
+        Map<Automaton, Automaton> indexGroupAutomatons = indexGroupAutomatons(
+            combineIndexGroups && checkForIndexPatterns.stream().anyMatch(Automatons::isLuceneRegex)
+        );
         for (String forIndexPattern : checkForIndexPatterns) {
             Automaton checkIndexAutomaton = Automatons.patterns(forIndexPattern);
             if (false == allowRestrictedIndices && false == isConcreteRestrictedIndex(forIndexPattern)) {
                 checkIndexAutomaton = Automatons.minusAndMinimize(checkIndexAutomaton, restrictedIndices.getAutomaton());
             }
             if (false == Operations.isEmpty(checkIndexAutomaton)) {
                 Automaton allowedIndexPrivilegesAutomaton = null;
-                for (Group group : groups) {
-                    final Automaton groupIndexAutomaton = predicateCache.computeIfAbsent(group, Group::getIndexMatcherAutomaton);
-                    if (Operations.subsetOf(checkIndexAutomaton, groupIndexAutomaton)) {
+                for (var indexAndPrivilegeAutomaton : indexGroupAutomatons.entrySet()) {
+                    if (Operations.subsetOf(checkIndexAutomaton, indexAndPrivilegeAutomaton.getValue())) {
                         if (allowedIndexPrivilegesAutomaton != null) {
                             allowedIndexPrivilegesAutomaton = Automatons.unionAndMinimize(
-                                Arrays.asList(allowedIndexPrivilegesAutomaton, group.privilege().getAutomaton())
+                                Arrays.asList(allowedIndexPrivilegesAutomaton, indexAndPrivilegeAutomaton.getKey())
                             );
                         } else {
-                            allowedIndexPrivilegesAutomaton = group.privilege().getAutomaton();
+                            allowedIndexPrivilegesAutomaton = indexAndPrivilegeAutomaton.getKey();
                         }
                     }
                 }
@@ -656,6 +676,61 @@ private static boolean containsPrivilegeThatGrantsMappingUpdatesForBwc(Group gro
         return group.privilege().name().stream().anyMatch(PRIVILEGE_NAME_SET_BWC_ALLOW_MAPPING_UPDATE::contains);
     }
 
+    /**
+     * Get all automatons for the index groups in this permission and optionally combine the index groups to enable checking if a set of
+     * index patterns specified using a regular expression grants a set of index privileges.
+     *
+     * <p>An index group is defined as a set of index patterns and a set of privileges (excluding field permissions and DLS queries).
+     * {@link IndicesPermission} consist of a set of index groups. For non-regular expression privilege checks, an index pattern is checked
+     * against each index group, to see if it's a sub-pattern of the index pattern for the group and then if that group grants some or all
+     * of the privileges requested. For regular expressions it's not sufficient to check per group since the index patterns covered by a
+     * group can be distinct sets and a regular expression can cover several distinct sets.
+     *
+     * <p>For example the two index groups: {"names": ["a"], "privileges": ["read", "create"]} and {"names": ["b"],
+     * "privileges": ["read","delete"]} will not match on ["\[ab]\"], while a single index group:
+     * {"names": ["a", "b"], "privileges": ["read"]} will. This happens because the index groups are evaluated against a request index
+     * pattern without first being combined. In the example above, the two index patterns should be combined to:
+     * {"names": ["a", "b"], "privileges": ["read"]} before being checked.
+     *
+     *
+     * @param combine combine index groups to allow for checking against regular expressions
+     *
+     * @return a map of all index and privilege pattern automatons
+     */
+    private Map<Automaton, Automaton> indexGroupAutomatons(boolean combine) {
+        // Map of privilege automaton object references (cached by IndexPrivilege::CACHE)
+        Map<Automaton, Automaton> allAutomatons = new HashMap<>();
+        for (Group group : groups) {
+            Automaton indexAutomaton = group.getIndexMatcherAutomaton();
+            allAutomatons.compute(
+                group.privilege().getAutomaton(),
+                (key, value) -> value == null ? indexAutomaton : Automatons.unionAndMinimize(List.of(value, indexAutomaton))
+            );
+            if (combine) {
+                List<Tuple<Automaton, Automaton>> combinedAutomatons = new ArrayList<>();
+                for (var indexAndPrivilegeAutomatons : allAutomatons.entrySet()) {
+                    Automaton intersectingPrivileges = Operations.intersection(
+                        indexAndPrivilegeAutomatons.getKey(),
+                        group.privilege().getAutomaton()
+                    );
+                    if (Operations.isEmpty(intersectingPrivileges) == false) {
+                        Automaton indexPatternAutomaton = Automatons.unionAndMinimize(
+                            List.of(indexAndPrivilegeAutomatons.getValue(), indexAutomaton)
+                        );
+                        combinedAutomatons.add(new Tuple<>(intersectingPrivileges, indexPatternAutomaton));
+                    }
+                }
+                combinedAutomatons.forEach(
+                    automatons -> allAutomatons.compute(
+                        automatons.v1(),
+                        (key, value) -> value == null ? automatons.v2() : Automatons.unionAndMinimize(List.of(value, automatons.v2()))
+                    )
+                );
+            }
+        }
+        return allAutomatons;
+    }
+
     public static class Group {
         public static final Group[] EMPTY_ARRAY = new Group[0];
 

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

@@ -233,7 +233,7 @@ private Builder(RestrictedIndices restrictedIndices, String[] names) {
         }
 
         public Builder cluster(Set<String> privilegeNames, Iterable<ConfigurableClusterPrivilege> configurableClusterPrivileges) {
-            ClusterPermission.Builder builder = ClusterPermission.builder();
+            ClusterPermission.Builder builder = new ClusterPermission.Builder(restrictedIndices);
             if (privilegeNames.isEmpty() == false) {
                 for (String name : privilegeNames) {
                     builder = ClusterPrivilegeResolver.resolve(name).buildPermission(builder);

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/ConfigurableClusterPrivilege.java

@@ -41,7 +41,8 @@ public interface ConfigurableClusterPrivilege extends NamedWriteable, ToXContent
      */
     enum Category {
         APPLICATION(new ParseField("application")),
-        PROFILE(new ParseField("profile"));
+        PROFILE(new ParseField("profile")),
+        ROLE(new ParseField("role"));
 
         public final ParseField field;