[CCR] Change unfollow API's privilege scheme. (#34175)

Unfollow should be allowed / disallowed on a per index level instead of
cluster level.

Also renamed `create_follow_index` index privilege to
`manage_follow_index` privilege and include unfollow and close APIs.

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

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

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

@@ -104,16 +104,20 @@ public class FollowIndexSecurityIT extends ESRestTestCase {
                 assertThat(countCcrNodeTasks(), equalTo(0));
             });
 
-            // User does not have create_follow_index index privilege for 'unallowedIndex':
+            assertOK(client().performRequest(new Request("POST", "/" + allowedIndex + "/_close")));
-            Exception e = expectThrows(ResponseException.class,
+            assertOK(client().performRequest(new Request("POST", "/" + allowedIndex + "/_ccr/unfollow")));
-                () -> follow("leader_cluster:" + unallowedIndex, unallowedIndex));
+            Exception e = expectThrows(ResponseException.class, () -> resumeFollow("leader_cluster:" + allowedIndex, allowedIndex));
+            assertThat(e.getMessage(), containsString("follow index [" + allowedIndex + "] does not have ccr metadata"));
+
+            // User does not have manage_follow_index index privilege for 'unallowedIndex':
+            e = expectThrows(ResponseException.class, () -> follow("leader_cluster:" + unallowedIndex, unallowedIndex));
             assertThat(e.getMessage(),
                 containsString("action [indices:admin/xpack/ccr/put_follow] 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)));
 
-            // User does have create_follow_index index privilege on 'allowed' index,
+            // User does have manage_follow_index index privilege on 'allowed' index,
             // but not read / monitor roles on 'disallowed' index:
             e = expectThrows(ResponseException.class,
                 () -> follow("leader_cluster:" + unallowedIndex, allowedIndex));
@@ -131,6 +135,10 @@ public class FollowIndexSecurityIT extends ESRestTestCase {
                 "privilege for action [indices:data/read/xpack/ccr/shard_changes] is missing"));
             assertThat(indexExists(adminClient(), unallowedIndex), is(false));
             assertBusy(() -> assertThat(countCcrNodeTasks(), equalTo(0)));
+
+            e = expectThrows(ResponseException.class,
+                () -> client().performRequest(new Request("POST", "/" + unallowedIndex + "/_ccr/unfollow")));
+            assertThat(e.getMessage(), containsString("action [indices:admin/xpack/ccr/unfollow] is unauthorized for user [test_ccr]"));
         }
     }
 

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

@@ -80,6 +80,12 @@ public class FollowIndexIT extends ESRestTestCase {
             }
             assertBusy(() -> verifyDocuments(followIndexName, numDocs + 3));
             assertBusy(() -> verifyCcrMonitoring(leaderIndexName, followIndexName));
+
+            pauseFollow(followIndexName);
+            assertOK(client().performRequest(new Request("POST", "/" + followIndexName + "/_close")));
+            assertOK(client().performRequest(new Request("POST", "/" + followIndexName + "/_ccr/unfollow")));
+            Exception e = expectThrows(ResponseException.class, () -> resumeFollow("leader_cluster:" + leaderIndexName, followIndexName));
+            assertThat(e.getMessage(), containsString("follow index [" + followIndexName + "] does not have ccr metadata"));
         }
     }
 

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

@@ -8,6 +8,8 @@ package org.elasticsearch.xpack.core.ccr.action;
 
 import org.elasticsearch.action.Action;
 import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -20,7 +22,7 @@ import static org.elasticsearch.action.ValidateActions.addValidationError;
 public class UnfollowAction extends Action<AcknowledgedResponse> {
 
     public static final UnfollowAction INSTANCE = new UnfollowAction();
-    public static final String NAME = "cluster:admin/xpack/ccr/unfollow";
+    public static final String NAME = "indices:admin/xpack/ccr/unfollow";
 
     private UnfollowAction() {
         super(NAME);
@@ -31,7 +33,7 @@ public class UnfollowAction extends Action<AcknowledgedResponse> {
         return new AcknowledgedResponse();
     }
 
-    public static class Request extends AcknowledgedRequest<Request> {
+    public static class Request extends AcknowledgedRequest<Request> implements IndicesRequest {
 
         private final String followerIndex;
 
@@ -48,6 +50,16 @@ public class UnfollowAction extends Action<AcknowledgedResponse> {
             return followerIndex;
         }
 
+        @Override
+        public String[] indices() {
+            return new String[] {followerIndex};
+        }
+
+        @Override
+        public IndicesOptions indicesOptions() {
+            return IndicesOptions.strictSingleIndexNoExpandForbidClosed();
+        }
+
         @Override
         public ActionRequestValidationException validate() {
             ActionRequestValidationException e = null;

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

@@ -9,6 +9,7 @@ import org.apache.lucene.util.automaton.Automaton;
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsAction;
 import org.elasticsearch.action.admin.indices.alias.exists.AliasesExistAction;
 import org.elasticsearch.action.admin.indices.alias.get.GetAliasesAction;
+import org.elasticsearch.action.admin.indices.close.CloseIndexAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexAction;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
 import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsAction;
@@ -22,6 +23,7 @@ import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryAction
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.xpack.core.ccr.action.PutFollowAction;
+import org.elasticsearch.xpack.core.ccr.action.UnfollowAction;
 import org.elasticsearch.xpack.core.security.support.Automatons;
 
 import java.util.Arrays;
@@ -56,7 +58,8 @@ 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(PutFollowAction.NAME);
+    private static final Automaton MANAGE_FOLLOW_INDEX_AUTOMATON = patterns(PutFollowAction.NAME, UnfollowAction.NAME,
+        CloseIndexAction.NAME);
 
     public static final IndexPrivilege NONE =                new IndexPrivilege("none",             Automatons.EMPTY);
     public static final IndexPrivilege ALL =                 new IndexPrivilege("all",              ALL_AUTOMATON);
@@ -71,7 +74,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);
+    public static final IndexPrivilege MANAGE_FOLLOW_INDEX = new IndexPrivilege("manage_follow_index", MANAGE_FOLLOW_INDEX_AUTOMATON);
 
     private static final Map<String, IndexPrivilege> VALUES = MapBuilder.<String, IndexPrivilege>newMapBuilder()
             .put("none", NONE)
@@ -87,7 +90,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)
+            .put("manage_follow_index", MANAGE_FOLLOW_INDEX)
             .immutableMap();
 
     public static final Predicate<String> ACTION_MATCHER = ALL.predicate();