Allow users with search priv to get async status

This changs RBAC engine so users with access to submit async searches
also have permission to retrive async search status. It relies on the
fact that async search will check that the user owns the async search
task

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

@@ -105,6 +105,7 @@ public class AsyncSearchSecurityIT extends ESRestTestCase {
         .user("user2", "x-pack-test-password", "user2", false)
         .user("user-dls", "x-pack-test-password", "user-dls", false)
         .user("user-cancel", "x-pack-test-password", "user-cancel", false)
+        .user("user-monitor", "x-pack-test-password", "user-monitor", false)
         .build();
 
     @Override
@@ -174,24 +175,43 @@ private void testCase(String user, String other) throws Exception {
             Response submitResp = submitAsyncSearch(indexName, "foo:bar", TimeValue.timeValueSeconds(10), user);
             assertOK(submitResp);
             String id = extractResponseId(submitResp);
+
+            Response statusResp = getAsyncStatus(id, user);
+            assertOK(statusResp);
+
             Response getResp = getAsyncSearch(id, user);
             assertOK(getResp);
 
-            // other cannot access the result
-            ResponseException exc = expectThrows(ResponseException.class, () -> getAsyncSearch(id, other));
+            // other (user) cannot access the status
+            ResponseException exc = expectThrows(ResponseException.class, () -> getAsyncStatus(id, other));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), equalTo(404));
+
+            // other (user) cannot access the result
+            exc = expectThrows(ResponseException.class, () -> getAsyncSearch(id, other));
             assertThat(exc.getResponse().getStatusLine().getStatusCode(), equalTo(404));
 
             // user-cancel cannot access the result
             exc = expectThrows(ResponseException.class, () -> getAsyncSearch(id, "user-cancel"));
             assertThat(exc.getResponse().getStatusLine().getStatusCode(), equalTo(404));
 
+            // user-monitor can access the status
+            assertOK(getAsyncStatus(id, "user-monitor"));
+
+            // user-monitor cannot access the result
+            exc = expectThrows(ResponseException.class, () -> getAsyncSearch(id, "user-monitor"));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), equalTo(404));
+
             // other cannot delete the result
             exc = expectThrows(ResponseException.class, () -> deleteAsyncSearch(id, other));
             assertThat(exc.getResponse().getStatusLine().getStatusCode(), equalTo(404));
 
-            // other and user cannot access the result from direct get calls
+            // user-monitor cannot delete the result
+            exc = expectThrows(ResponseException.class, () -> deleteAsyncSearch(id, "user-monitor"));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), equalTo(404));
+
+            // none of the users can access the result from direct get calls on the index
             AsyncExecutionId searchId = AsyncExecutionId.decode(id);
-            for (String runAs : new String[] { user, other }) {
+            for (String runAs : new String[] { user, other, "user-monitor", "user-cancel" }) {
                 exc = expectThrows(ResponseException.class, () -> get(ASYNC_RESULTS_INDEX, searchId.getDocId(), runAs));
                 assertThat(exc.getResponse().getStatusLine().getStatusCode(), equalTo(403));
                 assertThat(exc.getMessage(), containsString("unauthorized"));
@@ -402,6 +422,12 @@ static Response submitAsyncSearch(String indexName, String query, TimeValue wait
         return client().performRequest(request);
     }
 
+    static Response getAsyncStatus(String id, String user) throws IOException {
+        final Request request = new Request("GET", "/_async_search/status/" + id);
+        setRunAsHeader(request, user);
+        return client().performRequest(request);
+    }
+
     static Response getAsyncSearch(String id, String user) throws IOException {
         final Request request = new Request("GET", "/_async_search/" + id);
         setRunAsHeader(request, user);

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

@@ -59,3 +59,7 @@ user-dls:
 user-cancel:
   cluster:
     - cancel_task
+
+user-monitor:
+  cluster:
+    - monitor

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

@@ -46,6 +46,7 @@
 import org.elasticsearch.xpack.core.eql.EqlAsyncActionNames;
 import org.elasticsearch.xpack.core.esql.EsqlAsyncActionNames;
 import org.elasticsearch.xpack.core.search.action.GetAsyncSearchAction;
+import org.elasticsearch.xpack.core.search.action.GetAsyncStatusAction;
 import org.elasticsearch.xpack.core.search.action.SubmitAsyncSearchAction;
 import org.elasticsearch.xpack.core.security.action.apikey.GetApiKeyAction;
 import org.elasticsearch.xpack.core.security.action.apikey.GetApiKeyRequest;
@@ -188,6 +189,10 @@ public void authorizeClusterAction(
                 listener.onResponse(AuthorizationResult.granted());
             } else if (checkSameUserPermissions(requestInfo.getAction(), requestInfo.getRequest(), requestInfo.getAuthentication())) {
                 listener.onResponse(AuthorizationResult.granted());
+            } else if (GetAsyncStatusAction.NAME.equals(requestInfo.getAction()) && role.checkIndicesAction(SubmitAsyncSearchAction.NAME)) {
+                // Users who are allowed to submit async searches are allowed to check the status of those searches
+                // Search ownership will be checked by AsyncSearchSecurity
+                listener.onResponse(AuthorizationResult.granted());
             } else {
                 listener.onResponse(AuthorizationResult.deny());
             }