logstash 5.0 fix

.travis.yml

@@ -1,7 +1,9 @@
 sudo: false
 language: ruby
 cache: bundler
+jdk:
+  - oraclejdk8
 rvm:
-  - jruby-1.7.23
+  - jruby-1.7.25
 script: 
   - bundle exec rspec spec

CHANGELOG.md

@@ -1,3 +1,5 @@
+# 3.0.1
+  - Breaking: Updated plugin to use new Java Event APIs for Elastic Stack/Logstash 5.x
 # 1.0.0 (fork from CEF 2.1.3)
   - Slightly changed the format that the codec is available for the IBM Qradar Log Event Extended Format (LEEF)
 # 2.1.3

lib/logstash/codecs/leef.rb

@@ -10,9 +10,13 @@
 
 class LogStash::Codecs::LEEF < LogStash::Codecs::Base
   config_name "leef"
+
+  # Field is to disable the leef header, which can be helpful for generating WCollect messages over syslog
+  config :leefheader, :validate => boolean, :default => true
 
   # Field to enable the default syslog header, which uses the default `%{host}` field for hostname and the timestamp is generated by the codec parsing time. If no value is set the hostname is set to the `hostname` value where logstash is running.
   config :syslogheader, :validate => :boolean, :default => true
+  config :sysloghost, :validate => :string, :default => "logstash"
 
   # Device vendor field in LEEF header. The new value can include `%{foo}` strings
   # to help you build a new value from other parts of the event.
@@ -24,7 +28,7 @@ class LogStash::Codecs::LEEF < LogStash::Codecs::Base
 
   # Device version field in LEEF header. The new value can include `%{foo}` strings
   # to help you build a new value from other parts of the event.
-  config :version, :validate => :string, :default => "2.3.3"
+  config :version, :validate => :string, :default => "5.0.0"
 
   # EventID field in LEEF header. The new value can include `%{foo}` strings
   # to help you build a new value from other parts of the event.
@@ -55,11 +59,19 @@ class LogStash::Codecs::LEEF < LogStash::Codecs::Base
   # Fields to be included in LEEF extension part as key/value pairs
   config :fields, :validate => :array, :default => []
 
+HEADER_FIELDS = ['leef_version', 'leef_vendor', 'leef_product', 'leef_device_version', 'leef_eventid']
+
   public
   def initialize(params={})
     super(params)
   end
 
+private
+def store_header_field(event,field_name,field_data)
+    #Unescape pipes and backslash in header fields
+    event.set(field_name,field_data.gsub(/\\\|/, '|').gsub(/\\\\/, '\\')) unless field_data.nil?
+end
+
   public
   def decode(data)
     # Strip any quotations at the start and end, flex connectors seem to send this
@@ -73,25 +85,24 @@ def decode(data)
     # gives an "SyntaxError: (RegexpError) invalid pattern in look-behind" for the variable length look behind.
     # Therefore one edge case is not handled properly: \\| (this should split, but it does not, because the escaped \ is not recognized)
     # TODO: To solve all unescaping cases, regex is not suitable. A little parse should be written.
-    event['leef_version'], event['leef_vendor'], event['leef_product'], event['leef_device_version'], event['leef_eventid'], *message = data.split /(?<=[^\\]\\\\)[\|]|(?<!\\)[\|]/
-    message = message.join('|')
-
-    # Unescape pipes and backslash in header fields
-    event['leef_version'] = event['leef_version'].gsub(/\\\|/, '|').gsub(/\\\\/, '\\')
-    event['leef_vendor'] = event['leef_vendor'].gsub(/\\\|/, '|').gsub(/\\\\/, '\\')
-    event['leef_product'] = event['leef_product'].gsub(/\\\|/, '|').gsub(/\\\\/, '\\')
-    event['leef_device_version'] = event['leef_device_version'].gsub(/\\\|/, '|').gsub(/\\\\/, '\\')
-    event['leef_eventid'] = event['leef_eventid'].gsub(/\\\|/, '|').gsub(/\\\\/, '\\')
-    #event['leef_name'] = event['leef_name'].gsub(/\\\|/, '|').gsub(/\\\\/, '\\')
-    #event['leef_severity'] = event['leef_severity'].gsub(/\\\|/, '|').gsub(/\\\\/, '\\') unless event['leef_severity'].nil?
+    split_data = data.split /(?<=[^\\]\\\\)[\|]|(?<!\\)[\|]/
+
+    # Store header fields
+    HEADER_FIELDS.each_with_index do |field_name, index|
+     store_header_field(event,field_name,split_data[index])
+    end
+    # Remainder is message
+    message = split_data[HEADER_FIELDS.size..-1].join('|')
 
     # Try and parse out the syslog header if there is one
