Fix shard follow task cleaner under security (elastic#52347)

The shard follow task cleaner executes on behalf of the user to clean up
a shard follow task after the follower index has been
deleted. Otherwise, these persistent tasks are left laying around, and
they fail to execute because the follower index has been deleted. In the
face of security, attempts to complete these persistent tasks would
fail.  This is because these cleanups are executed under the system
context (this makes sense, they are happening on behalf of the user
after the user has executed an action) but the system role was never
granted the permission for persistent task completion. This commit
addresses this by adding this cluster privilege to the system role.

Author: jasontedor

x-pack/plugin/ccr/qa/security/follower-roles.yml

@@ -8,3 +8,10 @@ ccruser:
         - read
         - write
         - manage_follow_index
+    - names: [ 'clean-follower' ]
+      privileges:
+        - monitor
+        - read
+        - write
+        - delete_index
+        - manage_follow_index

x-pack/plugin/ccr/qa/security/leader-roles.yml

@@ -2,7 +2,7 @@ ccruser:
   cluster:
     - read_ccr
   indices:
-    - names: [ 'allowed-index', 'forget-leader', 'logs-eu-*' ]
+    - names: [ 'allowed-index', 'clean-leader', 'forget-leader', 'logs-eu-*' ]
       privileges:
         - monitor
         - read

x-pack/plugin/ccr/qa/security/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexSecurityIT.java

@@ -225,4 +225,32 @@ public void testForgetFollower() throws IOException {
         }
     }
 
+    public void testCleanShardFollowTaskAfterDeleteFollower() throws Exception {
+        final String cleanLeader = "clean-leader";
+        final String cleanFollower = "clean-follower";
+        if ("leader".equals(targetCluster)) {
+            logger.info("running against leader cluster");
+            final Settings indexSettings = Settings.builder()
+                .put("index.number_of_replicas", 0)
+                .put("index.number_of_shards", 1)
+                .put("index.soft_deletes.enabled", true)
+                .build();
+            createIndex(cleanLeader, indexSettings);
+        } else {
+            logger.info("running against follower cluster");
+            followIndex(client(), "leader_cluster", cleanLeader, cleanFollower);
+
+            final Request request = new Request("DELETE", "/" + cleanFollower);
+            final Response response = client().performRequest(request);
+            assertOK(response);
+            // the shard follow task should have been cleaned up on behalf of the user, see ShardFollowTaskCleaner
+            assertBusy(() -> {
+                Map<String, Object> clusterState = toMap(adminClient().performRequest(new Request("GET", "/_cluster/state")));
+                List<?> tasks = (List<?>) XContentMapValues.extractValue("metadata.persistent_tasks.tasks", clusterState);
+                assertThat(tasks.size(), equalTo(0));
+                assertThat(countCcrNodeTasks(), equalTo(0));
+            });
+        }
+    }
+
 }

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTaskCleaner.java

@@ -68,8 +68,12 @@ public void clusterChanged(final ClusterChangedEvent event) {
             CompletionPersistentTaskAction.Request request =
                 new CompletionPersistentTaskAction.Request(persistentTask.getId(), persistentTask.getAllocationId(), infe);
             threadPool.generic().submit(() -> {
+                /*
+                 * We are executing under the system context, on behalf of the user to clean up the shard follow task after the follower
+                 * index was deleted. This is why the system role includes the privilege for persistent task completion.
+                 */
+                assert threadPool.getThreadContext().isSystemContext();
                 client.execute(CompletionPersistentTaskAction.INSTANCE, request, new ActionListener<PersistentTaskResponse>() {
-
                     @Override
                     public void onResponse(PersistentTaskResponse persistentTaskResponse) {
                         logger.debug("task [{}] cleaned up", persistentTask.getId());

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

@@ -8,6 +8,7 @@
 import org.elasticsearch.index.seqno.RetentionLeaseActions;
 import org.elasticsearch.index.seqno.RetentionLeaseBackgroundSyncAction;
 import org.elasticsearch.index.seqno.RetentionLeaseSyncAction;
+import org.elasticsearch.persistent.CompletionPersistentTaskAction;
 import org.elasticsearch.transport.TransportActionProxy;
 import org.elasticsearch.xpack.core.security.support.Automatons;
 
@@ -33,7 +34,8 @@ public final class SystemPrivilege extends Privilege {
         RetentionLeaseActions.Add.ACTION_NAME + "*", // needed for CCR to add retention leases
         RetentionLeaseActions.Remove.ACTION_NAME + "*", // needed for CCR to remove retention leases
         RetentionLeaseActions.Renew.ACTION_NAME + "*", // needed for CCR to renew retention leases
-        "indices:admin/settings/update" // needed for DiskThresholdMonitor.markIndicesReadOnly
+        "indices:admin/settings/update", // needed for DiskThresholdMonitor.markIndicesReadOnly
+        CompletionPersistentTaskAction.NAME // needed for ShardFollowTaskCleaner
     );
 
     private static final Predicate<String> PREDICATE = (action) -> {