Refactor remote cluster interceptor and tests

x-pack/plugin/security/qa/rcs-extension/src/main/java/org/elasticsearch/xpack/security/rcs/extension/TestRemoteClusterSecurityExtension.java

@@ -10,25 +10,20 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.DestructiveOperations;
 import org.elasticsearch.common.settings.Setting;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.ssl.SslConfiguration;
-import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportInterceptor;
 import org.elasticsearch.transport.TransportRequest;
-import org.elasticsearch.xpack.core.XPackSettings;
 import org.elasticsearch.xpack.core.security.SecurityContext;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
-import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.core.ssl.SslProfile;
 import org.elasticsearch.xpack.security.authc.RemoteClusterAuthenticationService;
 import org.elasticsearch.xpack.security.transport.RemoteClusterTransportInterceptor;
 import org.elasticsearch.xpack.security.transport.ServerTransportFilter;
 import org.elasticsearch.xpack.security.transport.extension.RemoteClusterSecurityExtension;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 public class TestRemoteClusterSecurityExtension implements RemoteClusterSecurityExtension {
 
@@ -58,40 +53,16 @@ public boolean isRemoteClusterConnection(Transport.Connection connection) {
                 return false;
             }
 
-            public boolean hasRemoteClusterAccessHeadersInContext(SecurityContext securityContext) {
-                return false;
-            }
-
             @Override
-            public Map<String, ServerTransportFilter> getProfileTransportFilters(
-                Map<String, SslProfile> profileConfigurations,
+            public Optional<ServerTransportFilter> getRemoteProfileTransportFilter(
+                SslProfile sslProfile,
                 DestructiveOperations destructiveOperations
             ) {
-                Map<String, ServerTransportFilter> profileFilters = Maps.newMapWithExpectedSize(profileConfigurations.size() + 1);
-                Settings settings = components.settings();
-                final boolean transportSSLEnabled = XPackSettings.TRANSPORT_SSL_ENABLED.get(settings);
-
-                for (Map.Entry<String, SslProfile> entry : profileConfigurations.entrySet()) {
-                    final String profileName = entry.getKey();
-                    final SslProfile sslProfile = entry.getValue();
-                    final SslConfiguration profileConfiguration = sslProfile.configuration();
-                    profileFilters.put(
-                        profileName,
-                        new ServerTransportFilter(
-                            components.authenticationService(),
-                            components.authorizationService(),
-                            components.threadPool().getThreadContext(),
-                            transportSSLEnabled && SSLService.isSSLClientAuthEnabled(profileConfiguration),
-                            destructiveOperations,
-                            components.securityContext()
-                        )
-                    );
-                }
-                // We need to register here the default security
-                // server transport filter which ensures that all
-                // incoming transport requests are properly
-                // authenticated and authorized.
-                return Collections.unmodifiableMap(profileFilters);
+                return Optional.empty();
+            }
+
+            public boolean hasRemoteClusterAccessHeadersInContext(SecurityContext securityContext) {
+                return false;
             }
 
         };

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

@@ -1207,6 +1207,8 @@ Collection<Object> createComponents(
             new SecurityServerTransportInterceptor(
                 settings,
                 threadPool,
+                authcService.get(),
+                authzService,
                 getSslService(),
                 securityContext.get(),
                 destructiveOperations,

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/CrossClusterAccessTransportInterceptor.java

@@ -15,7 +15,6 @@
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.ssl.SslConfiguration;
-import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.license.XPackLicenseState;
@@ -49,13 +48,11 @@
 import org.elasticsearch.xpack.security.authc.CrossClusterAccessHeaders;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
 
-import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Function;
 
 import static org.elasticsearch.core.Strings.format;
-import static org.elasticsearch.transport.RemoteClusterPortSettings.REMOTE_CLUSTER_PROFILE;
 import static org.elasticsearch.transport.RemoteClusterPortSettings.REMOTE_CLUSTER_SERVER_ENABLED;
 import static org.elasticsearch.transport.RemoteClusterPortSettings.TRANSPORT_VERSION_ADVANCED_REMOTE_CLUSTER_SECURITY;
 
@@ -83,7 +80,6 @@ public class CrossClusterAccessTransportInterceptor implements RemoteClusterTran
     private final Function<Transport.Connection, Optional<RemoteClusterAliasWithCredentials>> remoteClusterCredentialsResolver;
     private final CrossClusterAccessAuthenticationService crossClusterAccessAuthcService;
     private final CrossClusterApiKeySignatureManager crossClusterApiKeySignatureManager;
-    private final AuthenticationService authcService;
     private final AuthorizationService authzService;
     private final XPackLicenseState licenseState;
     private final SecurityContext securityContext;
@@ -128,7 +124,6 @@ public CrossClusterAccessTransportInterceptor(
         this.remoteClusterCredentialsResolver = remoteClusterCredentialsResolver;
         this.crossClusterAccessAuthcService = crossClusterAccessAuthcService;
         this.crossClusterApiKeySignatureManager = crossClusterApiKeySignatureManager;
-        this.authcService = authcService;
         this.authzService = authzService;
         this.licenseState = licenseState;
         this.securityContext = securityContext;
@@ -328,51 +323,27 @@ public boolean isRemoteClusterConnection(Transport.Connection connection) {
     }
 
     @Override
-    public Map<String, ServerTransportFilter> getProfileTransportFilters(
-        Map<String, SslProfile> profileConfigurations,
+    public Optional<ServerTransportFilter> getRemoteProfileTransportFilter(
+        SslProfile sslProfile,
         DestructiveOperations destructiveOperations
     ) {
-        Map<String, ServerTransportFilter> profileFilters = Maps.newMapWithExpectedSize(profileConfigurations.size() + 1);
-
-        final boolean transportSSLEnabled = XPackSettings.TRANSPORT_SSL_ENABLED.get(settings);
-        final boolean remoteClusterPortEnabled = REMOTE_CLUSTER_SERVER_ENABLED.get(settings);
+        final SslConfiguration profileConfiguration = sslProfile.configuration();
+        final boolean remoteClusterServerEnabled = REMOTE_CLUSTER_SERVER_ENABLED.get(settings);
         final boolean remoteClusterServerSSLEnabled = XPackSettings.REMOTE_CLUSTER_SERVER_SSL_ENABLED.get(settings);
-
-        for (Map.Entry<String, SslProfile> entry : profileConfigurations.entrySet()) {
-            final String profileName = entry.getKey();
-            final SslProfile sslProfile = entry.getValue();
-            final SslConfiguration profileConfiguration = sslProfile.configuration();
-            assert profileConfiguration != null : "Ssl Profile [" + sslProfile + "] for [" + profileName + "] has a null configuration";
-            final boolean useRemoteClusterProfile = remoteClusterPortEnabled && profileName.equals(REMOTE_CLUSTER_PROFILE);
-            if (useRemoteClusterProfile) {
-                profileFilters.put(
-                    profileName,
-                    new CrossClusterAccessServerTransportFilter(
-                        crossClusterAccessAuthcService,
-                        authzService,
-                        threadPool.getThreadContext(),
-                        remoteClusterServerSSLEnabled && SSLService.isSSLClientAuthEnabled(profileConfiguration),
-                        destructiveOperations,
-                        securityContext,
-                        licenseState
-                    )
-                );
-            } else {
-                profileFilters.put(
-                    profileName,
-                    new ServerTransportFilter(
-                        authcService,
-                        authzService,
-                        threadPool.getThreadContext(),
-                        transportSSLEnabled && SSLService.isSSLClientAuthEnabled(profileConfiguration),
-                        destructiveOperations,
-                        securityContext
-                    )
-                );
-            }
+        if (remoteClusterServerEnabled) {
+            return Optional.of(
+                new CrossClusterAccessServerTransportFilter(
+                    crossClusterAccessAuthcService,
+                    authzService,
+                    threadPool.getThreadContext(),
+                    remoteClusterServerSSLEnabled && SSLService.isSSLClientAuthEnabled(profileConfiguration),
+                    destructiveOperations,
+                    securityContext,
+                    licenseState
+                )
+            );
         }
-
-        return Collections.unmodifiableMap(profileFilters);
+        return Optional.empty();
     }
 
     @Override

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/RemoteClusterTransportInterceptor.java

@@ -8,12 +8,13 @@
 package org.elasticsearch.xpack.security.transport;
 
 import org.elasticsearch.action.support.DestructiveOperations;
+import org.elasticsearch.transport.RemoteClusterPortSettings;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportInterceptor;
 import org.elasticsearch.xpack.core.security.SecurityContext;
 import org.elasticsearch.xpack.core.ssl.SslProfile;
 
-import java.util.Map;
+import java.util.Optional;
 
 /**
  * Allows to provide remote cluster interception that's capable of intercepting remote connections
@@ -32,16 +33,20 @@ public interface RemoteClusterTransportInterceptor {
     boolean isRemoteClusterConnection(Transport.Connection connection);
 
     /**
-     * Allows interceptors to provide a custom {@link ServerTransportFilter} implementations per transport profile.
-     * The transport filter is called on the receiver side to filter incoming requests
-     * and execute authentication and authorization for all requests.
+     * Allows interceptors to provide a custom {@link ServerTransportFilter} implementation
+     * for intercepting requests for {@link RemoteClusterPortSettings#REMOTE_CLUSTER_PROFILE}
+     * transport profile.
+     * <p>
+     * The transport filter is called on the receiver side to filter incoming remote cluster requests
+     * and to execute authentication and authorization for all incoming requests.
+     * <p>
+     * This method is only called when setting {@link RemoteClusterPortSettings#REMOTE_CLUSTER_SERVER_ENABLED}
+     * is set to {@code true}.
      *
-     * @return map of {@link ServerTransportFilter}s per transport profile name
+     * @return a custom {@link ServerTransportFilter}s for the given transport profile,
+     *         or an empty optional to fall back to the default transport filter
      */
-    Map<String, ServerTransportFilter> getProfileTransportFilters(
-        Map<String, SslProfile> profileConfigurations,
-        DestructiveOperations destructiveOperations
-    );
+    Optional<ServerTransportFilter> getRemoteProfileTransportFilter(SslProfile sslProfile, DestructiveOperations destructiveOperations);
 
     /**
      * Returns {@code true} if any of the remote cluster access headers are in the security context.

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java

@@ -12,6 +12,8 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.DestructiveOperations;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.ssl.SslConfiguration;
+import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.RunOnce;
@@ -29,44 +31,92 @@
 import org.elasticsearch.transport.TransportResponseHandler;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.transport.TransportService.ContextRestoreResponseHandler;
+import org.elasticsearch.xpack.core.XPackSettings;
 import org.elasticsearch.xpack.core.security.SecurityContext;
 import org.elasticsearch.xpack.core.security.transport.ProfileConfigurations;
 import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.core.ssl.SslProfile;
+import org.elasticsearch.xpack.security.authc.AuthenticationService;
+import org.elasticsearch.xpack.security.authz.AuthorizationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationUtils;
 import org.elasticsearch.xpack.security.authz.PreAuthorizationUtils;
 
+import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.Executor;
 
 import static org.elasticsearch.core.Strings.format;
+import static org.elasticsearch.transport.RemoteClusterPortSettings.REMOTE_CLUSTER_PROFILE;
 
 public class SecurityServerTransportInterceptor implements TransportInterceptor {
 
     private static final Logger logger = LogManager.getLogger(SecurityServerTransportInterceptor.class);
 
+    private final AuthenticationService authcService;
+    private final AuthorizationService authzService;
     private final RemoteClusterTransportInterceptor remoteClusterTransportInterceptor;
     private final Map<String, ServerTransportFilter> profileFilters;
     private final ThreadPool threadPool;
     private final SecurityContext securityContext;
+    private final Settings settings;
 
     public SecurityServerTransportInterceptor(
         Settings settings,
         ThreadPool threadPool,
+        AuthenticationService authcService,
+        AuthorizationService authzService,
         SSLService sslService,
         SecurityContext securityContext,
         DestructiveOperations destructiveOperations,
         RemoteClusterTransportInterceptor remoteClusterTransportInterceptor
-
     ) {
         this.remoteClusterTransportInterceptor = remoteClusterTransportInterceptor;
         this.securityContext = securityContext;
         this.threadPool = threadPool;
+        this.settings = settings;
+        this.authcService = authcService;
+        this.authzService = authzService;
         final Map<String, SslProfile> profileConfigurations = ProfileConfigurations.get(settings, sslService, false);
-        this.profileFilters = this.remoteClusterTransportInterceptor.getProfileTransportFilters(
-            profileConfigurations,
-            destructiveOperations
-        );
+        this.profileFilters = initializeProfileFilters(profileConfigurations, destructiveOperations);
+    }
+
+    private Map<String, ServerTransportFilter> initializeProfileFilters(
+        final Map<String, SslProfile> profileConfigurations,
+        final DestructiveOperations destructiveOperations
+    ) {
+        final Map<String, ServerTransportFilter> profileFilters = Maps.newMapWithExpectedSize(profileConfigurations.size() + 1);
+        final boolean transportSSLEnabled = XPackSettings.TRANSPORT_SSL_ENABLED.get(settings);
+
+        for (Map.Entry<String, SslProfile> entry : profileConfigurations.entrySet()) {
+            final String profileName = entry.getKey();
+            final SslProfile sslProfile = entry.getValue();
+            if (profileName.equals(REMOTE_CLUSTER_PROFILE)) {
+                var remoteProfileTransportFilter = this.remoteClusterTransportInterceptor.getRemoteProfileTransportFilter(
+                    sslProfile,
+                    destructiveOperations
+                );
+                if (remoteProfileTransportFilter.isPresent()) {
+                    profileFilters.put(profileName, remoteProfileTransportFilter.get());
+                    continue;
+                }
+            }
+
+            final SslConfiguration profileConfiguration = sslProfile.configuration();
+            assert profileConfiguration != null : "SSL Profile [" + sslProfile + "] for [" + profileName + "] has a null configuration";
+            profileFilters.put(
+                profileName,
+                new ServerTransportFilter(
+                    authcService,
+                    authzService,
+                    threadPool.getThreadContext(),
+                    transportSSLEnabled && SSLService.isSSLClientAuthEnabled(profileConfiguration),
+                    destructiveOperations,
+                    securityContext
+                )
+            );
+        }
+
+        return Collections.unmodifiableMap(profileFilters);
     }
 
     @Override

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/ServerTransportFilter.java

@@ -62,11 +62,12 @@ public ServerTransportFilter(
     }
 
     /**
-     * Called just after the given request was received by the transport. Any exception
-     * thrown by this method will stop the request from being handled and the error will
-     * be sent back to the sender.
+     * Called just after the given request was received by the transport service.
+     * <p>
+     * Any exception thrown by this method will stop the request from being handled
+     * and the error will be sent back to the sender.
      */
-    void inbound(String action, TransportRequest request, TransportChannel transportChannel, ActionListener<Void> listener) {
+    public void inbound(String action, TransportRequest request, TransportChannel transportChannel, ActionListener<Void> listener) {
         if (TransportCloseIndexAction.NAME.equals(action)
             || OpenIndexAction.NAME.equals(action)
             || TransportDeleteIndexAction.TYPE.name().equals(action)) {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/AbstractServerTransportFilterTests.java

@@ -0,0 +1,34 @@
+/*
+ * 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.transport;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+import org.mockito.stubbing.Answer;
+
+import static org.hamcrest.Matchers.arrayWithSize;
+
+public abstract class AbstractServerTransportFilterTests extends ESTestCase {
+
+    protected static Answer<Class<Void>> getAnswer(Authentication authentication) {
+        return getAnswer(authentication, false);
+    }
+
+    protected static Answer<Class<Void>> getAnswer(Authentication authentication, boolean crossClusterAccess) {
+        return i -> {
+            final Object[] args = i.getArguments();
+            assertThat(args, arrayWithSize(crossClusterAccess ? 3 : 4));
+            @SuppressWarnings("unchecked")
+            ActionListener<Authentication> callback = (ActionListener<Authentication>) args[args.length - 1];
+            callback.onResponse(authentication);
+            return Void.TYPE;
+        };
+    }
+
+}