Failure Store Access Authorization (#123986)

This PR implements authorization logic for failure store access. It
builds on https://github.com/elastic/elasticsearch/pull/122715.

Access to the failure store is granted by two privileges:
`read_failure_store` and `manage_failure_store`. Either of these
privileges lets a user access a failure store via the `::failures`
selector, as well as access its backing failure indices. 
`read_failure_store` grants read access (for example to search documents
in a failure store), `manage_failure_store` grants access to write
operations, such as rollover. Users with only `read` or `manage` on a
data stream do not get failure store access. Vice versa, users with
`read_failure_store` and `manage_failure_store` do not get access to
regular data in a data stream. 

The PR implements this by making authorization logic selector-aware. It
involves two main changes:

1. Index permission groups now compare the selector under which an index resource is accessed to the selector associated with the group.
2. The `AuthorizedIndices` interface likewise uses selectors to decide which indices to treat as authorized. This part of the change requires a sizable refactor and changes to the interface. 

The high-level behavior for selector-aware search is as follows:

For a user with `read_failure_store` over data stream `logs`:

- `POST /logs::failures/_search` returns the documents in the failure store.
- `POST /logs/_search` returns a 403.
- `POST /logs/_search?ignore_unavailable=true` and `POST /*/_search` return an empty result.

Similarly, for a user with `read` over data stream `logs`:

- `POST /logs::failures/_search` returns a 403.
- `POST /logs/_search` returns documents in the data stream.
- `POST /logs::failures/_search?ignore_unavailable=true` and `POST /*::failures/_search` return an empty result.

A user with both `read` and `read_failure_store` over data stream `logs`
gets access to both `POST /logs::failures/_search` and `POST
/logs/_search`.

The index privilege `all` automatically grants access to both data and
the failures store, as well as all hypothetical future selectors. 

Resolves: ES-10873

plugins/examples/security-authorization-engine/src/main/java/org/elasticsearch/example/CustomAuthorizationEngine.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.example;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
@@ -35,7 +36,6 @@ import org.elasticsearch.xpack.core.security.user.User;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.function.Supplier;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -119,19 +119,19 @@ public class CustomAuthorizationEngine implements AuthorizationEngine {
     ) {
         if (isSuperuser(requestInfo.getAuthentication().getEffectiveSubject().getUser())) {
             listener.onResponse(new AuthorizedIndices() {
-                public Supplier<Set<String>> all() {
+                public Set<String> all(IndexComponentSelector selector) {
                     return () -> indicesLookup.keySet();
                 }
-                public boolean check(String name) {
+                public boolean check(String name, IndexComponentSelector selector) {
                     return indicesLookup.containsKey(name);
                 }
             });
         } else {
             listener.onResponse(new AuthorizedIndices() {
-                public Supplier<Set<String>> all() {
+                public Set<String> all(IndexComponentSelector selector) {
                     return () -> Set.of();
                 }
-                public boolean check(String name) {
+                public boolean check(String name, IndexComponentSelector selector) {
                     return false;
                 }
             });

server/src/main/java/org/elasticsearch/action/support/IndexComponentSelector.java

@@ -72,6 +72,24 @@ public enum IndexComponentSelector implements Writeable {
         return KEY_REGISTRY.get(key);
     }
 
+    /**
+     * Like {@link #getByKey(String)} but throws an exception if the key is not recognised.
+     * @return the selector if recognized. `null` input will return `DATA`.
+     * @throws IllegalArgumentException if the key was not recognised.
+     */
+    public static IndexComponentSelector getByKeyOrThrow(@Nullable String key) {
+        if (key == null) {
+            return DATA;
+        }
+        IndexComponentSelector selector = getByKey(key);
+        if (selector == null) {
+            throw new IllegalArgumentException(
+                "Unknown key of index component selector [" + key + "], available options are: " + KEY_REGISTRY.keySet()
+            );
+        }
+        return selector;
+    }
+
     public static IndexComponentSelector read(StreamInput in) throws IOException {
         byte id = in.readByte();
         if (in.getTransportVersion().onOrAfter(TransportVersions.REMOVE_ALL_APPLICABLE_SELECTOR)

server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstraction.java

@@ -100,6 +100,13 @@ public interface IndexAbstraction {
         return false;
     }
 
+    /**
+     * @return whether this index abstraction is a failure index of a data stream
+     */
+    default boolean isFailureIndexOfDataStream() {
+        return false;
+    }
+
     /**
      * An index abstraction type.
      */
@@ -183,6 +190,11 @@ public interface IndexAbstraction {
             return dataStream;
         }
 
+        @Override
+        public boolean isFailureIndexOfDataStream() {
+            return getParentDataStream() != null && getParentDataStream().isFailureStoreIndex(getName());
+        }
+
         @Override
         public boolean isHidden() {
             return isHidden;

server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstractionResolver.java

@@ -22,8 +22,8 @@ import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.function.Predicate;
+import java.util.function.BiPredicate;
-import java.util.function.Supplier;
+import java.util.function.Function;
 
 public class IndexAbstractionResolver {
 
@@ -37,8 +37,8 @@ public class IndexAbstractionResolver {
         Iterable<String> indices,
         IndicesOptions indicesOptions,
         ProjectMetadata projectMetadata,
-        Supplier<Set<String>> allAuthorizedAndAvailable,
+        Function<IndexComponentSelector, Set<String>> allAuthorizedAndAvailableBySelector,
-        Predicate<String> isAuthorized,
+        BiPredicate<String, IndexComponentSelector> isAuthorized,
         boolean includeDataStreams
     ) {
         List<String> finalIndices = new ArrayList<>();
@@ -64,6 +64,7 @@ public class IndexAbstractionResolver {
                 );
             }
             indexAbstraction = expressionAndSelector.v1();
+            IndexComponentSelector selector = IndexComponentSelector.getByKeyOrThrow(selectorString);
 
             // we always need to check for date math expressions
             indexAbstraction = IndexNameExpressionResolver.resolveDateMathExpression(indexAbstraction);
@@ -71,7 +72,7 @@ public class IndexAbstractionResolver {
             if (indicesOptions.expandWildcardExpressions() && Regex.isSimpleMatchPattern(indexAbstraction)) {
                 wildcardSeen = true;
                 Set<String> resolvedIndices = new HashSet<>();
-                for (String authorizedIndex : allAuthorizedAndAvailable.get()) {
+                for (String authorizedIndex : allAuthorizedAndAvailableBySelector.apply(selector)) {
                     if (Regex.simpleMatch(indexAbstraction, authorizedIndex)
                         && isIndexVisible(
                             indexAbstraction,
@@ -102,7 +103,7 @@ public class IndexAbstractionResolver {
                 resolveSelectorsAndCollect(indexAbstraction, selectorString, indicesOptions, resolvedIndices, projectMetadata);
                 if (minus) {
                     finalIndices.removeAll(resolvedIndices);
-                } else if (indicesOptions.ignoreUnavailable() == false || isAuthorized.test(indexAbstraction)) {
+                } else if (indicesOptions.ignoreUnavailable() == false || isAuthorized.test(indexAbstraction, selector)) {
                     // Unauthorized names are considered unavailable, so if `ignoreUnavailable` is `true` they should be silently
                     // discarded from the `finalIndices` list. Other "ways of unavailable" must be handled by the action
                     // handler, see: https://github.com/elastic/elasticsearch/issues/90215

server/src/main/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolver.java

@@ -28,6 +28,7 @@ import org.elasticsearch.common.time.DateUtils;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.core.Assertions;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Predicates;
 import org.elasticsearch.core.Tuple;
@@ -1001,6 +1002,14 @@ public class IndexNameExpressionResolver {
         return expression.contains(SelectorResolver.SELECTOR_SEPARATOR);
     }
 
+    public static boolean hasSelector(@Nullable String expression, IndexComponentSelector selector) {
+        Objects.requireNonNull(selector, "null selectors not supported");
+        if (expression == null) {
+            return false;
+        }
+        return expression.endsWith(SelectorResolver.SELECTOR_SEPARATOR + selector.getKey());
+    }
+
     /**
      * @return If the specified string is a selector expression then this method returns the base expression and its selector part.
      */
@@ -1022,6 +1031,14 @@ public class IndexNameExpressionResolver {
             : (baseExpression + SelectorResolver.SELECTOR_SEPARATOR + selectorExpression);
     }
 
+    public static void assertExpressionHasNullOrDataSelector(String expression) {
+        if (Assertions.ENABLED) {
+            var tuple = splitSelectorExpression(expression);
+            assert tuple.v2() == null || IndexComponentSelector.DATA.getKey().equals(tuple.v2())
+                : "Expected expression [" + expression + "] to have a data selector but found [" + tuple.v2() + "]";
+        }
+    }
+
     /**
      * Resolve an array of expressions to the set of indices and aliases that these expressions match.
      */

server/src/test/java/org/elasticsearch/cluster/metadata/IndexAbstractionResolverTests.java

@@ -27,7 +27,6 @@ import java.io.UncheckedIOException;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Supplier;
 
 import static org.elasticsearch.index.mapper.MapperService.SINGLE_MAPPING_NAME;
 import static org.elasticsearch.indices.SystemIndices.EXTERNAL_SYSTEM_INDEX_ACCESS_CONTROL_HEADER_KEY;
@@ -48,7 +47,7 @@ public class IndexAbstractionResolverTests extends ESTestCase {
     private String dateTimeIndexTomorrow;
 
     // Only used when resolving wildcard expressions
-    private final Supplier<Set<String>> defaultMask = () -> Set.of("index1", "index2", "data-stream1");
+    private final Set<String> defaultMask = Set.of("index1", "index2", "data-stream1");
 
     @Override
     public void setUp() throws Exception {
@@ -215,13 +214,11 @@ public class IndexAbstractionResolverTests extends ESTestCase {
 
     public void testIsIndexVisible() {
         assertThat(isIndexVisible("index1", null), is(true));
-        assertThat(isIndexVisible("index1", "*"), is(true));
         assertThat(isIndexVisible("index1", "data"), is(true));
         assertThat(isIndexVisible("index1", "failures"), is(false)); // *
         // * Indices don't have failure components so the failure component is not visible
 
         assertThat(isIndexVisible("data-stream1", null), is(true));
-        assertThat(isIndexVisible("data-stream1", "*"), is(true));
         assertThat(isIndexVisible("data-stream1", "data"), is(true));
         assertThat(isIndexVisible("data-stream1", "failures"), is(true));
     }
@@ -290,14 +287,14 @@ public class IndexAbstractionResolverTests extends ESTestCase {
             indexAbstractionResolver = new IndexAbstractionResolver(indexNameExpressionResolver);
 
             // this covers the GET * case -- with system access, you can see everything
-            assertThat(isIndexVisible("other", "*"), is(true));
+            assertThat(isIndexVisible("other", null), is(true));
-            assertThat(isIndexVisible(".foo", "*"), is(true));
+            assertThat(isIndexVisible(".foo", null), is(true));
-            assertThat(isIndexVisible(".bar", "*"), is(true));
+            assertThat(isIndexVisible(".bar", null), is(true));
 
             // but if you don't ask for hidden and aliases, you won't see hidden indices or aliases, naturally
-            assertThat(isIndexVisible("other", "*", noHiddenNoAliases), is(true));
+            assertThat(isIndexVisible("other", null, noHiddenNoAliases), is(true));
-            assertThat(isIndexVisible(".foo", "*", noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible(".foo", null, noHiddenNoAliases), is(false));
-            assertThat(isIndexVisible(".bar", "*", noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible(".bar", null, noHiddenNoAliases), is(false));
         }
 
         {
@@ -311,14 +308,14 @@ public class IndexAbstractionResolverTests extends ESTestCase {
             indexAbstractionResolver = new IndexAbstractionResolver(indexNameExpressionResolver);
 
             // this covers the GET * case -- without system access, you can't see everything
-            assertThat(isIndexVisible("other", "*"), is(true));
+            assertThat(isIndexVisible("other", null), is(true));
-            assertThat(isIndexVisible(".foo", "*"), is(false));
+            assertThat(isIndexVisible(".foo", null), is(false));
-            assertThat(isIndexVisible(".bar", "*"), is(false));
+            assertThat(isIndexVisible(".bar", null), is(false));
 
             // no difference here in the datastream case, you can't see these then, either
-            assertThat(isIndexVisible("other", "*", noHiddenNoAliases), is(true));
+            assertThat(isIndexVisible("other", null, noHiddenNoAliases), is(true));
-            assertThat(isIndexVisible(".foo", "*", noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible(".foo", null, noHiddenNoAliases), is(false));
-            assertThat(isIndexVisible(".bar", "*", noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible(".bar", null, noHiddenNoAliases), is(false));
         }
 
         {
@@ -333,14 +330,14 @@ public class IndexAbstractionResolverTests extends ESTestCase {
             indexAbstractionResolver = new IndexAbstractionResolver(indexNameExpressionResolver);
 
             // this covers the GET * case -- with product (only) access, you can't see everything
-            assertThat(isIndexVisible("other", "*"), is(true));
+            assertThat(isIndexVisible("other", null), is(true));
-            assertThat(isIndexVisible(".foo", "*"), is(false));
+            assertThat(isIndexVisible(".foo", null), is(false));
-            assertThat(isIndexVisible(".bar", "*"), is(false));
+            assertThat(isIndexVisible(".bar", null), is(false));
 
             // no difference here in the datastream case, you can't see these then, either
-            assertThat(isIndexVisible("other", "*", noHiddenNoAliases), is(true));
+            assertThat(isIndexVisible("other", null, noHiddenNoAliases), is(true));
-            assertThat(isIndexVisible(".foo", "*", noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible(".foo", null, noHiddenNoAliases), is(false));
-            assertThat(isIndexVisible(".bar", "*", noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible(".bar", null, noHiddenNoAliases), is(false));
         }
     }
 
@@ -366,8 +363,15 @@ public class IndexAbstractionResolverTests extends ESTestCase {
         return resolveAbstractions(expressions, IndicesOptions.strictExpandOpen(), defaultMask);
     }
 
-    private List<String> resolveAbstractions(List<String> expressions, IndicesOptions indicesOptions, Supplier<Set<String>> mask) {
+    private List<String> resolveAbstractions(List<String> expressions, IndicesOptions indicesOptions, Set<String> mask) {
-        return indexAbstractionResolver.resolveIndexAbstractions(expressions, indicesOptions, projectMetadata, mask, (idx) -> true, true);
+        return indexAbstractionResolver.resolveIndexAbstractions(
+            expressions,
+            indicesOptions,
+            projectMetadata,
+            (ignored) -> mask,
+            (ignored, nothing) -> true,
+            true
+        );
     }
 
     private boolean isIndexVisible(String index, String selector) {

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

@@ -11,8 +11,11 @@ import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.action.support.SubscribableListener;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
@@ -39,7 +42,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
-import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.action.ValidateActions.addValidationError;
@@ -281,22 +283,23 @@ public interface AuthorizationEngine {
     }
 
     /**
-     * Used to retrieve index-like resources that the user has access to, for a specific access action type,
+     * Used to retrieve index-like resources that the user has access to, for a specific access action type and selector,
      * at a specific point in time (for a fixed cluster state view).
      * It can also be used to check if a specific resource name is authorized (access to the resource name
      * can be authorized even if it doesn't exist).
      */
     interface AuthorizedIndices {
         /**
-         * Returns all the index-like resource names that are available and accessible for an action type by a user,
+         * Returns all the index-like resource names that are available and accessible for an action type and selector by a user,
          * at a fixed point in time (for a single cluster state view).
+         * The result is cached and subsequent calls to this method are idempotent.
          */
-        Supplier<Set<String>> all();
+        Set<String> all(IndexComponentSelector selector);
 
         /**
          * Checks if an index-like resource name is authorized, for an action by a user. The resource might or might not exist.
          */
-        boolean check(String name);
+        boolean check(String name, IndexComponentSelector selector);
     }
 
     /**
@@ -366,6 +369,31 @@ public interface AuthorizationEngine {
                 && 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/accesscontrol/IndicesAccessControl.java

@@ -6,13 +6,13 @@
  */
 package org.elasticsearch.xpack.core.security.authz.accesscontrol;
 
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.CachedSupplier;
 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.xpack.core.security.authz.IndicesAndAliasesResolverField;
 import org.elasticsearch.xpack.core.security.authz.permission.DocumentPermissions;
 import org.elasticsearch.xpack.core.security.authz.permission.FieldPermissions;
@@ -56,13 +56,15 @@ public class IndicesAccessControl {
     }
 
     /**
-     * @return The document and field permissions for an index if exist, otherwise <code>null</code> is returned.
+     * @return The document and field permissions for an index if they exist, otherwise <code>null</code> is returned.
      *         If <code>null</code> is being returned this means that there are no field or document level restrictions.
      */
     @Nullable
     public IndexAccessControl getIndexPermissions(String index) {
-        Tuple<String, String> indexAndSelector = IndexNameExpressionResolver.splitSelectorExpression(index);
+        assert false == IndexNameExpressionResolver.hasSelectorSuffix(index)
-        return this.getAllIndexPermissions().get(indexAndSelector.v1());
+            || IndexNameExpressionResolver.hasSelector(index, IndexComponentSelector.FAILURES)
+            : "index name [" + index + "] cannot have explicit selector other than ::failures";
+        return getAllIndexPermissions().get(index);
     }
 
     public boolean hasIndexPermissions(String index) {

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

@@ -142,17 +142,33 @@ public final class IndicesPermission {
     }
 
     private IsResourceAuthorizedPredicate buildIndexMatcherPredicateForAction(String action) {
-        final Set<String> ordinaryIndices = new HashSet<>();
+        final Set<String> dataAccessOrdinaryIndices = new HashSet<>();
-        final Set<String> restrictedIndices = new HashSet<>();
+        final Set<String> failuresAccessOrdinaryIndices = new HashSet<>();
+        final Set<String> dataAccessRestrictedIndices = new HashSet<>();
+        final Set<String> failuresAccessRestrictedIndices = new HashSet<>();
         final Set<String> grantMappingUpdatesOnIndices = new HashSet<>();
         final Set<String> grantMappingUpdatesOnRestrictedIndices = new HashSet<>();
         final boolean isMappingUpdateAction = isMappingUpdateAction(action);
         for (final Group group : groups) {
             if (group.actionMatcher.test(action)) {
+                final List<String> indexList = Arrays.asList(group.indices());
+                final boolean dataAccess = group.checkSelector(IndexComponentSelector.DATA);
+                final boolean failuresAccess = group.checkSelector(IndexComponentSelector.FAILURES);
+                assert dataAccess || failuresAccess : "group must grant access at least one of [DATA, FAILURES] selectors";
                 if (group.allowRestrictedIndices) {
-                    restrictedIndices.addAll(Arrays.asList(group.indices()));
+                    if (dataAccess) {
+                        dataAccessRestrictedIndices.addAll(indexList);
+                    }
+                    if (failuresAccess) {
+                        failuresAccessRestrictedIndices.addAll(indexList);
+                    }
                 } else {
-                    ordinaryIndices.addAll(Arrays.asList(group.indices()));
+                    if (dataAccess) {
+                        dataAccessOrdinaryIndices.addAll(indexList);
+                    }
+                    if (failuresAccess) {
+                        failuresAccessOrdinaryIndices.addAll(indexList);
+                    }
                 }
             } else if (isMappingUpdateAction && containsPrivilegeThatGrantsMappingUpdatesForBwc(group)) {
                 // special BWC case for certain privileges: allow put mapping on indices and aliases (but not on data streams), even if
@@ -164,30 +180,44 @@ public final class IndicesPermission {
                 }
             }
         }
-        final StringMatcher nameMatcher = indexMatcher(ordinaryIndices, restrictedIndices);
+        final StringMatcher dataAccessNameMatcher = indexMatcher(dataAccessOrdinaryIndices, dataAccessRestrictedIndices);
+        final StringMatcher failuresAccessNameMatcher = indexMatcher(failuresAccessOrdinaryIndices, failuresAccessRestrictedIndices);
         final StringMatcher bwcSpecialCaseMatcher = indexMatcher(grantMappingUpdatesOnIndices, grantMappingUpdatesOnRestrictedIndices);
-        return new IsResourceAuthorizedPredicate(nameMatcher, bwcSpecialCaseMatcher);
+        return new IsResourceAuthorizedPredicate(dataAccessNameMatcher, failuresAccessNameMatcher, bwcSpecialCaseMatcher);
     }
 
     /**
      * This encapsulates the authorization test for resources.
      * There is an additional test for resources that are missing or that are not a datastream or a backing index.
      */
-    public static class IsResourceAuthorizedPredicate implements BiPredicate<String, IndexAbstraction> {
+    public static class IsResourceAuthorizedPredicate {
 
-        private final BiPredicate<String, IndexAbstraction> biPredicate;
+        private final BiPredicate<String, IndexAbstraction> isAuthorizedForDataAccess;
+        private final BiPredicate<String, IndexAbstraction> isAuthorizedForFailuresAccess;
 
         // public for tests
-        public IsResourceAuthorizedPredicate(StringMatcher resourceNameMatcher, StringMatcher additionalNonDatastreamNameMatcher) {
+        public IsResourceAuthorizedPredicate(
+            StringMatcher dataResourceNameMatcher,
+            StringMatcher failuresResourceNameMatcher,
+            StringMatcher additionalNonDatastreamNameMatcher
+        ) {
             this((String name, @Nullable IndexAbstraction indexAbstraction) -> {
                 assert indexAbstraction == null || name.equals(indexAbstraction.getName());
-                return resourceNameMatcher.test(name)
+                return dataResourceNameMatcher.test(name)
                     || (isPartOfDatastream(indexAbstraction) == false && additionalNonDatastreamNameMatcher.test(name));
+            }, (String name, @Nullable IndexAbstraction indexAbstraction) -> {
+                assert indexAbstraction == null || name.equals(indexAbstraction.getName());
+                // we can't enforce that the abstraction is part of a data stream since we need to account for non-existent resources
+                return failuresResourceNameMatcher.test(name);
             });
         }
 
-        private IsResourceAuthorizedPredicate(BiPredicate<String, IndexAbstraction> biPredicate) {
+        private IsResourceAuthorizedPredicate(
-            this.biPredicate = biPredicate;
+            BiPredicate<String, IndexAbstraction> isAuthorizedForDataAccess,
+            BiPredicate<String, IndexAbstraction> isAuthorizedForFailuresAccess
+        ) {
+            this.isAuthorizedForDataAccess = isAuthorizedForDataAccess;
+            this.isAuthorizedForFailuresAccess = isAuthorizedForFailuresAccess;
         }
 
         /**
@@ -195,18 +225,25 @@ public final class IndicesPermission {
         * return a new {@link IsResourceAuthorizedPredicate} instance that is equivalent to the conjunction of
         * authorization tests of that other instance and this one.
         */
-        @Override
+        public final IsResourceAuthorizedPredicate and(IsResourceAuthorizedPredicate other) {
-        public final IsResourceAuthorizedPredicate and(BiPredicate<? super String, ? super IndexAbstraction> other) {
+            return new IsResourceAuthorizedPredicate(
-            return new IsResourceAuthorizedPredicate(this.biPredicate.and(other));
+                this.isAuthorizedForDataAccess.and(other.isAuthorizedForDataAccess),
+                this.isAuthorizedForFailuresAccess.and(other.isAuthorizedForFailuresAccess)
+            );
+        }
+
+        // TODO remove me (this has >700 usages in tests which would make for a horrible diff; will remove this once the main PR is merged)
+        public boolean test(IndexAbstraction indexAbstraction) {
+            return test(indexAbstraction.getName(), indexAbstraction, IndexComponentSelector.DATA);
         }
 
         /**
          * Verifies if access is authorized to the given {@param indexAbstraction} resource.
-         * The resource must exist. Otherwise, use the {@link #test(String, IndexAbstraction)} method.
+         * The resource must exist. Otherwise, use the {@link #test(String, IndexAbstraction, IndexComponentSelector)} method.
          * Returns {@code true} if access to the given resource is authorized or {@code false} otherwise.
          */
-        public final boolean test(IndexAbstraction indexAbstraction) {
+        public boolean test(IndexAbstraction indexAbstraction, IndexComponentSelector selector) {
-            return test(indexAbstraction.getName(), indexAbstraction);
+            return test(indexAbstraction.getName(), indexAbstraction, selector);
         }
 
         /**
@@ -215,9 +252,10 @@ public final class IndicesPermission {
          * if it doesn't.
          * Returns {@code true} if access to the given resource is authorized or {@code false} otherwise.
          */
-        @Override
+        public boolean test(String name, @Nullable IndexAbstraction indexAbstraction, IndexComponentSelector selector) {
-        public boolean test(String name, @Nullable IndexAbstraction indexAbstraction) {
+            return IndexComponentSelector.FAILURES.equals(selector)
-            return biPredicate.test(name, indexAbstraction);
+                ? isAuthorizedForFailuresAccess.test(name, indexAbstraction)
+                : isAuthorizedForDataAccess.test(name, indexAbstraction);
         }
 
         private static boolean isPartOfDatastream(IndexAbstraction indexAbstraction) {
@@ -283,6 +321,7 @@ public final class IndicesPermission {
             combineIndexGroups && checkForIndexPatterns.stream().anyMatch(Automatons::isLuceneRegex)
         );
         for (String forIndexPattern : checkForIndexPatterns) {
+            IndexNameExpressionResolver.assertExpressionHasNullOrDataSelector(forIndexPattern);
             Automaton checkIndexAutomaton = Automatons.patterns(forIndexPattern);
             if (false == allowRestrictedIndices && false == isConcreteRestrictedIndex(forIndexPattern)) {
                 checkIndexAutomaton = Automatons.minusAndMinimize(checkIndexAutomaton, restrictedIndices.getAutomaton());
@@ -338,9 +377,12 @@ public final class IndicesPermission {
     }
 
     public Automaton allowedActionsMatcher(String index) {
+        Tuple<String, String> tuple = IndexNameExpressionResolver.splitSelectorExpression(index);
+        String indexName = tuple.v1();
+        IndexComponentSelector selector = IndexComponentSelector.getByKey(tuple.v2());
         List<Automaton> automatonList = new ArrayList<>();
         for (Group group : groups) {
-            if (group.indexNameMatcher.test(index)) {
+            if (group.checkSelector(selector) && group.indexNameMatcher.test(indexName)) {
                 automatonList.add(group.privilege.getAutomaton());
             }
         }
@@ -373,15 +415,6 @@ public final class IndicesPermission {
             assert name != null : "Resource name cannot be null";
             assert abstraction == null || abstraction.getName().equals(name)
                 : "Index abstraction has unexpected name [" + abstraction.getName() + "] vs [" + name + "]";
-            assert abstraction == null
-                || selector == null
-                || IndexComponentSelector.FAILURES.equals(selector) == false
-                || abstraction.isDataStreamRelated()
-                : "Invalid index component selector ["
-                    + selector.getKey()
-                    + "] applied to abstraction of type ["
-                    + abstraction.getType()
-                    + "]";
             this.name = name;
             this.indexAbstraction = abstraction;
             this.selector = selector;
@@ -411,11 +444,19 @@ public final class IndicesPermission {
         public boolean checkIndex(Group group) {
             final DataStream ds = indexAbstraction == null ? null : indexAbstraction.getParentDataStream();
             if (ds != null) {
-                if (group.checkIndex(ds.getName())) {
+                if (indexAbstraction.isFailureIndexOfDataStream()) {
+                    // failure indices are special: when accessed directly (not through ::failures on parent data stream) they are accessed
+                    // implicitly as data. However, authz to the parent data stream happens via the failures selector
+                    if (group.checkSelector(IndexComponentSelector.FAILURES) && group.checkIndex(ds.getName())) {
                         return true;
                     }
+                } else if (IndexComponentSelector.DATA.equals(selector) || selector == null) {
+                    if (group.checkSelector(IndexComponentSelector.DATA) && group.checkIndex(ds.getName())) {
+                        return true;
                     }
-            return group.checkIndex(name);
+                } // we don't support granting access to a backing index with a failure selector via the parent data stream
+            }
+            return group.checkSelector(selector) && group.checkIndex(name);
         }
 
         /**
@@ -478,6 +519,13 @@ public final class IndicesPermission {
         public boolean canHaveBackingIndices() {
             return indexAbstraction != null && indexAbstraction.getType() != IndexAbstraction.Type.CONCRETE_INDEX;
         }
+
+        public String nameWithSelector() {
+            String combined = IndexNameExpressionResolver.combineSelector(name, selector);
+            assert false != IndexComponentSelector.FAILURES.equals(selector) || name.equals(combined)
+                : "Only failures selectors should result in explicit selectors suffix";
+            return combined;
+        }
     }
 
     /**
@@ -500,18 +548,20 @@ public final class IndicesPermission {
         int totalResourceCount = 0;
         Map<String, IndexAbstraction> lookup = metadata.getIndicesLookup();
         for (String indexOrAlias : requestedIndicesOrAliases) {
-            // Remove any selectors from abstraction name. Discard them for this check as we do not have access control for them (yet)
+            // Remove any selectors from abstraction name. Access control is based on the `selector` field of the IndexResource
             Tuple<String, String> expressionAndSelector = IndexNameExpressionResolver.splitSelectorExpression(indexOrAlias);
             indexOrAlias = expressionAndSelector.v1();
             IndexComponentSelector selector = expressionAndSelector.v2() == null
                 ? null
                 : IndexComponentSelector.getByKey(expressionAndSelector.v2());
             final IndexResource resource = new IndexResource(indexOrAlias, lookup.get(indexOrAlias), selector);
-            resources.put(resource.name, resource);
+            // We can't use resource.name here because we may be accessing a data stream _and_ its failure store,
+            // where the selector-free name is the same for both and thus ambiguous.
+            resources.put(resource.nameWithSelector(), resource);
             totalResourceCount += resource.size(lookup);
         }
 
-        final boolean overallGranted = isActionGranted(action, resources);
+        final boolean overallGranted = isActionGranted(action, resources.values());
         final int finalTotalResourceCount = totalResourceCount;
         final Supplier<Map<String, IndicesAccessControl.IndexAccessControl>> indexPermissions = () -> buildIndicesAccessControl(
             action,
@@ -540,10 +590,11 @@ public final class IndicesPermission {
 
         final boolean isMappingUpdateAction = isMappingUpdateAction(action);
 
-        for (IndexResource resource : requestedResources.values()) {
+        for (Map.Entry<String, IndexResource> resourceEntry : requestedResources.entrySet()) {
             // true if ANY group covers the given index AND the given action
             boolean granted = false;
-
+            final String resourceName = resourceEntry.getKey();
+            final IndexResource resource = resourceEntry.getValue();
             final Collection<String> concreteIndices = resource.resolveConcreteIndices(metadata);
             for (Group group : groups) {
                 // the group covers the given index OR the given index is a backing index and the group covers the parent data stream
@@ -590,9 +641,9 @@ public final class IndicesPermission {
                                 roleQueriesByIndex.put(index, docPermissions);
                             }
 
-                            if (index.equals(resource.name) == false) {
+                            if (index.equals(resourceName) == false) {
-                                fieldPermissionsByIndex.put(resource.name, fieldPermissions);
+                                fieldPermissionsByIndex.put(resourceName, fieldPermissions);
-                                roleQueriesByIndex.put(resource.name, docPermissions);
+                                roleQueriesByIndex.put(resourceName, docPermissions);
                             }
                         }
                     }
@@ -600,10 +651,11 @@ public final class IndicesPermission {
             }
 
             if (granted) {
-                grantedResources.add(resource.name);
+                grantedResources.add(resourceName);
+
                 if (resource.canHaveBackingIndices()) {
                     for (String concreteIndex : concreteIndices) {
-                        // If the name appear directly as part of the requested indices, it takes precedence over implicit access
+                        // If the name appears directly as part of the requested indices, it takes precedence over implicit access
                         if (false == requestedResources.containsKey(concreteIndex)) {
                             grantedResources.add(concreteIndex);
                         }
@@ -639,11 +691,11 @@ public final class IndicesPermission {
      * Returns {@code true} if action is granted for all {@code requestedResources}.
      * If action is not granted for at least one resource, this method will return {@code false}.
      */
-    private boolean isActionGranted(final String action, final Map<String, IndexResource> requestedResources) {
+    private boolean isActionGranted(final String action, final Collection<IndexResource> requestedResources) {
 
         final boolean isMappingUpdateAction = isMappingUpdateAction(action);
 
-        for (IndexResource resource : requestedResources.values()) {
+        for (IndexResource resource : requestedResources) {
             // true if ANY group covers the given index AND the given action
             boolean granted = false;
             // true if ANY group, which contains certain ingest privileges, covers the given index AND the action is a mapping update for
@@ -758,6 +810,11 @@ public final class IndicesPermission {
         // 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)) {
+                continue;
+            }
             Automaton indexAutomaton = group.getIndexMatcherAutomaton();
             allAutomatons.compute(
                 group.privilege().getAutomaton(),
@@ -864,8 +921,8 @@ public final class IndicesPermission {
             return query != null;
         }
 
-        public boolean checkSelector(IndexComponentSelector selector) {
+        public boolean checkSelector(@Nullable IndexComponentSelector selector) {
-            return selectorPredicate.test(selector);
+            return selectorPredicate.test(selector == null ? IndexComponentSelector.DATA : selector);
         }
 
         public boolean allowRestrictedIndices() {

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

@@ -48,11 +48,13 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.SortedMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
@@ -86,6 +88,7 @@ public final class IndexPrivilege extends Privilege {
         ResolveIndexAction.NAME,
         TransportResolveClusterAction.NAME
     );
+    private static final Automaton READ_FAILURE_STORE_AUTOMATON = patterns("indices:data/read/*", ResolveIndexAction.NAME);
     private static final Automaton READ_CROSS_CLUSTER_AUTOMATON = patterns(
         "internal:transport/proxy/indices:data/read/*",
         TransportClusterSearchShardsAction.TYPE.name(),
@@ -183,12 +186,6 @@ public final class IndexPrivilege extends Privilege {
 
     public static final IndexPrivilege NONE = new IndexPrivilege("none", Automatons.EMPTY);
     public static final IndexPrivilege ALL = new IndexPrivilege("all", ALL_AUTOMATON, IndexComponentSelectorPredicate.ALL);
-    public static final IndexPrivilege READ_FAILURE_STORE = new IndexPrivilege(
-        "read_failure_store",
-        // TODO use READ_AUTOMATON here in authorization follow-up
-        Automatons.EMPTY,
-        IndexComponentSelectorPredicate.FAILURES
-    );
     public static final IndexPrivilege READ = new IndexPrivilege("read", READ_AUTOMATON);
     public static final IndexPrivilege READ_CROSS_CLUSTER = new IndexPrivilege("read_cross_cluster", READ_CROSS_CLUSTER_AUTOMATON);
     public static final IndexPrivilege CREATE = new IndexPrivilege("create", CREATE_AUTOMATON);
@@ -219,11 +216,29 @@ public final class IndexPrivilege extends Privilege {
         CROSS_CLUSTER_REPLICATION_INTERNAL_AUTOMATON
     );
 
+    public static final IndexPrivilege READ_FAILURE_STORE = new IndexPrivilege(
+        "read_failure_store",
+        READ_FAILURE_STORE_AUTOMATON,
+        IndexComponentSelectorPredicate.FAILURES
+    );
+    public static final IndexPrivilege MANAGE_FAILURE_STORE = new IndexPrivilege(
+        "manage_failure_store",
+        MANAGE_AUTOMATON,
+        IndexComponentSelectorPredicate.FAILURES
+    );
+
     /**
      * If you are adding a new named index privilege, also add it to the
      * <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/security-privileges.html#privileges-list-indices">docs</a>.
      */
-    private static final Map<String, IndexPrivilege> VALUES = sortByAccessLevel(
+    private static final Map<String, IndexPrivilege> VALUES = combineSortedInOrder(
+        sortByAccessLevel(
+            Stream.of(
+                DataStream.isFailureStoreFeatureFlagEnabled() ? entry("read_failure_store", READ_FAILURE_STORE) : null,
+                DataStream.isFailureStoreFeatureFlagEnabled() ? entry("manage_failure_store", MANAGE_FAILURE_STORE) : null
+            ).filter(Objects::nonNull).collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue))
+        ),
+        sortByAccessLevel(
             Stream.of(
                 entry("none", NONE),
                 entry("all", ALL),
@@ -246,11 +261,26 @@ public final class IndexPrivilege extends Privilege {
                 entry("maintenance", MAINTENANCE),
                 entry("auto_configure", AUTO_CONFIGURE),
                 entry("cross_cluster_replication", CROSS_CLUSTER_REPLICATION),
-            entry("cross_cluster_replication_internal", CROSS_CLUSTER_REPLICATION_INTERNAL),
+                entry("cross_cluster_replication_internal", CROSS_CLUSTER_REPLICATION_INTERNAL)
-            DataStream.isFailureStoreFeatureFlagEnabled() ? entry("read_failure_store", READ_FAILURE_STORE) : null
+            ).collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue))
-        ).filter(Objects::nonNull).collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue))
+        )
     );
 
+    private static Map<String, IndexPrivilege> combineSortedInOrder(
+        SortedMap<String, IndexPrivilege> first,
+        SortedMap<String, IndexPrivilege> second
+    ) {
+        if (first.isEmpty()) {
+            return second;
+        }
+        if (second.isEmpty()) {
+            return first;
+        }
+        final Map<String, IndexPrivilege> combined = new LinkedHashMap<>(first);
+        combined.putAll(second);
+        return Collections.unmodifiableMap(combined);
+    }
+
     public static final Predicate<String> ACTION_MATCHER = ALL.predicate();
     public static final Predicate<String> CREATE_INDEX_MATCHER = CREATE_INDEX.predicate();
 

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

@@ -28,6 +28,7 @@ import org.elasticsearch.action.downsample.DownsampleAction;
 import org.elasticsearch.action.index.TransportIndexAction;
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.search.TransportSearchScrollAction;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.index.reindex.ReindexAction;
 import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.ilm.action.ILMActions;
@@ -245,12 +246,25 @@ public class InternalUsers {
         new RoleDescriptor(
             UsernamesField.LAZY_ROLLOVER_ROLE,
             new String[] {},
-            new RoleDescriptor.IndicesPrivileges[] {
+            DataStream.isFailureStoreFeatureFlagEnabled()
+                ? new RoleDescriptor.IndicesPrivileges[] {
                     RoleDescriptor.IndicesPrivileges.builder()
                         .indices("*")
                         .privileges(LazyRolloverAction.NAME)
                         .allowRestrictedIndices(true)
-                    .build() },
+                        .build(),
+                    RoleDescriptor.IndicesPrivileges.builder()
+                        .indices("*")
+                        // needed to rollover failure store
+                        .privileges("manage_failure_store")
+                        .allowRestrictedIndices(true)
+                        .build() }
+                : new RoleDescriptor.IndicesPrivileges[] {
+                    RoleDescriptor.IndicesPrivileges.builder()
+                        .indices("*")
+                        .privileges(LazyRolloverAction.NAME)
+                        .allowRestrictedIndices(true)
+                        .build(), },
             null,
             null,
             new String[] {},

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/action/profile/ProfileHasPrivilegesRequestTests.java

@@ -18,6 +18,8 @@ import org.elasticsearch.xpack.core.security.authz.privilege.IndexPrivilege;
 
 import java.util.List;
 import java.util.Locale;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.hasItem;
@@ -134,13 +136,20 @@ public class ProfileHasPrivilegesRequestTests extends AbstractWireSerializingTes
         )];
         for (int i = 0; i < indicesPrivileges.length; i++) {
             indicesPrivileges[i] = RoleDescriptor.IndicesPrivileges.builder()
-                .privileges(randomSubsetOf(randomIntBetween(1, 5), IndexPrivilege.names()))
+                .privileges(randomSubsetOf(randomIntBetween(1, 5), validPrivilegeNames()))
                 .indices(randomList(1, 3, () -> randomAlphaOfLengthBetween(2, 8) + (randomBoolean() ? "*" : "")))
                 .build();
         }
         return indicesPrivileges;
     }
 
+    private static Set<String> validPrivilegeNames() {
+        return IndexPrivilege.names()
+            .stream()
+            .filter(name -> false == name.equals("read_failure_store") && false == name.equals("manage_failure_store"))
+            .collect(Collectors.toSet());
+    }
+
     private static RoleDescriptor.ApplicationResourcePrivileges[] randomApplicationResourcePrivileges(boolean allowEmpty) {
         RoleDescriptor.ApplicationResourcePrivileges[] appPrivileges = new RoleDescriptor.ApplicationResourcePrivileges[randomIntBetween(
             allowEmpty ? 0 : 1,

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

@@ -529,22 +529,22 @@ public class LimitedRoleTests extends ESTestCase {
 
     public void testAllowedIndicesMatcher() {
         Role fromRole = Role.builder(EMPTY_RESTRICTED_INDICES, "a-role").add(IndexPrivilege.READ, "ind-1*").build();
-        assertThat(fromRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1")), is(true));
+        assertThat(fromRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1"), null), is(true));
-        assertThat(fromRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11")), is(true));
+        assertThat(fromRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11"), null), is(true));
-        assertThat(fromRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2")), is(false));
+        assertThat(fromRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2"), null), is(false));
 
         {
             Role limitedByRole = Role.builder(EMPTY_RESTRICTED_INDICES, "limited-role").add(IndexPrivilege.READ, "ind-1", "ind-2").build();
             assertThat(
-                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1")),
+                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1"), null),
                 is(true)
             );
             assertThat(
-                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11")),
+                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11"), null),
                 is(false)
             );
             assertThat(
-                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2")),
+                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2"), null),
                 is(true)
             );
             Role role;
@@ -553,18 +553,18 @@ public class LimitedRoleTests extends ESTestCase {
             } else {
                 role = fromRole.limitedBy(limitedByRole);
             }
-            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1")), is(true));
+            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1"), null), is(true));
-            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11")), is(false));
+            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11"), null), is(false));
-            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2")), is(false));
+            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2"), null), is(false));
         }
         {
             Role limitedByRole = Role.builder(EMPTY_RESTRICTED_INDICES, "limited-role").add(IndexPrivilege.READ, "ind-*").build();
             assertThat(
-                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1")),
+                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1"), null),
                 is(true)
             );
             assertThat(
-                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2")),
+                limitedByRole.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2"), null),
                 is(true)
             );
             Role role;
@@ -573,16 +573,16 @@ public class LimitedRoleTests extends ESTestCase {
             } else {
                 role = fromRole.limitedBy(limitedByRole);
             }
-            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1")), is(true));
+            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1"), null), is(true));
-            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2")), is(false));
+            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2"), null), is(false));
         }
     }
 
     public void testAllowedIndicesMatcherWithNestedRole() {
         Role role = Role.builder(EMPTY_RESTRICTED_INDICES, "a-role").add(IndexPrivilege.READ, "ind-1*").build();
-        assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1")), is(true));
+        assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1"), null), is(true));
-        assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11")), is(true));
+        assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11"), null), is(true));
-        assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2")), is(false));
+        assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2"), null), is(false));
 
         final int depth = randomIntBetween(2, 4);
         boolean index11Excluded = false;
@@ -598,12 +598,12 @@ public class LimitedRoleTests extends ESTestCase {
             } else {
                 role = role.limitedBy(limitedByRole);
             }
-            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1")), is(true));
+            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-1"), null), is(true));
             assertThat(
-                role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11")),
+                role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-11"), null),
                 is(false == index11Excluded)
             );
-            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2")), is(false));
+            assertThat(role.allowedIndicesMatcher(TransportSearchAction.TYPE.name()).test(mockIndexAbstraction("ind-2"), null), is(false));
         }
     }
 
@@ -646,6 +646,101 @@ public class LimitedRoleTests extends ESTestCase {
         assertThat(rolePredicate.test(TransportBulkAction.NAME), is(false));
     }
 
+    public void testAllowedActionsMatcherWithSelectors() {
+        Role fromRole = Role.builder(EMPTY_RESTRICTED_INDICES, "fromRole")
+            .add(IndexPrivilege.READ_FAILURE_STORE, "ind*")
+            .add(IndexPrivilege.READ, "ind*")
+            .add(IndexPrivilege.READ_FAILURE_STORE, "metric")
+            .add(IndexPrivilege.READ, "logs")
+            .build();
+        Automaton fromRoleAutomaton = fromRole.allowedActionsMatcher("index1");
+        Predicate<String> fromRolePredicate = Automatons.predicate(fromRoleAutomaton);
+        assertThat(fromRolePredicate.test(TransportSearchAction.TYPE.name()), is(true));
+
+        fromRoleAutomaton = fromRole.allowedActionsMatcher("index1::failures");
+        fromRolePredicate = Automatons.predicate(fromRoleAutomaton);
+        assertThat(fromRolePredicate.test(TransportSearchAction.TYPE.name()), is(true));
+
+        fromRoleAutomaton = fromRole.allowedActionsMatcher("metric");
+        fromRolePredicate = Automatons.predicate(fromRoleAutomaton);
+        assertThat(fromRolePredicate.test(TransportSearchAction.TYPE.name()), is(false));
+
+        fromRoleAutomaton = fromRole.allowedActionsMatcher("metric::failures");
+        fromRolePredicate = Automatons.predicate(fromRoleAutomaton);
+        assertThat(fromRolePredicate.test(TransportSearchAction.TYPE.name()), is(true));
+
+        fromRoleAutomaton = fromRole.allowedActionsMatcher("logs");
+        fromRolePredicate = Automatons.predicate(fromRoleAutomaton);
+        assertThat(fromRolePredicate.test(TransportSearchAction.TYPE.name()), is(true));
+
+        fromRoleAutomaton = fromRole.allowedActionsMatcher("logs::failures");
+        fromRolePredicate = Automatons.predicate(fromRoleAutomaton);
+        assertThat(fromRolePredicate.test(TransportSearchAction.TYPE.name()), is(false));
+
+        Role limitedByRole = Role.builder(EMPTY_RESTRICTED_INDICES, "limitedRole")
+            .add(IndexPrivilege.READ, "index1", "index2")
+            .add(IndexPrivilege.READ_FAILURE_STORE, "index3")
+            .build();
+        Automaton limitedByRoleAutomaton = limitedByRole.allowedActionsMatcher("index1");
+        Predicate<String> limitedByRolePredicated = Automatons.predicate(limitedByRoleAutomaton);
+        assertThat(limitedByRolePredicated.test(TransportSearchAction.TYPE.name()), is(true));
+
+        limitedByRoleAutomaton = limitedByRole.allowedActionsMatcher("index1");
+        limitedByRolePredicated = Automatons.predicate(limitedByRoleAutomaton);
+        assertThat(limitedByRolePredicated.test(TransportSearchAction.TYPE.name()), is(true));
+
+        limitedByRoleAutomaton = limitedByRole.allowedActionsMatcher("index1::failures");
+        limitedByRolePredicated = Automatons.predicate(limitedByRoleAutomaton);
+        assertThat(limitedByRolePredicated.test(TransportSearchAction.TYPE.name()), is(false));
+
+        limitedByRoleAutomaton = limitedByRole.allowedActionsMatcher("index3");
+        limitedByRolePredicated = Automatons.predicate(limitedByRoleAutomaton);
+        assertThat(limitedByRolePredicated.test(TransportSearchAction.TYPE.name()), is(false));
+
+        limitedByRoleAutomaton = limitedByRole.allowedActionsMatcher("index3::failures");
+        limitedByRolePredicated = Automatons.predicate(limitedByRoleAutomaton);
+        assertThat(limitedByRolePredicated.test(TransportSearchAction.TYPE.name()), is(true));
+
+        Role role;
+        if (randomBoolean()) {
+            role = limitedByRole.limitedBy(fromRole);
+        } else {
+            role = fromRole.limitedBy(limitedByRole);
+        }
+
+        Automaton roleAutomaton = role.allowedActionsMatcher("index1");
+        Predicate<String> rolePredicate = Automatons.predicate(roleAutomaton);
+        assertThat(rolePredicate.test(TransportSearchAction.TYPE.name()), is(true));
+
+        roleAutomaton = role.allowedActionsMatcher("index1::failures");
+        rolePredicate = Automatons.predicate(roleAutomaton);
+        assertThat(rolePredicate.test(TransportSearchAction.TYPE.name()), is(false));
+
+        roleAutomaton = role.allowedActionsMatcher("index3");
+        rolePredicate = Automatons.predicate(roleAutomaton);
+        assertThat(rolePredicate.test(TransportSearchAction.TYPE.name()), is(false));
+
+        roleAutomaton = role.allowedActionsMatcher("index3::failures");
+        rolePredicate = Automatons.predicate(roleAutomaton);
+        assertThat(rolePredicate.test(TransportSearchAction.TYPE.name()), is(true));
+
+        roleAutomaton = role.allowedActionsMatcher("metric");
+        rolePredicate = Automatons.predicate(roleAutomaton);
+        assertThat(rolePredicate.test(TransportSearchAction.TYPE.name()), is(false));
+
+        roleAutomaton = role.allowedActionsMatcher("metric::failures");
+        rolePredicate = Automatons.predicate(roleAutomaton);
+        assertThat(rolePredicate.test(TransportSearchAction.TYPE.name()), is(false));
+
+        roleAutomaton = role.allowedActionsMatcher("logs");
+        rolePredicate = Automatons.predicate(roleAutomaton);
+        assertThat(rolePredicate.test(TransportSearchAction.TYPE.name()), is(false));
+
+        roleAutomaton = role.allowedActionsMatcher("logs::failures");
+        rolePredicate = Automatons.predicate(roleAutomaton);
+        assertThat(rolePredicate.test(TransportSearchAction.TYPE.name()), is(false));
+    }
+
     public void testCheckClusterPrivilege() {
         Role fromRole = Role.builder(EMPTY_RESTRICTED_INDICES, "a-role")
             .cluster(Collections.singleton("manage_security"), Collections.emptyList())

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/user/InternalUsersTests.java

@@ -376,7 +376,7 @@ public class InternalUsersTests extends ESTestCase {
         final IndexAbstraction.ConcreteIndex index = new IndexAbstraction.ConcreteIndex(metadata);
         assertThat(
             "Role " + role + ", action " + action + " access to " + indexName,
-            role.allowedIndicesMatcher(action).test(index),
+            role.allowedIndicesMatcher(action).test(index, null),
             is(expectedValue)
         );
     }

x-pack/plugin/downsample/src/main/java/org/elasticsearch/xpack/downsample/TransportDownsampleAction.java

@@ -34,6 +34,7 @@ import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexMetadata.DownsampleTaskStatus;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.MetadataCreateIndexService;
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
@@ -209,7 +210,8 @@ public class TransportDownsampleAction extends AcknowledgedTransportMasterNodeAc
     ) {
         long startTime = client.threadPool().relativeTimeInMillis();
         String sourceIndexName = request.getSourceIndex();
-
+        IndexNameExpressionResolver.assertExpressionHasNullOrDataSelector(sourceIndexName);
+        IndexNameExpressionResolver.assertExpressionHasNullOrDataSelector(request.getTargetIndex());
         final IndicesAccessControl indicesAccessControl = threadContext.getTransient(AuthorizationServiceField.INDICES_PERMISSIONS_KEY);
         if (indicesAccessControl != null) {
             final IndicesAccessControl.IndexAccessControl indexPermissions = indicesAccessControl.getIndexPermissions(sourceIndexName);

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

@@ -2211,6 +2211,7 @@ public class Security extends Plugin
                     return FieldPredicate.ACCEPT_ALL;
                 }
                 assert indicesAccessControl.isGranted();
+                IndexNameExpressionResolver.assertExpressionHasNullOrDataSelector(index);
                 IndicesAccessControl.IndexAccessControl indexPermissions = indicesAccessControl.getIndexPermissions(index);
                 if (indexPermissions == null) {
                     return FieldPredicate.ACCEPT_ALL;

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

@@ -48,8 +48,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.function.Predicate;
+import java.util.function.BiPredicate;
-import java.util.function.Supplier;
 
 import static org.elasticsearch.xpack.core.security.authz.IndicesAndAliasesResolverField.NO_INDEX_PLACEHOLDER;
 
@@ -323,7 +322,8 @@ class IndicesAndAliasesResolver {
                     );
                 }
                 if (indicesOptions.expandWildcardExpressions()) {
-                    for (String authorizedIndex : authorizedIndices.all().get()) {
+                    IndexComponentSelector selector = IndexComponentSelector.getByKeyOrThrow(allIndicesPatternSelector);
+                    for (String authorizedIndex : authorizedIndices.all(selector)) {
                         if (IndexAbstractionResolver.isIndexVisible(
                             "*",
                             allIndicesPatternSelector,
@@ -352,7 +352,7 @@ class IndicesAndAliasesResolver {
                     split.getLocal(),
                     indicesOptions,
                     projectMetadata,
-                    authorizedIndices.all(),
+                    authorizedIndices::all,
                     authorizedIndices::check,
                     indicesRequest.includeDataStreams()
                 );
@@ -389,7 +389,7 @@ class IndicesAndAliasesResolver {
             if (aliasesRequest.expandAliasesWildcards()) {
                 List<String> aliases = replaceWildcardsWithAuthorizedAliases(
                     aliasesRequest.aliases(),
-                    loadAuthorizedAliases(authorizedIndices.all(), projectMetadata)
+                    loadAuthorizedAliases(authorizedIndices, projectMetadata)
                 );
                 aliasesRequest.replaceAliases(aliases.toArray(new String[aliases.size()]));
             }
@@ -431,8 +431,13 @@ class IndicesAndAliasesResolver {
      * request's concrete index is not in the list of authorized indices, then we need to look to
      * see if this can be authorized against an alias
      */
-    static String getPutMappingIndexOrAlias(PutMappingRequest request, Predicate<String> isAuthorized, ProjectMetadata projectMetadata) {
+    static String getPutMappingIndexOrAlias(
+        PutMappingRequest request,
+        BiPredicate<String, IndexComponentSelector> isAuthorized,
+        ProjectMetadata projectMetadata
+    ) {
         final String concreteIndexName = request.getConcreteIndex().getName();
+        assert IndexNameExpressionResolver.hasSelectorSuffix(concreteIndexName) == false : "selectors are not allowed in this context";
 
         // validate that the concrete index exists, otherwise there is no remapping that we could do
         final IndexAbstraction indexAbstraction = projectMetadata.getIndicesLookup().get(concreteIndexName);
@@ -447,7 +452,8 @@ class IndicesAndAliasesResolver {
                     + indexAbstraction.getType().getDisplayName()
                     + "], but a concrete index is expected"
             );
-        } else if (isAuthorized.test(concreteIndexName)) {
+            // we know this is implicit data access (as opposed to another selector) so the default selector check is correct
+        } else if (isAuthorized.test(concreteIndexName, IndexComponentSelector.DATA)) {
             // user is authorized to put mappings for this index
             resolvedAliasOrIndex = concreteIndexName;
         } else {
@@ -456,7 +462,12 @@ class IndicesAndAliasesResolver {
             Map<String, List<AliasMetadata>> foundAliases = projectMetadata.findAllAliases(new String[] { concreteIndexName });
             List<AliasMetadata> aliasMetadata = foundAliases.get(concreteIndexName);
             if (aliasMetadata != null) {
-                Optional<String> foundAlias = aliasMetadata.stream().map(AliasMetadata::alias).filter(isAuthorized).filter(aliasName -> {
+                Optional<String> foundAlias = aliasMetadata.stream().map(AliasMetadata::alias).filter(aliasName -> {
+                    // we know this is implicit data access (as opposed to another selector) so the default selector check is correct
+                    assert IndexNameExpressionResolver.hasSelectorSuffix(aliasName) == false : "selectors are not allowed in this context";
+                    if (false == isAuthorized.test(aliasName, IndexComponentSelector.DATA)) {
+                        return false;
+                    }
                     IndexAbstraction alias = projectMetadata.getIndicesLookup().get(aliasName);
                     List<Index> indices = alias.getIndices();
                     if (indices.size() == 1) {
@@ -476,10 +487,13 @@ class IndicesAndAliasesResolver {
         return resolvedAliasOrIndex;
     }
 
-    private static List<String> loadAuthorizedAliases(Supplier<Set<String>> authorizedIndices, ProjectMetadata projectMetadata) {
+    private static List<String> loadAuthorizedAliases(
+        AuthorizationEngine.AuthorizedIndices authorizedIndices,
+        ProjectMetadata projectMetadata
+    ) {
         List<String> authorizedAliases = new ArrayList<>();
         SortedMap<String, IndexAbstraction> existingAliases = projectMetadata.getIndicesLookup();
-        for (String authorizedIndex : authorizedIndices.get()) {
+        for (String authorizedIndex : authorizedIndices.all(IndexComponentSelector.DATA)) {
             IndexAbstraction indexAbstraction = existingAliases.get(authorizedIndex);
             if (indexAbstraction != null && indexAbstraction.getType() == IndexAbstraction.Type.ALIAS) {
                 authorizedAliases.add(authorizedIndex);
@@ -498,6 +512,7 @@ class IndicesAndAliasesResolver {
         }
 
         for (String aliasExpression : aliases) {
+            IndexNameExpressionResolver.assertExpressionHasNullOrDataSelector(aliasExpression);
             boolean include = true;
             if (aliasExpression.charAt(0) == '-') {
                 include = false;

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

@@ -31,6 +31,7 @@ import org.elasticsearch.action.search.TransportClearScrollAction;
 import org.elasticsearch.action.search.TransportClosePointInTimeAction;
 import org.elasticsearch.action.search.TransportMultiSearchAction;
 import org.elasticsearch.action.search.TransportSearchScrollAction;
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.action.termvectors.MultiTermVectorsAction;
 import org.elasticsearch.cluster.metadata.DataStream;
@@ -107,6 +108,7 @@ import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Set;
 import java.util.TreeSet;
+import java.util.function.BiPredicate;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -882,42 +884,76 @@ public class RBACEngine implements AuthorizationEngine {
             // TODO: can this be done smarter? I think there are usually more indices/aliases in the cluster then indices defined a roles?
             if (includeDataStreams) {
                 for (IndexAbstraction indexAbstraction : lookup.values()) {
-                    if (predicate.test(indexAbstraction)) {
+                    // failure indices are special: when accessed directly (not through ::failures on parent data stream) they are accessed
+                    // implicitly as data. However, authz to the parent data stream happens via the failures selector
+                    if (indexAbstraction.isFailureIndexOfDataStream()
+                        && predicate.test(indexAbstraction.getParentDataStream(), IndexComponentSelector.FAILURES)) {
+                        indicesAndAliases.add(indexAbstraction.getName());
+                        // we know this is a failure index, and it's authorized so no need to check further
+                        continue;
+                    }
+                    if (predicate.test(indexAbstraction, IndexComponentSelector.DATA)) {
                         indicesAndAliases.add(indexAbstraction.getName());
                         if (indexAbstraction.getType() == IndexAbstraction.Type.DATA_STREAM) {
                             // add data stream and its backing indices for any authorized data streams
                             for (Index index : indexAbstraction.getIndices()) {
                                 indicesAndAliases.add(index.getName());
                             }
-                            // TODO: We need to limit if a data stream's failure indices should return here.
-                            for (Index index : ((DataStream) indexAbstraction).getFailureIndices()) {
-                                indicesAndAliases.add(index.getName());
-                            }
                         }
                     }
                 }
             } else {
                 for (IndexAbstraction indexAbstraction : lookup.values()) {
-                    if (indexAbstraction.getType() != IndexAbstraction.Type.DATA_STREAM && predicate.test(indexAbstraction)) {
+                    if (indexAbstraction.getType() != IndexAbstraction.Type.DATA_STREAM
+                        // data check is correct, even for failure indices -- in this context, we treat them as regular indices, and
+                        // only include them if direct data access to them is granted (e.g., if a role has `read` over "*")
+                        && predicate.test(indexAbstraction, IndexComponentSelector.DATA)) {
                         indicesAndAliases.add(indexAbstraction.getName());
                     }
                 }
             }
             timeChecker.accept(indicesAndAliases);
             return indicesAndAliases;
-        }, name -> {
+        }, () -> {
+            // TODO handle time checking in a follow-up
+            Set<String> indicesAndAliases = new HashSet<>();
+            if (includeDataStreams) {
+                for (IndexAbstraction indexAbstraction : lookup.values()) {
+                    if (indexAbstraction.getType() == IndexAbstraction.Type.DATA_STREAM
+                        && predicate.test(indexAbstraction, IndexComponentSelector.FAILURES)) {
+                        indicesAndAliases.add(indexAbstraction.getName());
+                        // add data stream and its backing failure indices for any authorized data streams
+                        for (Index index : ((DataStream) indexAbstraction).getFailureIndices()) {
+                            indicesAndAliases.add(index.getName());
+                        }
+                    }
+                }
+            }
+            return indicesAndAliases;
+        }, (name, selector) -> {
             final IndexAbstraction indexAbstraction = lookup.get(name);
             if (indexAbstraction == null) {
                 // test access (by name) to a resource that does not currently exist
                 // the action handler must handle the case of accessing resources that do not exist
-                return predicate.test(name, null);
+                return predicate.test(name, null, selector);
-            } else {
+            }
             // We check the parent data stream first if there is one. For testing requested indices, this is most likely
             // more efficient than checking the index name first because we recommend grant privileges over data stream
             // instead of backing indices.
-                return (indexAbstraction.getParentDataStream() != null && predicate.test(indexAbstraction.getParentDataStream()))
+            if (indexAbstraction.getParentDataStream() != null) {
-                    || predicate.test(indexAbstraction);
+                if (indexAbstraction.isFailureIndexOfDataStream()) {
+                    // access to failure indices is authorized via failures-based selectors on the parent data stream _not_ via data
+                    // ones
+                    if (predicate.test(indexAbstraction.getParentDataStream(), IndexComponentSelector.FAILURES)) {
+                        return true;
                     }
+                } else if (IndexComponentSelector.DATA.equals(selector) || selector == null) {
+                    if (predicate.test(indexAbstraction.getParentDataStream(), IndexComponentSelector.DATA)) {
+                        return true;
+                    }
+                } // we don't support granting access to a backing index with a failure selector via the parent data stream
+            }
+            return predicate.test(indexAbstraction, selector);
         });
     }
 
@@ -1042,23 +1078,32 @@ public class RBACEngine implements AuthorizationEngine {
     }
 
     static final class AuthorizedIndices implements AuthorizationEngine.AuthorizedIndices {
+        private final CachedSupplier<Set<String>> authorizedAndAvailableDataResources;
+        private final CachedSupplier<Set<String>> authorizedAndAvailableFailuresResources;
+        private final BiPredicate<String, IndexComponentSelector> isAuthorizedPredicate;
 
-        private final CachedSupplier<Set<String>> allAuthorizedAndAvailableSupplier;
+        AuthorizedIndices(
-        private final Predicate<String> isAuthorizedPredicate;
+            Supplier<Set<String>> authorizedAndAvailableDataResources,
-
+            Supplier<Set<String>> authorizedAndAvailableFailuresResources,
-        AuthorizedIndices(Supplier<Set<String>> allAuthorizedAndAvailableSupplier, Predicate<String> isAuthorizedPredicate) {
+            BiPredicate<String, IndexComponentSelector> isAuthorizedPredicate
-            this.allAuthorizedAndAvailableSupplier = CachedSupplier.wrap(allAuthorizedAndAvailableSupplier);
+        ) {
+            this.authorizedAndAvailableDataResources = CachedSupplier.wrap(authorizedAndAvailableDataResources);
+            this.authorizedAndAvailableFailuresResources = CachedSupplier.wrap(authorizedAndAvailableFailuresResources);
             this.isAuthorizedPredicate = Objects.requireNonNull(isAuthorizedPredicate);
         }
 
         @Override
-        public Supplier<Set<String>> all() {
+        public Set<String> all(IndexComponentSelector selector) {
-            return allAuthorizedAndAvailableSupplier;
+            Objects.requireNonNull(selector, "must specify a selector to get authorized indices");
+            return IndexComponentSelector.FAILURES.equals(selector)
+                ? authorizedAndAvailableFailuresResources.get()
+                : authorizedAndAvailableDataResources.get();
         }
 
         @Override
-        public boolean check(String name) {
+        public boolean check(String name, IndexComponentSelector selector) {
-            return this.isAuthorizedPredicate.test(name);
+            Objects.requireNonNull(selector, "must specify a selector for authorization check");
+            return isAuthorizedPredicate.test(name, selector);
         }
     }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/rest/action/privilege/RestGetBuiltinPrivilegesAction.java

@@ -42,7 +42,7 @@ public class RestGetBuiltinPrivilegesAction extends SecurityBaseRestHandler {
 
     private static final Logger logger = LogManager.getLogger(RestGetBuiltinPrivilegesAction.class);
     // TODO remove this once we can update docs tests again
-    private static final Set<String> FAILURE_STORE_PRIVILEGES_TO_EXCLUDE = Set.of("read_failure_store");
+    private static final Set<String> FAILURE_STORE_PRIVILEGES_TO_EXCLUDE = Set.of("read_failure_store", "manage_failure_store");
     private final GetBuiltinPrivilegesResponseTranslator responseTranslator;
 
     public RestGetBuiltinPrivilegesAction(

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

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.security.authz;
 
 import org.elasticsearch.action.admin.indices.resolve.ResolveIndexAction;
 import org.elasticsearch.action.search.TransportSearchAction;
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.cluster.metadata.AliasMetadata;
 import org.elasticsearch.cluster.metadata.DataStream;
@@ -55,7 +56,7 @@ public class AuthorizedIndicesTests extends ESTestCase {
             Metadata.EMPTY_METADATA.getProject().getIndicesLookup(),
             () -> ignore -> {}
         );
-        assertTrue(authorizedIndices.all().get().isEmpty());
+        assertTrue(authorizedIndices.all(IndexComponentSelector.DATA).isEmpty());
     }
 
     public void testAuthorizedIndicesUserWithSomeRoles() {
@@ -115,15 +116,15 @@ public class AuthorizedIndicesTests extends ESTestCase {
             metadata.getProject().getIndicesLookup(),
             () -> ignore -> {}
         );
-        assertThat(authorizedIndices.all().get(), containsInAnyOrder("a1", "a2", "aaaaaa", "b", "ab"));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), containsInAnyOrder("a1", "a2", "aaaaaa", "b", "ab"));
-        assertThat(authorizedIndices.all().get(), not(contains("bbbbb")));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains("bbbbb")));
-        assertThat(authorizedIndices.check("bbbbb"), is(false));
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.DATA), is(false));
-        assertThat(authorizedIndices.all().get(), not(contains("ba")));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains("ba")));
-        assertThat(authorizedIndices.check("ba"), is(false));
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.DATA), is(false));
-        assertThat(authorizedIndices.all().get(), not(contains(internalSecurityIndex)));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains(internalSecurityIndex)));
-        assertThat(authorizedIndices.check(internalSecurityIndex), is(false));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.DATA), is(false));
-        assertThat(authorizedIndices.all().get(), not(contains(SecuritySystemIndices.SECURITY_MAIN_ALIAS)));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains(SecuritySystemIndices.SECURITY_MAIN_ALIAS)));
-        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS), is(false));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.DATA), is(false));
     }
 
     public void testAuthorizedIndicesUserWithSomeRolesEmptyMetadata() {
@@ -134,7 +135,7 @@ public class AuthorizedIndicesTests extends ESTestCase {
             Metadata.EMPTY_METADATA.getProject().getIndicesLookup(),
             () -> ignore -> {}
         );
-        assertTrue(authorizedIndices.all().get().isEmpty());
+        assertTrue(authorizedIndices.all(IndexComponentSelector.DATA).isEmpty());
     }
 
     public void testSecurityIndicesAreRemovedFromRegularUser() {
@@ -145,7 +146,7 @@ public class AuthorizedIndicesTests extends ESTestCase {
             Metadata.EMPTY_METADATA.getProject().getIndicesLookup(),
             () -> ignore -> {}
         );
-        assertTrue(authorizedIndices.all().get().isEmpty());
+        assertTrue(authorizedIndices.all(IndexComponentSelector.DATA).isEmpty());
     }
 
     public void testSecurityIndicesAreRestrictedForDefaultRole() {
@@ -177,13 +178,13 @@ public class AuthorizedIndicesTests extends ESTestCase {
             metadata.getProject().getIndicesLookup(),
             () -> ignore -> {}
         );
-        assertThat(authorizedIndices.all().get(), containsInAnyOrder("an-index", "another-index"));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), containsInAnyOrder("an-index", "another-index"));
-        assertThat(authorizedIndices.check("an-index"), is(true));
+        assertThat(authorizedIndices.check("an-index", IndexComponentSelector.DATA), is(true));
-        assertThat(authorizedIndices.check("another-index"), is(true));
+        assertThat(authorizedIndices.check("another-index", IndexComponentSelector.DATA), is(true));
-        assertThat(authorizedIndices.all().get(), not(contains(internalSecurityIndex)));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains(internalSecurityIndex)));
-        assertThat(authorizedIndices.check(internalSecurityIndex), is(false));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.DATA), is(false));
-        assertThat(authorizedIndices.all().get(), not(contains(SecuritySystemIndices.SECURITY_MAIN_ALIAS)));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains(SecuritySystemIndices.SECURITY_MAIN_ALIAS)));
-        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS), is(false));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.DATA), is(false));
     }
 
     public void testSecurityIndicesAreNotRemovedFromUnrestrictedRole() {
@@ -216,7 +217,7 @@ public class AuthorizedIndicesTests extends ESTestCase {
             () -> ignore -> {}
         );
         assertThat(
-            authorizedIndices.all().get(),
+            authorizedIndices.all(IndexComponentSelector.DATA),
             containsInAnyOrder("an-index", "another-index", SecuritySystemIndices.SECURITY_MAIN_ALIAS, internalSecurityIndex)
         );
 
@@ -227,7 +228,7 @@ public class AuthorizedIndicesTests extends ESTestCase {
             () -> ignore -> {}
         );
         assertThat(
-            authorizedIndicesSuperUser.all().get(),
+            authorizedIndicesSuperUser.all(IndexComponentSelector.DATA),
             containsInAnyOrder("an-index", "another-index", SecuritySystemIndices.SECURITY_MAIN_ALIAS, internalSecurityIndex)
         );
     }
@@ -274,10 +275,92 @@ public class AuthorizedIndicesTests extends ESTestCase {
                 true
             )
             .put(new IndexMetadata.Builder(backingIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(DataStreamTestHelper.newInstance("adatastream1", List.of(new Index(backingIndex, "_na_"))))
+            .build();
+        final PlainActionFuture<Role> future = new PlainActionFuture<>();
+        final Set<RoleDescriptor> descriptors = Sets.newHashSet(aStarRole, bRole);
+        CompositeRolesStore.buildRoleFromDescriptors(
+            descriptors,
+            new FieldPermissionsCache(Settings.EMPTY),
+            null,
+            RESTRICTED_INDICES,
+            future
+        );
+        Role roles = future.actionGet();
+        AuthorizedIndices authorizedIndices = RBACEngine.resolveAuthorizedIndicesFromRole(
+            roles,
+            getRequestInfo(TransportSearchAction.TYPE.name()),
+            metadata.getProject().getIndicesLookup(),
+            () -> ignore -> {}
+        );
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), containsInAnyOrder("a1", "a2", "aaaaaa", "b", "ab"));
+        for (String resource : List.of("a1", "a2", "aaaaaa", "b", "ab")) {
+            assertThat(authorizedIndices.check(resource, IndexComponentSelector.DATA), is(true));
+        }
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains("bbbbb")));
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains("ba")));
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.DATA), is(false));
+        // due to context, datastreams are excluded from wildcard expansion
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains("adatastream1")));
+        // but they are authorized when explicitly tested (they are not "unavailable" for the Security filter)
+        assertThat(authorizedIndices.check("adatastream1", IndexComponentSelector.DATA), is(true));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains(internalSecurityIndex)));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains(SecuritySystemIndices.SECURITY_MAIN_ALIAS)));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.DATA), is(false));
+    }
+
+    public void testDataStreamsAreNotIncludedInAuthorizedIndicesWithFailuresSelectorAndAllPrivilege() {
+        assumeTrue("requires failure store", DataStream.isFailureStoreFeatureFlagEnabled());
+        RoleDescriptor aStarRole = new RoleDescriptor(
+            "a_star",
+            null,
+            new IndicesPrivileges[] { IndicesPrivileges.builder().indices("a*").privileges("all").build() },
+            null
+        );
+        RoleDescriptor bRole = new RoleDescriptor(
+            "b",
+            null,
+            new IndicesPrivileges[] { IndicesPrivileges.builder().indices("b").privileges("READ").build() },
+            null
+        );
+        Settings indexSettings = Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, IndexVersion.current()).build();
+        final String internalSecurityIndex = randomFrom(
+            TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_6,
+            TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_7
+        );
+        String backingIndex = DataStream.getDefaultBackingIndexName("adatastream1", 1);
+        String failureIndex = DataStream.getDefaultFailureStoreName("adatastream1", 1, 1);
+        Metadata metadata = Metadata.builder()
+            .put(new IndexMetadata.Builder("a1").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("a2").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("aaaaaa").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("bbbbb").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(
+                new IndexMetadata.Builder("b").settings(indexSettings)
+                    .numberOfShards(1)
+                    .numberOfReplicas(0)
+                    .putAlias(new AliasMetadata.Builder("ab").build())
+                    .putAlias(new AliasMetadata.Builder("ba").build())
+                    .build(),
+                true
+            )
+            .put(
+                new IndexMetadata.Builder(internalSecurityIndex).settings(indexSettings)
+                    .numberOfShards(1)
+                    .numberOfReplicas(0)
+                    .putAlias(new AliasMetadata.Builder(SecuritySystemIndices.SECURITY_MAIN_ALIAS).build())
+                    .build(),
+                true
+            )
+            .put(new IndexMetadata.Builder(backingIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder(failureIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
             .put(
                 DataStreamTestHelper.newInstance(
                     "adatastream1",
-                    List.of(new Index(DataStream.getDefaultBackingIndexName("adatastream1", 1), "_na_"))
+                    List.of(new Index(backingIndex, "_na_")),
+                    List.of(new Index(failureIndex, "_na_"))
                 )
             )
             .build();
@@ -297,22 +380,279 @@ public class AuthorizedIndicesTests extends ESTestCase {
             metadata.getProject().getIndicesLookup(),
             () -> ignore -> {}
         );
-        assertThat(authorizedIndices.all().get(), containsInAnyOrder("a1", "a2", "aaaaaa", "b", "ab"));
+        assertAuthorizedFor(authorizedIndices, IndexComponentSelector.DATA, "a1", "a2", "aaaaaa", "b", "ab");
-        for (String resource : List.of("a1", "a2", "aaaaaa", "b", "ab")) {
+        assertAuthorizedFor(authorizedIndices, IndexComponentSelector.FAILURES);
-            assertThat(authorizedIndices.check(resource), is(true));
+
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.FAILURES), is(false));
+
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.FAILURES), is(false));
+
+        // data are authorized when explicitly tested (they are not "unavailable" for the Security filter)
+        assertThat(authorizedIndices.check("adatastream1", IndexComponentSelector.DATA), is(true));
+        assertThat(authorizedIndices.check("adatastream1", IndexComponentSelector.FAILURES), is(true));
+
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.FAILURES), is(false));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.FAILURES), is(false));
     }
-        assertThat(authorizedIndices.all().get(), not(contains("bbbbb")));
+
-        assertThat(authorizedIndices.check("bbbbb"), is(false));
+    public void testDataStreamsAreIncludedInAuthorizedIndicesWithFailuresSelectorAndAllPrivilege() {
-        assertThat(authorizedIndices.all().get(), not(contains("ba")));
+        assumeTrue("requires failure store", DataStream.isFailureStoreFeatureFlagEnabled());
-        assertThat(authorizedIndices.check("ba"), is(false));
+        RoleDescriptor aStarRole = new RoleDescriptor(
-        // due to context, datastreams are excluded from wildcard expansion
+            "a_star",
-        assertThat(authorizedIndices.all().get(), not(contains("adatastream1")));
+            null,
-        // but they are authorized when explicitly tested (they are not "unavailable" for the Security filter)
+            new IndicesPrivileges[] { IndicesPrivileges.builder().indices("a*").privileges("all").build() },
-        assertThat(authorizedIndices.check("adatastream1"), is(true));
+            null
-        assertThat(authorizedIndices.all().get(), not(contains(internalSecurityIndex)));
+        );
-        assertThat(authorizedIndices.check(internalSecurityIndex), is(false));
+        RoleDescriptor bRole = new RoleDescriptor(
-        assertThat(authorizedIndices.all().get(), not(contains(SecuritySystemIndices.SECURITY_MAIN_ALIAS)));
+            "b",
-        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS), is(false));
+            null,
+            new IndicesPrivileges[] { IndicesPrivileges.builder().indices("b").privileges("READ").build() },
+            null
+        );
+        Settings indexSettings = Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, IndexVersion.current()).build();
+        final String internalSecurityIndex = randomFrom(
+            TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_6,
+            TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_7
+        );
+        String backingIndex = DataStream.getDefaultBackingIndexName("adatastream1", 1);
+        String failureIndex = DataStream.getDefaultFailureStoreName("adatastream1", 1, 1);
+        Metadata metadata = Metadata.builder()
+            .put(new IndexMetadata.Builder("a1").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("a2").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("aaaaaa").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("bbbbb").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(
+                new IndexMetadata.Builder("b").settings(indexSettings)
+                    .numberOfShards(1)
+                    .numberOfReplicas(0)
+                    .putAlias(new AliasMetadata.Builder("ab").build())
+                    .putAlias(new AliasMetadata.Builder("ba").build())
+                    .build(),
+                true
+            )
+            .put(
+                new IndexMetadata.Builder(internalSecurityIndex).settings(indexSettings)
+                    .numberOfShards(1)
+                    .numberOfReplicas(0)
+                    .putAlias(new AliasMetadata.Builder(SecuritySystemIndices.SECURITY_MAIN_ALIAS).build())
+                    .build(),
+                true
+            )
+            .put(new IndexMetadata.Builder(backingIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder(failureIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(
+                DataStreamTestHelper.newInstance(
+                    "adatastream1",
+                    List.of(new Index(backingIndex, "_na_")),
+                    List.of(new Index(failureIndex, "_na_"))
+                )
+            )
+            .build();
+        final PlainActionFuture<Role> future = new PlainActionFuture<>();
+        final Set<RoleDescriptor> descriptors = Sets.newHashSet(aStarRole, bRole);
+        CompositeRolesStore.buildRoleFromDescriptors(
+            descriptors,
+            new FieldPermissionsCache(Settings.EMPTY),
+            null,
+            RESTRICTED_INDICES,
+            future
+        );
+        Role roles = future.actionGet();
+        TransportRequest request = new ResolveIndexAction.Request(new String[] { "a*" });
+        AuthorizationEngine.RequestInfo requestInfo = getRequestInfo(request, TransportSearchAction.TYPE.name());
+        AuthorizedIndices authorizedIndices = RBACEngine.resolveAuthorizedIndicesFromRole(
+            roles,
+            requestInfo,
+            metadata.getProject().getIndicesLookup(),
+            () -> ignore -> {}
+        );
+        assertAuthorizedFor(
+            authorizedIndices,
+            IndexComponentSelector.DATA,
+            "a1",
+            "a2",
+            "aaaaaa",
+            "b",
+            "ab",
+            "adatastream1",
+            backingIndex,
+            failureIndex
+        );
+        assertAuthorizedFor(authorizedIndices, IndexComponentSelector.FAILURES, "adatastream1", failureIndex);
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains("bbbbb")));
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.FAILURES), is(false));
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.FAILURES), is(false));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.FAILURES), is(false));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.FAILURES), is(false));
+    }
+
+    public void testDataStreamsAreIncludedInAuthorizedIndicesWithFailuresSelector() {
+        assumeTrue("requires failure store", DataStream.isFailureStoreFeatureFlagEnabled());
+        RoleDescriptor aReadFailuresStarRole = new RoleDescriptor(
+            "a_read_failure_store",
+            null,
+            new IndicesPrivileges[] { IndicesPrivileges.builder().indices("a*").privileges("read_failure_store").build() },
+            null
+        );
+        RoleDescriptor aReadRole = new RoleDescriptor(
+            "a_read",
+            null,
+            new IndicesPrivileges[] { IndicesPrivileges.builder().indices("a*").privileges("read").build() },
+            null
+        );
+        Settings indexSettings = Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, IndexVersion.current()).build();
+        final String internalSecurityIndex = randomFrom(
+            TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_6,
+            TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_7
+        );
+        String backingIndex = DataStream.getDefaultBackingIndexName("adatastream1", 1);
+        String failureIndex = DataStream.getDefaultFailureStoreName("adatastream1", 1, 1);
+        Metadata metadata = Metadata.builder()
+            .put(new IndexMetadata.Builder("a1").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("a2").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("aaaaaa").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("bbbbb").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(
+                new IndexMetadata.Builder(internalSecurityIndex).settings(indexSettings)
+                    .numberOfShards(1)
+                    .numberOfReplicas(0)
+                    .putAlias(new AliasMetadata.Builder(SecuritySystemIndices.SECURITY_MAIN_ALIAS).build())
+                    .build(),
+                true
+            )
+            .put(new IndexMetadata.Builder(backingIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder(failureIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(
+                DataStreamTestHelper.newInstance(
+                    "adatastream1",
+                    List.of(new Index(backingIndex, "_na_")),
+                    List.of(new Index(failureIndex, "_na_"))
+                )
+            )
+            .build();
+        final PlainActionFuture<Role> future = new PlainActionFuture<>();
+        final Set<RoleDescriptor> descriptors = Sets.newHashSet(aReadFailuresStarRole, aReadRole);
+        CompositeRolesStore.buildRoleFromDescriptors(
+            descriptors,
+            new FieldPermissionsCache(Settings.EMPTY),
+            null,
+            RESTRICTED_INDICES,
+            future
+        );
+        Role roles = future.actionGet();
+        TransportRequest request = new ResolveIndexAction.Request(new String[] { "a*" });
+        AuthorizationEngine.RequestInfo requestInfo = getRequestInfo(request, TransportSearchAction.TYPE.name());
+        AuthorizedIndices authorizedIndices = RBACEngine.resolveAuthorizedIndicesFromRole(
+            roles,
+            requestInfo,
+            metadata.getProject().getIndicesLookup(),
+            () -> ignore -> {}
+        );
+        assertAuthorizedFor(
+            authorizedIndices,
+            IndexComponentSelector.DATA,
+            "a1",
+            "a2",
+            "aaaaaa",
+            "adatastream1",
+            backingIndex,
+            failureIndex
+        );
+        assertAuthorizedFor(authorizedIndices, IndexComponentSelector.FAILURES, "adatastream1", failureIndex);
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains("bbbbb")));
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.FAILURES), is(false));
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.FAILURES), is(false));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.FAILURES), is(false));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.FAILURES), is(false));
+    }
+
+    public void testDataStreamsAreNotIncludedInAuthorizedIndicesWithFailuresSelector() {
+        assumeTrue("requires failure store", DataStream.isFailureStoreFeatureFlagEnabled());
+        RoleDescriptor aReadFailuresStarRole = new RoleDescriptor(
+            "a_read_failure_store",
+            null,
+            new IndicesPrivileges[] { IndicesPrivileges.builder().indices("a*").privileges("read_failure_store").build() },
+            null
+        );
+        RoleDescriptor aReadRole = new RoleDescriptor(
+            "a_read",
+            null,
+            new IndicesPrivileges[] { IndicesPrivileges.builder().indices("a*").privileges("read").build() },
+            null
+        );
+        Settings indexSettings = Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, IndexVersion.current()).build();
+        final String internalSecurityIndex = randomFrom(
+            TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_6,
+            TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_7
+        );
+        String backingIndex = DataStream.getDefaultBackingIndexName("adatastream1", 1);
+        String failureIndex = DataStream.getDefaultFailureStoreName("adatastream1", 1, 1);
+        Metadata metadata = Metadata.builder()
+            .put(new IndexMetadata.Builder("a1").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("a2").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("aaaaaa").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder("bbbbb").settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(
+                new IndexMetadata.Builder(internalSecurityIndex).settings(indexSettings)
+                    .numberOfShards(1)
+                    .numberOfReplicas(0)
+                    .putAlias(new AliasMetadata.Builder(SecuritySystemIndices.SECURITY_MAIN_ALIAS).build())
+                    .build(),
+                true
+            )
+            .put(new IndexMetadata.Builder(backingIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(new IndexMetadata.Builder(failureIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
+            .put(
+                DataStreamTestHelper.newInstance(
+                    "adatastream1",
+                    List.of(new Index(backingIndex, "_na_")),
+                    List.of(new Index(failureIndex, "_na_"))
+                )
+            )
+            .build();
+        final PlainActionFuture<Role> future = new PlainActionFuture<>();
+        final Set<RoleDescriptor> descriptors = Sets.newHashSet(aReadFailuresStarRole, aReadRole);
+        CompositeRolesStore.buildRoleFromDescriptors(
+            descriptors,
+            new FieldPermissionsCache(Settings.EMPTY),
+            null,
+            RESTRICTED_INDICES,
+            future
+        );
+        Role roles = future.actionGet();
+        AuthorizedIndices authorizedIndices = RBACEngine.resolveAuthorizedIndicesFromRole(
+            roles,
+            getRequestInfo(TransportSearchAction.TYPE.name()),
+            metadata.getProject().getIndicesLookup(),
+            () -> ignore -> {}
+        );
+        assertAuthorizedFor(authorizedIndices, IndexComponentSelector.DATA, "a1", "a2", "aaaaaa");
+        assertAuthorizedFor(authorizedIndices, IndexComponentSelector.FAILURES);
+
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.FAILURES), is(false));
+
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.FAILURES), is(false));
+
+        // data are authorized when explicitly tested (they are not "unavailable" for the Security filter)
+        assertThat(authorizedIndices.check("adatastream1", IndexComponentSelector.DATA), is(true));
+        assertThat(authorizedIndices.check("adatastream1", IndexComponentSelector.FAILURES), is(true));
+
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.FAILURES), is(false));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.FAILURES), is(false));
     }
 
     public void testDataStreamsAreIncludedInAuthorizedIndices() {
@@ -357,12 +697,7 @@ public class AuthorizedIndicesTests extends ESTestCase {
                 true
             )
             .put(new IndexMetadata.Builder(backingIndex).settings(indexSettings).numberOfShards(1).numberOfReplicas(0).build(), true)
-            .put(
+            .put(DataStreamTestHelper.newInstance("adatastream1", List.of(new Index(backingIndex, "_na_"))))
-                DataStreamTestHelper.newInstance(
-                    "adatastream1",
-                    List.of(new Index(DataStream.getDefaultBackingIndexName("adatastream1", 1), "_na_"))
-                )
-            )
             .build();
         final PlainActionFuture<Role> future = new PlainActionFuture<>();
         final Set<RoleDescriptor> descriptors = Sets.newHashSet(aStarRole, bRole);
@@ -382,15 +717,18 @@ public class AuthorizedIndicesTests extends ESTestCase {
             metadata.getProject().getIndicesLookup(),
             () -> ignore -> {}
         );
-        assertThat(authorizedIndices.all().get(), containsInAnyOrder("a1", "a2", "aaaaaa", "b", "ab", "adatastream1", backingIndex));
+        assertThat(
-        assertThat(authorizedIndices.all().get(), not(contains("bbbbb")));
+            authorizedIndices.all(IndexComponentSelector.DATA),
-        assertThat(authorizedIndices.check("bbbbb"), is(false));
+            containsInAnyOrder("a1", "a2", "aaaaaa", "b", "ab", "adatastream1", backingIndex)
-        assertThat(authorizedIndices.all().get(), not(contains("ba")));
+        );
-        assertThat(authorizedIndices.check("ba"), is(false));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains("bbbbb")));
-        assertThat(authorizedIndices.all().get(), not(contains(internalSecurityIndex)));
+        assertThat(authorizedIndices.check("bbbbb", IndexComponentSelector.DATA), is(false));
-        assertThat(authorizedIndices.check(internalSecurityIndex), is(false));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains("ba")));
-        assertThat(authorizedIndices.all().get(), not(contains(SecuritySystemIndices.SECURITY_MAIN_ALIAS)));
+        assertThat(authorizedIndices.check("ba", IndexComponentSelector.DATA), is(false));
-        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS), is(false));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains(internalSecurityIndex)));
+        assertThat(authorizedIndices.check(internalSecurityIndex, IndexComponentSelector.DATA), is(false));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(contains(SecuritySystemIndices.SECURITY_MAIN_ALIAS)));
+        assertThat(authorizedIndices.check(SecuritySystemIndices.SECURITY_MAIN_ALIAS, IndexComponentSelector.DATA), is(false));
     }
 
     public static AuthorizationEngine.RequestInfo getRequestInfo(String action) {
@@ -410,4 +748,15 @@ public class AuthorizedIndicesTests extends ESTestCase {
             null
         );
     }
+
+    private static void assertAuthorizedFor(
+        AuthorizedIndices authorizedIndices,
+        IndexComponentSelector selector,
+        String... expectedIndices
+    ) {
+        assertThat(authorizedIndices.all(selector), containsInAnyOrder(expectedIndices));
+        for (String resource : expectedIndices) {
+            assertThat(authorizedIndices.check(resource, selector), is(true));
+        }
+    }
 }

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

@@ -30,6 +30,7 @@ import org.elasticsearch.action.search.SearchShardsRequest;
 import org.elasticsearch.action.search.TransportMultiSearchAction;
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.search.TransportSearchShardsAction;
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.termvectors.MultiTermVectorsRequest;
@@ -1941,7 +1942,11 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         String index = "logs-00003"; // write index
         PutMappingRequest request = new PutMappingRequest(Strings.EMPTY_ARRAY).setConcreteIndex(new Index(index, UUIDs.base64UUID()));
         assert projectMetadata.getIndicesLookup().get("logs-alias").getIndices().size() == 3;
-        String putMappingIndexOrAlias = IndicesAndAliasesResolver.getPutMappingIndexOrAlias(request, "logs-alias"::equals, projectMetadata);
+        String putMappingIndexOrAlias = IndicesAndAliasesResolver.getPutMappingIndexOrAlias(
+            request,
+            (i, s) -> i.equals("logs-alias"),
+            projectMetadata
+        );
         String message = "user is authorized to access `logs-alias` and the put mapping request is for a write index"
             + "so this should have returned the alias name";
         assertEquals(message, "logs-alias", putMappingIndexOrAlias);
@@ -1951,7 +1956,11 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         String index = "logs-00002"; // read index
         PutMappingRequest request = new PutMappingRequest(Strings.EMPTY_ARRAY).setConcreteIndex(new Index(index, UUIDs.base64UUID()));
         assert projectMetadata.getIndicesLookup().get("logs-alias").getIndices().size() == 3;
-        String putMappingIndexOrAlias = IndicesAndAliasesResolver.getPutMappingIndexOrAlias(request, "logs-alias"::equals, projectMetadata);
+        String putMappingIndexOrAlias = IndicesAndAliasesResolver.getPutMappingIndexOrAlias(
+            request,
+            (i, s) -> i.equals("logs-alias"),
+            projectMetadata
+        );
         String message = "user is authorized to access `logs-alias` and the put mapping request is for a read index"
             + "so this should have returned the concrete index as fallback";
         assertEquals(message, index, putMappingIndexOrAlias);
@@ -2225,14 +2234,14 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         List<String> dataStreams = List.of("logs-foo", "logs-foobar");
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
         for (String dsName : dataStreams) {
-            assertThat(authorizedIndices.all().get(), hasItem(dsName));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(dsName));
-            assertThat(authorizedIndices.check(dsName), is(true));
+            assertThat(authorizedIndices.check(dsName, IndexComponentSelector.DATA), is(true));
             DataStream dataStream = projectMetadata.dataStreams().get(dsName);
-            assertThat(authorizedIndices.all().get(), hasItem(dsName));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(dsName));
-            assertThat(authorizedIndices.check(dsName), is(true));
+            assertThat(authorizedIndices.check(dsName, IndexComponentSelector.DATA), is(true));
             for (Index i : dataStream.getIndices()) {
-                assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+                assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-                assertThat(authorizedIndices.check(i.getName()), is(true));
+                assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
             }
         }
 
@@ -2264,14 +2273,14 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         // data streams and their backing indices should _not_ be in the authorized list since the backing indices
         // do not match the requested name
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
-        assertThat(authorizedIndices.all().get(), hasItem(dataStreamName));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(dataStreamName));
-        assertThat(authorizedIndices.check(dataStreamName), is(true));
+        assertThat(authorizedIndices.check(dataStreamName, IndexComponentSelector.DATA), is(true));
         DataStream dataStream = projectMetadata.dataStreams().get(dataStreamName);
-        assertThat(authorizedIndices.all().get(), hasItem(dataStreamName));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(dataStreamName));
-        assertThat(authorizedIndices.check(dataStreamName), is(true));
+        assertThat(authorizedIndices.check(dataStreamName, IndexComponentSelector.DATA), is(true));
         for (Index i : dataStream.getIndices()) {
-            assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-            assertThat(authorizedIndices.check(i.getName()), is(true));
+            assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
         }
 
         // neither data streams nor their backing indices will be in the resolved list since the backing indices do not match the
@@ -2299,11 +2308,11 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, TransportSearchAction.TYPE.name(), request);
         for (String dsName : expectedDataStreams) {
             DataStream dataStream = projectMetadata.dataStreams().get(dsName);
-            assertThat(authorizedIndices.all().get(), hasItem(dsName));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(dsName));
-            assertThat(authorizedIndices.check(dsName), is(true));
+            assertThat(authorizedIndices.check(dsName, IndexComponentSelector.DATA), is(true));
             for (Index i : dataStream.getIndices()) {
-                assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+                assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-                assertThat(authorizedIndices.check(i.getName()), is(true));
+                assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
             }
         }
 
