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.
+This flag will be deprecated in the next major release and the `tags` field will only allow a string or an array of string.
+| `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'
+      logger.warn(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,21 @@ 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.
         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

@@ -40,6 +40,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import static org.logstash.ObjectMappers.CBOR_MAPPER;
 import static org.logstash.ObjectMappers.JSON_MAPPER;
@@ -62,9 +63,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 +114,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)) {
+                tagFailTags(tags);
+                initTag(TAGS_FAILURE_TAG);
+            }
+        }
+
         Object providedTimestamp = data.get(TIMESTAMP);
         // keep reference to the parsedTimestamp for tagging below
         Timestamp parsedTimestamp = initTimestamp(providedTimestamp);
@@ -200,6 +217,16 @@ 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)) {
+                setTagsAndFailTags(value);
+                return;
+            } else if (isTagsWithMap(field)) {
+                setTagsAndFailTagsWithMap(field, value);
+                return;
+            }
+        }
+
         switch (field.type()) {
             case FieldReference.META_PARENT:
                 // ConvertedMap.newFromMap already does valuefication
@@ -213,6 +240,62 @@ 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;
+    }
+
+
+    private void setTagsAndFailTags(final Object value) {
+        tag(TAGS_FAILURE_TAG);
+        tagFailTags(Valuefier.convert(value));
+    }
+
+    /**
+     * handle key/val pair in `tags`, add _tagsparsefailure to `tags`.
+     * rename FieldReference from `tags` to `_tags`. If existing `_tags` is a list, append value to `_tags`,
+     * eg. [tags][key] -> [_tags][list_index][key]
+     * @param field FieldReference of `tags` with map path
+     * @param value
+     */
+    private void setTagsAndFailTagsWithMap(final FieldReference field, final Object value) {
+        tag(TAGS_FAILURE_TAG);
+
+        List<String> paths = new ArrayList<>();
+        paths.add(TAGS_FAILURE);
+
+        // take list size of _tags as index
+        final Object failTags = Accessors.get(data, TAGS_FAILURE_FIELD);
+        int index = (failTags instanceof List)? ((List) failTags).size() : 0;
+        paths.add(Integer.toString(index));
+
+        // rebuild path & key
+        for(int i = 1; i < field.getPath().length; i++) {
+            paths.add(field.getPath()[i]);
+        }
+        if (field.getKey() != null) {
+            paths.add(field.getKey());
+        }
+
+        String renamedFieldRef = paths.stream().collect(Collectors.joining("][", "[", "]"));
+        final FieldReference renamedField = FieldReference.from(renamedFieldRef);
+        Accessors.set(data, renamedField, Valuefier.convert(value));
+    }
+
     @Override
     public boolean includes(final String field) {
         return includes(FieldReference.from(field));
@@ -483,6 +566,28 @@ private void appendTag(final List<String> tags, final String tag) {
         }
     }
 
+    @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 +600,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);