Get remote access privileges for authenticated subject (#91734)

This PR adds support for resolving a remote access role descriptors
intersection for an authenticated subject, on the querying cluster. It
does so by using the role info available in authorization information
(for RBACEngine) and provides an interface method for custom engines to
likewise implement support. The intersection represents the indices
privileges a given user has towards a target remote cluster.

This functionality is not hooked up to any active flows yet. It will be
used in the security transport interceptor to construct the remote
access header to be sent from the querying cluster to the fulfilling
cluster for RCS 2.0. The POC PR highlights the tentative usage of
functionality added in this PR.

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

@@ -236,6 +236,18 @@ public interface AuthorizationEngine {
      */
     void getUserPrivileges(AuthorizationInfo authorizationInfo, ActionListener<GetUserPrivilegesResponse> listener);
 
+    /**
+     * Retrieve remote access privileges for a given target cluster, from the provided authorization information, to be sent together
+     * with a cross-cluster request (e.g. CCS) from an originating cluster to the target cluster.
+     */
+    default void getRemoteAccessRoleDescriptorsIntersection(
+        final String remoteClusterAlias,
+        final AuthorizationInfo authorizationInfo,
+        final ActionListener<RoleDescriptorsIntersection> listener
+    ) {
+        throw new UnsupportedOperationException("retrieving remote access role descriptors is not supported by this authorization engine");
+    }
+
     /**
      * Interface for objects that contains the information needed to authorize a request
      */

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

@@ -1034,7 +1034,7 @@ public class RoleDescriptor implements ToXContentObject, Writeable {
      * A class representing permissions for a group of indices mapped to
      * privileges, field permissions, and a query.
      */
-    public static class IndicesPrivileges implements ToXContentObject, Writeable {
+    public static class IndicesPrivileges implements ToXContentObject, Writeable, Comparable<IndicesPrivileges> {
 
         private static final IndicesPrivileges[] NONE = new IndicesPrivileges[0];
 
@@ -1214,6 +1214,35 @@ public class RoleDescriptor implements ToXContentObject, Writeable {
             privileges.writeTo(out);
         }
 
+        @Override
+        public int compareTo(IndicesPrivileges o) {
+            if (this == o) {
+                return 0;
+            }
+            int cmp = Boolean.compare(allowRestrictedIndices, o.allowRestrictedIndices);
+            if (cmp != 0) {
+                return cmp;
+            }
+            cmp = Arrays.compare(indices, o.indices);
+            if (cmp != 0) {
+                return cmp;
+            }
+            cmp = Arrays.compare(privileges, o.privileges);
+            if (cmp != 0) {
+                return cmp;
+            }
+            cmp = Objects.compare(query, o.query, Comparator.nullsFirst(BytesReference::compareTo));
+            if (cmp != 0) {
+                return cmp;
+            }
+            cmp = Arrays.compare(grantedFields, o.grantedFields);
+            if (cmp != 0) {
+                return cmp;
+            }
+            cmp = Arrays.compare(deniedFields, o.deniedFields);
+            return cmp;
+        }
+
         public static class Builder {
 
             private IndicesPrivileges indicesPrivileges = new IndicesPrivileges();

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

@@ -18,11 +18,14 @@ import org.elasticsearch.xcontent.XContentParser;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 
 public record RoleDescriptorsIntersection(Collection<Set<RoleDescriptor>> roleDescriptorsList) implements ToXContentObject, Writeable {
 
+    public static RoleDescriptorsIntersection EMPTY = new RoleDescriptorsIntersection(Collections.emptyList());
+
     public RoleDescriptorsIntersection(StreamInput in) throws IOException {
         this(in.readImmutableList(inner -> inner.readSet(RoleDescriptor::new)));
     }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authc/AuthenticationTestHelper.java

@@ -90,6 +90,10 @@ public class AuthenticationTestHelper {
         );
     }
 
+    public static User randomInternalUser() {
+        return ESTestCase.randomFrom(INTERNAL_USERS);
+    }
+
     public static User userWithRandomMetadataAndDetails(final String username, final String... roles) {
         return new User(
             username,

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/RoleDescriptorTests.java

@@ -48,7 +48,9 @@ import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.emptyArray;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.lessThan;
 import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.core.Is.is;
 
@@ -655,6 +657,79 @@ public class RoleDescriptorTests extends ESTestCase {
         );
     }
 
+    public void testIndicesPrivilegesCompareTo() {
+        final RoleDescriptor.IndicesPrivileges indexPrivilege = randomIndicesPrivilegesBuilder().build();
+        @SuppressWarnings({ "EqualsWithItself" })
+        final int actual = indexPrivilege.compareTo(indexPrivilege);
+        assertThat(actual, equalTo(0));
+        assertThat(
+            indexPrivilege.compareTo(
+                RoleDescriptor.IndicesPrivileges.builder()
+                    .indices(indexPrivilege.getIndices().clone())
+                    .privileges(indexPrivilege.getPrivileges().clone())
+                    // test for both cases when the query is the same instance or a copy
+                    .query(
+                        (indexPrivilege.getQuery() == null || randomBoolean())
+                            ? indexPrivilege.getQuery()
+                            : new BytesArray(indexPrivilege.getQuery().toBytesRef())
+                    )
+                    .grantedFields(indexPrivilege.getGrantedFields() == null ? null : indexPrivilege.getGrantedFields().clone())
+                    .deniedFields(indexPrivilege.getDeniedFields() == null ? null : indexPrivilege.getDeniedFields().clone())
+                    .allowRestrictedIndices(indexPrivilege.allowRestrictedIndices())
+                    .build()
+            ),
+            equalTo(0)
+        );
+
+        RoleDescriptor.IndicesPrivileges first = randomIndicesPrivilegesBuilder().allowRestrictedIndices(false).build();
+        RoleDescriptor.IndicesPrivileges second = randomIndicesPrivilegesBuilder().allowRestrictedIndices(true).build();
+        assertThat(first.compareTo(second), lessThan(0));
+        assertThat(second.compareTo(first), greaterThan(0));
+
+        first = randomIndicesPrivilegesBuilder().indices("a", "b").build();
+        second = randomIndicesPrivilegesBuilder().indices("b", "a").allowRestrictedIndices(first.allowRestrictedIndices()).build();
+        assertThat(first.compareTo(second), lessThan(0));
+        assertThat(second.compareTo(first), greaterThan(0));
+
+        first = randomIndicesPrivilegesBuilder().privileges("read", "write").build();
+        second = randomIndicesPrivilegesBuilder().allowRestrictedIndices(first.allowRestrictedIndices())
+            .privileges("write", "read")
+            .indices(first.getIndices())
+            .build();
+        assertThat(first.compareTo(second), lessThan(0));
+        assertThat(second.compareTo(first), greaterThan(0));
+
+        first = randomIndicesPrivilegesBuilder().query(randomBoolean() ? null : "{\"match\":{\"field-a\":\"a\"}}").build();
+        second = randomIndicesPrivilegesBuilder().allowRestrictedIndices(first.allowRestrictedIndices())
+            .query("{\"match\":{\"field-b\":\"b\"}}")
+            .indices(first.getIndices())
+            .privileges(first.getPrivileges())
+            .build();
+        assertThat(first.compareTo(second), lessThan(0));
+        assertThat(second.compareTo(first), greaterThan(0));
+
+        first = randomIndicesPrivilegesBuilder().grantedFields(randomBoolean() ? null : new String[] { "a", "b" }).build();
+        second = randomIndicesPrivilegesBuilder().allowRestrictedIndices(first.allowRestrictedIndices())
+            .grantedFields("b", "a")
+            .indices(first.getIndices())
+            .privileges(first.getPrivileges())
+            .query(first.getQuery())
+            .build();
+        assertThat(first.compareTo(second), lessThan(0));
+        assertThat(second.compareTo(first), greaterThan(0));
+
+        first = randomIndicesPrivilegesBuilder().deniedFields(randomBoolean() ? null : new String[] { "a", "b" }).build();
+        second = randomIndicesPrivilegesBuilder().allowRestrictedIndices(first.allowRestrictedIndices())
+            .deniedFields("b", "a")
+            .indices(first.getIndices())
+            .privileges(first.getPrivileges())
+            .query(first.getQuery())
+            .grantedFields(first.getGrantedFields())
+            .build();
+        assertThat(first.compareTo(second), lessThan(0));
+        assertThat(second.compareTo(first), greaterThan(0));
+    }
+
     public void testGlobalPrivilegesOrdering() throws IOException {
         final String roleName = randomAlphaOfLengthBetween(3, 30);
         final String[] applicationNames = generateRandomStringArray(3, randomIntBetween(0, 3), false, true);
@@ -892,30 +967,34 @@ public class RoleDescriptorTests extends ESTestCase {
         );
     }
 
-    private static RoleDescriptor.IndicesPrivileges[] randomIndicesPriveleges() {
+    public static RoleDescriptor.IndicesPrivileges[] randomIndicesPriveleges() {
         final RoleDescriptor.IndicesPrivileges[] indexPrivileges = new RoleDescriptor.IndicesPrivileges[randomIntBetween(0, 3)];
         for (int i = 0; i < indexPrivileges.length; i++) {
-            final RoleDescriptor.IndicesPrivileges.Builder builder = RoleDescriptor.IndicesPrivileges.builder()
-                .privileges(randomSubsetOf(randomIntBetween(1, 4), IndexPrivilege.names()))
-                .indices(generateRandomStringArray(5, randomIntBetween(3, 9), false, false))
-                .allowRestrictedIndices(randomBoolean());
-            if (randomBoolean()) {
-                builder.query(
-                    randomBoolean()
-                        ? "{ \"term\": { \"" + randomAlphaOfLengthBetween(3, 24) + "\" : \"" + randomAlphaOfLengthBetween(3, 24) + "\" }"
-                        : "{ \"match_all\": {} }"
-                );
-            }
+            indexPrivileges[i] = randomIndicesPrivilegesBuilder().build();
+        }
+        return indexPrivileges;
+    }
+
+    private static RoleDescriptor.IndicesPrivileges.Builder randomIndicesPrivilegesBuilder() {
+        final RoleDescriptor.IndicesPrivileges.Builder builder = RoleDescriptor.IndicesPrivileges.builder()
+            .privileges(randomSubsetOf(randomIntBetween(1, 4), IndexPrivilege.names()))
+            .indices(generateRandomStringArray(5, randomIntBetween(3, 9), false, false))
+            .allowRestrictedIndices(randomBoolean());
+        if (randomBoolean()) {
+            builder.query(
+                randomBoolean()
+                    ? "{ \"term\": { \"" + randomAlphaOfLengthBetween(3, 24) + "\" : \"" + randomAlphaOfLengthBetween(3, 24) + "\" }"
+                    : "{ \"match_all\": {} }"
+            );
+        }
+        if (randomBoolean()) {
             if (randomBoolean()) {
-                if (randomBoolean()) {
-                    builder.grantedFields("*");
-                    builder.deniedFields(generateRandomStringArray(4, randomIntBetween(4, 9), false, false));
-                } else {
-                    builder.grantedFields(generateRandomStringArray(4, randomIntBetween(4, 9), false, false));
-                }
+                builder.grantedFields("*");
+                builder.deniedFields(generateRandomStringArray(4, randomIntBetween(4, 9), false, false));
+            } else {
+                builder.grantedFields(generateRandomStringArray(4, randomIntBetween(4, 9), false, false));
             }
-            indexPrivileges[i] = builder.build();
         }
-        return indexPrivileges;
+        return builder;
     }
 }

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

@@ -61,6 +61,7 @@ import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.RequestIn
 import org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField;
 import org.elasticsearch.xpack.core.security.authz.ResolvedIndices;
 import org.elasticsearch.xpack.core.security.authz.RestrictedIndices;
+import org.elasticsearch.xpack.core.security.authz.RoleDescriptorsIntersection;
 import org.elasticsearch.xpack.core.security.authz.accesscontrol.IndicesAccessControl;
 import org.elasticsearch.xpack.core.security.authz.privilege.ApplicationPrivilegeDescriptor;
 import org.elasticsearch.xpack.core.security.authz.privilege.ClusterPrivilegeResolver;
@@ -202,6 +203,47 @@ public class AuthorizationService {
         authorizationEngine.getUserPrivileges(authorizationInfo, wrapPreservingContext(listener, threadContext));
     }
 
+    public void retrieveRemoteAccessRoleDescriptorsIntersection(
+        final String remoteClusterAlias,
+        final Subject subject,
+        final ActionListener<RoleDescriptorsIntersection> listener
+    ) {
+        if (SystemUser.is(subject.getUser())) {
+            final String message = "the user ["
+                + subject.getUser().principal()
+                + "] is the system user and we should never try to retrieve its remote access roles descriptors";
+            assert false : message;
+            listener.onFailure(new IllegalArgumentException(message));
+            return;
+        }
+
+        final AuthorizationEngine authorizationEngine = getAuthorizationEngineForSubject(subject);
+        final AuthorizationInfo authorizationInfo = threadContext.getTransient(AUTHORIZATION_INFO_KEY);
+        if (authorizationInfo != null) {
+            authorizationEngine.getRemoteAccessRoleDescriptorsIntersection(
+                remoteClusterAlias,
+                authorizationInfo,
+                wrapPreservingContext(listener, threadContext)
+            );
+        } else {
+            assert isInternal(subject.getUser())
+                : "authorization info must be available in thread context for all users other than internal users";
+            authorizationEngine.resolveAuthorizationInfo(
+                subject,
+                wrapPreservingContext(
+                    listener.delegateFailure(
+                        (delegatedLister, resolvedAuthzInfo) -> authorizationEngine.getRemoteAccessRoleDescriptorsIntersection(
+                            remoteClusterAlias,
+                            resolvedAuthzInfo,
+                            wrapPreservingContext(delegatedLister, threadContext)
+                        )
+                    ),
+                    threadContext
+                )
+            );
+        }
+    }
+
     /**
      * Verifies that the given user can execute the given request (and action). If the user doesn't
      * have the appropriate privileges for this action/request, an {@link ElasticsearchSecurityException}

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

@@ -61,6 +61,7 @@ import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine;
 import org.elasticsearch.xpack.core.security.authz.IndicesAndAliasesResolverField;
 import org.elasticsearch.xpack.core.security.authz.ResolvedIndices;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
+import org.elasticsearch.xpack.core.security.authz.RoleDescriptorsIntersection;
 import org.elasticsearch.xpack.core.security.authz.accesscontrol.IndicesAccessControl;
 import org.elasticsearch.xpack.core.security.authz.permission.FieldPermissionsCache;
 import org.elasticsearch.xpack.core.security.authz.permission.FieldPermissionsDefinition;
@@ -82,6 +83,7 @@ import org.elasticsearch.xpack.core.sql.SqlAsyncActionNames;
 import org.elasticsearch.xpack.security.authc.esnative.ReservedRealm;
 import org.elasticsearch.xpack.security.authz.store.CompositeRolesStore;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -118,6 +120,8 @@ public class RBACEngine implements AuthorizationEngine {
     private static final String DELETE_SUB_REQUEST_REPLICA = DeleteAction.NAME + "[r]";
 
     private static final Logger logger = LogManager.getLogger(RBACEngine.class);
+    // TODO move once we have a dedicated class for RCS 2.0 constants
+    public static final String REMOTE_USER_ROLE_NAME = "_remote_user";
 
     private final Settings settings;
     private final CompositeRolesStore rolesStore;
@@ -657,6 +661,94 @@ public class RBACEngine implements AuthorizationEngine {
         }
     }
 
+    @Override
+    public void getRemoteAccessRoleDescriptorsIntersection(
+        final String remoteClusterAlias,
+        final AuthorizationInfo authorizationInfo,
+        final ActionListener<RoleDescriptorsIntersection> listener
+    ) {
+        if (authorizationInfo instanceof RBACAuthorizationInfo == false) {
+            listener.onFailure(
+                new IllegalArgumentException("unsupported authorization info:" + authorizationInfo.getClass().getSimpleName())
+            );
+            return;
+        }
+
+        final Role role = ((RBACAuthorizationInfo) authorizationInfo).getRole();
+        final RemoteIndicesPermission remoteIndicesPermission;
+        try {
+            remoteIndicesPermission = role.remoteIndices().forCluster(remoteClusterAlias);
+        } catch (UnsupportedOperationException e) {
+            // TODO we will need to implement this to support API keys with assigned role descriptors
+            listener.onFailure(
+                new IllegalArgumentException(
+                    "cannot retrieve remote access role descriptors for API keys with assigned role descriptors.",
+                    e
+                )
+            );
+            return;
+        }
+
+        if (remoteIndicesPermission.remoteIndicesGroups().isEmpty()) {
+            listener.onResponse(RoleDescriptorsIntersection.EMPTY);
+            return;
+        }
+
+        final List<RoleDescriptor.IndicesPrivileges> indicesPrivileges = new ArrayList<>();
+        for (RemoteIndicesPermission.RemoteIndicesGroup remoteIndicesGroup : remoteIndicesPermission.remoteIndicesGroups()) {
+            for (IndicesPermission.Group indicesGroup : remoteIndicesGroup.indicesPermissionGroups()) {
+                indicesPrivileges.add(toIndicesPrivileges(indicesGroup));
+            }
+        }
+
+        listener.onResponse(
+            new RoleDescriptorsIntersection(
+                List.of(
+                    Set.of(
+                        new RoleDescriptor(
+                            REMOTE_USER_ROLE_NAME,
+                            null,
+                            // 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
+                            indicesPrivileges.stream().sorted().toArray(RoleDescriptor.IndicesPrivileges[]::new),
+                            null,
+                            null,
+                            null,
+                            null,
+                            null
+                        )
+                    )
+                )
+            )
+        );
+    }
+
+    private static RoleDescriptor.IndicesPrivileges toIndicesPrivileges(final IndicesPermission.Group indicesGroup) {
+        final Set<BytesReference> queries = indicesGroup.getQuery();
+        final Set<FieldPermissionsDefinition.FieldGrantExcludeGroup> fieldGrantExcludeGroups = getFieldGrantExcludeGroups(indicesGroup);
+        assert queries == null || queries.size() <= 1
+            : "translation from an indices permission group to indices privileges supports up to one DLS query but multiple queries found";
+        assert fieldGrantExcludeGroups.size() <= 1
+            : "translation from an indices permission group to indices privileges supports up to one FLS field-grant-exclude group"
+                + " but multiple groups found";
+
+        final BytesReference query = (queries == null || false == queries.iterator().hasNext()) ? null : queries.iterator().next();
+        final RoleDescriptor.IndicesPrivileges.Builder builder = RoleDescriptor.IndicesPrivileges.builder()
+            // Sort because these index privileges will be part of role descriptors that may be cached in raw byte form;
+            // we need deterministic order to ensure cache hits for equivalent role descriptors
+            .indices(Arrays.stream(indicesGroup.indices()).sorted().collect(Collectors.toList()))
+            .privileges(indicesGroup.privilege().name().stream().sorted().collect(Collectors.toList()))
+            .allowRestrictedIndices(indicesGroup.allowRestrictedIndices())
+            .query(query);
+        if (false == fieldGrantExcludeGroups.isEmpty()) {
+            final FieldPermissionsDefinition.FieldGrantExcludeGroup fieldGrantExcludeGroup = fieldGrantExcludeGroups.iterator().next();
+            builder.grantedFields(fieldGrantExcludeGroup.getGrantedFields()).deniedFields(fieldGrantExcludeGroup.getExcludedFields());
+        }
+
+        return builder.build();
+    }
+
     static GetUserPrivilegesResponse buildUserPrivilegesResponseObject(Role userRole) {
         logger.trace(() -> "List privileges for role [" + arrayToCommaDelimitedString(userRole.names()) + "]");
 
@@ -721,24 +813,27 @@ public class RBACEngine implements AuthorizationEngine {
 
     private static GetUserPrivilegesResponse.Indices toIndices(final IndicesPermission.Group group) {
         final Set<BytesReference> queries = group.getQuery() == null ? Collections.emptySet() : group.getQuery();
-        final Set<FieldPermissionsDefinition.FieldGrantExcludeGroup> fieldSecurity;
+        final Set<FieldPermissionsDefinition.FieldGrantExcludeGroup> fieldSecurity = getFieldGrantExcludeGroups(group);
+        return new GetUserPrivilegesResponse.Indices(
+            Arrays.asList(group.indices()),
+            group.privilege().name(),
+            fieldSecurity,
+            queries,
+            group.allowRestrictedIndices()
+        );
+    }
+
+    private static Set<FieldPermissionsDefinition.FieldGrantExcludeGroup> getFieldGrantExcludeGroups(IndicesPermission.Group group) {
         if (group.getFieldPermissions().hasFieldLevelSecurity()) {
             final List<FieldPermissionsDefinition> fieldPermissionsDefinitions = group.getFieldPermissions()
                 .getFieldPermissionsDefinitions();
             assert fieldPermissionsDefinitions.size() == 1
                 : "limited-by field must not exist since we do not support reporting user privileges for limited roles";
             final FieldPermissionsDefinition definition = fieldPermissionsDefinitions.get(0);
-            fieldSecurity = definition.getFieldGrantExcludeGroups();
+            return definition.getFieldGrantExcludeGroups();
         } else {
-            fieldSecurity = Collections.emptySet();
+            return Collections.emptySet();
         }
-        return new GetUserPrivilegesResponse.Indices(
-            Arrays.asList(group.indices()),
-            group.privilege().name(),
-            fieldSecurity,
-            queries,
-            group.allowRestrictedIndices()
-        );
     }
 
     static AuthorizedIndices resolveAuthorizedIndicesFromRole(

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizationServiceIntegTests.java

@@ -0,0 +1,200 @@
+/*
+ * 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.security.authz;
+
+import org.elasticsearch.action.LatchedActionListener;
+import org.elasticsearch.action.support.ActionTestUtils;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.test.SecurityIntegTestCase;
+import org.elasticsearch.transport.TcpTransport;
+import org.elasticsearch.xpack.core.security.SecurityContext;
+import org.elasticsearch.xpack.core.security.action.user.AuthenticateAction;
+import org.elasticsearch.xpack.core.security.action.user.AuthenticateRequest;
+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.RoleDescriptorsIntersection;
+import org.elasticsearch.xpack.core.security.authz.privilege.ClusterPrivilegeResolver;
+import org.elasticsearch.xpack.core.security.authz.privilege.IndexPrivilege;
+import org.elasticsearch.xpack.core.security.support.NativeRealmValidationUtil;
+import org.elasticsearch.xpack.core.security.user.SystemUser;
+import org.elasticsearch.xpack.core.security.user.User;
+import org.elasticsearch.xpack.security.audit.AuditUtil;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField.AUTHORIZATION_INFO_KEY;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
+
+public class AuthorizationServiceIntegTests extends SecurityIntegTestCase {
+
+    @Override
+    protected boolean addMockHttpTransport() {
+        return false; // need real http
+    }
+
+    public void testRetrieveRemoteAccessRoleDescriptorsIntersectionForNonInternalUser() throws IOException, InterruptedException {
+        assumeTrue("untrusted remote cluster feature flag must be enabled", TcpTransport.isUntrustedRemoteClusterEnabled());
+
+        final String concreteClusterAlias = randomAlphaOfLength(10);
+        final String roleName = randomAlphaOfLength(5);
+        getSecurityClient().putRole(
+            new RoleDescriptor(
+                roleName,
+                randomSubsetOf(ClusterPrivilegeResolver.names()).toArray(String[]::new),
+                randomBoolean()
+                    ? null
+                    : new RoleDescriptor.IndicesPrivileges[] {
+                        RoleDescriptor.IndicesPrivileges.builder()
+                            .privileges(randomSubsetOf(randomIntBetween(1, 4), IndexPrivilege.names()))
+                            .indices(generateRandomStringArray(5, randomIntBetween(3, 9), false, false))
+                            .allowRestrictedIndices(randomBoolean())
+                            .build() },
+                null,
+                null,
+                generateRandomStringArray(5, randomIntBetween(2, 8), false, true),
+                null,
+                null,
+                new RoleDescriptor.RemoteIndicesPrivileges[] {
+                    new RoleDescriptor.RemoteIndicesPrivileges(
+                        RoleDescriptor.IndicesPrivileges.builder()
+                            .indices(shuffledList(List.of("index1", "index2")))
+                            .privileges(shuffledList(List.of("read", "write")))
+                            .build(),
+                        randomNonEmptySubsetOf(List.of(concreteClusterAlias, "*")).toArray(new String[0])
+                    ) }
+            )
+        );
+        final String nodeName = internalCluster().getRandomNodeName();
+        final ThreadContext threadContext = internalCluster().getInstance(SecurityContext.class, nodeName).getThreadContext();
+        final AuthorizationService authzService = internalCluster().getInstance(AuthorizationService.class, nodeName);
+        final Authentication authentication = Authentication.newRealmAuthentication(
+            new User(randomAlphaOfLengthBetween(5, 16), roleName),
+            new Authentication.RealmRef(
+                randomBoolean() ? "realm" : randomAlphaOfLengthBetween(1, 16),
+                randomAlphaOfLengthBetween(5, 16),
+                nodeName
+            )
+        );
+        final RoleDescriptorsIntersection actual = authorizeThenRetrieveRemoteAccessDescriptors(
+            threadContext,
+            authzService,
+            authentication,
+            concreteClusterAlias
+        );
+        final String generatedRoleName = actual.roleDescriptorsList().iterator().next().iterator().next().getName();
+        assertNull(NativeRealmValidationUtil.validateRoleName(generatedRoleName, false));
+        assertThat(generatedRoleName, not(equalTo(roleName)));
+        assertThat(
+            actual,
+            equalTo(
+                new RoleDescriptorsIntersection(
+                    List.of(
+                        Set.of(
+                            new RoleDescriptor(
+                                generatedRoleName,
+                                null,
+                                new RoleDescriptor.IndicesPrivileges[] {
+                                    RoleDescriptor.IndicesPrivileges.builder()
+                                        .indices("index1", "index2")
+                                        .privileges("read", "write")
+                                        .build() },
+                                null,
+                                null,
+                                null,
+                                null,
+                                null
+                            )
+                        )
+                    )
+                )
+            )
+        );
+    }
+
+    public void testRetrieveRemoteAccessRoleDescriptorsIntersectionForInternalUser() throws InterruptedException {
+        assumeTrue("untrusted remote cluster feature flag must be enabled", TcpTransport.isUntrustedRemoteClusterEnabled());
+
+        final String nodeName = internalCluster().getRandomNodeName();
+        final ThreadContext threadContext = internalCluster().getInstance(SecurityContext.class, nodeName).getThreadContext();
+        final AuthorizationService authzService = internalCluster().getInstance(AuthorizationService.class, nodeName);
+        final Authentication authentication = AuthenticationTestHelper.builder()
+            .internal(randomValueOtherThan(SystemUser.INSTANCE, AuthenticationTestHelper::randomInternalUser))
+            .build();
+        final String concreteClusterAlias = randomAlphaOfLength(10);
+
+        // For internal users, we support the situation where there is no authorization information populated in thread context
+        // We test both scenarios, one where we don't authorize and don't have authorization info in thread context, and one where we do
+        if (randomBoolean()) {
+            assertThat(threadContext.getTransient(AUTHORIZATION_INFO_KEY), nullValue());
+            final CountDownLatch latch = new CountDownLatch(1);
+            final AtomicReference<RoleDescriptorsIntersection> actual = new AtomicReference<>();
+            authzService.retrieveRemoteAccessRoleDescriptorsIntersection(
+                concreteClusterAlias,
+                authentication.getEffectiveSubject(),
+                new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(newValue -> {
+                    assertThat(threadContext.getTransient(AUTHORIZATION_INFO_KEY), nullValue());
+                    actual.set(newValue);
+                }), latch)
+            );
+            latch.await();
+            assertThat(actual.get(), equalTo(RoleDescriptorsIntersection.EMPTY));
+            // Validate original authz info is restored to null after call complete
+            assertThat(threadContext.getTransient(AUTHORIZATION_INFO_KEY), nullValue());
+        } else {
+            assertThat(
+                authorizeThenRetrieveRemoteAccessDescriptors(threadContext, authzService, authentication, concreteClusterAlias),
+                equalTo(RoleDescriptorsIntersection.EMPTY)
+            );
+        }
+    }
+
+    private RoleDescriptorsIntersection authorizeThenRetrieveRemoteAccessDescriptors(
+        final ThreadContext threadContext,
+        final AuthorizationService authzService,
+        final Authentication authentication,
+        final String concreteClusterAlias
+    ) throws InterruptedException {
+        try (var ignored = threadContext.stashContext()) {
+            assertThat(threadContext.getTransient(AUTHORIZATION_INFO_KEY), nullValue());
+            final AtomicReference<RoleDescriptorsIntersection> actual = new AtomicReference<>();
+            final CountDownLatch latch = new CountDownLatch(1);
+            // A request ID is set during authentication and is required for authorization; since we are not authenticating, set it
+            // explicitly
+            AuditUtil.generateRequestId(threadContext);
+            // Authorize to populate thread context with authz info
+            // Note that if the outer listener throws, we will not count down on the latch, however, we also won't get to the await call
+            // since the exception will be thrown before -- so no deadlock
+            authzService.authorize(
+                authentication,
+                AuthenticateAction.INSTANCE.name(),
+                AuthenticateRequest.INSTANCE,
+                ActionTestUtils.assertNoFailureListener(nothing -> {
+                    authzService.retrieveRemoteAccessRoleDescriptorsIntersection(
+                        concreteClusterAlias,
+                        authentication.getEffectiveSubject(),
+                        new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(newValue -> {
+                            assertThat(threadContext.getTransient(AUTHORIZATION_INFO_KEY), not(nullValue()));
+                            actual.set(newValue);
+                        }), latch)
+                    );
+                })
+            );
+            latch.await();
+            // Validate original authz info is restored after call complete
+            assertThat(threadContext.getTransient(AUTHORIZATION_INFO_KEY), nullValue());
+            return actual.get();
+        }
+    }
+}

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/RBACEngineTests.java

@@ -25,12 +25,14 @@ import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.license.GetLicenseAction;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.core.XPackPlugin;
@@ -60,8 +62,10 @@ import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.Authoriza
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.AuthorizedIndices;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.PrivilegesCheckResult;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.PrivilegesToCheck;
+import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor.ApplicationResourcePrivileges;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor.IndicesPrivileges;
+import org.elasticsearch.xpack.core.security.authz.RoleDescriptorsIntersection;
 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.FieldPermissions;
@@ -88,13 +92,16 @@ import org.junit.Before;
 import org.mockito.Mockito;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Locale;
+import java.util.Objects;
 import java.util.Set;
 import java.util.TreeMap;
+import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyMap;
@@ -123,6 +130,7 @@ import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
@@ -1576,6 +1584,201 @@ public class RBACEngineTests extends ESTestCase {
         assertThat(e.getCause(), sameInstance(unsupportedOperationException));
     }
 
+    public void testGetRemoteAccessRoleDescriptorsIntersection() throws ExecutionException, InterruptedException {
+        assumeTrue("untrusted remote cluster feature flag must be enabled", TcpTransport.isUntrustedRemoteClusterEnabled());
+
+        final RemoteIndicesPermission.Builder remoteIndicesBuilder = RemoteIndicesPermission.builder();
+        final String concreteClusterAlias = randomAlphaOfLength(10);
+        final int numGroups = randomIntBetween(1, 3);
+        final List<IndicesPrivileges> expectedIndicesPrivileges = new ArrayList<>();
+        for (int i = 0; i < numGroups; i++) {
+            final String[] indexNames = Objects.requireNonNull(generateRandomStringArray(3, 10, false, false));
+            final boolean allowRestrictedIndices = randomBoolean();
+            final boolean hasFls = randomBoolean();
+            final FieldPermissionsDefinition.FieldGrantExcludeGroup group = hasFls
+                ? randomFieldGrantExcludeGroup()
+                : new FieldPermissionsDefinition.FieldGrantExcludeGroup(null, null);
+            final BytesReference query = randomBoolean() ? randomDlsQuery() : null;
+            final IndicesPrivileges.Builder builder = IndicesPrivileges.builder()
+                .indices(Arrays.stream(indexNames).sorted().collect(Collectors.toList()))
+                .privileges("read")
+                .allowRestrictedIndices(allowRestrictedIndices)
+                .query(query);
+            if (hasFls) {
+                builder.grantedFields(group.getGrantedFields());
+                builder.deniedFields(group.getExcludedFields());
+            }
+            expectedIndicesPrivileges.add(builder.build());
+            remoteIndicesBuilder.addGroup(
+                Set.of(randomFrom(concreteClusterAlias, "*")),
+                IndexPrivilege.READ,
+                new FieldPermissions(new FieldPermissionsDefinition(Set.of(group))),
+                query == null ? null : Set.of(query),
+                allowRestrictedIndices,
+                indexNames
+            );
+        }
+        final String mismatchedConcreteClusterAlias = randomValueOtherThan(concreteClusterAlias, () -> randomAlphaOfLength(10));
+        // Add some groups that don't match the alias
+        final int numMismatchedGroups = randomIntBetween(0, 3);
+        for (int i = 0; i < numMismatchedGroups; i++) {
+            remoteIndicesBuilder.addGroup(
+                Set.of(mismatchedConcreteClusterAlias),
+                IndexPrivilege.READ,
+                new FieldPermissions(
+                    new FieldPermissionsDefinition(Set.of(new FieldPermissionsDefinition.FieldGrantExcludeGroup(null, null)))
+                ),
+                null,
+                randomBoolean(),
+                generateRandomStringArray(3, 10, false, false)
+            );
+        }
+
+        final Role role = mockRoleWithRemoteIndices(remoteIndicesBuilder.build());
+        final RBACAuthorizationInfo authorizationInfo = mock(RBACAuthorizationInfo.class);
+        when(authorizationInfo.getRole()).thenReturn(role);
+
+        final PlainActionFuture<RoleDescriptorsIntersection> future = new PlainActionFuture<>();
+        engine.getRemoteAccessRoleDescriptorsIntersection(concreteClusterAlias, authorizationInfo, future);
+        final RoleDescriptorsIntersection actual = future.get();
+
+        assertThat(
+            actual,
+            equalTo(
+                new RoleDescriptorsIntersection(
+                    List.of(
+                        Set.of(
+                            new RoleDescriptor(
+                                RBACEngine.REMOTE_USER_ROLE_NAME,
+                                null,
+                                expectedIndicesPrivileges.stream().sorted().toArray(RoleDescriptor.IndicesPrivileges[]::new),
+                                null,
+                                null,
+                                null,
+                                null,
+                                null
+                            )
+                        )
+                    )
+                )
+            )
+        );
+        verify(role, times(1)).remoteIndices();
+        verifyNoMoreInteractions(role);
+    }
+
+    public void testGetRemoteAccessRoleDescriptorsIntersectionHasDeterministicOrderForIndicesPrivileges() throws ExecutionException,
+        InterruptedException {
+        assumeTrue("untrusted remote cluster feature flag must be enabled", TcpTransport.isUntrustedRemoteClusterEnabled());
+
+        final RemoteIndicesPermission.Builder remoteIndicesBuilder = RemoteIndicesPermission.builder();
+        final String concreteClusterAlias = randomAlphaOfLength(10);
+        final int numGroups = randomIntBetween(2, 5);
+        for (int i = 0; i < numGroups; i++) {
+            remoteIndicesBuilder.addGroup(
+                Set.copyOf(randomNonEmptySubsetOf(List.of(concreteClusterAlias, "*"))),
+                IndexPrivilege.get(Set.copyOf(randomSubsetOf(randomIntBetween(1, 4), IndexPrivilege.names()))),
+                new FieldPermissions(
+                    new FieldPermissionsDefinition(
+                        Set.of(
+                            randomBoolean()
+                                ? randomFieldGrantExcludeGroup()
+                                : new FieldPermissionsDefinition.FieldGrantExcludeGroup(null, null)
+                        )
+                    )
+                ),
+                randomBoolean() ? Set.of(randomDlsQuery()) : null,
+                randomBoolean(),
+                generateRandomStringArray(3, 10, false, false)
+            );
+        }
+        final RemoteIndicesPermission permissions = remoteIndicesBuilder.build();
+        List<RemoteIndicesPermission.RemoteIndicesGroup> remoteIndicesGroups = permissions.remoteIndicesGroups();
+        final Role role1 = mockRoleWithRemoteIndices(permissions);
+        final RBACAuthorizationInfo authorizationInfo1 = mock(RBACAuthorizationInfo.class);
+        when(authorizationInfo1.getRole()).thenReturn(role1);
+        final PlainActionFuture<RoleDescriptorsIntersection> future1 = new PlainActionFuture<>();
+        engine.getRemoteAccessRoleDescriptorsIntersection(concreteClusterAlias, authorizationInfo1, future1);
+        final RoleDescriptorsIntersection actual1 = future1.get();
+        verify(role1, times(1)).remoteIndices();
+        verifyNoMoreInteractions(role1);
+
+        // Randomize the order of both remote indices groups and each of the indices permissions groups each group holds
+        final RemoteIndicesPermission shuffledPermissions = new RemoteIndicesPermission(
+            shuffledList(
+                remoteIndicesGroups.stream()
+                    .map(
+                        group -> new RemoteIndicesPermission.RemoteIndicesGroup(
+                            group.remoteClusterAliases(),
+                            shuffledList(group.indicesPermissionGroups())
+                        )
+                    )
+                    .toList()
+            )
+        );
+        final Role role2 = mockRoleWithRemoteIndices(shuffledPermissions);
+        final RBACAuthorizationInfo authorizationInfo2 = mock(RBACAuthorizationInfo.class);
+        when(authorizationInfo2.getRole()).thenReturn(role2);
+        final PlainActionFuture<RoleDescriptorsIntersection> future2 = new PlainActionFuture<>();
+        engine.getRemoteAccessRoleDescriptorsIntersection(concreteClusterAlias, authorizationInfo2, future2);
+        final RoleDescriptorsIntersection actual2 = future2.get();
+
+        verify(role2, times(1)).remoteIndices();
+        verifyNoMoreInteractions(role2);
+        assertThat(actual1, equalTo(actual2));
+        assertThat(actual1.roleDescriptorsList().iterator().next().iterator().next().getIndicesPrivileges().length, equalTo(numGroups));
+    }
+
+    public void testGetRemoteAccessRoleDescriptorsIntersectionWithoutMatchingGroups() throws ExecutionException, InterruptedException {
+        assumeTrue("untrusted remote cluster feature flag must be enabled", TcpTransport.isUntrustedRemoteClusterEnabled());
+
+        final String concreteClusterAlias = randomAlphaOfLength(10);
+        final Role role = mockRoleWithRemoteIndices(
+            RemoteIndicesPermission.builder()
+                .addGroup(
+                    Set.of(concreteClusterAlias),
+                    IndexPrivilege.READ,
+                    new FieldPermissions(new FieldPermissionsDefinition(null, null)),
+                    null,
+                    randomBoolean(),
+                    generateRandomStringArray(3, 10, false, false)
+                )
+                .build()
+        );
+        final RBACAuthorizationInfo authorizationInfo = mock(RBACAuthorizationInfo.class);
+        when(authorizationInfo.getRole()).thenReturn(role);
+
+        final PlainActionFuture<RoleDescriptorsIntersection> future = new PlainActionFuture<>();
+        engine.getRemoteAccessRoleDescriptorsIntersection(
+            randomValueOtherThan(concreteClusterAlias, () -> randomAlphaOfLength(10)),
+            authorizationInfo,
+            future
+        );
+        final RoleDescriptorsIntersection actual = future.get();
+        assertThat(actual, equalTo(RoleDescriptorsIntersection.EMPTY));
+        verify(role, times(1)).remoteIndices();
+        verifyNoMoreInteractions(role);
+    }
+
+    public void testGetRemoteAccessRoleDescriptorsIntersectionWithoutRemoteIndicesPermissions() throws ExecutionException,
+        InterruptedException {
+        assumeTrue("untrusted remote cluster feature flag must be enabled", TcpTransport.isUntrustedRemoteClusterEnabled());
+
+        final String concreteClusterAlias = randomAlphaOfLength(10);
+        final Role role = mockRoleWithRemoteIndices(RemoteIndicesPermission.NONE);
+        final RBACAuthorizationInfo authorizationInfo = mock(RBACAuthorizationInfo.class);
+        when(authorizationInfo.getRole()).thenReturn(role);
+
+        final PlainActionFuture<RoleDescriptorsIntersection> future = new PlainActionFuture<>();
+        engine.getRemoteAccessRoleDescriptorsIntersection(
+            randomValueOtherThan(concreteClusterAlias, () -> randomAlphaOfLength(10)),
+            authorizationInfo,
+            future
+        );
+        final RoleDescriptorsIntersection actual = future.get();
+        assertThat(actual, equalTo(RoleDescriptorsIntersection.EMPTY));
+    }
+
     private GetUserPrivilegesResponse.Indices findIndexPrivilege(Set<GetUserPrivilegesResponse.Indices> indices, String name) {
         return indices.stream().filter(i -> i.getIndices().contains(name)).findFirst().get();
     }
@@ -1667,4 +1870,26 @@ public class RBACEngineTests extends ESTestCase {
     private static MapBuilder<String, Boolean> mapBuilder() {
         return MapBuilder.newMapBuilder();
     }
+
+    private BytesArray randomDlsQuery() {
+        return new BytesArray(
+            "{ \"term\": { \"" + randomAlphaOfLengthBetween(3, 24) + "\" : \"" + randomAlphaOfLengthBetween(3, 24) + "\" }"
+        );
+    }
+
+    private FieldPermissionsDefinition.FieldGrantExcludeGroup randomFieldGrantExcludeGroup() {
+        return new FieldPermissionsDefinition.FieldGrantExcludeGroup(generateRandomStringArray(3, 10, false, false), new String[] {});
+    }
+
+    private Role mockRoleWithRemoteIndices(final RemoteIndicesPermission remoteIndicesPermission) {
+        final Role role = mock(Role.class);
+        final String[] roleNames = generateRandomStringArray(3, 10, false, false);
+        when(role.names()).thenReturn(roleNames);
+        when(role.cluster()).thenReturn(ClusterPermission.NONE);
+        when(role.indices()).thenReturn(IndicesPermission.NONE);
+        when(role.application()).thenReturn(ApplicationPermission.NONE);
+        when(role.runAs()).thenReturn(RunAsPermission.NONE);
+        when(role.remoteIndices()).thenReturn(remoteIndicesPermission);
+        return role;
+    }
 }