@@ -2340,11 +2349,11 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
 
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, TransportSearchAction.TYPE.name(), request);
         // data streams and their backing indices should be in the authorized list
-        assertThat(authorizedIndices.all().get(), hasItem(dataStreamName));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(dataStreamName));
-        assertThat(authorizedIndices.check(dataStreamName), is(true));
+        assertThat(authorizedIndices.check(dataStreamName, IndexComponentSelector.DATA), is(true));
         for (Index i : dataStream.getIndices()) {
-            assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-            assertThat(authorizedIndices.check(i.getName()), is(true));
+            assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
         }
 
         ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(
@@ -2373,15 +2382,15 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, TransportSearchAction.TYPE.name(), request);
         for (String dsName : expectedDataStreams) {
             DataStream dataStream = projectMetadata.dataStreams().get(dsName);
-            assertThat(authorizedIndices.all().get(), hasItem(dsName));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(dsName));
-            assertThat(authorizedIndices.check(dsName), is(true));
+            assertThat(authorizedIndices.check(dsName, IndexComponentSelector.DATA), is(true));
             for (Index i : dataStream.getIndices()) {
-                assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+                assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-                assertThat(authorizedIndices.check(i.getName()), is(true));
+                assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
             }
             for (Index i : dataStream.getFailureIndices()) {
-                assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+                assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-                assertThat(authorizedIndices.check(i.getName()), is(true));
+                assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
             }
         }
 
