Consolidate use of RemoteAccessHeaders (#93662)

This refactor PR extracts the RemoteAccessHeaders class and re-uses it
between querying-cluster and fulfilling-cluster side code.

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

@@ -56,7 +56,11 @@ public final class RemoteAccessAuthentication {
     }
 
     public static RemoteAccessAuthentication readFromContext(final ThreadContext ctx) throws IOException {
-        return decode(ctx.getHeader(REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY));
+        final String header = ctx.getHeader(REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY);
+        if (header == null) {
+            throw new IllegalArgumentException("remote access header [" + REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY + "] is required");
+        }
+        return decode(header);
     }
 
     public Authentication getAuthentication() {
@@ -212,5 +216,23 @@ public final class RemoteAccessAuthentication {
         public String toString() {
             return "RoleDescriptorsBytes{" + "rawBytes=" + rawBytes + '}';
         }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            if (false == super.equals(o)) return false;
+
+            RoleDescriptorsBytes that = (RoleDescriptorsBytes) o;
+
+            return Objects.equals(rawBytes, that.rawBytes);
+        }
+
+        @Override
+        public int hashCode() {
+            int result = super.hashCode();
+            result = 31 * result + (rawBytes != null ? rawBytes.hashCode() : 0);
+            return result;
+        }
     }
 }

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

@@ -23,6 +23,7 @@ import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.xpack.core.security.authc.RemoteAccessAuthentication.REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY;
 import static org.elasticsearch.xpack.core.security.authz.RoleDescriptorTests.randomUniquelyNamedRoleDescriptors;
 import static org.hamcrest.Matchers.equalTo;
 
@@ -58,6 +59,14 @@ public class RemoteAccessAuthenticationTests extends ESTestCase {
         assertThat(actualRoleDescriptors, equalTo(expectedRoleDescriptors));
     }
 
+    public void testThrowsOnMissingEntry() {
+        var actual = expectThrows(
+            IllegalArgumentException.class,
+            () -> RemoteAccessAuthentication.readFromContext(new ThreadContext(Settings.EMPTY))
+        );
+        assertThat(actual.getMessage(), equalTo("remote access header [" + REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY + "] is required"));
+    }
+
     private RoleDescriptorsIntersection randomRoleDescriptorsIntersection() {
         return new RoleDescriptorsIntersection(randomList(0, 3, () -> Set.copyOf(randomUniquelyNamedRoleDescriptors(0, 1))));
     }

x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/RemoteClusterSecurityRestIT.java

@@ -183,6 +183,7 @@ public class RemoteClusterSecurityRestIT extends AbstractRemoteClusterSecurityTe
         return client().performRequest(request);
     }
 
+    // TODO centralize common usage of this across all tests
     private static String randomEncodedApiKey() {
         return Base64.getEncoder().encodeToString((UUIDs.base64UUID() + ":" + UUIDs.base64UUID()).getBytes(StandardCharsets.UTF_8));
     }

x-pack/plugin/security/qa/security-trial/src/javaRestTest/java/org/elasticsearch/xpack/security/remoteaccess/RemoteAccessHeadersForCcsRestIT.java

@@ -53,8 +53,8 @@ import org.elasticsearch.xpack.core.security.authz.RoleDescriptorsIntersection;
 import org.elasticsearch.xpack.core.security.user.SystemUser;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.security.SecurityOnTrialLicenseRestTestCase;
+import org.elasticsearch.xpack.security.authc.RemoteAccessHeaders;
 import org.elasticsearch.xpack.security.authz.RBACEngine;
-import org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -393,9 +393,9 @@ public class RemoteAccessHeadersForCcsRestIT extends SecurityOnTrialLicenseRestT
     }
 
     private void assertContainsRemoteClusterCredential(String clusterCredential, CapturedActionWithHeaders actual) {
-        assertThat(actual.headers(), hasKey(SecurityServerTransportInterceptor.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY));
+        assertThat(actual.headers(), hasKey(RemoteAccessHeaders.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY));
         assertThat(
-            actual.headers().get(SecurityServerTransportInterceptor.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY),
+            actual.headers().get(RemoteAccessHeaders.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY),
             equalTo("ApiKey " + clusterCredential)
         );
     }
@@ -480,7 +480,7 @@ public class RemoteAccessHeadersForCcsRestIT extends SecurityOnTrialLicenseRestT
             actualHeaders.keySet(),
             containsInAnyOrder(
                 RemoteAccessAuthentication.REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY,
-                SecurityServerTransportInterceptor.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY
+                RemoteAccessHeaders.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY
             )
         );
     }

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

