performing-network-traffic-analysis-with-zeek

name	performing-network-traffic-analysis-with-zeek
description	部署 Zeek 网络安全监控器，捕获、解析和分析网络流量元数据，用于威胁检测、异常识别和取证调查。
domain	cybersecurity
subdomain	network-security
tags	["zeek","network-monitoring","traffic-analysis","ids","nids","pcap","threat-detection","forensics","siem-integration"]
version	1.0
author	mahipal
license	Apache-2.0

使用 Zeek 进行网络流量分析

概述

Zeek（原名 Bro）是一款开源网络分析框架，以被动网络安全监控器模式运行。与传统的基于签名的 IDS 工具不同，Zeek 从观测到的网络流量中生成高保真度的结构化日志，为 HTTP、DNS、TLS、SSH、SMTP、FTP 等数十种协议捕获详细元数据。Zeek 可扩展的脚本语言支持自定义检测逻辑、行为分析和自动响应。本技能涵盖部署 Zeek、理解其日志架构、编写自定义检测脚本以及与 SIEM 平台集成输出。

前置条件

• 配备 4 核以上 CPU 和 8GB 以上内存的 Linux 服务器（Ubuntu 22.04+ 或 CentOS 8+）
• 已配置用于流量捕获的网络 TAP 或 SPAN 端口镜像
• 已安装 Zeek 6.0+（通过包管理器或源码编译）
• 具备用于数据包捕获的 root 或 capture 组权限
• 用于日志摄入的 SIEM 平台（Splunk、ELK Stack 或 QRadar）

核心概念

Zeek 架构

Zeek 以两种主要模式运行：

1. 实时捕获 — 实时监控一个或多个网络接口上的流量
2. 离线分析 — 处理保存的 PCAP 文件以进行回溯分析

处理管道由以下部分组成：

• 数据包捕获层 — 从接口或 PCAP 文件读取原始数据包
• 事件引擎 — 重组 TCP 流并生成协议事件
• 脚本解释器 — 执行处理事件并生成日志的 Zeek 脚本
• 日志框架 — 以 TSV、JSON 或自定义格式写入结构化日志

日志架构

Zeek 生成按协议分类的日志文件：

日志文件	描述
conn.log	TCP/UDP/ICMP 连接摘要，包含持续时间、字节数和状态
dns.log	DNS 查询和响应，包含查询类型、答案和 TTL
http.log	HTTP 请求/响应，包含 URI、用户代理和 MIME 类型
ssl.log	TLS 握手详情，包含证书链、JA3/JA3S 指纹
files.log	文件传输，包含 MIME 类型和哈希值（MD5、SHA1、SHA256）
notice.log	Zeek 检测脚本生成的告警
weird.log	协议异常和意外行为
x509.log	TLS 连接的证书详情
smtp.log	邮件元数据，包含发件人、收件人和主题
ssh.log	SSH 连接详情和认证结果
pe.log	PE（可移植可执行文件）文件元数据
dpd.log	动态协议检测失败记录

步骤

步骤 1：安装和配置 Zeek

# 在 Ubuntu 上安装 Zeek
sudo apt-get install -y zeek

# 或从 Zeek 仓库安装
echo 'deb http://download.opensuse.org/repositories/security:/zeek/xUbuntu_22.04/ /' | \
    sudo tee /etc/apt/sources.list.d/zeek.list
sudo apt-get update && sudo apt-get install -y zeek-lts

# 验证安装
zeek --version

在 /opt/zeek/etc/node.cfg 中配置节点布局：

[manager]
type=manager
host=localhost

[proxy-1]
type=proxy
host=localhost

[worker-1]
type=worker
host=localhost
interface=eth0
lb_method=pf_ring
lb_procs=4

[worker-2]
type=worker
host=localhost
interface=eth1
lb_method=pf_ring
lb_procs=4

在 /opt/zeek/etc/networks.cfg 中配置网络定义：

# 内网地址段
10.0.0.0/8         Private RFC1918
172.16.0.0/12      Private RFC1918
192.168.0.0/16     Private RFC1918

步骤 2：配置日志和输出

编辑 /opt/zeek/share/zeek/site/local.zeek：

