Add manage_own_api_key cluster privilege (#45696)

This commit adds `manage_own_api_key` cluster privilege which
only allows api key cluster actions on API keys owned by the
currently authenticated user.

Relates: #40031

x-pack/docs/en/rest-api/security/get-builtin-privileges.asciidoc

@@ -75,6 +75,7 @@ A successful call returns an object with "cluster" and "index" fields.
     "manage_ingest_pipelines",
     "manage_ml",
     "manage_oidc",
+    "manage_own_api_key",
     "manage_pipeline",
     "manage_rollup",
     "manage_saml",

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/ClusterPermission.java

@@ -158,32 +158,60 @@ public interface PermissionCheck {
         boolean implies(PermissionCheck otherPermissionCheck);
     }
 
-    // Automaton based permission check
-    private static class AutomatonPermissionCheck implements PermissionCheck {
+    /**
+     * Base for implementing cluster action based {@link PermissionCheck}.
+     * It enforces the checks at cluster action level and then hands it off to the implementations
+     * to enforce checks based on {@link TransportRequest} and/or {@link Authentication}.
+     */
+    public abstract static class ActionBasedPermissionCheck implements PermissionCheck {
         private final Automaton automaton;
         private final Predicate<String> actionPredicate;
 
-        AutomatonPermissionCheck(final Automaton automaton) {
+        public ActionBasedPermissionCheck(final Automaton automaton) {
             this.automaton = automaton;
             this.actionPredicate = Automatons.predicate(automaton);
         }
 
         @Override
-        public boolean check(final String action, final TransportRequest request, final Authentication authentication) {
-            return actionPredicate.test(action);
+        public final boolean check(final String action, final TransportRequest request, final Authentication authentication) {
+            return actionPredicate.test(action) && extendedCheck(action, request, authentication);
         }
 
+        protected abstract boolean extendedCheck(String action, TransportRequest request, Authentication authentication);
+
         @Override
-        public boolean implies(final PermissionCheck permissionCheck) {
-            if (permissionCheck instanceof AutomatonPermissionCheck) {
-                return Operations.subsetOf(((AutomatonPermissionCheck) permissionCheck).automaton, this.automaton);
+        public final boolean implies(final PermissionCheck permissionCheck) {
+            if (permissionCheck instanceof ActionBasedPermissionCheck) {
+                return Operations.subsetOf(((ActionBasedPermissionCheck) permissionCheck).automaton, this.automaton) &&
+                    doImplies((ActionBasedPermissionCheck) permissionCheck);
             }
             return false;
         }
+
+        protected abstract boolean doImplies(ActionBasedPermissionCheck permissionCheck);
+    }
+
+    // Automaton based permission check
+    private static class AutomatonPermissionCheck extends ActionBasedPermissionCheck {
+
+        AutomatonPermissionCheck(final Automaton automaton) {
+            super(automaton);
+        }
+
+        @Override
+        protected boolean extendedCheck(String action, TransportRequest request, Authentication authentication) {
+            return true;
+        }
+
+        @Override
+        protected boolean doImplies(ActionBasedPermissionCheck permissionCheck) {
+            return permissionCheck instanceof AutomatonPermissionCheck;
+        }
+
     }
 
     // action, request based permission check
-    private static class ActionRequestBasedPermissionCheck extends AutomatonPermissionCheck {
+    private static class ActionRequestBasedPermissionCheck extends ActionBasedPermissionCheck {
         private final ClusterPrivilege clusterPrivilege;
         private final Predicate<TransportRequest> requestPredicate;
 
@@ -195,18 +223,16 @@ private static class ActionRequestBasedPermissionCheck extends AutomatonPermissi
         }
 
         @Override
-        public boolean check(final String action, final TransportRequest request, final Authentication authentication) {
-            return super.check(action, request, authentication) && requestPredicate.test(request);
+        protected boolean extendedCheck(String action, TransportRequest request, Authentication authentication) {
+            return requestPredicate.test(request);
         }
 
         @Override
-        public boolean implies(final PermissionCheck permissionCheck) {
-            if (super.implies(permissionCheck)) {
-                if (permissionCheck instanceof ActionRequestBasedPermissionCheck) {
-                    final ActionRequestBasedPermissionCheck otherCheck =
-                        (ActionRequestBasedPermissionCheck) permissionCheck;
-                    return this.clusterPrivilege.equals(otherCheck.clusterPrivilege);
-                }
+        protected boolean doImplies(final ActionBasedPermissionCheck permissionCheck) {
+            if (permissionCheck instanceof ActionRequestBasedPermissionCheck) {
+                final ActionRequestBasedPermissionCheck otherCheck =
+                    (ActionRequestBasedPermissionCheck) permissionCheck;
+                return this.clusterPrivilege.equals(otherCheck.clusterPrivilege);
             }
             return false;
         }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/ClusterPrivilegeResolver.java

@@ -103,6 +103,8 @@ public class ClusterPrivilegeResolver {
     public static final NamedClusterPrivilege MANAGE_SLM = new ActionClusterPrivilege("manage_slm", MANAGE_SLM_PATTERN);
     public static final NamedClusterPrivilege READ_SLM = new ActionClusterPrivilege("read_slm", READ_SLM_PATTERN);
 
+    public static final NamedClusterPrivilege MANAGE_OWN_API_KEY = ManageOwnApiKeyClusterPrivilege.INSTANCE;
+
     private static final Map<String, NamedClusterPrivilege> VALUES = Stream.of(
         NONE,
         ALL,
@@ -131,7 +133,8 @@ public class ClusterPrivilegeResolver {
         MANAGE_ILM,
         READ_ILM,
         MANAGE_SLM,
-        READ_SLM).collect(Collectors.toUnmodifiableMap(NamedClusterPrivilege::name, Function.identity()));
+        READ_SLM,
+        MANAGE_OWN_API_KEY).collect(Collectors.toUnmodifiableMap(NamedClusterPrivilege::name, Function.identity()));
 
     /**
      * Resolves a {@link NamedClusterPrivilege} from a given name if it exists.

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/ManageOwnApiKeyClusterPrivilege.java

@@ -0,0 +1,106 @@
+/*
+ *
+ *  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.authz.privilege;
+
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.xpack.core.security.action.CreateApiKeyRequest;
+import org.elasticsearch.xpack.core.security.action.GetApiKeyRequest;
+import org.elasticsearch.xpack.core.security.action.InvalidateApiKeyRequest;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+import org.elasticsearch.xpack.core.security.authz.permission.ClusterPermission;
+import org.elasticsearch.xpack.core.security.support.Automatons;
+
+/**
+ * Named cluster privilege for managing API keys owned by the current authenticated user.
+ */
+public class ManageOwnApiKeyClusterPrivilege implements NamedClusterPrivilege {
+    public static final ManageOwnApiKeyClusterPrivilege INSTANCE = new ManageOwnApiKeyClusterPrivilege();
+    private static final String PRIVILEGE_NAME = "manage_own_api_key";
+    private static final String API_KEY_REALM_TYPE = "_es_api_key";
+    private static final String API_KEY_ID_KEY = "_security_api_key_id";
+
+    private ManageOwnApiKeyClusterPrivilege() {
+    }
+
+    @Override
+    public String name() {
+        return PRIVILEGE_NAME;
+    }
+
+    @Override
+    public ClusterPermission.Builder buildPermission(ClusterPermission.Builder builder) {
+        return builder.add(this, ManageOwnClusterPermissionCheck.INSTANCE);
+    }
+
+    private static final class ManageOwnClusterPermissionCheck extends ClusterPermission.ActionBasedPermissionCheck {
+        public static final ManageOwnClusterPermissionCheck INSTANCE = new ManageOwnClusterPermissionCheck();
+
+        private ManageOwnClusterPermissionCheck() {
+            super(Automatons.patterns("cluster:admin/xpack/security/api_key/*"));
+        }
+
+        @Override
+        protected boolean extendedCheck(String action, TransportRequest request, Authentication authentication) {
+            if (request instanceof CreateApiKeyRequest) {
+                return true;
+            } else if (request instanceof GetApiKeyRequest) {
+                final GetApiKeyRequest getApiKeyRequest = (GetApiKeyRequest) request;
+                return checkIfUserIsOwnerOfApiKeys(authentication, getApiKeyRequest.getApiKeyId(), getApiKeyRequest.getUserName(),
+                    getApiKeyRequest.getRealmName(), getApiKeyRequest.ownedByAuthenticatedUser());
+            } else if (request instanceof InvalidateApiKeyRequest) {
+                final InvalidateApiKeyRequest invalidateApiKeyRequest = (InvalidateApiKeyRequest) request;
+                return checkIfUserIsOwnerOfApiKeys(authentication, invalidateApiKeyRequest.getId(),
+                    invalidateApiKeyRequest.getUserName(), invalidateApiKeyRequest.getRealmName(),
+                    invalidateApiKeyRequest.ownedByAuthenticatedUser());
+            }
+            throw new IllegalArgumentException(
+                "manage own api key privilege only supports API key requests (not " + request.getClass().getName() + ")");
+        }
+
+        @Override
+        protected boolean doImplies(ClusterPermission.ActionBasedPermissionCheck permissionCheck) {
+            return permissionCheck instanceof ManageOwnClusterPermissionCheck;
+        }
+
+        private boolean checkIfUserIsOwnerOfApiKeys(Authentication authentication, String apiKeyId, String username, String realmName,
+                                                    boolean ownedByAuthenticatedUser) {
+            if (isCurrentAuthenticationUsingSameApiKeyIdFromRequest(authentication, apiKeyId)) {
+                return true;
+            } else {
+                /*
+                 * TODO bizybot we need to think on how we can propagate appropriate error message to the end user when username, realm name
+                 *   is missing. This is similar to the problem of propagating right error messages in case of access denied.
+                 */
+                if (authentication.getAuthenticatedBy().getType().equals(API_KEY_REALM_TYPE)) {
+                    // API key cannot own any other API key so deny access
+                    return false;
+                } else if (ownedByAuthenticatedUser) {
+                    return true;
+                } else if (Strings.hasText(username) && Strings.hasText(realmName)) {
+                    final String authenticatedUserPrincipal = authentication.getUser().principal();
+                    final String authenticatedUserRealm = authentication.getAuthenticatedBy().getName();
+                    return username.equals(authenticatedUserPrincipal) && realmName.equals(authenticatedUserRealm);
+                }
+            }
+            return false;
+        }
+
+        private boolean isCurrentAuthenticationUsingSameApiKeyIdFromRequest(Authentication authentication, String apiKeyId) {
+            if (authentication.getAuthenticatedBy().getType().equals(API_KEY_REALM_TYPE)) {
+                // API key id from authentication must match the id from request
+                final String authenticatedApiKeyId = (String) authentication.getMetadata().get(API_KEY_ID_KEY);
+                if (Strings.hasText(apiKeyId)) {
+                    return apiKeyId.equals(authenticatedApiKeyId);
+                }
+            }
+            return false;
+        }
+    }
+}

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/permission/ClusterPermissionTests.java

@@ -276,7 +276,7 @@ public int hashCode() {
         @Override
         public String toString() {
             return "MockConfigurableClusterPrivilege{" +
-                "requestAuthnPredicate=" + requestPredicate +
+                "requestPredicate=" + requestPredicate +
                 '}';
         }
 

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/privilege/ManageOwnApiKeyClusterPrivilegeTests.java

@@ -0,0 +1,110 @@
+/*
+ *
+ * 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.authz.privilege;
+
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.xpack.core.security.action.GetApiKeyRequest;
+import org.elasticsearch.xpack.core.security.action.InvalidateApiKeyRequest;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+import org.elasticsearch.xpack.core.security.authz.permission.ClusterPermission;
+import org.elasticsearch.xpack.core.security.user.User;
+
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ManageOwnApiKeyClusterPrivilegeTests extends ESTestCase {
+
+    public void testAuthenticationWithApiKeyAllowsAccessToApiKeyActionsWhenItIsOwner() {
+        final ClusterPermission clusterPermission =
+            ManageOwnApiKeyClusterPrivilege.INSTANCE.buildPermission(ClusterPermission.builder()).build();
+
+        final String apiKeyId = randomAlphaOfLengthBetween(4, 7);
+        final Authentication authentication = createMockAuthentication("joe","_es_api_key", "_es_api_key",
+            Map.of("_security_api_key_id", apiKeyId));
+        final TransportRequest getApiKeyRequest = GetApiKeyRequest.usingApiKeyId(apiKeyId, randomBoolean());
+        final TransportRequest invalidateApiKeyRequest = InvalidateApiKeyRequest.usingApiKeyId(apiKeyId, randomBoolean());
+
+        assertTrue(clusterPermission.check("cluster:admin/xpack/security/api_key/get", getApiKeyRequest, authentication));
+        assertTrue(clusterPermission.check("cluster:admin/xpack/security/api_key/invalidate", invalidateApiKeyRequest, authentication));
+        assertFalse(clusterPermission.check("cluster:admin/something", mock(TransportRequest.class), authentication));
+    }
+
+    public void testAuthenticationWithApiKeyDeniesAccessToApiKeyActionsWhenItIsNotOwner() {
+        final ClusterPermission clusterPermission =
+            ManageOwnApiKeyClusterPrivilege.INSTANCE.buildPermission(ClusterPermission.builder()).build();
+
+        final String apiKeyId = randomAlphaOfLengthBetween(4, 7);
+        final Authentication authentication = createMockAuthentication("joe","_es_api_key", "_es_api_key",
+            Map.of("_security_api_key_id", randomAlphaOfLength(7)));
+        final TransportRequest getApiKeyRequest = GetApiKeyRequest.usingApiKeyId(apiKeyId, randomBoolean());
+        final TransportRequest invalidateApiKeyRequest = InvalidateApiKeyRequest.usingApiKeyId(apiKeyId, randomBoolean());
+
+        assertFalse(clusterPermission.check("cluster:admin/xpack/security/api_key/get", getApiKeyRequest, authentication));
+        assertFalse(clusterPermission.check("cluster:admin/xpack/security/api_key/invalidate", invalidateApiKeyRequest, authentication));
+    }
+
+    public void testAuthenticationWithUserAllowsAccessToApiKeyActionsWhenItIsOwner() {
+        final ClusterPermission clusterPermission =
+            ManageOwnApiKeyClusterPrivilege.INSTANCE.buildPermission(ClusterPermission.builder()).build();
+
+        final Authentication authentication = createMockAuthentication("joe","realm1", "native", Map.of());
+        final TransportRequest getApiKeyRequest = GetApiKeyRequest.usingRealmAndUserName("realm1", "joe");
+        final TransportRequest invalidateApiKeyRequest = InvalidateApiKeyRequest.usingRealmAndUserName("realm1", "joe");
+
+        assertTrue(clusterPermission.check("cluster:admin/xpack/security/api_key/get", getApiKeyRequest, authentication));
+        assertTrue(clusterPermission.check("cluster:admin/xpack/security/api_key/invalidate", invalidateApiKeyRequest, authentication));
+        assertFalse(clusterPermission.check("cluster:admin/something", mock(TransportRequest.class), authentication));
+    }
+
+    public void testAuthenticationWithUserAllowsAccessToApiKeyActionsWhenItIsOwner_WithOwnerFlagOnly() {
+        final ClusterPermission clusterPermission =
+            ManageOwnApiKeyClusterPrivilege.INSTANCE.buildPermission(ClusterPermission.builder()).build();
+
+        final Authentication authentication = createMockAuthentication("joe","realm1", "native", Map.of());
+        final TransportRequest getApiKeyRequest = GetApiKeyRequest.forOwnedApiKeys();
+        final TransportRequest invalidateApiKeyRequest = InvalidateApiKeyRequest.forOwnedApiKeys();
+
+        assertTrue(clusterPermission.check("cluster:admin/xpack/security/api_key/get", getApiKeyRequest, authentication));
+        assertTrue(clusterPermission.check("cluster:admin/xpack/security/api_key/invalidate", invalidateApiKeyRequest, authentication));
+        assertFalse(clusterPermission.check("cluster:admin/something", mock(TransportRequest.class), authentication));
+    }
+
+    public void testAuthenticationWithUserDeniesAccessToApiKeyActionsWhenItIsNotOwner() {
+        final ClusterPermission clusterPermission =
+            ManageOwnApiKeyClusterPrivilege.INSTANCE.buildPermission(ClusterPermission.builder()).build();
+
+        final Authentication authentication = createMockAuthentication("joe", "realm1", "native", Map.of());
+        final TransportRequest getApiKeyRequest = randomFrom(
+            GetApiKeyRequest.usingRealmAndUserName("realm1", randomAlphaOfLength(7)),
+            GetApiKeyRequest.usingRealmAndUserName(randomAlphaOfLength(5), "joe"),
+            new GetApiKeyRequest(randomAlphaOfLength(5), randomAlphaOfLength(7), null, null, false));
+        final TransportRequest invalidateApiKeyRequest = randomFrom(
+            InvalidateApiKeyRequest.usingRealmAndUserName("realm1", randomAlphaOfLength(7)),
+            InvalidateApiKeyRequest.usingRealmAndUserName(randomAlphaOfLength(5), "joe"),
+            new InvalidateApiKeyRequest(randomAlphaOfLength(5), randomAlphaOfLength(7), null, null, false));
+
+        assertFalse(clusterPermission.check("cluster:admin/xpack/security/api_key/get", getApiKeyRequest, authentication));
+        assertFalse(clusterPermission.check("cluster:admin/xpack/security/api_key/invalidate", invalidateApiKeyRequest, authentication));
+    }
+
+    private Authentication createMockAuthentication(String username, String realmName, String realmType, Map<String, Object> metadata) {
+        final User user = new User(username);
+        final Authentication authentication = mock(Authentication.class);
+        final Authentication.RealmRef authenticatedBy = mock(Authentication.RealmRef.class);
+        when(authentication.getUser()).thenReturn(user);
+        when(authentication.getAuthenticatedBy()).thenReturn(authenticatedBy);
+        when(authenticatedBy.getName()).thenReturn(realmName);
+        when(authenticatedBy.getType()).thenReturn(realmType);
+        when(authentication.getMetadata()).thenReturn(metadata);
+        return authentication;
+    }
+}

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/TransportGetApiKeyAction.java

@@ -50,7 +50,7 @@ protected void doExecute(Task task, GetApiKeyRequest request, ActionListener<Get
             assert realm == null;
             // restrict username and realm to current authenticated user.
             username = authentication.getUser().principal();
-            realm = authentication.getAuthenticatedBy().getName();
+            realm = ApiKeyService.getCreatorRealmName(authentication);
         }
 
         apiKeyService.getApiKeys(realm, username, apiKeyName, apiKeyId, listener);

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/TransportInvalidateApiKeyAction.java

@@ -50,7 +50,7 @@ protected void doExecute(Task task, InvalidateApiKeyRequest request, ActionListe
             assert realm == null;
             // restrict username and realm to current authenticated user.
             username = authentication.getUser().principal();
-            realm = authentication.getAuthenticatedBy().getName();
+            realm = ApiKeyService.getCreatorRealmName(authentication);
         }
 
         apiKeyService.invalidateApiKeys(realm, username, apiKeyName, apiKeyId, listener);