Fix owner user realm check for API key authentication (elastic#84325)

API Key can run-as since elastic#79809. There are places in the code where we
assume API key cannot run-as. Most of them are corrected in elastic#81564.
But there are still a few things got missed. This PR fixes the methods
for checking owner user realm for API key.

This means, when API Keys "running-as" (impersonating other users),
we do not expose the authenticating key ID and name to the end-user
such as the Authenticate API and the SetSecurityUseringest processor.
Only the effective user is revealed, just like in the regular case of a realm
user run as. For audit logging, the key's ID and name are not exposed
either. But this is mainly because there are no existing fields suitable for
these information. We do intend to add them later (elastic#84394) because
auditing logging is to consumed by system admin instead of end-users.

Note the resource sharing check (canAccessResourcesOf) also needs to be
fixed, this will be handled by elastic#84277

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

@@ -253,7 +253,7 @@ public void toXContentFragment(XContentBuilder builder) throws IOException {
         builder.array(User.Fields.ROLES.getPreferredName(), user.roles());
         builder.field(User.Fields.FULL_NAME.getPreferredName(), user.fullName());
         builder.field(User.Fields.EMAIL.getPreferredName(), user.email());
-        if (isAuthenticatedWithServiceAccount()) {
+        if (isServiceAccount()) {
             final String tokenName = (String) getMetadata().get(ServiceAccountSettings.TOKEN_NAME_FIELD);
             assert tokenName != null : "token name cannot be null";
             final String tokenSource = (String) getMetadata().get(ServiceAccountSettings.TOKEN_SOURCE_FIELD);
@@ -279,7 +279,7 @@ public void toXContentFragment(XContentBuilder builder) throws IOException {
         }
         builder.endObject();
         builder.field(User.Fields.AUTHENTICATION_TYPE.getPreferredName(), getAuthenticationType().name().toLowerCase(Locale.ROOT));
-        if (isAuthenticatedWithApiKey()) {
+        if (isApiKey()) {
             this.assertApiKeyMetadata();
             final String apiKeyId = (String) this.metadata.get(AuthenticationField.API_KEY_ID_KEY);
             final String apiKeyName = (String) this.metadata.get(AuthenticationField.API_KEY_NAME_KEY);

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

@@ -8,9 +8,16 @@
 package org.elasticsearch.xpack.core.security.authc;
 
 import org.elasticsearch.Version;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.VersionUtils;
+import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.core.security.action.service.TokenInfo;
 import org.elasticsearch.xpack.core.security.authc.Authentication.AuthenticationType;
 import org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef;
@@ -24,12 +31,15 @@
 import org.elasticsearch.xpack.core.security.user.XPackSecurityUser;
 import org.elasticsearch.xpack.core.security.user.XPackUser;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ANONYMOUS_REALM_NAME;
@@ -38,7 +48,10 @@
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ATTACH_REALM_TYPE;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.FALLBACK_REALM_NAME;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.FALLBACK_REALM_TYPE;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
 
 public class AuthenticationTests extends ESTestCase {
 
@@ -168,6 +181,41 @@ public void testIsServiceAccount() {
         }
     }
 
+    public void testToXContentWithApiKey() throws IOException {
+        final String apiKeyId = randomAlphaOfLength(20);
+        final Authentication authentication1 = randomApiKeyAuthentication(randomUser(), apiKeyId);
+        final String apiKeyName = (String) authentication1.getMetadata().get(AuthenticationField.API_KEY_NAME_KEY);
+        runWithAuthenticationToXContent(
+            authentication1,
+            m -> assertThat(
+                m,
+                hasEntry("api_key", apiKeyName != null ? Map.of("id", apiKeyId, "name", apiKeyName) : Map.of("id", apiKeyId))
+            )
+        );
+
+        final Authentication authentication2 = toRunAs(authentication1, randomUser(), randomRealm());
+        runWithAuthenticationToXContent(authentication2, m -> assertThat(m, not(hasKey("api_key"))));
+    }
+
+    public void testToXContentWithServiceAccount() throws IOException {
+        final Authentication authentication1 = randomServiceAccountAuthentication();
+        final String tokenName = (String) authentication1.getMetadata().get(ServiceAccountSettings.TOKEN_NAME_FIELD);
+        final String tokenType = ServiceAccountSettings.REALM_TYPE
+            + "_"
+            + authentication1.getMetadata().get(ServiceAccountSettings.TOKEN_SOURCE_FIELD);
+        runWithAuthenticationToXContent(
+            authentication1,
+            m -> assertThat(m, hasEntry("token", Map.of("name", tokenName, "type", tokenType)))
+        );
+    }
+
+    private void runWithAuthenticationToXContent(Authentication authentication, Consumer<Map<String, Object>> consumer) throws IOException {
+        try (XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
+            authentication.toXContent(builder, ToXContent.EMPTY_PARAMS);
+            consumer.accept(XContentHelper.convertToMap(BytesReference.bytes(builder), false, XContentType.JSON).v2());
+        }
+    }
+
     private void checkCanAccessResources(Authentication authentication0, Authentication authentication1) {
         if (authentication0.getAuthenticationType() == authentication1.getAuthenticationType()
             || EnumSet.of(AuthenticationType.REALM, AuthenticationType.TOKEN)
@@ -243,6 +291,11 @@ public static Authentication randomApiKeyAuthentication(User user, String apiKey
         final HashMap<String, Object> metadata = new HashMap<>();
         metadata.put(AuthenticationField.API_KEY_ID_KEY, apiKeyId);
         metadata.put(AuthenticationField.API_KEY_NAME_KEY, randomBoolean() ? null : randomAlphaOfLengthBetween(1, 16));
+        metadata.put(AuthenticationField.API_KEY_CREATOR_REALM_NAME, AuthenticationField.API_KEY_CREATOR_REALM_NAME);
+        metadata.put(AuthenticationField.API_KEY_CREATOR_REALM_TYPE, AuthenticationField.API_KEY_CREATOR_REALM_TYPE);
+        metadata.put(AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY, new BytesArray("{}"));
+        metadata.put(AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, new BytesArray("""
+            {"x":{"cluster":["all"],"indices":[{"names":["index*"],"privileges":["all"]}]}}"""));
         return new Authentication(
             user,
             apiKeyRealm,
@@ -304,6 +357,22 @@ public static Authentication toToken(Authentication authentication) {
         return newTokenAuthentication;
     }
 
+    public static Authentication toRunAs(Authentication authentication, User runAs, @Nullable RealmRef lookupRealmRef) {
+        Objects.requireNonNull(runAs);
+        assert false == runAs.isRunAs();
+        assert false == authentication.getUser().isRunAs();
+        assert AuthenticationType.REALM == authentication.getAuthenticationType()
+            || AuthenticationType.API_KEY == authentication.getAuthenticationType();
+        return new Authentication(
+            new User(runAs, authentication.getUser()),
+            authentication.getAuthenticatedBy(),
+            lookupRealmRef,
+            authentication.getVersion(),
+            authentication.getAuthenticationType(),
+            authentication.getMetadata()
+        );
+    }
+
     private boolean realmIsSingleton(RealmRef realmRef) {
         return Set.of(FileRealmSettings.TYPE, NativeRealmSettings.TYPE).contains(realmRef.getType());
     }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrail.java

@@ -1470,13 +1470,13 @@ private void setThreadContextField(ThreadContext threadContext, String threadCon
         LogEntryBuilder withAuthentication(Authentication authentication) {
             logEntry.with(PRINCIPAL_FIELD_NAME, authentication.getUser().principal());
             logEntry.with(AUTHENTICATION_TYPE_FIELD_NAME, authentication.getAuthenticationType().toString());
-            if (authentication.isAuthenticatedWithApiKey()) {
+            if (authentication.isApiKey()) {
                 logEntry.with(API_KEY_ID_FIELD_NAME, (String) authentication.getMetadata().get(AuthenticationField.API_KEY_ID_KEY));
                 String apiKeyName = (String) authentication.getMetadata().get(AuthenticationField.API_KEY_NAME_KEY);
                 if (apiKeyName != null) {
                     logEntry.with(API_KEY_NAME_FIELD_NAME, apiKeyName);
                 }
-                String creatorRealmName = (String) authentication.getMetadata().get(AuthenticationField.API_KEY_CREATOR_REALM_NAME);
+                final String creatorRealmName = ApiKeyService.getCreatorRealmName(authentication);
                 if (creatorRealmName != null) {
                     // can be null for API keys created before version 7.7
                     logEntry.with(PRINCIPAL_REALM_FIELD_NAME, creatorRealmName);
@@ -1485,11 +1485,15 @@ LogEntryBuilder withAuthentication(Authentication authentication) {
                 if (authentication.getUser().isRunAs()) {
                     logEntry.with(PRINCIPAL_REALM_FIELD_NAME, authentication.getLookedUpBy().getName())
                         .with(PRINCIPAL_RUN_BY_FIELD_NAME, authentication.getUser().authenticatedUser().principal())
+                        // API key can run-as, when that happens, the following field will be _es_api_key,
+                        // not the API key owner user's realm.
                         .with(PRINCIPAL_RUN_BY_REALM_FIELD_NAME, authentication.getAuthenticatedBy().getName());
+                    // TODO: API key can run-as which means we could use extra fields (#84394)
                 } else {
                     logEntry.with(PRINCIPAL_REALM_FIELD_NAME, authentication.getAuthenticatedBy().getName());
                 }
             }
+            // TODO: service token info is logged in a separate authentication field (#84394)
             if (authentication.isAuthenticatedWithServiceAccount()) {
                 logEntry.with(SERVICE_TOKEN_NAME_FIELD_NAME, (String) authentication.getMetadata().get(TOKEN_NAME_FIELD))
                     .with(

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

@@ -1335,31 +1335,29 @@ AtomicLong getLastEvictionCheckedAt() {
     }
 
     /**
-     * Returns realm name for the authenticated user.
-     * If the user is authenticated by realm type {@value AuthenticationField#API_KEY_REALM_TYPE}
-     * then it will return the realm name of user who created this API key.
+     * Returns realm name of the owner user of an API key if the effective user is an API Key.
+     * If the effective user is not an API key, it just returns the source realm name.
      *
      * @param authentication {@link Authentication}
      * @return realm name
      */
     public static String getCreatorRealmName(final Authentication authentication) {
-        if (authentication.isAuthenticatedWithApiKey()) {
+        if (authentication.isApiKey()) {
             return (String) authentication.getMetadata().get(AuthenticationField.API_KEY_CREATOR_REALM_NAME);
         } else {
             return authentication.getSourceRealm().getName();
         }
     }
 
     /**
-     * Returns realm type for the authenticated user.
-     * If the user is authenticated by realm type {@value AuthenticationField#API_KEY_REALM_TYPE}
-     * then it will return the realm name of user who created this API key.
+     * Returns realm type of the owner user of an API key if the effective user is an API Key.
+     * If the effective user is not an API key, it just returns the source realm type.
      *
      * @param authentication {@link Authentication}
      * @return realm type
      */
     public static String getCreatorRealmType(final Authentication authentication) {
-        if (authentication.isAuthenticatedWithApiKey()) {
+        if (authentication.isApiKey()) {
             return (String) authentication.getMetadata().get(AuthenticationField.API_KEY_CREATOR_REALM_TYPE);
         } else {
             return authentication.getSourceRealm().getType();

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

@@ -140,7 +140,7 @@ public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
                     }
                     break;
                 case API_KEY:
-                    if (authentication.isAuthenticatedWithApiKey()) {
+                    if (authentication.isApiKey()) {
                         final String apiKey = "api_key";
                         final Object existingApiKeyField = userObject.get(apiKey);
                         @SuppressWarnings("unchecked")