gistfile1.txt

input {
  s3 {
    "access_key_id"  => ""
    "secret_access_key" => ""
    "bucket" => "birahoneypot"
    "interval" => 1440
    "sincedb_path" => "/var/lib/logstash/sincedb/ufw.db"
    "prefix" => "ufw/ufw.log."
    "delete" => true
  }
}

filter {

    grok {
        match => { "message" => "%{DATA:timestamp} %{SYSLOGHOST:hostname} %{DATA:program}(?:\[%{POSINT:pid}\])?: %{GREEDYDATA:ufw_message}" }
    }

    grok {
        patterns_dir => ["/etc/logstash/conf.d"]
        patterns_files_glob => "iptables.pat"
        match => { "ufw_message" => "\[%{DATA}\] \[UFW %{WORD:action}\] IN=%{DATA:i_int} OUT=%{DATA:o_int} %{GREEDYDATA:data}"}
        remove_field => ["ufw_message"]
    }

    if [i_int] == "eth0" {
        grok {
            patterns_dir => ["/etc/logstash/conf.d"]
            patterns_files_glob => "*.pat"
            match => { "data" => "%{IPTABLES}"}
            remove_field => ["data"]
        }
    }

    if [o_int] == "eth0" {
        grok {
            match => {"data" => "SRC=%{IP:src_ip} DST=%{IP:dest_ip} %{GREEDYDATA:tcp_opts} PROTO=%{WORD:protocol} SPT=%{INT:src_port} DPT=%{INT:dst_port} %{GREEDYDATA:tcp_opts}"}
            remove_field => ["data"]
        }
    }

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

    geoip {
        source => ["src_ip"]
    }

    mutate {
        copy => {"src_ip" => "src_reverse_dns"}
    }

    dns {
        reverse => ["src_reverse_dns"]
        action => replace
    }

    mutate {
        remove_field => [ "message", "ufw_message"]
    }
}

output {
    elasticsearch {
        hosts => ["http://localhost:9200"]
        index => "honeypot-ufw-%{+YYYY.MM.dd}"
    }
}