Add new internal cross-project search user (#135531)

This PR introduces a dedicated `_cross_project_search` user to 
perform internal orchestration actions for cross project operations, 
like establishing remote cluster handshakes, cancelling cross-
cluster searches, etc...

Author: slobodanadamovic

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/user/InternalUsers.java

@@ -29,6 +29,8 @@
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.search.TransportSearchScrollAction;
 import org.elasticsearch.index.reindex.ReindexAction;
+import org.elasticsearch.tasks.TaskCancellationService;
+import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.ilm.action.ILMActions;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
@@ -294,6 +296,29 @@ public class InternalUsers {
         )
     );
 
+    /**
+     * Internal user that can manage a cross-project connections (e.g. handshake)
+     * and searches (e.g. cancelling).
+     */
+    public static final InternalUser CROSS_PROJECT_SEARCH_USER = new InternalUser(
+        UsernamesField.CROSS_PROJECT_SEARCH_USER_NAME,
+        new RoleDescriptor(
+            UsernamesField.CROSS_PROJECT_SEARCH_ROLE_NAME,
+            new String[] {
+                RemoteClusterService.REMOTE_CLUSTER_HANDSHAKE_ACTION_NAME,
+                TaskCancellationService.REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
+                TaskCancellationService.REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME,
+                "cluster:internal:data/read/esql/open_exchange",
+                "cluster:internal:data/read/esql/exchange" },
+            null,
+            null,
+            null,
+            null,
+            MetadataUtils.DEFAULT_RESERVED_METADATA,
+            Map.of()
+        )
+    );
+
     public static final SystemUser SYSTEM_USER = SystemUser.INSTANCE;
 
     private static final Map<String, InternalUser> INTERNAL_USERS;
@@ -309,7 +334,8 @@ public class InternalUsers {
             DATA_STREAM_LIFECYCLE_USER,
             REINDEX_DATA_STREAM_USER,
             SYNONYMS_USER,
-            LAZY_ROLLOVER_USER
+            LAZY_ROLLOVER_USER,
+            CROSS_PROJECT_SEARCH_USER
         ).collect(Collectors.toUnmodifiableMap(InternalUser::principal, Function.identity()));
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/user/UsernamesField.java

@@ -34,6 +34,8 @@ public final class UsernamesField {
     public static final String STORAGE_ROLE_NAME = "_storage";
     public static final String SYNONYMS_USER_NAME = "_synonyms";
     public static final String SYNONYMS_ROLE_NAME = "_synonyms";
+    public static final String CROSS_PROJECT_SEARCH_USER_NAME = "_cross_project_search";
+    public static final String CROSS_PROJECT_SEARCH_ROLE_NAME = "_cross_project_search";
 
     public static final String REMOTE_MONITORING_NAME = "remote_monitoring_user";
     public static final String REMOTE_MONITORING_COLLECTION_ROLE = "remote_monitoring_collector";

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/user/InternalUsersTests.java

@@ -39,15 +39,22 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.reindex.ReindexAction;
+import org.elasticsearch.tasks.TaskCancellationService;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.XPackPlugin;
+import org.elasticsearch.xpack.core.action.XPackInfoAction;
 import org.elasticsearch.xpack.core.ml.action.UpdateJobAction;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationTestHelper;
+import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.permission.ApplicationPermission;
 import org.elasticsearch.xpack.core.security.authz.permission.ClusterPermission;
 import org.elasticsearch.xpack.core.security.authz.permission.FieldPermissionsCache;
+import org.elasticsearch.xpack.core.security.authz.permission.IndicesPermission;
+import org.elasticsearch.xpack.core.security.authz.permission.RemoteClusterPermissions;
 import org.elasticsearch.xpack.core.security.authz.permission.RemoteIndicesPermission;
 import org.elasticsearch.xpack.core.security.authz.permission.Role;
 import org.elasticsearch.xpack.core.security.authz.permission.RunAsPermission;
@@ -57,11 +64,13 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 
 import static org.elasticsearch.xpack.core.security.test.TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_7;
 import static org.elasticsearch.xpack.core.security.test.TestRestrictedIndices.INTERNAL_SECURITY_TOKENS_INDEX_7;
 import static org.elasticsearch.xpack.core.security.test.TestRestrictedIndices.SECURITY_MAIN_ALIAS;
 import static org.elasticsearch.xpack.core.security.test.TestRestrictedIndices.SECURITY_TOKENS_ALIAS;
+import static org.elasticsearch.xpack.core.security.user.UsernamesField.CROSS_PROJECT_SEARCH_USER_NAME;
 import static org.elasticsearch.xpack.core.security.user.UsernamesField.REINDEX_DATA_STREAM_NAME;
 import static org.hamcrest.Matchers.arrayContaining;
 import static org.hamcrest.Matchers.equalTo;
@@ -358,6 +367,47 @@ public void testRegularUser() {
         expectThrows(IllegalStateException.class, () -> InternalUsers.getUser(username));
     }
 
+    public void testCrossProjectSearchUser() {
+        final InternalUser crossProjectSearchUser = InternalUsers.CROSS_PROJECT_SEARCH_USER;
+        assertThat(InternalUsers.getUser(CROSS_PROJECT_SEARCH_USER_NAME), is(crossProjectSearchUser));
+
+        assertThat(crossProjectSearchUser.getRemoteAccessRoleDescriptor().isPresent(), equalTo(false));
+
+        Optional<RoleDescriptor> localClusterRoleDescriptor = crossProjectSearchUser.getLocalClusterRoleDescriptor();
+        assertThat(localClusterRoleDescriptor.isPresent(), equalTo(true));
+        assertThat(localClusterRoleDescriptor.get().getMetadata(), equalTo(MetadataUtils.DEFAULT_RESERVED_METADATA));
+
+        final SimpleRole role = getLocalClusterRole(crossProjectSearchUser);
+
+        assertThat(role.indices(), is(IndicesPermission.NONE));
+        assertThat(role.runAs(), is(RunAsPermission.NONE));
+        assertThat(role.application(), is(ApplicationPermission.NONE));
+        assertThat(role.remoteIndices(), is(RemoteIndicesPermission.NONE));
+        assertThat(role.remoteCluster(), is(RemoteClusterPermissions.NONE));
+        assertThat(role.hasFieldOrDocumentLevelSecurity(), is(false));
+        assertThat(role.hasWorkflowsRestriction(), is(false));
+
+        final List<String> allowedClusterActions = List.of(
+            RemoteClusterService.REMOTE_CLUSTER_HANDSHAKE_ACTION_NAME,
+            TaskCancellationService.REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
+            TaskCancellationService.REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME,
+            "cluster:internal:data/read/esql/open_exchange",
+            "cluster:internal:data/read/esql/exchange"
+        );
+
+        for (String clusterAction : allowedClusterActions) {
+            checkClusterAccess(crossProjectSearchUser, role, clusterAction, true);
+        }
+
+        checkClusterAccess(
+            crossProjectSearchUser,
+            role,
+            randomFrom(ClusterStateAction.NAME, XPackInfoAction.NAME, TransportService.HANDSHAKE_ACTION_NAME),
+            false
+        );
+
+    }
+
     private static SimpleRole getLocalClusterRole(InternalUser internalUser) {
         final FieldPermissionsCache fieldPermissionsCache = new FieldPermissionsCache(Settings.EMPTY);
         return Role.buildFromRoleDescriptor(