-    if event['leef_version'].include? ' '
-      event['syslog'], unused, event['leef_version'] = event['leef_version'].rpartition(' ')
+    if event.get('leef_version').include? ' '
+     split_leef_version= event.get('leef_version').rpartition(' ')
+     event.set('syslog', split_leef_version[0])
+     event.set('leef_version',split_leef_version[2])
     end
 
     # Get rid of the LEEF bit in the version
-    event['leef_version'] = event['leef_version'].sub /^LEEF:/, ''
+    event.set('leef_version', event.get('leef_version').sub(/^LEEF:/, ''))
 
     # Strip any whitespace from the message
     if not message.nil? and message.include? '='
@@ -120,15 +131,17 @@ def decode(data)
   public
   def encode(event)
     # "LEEF:1.0|Elastic|Logstash|2.3.3|EventID|"
-    
+
     if self.class.get_config["syslogheader"][:default] == true
-	time = Time.new
-	syslogtime = time.strftime("%b %d %H:%M:%S")
-	sysloghost = sanitize_header_field(event.sprintf(@host))
+        time = Time.new
+        syslogtime = time.strftime("%b %d %H:%M:%S")
+        syslogtime = "<13>" + syslogtime
+        sysloghost = sanitize_header_field(event.sprintf(@sysloghost))
         if sysloghost == ""
-		sysloghost = Socket.gethostname
-	end
-    end 
+                sysloghost = Socket.gethostname
+        end
+    end
+
     vendor = sanitize_header_field(event.sprintf(@vendor))
     vendor = self.class.get_config["vendor"][:default] if vendor == ""
 
@@ -153,19 +166,26 @@ def encode(event)
 
     # Should also probably set the fields sent
 
-    if @syslogheader == true   
-  	sheader = [syslogtime, sysloghost].join(" ")
-	header = ["LEEF:1.0", vendor, product, version, eventid].join("|")
-    	values = @fields.map {|fieldname| get_value(fieldname, event)}.compact.join("       ")
-
-	@on_event.call(event, "#{sheader} #{header}|#{values}\n")
-
+    if @syslogheader == true && @leefheader == true
+        sheader = [syslogtime, sysloghost].join(" ")
+        header = ["LEEF:1.0", vendor, product, version, eventid].join("|")
+        values = @fields.map {|fieldname| get_value(fieldname, event)}.compact.join("	")
+
+        @on_event.call(event, "#{sheader} #{header}|#{values}\n")
+
+    elsif @syslogheader == true && @leefheader == false
+        sheader = [syslogtime, sysloghost].join(" ")
+        values = @fields.map {|fieldname| get_value(fieldname, event)}.compact.join("	")
+        @on_event.call(event, "#{sheader} #{values}\n")
+    elsif @syslogheader == false && @leefheader == false
+        values = @fields.map {|fieldname| get_value(fieldname, event)}.compact.join("	")
+        @on_event.call(event, "#{values}\n")
     else
+        # default behaviour LEEF
+        header = ["LEEF:1.0", vendor, product, version, eventid].join("|")
+        values = @fields.map {|fieldname| get_value(fieldname, event)}.compact.join("	")
 
-    	header = ["LEEF:1.0", vendor, product, version, eventid].join("|")
-    	values = @fields.map {|fieldname| get_value(fieldname, event)}.compact.join("	")
-
-    	@on_event.call(event, "#{header}|#{values}\n")
+        @on_event.call(event, "#{header}|#{values}\n")
     end
   end
 
@@ -222,7 +242,7 @@ def sanitize_extension_val(value)
   end
 
   def get_value(fieldname, event)
-    val = event[fieldname]
+    val = event.get(fieldname)
 
     return nil if val.nil?
 

logstash-codec-leef.gemspec

@@ -1,7 +1,8 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-codec-leef'
-  s.version         = '1.0.1'
+  s.version         = '3.0.1'
+  s.platform	    = 'java'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "LEEF codec to parse and encode LEEF formated logs"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
@@ -20,7 +21,7 @@ Gem::Specification.new do |s|
   s.metadata = { "logstash_plugin" => "true", "logstash_group" => "codec" }
 
   # Gem dependencies
-  s.add_runtime_dependency "logstash-core-plugin-api", "~> 1.0"
+  s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
 
   s.add_development_dependency 'logstash-devutils'
 end

