beacon_detect.py

from datetime import datetime, timedelta
import random
import statistics
from ai_triage import ai_triage_alert


def detect_beaconing(
    connection_log, interval_tolerance_seconds=15, min_connections=5
):
    """connection_log: list of dicts like {"host": "10.0.0.5", "dest_ip": "1.2.3.4", "timestamp": datetime}

    Groups connections by (host, dest_ip) pair and flags ones with suspiciously
    regular timing (low variance between connection intervals) - a classic
    beaconing sign.
    """
    from collections import defaultdict

    grouped = defaultdict(list)
    for conn in connection_log:
        grouped[(conn["host"], conn["dest_ip"])].append(conn["timestamp"])

    flagged = []
    for (host, dest_ip), timestamps in grouped.items():
        if len(timestamps) < min_connections:
            continue

        timestamps.sort()
        intervals = [
            (timestamps[i + 1] - timestamps[i]).total_seconds()
            for i in range(len(timestamps) - 1)
        ]

        if len(intervals) < 2:
            continue

        stdev = statistics.stdev(intervals)
        avg_interval = statistics.mean(intervals)

        # Low standard deviation relative to the average interval = suspiciously regular = likely automated
        if stdev < interval_tolerance_seconds and avg_interval > 10:
            flagged.append({
                "host": host,
                "dest_ip": dest_ip,
                "connection_count": len(timestamps),
                "avg_interval_seconds": round(avg_interval, 1),
                "interval_stdev": round(stdev, 2),
            })

    return flagged


def build_demo_connection_log():
    """Simulates one host beaconing every ~60s, and one host with normal random browsing traffic."""
    base = datetime.now()
    log = []

    # Suspicious: near-perfect 60s interval -> looks like beaconing
    for i in range(10):
        log.append({
            "host": "10.0.0.5",
            "dest_ip": "185.220.101.4",
            "timestamp": base + timedelta(seconds=60 * i + (i % 2)),
        })

    # Normal: random browsing to a clean IP, irregular timing
    t = base
    for _ in range(8):
        t += timedelta(seconds=random.randint(5, 300))
        log.append({"host": "10.0.0.9", "dest_ip": "8.8.8.8", "timestamp": t})

    return log


if __name__ == "__main__":
    conn_log = build_demo_connection_log()
    flagged = detect_beaconing(conn_log)

    print(
        f"Scanned {len(conn_log)} connection events, found"
        f" {len(flagged)} beaconing pattern(s):\n"
    )

    for f in flagged:
        print(f)
        alert = {"type": "beaconing", "host": f["host"], "ip": f["dest_ip"]}
        result = ai_triage_alert(alert)
        print(f" -> Triaged as {result['severity']}: {result['reason']}\n")