[9.2] Handle ._original and._ignore_malformed stored fields correctly with fls (#137527)

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

Backporting #137442 to 9.2 branch.

* fixed compile error

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;
@@ -571,7 +572,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

@@ -41,7 +41,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;
@@ -55,6 +57,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
@@ -69,46 +72,52 @@ 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,
-        IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat
+        IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat,
+        Function<String, Boolean> isMapped
     ) throws IOException {
-        return new FieldSubsetDirectoryReader(in, filter, ignoredSourceFormat);
+        return new FieldSubsetDirectoryReader(in, filter, ignoredSourceFormat, isMapped);
     }
 
     // wraps subreaders with fieldsubsetreaders.
     static class FieldSubsetDirectoryReader extends FilterDirectoryReader {
 
         private final CharacterRunAutomaton filter;
         private final IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat;
+        private final Function<String, Boolean> isMapped;
 
         FieldSubsetDirectoryReader(
             DirectoryReader in,
             final CharacterRunAutomaton filter,
-            final IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat
+            final IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat,
+            Function<String, Boolean> isMapped
         ) throws IOException {
             super(in, new FilterDirectoryReader.SubReaderWrapper() {
                 @Override
                 public LeafReader wrap(LeafReader reader) {
                     try {
-                        return new FieldSubsetReader(reader, filter, ignoredSourceFormat);
+                        return new FieldSubsetReader(reader, filter, ignoredSourceFormat, isMapped);
                     } catch (IOException e) {
                         throw new UncheckedIOException(e);
                     }
                 }
             });
             this.filter = filter;
             this.ignoredSourceFormat = ignoredSourceFormat;
+            this.isMapped = isMapped;
             verifyNoOtherFieldSubsetDirectoryReaderIsWrapped(in);
         }
 
         @Override
         protected DirectoryReader doWrapDirectoryReader(DirectoryReader in) throws IOException {
-            return new FieldSubsetDirectoryReader(in, filter, ignoredSourceFormat);
+            return new FieldSubsetDirectoryReader(in, filter, ignoredSourceFormat, isMapped);
         }
 
         /** Return the automaton that is used to filter fields. */
@@ -145,12 +154,25 @@ public CacheHelper getReaderCacheHelper() {
     /**
      * Wrap a single segment, exposing a subset of its fields.
      */
-    FieldSubsetReader(LeafReader in, CharacterRunAutomaton filter, IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat)
-        throws IOException {
+    FieldSubsetReader(
+        LeafReader in,
+        CharacterRunAutomaton filter,
+        IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat,
+        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,9 +96,11 @@ public DirectoryReader apply(final DirectoryReader reader) {
                 }
             }
 
-            var indexVersionCreated = searchExecutionContextProvider.apply(shardId).indexVersionCreated();
+            var searchContext = searchExecutionContextProvider.apply(shardId);
+            var indexVersionCreated = searchContext.indexVersionCreated();
+            Function<String, Boolean> isMapped = searchContext::isFieldMapped;
 
-            return permissions.getFieldPermissions().filter(wrappedReader, indexVersionCreated);
+            return permissions.getFieldPermissions().filter(wrappedReader, indexVersionCreated, 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

@@ -34,6 +34,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 /**
@@ -245,11 +246,17 @@ public boolean hasFieldLevelSecurity() {
     }
 
     /** Return a wrapped reader that only exposes allowed fields. */
-    public DirectoryReader filter(DirectoryReader reader, IndexVersion indexVersionCreated) throws IOException {
+    public DirectoryReader filter(DirectoryReader reader, IndexVersion indexVersionCreated, Function<String, Boolean> isMapped)
+        throws IOException {
         if (hasFieldLevelSecurity() == false) {
             return reader;
         }
-        return FieldSubsetReader.wrap(reader, permittedFieldsAutomaton, IgnoredSourceFieldMapper.ignoredSourceFormat(indexVersionCreated));
+        return FieldSubsetReader.wrap(
+            reader,
+            permittedFieldsAutomaton,
+            IgnoredSourceFieldMapper.ignoredSourceFormat(indexVersionCreated),
+            isMapped
+        );
     }
 
     Automaton getIncludeAutomaton() {