Ensure TermsEnum action works correctly with API keys (#91170)

An API key's permission is bounded by its owner user's permission. When
checking for DLS access, both the key's permission and the owner user's
permission must be consulted. The access is granted only when it is
granted by both. This PR ensures this logic is correctly enforced by the
termsEnum action.

docs/changelog/91170.yaml

@@ -0,0 +1,5 @@
+pr: 91170
+summary: Ensure `TermsEnum` action works correctly with API keys
+area: Authorization
+type: bug
+issues: []

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/termsenum/action/TransportTermsEnumAction.java

@@ -429,28 +429,44 @@ public class TransportTermsEnumAction extends HandledTransportAction<TermsEnumRe
                     Collections.emptyMap()
                 );
 
-                // Current user has potentially many roles and therefore potentially many queries
-                // defining sets of docs accessible
-                Set<BytesReference> queries = indexAccessControl.getDocumentPermissions().getQueries();
-                for (BytesReference querySource : queries) {
-                    QueryBuilder queryBuilder = DLSRoleQueryValidator.evaluateAndVerifyRoleQuery(
-                        querySource,
-                        scriptService,
-                        queryShardContext.getParserConfig().registry(),
-                        securityContext.getUser()
+                // Unfiltered access is allowed when both base role and limiting role allow it.
+                return hasMatchAllEquivalent(indexAccessControl.getDocumentPermissions().getQueries(), securityContext, queryShardContext)
+                    && hasMatchAllEquivalent(
+                        indexAccessControl.getDocumentPermissions().getLimitedByQueries(),
+                        securityContext,
+                        queryShardContext
                     );
-                    QueryBuilder rewrittenQueryBuilder = Rewriteable.rewrite(queryBuilder, queryShardContext);
-                    if (rewrittenQueryBuilder instanceof MatchAllQueryBuilder) {
-                        // One of the roles assigned has "all" permissions - allow unfettered access to termsDict
-                        return true;
-                    }
-                }
-                return false;
             }
         }
         return true;
     }
 
+    private boolean hasMatchAllEquivalent(
+        Set<BytesReference> queries,
+        SecurityContext securityContext,
+        SearchExecutionContext queryShardContext
+    ) throws IOException {
+        if (queries == null) {
+            return true;
+        }
+        // Current user has potentially many roles and therefore potentially many queries
+        // defining sets of docs accessible
+        for (BytesReference querySource : queries) {
+            QueryBuilder queryBuilder = DLSRoleQueryValidator.evaluateAndVerifyRoleQuery(
+                querySource,
+                scriptService,
+                queryShardContext.getParserConfig().registry(),
+                securityContext.getUser()
+            );
+            QueryBuilder rewrittenQueryBuilder = Rewriteable.rewrite(queryBuilder, queryShardContext);
+            if (rewrittenQueryBuilder instanceof MatchAllQueryBuilder) {
+                // One of the roles assigned has "all" permissions - allow unfettered access to termsDict
+                return true;
+            }
+        }
+        return false;
+    }
+
     private boolean canMatchShard(ShardId shardId, NodeTermsEnumRequest req) throws IOException {
         if (req.indexFilter() == null || req.indexFilter() instanceof MatchAllQueryBuilder) {
             return true;

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

@@ -33,6 +33,7 @@ setup:
         name: "dls_all_role"
         body:  >
             {
+              "cluster": [ "manage_own_api_key" ],
               "indices": [
                 { "names": ["test_security"], "privileges": ["read"], "query": "{\"term\": {\"ck\": \"const\"}}" }
               ]
@@ -82,6 +83,7 @@ setup:
         name: "dls_some_role"
         body:  >
             {
+              "cluster": [ "manage_own_api_key" ],
               "indices": [
                 { "names": ["test_security"], "privileges": ["read"], "query": "{\"term\": {\"foo\": \"bar_dls\"}}" }
               ]
@@ -560,4 +562,90 @@ teardown:
         body:  {"field": "foo", "string":"b"}
   - length: {terms: 0}
 
+---
+"Test security with API keys":
+
+  - do:
+      headers: { Authorization: "Basic ZGxzX2FsbF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # dls_all_user
+      security.create_api_key:
+        body: >
+          {
+            "name": "dls_all_user_good_key",
+            "role_descriptors": {
+              "role-a": {
+                "index": [
+                  {
+                    "names": ["test_security"],
+                    "privileges": ["read"],
+                    "query": "{\"term\": {\"ck\": \"const\"}}"
+                  }
+                ]
+              }
+            }
+          }
+  - match: { name: "dls_all_user_good_key" }
+  - set: { encoded: login_creds}
+  - do:
+      headers:
+        Authorization: ApiKey ${login_creds}  # dls_all_user good API key sees all documents
+      terms_enum:
+        index: test_security
+        body: { "field": "foo", "string": "b" }
+  - length: { terms: 1 }
 
+  - do:
+      headers: { Authorization: "Basic ZGxzX2FsbF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # dls_all_user
+      security.create_api_key:
+        body: >
+          {
+            "name": "dls_all_user_bad_key",
+            "role_descriptors": {
+              "role-a": {
+                "index": [
+                  {
+                    "names": ["test_security"],
+                    "privileges": ["read"],
+                    "query": "{\"term\": {\"foo\": \"bar_dls\"}}"
+                  }
+                ]
+              }
+            }
+          }
+  - match: { name: "dls_all_user_bad_key" }
+  - set: { encoded: login_creds}
+  - do:
+      headers:
+        Authorization: ApiKey ${login_creds}  # dls_all_user bad API key sees selected docs
+      terms_enum:
+        index: test_security
+        body: { "field": "foo", "string": "b" }
+  - length: { terms: 0 }
+
+  - do:
+      headers: { Authorization: "Basic ZGxzX3NvbWVfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" } # dls_some_user
+      # Create an API key with all DLS permissions, but it is still bounded by the owner user's permissions
+      security.create_api_key:
+        body: >
+          {
+            "name": "dls_some_user_key",
+            "role_descriptors": {
+              "role-a": {
+                "index": [
+                  {
+                    "names": ["test_security"],
+                    "privileges": ["read"],
+                    "query": "{\"term\": {\"ck\": \"const\"}}"
+                  }
+                ]
+              }
+            }
+          }
+  - match: { name: "dls_some_user_key" }
+  - set: { encoded: login_creds}
+  - do:
+      headers:
+        Authorization: ApiKey ${login_creds}  # dls_some_user's API key sees selected user regardless of the key's role descriptor
+      terms_enum:
+        index: test_security
+        body: { "field": "foo", "string": "b" }
+  - length: { terms: 0 }