Teach field subset reader about multi-fields.

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/accesscontrol/FieldSubsetReader.java

@@ -54,6 +54,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Function;
 
 /**
  * A {@link FilterLeafReader} that exposes only a subset
@@ -68,36 +69,42 @@ public final class FieldSubsetReader extends SequentialStoredFieldsLeafReader {
      * Note that for convenience, the returned reader
      * can be used normally (e.g. passed to {@link DirectoryReader#openIfChanged(DirectoryReader)})
      * and so on.
-     * @param in reader to filter
-     * @param filter fields to filter.
+     *
+     * @param in             reader to filter
+     * @param filter         fields to filter.
+     * @param getParentField
      */
-    public static DirectoryReader wrap(DirectoryReader in, CharacterRunAutomaton filter) throws IOException {
-        return new FieldSubsetDirectoryReader(in, filter);
+    public static DirectoryReader wrap(DirectoryReader in, CharacterRunAutomaton filter, Function<String, String> getParentField)
+        throws IOException {
+        return new FieldSubsetDirectoryReader(in, filter, getParentField);
     }
 
     // wraps subreaders with fieldsubsetreaders.
     static class FieldSubsetDirectoryReader extends FilterDirectoryReader {
 
         private final CharacterRunAutomaton filter;
+        private final Function<String, String> getParentField;
 
-        FieldSubsetDirectoryReader(DirectoryReader in, final CharacterRunAutomaton filter) throws IOException {
+        FieldSubsetDirectoryReader(DirectoryReader in, final CharacterRunAutomaton filter, Function<String, String> getParentField)
+            throws IOException {
             super(in, new FilterDirectoryReader.SubReaderWrapper() {
                 @Override
                 public LeafReader wrap(LeafReader reader) {
                     try {
-                        return new FieldSubsetReader(reader, filter);
+                        return new FieldSubsetReader(reader, filter, getParentField);
                     } catch (IOException e) {
                         throw new UncheckedIOException(e);
                     }
                 }
             });
             this.filter = filter;
+            this.getParentField = getParentField;
             verifyNoOtherFieldSubsetDirectoryReaderIsWrapped(in);
         }
 
         @Override
         protected DirectoryReader doWrapDirectoryReader(DirectoryReader in) throws IOException {
-            return new FieldSubsetDirectoryReader(in, filter);
+            return new FieldSubsetDirectoryReader(in, filter, getParentField);
         }
 
         /** Return the automaton that is used to filter fields. */
@@ -125,6 +132,7 @@ public CacheHelper getReaderCacheHelper() {
 
     /** List of filtered fields */
     private final FieldInfos fieldInfos;
+    private final Function<String, String> getParentField;
     /** An automaton that only accepts authorized fields. */
     private final CharacterRunAutomaton filter;
     /** {@link Terms} cache with filtered stats for the {@link FieldNamesFieldMapper} field. */
@@ -133,15 +141,20 @@ public CacheHelper getReaderCacheHelper() {
     /**
      * Wrap a single segment, exposing a subset of its fields.
      */
-    FieldSubsetReader(LeafReader in, CharacterRunAutomaton filter) throws IOException {
+    FieldSubsetReader(LeafReader in, CharacterRunAutomaton filter, Function<String, String> getParentField) throws IOException {
         super(in);
         ArrayList<FieldInfo> filteredInfos = new ArrayList<>();
         for (FieldInfo fi : in.getFieldInfos()) {
             if (filter.run(fi.name)) {
+                String parentField = getParentField.apply(fi.name);
+                if (parentField != null && filter.run(parentField) == false) {
+                    continue;
+                }
                 filteredInfos.add(fi);
             }
         }
         fieldInfos = new FieldInfos(filteredInfos.toArray(new FieldInfo[filteredInfos.size()]));
+        this.getParentField = getParentField;
         this.filter = filter;
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/accesscontrol/SecurityIndexReaderWrapper.java

@@ -96,7 +96,9 @@ public DirectoryReader apply(final DirectoryReader reader) {
                 }
             }
 
-            return permissions.getFieldPermissions().filter(wrappedReader);
+            var mappingLookup = searchExecutionContextProvider.apply(shardId).getMappingLookup();
+            Function<String, String> getParentField = mappingLookup::parentField;
+            return permissions.getFieldPermissions().filter(wrappedReader, getParentField);
         } catch (IOException e) {
             logger.error("Unable to apply field level security");
             throw ExceptionsHelper.convertToElastic(e);

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

@@ -32,6 +32,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 /**
@@ -243,11 +244,11 @@ public boolean hasFieldLevelSecurity() {
     }
 
     /** Return a wrapped reader that only exposes allowed fields. */
-    public DirectoryReader filter(DirectoryReader reader) throws IOException {
+    public DirectoryReader filter(DirectoryReader reader, Function<String, String> getParentField) throws IOException {
         if (hasFieldLevelSecurity() == false) {
             return reader;
         }
-        return FieldSubsetReader.wrap(reader, permittedFieldsAutomaton);
+        return FieldSubsetReader.wrap(reader, permittedFieldsAutomaton, getParentField);
     }
 
     Automaton getIncludeAutomaton() {