@@ -2412,16 +2421,16 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         // data streams and their backing indices should _not_ be in the authorized list since the backing indices
         // did not match the requested pattern and the request does not support data streams
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
-        assertThat(authorizedIndices.all().get(), hasItem(dataStreamName));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(dataStreamName));
-        assertThat(authorizedIndices.check(dataStreamName), is(true));
+        assertThat(authorizedIndices.check(dataStreamName, IndexComponentSelector.DATA), is(true));
         DataStream dataStream = projectMetadata.dataStreams().get(dataStreamName);
         for (Index i : dataStream.getIndices()) {
-            assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-            assertThat(authorizedIndices.check(i.getName()), is(true));
+            assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
         }
         for (Index i : dataStream.getFailureIndices()) {
-            assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-            assertThat(authorizedIndices.check(i.getName()), is(true));
+            assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
         }
 
         // neither data streams nor their backing indices will be in the resolved list since the request does not support data streams
@@ -2452,19 +2461,19 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         // data streams should _not_ be in the authorized list but their backing indices that matched both the requested pattern
         // and the authorized pattern should be in the list
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
-        assertThat(authorizedIndices.all().get(), not(hasItem("logs-foobar")));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(hasItem("logs-foobar")));
-        assertThat(authorizedIndices.check("logs-foobar"), is(false));
+        assertThat(authorizedIndices.check("logs-foobar", IndexComponentSelector.DATA), is(false));
         DataStream dataStream = projectMetadata.dataStreams().get("logs-foobar");
