Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

Overview

Zeek (formerly Bro) is an open-source network analysis framework that operates as a passive network security monitor. Unlike traditional signature-based IDS tools, Zeek generates high-fidelity structured logs from observed network traffic, capturing detailed metadata for protocols including HTTP, DNS, TLS, SSH, SMTP, FTP, and dozens more. Zeek's extensible scripting language enables custom detection logic, behavioral analysis, and automated response. This skill covers deploying Zeek, understanding its log architecture, writing custom detection scripts, and integrating outputs with SIEM platforms.

When to Use

When conducting security assessments that involve performing network traffic analysis with zeek
When following incident response procedures for related security events
When performing scheduled security testing or auditing activities
When validating security controls through hands-on testing

Prerequisites

Linux server (Ubuntu 22.04+ or CentOS 8+) with 4+ CPU cores and 8GB+ RAM
Network TAP or SPAN port mirroring configured for traffic capture
Zeek 6.0+ installed (via package manager or source compilation)
Root or capture group privileges for packet capture
SIEM platform (Splunk, ELK Stack, or QRadar) for log ingestion

Core Concepts

Zeek Architecture

Zeek operates in two main modes:

Live Capture - Monitors traffic in real-time on one or more network interfaces
Offline Analysis - Processes saved PCAP files for retrospective analysis

The processing pipeline consists of:

Packet Capture Layer - Reads raw packets from interfaces or PCAP files
Event Engine - Reassembles TCP streams and generates protocol events
Script Interpreter - Executes Zeek scripts that process events and generate logs
Log Framework - Writes structured logs in TSV, JSON, or custom formats

Log Architecture

Zeek generates protocol-specific log files:

Log File	Description
`conn.log`	TCP/UDP/ICMP connection summaries with duration, bytes, state
`dns.log`	DNS queries and responses with query type, answers, TTL
`http.log`	HTTP requests/responses with URIs, user agents, MIME types
`ssl.log`	TLS handshake details including certificate chain, JA3/JA3S
`files.log`	File transfers with MIME types, hashes (MD5, SHA1, SHA256)
`notice.log`	Alerts generated by Zeek detection scripts
`weird.log`	Protocol anomalies and unexpected behaviors
`x509.log`	Certificate details from TLS connections
`smtp.log`	Email metadata including sender, recipient, subject
`ssh.log`	SSH connection details and authentication results
`pe.log`	Portable Executable file metadata
`dpd.log`	Dynamic Protocol Detection failures

Workflow

Step 1: Install and Configure 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

Configure the node layout in /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

Configure network definitions in /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

Step 2: Configure Logging and Output

Edit /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 = "soc@example.com";

Step 3: Write Custom Detection Scripts

Create detection scripts for common threats:

Detect DNS Tunneling (/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]);
            }
        }
    }
}

Detect Beaconing Behavior (/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]);
            }
        }
    }
}

Step 4: Configure Intel Framework

