[Failure Store] Has Privileges API (elastic#125329)

This PR adds support for checking access to the failure store via the
Has Privileges API.

To check access for a data stream `logs`, a request must query for a
concrete named privilege, `read_failure_store` or
`manage_failure_store`, e.g., a request to the HasPrivileges API by a
user with `read_failure_store` over `logs`:

```
POST /_security/user/_has_privileges
{
    "index": [
        {
            "names": ["logs"],
            "privileges": ["read_failure_store", "read", "indices:data/read/*"]
        }
    ]
}
```

Returns:
```

{     "username": "<...>",     "has_all_requested": false,
"cluster": {},     "index": {         "logs": {
"read_failure_store": true,             "read": false, <1>
"indices:data/read/*": false <2>         }     },     "application": {}
}

```
Note that `<1>` and `<2>` are both `false` since `read` is not covered by `read_failure_store` and neither are any raw actions like `indices:data/read/*` since these implicitly correspond to data access.

Selectors are not allowed in the index patterns of HasPrivileges requests to avoid ambiguities such as checking `read` on `logs::failures` as well as the ambiguity of index patterns that are regular expressions.

(cherry picked from commit 0e0214d)

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

Author: n1v0lg

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

@@ -339,14 +339,27 @@ public ActionRequestValidationException validate(ActionRequestValidationExceptio
             if (index == null) {
                 validationException = addValidationError("indexPrivileges must not be null", validationException);
             } else {
-                for (int i = 0; i < index.length; i++) {
-                    BytesReference query = index[i].getQuery();
+                for (RoleDescriptor.IndicesPrivileges indicesPrivileges : index) {
+                    BytesReference query = indicesPrivileges.getQuery();
                     if (query != null) {
                         validationException = addValidationError(
                             "may only check index privileges without any DLS query [" + query.utf8ToString() + "]",
                             validationException
                         );
                     }
+                    if (DataStream.isFailureStoreFeatureFlagEnabled()) {
+                        // best effort prevent users from attempting to use selectors in privilege check
+                        for (String indexPattern : indicesPrivileges.getIndices()) {
+                            if (IndexNameExpressionResolver.hasSelector(indexPattern, IndexComponentSelector.FAILURES)
+                                || IndexNameExpressionResolver.hasSelector(indexPattern, IndexComponentSelector.DATA)) {
+                                validationException = addValidationError(
+                                    "may only check index privileges without selectors in index patterns [" + indexPattern + "]",
+                                    validationException
+                                );
+                                break;
+                            }
+                        }
+                    }
                 }
             }
             if (application == null) {
@@ -368,31 +381,6 @@ public ActionRequestValidationException validate(ActionRequestValidationExceptio
                 && application.length == 0) {
                 validationException = addValidationError("must specify at least one privilege", validationException);
             }
-            if (index != null) {
-                // no need to validate failure-store related constraints if it's not enabled
-                if (DataStream.isFailureStoreFeatureFlagEnabled()) {
-                    for (RoleDescriptor.IndicesPrivileges indexPrivilege : index) {
-                        if (indexPrivilege.getIndices() != null
-                            && Arrays.stream(indexPrivilege.getIndices())
-                                // best effort prevent users from attempting to check failure selectors
-                                .anyMatch(idx -> IndexNameExpressionResolver.hasSelector(idx, IndexComponentSelector.FAILURES))) {
-                            validationException = addValidationError(
-                                // TODO adjust message once HasPrivileges check supports checking failure store privileges
-                                "failures selector is not supported in index patterns",
-                                validationException
-                            );
-                        }
-                        if (indexPrivilege.getPrivileges() != null
-                            && Arrays.stream(indexPrivilege.getPrivileges())
-                                .anyMatch(p -> "read_failure_store".equals(p) || "manage_failure_store".equals(p))) {
-                            validationException = addValidationError(
-                                "checking failure store privileges is not supported",
-                                validationException
-                            );
-                        }
-                    }
-                }
-            }
             return validationException;
         }
 

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

@@ -317,36 +317,58 @@ public boolean checkResourcePrivileges(
         @Nullable ResourcePrivilegesMap.Builder resourcePrivilegesMapBuilder
     ) {
         boolean allMatch = true;
-        Map<Automaton, Automaton> indexGroupAutomatons = indexGroupAutomatons(
-            combineIndexGroups && checkForIndexPatterns.stream().anyMatch(Automatons::isLuceneRegex)
+        Map<Automaton, Automaton> indexGroupAutomatonsForDataSelector = indexGroupAutomatons(
+            combineIndexGroups && checkForIndexPatterns.stream().anyMatch(Automatons::isLuceneRegex),
+            IndexComponentSelector.DATA
         );
+        // optimization: if there are no failures selector privileges in the set of privileges to check, we can skip building
+        // the automaton map
+        final boolean containsPrivilegesForFailuresSelector = containsPrivilegesForFailuresSelector(checkForPrivileges);
+        Map<Automaton, Automaton> indexGroupAutomatonsForFailuresSelector = false == containsPrivilegesForFailuresSelector
+            ? Map.of()
+            : indexGroupAutomatons(
+                combineIndexGroups && checkForIndexPatterns.stream().anyMatch(Automatons::isLuceneRegex),
+                IndexComponentSelector.FAILURES
+            );
         for (String forIndexPattern : checkForIndexPatterns) {
-            IndexNameExpressionResolver.assertExpressionHasNullOrDataSelector(forIndexPattern);
             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 (var indexAndPrivilegeAutomaton : indexGroupAutomatons.entrySet()) {
-                    if (Operations.subsetOf(checkIndexAutomaton, indexAndPrivilegeAutomaton.getValue())) {
-                        if (allowedIndexPrivilegesAutomaton != null) {
-                            allowedIndexPrivilegesAutomaton = Automatons.unionAndMinimize(
-                                Arrays.asList(allowedIndexPrivilegesAutomaton, indexAndPrivilegeAutomaton.getKey())
-                            );
-                        } else {
-                            allowedIndexPrivilegesAutomaton = indexAndPrivilegeAutomaton.getKey();
-                        }
-                    }
-                }
+                Automaton allowedPrivilegesAutomatonForDataSelector = getIndexPrivilegesAutomaton(
+                    indexGroupAutomatonsForDataSelector,
+                    checkIndexAutomaton
+                );
+                Automaton allowedPrivilegesAutomatonForFailuresSelector = getIndexPrivilegesAutomaton(
+                    indexGroupAutomatonsForFailuresSelector,
+                    checkIndexAutomaton
+                );
                 for (String privilege : checkForPrivileges) {
-                    IndexPrivilege indexPrivilege = IndexPrivilege.get(privilege);
-                    if (allowedIndexPrivilegesAutomaton != null
-                        && Operations.subsetOf(indexPrivilege.getAutomaton(), allowedIndexPrivilegesAutomaton)) {
+                    final IndexPrivilege indexPrivilege = IndexPrivilege.get(privilege);
+                    final boolean checkWithDataSelector = indexPrivilege.getSelectorPredicate().test(IndexComponentSelector.DATA);
+                    final boolean checkWithFailuresSelector = indexPrivilege.getSelectorPredicate().test(IndexComponentSelector.FAILURES);
+                    assert checkWithDataSelector || checkWithFailuresSelector
+                        : "index privilege must map to at least one of [data, failures] selectors";
+                    assert containsPrivilegesForFailuresSelector
+                        || indexPrivilege.getSelectorPredicate() != IndexComponentSelectorPredicate.FAILURES
+                        : "no failures access privileges should be present in the set of privileges to check";
+                    final Automaton automatonToCheck = indexPrivilege.getAutomaton();
+                    if (checkWithDataSelector
+                        && allowedPrivilegesAutomatonForDataSelector != null
+                        && Automatons.subsetOf(automatonToCheck, allowedPrivilegesAutomatonForDataSelector)) {
                         if (resourcePrivilegesMapBuilder != null) {
                             resourcePrivilegesMapBuilder.addResourcePrivilege(forIndexPattern, privilege, Boolean.TRUE);
                         }
-                    } else {
+                    } else if (checkWithFailuresSelector
+                        && allowedPrivilegesAutomatonForFailuresSelector != null
+                        && Automatons.subsetOf(automatonToCheck, allowedPrivilegesAutomatonForFailuresSelector)) {
+                            if (resourcePrivilegesMapBuilder != null) {
+                                resourcePrivilegesMapBuilder.addResourcePrivilege(forIndexPattern, privilege, Boolean.TRUE);
+                            }
+                        }
+                    // comment to force correct else-block indent
+                    else {
                         if (resourcePrivilegesMapBuilder != null) {
                             resourcePrivilegesMapBuilder.addResourcePrivilege(forIndexPattern, privilege, Boolean.FALSE);
                             allMatch = false;
@@ -806,13 +828,11 @@ private static boolean containsPrivilegeThatGrantsMappingUpdatesForBwc(Group gro
      *
      * @return a map of all index and privilege pattern automatons
      */
-    private Map<Automaton, Automaton> indexGroupAutomatons(boolean combine) {
+    private Map<Automaton, Automaton> indexGroupAutomatons(boolean combine, IndexComponentSelector selector) {
         // Map of privilege automaton object references (cached by IndexPrivilege::CACHE)
         Map<Automaton, Automaton> allAutomatons = new HashMap<>();
         for (Group group : groups) {
-            // TODO support failure store privileges
-            // we also check that the group does not support data access to avoid erroneously filtering out `all` privilege groups
-            if (group.checkSelector(IndexComponentSelector.FAILURES) && false == group.checkSelector(IndexComponentSelector.DATA)) {
+            if (false == group.checkSelector(selector)) {
                 continue;
             }
             Automaton indexAutomaton = group.getIndexMatcherAutomaton();
@@ -845,6 +865,41 @@ private Map<Automaton, Automaton> indexGroupAutomatons(boolean combine) {
         return allAutomatons;
     }
 
+    private static boolean containsPrivilegesForFailuresSelector(Set<String> checkForPrivileges) {
+        for (String privilege : checkForPrivileges) {
+            // use `getNamedOrNull` since only a named privilege can be a failures-only privilege (raw action names are always data access)
+            IndexPrivilege named = IndexPrivilege.getNamedOrNull(privilege);
+            // note: we are looking for failures-only privileges here, not `all` which does cover failures but is not a failures-only
+            // privilege
+            if (named != null && named.getSelectorPredicate() == IndexComponentSelectorPredicate.FAILURES) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    @Nullable
+    private static Automaton getIndexPrivilegesAutomaton(Map<Automaton, Automaton> indexGroupAutomatons, Automaton checkIndexAutomaton) {
+        if (indexGroupAutomatons.isEmpty()) {
+            return null;
+        }
+        Automaton allowedPrivilegesAutomaton = null;
+        for (Map.Entry<Automaton, Automaton> indexAndPrivilegeAutomaton : indexGroupAutomatons.entrySet()) {
+            Automaton indexNameAutomaton = indexAndPrivilegeAutomaton.getValue();
+            if (Automatons.subsetOf(checkIndexAutomaton, indexNameAutomaton)) {
+                Automaton privilegesAutomaton = indexAndPrivilegeAutomaton.getKey();
+                if (allowedPrivilegesAutomaton != null) {
+                    allowedPrivilegesAutomaton = Automatons.unionAndMinimize(
+                        Arrays.asList(allowedPrivilegesAutomaton, privilegesAutomaton)
+                    );
+                } else {
+                    allowedPrivilegesAutomaton = privilegesAutomaton;
+                }
+            }
+        }
+        return allowedPrivilegesAutomaton;
+    }
+
     public static class Group {
         public static final Group[] EMPTY_ARRAY = new Group[0];