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" }
四 添加图像
设置收集的关键字
很直观的看到登录成功或者失败的次数