Guard reserved tags field against incorrect use

docker/data/logstash/env2yaml/env2yaml.go

@@ -3,16 +3,15 @@
 // Merge environment variables into logstash.yml.
 // For example, running Docker with:
 //
-//   docker run -e pipeline.workers=6
+//	docker run -e pipeline.workers=6
 //
 // or
 //
-//   docker run -e PIPELINE_WORKERS=6
+//	docker run -e PIPELINE_WORKERS=6
 //
 // will cause logstash.yml to contain the line:
 //
-//   pipeline.workers: 6
-//
+//	pipeline.workers: 6
 package main
 
 import (
@@ -72,6 +71,7 @@ func normalizeSetting(setting string) (string, error) {
 		"config.debug",
 		"config.support_escapes",
 		"config.field_reference.escape_style",
+		"event_api.tags.illegal",
 		"queue.type",
 		"path.queue",
 		"queue.page_capacity",

docs/static/settings-file.asciidoc

@@ -346,4 +346,10 @@ separating each log lines per pipeline could be helpful in case you need to trou
 | `allow_superuser`
 | Setting to `true` to allow or `false` to block running Logstash as a superuser.
 | `true`
+
+| `event_api.tags.illegal`
+| When set to `warn`, allow illegal value assignment to the reserved `tags` field.
+When set to `rename`, illegal value in `tags` will be moved to `_tags`. A tag `_tagsparsefailure` is added to `tags` field to indicate the illegal assignment.
+Setting this flag to `warn` is deprecated and will be removed in a future release.
+| `rename`
 |=======================================================================

logstash-core/lib/logstash/environment.rb

@@ -51,6 +51,7 @@ module Environment
          Setting::TimeValue.new("config.reload.interval", "3s"), # in seconds
            Setting::Boolean.new("config.support_escapes", false),
             Setting::String.new("config.field_reference.escape_style", "none", true, %w(none percent ampersand)),
+            Setting::String.new("event_api.tags.illegal", "rename", true, %w(rename warn)),
            Setting::Boolean.new("metric.collect", true),
             Setting::String.new("metric.timers", "delayed", true, %w(delayed live)),
             Setting::String.new("pipeline.id", "main"),

logstash-core/lib/logstash/runner.rb

@@ -91,6 +91,11 @@ class LogStash::Runner < Clamp::StrictCommand
          :default => LogStash::SETTINGS.get_default("config.field_reference.escape_style"),
          :attribute_name => "config.field_reference.escape_style"
 
+  option ["--event_api.tags.illegal"], "STRING",
+         I18n.t("logstash.runner.flag.event_api.tags.illegal"),
+         :default => LogStash::SETTINGS.get_default("event_api.tags.illegal"),
+         :attribute_name => "event_api.tags.illegal"
+
   # Module settings
   option ["--modules"], "MODULES",
     I18n.t("logstash.runner.flag.modules"),
@@ -337,6 +342,12 @@ def execute
     logger.debug("Setting global FieldReference escape style: #{field_reference_escape_style}")
     org.logstash.FieldReference::set_escape_style(field_reference_escape_style)
 
+    tags_illegal_setting = settings.get_setting('event_api.tags.illegal').value
+    if tags_illegal_setting == 'warn'
+      deprecation_logger.deprecated(I18n.t("logstash.runner.tags-illegal-warning"))
+      org.logstash.Event::set_illegal_tags_action(tags_illegal_setting)
+    end
+
     return start_shell(setting("interactive"), binding) if setting("interactive")
 
     module_parser = LogStash::Modules::CLIParser.new(setting("modules_list"), setting("modules_variable_list"))

logstash-core/locales/en.yml

@@ -147,6 +147,9 @@ en:
       configtest-flag-information: |-
         You may be interested in the '--configtest' flag which you can use to validate
         logstash's configuration before you choose to restart a running system.
+      tags-illegal-warning: >-
+        Setting `event_api.tags.illegal` to `warn` allows illegal values in the reserved `tags` field, which may crash pipeline unexpectedly.
+        This flag value is deprecated and may be removed in a future release.
       # YAML named reference to the logstash.runner.configuration
       # so we can later alias it from logstash.agent.configuration
       configuration: &runner_configuration
@@ -252,6 +255,22 @@ en:
              HTML-style ampersand-hash encoding notation
              representing decimal unicode codepoints
              (`[` is `[`; `]` is `]`).
+        event_api:
+          tags:
+            illegal: |+
+              The top-level `tags` field is reserved, and may only contain a
+              single `string` or an array of `string`s -- other values will cause
+              subsequent access of the `tags` field to crash the pipeline.
+              This flag controls how the Event API handles a `tags` field that is
+              an illegal shape, such as a key-value map.
+
+              Available options are:
+               - `rename`: illegal value in `tags` will be moved to `_tags`.
+                 A tag `_tagsparsefailure` is added to `tags` field to 
+                 indicate the illegal assignment. This is the default option.
+               - `warn`: allow illegal value assignment and print warning
+                 at startup. This option is deprecated and slated
+                 for removal.
         modules: |+
           Load Logstash modules.
           Modules can be defined using multiple instances

logstash-core/spec/logstash/filters/base_spec.rb

@@ -379,8 +379,8 @@ def filter(event)
     }
     CONFIG
 
-    sample_one("type" => "noop", "go" => "away", "tags" => {"blackhole" => "go"}) do
-      expect(subject.get("[tags][blackhole]")).to eq("go")
+    sample_one("type" => "noop", "go" => "away", "tags" => "blackhole") do
+      expect(subject.get("[tags]")).to eq("blackhole")
     end
 
   end
@@ -393,8 +393,8 @@ def filter(event)
         }
       CONFIG
 
