[UIAM] Cloud API key authentication metadata and validations

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

@@ -748,14 +748,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 +791,30 @@ 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));
+
+        if (isApiKey() || isCrossClusterAccess() || isCloudApiKey()) {
+            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);
+            }
+            if (isCloudApiKey()) {
+                final boolean internal = (boolean) metadata.get(AuthenticationField.API_KEY_INTERNAL_KEY);
+                apiKeyField.put("internal", internal);
+            }
+            final String managedBy = (String) metadata.get(AuthenticationField.API_KEY_MANAGED_BY_KEY);
+            if (managedBy != null) {
+                apiKeyField.put("managed_by", managedBy);
             } else {
-                builder.field("api_key", Map.of("id", apiKeyId, "name", apiKeyName));
+                apiKeyField.put(
+                    "managed_by",
+                    isCloudApiKey() ? ManagedBy.CLOUD.getDisplayName() : ManagedBy.ELASTICSEARCH.getDisplayName()
+                );
             }
+            builder.field("api_key", Collections.unmodifiableMap(apiKeyField));
         }
-        // TODO cloud API key fields such as managed_by
     }
 
     public static Authentication getAuthenticationFromCrossClusterAccessMetadata(Authentication authentication) {
@@ -924,11 +939,8 @@ 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
-            return;
-        }
-        checkConsistencyForApiKeyAuthenticatingSubject("API key");
+        final String prefixMessage = authenticatingRealm.isCloudApiKeyRealm() ? "Cloud API key" : "API key";
+        checkConsistencyForApiKeyAuthenticatingSubject(prefixMessage);
         if (Subject.Type.CROSS_CLUSTER_ACCESS == authenticatingSubject.getType()) {
             if (authenticatingSubject.getMetadata().get(CROSS_CLUSTER_ACCESS_AUTHENTICATION_KEY) == null) {
                 throw new IllegalArgumentException(
@@ -1015,7 +1027,9 @@ private void checkConsistencyForApiKeyAuthenticatingSubject(String prefixMessage
         final RealmRef authenticatingRealm = authenticatingSubject.getRealm();
         checkNoDomain(authenticatingRealm, prefixMessage);
         checkNoInternalUser(authenticatingSubject, prefixMessage);
-        checkNoRole(authenticatingSubject, prefixMessage);
+        if (Subject.Type.CLOUD_API_KEY != authenticatingSubject.getType()) {
+            checkNoRole(authenticatingSubject, prefixMessage);
+        }
         if (authenticatingSubject.getMetadata().get(AuthenticationField.API_KEY_ID_KEY) == null) {
             throw new IllegalArgumentException(prefixMessage + " authentication requires metadata to contain a non-null API key ID");
         }
@@ -1057,7 +1071,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 +1082,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 +1665,20 @@ public enum AuthenticationType {
         INTERNAL
     }
 
+    /**
+     *  Indicates if credentials are managed by Elasticsearch or by the Cloud.
+     */
+    public enum ManagedBy {
+
+        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,8 @@ 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_MANAGED_BY_KEY = "_security_api_key_managed_by";
+    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 isTheSameApiKey(resourceCreatorSubject);
+            } 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(
@@ -169,6 +199,13 @@ private Map<String, String> getErrorMessageToEncodedAuthentication() throws IOEx
                     Authentication.AuthenticationType.API_KEY
                 )
             ),
+            entry(
+                "Cloud API key authentication requires metadata to contain a non-null API key ID",
+                encodeAuthentication(
+                    new Subject(new User("username", "role1"), Authentication.RealmRef.newCloudApiKeyRealmRef("node")),
+                    Authentication.AuthenticationType.API_KEY
+                )
+            ),
             entry(
                 "Cross cluster access authentication requires metadata to contain "
                     + "a non-null serialized cross cluster access authentication field",
@@ -330,6 +367,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/AuthenticationSerializationTests.java

@@ -117,7 +117,10 @@ public void testWriteToWithCrossClusterAccessThrowsOnUnsupportedVersion() throws
 
     public void testWriteToAndReadFromWithCloudApiKeyAuthentication() throws Exception {
         final Authentication authentication = Authentication.newCloudApiKeyAuthentication(
-            AuthenticationResult.success(new User(randomAlphanumericOfLength(5), "superuser"), Map.of()),
+            AuthenticationResult.success(
+                new User(randomAlphanumericOfLength(5), "superuser"),
+                Map.of(AuthenticationField.API_KEY_ID_KEY, "test-api-key-id")
+            ),
             randomAlphanumericOfLength(10)
         );
 
@@ -134,7 +137,10 @@ public void testWriteToAndReadFromWithCloudApiKeyAuthentication() throws Excepti
 
     public void testWriteToWithCloudApiKeyThrowsOnUnsupportedVersion() {
         final Authentication authentication = Authentication.newCloudApiKeyAuthentication(
-            AuthenticationResult.success(new User(randomAlphanumericOfLength(5), "superuser"), Map.of()),
+            AuthenticationResult.success(
+                new User(randomAlphanumericOfLength(5), "superuser"),
+                Map.of(AuthenticationField.API_KEY_ID_KEY, "test-api-key-id")
+            ),
             randomAlphanumericOfLength(10)
         );
 

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

@@ -244,6 +244,30 @@ 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, String apiKeyId) {
+        if (apiKeyId == null) {
+            apiKeyId = ESTestCase.randomAlphanumericOfLength(64);
+        }
+        if (user == null) {
+            user = randomUser();
+        }
+        Map<String, Object> metadata = Map.of(AuthenticationField.API_KEY_ID_KEY, apiKeyId);
+
+        return Authentication.newCloudApiKeyAuthentication(
+            AuthenticationResult.success(user, metadata),
+            ESTestCase.randomAlphaOfLengthBetween(3, 8)
+        );
+
+    }
+
     public static CrossClusterAccessSubjectInfo randomCrossClusterAccessSubjectInfo(
         RoleDescriptorsIntersection roleDescriptorsIntersection
     ) {

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

@@ -113,17 +113,27 @@ public void testCanAccessResourcesOf() {
             randomAuthentication(user1, realm1),
             randomApiKeyAuthentication(user1, randomAlphaOfLengthBetween(10, 20))
         );
+        assertCannotAccessResources(
+            randomAuthentication(user1, realm1),
+            randomCloudApiKeyAuthentication(user1, randomAlphaOfLengthBetween(10, 20))
+        );
 
         // Same API key ID are the same owner
         final String apiKeyId1 = randomAlphaOfLengthBetween(10, 20);
         assertCanAccessResources(randomApiKeyAuthentication(user1, apiKeyId1), randomApiKeyAuthentication(user1, apiKeyId1));
+        assertCanAccessResources(randomCloudApiKeyAuthentication(user1, apiKeyId1), randomCloudApiKeyAuthentication(user1, apiKeyId1));
 
         // Two API keys (2 API key IDs) are not the same owner
         final String apiKeyId2 = randomValueOtherThan(apiKeyId1, () -> randomAlphaOfLengthBetween(10, 20));
         assertCannotAccessResources(
             randomApiKeyAuthentication(randomFrom(user1, user2), apiKeyId1),
             randomApiKeyAuthentication(randomFrom(user1, user2), apiKeyId2)
         );
+        assertCannotAccessResources(
+            randomCloudApiKeyAuthentication(randomFrom(user1, user2), apiKeyId1),
+            randomCloudApiKeyAuthentication(randomFrom(user1, user2), apiKeyId2)
+        );
+
         final User user3 = randomValueOtherThanMany(
             u -> u.principal().equals(user1.principal()) || u.principal().equals(user2.principal()),
             AuthenticationTests::randomUser
@@ -134,6 +144,11 @@ public void testCanAccessResourcesOf() {
             randomApiKeyAuthentication(user1, apiKeyId1).runAs(user2, realm2),
             randomApiKeyAuthentication(user1, apiKeyId1).runAs(user3, realm2)
         );
+        // Same cloud API key but run-as different user are not the same owner
+        assertCannotAccessResources(
+            randomCloudApiKeyAuthentication(user1, apiKeyId1).runAs(user2, realm2),
+            randomCloudApiKeyAuthentication(user1, apiKeyId1).runAs(user3, realm2)
+        );
 
         // Same or different API key run-as the same user are the same owner
         assertCanAccessResources(
@@ -144,6 +159,15 @@ public void testCanAccessResourcesOf() {
             randomApiKeyAuthentication(user1, apiKeyId1).runAs(user3, realm2),
             randomApiKeyAuthentication(user2, apiKeyId2).runAs(user3, realm2)
         );
+        // Same or different cloud API key run-as the same user are the same owner
+        assertCanAccessResources(
+            randomCloudApiKeyAuthentication(user1, apiKeyId1).runAs(user3, realm2),
+            randomCloudApiKeyAuthentication(user1, apiKeyId1).runAs(user3, realm2)
+        );
+        assertCanAccessResources(
+            randomCloudApiKeyAuthentication(user1, apiKeyId1).runAs(user3, realm2),
+            randomCloudApiKeyAuthentication(user2, apiKeyId2).runAs(user3, realm2)
+        );
     }
 
     public void testCrossClusterAccessCanAccessResourceOf() throws IOException {
@@ -776,7 +800,12 @@ public void testToXContentWithApiKey() throws IOException {
             authentication1,
             m -> assertThat(
                 m,
-                hasEntry("api_key", apiKeyName != null ? Map.of("id", apiKeyId, "name", apiKeyName) : Map.of("id", apiKeyId))
+                hasEntry(
+                    "api_key",
+                    apiKeyName != null
+                        ? Map.of("id", apiKeyId, "name", apiKeyName, "managed_by", "elasticsearch")
+                        : Map.of("id", apiKeyId, "managed_by", "elasticsearch")
+                )
             )
         );
 
@@ -796,7 +825,12 @@ public void testToXContentWithCrossClusterAccess() throws IOException {
             authentication,
             m -> assertThat(
                 m,
-                hasEntry("api_key", apiKeyName != null ? Map.of("id", apiKeyId, "name", apiKeyName) : Map.of("id", apiKeyId))
+                hasEntry(
+                    "api_key",
+                    apiKeyName != null
+                        ? Map.of("id", apiKeyId, "name", apiKeyName, "managed_by", "elasticsearch")
+                        : Map.of("id", apiKeyId, "managed_by", "elasticsearch")
+                )
             )
         );
     }
@@ -1313,6 +1347,10 @@ public static Authentication randomAuthentication(User user, RealmRef realmRef,
         return AuthenticationTestHelper.builder().user(user).realmRef(realmRef).transportVersion(version).metadata(metadata).build(isRunAs);
     }
 
+    private static Authentication randomCloudApiKeyAuthentication(User user, String apiKeyId) {
+        return AuthenticationTestHelper.randomCloudApiKeyAuthentication(user, apiKeyId);
+    }
+
     public static Authentication randomApiKeyAuthentication(User user, String apiKeyId) {
         return randomApiKeyAuthentication(
             user,