-        assertThat(authorizedIndices.all().get(), not(hasItem(indexName)));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(hasItem(indexName)));
         // request pattern is subset of the authorized pattern, but be aware that patterns are never passed to #check in main code
-        assertThat(authorizedIndices.check(indexName), is(true));
+        assertThat(authorizedIndices.check(indexName, IndexComponentSelector.DATA), is(true));
         for (Index i : dataStream.getIndices()) {
-            assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-            assertThat(authorizedIndices.check(i.getName()), is(true));
+            assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
         }
         for (Index i : dataStream.getFailureIndices()) {
-            assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-            assertThat(authorizedIndices.check(i.getName()), is(true));
+            assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
         }
 
         // only the backing indices will be in the resolved list since the request does not support data streams
@@ -2492,14 +2501,14 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         // data streams should _not_ be in the authorized list but a single backing index that matched the requested pattern
         // and the authorized name should be in the list
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
-        assertThat(authorizedIndices.all().get(), not(hasItem("logs-foobar")));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(hasItem("logs-foobar")));
-        assertThat(authorizedIndices.check("logs-foobar"), is(false));
+        assertThat(authorizedIndices.check("logs-foobar", IndexComponentSelector.DATA), is(false));
 
         String expectedIndex = failureStore
             ? DataStream.getDefaultFailureStoreName("logs-foobar", 1, System.currentTimeMillis())
             : DataStream.getDefaultBackingIndexName("logs-foobar", 1);
