Exfiltration Patterns

#!/usr/bin/env python3
"""
Multi-signal exfiltration detection from Zeek conn.log and dns.log.
Checks: large transfers, upload ratio, DNS volume, protocol anomalies.
"""
import sys
import math
from collections import defaultdict
from ipaddress import ip_address, ip_network
from dataclasses import dataclass, field
from typing import List

PRIVATE = [ip_network(n) for n in ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"]]

CLOUD_KEYWORDS = [
    "mega.nz", "backblaze.com", "onedrive.live.com", "dropbox.com",
    "drive.google.com", "wetransfer.com", "gofile.io", "anonfiles.com",
    "transfer.sh", "file.io"
]

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

def entropy(data: str) -> float:
    from collections import Counter
    c = Counter(data)
    total = len(data)
    if total == 0:
        return 0.0
    return -sum((v/total)*math.log2(v/total) for v in c.values())

@dataclass
class ConnRecord:
    src: str
    dst: str
    dport: int
    duration: float
    bytes_orig: int
    bytes_resp: int
    service: str
    ts: float


def parse_zeek_conn_log(filepath: str) -> List[ConnRecord]:
    records = []
    with open(filepath) as f:
        for line in f:
            if line.startswith('#'):
                continue
            parts = line.strip().split('\t')
            if len(parts) < 10:
                continue
            try:
                records.append(ConnRecord(
                    ts=float(parts[0]),
                    src=parts[2],
                    dst=parts[4],
                    dport=int(parts[5]) if parts[5].isdigit() else 0,
                    duration=float(parts[8]) if parts[8] != '-' else 0.0,
                    bytes_orig=int(parts[9]) if parts[9].isdigit() else 0,
                    bytes_resp=int(parts[10]) if len(parts) > 10 and parts[10].isdigit() else 0,
                    service=parts[7] if len(parts) > 7 else '-',
                ))
            except (ValueError, IndexError):
                pass
    return records


def detect_exfil(records: List[ConnRecord]) -> None:
    # Group outbound connections by src → dst
    by_pair = defaultdict(list)
    for r in records:
        if is_private(r.src) and not is_private(r.dst):
            by_pair[(r.src, r.dst, r.dport)].append(r)

    findings = []

    for (src, dst, dport), recs in by_pair.items():
        total_orig = sum(r.bytes_orig for r in recs)
        total_resp = sum(r.bytes_resp for r in recs)
        total = total_orig + total_resp

        if total == 0:
            continue

        upload_ratio = total_orig / total

        # Signal 1: Large total transfer
        if total_orig > 10_000_000:  # >10 MB outbound
            findings.append({
                "type": "LARGE_OUTBOUND",
                "score": min(100, int(total_orig / 100_000)),
                "src": src, "dst": dst, "port": dport,
                "detail": f"{total_orig/1_000_000:.1f}MB sent",
            })

        # Signal 2: High upload ratio (more sent than received)
        if upload_ratio > 0.85 and total_orig > 1_000_000:
            findings.append({
                "type": "HIGH_UPLOAD_RATIO",
                "score": int(upload_ratio * 80),
                "src": src, "dst": dst, "port": dport,
                "detail": f"ratio={upload_ratio:.2f} ({total_orig/1024:.0f}KB sent vs {total_resp/1024:.0f}KB recv)",
            })

    findings.sort(key=lambda x: -x["score"])
    print(f"\n{'Score':>6}  {'Type':<25}  {'Src':<15}  {'Dst':<15}  {'Port':>5}  Detail")
    print("-" * 90)
    for f in findings[:20]:
        print(f"{f['score']:>6}  {f['type']:<25}  {f['src']:<15}  {f['dst']:<15}  "
              f"{f['port']:>5}  {f['detail']}")


if __name__ == "__main__":
    path = sys.argv[1] if len(sys.argv) > 1 else "/opt/zeek/logs/current/conn.log"
    print(f"Loading {path}...")
    records = parse_zeek_conn_log(path)
    print(f"Loaded {len(records)} connection records")
    detect_exfil(records)