@@ -980,12 +980,15 @@ public class ApiKeyService {
         return getCredentialsFromHeader(threadContext.getHeader("Authorization"));
     }
 
-    ApiKeyCredentials getCredentialsFromHeader(final String header) {
+    static ApiKeyCredentials getCredentialsFromHeader(final String header) {
-        assert isEnabled() : "API keys must be enabled";
         return parseApiKey(Authenticator.extractCredentialFromHeaderValue(header, "ApiKey"));
     }
 
-    private ApiKeyCredentials parseApiKey(SecureString apiKeyString) {
+    public static String withApiKeyPrefix(final String encodedApiKey) {
+        return "ApiKey " + encodedApiKey;
+    }
+
+    private static ApiKeyCredentials parseApiKey(SecureString apiKeyString) {
         if (apiKeyString != null) {
             final byte[] decodedApiKeyCredBytes = Base64.getDecoder().decode(CharArrays.toUtf8Bytes(apiKeyString.getChars()));
             char[] apiKeyCredChars = null;
@@ -1087,6 +1090,7 @@ public class ApiKeyService {
         public void clearCredentials() {
             close();
         }
+
     }
 
     private static class ApiKeyLoggingDeprecationHandler implements DeprecationHandler {

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

@@ -34,7 +34,7 @@ import java.util.function.Supplier;
 
 import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.xpack.core.security.authc.RemoteAccessAuthentication.REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY;
-import static org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY;
+import static org.elasticsearch.xpack.security.authc.RemoteAccessHeaders.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY;
 
 public class RemoteAccessAuthenticationService {
 
@@ -87,7 +87,8 @@ public class RemoteAccessAuthenticationService {
 
         final RemoteAccessHeaders remoteAccessHeaders;
         try {
-            remoteAccessHeaders = extractRemoteAccessHeaders(threadContext);
+            apiKeyService.ensureEnabled();
+            remoteAccessHeaders = RemoteAccessHeaders.readFromContext(threadContext);
         } catch (Exception ex) {
             withRequestProcessingFailure(authcContext, ex, listener);
             return;
@@ -104,7 +105,7 @@ public class RemoteAccessAuthenticationService {
             )
         ) {
             final Supplier<ThreadContext.StoredContext> storedContextSupplier = threadContext.newRestorableContext(false);
-            authcContext.addAuthenticationToken(remoteAccessHeaders.clusterCredential());
+            authcContext.addAuthenticationToken(remoteAccessHeaders.clusterCredentials());
             authenticationService.authenticate(
                 authcContext,
                 new ContextPreservingActionListener<>(storedContextSupplier, ActionListener.wrap(authentication -> {
@@ -122,7 +123,7 @@ public class RemoteAccessAuthenticationService {
         }
     }
 
-    private static RemoteAccessAuthentication maybeRewriteForSystemUser(RemoteAccessAuthentication remoteAccessAuthentication)
+    private static RemoteAccessAuthentication maybeRewriteForSystemUser(final RemoteAccessAuthentication remoteAccessAuthentication)
         throws IOException {
         final Subject receivedEffectiveSubject = remoteAccessAuthentication.getAuthentication().getEffectiveSubject();
         final User user = receivedEffectiveSubject.getUser();
@@ -148,29 +149,6 @@ public class RemoteAccessAuthenticationService {
         return authenticationService;
     }
 
-    private record RemoteAccessHeaders(
-        ApiKeyService.ApiKeyCredentials clusterCredential,
-        RemoteAccessAuthentication remoteAccessAuthentication
-    ) {}
-
-    private RemoteAccessHeaders extractRemoteAccessHeaders(final ThreadContext threadContext) throws IOException {
-        apiKeyService.ensureEnabled();
-        final String clusterCredentialHeader = threadContext.getHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY);
-        if (clusterCredentialHeader == null) {
-            throw new IllegalArgumentException("remote access header [" + REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY + "] is required");
-        }
-        final ApiKeyService.ApiKeyCredentials apiKeyCredential = apiKeyService.getCredentialsFromHeader(clusterCredentialHeader);
-        if (apiKeyCredential == null) {
-            throw new IllegalArgumentException(
-                "remote access header [" + REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY + "] value must be a valid API key credential"
-            );
-        }
-        if (threadContext.getHeader(REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY) == null) {
-            throw new IllegalArgumentException("remote access header [" + REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY + "] is required");
-        }
-        return new RemoteAccessHeaders(apiKeyCredential, RemoteAccessAuthentication.readFromContext(threadContext));
-    }
-
     private void validate(final RemoteAccessAuthentication remoteAccessAuthentication) {
         final Subject effectiveSubject = remoteAccessAuthentication.getAuthentication().getEffectiveSubject();
         for (RemoteAccessAuthentication.RoleDescriptorsBytes roleDescriptorsBytes : remoteAccessAuthentication

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

@@ -0,0 +1,76 @@
+/*
+ * 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.common.util.concurrent.ThreadContext;
+import org.elasticsearch.xpack.core.security.authc.RemoteAccessAuthentication;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public final class RemoteAccessHeaders {
+
+    public static final String REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY = "_remote_access_cluster_credential";
+    private final String clusterCredentialsHeader;
+    private final RemoteAccessAuthentication remoteAccessAuthentication;
+
+    public RemoteAccessHeaders(String clusterCredentialsHeader, RemoteAccessAuthentication remoteAccessAuthentication) {
+        assert clusterCredentialsHeader.startsWith("ApiKey ") : "cluster credentials headers must start with [ApiKey ]";
+        this.clusterCredentialsHeader = clusterCredentialsHeader;
+        this.remoteAccessAuthentication = remoteAccessAuthentication;
+    }
+
+    public void writeToContext(final ThreadContext ctx) throws IOException {
+        ctx.putHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY, clusterCredentialsHeader);
+        remoteAccessAuthentication.writeToContext(ctx);
+    }
+
+    public static RemoteAccessHeaders readFromContext(final ThreadContext ctx) throws IOException {
+        final String clusterCredentialsHeader = ctx.getHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY);
+        if (clusterCredentialsHeader == null) {
+            throw new IllegalArgumentException("remote access header [" + REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY + "] is required");
+        }
+        // Invoke parsing logic to validate that the header decodes to a valid API key credential
+        // Call `close` since the returned value is an auto-closable
+        parseClusterCredentialsHeader(clusterCredentialsHeader).close();
+        return new RemoteAccessHeaders(clusterCredentialsHeader, RemoteAccessAuthentication.readFromContext(ctx));
+    }
+
+    public ApiKeyService.ApiKeyCredentials clusterCredentials() {
+        return parseClusterCredentialsHeader(clusterCredentialsHeader);
+    }
+
+    private static ApiKeyService.ApiKeyCredentials parseClusterCredentialsHeader(final String header) {
+        try {
+            return Objects.requireNonNull(ApiKeyService.getCredentialsFromHeader(header));
+        } catch (Exception ex) {
+            throw new IllegalArgumentException(
+                "remote access header [" + REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY + "] value must be a valid API key credential",
+                ex
+            );
+        }
+    }
+
+    public RemoteAccessAuthentication remoteAccessAuthentication() {
+        return remoteAccessAuthentication;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == this) return true;
+        if (obj == null || obj.getClass() != this.getClass()) return false;
+        var that = (RemoteAccessHeaders) obj;
+        return Objects.equals(this.clusterCredentialsHeader, that.clusterCredentialsHeader)
+            && Objects.equals(this.remoteAccessAuthentication, that.remoteAccessAuthentication);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(clusterCredentialsHeader, remoteAccessAuthentication);
+    }
+}

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

@@ -21,7 +21,7 @@ import java.util.HashSet;
 import java.util.Set;
 
 import static org.elasticsearch.xpack.core.security.authc.RemoteAccessAuthentication.REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY;
-import static org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY;
+import static org.elasticsearch.xpack.security.authc.RemoteAccessHeaders.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY;
 
 final class RemoteAccessServerTransportFilter extends ServerTransportFilter {
     // pkg-private for testing

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

@@ -14,6 +14,7 @@ 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 org.elasticsearch.xpack.security.authc.ApiKeyService;
 
 import java.util.Map;
 
@@ -47,7 +48,8 @@ public class RemoteClusterAuthorizationResolver {
 
     public String resolveAuthorization(final String clusterAlias) {
         if (TcpTransport.isUntrustedRemoteClusterEnabled()) {
-            return this.apiKeys.get(clusterAlias);
+            final String apiKey = apiKeys.get(clusterAlias);
+            return apiKey == null ? null : ApiKeyService.withApiKeyPrefix(apiKey);
         }
         return null;
     }

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

@@ -49,6 +49,7 @@ import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.security.authc.AuthenticationService;
 import org.elasticsearch.xpack.security.authc.RemoteAccessAuthenticationService;
+import org.elasticsearch.xpack.security.authc.RemoteAccessHeaders;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationUtils;
 import org.elasticsearch.xpack.security.authz.PreAuthorizationUtils;
@@ -67,7 +68,6 @@ import static org.elasticsearch.transport.RemoteClusterPortSettings.REMOTE_CLUST
 
 public class SecurityServerTransportInterceptor implements TransportInterceptor {
 
-    public static final String REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY = "_remote_access_cluster_credential";
     private static final TransportVersion VERSION_REMOTE_ACCESS_HEADERS = TransportVersion.V_8_7_0;
     private static final Logger logger = LogManager.getLogger(SecurityServerTransportInterceptor.class);
     // package private for testing
@@ -196,7 +196,7 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
             }
 
             private void assertNoRemoteAccessHeadersInContext() {
-                assert securityContext.getThreadContext().getHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY) == null
+                assert securityContext.getThreadContext().getHeader(RemoteAccessHeaders.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY) == null
                     : "remote access headers should not be in security context";
                 assert securityContext.getThreadContext()
                     .getHeader(RemoteAccessAuthentication.REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY) == null
@@ -366,11 +366,13 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
                 final User user = authentication.getEffectiveSubject().getUser();
                 if (SystemUser.is(user)) {
                     try (ThreadContext.StoredContext ignored = threadContext.stashContext()) {
-                        remoteAccessCredentials.writeToContext(threadContext);
+                        new RemoteAccessHeaders(
-                        // Access control is handled differently for the system user. Privileges are defined by the fulfilling cluster,
+                            remoteAccessCredentials.credentials(),
-                        // so we pass an empty role descriptors intersection here and let the receiver resolve privileges based on the
+                            // Access control is handled differently for the system user. Privileges are defined by the fulfilling cluster,
-                        // authentication instance
+                            // so we pass an empty role descriptors intersection here and let the receiver resolve privileges based on the
-                        new RemoteAccessAuthentication(authentication, RoleDescriptorsIntersection.EMPTY).writeToContext(threadContext);
+                            // authentication instance
+                            new RemoteAccessAuthentication(authentication, RoleDescriptorsIntersection.EMPTY)
+                        ).writeToContext(threadContext);
                         sender.sendRequest(connection, action, request, options, contextRestoreHandler);
                     } catch (IOException e) {
                         contextRestoreHandler.handleException(new SendRequestTransportException(connection.getNode(), action, e));
@@ -385,8 +387,10 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
                         authentication.getEffectiveSubject(),
                         ActionListener.wrap(roleDescriptorsIntersection -> {
                             try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
-                                remoteAccessCredentials.writeToContext(threadContext);
+                                new RemoteAccessHeaders(
-                                new RemoteAccessAuthentication(authentication, roleDescriptorsIntersection).writeToContext(threadContext);
+                                    remoteAccessCredentials.credentials(),
+                                    new RemoteAccessAuthentication(authentication, roleDescriptorsIntersection)
+                                ).writeToContext(threadContext);
                                 sender.sendRequest(connection, action, request, options, contextRestoreHandler);
                             }
                         }, e -> contextRestoreHandler.handleException(new SendRequestTransportException(connection.getNode(), action, e)))
@@ -394,15 +398,8 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
                 }
             }
 
-            record RemoteAccessCredentials(String clusterAlias, String credentials) {
+            // TODO move this to `RemoteClusterAuthorizationResolver` and have `resolveAuthorization` return it
-                void writeToContext(final ThreadContext ctx) {
+            record RemoteAccessCredentials(String clusterAlias, String credentials) {}
-                    ctx.putHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY, withApiKeyPrefix(credentials));
-                }
-
-                private String withApiKeyPrefix(final String clusterCredential) {
-                    return "ApiKey " + clusterCredential;
-                }
-            }
         };
     }
 

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

@@ -0,0 +1,114 @@
+/*
+ * 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.common.UUIDs;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationTestHelper;
+import org.elasticsearch.xpack.core.security.authz.RoleDescriptorsIntersection;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Set;
+
+import static org.elasticsearch.xpack.core.security.authz.RoleDescriptorTests.randomUniquelyNamedRoleDescriptors;
+import static org.elasticsearch.xpack.security.authc.RemoteAccessHeaders.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY;
+import static org.hamcrest.Matchers.equalTo;
+
+public class RemoteAccessHeadersTests extends ESTestCase {
+
+    public void testWriteReadContextRoundtrip() throws IOException {
+        final ThreadContext ctx = new ThreadContext(Settings.EMPTY);
+        final var expected = new RemoteAccessHeaders(
+            randomEncodedApiKeyHeader(),
+            AuthenticationTestHelper.randomRemoteAccessAuthentication(randomRoleDescriptorsIntersection())
+        );
+
+        expected.writeToContext(ctx);
+        final RemoteAccessHeaders actual = RemoteAccessHeaders.readFromContext(ctx);
+
+        assertThat(actual.remoteAccessAuthentication(), equalTo(expected.remoteAccessAuthentication()));
+        assertThat(actual.clusterCredentials().getId(), equalTo(expected.clusterCredentials().getId()));
+        assertThat(actual.clusterCredentials().getKey().toString(), equalTo(expected.clusterCredentials().getKey().toString()));
+    }
+
+    public void testClusterCredentialsReturnsValidApiKey() {
+        final String id = UUIDs.randomBase64UUID();
+        final String key = UUIDs.randomBase64UUID();
+        final String encodedApiKey = encodedApiKeyWithPrefix(id, key);
+        final var headers = new RemoteAccessHeaders(
+            encodedApiKey,
+            AuthenticationTestHelper.randomRemoteAccessAuthentication(randomRoleDescriptorsIntersection())
+        );
+
+        final ApiKeyService.ApiKeyCredentials actual = headers.clusterCredentials();
+
+        assertThat(actual.getId(), equalTo(id));
+        assertThat(actual.getKey().toString(), equalTo(key));
+    }
+
+    public void testReadOnInvalidApiKeyValueThrows() throws IOException {
+        final ThreadContext ctx = new ThreadContext(Settings.EMPTY);
+        final var expected = new RemoteAccessHeaders(
+            randomFrom("ApiKey abc", "ApiKey id:key", "ApiKey ", "ApiKey  "),
+            AuthenticationTestHelper.randomRemoteAccessAuthentication(randomRoleDescriptorsIntersection())
+        );
+
+        expected.writeToContext(ctx);
+        var actual = expectThrows(IllegalArgumentException.class, () -> RemoteAccessHeaders.readFromContext(ctx));
+
+        assertThat(
+            actual.getMessage(),
+            equalTo("remote access header [" + REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY + "] value must be a valid API key credential")
+        );
+    }
+
+    public void testReadOnHeaderWithMalformedPrefixThrows() throws IOException {
+        final ThreadContext ctx = new ThreadContext(Settings.EMPTY);
+        AuthenticationTestHelper.randomRemoteAccessAuthentication(randomRoleDescriptorsIntersection()).writeToContext(ctx);
+        final String encodedApiKey = encodedApiKey(UUIDs.randomBase64UUID(), UUIDs.randomBase64UUID());
+        ctx.putHeader(
+            REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY,
+            randomFrom(
+                // missing space
+                "ApiKey" + encodedApiKey,
+                // no prefix
+                encodedApiKey,
+                // wrong prefix
+                "Bearer " + encodedApiKey
+            )
+        );
+
+        var actual = expectThrows(IllegalArgumentException.class, () -> RemoteAccessHeaders.readFromContext(ctx));
+
+        assertThat(
+            actual.getMessage(),
+            equalTo("remote access header [" + REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY + "] value must be a valid API key credential")
+        );
+    }
+
+    private static RoleDescriptorsIntersection randomRoleDescriptorsIntersection() {
+        return new RoleDescriptorsIntersection(randomList(0, 3, () -> Set.copyOf(randomUniquelyNamedRoleDescriptors(0, 1))));
+    }
+
+    // TODO centralize common usage of this across all tests
+    private static String randomEncodedApiKeyHeader() {
+        return encodedApiKeyWithPrefix(UUIDs.randomBase64UUID(), UUIDs.randomBase64UUID());
+    }
+
+    private static String encodedApiKeyWithPrefix(String id, String key) {
+        return "ApiKey " + encodedApiKey(id, key);
+    }
+
+    private static String encodedApiKey(String id, String key) {
+        return Base64.getEncoder().encodeToString((id + ":" + key).getBytes(StandardCharsets.UTF_8));
+    }
+}

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/remoteaccess/RemoteAccessAuthenticationServiceIntegTests.java

@@ -26,7 +26,9 @@ import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptorTests;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptorsIntersection;
 import org.elasticsearch.xpack.core.security.user.SystemUser;
+import org.elasticsearch.xpack.security.authc.ApiKeyService;
 import org.elasticsearch.xpack.security.authc.RemoteAccessAuthenticationService;
+import org.elasticsearch.xpack.security.authc.RemoteAccessHeaders;
 import org.junit.BeforeClass;
 
 import java.io.IOException;
@@ -38,7 +40,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.function.Consumer;
 
 import static org.elasticsearch.xpack.core.security.authc.RemoteAccessAuthentication.REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY;
-import static org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY;
+import static org.elasticsearch.xpack.security.authc.RemoteAccessHeaders.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
@@ -65,7 +67,8 @@ public class RemoteAccessAuthenticationServiceIntegTests extends SecurityIntegTe
         }
 
         try (var ignored = threadContext.stashContext()) {
-            threadContext.putHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY, "abc");
+            new RemoteAccessHeaders(ApiKeyService.withApiKeyPrefix("abc"), AuthenticationTestHelper.randomRemoteAccessAuthentication())
+                .writeToContext(threadContext);
             authenticateAndAssertExpectedErrorMessage(
                 service,
                 msg -> assertThat(
@@ -82,7 +85,7 @@ public class RemoteAccessAuthenticationServiceIntegTests extends SecurityIntegTe
         try (var ignored = threadContext.stashContext()) {
             final String randomApiKey = Base64.getEncoder()
                 .encodeToString((UUIDs.base64UUID() + ":" + UUIDs.base64UUID()).getBytes(StandardCharsets.UTF_8));
-            threadContext.putHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY, "ApiKey " + randomApiKey);
+            threadContext.putHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY, ApiKeyService.withApiKeyPrefix(randomApiKey));
             authenticateAndAssertExpectedErrorMessage(
                 service,
                 msg -> assertThat(msg, equalTo("remote access header [" + REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY + "] is required"))
@@ -90,11 +93,13 @@ public class RemoteAccessAuthenticationServiceIntegTests extends SecurityIntegTe
         }
 
         try (var ignored = threadContext.stashContext()) {
-            threadContext.putHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY, "ApiKey " + encodedRemoteAccessApiKey);
             final var internalUser = randomValueOtherThan(SystemUser.INSTANCE, AuthenticationTestHelper::randomInternalUser);
-            new RemoteAccessAuthentication(
+            new RemoteAccessHeaders(
-                AuthenticationTestHelper.builder().internal(internalUser).build(),
+                encodedRemoteAccessApiKey,
-                RoleDescriptorsIntersection.EMPTY
+                new RemoteAccessAuthentication(
+                    AuthenticationTestHelper.builder().internal(internalUser).build(),
+                    RoleDescriptorsIntersection.EMPTY
+                )
             ).writeToContext(threadContext);
             authenticateAndAssertExpectedErrorMessage(
                 service,
@@ -106,16 +111,18 @@ public class RemoteAccessAuthenticationServiceIntegTests extends SecurityIntegTe
         }
 
         try (var ignored = threadContext.stashContext()) {
-            threadContext.putHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY, "ApiKey " + encodedRemoteAccessApiKey);
+            new RemoteAccessHeaders(
-            AuthenticationTestHelper.randomRemoteAccessAuthentication(
+                encodedRemoteAccessApiKey,
-                new RoleDescriptorsIntersection(
+                AuthenticationTestHelper.randomRemoteAccessAuthentication(
-                    randomValueOtherThanMany(
+                    new RoleDescriptorsIntersection(
-                        rd -> false == (rd.hasClusterPrivileges()
+                        randomValueOtherThanMany(
-                            || rd.hasApplicationPrivileges()
+                            rd -> false == (rd.hasClusterPrivileges()
-                            || rd.hasConfigurableClusterPrivileges()
+                                || rd.hasApplicationPrivileges()
-                            || rd.hasRunAs()
+                                || rd.hasConfigurableClusterPrivileges()
-                            || rd.hasRemoteIndicesPrivileges()),
+                                || rd.hasRunAs()
-                        () -> RoleDescriptorTests.randomRoleDescriptor()
+                                || rd.hasRemoteIndicesPrivileges()),
+                            () -> RoleDescriptorTests.randomRoleDescriptor()
+                        )
                     )
                 )
             ).writeToContext(threadContext);
@@ -131,9 +138,12 @@ public class RemoteAccessAuthenticationServiceIntegTests extends SecurityIntegTe
         }
 
         try (var ignored = threadContext.stashContext()) {
-            threadContext.putHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY, "ApiKey " + encodedRemoteAccessApiKey);
             Authentication authentication = AuthenticationTestHelper.builder().apiKey().build();
-            new RemoteAccessAuthentication(authentication, RoleDescriptorsIntersection.EMPTY).writeToContext(threadContext);
+            new RemoteAccessHeaders(
+                encodedRemoteAccessApiKey,
+                new RemoteAccessAuthentication(authentication, RoleDescriptorsIntersection.EMPTY)
+            ).writeToContext(threadContext);
+
             authenticateAndAssertExpectedErrorMessage(
                 service,
                 msg -> assertThat(
@@ -151,16 +161,17 @@ public class RemoteAccessAuthenticationServiceIntegTests extends SecurityIntegTe
     }
 
     public void testSystemUserIsMappedToCrossClusterInternalRole() throws InterruptedException, IOException, ExecutionException {
-        final String encodedRemoteAccessApiKey = getEncodedRemoteAccessApiKey();
         final String nodeName = internalCluster().getRandomNodeName();
         final ThreadContext threadContext = internalCluster().getInstance(SecurityContext.class, nodeName).getThreadContext();
         final RemoteAccessAuthenticationService service = internalCluster().getInstance(RemoteAccessAuthenticationService.class, nodeName);
 
         try (var ignored = threadContext.stashContext()) {
-            threadContext.putHeader(REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY, "ApiKey " + encodedRemoteAccessApiKey);
+            new RemoteAccessHeaders(
-            new RemoteAccessAuthentication(
+                getEncodedRemoteAccessApiKey(),
-                AuthenticationTestHelper.builder().internal(SystemUser.INSTANCE).build(),
+                new RemoteAccessAuthentication(
-                new RoleDescriptorsIntersection(new RoleDescriptor("role", null, null, null, null, null, null, null))
+                    AuthenticationTestHelper.builder().internal(SystemUser.INSTANCE).build(),
+                    new RoleDescriptorsIntersection(new RoleDescriptor("role", null, null, null, null, null, null, null))
+                )
             ).writeToContext(threadContext);
 
             final PlainActionFuture<Authentication> future = new PlainActionFuture<>();
@@ -181,7 +192,9 @@ public class RemoteAccessAuthenticationServiceIntegTests extends SecurityIntegTe
 
     private String getEncodedRemoteAccessApiKey() {
         final CreateApiKeyResponse response = new CreateApiKeyRequestBuilder(client().admin().cluster()).setName("remote_access_key").get();
-        return Base64.getEncoder().encodeToString((response.getId() + ":" + response.getKey()).getBytes(StandardCharsets.UTF_8));
+        return ApiKeyService.withApiKeyPrefix(
+            Base64.getEncoder().encodeToString((response.getId() + ":" + response.getKey()).getBytes(StandardCharsets.UTF_8))
+        );
     }
 
     private void authenticateAndAssertExpectedErrorMessage(

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

@@ -21,6 +21,7 @@ import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TcpTransport;
+import org.elasticsearch.xpack.security.authc.ApiKeyService;
 import org.junit.BeforeClass;
 
 import java.nio.charset.StandardCharsets;
@@ -68,7 +69,10 @@ public class RemoteClusterAuthorizationResolverTests extends ESTestCase {
             initialSettings,
             this.clusterService.getClusterSettings()
         );
-        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA), is(equalTo("initialize")));
+        assertThat(
+            remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA),
+            is(equalTo(ApiKeyService.withApiKeyPrefix("initialize")))
+        );
         assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB), is(nullValue()));
         assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterDoesNotExist), is(nullValue()));
         final DiscoveryNode masterNodeA = this.clusterService.state().nodes().getMasterNode();
@@ -81,8 +85,14 @@ public class RemoteClusterAuthorizationResolverTests extends ESTestCase {
             .build();
         final ClusterState newClusterState1 = createClusterState(clusterNameA, masterNodeA, newSettingsAddClusterB);
         ClusterServiceUtils.setState(this.clusterService, newClusterState1);
-        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA), is(equalTo("addB")));
+        assertThat(
-        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB), is(equalTo(clusterBapiKey1)));
+            remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA),
+            is(equalTo(ApiKeyService.withApiKeyPrefix("addB")))
+        );
+        assertThat(
+            remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB),
+            is(equalTo(ApiKeyService.withApiKeyPrefix(clusterBapiKey1)))
+        );
         assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterDoesNotExist), is(nullValue()));
 
         // Change clusterB authorization setting
@@ -93,8 +103,14 @@ public class RemoteClusterAuthorizationResolverTests extends ESTestCase {
             .build();
         final ClusterState newClusterState2 = createClusterState(clusterNameA, masterNodeA, newSettingsUpdateClusterB);
         ClusterServiceUtils.setState(this.clusterService, newClusterState2);
-        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA), is(equalTo("editB")));
+        assertThat(
-        assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB), is(equalTo(clusterBapiKey2)));
+            remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameA),
+            is(equalTo(ApiKeyService.withApiKeyPrefix("editB")))
+        );
+        assertThat(
+            remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB),
+            is(equalTo(ApiKeyService.withApiKeyPrefix(clusterBapiKey2)))
+        );
         assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterDoesNotExist), is(nullValue()));
 
         // Remove clusterB authorization setting
@@ -106,7 +122,10 @@ public class RemoteClusterAuthorizationResolverTests extends ESTestCase {
         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(clusterNameA),
+            is(equalTo(ApiKeyService.withApiKeyPrefix("omitB")))
+        );
         assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterNameB), is(nullValue()));
         assertThat(remoteClusterAuthorizationResolver.resolveAuthorization(clusterDoesNotExist), is(nullValue()));
     }

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

@@ -58,6 +58,7 @@ import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.core.security.user.XPackSecurityUser;
 import org.elasticsearch.xpack.core.security.user.XPackUser;
 import org.elasticsearch.xpack.core.ssl.SSLService;
+import org.elasticsearch.xpack.security.authc.ApiKeyService;
 import org.elasticsearch.xpack.security.authc.AuthenticationService;
 import org.elasticsearch.xpack.security.authc.RemoteAccessAuthenticationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
@@ -81,8 +82,8 @@ import static org.elasticsearch.xpack.core.ClientHelper.SECURITY_PROFILE_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.TRANSFORM_ORIGIN;
 import static org.elasticsearch.xpack.core.security.authc.RemoteAccessAuthentication.REMOTE_ACCESS_AUTHENTICATION_HEADER_KEY;
 import static org.elasticsearch.xpack.core.security.authz.RoleDescriptorTests.randomUniquelyNamedRoleDescriptors;
+import static org.elasticsearch.xpack.security.authc.RemoteAccessHeaders.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY;
 import static org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor.REMOTE_ACCESS_ACTION_ALLOWLIST;
-import static org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor.REMOTE_ACCESS_CLUSTER_CREDENTIAL_HEADER_KEY;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.equalTo;
@@ -581,7 +582,7 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
             : AuthenticationTestHelper.builder().user(new User(randomAlphaOfLengthBetween(3, 10), randomRoles())).realm().build();
         authentication.writeToContext(threadContext);
         final RemoteClusterAuthorizationResolver remoteClusterAuthorizationResolver = mock(RemoteClusterAuthorizationResolver.class);
-        final String remoteClusterCredential = randomAlphaOfLengthBetween(10, 42);
+        final String remoteClusterCredential = ApiKeyService.withApiKeyPrefix(randomAlphaOfLengthBetween(10, 42));
         when(remoteClusterAuthorizationResolver.resolveAuthorization(any())).thenReturn(remoteClusterCredential);
         final String remoteClusterAlias = randomAlphaOfLengthBetween(5, 10);
         final AuthorizationService authzService = mock(AuthorizationService.class);
@@ -664,7 +665,7 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
             listenerCaptor.getValue().onResponse(expectedRoleDescriptorsIntersection);
         }
         assertTrue(calledWrappedSender.get());
-        assertThat(sentCredential.get(), equalTo("ApiKey " + remoteClusterCredential));
+        assertThat(sentCredential.get(), equalTo(remoteClusterCredential));
         assertThat(
             sentRemoteAccessAuthentication.get(),
             equalTo(new RemoteAccessAuthentication(authentication, expectedRoleDescriptorsIntersection))
@@ -693,7 +694,7 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
         }
         final RemoteClusterAuthorizationResolver remoteClusterAuthorizationResolver = mock(RemoteClusterAuthorizationResolver.class);
         when(remoteClusterAuthorizationResolver.resolveAuthorization(any())).thenReturn(
-            noCredential ? null : randomAlphaOfLengthBetween(10, 42)
+            noCredential ? null : ApiKeyService.withApiKeyPrefix(randomAlphaOfLengthBetween(10, 42))
         );
         final AuthenticationTestHelper.AuthenticationTestBuilder builder = AuthenticationTestHelper.builder();
         final Authentication authentication;
@@ -773,7 +774,7 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
             .build();
         authentication.writeToContext(threadContext);
         final RemoteClusterAuthorizationResolver remoteClusterAuthorizationResolver = mock(RemoteClusterAuthorizationResolver.class);
-        final String remoteClusterCredential = randomAlphaOfLengthBetween(10, 42);
+        final String remoteClusterCredential = ApiKeyService.withApiKeyPrefix(randomAlphaOfLengthBetween(10, 42));
         when(remoteClusterAuthorizationResolver.resolveAuthorization(any())).thenReturn(remoteClusterCredential);
         final String remoteClusterAlias = randomAlphaOfLengthBetween(5, 10);
 
@@ -981,4 +982,5 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
     private static Consumer<ThreadContext.StoredContext> anyConsumer() {
         return any(Consumer.class);
     }
+
 }