-        assertThat(authorizedIndices.all().get(), hasItem(expectedIndex));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(expectedIndex));
-        assertThat(authorizedIndices.check(expectedIndex), is(true));
+        assertThat(authorizedIndices.check(expectedIndex, IndexComponentSelector.DATA), is(true));
 
         // only the single backing index will be in the resolved list since the request does not support data streams
         // but one of the backing indices matched the requested pattern
@@ -2524,19 +2533,19 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         // data streams should _not_ be in the authorized list but their backing indices that matched both the requested pattern
         // and the authorized pattern should be in the list
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
-        assertThat(authorizedIndices.all().get(), not(hasItem("logs-foobar")));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(hasItem("logs-foobar")));
-        assertThat(authorizedIndices.check("logs-foobar"), is(false));
+        assertThat(authorizedIndices.check("logs-foobar", IndexComponentSelector.DATA), is(false));
         DataStream dataStream = projectMetadata.dataStreams().get("logs-foobar");
-        assertThat(authorizedIndices.all().get(), not(hasItem(indexName)));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(hasItem(indexName)));
         // request pattern is subset of the authorized pattern, but be aware that patterns are never passed to #check in main code
-        assertThat(authorizedIndices.check(indexName), is(true));
+        assertThat(authorizedIndices.check(indexName, IndexComponentSelector.DATA), is(true));
         for (Index i : dataStream.getIndices()) {
-            assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-            assertThat(authorizedIndices.check(i.getName()), is(true));
+            assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
         }
         for (Index i : dataStream.getFailureIndices()) {
-            assertThat(authorizedIndices.all().get(), hasItem(i.getName()));
+            assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(i.getName()));
-            assertThat(authorizedIndices.check(i.getName()), is(true));
+            assertThat(authorizedIndices.check(i.getName(), IndexComponentSelector.DATA), is(true));
         }
 
         // only the backing indices will be in the resolved list since the request does not support data streams
