Load/reload API Keys for SecurityServerTransportInterceptor to use for Transport requests to remote clusters (#90890)

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

@@ -334,6 +334,7 @@ import org.elasticsearch.xpack.security.rest.action.user.RestSetEnabledAction;
 import org.elasticsearch.xpack.security.support.CacheInvalidatorRegistry;
 import org.elasticsearch.xpack.security.support.ExtensionComponents;
 import org.elasticsearch.xpack.security.support.SecuritySystemIndices;
+import org.elasticsearch.xpack.security.transport.RemoteClusterAuthorizationResolver;
 import org.elasticsearch.xpack.security.transport.SecurityHttpSettings;
 import org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor;
 import org.elasticsearch.xpack.security.transport.filter.IPFilter;
@@ -897,6 +898,12 @@ public class Security extends Plugin
 
         ipFilter.set(new IPFilter(settings, auditTrailService, clusterService.getClusterSettings(), getLicenseState()));
         components.add(ipFilter.get());
+
+        final RemoteClusterAuthorizationResolver remoteClusterAuthorizationResolver = new RemoteClusterAuthorizationResolver(
+            settings,
+            clusterService.getClusterSettings()
+        );
+
         DestructiveOperations destructiveOperations = new DestructiveOperations(settings, clusterService.getClusterSettings());
         securityInterceptor.set(
             new SecurityServerTransportInterceptor(
@@ -906,7 +913,8 @@ public class Security extends Plugin
                 authzService,
                 getSslService(),
                 securityContext.get(),
-                destructiveOperations
+                destructiveOperations,
+                remoteClusterAuthorizationResolver
             )
         );
 

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

@@ -0,0 +1,64 @@
+/*
+ * 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.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
+import org.elasticsearch.transport.TcpTransport;
+
+import java.util.Map;
+
+import static org.elasticsearch.transport.RemoteClusterService.REMOTE_CLUSTER_AUTHORIZATION;
+
+public class RemoteClusterAuthorizationResolver {
+
+    private static final Logger LOGGER = LogManager.getLogger(RemoteClusterAuthorizationResolver.class);
+
+    private final Map<String, String> apiKeys = ConcurrentCollections.newConcurrentMap();
+
+    /**
+     * Initialize load and reload REMOTE_CLUSTER_AUTHORIZATION values.
+     * @param settings Contains zero, one, or many values of REMOTE_CLUSTER_AUTHORIZATION literal values.
+     * @param clusterSettings Contains one affix setting REMOTE_CLUSTER_AUTHORIZATION.
+     */
+    public RemoteClusterAuthorizationResolver(final Settings settings, final ClusterSettings clusterSettings) {
+        if (TcpTransport.isUntrustedRemoteClusterEnabled()) {
+            for (final Map.Entry<String, String> entry : REMOTE_CLUSTER_AUTHORIZATION.getAsMap(settings).entrySet()) {
+                if (Strings.isEmpty(entry.getValue()) == false) {
+                    this.updateAuthorization(entry.getKey(), entry.getValue());
+                }
+            }
+            clusterSettings.addAffixUpdateConsumer(
+                REMOTE_CLUSTER_AUTHORIZATION,
+                this::updateAuthorization,
+                (clusterAlias, authorization) -> {}
+            );
+        }
+    }
+
+    public String resolveAuthorization(final String clusterAlias) {
+        if (TcpTransport.isUntrustedRemoteClusterEnabled()) {
+            return this.apiKeys.get(clusterAlias);
+        }
+        return null;
+    }
+
+    private void updateAuthorization(final String clusterAlias, final String authorization) {
+        if (Strings.isEmpty(authorization)) {
+            apiKeys.remove(clusterAlias);
+            LOGGER.debug("Authorization value for clusterAlias {} removed", clusterAlias);
+        } else {
+            final boolean notFound = Strings.isEmpty(apiKeys.put(clusterAlias, authorization));
+            LOGGER.debug("Authorization value for clusterAlias {} {}", clusterAlias, (notFound ? "added" : "updated"));
+        }
+    }
+}

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

@@ -54,6 +54,7 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
     private final ThreadPool threadPool;
     private final Settings settings;
     private final SecurityContext securityContext;
