Handle cross-cluster authentication in authorization info report (#95203)

This PR adds support for reporting authorization information for
cross-cluster authentication.

Note: Labelled as `>non-issue` because the cross-cluster-access is still
behind feature flag.

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/xcontent/XContentUtils.java

@@ -11,6 +11,7 @@ import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationField;
 import org.elasticsearch.xpack.core.security.authc.Subject;
 import org.elasticsearch.xpack.core.security.authc.support.AuthenticationContextSerializer;
@@ -21,6 +22,8 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.CROSS_CLUSTER_ACCESS_AUTHENTICATION_KEY;
+
 public class XContentUtils {
 
     private XContentUtils() {}
@@ -102,26 +105,40 @@ public class XContentUtils {
             return;
         }
         builder.startObject("authorization");
-        switch (authenticationSubject.getType()) {
-            case USER -> builder.array(User.Fields.ROLES.getPreferredName(), authenticationSubject.getUser().roles());
+        addSubjectInfo(builder, authenticationSubject);
+        builder.endObject();
+    }
+
+    private static void addSubjectInfo(XContentBuilder builder, Subject subject) throws IOException {
+        switch (subject.getType()) {
+            case USER -> builder.array(User.Fields.ROLES.getPreferredName(), subject.getUser().roles());
             case API_KEY -> {
-                builder.startObject("api_key");
-                Map<String, Object> metadata = authenticationSubject.getMetadata();
-                builder.field("id", metadata.get(AuthenticationField.API_KEY_ID_KEY));
-                Object name = metadata.get(AuthenticationField.API_KEY_NAME_KEY);
-                if (name instanceof String) {
-                    builder.field("name", name);
-                }
-                builder.endObject();
+                addApiKeyInfo(builder, subject);
             }
-            case SERVICE_ACCOUNT -> builder.field("service_account", authenticationSubject.getUser().principal());
+            case SERVICE_ACCOUNT -> builder.field("service_account", subject.getUser().principal());
             case CROSS_CLUSTER_ACCESS -> {
-                // TODO handle cross cluster access authentication
-                final String message = "cross cluster access authentication is not yet supported";
-                assert false : message;
-                throw new UnsupportedOperationException(message);
+                builder.startObject("cross_cluster_access");
+                {
+                    addApiKeyInfo(builder, subject);
+                    builder.startObject("remote_authorization");
+                    final var innerAuthentication = (Authentication) subject.getMetadata().get(CROSS_CLUSTER_ACCESS_AUTHENTICATION_KEY);
+                    assert innerAuthentication != null && false == innerAuthentication.isCrossClusterAccess();
+                    addSubjectInfo(builder, innerAuthentication.getEffectiveSubject());
+                    builder.endObject();
+                }
+                builder.endObject();
             }
         }
+    }
+
+    private static void addApiKeyInfo(XContentBuilder builder, Subject authenticationSubject) throws IOException {
+        builder.startObject("api_key");
+        Map<String, Object> metadata = authenticationSubject.getMetadata();
+        builder.field("id", metadata.get(AuthenticationField.API_KEY_ID_KEY));
+        Object name = metadata.get(AuthenticationField.API_KEY_NAME_KEY);
+        if (name instanceof String) {
+            builder.field("name", name);
+        }
         builder.endObject();
     }
 

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/xcontent/XContentUtilsTests.java

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.core.security.xcontent;
 
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.json.JsonXContent;
@@ -22,6 +23,9 @@ import java.util.Arrays;
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ID_KEY;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_NAME_KEY;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.CROSS_CLUSTER_ACCESS_AUTHENTICATION_KEY;
 import static org.hamcrest.Matchers.equalTo;
 
 public class XContentUtilsTests extends ESTestCase {
@@ -52,7 +56,7 @@ public class XContentUtilsTests extends ESTestCase {
         String apiKeyName = randomAlphaOfLengthBetween(1, 16);
         AuthenticationTestBuilder builder = AuthenticationTestHelper.builder()
             .apiKey(apiKeyId)
-            .metadata(Map.of(AuthenticationField.API_KEY_NAME_KEY, apiKeyName));
+            .metadata(Map.of(API_KEY_NAME_KEY, apiKeyName));
         Authentication authentication = builder.build();
         String json = generateJson(Map.of(AuthenticationField.AUTHENTICATION_KEY, authentication.encode()));
         assertThat(json, equalTo("{\"authorization\":{\"api_key\":{\"id\":\"" + apiKeyId + "\",\"name\":\"" + apiKeyName + "\"}}}"));
@@ -67,6 +71,45 @@ public class XContentUtilsTests extends ESTestCase {
         assertThat(json, equalTo("{\"authorization\":{\"service_account\":\"" + account + "\"}}"));
     }
 
+    public void testAddAuthorizationInfoWithCrossClusterAccess() throws IOException {
+        final Authentication authentication = AuthenticationTestHelper.builder().crossClusterAccess().build();
+        final var apiKeyName = (String) authentication.getAuthenticatingSubject().getMetadata().get(API_KEY_NAME_KEY);
+        final var innerAuthentication = (Authentication) authentication.getAuthenticatingSubject()
+            .getMetadata()
+            .get(CROSS_CLUSTER_ACCESS_AUTHENTICATION_KEY);
+
+        // Rely on the target function itself to generate the json string for inner authentication.
+        // This is OK because other subject variants are tested elsewhere. We are only interested in the cross cluster variant here.
+        String innerAuthenticationString = generateJson(Map.of(AuthenticationField.AUTHENTICATION_KEY, innerAuthentication.encode()));
+        innerAuthenticationString = innerAuthenticationString.replace("{\"authorization\":", "");
+        innerAuthenticationString = innerAuthenticationString.substring(0, innerAuthenticationString.length() - 1);
+
+        String json = generateJson(Map.of(AuthenticationField.AUTHENTICATION_KEY, authentication.encode()));
+        assertThat(
+            json,
+            equalTo(
+                XContentHelper.stripWhitespace(
+                    Strings.format(
+                        """
+                            {
+                              "authorization": {
+                                "cross_cluster_access": {
+                                  "api_key": {
+                                    "id": "%s"%s
+                                  },
+                                  "remote_authorization": %s
+                                }
+                              }
+                            }""",
+                        authentication.getAuthenticatingSubject().getMetadata().get(API_KEY_ID_KEY),
+                        apiKeyName == null ? "" : ",\"name\":\"" + apiKeyName + "\"",
+                        innerAuthenticationString
+                    )
+                )
+            )
+        );
+    }
+
     public void testAddAuthorizationInfoWithCorruptData() throws IOException {
         String json = generateJson(Map.of(AuthenticationField.AUTHENTICATION_KEY, "corrupt"));
         assertThat(json, equalTo("{}"));