freeBuf
流量分析工具(二):centos7 + suricata + redis + ELK
2023-04-20 15:36:33
所属地 江苏省

centos7+suricata+redis+ELK

架构设计

1. 需求

目前有一个针对网络流量的攻击日志回溯项目,要求部署一套支持全流量解析并快速检索的系统,计划是解析实时流量并将解析后的json日志存储到 ELK中。

  1. 要求包含比较详细的应用层信息,比如 HTTP 的URL、请求头、响应头、请求体和响应体。

  2. 支持实时抓包,并存储成pcap的功能,以供本地使用wireshark回溯分析。

  3. 支持在大流量背景下的全包存储能力,类似于tcpdump的抓包保存pcap功能。

  4. 支持网络流量的威胁检测能力

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_alertsuricata_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
被以下专辑收录,发现更多精彩内容
+ 收入我的专辑
+ 加入我的收藏
文章目录