+    private final RemoteClusterAuthorizationResolver remoteClusterAuthorizationResolver;
 
     public SecurityServerTransportInterceptor(
         Settings settings,
@@ -62,7 +63,8 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
         AuthorizationService authzService,
         SSLService sslService,
         SecurityContext securityContext,
-        DestructiveOperations destructiveOperations
+        DestructiveOperations destructiveOperations,
+        RemoteClusterAuthorizationResolver remoteClusterAuthorizationResolver
     ) {
         this.settings = settings;
         this.threadPool = threadPool;
@@ -71,6 +73,7 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
         this.sslService = sslService;
         this.securityContext = securityContext;
         this.profileFilters = initializeProfileFilters(destructiveOperations);
+        this.remoteClusterAuthorizationResolver = remoteClusterAuthorizationResolver;
     }
 
     @Override

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

@@ -0,0 +1,134 @@
+/*
+ * 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.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.cluster.routing.RoutingTable;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.settings.SecureString;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.ClusterServiceUtils;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.threadpool.TestThreadPool;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TcpTransport;
+import org.junit.BeforeClass;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+
+public class RemoteClusterAuthorizationResolverTests extends ESTestCase {
+
+    private ThreadPool threadPool;
+    private ClusterService clusterService;
+
+    @BeforeClass
+    public static void checkFeatureFlag() {
+        assumeTrue("untrusted remote cluster feature flag must be enabled", TcpTransport.isUntrustedRemoteClusterEnabled());
+    }
+
+    @Override
+    public void setUp() throws Exception {
+        super.setUp();
+        this.threadPool = new TestThreadPool(getTestName());
+        this.clusterService = ClusterServiceUtils.createClusterService(this.threadPool);
+    }
+
+    @Override
+    public void tearDown() throws Exception {
+        super.tearDown();
+        this.clusterService.close();
+        terminate(this.threadPool);
+    }
+
+    public void testRemoteClusterApiKeyChanges() {
+        final String clusterNameA = "clusterA";
+        final String clusterNameB = "clusterB";
+        final String clusterDoesNotExist = randomAlphaOfLength(10);
+        final Settings.Builder initialSettingsBuilder = Settings.builder();
+        initialSettingsBuilder.put("cluster.remote." + clusterNameA + ".authorization", "initialize");
+        if (randomBoolean()) {
+            initialSettingsBuilder.put("cluster.remote." + clusterNameB + ".authorization", "");
+        }
+        final Settings initialSettings = initialSettingsBuilder.build();
+        RemoteClusterAuthorizationResolver remoteClusterAuthorizationResolver = new RemoteClusterAuthorizationResolver(
+            initialSettings,
+            this.clusterService.getClusterSettings()
+        );
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA), is(equalTo("initialize")));
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB), is(nullValue()));
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterDoesNotExist), is(nullValue()));
+        final DiscoveryNode masterNodeA = this.clusterService.state().nodes().getMasterNode();
+
+        // Add clusterB authorization setting
+        final String clusterBapiKey1 = randomApiKey();
+        final Settings newSettingsAddClusterB = Settings.builder()
+            .put("cluster.remote." + clusterNameA + ".authorization", "addB")
+            .put("cluster.remote." + clusterNameB + ".authorization", clusterBapiKey1)
+            .build();
+        final ClusterState newClusterState1 = createClusterState(clusterNameA, masterNodeA, newSettingsAddClusterB);
+        ClusterServiceUtils.setState(this.clusterService, newClusterState1);
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA), is(equalTo("addB")));
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB), is(equalTo(clusterBapiKey1)));
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterDoesNotExist), is(nullValue()));
+
+        // Change clusterB authorization setting
+        final String clusterBapiKey2 = randomApiKey();
+        final Settings newSettingsUpdateClusterB = Settings.builder()
+            .put("cluster.remote." + clusterNameA + ".authorization", "editB")
+            .put("cluster.remote." + clusterNameB + ".authorization", clusterBapiKey2)
+            .build();
+        final ClusterState newClusterState2 = createClusterState(clusterNameA, masterNodeA, newSettingsUpdateClusterB);
+        ClusterServiceUtils.setState(this.clusterService, newClusterState2);
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA), is(equalTo("editB")));
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB), is(equalTo(clusterBapiKey2)));
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterDoesNotExist), is(nullValue()));
+
+        // Remove clusterB authorization setting
+        final Settings.Builder newSettingsOmitClusterBBuilder = Settings.builder();
+        newSettingsOmitClusterBBuilder.put("cluster.remote." + clusterNameA + ".authorization", "omitB");
+        if (randomBoolean()) {
+            initialSettingsBuilder.put("cluster.remote." + clusterNameB + ".authorization", "");
+        }
+        final Settings newSettingsOmitClusterB = newSettingsOmitClusterBBuilder.build();
+        final ClusterState newClusterState3 = createClusterState(clusterNameA, masterNodeA, newSettingsOmitClusterB);
+        ClusterServiceUtils.setState(this.clusterService, newClusterState3);
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA), is(equalTo("omitB")));
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB), is(nullValue()));
+        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterDoesNotExist), is(nullValue()));
+    }
+
+    private static ClusterState createClusterState(final String clusterName, final DiscoveryNode masterNode, final Settings newSettings) {
+        final DiscoveryNodes.Builder discoBuilder = DiscoveryNodes.builder();
+        discoBuilder.add(masterNode);
+        discoBuilder.masterNodeId(masterNode.getId());
+
+        final ClusterState.Builder state = ClusterState.builder(new ClusterName(clusterName));
+        state.nodes(discoBuilder);
+        state.metadata(Metadata.builder().persistentSettings(newSettings).generateClusterUuidIfNeeded());
+        state.routingTable(RoutingTable.builder().build());
+        return state.build();
+    }
+
+    private String randomApiKey() {
+        final String id = "apikey_" + randomAlphaOfLength(6);
+        // Sufficient for testing. See ApiKeyService and ApiKeyService.ApiKeyCredentials for actual API Key generation.
+        try (SecureString secret = UUIDs.randomBase64UUIDSecureString()) {
+            final String apiKey = id + ":" + secret;
+            return Base64.getEncoder().withoutPadding().encodeToString(apiKey.getBytes(StandardCharsets.UTF_8));
+        }
+    }
+}

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

@@ -111,7 +111,8 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
             new DestructiveOperations(
                 Settings.EMPTY,
                 new ClusterSettings(Settings.EMPTY, Collections.singleton(DestructiveOperations.REQUIRES_NAME_SETTING))
-            )
+            ),
+            new RemoteClusterAuthorizationResolver(settings, clusterService.getClusterSettings())
         );
         ClusterServiceUtils.setState(clusterService, clusterService.state()); // force state update to trigger listener
 
@@ -160,7 +161,8 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
             new DestructiveOperations(
                 Settings.EMPTY,
                 new ClusterSettings(Settings.EMPTY, Collections.singleton(DestructiveOperations.REQUIRES_NAME_SETTING))
-            )
+            ),
+            new RemoteClusterAuthorizationResolver(settings, clusterService.getClusterSettings())
         );
         ClusterServiceUtils.setState(clusterService, clusterService.state()); // force state update to trigger listener
 
@@ -202,7 +204,8 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
             new DestructiveOperations(
                 Settings.EMPTY,
                 new ClusterSettings(Settings.EMPTY, Collections.singleton(DestructiveOperations.REQUIRES_NAME_SETTING))
-            )
+            ),
+            new RemoteClusterAuthorizationResolver(settings, clusterService.getClusterSettings())
         ) {
             @Override
             void assertNoAuthentication(String action) {}
@@ -262,7 +265,8 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
             new DestructiveOperations(
                 Settings.EMPTY,
                 new ClusterSettings(Settings.EMPTY, Collections.singleton(DestructiveOperations.REQUIRES_NAME_SETTING))
-            )
+            ),
+            new RemoteClusterAuthorizationResolver(settings, clusterService.getClusterSettings())
         );
         ClusterServiceUtils.setState(clusterService, clusterService.state()); // force state update to trigger listener
 
@@ -328,7 +332,8 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
             new DestructiveOperations(
                 Settings.EMPTY,
                 new ClusterSettings(Settings.EMPTY, Collections.singleton(DestructiveOperations.REQUIRES_NAME_SETTING))
-            )
+            ),
+            new RemoteClusterAuthorizationResolver(settings, clusterService.getClusterSettings())
         );
         ClusterServiceUtils.setState(clusterService, clusterService.state()); // force state update to trigger listener
 
@@ -390,7 +395,8 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
             new DestructiveOperations(
                 Settings.EMPTY,
                 new ClusterSettings(Settings.EMPTY, Collections.singleton(DestructiveOperations.REQUIRES_NAME_SETTING))
-            )
+            ),
+            new RemoteClusterAuthorizationResolver(settings, clusterService.getClusterSettings())
         );
 
         final AtomicBoolean calledWrappedSender = new AtomicBoolean(false);