# 加载标准检测脚本
@load base/protocols/conn
@load base/protocols/dns
@load base/protocols/http
@load base/protocols/ssl
@load base/protocols/ssh
@load base/protocols/smtp
@load base/protocols/ftp

# 加载文件分析
@load base/files/hash-all-files
@load base/files/extract-all-files

# 加载检测框架
@load base/frameworks/notice
@load base/frameworks/intel
@load base/frameworks/files
@load base/frameworks/software

# 加载额外协议分析器
@load policy/protocols/ssl/validate-certs
@load policy/protocols/ssl/log-hostcerts-only
@load policy/protocols/ssh/detect-bruteforcing
@load policy/protocols/dns/detect-external-names
@load policy/protocols/http/detect-sqli

# 启用 JA3 指纹
@load policy/protocols/ssl/ja3

# 启用 JSON 输出以便 SIEM 摄入
@load policy/tuning/json-logs

redef LogAscii::use_json = T;

# 配置文件提取目录
redef FileExtract::prefix = "/opt/zeek/extracted/";

# 设置告警邮件
redef Notice::mail_dest = "[e-mail]";

步骤 3：编写自定义检测脚本

为常见威胁创建检测脚本：

检测 DNS 隧道（/opt/zeek/share/zeek/site/detect-dns-tunnel.zeek）：

@load base/protocols/dns

module DNSTunnel;

export {
    redef enum Notice::Type += {
        DNS_Tunnel_Suspected
    };

    # 可疑 DNS 查询长度阈值
    const query_len_threshold = 50 &redef;

    # 按主机和域跟踪查询数量
    global dns_query_counts: table[addr, string] of count &default=0 &create_expire=5min;

    # 高查询量阈值
    const query_volume_threshold = 100 &redef;
}

event dns_request(c: connection, msg: dns_msg, query: string, qtype: count, qclass: count)
{
    if ( |query| > query_len_threshold )
    {
        local parts = split_string(query, /\./);
        if ( |parts| > 3 )
        {
            local base_domain = cat(parts[|parts|-2], ".", parts[|parts|-1]);
            dns_query_counts[c$id$orig_h, base_domain] += 1;

            if ( dns_query_counts[c$id$orig_h, base_domain] > query_volume_threshold )
            {
                NOTICE([$note=DNS_Tunnel_Suspected,
                        $msg=fmt("可能存在 DNS 隧道：%s 对 %s 发出大量长查询",
                                 c$id$orig_h, base_domain),
                        $conn=c,
                        $identifier=cat(c$id$orig_h, base_domain),
                        $suppress_for=30min]);
            }
        }
    }
}

检测信标行为（/opt/zeek/share/zeek/site/detect-beaconing.zeek）：

@load base/protocols/conn

module Beaconing;

export {
    redef enum Notice::Type += {
        C2_Beacon_Detected
    };

    # 跟踪连接间隔
    global conn_intervals: table[addr, addr, port] of vector of time &create_expire=1hr;

    const min_connections = 20 &redef;
    const jitter_threshold = 0.15 &redef;
}

event connection_state_remove(c: connection)
{
    if ( c$id$resp_p == 80/tcp || c$id$resp_p == 443/tcp )
    {
        local key = [c$id$orig_h, c$id$resp_h, c$id$resp_p];

        if ( key !in conn_intervals )
            conn_intervals[key] = vector();

        conn_intervals[key] += network_time();

        if ( |conn_intervals[key]| >= min_connections )
        {
            local intervals: vector of interval = vector();
            local i = 1;
            while ( i < |conn_intervals[key]| )
            {
                intervals += conn_intervals[key][i] - conn_intervals[key][i-1];
                i += 1;
            }

            local sum_val = 0.0;
            for ( idx in intervals )
                sum_val += interval_to_double(intervals[idx]);

            local mean_val = sum_val / |intervals|;

            local variance = 0.0;
            for ( idx in intervals )
            {
                local diff = interval_to_double(intervals[idx]) - mean_val;
                variance += diff * diff;
            }
            variance = variance / |intervals|;
            local stddev = sqrt(variance);

            if ( mean_val > 0 && (stddev / mean_val) < jitter_threshold )
            {
                NOTICE([$note=C2_Beacon_Detected,
                        $msg=fmt("可能存在 C2 信标：%s -> %s:%s（间隔=%.1fs，抖动=%.2f）",
                                 c$id$orig_h, c$id$resp_h, c$id$resp_p,
                                 mean_val, stddev/mean_val),
                        $conn=c,
                        $identifier=cat(c$id$orig_h, c$id$resp_h),
                        $suppress_for=1hr]);
            }
        }
    }
}

