model the data (non tested)

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/role/PutRoleRequest.java

@@ -125,6 +125,7 @@ public void addRemoteIndex(
         );
     }
 
+    //TODO: who calls this ?
     public void addRemoteCluster(final String[] privileges, final String[] remoteClusters) {
         remoteClusterPrivileges.add(new RoleDescriptor.RemoteClusterPrivileges(remoteClusters, privileges));
     }

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

@@ -31,6 +31,7 @@
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.PrivilegesToCheck;
 import org.elasticsearch.xpack.core.security.authz.permission.FieldPermissionsCache;
 import org.elasticsearch.xpack.core.security.authz.permission.FieldPermissionsDefinition;
+import org.elasticsearch.xpack.core.security.authz.permission.RemoteClusterPermissions;
 import org.elasticsearch.xpack.core.security.authz.privilege.ConfigurableClusterPrivilege;
 import org.elasticsearch.xpack.core.security.authz.privilege.ConfigurableClusterPrivileges;
 import org.elasticsearch.xpack.core.security.support.Validation;
@@ -1070,22 +1071,33 @@ private static ApplicationResourcePrivileges parseApplicationPrivilege(String ro
         return builder.build();
     }
 
+    //TODO: can this be replace completely by RemoteClusterPermissions.RemoteClusterGroup ?
     public static final class RemoteClusterPrivileges implements Writeable, ToXContentObject {
 
         private static final RemoteClusterPrivileges[] NONE = new RemoteClusterPrivileges[0];
-        private final String[] clusterPrivileges;
-        private final String[] remoteClusters;
+        private final RemoteClusterPermissions.RemoteClusterGroup remoteClusterGroup;
 
         public RemoteClusterPrivileges(String[] clusterPrivileges, String[] remoteClusters) {
-            this.clusterPrivileges = clusterPrivileges;
-            this.remoteClusters = remoteClusters;
+            remoteClusterGroup = new RemoteClusterPermissions.RemoteClusterGroup(clusterPrivileges, remoteClusters);
+        }
+
+        public String[] clusterPrivileges() {
+            return remoteClusterGroup.clusterPrivileges();
+        }
+
+        public String[] remoteClusters() {
+            return remoteClusterGroup.remoteClusterAliases();
+        }
+
+        public RemoteClusterPermissions.RemoteClusterGroup remoteClusterGroup() {
+            return remoteClusterGroup;
         }
 
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();
-            builder.array(Fields.PRIVILEGES.getPreferredName(), clusterPrivileges);
-            builder.array(Fields.CLUSTERS.getPreferredName(), remoteClusters);
+            builder.array(Fields.PRIVILEGES.getPreferredName(), remoteClusterGroup.clusterPrivileges());
+            builder.array(Fields.CLUSTERS.getPreferredName(), remoteClusterGroup.remoteClusterAliases());
             builder.endObject();
             return builder;
         }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/LimitedRole.java

@@ -75,6 +75,11 @@ public RemoteIndicesPermission remoteIndices() {
         throw new UnsupportedOperationException("cannot retrieve remote indices permission on limited role");
     }
 
