Introduce remote cluster security interceptor and authenticator (elastic#134245)

This refactoring is meant to be purely structural, with no functional changes. 
It introduces two interfaces:
- `RemoteClusterAuthenticationService` - extracted based on the existing `CrossClusterAccessAuthenticationService`
- `RemoteClusterTransportInterceptor` - which aims to abstract and move all "cross-cluster access" logic from `SecurityServerTransportInterceptor` 

This is prerequisite for making remote cluster security logic pluggable (which I plan to add in a followup PR).

Relates to ES-12801

Author: slobodanadamovic

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/core/security/transport/netty4/SecurityNetty4Transport.java

@@ -47,7 +47,7 @@
 import org.elasticsearch.xpack.core.security.transport.SecurityTransportExceptionHandler;
 import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.core.ssl.SslProfile;
-import org.elasticsearch.xpack.security.authc.CrossClusterAccessAuthenticationService;
+import org.elasticsearch.xpack.security.authc.RemoteClusterAuthenticationService;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
@@ -79,7 +79,7 @@ public class SecurityNetty4Transport extends Netty4Transport {
     private final boolean remoteClusterServerSslEnabled;
     private final SslProfile remoteClusterClientSslProfile;
     private final RemoteClusterClientBootstrapOptions remoteClusterClientBootstrapOptions;
-    private final CrossClusterAccessAuthenticationService crossClusterAccessAuthenticationService;
+    private final RemoteClusterAuthenticationService remoteClusterAuthenticationService;
 
     public SecurityNetty4Transport(
         final Settings settings,
@@ -91,7 +91,7 @@ public SecurityNetty4Transport(
         final CircuitBreakerService circuitBreakerService,
         final SSLService sslService,
         final SharedGroupFactory sharedGroupFactory,
-        final CrossClusterAccessAuthenticationService crossClusterAccessAuthenticationService
+        final RemoteClusterAuthenticationService remoteClusterAuthenticationService
     ) {
         super(
             settings,
@@ -103,7 +103,7 @@ public SecurityNetty4Transport(
             circuitBreakerService,
             sharedGroupFactory
         );
-        this.crossClusterAccessAuthenticationService = crossClusterAccessAuthenticationService;
+        this.remoteClusterAuthenticationService = remoteClusterAuthenticationService;
         this.exceptionHandler = new SecurityTransportExceptionHandler(logger, lifecycle, (c, e) -> super.onException(c, e));
         this.sslService = sslService;
         this.transportSslEnabled = XPackSettings.TRANSPORT_SSL_ENABLED.get(settings);
@@ -180,7 +180,7 @@ protected void headerReceived(Header header) {
                         channel.config().setAutoRead(false);
                         // this prevents thread-context changes to propagate beyond the validation, as netty worker threads are reused
                         try (ThreadContext.StoredContext ignore = threadPool.getThreadContext().newStoredContext()) {
-                            crossClusterAccessAuthenticationService.tryAuthenticate(
+                            remoteClusterAuthenticationService.authenticateHeaders(
                                 header.getRequestHeaders(),
                                 ActionListener.runAfter(ActionListener.wrap(aVoid -> {
                                     // authn is successful -> NOOP (the complete request will be subsequently authn & authz & audited)

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

@@ -417,6 +417,8 @@
 import org.elasticsearch.xpack.security.support.ReloadableSecurityComponent;
 import org.elasticsearch.xpack.security.support.SecurityMigrations;
 import org.elasticsearch.xpack.security.support.SecuritySystemIndices;
+import org.elasticsearch.xpack.security.transport.CrossClusterAccessTransportInterceptor;
+import org.elasticsearch.xpack.security.transport.RemoteClusterTransportInterceptor;
 import org.elasticsearch.xpack.security.transport.SecurityHttpSettings;
 import org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor;
 import org.elasticsearch.xpack.security.transport.filter.IPFilter;
@@ -1164,17 +1166,24 @@ Collection<Object> createComponents(
         DestructiveOperations destructiveOperations = new DestructiveOperations(settings, clusterService.getClusterSettings());
         crossClusterAccessAuthcService.set(new CrossClusterAccessAuthenticationService(clusterService, apiKeyService, authcService.get()));
         components.add(crossClusterAccessAuthcService.get());
+
+        RemoteClusterTransportInterceptor remoteClusterTransportInterceptor = new CrossClusterAccessTransportInterceptor(
+            settings,
+            threadPool,
+            authcService.get(),
+            authzService,
+            securityContext.get(),
+            crossClusterAccessAuthcService.get(),
+            getLicenseState()
+        );
         securityInterceptor.set(
             new SecurityServerTransportInterceptor(
                 settings,
                 threadPool,
-                authcService.get(),
-                authzService,
                 getSslService(),
                 securityContext.get(),
                 destructiveOperations,
-                crossClusterAccessAuthcService.get(),
-                getLicenseState()
+                remoteClusterTransportInterceptor
             )
         );
 

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

@@ -34,7 +34,7 @@
 import static org.elasticsearch.xpack.core.security.authc.CrossClusterAccessSubjectInfo.CROSS_CLUSTER_ACCESS_SUBJECT_INFO_HEADER_KEY;
 import static org.elasticsearch.xpack.security.authc.CrossClusterAccessHeaders.CROSS_CLUSTER_ACCESS_CREDENTIALS_HEADER_KEY;
 
-public class CrossClusterAccessAuthenticationService {
+public class CrossClusterAccessAuthenticationService implements RemoteClusterAuthenticationService {
 
     private static final Logger logger = LogManager.getLogger(CrossClusterAccessAuthenticationService.class);
 
@@ -52,6 +52,7 @@ public CrossClusterAccessAuthenticationService(
         this.authenticationService = authenticationService;
     }
 
+    @Override
     public void authenticate(final String action, final TransportRequest request, final ActionListener<Authentication> listener) {
         final ThreadContext threadContext = clusterService.threadPool().getThreadContext();
         final CrossClusterAccessHeaders crossClusterAccessHeaders;
@@ -117,7 +118,8 @@ public void authenticate(final String action, final TransportRequest request, fi
         }
     }
 
-    public void tryAuthenticate(Map<String, String> headers, ActionListener<Void> listener) {
+    @Override
+    public void authenticateHeaders(Map<String, String> headers, ActionListener<Void> listener) {
         final ApiKeyService.ApiKeyCredentials credentials;
         try {
             credentials = extractApiKeyCredentialsFromHeaders(headers);
@@ -128,7 +130,8 @@ public void tryAuthenticate(Map<String, String> headers, ActionListener<Void> li
         tryAuthenticate(credentials, listener);
     }
 
-    public void tryAuthenticate(ApiKeyService.ApiKeyCredentials credentials, ActionListener<Void> listener) {
+    // package-private for testing
+    void tryAuthenticate(ApiKeyService.ApiKeyCredentials credentials, ActionListener<Void> listener) {
         Objects.requireNonNull(credentials);
         apiKeyService.tryAuthenticate(clusterService.threadPool().getThreadContext(), credentials, ActionListener.wrap(authResult -> {
             if (authResult.isAuthenticated()) {

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

@@ -0,0 +1,45 @@
+/*
+ * 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.authc;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+
+import java.util.Map;
+
+/**
+ * Service interface for authenticating remote cluster requests.
+ *
+ * <p>
+ * This service handles authentication for cross-cluster requests.
+ * It provides methods to authenticate both full transport requests
+ * and credential headers only.
+ */
+public interface RemoteClusterAuthenticationService {
+
+    /**
+     * Called to authenticates a remote cluster transport request.
+     *
+     * @param action the transport action being performed
+     * @param request the transport request containing authentication headers
+     * @param listener callback to receive the authenticated {@link Authentication}
+     *                 object on success, or an exception on failure
+     */
+    void authenticate(String action, TransportRequest request, ActionListener<Authentication> listener);
+
+    /**
+     * Called early (after transport headers were received) to authenticate a remote cluster transport request.
+     *
+     * @param headers map of request headers containing authentication credentials
+     * @param listener callback to receive {@code null} on successful authentication,
+     *                 or an exception on authentication failure
+     */
+    void authenticateHeaders(Map<String, String> headers, ActionListener<Void> listener);
+
+}

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

@@ -23,7 +23,6 @@
 
 import java.util.Arrays;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 
 public final class PreAuthorizationUtils {
@@ -118,17 +117,17 @@ private static boolean shouldPreAuthorizeChildActionOfParent(final String parent
     }
 
     public static boolean shouldRemoveParentAuthorizationFromThreadContext(
-        Optional<String> remoteClusterAlias,
         String childAction,
-        SecurityContext securityContext
+        SecurityContext securityContext,
+        boolean isRemoteClusterRequest
     ) {
         final ParentActionAuthorization parentAuthorization = securityContext.getParentAuthorization();
         if (parentAuthorization == null) {
             // Nothing to remove.
             return false;
         }
 
-        if (remoteClusterAlias.isPresent()) {
+        if (isRemoteClusterRequest) {
             // We never want to send the parent authorization header to remote clusters.
             return true;
         }