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 will be deprecated in the next major release and `tags` field will only allow a string or an array of string.
       # YAML named reference to the logstash.runner.configuration
       # so we can later alias it from logstash.agent.configuration
       configuration: &runner_configuration
@@ -252,6 +255,19 @@ en:
              HTML-style ampersand-hash encoding notation
              representing decimal unicode codepoints
              (`[` is `[`; `]` is `]`).
+        event_api:
+          tags:
+            illegal: |+
+              The reserved field `tags` allow string and string[] since v7.9.
+              Assigning other types could crash the pipeline. This allows you to
+              change the behavior when `tags` get assigned with illegal type, eg. 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

@@ -36,6 +36,7 @@
 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;
@@ -62,8 +63,14 @@ 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_FIELD = "_tags";
 
-    private static final FieldReference TAGS_FIELD = FieldReference.from("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 Logger logger = LogManager.getLogger(Event.class);
 
@@ -106,6 +113,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 (tags instanceof ConvertedMap) {
+                this.setField(TAGS_FAILURE_FIELD, tags);
+                initTag(TAGS_FAILURE_TAG);
+            }
+        }
+
         Object providedTimestamp = data.get(TIMESTAMP);
         // keep reference to the parsedTimestamp for tagging below
         Timestamp parsedTimestamp = initTimestamp(providedTimestamp);
@@ -209,8 +225,28 @@ public void setField(final FieldReference field, final Object value) {
                 Accessors.set(metadata, field, Valuefier.convert(value));
                 break;
             default:
-                Accessors.set(data, field, Valuefier.convert(value));
+                final FieldReference renamedField = renameIllegalTags(field);
+                Accessors.set(data, renamedField, Valuefier.convert(value));
+        }
+    }
+
+    /**
+     * if the field is a reserved `tags` field with map structure,
+     * construct a FieldReference pointing to `_tag` field and add _tagsparsefailure to `tags` field,
+     * otherwise return the original field.
+     * @param field
+     * @return Renamed {@link FieldReference} or original field
+     */
+    private FieldReference renameIllegalTags(final FieldReference field) {
+        if (ILLEGAL_TAGS_ACTION == IllegalTagsAction.RENAME &&
+                field.getPath() != null && field.getPath().length > 0 && field.getPath()[0].equals(TAGS)) {
+            tag(TAGS_FAILURE_TAG);
+            String[] failTagsPath = Arrays.copyOf(field.getPath(), field.getPath().length);
+            failTagsPath[0] = TAGS_FAILURE_FIELD;
+            return new FieldReference(failTagsPath, field.getKey(), field.type());
         }
+
+        return field;
     }
 
     @Override
@@ -495,6 +531,10 @@ 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());
+    }
+
     @Override
     public byte[] serialize() throws JsonProcessingException {
         final Map<String, Map<String, Object>> map = new HashMap<>(2, 1.0F);

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

@@ -126,7 +126,7 @@ public static void setEscapeStyle(final String escapeStyleSpec) {
      */
     private final int type;
 
-    private FieldReference(final String[] path, final String key, final int type) {
+    public FieldReference(final String[] path, final String key, final int type) {
         this.key = ConvertedMap.internStringForUseAsKey(key);
         ConvertedMap.internStringsForUseAsKeys(path);
         this.path = path;

logstash-core/src/test/java/org/logstash/EventTest.java

@@ -516,4 +516,53 @@ public void removeMetadata() {
         assertTrue(event.getMetadata().isEmpty());
         assertFalse(event.includes("[@metadata][foo]"));
     }
+
+    @Test
+    public void setTopLevelTagsWithMap() {
+        Event.setIllegalTagsAction(Event.IllegalTagsAction.RENAME.toString());
+        final Event event = new Event();
+        event.setField("[tags][foo]", "bar");
+
+        assertEquals(event.getField(Event.TAGS), Collections.singletonList(Event.TAGS_FAILURE_TAG));
+        assertEquals(event.getField("[_tags][foo]"), "bar");
+    }
+
+    @Test
+    public void createEventWithTopLevelTagsWithMap() {
+        Event.setIllegalTagsAction(Event.IllegalTagsAction.RENAME.toString());
+        Map<String, Object> inner = new HashMap<>();
+        inner.put("poison", "true");
+        Map<String, Object> data = new HashMap<>();
+        data.put("tags", inner);
+        final Event event = new Event(data);
+
+        assertEquals(event.getField(Event.TAGS), Collections.singletonList(Event.TAGS_FAILURE_TAG));
+        assertEquals(event.getField("[_tags][poison]"), "true");
+    }
+
+
+    @Test
+    public void allowTopLevelTagsWithMap() {
+        Event.setIllegalTagsAction(Event.IllegalTagsAction.WARN.toString());
+        final Event event = new Event();
+        event.setField("[tags][foo]", "bar");
+        Event.setIllegalTagsAction(Event.IllegalTagsAction.RENAME.toString());
+
+        assertNull(event.getField(Event.TAGS_FAILURE_FIELD));
+        assertEquals(event.getField("[tags][foo]"), "bar");
+    }
+
+    @Test
+    public void allowCreatingEventWithTopLevelTagsWithMap() {
+        Event.setIllegalTagsAction(Event.IllegalTagsAction.WARN.toString());
+        Map<String, Object> inner = new HashMap<>();
+        inner.put("poison", "true");
+        Map<String, Object> data = new HashMap<>();
+        data.put("tags", inner);
+        final Event event = new Event(data);
+        Event.setIllegalTagsAction(Event.IllegalTagsAction.RENAME.toString());
+
+        assertNull(event.getField(Event.TAGS_FAILURE_FIELD));
+        assertEquals(event.getField("[tags][poison]"), "true");
+    }
 }