[9.1] Handle ._original and._ignore_malformed stored fields correctly with flu (#137541)

* [9.1] Handle ._original and._ignore_malformed stored fields correctly with fls

Backporting #137442 to 9.1 branch.
* fixed compile error

* [CI] Auto commit changes from spotless

* fixed mocking

---------

Co-authored-by: elasticsearchmachine <[email protected]>

Author: martijnvg

docs/changelog/137442.yaml

@@ -0,0 +1,5 @@
+pr: 137442
+summary: Handle ._original stored fields with fls
+area: "Authorization"
+type: bug
+issues: []

modules/mapper-extras/src/main/java/org/elasticsearch/index/mapper/extras/MatchOnlyTextFieldMapper.java

@@ -208,7 +208,7 @@ public MatchOnlyTextFieldType(
             super(name, true, false, false, tsi, meta);
             this.indexAnalyzer = Objects.requireNonNull(indexAnalyzer);
             this.textFieldType = new TextFieldType(name, isSyntheticSource, syntheticSourceDelegate);
-            this.originalName = isSyntheticSource ? name + "._original" : null;
+            this.originalName = isSyntheticSource ? name + KeywordFieldMapper.FALLBACK_FIELD_NAME_SUFFIX : null;
             this.withinMultiField = withinMultiField;
             this.storedFieldInBinaryFormat = storedFieldInBinaryFormat;
         }

server/src/main/java/org/elasticsearch/index/mapper/IgnoreMalformedStoredValues.java

@@ -26,6 +26,9 @@
  * {@code _source}.
  */
 public abstract class IgnoreMalformedStoredValues {
+
+    public static final String IGNORE_MALFORMED_FIELD_NAME_SUFFIX = "._ignore_malformed";
+
     /**
      * Creates a stored field that stores malformed data to be used in synthetic source.
      * Name of the stored field is original name of the field with added conventional suffix.
@@ -143,6 +146,6 @@ public void reset() {
     }
 
     public static String name(String fieldName) {
-        return fieldName + "._ignore_malformed";
+        return fieldName + IGNORE_MALFORMED_FIELD_NAME_SUFFIX;
     }
 }

server/src/main/java/org/elasticsearch/index/mapper/KeywordFieldMapper.java

@@ -103,6 +103,7 @@ public final class KeywordFieldMapper extends FieldMapper {
 
     public static final String CONTENT_TYPE = "keyword";
     private static final String HOST_NAME = "host.name";
+    public static final String FALLBACK_FIELD_NAME_SUFFIX = "._original";
 
     public static class Defaults {
         public static final FieldType FIELD_TYPE;
@@ -537,7 +538,7 @@ public KeywordFieldType(
             this.isSyntheticSource = isSyntheticSource;
             this.indexSortConfig = builder.indexSortConfig;
             this.hasDocValuesSkipper = DocValuesSkipIndexType.NONE.equals(fieldType.docValuesSkipIndexType()) == false;
-            this.originalName = isSyntheticSource ? name + "._original" : null;
+            this.originalName = isSyntheticSource ? name + FALLBACK_FIELD_NAME_SUFFIX : null;
         }
 
         public KeywordFieldType(String name) {

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

@@ -40,7 +40,9 @@
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.mapper.FieldNamesFieldMapper;
+import org.elasticsearch.index.mapper.IgnoreMalformedStoredValues;
 import org.elasticsearch.index.mapper.IgnoredSourceFieldMapper;
+import org.elasticsearch.index.mapper.KeywordFieldMapper;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.transport.Transports;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -54,6 +56,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 +71,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 isMapped whether a field is mapped or not.
      */
-    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, Boolean> isMapped)
+        throws IOException {
+        return new FieldSubsetDirectoryReader(in, filter, isMapped);
     }
 
     // wraps subreaders with fieldsubsetreaders.
     static class FieldSubsetDirectoryReader extends FilterDirectoryReader {
 
         private final CharacterRunAutomaton filter;
+        private final Function<String, Boolean> isMapped;
 
-        FieldSubsetDirectoryReader(DirectoryReader in, final CharacterRunAutomaton filter) throws IOException {
+        FieldSubsetDirectoryReader(DirectoryReader in, final CharacterRunAutomaton filter, Function<String, Boolean> isMapped)
+            throws IOException {
             super(in, new FilterDirectoryReader.SubReaderWrapper() {
                 @Override
                 public LeafReader wrap(LeafReader reader) {
                     try {
-                        return new FieldSubsetReader(reader, filter);
+                        return new FieldSubsetReader(reader, filter, isMapped);
                     } catch (IOException e) {
                         throw new UncheckedIOException(e);
                     }
                 }
             });
             this.filter = filter;
+            this.isMapped = isMapped;
             verifyNoOtherFieldSubsetDirectoryReaderIsWrapped(in);
         }
 
         @Override
         protected DirectoryReader doWrapDirectoryReader(DirectoryReader in) throws IOException {
-            return new FieldSubsetDirectoryReader(in, filter);
+            return new FieldSubsetDirectoryReader(in, filter, isMapped);
         }
 
         /** Return the automaton that is used to filter fields. */
@@ -133,11 +142,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, Boolean> isMapped) throws IOException {
         super(in);
         ArrayList<FieldInfo> filteredInfos = new ArrayList<>();
         for (FieldInfo fi : in.getFieldInfos()) {
-            if (filter.run(fi.name)) {
+            String name = fi.name;
+            if (fi.getName().endsWith(KeywordFieldMapper.FALLBACK_FIELD_NAME_SUFFIX) && isMapped.apply(fi.getName()) == false) {
+                name = fi.getName().substring(0, fi.getName().length() - KeywordFieldMapper.FALLBACK_FIELD_NAME_SUFFIX.length());
+            }
+            if (fi.getName().endsWith(IgnoreMalformedStoredValues.IGNORE_MALFORMED_FIELD_NAME_SUFFIX)
+                && isMapped.apply(fi.getName()) == false) {
+                name = fi.getName()
+                    .substring(0, fi.getName().length() - IgnoreMalformedStoredValues.IGNORE_MALFORMED_FIELD_NAME_SUFFIX.length());
+            }
+            if (filter.run(name)) {
                 filteredInfos.add(fi);
             }
         }

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

@@ -96,7 +96,10 @@ public DirectoryReader apply(final DirectoryReader reader) {
                 }
             }
 
-            return permissions.getFieldPermissions().filter(wrappedReader);
+            var searchContext = searchExecutionContextProvider.apply(shardId);
+            Function<String, Boolean> isMapped = searchContext::isFieldMapped;
+
+            return permissions.getFieldPermissions().filter(wrappedReader, isMapped);
         } 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, Boolean> isMapped) throws IOException {
         if (hasFieldLevelSecurity() == false) {
             return reader;
         }
-        return FieldSubsetReader.wrap(reader, permittedFieldsAutomaton);
+        return FieldSubsetReader.wrap(reader, permittedFieldsAutomaton, isMapped);
     }
 
     Automaton getIncludeAutomaton() {