Logstash-mail.conf

input {
 redis {
    host => 'redis1.hgv.iss.local'
    data_type => list
    key => logpassing
  }
  redis {
    host => 'redis2.hgv.iss.local'
    data_type => list
    key => logpassing
  }
  redis {
    host => 'redis3.hgv.iss.local'
    data_type => list
    key => logpassing
  }
}

filter {
    grok {
      patterns_dir => "/opt/logstash/patterns/"
      break_on_match => false
      keep_empty_captures => true
        match => ["message", "(%{SYSLOGBASE} )(%{EXIM_MULTILINE} )(%{GREEDYDATA})"]
        add_field => { "multiline_message" => true }
    }

    if [multiline_message] {
        drop { }
    }


    # Multiple IF are needed to keep the exclude readable
    # I have devided them in sections on where the are used.
    # System
    if [message] =~ /Bad HELO|no (host|IP) (name|address) found for (host|IP address)|SMTP (command timeout|protocol synchronization error)|too many syntax or protocol errors|unexpected disconnection|TLS error on connection|rejected connection|whitelist=mx-trusted.white.dnsbl.ziggo.local|refused: too many connections|Warning: Delaying blacklisted|daemon started|Spool file is locked|message abandoned|rejected HELO from|uitlees|adreskwaliteit|totaal|too many nonmail commands|lost while reading message data/ {
        drop { }
    }
    if [message] =~ /prefork: child states|spamd: (result|clean message|checking message|setuid|connection from localhost|identified spam)|R=feed_spamanalyser T=netcat|DecodeShortURLs|handled cleanup of child pid|rejected EHLO from|server successfully spawned child process|Unrouteable address|suspected mail loop|Unfrozen by forced delivery|Frozen|LMTP error after DATA|Use of uninitialized value|bayes db version 0 is not able to be used|autolearn=ham|spamd: server pid|killed successfully|cannot parse spamd output|server pid|removing stderr method|has dependency|restarting|sysread(16)|compiled rules|server started on port 783/ {
        drop { }
    }
    
    # Client ERROR's
    if [message] =~ /(Start|End) queue run|SIGHUP received|more than 100 messages|host lookup did not complete|Could not complete sender verify|server reached --max-children setting|all relevant MX records point to non-existent hosts|error in redirect data|an MX or SRV record indicated no SMTP service|unqualified recipient rejected|SMTP connection lost after final dot|discarded by RCPT ACL|error reading from spamd socket|port 783 failed|unqualified sender rejected|all spamd servers failed/ {
        drop { }
    }
    #
    # Custom ziggo log tags
    if [message] =~ /RD001|RD004|RD005|RD007|DW001|MD004|DH003|DH007|EG010|EG0013|EG0014|RW002|DH005|B002|MD001|MD002|RD008|DH001|DH002|DH006/ {
        drop { }
    }
    #
    # Undefined stuff
    if [message] =~ /je poort 587 gebruikt|AAA (2|6)s delay for auth|AAA warning AUTH used twice|message too big|rejected (VRFY|EXPN)/ {
        drop { }
    }
    #
    # LDAP stuff
    if [message] =~ /LDAP search failed - error 50/ {
        drop {}
    }

    # Really, really dirty hack to workaround bug in grok code
    # which won't handle multiple matches on the same field
    mutate {
        add_field => {
            "message_1" => "%{message}"
            "message_2" => "%{message}"
            "message_3" => "%{message}"
            "message_4" => "%{message}"
            "message_5" => "%{message}"
            "message_6" => "%{message}"
            "message_7" => "%{message}"
            "message_8" => "%{message}"
            "message_9" => "%{message}"
            "message_10" => "%{message}"
            "message_11" => "%{message}"
            "message_12" => "%{message}"
            "message_13" => "%{message}"
            "message_14" => "%{message}"
            "message_15" => "%{message}"
        }
    }

  grok {
    patterns_dir => "/etc/logstash/patterns/"
    break_on_match => true
    keep_empty_captures => true
    match => [
        "message_1", "(%{EXIM_LOG_BASE})(%{EXIM_REMOTE_HOST} )(%{EXIM_PROTOCOL} )(%{EXIM_MSG_SIZE} )(%{EXIM_HEADER_ID} )(%{NOTSPACE:for} )(%{EMAILADDRESS:env_rcpt})"
        ]
    match => [
        "message_2", "(%{EXIM_LOG_BASE})(%{EXIM_REMOTE_HOST} )(%{EXIM_PROTOCOL} )(%{NOTSPACE:tls_info} )(%{EXIM_AUTH_LOGIN:login} )(%{EXIM_MSG_SIZE } )(for )(%{EMAILADDRESS:env_rcpt})"
        ]
    match => [
        "message_3", "(%{EXIM_LOG_BASE})(%{EXIM_REMOTE_HOST} )(%{EXIM_PROTOCOL} )(%{NOTSPACE:tls_info} )((%{EXIM_AUTH_LOGIN:login} )|(%{EXIM_AUTH_LOGIN:login}  ))(%{EXIM_MSG_SIZE} )(%{EXIM_HEADER_ID} )(%{NOTSPACE:for} )(%{EMAILADDRESS:env_rcpt})"
        ]
    match => [
        "message_4", "(%{SYSLOGBASE} )(%{EXIM_DATE} )(%{EXIM_REMOTE_HOST} )(%{EXIM_FOR:env_sender} )(rejected RCPT <%{EMAILADDRESS}>: Auth-user )(%{EMAILADDRESS:login} )(%{GREEDYDATA:reason})"
        ]
    match => [
        "message_5", "(%{SYSLOGBASE} )(%{EXIM_DATE} )(%{EXIM_MSGID:exim_msg_id} )(%{EXIM_MSG_STATE:exim_msg_state})"
        ]
    match => [
        "message_6", "(%{SYSLOGBASE} )(%{EXIM_DATE} )(%{EXIM_MSGID:exim_msg_id} )(%{EXIM_REMOTE_HOST} )(Warning: DW001 Spam: host=%{IP:ip} )(login=%{EMAILADDRESS:login} )(%{GREEDYDATA})"
        ]
    match => [
        "message_7", "(%{EXIM_LOG_BASE})(%{EXIM_ROUTE} )(%{EXIM_TRANSPORT} )(%{EXIM_REMOTE_HOST})"
        ]
    match => [
        "message_8", "(%{EXIM_LOG_BASE})(%{EXIM_CAP_EMAIL} )(%{EXIM_ROUTE} )(%{EXIM_TRANSPORT} )(%{EXIM_REMOTE_HOST})"
        ]
    match => [
        "message_9", "(%{SYSLOGBASE} )(%{EXIM_DATE} )(%{EXIM_MSGID:exim_msg_id} )(%{EXIM_REMOTE_HOST} )(%{GREEDYDATA})"
        ]
    match => [
        "message_10", "(%{SYSLOGBASE} )(%{EXIM_DATE} )(%{EXIM_AUTH_FAIL:auth_fail} )(%{NOTSPACE:remote_hostname} )(\(%{NOTSPACE:remote_heloname}\) )?(\[%{IP:remote_host}\]\: )(%{EXIM_ERR_NUM} )(%{EXIM_SET_ID})"
        ]
    match => [
        "message_11", "(%{EXIM_LOG_BASE})(%{EXIM_REMOTE_HOST} )(%{EXIM_PROTOCOL} )(%{EXIM_MSG_SIZE} )(%{NOTSPACE:for} )(%{EMAILADDRESS:env_rcpt})"
        ]
    match => [
        "message_12", "(%{SYSLOGBASE} )(%{EXIM_DATE} )(%{EXIM_MSGID:exim_msg_id} )(%{EXIM_REMOTE_HOST} )(%{EXIM_DEFERRED:reason})"
        ]
    match => [
        "message_13", "(%{SYSLOGBASE} )(%{EXIM_DATE} )(%{EXIM_AUTH_FAIL:auth_fail} )(%{NOTSPACE:remote_hostname} )(\(%{NOTSPACE:remote_heloname}\) )?(\[%{IP:remote_host}\]\: )(%{EXIM_ERR_NUM})"
        ]
    match => [
        "message_14", "(%{EXIM_LOG_BASE})(%{EXIM_REMOTE_HOST} )(%{EXIM_PROTOCOL} )(%{EXIM_MSG_SIZE})(%{GREEDYDATA})"
        ]
    match => [
        "message_15", "(%{SYSLOGBASE} )(%{EXIM_DATE} )(%{EXIM_MSGID:exim_msg_id} )(%{EXIM_FLAGS:exim_flags} )(%{EXIM_BLACKHOLE:blackhole} )(%{GREEDYDATA:reason})"
        ]
  }
 
  ########### End main grok match
  date {
    match => [ "timestamp", "MMM dd HH:mm:ss", "MMM  d HH:mm:ss", "ISO8601" ]
  }

    if [program] == "spamd" {
        drop { }
    }

    if ![exim_msg_state] {
        mutate {
            add_field => { "exim_msg_state" => "undefined" }
        }
    }

    if [exim_msg_id] and ![exim_flags] and [remote_host] {
        mutate {
            update => [ "exim_msg_state", "connection" ]
        }
    }

    if [exim_flags] == "<=" {
        mutate {
            update => [ "exim_msg_state", "received" ]
            rename => [ "env_email", "env_sender" ]
        }
    }
    if [exim_flags] == "=>" or [exim_flags] == "->" {
        mutate {
            update => [ "exim_msg_state", "delivered" ]
            rename => [ "env_email", "env_rcpt" ]
        }
    }
    if [exim_flags] == ">>" {
        mutate {
            update => [ "exim_msg_state", "cutthrough_delivery" ]
        }
    } 
    if [exim_flags] == "*>" {
        mutate {
            update => [ "exim_msg_state", "suppressed_delivery" ]
        }
    } 
    if [exim_flags] == "==" {
        mutate {
            update => [ "exim_msg_state", "deferred" ]
        }
    } 
    if [exim_flags] == "**" {
        mutate {
            update => [ "exim_msg_state", "failed" ]
        }
    } 
    if "Completed" in [message] {
        mutate {
            update => [ "exim_msg_state", "completed" ]
        }
    }

    if [auth_fail] {
        mutate {
            update => [ "exim_msg_state", "failed_login" ]
        }
    }

    if [reason] {
        if [reason] =~ /wordt misbruikt vanuit/ {
            mutate {
                update => [ "exim_msg_state", "deferred" ]
            }
        } else if [reason] =~ "deferred" {
            mutate {
                update => [ "exim_msg_state", "deferred" ]
            }
        } else if [reason] =~ /blocked/ {
            mutate {
                update => [ "exim_msg_state", "blocked_account" ]
            }
        } else if [reason] =~ /zie spam vanaf ip adressen buiten ons netwerk/ {
            mutate {
                update => [ "exim_msg_state", "blocked_account" ]
            }
        }
    }

    if [exim_msg_state] == "delivered" and [exim_flags] == "<-" {
        elasticsearch {
            hosts => [ "loganalyse3.tb.mail.iss.local" ]
            query => 'exim_msg_id:"%{exim_msg_id}"'
            fields => [ "env_sender", "env_sender", "remote_host", "remote_host", "remote_hostname", "remote_hostname" ]
            sort => "ignore_unmapped:true"
            #fail_on_error => "false"
        }
        mutate {
            remove_field => [ "query_failed" ]
        }
    }

    if [logsource] =~ /smtp(1|2).(tb|gn).*/ {
        mutate {
            add_field => { "host_type" => "auth_smtp" }
        }
    }
    if [logsource] =~ /^smtp[3..7].(tb|gn).*/ {
        mutate {
            add_field => { "host_type" => "smtp" }
        }
    }
    if [logsource] =~ /^mx/ {
        mutate {
            add_field => { "host_type" => "mx" }
        }
    }
    if [logsource] =~ /^smtpq*.(gn|tb).*/ {
        mutate {
            add_field => { "host_type" => "smtpq" }
        }
    }

    # Remove the really, really dirty hack to workaround bug in grok code
    # which won't handle multiple matches on the same field
    mutate {
        remove_field => [ "message_1","message_2","message_3","message_4","message_5","message_6","message_7", "message_8","message_9","message_10","message_11","message_12","message_13","message_14","message_15" ]
    }

    # GeoIP data to the ES so we can build a bettermap to see
    # Where the connections are comming from
    if [remote_host] {
        if [remote_host] !~ "(^127\.0\.0\.1)|(^10\.)|(^172\.1[6-9]\.)|(^172\.2[0-9]\.)|(^172\.3[0-1]\.)|(^192\.168\.)|(^212\.54\.42\.(10|13|17))" {
            geoip {
                source => "remote_host"
            }
            mutate {
                add_field => [ "coords", "%{[geoip][longitude]}",
                               "tmplat", "%{[geoip][latitude]}" ]
            }
            mutate {
                merge => [ "coords", "tmplat" ]
            }
            mutate {
                convert => [ "coords", "float" ]
                remove_field => [ "tmplat" ]
            }
        } 
    }

#    # Temp stuff to filter out "good" event to develop on the undefined and bad events.
#    if [exim_msg_state] == "completed" {
#        drop {}
#    } else if [exim_msg_state] == "received" {
#        drop {}
#    } else if [exim_msg_state] == "delivered" {
#        drop {}
#    } else if [exim_msg_state] == "connection" {
#        drop {}
#    }
#    } else if [exim_msg_state] == "failed_login" {
#        drop {}
#    } else if [exim_msg_state] == "blocked_account" {
#        drop {}
#    } else if [exim_msg_state] == "deferred" {
#        drop {}
#    }
}


output {
  elasticsearch {
    cluster => mailtools
    host => "elasticsearch1.hgv.iss.local"
    index_type => "%{exim_msg_state}"
    index => "exim-%{+YYYY.MM.dd}"
    flush_size => 1
  }
#  stdout {
#    codec => rubydebug
#  }
}