[Failure Store] Authorization denial messages (#125757)

This PR makes authorization denial messages account for privileges that
grant access to the failure store. This is a minimal implementation that
only displays information around failure store privileges for requests
that include concrete names with the `::failures` selector. This avoids
including irrelevant information in regular non-failures requests. We
can improve on this in follow ups. 

Closes: ES-11158

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

@@ -482,10 +482,13 @@ public final class IndexPrivilege extends Privilege {
      * @see Privilege#sortByAccessLevel
      */
     public static Collection<String> findPrivilegesThatGrant(String action) {
+        return findPrivilegesThatGrant(action, p -> p.getSelectorPredicate().test(IndexComponentSelector.DATA));
+    }
+
+    public static Collection<String> findPrivilegesThatGrant(String action, Predicate<IndexPrivilege> preCondition) {
         return VALUES.entrySet()
             .stream()
-            // Only include privileges that grant data access; failures access is handled separately in authorization failure messages
-            .filter(e -> e.getValue().selectorPredicate.test(IndexComponentSelector.DATA))
+            .filter(e -> preCondition.test(e.getValue()))
             .filter(e -> e.getValue().predicate.test(action))
             .map(Map.Entry::getKey)
             .toList();

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

@@ -23,6 +23,7 @@ import org.elasticsearch.xpack.core.transform.action.GetCheckpointAction;
 import java.util.Collection;
 import java.util.List;
 import java.util.Set;
+import java.util.function.Predicate;
 
 import static org.elasticsearch.xpack.core.security.authz.privilege.IndexPrivilege.findPrivilegesThatGrant;
 import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -68,6 +69,16 @@ public class IndexPrivilegeTests extends ESTestCase {
             equalTo(List.of("monitor", "cross_cluster_replication", "manage", "all"))
         );
         assertThat(findPrivilegesThatGrant(RefreshAction.NAME), equalTo(List.of("maintenance", "manage", "all")));
+
+        if (DataStream.isFailureStoreFeatureFlagEnabled()) {
+            Predicate<IndexPrivilege> failuresOnly = p -> p.getSelectorPredicate() == IndexComponentSelectorPredicate.FAILURES;
+            assertThat(findPrivilegesThatGrant(TransportSearchAction.TYPE.name(), failuresOnly), equalTo(List.of("read_failure_store")));
+            assertThat(findPrivilegesThatGrant(TransportIndexAction.NAME, failuresOnly), equalTo(List.of()));
+            assertThat(findPrivilegesThatGrant(TransportUpdateAction.NAME, failuresOnly), equalTo(List.of()));
+            assertThat(findPrivilegesThatGrant(TransportDeleteAction.NAME, failuresOnly), equalTo(List.of()));
+            assertThat(findPrivilegesThatGrant(IndicesStatsAction.NAME, failuresOnly), equalTo(List.of("manage_failure_store")));
+            assertThat(findPrivilegesThatGrant(RefreshAction.NAME, failuresOnly), equalTo(List.of("manage_failure_store")));
+        }
     }
 
     public void testGet() {

x-pack/plugin/security/qa/security-trial/src/javaRestTest/java/org/elasticsearch/xpack/security/failurestore/FailureStoreSecurityRestIT.java

@@ -33,6 +33,7 @@ import org.elasticsearch.test.rest.ObjectPath;
 import org.elasticsearch.xcontent.json.JsonXContent;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.security.SecurityOnTrialLicenseRestTestCase;
+import org.hamcrest.Matcher;
 import org.junit.Before;
 import org.junit.ClassRule;
 
@@ -1092,6 +1093,12 @@ public class FailureStoreSecurityRestIT extends ESRestTestCase {
                     case FAILURE_STORE_ACCESS, BACKING_INDEX_DATA_ACCESS, BACKING_INDEX_FAILURE_ACCESS, FAILURE_INDEX_DATA_ACCESS,
                         FAILURE_INDEX_FAILURE_ACCESS:
                         expectThrows(user, request, 403);
+                        // also check authz message
+                        expectThrowsUnauthorized(
+                            user,
+                            request,
+                            containsString("this action is granted by the index privileges [read,all]")
+                        );
                         break;
                     default:
                         fail("must cover user: " + user);
@@ -1277,6 +1284,15 @@ public class FailureStoreSecurityRestIT extends ESRestTestCase {
                     case DATA_ACCESS, STAR_READ_ONLY_ACCESS, BACKING_INDEX_DATA_ACCESS, BACKING_INDEX_FAILURE_ACCESS,
                         FAILURE_INDEX_FAILURE_ACCESS, FAILURE_INDEX_DATA_ACCESS:
                         expectThrows(user, request, 403);
+                        // also check authz message
+                        expectThrowsUnauthorized(
+                            user,
+                            request,
+                            containsString(
+                                "this action is granted by the index privileges [read,all] for data access, "
+                                    + "or by [read_failure_store] for access with the [::failures] selector"
+                            )
+                        );
                         break;
                     case ADMIN_USER, FAILURE_STORE_ACCESS, BOTH_ACCESS:
                         expectSearch(user, request, failuresDocId);
@@ -2255,6 +2271,12 @@ public class FailureStoreSecurityRestIT extends ESRestTestCase {
         assertThat(ex.getResponse().getStatusLine().getStatusCode(), equalTo(statusCode));
     }
 
+    private void expectThrowsUnauthorized(String user, Search search, Matcher<String> errorMatcher) {
+        ResponseException ex = expectThrows(ResponseException.class, () -> performRequestMaybeUsingApiKey(user, search.toSearchRequest()));
+        assertThat(ex.getResponse().getStatusLine().getStatusCode(), equalTo(403));
+        assertThat(ex.getMessage(), errorMatcher);
+    }
+
     private void expectThrows(String user, Search search, int statusCode) {
         expectThrows(() -> performRequest(user, search.toSearchRequest()), statusCode);
         expectThrows(() -> performRequest(user, search.toAsyncSearchRequest()), statusCode);

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

@@ -7,6 +7,8 @@
 
 package org.elasticsearch.xpack.security.authz;
 
+import org.elasticsearch.action.support.IndexComponentSelector;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.Nullable;
@@ -24,6 +26,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.SortedSet;
+import java.util.function.Predicate;
 
 import static org.elasticsearch.common.Strings.collectionToCommaDelimitedString;
 import static org.elasticsearch.xpack.security.audit.logfile.LoggingAuditTrail.PRINCIPAL_ROLES_FIELD_NAME;
@@ -90,22 +93,47 @@ public interface AuthorizationDenialMessages {
 
             if (ClusterPrivilegeResolver.isClusterAction(action)) {
                 final Collection<String> privileges = findClusterPrivilegesThatGrant(authentication, action, request);
-                if (privileges != null && privileges.size() > 0) {
+                if (privileges != null && false == privileges.isEmpty()) {
                     message = message
                         + ", this action is granted by the cluster privileges ["
                         + collectionToCommaDelimitedString(privileges)
                         + "]";
                 }
             } else if (isIndexAction(action)) {
-                final Collection<String> privileges = findIndexPrivilegesThatGrant(action);
-                if (privileges != null && privileges.size() > 0) {
+                // this includes `all`
+                final Collection<String> privileges = findIndexPrivilegesThatGrant(
+                    action,
+                    p -> p.getSelectorPredicate().test(IndexComponentSelector.DATA)
+                );
+                // this is an invariant since `all` is included in the above so the only way
+                // we can get an empty result here is a bogus action, which will never be covered by a failures privilege
+                assert false == privileges.isEmpty()
+                    || findIndexPrivilegesThatGrant(
+                        action,
+                        p -> p.getSelectorPredicate().test(IndexComponentSelector.FAILURES)
+                            && false == p.getSelectorPredicate().test(IndexComponentSelector.DATA)
+                    ).isEmpty()
+                    : "action [" + action + "] is not covered by any regular index privilege, only by failures-selector privileges";
+
+                if (false == privileges.isEmpty()) {
                     message = message
                         + ", this action is granted by the index privileges ["
                         + collectionToCommaDelimitedString(privileges)
                         + "]";
+
+                    final Collection<String> privilegesForFailuresOnly = findIndexPrivilegesThatGrant(
+                        action,
+                        p -> p.getSelectorPredicate().test(IndexComponentSelector.FAILURES)
+                            && false == p.getSelectorPredicate().test(IndexComponentSelector.DATA)
+                    );
+                    if (false == privilegesForFailuresOnly.isEmpty() && hasIndicesWithFailuresSelector(request)) {
+                        message = message
+                            + " for data access, or by ["
+                            + collectionToCommaDelimitedString(privilegesForFailuresOnly)
+                            + "] for access with the [::failures] selector";
+                    }
                 }
             }
-
             return message;
         }
 
@@ -132,14 +160,27 @@ public interface AuthorizationDenialMessages {
             return ClusterPrivilegeResolver.findPrivilegesThatGrant(action, request, authentication);
         }
 
-        protected Collection<String> findIndexPrivilegesThatGrant(String action) {
-            return IndexPrivilege.findPrivilegesThatGrant(action);
+        protected Collection<String> findIndexPrivilegesThatGrant(String action, Predicate<IndexPrivilege> preCondition) {
+            return IndexPrivilege.findPrivilegesThatGrant(action, preCondition);
         }
 
         private String remoteClusterText(@Nullable String clusterAlias) {
             return Strings.format("towards remote cluster%s ", clusterAlias == null ? "" : " [" + clusterAlias + "]");
         }
 
+        private boolean hasIndicesWithFailuresSelector(TransportRequest request) {
+            String[] indices = AuthorizationEngine.RequestInfo.indices(request);
+            if (indices == null) {
+                return false;
+            }
+            for (String index : indices) {
+                if (IndexNameExpressionResolver.hasSelector(index, IndexComponentSelector.FAILURES)) {
+                    return true;
+                }
+            }
+            return false;
+        }
+
         private String authenticatedUserDescription(Authentication authentication) {
             String userText = (authentication.isServiceAccount() ? "service account" : "user")
                 + " ["

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

@@ -7,9 +7,12 @@
 
 package org.elasticsearch.xpack.security.authz;
 
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationField;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationTestHelper;
@@ -23,6 +26,7 @@ import java.util.Map;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.equalTo;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -233,6 +237,94 @@ public class AuthorizationDenialMessagesTests extends ESTestCase {
         );
     }
 
+    public void testActionDeniedWithFailuresAndCorrectActionIncludesFailuresMessage() {
+        assumeTrue("failure store required", DataStream.isFailureStoreFeatureFlagEnabled());
+
+        Authentication authentication = AuthenticationTestHelper.builder().build();
+
+        final String action = "indices:data/read/" + randomAlphaOfLengthBetween(0, 8);
+        SearchRequest request = mock(SearchRequest.class);
+        for (List<String> requestedIndices : List.of(
+            List.of(randomAlphaOfLength(5) + "::failures"),
+            List.of(randomAlphaOfLength(5) + "::failures", randomAlphaOfLength(5)),
+            List.of(randomAlphaOfLength(5), randomAlphaOfLength(5) + "::failures")
+        )) {
+            when(request.indices()).thenReturn(requestedIndices.toArray(new String[0]));
+            assertThat(
+                "for [" + Strings.collectionToCommaDelimitedString(requestedIndices) + "]",
+                denialMessages.actionDenied(authentication, null, action, request, null),
+                endsWith(
+                    "this action is granted by the index privileges [read,all] for data access, "
+                        + "or by [read_failure_store] for access with the [::failures] selector"
+                )
+            );
+        }
+    }
+
+    public void testActionDeniedWithNonMatchingActionFailuresOmitsFailuresMessage() {
+        assumeTrue("failure store required", DataStream.isFailureStoreFeatureFlagEnabled());
+
+        Authentication authentication = AuthenticationTestHelper.builder().build();
+
+        // granted only by all, so selector message is omitted
+        final String action = "indices:/some/action/" + randomAlphaOfLengthBetween(0, 8);
+        SearchRequest request = mock(SearchRequest.class);
+        for (List<String> requestedIndices : List.of(
+            List.of(randomAlphaOfLength(5) + "::failures"),
+            List.of(randomAlphaOfLength(5) + "::failures", randomAlphaOfLength(5)),
+            List.of(randomAlphaOfLength(5), randomAlphaOfLength(5) + "::failures")
+        )) {
+            when(request.indices()).thenReturn(requestedIndices.toArray(new String[0]));
+            assertThat(
+                "for [" + Strings.collectionToCommaDelimitedString(requestedIndices) + "]",
+                denialMessages.actionDenied(authentication, null, action, request, null),
+                endsWith("this action is granted by the index privileges [all]")
+            );
+        }
+    }
+
+    public void testActionDeniedWithoutFailuresOmitsFailuresMessage() {
+        assumeTrue("failure store required", DataStream.isFailureStoreFeatureFlagEnabled());
+
+        Authentication authentication = AuthenticationTestHelper.builder().build();
+
+        final String action = "indices:data/read/" + randomAlphaOfLengthBetween(0, 8);
+        SearchRequest request = mock(SearchRequest.class);
+        for (List<String> requestedIndices : List.of(
+            List.<String>of(),
+            List.of(randomAlphaOfLength(5)),
+            List.of(randomAlphaOfLength(5), randomAlphaOfLength(5))
+        )) {
+            when(request.indices()).thenReturn(requestedIndices.toArray(new String[0]));
+            assertThat(
+                "for [" + Strings.collectionToCommaDelimitedString(requestedIndices) + "]",
+                denialMessages.actionDenied(authentication, null, action, request, null),
+                endsWith("this action is granted by the index privileges [read,all]")
+            );
+        }
+    }
+
+    public void testActionDeniedWithoutIndicesOmitsFailuresMessage() {
+        assumeTrue("failure store required", DataStream.isFailureStoreFeatureFlagEnabled());
+
+        Authentication authentication = AuthenticationTestHelper.builder().build();
+
+        final String action = "indices:data/read/" + randomAlphaOfLengthBetween(0, 8);
+        // not an IndicesRequest
+        TransportRequest request = mock(TransportRequest.class);
+        for (List<String> requestedIndices : List.of(
+            List.<String>of(),
+            List.of(randomAlphaOfLength(5)),
+            List.of(randomAlphaOfLength(5), randomAlphaOfLength(5))
+        )) {
+            assertThat(
+                "for [" + Strings.collectionToCommaDelimitedString(requestedIndices) + "]",
+                denialMessages.actionDenied(authentication, null, action, request, null),
+                endsWith("this action is granted by the index privileges [read,all]")
+            );
+        }
+    }
+
     public void testSuccessfulAuthenticationDescription() {
         final Authentication authentication1 = AuthenticationTestHelper.builder().realm().build(false);
         assertThat(