@@ -2567,10 +2576,10 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
             ? DataStream.getDefaultFailureStoreName("logs-foobar", 1, System.currentTimeMillis())
             : DataStream.getDefaultBackingIndexName("logs-foobar", 1);
         final AuthorizedIndices authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
-        assertThat(authorizedIndices.all().get(), not(hasItem("logs-foobar")));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), not(hasItem("logs-foobar")));
-        assertThat(authorizedIndices.check("logs-foobar"), is(false));
+        assertThat(authorizedIndices.check("logs-foobar", IndexComponentSelector.DATA), is(false));
-        assertThat(authorizedIndices.all().get(), hasItem(expectedIndex));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(expectedIndex));
-        assertThat(authorizedIndices.check(expectedIndex), is(true));
+        assertThat(authorizedIndices.check(expectedIndex, IndexComponentSelector.DATA), is(true));
 
         // only the single backing index will be in the resolved list since the request does not support data streams
         // but one of the backing indices matched the requested pattern

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

@@ -19,6 +19,7 @@ import org.elasticsearch.action.delete.TransportDeleteAction;
 import org.elasticsearch.action.index.TransportIndexAction;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.TransportSearchAction;
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.client.internal.Client;
@@ -1444,14 +1445,14 @@ public class RBACEngineTests extends ESTestCase {
             lookup,
             () -> ignore -> {}
         );
