Tasks: Only require task permissions

Right now using the `GET /_tasks/<taskid>` API and causing a task to opt
in to saving its result after being completed requires permissions on
the `.tasks` index. When we built this we thought that that was fine,
but we've since moved towards not leaking details like "persisting task
results after the task is completed is done by saving them into an index
named `.tasks`." A more modern way of doing this would be to save the
tasks into the index "under the hood" and to have APIs to manage the
saved tasks. This is the first step down that road: it drops the
requirement to have permissions to interact with the `.tasks` index when
fetching task statuses and when persisting statuses beyond the lifetime
of the task.

In particular, this moves the concept of the "origin" of an action into
a more prominent place in the Elasticsearch server. The origin of an
action is ignored by the server, but the security plugin uses the origin
to make requests on behalf of a user in such a way that the user need
not have permissions to perform these actions. It *can* be made to be
fairly precise. More specifically, we can create an internal user just
for the tasks API that just has permission to interact with the `.tasks`
index. This change doesn't do that, instead, it uses the ubiquitus
"xpack" user which has most permissions because it is simpler. Adding
the tasks user is something I'd like to get to in a follow up change.

Instead, the majority of this change is about moving the "origin"
concept from the security portion of x-pack into the server. This should
allow any code to use the origin. To keep the change managable I've also
opted to deprecate rather than remove the "origin" helpers in the
security code. Removing them is almost entirely mechanical and I'd like
to that in a follow up as well.

Relates to elastic#35573

Author: nik9000

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/get/GetTaskAction.java