Load threat intelligence feeds into 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 file format (/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

Step 5: Deploy and Operate

# 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

Step 6: SIEM Integration

Filebeat configuration for ELK Stack:

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-*"

Analysis Techniques

Connection Analysis

# 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 Analysis

# 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()'

Best Practices

TAP Over SPAN - Use network TAPs instead of SPAN ports to avoid packet loss under load
Worker Scaling - Assign 1 Zeek worker per 1 Gbps of monitored traffic
AF_PACKET Clusters - Use AF_PACKET with load balancing for multi-core processing
Log Rotation - Configure automatic log rotation and archival (default: hourly)
Intel Updates - Automate threat intelligence feed updates at least daily
Packet Loss Monitoring - Monitor capture_loss.log for dropped packets
Custom Scripts - Develop organization-specific detections based on threat landscape

References

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.7 KB

API Reference — Performing Network Traffic Analysis with Zeek

Libraries Used

pathlib: Read Zeek TSV log files
subprocess: Execute Zeek on PCAP files
collections.Counter: Traffic pattern aggregation

CLI Interface

python agent.py conn --log conn.log
python agent.py dns --log dns.log
python agent.py http --log http.log
python agent.py notice --log notice.log
python agent.py run --pcap capture.pcap [--output-dir /tmp/zeek_output]

Core Functions

`parse_zeek_log(log_file)` — Generic Zeek TSV parser

Parses #fields header and data rows. Returns headers and record list.

`analyze_conn_log(conn_log)` — Connection analysis

Statistics: protocols, services, top IPs/ports, total bytes, long connections (>1hr).

`analyze_dns_log(dns_log)` — DNS query analysis

Detects: long queries (>50 chars), TXT queries, NXDOMAIN responses. Flags potential DNS tunneling indicators.

`analyze_http_log(http_log)` — Web traffic analysis

Tracks: methods, status codes, top hosts, user agents. Flags suspicious UAs: curl, wget, python, powershell, certutil, bitsadmin.

`analyze_notice_log(notice_log)` — Security alert review

Parses Zeek notice.log for detected security events.

`run_zeek_on_pcap(pcap_file, output_dir)` — Generate Zeek logs from PCAP

Executes Zeek against PCAP to produce conn.log, dns.log, http.log, etc.

Zeek Log Fields

Log	Key Fields
conn.log	id.orig_h, id.resp_h, id.resp_p, proto, service, duration, orig_bytes
dns.log	query, qtype_name, rcode_name
http.log	method, host, uri, status_code, user_agent
notice.log	note, msg, src, dst

Dependencies

System: zeek (for PCAP processing) No Python packages required.

Scripts 1

agent.py6.7 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
"""Agent for performing network traffic analysis with Zeek (Bro) log files."""

import json
import argparse
import subprocess
from pathlib import Path
from collections import Counter


def parse_zeek_log(log_file, delimiter="\t"):
    """Parse a Zeek TSV log file into structured records."""
    lines = Path(log_file).read_text(encoding="utf-8", errors="replace").splitlines()
    headers = []
    records = []
    for line in lines:
        if line.startswith("#fields"):
            headers = line.split(delimiter)[1:]
        elif line.startswith("#"):
            continue
        elif headers:
            values = line.split(delimiter)
            record = dict(zip(headers, values))
            records.append(record)
    return headers, records


def analyze_conn_log(conn_log):
    """Analyze Zeek conn.log for network connection patterns."""
    headers, records = parse_zeek_log(conn_log)
    total = len(records)
    protocols = Counter(r.get("proto", "") for r in records)
    services = Counter(r.get("service", "-") for r in records)
    src_ips = Counter(r.get("id.orig_h", "") for r in records)
    dst_ips = Counter(r.get("id.resp_h", "") for r in records)
    dst_ports = Counter(r.get("id.resp_p", "") for r in records)
    total_bytes = sum(int(r.get("orig_bytes", 0) or 0) + int(r.get("resp_bytes", 0) or 0) for r in records)
    long_connections = [r for r in records if float(r.get("duration", 0) or 0) > 3600]
    return {
        "log_file": conn_log, "total_connections": total,
        "protocols": dict(protocols), "services": dict(services.most_common(10)),
        "top_src_ips": dict(src_ips.most_common(10)),
        "top_dst_ips": dict(dst_ips.most_common(10)),
        "top_dst_ports": dict(dst_ports.most_common(15)),
        "total_bytes": total_bytes,
        "long_connections": len(long_connections),
    }


def analyze_dns_log(dns_log):
    """Analyze Zeek dns.log for DNS query patterns and anomalies."""
    headers, records = parse_zeek_log(dns_log)
    queries = Counter(r.get("query", "") for r in records)
    qtypes = Counter(r.get("qtype_name", r.get("qtype", "")) for r in records)
    rcodes = Counter(r.get("rcode_name", r.get("rcode", "")) for r in records)
    long_queries = [r for r in records if len(r.get("query", "")) > 50]
    txt_queries = [r for r in records if r.get("qtype_name", "") == "TXT"]
    nxdomain = [r for r in records if r.get("rcode_name", "") == "NXDOMAIN"]
    top_domains = Counter()
    for r in records:
        query = r.get("query", "")
        parts = query.rsplit(".", 2)
        if len(parts) >= 2:
            top_domains[".".join(parts[-2:])] += 1
    return {
        "log_file": dns_log, "total_queries": len(records),
        "query_types": dict(qtypes),
        "response_codes": dict(rcodes),
        "top_queried_domains": dict(top_domains.most_common(15)),
        "long_queries": len(long_queries),
        "txt_queries": len(txt_queries),
        "nxdomain_count": len(nxdomain),
        "potential_tunneling": len(long_queries) + len(txt_queries),
    }


def analyze_http_log(http_log):
    """Analyze Zeek http.log for web traffic patterns."""
    headers, records = parse_zeek_log(http_log)
    methods = Counter(r.get("method", "") for r in records)
    status_codes = Counter(r.get("status_code", "") for r in records)
    hosts = Counter(r.get("host", "") for r in records)
    user_agents = Counter(r.get("user_agent", "")[:100] for r in records)
    suspicious_ua = [r for r in records if any(kw in r.get("user_agent", "").lower()
                     for kw in ["curl", "wget", "python", "powershell", "certutil", "bitsadmin"])]
    return {
        "log_file": http_log, "total_requests": len(records),
        "methods": dict(methods), "status_codes": dict(status_codes),
        "top_hosts": dict(hosts.most_common(15)),
        "top_user_agents": dict(user_agents.most_common(10)),
        "suspicious_user_agents": len(suspicious_ua),
        "suspicious_requests": [{"host": r.get("host"), "uri": r.get("uri", "")[:100],
                                  "ua": r.get("user_agent", "")[:100]} for r in suspicious_ua[:10]],
    }


def analyze_notice_log(notice_log):
    """Analyze Zeek notice.log for security alerts."""
    headers, records = parse_zeek_log(notice_log)
    notice_types = Counter(r.get("note", r.get("msg", "")) for r in records)
    return {
        "log_file": notice_log, "total_notices": len(records),
        "notice_types": dict(notice_types),
        "notices": [{"note": r.get("note"), "msg": r.get("msg", "")[:200],
                      "src": r.get("src", r.get("id.orig_h", "")),
                      "dst": r.get("dst", r.get("id.resp_h", ""))} for r in records[:20]],
    }


def run_zeek_on_pcap(pcap_file, output_dir="/tmp/zeek_output"):
    """Run Zeek on a PCAP file to generate logs."""
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    cmd = ["zeek", "-r", pcap_file, f"Log::default_logdir={output_dir}"]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=300, cwd=output_dir)
        logs = list(Path(output_dir).glob("*.log"))
        return {
            "pcap_file": pcap_file, "output_dir": output_dir,
            "logs_generated": [l.name for l in logs],
            "success": result.returncode == 0,
            "stderr": result.stderr[:300] if result.stderr else "",
        }
    except FileNotFoundError:
        return {"error": "zeek not found in PATH"}
    except Exception as e:
        return {"error": str(e)}


def main():
    parser = argparse.ArgumentParser(description="Zeek Network Traffic Analysis Agent")
    sub = parser.add_subparsers(dest="command")
    c = sub.add_parser("conn", help="Analyze conn.log")
    c.add_argument("--log", required=True)
    d = sub.add_parser("dns", help="Analyze dns.log")
    d.add_argument("--log", required=True)
    h = sub.add_parser("http", help="Analyze http.log")
    h.add_argument("--log", required=True)
    n = sub.add_parser("notice", help="Analyze notice.log")
    n.add_argument("--log", required=True)
    r = sub.add_parser("run", help="Run Zeek on PCAP")
    r.add_argument("--pcap", required=True)
    r.add_argument("--output-dir", default="/tmp/zeek_output")
    args = parser.parse_args()
    if args.command == "conn":
        result = analyze_conn_log(args.log)
    elif args.command == "dns":
        result = analyze_dns_log(args.log)
    elif args.command == "http":
        result = analyze_http_log(args.log)
    elif args.command == "notice":
        result = analyze_notice_log(args.log)
    elif args.command == "run":
        result = run_zeek_on_pcap(args.pcap, args.output_dir)
    else:
        parser.print_help()
        return
    print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()

Keep exploring