performing-network-traffic-analysis-with-zeek

使用 Zeek 执行网络流量分析

概述

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

何时使用

• 当进行涉及使用 zeek 执行网络流量分析的安全评估时
• 当遵循相关安全事件的事件响应流程时
• 当执行计划内的安全测试或审计活动时
• 当通过动手测试验证安全控制措施时

前提条件

• Linux 服务器（Ubuntu 22.04+ 或 CentOS 8+），具有 4+ CPU 核心和 8GB+ 内存
• 已配置用于流量捕获的网络 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	可移植可执行文件元数据
dpd.log	动态协议检测失败

工作流

步骤 1：安装和配置 Zeek

# Install Zeek on Ubuntu
sudo apt-get install -y zeek

# Or install from Zeek repository
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

# Verify installation
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 中配置网络定义：

# Internal network ranges
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 standard detection scripts
@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 file analysis
@load base/files/hash-all-files
@load base/files/extract-all-files

# Load detection frameworks
@load base/frameworks/notice
@load base/frameworks/intel
@load base/frameworks/files
@load base/frameworks/software

# Load additional protocol analyzers
@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

# Enable JA3 fingerprinting
@load policy/protocols/ssl/ja3

# Enable JSON output for SIEM ingestion
@load policy/tuning/json-logs

redef LogAscii::use_json = T;

# Configure file extraction directory
redef FileExtract::prefix = "/opt/zeek/extracted/";

# Set notice email
redef Notice::mail_dest = "[email protected]";

步骤 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
    };

    # Threshold for suspicious DNS query length
    const query_len_threshold = 50 &redef;

    # Track query counts per host per domain
    global dns_query_counts: table[addr, string] of count &default=0 &create_expire=5min;

    # High query volume threshold
    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("Possible DNS tunneling: %s queries to %s with long query names",
                                 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
    };

    # Track connection intervals
    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;
            }

            # Calculate mean and standard deviation
            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("Possible C2 beaconing: %s -> %s:%s (interval=%.1fs, jitter=%.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：配置 Intel 框架

将威胁情报源加载到 Zeek 中：

# In 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",
};

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：部署和运行

# Deploy Zeek cluster
sudo /opt/zeek/bin/zeekctl deploy

# Check cluster status
sudo /opt/zeek/bin/zeekctl status

# Process offline PCAP
zeek -r capture.pcap local.zeek

# View logs
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

# Search for specific connections
cat /opt/zeek/logs/current/dns.log | zeek-cut query answers | grep -i "suspicious"

# Rotate logs
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-*"

分析技术

连接分析

# Find top talkers by bytes
cat conn.log | zeek-cut id.orig_h orig_bytes | sort -t$'\t' -k2 -rn | head -20

# Find long-duration connections (potential C2)
cat conn.log | zeek-cut id.orig_h id.resp_h id.resp_p duration | awk '$4 > 3600' | sort -t$'\t' -k4 -rn

# Find connections with unusual ports
cat conn.log | zeek-cut id.resp_p proto | sort | uniq -c | sort -rn | head -30

TLS 分析

# Find self-signed certificates
cat ssl.log | zeek-cut server_name validation_status | grep "self signed"

# Extract JA3 fingerprints for known malware
cat ssl.log | zeek-cut ja3 server_name | sort | uniq -c | sort -rn

# Find expired certificates
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 文档
• Zeek 脚本参考
• Zeek 情报框架
• CISA Zeek 资源
• Zeek GitHub 仓库