spec/codecs/leef_spec.rb

@@ -213,23 +213,23 @@
 
     def validate(e) 
       insist { e.is_a?(LogStash::Event) }
-      insist { e['leef_version'] } == "1.0"
-      insist { e['leef_device_version'] } == "1.0"
-      insist { e['leef_eventid'] } == "100"
+      insist { e.get('leef_version') } == "1.0"
+      insist { e.get('leef_device_version') } == "1.0"
+      insist { e.get('leef_eventid') } == "100"
     end
 
     it "should parse the leef headers" do
       subject.decode(message) do |e|
         validate(e)
-        ext = e['leef_ext']
-        insist { e["leef_vendor"] } == "security"
-        insist { e["leef_product"] } == "threatmanager"
+        ext = e.get('leef_ext')
+        insist { e.get("leef_vendor") } == "security"
+        insist { e.get("leef_product") } == "threatmanager"
       end
     end
 
     it "should parse the leef body" do
       subject.decode(message) do |e|
-        ext = e['leef_ext']
+        ext = e.get('leef_ext')
         insist { ext['src'] } == "10.0.0.192"
         insist { ext['dst'] } == "12.121.122.82"
         insist { ext['spt'] } == "1232"
@@ -240,16 +240,16 @@ def validate(e)
     it "should be OK with no extension dictionary" do
       subject.decode(no_ext) do |e|
         validate(e)
-        insist { e["leef_ext"] } == nil
+        insist { e.get("leef_ext") } == nil
       end 
     end
 
     let (:missing_headers) { "LEEF:1.0|||1.0|100|src=10.0.0.192 dst=12.121.122.82 spt=1232" }
     it "should be OK with missing LEEF headers (multiple pipes in sequence)" do
       subject.decode(missing_headers) do |e|
         validate(e)
-        insist { e["leef_vendor"] } == ""
-        insist { e["leef_product"] } == ""
+        insist { e.get("leef_vendor") } == ""
+        insist { e.get("leef_product") } == ""
       end 
     end
 
@@ -263,73 +263,73 @@ def validate(e)
     let (:escaped_pipes) { 'LEEF:1.0|security|threatmanager|1.0|100|moo=this\|has an escaped pipe' }
     it "should be OK with escaped pipes in the message" do
       subject.decode(escaped_pipes) do |e|
-        ext = e['leef_ext']
+        ext = e.get('leef_ext')
         insist { ext['moo'] } == 'this\|has an escaped pipe'
       end 
     end
 
     let (:pipes_in_message) {'LEEF:1.0|security|threatmanager|1.0|100|moo=this|has an pipe'}
     it "should be OK with not escaped pipes in the message" do
       subject.decode(pipes_in_message) do |e|
-        ext = e['leef_ext']
+        ext = e.get('leef_ext')
         insist { ext['moo'] } == 'this|has an pipe'
       end
     end
 
     let (:escaped_equal_in_message) {'LEEF:1.0|security|threatmanager|1.0|100|moo=this \=has escaped \= equals\='}
     it "should be OK with escaped equal in the message" do
       subject.decode(escaped_equal_in_message) do |e|
-        ext = e['leef_ext']
+        ext = e.get('leef_ext')
         insist { ext['moo'] } == 'this =has escaped = equals='
       end
     end
 
     let (:escaped_backslash_in_header) {'LEEF:1.0|secu\\\\rity|threat\\\\manager|1.\\\\0|10\\\\0|'}
     it "should be OK with escaped backslash in the headers" do
       subject.decode(escaped_backslash_in_header) do |e|
-        insist { e["leef_version"] } == '1.0'
-        insist { e["leef_vendor"] } == 'secu\\rity'
-        insist { e["leef_product"] } == 'threat\\manager'
-        insist { e["leef_device_version"] } == '1.\\0'
-        insist { e["leef_eventid"] } == '10\\0'
+        insist { e.get("leef_version") } == '1.0'
+        insist { e.get("leef_vendor") } == 'secu\\rity'
+        insist { e.get("leef_product") } == 'threat\\manager'
+        insist { e.get("leef_device_version") } == '1.\\0'
+        insist { e.get("leef_eventid") } == '10\\0'
       end
     end
 
     let (:escaped_backslash_in_header_edge_case) {'LEEF:1.0|security\\\\\\||threatmanager\\\\|1.0|100|'}
     it "should be OK with escaped backslash in the headers (edge case: escaped slash in front of pipe)" do
       subject.decode(escaped_backslash_in_header_edge_case) do |e|
         validate(e)
