clean up

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -185,9 +185,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INDEX_REQUEST_REMOVE_METERING = def(8_780_00_0);
     public static final TransportVersion CPU_STAT_STRING_PARSING = def(8_781_00_0);
 
-
-
-    //FIXME: before merging this PR, make sure to update the transport version correctly
+    // FIXME: before merging this PR, make sure to update the transport version correctly
     public static final TransportVersion ROLE_MONITOR_STATS = def(8_882_00_0);
 
     /*

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

@@ -25,11 +25,9 @@
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
-import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentType;
-import org.elasticsearch.xcontent.json.JsonXContent;
 import org.elasticsearch.xpack.core.security.action.profile.Profile;
 import org.elasticsearch.xpack.core.security.authc.CrossClusterAccessSubjectInfo.RoleDescriptorsBytes;
 import org.elasticsearch.xpack.core.security.authc.RealmConfig.RealmIdentifier;
@@ -46,12 +44,10 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
-import java.util.ArrayList;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -239,8 +235,8 @@ public Authentication maybeRewriteForOlderVersion(TransportVersion olderVersion)
                     + "]"
             );
         }
-
         final Map<String, Object> newMetadata = maybeRewriteMetadata(olderVersion, this);
+
         final Authentication newAuthentication;
         if (isRunAs()) {
             // The lookup user for run-as currently doesn't have authentication metadata associated with them because
@@ -278,12 +274,23 @@ public Authentication maybeRewriteForOlderVersion(TransportVersion olderVersion)
     }
 
     private static Map<String, Object> maybeRewriteMetadata(TransportVersion olderVersion, Authentication authentication) {
-        if (authentication.isAuthenticatedAsApiKey()) {
-            return maybeRewriteMetadataForApiKeyRoleDescriptors(olderVersion, authentication);
-        } else if (authentication.isCrossClusterAccess()) {
-            return maybeRewriteMetadataForCrossClusterAccessAuthentication(olderVersion, authentication);
-        } else {
-            return authentication.getAuthenticatingSubject().getMetadata();
+        try {
+            if (authentication.isAuthenticatedAsApiKey()) {
+                return maybeRewriteMetadataForApiKeyRoleDescriptors(olderVersion, authentication);
+            } else if (authentication.isCrossClusterAccess()) {
+                return maybeRewriteMetadataForCrossClusterAccessAuthentication(olderVersion, authentication);
+            } else {
+                return authentication.getAuthenticatingSubject().getMetadata();
+            }
+        } catch (Exception e) {
+            // CCS workflows may swallow the exception message making this difficult to troubleshoot, so we explicitly log and re-throw
+            // here. It may result in duplicate logs, so we only log the message at warn level.
+            if (logger.isDebugEnabled()) {
+                logger.debug("Un-expected exception thrown while rewriting metadata. This is likely a bug.", e);
+            } else {
+                logger.warn("Un-expected exception thrown while rewriting metadata. This is likely a bug [" + e.getMessage() + "]");
+            }
+            throw e;
         }
     }
 
@@ -1302,7 +1309,6 @@ private static Map<String, Object> maybeRewriteMetadataForApiKeyRoleDescriptors(
         TransportVersion streamVersion,
         Authentication authentication
     ) {
-        System.out.println("***************  maybeRewriteMetadataForApiKeyRoleDescriptors  ***************");
         Map<String, Object> metadata = authentication.getAuthenticatingSubject().getMetadata();
         // If authentication user is an API key or a token created by an API key,
         // regardless whether it has run-as, the metadata must contain API key role descriptors
@@ -1344,29 +1350,23 @@ private static Map<String, Object> maybeRewriteMetadataForApiKeyRoleDescriptors(
                         (BytesReference) metadata.get(AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY)
                     )
                 );
-            }
-
-            //TODO: this needs to be in the else so we don't add back the remote_cluster field
-            // the current cluster understands remote_cluster field in role descriptors, so check each permission and remove as needed
-            if (authentication.getEffectiveSubject().getTransportVersion().onOrAfter(ROLE_REMOTE_CLUSTER_PRIVS)){
-
-
-                RemoteClusterPermissions.getSupportedRemoteClusterPermissions();
-
+            } else if (streamVersion.onOrAfter(ROLE_REMOTE_CLUSTER_PRIVS)) {
+                // the remote cluster understands remote_cluster field in role descriptors, so check each permission and remove as needed
                 metadata = new HashMap<>(metadata);
                 metadata.put(
                     AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY,
                     maybeRemoveRemoteClusterPrivilegesFromRoleDescriptors(
-                        (BytesReference) metadata.get(AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY), streamVersion
+                        (BytesReference) metadata.get(AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY),
+                        streamVersion
                     )
                 );
                 metadata.put(
                     AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
                     maybeRemoveRemoteClusterPrivilegesFromRoleDescriptors(
-                        (BytesReference) metadata.get(AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY), streamVersion
+                        (BytesReference) metadata.get(AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY),
+                        streamVersion
                     )
                 );
-
             }
 
             if (authentication.getEffectiveSubject().getTransportVersion().onOrAfter(VERSION_API_KEY_ROLES_AS_BYTES)
@@ -1495,7 +1495,6 @@ static BytesReference maybeRemoveRemoteClusterPrivilegesFromRoleDescriptors(
         if (roleDescriptorsBytes == null || roleDescriptorsBytes.length() == 0) {
             return roleDescriptorsBytes;
         }
-
         final Map<String, Object> roleDescriptorsMap = convertRoleDescriptorsBytesToMap(roleDescriptorsBytes);
         final Map<String, Object> roleDescriptorsMapMutated = new HashMap<>(roleDescriptorsMap);
         final AtomicBoolean modified = new AtomicBoolean(false);

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

@@ -839,8 +839,9 @@ private static RemoteClusterPermissions parseRemoteCluster(final String roleName
                         Arrays.stream(privileges).map(s -> s.toLowerCase(Locale.ROOT)).collect(Collectors.toSet())
                     )) {
                         final String message = String.format(
-                            "failed to parse remote_cluster for role [%s]. " +
-                                "[%s] are the only values allowed for [%s] within [remote_cluster]",
+                            Locale.ROOT,
+                            "failed to parse remote_cluster for role [%s]. "
+                                + "[%s] are the only values allowed for [%s] within [remote_cluster]",
                             roleName,
                             RemoteClusterPermissions.getSupportedRemoteClusterPermissions(),
                             currentFieldName
@@ -852,6 +853,7 @@ private static RemoteClusterPermissions parseRemoteCluster(final String roleName
                     clusters = readStringArray(roleName, parser, false);
                 } else {
                     final String message = String.format(
+                        Locale.ROOT,
                         "failed to parse remote_cluster for role [%s]. unexpected field [%s]",
                         roleName,
                         currentFieldName

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/RemoteClusterPermissionGroup.java

@@ -13,7 +13,6 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
-import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.support.StringMatcher;
 
 import java.io.IOException;
@@ -46,12 +45,11 @@ public RemoteClusterPermissionGroup(StreamInput in) throws IOException {
         remoteClusterAliasMatcher = StringMatcher.of(remoteClusterAliases);
     }
 
-
-    public RemoteClusterPermissionGroup(Map<String, List<String>> remoteClusterGroup)  {
+    public RemoteClusterPermissionGroup(Map<String, List<String>> remoteClusterGroup) {
         assert remoteClusterGroup.get(PRIVILEGES.getPreferredName()) != null : "privileges must be non-null";
         assert remoteClusterGroup.get(CLUSTERS.getPreferredName()) != null : "clusters must be non-null";
-        clusterPrivileges =  remoteClusterGroup.get(PRIVILEGES.getPreferredName()).toArray(new String[0]);
-        remoteClusterAliases =  remoteClusterGroup.get(CLUSTERS.getPreferredName()).toArray(new String[0]);
+        clusterPrivileges = remoteClusterGroup.get(PRIVILEGES.getPreferredName()).toArray(new String[0]);
+        remoteClusterAliases = remoteClusterGroup.get(CLUSTERS.getPreferredName()).toArray(new String[0]);
         remoteClusterAliasMatcher = StringMatcher.of(remoteClusterAliases);
     }
 
@@ -67,10 +65,14 @@ public RemoteClusterPermissionGroup(String[] clusterPrivileges, String[] remoteC
             throw new IllegalArgumentException("remote cluster groups must not be null or empty");
         }
         if (Arrays.stream(clusterPrivileges).anyMatch(s -> Strings.hasText(s) == false)) {
-            throw new IllegalArgumentException("remote_cluster privileges must contain valid non-empty, non-null values");
+            throw new IllegalArgumentException(
+                "remote_cluster privileges must contain valid non-empty, non-null values " + Arrays.toString(clusterPrivileges)
+            );
         }
         if (Arrays.stream(remoteClusterAliases).anyMatch(s -> Strings.hasText(s) == false)) {
-            throw new IllegalArgumentException("remote_cluster clusters aliases must contain valid non-empty, non-null values");
+            throw new IllegalArgumentException(
+                "remote_cluster clusters aliases must contain valid non-empty, non-null values " + Arrays.toString(remoteClusterAliases)
+            );
         }
 
         this.clusterPrivileges = clusterPrivileges;
@@ -100,10 +102,17 @@ public String[] remoteClusterAliases() {
         return Arrays.copyOf(remoteClusterAliases, remoteClusterAliases.length);
     }
 
-
+    /**
+     * Converts the group to a map representation.
+     * @return A map representation of the group.
+     */
     public Map<String, List<String>> toMap() {
-        return Map.of(PRIVILEGES.getPreferredName(), Arrays.asList(clusterPrivileges),
-            CLUSTERS.getPreferredName(), Arrays.asList(remoteClusterAliases));
+        return Map.of(
+            PRIVILEGES.getPreferredName(),
+            Arrays.asList(clusterPrivileges),
+            CLUSTERS.getPreferredName(),
+            Arrays.asList(remoteClusterAliases)
+        );
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/RemoteClusterPermissions.java

@@ -36,9 +36,10 @@
 
 /**
  * Represents the set of permissions for remote clusters. This is intended to be the model for both the {@link RoleDescriptor}
- * and {@link Role}. This model is not intended to be sent to a remote cluster, but can be (wire) serialized within a single cluster
- * as well as the Xcontent serialization for the REST API and persistence of the role in the security index. The privileges modeled here
- * will be converted to the appropriate cluster privileges when sent to a remote cluster.
+ * and {@link Role}. This model is not intended to support RCS 2.0 where these remote cluster permissions are converted to local cluster
+ * permissions {@link #privilegeNames(String, TransportVersion)} ) before sent to the remote cluster. This model also supports RCS 1.0 where
+ * these permissions are sent to the remote API keys {@link #toMap()}. In both cases the outbound transport version will be used to
+ * remove permissions that are not available to elder remote clusters.
  * For example, on the local/querying cluster this model represents the following:
  * <code>
  * "remote_cluster" : [
@@ -52,15 +53,18 @@
  *         }
  *     ]
  * </code>
- * when sent to the remote cluster "clusterA", the privileges will be converted to the appropriate cluster privileges. For example:
+ * (RCS 2.0) when sent to the remote cluster "clusterA", the privileges will be converted to the appropriate cluster privileges.
+ * For example:
  * <code>
  *   "cluster": ["foo"]
  * </code>
- * and when sent to the remote cluster "clusterB", the privileges will be converted to the appropriate cluster privileges. For example:
+ * and (RCS 2.0) when sent to the remote cluster "clusterB", the privileges will be converted to the appropriate cluster privileges.
+ * For example:
  * <code>
  *   "cluster": ["bar"]
  * </code>
- * If the remote cluster does not support the privilege, as determined by the remote cluster version, the privilege will be not be sent.
+ * (RCS 1.0 and RCS 2.0) If the remote cluster does not support the privilege, as determined by the remote cluster version,
+ * the privilege will be not be sent. Upstream code performs the removal, but this class owns the business logic for how to remove.
  */
 public class RemoteClusterPermissions implements NamedWriteable, ToXContentObject {
 
@@ -88,7 +92,7 @@ public RemoteClusterPermissions(StreamInput in) throws IOException {
         remoteClusterPermissionGroups = in.readNamedWriteableCollectionAsList(RemoteClusterPermissionGroup.class);
     }
 
-    public RemoteClusterPermissions(List<Map<String, List<String>>> remoteClusters){
+    public RemoteClusterPermissions(List<Map<String, List<String>>> remoteClusters) {
         remoteClusterPermissionGroups = new ArrayList<>();
         for (Map<String, List<String>> remoteCluster : remoteClusters) {
             RemoteClusterPermissionGroup remoteClusterPermissionGroup = new RemoteClusterPermissionGroup(remoteCluster);
@@ -109,27 +113,69 @@ public RemoteClusterPermissions addGroup(RemoteClusterPermissionGroup remoteClus
         return this;
     }
 
+    /**
+     * Will remove any unsupported privileges for the provided outbound version. This method will not modify the current instance.
+     * This is useful for RCS 1.0 and API keys to help ensure that we don't send unsupported privileges to the remote cluster.
+     * @param outboundVersion The version by which to remove unsupported privileges, this is typically the version of the remote cluster
+     * @return a new instance of RemoteClusterPermissions with the unsupported privileges removed
+     */
     public RemoteClusterPermissions removeUnsupportedPrivileges(TransportVersion outboundVersion) {
-        System.out.println("************ [DEBUG] RemoteClusterPermissions.removeUnsupportedPrivileges");
+
         RemoteClusterPermissions copyForOutBoundVersion = new RemoteClusterPermissions();
+
+        // TODO: centralize this and put in a poor mans cache
+        // find all the privileges that are allowed for the remote cluster version
+        Set<String> allowedPermissionsPerVersion = allowedRemoteClusterPermissions.entrySet()
+            .stream()
+            .filter((entry) -> entry.getKey().onOrBefore(outboundVersion))
+            .map(Map.Entry::getValue)
+            .flatMap(Set::stream)
+            .map(s -> s.toLowerCase(Locale.ROOT))
+            .collect(Collectors.toSet());
+
         for (RemoteClusterPermissionGroup group : remoteClusterPermissionGroups) {
-            //TODO: actually implement this method, this is a hack to make the test pass for now
-            if(outboundVersion.onOrBefore(ROLE_MONITOR_STATS)){
-                Set<String> a = new HashSet<>(Arrays.asList(group.clusterPrivileges()));
-                a.remove("monitor_stats");
-                copyForOutBoundVersion.addGroup(
-                    new RemoteClusterPermissionGroup(a.toArray(new String[0]), group.remoteClusterAliases())
+            String[] privileges = group.clusterPrivileges();
+            List<String> privilegesMutated = new ArrayList<>(privileges.length);
+            for (String privilege : privileges) {
+                if (allowedPermissionsPerVersion.contains(privilege.toLowerCase(Locale.ROOT))) {
+                    privilegesMutated.add(privilege);
+                }
+            }
+            if (privilegesMutated.isEmpty() == false) {
+                RemoteClusterPermissionGroup mutatedGroup = new RemoteClusterPermissionGroup(
+                    privilegesMutated.toArray(new String[0]),
+                    group.remoteClusterAliases()
                 );
-            }else{
-                copyForOutBoundVersion.addGroup(group);
+                copyForOutBoundVersion.addGroup(mutatedGroup);
+                if (logger.isDebugEnabled()) {
+                    if (group.equals(mutatedGroup) == false) {
+                        logger.debug(
+                            "Removed unsupported remote cluster permissions {} for remote cluster [{}]. "
+                                + "Due to the remote cluster version, only the following permissions are allowed: {}",
+                            privilegesMutated,
+                            group.remoteClusterAliases(),
+                            allowedPermissionsPerVersion
+                        );
+                    }
+                }
+            } else {
+                if (logger.isDebugEnabled()) {
+                    logger.debug(
+                        "Removed all remote cluster permissions for remote cluster [{}]. "
+                            + "Due to the remote cluster version, only the following permissions are allowed: {}",
+                        group.remoteClusterAliases(),
+                        allowedPermissionsPerVersion
+                    );
+                }
             }
         }
         return copyForOutBoundVersion;
     }
 
     /**
-     * Gets the privilege names for the remote cluster. This method will collapse all groups to single String[] all lowercase
-     * and will only return the appropriate privileges for the provided remote cluster version.
+     * Gets all the privilege names for the remote cluster. This method will collapse all groups to single String[] all lowercase
+     * and will only return the appropriate privileges for the provided remote cluster version. This is useful for RCS 2.0 to ensure
+     * that we properly convert all the remote_cluster -> cluster privileges per remote cluster.
      */
     public String[] privilegeNames(final String remoteClusterAlias, TransportVersion remoteClusterVersion) {
 
@@ -168,6 +214,10 @@ public String[] privilegeNames(final String remoteClusterAlias, TransportVersion
         return allowedPrivileges.stream().sorted().toArray(String[]::new);
     }
 
+    /**
+     * Converts this object to it's {@link Map} representation.
+     * @return a list of maps representing the remote cluster permissions
+     */
     public List<Map<String, List<String>>> toMap() {
         return remoteClusterPermissionGroups.stream().map(RemoteClusterPermissionGroup::toMap).toList();
     }
@@ -256,5 +306,4 @@ public String getWriteableName() {
         return NAME;
     }
 
-
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/store/RoleReferenceIntersection.java

@@ -36,7 +36,6 @@ public List<RoleReference> getRoleReferences() {
         return roleReferences;
     }
 
-
     public void buildRole(BiConsumer<RoleReference, ActionListener<Role>> singleRoleBuilder, ActionListener<Role> roleActionListener) {
         final GroupedActionListener<Role> roleGroupedActionListener = new GroupedActionListener<>(
             roleReferences.size(),