威胁猎杀实战（一）：平台 Secer's Blog - 记录互联网安全历程与个人成长经历

在国内Threat Hunting常被翻译成威胁追踪或威胁狩猎，我们认为：“未知攻焉知防，未知防焉知攻”。蓝方并不一定要处于被动防守的状态，完全可以主动猎杀对手！

本文是威胁猎杀实战系列的第一篇，按照本文的操作步骤，只需几次Copy&Paste即可搭建一套基于Elastic Stack的威胁猎杀平台。在后面的文章我们会进一步完善我们的平台。

NSM架构

1.部署Elastic Stack（容器化）

$ echo "nameserver 9.9.9.9" > /etc/resolv.conf
$ git clone https://github.com/Zer0d0y/docker-elk.git
$ docker-compose build && docker-compose up -d
访问Kibana web UI：http://localhost:5601
完整指南参考
https://github.com/Zer0d0y/docker-elk

2.部署Bro

2.1 安装

方式一：使用官方提供的Binary软件包

Ubuntu 16.04：
$ wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Install_Bro_from_binary.sh$ chmod +x Install_Bro_from_binary.sh
&& ./Install_Bro_from_binary.sh

Bro repository提供5个Binary软件包：

Bro，包含meta-package
bro-core，包含Bro core和scripts
broctl，包含Bro control
libbroccoli和libbroccoli-dev，包含libbroccoli及其开发头文件

Ubuntu 16.04：
$ wget -nv http://download.opensuse.org/repositories/network:bro/xUbuntu_16.04/Release.key -O Release.key
$ sudo apt-key add - < Release.key
$ sudo apt-get update
$ sudo sh -c "echo 'deb http://download.opensuse.org/repositories/network:/bro/xUbuntu_16.04/ /' > /etc/apt/sources.list.d/bro.list"
$ sudo apt-get update
$ sudo apt-get install bro
# 注意：官方同时提供nightly binary builds：https://www.bro.org/download/nightly-packages.html

方式二：源码安装

依赖软件包：

$ cmake make gcc g++ flex bison libpcap-dev libssl-dev python-dev swig zlib1g-dev

其他依赖软件包（可选）：

参考：https://www.bro.org/sphinx/install/install.html#id6

Ubuntu 16.04：
$ wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Install_Bro_from_source.sh
$ chmod +x Install_Bro_from_source.sh && ./Install_Bro_from_source.sh# 
注意：也可以安装Bro开发版：https://www.bro.org/sphinx/install/install.html#id9

方式三：容器化方式（Docker）

参考：https://github.com/bro/bro-docker

2.2 配置

2.2.1 Bro配置文件

$PREFIX == 默认值：/opt/bro或/usr/local/bro
配置监听网络接口：$PREFIX/etc/node.cfg
配置本地网络地址：$PREFIX/etc/networks.cfg
主配置文件：$PREFIX/etc/broctl.cfg
# 完整配置参考：https://www.zer0d0y.info/post/Bro-plus-ELK/

2.2.2 使用systemd管理Bro

# 修改Bro接口名称
$ INAME=$(ip -o link show | sed -rn '/^[0-9]+: en/{s/.: ([^:]*):.*/\1/p}')
$ sed -i "s/eth0/$INAME/g" /usr/local/bro/etc/node.cfg
$ wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Bro_systemd.service -O /etc/systemd/system/bro.service
$ systemctl daemon-reload   
$ systemctl enable bro  
$ systemctl start bro

3.整合Elastic Stack，[Kafka]和Bro

3.1 Bro日志101

conn.log -- IP, TCP, UDP, ICMP
dhcp.log -- DHCPdns.log -- DNS查询/响应
ftp.log -- FTP请求/响应
http.log -- HTTP请求/响应
files.log -- 文件还原
mysql.log -- MySQL
irc.log -- IRC
radius.log -- RADIUS认证
kerberos.log -- Kerberos认证
sip.log -- SIP协议
smtp.log -- SMTP事务
ssl.log -- SSL握手
ssh.log -- SSH握手
syslog.log -- Syslog消息
tunnel.log -- 封装隧道的细节
Microsoft相关的日志
dce_rpc.log -- DCE/RPC消息
ntlm.log -- NTLMrdp.log -- 远程桌面 (RDP)
smb_files.log -- SMB文件传输
smb_mapping.log -- SMB管道

# 详细解释：https://github.com/corelight/bro-cheatsheets

3.2 使用Elastic Stack直接处理Bro的csv格式日志

# 注意事项
1.端口开放（--> 防火墙）：
    elasticsearch:9200
    Logstash:5044
    Kibana:5061
    
