【发布时间】:2015-04-08 12:24:26
【问题描述】:
我已经使用安装在一系列不同 Linux 机器上的 logstash-forwarder 创建了一个用于日志收集的 ELK 堆栈,而且效果非常好。
我现在正在考虑将 suricata 安装到主 ELK 堆栈上以开始使用 IDS/IPS 功能
我的第一个问题是我只需要在主 ELK 盒子上安装 suricata 并更改此盒子上的 conf 文件以及 logtash-forwarders,所以 suricata 只需要安装在一个盒子上?
其次,我正在尝试更改 conf 文件以允许使用 suricata,因此我在下面列出了我的 logstash 和 logstash 转发器的 conf 文件
文件 13-suricata.conf 是我尝试将其带入 logstash conf 文件,但我不确定这是否是正确的方法,我什至不确定如何处理 logstash-forwarder conf?
任何帮助都会很棒
/etc/logstash/conf.d$ ls
01-lumberjack-input.conf 11-sshlog.conf 13-suricata.conf
10-syslog.conf 12-apache.conf 30-lumberjack-output.conf
01-lumberjack-input.conf
input {
lumberjack {
port => 5000
type => "logs"
ssl_certificate => "/etc/pki/tls/certs/logstash-forwarder.crt"
ssl_key => "/etc/pki/tls/private/logstash-forwarder.key"
}
}
10-syslog.conf
过滤器{ 如果 [类型] == "系统日志" {
grok {
match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
add_field => [ "received_at", "%{@timestamp}" ]
add_field => [ "received_from", "%{host}" ]
}
syslog_pri { }
date {
match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]
}
} }
11-sshlog.conf
filter {
if [type] == "sshlog" {
grok {
type => "sshlog"
match => {"message" => "Failed password for (invalid user |)%{USERNAME:username} from %{IP:src_ip} port %{BASE10NUM:port} ssh2"}
add_tag => "ssh_brute_force_attack"
}
grok {
type => "sshlog"
match => {"message" => "Accepted password for %{USERNAME:username} from %{IP:src_ip} port %{BASE10NUM:port} ssh2"}
add_tag => "ssh_sucessful_login"
}
geoip {
source => "src_ip"
}
}
}
12-apache.conf
filter {
if [type] == "apache-access" {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
}
}
13-suricata.conf
filter {
if [type] == "SuricataIDPS" {
date {
match => [ "timestamp", "ISO8601" ]
}
ruby {
code => "if event['event_type'] == 'fileinfo'; event['fileinfo']['type']=event['fileinfo']['magic'].to_s.split(',')[0]; end;"
}
}
if [src_ip] {
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]}" ]
}
mutate {
convert => [ "[geoip][coordinates]", "float" ]
}
if ![geoip.ip] {
if [dest_ip] {
geoip {
source => "dest_ip"
target => "geoip"
#database => "/opt/logstash/vendor/geoip/GeoLiteCity.dat"
add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
}
mutate {
convert => [ "[geoip][coordinates]", "float" ]
}
}
}
}
}
30-lumberjack-output.conf
output {
elasticsearch { host => localhost }
stdout { codec => rubydebug }
}
logstash fordwarer 配置
"files": [
{
"paths": [
"/var/log/syslog",
"/var/log/auth.log"
],
"fields": { "type": "syslog" }
},
# An array of hashes. Each hash tells what paths to watch and
# what fields to annotate on events from those paths.
#{
#"paths": [
# single paths are fine
#"/var/log/messages",
# globs are fine too, they will be periodically evaluated
# to see if any new files match the wildcard.
#"/var/log/*.log"
#],
# A dictionary of fields to annotate on each event.
#"fields": { "type": "syslog" }
#}, {
# A path of "-" means stdin.
#"paths": [ "-" ],
#"fields": { "type": "stdin" }
# },
{
"paths": [
"/var/log/apache2/*.log"
],
"fields": { "type": "apache-access" }
},
{
"paths": [
"/var/log/auth*.log"
],
"fields": { "type": "sshlog" }
}
"files": [
{
"paths": [ "/var/log/suricata/eve.json" ],
"fields": { "type": "suricata" }
}
]
}
【问题讨论】:
标签: logging elasticsearch logstash logstash-grok