[POC] RCS 2.0 - fulfilling cluster handles cross cluster requests

build-tools-internal/src/main/groovy/elasticsearch.run-ccs.gradle

@@ -17,6 +17,23 @@ def fulfillingCluster = testClusters.register('fulfilling-cluster') {
   setting 'xpack.watcher.enabled', 'false'
   setting 'xpack.ml.enabled', 'false'
   setting 'xpack.license.self_generated.type', 'trial'
+  setting 'remote_cluster.enabled', 'true'
+  // TODO
+  setting 'remote_cluster.port', '9901'
+
+  user username: 'elastic-admin', password: 'elastic-password', role: '_es_test_root'
+}
+
+def fulfillingCluster2 = testClusters.register('fulfilling-cluster2') {
+  testDistribution = providers.systemProperty('run.distribution').orElse('default').get()
+
+  setting 'xpack.security.enabled', 'true'
+  setting 'xpack.watcher.enabled', 'false'
+  setting 'xpack.ml.enabled', 'false'
+  setting 'xpack.license.self_generated.type', 'trial'
+  setting 'remote_cluster.enabled', 'true'
+  // TODO
+  setting 'remote_cluster.port', '9902'
 
   user username: 'elastic-admin', password: 'elastic-password', role: '_es_test_root'
 }
@@ -29,27 +46,38 @@ def queryingCluster = testClusters.register('querying-cluster') {
   setting 'xpack.ml.enabled', 'false'
   setting 'xpack.license.self_generated.type', 'trial'
   setting 'cluster.remote.connections_per_cluster', "1"
+
   user username: 'elastic-admin', password: 'elastic-password', role: '_es_test_root'
+  user username: 'ccs-admin', password: 'elastic-password', role: '_ccs_admin'
 }
 
-
 tasks.register("run-ccs", RunTask) {
+  useCluster fulfillingCluster2
   useCluster fulfillingCluster
   useCluster queryingCluster
   doFirst {
     queryingCluster.get().getNodes().each { node ->
       if (proxyMode) {
         node.setting('cluster.remote.my_remote_cluster.mode', 'proxy')
-        node.setting('cluster.remote.my_remote_cluster.proxy_address', "\"${fulfillingCluster.get().getAllTransportPortURI().get(0)}\"")
+        node.setting('cluster.remote.my_remote_cluster.proxy_address', "\"${fulfillingCluster.get().getAllRemoteAccessPortURI().get(0)}\"")
+        node.setting('cluster.remote.my_remote_cluster2.mode', 'proxy')
+        node.setting('cluster.remote.my_remote_cluster2.proxy_address', "\"${fulfillingCluster2.get().getAllRemoteAccessPortURI().get(0)}\"")
       } else {
-        node.setting('cluster.remote.my_remote_cluster.seeds', fulfillingCluster.get().getAllTransportPortURI().collect { "\"$it\"" }.toString())
+        node.setting('cluster.remote.my_remote_cluster.seeds', fulfillingCluster.get().getAllRemoteAccessPortURI().collect { "\"$it\"" }.toString())
+        node.setting('cluster.remote.my_remote_cluster2.seeds', fulfillingCluster2.get().getAllRemoteAccessPortURI().collect { "\"$it\"" }.toString())
       }
     }
     queryingCluster.get().restart() //updates the on-disk config
 
     println "** Querying cluster HTTP endpoints are: ${-> queryingCluster.get().allHttpSocketURI.join(",")}"
     println "** Querying cluster transport endpoints are: ${-> queryingCluster.get().getAllTransportPortURI().join(",")}"
+
     println "** Fulfilling cluster HTTP endpoints are: ${-> fulfillingCluster.get().allHttpSocketURI.join(",")}"
     println "** Fulfilling cluster transport endpoints are: ${-> fulfillingCluster.get().getAllTransportPortURI().join(",")}"
+    println "** Fulfilling cluster remote access endpoints are: ${-> fulfillingCluster.get().getAllRemoteAccessPortURI().join(",")}"
+
+    println "** Fulfilling cluster 2 HTTP endpoints are: ${-> fulfillingCluster2.get().allHttpSocketURI.join(",")}"
+    println "** Fulfilling cluster 2 transport endpoints are: ${-> fulfillingCluster2.get().getAllTransportPortURI().join(",")}"
+    println "** Fulfilling cluster 2 remote access endpoints are: ${-> fulfillingCluster2.get().getAllRemoteAccessPortURI().join(",")}"
   }
 }

build-tools/src/main/resources/roles.yml

@@ -11,3 +11,11 @@ _es_test_root:
     - application: "*"
       privileges: [ "*" ]
       resources: [ "*" ]
