[UIAM] Cloud API key authentication metadata and validations (elastic#129227)

A followup to elastic#128440, which introduces a new `managed_by` field (`<1>`) that will be returned in the response of the Authenticate API.


Besides `managed_by` field, it also captures additional `internal` field (`<2>`) for cloud API key authentication and exposes it as part of the `api_key` fields.

```json
{
  "username": "omSAd5YBK3gZiBcD-GvX", 
  "roles": [ "viewer" ],
  "metadata": {
    ...
  },
  "enabled": true,
  "authentication_realm": {
    "name": "_cloud_api_key",
    "type": "_cloud_api_key"
  },
  "lookup_realm": {
    "name": "_cloud_api_key",
    "type": "_cloud_api_key"
  },
  "authentication_type": "api_key",
  "api_key": { 
    "id": "omSAd5YBK3gZiBcD-GvX",
    "name": "my cloud API key",
    "managed_by": "cloud", <1>
    "internal": false <2>
  }
}

```


- Additionally it implements the `Authentication#canAccessResourcesOf` for the cloud API keys. Ownership check allows access only to the same cloud API key.

- And lastly, adds a consistency check for cloud API keys in `Authentication#checkConsistencyForApiKeyAuthenticationType`.

Author: slobodanadamovic

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

@@ -68,8 +68,10 @@
 import static org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef.newServiceAccountRealmRef;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ANONYMOUS_REALM_NAME;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ANONYMOUS_REALM_TYPE;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_REALM_NAME;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_REALM_TYPE;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ATTACH_REALM_NAME;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ATTACH_REALM_TYPE;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.CLOUD_API_KEY_REALM_NAME;
@@ -569,6 +571,11 @@ public boolean supportsRunAs(@Nullable AnonymousUser anonymousUser) {
             return false;
         }
 
+        // We may allow cloud API keys to run-as in the future, but for now there is no requirement
+        if (isCloudApiKey()) {
+            return false;
+        }
+
         // There is no reason for internal users to run-as. This check prevents either internal user itself
         // or a token created for it (though no such thing in current code) to run-as.
         if (getEffectiveSubject().getUser() instanceof InternalUser) {
@@ -748,14 +755,15 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
      */
     public void toXContentFragment(XContentBuilder builder) throws IOException {
         final User user = effectiveSubject.getUser();
+        final Map<String, Object> metadata = getAuthenticatingSubject().getMetadata();
         builder.field(User.Fields.USERNAME.getPreferredName(), user.principal());
         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 (isServiceAccount()) {
-            final String tokenName = (String) getAuthenticatingSubject().getMetadata().get(ServiceAccountSettings.TOKEN_NAME_FIELD);
+            final String tokenName = (String) metadata.get(ServiceAccountSettings.TOKEN_NAME_FIELD);
             assert tokenName != null : "token name cannot be null";
-            final String tokenSource = (String) getAuthenticatingSubject().getMetadata().get(ServiceAccountSettings.TOKEN_SOURCE_FIELD);
+            final String tokenSource = (String) metadata.get(ServiceAccountSettings.TOKEN_SOURCE_FIELD);
             assert tokenSource != null : "token source cannot be null";
             builder.field(
                 User.Fields.TOKEN.getPreferredName(),
@@ -790,16 +798,31 @@ public void toXContentFragment(XContentBuilder builder) throws IOException {
         }
         builder.endObject();
         builder.field(User.Fields.AUTHENTICATION_TYPE.getPreferredName(), getAuthenticationType().name().toLowerCase(Locale.ROOT));
+
         if (isApiKey() || isCrossClusterAccess()) {
-            final String apiKeyId = (String) getAuthenticatingSubject().getMetadata().get(AuthenticationField.API_KEY_ID_KEY);
-            final String apiKeyName = (String) getAuthenticatingSubject().getMetadata().get(AuthenticationField.API_KEY_NAME_KEY);
-            if (apiKeyName == null) {
-                builder.field("api_key", Map.of("id", apiKeyId));
-            } else {
-                builder.field("api_key", Map.of("id", apiKeyId, "name", apiKeyName));
+            final String apiKeyId = (String) metadata.get(AuthenticationField.API_KEY_ID_KEY);
+            final String apiKeyName = (String) metadata.get(AuthenticationField.API_KEY_NAME_KEY);
+            final Map<String, Object> apiKeyField = new HashMap<>();
+            apiKeyField.put("id", apiKeyId);
+            if (apiKeyName != null) {
+                apiKeyField.put("name", apiKeyName);
+            }
+            apiKeyField.put("managed_by", CredentialManagedBy.ELASTICSEARCH.getDisplayName());
+            builder.field("api_key", Collections.unmodifiableMap(apiKeyField));
+
+        } else if (isCloudApiKey()) {
+            final String apiKeyId = user.principal();
+            final String apiKeyName = (String) user.metadata().get(AuthenticationField.API_KEY_NAME_KEY);
+            final boolean internal = (boolean) user.metadata().get(AuthenticationField.API_KEY_INTERNAL_KEY);
+            final Map<String, Object> apiKeyField = new HashMap<>();
+            apiKeyField.put("id", apiKeyId);
+            if (apiKeyName != null) {
+                apiKeyField.put("name", apiKeyName);
             }
+            apiKeyField.put("internal", internal);
+            apiKeyField.put("managed_by", CredentialManagedBy.CLOUD.getDisplayName());
+            builder.field("api_key", Collections.unmodifiableMap(apiKeyField));
         }
-        // TODO cloud API key fields such as managed_by
     }
 
     public static Authentication getAuthenticationFromCrossClusterAccessMetadata(Authentication authentication) {
@@ -924,10 +947,11 @@ private void checkConsistencyForApiKeyAuthenticationType() {
                 Strings.format("API key authentication cannot have realm type [%s]", authenticatingRealm.type)
             );
         }
-        if (authenticatingRealm.isCloudApiKeyRealm()) {
-            // TODO consistency check for cloud API keys
+        if (authenticatingSubject.getType() == Subject.Type.CLOUD_API_KEY) {
+            checkConsistencyForCloudApiKeyAuthenticatingSubject("Cloud API key");
             return;
         }
+
         checkConsistencyForApiKeyAuthenticatingSubject("API key");
         if (Subject.Type.CROSS_CLUSTER_ACCESS == authenticatingSubject.getType()) {
             if (authenticatingSubject.getMetadata().get(CROSS_CLUSTER_ACCESS_AUTHENTICATION_KEY) == null) {
@@ -1021,6 +1045,18 @@ private void checkConsistencyForApiKeyAuthenticatingSubject(String prefixMessage
         }
     }
 
+    private void checkConsistencyForCloudApiKeyAuthenticatingSubject(String prefixMessage) {
+        final RealmRef authenticatingRealm = authenticatingSubject.getRealm();
+        checkNoDomain(authenticatingRealm, prefixMessage);
+        checkNoInternalUser(authenticatingSubject, prefixMessage);
+        checkNoRunAs(this, prefixMessage);
+        if (authenticatingSubject.getMetadata().get(CROSS_CLUSTER_ACCESS_ROLE_DESCRIPTORS_KEY) != null
+            || authenticatingSubject.getMetadata().get(API_KEY_ROLE_DESCRIPTORS_KEY) != null
+            || authenticatingSubject.getMetadata().get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY) != null) {
+            throw new IllegalArgumentException(prefixMessage + " authentication cannot contain a role descriptors metadata field");
+        }
+    }
+
     private static void checkNoInternalUser(Subject subject, String prefixMessage) {
         if (subject.getUser() instanceof InternalUser) {
             throw new IllegalArgumentException(
@@ -1057,7 +1093,8 @@ private static boolean hasSyntheticRealmNameOrType(@Nullable RealmRef realmRef)
             ANONYMOUS_REALM_NAME,
             FALLBACK_REALM_NAME,
             ATTACH_REALM_NAME,
-            CROSS_CLUSTER_ACCESS_REALM_NAME
+            CROSS_CLUSTER_ACCESS_REALM_NAME,
+            CLOUD_API_KEY_REALM_NAME
         ).contains(realmRef.getName())) {
             return true;
         }
@@ -1067,7 +1104,8 @@ private static boolean hasSyntheticRealmNameOrType(@Nullable RealmRef realmRef)
             ANONYMOUS_REALM_TYPE,
             FALLBACK_REALM_TYPE,
             ATTACH_REALM_TYPE,
-            CROSS_CLUSTER_ACCESS_REALM_TYPE
+            CROSS_CLUSTER_ACCESS_REALM_TYPE,
+            CLOUD_API_KEY_REALM_TYPE
         ).contains(realmRef.getType())) {
             return true;
         }
@@ -1649,6 +1687,20 @@ public enum AuthenticationType {
         INTERNAL
     }
 
+    /**
+     *  Indicates if credentials are managed by Elasticsearch or by the Cloud.
+     */
+    public enum CredentialManagedBy {
+
+        CLOUD,
+
+        ELASTICSEARCH;
+
+        public String getDisplayName() {
+            return name().toLowerCase(Locale.ROOT);
+        }
+    }
+
     public static class AuthenticationSerializationHelper {
 
         private AuthenticationSerializationHelper() {}

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

@@ -23,6 +23,7 @@ public final class AuthenticationField {
     public static final String API_KEY_CREATOR_REALM_TYPE = "_security_api_key_creator_realm_type";
     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_INTERNAL_KEY = "_security_api_key_internal";
     public static final String API_KEY_TYPE_KEY = "_security_api_key_type";
     public static final String API_KEY_METADATA_KEY = "_security_api_key_metadata";
     public static final String API_KEY_ROLE_DESCRIPTORS_KEY = "_security_api_key_role_descriptors";

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

@@ -34,6 +34,7 @@
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.CROSS_CLUSTER_ACCESS_AUTHENTICATION_KEY;
 import static org.elasticsearch.xpack.core.security.authc.Subject.Type.API_KEY;
+import static org.elasticsearch.xpack.core.security.authc.Subject.Type.CLOUD_API_KEY;
 import static org.elasticsearch.xpack.core.security.authc.Subject.Type.CROSS_CLUSTER_ACCESS;
 
 /**
@@ -137,6 +138,13 @@ public boolean canAccessResourcesOf(Subject resourceCreatorSubject) {
                 // A cross cluster access subject can never share resources with non-cross cluster access
                 return false;
             }
+        } else if (eitherIsACloudApiKey(resourceCreatorSubject)) {
+            if (bothAreCloudApiKeys(resourceCreatorSubject)) {
+                return getUser().principal().equals(resourceCreatorSubject.getUser().principal());
+            } else {
+                // a cloud API Key cannot access resources created by non-Cloud API Keys or vice versa
+                return false;
+            }
         } else {
             if (false == getUser().principal().equals(resourceCreatorSubject.getUser().principal())) {
                 return false;
@@ -191,6 +199,14 @@ private boolean bothAreCrossClusterAccess(Subject resourceCreatorSubject) {
         return CROSS_CLUSTER_ACCESS.equals(getType()) && CROSS_CLUSTER_ACCESS.equals(resourceCreatorSubject.getType());
     }
 
+    private boolean eitherIsACloudApiKey(Subject resourceCreatorSubject) {
+        return CLOUD_API_KEY.equals(getType()) || CLOUD_API_KEY.equals(resourceCreatorSubject.getType());
+    }
+
+    private boolean bothAreCloudApiKeys(Subject resourceCreatorSubject) {
+        return CLOUD_API_KEY.equals(getType()) && CLOUD_API_KEY.equals(resourceCreatorSubject.getType());
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

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

@@ -148,13 +148,43 @@ private Map<String, String> getErrorMessageToEncodedAuthentication() throws IOEx
                     Authentication.AuthenticationType.API_KEY
                 )
             ),
+            entry(
+                "Cloud API key authentication cannot have domain",
+                encodeAuthentication(
+                    new Subject(
+                        userFoo,
+                        new Authentication.RealmRef(
+                            AuthenticationField.CLOUD_API_KEY_REALM_NAME,
+                            AuthenticationField.CLOUD_API_KEY_REALM_TYPE,
+                            "node",
+                            new RealmDomain(
+                                "domain1",
+                                Set.of(
+                                    new RealmConfig.RealmIdentifier(
+                                        AuthenticationField.CLOUD_API_KEY_REALM_NAME,
+                                        AuthenticationField.CLOUD_API_KEY_REALM_TYPE
+                                    )
+                                )
+                            )
+                        )
+                    ),
+                    Authentication.AuthenticationType.API_KEY
+                )
+            ),
             entry(
                 "API key authentication cannot have internal user [_xpack]",
                 encodeAuthentication(
                     new Subject(InternalUsers.XPACK_USER, Authentication.RealmRef.newApiKeyRealmRef("node")),
                     Authentication.AuthenticationType.API_KEY
                 )
             ),
+            entry(
+                "Cloud API key authentication cannot have internal user [_xpack]",
+                encodeAuthentication(
+                    new Subject(InternalUsers.XPACK_USER, Authentication.RealmRef.newCloudApiKeyRealmRef("node")),
+                    Authentication.AuthenticationType.API_KEY
+                )
+            ),
             entry(
                 "API key authentication user must have no role",
                 encodeAuthentication(
@@ -237,6 +267,33 @@ private Map<String, String> getErrorMessageToEncodedAuthentication() throws IOEx
                     Authentication.AuthenticationType.API_KEY
                 )
             ),
+            entry(
+                "Cloud API key authentication cannot contain a role descriptors metadata field",
+                encodeAuthentication(
+                    new Subject(
+                        new User("api_key_id", "role1"),
+                        Authentication.RealmRef.newCloudApiKeyRealmRef("node"),
+                        TransportVersion.current(),
+                        Map.of(
+                            randomFrom(
+                                AuthenticationField.CROSS_CLUSTER_ACCESS_ROLE_DESCRIPTORS_KEY,
+                                AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY,
+                                AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY
+                            ),
+                            List.of()
+                        )
+                    ),
+                    Authentication.AuthenticationType.API_KEY
+                )
+            ),
+            entry(
+                "Cloud API key authentication cannot run-as other user",
+                encodeAuthentication(
+                    new Subject(userBar, realm2),
+                    new Subject(userFoo, Authentication.RealmRef.newCloudApiKeyRealmRef("node")),
+                    Authentication.AuthenticationType.API_KEY
+                )
+            ),
             // Authentication type: Realm
             entry(
                 "Realm authentication must have subject type of user",
@@ -330,6 +387,14 @@ private Map<String, String> getErrorMessageToEncodedAuthentication() throws IOEx
                     new Subject(userFoo, realm1),
                     Authentication.AuthenticationType.REALM
                 )
+            ),
+            entry(
+                "Run-as subject type cannot be [CLOUD_API_KEY]",
+                encodeAuthentication(
+                    new Subject(userBar, Authentication.RealmRef.newCloudApiKeyRealmRef("node")),
+                    new Subject(userFoo, realm1),
+                    Authentication.AuthenticationType.REALM
+                )
             )
         );
     }

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

@@ -244,6 +244,48 @@ public static String randomInternalRoleName() {
         );
     }
 
+    public static Authentication randomCloudApiKeyAuthentication() {
+        return randomCloudApiKeyAuthentication(null, null);
+    }
+
+    public static Authentication randomCloudApiKeyAuthentication(String apiKeyId) {
+        return randomCloudApiKeyAuthentication(null, apiKeyId);
+    }
+
+    public static Authentication randomCloudApiKeyAuthentication(User user) {
+        return randomCloudApiKeyAuthentication(user, null);
+    }
+
+    public static Authentication randomCloudApiKeyAuthentication(User user, String apiKeyId) {
+        if (apiKeyId == null) {
+            apiKeyId = user != null ? user.principal() : ESTestCase.randomAlphanumericOfLength(64);
+        }
+        final Map<String, Object> metadata = ESTestCase.randomBoolean()
+            ? null
+            : Map.ofEntries(
+                Map.entry(AuthenticationField.API_KEY_NAME_KEY, ESTestCase.randomAlphanumericOfLength(64)),
+                Map.entry(AuthenticationField.API_KEY_INTERNAL_KEY, ESTestCase.randomBoolean())
+            );
+        if (user == null) {
+            user = new User(
+                apiKeyId,
+                ESTestCase.randomArray(1, 3, String[]::new, () -> "role_" + ESTestCase.randomAlphaOfLengthBetween(3, 8)),
+                null,
+                null,
+                metadata,
+                true
+            );
+        }
+
+        assert user.principal().equals(apiKeyId) : "user principal must match cloud API key ID";
+
+        return Authentication.newCloudApiKeyAuthentication(
+            AuthenticationResult.success(user, metadata),
+            "node_" + ESTestCase.randomAlphaOfLengthBetween(3, 8)
+        );
+
+    }
+
     public static CrossClusterAccessSubjectInfo randomCrossClusterAccessSubjectInfo(
         RoleDescriptorsIntersection roleDescriptorsIntersection
     ) {