-        assertThat(authorizedIndices.all().get(), hasItem(dataStreamName));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA), hasItem(dataStreamName));
-        assertThat(authorizedIndices.check(dataStreamName), is(true));
+        assertThat(authorizedIndices.check(dataStreamName, IndexComponentSelector.DATA), is(true));
         assertThat(
-            authorizedIndices.all().get(),
+            authorizedIndices.all(IndexComponentSelector.DATA),
-            hasItems(backingIndices.stream().map(im -> im.getIndex().getName()).collect(Collectors.toList()).toArray(Strings.EMPTY_ARRAY))
+            hasItems(backingIndices.stream().map(im -> im.getIndex().getName()).toList().toArray(Strings.EMPTY_ARRAY))
         );
         for (String index : backingIndices.stream().map(im -> im.getIndex().getName()).toList()) {
-            assertThat(authorizedIndices.check(index), is(true));
+            assertThat(authorizedIndices.check(index, IndexComponentSelector.DATA), is(true));
         }
     }
 
@@ -1487,7 +1488,8 @@ public class RBACEngineTests extends ESTestCase {
             lookup,
             () -> ignore -> {}
         );
-        assertThat(authorizedIndices.all().get().isEmpty(), is(true));
+        assertThat(authorizedIndices.all(IndexComponentSelector.DATA).isEmpty(), is(true));
+        assertThat(authorizedIndices.all(IndexComponentSelector.FAILURES).isEmpty(), is(true));
     }
 
     public void testNoInfiniteRecursionForRBACAuthorizationInfoHashCode() {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/accesscontrol/IndicesPermissionTests.java

@@ -11,6 +11,7 @@ import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.admin.indices.mapping.put.TransportAutoPutMappingAction;
 import org.elasticsearch.action.admin.indices.mapping.put.TransportPutMappingAction;
 import org.elasticsearch.action.search.TransportSearchAction;
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.cluster.metadata.AliasMetadata;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
@@ -185,6 +186,269 @@ public class IndicesPermissionTests extends ESTestCase {
 
     }
 
+    public void testAuthorizeDataStreamAccessWithFailuresSelector() {
+        Metadata.Builder builder = Metadata.builder();
+        String dataStreamName = randomAlphaOfLength(6);
+        int numBackingIndices = randomIntBetween(1, 3);
+        List<IndexMetadata> backingIndices = new ArrayList<>();
+        for (int backingIndexNumber = 1; backingIndexNumber <= numBackingIndices; backingIndexNumber++) {
+            backingIndices.add(createBackingIndexMetadata(DataStream.getDefaultBackingIndexName(dataStreamName, backingIndexNumber)));
+        }
+        DataStream ds = DataStreamTestHelper.newInstance(
+            dataStreamName,
+            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList())
+        );
+        builder.put(ds);
+        for (IndexMetadata index : backingIndices) {
+            builder.put(index, false);
+        }
+        var metadata = builder.build().getProject();
+        FieldPermissionsCache fieldPermissionsCache = new FieldPermissionsCache(Settings.EMPTY);
+
+        for (var privilege : List.of(IndexPrivilege.ALL, IndexPrivilege.READ)) {
+            Role role = Role.builder(RESTRICTED_INDICES, "_role")
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    privilege,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .build();
+            IndicesAccessControl permissions = role.authorize(
+                TransportSearchAction.TYPE.name(),
+                Sets.newHashSet(randomFrom(dataStreamName, dataStreamName + "::data")),
+                metadata,
+                fieldPermissionsCache
+            );
+            assertThat("for privilege " + privilege, permissions.isGranted(), is(true));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(dataStreamName + "::failures"), is(false));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(dataStreamName), is(true));
+        }
+
+        for (var privilege : List.of(IndexPrivilege.ALL, IndexPrivilege.READ_FAILURE_STORE)) {
+            Role role = Role.builder(RESTRICTED_INDICES, "_role")
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    privilege,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .build();
+            IndicesAccessControl permissions = role.authorize(
+                TransportSearchAction.TYPE.name(),
+                Sets.newHashSet(dataStreamName + "::failures"),
+                metadata,
+                fieldPermissionsCache
+            );
+            assertThat("for privilege " + privilege, permissions.isGranted(), is(true));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(dataStreamName + "::failures"), is(true));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(dataStreamName), is(false));
+        }
+
+        for (var privilege : List.of(IndexPrivilege.ALL, IndexPrivilege.READ_FAILURE_STORE)) {
+            Role role = Role.builder(RESTRICTED_INDICES, "_role")
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    privilege,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .build();
+            IndicesAccessControl permissions = role.authorize(
+                TransportSearchAction.TYPE.name(),
+                Sets.newHashSet(dataStreamName + "::failures"),
+                metadata,
+                fieldPermissionsCache
+            );
+
+            assertThat("for privilege " + privilege, permissions.isGranted(), is(true));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(dataStreamName + "::failures"), is(true));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(dataStreamName), is(false));
+        }
+
+        {
+            Role role = Role.builder(RESTRICTED_INDICES, "_role")
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    IndexPrivilege.READ,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    IndexPrivilege.READ_FAILURE_STORE,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .build();
+            IndicesAccessControl permissions = role.authorize(
+                TransportSearchAction.TYPE.name(),
+                Sets.newHashSet(randomFrom(dataStreamName, dataStreamName + "::data"), dataStreamName + "::failures"),
+                metadata,
+                fieldPermissionsCache
+            );
+            assertThat(permissions.isGranted(), is(true));
+            assertThat(permissions.hasIndexPermissions(dataStreamName + "::failures"), is(true));
+            assertThat(permissions.hasIndexPermissions(dataStreamName), is(true));
+        }
+
+        {
+            Role role = Role.builder(RESTRICTED_INDICES, "_role")
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    IndexPrivilege.ALL,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .build();
+            IndicesAccessControl permissions = role.authorize(
+                TransportSearchAction.TYPE.name(),
+                Sets.newHashSet(randomFrom(dataStreamName, dataStreamName + "::data"), dataStreamName + "::failures"),
+                metadata,
+                fieldPermissionsCache
+            );
+            assertThat("for privilege " + IndexPrivilege.ALL, permissions.isGranted(), is(true));
+            assertThat("for privilege " + IndexPrivilege.ALL, permissions.hasIndexPermissions(dataStreamName + "::failures"), is(true));
+            assertThat("for privilege " + IndexPrivilege.ALL, permissions.hasIndexPermissions(dataStreamName), is(true));
+        }
+        {
+            Role role = Role.builder(RESTRICTED_INDICES, "_role")
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    IndexPrivilege.READ_FAILURE_STORE,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .build();
+            IndicesAccessControl permissions = role.authorize(
+                TransportSearchAction.TYPE.name(),
+                Sets.newHashSet(randomFrom(dataStreamName, dataStreamName + "::data"), dataStreamName + "::failures"),
+                metadata,
+                fieldPermissionsCache
+            );
+            assertThat("for privilege " + IndexPrivilege.READ_FAILURE_STORE, permissions.isGranted(), is(false));
+            assertThat(
+                "for privilege " + IndexPrivilege.READ_FAILURE_STORE,
+                permissions.hasIndexPermissions(dataStreamName + "::failures"),
+                is(true)
+            );
+            assertThat("for privilege " + IndexPrivilege.READ_FAILURE_STORE, permissions.hasIndexPermissions(dataStreamName), is(false));
+        }
+
+        {
+            Role role = Role.builder(RESTRICTED_INDICES, "_role")
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    IndexPrivilege.READ,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .build();
+            IndicesAccessControl permissions = role.authorize(
+                TransportSearchAction.TYPE.name(),
+                Sets.newHashSet(randomFrom(dataStreamName, dataStreamName + "::data"), dataStreamName + "::failures"),
+                metadata,
+                fieldPermissionsCache
+            );
+            assertThat("for privilege " + IndexPrivilege.READ, permissions.isGranted(), is(false));
+            assertThat("for privilege " + IndexPrivilege.READ, permissions.hasIndexPermissions(dataStreamName + "::failures"), is(false));
+            assertThat("for privilege " + IndexPrivilege.READ, permissions.hasIndexPermissions(dataStreamName), is(true));
+        }
+    }
+
+    public void testAuthorizeDataStreamFailureIndices() {
+        Metadata.Builder builder = Metadata.builder();
+        String dataStreamName = randomAlphaOfLength(6);
+        int numBackingIndices = randomIntBetween(1, 3);
+        List<IndexMetadata> backingIndices = new ArrayList<>();
+        for (int backingIndexNumber = 1; backingIndexNumber <= numBackingIndices; backingIndexNumber++) {
+            backingIndices.add(createBackingIndexMetadata(DataStream.getDefaultBackingIndexName(dataStreamName, backingIndexNumber)));
+        }
+        List<IndexMetadata> failureIndices = new ArrayList<>();
+        int numFailureIndices = randomIntBetween(1, 3);
+        for (int failureIndexNumber = 1; failureIndexNumber <= numFailureIndices; failureIndexNumber++) {
+            failureIndices.add(createBackingIndexMetadata(DataStream.getDefaultFailureStoreName(dataStreamName, failureIndexNumber, 1L)));
+        }
+        DataStream ds = DataStreamTestHelper.newInstance(
+            dataStreamName,
+            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()),
+            failureIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList())
+        );
+        builder.put(ds);
+        for (IndexMetadata index : backingIndices) {
+            builder.put(index, false);
+        }
+        for (IndexMetadata index : failureIndices) {
+            builder.put(index, false);
+        }
+        var metadata = builder.build().getProject();
+        FieldPermissionsCache fieldPermissionsCache = new FieldPermissionsCache(Settings.EMPTY);
+
+        for (var privilege : List.of(IndexPrivilege.READ)) {
+            Role role = Role.builder(RESTRICTED_INDICES, "_role")
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    privilege,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .build();
+            String failureIndex = randomFrom(failureIndices).getIndex().getName();
+            IndicesAccessControl permissions = role.authorize(
+                TransportSearchAction.TYPE.name(),
+                Sets.newHashSet(failureIndex),
+                metadata,
+                fieldPermissionsCache
+            );
+            assertThat("for privilege " + privilege, permissions.isGranted(), is(false));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(failureIndex), is(false));
+
+            String dataIndex = randomFrom(backingIndices).getIndex().getName();
+            permissions = role.authorize(TransportSearchAction.TYPE.name(), Sets.newHashSet(dataIndex), metadata, fieldPermissionsCache);
+
+            assertThat("for privilege " + privilege, permissions.isGranted(), is(true));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(dataIndex), is(true));
+        }
+
+        for (var privilege : List.of(IndexPrivilege.READ_FAILURE_STORE)) {
+            Role role = Role.builder(RESTRICTED_INDICES, "_role")
+                .add(
+                    new FieldPermissions(fieldPermissionDef(null, null)),
+                    null,
+                    privilege,
+                    randomBoolean(),
+                    randomFrom(dataStreamName, dataStreamName + "*")
+                )
+                .build();
+
+            String failureIndex = randomFrom(failureIndices).getIndex().getName();
+            IndicesAccessControl permissions = role.authorize(
+                TransportSearchAction.TYPE.name(),
+                Sets.newHashSet(failureIndex),
+                metadata,
+                fieldPermissionsCache
+            );
+
+            assertThat("for privilege " + privilege, permissions.isGranted(), is(true));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(failureIndex), is(true));
+
+            String dataIndex = randomFrom(backingIndices).getIndex().getName();
+            permissions = role.authorize(TransportSearchAction.TYPE.name(), Sets.newHashSet(dataIndex), metadata, fieldPermissionsCache);
+
+            assertThat("for privilege " + privilege, permissions.isGranted(), is(false));
+            assertThat("for privilege " + privilege, permissions.hasIndexPermissions(dataIndex), is(false));
+        }
+    }
+
     public void testAuthorizeMultipleGroupsMixedDls() {
         IndexMetadata.Builder imbBuilder = IndexMetadata.builder("_index")
             .settings(indexSettings(IndexVersion.current(), 1, 1))
@@ -705,14 +969,15 @@ public class IndicesPermissionTests extends ESTestCase {
         );
         IndicesPermission.IsResourceAuthorizedPredicate predicate = new IndicesPermission.IsResourceAuthorizedPredicate(
             StringMatcher.of("other"),
+            StringMatcher.of(),
             StringMatcher.of(dataStreamName, backingIndex.getName(), concreteIndex.getName(), alias.getName())
         );
         assertThat(predicate.test(dataStream), is(false));
         // test authorization for a missing resource with the datastream's name
-        assertThat(predicate.test(dataStream.getName(), null), is(true));
+        assertThat(predicate.test(dataStream.getName(), null, IndexComponentSelector.DATA), is(true));
         assertThat(predicate.test(backingIndex), is(false));
         // test authorization for a missing resource with the backing index's name
-        assertThat(predicate.test(backingIndex.getName(), null), is(true));
+        assertThat(predicate.test(backingIndex.getName(), null, IndexComponentSelector.DATA), is(true));
         assertThat(predicate.test(concreteIndex), is(true));
         assertThat(predicate.test(alias), is(true));
     }
