Security Skills Atlas
security operations

Detecting Beaconing Patterns with Zeek

Performs statistical analysis of Zeek conn.log connection intervals to detect C2 beaconing patterns. Uses the ZAT library to load Zeek logs into Pandas DataFrames, calculates inter-arrival time standard deviation, and flags periodic connections with low jitter. Use when hunting for command-and-control callbacks in network data.

c2-beaconingconn-log-analysisnetwork-securitystatistical-analysisthreat-huntingzatzeek
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1008 on the official MITRE ATT&CK siteT1071.001 on the official MITRE ATT&CK siteT1071.004 on the official MITRE ATT&CK siteT1095 on the official MITRE ATT&CK siteT1573 on the official MITRE ATT&CK site

NIST CSF 2.0

DE.AE-02 on the official NIST CSF 2.0 siteDE.CM-01 on the official NIST CSF 2.0 siteGV.OV-01 on the official NIST CSF 2.0 siteRS.MA-01 on the official NIST CSF 2.0 site

When to Use

  • When investigating security incidents that require detecting beaconing patterns with zeek
  • When building detection rules or threat hunting queries for this domain
  • When SOC analysts need structured procedures for this analysis type
  • When validating security monitoring coverage for related attack techniques

Prerequisites

  • Familiarity with security operations concepts and tools
  • Access to a test or lab environment for safe execution
  • Python 3.8+ with required dependencies installed
  • Appropriate authorization for any testing activities

Instructions

Load Zeek conn.log data using ZAT (Zeek Analysis Tools), group connections by source/destination pairs, and compute timing statistics to identify beaconing.

from zat.log_to_dataframe import LogToDataFrame
import numpy as np
 
log_to_df = LogToDataFrame()
conn_df = log_to_df.create_dataframe('/path/to/conn.log')
 
# Group by src/dst pair and calculate inter-arrival time
for (src, dst), group in conn_df.groupby(['id.orig_h', 'id.resp_h']):
    times = group['ts'].sort_values()
    intervals = times.diff().dt.total_seconds().dropna()
    if len(intervals) > 10:
        std_dev = np.std(intervals)
        mean_interval = np.mean(intervals)
        # Low std_dev relative to mean = likely beaconing

Key analysis steps:

  1. Parse Zeek conn.log into DataFrame with ZAT LogToDataFrame
  2. Group connections by source IP and destination IP pairs
  3. Calculate inter-arrival time intervals between consecutive connections
  4. Compute standard deviation and coefficient of variation
  5. Flag pairs with low coefficient of variation as potential beacons

Examples

from zat.log_to_dataframe import LogToDataFrame
log_to_df = LogToDataFrame()
df = log_to_df.create_dataframe('conn.log')
print(df[['id.orig_h', 'id.resp_h', 'ts', 'duration']].head())
Source materials

References and resources

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

References 1

api-reference.md1.8 KB

API Reference: Detecting Beaconing Patterns with Zeek

ZAT (Zeek Analysis Tools)

from zat.log_to_dataframe import LogToDataFrame
from zat import zeek_log_reader
from zat.utils import dataframe_to_matrix
 
# Load conn.log into DataFrame
log_to_df = LogToDataFrame()
conn_df = log_to_df.create_dataframe('/path/to/conn.log')
 
# Select specific columns
conn_df = log_to_df.create_dataframe('conn.log',
    usecols=['id.orig_h', 'id.resp_h', 'id.resp_p', 'ts', 'duration'])
 
# Read rows as dicts (streaming)
reader = zeek_log_reader.ZeekLogReader('conn.log')
for row in reader.readrows():
    print(row)
 
# Tail mode (live monitoring)
reader = zeek_log_reader.ZeekLogReader('conn.log', tail=True)
for row in reader.readrows():
    process(row)
 
# Convert to matrix for ML
to_matrix = dataframe_to_matrix.DataFrameToMatrix()
matrix = to_matrix.fit_transform(conn_df[features])

Beaconing Detection Math

import numpy as np
 
intervals = times.diff().dt.total_seconds().dropna().values
std_dev = np.std(intervals)
mean_val = np.mean(intervals)
cv = std_dev / mean_val  # Coefficient of Variation
# cv < 0.3 = likely beacon (low jitter relative to interval)

Key Zeek Log Fields

Log Key Fields
conn.log id.orig_h, id.resp_h, id.resp_p, ts, duration, orig_bytes
dns.log id.orig_h, query, qtype_name, answers, ts
ssl.log id.orig_h, server_name, ja3, ja3s, ts

