entry_mutator: keep trusted fields when stripping underscores

README.md

@@ -143,6 +143,13 @@ If true, lowercase all non-mapped fields. Defaults to false.
 
 If true, strip leading underscores from all non-mapped fields. Defaults to false.
 
+Journald reserves the leading underscore for its trusted fields, which a client cannot forge,
+ while a client is free to send a user field with the same name minus the underscore.
+Stripping makes both names collide, so the trusted field wins and the user field of that name is dropped.
+For example, if an entry holds `_SYSTEMD_UNIT` and a client supplied `SYSTEMD_UNIT`,
+the result only holds the trusted value.
+Map the trusted field to another name with `field_map` if you need to keep both.
+
 ### Filter Example
 
 Given a systemd journal source entry:

lib/fluent/plugin/systemd/entry_mutator.rb

@@ -15,6 +15,7 @@
 #   limitations under the License.
 
 require 'fluent/config/error'
+require 'systemd/journal/fields'
 
 module Fluent
   module Plugin
@@ -33,14 +34,16 @@ module Plugin
     #   "<new_field1>" => ["<source_field1>", "<source_field2>"],
     #   "<new_field2>" => ["<source_field2>"]
     # }
-    class SystemdEntryMutator
+    class SystemdEntryMutator # rubocop:disable Metrics/ClassLength
       Options = Struct.new(
         :field_map,
         :field_map_strict,
         :fields_lowercase,
         :fields_strip_underscores
       )
 
+      TRUSTED_FIELDS = (Systemd::Journal::TRUSTED_FIELDS + Systemd::Journal::KERNEL_FIELDS).freeze
+
       def self.default_opts
         Options.new({}, false, false, false)
       end
@@ -103,13 +106,20 @@ def map_fields(entry)
       # mapped - Optional hash that represents a previously mapped entry to
       #          which the formatted fields will be added
       def format_fields(entry, mapped = nil)
+        reserved = reserved_field_names(entry)
         entry.each_with_object(mapped || {}) do |(fld, val), formatted_entry|
           # don't mess with explicitly mapped fields
           next if @map_src_fields.include?(fld)
 
-          fld = format_field_name(fld)
+          name = format_field_name(fld)
+          # A client may send `SYSTEMD_UNIT` but never `_SYSTEMD_UNIT`, so the
+          # trusted field keeps the name when stripping underscores makes the
+          # two collide. Otherwise any local process could fake the journal
+          # metadata that the trusted fields are supposed to guarantee.
+          next if !fld.start_with?('_') && reserved.include?(name)
+
           # account for mapping (appending) to an existing systemd field
-          formatted_entry[fld] = join_if_needed([val, mapped[fld]])
+          formatted_entry[name] = join_if_needed([val, mapped[name]])
         end
       end
 
@@ -128,6 +138,25 @@ def join_if_needed(values)
         values.join(' ')
       end
 
+      def reserved_field_names(entry)
+        return [] unless @opts.fields_strip_underscores
+
+        trusted_field_names(entry).each_with_object([]) do |fld, names|
+          name = format_field_name(fld)
+          next if @map_src_fields.include?(fld) && !Array(@opts.field_map[fld]).include?(name)
+
+          names << name
+        end
+      end
+
+      # Journald never lets a client send a leading underscore, so the known
+      # trusted names stay reserved even when this entry does not carry them.
+      def trusted_field_names(entry)
+        entry.each_with_object(TRUSTED_FIELDS.dup) do |(fld, _val), flds|
+          flds << fld if fld.start_with?('_')
+        end
+      end
+
       def format_field_name(name)
         name = name.gsub(/\A_+/, '') if @opts.fields_strip_underscores
         name = name.downcase if @opts.fields_lowercase

test/plugin/systemd/test_entry_mutator.rb

@@ -41,6 +41,15 @@ class EntryTestData
   # string json form of `FIELD_MAP`
   FIELD_MAP_JSON = JSON.generate(FIELD_MAP).freeze
 
+  # entry where a client sent user fields named after the trusted fields
+  SPOOFED_ENTRY = {
+    '_SYSTEMD_UNIT' => 'user@1000.service',
+    'SYSTEMD_UNIT' => 'sshd.service',
+    '_PID' => '777',
+    'PID' => '1',
+    'MESSAGE' => 'Accepted publickey for root'
+  }.freeze
+
   # expected entry mutation results
   EXPECTED = {
     no_transform: {
@@ -175,6 +184,34 @@ class EntryMutatorTest < Test::Unit::TestCase
     ]
   }
 
+  # mutate test data for `SPOOFED_ENTRY`, same form as `@mutate_tests`.
+  # A mutator without options skips formatting, so the "no stripping" case
+  # turns on `fields_lowercase` to reach the same code with stripping off.
+  @spoofed_tests = {
+    trusted_field_wins: [
+      { fields_strip_underscores: true },
+      { 'SYSTEMD_UNIT' => 'user@1000.service', 'PID' => '777', 'MESSAGE' => 'Accepted publickey for root' }
+    ],
+    trusted_field_wins_lowercased: [
+      { fields_strip_underscores: true, fields_lowercase: true },
+      { 'systemd_unit' => 'user@1000.service', 'pid' => '777', 'message' => 'Accepted publickey for root' }
+    ],
+    user_fields_kept_without_stripping: [
+      { fields_lowercase: true },
+      {
+        '_systemd_unit' => 'user@1000.service', 'systemd_unit' => 'sshd.service',
+        '_pid' => '777', 'pid' => '1', 'message' => 'Accepted publickey for root'
+      }
+    ],
+    mapped_trusted_field_frees_the_name: [
+      { field_map: { '_SYSTEMD_UNIT' => 'unit' }, fields_strip_underscores: true },
+      {
+        'unit' => 'user@1000.service', 'SYSTEMD_UNIT' => 'sshd.service',
+        'PID' => '777', 'MESSAGE' => 'Accepted publickey for root'
+      }
+    ]
+  }
+
   data(@validation_tests)
   def test_validation(opt)
     assert_raise Fluent::ConfigError do
@@ -213,4 +250,14 @@ def test_mutate_with_hash_entry(data)
     mutated = m.run(EntryTestData::ENTRY.to_h)
     assert_equal(expected, mutated)
   end
+
+  # tests using an entry with user fields named after the trusted fields
+
+  data(@spoofed_tests)
+  def test_mutate_with_spoofed_entry(data)
+    options, expected = data
+    m = Fluent::Plugin::SystemdEntryMutator.new(**options)
+    mutated = m.run(EntryTestData::SPOOFED_ENTRY)
+    assert_equal(expected, mutated)
+  end
 end