Fix "unexpected field [remote_cluster]" for CCS (RCS 1.0) when using API key that references remote_cluster (elastic#112226) (elastic#112259)

This commit will remove the "remote_cluster" from the role descriptor of API keys that is sent to elder clusters for RCS 1.0. 
This will allow API keys created in 8.15.0 that reference "remote_cluster" to work when sent to an elder cluster. 
The API key could either explicitly reference "remote_cluster" or implicitly reference it via the limited by permissions of the superuser built in role.

Note this in reference to the standard API key, not cross cluster API key. The cross cluster API already removes this for elder clusters.

fixes: elastic#112222
related: elastic#107493

Author: jakelandis

docs/changelog/112226.yaml

@@ -0,0 +1,6 @@
+pr: 112226
+summary: "Fix \"unexpected field [remote_cluster]\" for CCS (RCS 1.0) when using API\
+  \ key that references `remote_cluster`"
+area: Security
+type: bug
+issues: []

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

@@ -54,6 +54,7 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static org.elasticsearch.TransportVersions.ROLE_REMOTE_CLUSTER_PRIVS;
 import static org.elasticsearch.transport.RemoteClusterPortSettings.TRANSPORT_VERSION_ADVANCED_REMOTE_CLUSTER_SECURITY;
 import static org.elasticsearch.xcontent.ConstructingObjectParser.constructorArg;
 import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstructorArg;
@@ -1319,6 +1320,24 @@ private static Map<String, Object> maybeRewriteMetadataForApiKeyRoleDescriptors(
                     )
                 );
             }
+
+            if (authentication.getEffectiveSubject().getTransportVersion().onOrAfter(ROLE_REMOTE_CLUSTER_PRIVS)
+                && streamVersion.before(ROLE_REMOTE_CLUSTER_PRIVS)) {
+                metadata = new HashMap<>(metadata);
+                metadata.put(
+                    AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY,
+                    maybeRemoveRemoteClusterFromRoleDescriptors(
+                        (BytesReference) metadata.get(AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY)
+                    )
+                );
+                metadata.put(
+                    AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
+                    maybeRemoveRemoteClusterFromRoleDescriptors(
+                        (BytesReference) metadata.get(AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY)
+                    )
+                );
+            }
+
             if (authentication.getEffectiveSubject().getTransportVersion().onOrAfter(VERSION_API_KEY_ROLES_AS_BYTES)
                 && streamVersion.before(VERSION_API_KEY_ROLES_AS_BYTES)) {
                 metadata = new HashMap<>(metadata);
@@ -1397,7 +1416,15 @@ private static BytesReference convertRoleDescriptorsMapToBytes(Map<String, Objec
         }
     }
 
