收集syslog日志

可将第三方平台的syslog日志发送到logstash过滤，然后给es存储。
一开始建议不要做条件和过滤，直接放通514端口给syslog，然后直接输出到es，在kibana上获取到源日志，再根据源日志编写配置。

这里以某数据库安全审计平台的syslog为例

1 input.conf

input {
  #sangforVPN syslog
  syslog {
    port => 514  #端口可加双引号，也可以不加
  }
}

2 syslog_out.conf

源日志
: reqtime: 2020-08-04 03:10:13 engine_name: aefqeg-(oracle-18.111.214.103) hostname: WIN-R241P1N4VV osuser: appname: client_mac: BC-B3-11-F2-62-A0 client_ip: 19.111.228.170 client_port: 28994 db_mac: 00-50-56-A3-4C-7B db_ip: 18.111.214.103 db_port: 1521 dbType: 0 dbname: ora11g dbuser: U0235324 sql_catalog: 12 table: d2_peopleinfo field: sql: truncate table d1_peopleinfo reply: SUCCESS. Affected Rows : 0 rowCount: 0 status: 5 pattern: truncate table d2_peopleinfo policy: ȱʡOracle\xB2\xDF\xC2\xD4 rule: [{"name":"ȱʡOracle\xB2\xDF\xC2\xD4->\xB8\xDFΣ\xB2\xD9\xD7\xF7-TRUNCATE","action":1,"threat":3}] threat: 3 log_level: 4 terminal_ip: url: app_user: \n

配置

filter {

    grok {
            keep_empty_captures => true
                #设置两个match，上面匹配不成功，就匹配下面这个。
            match => ["message", ":\sreqtime:\s(?<log_time>%{YEAR}-%{MONTHNUM}-%{MONTHDAY}\s+%{TIME})\sengine_name:.*\((?<db_type>.*)\-.*\)\shostname.*client_mac:\s(%{MAC:client_mac}|)\s+client_ip:\s(%{IP:client_ip:}|)\s+client_port:\s(%{NUMBER:client_port}|)\s+db_mac:\s(%{MAC:db_mac}|)\s+db_ip:\s(%{IP:db_ip}|)\s+db_port:\s(%{NUMBER:db_port}|)\s.*\sdbname:\s((?<db_name>.*)|)\s+dbuser:\s((?<db_user>.*)|)\s+sql_catalog.*table:\s((?<table>.*)|)\s+field.*\ssql:\s((?<sql>.*)|)\s+reply:\s(?<reply>.*)\srowCount.*"]
            match => ["message", ":\sreqtime:\s(?<log_time>%{YEAR}-%{MONTHNUM}-%{MONTHDAY}\s+%{TIME})\sengine_name:\s(?<db_type>.*)_[0-9]{1,3}\s.*\sclient_mac:\s(%{MAC:client_mac}|)\s+client_ip:\s(%{IP:client_ip:}|)\s+client_port:\s(%{NUMBER:client_port}|)\s+db_mac:\s(%{MAC:db_mac}|)\s+db_ip:\s(%{IP:db_ip}|)\s+db_port:\s(%{NUMBER:db_port}|)\s.*\sdbname:\s((?<db_name>.*)|)dbuser:\s((?<db_user>.*)|)sql_catalog.*table:\s((?<table>.*)|)\s+field.*\ssql:\s((?<sql>.*)|)\s+reply:\s(?<reply>.*)\srowCount.*"]
            overwrite => ["timestamp"]
    }

    date {
            match => ["timestamp", "yyyy-MM-dd HH:mm:ss"]
            target => "@timestamp"
            remove_field => "timestamp"
    }
}