Fix: LS failing with ssl_peer_metadata => true (#431)

Author: kares

CHANGELOG.md

@@ -1,4 +1,5 @@
-## x.y.z
+## 6.2.1
+ - Fix: LS failing with `ssl_peer_metadata => true` [#431](https://github.com/logstash-plugins/logstash-input-beats/pull/431)
  - [DOC] described `executor_threads` configuration parameter [#421](https://github.com/logstash-plugins/logstash-input-beats/pull/421)
 
 ## 6.2.0

VERSION

@@ -1 +1 @@
-6.2.0
+6.2.1

lib/logstash/inputs/beats.rb

@@ -129,6 +129,7 @@ class LogStash::Inputs::Beats < LogStash::Inputs::Base
   config :executor_threads, :validate => :number, :default => LogStash::Config::CpuCoreStrategy.maximum
 
   attr_reader :field_hostname, :field_hostip
+  attr_reader :field_tls_protocol_version, :field_tls_peer_subject, :field_tls_cipher
 
   def register
     # For Logstash 2.4 we need to make sure that the logger is correctly set for the
@@ -167,10 +168,10 @@ def register
 
     # define ecs name mapping
     @field_hostname = ecs_select[disabled: "host", v1: "[@metadata][input][beats][host][name]"]
-    @field_hostip   = ecs_select[disabled: "[@metadata][ip_address]", v1: "[@metadata][input][beats][host][ip]"]
-    @field_tls_protocol_version   = ecs_select[disabled: "[@metadata][tls_peer][protocol]", v1: "[@metadata][input][beats][tls][version_protocol]"]
-    @field_tls_peer_subject   = ecs_select[disabled: "[@metadata][tls_peer][subject]", v1: "[@metadata][input][beats][tls][client][subject]"]
-    @field_tls_cipher   = ecs_select[disabled: "[@metadata][tls_peer][cipher_suite]", v1: "[@metadata][input][beats][tls][cipher]"]
+    @field_hostip = ecs_select[disabled: "[@metadata][ip_address]", v1: "[@metadata][input][beats][host][ip]"]
+    @field_tls_protocol_version = ecs_select[disabled: "[@metadata][tls_peer][protocol]", v1: "[@metadata][input][beats][tls][version_protocol]"]
+    @field_tls_peer_subject = ecs_select[disabled: "[@metadata][tls_peer][subject]", v1: "[@metadata][input][beats][tls][client][subject]"]
+    @field_tls_cipher = ecs_select[disabled: "[@metadata][tls_peer][cipher_suite]", v1: "[@metadata][input][beats][tls][cipher]"]
 
     @logger.info("Starting input listener", :address => "#{@host}:#{@port}")
 

lib/logstash/inputs/beats/message_listener.rb

@@ -132,7 +132,7 @@ def extract_tls_peer(hash, ctx)
         tls_session = ctx.channel().pipeline().get("ssl-handler").engine().getSession()
         tls_verified = true
 
-        if not @input.client_authentication_required?
+        unless @input.client_authentication_required?
           # throws SSLPeerUnverifiedException if unverified
           begin
             tls_session.getPeerCertificates()
@@ -144,18 +144,16 @@ def extract_tls_peer(hash, ctx)
           end
         end
 
+        meta_data = hash['@metadata'] ||= {}
+
         if tls_verified
-          set_nested(hash, @field_tls_protocol_version, tls_session.getProtocol())
-          set_nested(hash, @field_tls_peer_subject, tls_session.getPeerPrincipal().getName())
-          set_nested(hash, @field_tls_cipher, tls_session.getCipherSuite())
+          meta_data['tls_peer'] = { :status => "verified" }
 
-          hash['@metadata']['tls_peer'] = {
-            :status       => "verified"
-          }
+          set_nested(hash, input.field_tls_protocol_version, tls_session.getProtocol())
+          set_nested(hash, input.field_tls_peer_subject, tls_session.getPeerPrincipal().getName())
+          set_nested(hash, input.field_tls_cipher, tls_session.getCipherSuite())
         else
-          hash['@metadata']['tls_peer'] = {
-            :status     => "unverified"
-          }
+          meta_data['tls_peer'] = { :status => "unverified" }
         end
       end
     end
@@ -166,9 +164,6 @@ def set_nested(hash, field_name, value)
       field_ref = Java::OrgLogstash::FieldReference.from(field_name)
       # create @metadata sub-hash if needed
       if field_ref.type == Java::OrgLogstash::FieldReference::META_CHILD
-        unless hash.key?("@metadata")
-          hash["@metadata"] = {}
-        end
         nesting_hash = hash["@metadata"]
       else
         nesting_hash = hash

spec/inputs/beats_spec.rb

@@ -12,26 +12,28 @@
   let(:connection) { double("connection") }
   let(:certificate) { BeatsInputTest.certificate }
   let(:port) { BeatsInputTest.random_port }
+  let(:client_inactivity_timeout) { 400 }
+  let(:threads) { 1 + rand(9) }
   let(:queue)  { Queue.new }
   let(:config)   do
     {
-        "port" => 0,
+        "port" => port,
         "ssl_certificate" => certificate.ssl_cert,
         "ssl_key" => certificate.ssl_key,
+        "client_inactivity_timeout" => client_inactivity_timeout,
+        "executor_threads" => threads,
         "type" => "example",
         "tags" => "beats"
     }
   end
 
+  subject(:plugin) { LogStash::Inputs::Beats.new(config) }
+
   context "#register" do
     context "host related configuration" do
-      let(:config) { super().merge("host" => host, "port" => port, "client_inactivity_timeout" => client_inactivity_timeout, "executor_threads" => threads) }
+      let(:config) { super().merge("host" => host, "port" => port) }
       let(:host) { "192.168.1.20" }
-      let(:port) { 9000 }
-      let(:client_inactivity_timeout) { 400 }
-      let(:threads) { 10 }
-
-      subject(:plugin) { LogStash::Inputs::Beats.new(config) }
+      let(:port) { 9001 }
 
       it "sends the required options to the server" do
         expect(org.logstash.beats.Server).to receive(:new).with(host, port, client_inactivity_timeout, threads)
@@ -158,9 +160,80 @@
 
       it "raise a ConfigurationError when multiline codec is set" do
         plugin = LogStash::Inputs::Beats.new(config)
-        expect {plugin.register}.to raise_error(LogStash::ConfigurationError, "Multiline codec with beats input is not supported. Please refer to the beats documentation for how to best manage multiline data. See https://www.elastic.co/guide/en/beats/filebeat/current/multiline-examples.html")
+        expect { plugin.register }.to raise_error(LogStash::ConfigurationError, "Multiline codec with beats input is not supported. Please refer to the beats documentation for how to best manage multiline data. See https://www.elastic.co/guide/en/beats/filebeat/current/multiline-examples.html")
+      end
+    end
+  end
+
+  context "tls meta-data" do
+    let(:config) { super().merge("host" => host, "ssl_peer_metadata" => true, "ssl_certificate_authorities" => [ certificate.ssl_cert ]) }
+    let(:host) { "192.168.1.20" }
+    let(:port) { 9002 }
+
+    let(:queue) { Queue.new }
+    let(:event) { LogStash::Event.new }
+
+    subject(:plugin) { LogStash::Inputs::Beats.new(config) }
+
+    before do
+      @server = org.logstash.beats.Server.new(host, port, client_inactivity_timeout, threads)
+      expect( org.logstash.beats.Server ).to receive(:new).with(host, port, client_inactivity_timeout, threads).and_return @server
+      expect( @server ).to receive(:listen)
+
+      subject.register
+      subject.run(queue) # listen does nothing
+      @message_listener = @server.getMessageListener
+
+      allow( ssl_engine = double('ssl_engine') ).to receive(:getSession).and_return ssl_session
+      allow( ssl_handler = double('ssl-handler') ).to receive(:engine).and_return ssl_engine
+      allow( pipeline = double('pipeline') ).to receive(:get).and_return ssl_handler
+      allow( @channel = double('channel') ).to receive(:pipeline).and_return pipeline
+    end
+
+    let(:ctx) do
+      Java::io.netty.channel.ChannelHandlerContext.impl do |method, *args|
+        fail("unexpected #{method}( #{args} )") unless method.eql?(:channel)
+        @channel
       end
     end
+
+    let(:ssl_session) do
+      Java::javax.net.ssl.SSLSession.impl do |method, *args|
+        case method
+        when :getPeerCertificates
+          [].to_java(java.security.cert.Certificate)
+        when :getProtocol
+          'TLS-Mock'
+        when :getCipherSuite
+          'SSL_NULL_WITH_TEST_SPEC'
+        when :getPeerPrincipal
+          javax.security.auth.x500.X500Principal.new('CN=TEST, OU=RSpec, O=Logstash, C=NL', {})
+        else
+          fail("unexpected #{method}( #{args} )")
+        end
+      end
+    end
+
+    let(:ssl_session_peer_principal) do
+      javax.security.auth.x500.X500Principal
+    end
+
+    let(:message) do
+      org.logstash.beats.Message.new(0, java.util.HashMap.new('foo' => 'bar'))
+    end
+
+    it 'sets tls fields' do
+      @message_listener.onNewMessage(ctx, message)
+
+      expect( queue.size ).to be 1
+      expect( event = queue.pop ).to be_a LogStash::Event
+
+      expect( event.get('[@metadata][tls_peer][status]') ).to eql 'verified'
+
+      expect( event.get('[@metadata][tls_peer][protocol]') ).to eql 'TLS-Mock'
+      expect( event.get('[@metadata][tls_peer][cipher_suite]') ).to eql 'SSL_NULL_WITH_TEST_SPEC'
+      expect( event.get('[@metadata][tls_peer][subject]') ).to eql 'CN=TEST,OU=RSpec,O=Logstash,C=NL'
+    end
   end
 
   context "when interrupting the plugin" do

src/main/java/org/logstash/beats/Server.java

@@ -85,6 +85,14 @@ private void shutdown(){
         }
     }
 
+    /**
+     * @note used in tests
+     * @return the message listener
+     */
+    public IMessageListener getMessageListener() {
+        return messageListener;
+    }
+
     public void setMessageListener(IMessageListener listener) {
         messageListener = listener;
     }