Expose more authentication info to ingest pipeline (elastic#51305)

The changes add more granularity for identiying the data ingestion user.
The ingest pipeline can now be configure to record authentication realm and
type. It can also record API key name and ID when one is in use. 
This improves traceability when data are being ingested from multiple agents
and will become more relevant with the incoming support of required
pipelines (elastic#46847)

Resolves: elastic#49106

docs/reference/ingest/processors/set-security-user.asciidoc

@@ -1,18 +1,25 @@
 [[ingest-node-set-security-user-processor]]
 === Set Security User Processor
-Sets user-related details (such as `username`,  `roles`, `email`, `full_name`
-and `metadata` ) from the current
+Sets user-related details (such as `username`,  `roles`, `email`, `full_name`,
+`metadata`, `api_key`, `realm` and `authentication_type`) from the current
 authenticated user to the current document by pre-processing the ingest.
+The `api_key` property exists only if the user authenticates with an
+API key. It is an object containing the `id` and `name` fields of the API key.
+The `realm` property is also an object with two fields, `name` and `type`.
+When using API key authentication, the `realm` property refers to the realm
+from which the API key is created.
+The `authentication_type` property is a string that can take value from
+`REALM`, `API_KEY`, `TOKEN` and `ANONYMOUS`.
 
 IMPORTANT: Requires an authenticated user for the index request.
 
 [[set-security-user-options]]
 .Set Security User Options
 [options="header"]
 |======
-| Name          | Required  | Default                                                   | Description
-| `field`       | yes       | -                                                         | The field to store the user information into.
-| `properties`  | no        | [`username`, `roles`, `email`, `full_name`, `metadata`]   | Controls what user related properties are added to the `field`.
+| Name          | Required  | Default                                                                                              | Description
+| `field`       | yes       | -                                                                                                    | The field to store the user information into.
+| `properties`  | no        | [`username`, `roles`, `email`, `full_name`, `metadata`, `api_key`, `realm`, `authentication_type`]   | Controls what user related properties are added to the `field`.
 include::common-options.asciidoc[]
 |======
 

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

@@ -81,6 +81,10 @@ public RealmRef getLookedUpBy() {
         return lookedUpBy;
     }
 
+    public RealmRef getSourceRealm() {
+        return lookedUpBy == null ? authenticatedBy : lookedUpBy;
+    }
+
     public Version getVersion() {
         return version;
     }

x-pack/plugin/core/src/main/resources/security-index-template-7.json

@@ -194,6 +194,9 @@
             },
             "realm" : {
               "type" : "keyword"
+            },
+            "realm_type" : {
+              "type" : "keyword"
             }
           }
         },

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

@@ -0,0 +1,33 @@
+/*
+ *
+ *  * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ *  * or more contributor license agreements. Licensed under the Elastic License;
+ *  * you may not use this file except in compliance with the Elastic License.
+ *
+ */
+
+package org.elasticsearch.xpack.core.security.authc;
+
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.security.user.User;
+
+public class AuthenticationTests extends ESTestCase {
+
+    public void testWillGetLookedUpByWhenItExists() {
+        final Authentication.RealmRef authenticatedBy = new Authentication.RealmRef("auth_by", "auth_by_type", "node");
+        final Authentication.RealmRef lookedUpBy = new Authentication.RealmRef("lookup_by", "lookup_by_type", "node");
+        final Authentication authentication = new Authentication(
+            new User("user"), authenticatedBy, lookedUpBy);
+
+        assertEquals(lookedUpBy, authentication.getSourceRealm());
+    }
+
+    public void testWillGetAuthenticateByWhenLookupIsNull() {
+        final Authentication.RealmRef authenticatedBy = new Authentication.RealmRef("auth_by", "auth_by_type", "node");
+        final Authentication authentication = new Authentication(
+            new User("user"), authenticatedBy, null);
+
+        assertEquals(authenticatedBy, authentication.getSourceRealm());
+    }
+
+}

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