-      sample_one("type" => "noop", "tags" => {"blackhole" => "go"}) do
-        expect(subject.get("[tags][blackhole]")).to eq("go")
+      sample_one("type" => "noop", "tags" => "blackhole") do
+        expect(subject.get("[tags]")).to eq("blackhole")
       end
     end
   end

logstash-core/src/main/java/org/logstash/Event.java

@@ -36,10 +36,13 @@
 import java.io.Serializable;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.logstash.ObjectMappers.CBOR_MAPPER;
 import static org.logstash.ObjectMappers.JSON_MAPPER;
@@ -62,9 +65,16 @@ public final class Event implements Cloneable, Queueable, co.elastic.logstash.ap
     public static final String VERSION_ONE = "1";
     private static final String DATA_MAP_KEY = "DATA";
     private static final String META_MAP_KEY = "META";
+    public static final String TAGS = "tags";
+    public static final String TAGS_FAILURE_TAG = "_tagsparsefailure";
+    public static final String TAGS_FAILURE = "_tags";
+
+    enum IllegalTagsAction { RENAME, WARN }
+    private static IllegalTagsAction ILLEGAL_TAGS_ACTION = IllegalTagsAction.RENAME;
+
+    private static final FieldReference TAGS_FIELD = FieldReference.from(TAGS);
+    private static final FieldReference TAGS_FAILURE_FIELD = FieldReference.from(TAGS_FAILURE);
 
-    private static final FieldReference TAGS_FIELD = FieldReference.from("tags");
-
     private static final Logger logger = LogManager.getLogger(Event.class);
 
     public Event()
@@ -106,6 +116,15 @@ public Event(ConvertedMap data) {
         }
         this.cancelled = false;
 
+        // guard tags field from key/value map, only string or list is allowed
+        if (ILLEGAL_TAGS_ACTION == IllegalTagsAction.RENAME) {
+            final Object tags = Accessors.get(data, TAGS_FIELD);
+            if (!isLegalTagValue(tags)) {
+                initFailTag(tags);
+                initTag(TAGS_FAILURE_TAG);
+            }
+        }
+
         Object providedTimestamp = data.get(TIMESTAMP);
         // keep reference to the parsedTimestamp for tagging below
         Timestamp parsedTimestamp = initTimestamp(providedTimestamp);
