cowrie/doc/elk/logstash-cowrie.conf

input {
       # this is the actual live log file to monitor
       file {
              path => ["/home/cowrie/cowrie-git/log/cowrie.json"]
              codec => json
              type => "cowrie"
       }
       # this is to send old logs to for reprocessing
       #tcp {
       #       port => 3333
       #       type => "cowrie"
       #}
}

filter {
    if [type] == "cowrie" {

        json {
            source => message
        }

        date {
            match => [ "timestamp", "ISO8601" ]
        }

        if [src_ip]  {

            dns {
                reverse => [ "src_host", "src_ip" ]
                action => "append"
            }

            geoip {
                source => "src_ip"
                target => "geoip"
                database => "/opt/logstash/vendor/geoip/GeoLiteCity.dat"
                add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
                add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}"  ]
            }

            geoip {
                source => "src_ip"
                database => "/opt/logstash/vendor/geoip/GeoIPASNum.dat"
            }

            mutate {
                convert => [ "[geoip][coordinates]", "float" ]
            }
        }
    }
}

output {
    if [type] == "cowrie" {
        elasticsearch {
            hosts => ["localhost:9200"]
        }
        file {
            path => "/tmp/cowrie-logstash.log"
            codec => json
        }
        stdout {
            codec => rubydebug
        }
    }
}