[CCR] Add create_follow_index privilege (#33559)

This is a new index privilege that the user needs to have in the follow cluster.
This privilege is required in addition to the `manage_ccr` cluster privilege in
order to execute the create and follow api.

Closes #33555

x-pack/plugin/ccr/qa/multi-cluster-with-security/roles.yml

@@ -7,3 +7,4 @@ ccruser:
         - monitor
         - read
         - write
+        - create_follow_index

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

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.ccr;
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.Strings;
@@ -18,6 +19,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.test.rest.ESRestTestCase;
 
 import java.io.IOException;
@@ -26,7 +28,9 @@ import java.util.Map;
 
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
 
 public class FollowIndexSecurityIT extends ESRestTestCase {
 
@@ -96,16 +100,19 @@ public class FollowIndexSecurityIT extends ESRestTestCase {
                 assertThat(countCcrNodeTasks(), equalTo(0));
             });
 
-            createAndFollowIndex("leader_cluster:" + unallowedIndex, unallowedIndex);
+            Exception e = expectThrows(ResponseException.class,
-            // Verify that nothing has been replicated and no node tasks are running
+                () -> createAndFollowIndex("leader_cluster:" + unallowedIndex, unallowedIndex));
-            // These node tasks should have been failed due to the fact that the user
+            assertThat(e.getMessage(),
-            // has no sufficient priviledges.
+                containsString("action [indices:admin/xpack/ccr/create_and_follow_index] is unauthorized for user [test_ccr]"));
+            // Verify that the follow index has not been created and no node tasks are running
+            assertThat(indexExists(adminClient(), unallowedIndex), is(false));
             assertBusy(() -> assertThat(countCcrNodeTasks(), equalTo(0)));
-            verifyDocuments(adminClient(), unallowedIndex, 0);
 
-            followIndex("leader_cluster:" + unallowedIndex, unallowedIndex);
+            e = expectThrows(ResponseException.class,
+                () -> followIndex("leader_cluster:" + unallowedIndex, unallowedIndex));
+            assertThat(e.getMessage(), containsString("follow index [" + unallowedIndex + "] does not exist"));
+            assertThat(indexExists(adminClient(), unallowedIndex), is(false));
             assertBusy(() -> assertThat(countCcrNodeTasks(), equalTo(0)));
-            verifyDocuments(adminClient(), unallowedIndex, 0);
         }
     }
 
@@ -191,4 +198,9 @@ public class FollowIndexSecurityIT extends ESRestTestCase {
         assertOK(adminClient().performRequest(request));
     }
 
+    private static boolean indexExists(RestClient client, String index) throws IOException {
+        Response response = client.performRequest(new Request("HEAD", "/" + index));
+        return RestStatus.OK.getStatus() == response.getStatusLine().getStatusCode();
+    }
+
 }

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

@@ -12,9 +12,11 @@ import org.elasticsearch.action.Action;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.ActiveShardsObserver;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.client.Client;
@@ -52,7 +54,7 @@ import java.util.Objects;
 public class CreateAndFollowIndexAction extends Action<CreateAndFollowIndexAction.Response> {
 
     public static final CreateAndFollowIndexAction INSTANCE = new CreateAndFollowIndexAction();
-    public static final String NAME = "cluster:admin/xpack/ccr/create_and_follow_index";
+    public static final String NAME = "indices:admin/xpack/ccr/create_and_follow_index";
 
     private CreateAndFollowIndexAction() {
         super(NAME);
@@ -63,7 +65,7 @@ public class CreateAndFollowIndexAction extends Action<CreateAndFollowIndexActio
         return new Response();
     }
 
-    public static class Request extends AcknowledgedRequest<Request> {
+    public static class Request extends AcknowledgedRequest<Request> implements IndicesRequest {
 
         private FollowIndexAction.Request followRequest;
 
@@ -83,6 +85,16 @@ public class CreateAndFollowIndexAction extends Action<CreateAndFollowIndexActio
             return followRequest.validate();
         }
 
+        @Override
+        public String[] indices() {
+            return new String[]{followRequest.getFollowerIndex()};
+        }
+
+        @Override
+        public IndicesOptions indicesOptions() {
+            return IndicesOptions.strictSingleIndexNoExpandForbidClosed();
+        }
+
         @Override
         public void readFrom(StreamInput in) throws IOException {
             super.readFrom(in);

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

@@ -55,6 +55,7 @@ public final class IndexPrivilege extends Privilege {
     private static final Automaton VIEW_METADATA_AUTOMATON = patterns(GetAliasesAction.NAME, AliasesExistAction.NAME,
             GetIndexAction.NAME, IndicesExistsAction.NAME, GetFieldMappingsAction.NAME + "*", GetMappingsAction.NAME,
             ClusterSearchShardsAction.NAME, TypesExistsAction.NAME, ValidateQueryAction.NAME + "*", GetSettingsAction.NAME);
+    private static final Automaton CREATE_FOLLOW_INDEX_AUTOMATON = patterns("indices:admin/xpack/ccr/create_and_follow_index");
 
     public static final IndexPrivilege NONE =                new IndexPrivilege("none",             Automatons.EMPTY);
     public static final IndexPrivilege ALL =                 new IndexPrivilege("all",              ALL_AUTOMATON);
@@ -69,6 +70,7 @@ public final class IndexPrivilege extends Privilege {
     public static final IndexPrivilege DELETE_INDEX =        new IndexPrivilege("delete_index",        DELETE_INDEX_AUTOMATON);
     public static final IndexPrivilege CREATE_INDEX =        new IndexPrivilege("create_index",        CREATE_INDEX_AUTOMATON);
     public static final IndexPrivilege VIEW_METADATA =       new IndexPrivilege("view_index_metadata", VIEW_METADATA_AUTOMATON);
+    public static final IndexPrivilege CREATE_FOLLOW_INDEX = new IndexPrivilege("create_follow_index", CREATE_FOLLOW_INDEX_AUTOMATON);
 
     private static final Map<String, IndexPrivilege> VALUES = MapBuilder.<String, IndexPrivilege>newMapBuilder()
             .put("none", NONE)
@@ -84,6 +86,7 @@ public final class IndexPrivilege extends Privilege {
             .put("delete_index", DELETE_INDEX)
             .put("view_index_metadata", VIEW_METADATA)
             .put("read_cross_cluster", READ_CROSS_CLUSTER)
+            .put("create_follow_index", CREATE_FOLLOW_INDEX)
             .immutableMap();
 
     public static final Predicate<String> ACTION_MATCHER = ALL.predicate();