+
+#_ccs_admin:
+#  cluster: [ "ALL" ]
+#  remote_indices:
+#    - names: [ "my-index-*" ]
+#      privileges: [ "READ", "READ_CROSS_CLUSTER" ]
+#      clusters: [ "*remote*" ]
+

server/src/main/java/org/elasticsearch/transport/RemoteConnectionManager.java

@@ -83,7 +83,14 @@ public void removeListener(TransportConnectionListener listener) {
 
     @Override
     public void openConnection(DiscoveryNode node, ConnectionProfile profile, ActionListener<Transport.Connection> listener) {
-        delegate.openConnection(node, profile, listener);
+        delegate.openConnection(
+            node,
+            profile,
+            ActionListener.wrap(
+                connection -> listener.onResponse(new InternalRemoteConnection(connection, clusterAlias)),
+                listener::onFailure
+            )
+        );
     }
 
     @Override

server/src/test/java/org/elasticsearch/transport/RemoteConnectionManagerTests.java

@@ -19,6 +19,7 @@
 import java.net.InetAddress;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasItems;
@@ -89,7 +90,7 @@ public void testGetConnection() {
         assertEquals(1, versions.size());
     }
 
-    public void testResolveRemoteClusterAlias() {
+    public void testResolveRemoteClusterAlias() throws ExecutionException, InterruptedException {
         DiscoveryNode remoteNode1 = new DiscoveryNode("remote-node-1", address, Version.CURRENT);
         PlainActionFuture<Void> future = PlainActionFuture.newFuture();
         remoteConnectionManager.connectToRemoteClusterNode(remoteNode1, validator, future);
@@ -105,6 +106,10 @@ public void testResolveRemoteClusterAlias() {
         Transport.Connection proxyConnection = remoteConnectionManager.getConnection(remoteNode2);
         assertThat(proxyConnection, instanceOf(RemoteConnectionManager.ProxyConnection.class));
         assertThat(RemoteConnectionManager.resolveRemoteClusterAlias(proxyConnection).get(), equalTo("remote-cluster"));
+
+        PlainActionFuture<Transport.Connection> future2 = PlainActionFuture.newFuture();
+        remoteConnectionManager.openConnection(remoteNode1, null, future2);
+        assertThat(RemoteConnectionManager.resolveRemoteClusterAlias(future2.get()).get(), equalTo("remote-cluster"));
     }
 
     private static class TestRemoteConnection extends CloseableConnection {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/Authentication.java

@@ -53,6 +53,7 @@
 import static org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef.newApiKeyRealmRef;
 import static org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef.newInternalAttachRealmRef;
 import static org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef.newInternalFallbackRealmRef;
+import static org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef.newRemoteAccessRealmRef;
 import static org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef.newServiceAccountRealmRef;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ANONYMOUS_REALM_NAME;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ANONYMOUS_REALM_TYPE;
@@ -62,6 +63,8 @@
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ATTACH_REALM_TYPE;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.FALLBACK_REALM_NAME;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.FALLBACK_REALM_TYPE;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.REMOTE_ACCESS_REALM_NAME;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.REMOTE_ACCESS_REALM_TYPE;
 import static org.elasticsearch.xpack.core.security.authc.RealmDomain.REALM_DOMAIN_PARSER;
 
 /**
@@ -509,6 +512,7 @@ public void writeTo(StreamOutput out) throws IOException {
      */
     public boolean canAccessResourcesOf(Authentication resourceCreatorAuthentication) {
         // if we introduce new authentication types in the future, it is likely that we'll need to revisit this method
+
         assert EnumSet.of(
             Authentication.AuthenticationType.REALM,
             Authentication.AuthenticationType.API_KEY,
@@ -792,6 +796,10 @@ static RealmRef newApiKeyRealmRef(String nodeName) {
             // no domain for API Key tokens
             return new RealmRef(API_KEY_REALM_NAME, API_KEY_REALM_TYPE, nodeName, null);
         }
+
+        static RealmRef newRemoteAccessRealmRef(String apiKeyId, String nodeName) {
+            return new RealmRef(REMOTE_ACCESS_REALM_NAME + "_" + apiKeyId, REMOTE_ACCESS_REALM_TYPE, nodeName, null);
+        }
     }
 
     public static boolean isFileOrNativeRealm(String realmType) {
@@ -882,6 +890,33 @@ public static Authentication newApiKeyAuthentication(AuthenticationResult<User>
         return authentication;
     }
 
+    public static Authentication newRemoteAccessAuthentication(
+        AuthenticationResult<User> authResult,
+        Authentication receivedAuthentication,
+        String nodeName
+    ) {
+        assert authResult.isAuthenticated() : "API Key authn for remote access result must be successful";
+        final User receivedUser = receivedAuthentication.getEffectiveSubject().getUser();
+        assert receivedUser.enabled();
+        final User user = new User(
+            // Consider including API key ID as well
+            receivedUser.principal(),
+            new String[0],
+            receivedUser.fullName(),
+            receivedUser.email(),
+            receivedUser.metadata(),
+            true
+        );
+        final User apiKeyUser = authResult.getValue();
+        final Authentication.RealmRef authenticatedBy = newRemoteAccessRealmRef(apiKeyUser.principal(), nodeName);
+        final Authentication authentication = new Authentication(
+            new Subject(user, authenticatedBy, Version.CURRENT, authResult.getMetadata()),
+            AuthenticationType.API_KEY
+        );
+        assert false == authentication.isAssignedToDomain();
+        return authentication;
+    }
+
     private static RealmRef maybeRewriteRealmRef(Version streamVersion, RealmRef realmRef) {
         if (realmRef != null && realmRef.getDomain() != null && streamVersion.before(VERSION_REALM_DOMAINS)) {
             logger.info("Rewriting realm [" + realmRef + "] without domain");

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/AuthenticationField.java

@@ -33,5 +33,9 @@ public final class AuthenticationField {
     public static final String ATTACH_REALM_NAME = "__attach";
     public static final String ATTACH_REALM_TYPE = "__attach";
 
+    public static final String REMOTE_ACCESS_REALM_NAME = "_es_remote_access";
+    public static final String REMOTE_ACCESS_REALM_TYPE = "_es_remote_access";
+    public static final String REMOTE_ACCESS_ROLE_DESCRIPTORS_KEY = "_security_remote_access_role_descriptors";
+
     private AuthenticationField() {}
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/RemoteAccessAuthentication.java

@@ -91,7 +91,7 @@ public String toString() {
             + '}';
     }
 
-    private static List<RoleDescriptorsBytes> toRoleDescriptorsBytesList(final RoleDescriptorsIntersection roleDescriptorsIntersection)
+    public static List<RoleDescriptorsBytes> toRoleDescriptorsBytesList(final RoleDescriptorsIntersection roleDescriptorsIntersection)
         throws IOException {
         // If we ever lift this restriction, we need to ensure that the serialization of each set of role descriptors to raw bytes is
         // deterministic. We can do so by sorting the role descriptors before serializing.

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/Subject.java

@@ -7,6 +7,8 @@
 
 package org.elasticsearch.xpack.core.security.authc;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchSecurityException;
 import org.elasticsearch.Version;
 import org.elasticsearch.common.bytes.BytesArray;
@@ -19,12 +21,15 @@
 import org.elasticsearch.xpack.core.security.user.AnonymousUser;
 import org.elasticsearch.xpack.core.security.user.User;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
 import static org.elasticsearch.xpack.core.security.authc.Authentication.VERSION_API_KEY_ROLES_AS_BYTES;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.REMOTE_ACCESS_ROLE_DESCRIPTORS_KEY;
 import static org.elasticsearch.xpack.core.security.authc.Subject.Type.API_KEY;
 
 /**
@@ -35,10 +40,13 @@
  */
 public class Subject {
 
+    private static final Logger logger = LogManager.getLogger(Subject.class);
+
     public enum Type {
         USER,
         API_KEY,
         SERVICE_ACCOUNT,
+        REMOTE_ACCESS,
     }
 
     private final Version version;
@@ -65,6 +73,9 @@ public Subject(User user, Authentication.RealmRef realm, Version version, Map<St
         } else if (ServiceAccountSettings.REALM_TYPE.equals(realm.getType())) {
             assert ServiceAccountSettings.REALM_NAME.equals(realm.getName()) : "service account realm name mismatch";
             this.type = Type.SERVICE_ACCOUNT;
+        } else if (AuthenticationField.REMOTE_ACCESS_REALM_TYPE.equals(realm.getType())) {
+            assert realm.getName().startsWith(AuthenticationField.REMOTE_ACCESS_REALM_NAME) : "remote access realm name prefix mismatch";
+            this.type = Type.REMOTE_ACCESS;
         } else {
             this.type = Type.USER;
         }
@@ -99,6 +110,8 @@ public RoleReferenceIntersection getRoleReferenceIntersection(@Nullable Anonymou
                 return buildRoleReferencesForApiKey();
             case SERVICE_ACCOUNT:
                 return new RoleReferenceIntersection(new RoleReference.ServiceAccountRoleReference(user.principal()));
+            case REMOTE_ACCESS:
+                return buildRoleReferencesForRemoteAccess();
             default:
                 assert false : "unknown subject type: [" + type + "]";
                 throw new IllegalStateException("unknown subject type: [" + type + "]");
@@ -220,6 +233,23 @@ private RoleReferenceIntersection buildRoleReferencesForApiKey() {
         );
     }
 
+    private RoleReferenceIntersection buildRoleReferencesForRemoteAccess() {
+        final List<RoleReference> roleReferences = new ArrayList<>(buildRoleReferencesForApiKey().getRoleReferences());
+        @SuppressWarnings("unchecked")
+        final List<RemoteAccessAuthentication.RoleDescriptorsBytes> remoteAccessRoleDescriptorsBytes = (List<
+            RemoteAccessAuthentication.RoleDescriptorsBytes>) metadata.get(REMOTE_ACCESS_ROLE_DESCRIPTORS_KEY);
+        if (remoteAccessRoleDescriptorsBytes.isEmpty()) {
+            // TODO hack. fail earlier, and assert here instead
+            return new RoleReferenceIntersection(List.of(new RoleReference.NamedRoleReference(new String[] {})));
+        }
+        // TODO handle this once we support API keys as querying subjects
+        assert remoteAccessRoleDescriptorsBytes.size() == 1;
+        for (RemoteAccessAuthentication.RoleDescriptorsBytes roleDescriptorsBytes : remoteAccessRoleDescriptorsBytes) {
+            roleReferences.add(new RoleReference.RemoteAccessRoleReference(roleDescriptorsBytes));
+        }
+        return new RoleReferenceIntersection(List.copyOf(roleReferences));
+    }
+
     private static boolean isEmptyRoleDescriptorsBytes(BytesReference roleDescriptorsBytes) {
         return roleDescriptorsBytes == null || (roleDescriptorsBytes.length() == 2 && "{}".equals(roleDescriptorsBytes.utf8ToString()));
     }

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

@@ -124,9 +124,7 @@ public IndicesAccessControl authorize(
      */
     @Override
     public IsResourceAuthorizedPredicate allowedIndicesMatcher(String action) {
-        IsResourceAuthorizedPredicate predicate = baseRole.indices().allowedIndicesMatcher(action);
-        predicate = predicate.and(limitedByRole.indices().allowedIndicesMatcher(action));
-        return predicate;
+        return baseRole.allowedIndicesMatcher(action).and(limitedByRole.allowedIndicesMatcher(action));
     }
 
     @Override

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

@@ -10,6 +10,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.hash.MessageDigests;
+import org.elasticsearch.xpack.core.security.authc.RemoteAccessAuthentication;
 
 import java.util.HashSet;
 import java.util.List;
@@ -116,6 +117,38 @@ public ApiKeyRoleType getRoleType() {
         }
     }
 
+    final class RemoteAccessRoleReference implements RoleReference {
+
+        private final RemoteAccessAuthentication.RoleDescriptorsBytes roleDescriptorsBytes;
+        private RoleKey id = null;
+
+        public RemoteAccessRoleReference(RemoteAccessAuthentication.RoleDescriptorsBytes roleDescriptorsBytes) {
+            this.roleDescriptorsBytes = roleDescriptorsBytes;
+        }
+
+        @Override
+        public RoleKey id() {
+            // Hashing can be expensive. memorize the result in case the method is called multiple times.
+            if (id == null) {
+                final String roleDescriptorsHash = MessageDigests.toHexString(
+                    MessageDigests.digest(roleDescriptorsBytes, MessageDigests.sha256())
+                );
+                id = new RoleKey(Set.of("remote_access:" + roleDescriptorsHash), "remote_access");
+            }
+            return id;
+        }
+
+        @Override
+        public void resolve(RoleReferenceResolver resolver, ActionListener<RolesRetrievalResult> listener) {
+            resolver.resolveRemoteAccessRoleReference(this, listener);
+        }
+
+        public RemoteAccessAuthentication.RoleDescriptorsBytes getRoleDescriptorsBytes() {
+            return roleDescriptorsBytes;
+        }
+
+    }
+
     /**
      * Same as {@link ApiKeyRoleReference} but for BWC purpose (prior to v7.9.0)
      */

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

@@ -19,6 +19,11 @@ public interface RoleReferenceResolver {
 
     void resolveApiKeyRoleReference(RoleReference.ApiKeyRoleReference apiKeyRoleReference, ActionListener<RolesRetrievalResult> listener);
 
+    void resolveRemoteAccessRoleReference(
+        RoleReference.RemoteAccessRoleReference remoteAccessRoleReference,
+        ActionListener<RolesRetrievalResult> listener
+    );
+
     void resolveBwcApiKeyRoleReference(
         RoleReference.BwcApiKeyRoleReference bwcApiKeyRoleReference,
         ActionListener<RolesRetrievalResult> listener