Support multi-intersection for FieldPermissions (#91169)

This PR is the 2nd half of updating DocumentPermissions and FieldPermissions
 to support multi-level of limiting similar to LimitedRole (since #81403). 
Instead of hard coding fieldsDefinition and limitedByFieldsDefinition, 
this PR replaces them with a list of fieldsDefinitions which can accomodate 
multiple of them (more than 2).

Relates: #91151

Author: ywangd

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

@@ -18,7 +18,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.regex.Regex;
-import org.elasticsearch.core.Nullable;
+import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.xpack.core.security.authz.accesscontrol.FieldSubsetReader;
 import org.elasticsearch.xpack.core.security.authz.permission.FieldPermissionsDefinition.FieldGrantExcludeGroup;
 import org.elasticsearch.xpack.core.security.authz.support.SecurityQueryTemplateEvaluator.DlsQueryEvaluationContext;
@@ -61,9 +61,8 @@ public final class FieldPermissions implements Accountable, CacheKey {
         BASE_HASHSET_ENTRY_SIZE = mapEntryShallowSize + 2 * RamUsageEstimator.NUM_BYTES_OBJECT_REF;
     }
 
-    private final FieldPermissionsDefinition fieldPermissionsDefinition;
-    @Nullable
-    private final FieldPermissionsDefinition limitedByFieldPermissionsDefinition;
+    private final List<FieldPermissionsDefinition> fieldPermissionsDefinitions;
+
     // an automaton that represents a union of one more sets of permitted and denied fields
     private final CharacterRunAutomaton permittedFieldsAutomaton;
     private final boolean permittedFieldsAutomatonIsTotal;
@@ -72,7 +71,7 @@ public final class FieldPermissions implements Accountable, CacheKey {
     private final long ramBytesUsed;
 
     /** Constructor that does not enable field-level security: all fields are accepted. */
-    public FieldPermissions() {
+    private FieldPermissions() {
         this(new FieldPermissionsDefinition(null, null), Automatons.MATCH_ALL);
     }
 
@@ -85,33 +84,33 @@ public FieldPermissions(FieldPermissionsDefinition fieldPermissionsDefinition) {
     /** Constructor that enables field-level security based on include/exclude rules. Exclude rules
      *  have precedence over include rules. */
     FieldPermissions(FieldPermissionsDefinition fieldPermissionsDefinition, Automaton permittedFieldsAutomaton) {
-        this(fieldPermissionsDefinition, null, permittedFieldsAutomaton);
+        this(
+            List.of(Objects.requireNonNull(fieldPermissionsDefinition, "field permission definition cannot be null")),
+            permittedFieldsAutomaton
+        );
     }
 
     /** Constructor that enables field-level security based on include/exclude rules. Exclude rules
      *  have precedence over include rules. */
-    private FieldPermissions(
-        FieldPermissionsDefinition fieldPermissionsDefinition,
-        @Nullable FieldPermissionsDefinition limitedByFieldPermissionsDefinition,
-        Automaton permittedFieldsAutomaton
-    ) {
+    private FieldPermissions(List<FieldPermissionsDefinition> fieldPermissionsDefinitions, Automaton permittedFieldsAutomaton) {
         if (permittedFieldsAutomaton.isDeterministic() == false && permittedFieldsAutomaton.getNumStates() > 1) {
             // we only accept deterministic automata so that the CharacterRunAutomaton constructor
             // directly wraps the provided automaton
             throw new IllegalArgumentException("Only accepts deterministic automata");
         }
-        this.fieldPermissionsDefinition = Objects.requireNonNull(fieldPermissionsDefinition, "field permission definition cannot be null");
-        this.limitedByFieldPermissionsDefinition = limitedByFieldPermissionsDefinition;
+        this.fieldPermissionsDefinitions = Objects.requireNonNull(
+            fieldPermissionsDefinitions,
+            "field permission definitions cannot be null"
+        );
         this.originalAutomaton = permittedFieldsAutomaton;
         this.permittedFieldsAutomaton = new CharacterRunAutomaton(permittedFieldsAutomaton);
         // we cache the result of isTotal since this might be a costly operation
         this.permittedFieldsAutomatonIsTotal = Operations.isTotal(permittedFieldsAutomaton);
 
         long ramBytesUsed = BASE_FIELD_PERM_DEF_BYTES;
-        ramBytesUsed += ramBytesUsedForFieldPermissionsDefinition(this.fieldPermissionsDefinition);
-        if (this.limitedByFieldPermissionsDefinition != null) {
-            ramBytesUsed += ramBytesUsedForFieldPermissionsDefinition(this.limitedByFieldPermissionsDefinition);
-        }
+        ramBytesUsed += this.fieldPermissionsDefinitions.stream()
+            .mapToLong(FieldPermissions::ramBytesUsedForFieldPermissionsDefinition)
+            .sum();
         ramBytesUsed += permittedFieldsAutomaton.ramBytesUsed();
         ramBytesUsed += runAutomatonRamBytesUsed(permittedFieldsAutomaton);
         this.ramBytesUsed = ramBytesUsed;
@@ -199,12 +198,16 @@ public static Automaton buildPermittedFieldsAutomaton(final String[] grantedFiel
      */
     public FieldPermissions limitFieldPermissions(FieldPermissions limitedBy) {
         if (hasFieldLevelSecurity() && limitedBy != null && limitedBy.hasFieldLevelSecurity()) {
+            // TODO: cache the automaton computation with FieldPermissionsCache
             Automaton _permittedFieldsAutomaton = Automatons.intersectAndMinimize(getIncludeAutomaton(), limitedBy.getIncludeAutomaton());
-            return new FieldPermissions(fieldPermissionsDefinition, limitedBy.fieldPermissionsDefinition, _permittedFieldsAutomaton);
+            return new FieldPermissions(
+                CollectionUtils.concatLists(fieldPermissionsDefinitions, limitedBy.fieldPermissionsDefinitions),
+                _permittedFieldsAutomaton
+            );
         } else if (limitedBy != null && limitedBy.hasFieldLevelSecurity()) {
-            return new FieldPermissions(limitedBy.getFieldPermissionsDefinition(), limitedBy.getIncludeAutomaton());
+            return new FieldPermissions(limitedBy.fieldPermissionsDefinitions, limitedBy.getIncludeAutomaton());
         } else if (hasFieldLevelSecurity()) {
-            return new FieldPermissions(this.getFieldPermissionsDefinition(), getIncludeAutomaton());
+            return new FieldPermissions(fieldPermissionsDefinitions, getIncludeAutomaton());
         }
         return FieldPermissions.DEFAULT;
     }
@@ -217,23 +220,13 @@ public boolean grantsAccessTo(String fieldName) {
         return permittedFieldsAutomatonIsTotal || permittedFieldsAutomaton.run(fieldName);
     }
 
-    public FieldPermissionsDefinition getFieldPermissionsDefinition() {
-        return fieldPermissionsDefinition;
-    }
-
-    public FieldPermissionsDefinition getLimitedByFieldPermissionsDefinition() {
-        return limitedByFieldPermissionsDefinition;
+    public List<FieldPermissionsDefinition> getFieldPermissionsDefinitions() {
+        return fieldPermissionsDefinitions;
     }
 
     @Override
     public void buildCacheKey(StreamOutput out, DlsQueryEvaluationContext context) throws IOException {
-        fieldPermissionsDefinition.buildCacheKey(out, context);
-        if (limitedByFieldPermissionsDefinition != null) {
-            out.writeBoolean(true);
-            limitedByFieldPermissionsDefinition.buildCacheKey(out, context);
-        } else {
-            out.writeBoolean(false);
-        }
+        out.writeCollection(fieldPermissionsDefinitions, (o, fpd) -> fpd.buildCacheKey(o, context));
     }
 
     /** Return whether field-level security is enabled, ie. whether any field might be filtered out. */
@@ -259,13 +252,12 @@ public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) return false;
         FieldPermissions that = (FieldPermissions) o;
         return permittedFieldsAutomatonIsTotal == that.permittedFieldsAutomatonIsTotal
-            && fieldPermissionsDefinition.equals(that.fieldPermissionsDefinition)
-            && Objects.equals(limitedByFieldPermissionsDefinition, that.limitedByFieldPermissionsDefinition);
+            && fieldPermissionsDefinitions.equals(that.fieldPermissionsDefinitions);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(fieldPermissionsDefinition, limitedByFieldPermissionsDefinition, permittedFieldsAutomatonIsTotal);
+        return Objects.hash(fieldPermissionsDefinitions, permittedFieldsAutomatonIsTotal);
     }
 
     @Override

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

@@ -72,26 +72,25 @@ public FieldPermissions getFieldPermissions(FieldPermissionsDefinition fieldPerm
     }
 
     /**
-     * Returns a field permissions object that corresponds to the merging of the given field permissions and caches the instance if one was
-     * not found in the cache.
+     * Returns a field permissions object that corresponds to the union of the given field permissions.
+     * Union means a field is granted if it is granted by any of the FieldPermissions from the given
+     * collection.
+     * The returned instance is cached if one was not found in the cache.
      */
-    FieldPermissions getFieldPermissions(Collection<FieldPermissions> fieldPermissionsCollection) {
+    FieldPermissions union(Collection<FieldPermissions> fieldPermissionsCollection) {
         Optional<FieldPermissions> allowAllFieldPermissions = fieldPermissionsCollection.stream()
             .filter(((Predicate<FieldPermissions>) (FieldPermissions::hasFieldLevelSecurity)).negate())
             .findFirst();
         return allowAllFieldPermissions.orElseGet(() -> {
             final Set<FieldGrantExcludeGroup> fieldGrantExcludeGroups = new HashSet<>();
             for (FieldPermissions fieldPermissions : fieldPermissionsCollection) {
-                final FieldPermissionsDefinition definition = fieldPermissions.getFieldPermissionsDefinition();
-                final FieldPermissionsDefinition limitedByDefinition = fieldPermissions.getLimitedByFieldPermissionsDefinition();
-                if (definition == null) {
-                    throw new IllegalArgumentException("Expected field permission definition, but found null");
-                } else if (limitedByDefinition != null) {
+                final List<FieldPermissionsDefinition> fieldPermissionsDefinitions = fieldPermissions.getFieldPermissionsDefinitions();
+                if (fieldPermissionsDefinitions.size() != 1) {
                     throw new IllegalArgumentException(
-                        "Expected no limited-by field permission definition, but found [" + limitedByDefinition + "]"
+                        "Expected a single field permission definition, but found [" + fieldPermissionsDefinitions + "]"
                     );
                 }
-                fieldGrantExcludeGroups.addAll(definition.getFieldGrantExcludeGroups());
+                fieldGrantExcludeGroups.addAll(fieldPermissionsDefinitions.get(0).getFieldGrantExcludeGroups());
             }
             final FieldPermissionsDefinition combined = new FieldPermissionsDefinition(fieldGrantExcludeGroups);
             try {

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

@@ -494,7 +494,7 @@ && containsPrivilegeThatGrantsMappingUpdatesForBwc(group))) {
             if (indexFieldPermissions != null && indexFieldPermissions.isEmpty() == false) {
                 fieldPermissions = indexFieldPermissions.size() == 1
                     ? indexFieldPermissions.iterator().next()
-                    : fieldPermissionsCache.getFieldPermissions(indexFieldPermissions);
+                    : fieldPermissionsCache.union(indexFieldPermissions);
             } else {
                 fieldPermissions = FieldPermissions.DEFAULT;
             }
@@ -609,6 +609,7 @@ public static class Group {
         private final String[] indices;
         private final StringMatcher indexNameMatcher;
         private final Supplier<Automaton> indexNameAutomaton;
+        // TODO: Use FieldPermissionsDefinition instead of FieldPermissions. The former is a better counterpart to query
         private final FieldPermissions fieldPermissions;
         private final Set<BytesReference> query;
         // by default certain restricted indices are exempted when granting privileges, as they should generally be hidden for ordinary

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/accesscontrol/SecurityIndexReaderWrapperIntegrationTests.java

@@ -159,7 +159,7 @@ public void testDLS() throws Exception {
         for (int i = 0; i < numValues; i++) {
             String termQuery = "{\"term\": {\"field\": \"" + values[i] + "\"} }";
             IndicesAccessControl.IndexAccessControl indexAccessControl = new IndicesAccessControl.IndexAccessControl(
-                new FieldPermissions(),
+                FieldPermissions.DEFAULT,
                 DocumentPermissions.filteredBy(singleton(new BytesArray(termQuery)))
             );
             SecurityIndexReaderWrapper wrapper = new SecurityIndexReaderWrapper(
@@ -224,15 +224,15 @@ public void testDLSWithLimitedPermissions() throws Exception {
         queries.add(new BytesArray("{\"terms\" : { \"f2\" : [\"fv22\"] } }"));
         queries.add(new BytesArray("{\"terms\" : { \"f2\" : [\"fv32\"] } }"));
         IndicesAccessControl.IndexAccessControl indexAccessControl = new IndicesAccessControl.IndexAccessControl(
-            new FieldPermissions(),
+            FieldPermissions.DEFAULT,
             DocumentPermissions.filteredBy(queries)
         );
         queries = singleton(new BytesArray("{\"terms\" : { \"f1\" : [\"fv11\", \"fv21\", \"fv31\"] } }"));
         if (restrictiveLimitedIndexPermissions) {
             queries = singleton(new BytesArray("{\"terms\" : { \"f1\" : [\"fv11\", \"fv31\"] } }"));
         }
         IndicesAccessControl.IndexAccessControl limitedIndexAccessControl = new IndicesAccessControl.IndexAccessControl(
-            new FieldPermissions(),
+            FieldPermissions.DEFAULT,
             DocumentPermissions.filteredBy(queries)
         );
         IndexSettings indexSettings = IndexSettingsModule.newIndexSettings(

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

@@ -44,9 +44,7 @@ public void testMergeFieldPermissions() {
         FieldPermissions fieldPermissions2 = randomBoolean()
             ? fieldPermissionsCache.getFieldPermissions(allowed2, denied2)
             : new FieldPermissions(fieldPermissionDef(allowed2, denied2));
-        FieldPermissions mergedFieldPermissions = fieldPermissionsCache.getFieldPermissions(
-            Arrays.asList(fieldPermissions1, fieldPermissions2)
-        );
+        FieldPermissions mergedFieldPermissions = fieldPermissionsCache.union(Arrays.asList(fieldPermissions1, fieldPermissions2));
         assertTrue(mergedFieldPermissions.grantsAccessTo(allowedPrefix1 + "b"));
         assertTrue(mergedFieldPermissions.grantsAccessTo(allowedPrefix2 + "b"));
         assertFalse(mergedFieldPermissions.grantsAccessTo(denied1[0]));
@@ -62,7 +60,7 @@ public void testMergeFieldPermissions() {
         fieldPermissions2 = randomBoolean()
             ? fieldPermissionsCache.getFieldPermissions(allowed2, denied2)
             : new FieldPermissions(fieldPermissionDef(allowed2, denied2));
-        mergedFieldPermissions = fieldPermissionsCache.getFieldPermissions(Arrays.asList(fieldPermissions1, fieldPermissions2));
+        mergedFieldPermissions = fieldPermissionsCache.union(Arrays.asList(fieldPermissions1, fieldPermissions2));
         assertFalse(mergedFieldPermissions.hasFieldLevelSecurity());
 
         allowed1 = new String[] {};
@@ -75,7 +73,7 @@ public void testMergeFieldPermissions() {
         fieldPermissions2 = randomBoolean()
             ? fieldPermissionsCache.getFieldPermissions(allowed2, denied2)
             : new FieldPermissions(fieldPermissionDef(allowed2, denied2));
-        mergedFieldPermissions = fieldPermissionsCache.getFieldPermissions(Arrays.asList(fieldPermissions1, fieldPermissions2));
+        mergedFieldPermissions = fieldPermissionsCache.union(Arrays.asList(fieldPermissions1, fieldPermissions2));
         for (String field : allowed2) {
             assertTrue(mergedFieldPermissions.grantsAccessTo(field));
         }
@@ -93,7 +91,7 @@ public void testMergeFieldPermissions() {
         fieldPermissions2 = randomBoolean()
             ? fieldPermissionsCache.getFieldPermissions(allowed2, denied2)
             : new FieldPermissions(fieldPermissionDef(allowed2, denied2));
-        mergedFieldPermissions = fieldPermissionsCache.getFieldPermissions(Arrays.asList(fieldPermissions1, fieldPermissions2));
+        mergedFieldPermissions = fieldPermissionsCache.union(Arrays.asList(fieldPermissions1, fieldPermissions2));
         assertTrue(mergedFieldPermissions.grantsAccessTo("a"));
         assertTrue(mergedFieldPermissions.grantsAccessTo("b"));
 
@@ -107,7 +105,7 @@ public void testMergeFieldPermissions() {
         fieldPermissions2 = randomBoolean()
             ? fieldPermissionsCache.getFieldPermissions(allowed2, denied2)
             : new FieldPermissions(fieldPermissionDef(allowed2, denied2));
-        mergedFieldPermissions = fieldPermissionsCache.getFieldPermissions(Arrays.asList(fieldPermissions1, fieldPermissions2));
+        mergedFieldPermissions = fieldPermissionsCache.union(Arrays.asList(fieldPermissions1, fieldPermissions2));
         assertTrue(mergedFieldPermissions.grantsAccessTo("a"));
         assertTrue(mergedFieldPermissions.grantsAccessTo("b"));
         assertFalse(mergedFieldPermissions.grantsAccessTo("aa"));
@@ -126,7 +124,7 @@ public void testNonFlsAndFlsMerging() {
         FieldPermissionsCache cache = new FieldPermissionsCache(Settings.EMPTY);
         for (int i = 0; i < scaledRandomIntBetween(1, 12); i++) {
             Collections.shuffle(permissionsList, random());
-            FieldPermissions result = cache.getFieldPermissions(permissionsList);
+            FieldPermissions result = cache.union(permissionsList);
             assertFalse(result.hasFieldLevelSecurity());
         }
     }