beats-conf.conf

input {
beats {
    port => 5044
  }
}

filter{
if [sensor_name] == "dionaea"{
 json{
	source => message
	}
 mutate {
      rename => {
        "dst_ip" => "dest_ip"
      }
      gsub => [
        "src_ip", "::ffff:", "",
        "dest_ip", "::ffff:", ""
      ]
    }
 if [credentials] {
      mutate {
        add_field => {
          "username_dionaea" => "%{[credentials][username]}"
          "password_dionaea" => "%{[credentials][password]}"
        }
      }
    }
}

if [sensor_name] == "cowrie"{
 json{
	source => message
	}
 date {
        match => [ "timestamp", "ISO8601" ]
   	}
 if [src_ip] {
        mutate {
            add_field => {
			 "src_host" => "%{src_ip}" 
			}
	    rename => {
                        "protocol" => "connection.protocol"
                        }
		}

        dns {
            reverse => [ "src_host" ]

            nameserver => [ "8.8.8.8", "8.8.4.4" ]
            action => "replace"
            hit_cache_size => 4096
            hit_cache_ttl => 900
            failed_cache_size => 512
            failed_cache_ttl => 900
        	}
        }
}
if [sensor_name] == "malware" {
   json {
          source => message
}
date{
        match => [ "timestamp", "ISO8601" ]
}

}

if [src_ip]  {
                geoip {
                  cache_size => 10000
                  source => "src_ip"
                  database => "/etc/logstash/vendor/geoip/GeoLite2-City.mmdb"
                }
                geoip {
                  cache_size => 10000
                  source => "src_ip"
                  database => "/etc/logstash/vendor/geoip/GeoLite2-ASN.mmdb"
                }
                translate {
                  refresh_interval => 86400
                  field => "src_ip"
                  destination => "ip_rep"
                  dictionary_path => "/etc/logstash/vendor/geoip/iprep.yaml"
    }
}

mutate {
            remove_tag => [ "beats_input_codec_plain_applied"]

            # cut useless fields added by filebeat, if you use it of course
            remove_field => [ "source", "offset", "beat", "message" ]
        }

}
output {
    elasticsearch {
    hosts => ["10.33.109.81:9200"]
        index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
 }
    if [sensor_name] == "malware" {
        mongodb {
                collection => "malware"
                database => "honeypot"
                uri => "mongodb://127.0.0.1:27017/honeypot"
                codec => "json"
        }
    }

    if [sensor_name] == "cowrie" {
        mongodb{
                collection => "cowrie"
                database => "honeypot"
                uri => "mongodb://127.0.0.1:27017/honeypot"
                codec => "json"

        }
    }
    if [sensor_name] == "dionaea" {
        mongodb{
                collection => "dionaea"
                database => "honeypot"
                uri => "mongodb://127.0.0.1:27017/honeypot"
                codec => "json"
        }
    }

    stdout {
            codec => rubydebug
    }
}