@@ -720,10 +985,12 @@ public class IndicesPermissionTests extends ESTestCase {
     public void testResourceAuthorizedPredicateAnd() {
         IndicesPermission.IsResourceAuthorizedPredicate predicate1 = new IndicesPermission.IsResourceAuthorizedPredicate(
             StringMatcher.of("c", "a"),
+            StringMatcher.of(),
             StringMatcher.of("b", "d")
         );
         IndicesPermission.IsResourceAuthorizedPredicate predicate2 = new IndicesPermission.IsResourceAuthorizedPredicate(
             StringMatcher.of("c", "b"),
+            StringMatcher.of(),
             StringMatcher.of("a", "d")
         );
         Metadata.Builder mb = Metadata.builder(
@@ -754,6 +1021,75 @@ public class IndicesPermissionTests extends ESTestCase {
         assertThat(predicate.test(concreteIndexD), is(true));
     }
 
+    public void testResourceAuthorizedPredicateAndWithFailures() {
+        IndicesPermission.IsResourceAuthorizedPredicate predicate1 = new IndicesPermission.IsResourceAuthorizedPredicate(
+            StringMatcher.of("c", "a"),
+            StringMatcher.of("e", "f"),
+            StringMatcher.of("b", "d")
+        );
+        IndicesPermission.IsResourceAuthorizedPredicate predicate2 = new IndicesPermission.IsResourceAuthorizedPredicate(
+            StringMatcher.of("c", "b"),
+            StringMatcher.of("a", "f", "g"),
+            StringMatcher.of("a", "d")
+        );
+        Metadata.Builder mb = Metadata.builder(
+            DataStreamTestHelper.getClusterStateWithDataStreams(
+                List.of(
+                    Tuple.tuple("a", 1),
+                    Tuple.tuple("b", 1),
+                    Tuple.tuple("c", 1),
+                    Tuple.tuple("d", 1),
+                    Tuple.tuple("e", 1),
+                    Tuple.tuple("f", 1)
+                ),
+                List.of(),
+                Instant.now().toEpochMilli(),
+                builder().build(),
+                1
+            ).getMetadata()
+        );
+        DataStream dataStreamA = mb.dataStream("a");
+        DataStream dataStreamB = mb.dataStream("b");
+        DataStream dataStreamC = mb.dataStream("c");
+        DataStream dataStreamD = mb.dataStream("d");
+        DataStream dataStreamE = mb.dataStream("e");
+        DataStream dataStreamF = mb.dataStream("f");
+        IndexAbstraction concreteIndexA = concreteIndexAbstraction("a");
+        IndexAbstraction concreteIndexB = concreteIndexAbstraction("b");
+        IndexAbstraction concreteIndexC = concreteIndexAbstraction("c");
+        IndexAbstraction concreteIndexD = concreteIndexAbstraction("d");
+        IndexAbstraction concreteIndexE = concreteIndexAbstraction("e");
+        IndexAbstraction concreteIndexF = concreteIndexAbstraction("f");
+        IndicesPermission.IsResourceAuthorizedPredicate predicate = predicate1.and(predicate2);
+        assertThat(predicate.test(dataStreamA), is(false));
+        assertThat(predicate.test(dataStreamB), is(false));
+        assertThat(predicate.test(dataStreamC), is(true));
+        assertThat(predicate.test(dataStreamD), is(false));
+        assertThat(predicate.test(dataStreamE), is(false));
+        assertThat(predicate.test(dataStreamF), is(false));
+
+        assertThat(predicate.test(dataStreamA, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(dataStreamB, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(dataStreamC, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(dataStreamD, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(dataStreamE, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(dataStreamF, IndexComponentSelector.FAILURES), is(true));
+
+        assertThat(predicate.test(concreteIndexA), is(true));
+        assertThat(predicate.test(concreteIndexB), is(true));
+        assertThat(predicate.test(concreteIndexC), is(true));
+        assertThat(predicate.test(concreteIndexD), is(true));
+        assertThat(predicate.test(concreteIndexE), is(false));
+        assertThat(predicate.test(concreteIndexF), is(false));
+
+        assertThat(predicate.test(concreteIndexA, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(concreteIndexB, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(concreteIndexC, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(concreteIndexD, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(concreteIndexE, IndexComponentSelector.FAILURES), is(false));
+        assertThat(predicate.test(concreteIndexF, IndexComponentSelector.FAILURES), is(true));
+    }
+
     private static IndexAbstraction concreteIndexAbstraction(String name) {
         return new IndexAbstraction.ConcreteIndex(
             IndexMetadata.builder(name).settings(indexSettings(IndexVersion.current(), 1, 0)).build()