Fix DLS/FLS permission for the submit async search action (#59693)

The submit async search action should not populate the thread context
DLS/FLS permission set, because it is not currently authorised as an "indices request"
and hence the permission set that it builds is incomplete and it overrides the
DLS/FLS permission set of the actual spawned search request (which is built correctly).

x-pack/plugin/async-search/qa/security/build.gradle

@@ -17,4 +17,5 @@ testClusters.integTest {
   user username: "test-admin", password: 'x-pack-test-password', role: "test-admin"
   user username: "user1", password: 'x-pack-test-password', role: "user1"
   user username: "user2", password: 'x-pack-test-password', role: "user2"
+  user username: "user_dls", password: 'x-pack-test-password', role: "user_dls"
 }

x-pack/plugin/async-search/qa/security/roles.yml

@@ -31,3 +31,27 @@ user2:
         - write
         - create_index
         - indices:admin/refresh
+
+user_dls:
+  cluster:
+    - cluster:monitor/main
+  indices:
+    - names:
+        - 'index*'
+      privileges:
+        - read
+      field_security:
+        grant:
+          - baz
+      query: |
+        {
+          "bool": {
+            "must_not": [
+              {
+                "match": {
+                  "foo": "bar"
+                }
+              }
+            ]
+          }
+        }

x-pack/plugin/async-search/qa/security/src/test/java/org/elasticsearch/xpack/search/AsyncSearchSecurityIT.java

@@ -11,25 +11,34 @@
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.client.asyncsearch.AsyncSearchResponse;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.test.rest.ESRestTestCase;
 import org.elasticsearch.xpack.core.async.AsyncExecutionId;
+import org.hamcrest.CustomMatcher;
 import org.junit.Before;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.elasticsearch.xpack.core.XPackPlugin.ASYNC_RESULTS_INDEX;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationServiceField.RUN_AS_USER_HEADER;
 import static org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
-import static org.elasticsearch.xpack.core.XPackPlugin.ASYNC_RESULTS_INDEX;
+import static org.hamcrest.Matchers.arrayContainingInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
@@ -49,6 +58,8 @@ protected Settings restClientSettings() {
     public void indexDocuments() throws IOException {
         createIndex("index", Settings.EMPTY);
         index("index", "0", "foo", "bar");
+        index("index", "1", "bar", "baz");
+        index("index", "2", "baz", "boo");
         refresh("index");
 
         createIndex("index-user1", Settings.EMPTY);
@@ -57,9 +68,51 @@ public void indexDocuments() throws IOException {
 
         createIndex("index-user2", Settings.EMPTY);
         index("index-user2", "0", "foo", "bar");
+        index("index-user2", "1", "bar", "baz");
         refresh("index-user2");
     }
 
+    public void testWithDlsAndFls() throws Exception {
+        Response submitResp = submitAsyncSearch("*", "*", TimeValue.timeValueSeconds(10), "user_dls");
+        assertOK(submitResp);
+        String id = extractResponseId(submitResp);
+        Response getResp = getAsyncSearch(id, "user_dls");
+        AsyncSearchResponse searchResponse = AsyncSearchResponse.fromXContent(XContentHelper.createParser(NamedXContentRegistry.EMPTY,
+                LoggingDeprecationHandler.INSTANCE,
+                new BytesArray(EntityUtils.toByteArray(getResp.getEntity())),
+                XContentType.JSON));
+        SearchHit[] hits = searchResponse.getSearchResponse().getHits().getHits();
+
+        assertThat(hits, arrayContainingInAnyOrder(
+                new CustomMatcher<SearchHit>("\"index\" doc 1 matcher") {
+                    @Override
+                    public boolean matches(Object actual) {
+                        SearchHit hit = (SearchHit) actual;
+                        return "index".equals(hit.getIndex()) &&
+                                "1".equals(hit.getId()) &&
+                                hit.getSourceAsMap().isEmpty();
+                    }
+                },
+                new CustomMatcher<SearchHit>("\"index\" doc 2 matcher") {
+                    @Override
+                    public boolean matches(Object actual) {
+                        SearchHit hit = (SearchHit) actual;
+                        return "index".equals(hit.getIndex()) &&
+                                "2".equals(hit.getId()) &&
+                                "boo".equals(hit.getSourceAsMap().get("baz"));
+                    }
+                },
+                new CustomMatcher<SearchHit>("\"index-user2\" doc 1 matcher") {
+                    @Override
+                    public boolean matches(Object actual) {
+                        SearchHit hit = (SearchHit) actual;
+                        return "index-user2".equals(hit.getIndex()) &&
+                                "1".equals(hit.getId()) &&
+                                hit.getSourceAsMap().isEmpty();
+                    }
+                }));
+    }
+
     public void testWithUsers() throws Exception {
         testCase("user1", "user2");
         testCase("user2", "user1");
@@ -103,6 +156,12 @@ static String extractResponseId(Response response) throws IOException {
         return (String) map.get("id");
     }
 
+    @SuppressWarnings("unchecked")
+    static List<Map<String, Map<String, Object>>> extractHits(Map<String, Object> respMap) {
+        Map<String, Object> response = ((Map<String, Object>) respMap.get("response"));
+        return ((List<Map<String, Map<String, Object>>>)((Map<String, Object>) response.get("hits")).get("hits"));
+    }
+
     static void index(String index, String id, Object... fields) throws IOException {
         XContentBuilder document = jsonBuilder().startObject();
         for (int i = 0; i < fields.length; i += 2) {

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

@@ -269,8 +269,16 @@ public void authorizeIndexAction(RequestInfo requestInfo, AuthorizationInfo auth
                     listener.onResponse(new IndexAuthorizationResult(true, IndicesAccessControl.ALLOW_NO_INDICES));
                 }
             } else if (isAsyncRelatedAction(action)) {
-                //index-level permissions are handled by the search action that is triggered internally by the submit API.
-                listener.onResponse(new IndexAuthorizationResult(true, IndicesAccessControl.ALLOW_NO_INDICES));
+                if (SubmitAsyncSearchAction.NAME.equals(action)) {
+                    // authorize submit async search but don't fill in the DLS/FLS permissions
+                    // the `null` IndicesAccessControl parameter indicates that this action has *not* determined
+                    // which DLS/FLS controls should be applied to this action
+                    listener.onResponse(new IndexAuthorizationResult(true, null));
+                } else {
+                    // async-search actions other than submit have a custom security layer that checks if the current user is
+                    // the same as the user that submitted the original request so no additional checks are needed here.
+                    listener.onResponse(new IndexAuthorizationResult(true, IndicesAccessControl.ALLOW_NO_INDICES));
+                }
             } else {
                 assert false : "only scroll and async-search related requests are known indices api that don't " +
                     "support retrieving the indices they relate to";