Fix so non super users can also create API keys (#40028) (#40286)

When creating API keys we check for if API key with
the same key name already exists and fail the request if it does.
The check should have been performed with XPackSecurityUser
instead of the authenticated user. This caused the request to fail
in case of the non-super user trying to create an API key.
This commit fixes by executing search action with SECURITY_ORIGIN
so it can be executed with XPackSecurityUser.
Also fixed the Rest test to avoid using a user with `super_user` role.

Closes #40029

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

@@ -22,6 +22,7 @@ import org.elasticsearch.action.get.GetRequest;
 import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.index.IndexAction;
 import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchAction;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
 import org.elasticsearch.action.update.UpdateRequest;
@@ -190,8 +191,44 @@ public class ApiKeyService {
              * this check is best effort as there could be two nodes executing search and
              * then index concurrently allowing a duplicate name.
              */
-            findApiKeyForApiKeyName(request.getName(), true, true, ActionListener.wrap(apiKeyIds -> {
-                if (apiKeyIds.isEmpty()) {
+            checkDuplicateApiKeyNameAndCreateApiKey(authentication, request, roleDescriptorSet, listener);
+        }
+    }
+
+    private void checkDuplicateApiKeyNameAndCreateApiKey(Authentication authentication, CreateApiKeyRequest request,
+                                                         Set<RoleDescriptor> roleDescriptorSet,
+                                                         ActionListener<CreateApiKeyResponse> listener) {
+        final BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
+                .filter(QueryBuilders.termQuery("doc_type", "api_key"))
+                .filter(QueryBuilders.termQuery("name", request.getName()))
+                .filter(QueryBuilders.termQuery("api_key_invalidated", false));
+        final BoolQueryBuilder expiredQuery = QueryBuilders.boolQuery()
+                .should(QueryBuilders.rangeQuery("expiration_time").lte(Instant.now().toEpochMilli()))
+                .should(QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery("expiration_time")));
+        boolQuery.filter(expiredQuery);
+
+        final SearchRequest searchRequest = client.prepareSearch(SECURITY_INDEX_NAME)
+            .setScroll(DEFAULT_KEEPALIVE_SETTING.get(settings))
+            .setQuery(boolQuery)
+            .setVersion(false)
+            .setSize(1)
+            .request();
+        securityIndex.prepareIndexIfNeededThenExecute(listener::onFailure, () ->
+        executeAsyncWithOrigin(client, SECURITY_ORIGIN, SearchAction.INSTANCE, searchRequest,
+                ActionListener.wrap(
+                        indexResponse -> {
+                            if (indexResponse.getHits().getTotalHits().value > 0) {
+                                listener.onFailure(traceLog("create api key", new ElasticsearchSecurityException(
+                                        "Error creating api key as api key with name [{}] already exists", request.getName())));
+                            } else {
+                                createApiKeyAndIndexIt(authentication, request, roleDescriptorSet, listener);
+                            }
+                        },
+                        listener::onFailure)));
+    }
+
+    private void createApiKeyAndIndexIt(Authentication authentication, CreateApiKeyRequest request, Set<RoleDescriptor> roleDescriptorSet,
+                                        ActionListener<CreateApiKeyResponse> listener) {
         final Instant created = clock.instant();
         final Instant expiration = getApiKeyExpiration(created, request);
         final SecureString apiKey = UUIDs.randomBase64UUIDSecureString();
@@ -264,12 +301,6 @@ public class ApiKeyService {
         } finally {
             Arrays.fill(keyHash, (char) 0);
         }
-                } else {
-                    listener.onFailure(traceLog("create api key", new ElasticsearchSecurityException(
-                            "Error creating api key as api key with name [{}] already exists", request.getName())));
-                }
-            }, listener::onFailure));
-        }
     }
 
     /**

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

@@ -7,13 +7,34 @@ setup:
       cluster.health:
           wait_for_status: yellow
 
+  - do:
+      security.put_role:
+        name: "admin_role"
+        body:  >
+            {
+              "cluster": ["all"],
+              "indices": [
+                {
+                  "names": "*",
+                  "privileges": ["all"]
+                }
+              ],
+              "applications": [
+                {
+                  "application": "myapp",
+                  "privileges": ["*"],
+                  "resources": ["*"]
+                }
+              ]
+            }
+
   - do:
       security.put_user:
         username: "api_key_user"
         body:  >
           {
             "password" : "x-pack-test-password",
-            "roles" : [ "superuser" ],
+            "roles" : [ "admin_role" ],
             "full_name" : "API key user"
           }
 
@@ -38,6 +59,11 @@ setup:
 
 ---
 teardown:
+  - do:
+      security.delete_role:
+        name: "admin_role"
+        ignore: 404
+
   - do:
       security.delete_user:
         username: "api_key_user"
@@ -54,7 +80,7 @@ teardown:
 
   - do:
       headers:
-        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk"
+        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" # api_key_user
       security.create_api_key:
         body:  >
             {
@@ -105,7 +131,7 @@ teardown:
 
   - do:
       headers:
-        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk"
+        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" # api_key_user
       security.create_api_key:
         body:  >
             {
@@ -140,8 +166,6 @@ teardown:
   - set: { name: api_key_name }
 
   - do:
-      headers:
-        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk"
       security.get_api_key:
         id: "$api_key_id"
   - match: { "api_keys.0.id": "$api_key_id" }
@@ -157,7 +181,7 @@ teardown:
 
   - do:
       headers:
-        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk"
+        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" # api_key_user
       security.create_api_key:
         body:  >
             {
@@ -174,8 +198,6 @@ teardown:
   - transform_and_set: { login_creds: "#base64EncodeCredentials(id,api_key)" }
 
   - do:
-      headers:
-        Authorization: Apikey ${login_creds}
       security.invalidate_api_key:
         body:  >
             {
@@ -193,7 +215,7 @@ teardown:
 
   - do:
       headers:
-        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk"
+        Authorization: "Basic YXBpX2tleV91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" # api_key_user
       security.create_api_key:
         body:  >
             {