Beaconing Detection in Python

#!/usr/bin/env python3
"""
Beaconing detector for network forensics.
Input: Zeek conn.log or tshark CSV of SYN timestamps
Output: ranked list of beaconing candidates with CV scores
"""
import subprocess, sys, csv, json, math
from collections import defaultdict
from pathlib import Path

# ── CONFIG ────────────────────────────────────────────────────────────
CV_THRESHOLD = 0.20      # Lower CV = more regular = more suspicious
MIN_CONNECTIONS = 10     # Minimum connections to compute meaningful CV
MIN_DURATION_HOURS = 0.5 # Ignore sessions shorter than 30 minutes
EXCLUDE_PORTS = {123, 53, 67, 68, 443, 80}  # NTP, DNS, DHCP worth investigating separately

# ── DATA EXTRACTION ───────────────────────────────────────────────────
def extract_from_pcap(pcap_file: str) -> dict:
    """
    Extract SYN timestamps using tshark.
    Returns dict: {(src_ip, dst_ip, dst_port): [timestamp1, timestamp2, ...]}
    """
    cmd = [
        "tshark", "-r", pcap_file, "-n",
        "-Y", "tcp.flags.syn==1 and not tcp.flags.ack",
        "-T", "fields",
        "-E", "separator=\t",
        "-e", "frame.time_epoch",
        "-e", "ip.src",
        "-e", "ip.dst",
        "-e", "tcp.dstport",
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    flows = defaultdict(list)
    for line in result.stdout.splitlines():
        parts = line.split("\t")
        if len(parts) < 4:
            continue
        try:
            ts, src, dst, dport = float(parts[0]), parts[1], parts[2], int(parts[3])
            if dport not in EXCLUDE_PORTS:
                flows[(src, dst, dport)].append(ts)
        except (ValueError, IndexError):
            pass
    return flows

def extract_from_zeek_conn_log(conn_log: str) -> dict:
    """Extract from Zeek conn.log (tab-separated)."""
    flows = defaultdict(list)
    with open(conn_log) as f:
        for line in f:
            if line.startswith("#"):
                continue
            parts = line.strip().split("\t")
            if len(parts) < 7:
                continue
            try:
                ts = float(parts[0])
                src_ip = parts[2]
                dst_ip = parts[4]
                dst_port = int(parts[5])
                proto = parts[6]
                if proto == "tcp" and dst_port not in EXCLUDE_PORTS:
                    flows[(src_ip, dst_ip, dst_port)].append(ts)
            except (ValueError, IndexError):
                pass
    return flows

# ── ANALYSIS ──────────────────────────────────────────────────────────
def compute_cv(timestamps: list) -> tuple[float, float, float, int]:
    """Returns (mean, stddev, cv, count) or (None, None, None, count)."""
    n = len(timestamps)
    if n < MIN_CONNECTIONS:
        return None, None, None, n
    intervals = [timestamps[i+1] - timestamps[i] for i in range(n-1)]
    # Filter out negative intervals (clock skew or out-of-order) and >1 hour gaps
    intervals = [iv for iv in intervals if 0 < iv < 3600]
    if len(intervals) < MIN_CONNECTIONS - 1:
        return None, None, None, n
    mean = sum(intervals) / len(intervals)
    if mean < 1:  # Ignore sub-second mean intervals (port scans, not beacons)
        return None, None, None, n
    variance = sum((iv - mean) ** 2 for iv in intervals) / len(intervals)
    stddev = math.sqrt(variance)
    cv = stddev / mean
    return mean, stddev, cv, n

def analyze_beaconing(flows: dict) -> list:
    """Analyze all flows and return ranked beaconing candidates."""
    results = []
    for (src, dst, dport), timestamps in flows.items():
        timestamps.sort()
        # Check minimum duration
        duration_hours = (timestamps[-1] - timestamps[0]) / 3600
        if duration_hours < MIN_DURATION_HOURS:
            continue

        mean, stddev, cv, count = compute_cv(timestamps)
        if cv is None:
            continue

        if cv <= CV_THRESHOLD:
            confidence = "HIGH" if cv < 0.1 else "MEDIUM" if cv < 0.2 else "LOW"
            results.append({
                "src": src,
                "dst": dst,
                "dstport": dport,
                "count": count,
                "cv": round(cv, 4),
                "mean_interval_sec": round(mean, 1),
                "stddev_sec": round(stddev, 2),
                "duration_hours": round(duration_hours, 1),
                "confidence": confidence,
            })

    return sorted(results, key=lambda x: x["cv"])

# ── MAIN ──────────────────────────────────────────────────────────────
if __name__ == "__main__":
    input_file = sys.argv[1]

    if input_file.endswith(".pcap") or input_file.endswith(".pcapng"):
        print(f"Extracting from PCAP: {input_file}")
        flows = extract_from_pcap(input_file)
    else:
        print(f"Reading Zeek conn.log: {input_file}")
        flows = extract_from_zeek_conn_log(input_file)

    print(f"Analyzing {len(flows)} unique flows...")
    candidates = analyze_beaconing(flows)

    print(f"\n=== Beaconing Candidates (CV ≤ {CV_THRESHOLD}) ===")
    print(f"{'Confidence':8}  {'CV':6}  {'Interval':8}  {'Count':6}  {'Hours':6}  Connection")
    print("-" * 80)
    for c in candidates[:30]:
        print(f"{c['confidence']:8}  {c['cv']:6.4f}  {c['mean_interval_sec']:8.1f}s  "
              f"{c['count']:6d}  {c['duration_hours']:6.1f}h  "
              f"{c['src']} → {c['dst']}:{c['dstport']}")

    # Write full results
    out_file = "beaconing_candidates.json"
    with open(out_file, "w") as f:
        json.dump(candidates, f, indent=2)
    print(f"\nFull results: {out_file}")