2."index => "bro_logs-%{+YYYY.MM.dd}"",其中index名称必须小写
3.创建Index Patterns前必须有对应Bro的日志，否则会导致Field不全
# 注意事项

# 软件环境
Elastic Stack 6.4
bro version 2.5.4


# 方式一：使用Filebeat处理Bro日志，
数据流：
Bro --> Filebeat --> ELK(Logstash)


1.安装Filebeat

Ubuntu 16.04：
$ wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Install_Filebeat.sh
$ chmod +x Install_Filebeat.sh && ./Install_Filebeat.sh

2.配置ELK(Logstash)接收来自FileBeat收集的Bro日志
# 注意：此命令在ELK主机上执行
$ wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Deploy_Bro_Filebeat_Logstash.sh
$ chmod +x Deploy_Bro_Filebeat_Logstash.sh && ./Deploy_Bro_Filebeat_Logstash.sh
$ sed -i 's/8.8.8.8/ELK IP/g' Bro_Filebeat_Logstash.conf
$ systemctl start logstash.service


3.配置Filebeat处理Bro日志
$ wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Deploy_Filebeat.sh
$ chmod +x Deploy_Filebeat.sh && ./Deploy_Filebeat.sh
$ sed -i 's/8.8.8.8/ELK logstash IP/g' /etc/filebeat/filebeat.yml
$ service filebeat start

4.访问Kibana web UI：http://localhost:5601，添加"Index Patterns"
正常情况下，字段(Fields) >= 218


# 方式二：使用Logstash处理Bro日志，
数据流：
Bro --> Logstash --> ELK(Elasticsearch)

1.安装Logstash
Ubuntu 16.04：
$ wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Install_Logstash.sh
$ chmod +x Install_Logstash.sh && ./Install_Logstash.sh

2.配置Logstash处理Bro日志
# 注意：如ELK和Bro不在同一台服务器上，需要修改配置文件中elasticsearch的值，如： hosts => ["ELK IP:9200"]
# sed -i 's/localhost/ELK IP/g' bro*.conf

$ cd /etc/logstash/conf.d
$ wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Deploy_Logstash.sh
$ chmod +x Deploy_Logstash.sh && ./Deploy_Logstash.sh
$ rm -f Deploy_Logstash.sh

3.访问Kibana web UI：http://localhost:5601，添加“Index Patterns”

# 调试&排错
## Logstash
$ mkdir -p /root/xxx/logs && cd /root/xxx
$ /usr/share/logstash/bin/logstash -f xxx.conf --path.logs /root/xxx/logs --log.level=debug --config.debug --config.test_and_exit

$ /usr/share/logstash/bin/logstash -f nmap-logstash.conf --path.logs /root/xxx/logs/ --log.level=debug --config.debug 2>&1 | tee /root/xxx/logs/101
## FileBeat
$ filebeat -e -d "*" -c /etc/filebeat/filebeat.yml

# 容器化ELK项目对应配置（https://github.com/Zer0d0y/docker-elk）
1.docker-elk/docker-compose.yml
  logstash:
    ports:
      - "5044:5044"

2.docker-elk/logstash/pipeline/bro_logs.conf
3.docker-compose build

3.3 使用Elastic Stack + Kafka处理Bro的json格式日志

数据流：Bro --> Kafka --> Logstash --> ELK(Elasticsearch)

3.3.1 安装Kafka

# 软件环境：
# Ubuntu 16.04
# Elastic Stack 6.4
# Bro 2.5.5
# Kafka 2.12
# librdkafka-0.9.4

# 1.安装Kafka
# 创建临时目录
mkdir /src && cd /src

# 下载&验证kafka
wget https://archive.apache.org/dist/kafka/1.0.0/kafka_2.12-1.0.0.tgz
wget https://archive.apache.org/dist/kafka/1.0.0/kafka_2.12-1.0.0.tgz.asc
gpg --recv-keys  3B417B9B 
gpg -v kafka_2.12-1.0.0.tgz.asc

# 安装&启动kafka服务
tar -xf kafka_2.12-1.0.0.tgz
sudo mv kafka_2.12-1.0.0 /opt/kafka
sudo sed -i '/^log.dirs/{s/=.*//;}' /opt/kafka/config/server.properties
sudo sed -i 's/^log.dirs/log.dirs=\/var\/lib\/kafka/' /opt/kafka/config/server.properties
sudo sed -i '$alisteners=bro://BRO所在机器的IP地址:9092' /opt/kafka/config/server.properties 

cat > /etc/systemd/system/kafka.service << EOF
[Unit]
Description=Kafka Service
Wants=network.target
After=zookeeper.target

