ssh登录日志收集

一 创建logstash grok 过滤规则

cd /usr/share/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-4.1.2/patterns

#cat  ssh
SECURELOG %{WORD:program}[%{DATA:pid}]: %{WORD:status} password for ?(invalid user)? %{WORD:USER} from %{DATA:IP} port
SYSLOGPAMSESSION %{SYSLOGBASE} (?=%{GREEDYDATA:message})%{WORD:pam_module}(%{DATA:pam_caller}): session %{WORD:pam_session_state} for user %{USERNAME:username}(?: by %{GREEDYDATA:pam_by})?
SYSLOGBASE2 (?:%{SYSLOGTIMESTAMP:timestamp}|%{TIMESTAMP_ISO8601:timestamp8601}) (?:%{SYSLOGFACILITY} )?%{SYSLOGHOST:logsource} %{SYSLOGPROG}:

二 配置logstash配置文件

input {
    file {
        type => "seclog"
        path => "/var/log/secure"
   }
}
filter {
if [type] == "seclog" {
    grok {
        match => { "message" => "%{SYSLOGPAMSESSION}" }
        match => { "message" => "%{SECURELOG}" }
        match => { "message" => "%{SYSLOGBASE2}" }
    }
    }
    if ([status] == "Accepted") {
        mutate {
        add_tag => ["Success"]
        }
    }
    else if ([status] == "Failed") {
        mutate {
        add_tag => ["Failed"]
        }
    }
}
output {
    stdout { codec => rubydebug }
    elasticsearch {
        hosts => "elk.test.com:9200"
        index => "sshd_log-%{+YYYY.MM}"
    }
}

三 输出的日志格式

"path" => "/var/log/secure",
"@timestamp" => 2017-12-04T06:15:14.038Z,
"@version" => "1",
"host" => "elk.test.com",
"pid" => "12095",
"program" => "sshd",
"message" => "Dec 4 14:15:13 elk sshd[12095]: Address 192.168.216.1 maps to localhost, but this does not map back to the address - POSSIBLE BREAK-IN ATTEMPT!",
"type" => "seclog",
"logsource" => "elk",
"timestamp" => "Dec 4 14:15:13"
}

四 添加图像

设置收集的关键字

很直观的看到登录成功或者失败的次数