@@ -25,6 +25,7 @@
  * Action for retrieving a list of currently running tasks
  */
 public class GetTaskAction extends Action<GetTaskResponse> {
+    public static final String TASKS_ORIGIN = "tasks";
 
     public static final GetTaskAction INSTANCE = new GetTaskAction();
     public static final String NAME = "cluster:monitor/task/get";

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/get/TransportGetTaskAction.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.client.OriginSettingClient;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
@@ -51,6 +52,7 @@
 
 import java.io.IOException;
 
+import static org.elasticsearch.action.admin.cluster.node.tasks.get.GetTaskAction.TASKS_ORIGIN;
 import static org.elasticsearch.action.admin.cluster.node.tasks.list.TransportListTasksAction.waitForCompletionTimeout;
 
 /**
@@ -210,7 +212,7 @@ void getFinishedTaskFromIndex(Task thisTask, GetTaskRequest request, ActionListe
         GetRequest get = new GetRequest(TaskResultsService.TASK_INDEX, TaskResultsService.TASK_TYPE,
                 request.getTaskId().toString());
         get.setParentTask(clusterService.localNode().getId(), thisTask.getId());
-        client.get(get, new ActionListener<GetResponse>() {
+        new OriginSettingClient(client, TASKS_ORIGIN).get(get, new ActionListener<GetResponse>() {
             @Override
             public void onResponse(GetResponse getResponse) {
                 try {

server/src/main/java/org/elasticsearch/client/OriginSettingClient.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.client;
+
+import org.elasticsearch.action.Action;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.support.ContextPreservingActionListener;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+
+import java.util.function.Supplier;
+
+/**
+ * A {@linkplain Client} that sends requests with the
+ * {@link ThreadContext#stashWithOrigin origin} set to a particular
+ * value and calls its {@linkplain ActionListener} in its original
+ * {@link ThreadContext}.
+ */
+public final class OriginSettingClient extends FilterClient {
+
+    private final String origin;
+
+    public OriginSettingClient(Client in, String origin) {
+        super(in);
+        this.origin = origin;
+    }
+
+    @Override
+    protected <Request extends ActionRequest, Response extends ActionResponse>
+            void doExecute(Action<Response> action, Request request, ActionListener<Response> listener) {
+        final Supplier<ThreadContext.StoredContext> supplier = in().threadPool().getThreadContext().newRestorableContext(false);
+        try (ThreadContext.StoredContext ignore = in().threadPool().getThreadContext().stashWithOrigin(origin)) {
+            super.doExecute(action, request, new ContextPreservingActionListener<>(supplier, listener));
+        }
+    }
+}

server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java

@@ -22,6 +22,8 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.lucene.util.CloseableThreadLocal;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.action.support.ContextPreservingActionListener;
+import org.elasticsearch.client.OriginSettingClient;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -85,6 +87,12 @@ public final class ThreadContext implements Closeable, Writeable {
 
     public static final String PREFIX = "request.headers";
     public static final Setting<Settings> DEFAULT_HEADERS_SETTING = Setting.groupSetting(PREFIX + ".", Property.NodeScope);
+
+    /**
+     * Name for the {@link #stashWithOrigin origin} attribute.
+     */
+    public static final String ACTION_ORIGIN_TRANSIENT_NAME = "action.origin";
+
     private static final Logger logger = LogManager.getLogger(ThreadContext.class);
     private static final ThreadContextStruct DEFAULT_CONTEXT = new ThreadContextStruct();
     private final Map<String, String> defaultHeader;
@@ -119,14 +127,39 @@ public void close() throws IOException {
 
     /**
      * Removes the current context and resets a default context. The removed context can be
-     * restored when closing the returned {@link StoredContext}
+     * restored by closing the returned {@link StoredContext}.
      */
     public StoredContext stashContext() {
         final ThreadContextStruct context = threadLocal.get();
         threadLocal.set(null);
         return () -> threadLocal.set(context);
     }
 
+    /**
+     * Removes the current context and resets a default context marked with as
+     * originating from the supplied string. The removed context can be
+     * restored by closing the returned {@link StoredContext}. Callers should
+     * be careful to save the current context before calling this method and
+     * restore it any listeners, likely with
+     * {@link ContextPreservingActionListener}. Use {@link OriginSettingClient}
+     * which can be used to do this automatically.
+     * <p>
+     * Without security the origin is ignored, but security uses it to authorize
+     * actions that are made up of many sub-actions. These actions call
+     * {@link #stashWithOrigin} before performing on behalf of a user that
+     * should be allowed even if the user doesn't have permission to perform
+     * those actions on their own.
+     * <p>
+     * For example, a user might not have permission to GET from the tasks index
+     * but the tasks API will perform a get on their behalf using this method
+     * if it can't find the task in memory.
+     */
+    public StoredContext stashWithOrigin(String origin) {
+        final ThreadContext.StoredContext storedContext = stashContext();
+        putTransient(ACTION_ORIGIN_TRANSIENT_NAME, origin);
+        return storedContext;
+    }
+
     /**
      * Removes the current context and resets a new context that contains a merge of the current headers and the given headers.
      * The removed context can be restored when closing the returned {@link StoredContext}. The merge strategy is that headers

server/src/main/java/org/elasticsearch/persistent/PersistentTasksService.java

@@ -23,22 +23,20 @@
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksResponse;
-import org.elasticsearch.action.support.ContextPreservingActionListener;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.client.OriginSettingClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.node.NodeClosedException;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.function.Predicate;
-import java.util.function.Supplier;
 
 /**
  * This service is used by persistent tasks and allocated persistent tasks to communicate changes
@@ -47,15 +45,14 @@
  */
 public class PersistentTasksService extends AbstractComponent {
 
-    private static final String ACTION_ORIGIN_TRANSIENT_NAME = "action.origin";
     private static final String PERSISTENT_TASK_ORIGIN = "persistent_tasks";
 
     private final Client client;
     private final ClusterService clusterService;
     private final ThreadPool threadPool;
 
     public PersistentTasksService(ClusterService clusterService, ThreadPool threadPool, Client client) {
-        this.client = client;
+        this.client = new OriginSettingClient(client, PERSISTENT_TASK_ORIGIN);
         this.clusterService = clusterService;
         this.threadPool = threadPool;
     }
@@ -95,12 +92,7 @@ void sendCancelRequest(final long taskId, final String reason, final ActionListe
         request.setTaskId(new TaskId(clusterService.localNode().getId(), taskId));
         request.setReason(reason);
         try {
-            final ThreadContext threadContext = client.threadPool().getThreadContext();
-            final Supplier<ThreadContext.StoredContext> supplier = threadContext.newRestorableContext(false);
-
-            try (ThreadContext.StoredContext ignore = stashWithOrigin(threadContext, PERSISTENT_TASK_ORIGIN)) {
-                client.admin().cluster().cancelTasks(request, new ContextPreservingActionListener<>(supplier, listener));
-            }
+            client.admin().cluster().cancelTasks(request, listener);
         } catch (Exception e) {
             listener.onFailure(e);
         }
@@ -137,14 +129,8 @@ public void sendRemoveRequest(final String taskId, final ActionListener<Persiste
     private <Req extends ActionRequest, Resp extends PersistentTaskResponse>
         void execute(final Req request, final Action<Resp> action, final ActionListener<PersistentTask<?>> listener) {
             try {
-                final ThreadContext threadContext = client.threadPool().getThreadContext();
-                final Supplier<ThreadContext.StoredContext> supplier = threadContext.newRestorableContext(false);
-
-                try (ThreadContext.StoredContext ignore = stashWithOrigin(threadContext, PERSISTENT_TASK_ORIGIN)) {
-                    client.execute(action, request,
-                        new ContextPreservingActionListener<>(supplier,
-                            ActionListener.wrap(r -> listener.onResponse(r.getTask()), listener::onFailure)));
-                }
+                client.execute(action, request,
+                        ActionListener.wrap(r -> listener.onResponse(r.getTask()), listener::onFailure));
             } catch (Exception e) {
                 listener.onFailure(e);
             }
@@ -230,10 +216,4 @@ default void onTimeout(TimeValue timeout) {
             onFailure(new IllegalStateException("Timed out when waiting for persistent task after " + timeout));
         }
     }
-
-    public static ThreadContext.StoredContext stashWithOrigin(ThreadContext threadContext, String origin) {
-        final ThreadContext.StoredContext storedContext = threadContext.stashContext();
-        threadContext.putTransient(ACTION_ORIGIN_TRANSIENT_NAME, origin);
-        return storedContext;
-    }
 }

server/src/main/java/org/elasticsearch/tasks/TaskResultsService.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.client.OriginSettingClient;
 import org.elasticsearch.client.Requests;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
@@ -70,7 +71,7 @@ public class TaskResultsService extends AbstractComponent {
 
     @Inject
     public TaskResultsService(Client client, ClusterService clusterService) {
-        this.client = client;
+        this.client = new OriginSettingClient(client, "tasks");
         this.clusterService = clusterService;
     }
 

server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TasksIT.java

@@ -27,14 +27,14 @@
 import org.elasticsearch.action.TaskOperationFailure;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction;
 import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksResponse;
+import org.elasticsearch.action.admin.cluster.node.tasks.get.GetTaskRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.get.GetTaskResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksAction;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
 import org.elasticsearch.action.admin.indices.refresh.RefreshAction;
 import org.elasticsearch.action.admin.indices.upgrade.post.UpgradeAction;
 import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryAction;
 import org.elasticsearch.action.bulk.BulkAction;
-import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.index.IndexAction;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.search.SearchAction;
@@ -85,7 +85,6 @@
 import static org.elasticsearch.common.unit.TimeValue.timeValueMillis;
 import static org.elasticsearch.common.unit.TimeValue.timeValueSeconds;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_MAX_HEADER_SIZE;
-import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSearchResponse;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertThrows;
@@ -725,12 +724,6 @@ public void testTasksWaitForAllTask() throws Exception {
     }
 
     public void testTaskStoringSuccesfulResult() throws Exception {
-        // Randomly create an empty index to make sure the type is created automatically
-        if (randomBoolean()) {
-            logger.info("creating an empty results index with custom settings");
-            assertAcked(client().admin().indices().prepareCreate(TaskResultsService.TASK_INDEX));
-        }
-
         registerTaskManageListeners(TestTaskPlugin.TestTaskAction.NAME);  // we need this to get task id of the process
 
         // Start non-blocking test task
@@ -743,23 +736,20 @@ public void testTaskStoringSuccesfulResult() throws Exception {
         TaskInfo taskInfo = events.get(0);
         TaskId taskId = taskInfo.getTaskId();
 
-        GetResponse resultDoc = client()
-                .prepareGet(TaskResultsService.TASK_INDEX, TaskResultsService.TASK_TYPE, taskId.toString()).get();
-        assertTrue(resultDoc.isExists());
-
-        Map<String, Object> source = resultDoc.getSource();
-        @SuppressWarnings("unchecked")
-        Map<String, Object> task = (Map<String, Object>) source.get("task");
-        assertEquals(taskInfo.getTaskId().getNodeId(), task.get("node"));
-        assertEquals(taskInfo.getAction(), task.get("action"));
-        assertEquals(Long.toString(taskInfo.getId()), task.get("id").toString());
-
-        @SuppressWarnings("unchecked")
-        Map<String, Object> result = (Map<String, Object>) source.get("response");
+        TaskResult taskResult = client().admin().cluster()
+                .getTask(new GetTaskRequest().setTaskId(taskId)).get().getTask();
+        assertTrue(taskResult.isCompleted());
+        assertNull(taskResult.getError());
+
+        assertEquals(taskInfo.getTaskId(), taskResult.getTask().getTaskId());
+        assertEquals(taskInfo.getType(), taskResult.getTask().getType());
+        assertEquals(taskInfo.getAction(), taskResult.getTask().getAction());
+        assertEquals(taskInfo.getDescription(), taskResult.getTask().getDescription());
+        assertEquals(taskInfo.getStartTime(), taskResult.getTask().getStartTime());
+        assertEquals(taskInfo.getHeaders(), taskResult.getTask().getHeaders());
+        Map<?, ?> result = taskResult.getResponseAsMap();
         assertEquals("0", result.get("failure_count").toString());
 
-        assertNull(source.get("failure"));
-
         assertNoFailures(client().admin().indices().prepareRefresh(TaskResultsService.TASK_INDEX).get());
 
         SearchResponse searchResponse = client().prepareSearch(TaskResultsService.TASK_INDEX)
@@ -797,25 +787,21 @@ public void testTaskStoringFailureResult() throws Exception {
         TaskInfo failedTaskInfo = events.get(0);
         TaskId failedTaskId = failedTaskInfo.getTaskId();
 
-        GetResponse failedResultDoc = client()
-            .prepareGet(TaskResultsService.TASK_INDEX, TaskResultsService.TASK_TYPE, failedTaskId.toString())
-            .get();
-        assertTrue(failedResultDoc.isExists());
-
-        Map<String, Object> source = failedResultDoc.getSource();
-        @SuppressWarnings("unchecked")
-        Map<String, Object> task = (Map<String, Object>) source.get("task");
-        assertEquals(failedTaskInfo.getTaskId().getNodeId(), task.get("node"));
-        assertEquals(failedTaskInfo.getAction(), task.get("action"));
-        assertEquals(Long.toString(failedTaskInfo.getId()), task.get("id").toString());
-
-        @SuppressWarnings("unchecked")
-        Map<String, Object> error = (Map<String, Object>) source.get("error");
+        TaskResult taskResult = client().admin().cluster()
+                .getTask(new GetTaskRequest().setTaskId(failedTaskId)).get().getTask();
+        assertTrue(taskResult.isCompleted());
+        assertNull(taskResult.getResponse());
+
+        assertEquals(failedTaskInfo.getTaskId(), taskResult.getTask().getTaskId());
+        assertEquals(failedTaskInfo.getType(), taskResult.getTask().getType());
+        assertEquals(failedTaskInfo.getAction(), taskResult.getTask().getAction());
+        assertEquals(failedTaskInfo.getDescription(), taskResult.getTask().getDescription());
+        assertEquals(failedTaskInfo.getStartTime(), taskResult.getTask().getStartTime());
+        assertEquals(failedTaskInfo.getHeaders(), taskResult.getTask().getHeaders());
+        Map<?, ?> error = (Map<?, ?>) taskResult.getErrorAsMap();
         assertEquals("Simulating operation failure", error.get("reason"));
         assertEquals("illegal_state_exception", error.get("type"));
 
-        assertNull(source.get("result"));
-
         GetTaskResponse getResponse = expectFinishedTask(failedTaskId);
         assertNull(getResponse.getTask().getResponse());
         assertEquals(error, getResponse.getTask().getErrorAsMap());