@@ -108,9 +108,11 @@ public class ApiKeyService {
     private static final Logger logger = LogManager.getLogger(ApiKeyService.class);
     private static final DeprecationLogger deprecationLogger = new DeprecationLogger(logger);
     public static final String API_KEY_ID_KEY = "_security_api_key_id";
+    public static final String API_KEY_NAME_KEY = "_security_api_key_name";
     public static final String API_KEY_REALM_NAME = "_es_api_key";
     public static final String API_KEY_REALM_TYPE = "_es_api_key";
-    public static final String API_KEY_CREATOR_REALM = "_security_api_key_creator_realm";
+    public static final String API_KEY_CREATOR_REALM_NAME = "_security_api_key_creator_realm_name";
+    public static final String API_KEY_CREATOR_REALM_TYPE = "_security_api_key_creator_realm_type";
     static final String API_KEY_ROLE_DESCRIPTORS_KEY = "_security_api_key_role_descriptors";
     static final String API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY = "_security_api_key_limited_by_role_descriptors";
 
@@ -279,8 +281,8 @@ XContentBuilder newDocument(SecureString apiKey, String name, Authentication aut
             .startObject("creator")
             .field("principal", authentication.getUser().principal())
             .field("metadata", authentication.getUser().metadata())
-            .field("realm", authentication.getLookedUpBy() == null ?
-                authentication.getAuthenticatedBy().getName() : authentication.getLookedUpBy().getName())
+            .field("realm", authentication.getSourceRealm().getName())
+            .field("realm_type", authentication.getSourceRealm().getType())
             .endObject()
             .endObject();
 
@@ -509,10 +511,12 @@ private void validateApiKeyExpiration(Map<String, Object> source, ApiKeyCredenti
                 : limitedByRoleDescriptors.keySet().toArray(Strings.EMPTY_ARRAY);
             final User apiKeyUser = new User(principal, roleNames, null, null, metadata, true);
             final Map<String, Object> authResultMetadata = new HashMap<>();
-            authResultMetadata.put(API_KEY_CREATOR_REALM, creator.get("realm"));
+            authResultMetadata.put(API_KEY_CREATOR_REALM_NAME, creator.get("realm"));
+            authResultMetadata.put(API_KEY_CREATOR_REALM_TYPE, creator.get("realm_type"));
             authResultMetadata.put(API_KEY_ROLE_DESCRIPTORS_KEY, roleDescriptors);
             authResultMetadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, limitedByRoleDescriptors);
             authResultMetadata.put(API_KEY_ID_KEY, credentials.getId());
+            authResultMetadata.put(API_KEY_NAME_KEY, source.get("name"));
             listener.onResponse(AuthenticationResult.success(apiKeyUser, authResultMetadata));
         } else {
             listener.onResponse(AuthenticationResult.unsuccessful("api key is expired", null));
@@ -886,7 +890,7 @@ public void getApiKeys(String realmName, String username, String apiKeyName, Str
      */
     public static String getCreatorRealmName(final Authentication authentication) {
         if (authentication.getAuthenticatedBy().getType().equals(API_KEY_REALM_TYPE)) {
-            return (String) authentication.getMetadata().get(API_KEY_CREATOR_REALM);
+            return (String) authentication.getMetadata().get(API_KEY_CREATOR_REALM_NAME);
         } else {
             return authentication.getAuthenticatedBy().getName();
         }

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

@@ -11,6 +11,7 @@
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.user.User;
+import org.elasticsearch.xpack.security.authc.ApiKeyService;
 
 import java.util.Arrays;
 import java.util.EnumSet;
@@ -85,6 +86,54 @@ public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
                         userObject.put("metadata", user.metadata());
                     }
                     break;
+                case API_KEY:
+                    final String apiKey = "api_key";
+                    final Object existingApiKeyField = userObject.get(apiKey);
+                    @SuppressWarnings("unchecked")
+                    final Map<String, Object> apiKeyField =
+                        existingApiKeyField instanceof Map ? (Map<String, Object>) existingApiKeyField : new HashMap<>();
+                    Object apiKeyName = authentication.getMetadata().get(ApiKeyService.API_KEY_NAME_KEY);
+                    if (apiKeyName != null) {
+                        apiKeyField.put("name", apiKeyName);
+                    }
+                    Object apiKeyId = authentication.getMetadata().get(ApiKeyService.API_KEY_ID_KEY);
+                    if (apiKeyId != null) {
+                        apiKeyField.put("id", apiKeyId);
+                    }
+                    if (false == apiKeyField.isEmpty()) {
+                        userObject.put(apiKey, apiKeyField);
+                    }
+                    break;
+                case REALM:
+                    final String realmKey = "realm";
+                    final Object existingRealmField = userObject.get(realmKey);
+                    @SuppressWarnings("unchecked")
+                    final Map<String, Object> realmField =
+                        existingRealmField instanceof Map ? (Map<String, Object>) existingRealmField : new HashMap<>();
+
+                    final Object realmName, realmType;
+                    if (Authentication.AuthenticationType.API_KEY == authentication.getAuthenticationType()) {
+                        realmName = authentication.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_NAME);
+                        realmType = authentication.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_TYPE);
+                    } else {
+                        realmName = authentication.getSourceRealm().getName();
+                        realmType = authentication.getSourceRealm().getType();
+                    }
+                    if (realmName != null) {
+                        realmField.put("name", realmName);
+                    }
+                    if (realmType != null) {
+                        realmField.put("type", realmType);
+                    }
+                    if (false == realmField.isEmpty()) {
+                        userObject.put(realmKey, realmField);
+                    }
+                    break;
+                case AUTHENTICATION_TYPE:
+                    if (authentication.getAuthenticationType() != null) {
+                        userObject.put("authentication_type", authentication.getAuthenticationType().toString());
+                    }
+                    break;
                 default:
                     throw new UnsupportedOperationException("unsupported property [" + property + "]");
             }