步骤 4：配置情报框架

将威胁情报导入 Zeek：

# 在 local.zeek 中
@load frameworks/intel/seen
@load frameworks/intel/do_notice

redef Intel::read_files += {
    "/opt/zeek/intel/malicious-ips.intel",
    "/opt/zeek/intel/malicious-domains.intel",
    "/opt/zeek/intel/malicious-hashes.intel",
};

情报文件格式（/opt/zeek/intel/malicious-ips.intel）：

#fields	indicator	indicator_type	meta.source	meta.desc	meta.do_notice
198.51.100.50	Intel::ADDR	abuse.ch	Known C2 server	T
203.0.113.100	Intel::ADDR	threatfeed	Ransomware infrastructure	T

步骤 5：部署和运维

# 部署 Zeek 集群
sudo /opt/zeek/bin/zeekctl deploy

# 检查集群状态
sudo /opt/zeek/bin/zeekctl status

# 处理离线 PCAP
zeek -r capture.pcap local.zeek

# 查看日志
cat /opt/zeek/logs/current/conn.log | zeek-cut id.orig_h id.resp_h id.resp_p proto service duration orig_bytes resp_bytes

# 搜索特定连接
cat /opt/zeek/logs/current/dns.log | zeek-cut query answers | grep -i "suspicious"

# 轮转日志
sudo /opt/zeek/bin/zeekctl cron

步骤 6：SIEM 集成

用于 ELK Stack 的 Filebeat 配置：

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /opt/zeek/logs/current/*.log
    json.keys_under_root: true
    json.add_error_key: true
    fields:
      source: zeek
    fields_under_root: true

output.elasticsearch:
  hosts: ["https://elasticsearch:9200"]
  index: "zeek-%{+yyyy.MM.dd}"

setup.template.name: "zeek"
setup.template.pattern: "zeek-*"

分析技术

连接分析

# 按字节数查找高流量主机
cat conn.log | zeek-cut id.orig_h orig_bytes | sort -t$'\t' -k2 -rn | head -20

# 查找长时间连接（可能的 C2）
cat conn.log | zeek-cut id.orig_h id.resp_h id.resp_p duration | awk '$4 > 3600' | sort -t$'\t' -k4 -rn

# 查找使用异常端口的连接
cat conn.log | zeek-cut id.resp_p proto | sort | uniq -c | sort -rn | head -30

TLS 分析

# 查找自签名证书
cat ssl.log | zeek-cut server_name validation_status | grep "self signed"

# 提取已知恶意软件的 JA3 指纹
cat ssl.log | zeek-cut ja3 server_name | sort | uniq -c | sort -rn

# 查找过期证书
cat ssl.log | zeek-cut server_name not_valid_after | awk -F'\t' '$2 < systime()'

最佳实践

• TAP 优于 SPAN — 使用网络 TAP 而非 SPAN 端口以避免高负载时丢包
• Worker 扩展 — 每 1 Gbps 监控流量分配 1 个 Zeek worker
• AF_PACKET 集群 — 使用带负载均衡的 AF_PACKET 进行多核处理
• 日志轮转 — 配置自动日志轮转和归档（默认：每小时）
• 情报更新 — 至少每日自动更新威胁情报
• 丢包监控 — 监控 capture_loss.log 中的丢包情况
• 自定义脚本 — 根据组织的威胁场景开发定制检测逻辑

参考资料

• Zeek 文档：https://docs.zeek.org/en/master/
• Zeek 脚本参考：https://docs.zeek.org/en/master/script-reference/index.html
• Zeek 情报框架：https://docs.zeek.org/en/master/frameworks/intel.html
• Zeek GitHub 仓库：https://github.com/zeek/zeek