[Backport] Add more context to cluster access denied messages (#68263)

In #60357 we improved the error message when access to perform an
action on an index was denied by including the index name and the
privileges that would grant the action.

This commit extends the second part of that change (the list of
privileges that would resolve the problem) to situations when a
cluster action is denied.

This implementation for cluster privileges is slightly more complex
than that of index privileges because cluster privileges can be
dependent on parameters in the request, not just the action name.
For example, "manage_own_api_key" should be suggested as a matching
privilege when a user attempts to create an API key, or delete their
own API key, but should not be suggested when that same user attempts
to delete another user's API key.

Backport of: #66900

Author: tvernum

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

@@ -207,9 +207,14 @@ protected boolean extendedCheck(String action, TransportRequest request, Authent
 
         @Override
         protected boolean doImplies(ActionBasedPermissionCheck permissionCheck) {
-            return permissionCheck instanceof AutomatonPermissionCheck;
+            /*
+             * We know that "permissionCheck" has an automaton which is a subset of ours.
+             * Which means "permissionCheck" _cannot_ grant an action that we don't (see ActionBasedPermissionCheck#check)
+             * Since we grant _all_ requests on actions within our automaton, we must therefore grant _all_ actions+requests that the other
+             * permission check grants.
+             */
+            return true;
         }
-
     }
 
     // action, request based permission check

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

@@ -19,6 +19,7 @@ public class ActionClusterPrivilege implements NamedClusterPrivilege {
     private final String name;
     private final Set<String> allowedActionPatterns;
     private final Set<String> excludedActionPatterns;
+    private final ClusterPermission permission;
 
     /**
      * Constructor for {@link ActionClusterPrivilege} defining what cluster actions are accessible for the user with this privilege.
@@ -43,6 +44,7 @@ public ActionClusterPrivilege(final String name, final Set<String> allowedAction
         this.name = name;
         this.allowedActionPatterns = allowedActionPatterns;
         this.excludedActionPatterns = excludedActionPatterns;
+        this.permission = buildPermission(ClusterPermission.builder()).build();
     }
 
     @Override
@@ -62,4 +64,9 @@ public Set<String> getExcludedActionPatterns() {
     public ClusterPermission.Builder buildPermission(final ClusterPermission.Builder builder) {
         return builder.add(this, allowedActionPatterns, excludedActionPatterns);
     }
+
+    @Override
+    public ClusterPermission permission() {
+        return permission;
+    }
 }

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

@@ -18,6 +18,7 @@
 import org.elasticsearch.action.ingest.SimulatePipelineAction;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.xpack.core.ilm.action.GetLifecycleAction;
 import org.elasticsearch.xpack.core.ilm.action.GetStatusAction;
 import org.elasticsearch.xpack.core.ilm.action.StartILMAction;
@@ -28,15 +29,21 @@
 import org.elasticsearch.xpack.core.security.action.token.InvalidateTokenAction;
 import org.elasticsearch.xpack.core.security.action.token.RefreshTokenAction;
 import org.elasticsearch.xpack.core.security.action.user.HasPrivilegesAction;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.slm.action.GetSnapshotLifecycleAction;
 
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 /**
  * Translates cluster privilege names into concrete implementations
@@ -153,8 +160,7 @@ public class ClusterPrivilegeResolver {
     public static final NamedClusterPrivilege MANAGE_LOGSTASH_PIPELINES = new ActionClusterPrivilege("manage_logstash_pipelines",
         Collections.unmodifiableSet(Sets.newHashSet("cluster:admin/logstash/pipeline/*")));
 
-    private static final Map<String, NamedClusterPrivilege> VALUES = Collections.unmodifiableMap(
-        Stream.of(
+    private static final Map<String, NamedClusterPrivilege> VALUES = sortByAccessLevel(Arrays.asList(
         NONE,
         ALL,
         MONITOR,
@@ -193,7 +199,7 @@ public class ClusterPrivilegeResolver {
         DELEGATE_PKI,
         MANAGE_OWN_API_KEY,
         MANAGE_ENRICH,
-        MANAGE_LOGSTASH_PIPELINES).collect(Collectors.toMap(cp -> cp.name(), cp -> cp)));
+        MANAGE_LOGSTASH_PIPELINES));
 
     /**
      * Resolves a {@link NamedClusterPrivilege} from a given name if it exists.
@@ -234,4 +240,36 @@ private static String actionToPattern(String text) {
         return text + "*";
     }
 
+    /**
+     * Returns the names of privileges that grant the specified action and request, for the given authentication context.
+     * @return A collection of names, ordered (to the extent possible) from least privileged (e.g. {@link #MONITOR})
+     * to most privileged (e.g. {@link #ALL})
+     * @see #sortByAccessLevel(Collection)
+     * @see org.elasticsearch.xpack.core.security.authz.permission.ClusterPermission#check(String, TransportRequest, Authentication)
+     */
+    public static Collection<String> findPrivilegesThatGrant(String action, TransportRequest request, Authentication authentication) {
+        return Collections.unmodifiableList(VALUES.entrySet().stream()
+            .filter(e -> e.getValue().permission().check(action, request, authentication))
+            .map(Map.Entry::getKey)
+            .collect(Collectors.toList()));
+    }
+
+    /**
+     * Sorts the collection of privileges from least-privilege to most-privilege (to the extent possible),
+     * returning them in a sorted map keyed by name.
+     */
+    static SortedMap<String, NamedClusterPrivilege> sortByAccessLevel(Collection<NamedClusterPrivilege> privileges) {
+        // How many other privileges does this privilege imply. Those with a higher count are considered to be a higher privilege
+        final Map<String, Long> impliesCount = new HashMap<>(privileges.size());
+        privileges.forEach(priv -> impliesCount.put(priv.name(),
+            privileges.stream().filter(p2 -> p2 != priv && priv.permission().implies(p2.permission())).count())
+        );
+
+        final Comparator<String> compare = Comparator.<String>comparingLong(key -> impliesCount.getOrDefault(key, 0L))
+            .thenComparing(Comparator.naturalOrder());
+        final TreeMap<String, NamedClusterPrivilege> tree = new TreeMap<>(compare);
+        privileges.forEach(p -> tree.put(p.name(), p));
+        return Collections.unmodifiableSortedMap(tree);
+    }
+
 }

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

@@ -27,7 +27,10 @@ public class ManageOwnApiKeyClusterPrivilege implements NamedClusterPrivilege {
     private static final String PRIVILEGE_NAME = "manage_own_api_key";
     private static final String API_KEY_ID_KEY = "_security_api_key_id";
 
+    private final ClusterPermission permission;
+
     private ManageOwnApiKeyClusterPrivilege() {
+        permission = this.buildPermission(ClusterPermission.builder()).build();
     }
 
     @Override
@@ -40,6 +43,11 @@ public ClusterPermission.Builder buildPermission(ClusterPermission.Builder build
         return builder.add(this, ManageOwnClusterPermissionCheck.INSTANCE);
     }
 
+    @Override
+    public ClusterPermission permission() {
+        return permission;
+    }
+
     private static final class ManageOwnClusterPermissionCheck extends ClusterPermission.ActionBasedPermissionCheck {
         public static final ManageOwnClusterPermissionCheck INSTANCE = new ManageOwnClusterPermissionCheck();
 

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

@@ -8,11 +8,24 @@
 package org.elasticsearch.xpack.core.security.authz.privilege;
 
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
+import org.elasticsearch.xpack.core.security.authz.permission.ClusterPermission;
 
 /**
  * A {@link ClusterPrivilege} that has a name. The named cluster privileges can be referred simply by name within a
  * {@link RoleDescriptor#getClusterPrivileges()}.
  */
 public interface NamedClusterPrivilege extends ClusterPrivilege {
     String name();
+
+    /**
+     * Returns a permission that represents this privilege only.
+     * When building a role (or role-like object) that has many privileges, it is more efficient to build a shared permission using the
+     * {@link #buildPermission(ClusterPermission.Builder)} method instead. This method is intended to allow callers to interrogate the
+     * runtime permissions specifically granted by this privilege.
+     * It is acceptable (and encouraged) for implementations of this method to cache (or precompute) the {@link ClusterPermission}
+     * and return the same object on each call.
+     * @see #buildPermission(ClusterPermission.Builder)
+     */
+    ClusterPermission permission();
+
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/permission/ClusterPermissionTests.java

@@ -8,20 +8,24 @@
 package org.elasticsearch.xpack.core.security.authz.permission;
 
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authz.privilege.ClusterPrivilege;
 import org.elasticsearch.xpack.core.security.authz.privilege.ClusterPrivilegeResolver;
 import org.elasticsearch.xpack.core.security.authz.privilege.ConfigurableClusterPrivilege;
+import org.elasticsearch.xpack.core.security.authz.privilege.NamedClusterPrivilege;
 import org.junit.Before;
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.List;
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.is;
@@ -230,6 +234,33 @@ public void testClusterPermissionSubsetIsImpliedByAllClusterPermission() {
         assertThat(allClusterPermission.implies(otherClusterPermission), is(true));
     }
 
+    public void testImpliesOnSecurityPrivilegeHierarchy() {
+        final List<ClusterPermission> highToLow = CollectionUtils.arrayAsArrayList(
+            ClusterPrivilegeResolver.ALL.permission(),
+            ClusterPrivilegeResolver.MANAGE_SECURITY.permission(),
+            ClusterPrivilegeResolver.MANAGE_API_KEY.permission(),
+            ClusterPrivilegeResolver.MANAGE_OWN_API_KEY.permission()
+        );
+
+        for (int i = 0; i < highToLow.size(); i++) {
+            ClusterPermission high = highToLow.get(i);
+            for (int j = i; j < highToLow.size(); j++) {
+                ClusterPermission low = highToLow.get(j);
+                assertThat("Permission " + name(high) + " should imply " + name(low), high.implies(low), is(true));
+            }
+        }
+    }
+
+    private String name(ClusterPermission permission) {
+        return permission.privileges().stream().map(priv -> {
+            if (priv instanceof NamedClusterPrivilege) {
+                return ((NamedClusterPrivilege) priv).name();
+            } else {
+                return priv.toString();
+            }
+        }).collect(Collectors.joining(","));
+    }
+
     private static class MockConfigurableClusterPrivilege implements ConfigurableClusterPrivilege {
         private Predicate<TransportRequest> requestPredicate;
 

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/privilege/ClusterPrivilegeResolverTests.java

@@ -0,0 +1,39 @@
+/*
+ * 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.privilege;
+
+import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.SortedMap;
+
+import static org.hamcrest.Matchers.contains;
+
+public class ClusterPrivilegeResolverTests extends ESTestCase {
+
+    public void testSortByAccessLevel() throws Exception {
+        final List<NamedClusterPrivilege> privileges = CollectionUtils.arrayAsArrayList(
+            ClusterPrivilegeResolver.ALL,
+            ClusterPrivilegeResolver.MONITOR,
+            ClusterPrivilegeResolver.MANAGE,
+            ClusterPrivilegeResolver.MANAGE_OWN_API_KEY,
+            ClusterPrivilegeResolver.MANAGE_API_KEY,
+            ClusterPrivilegeResolver.MANAGE_SECURITY
+        );
+        Collections.shuffle(privileges, random());
+        final SortedMap<String, NamedClusterPrivilege> sorted = ClusterPrivilegeResolver.sortByAccessLevel(privileges);
+        // This is:
+        // "manage_own_api_key", "monitor" (neither of which grant anything else in the list), sorted by name
+        // "manage" and "manage_api_key",(which each grant 1 other privilege in the list), sorted by name
+        // "manage_security" and "all", sorted by access level ("all" implies "manage_security")
+        assertThat(sorted.keySet(), contains("manage_own_api_key", "monitor", "manage", "manage_api_key", "manage_security", "all"));
+    }
+
+}