@@ -200,6 +219,13 @@ public void setField(final String reference, final Object value) {
 
     @SuppressWarnings("unchecked")
     public void setField(final FieldReference field, final Object value) {
+        if (ILLEGAL_TAGS_ACTION == IllegalTagsAction.RENAME) {
+            if ((field.equals(TAGS_FIELD) && !isLegalTagValue(value)) ||
+                    isTagsWithMap(field)) {
+                throw new InvalidTypeException(field, value);
+            }
+        }
+
         switch (field.type()) {
             case FieldReference.META_PARENT:
                 // ConvertedMap.newFromMap already does valuefication
@@ -213,6 +239,25 @@ public void setField(final FieldReference field, final Object value) {
         }
     }
 
+    private boolean isTagsWithMap(final FieldReference field) {
+        return field.getPath() != null && field.getPath().length > 0 && field.getPath()[0].equals(TAGS);
+    }
+
+    private boolean isLegalTagValue(final Object value) {
+        if (value instanceof String || value instanceof RubyString || value == null) {
+            return true;
+        } else if (value instanceof List) {
+            for (Object item: (List) value) {
+                if (!(item instanceof String) && !(item instanceof RubyString)) {
+                    return false;
+                }
+            }
+            return true;
+        }
+
+        return false;
+    }
+
     @Override
     public boolean includes(final String field) {
         return includes(FieldReference.from(field));
@@ -483,6 +528,34 @@ private void appendTag(final List<String> tags, final String tag) {
         }
     }
 
+    private void initFailTag(final Object tag) {
+        final ConvertedList list = new ConvertedList(1);
+        list.add(tag);
+        Accessors.set(data, TAGS_FAILURE_FIELD, list);
+    }
+
+    @SuppressWarnings("unchecked")
+    private void tagFailTags(final Object tag) {
+        final Object failTags = Accessors.get(data, TAGS_FAILURE_FIELD);
+        if (failTags == null) {
+            final List<Object> list = new ArrayList<>(1);
+            appendFailTags(list, tag);
+        } else {
+            if (failTags instanceof List) {
+                appendFailTags((List<Object>) failTags, tag);
+            } else {
+                final List<Object> list = new ArrayList<>(2);
+                list.add(failTags);
+                appendFailTags(list, tag);
+            }
+        }
+    }
+
+    private void appendFailTags(final List<Object> failTags, final Object failTag) {
+        failTags.add(failTag);
+        Accessors.set(data, TAGS_FAILURE_FIELD, ConvertedList.newFromList(failTags));
+    }
+
     /**
      * Fallback for {@link Event#tag(String)} in case "tags" was populated by just a String value
      * and needs to be converted to a list before appending to it.
@@ -495,6 +568,14 @@ private void scalarTagFallback(final String existing, final String tag) {
         appendTag(tags, tag);
     }
 
+    public static void setIllegalTagsAction(final String action) {
+        ILLEGAL_TAGS_ACTION = IllegalTagsAction.valueOf(action.toUpperCase());
+    }
+
+    public static IllegalTagsAction getIllegalTagsAction() {
+        return ILLEGAL_TAGS_ACTION;
+    }
+
     @Override
     public byte[] serialize() throws JsonProcessingException {
         final Map<String, Map<String, Object>> map = new HashMap<>(2, 1.0F);
@@ -509,4 +590,25 @@ public static Event deserialize(byte[] data) throws IOException {
         }
         return fromSerializableMap(data);
     }
+
+    public static class InvalidTypeException extends RuntimeException {
+        private static final long serialVersionUID = 1L;
+
+        public InvalidTypeException(final FieldReference field, final Object value) {
+            super(String.format("Could not set the reserved tags field '%s' to value '%s'. " +
+                            "The tags field only accepts string or array of string.",
+                    getCanonicalFieldReference(field), value
+            ));
+        }
+
+        private static String getCanonicalFieldReference(final FieldReference field) {
+            List<String> path = new ArrayList<>();
+            if (field.type() == FieldReference.META_CHILD) {
+                path.add(METADATA);
+            }
+            path.addAll(List.of(field.getPath()));
+            path.add(field.getKey());
+            return path.stream().collect(Collectors.joining("][", "[", "]"));
+        }
+    }
 }

logstash-core/src/main/java/org/logstash/FieldReference.java

@@ -32,6 +32,8 @@
 import org.jruby.RubyString;
 import org.logstash.util.EscapeHandler;
 
+import javax.annotation.Nonnegative;
+
 /**
  * Represents a reference to another field of the event {@link Event}
  * */
@@ -167,6 +169,32 @@ public static boolean isValid(final String reference) {
         }
     }
 
+    /**
+     * Rebase the path of FieldReference to a new base.
+     *  [tags][key] to [_tags][key], rebaseOnto(1, List.of("_tags"))
+     *  [tags][path][key] to [_tags][key], rebaseOnto(2, List.of("_tags"))
+     * @param shiftLevel the number of levels to shift off of this {@code FieldReference}.
+     *                   A value of {@code 0} will _move_ the entire nesting onto the new base
+     *                   A value of {@code 1} will remove 1 top-level nesting before moving the remainder
+     *                   Specifying more levels than available nestings is an error condition.
+     * @param newBase a sequence of zero or more field nesting to prepend
+     * @return FieldReference
+     */
+    public FieldReference rebaseOnto(@Nonnegative int shiftLevel, final List<String> newBase) {
+        if (shiftLevel == 0 && newBase.isEmpty()) { return this; }
+
+        List<String> tokens = toTokens();
+        if (tokens.size() <= shiftLevel) {
+            throw new IndexOutOfBoundsException(String.format("cannot shift %s levels from field reference with %s tokens", shiftLevel, tokens.size()));
+        }
+
+        final List<String> modifiedPath = new ArrayList<>(newBase.size() + tokens.size() - shiftLevel);
+        modifiedPath.addAll(newBase);
+        modifiedPath.addAll(tokens.subList(shiftLevel, tokens.size()));
+
+        return FieldReference.fromTokens(modifiedPath);
+    }
+
     /**
      * Returns the type of this instance to allow for fast switch operations in
      * {@link Event#getUnconvertedField(FieldReference)} and
@@ -245,19 +273,38 @@ private static FieldReference parse(final CharSequence reference) {
                 .map(ESCAPE_HANDLER::unescape)
                 .collect(Collectors.toList());
 
+        return fromTokens(path);
+    }
+
+    private static FieldReference fromTokens(final List<String> path) {
         final String key = path.remove(path.size() - 1);
         final boolean empty = path.isEmpty();
         if (empty && key.equals(Event.METADATA)) {
             return new FieldReference(EMPTY_STRING_ARRAY, key, META_PARENT);
         } else if (!empty && path.get(0).equals(Event.METADATA)) {
             return new FieldReference(
-                path.subList(1, path.size()).toArray(EMPTY_STRING_ARRAY), key, META_CHILD
+                    path.subList(1, path.size()).toArray(EMPTY_STRING_ARRAY), key, META_CHILD
             );
         } else {
             return new FieldReference(path.toArray(EMPTY_STRING_ARRAY), key, DATA_CHILD);
         }
     }
 
+    /**
+     * @return a list of tokens that can round-trip through {@link FieldReference#fromTokens}
+     *         to produce an identical {@code FieldReference}.
+     */
+    private List<String> toTokens() {
+        // min size to avoid resizing: maybe_meta + path + key
+        final List<String> tokens = new ArrayList<>(path.length + 2);
+
+        if (type == META_CHILD) { tokens.add(Event.METADATA); }
+        tokens.addAll(Arrays.asList(path));
+        tokens.add(key);
+
+        return tokens;
+    }
+
     /**
      * The {@link StrictTokenizer} parses field-references in a strict manner; when illegal syntax is encountered,
      * the input is considered ambiguous and the reference is not expanded.