[CCR] Add create_follow_index privilege

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 elastic#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 @@
 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.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 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 void testFollowIndex() throws Exception {
                 assertThat(countCcrNodeTasks(), equalTo(0));
             });
 
-            createAndFollowIndex("leader_cluster:" + unallowedIndex, unallowedIndex);
-            // Verify that nothing has been replicated and no node tasks are running
-            // These node tasks should have been failed due to the fact that the user
-            // has no sufficient priviledges.
+            Exception e = expectThrows(ResponseException.class,
+                () -> createAndFollowIndex("leader_cluster:" + unallowedIndex, unallowedIndex));
+            assertThat(e.getMessage(),
+                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 @@ protected static void createIndex(String name, Settings settings, String mapping
         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

@@ -52,7 +52,7 @@
 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);

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();