官方公众号企业安全新浪微博
FreeBuf.COM网络安全行业门户，每日发布专业的安全资讯、技术剖析。
FreeBuf+小程序
流量分析工具（二）：centos7 + suricata + redis + ELK
- 关注
流量分析工具（二）：centos7 + suricata + redis + ELK
centos7+suricata+redis+ELK
架构设计
1. 需求
目前有一个针对网络流量的攻击日志回溯项目，要求部署一套支持全流量解析并快速检索的系统，计划是解析实时流量并将解析后的json日志存储到 ELK中。
要求包含比较详细的应用层信息，比如 HTTP 的URL、请求头、响应头、请求体和响应体。
支持实时抓包，并存储成pcap的功能，以供本地使用wireshark回溯分析。
支持在大流量背景下的全包存储能力，类似于tcpdump的抓包保存pcap功能。
支持网络流量的威胁检测能力
2. 为什么选择 surcata？
suricata 主要用于企业业务流量的实时分析，是一款非常常用的 IDS 产品，也支持IPS 的功能，通过参数配置即可打开。
suricata 可以完成实时流量的解析，也可以使用离线模式读取pcap。
可以将实时流量抓包存储成pcap，也可以将以下的应用层流量解析成json。
http, ftp, tls, smb, dns, dcerpc, ssh, smtp, imap, modbus, dnp3, enip, nfs, ikev2, krb5, ntp, dhcp, rfb, rdp, snmp, tftp, sip, http2
通过调研发现，suricata的应用层协议输出并不是全部的数据。比如，针对HTTP协议，suricata 无法识别存储HTTP的请求体和响应体，只能做到URL、请求头和响应头的解析和显示。
对比，同类型的产品 zeek，发现zeek也是默认不支持存储HTTP的请求体和响应体，但zeek的脚本功能可以自定义http的协议输出
参考 流量分析的瑞士军刀：Zeek：https://www.cnblogs.com/cscshi/p/16582940.html
与此同时，suricata的 lua脚本功能也可以配置输出HTTP的请求体和响应体，但是还有一个更简单的方式，suricata的规则告警输出的http报文默认包含 HTTP请求体和响应体，通过修改配置文件即可在告警日志中出现 Base64 编码的请求体和响应体。
types:
- alert:
http-body: yes # Requires metadata; enable dumping of HTTP body in Base64
# http-body-printable: yes # Requires metadata; enable dumping of HTTP body in printable format
使用的配置规则：
alert http any any -> any any (msg:"HTTP protocol"; http.protocol; content:"HTTP/"; sid:3;)
另外一个关键点是，在大流量背景下，zeek是通过各种不同的脚本配置输出日志的， 会比较早的遇到性能瓶颈，毕竟所有流量都会挨个经过所有的脚本。
suricata如果启用 lua 脚本支持的话，也会遇到大流量的性能瓶颈，但在不启用lua脚本且规则数量较少的情况下，可以轻松应对3G/s
规模的实时流量，当然了，这跟机器的配置也有关系。
3. 为什么选择redis作为消息队列？
suricata 支持将日志输出到文件、redis、syslog，但不支持输出到 kafka，但可以使用 reids 的 channel 模式，替代 kafka 的 topic 功能。
根据suricata支持的输出功能，通过网上搜索调研和阅读官方文档，统计有在五种比较好用的方式。
方式一：
使用suricata将日志输出到 eve.json文件，然后使用logstash加载 eve.json 文件，并写入到 elasticsearch中。这个也是最常用的方式，网上的教程大多数都是这个方式。
特点：不会造成数据丢失，但 json 文件过多的话会严重占用磁盘空间，需要手动定期删除。
方式二：
使用suricata将日志输出到 suricata的kafka插件，然后使用 kafka 写入到 elasticsearch中。
特点：kafka插件是python写的， 我没测试，有需要的话可以自己配置。
方式三：
改写suricata的输出功能的源码，配置支持将日志输出到 kafka，然后使用 kafka 写入到 elasticsearch中。
suricata 支持Kafka输出：https://www.jianshu.com/p/7877bed3684c
特点：改写源码，俺也不会啊，github有一个已经改写支持的kafka的suricata项目，但使用的是suricata4.0项目， 我没有测试。kafka-feature-3105-v5
方式四：
使用suricata将日志输出到syslog，然后使用logstash读取 syslog 数据，并写入到 elasticsearch中。
特点：syslog属于UDP协议的输出传输，会出现丢包情况，syslog没有消息队列的功能，如果后面的logstash或者elasticsearch服务异常的话，会导致数据丢失。
方式五：
使用suricata将日志输出到redis的 channel 模式，然后使用logstash读取 redis 的 channel数据，并写入到 elasticsearch中。
特点：替代kafka的消息队列，使用redis的channel模式。消息队列的好处是如果后面的服务异常，但是数据没有消费的话，将一直在缓存中等待消费。
对比参考连接 深入探究Kafka与Redis的对比https://www.dbs724.com/65282.html
考虑到大流量的解析和入库问题，磁盘存储问题，搭建的方便性问题，最终决定采用方式五。
一、依赖环境
系统环境：使用最小化安装的 centos7.8
suricata：源码安装 suricata-6.0.10.tar.gz
ELK：
elasticsearch-8.7.0-x86_64.rpm
kibana-8.7.0-x86_64.rpm
logstash-8.7.0-x86_64.rpm
需要依赖：
htp-0.5.17.tar.gz
LuaJIT-2.0.5.tar.gz
二、安装 suricata
1. 安装依赖文件
这里采用源码安装的方式进行，便于自定义目录及组件
yum install -y http://rpms.famillecollet.com/enterprise/remi-release-7.rpm
yum install -y python3-pip git redis net-tools gcc libpcap-devel pcre-devel libyaml-devel file-devel zlib-devel jansson-devel nss-devel libcap-ng-devel libnet-devel tar make libnetfilter_queue-devel lua-devel PyYAML cargo libevent-devel libffi-devel libmaxminddb-devel lz4-devel openssl-devel python3-devel rustc unzip hiredis-devel kernel-devel -y
2. 安装LuaJIT
用于替换suricata原生带有的 lua，效率更高，参考链接： lua与luaJit简介
wget http://luajit.org/download/LuaJIT-2.0.5.tar.gz
tar -zxf LuaJIT-2.0.5.tar.gz
cd LuaJIT-2.0.5/
make && make install
# 配置链接库
echo "/usr/local/lib" >> /etc/ld.so.conf
ldconfig
3. 安装HTP库
LibHTP是HTTP协议和相关位的安全感知解析器，总的来说，是suricata的必要组件
wget https://github.com/OISF/libhtp/releases/download/0.5.17/htp-0.5.17.tar.gz
tar -xzvf htp-0.5.17.tar.gz
cd htp-0.5.17
./configure
make && make install
4. 安装 suricata
Suricata是一个高性能的网络IDS、IPS和网络安全监控引擎。
Suricata安装在/usr/bin/下，相关配置在/etc/suricata/，日志输出在/var/log/suricata/下。
# 编译
tar zxvf suricata-6.0.10.tar.gz
cd suricata-6.0.10/
./configure \
--prefix=/usr \
--sysconfdir=/etc \
--localstatedir=/var \
--enable-nfqueue \
--enable-luajit \
--with-libluajit-includes=/usr/local/include/luajit-2.0/ \
--with-libluajit-libraries=/usr/local/lib/ \
--enable-redis \
--enable-hiredis \
--enable-profiling \
--enable-geoip \
--enable-rust
#--enable-lua \ # liblua and luajit 不能共存
#--enable-nfqueue #为内联IDP(启用NFQueue支持)启用NFQueue支持，一般开启Suricata的IPS模式时使用
#--enable-profiling #启用性能分析
其他组件的支持，仅供参考
# 加入PF_RING支持
--enable-pfring #启用本机Pf_Ring的支持
--with-libpfring-includes #libpfring的目录
--with-libpfring-libraries #libpfring的库目录
#Lua主要用来编写规则
--enable-luajit #启用Luajit(C语言编写的Lua代码解析器)支持
--with-libluajit-includes #libluajit的目录
--with-libluajit-libraries #libluajit的库目录
#使用scan(一个高性能的多重正则表达式匹配库)时，所需的库
--with-libhs-includes #libhs的目录
--with-libhs-libraries #libhs的库目录
--with-libnss-libraries #libnss的目录
--with-libnss-includes #libnss的库目录
--with-libnspr-libraries #libnspr的目录
--with-libnspr-includes #libnspr的库目录
# 安装编译时用到的依赖组件
yum -y install liblua5.1-dev #配置支持lua, --enable-lua
yum -y install libgeoip-dev #配置支持GeoIP, --enable-geoip
yum -y install rustc cargo #配置支持Rust, --enable-rust
安装
make
make install
# 生成配置文件，会在/etc/suricata/下面生成新的配置文件
make install-conf
# 生成规则规则
make install-rules
# 全部生成
make install-full
验证
# 查看编译信息
suricata --build-info
# 启用测试
suricata -T
后台启动
# 监听模式
suricata -i ens33 -c /etc/suricata/suricata_multi.yaml -v -k none --runmode autofp -D
# 离线读包模式
suricata -i ens33 -r pcap_source_data/ -v -k none --runmode autofp
-i ens33 # 指定网卡
-c /etc/suricata/suricata_multi.yaml # 指定配置文件
--runmode autofp # 运行模式
-D 后台运行
三、配置规则
在调试配置之前先编写一个规则文件，实现的功能很简单，就是将HTTP的协议全部输出成告警事件。
mkdir /etc/suricata/rules/
cd /etc/suricata/rules/
vim suricata.rules
alert http any any -> any any (msg:"HTTP protocol"; http.protocol; content:"HTTP/"; sid:3;)
在前面步骤中make install-conf，可以/etc/suricata/目录生成suricata的配置文件 suricata.yaml
修改配置文件，删除其他的规则，加载suricata.rules
default-rule-path: /etc/suricata/rules
rule-files:
- suricata.rules
测试启动suricata，查看是否出现报错
suricata -i ens33 -c /etc/suricata/suricata.yaml -T
没有报错的话就可以正式启动，修改配置文件，查看输出的文件。
suricata -i ens33 -c /etc/suricata/suricata.yaml -v -k none --runmode autofp
四、配置 suricata输出
1. 修改日志格式为json
默认输出位置在 /etc/suricata目录，修改文件 /etc/suricata/suricata.yaml。
vim /etc/suricata/suricata.yaml
# Define your logging outputs. If none are defined, or they are all
# disabled you will get the default: console output.
outputs:
- console:
enabled: yes
type: json
- file:
enabled: yes
level: info
filename: suricata.log
type: json
- syslog:
enabled: no
facility: local5
format: "[%i] <%d> -- "
type: json
2. 开启pcap输出功能
vim /etc/suricata/suricata.yaml
- pcap-log:
enabled: yes
filename: listen.pcap
limit: 1000mb
max-files: 100 # 每个1G，100个是100G
compression: none
mode: sguil # normal, multi or sguil.
dir: /data/suricata/pcaps/
use-stream-depth: no # "no" logs all packets
honor-pass-rules: no # If set to "yes", flows in which a pass rule matched will stop being logged.
3. 启用所有协议的日志输出
参考官网： https://suricata.readthedocs.io/en/suricata-6.0.10/output/eve/eve-json-format.html
4. 输出到 eve.json文件（本次不采用此模式）
输出到 eve.json 文件，可以很方便的调试配置文件的输出，可以通过文件的大小
在此用于解释多文件输出的用法，以方便理解后面的两个通道输出到 redis的过程。
参考官网：
EVE JSON输出：https://www.osgeo.cn/suricata/output/eve/eve-json-output.html#output-types
文件权限：https://www.osgeo.cn/suricata/output/eve/eve-json-output.html#file-permissions
JSON标志：https://www.osgeo.cn/suricata/output/eve/eve-json-output.html#json-flags
suricata支持在日志输出的位置配置不同的输出器，这里先配置输出到json文件，后续输出到ELK中时需要改成输出到 redis 的 channel 模式。
可选配置文件的rotate，如设置每60分钟创建新文件，防止单个文件过大。
可选配置多进程输出到文件，文件名会以进程ID命名。
需要提前创建好日志输出/data/suricata/protocol/目录，否则会启动失败。
vim /etc/suricata/suricata.yaml
outputs:
- eve-log:
enabled: yes
filetype: regular # 可选regular|syslog|unix_dgram|unix_stream|redis
filename: /data/suricata/protocol/alert-%Y-%m-%d-%H:%M:%S.json
filemode: 644
json:
preserve-order: yes
compact: yes
ensure-ascii: yes
escape-slash: yes
rotate-interval: 60m # 60 minutes
threaded: false
5. 输出多个 eve.json 文件（用于理解多通道输出）
多个记录器实例：https://www.osgeo.cn/suricata/output/eve/eve-json-output.html#multiple-logger-instances
官方支持将不同的协议日志输出到不同的文件中，在output中配置同级的 eve-log 即可，但是官方并建议使用很多个独立的日志文件输出。
官方示例如下：
outputs:
- eve-log:
enabled: yes
filetype: regular # 可选regular|syslog|unix_dgram|unix_stream|redis
filename: eve-ips.json
types:
- alert
- drop
- eve-log:
enabled: yes
filetype: regular # 可选regular|syslog|unix_dgram|unix_stream|redis
filename: eve-nsm.json
types:
- http
- dns
- tls
实现思路很简单，就是将 - eve-log: 并列多写几个，然后指定不同的输出即可。
复制配置文件到 suricata_multi.yaml，修改配置
vim /etc/suricata/suricata_multi.yaml
# 输出alert告警日志到 /data/suricata/protocol/alert-m-%d-%H:%M:%S.json，告警日志级别选择 Alert
- eve-log:
enabled: yes
filetype: regular #regular|syslog|unix_dgram|unix_stream|redis
filename: /data/suricata/protocol/alert-%Y-%m-%d-%H:%M:%S.json
filemode: 644
json:
preserve-order: yes
compact: yes
ensure-ascii: yes
escape-slash: yes
rotate-interval: 60m # 60 minutes
threaded: false
identity: "suricata"
facility: local5
level: Alert
metadata: yes
pcap-file: true
community-id: true
community-id-seed: 0
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
types:
- alert:
payload: no
packet: no
metadata: yes
http-body: yes
tagged-packets: yes
metadata:
app-layer: true
flow: false
rule:
metadata: false
raw: false
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
# 输出协议日志到 /data/suricata/protocol/protocol-%Y-%m-%d-%H:%M:%S.json 文件，告警日志级别选择 Info
- eve-log:
enabled: yes
filetype: regular #regular|syslog|unix_dgram|unix_stream|redis
filename: /data/suricata/protocol/protocol-%Y-%m-%d-%H:%M:%S.json
filemode: 644
json:
preserve-order: yes
compact: yes
ensure-ascii: yes
escape-slash: yes
rotate-interval: 60m # 60 minutes
threaded: false
identity: "suricata"
facility: local5
level: Info
ethernet: yes
metadata: yes
pcap-file: true
community-id: true
community-id-seed: 0
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
types:
- http:
extended: yes
dump-all-headers: both
- dns:
version: 2
enabled: yes
requests: yes
responses: yes
formats: [detailed, grouped]
types: [a, ns, md, mf, cname, soa, mb, mg, mr, null, wks, ptr, hinfo, minfo, mx, txt, rp, afsdb, x25, isdn, rt, nsap, nsapptr, sig, key, px, gpos, aaaa, loc, nxt, srv, atma, naptr, kx, cert, a6, dname, opt, apl, ds, sshfp, ipseckey, rrsig, nsec, dnskey, dhcid, nsec3, nsec3param, tlsa, hip, cds, cdnskey, spf, tkey, tsig, maila, any, uri]
- tls:
extended: yes
session-resumption: yes
custom: [subject, issuer, session_resumed, serial, fingerprint, sni, version, not_before, not_after, certificate, chain, ja3, ja3s]
测试启动
suricata -i ens33 -c /etc/suricata/suricata_multi.yaml -v -k none --runmode autofp
查看启动日志输出，存在 All AFP capture threads are running.即说明启动成功。
配置文件中的目录已经要先创建好，否则会启动失败。
在 /data/suricata/protocol可以看到输出的日志文件。
[root@NTA protocol]# cd /data/suricata/protocol/
[root@NTA protocol]# ls -l
total 22785064
-rw-r--r--. 1 root root 0 Apr 20 10:30 alert-2023-04-20-10:30:19.json
-rw-r--r--. 1 root root 0 Apr 20 10:33 alert-2023-04-20-10:33:17.json
-rw-r--r--. 1 root root 3865 Apr 20 10:33 protocol-2023-04-20-10:30:19.json
-rw-r--r--. 1 root root 3859 Apr 20 10:34 protocol-2023-04-20-10:33:17.json
通过分析结果可以看到，确实是已经输出了新的文件，文件内容就是json数据。
[root@NTA protocol]# tail protocol-2023-04-20-10:30:19.json -n1 | jq
{
"timestamp": "2023-04-20T10:33:15.336837+0800",
"flow_id": 1817757461430226,
"in_iface": "ens33",
"event_type": "flow",
"src_ip": "10.10.10.1",
"src_port": 61063,
"dest_ip": "10.10.10.135",
"dest_port": 22,
"proto": "TCP",
"flow": {
"pkts_toserver": 3,
"pkts_toclient": 3,
"bytes_toserver": 270,
"bytes_toclient": 162,
"start": "2023-04-20T10:30:30.835538+0800",
"end": "2023-04-20T10:32:30.831269+0800",
"age": 120,
"state": "new",
"reason": "shutdown",
"alerted": false
},
"ether": {
"dest_macs": [
"00:0c:29:4f:81:53"
],
"src_macs": [
"00:50:56:c0:00:08"
]
},
"community_id": "1:z05uY2pXcMpfBOoYDWsPvPs4hdk=",
"tcp": {
"tcp_flags": "00",
"tcp_flags_ts": "00",
"tcp_flags_tc": "00"
},
"host": "suricata_sensor"
}
到此，就说明整体的流程配置好了，既然有了json文件的输出，就可以使用logstash读取文件然后入库到 elasticsearch。
下一步，考虑到日志文件的存储会消耗磁盘空间，我们使用 redis 的订阅模式输出日志事件到logstash。
6. 将数据输出到 redis 的订阅模式（推荐使用）
复制配置文件为/etc/suricata/suricata_redis.yaml，修改配置文件输出到 redis的channel模式
redis 的输出 频道为 suricata_alert和suricata_protocol
vim /etc/suricata/suricata_redis.yaml
# Configure the type of alert (and other) logging you would like.
outputs:
- eve-log:
enabled: yes
filetype: redis #regular|syslog|unix_dgram|unix_stream|redis
identity: "suricata"
facility: local5
level: Alert ## possible levels: Emergency, Alert, Critical, Error, Warning, Notice, Info, Debug
ethernet: yes # log ethernet header in events when available
redis:
server: 127.0.0.1
port: 6379
async: true ## if redis replies are read asynchronously
mode: channel ## possible values: list|lpush (default), rpush, channel|publish
## lpush and rpush are using a Redis list. "list" is an alias for lpush
## publish is using a Redis channel. "channel" is an alias for publish
key: suricata_alert
pipelining:
enabled: yes
batch-size: 10
metadata: yes
pcap-file: true
community-id: true
community-id-seed: 0
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
types:
- alert:
payload: no
packet: no # enable dumping of packet (without stream segments)
metadata: yes # enable inclusion of app layer metadata with alert. Default yes
http-body: yes # Requires metadata; enable dumping of HTTP body in Base64
tagged-packets: yes
metadata:
app-layer: true
flow: false
rule:
metadata: false
raw: false
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
# Extensible Event Format (nicknamed EVE) event log in JSON format
- eve-log:
enabled: yes
filetype: redis #regular|syslog|unix_dgram|unix_stream|redis
identity: "suricata"
facility: local5
level: Info ## possible levels: Emergency, Alert, Critical, ## Error, Warning, Notice, Info, Debug
ethernet: yes # log ethernet header in events when available
redis:
server: 127.0.0.1
port: 6379
async: true
mode: channel ## possible values: list|lpush (default), rpush, channel|publish
## lpush and rpush are using a Redis list. "list" is an alias for lpush
## publish is using a Redis channel. "channel" is an alias for publish
key: suricata_protocol ## key or channel to use (default to suricata)
pipelining:
enabled: yes
batch-size: 10
metadata: yes
pcap-file: true
community-id: true
community-id-seed: 0
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
types:
- http:
extended: yes
dump-all-headers: both
- dns:
version: 2
enabled: yes
requests: yes
responses: yes
formats: [detailed, grouped]
types: [a, ns, md, mf, cname, soa, mb, mg, mr, null, wks, ptr, hinfo, minfo, mx, txt, rp, afsdb, x25, isdn, rt, nsap, nsapptr, sig, key, px, gpos, aaaa, loc, nxt, srv, atma, naptr, kx, cert, a6, dname, opt, apl, ds, sshfp, ipseckey, rrsig, nsec, dnskey, dhcid, nsec3, nsec3param, tlsa, hip, cds, cdnskey, spf, tkey, tsig, maila, any, uri]
- tls:
extended: yesd
session-resumption: yes
custom: [subject, issuer, session_resumed, serial, fingerprint, sni, version, not_before, not_after, certificate, chain, ja3, ja3s]
测试启动
suricata -i ens33 -c /etc/suricata/suricata_redis.yaml -v -k none --runmode autofp
查看启动日志输出，存在 All AFP capture threads are running.即说明启动成功。
配置文件中的目录已经要先创建好，否则会启动失败。
在redis启动消费消息队列，监听suricata的输出 频道为 suricata_alert
[root@NTA data]# redis-cli
127.0.0.1:6379> SUBSCRIBE suricata_alert
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "suricata_alert"
3) (integer) 1
虚拟机搭建的话可以使用科来的数据包播放器给虚拟机的网卡发包。
下载安装 科来网络分析系统：https://www.colasoft.com.cn/download/capsa.php
发包之后可以看到 redis 消费到了新的数据
1) "message"
2) "suricata_alert"
3) "{\"timestamp\":\"2023-04-20T10:58:20.279310+0800\",\"flow_id\":1915493846657675,\"in_iface\":\"ens33\",\"event_type\":\"alert\",\"src_ip\":\"10.0.88.40\",\"src_port\":64452,\"dest_ip\":\"45.32.125.185\",\"dest_port\":80,\"proto\":\"TCP\",\"ether\":{\"src_mac\":\"f4:5c:89:b1:1d:bd\",\"dest_mac\":\"60:da:83:36:bb:e9\"},\"community_id\":\"1:CtewaAnze05BxdS8vTRNBawnYwI=\",\"tx_id\":0,\"alert\":{\"action\":\"allowed\",\"gid\":1,\"signature_id\":3,\"rev\":0,\"signature\":\"HTTP protocol\",\"category\":\"\",\"severity\":3},\"http\":{\"hostname\":\"45.32.125.185\",\"url\":\"/\",\"http_user_agent\":\"python-requests/2.6.2 CPython/2.7.10 Darwin/18.0.0\",\"http_content_type\":\"text/html\",\"http_method\":\"GET\",\"protocol\":\"HTTP/1.1\",\"status\":401,\"length\":459,\"http_response_body\":\"PCFET0NUWVBFIEhUTUwgUFVCTElDICItLy9JRVRGLy9EVEQgSFRNTCAyLjAvL0VOIj4KPGh0bWw+PGhlYWQ+Cjx0aXRsZT40MDEgVW5hdXRob3JpemVkPC90aXRsZT4KPC9oZWFkPjxib2R5Pgo8aDE+VW5hdXRob3JpemVkPC9oMT4KPHA+VGhpcyBzZXJ2ZXIgY291bGQgbm90IHZlcmlmeSB0aGF0IHlvdQphcmUgYXV0aG9yaXplZCB0byBhY2Nlc3MgdGhlIGRvY3VtZW50CnJlcXVlc3RlZC4gIEVpdGhlciB5b3Ugc3VwcGxpZWQgdGhlIHdyb25nCmNyZWRlbnRpYWxzIChlLmcuLCBiYWQgcGFzc3dvcmQpLCBvciB5b3VyCmJyb3dzZXIgZG9lc24ndCB1bmRlcnN0YW5kIGhvdyB0byBzdXBwbHkKdGhlIGNyZWRlbnRpYWxzIHJlcXVpcmVkLjwvcD4KPGhyPgo8YWRkcmVzcz5BcGFjaGUvMi40LjcgKFVidW50dSkgU2VydmVyIGF0IDQ1LjMyLjEyNS4xODUgUG9ydCA4MDwvYWRkcmVzcz4KPC9ib2R5PjwvaHRtbD4K\"},\"app_proto\":\"http\",\"host\":\"suricata_sensor\"}"
7. logstash 订阅redis数据，并输出到 elasticsearch
ELK 的安装见上一章。流量分析工具（一）：centos7安装ELK 8.7 并配置密码
在此使用的elasticsearch已配置密码，需要在logstash里面配置。
由于是有两个频道输出日志，可以启用logstash 的 pipelines模式，配置文件/usr/share/logstash/config/pipelines.yml
# /usr/share/logstash/config/pipelines.yml
##唯一id(标识用的)
#- pipeline.id: logstash_test
##开启线程数量
# pipeline.workers: 10
##指定对应conf文件
# path.config: "/etc/logstash/conf.d/logstash_test.conf"
- pipeline.id: suricata_alert
pipeline.workers: 10
queue.type: persisted
path.config: "/etc/logstash/conf.d/logstash_alert.conf"
- pipeline.id: suricata_protocol
pipeline.workers: 10
queue.type: persisted
path.config: "/etc/logstash/conf.d/logstash_protocol.conf"
依据piplines读取的配置文件，编辑文件：
# /etc/logstash/conf.d/logstash_alert.conf
# 从redis读取
input {
redis {
data_type => "pattern_channel"
key => "suricata_alert"
host => "127.0.0.1"
port => 6379
threads => 10
}
}
# 从文件读取
# input
# {
# file
# {
# path => ["/data/suricata/protocol/alert-*.json"]
# codec => "json"
# # sincedb_path => "NULL" # windows 的选择
# sincedb_path => "/dev/null" # linux 的选择
# start_position => "beginning"
# }
# }
filter{
# 矫正 @timestamp 用于生成索引名的时间
ruby{
code => "event.set('n_logstashStamp', (event.get('@timestamp').time.localtime + 8*60*60).strftime('%Y-%m-%d %H:%M:%S'))"
}
date {
match => [ "n_logstashStamp", "yyyy-MM-dd HH:mm:ss" ]
target => "@timestamp"
}
mutate {
#将不需要的JSON字段过滤
remove_field => ["n_logstashStamp", "@version", "event", "log"]
}
}
# 输出elasticsearch，配置用户名和密码
output {
elasticsearch {
hosts => ["http://127.0.0.1:9200"]
index => "alert_%{+YYYYMMdd}"
user => elastic
password => "elastic_023"
timeout => 300
}
# /etc/logstash/conf.d/logstash_protocol.conf
# 从redis读取
input {
redis {
data_type => "pattern_channel"
key => "suricata_protocol"
host => "127.0.0.1"
port => 6379
threads => 10
}
}
# 从文件读取
# input
# {
# file
# {
# path => ["/data/suricata/protocol/protocol-*.json"]
# codec => "json"
# # sincedb_path => "NULL" # windows 的选择
# sincedb_path => "/dev/null" # linux 的选择
# start_position => "beginning"
# }
# }
filter{
# 矫正 @timestamp 用于生成索引名的时间
ruby{
code => "event.set('n_logstashStamp', (event.get('@timestamp').time.localtime + 8*60*60).strftime('%Y-%m-%d %H:%M:%S'))"
}
date {
match => [ "n_logstashStamp", "yyyy-MM-dd HH:mm:ss" ]
target => "@timestamp"
}
mutate {
#将不需要的JSON字段过滤
remove_field => ["n_logstashStamp", "@version", "event", "log"]
}
}
# 输出elasticsearch，配置用户名和密码
output {
elasticsearch {
hosts => ["http://127.0.0.1:9200"]
index => "protocol_%{+YYYYMMdd}"
user => elastic
password => "elastic_023"
timeout => 300
}
启用logstash，执行 /usr/share/logstash/bin/logstash ，启动程序
[root@NTA config]# /usr/share/logstash/bin/logstash
Using bundled JDK: /usr/share/logstash/jdk
.......
.......
.......
遇到redis 的消息之后，logstash会自动入库，可以在kibana中查看索引和日志数据。
# kibana的开发控制台中：
GET _cat/indices
yellow open protocol_20230419 h6kk5wMnThGfY96XXCcUnw 1 1 221962 0 46.8mb 46.8mb
yellow open alert_20230420 I4Ev2UCxRDuY6SHW3i6IAg 1 1 710 0 3.7mb 3.7mb
yellow open alert_20230419 9LlRi5yYQpy4kco2EbRenA 1 1 3130 0 805.7kb 805.7kb
yellow open protocol_20230420 AqKtF6DBTQ-u3iyjZxPjfQ 1 1 60 0 177.1kb 177.1kb
至此已经搭建完成。
8. suricata 全文配置文件
%YAML 1.1
---
vars:
address-groups:
HOME_NET: "[192.168.0.0/16,10.0.0.0/8,172.16.0.0/12]"
EXTERNAL_NET: "!$HOME_NET"
HTTP_SERVERS: "$HOME_NET"
SMTP_SERVERS: "$HOME_NET"
SQL_SERVERS: "$HOME_NET"
DNS_SERVERS: "$HOME_NET"
TELNET_SERVERS: "$HOME_NET"
AIM_SERVERS: "$EXTERNAL_NET"
DC_SERVERS: "$HOME_NET"
DNP3_SERVER: "$HOME_NET"
DNP3_CLIENT: "$HOME_NET"
MODBUS_CLIENT: "$HOME_NET"
MODBUS_SERVER: "$HOME_NET"
ENIP_CLIENT: "$HOME_NET"
ENIP_SERVER: "$HOME_NET"
port-groups:
HTTP_PORTS: "80"
SHELLCODE_PORTS: "!80"
ORACLE_PORTS: 1521
SSH_PORTS: 22
DNP3_PORTS: 20000
MODBUS_PORTS: 502
FILE_DATA_PORTS: "[$HTTP_PORTS,110,143]"
FTP_PORTS: 21
GENEVE_PORTS: 6081
VXLAN_PORTS: 4789
TEREDO_PORTS: 3544
default-log-dir: /etc/suricata/log/
stats:
enabled: yes
interval: 8
outputs:
- fast:
enabled: no
filename: fast.log
append: yes
- eve-log:
enabled: yes
filetype: redis #regular|syslog|unix_dgram|unix_stream|redis
identity: "suricata"
facility: local5
level: Alert ## possible levels: Emergency, Alert, Critical, Error, Warning, Notice, Info, Debug
ethernet: yes # log ethernet header in events when available
redis:
server: 127.0.0.1
port: 6379
async: true ## if redis replies are read asynchronously
mode: channel ## possible values: list|lpush (default), rpush, channel|publish
key: suricata_alert ## key or channel to use (default to suricata)
pipelining:
enabled: yes ## set enable to yes to enable query pipelining
batch-size: 10 ## number of entries to keep in buffer
metadata: yes
pcap-file: true
community-id: true
community-id-seed: 0
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
types:
- alert:
payload: no # enable dumping payload in Base64
packet: no # enable dumping of packet (without stream segments)
metadata: yes # enable inclusion of app layer metadata with alert. Default yes
http-body: yes # Requires metadata; enable dumping of HTTP body in Base64
tagged-packets: yes
metadata:
app-layer: true
flow: false
rule:
metadata: false
raw: false
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
- eve-log:
enabled: yes
filetype: redis #regular|syslog|unix_dgram|unix_stream|redis
identity: "suricata"
facility: local5
level: Info ## possible levels: Emergency, Alert, Critical,
ethernet: yes # log ethernet header in events when available
redis:
server: 127.0.0.1
port: 6379
async: true ## if redis replies are read asynchronously
mode: channel ## possible values: list|lpush (default), rpush, channel|publish
key: suricata_protocol ## key or channel to use (default to suricata)
pipelining:
enabled: yes ## set enable to yes to enable query pipelining
batch-size: 10 ## number of entries to keep in buffer
metadata: yes
pcap-file: true
community-id: true
community-id-seed: 0
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
types:
- anomaly:
enabled: no
types:
decode: yes
stream: yes
applayer: yes
packethdr: yes
- http:
extended: yes # enable this for extended logging information
dump-all-headers: both
- dns:
version: 2
enabled: yes
requests: yes
responses: yes
formats: [detailed, grouped]
types: [a, ns, md, mf, cname, soa, mb, mg, mr, null, wks, ptr, hinfo, minfo, mx, txt, rp, afsdb, x25, isdn, rt, nsap, nsapptr, sig, key, px, gpos, aaaa, loc, nxt, srv, atma, naptr, kx, cert, a6, dname, opt, apl, ds, sshfp, ipseckey, rrsig, nsec, dnskey, dhcid, nsec3, nsec3param, tlsa, hip, cds, cdnskey, spf, tkey, tsig, maila, any, uri]
- tls:
extended: yes # enable this for extended logging information
session-resumption: yes
custom: [subject, issuer, session_resumed, serial, fingerprint, sni, version, not_before, not_after, certificate, chain, ja3, ja3s]
- smtp:
extended: yes # enable this for extended logging information
custom: [received, x-mailer, x-originating-ip, relays, reply-to, bcc]
- ftp
- rdp
- nfs
- smb
- tftp
- ikev2
- dcerpc
- krb5
- snmp
- dhcp:
enabled: yes
extended: yes
- ssh
- mqtt:
passwords: yes # enable output of passwords
- http2
- flow
- metadata
- http-log:
enabled: no
filename: http.log
append: yes
extended: yes # enable this for extended logging information
custom: yes # enable the custom logging format (defined by customformat)
customformat: "%{%D-%H:%M:%S}t.%z %{X-Forwarded-For}i %H %m %h %u %s %B %a:%p -> %A:%P"
filetype: regular # 'regular', 'unix_stream' or 'unix_dgram'
- tls-log:
enabled: no # Log TLS connections.
filename: tls.log # File to store TLS logs.
append: yes
extended: yes # Log extended information like fingerprint
custom: yes # enabled the custom logging format (defined by customformat)
customformat: "%{%D-%H:%M:%S}t.%z %a:%p -> %A:%P %v %n %d %D"
filetype: regular # 'regular', 'unix_stream' or 'unix_dgram'
session-resumption: yes
- tls-store:
enabled: no
certs-log-dir: certs # directory to store the certificates files
- pcap-log:
enabled: yes
filename: listen.pcap
limit: 1000mb
max-files: 100 # 100G
compression: none
mode: sguil # normal, multi or sguil.
dir: /data/suricata/pcaps/
use-stream-depth: no #If set to "yes" packets seen after reaching stream inspection depth are ignored. "no" logs all packets
honor-pass-rules: no # If set to "yes", flows in which a pass rule matched will stop being logged.
- alert-debug:
enabled: no
filename: alert-debug.log
append: yes
- alert-prelude:
enabled: no
profile: suricata
log-packet-content: no
log-packet-header: yes
- stats:
enabled: yes
filename: stats.log
append: yes # append to file (yes) or overwrite it (no)
totals: yes # stats for all threads merged together
threads: no # per thread stats
- syslog:
enabled: no
facility: local5
- file-store:
version: 2
enabled: no
dir: /data/suricata/filestore
write-fileinfo: yes
force-filestore: yes
force-hash: [sha1, md5]
xff:
enabled: yes
mode: extra-data
deployment: reverse
header: X-Forwarded-For
- tcp-data:
enabled: no
type: both
filename: tcp-data.log
- http-body-data:
enabled: no
type: both
filename: http-data.log
- lua:
enabled: no
scripts:
logging:
default-log-level: notice
default-output-filter:
outputs:
- console:
enabled: yes
type: json
- file:
enabled: yes
level: info
filename: suricata.log
type: json
- syslog:
enabled: no
facility: local5
format: "[%i] <%d> -- "
type: json
af-packet:
- interface: ens33
threads: auto
cluster-id: 99
cluster-type: cluster_flow
defrag: yes
- interface: default
pcap:
- interface: ens33
- interface: default
pcap-file:
checksum-checks: auto
app-layer:
protocols:
rfb:
enabled: yes
detection-ports:
dp: 5900, 5901, 5902, 5903, 5904, 5905, 5906, 5907, 5908, 5909
mqtt:
enabled: yes
krb5:
enabled: yes
snmp:
enabled: yes
ikev2:
enabled: yes
tls:
enabled: yes
detection-ports:
dp: 443
ja3-fingerprints: yes
encryption-handling: bypass
dcerpc:
enabled: yes
ftp:
enabled: yes
rdp:
enabled: yes
ssh:
enabled: yes
http2:
enabled: yes
http1-rules: no
smtp:
enabled: yes
raw-extraction: no
mime:
decode-mime: yes
decode-base64: yes
decode-quoted-printable: yes
header-value-depth: 2000
extract-urls: yes
body-md5: no
inspected-tracker:
content-limit: 100000
content-inspect-min-size: 32768
content-inspect-window: 4096
imap:
enabled: detection-only
smb:
enabled: yes
detection-ports:
dp: 139, 445
nfs:
enabled: yes
tftp:
enabled: yes
dns:
tcp:
enabled: yes
detection-ports:
dp: 53
udp:
enabled: yes
detection-ports:
dp: 53
http:
enabled: yes
libhtp:
default-config:
personality: IDS
request-body-limit: 100kb
response-body-limit: 100kb
request-body-minimal-inspect-size: 32kb
request-body-inspect-window: 4kb
response-body-minimal-inspect-size: 40kb
response-body-inspect-window: 16kb
response-body-decompress-layer-limit: 2
http-body-inline: auto
swf-decompression:
enabled: yes
type: both
compress-depth: 100kb
decompress-depth: 100kb
double-decode-path: no
double-decode-query: no
server-config:
modbus:
enabled: no
detection-ports:
dp: 502
stream-depth: 0
dnp3:
enabled: no
detection-ports:
dp: 20000
enip:
enabled: no
detection-ports:
dp: 44818
sp: 44818
ntp:
enabled: yes
dhcp:
enabled: yes
sip:
enabled: yes
asn1-max-frames: 256
sensor-name: suricata_sensor
pid-file: /etc/suricata/suricata.pid
coredump:
max-dump: unlimited
host-mode: sniffer-only
runmode: autofp
default-packet-size: 1510
unix-command:
enabled: auto
geoip-database: /etc/suricata/GeoLite2-Country.mmdb
legacy:
uricontent: enabled
engine-analysis:
rules-fast-pattern: yes
rules: yes
pcre:
match-limit: 3500
match-limit-recursion: 1500
host-os-policy:
windows: [0.0.0.0/0]
bsd: []
bsd-right: []
old-linux: []
linux: []
old-solaris: []
solaris: []
hpux10: []
hpux11: []
irix: []
macos: []
vista: []
windows2k3: []
defrag:
memcap: 32mb
hash-size: 65536
trackers: 65535 # number of defragmented flows to follow
max-frags: 65535 # number of fragments to keep (higher than trackers)
prealloc: yes
timeout: 60
flow:
memcap: 128mb
hash-size: 65536
prealloc: 10000
emergency-recovery: 30
vlan:
use-for-tracking: true
flow-timeouts:
default:
new: 30
established: 300
closed: 0
bypassed: 100
emergency-new: 10
emergency-established: 100
emergency-closed: 0
emergency-bypassed: 50
tcp:
new: 60
established: 600
closed: 60
bypassed: 100
emergency-new: 5
emergency-established: 100
emergency-closed: 10
emergency-bypassed: 50
udp:
new: 30
established: 300
bypassed: 100
emergency-new: 10
emergency-established: 100
emergency-bypassed: 50
icmp:
new: 30
established: 300
bypassed: 100
emergency-new: 10
emergency-established: 100
emergency-bypassed: 50
stream:
memcap: 64mb
checksum-validation: yes # reject incorrect csums
inline: auto # auto will use inline mode in IPS mode, yes or no set it statically
reassembly:
memcap: 256mb
depth: 1mb # reassemble 1mb into a stream
toserver-chunk-size: 2560
toclient-chunk-size: 2560
randomize-chunk-size: yes
host:
hash-size: 4096
prealloc: 1000
memcap: 32mb
decoder:
teredo:
enabled: true
ports: $TEREDO_PORTS # syntax: '[3544, 1234]' or '3533' or 'any'.
vxlan:
enabled: true
ports: $VXLAN_PORTS # syntax: '[8472, 4789]' or '4789'.
vntag:
enabled: false
geneve:
enabled: true
ports: $GENEVE_PORTS # syntax: '[6081, 1234]' or '6081'.
detect:
profile: medium
custom-values:
toclient-groups: 3
toserver-groups: 25
sgh-mpm-context: auto
inspection-recursion-limit: 3000
prefilter:
default: mpm
grouping:
profiling:
grouping:
dump-to-disk: false
include-rules: false # very verbose
include-mpm-stats: false
mpm-algo: auto
spm-algo: auto
threading:
set-cpu-affinity: no
cpu-affinity:
- management-cpu-set:
cpu: [ 0 ] # include only these CPUs in affinity settings
- receive-cpu-set:
cpu: [ 0 ] # include only these CPUs in affinity settings
- worker-cpu-set:
cpu: [ "all" ]
mode: "exclusive"
prio:
low: [ 0 ]
medium: [ "1-2" ]
high: [ 3 ]
default: "medium"
detect-thread-ratio: 1.0
luajit:
states: 128
profiling:
rules:
enabled: no
filename: rule_perf.log
append: yes
limit: 10
json: yes
keywords:
enabled: no
filename: keyword_perf.log
append: yes
prefilter:
enabled: yes
filename: prefilter_perf.log
append: yes
rulegroups:
enabled: no
filename: rule_group_perf.log
append: yes
packets:
enabled: no
filename: packet_stats.log
append: yes
csv:
enabled: no
filename: packet_stats.csv
locks:
enabled: no
filename: lock_stats.log
append: yes
pcap-log:
enabled: no
filename: pcaplog_stats.log
append: yes
nfq:
nflog:
- group: 2
buffer-size: 18432
- group: default
qthreshold: 1
qtimeout: 100
max-size: 20000
capture:
netmap:
- interface: eth2
- interface: default
pfring:
- interface: ens33
threads: auto
cluster-id: 99
cluster-type: cluster_flow
- interface: default
ipfw:
napatech:
streams: ["0-3"]
enable-stream-stats: no
auto-config: yes
hardware-bypass: yes
inline: no
ports: [0-1,2-3]
hashmode: hash5tuplesorted
default-rule-path: /etc/suricata/rules
rule-files:
- suricata.rules
classification-file: /etc/suricata/classification.config
reference-config-file: /etc/suricata/reference.config
threshold-file: /etc/suricata/threshold.config
本文为 独立观点，未经允许不得转载，授权请联系FreeBuf客服小蜜蜂，微信：freebee2022
被以下专辑收录，发现更多精彩内容
+ 收入我的专辑
+ 加入我的收藏
相关推荐
- 0 文章数
- 0 关注者
文章目录