+    @Override
+    public RemoteClusterPermissions remoteCluster() {
+        throw new UnsupportedOperationException("cannot retrieve remote cluster permission on limited role");
+    }
+
     @Override
     public boolean hasWorkflowsRestriction() {
         return baseRole.hasWorkflowsRestriction() || limitedByRole.hasWorkflowsRestriction();

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/RemoteClusterPermissions.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.security.authz.permission;
+
+import org.elasticsearch.xpack.core.security.support.StringMatcher;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+
+public class RemoteClusterPermissions {
+
+    private final List<RemoteClusterGroup> remoteClusterGroups;
+
+    public static final RemoteClusterPermissions NONE = new RemoteClusterPermissions(List.of());
+
+    private RemoteClusterPermissions(List<RemoteClusterGroup> remoteClusterGroups) {
+        this.remoteClusterGroups = remoteClusterGroups;
+    }
+
+    public String[] privilegeNames(final String remoteClusterAlias) {
+        return
+            remoteClusterGroups.stream()
+                .filter(group -> group.hasPrivileges(remoteClusterAlias))
+                .flatMap(groups -> Arrays.stream(groups.clusterPrivileges)).distinct().sorted().toArray(String[]::new);
+    }
+
+    public boolean hasPrivileges(final String remoteClusterAlias) {
+        return remoteClusterGroups.stream()
+            .anyMatch(remoteIndicesGroup -> remoteIndicesGroup.hasPrivileges(remoteClusterAlias));
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static class Builder {
+        final List<RemoteClusterGroup> remoteClusterGroupsList; //aliases -> permissions
+
+        public Builder() {
+            this.remoteClusterGroupsList = new ArrayList<>();
+        }
+
+        public Builder addGroup(RemoteClusterGroup remoteClusterGroup)
+
+        {
+            remoteClusterGroupsList.add(remoteClusterGroup);
+            return this;
+        }
+
+        public RemoteClusterPermissions build() {
+            return new RemoteClusterPermissions(remoteClusterGroupsList);
+        }
+    }
+
+    //TODO: pull out to top level, implement toXContent, Writable
+    // and replace org.elasticsearch.xpack.core.security.authz.RoleDescriptor.RemoteClusterPrivileges
+    public record RemoteClusterGroup(
+        String[] clusterPrivileges,
+        String[] remoteClusterAliases,
+        StringMatcher remoteClusterAliasMatcher
+
+    ) {
+        public RemoteClusterGroup(String[] clusterPrivileges, String[] remoteClusterAliases) {
+            this(clusterPrivileges,remoteClusterAliases, StringMatcher.of(remoteClusterAliases));
+        }
+
+        public boolean hasPrivileges(final String remoteClusterAlias) {
+            return remoteClusterAliasMatcher.test(remoteClusterAlias);
+        }
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/Role.java

@@ -62,6 +62,8 @@ public interface Role {
 
     RemoteIndicesPermission remoteIndices();
 
+    RemoteClusterPermissions remoteCluster();
+
     boolean hasWorkflowsRestriction();
 
     /**
@@ -214,7 +216,8 @@ class Builder {
         private ClusterPermission cluster = ClusterPermission.NONE;
         private RunAsPermission runAs = RunAsPermission.NONE;
         private final List<IndicesPermissionGroupDefinition> groups = new ArrayList<>();
-        private final Map<Set<String>, List<IndicesPermissionGroupDefinition>> remoteGroups = new HashMap<>();
+        private final Map<Set<String>, List<IndicesPermissionGroupDefinition>> remoteIndicesGroups = new HashMap<>();
+        private final List<RoleDescriptor.RemoteClusterPrivileges> remoteClusterGroups = new ArrayList<>();
         private final List<Tuple<ApplicationPrivilege, Set<String>>> applicationPrivs = new ArrayList<>();
         private final RestrictedIndices restrictedIndices;
         private WorkflowsRestriction workflowsRestriction = WorkflowsRestriction.NONE;
@@ -259,19 +262,40 @@ public Builder add(
             return this;
         }
 
-        public Builder addRemoteGroup(
+        public Builder addRemoteIndicesGroup(
             final Set<String> remoteClusterAliases,
             final FieldPermissions fieldPermissions,
             final Set<BytesReference> query,
             final IndexPrivilege privilege,
             final boolean allowRestrictedIndices,
             final String... indices
         ) {
-            remoteGroups.computeIfAbsent(remoteClusterAliases, k -> new ArrayList<>())
+            remoteIndicesGroups.computeIfAbsent(remoteClusterAliases, k -> new ArrayList<>())
                 .add(new IndicesPermissionGroupDefinition(privilege, fieldPermissions, query, allowRestrictedIndices, indices));
             return this;
         }
 
+        public Builder addRemoteClusterGroup( RoleDescriptor.RemoteClusterPrivileges remoteClusterPrivileges ){
+
+            if (remoteClusterPrivileges.clusterPrivileges().length > 0) {
+                Set<String> namedClusterPrivileges = ClusterPrivilegeResolver.names();
+                for (String namedPrivilege : remoteClusterPrivileges.clusterPrivileges()) {
+                    if("monitor_enrich".equals(namedPrivilege) == false){
+                        //this should be enforced upstream while defining the role, but check here too just in case...
+                        throw new IllegalArgumentException("Only [monitor_enrich] is supported as a remote cluster privilege");
+                    }
+                    // this can never happen, but if we ever expand the list of remote cluster privileges then we want to ensure that only
+                    // named cluster privileges are supported
+                    if (namedClusterPrivileges.contains(namedPrivilege) == false) {
+                        throw new IllegalArgumentException("Unknown cluster privilege [" + namedPrivilege + "]");
+                    }
+
+                }
+                this.remoteClusterGroups.add(remoteClusterPrivileges);
+            }
+
+            return this;
+        }
         public Builder addApplicationPrivilege(ApplicationPrivilege privilege, Set<String> resources) {
             applicationPrivs.add(new Tuple<>(privilege, resources));
             return this;
@@ -304,12 +328,13 @@ public SimpleRole build() {
                 indices = indicesBuilder.build();
             }
 
-            final RemoteIndicesPermission remoteIndices;
-            if (remoteGroups.isEmpty()) {
-                remoteIndices = RemoteIndicesPermission.NONE;
+            final RemoteIndicesPermission remoteIndicesPermission;
+            if (remoteIndicesGroups.isEmpty()) {
+                remoteIndicesPermission = RemoteIndicesPermission.NONE;
             } else {
                 final RemoteIndicesPermission.Builder remoteIndicesBuilder = new RemoteIndicesPermission.Builder();
-                for (final Map.Entry<Set<String>, List<IndicesPermissionGroupDefinition>> remoteGroupEntry : remoteGroups.entrySet()) {
+                for (final Map.Entry<Set<String>, List<IndicesPermissionGroupDefinition>> remoteGroupEntry : remoteIndicesGroups
+                    .entrySet()) {
                     final var clusterAlias = remoteGroupEntry.getKey();
                     for (IndicesPermissionGroupDefinition group : remoteGroupEntry.getValue()) {
                         remoteIndicesBuilder.addGroup(
@@ -322,13 +347,33 @@ public SimpleRole build() {
                         );
                     }
                 }
-                remoteIndices = remoteIndicesBuilder.build();
+                remoteIndicesPermission = remoteIndicesBuilder.build();
+            }
+
+            final RemoteClusterPermissions remoteClusterPermissions;
+            if (remoteClusterGroups.isEmpty()) {
+                remoteClusterPermissions = RemoteClusterPermissions.NONE;
+            } else {
+                final RemoteClusterPermissions.Builder remoteClusterBuilder = new RemoteClusterPermissions.Builder();
+                for (RoleDescriptor.RemoteClusterPrivileges remoteClusterPrivilege : remoteClusterGroups) {
+                    remoteClusterBuilder.addGroup(remoteClusterPrivilege.remoteClusterGroup());
+                }
+                remoteClusterPermissions = remoteClusterBuilder.build();
             }
 
             final ApplicationPermission applicationPermission = applicationPrivs.isEmpty()
                 ? ApplicationPermission.NONE
                 : new ApplicationPermission(applicationPrivs);
-            return new SimpleRole(names, cluster, indices, applicationPermission, runAs, remoteIndices, workflowsRestriction);
+            return new SimpleRole(
+                names,
+                cluster,
+                indices,
+                applicationPermission,
+                runAs,
+                remoteIndicesPermission,
+                remoteClusterPermissions,
+                workflowsRestriction
+            );
         }
 
         private static class IndicesPermissionGroupDefinition {
@@ -394,7 +439,7 @@ static SimpleRole buildFromRoleDescriptor(
             assert Arrays.equals(new String[] { "*" }, clusterAliases)
                 : "reserved role should not define remote indices privileges for specific clusters";
             final RoleDescriptor.IndicesPrivileges indicesPrivileges = remoteIndicesPrivileges.indicesPrivileges();
-            builder.addRemoteGroup(
+            builder.addRemoteIndicesGroup(
                 Set.of(clusterAliases),
                 fieldPermissionsCache.getFieldPermissions(
                     new FieldPermissionsDefinition(indicesPrivileges.getGrantedFields(), indicesPrivileges.getDeniedFields())
@@ -406,6 +451,13 @@ static SimpleRole buildFromRoleDescriptor(
             );
         }
 
+        for(RoleDescriptor.RemoteClusterPrivileges remoteClusterPrivileges : roleDescriptor.getRemoteClusterPrivileges()) {
+            final String[] clusterAliases = remoteClusterPrivileges.remoteClusters();
+            assert Arrays.equals(new String[] { "*" }, clusterAliases)
+                : "reserved role should not define remote cluster privileges for specific clusters";
+            builder.addRemoteClusterGroup(remoteClusterPrivileges);
+        }
+
         for (RoleDescriptor.ApplicationResourcePrivileges applicationPrivilege : roleDescriptor.getApplicationPrivileges()) {
             ApplicationPrivilege.get(
                 applicationPrivilege.getApplication(),

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/SimpleRole.java

@@ -51,7 +51,8 @@ public class SimpleRole implements Role {
     private final IndicesPermission indices;
     private final ApplicationPermission application;
     private final RunAsPermission runAs;
-    private final RemoteIndicesPermission remoteIndices;
+    private final RemoteIndicesPermission remoteIndicesPermission;
+    private final RemoteClusterPermissions remoteClusterPermissions;
     private final WorkflowsRestriction workflowsRestriction;
 
     SimpleRole(
@@ -60,15 +61,17 @@ public class SimpleRole implements Role {
         IndicesPermission indices,
         ApplicationPermission application,
         RunAsPermission runAs,
-        RemoteIndicesPermission remoteIndices,
+        RemoteIndicesPermission remoteIndicesPermission,
+        RemoteClusterPermissions remoteClusterPermissions,
         WorkflowsRestriction workflowsRestriction
     ) {
         this.names = names;
         this.cluster = Objects.requireNonNull(cluster);
         this.indices = Objects.requireNonNull(indices);
         this.application = Objects.requireNonNull(application);
         this.runAs = Objects.requireNonNull(runAs);
-        this.remoteIndices = Objects.requireNonNull(remoteIndices);
+        this.remoteIndicesPermission = Objects.requireNonNull(remoteIndicesPermission);
+        this.remoteClusterPermissions = Objects.requireNonNull(remoteClusterPermissions);
         this.workflowsRestriction = Objects.requireNonNull(workflowsRestriction);
     }
 
@@ -99,7 +102,12 @@ public RunAsPermission runAs() {
 
     @Override
     public RemoteIndicesPermission remoteIndices() {
-        return remoteIndices;
+        return remoteIndicesPermission;
+    }
+
+    @Override
+    public RemoteClusterPermissions remoteCluster() {
+        return remoteClusterPermissions;
     }
 
     @Override
@@ -195,10 +203,13 @@ public IndicesAccessControl authorize(
 
     @Override
     public RoleDescriptorsIntersection getRoleDescriptorsIntersectionForRemoteCluster(final String remoteClusterAlias) {
-        final RemoteIndicesPermission remoteIndicesPermission = remoteIndices.forCluster(remoteClusterAlias);
-        if (remoteIndicesPermission.remoteIndicesGroups().isEmpty()) {
+        final RemoteIndicesPermission remoteIndicesPermission = this.remoteIndicesPermission.forCluster(remoteClusterAlias);
+
+        if (remoteIndicesPermission.remoteIndicesGroups().isEmpty()
+            && remoteClusterPermissions.hasPrivileges(remoteClusterAlias) == false) {
             return RoleDescriptorsIntersection.EMPTY;
         }
+        //TODO: test the case where only remote cluster permissions are present, but no indices permissions are defined (and vice versa)
         final List<RoleDescriptor.IndicesPrivileges> indicesPrivileges = new ArrayList<>();
         for (RemoteIndicesPermission.RemoteIndicesGroup remoteIndicesGroup : remoteIndicesPermission.remoteIndicesGroups()) {
             for (IndicesPermission.Group indicesGroup : remoteIndicesGroup.indicesPermissionGroups()) {
@@ -209,7 +220,7 @@ public RoleDescriptorsIntersection getRoleDescriptorsIntersectionForRemoteCluste
         return new RoleDescriptorsIntersection(
             new RoleDescriptor(
                 REMOTE_USER_ROLE_NAME,
-                null,
+                remoteClusterPermissions.privilegeNames(remoteClusterAlias),
                 // The role descriptors constructed here may be cached in raw byte form, using a hash of their content as a
                 // cache key; we therefore need deterministic order when constructing them here, to ensure cache hits for
                 // equivalent role descriptors