Add "manage_api_key" cluster privilege (#43728)

This adds a new cluster privilege for manage_api_key. Users with this
privilege are able to create new API keys (as a child of their own
user identity) and may also get and invalidate any/all API keys
(including those owned by other users).

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

@@ -39,6 +39,7 @@ public final class ClusterPrivilege extends Privilege {
             InvalidateTokenAction.NAME, RefreshTokenAction.NAME);
     private static final Automaton MANAGE_OIDC_AUTOMATON = patterns("cluster:admin/xpack/security/oidc/*");
     private static final Automaton MANAGE_TOKEN_AUTOMATON = patterns("cluster:admin/xpack/security/token/*");
+    private static final Automaton MANAGE_API_KEY_AUTOMATON = patterns("cluster:admin/xpack/security/api_key/*");
     private static final Automaton MONITOR_AUTOMATON = patterns("cluster:monitor/*");
     private static final Automaton MONITOR_ML_AUTOMATON = patterns("cluster:monitor/xpack/ml/*");
     private static final Automaton MONITOR_DATA_FRAME_AUTOMATON = patterns("cluster:monitor/data_frame/*");
@@ -84,6 +85,7 @@ public final class ClusterPrivilege extends Privilege {
     public static final ClusterPrivilege MANAGE_SECURITY =       new ClusterPrivilege("manage_security",     MANAGE_SECURITY_AUTOMATON);
     public static final ClusterPrivilege MANAGE_SAML =           new ClusterPrivilege("manage_saml",         MANAGE_SAML_AUTOMATON);
     public static final ClusterPrivilege MANAGE_OIDC =           new ClusterPrivilege("manage_oidc", MANAGE_OIDC_AUTOMATON);
+    public static final ClusterPrivilege MANAGE_API_KEY =        new ClusterPrivilege("manage_api_key", MANAGE_API_KEY_AUTOMATON);
     public static final ClusterPrivilege MANAGE_PIPELINE =       new ClusterPrivilege("manage_pipeline", "cluster:admin/ingest/pipeline/*");
     public static final ClusterPrivilege MANAGE_CCR =            new ClusterPrivilege("manage_ccr", MANAGE_CCR_AUTOMATON);
     public static final ClusterPrivilege READ_CCR =              new ClusterPrivilege("read_ccr", READ_CCR_AUTOMATON);
@@ -112,6 +114,7 @@ public final class ClusterPrivilege extends Privilege {
             entry("manage_security", MANAGE_SECURITY),
             entry("manage_saml", MANAGE_SAML),
             entry("manage_oidc", MANAGE_OIDC),
+            entry("manage_api_key", MANAGE_API_KEY),
             entry("manage_pipeline", MANAGE_PIPELINE),
             entry("manage_rollup", MANAGE_ROLLUP),
             entry("manage_ccr", MANAGE_CCR),

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

@@ -743,27 +743,29 @@ public class ApiKeyService {
             expiredQuery.should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery("expiration_time")));
             boolQuery.filter(expiredQuery);
         }
-        final SearchRequest request = client.prepareSearch(SECURITY_MAIN_ALIAS)
-            .setScroll(DEFAULT_KEEPALIVE_SETTING.get(settings))
-            .setQuery(boolQuery)
-            .setVersion(false)
-            .setSize(1000)
-            .setFetchSource(true)
-            .request();
-        securityIndex.checkIndexVersionThenExecute(listener::onFailure,
-            () -> ScrollHelper.fetchAllByEntity(client, request, listener,
-                        (SearchHit hit) -> {
-                            Map<String, Object> source = hit.getSourceAsMap();
-                            String name = (String) source.get("name");
-                            String id = hit.getId();
-                            Long creation = (Long) source.get("creation_time");
-                            Long expiration = (Long) source.get("expiration_time");
-                            Boolean invalidated = (Boolean) source.get("api_key_invalidated");
-                            String username = (String) ((Map<String, Object>) source.get("creator")).get("principal");
-                            String realm = (String) ((Map<String, Object>) source.get("creator")).get("realm");
-                            return new ApiKey(name, id, Instant.ofEpochMilli(creation),
-                                    (expiration != null) ? Instant.ofEpochMilli(expiration) : null, invalidated, username, realm);
-                        }));
+        try (ThreadContext.StoredContext ignore = client.threadPool().getThreadContext().stashWithOrigin(SECURITY_ORIGIN)) {
+            final SearchRequest request = client.prepareSearch(SECURITY_MAIN_ALIAS)
+                .setScroll(DEFAULT_KEEPALIVE_SETTING.get(settings))
+                .setQuery(boolQuery)
+                .setVersion(false)
+                .setSize(1000)
+                .setFetchSource(true)
+                .request();
+            securityIndex.checkIndexVersionThenExecute(listener::onFailure,
+                () -> ScrollHelper.fetchAllByEntity(client, request, listener,
+                    (SearchHit hit) -> {
+                        Map<String, Object> source = hit.getSourceAsMap();
+                        String name = (String) source.get("name");
+                        String id = hit.getId();
+                        Long creation = (Long) source.get("creation_time");
+                        Long expiration = (Long) source.get("expiration_time");
+                        Boolean invalidated = (Boolean) source.get("api_key_invalidated");
+                        String username = (String) ((Map<String, Object>) source.get("creator")).get("principal");
+                        String realm = (String) ((Map<String, Object>) source.get("creator")).get("realm");
+                        return new ApiKey(name, id, Instant.ofEpochMilli(creation),
+                            (expiration != null) ? Instant.ofEpochMilli(expiration) : null, invalidated, username, realm);
+                    }));
+        }
     }
 
     private void findApiKeyForApiKeyName(String apiKeyName, boolean filterOutInvalidatedKeys, boolean filterOutExpiredKeys,

x-pack/plugin/src/test/resources/rest-api-spec/test/api_key/10_basic.yml

@@ -12,7 +12,7 @@ setup:
         name: "admin_role"
         body:  >
             {
-              "cluster": ["all"],
+              "cluster": ["manage_api_key"],
               "indices": [
                 {
                   "names": "*",
@@ -166,6 +166,8 @@ teardown:
   - set: { name: api_key_name }
 
   - do:
+      headers:
+        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" # api_key_user
       security.get_api_key:
         id: "$api_key_id"
   - match: { "api_keys.0.id": "$api_key_id" }
@@ -198,6 +200,8 @@ teardown:
   - transform_and_set: { login_creds: "#base64EncodeCredentials(id,api_key)" }
 
   - do:
+      headers:
+        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" # api_key_user
       security.invalidate_api_key:
         body:  >
             {