Failure Store Access Authorization (#123986)

This PR implements authorization logic for failure store access. It
builds on #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

Author: n1v0lg

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 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 void loadAuthorizedIndices(
     ) {
         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 static IndexComponentSelector getByKey(String key) {
         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 @@ default boolean isDataStreamRelated() {
         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 DataStream getParentDataStream() {
             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.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.function.Predicate;
-import java.util.function.Supplier;
+import java.util.function.BiPredicate;
+import java.util.function.Function;
 
 public class IndexAbstractionResolver {
 
@@ -37,8 +37,8 @@ public List<String> resolveIndexAbstractions(
         Iterable<String> indices,
         IndicesOptions indicesOptions,
         ProjectMetadata projectMetadata,
-        Supplier<Set<String>> allAuthorizedAndAvailable,
-        Predicate<String> isAuthorized,
+        Function<IndexComponentSelector, Set<String>> allAuthorizedAndAvailableBySelector,
+        BiPredicate<String, IndexComponentSelector> isAuthorized,
         boolean includeDataStreams
     ) {
         List<String> finalIndices = new ArrayList<>();
@@ -64,14 +64,15 @@ public List<String> resolveIndexAbstractions(
                 );
             }
             indexAbstraction = expressionAndSelector.v1();
+            IndexComponentSelector selector = IndexComponentSelector.getByKeyOrThrow(selectorString);
 
             // we always need to check for date math expressions
             indexAbstraction = IndexNameExpressionResolver.resolveDateMathExpression(indexAbstraction);
 
             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 @@ && isIndexVisible(
                 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.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 static boolean hasSelectorSuffix(String expression) {
         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 static String combineSelectorExpression(String baseExpression, @Nullable
             : (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.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 void testResolveIndexAbstractions() {
 
     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 void testIsNetNewSystemIndexVisible() {
             indexAbstractionResolver = new IndexAbstractionResolver(indexNameExpressionResolver);
 
             // this covers the GET * case -- with system access, you can see everything
-            assertThat(isIndexVisible("other", "*"), is(true));
-            assertThat(isIndexVisible(".foo", "*"), is(true));
-            assertThat(isIndexVisible(".bar", "*"), is(true));
+            assertThat(isIndexVisible("other", null), is(true));
+            assertThat(isIndexVisible(".foo", null), 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(".foo", "*", noHiddenNoAliases), is(false));
-            assertThat(isIndexVisible(".bar", "*", noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible("other", null, noHiddenNoAliases), is(true));
+            assertThat(isIndexVisible(".foo", null, noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible(".bar", null, noHiddenNoAliases), is(false));
         }
 
         {
@@ -311,14 +308,14 @@ public void testIsNetNewSystemIndexVisible() {
             indexAbstractionResolver = new IndexAbstractionResolver(indexNameExpressionResolver);
 
             // this covers the GET * case -- without system access, you can't see everything
-            assertThat(isIndexVisible("other", "*"), is(true));
-            assertThat(isIndexVisible(".foo", "*"), is(false));
-            assertThat(isIndexVisible(".bar", "*"), is(false));
+            assertThat(isIndexVisible("other", null), is(true));
+            assertThat(isIndexVisible(".foo", null), 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(".foo", "*", noHiddenNoAliases), is(false));
-            assertThat(isIndexVisible(".bar", "*", noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible("other", null, noHiddenNoAliases), is(true));
+            assertThat(isIndexVisible(".foo", null, noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible(".bar", null, noHiddenNoAliases), is(false));
         }
 
         {
@@ -333,14 +330,14 @@ public void testIsNetNewSystemIndexVisible() {
             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(".foo", "*"), is(false));
-            assertThat(isIndexVisible(".bar", "*"), is(false));
+            assertThat(isIndexVisible("other", null), is(true));
+            assertThat(isIndexVisible(".foo", null), 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(".foo", "*", noHiddenNoAliases), is(false));
-            assertThat(isIndexVisible(".bar", "*", noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible("other", null, noHiddenNoAliases), is(true));
+            assertThat(isIndexVisible(".foo", null, noHiddenNoAliases), is(false));
+            assertThat(isIndexVisible(".bar", null, noHiddenNoAliases), is(false));
         }
     }
 
@@ -366,8 +363,15 @@ private List<String> resolveAbstractionsSelectorAllowed(List<String> expressions
         return resolveAbstractions(expressions, IndicesOptions.strictExpandOpen(), defaultMask);
     }
 
-    private List<String> resolveAbstractions(List<String> expressions, IndicesOptions indicesOptions, Supplier<Set<String>> mask) {
-        return indexAbstractionResolver.resolveIndexAbstractions(expressions, indicesOptions, projectMetadata, mask, (idx) -> true, true);
+    private List<String> resolveAbstractions(List<String> expressions, IndicesOptions indicesOptions, Set<String> mask) {
+        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.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.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 @@ default AuthorizationInfo getAuthenticatedUserAuthorizationInfo() {
     }
 
     /**
-     * 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 ActionRequestValidationException validate(ActionRequestValidationExceptio
                 && application.length == 0) {
                 validationException = addValidationError("must specify at least one privilege", validationException);
             }
+            if (index != null) {
+                // no need to validate failure-store related constraints if it's not enabled
+                if (DataStream.isFailureStoreFeatureFlagEnabled()) {
+                    for (RoleDescriptor.IndicesPrivileges indexPrivilege : index) {
+                        if (indexPrivilege.getIndices() != null
+                            && Arrays.stream(indexPrivilege.getIndices())
+                                // best effort prevent users from attempting to check failure selectors
+                                .anyMatch(idx -> IndexNameExpressionResolver.hasSelector(idx, IndexComponentSelector.FAILURES))) {
+                            validationException = addValidationError(
+                                // TODO adjust message once HasPrivileges check supports checking failure store privileges
+                                "failures selector is not supported in index patterns",
+                                validationException
+                            );
+                        }
+                        if (indexPrivilege.getPrivileges() != null
+                            && Arrays.stream(indexPrivilege.getPrivileges())
+                                .anyMatch(p -> "read_failure_store".equals(p) || "manage_failure_store".equals(p))) {
+                            validationException = addValidationError(
+                                "checking failure store privileges is not supported",
+                                validationException
+                            );
+                        }
+                    }
+                }
+            }
             return validationException;
         }