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`, Logstash events can't be created with an illegal value in `tags`. This value will be moved to `_tags` and a `_tagsparsefailure` tag is added to indicate the illegal operation. Doing `set` operation with illegal value will throw exception.
+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,23 @@ 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. Doing `set` operation with 
+                 illegal value will throw exception. 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

@@ -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)) {
+                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 +217,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 InvalidTagsTypeException(field, value);
+            }
+        }
+
         switch (field.type()) {
             case FieldReference.META_PARENT:
                 // ConvertedMap.newFromMap already does valuefication
@@ -213,6 +237,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 +526,12 @@ 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);
+    }
+
     /**
      * 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 +544,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 +566,21 @@ public static Event deserialize(byte[] data) throws IOException {
         }
         return fromSerializableMap(data);
     }
+
+    public static class InvalidTagsTypeException extends RuntimeException {
+        private static final long serialVersionUID = 1L;
+
+        public InvalidTagsTypeException(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<>(List.of(field.getPath()));
+            path.add(field.getKey());
+            return path.stream().collect(Collectors.joining("][", "[", "]"));
+        }
+    }
 }

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

@@ -245,13 +245,17 @@ 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);

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

@@ -43,6 +43,7 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.logstash.Event.getIllegalTagsAction;
 
 public final class EventTest extends RubyTestBase {
 
@@ -516,4 +517,87 @@ public void removeMetadata() {
         assertTrue(event.getMetadata().isEmpty());
         assertFalse(event.includes("[@metadata][foo]"));
     }
+
+    @Test(expected = Event.InvalidTagsTypeException.class)
+    public void setTagsWithMapShouldThrow() {
+        final Event event = new Event();
+        event.setField("[tags][foo]", "bar");
+    }
+
+    @Test
+    public void createEventWithTagsWithMapShouldRename() {
+        final Event event = new Event(Map.of("tags", Map.of("poison", "true")));
+
+        assertEquals(event.getField(Event.TAGS), Collections.singletonList(Event.TAGS_FAILURE_TAG));
+        assertEquals(event.getField("[_tags][0][poison]"), "true");
+    }
+
+    @Test(expected = Event.InvalidTagsTypeException.class)
+    public void setTagsWithNumberShouldThrow() {
+        final Event event = new Event();
+        event.setField("[tags]", 123L);
+    }
+
+    @Test
+    public void allowTopLevelTagsString() {
+        final Event event = new Event();
+        event.setField("[tags]", "bar");
+
+        assertNull(event.getField(Event.TAGS_FAILURE));
+        assertEquals(event.getField("[tags]"), "bar");
+
+        event.setField("[tags]", "foo");
+        assertEquals(event.getField("[tags]"), "foo");
+    }
+
+    @Test
+    public void createEventWithoutTagShouldHaveEmptyTags() {
+        final Event event = new Event(Map.of("world", "cup"));
+        assertNull(event.getField(Event.TAGS));
+        assertNull(event.getField(Event.TAGS_FAILURE));
+    }
+
+    @Test
+    public void allowTopLevelTagsListOfStrings() {
+        final Event event = new Event();
+        event.setField("[tags]", List.of("foo", "bar"));
+
+        assertNull(event.getField(Event.TAGS_FAILURE));
+        assertEquals(event.getField("[tags]"), List.of("foo", "bar"));
+    }
+
+    @Test
+    public void allowTopLevelTagsWithMap() {
+        withIllegalTagsAction(Event.IllegalTagsAction.WARN, () -> {
+            final Event event = new Event();
+            event.setField("[tags][foo]", "bar");
+
+            assertNull(event.getField(Event.TAGS_FAILURE));
+            assertEquals(event.getField("[tags][foo]"), "bar");
+        });
+    }
+
+    @Test
+    public void allowCreatingEventWithTopLevelTagsWithMap() {
+        withIllegalTagsAction(Event.IllegalTagsAction.WARN, () -> {
+            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);
+
+            assertNull(event.getField(Event.TAGS_FAILURE));
+            assertEquals(event.getField("[tags][poison]"), "true");
+        });
+    }
+
+    private void withIllegalTagsAction(final Event.IllegalTagsAction temporaryIllegalTagsAction, final Runnable runnable) {
+        final Event.IllegalTagsAction previous = getIllegalTagsAction();
+        try {
+            Event.setIllegalTagsAction(temporaryIllegalTagsAction.toString());
+            runnable.run();
+        } finally {
+            Event.setIllegalTagsAction(previous.toString());
+        }
+    }
 }