-        insist { e["leef_vendor"] } == 'security\\|'
-        insist { e["leef_product"] } == 'threatmanager\\'
+        insist { e.get("leef_vendor") } == 'security\\|'
+        insist { e.get("leef_product") } == 'threatmanager\\'
       end
     end
 
   #  let (:escaped_pipes_in_header) {'LEEF:1.0|secu\\|rity|threatmanager\\||1.\\|0|10\\|0|'}
   #  it "should be OK with escaped pipes in the headers" do
   #    subject.decode(escaped_pipes_in_header) do |e|
-  #      insist { e["leef_version"] } == '0'
-  #      insist { e["leef_vendor"] } == 'secu|rity'
-  #      insist { e["leef_product"] } == 'threatmanager|'
-  #      insist { e["leef_device_version"] } == '1.|0'
-  #      insist { e["leef_eventid"] } == '10|0'
+  #      insist { e.get("leef_version") } == '0'
+  #      insist { e.get("leef_vendor") } == 'secu|rity'
+  #      insist { e.get("leef_product") } == 'threatmanager|'
+  #      insist { e.get("leef_device_version") } == '1.|0'
+  #      insist { e.get("leef_eventid") } == '10|0'
   #    end
   #  end
 
     let (:escaped_pipes_in_header) {'LEEF:1.0|secu\\|rity|threatmanager\\||1.\\|0|10\\|0|'}
     it "should be OK with escaped pipes in the headers" do
       subject.decode(escaped_pipes_in_header) do |e|
-        insist { e["leef_version"] } == '1.0'
-        insist { e["leef_vendor"] } == 'secu|rity'
-        insist { e["leef_product"] } == 'threatmanager|'
-        insist { e["leef_device_version"] } == '1.|0'
-        insist { e["leef_eventid"] } == '10|0'
+        insist { e.get("leef_version") } == '1.0'
+        insist { e.get("leef_vendor") } == 'secu|rity'
+        insist { e.get("leef_product") } == 'threatmanager|'
+        insist { e.get("leef_device_version") } == '1.|0'
+        insist { e.get("leef_eventid") } == '10|0'
       end
     end
 
     let (:escaped_backslash_in_message) {'LEEF:1.0|security|threatmanager|1.0|100|moo=this \\\\has escaped \\\\ backslashs\\\\'}
     it "should be OK with escaped backslashs in the message" do
       subject.decode(escaped_backslash_in_message) do |e|
-        ext = e['leef_ext']
+        ext = e.get('leef_ext')
         insist { ext['moo'] } == 'this \\has escaped \\ backslashs\\'
       end
     end
@@ -338,15 +338,15 @@ def validate(e)
     it "should be OK with equal in the headers" do
       subject.decode(equal_in_header) do |e|
         validate(e)
-        insist { e["leef_product"] } == "threatmanager=equal"
+        insist { e.get("leef_product") } == "threatmanager=equal"
       end
     end
 
     let (:syslog) { "Syslogdate Sysloghost LEEF:1.0|security|threatmanager|1.0|100|src=10.0.0.192 dst=12.121.122.82 spt=1232" }
     it "Should detect headers before LEEF starts" do
       subject.decode(syslog) do |e|
         validate(e)
-        insist { e['syslog'] } == 'Syslogdate Sysloghost'
+        insist { e.get('syslog') } == 'Syslogdate Sysloghost'
       end 
     end
   end
@@ -366,11 +366,11 @@ def validate(e)
       event = LogStash::Event.new("leef_vendor" => "vendor", "leef_product" => "product", "leef_device_version" => "2.0", "leef_eventid" => "eventid", "foo" => "bar")
       codec.encode(event)
       codec.decode(results.first) do |e|
-        expect(e['leef_vendor']).to be == event['leef_vendor']
-        expect(e['leef_product']).to be == event['leef_product']
-        expect(e['leef_device_version']).to be == event['leef_device_version']
-        expect(e['leef_eventid']).to be == event['leef_eventid']
-        expect(e['leef_ext']['foo']).to be == event['foo']
+        expect(e.get('leef_vendor')).to be == event.get('leef_vendor')
+        expect(e.get('leef_product')).to be == event.get('leef_product')
+        expect(e.get('leef_device_version')).to be == event.get('leef_device_version')
+        expect(e.get('leef_eventid')).to be == event.get('leef_eventid')
+        expect(e.get('[leef_ext][foo]')).to be == event.get('foo')
       end
     end
   end