Beaconing Statistics at Scale
Production beaconing detection operates against Zeek conn.log containing millions of flows per day. At this scale, per-connection analysis in Python is too slow — you need to work with aggregated statistics, use efficient data structures, and run detection in near-real-time as logs are written. This chapter covers production-grade beaconing detection using Zeek + Python + periodic batch analysis, along with the statistical methods that reduce false positives in enterprise environments full of legitimate periodic traffic.
Your SOC receives Zeek conn.log from 500 endpoints. Each log generates roughly 2 million flow records per day — 1 billion flows per day across the enterprise. You need to run beaconing detection across all of this, produce a prioritized alert list within 15 minutes of each hour's data landing, and maintain a false positive rate low enough that your 3-person SOC team can review everything flagged each day. The implementation needs to handle this on a single server with 32 GB RAM.
Production Architecture
Production Beaconing Detection Architecture
═══════════════════════════════════════════════════════════════════
Data source: Zeek conn.log (or Elastic/Splunk search output)
Aggregation (hourly batch, per host×destination×port):
For each unique (src_ip, dst_ip, dst_port) tuple:
├── Collect all connection timestamps in the window
├── Compute: count, min, max, mean interval, stddev, CV
└── Store in a rolling 24-hour buffer
Scoring (multi-dimensional):
score = CV_score + duration_score + count_score + context_score
CV_score: 0-40 points based on CV value
duration: 0-20 points (longer observation window = more confidence)
count: 0-20 points (more connections = higher confidence)
context: deduct points for known-good destinations
False positive reduction layers:
Layer 1: Allowlist (CDN IP ranges, known NTP, internal infrastructure)
Layer 2: Asset baseline (is this connection pattern new or recurring?)
Layer 3: Process correlation (from endpoint telemetry if available)
Layer 4: Threat intel enrichment (VT/Shodan for destination)
Output: daily alert list, sorted by score, with context
Production Beaconing Detector
#!/usr/bin/env python3
"""
Production beaconing detector for Zeek conn.log at scale.
Processes N million flow records efficiently.
"""
import sys, math, json, ipaddress, re, time
from collections import defaultdict
from pathlib import Path
# ── CONFIG ─────────────────────────────────────────────────────────────
CV_HIGH = 0.10 # HIGH confidence beacon
CV_MEDIUM = 0.20 # MEDIUM confidence
CV_LOW = 0.35 # LOW confidence (include if count/duration high)
MIN_COUNT = 10 # Minimum connections for CV to be meaningful
MIN_DURATION_H = 1 # Minimum observation window in hours
MAX_INTERVAL_S = 7200 # Ignore gaps > 2 hours (host offline)
# ── ALLOWLISTING ────────────────────────────────────────────────────────
PRIVATE_NETS = [
ipaddress.ip_network(n) for n in [
"10.0.0.0/8","172.16.0.0/12","192.168.0.0/16",
"127.0.0.0/8","169.254.0.0/16"
]
]
KNOWN_GOOD_PORTS = {80, 443, 53, 123, 8080, 8443} # Still analyze, just lower priority
SKIP_PORTS = {67, 68, 5355, 137, 138, 139} # DHCP, LLMNR, NetBIOS noise
def is_private(ip_str: str) -> bool:
try:
addr = ipaddress.ip_address(ip_str)
return any(addr in net for net in PRIVATE_NETS)
except ValueError:
return True
# ── STATISTICS ─────────────────────────────────────────────────────────
def compute_cv(timestamps: list) -> tuple:
"""Fast CV computation; filters outlier gaps."""
n = len(timestamps)
if n < MIN_COUNT:
return None, None, None, n
intervals = []
for i in range(n - 1):
gap = timestamps[i + 1] - timestamps[i]
if 0 < gap < MAX_INTERVAL_S:
intervals.append(gap)
if len(intervals) < MIN_COUNT - 1:
return None, None, None, n
mean = sum(intervals) / len(intervals)
if mean < 0.5:
return None, None, None, n
variance = sum((x - mean) ** 2 for x in intervals) / len(intervals)
stddev = math.sqrt(variance)
cv = stddev / mean
return mean, stddev, cv, n
# ── ZEEK LOG READER ─────────────────────────────────────────────────────
def stream_zeek_conn_log(path: str):
"""Generator: yield one flow dict per line from Zeek conn.log."""
field_names = []
with open(path) as f:
for line in f:
if line.startswith("#fields"):
field_names = line.strip().split("\t")[1:]
elif not line.startswith("#"):
values = line.strip().split("\t")
if field_names and len(values) == len(field_names):
yield dict(zip(field_names, values))
# ── MAIN ANALYSIS ───────────────────────────────────────────────────────
def analyze(conn_log_path: str) -> list:
start = time.time()
# First pass: collect timestamps per flow key
# Memory-efficient: store only timestamps, not full records
flow_timestamps = defaultdict(list)
total = 0
for flow in stream_zeek_conn_log(conn_log_path):
total += 1
if total % 500000 == 0:
elapsed = time.time() - start
print(f" {total:,} flows processed ({elapsed:.1f}s)...", end="\r")
try:
proto = flow.get("proto", "")
if proto != "tcp":
continue
src = flow.get("id.orig_h", "")
dst = flow.get("id.resp_h", "")
dport = int(flow.get("id.resp_p", "0") or "0")
ts = float(flow.get("ts", "0") or "0")
# Skip noise
if dport in SKIP_PORTS:
continue
if is_private(dst):
continue
if not ts:
continue
key = (src, dst, dport)
flow_timestamps[key].append(ts)
except (ValueError, TypeError):
pass
elapsed = time.time() - start
print(f"\n {total:,} flows processed in {elapsed:.1f}s")
print(f" Unique flow tuples: {len(flow_timestamps):,}")
# Second pass: compute CV for each flow group
results = []
for (src, dst, dport), timestamps in flow_timestamps.items():
if len(timestamps) < MIN_COUNT:
continue
timestamps.sort()
duration_h = (timestamps[-1] - timestamps[0]) / 3600
if duration_h < MIN_DURATION_H:
continue
mean, stddev, cv, count = compute_cv(timestamps)
if cv is None:
continue
if cv > CV_LOW:
continue
# Scoring
if cv < CV_HIGH:
priority = "HIGH"
score = 90
elif cv < CV_MEDIUM:
priority = "MEDIUM"
score = 60
else:
priority = "LOW"
score = 30
# Bonus for long observation + high count
if duration_h > 6:
score += 10
if count > 50:
score += 10
if dport not in KNOWN_GOOD_PORTS:
score += 10
results.append({
"priority": priority,
"score": score,
"src": src,
"dst": dst,
"dport": dport,
"count": count,
"cv": round(cv, 4),
"mean_interval_s": round(mean, 1),
"duration_h": round(duration_h, 2),
})
return sorted(results, key=lambda x: -x["score"])
# ── ENTRY POINT ─────────────────────────────────────────────────────────
if __name__ == "__main__":
conn_log = sys.argv[1]
print(f"Beaconing detection on: {conn_log}")
candidates = analyze(conn_log)
print(f"\nTotal beaconing candidates: {len(candidates)}")
print(f"\n{'Priority':8} {'Score':6} {'CV':7} {'Int(s)':8} {'Count':6} {'Hours':6} Connection")
print("-" * 90)
for c in candidates[:40]:
print(f"{c['priority']:8} {c['score']:6d} {c['cv']:7.4f} {c['mean_interval_s']:8.1f}s "
f"{c['count']:6d} {c['duration_h']:6.1f}h {c['src']}→{c['dst']}:{c['dport']}")
out = "beaconing_results.json"
with open(out, "w") as f:
json.dump(candidates, f, indent=2)
print(f"\nFull results: {out}")Q & A
Q: The detector generates 200 HIGH priority alerts per day. My team can review 30. What do I do?
Two hundred HIGH alerts means your CV threshold or scoring is too permissive, or you have too many legitimate periodic applications. Reduction strategies: (1) Raise the CV threshold for HIGH: instead of CV < 0.10 = HIGH, try CV < 0.05 = HIGH. Cobalt Strike with 0% jitter is CV ≈ 0.001; it will still be flagged. (2) Build a daily allowlist: run the detector for a full week during a known-clean period. Every destination that appears in the top 200 every day is almost certainly a legitimate periodic application. Add those (src, dst, dport) tuples to an allowlist. The week-of-clean-data approach automatically discovers legitimate NTP, update, and telemetry traffic. (3) Require minimum duration: raise MIN_DURATION_H from 1 to 4 hours. Legitimate applications may have regular intervals but don't necessarily run for 4+ hours a day. (4) Correlate with threat intel: destinations that appear in VirusTotal or AbuseIPDB should be automatically promoted to HIGH regardless of CV — they've been seen as attack infrastructure. Clean destinations (older domains, trusted ASNs, Alexa top 100K) can be demoted. (5) Deduplicate by destination: if 50 different internal IPs are all beaconing to the same external IP with similar CVs, that's probably a software update client — alert on it once, not 50 times.