Anomaly Detection with ZAT + scikit-learn

from sklearn.ensemble import IsolationForest
odd_clf = IsolationForest(contamination=0.35)
odd_clf.fit(zeek_matrix)
anomalies = conn_df[odd_clf.predict(zeek_matrix) == -1]

References

Scripts 1

agent.py6.3 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Agent for detecting C2 beaconing patterns in Zeek conn.log data."""

import json
import argparse
from datetime import datetime

import numpy as np
import pandas as pd
from zat.log_to_dataframe import LogToDataFrame


def load_conn_log(log_path):
    """Load Zeek conn.log into a Pandas DataFrame using ZAT."""
    log_to_df = LogToDataFrame()
    df = log_to_df.create_dataframe(log_path)
    return df


def calculate_beacon_score(intervals):
    """Calculate a beacon score based on interval regularity."""
    if len(intervals) < 5:
        return 0.0
    std_dev = np.std(intervals)
    mean_val = np.mean(intervals)
    if mean_val == 0:
        return 0.0
    cv = std_dev / mean_val
    score = max(0, 1.0 - cv) * 100
    return round(score, 2)


def detect_beaconing(conn_df, min_connections=10, max_cv=0.3):
    """Detect beaconing by analyzing connection interval regularity."""
    conn_df = conn_df.sort_values("ts")
    beacons = []
    grouped = conn_df.groupby(["id.orig_h", "id.resp_h", "id.resp_p"])
    for (src, dst, port), group in grouped:
        if len(group) < min_connections:
            continue
        times = group["ts"].sort_values()
        intervals = times.diff().dt.total_seconds().dropna().values
        if len(intervals) < 5:
            continue
        std_dev = float(np.std(intervals))
        mean_interval = float(np.mean(intervals))
        if mean_interval == 0:
            continue
        cv = std_dev / mean_interval
        beacon_score = calculate_beacon_score(intervals)
        if cv <= max_cv:
            beacons.append({
                "src_ip": src,
                "dst_ip": dst,
                "dst_port": int(port) if not pd.isna(port) else 0,
                "connection_count": len(group),
                "mean_interval_sec": round(mean_interval, 2),
                "std_dev_sec": round(std_dev, 2),
                "coefficient_of_variation": round(cv, 4),
                "beacon_score": beacon_score,
                "first_seen": str(times.iloc[0]),
                "last_seen": str(times.iloc[-1]),
            })
    return sorted(beacons, key=lambda x: x["beacon_score"], reverse=True)


def detect_jitter_beaconing(conn_df, base_interval=60, jitter_pct=0.2, min_conns=10):
    """Detect beaconing with expected interval and jitter tolerance."""
    conn_df = conn_df.sort_values("ts")
    matches = []
    grouped = conn_df.groupby(["id.orig_h", "id.resp_h"])
    for (src, dst), group in grouped:
        if len(group) < min_conns:
            continue
        times = group["ts"].sort_values()
        intervals = times.diff().dt.total_seconds().dropna().values
        lower = base_interval * (1 - jitter_pct)
        upper = base_interval * (1 + jitter_pct)
        matching = np.sum((intervals >= lower) & (intervals <= upper))
        match_pct = matching / len(intervals)
        if match_pct > 0.7:
            matches.append({
                "src_ip": src,
                "dst_ip": dst,
                "connections": len(group),
                "matching_intervals": int(matching),
                "match_percentage": round(match_pct * 100, 1),
                "expected_interval": base_interval,
            })
    return matches


def analyze_dns_beaconing(dns_log_path, min_queries=20, max_cv=0.25):
    """Analyze Zeek dns.log for DNS-based beaconing patterns."""
    log_to_df = LogToDataFrame()
    dns_df = log_to_df.create_dataframe(dns_log_path)
    dns_df = dns_df.sort_values("ts")
    beacons = []
    grouped = dns_df.groupby(["id.orig_h", "query"])
    for (src, query), group in grouped:
        if len(group) < min_queries:
            continue
        times = group["ts"].sort_values()
        intervals = times.diff().dt.total_seconds().dropna().values
        if len(intervals) < 5:
            continue
        std_dev = float(np.std(intervals))
        mean_val = float(np.mean(intervals))
        if mean_val == 0:
            continue
        cv = std_dev / mean_val
        if cv <= max_cv:
            beacons.append({
                "src_ip": src,
                "query": query,
                "query_count": len(group),
                "mean_interval_sec": round(mean_val, 2),
                "std_dev_sec": round(std_dev, 2),
                "cv": round(cv, 4),
                "beacon_score": calculate_beacon_score(intervals),
            })
    return sorted(beacons, key=lambda x: x["beacon_score"], reverse=True)


def filter_whitelisted(beacons, whitelist_domains=None):
    """Remove known-good destinations from beacon results."""
    if not whitelist_domains:
        whitelist_domains = ["microsoft.com", "google.com", "amazonaws.com",
                            "cloudflare.com", "akamai.net"]
    filtered = []
    for b in beacons:
        dst = b.get("dst_ip", "") or b.get("query", "")
        if not any(w in dst for w in whitelist_domains):
            filtered.append(b)
    return filtered


def main():
    parser = argparse.ArgumentParser(description="Zeek Beaconing Detection Agent")
    parser.add_argument("--conn-log", help="Path to Zeek conn.log")
    parser.add_argument("--dns-log", help="Path to Zeek dns.log")
    parser.add_argument("--min-connections", type=int, default=10)
    parser.add_argument("--max-cv", type=float, default=0.3)
    parser.add_argument("--output", default="beacon_report.json")
    parser.add_argument("--action", choices=[
        "conn_beacon", "dns_beacon", "full_hunt"
    ], default="full_hunt")
    args = parser.parse_args()

    report = {"generated_at": datetime.utcnow().isoformat(), "findings": {}}

    if args.action in ("conn_beacon", "full_hunt") and args.conn_log:
        conn_df = load_conn_log(args.conn_log)
        beacons = detect_beaconing(conn_df, args.min_connections, args.max_cv)
        beacons = filter_whitelisted(beacons)
        report["findings"]["conn_beacons"] = beacons
        print(f"[+] Connection beacons detected: {len(beacons)}")

    if args.action in ("dns_beacon", "full_hunt") and args.dns_log:
        dns_beacons = analyze_dns_beaconing(args.dns_log, args.min_connections)
        dns_beacons = filter_whitelisted(dns_beacons)
        report["findings"]["dns_beacons"] = dns_beacons
        print(f"[+] DNS beacons detected: {len(dns_beacons)}")

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()
Keep exploring
Analyzing Azure Activity Logs for ThreatsQueries Azure Monitor activity logs and sign-in logs via azure-monitor-query to detect suspicious administrative operations, impossible travel, privilege escalation, and resource modifications. Builds KQL queries for threat hunting in Azure environments. Use when investigating suspicious Azure tenant activity or building cloud SIEM detections.Hunting Credential Stuffing AttacksDetects credential stuffing attacks by analyzing authentication logs for login velocity anomalies, ASN diversity, password spray patterns, and geographic distribution of failed logins. Uses statistical analysis on Splunk or raw log data. Use when investigating account takeover campaigns or building detection rules for auth abuse.Implementing Canary Tokens for Network Intrusion DetectionDeploys DNS, HTTP, and AWS API key canary tokens across network infrastructure to detect unauthorized access and lateral movement. Integrates with webhook alerting (Slack, Teams, email, generic HTTP) for real-time intrusion notifications. Provides automated token generation, placement strategies, and monitoring for enterprise network environments. Use when building deception-based network intrusion detection with Canarytokens.org and Thinkst Canary platforms.Analyzing API Gateway Access LogsParses API Gateway access logs (AWS API Gateway, Kong, Nginx) to detect BOLA/IDOR attacks, rate limit bypass, credential scanning, and injection attempts. Uses pandas for statistical analysis of request patterns and anomaly detection. Use when investigating API abuse or building API-specific threat detection rules.Analyzing Memory Forensics with LiME and VolatilityPerforms Linux memory acquisition using LiME (Linux Memory Extractor) kernel module and analysis with Volatility 3 framework. Extracts process lists, network connections, bash history, loaded kernel modules, and injected code from Linux memory images. Use when performing incident response on compromised Linux systems.Analyzing PowerShell Script Block LoggingParse Windows PowerShell Script Block Logs (Event ID 4104) from EVTX files to detect obfuscated commands, encoded payloads, and living-off-the-land techniques. Uses python-evtx to extract and reconstruct multi-block scripts, applies entropy analysis and pattern matching for Base64-encoded commands, Invoke-Expression abuse, download cradles, and AMSI bypass attempts.