+    static BytesReference maybeRemoveRemoteClusterFromRoleDescriptors(BytesReference roleDescriptorsBytes) {
+        return maybeRemoveTopLevelFromRoleDescriptors(roleDescriptorsBytes, RoleDescriptor.Fields.REMOTE_CLUSTER.getPreferredName());
+    }
+
     static BytesReference maybeRemoveRemoteIndicesFromRoleDescriptors(BytesReference roleDescriptorsBytes) {
+        return maybeRemoveTopLevelFromRoleDescriptors(roleDescriptorsBytes, RoleDescriptor.Fields.REMOTE_INDICES.getPreferredName());
+    }
+
+    static BytesReference maybeRemoveTopLevelFromRoleDescriptors(BytesReference roleDescriptorsBytes, String topLevelField) {
         if (roleDescriptorsBytes == null || roleDescriptorsBytes.length() == 0) {
             return roleDescriptorsBytes;
         }
@@ -1408,7 +1435,7 @@ static BytesReference maybeRemoveRemoteIndicesFromRoleDescriptors(BytesReference
             if (value instanceof Map) {
                 @SuppressWarnings("unchecked")
                 Map<String, Object> roleDescriptor = (Map<String, Object>) value;
-                boolean removed = roleDescriptor.remove(RoleDescriptor.Fields.REMOTE_INDICES.getPreferredName()) != null;
+                boolean removed = roleDescriptor.remove(topLevelField) != null;
                 if (removed) {
                     removedAtLeastOne.set(true);
                 }

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

@@ -1089,6 +1089,51 @@ public void testMaybeRewriteMetadataForApiKeyRoleDescriptorsWithRemoteIndices()
         );
     }
 
+    public void testMaybeRewriteMetadataForApiKeyRoleDescriptorsWithRemoteCluster() {
+        final String apiKeyId = randomAlphaOfLengthBetween(1, 10);
+        final String apiKeyName = randomAlphaOfLengthBetween(1, 10);
+        final Map<String, Object> metadata = Map.ofEntries(
+            entry(AuthenticationField.API_KEY_ID_KEY, apiKeyId),
+            entry(AuthenticationField.API_KEY_NAME_KEY, apiKeyName),
+            entry(AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY, new BytesArray("""
+                {"base_role":{"cluster":["all"],
+                 "remote_cluster":[{"privileges":["monitor_enrich"],"clusters":["*"]}]
+                }}""")),
+            entry(AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, new BytesArray("""
+                {"limited_by_role":{"cluster":["*"],
+                 "remote_cluster":[{"privileges":["monitor_enrich"],"clusters":["*"]}]
+                }}"""))
+        );
+
+        final Authentication original = AuthenticationTestHelper.builder()
+            .apiKey()
+            .metadata(metadata)
+            .transportVersion(TransportVersions.ROLE_REMOTE_CLUSTER_PRIVS)
+            .build();
+
+        // pick a version before that of the authentication instance to force a rewrite
+        final TransportVersion olderVersion = TransportVersionUtils.randomVersionBetween(
+            random(),
+            Authentication.VERSION_API_KEY_ROLES_AS_BYTES,
+            TransportVersionUtils.getPreviousVersion(original.getEffectiveSubject().getTransportVersion())
+        );
+
+        final Map<String, Object> rewrittenMetadata = original.maybeRewriteForOlderVersion(olderVersion)
+            .getEffectiveSubject()
+            .getMetadata();
+        assertThat(rewrittenMetadata.keySet(), equalTo(original.getAuthenticatingSubject().getMetadata().keySet()));
+        assertThat(
+            ((BytesReference) rewrittenMetadata.get(AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY)).toBytesRef(),
+            equalTo(new BytesArray("""
+                {"base_role":{"cluster":["all"]}}""").toBytesRef())
+        );
+        assertThat(
+            ((BytesReference) rewrittenMetadata.get(AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY)).toBytesRef(),
+            equalTo(new BytesArray("""
+                {"limited_by_role":{"cluster":["*"]}}""").toBytesRef())
+        );
+    }
+
     public void testMaybeRemoveRemoteIndicesFromRoleDescriptors() {
         final boolean includeClusterPrivileges = randomBoolean();
         final BytesReference roleWithoutRemoteIndices = new BytesArray(Strings.format("""

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

@@ -113,6 +113,12 @@ public void testBwcWithLegacyCrossClusterSearch() throws Exception {
                       "privileges": ["read", "read_cross_cluster"],
                       "clusters": ["my_remote_cluster"]
                     }
+                  ],
+                  "remote_cluster": [
+                    {
+                      "privileges": ["monitor_enrich"],
+                      "clusters": ["*"]
+                    }
                   ]
                 }""");
             assertOK(adminClient().performRequest(putRoleRequest));
@@ -157,6 +163,12 @@ public void testBwcWithLegacyCrossClusterSearch() throws Exception {
                           "privileges": ["read", "read_cross_cluster"],
                           "clusters": ["my_remote_*", "non_existing_remote_cluster"]
                         }
+                      ],
+                      "remote_cluster": [
+                        {
+                          "privileges": ["monitor_enrich"],
+                          "clusters": ["*"]
+                        }
                       ]
                     }
                   }