@@ -138,7 +187,10 @@ public enum Property {
         FULL_NAME,
         EMAIL,
         ROLES,
-        METADATA;
+        METADATA,
+        API_KEY,
+        REALM,
+        AUTHENTICATION_TYPE;
 
         static Property parse(String tag, String value) {
             try {

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

@@ -148,6 +148,10 @@ public void testAuthenticateWithApiKey() throws Exception {
         assertThat(auth.getStatus(), is(AuthenticationResult.Status.SUCCESS));
         assertThat(auth.getUser(), notNullValue());
         assertThat(auth.getUser().principal(), is("hulk"));
+        assertThat(auth.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_NAME), is("realm1"));
+        assertThat(auth.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_TYPE), is("native"));
+        assertThat(auth.getMetadata().get(ApiKeyService.API_KEY_ID_KEY), is(id));
+        assertThat(auth.getMetadata().get(ApiKeyService.API_KEY_NAME_KEY), is("test"));
     }
 
     public void testAuthenticationIsSkippedIfLicenseDoesNotAllowIt() throws Exception {
@@ -284,6 +288,7 @@ public void testValidateApiKey() throws Exception {
         Map<String, Object> creatorMap = new HashMap<>();
         creatorMap.put("principal", "test_user");
         creatorMap.put("realm", "realm1");
+        creatorMap.put("realm_type", "realm_type1");
         creatorMap.put("metadata", Collections.emptyMap());
         sourceMap.put("creator", creatorMap);
         sourceMap.put("api_key_invalidated", false);
@@ -302,7 +307,7 @@ public void testValidateApiKey() throws Exception {
         assertThat(result.getMetadata().get(ApiKeyService.API_KEY_ROLE_DESCRIPTORS_KEY), equalTo(sourceMap.get("role_descriptors")));
         assertThat(result.getMetadata().get(ApiKeyService.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY),
             equalTo(sourceMap.get("limited_by_role_descriptors")));
-        assertThat(result.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM), is("realm1"));
+        assertThat(result.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_NAME), is("realm1"));
 
         sourceMap.put("expiration_time", Clock.systemUTC().instant().plus(1L, ChronoUnit.HOURS).toEpochMilli());
         future = new PlainActionFuture<>();
@@ -316,7 +321,7 @@ public void testValidateApiKey() throws Exception {
         assertThat(result.getMetadata().get(ApiKeyService.API_KEY_ROLE_DESCRIPTORS_KEY), equalTo(sourceMap.get("role_descriptors")));
         assertThat(result.getMetadata().get(ApiKeyService.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY),
             equalTo(sourceMap.get("limited_by_role_descriptors")));
-        assertThat(result.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM), is("realm1"));
+        assertThat(result.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_NAME), is("realm1"));
 
         sourceMap.put("expiration_time", Clock.systemUTC().instant().minus(1L, ChronoUnit.HOURS).toEpochMilli());
         future = new PlainActionFuture<>();
@@ -561,6 +566,14 @@ public void testApiKeyCacheDisabled() {
         assertNull(cachedApiKeyHashResult);
     }
 
+    public void testWillAlwaysGetAuthenticationRealmName() {
+        final Authentication.RealmRef authenticatedBy = new Authentication.RealmRef("auth_by", "auth_by_type", "node");
+        final Authentication.RealmRef lookedUpBy = new Authentication.RealmRef("lookup_by", "lookup_by_type", "node");
+        final Authentication authentication = new Authentication(
+            new User("user"), authenticatedBy, lookedUpBy);
+        assertEquals("auth_by", ApiKeyService.getCreatorRealmName(authentication));
+    }
+
     private ApiKeyService createApiKeyService(Settings baseSettings) {
         final Settings settings = Settings.builder()
             .put(XPackSettings.API_KEY_SERVICE_ENABLED_SETTING.getKey(), true)