Beaconing via NetFlow

#!/usr/bin/env python3
"""
Detect C2 beaconing from NetFlow data using Coefficient of Variation.
Input: nfdump binary files or GoFlow2 JSON records.
Output: ranked list of potential beacons.
"""
import math
import subprocess
import json
from collections import defaultdict
from dataclasses import dataclass, field
from typing import List, Tuple
from ipaddress import ip_address, ip_network

PRIVATE_NETS = [
    ip_network("10.0.0.0/8"),
    ip_network("172.16.0.0/12"),
    ip_network("192.168.0.0/16"),
]

# Minimum requirements for a valid beacon candidate
MIN_FLOWS = 10            # Must see at least 10 connections
MIN_DURATION_H = 0.5      # Must span at least 30 minutes
MAX_CV = 0.15             # CV threshold (lower = more regular)
MIN_INTERVAL_S = 10       # Ignore sub-10-second "intervals" (jitter)
MAX_INTERVAL_S = 7200     # Ignore gaps > 2 hours (not C2 check-in)

ALLOWLIST_DSTS = {
    "8.8.8.8", "8.8.4.4",     # Google DNS
    "1.1.1.1", "1.0.0.1",     # Cloudflare DNS
    "13.107.4.50",              # Microsoft/Windows Update (sampled)
}

def is_private(ip_str: str) -> bool:
    try:
        addr = ip_address(ip_str)
        return any(addr in net for net in PRIVATE_NETS)
    except ValueError:
        return True


def compute_cv(intervals: List[float]) -> float:
    """Coefficient of Variation = stddev/mean. Lower = more regular = suspicious."""
    if len(intervals) < 3:
        return float('inf')
    mean = sum(intervals) / len(intervals)
    if mean == 0:
        return float('inf')
    variance = sum((x - mean) ** 2 for x in intervals) / len(intervals)
    return math.sqrt(variance) / mean


def load_nfdump_json(directory: str) -> List[dict]:
    """Read NetFlow records from nfdump files as JSON."""
    cmd = ["nfdump", "-R", directory, "-q", "-o", "json"]
    result = subprocess.run(cmd, capture_output=True, text=True)
    records = []
    for line in result.stdout.splitlines():
        try:
            records.append(json.loads(line))
        except json.JSONDecodeError:
            pass
    return records


@dataclass
class FlowTracker:
    timestamps: List[float] = field(default_factory=list)
    total_bytes_orig: int = 0
    dst_port: int = 0


def analyze_flows(records: List[dict]) -> List[Tuple]:
    """Analyze NetFlow records for beaconing behavior."""
    # Group flows by (src_ip, dst_ip, dst_port, proto)
    flows: dict = defaultdict(FlowTracker)

    for rec in records:
        src = rec.get("src", rec.get("SrcAddr", ""))
        dst = rec.get("dst", rec.get("DstAddr", ""))
        sport = int(rec.get("srcport", rec.get("SrcPort", 0)))
        dport = int(rec.get("dstport", rec.get("DstPort", 0)))
        proto = int(rec.get("proto", rec.get("Proto", 0)))
        ts = float(rec.get("ts", rec.get("TimeFlowStart", 0)))
        byt = int(rec.get("ibyt", rec.get("Bytes", 0)))

        # Only look at internal → external TCP connections
        if proto != 6:
            continue
        if not src or not dst:
            continue
        if not is_private(src) or is_private(dst):
            continue
        if dst in ALLOWLIST_DSTS:
            continue

        key = (src, dst, dport)
        tracker = flows[key]
        tracker.timestamps.append(ts)
        tracker.total_bytes_orig += byt
        tracker.dst_port = dport

    results = []
    for (src, dst, dport), tracker in flows.items():
        timestamps = sorted(tracker.timestamps)
        if len(timestamps) < MIN_FLOWS:
            continue

        duration = timestamps[-1] - timestamps[0]
        if duration < MIN_DURATION_H * 3600:
            continue

        intervals = [
            timestamps[i+1] - timestamps[i]
            for i in range(len(timestamps) - 1)
            if MIN_INTERVAL_S <= timestamps[i+1] - timestamps[i] <= MAX_INTERVAL_S
        ]
        if len(intervals) < MIN_FLOWS - 1:
            continue

        cv = compute_cv(intervals)
        if cv > MAX_CV:
            continue

        mean_interval = sum(intervals) / len(intervals)
        score = 100 * (1 - cv) * min(1.0, len(timestamps) / 50)

        results.append((
            score, cv, mean_interval, len(timestamps),
            duration / 3600, src, dst, dport,
            tracker.total_bytes_orig
        ))

    results.sort(key=lambda x: -x[0])
    return results


def print_results(results: List[Tuple]) -> None:
    print(f"\n{'Score':>6}  {'CV':>6}  {'Interval':>10}  {'N':>5}  {'Hrs':>5}  "
          f"{'Bytes':>10}  {'Src':<15}  {'Dst':<15}  {'Port':>5}")
    print("-" * 90)
    for score, cv, interval, n, hours, src, dst, dport, byt in results:
        print(f"{score:6.1f}  {cv:6.3f}  {interval:8.0f}s  {n:5d}  {hours:5.1f}  "
              f"{byt:10d}  {src:<15}  {dst:<15}  {dport:5d}")
    if not results:
        print("  No beacon candidates found.")


if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} ")
        sys.exit(1)

    directory = sys.argv[1]
    print(f"Loading NetFlow records from {directory}...")
    records = load_nfdump_json(directory)
    print(f"Loaded {len(records)} flow records.")

    results = analyze_flows(records)
    print(f"\nFound {len(results)} beacon candidates (CV <= {MAX_CV})")
    print_results(results[:25])