[Service]
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.pr
ExecReload=on-failure
Restart=always
User=root
Group=root
StandardOutput=syslog
StandardError=syslog

[Install]
WantedBy=multi-user.target
EOF

# 
sudo apt-get -y install zookeeperd
sudo systemctl enable zookeeper 
sudo systemctl start zookeeper
sudo systemctl daemon-reload
sudo systemctl enable kafka 
sudo systemctl start kafka

3.3.2 安装kafka插件（metron-bro-plugin-kafka）

## 安装librdkafka
curl -L https://github.com/edenhill/librdkafka/archive/v0.9.4.tar.gz | tar
cd librdkafka-0.9.4/ 
./configure --enable-sasl 
make 
sudo make install 
## 构建插件
### 先安装Bro 2.5.5
cd /src
wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Install_Bro_from_source.sh
chmod +x Install_Bro_from_source.sh && ./Install_Bro_from_source.sh


git clone https://github.com/apache/metron-bro-plugin-kafka.git
cd metron-bro-plugin-kafka
./configure --bro-dist=/src/bro-2.5.5/
make 
sudo make install
## 验证
/usr/local/bro/bin/bro -N Apache::Kafka

3.3.3 配置Bro把日志发送到Kafka

$ vi /usr/local/bro/share/bro/site/local.bro
@load /usr/local/bro/lib/bro/plugins/APACHE_KAFKA/scripts/Apache/Kafka/logs-to-kafka.bro
redef Kafka::topic_name = "";
redef Kafka::logs_to_send = set(Conn::LOG, HTTP::LOG, DNS::LOG, SMTP::LOG, SSL::LOG, Software::LOG, DHCP::LOG, FTP::LOG, IRC::LOG, Notice::LOG, X509::LOG, SSH::LOG,
redef Kafka::kafka_conf = table(["metadata.broker.list"] = "BRO所在机器的IP地址:9092");
redef Kafka::tag_json = T;

3.3.4 配置Logstash接收Kafka日志

## 先安装Logstash
$ wget https://github.com/tianyulab/Threat_Hunting_with_ELK/raw/master/Bro/Install_Logstash.sh
$ chmod +x Install_Logstash.sh && ./Install_Logstash.sh
$ echo config.reload.automatic: true |sudo tee -a /etc/logstash/logstash.yml
$ echo config.reload.interval: 3s |sudo tee -a /etc/logstash/logstash.yml
# 以Bro conn日志为例：
$ cat > /etc/logstash/conf.d/bro-conn.conf << EOF
input {
    kafka {
        topics => ["conn"]
        group_id => "bro_logstash"
            bootstrap_servers => "10.42.94.92:9092"
            codec => json
            type => "conn"
            auto_offset_reset => "earliest"
    }
}

output {
    if [type] == "conn" {
        elasticsearch {
            hosts => ["192.168.8.112:9200"]
            index => "bro-conn-%{+YYYY.MM.dd}"
        }
    }
}
EOF

3.3.5 一键部署脚本

$ wget 
# 修改10.42.94.92 --> 为Kafka监听IP
$ sed -i 's/10.42.94.92/Kafka监听IP/g' Deploy_Kafka_for_Bro.sh
# 修改192.168.8.112 --> 为Elasticsearch监听IP
$ sed -i 's/192.168.8.112/Elasticsearch监听IP/g' Deploy_Kafka_for_Bro.sh
# 修改"BRO所在机器的IP地址"为BRO所在机器的IP地址
$ sed -i 's/BRO所在机器的IP地址/BRO所在机器的IP地址/g' Deploy_Kafka_for_Bro.sh
$ sh -x Deploy_Kafka_for_Bro.sh

# 验证
$ sudo systemctl status zookeeper
$ sudo systemctl status kafka
$ systemctl status logstash
$ /usr/local/bro/bin/bro -N Apache::Kafka
$ /usr/local/bro/bin/broctl status
$ netstat -tunlp  | grep -E '2181|9092|9600'

# 安装过程排错
$ watch tail log.out
$ cat log.out | grep error
$ cat log.out | grep -B 10 "Configuring incomplete, errors occurred"
$ cat log.out | grep -i "cd librdkafka-0.9.4" -A 50 | more

# Kafka 排错
$ apt-get install kafkacat
$ kafkacat -b 192.168.8.115:9092 -t http -o end # "http"为Bro的kafka插件定义的"topics"
或
$ /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.8.115:9092 --topic http

致谢：

@HardenedLinux 团队

@Rock NSM团队

@Security Onion团队

威胁猎杀实战（一）：平台