detect.py

"""IR detection signals computed from the parsed flow table.

Everything here is heuristic and tunable. Defaults lean toward surfacing
candidates for an analyst to confirm, not toward silent auto-verdicts.
"""

from __future__ import annotations

import math

import threatintel
from netutils import is_internal, is_public

# --- thresholds (kept simple; could move to config later) ---------------------
SHARED_DEST_MIN_HOSTS = 2      # external IP contacted by >= N internal hosts
BEACON_MIN_EVENTS = 8          # need enough intervals for cadence to mean anything
BEACON_MAX_CV = 0.25           # coefficient of variation below this == regular
BEACON_MIN_MEAN_INTERVAL = 1.0 # ignore sub-second chatter as "beaconing"
FANOUT_MIN_DESTS = 30          # internal host talking to > N *non-web* external IPs
EXFIL_MIN_BYTES = 5_000_000    # only flag sizeable transfers
EXFIL_OUT_IN_RATIO = 5.0       # outbound:inbound byte ratio
SCAN_MIN_TARGETS = 20          # distinct dst (host,port) with no completed handshake

# Ordinary client traffic (web / DNS / NTP). A host hitting hundreds of these
# is just browsing, not fanning out — excluded from the fan-out signal.
COMMON_CLIENT_PORTS = {53, 80, 123, 443, 8080, 8443}


def per_host_stats(result):
    """Aggregate flows into per-host stats. Returns dict ip -> stats dict."""
    hosts: dict[str, dict] = {}

    def _h(ip):
        return hosts.setdefault(ip, {
            "ip": ip,
            "internal": is_internal(ip),
            "bytes_out": 0, "bytes_in": 0,
            "packets": 0,
            "peers": set(),
            "ext_peers": set(),
            "ext_peers_nonweb": set(),  # external peers on non-web/DNS ports
            "int_peers": set(),
            "dst_endpoints": set(),   # (dst, port) this host initiated toward
            "unacked_targets": set(), # SYN-only / RST targets (scan signal)
        })

    for f in result.flows.values():
        # f.src is the client (initiator), f.dst the server
        c, srv = _h(f.client_ip), _h(f.server_ip)
        c["bytes_out"] += f.c2s_bytes
        c["bytes_in"] += f.s2c_bytes
        srv["bytes_out"] += f.s2c_bytes
        srv["bytes_in"] += f.c2s_bytes
        c["packets"] += f.packets
        srv["packets"] += f.packets
        c["peers"].add(f.server_ip)
        srv["peers"].add(f.client_ip)
        if is_internal(f.server_ip):
            c["int_peers"].add(f.server_ip)
        else:
            c["ext_peers"].add(f.server_ip)
            if f.server_port not in COMMON_CLIENT_PORTS:
                c["ext_peers_nonweb"].add(f.server_ip)
        (srv["ext_peers"] if not is_internal(f.client_ip) else srv["int_peers"]).add(f.client_ip)
        c["dst_endpoints"].add((f.server_ip, f.server_port))
        # TCP connection that never completed (SYN seen, no ACK) or got RST
        if f.proto == "TCP" and (f.rst > 0 or (f.syn > 0 and f.ack == 0)):
            c["unacked_targets"].add((f.server_ip, f.server_port))

    return hosts


def shared_external_destinations(result):
    """External IPs contacted by multiple *internal* hosts — the headline C2
    indicator. Returns a list sorted by number of internal hosts, descending."""
    by_ext: dict[str, dict] = {}
    for f in result.flows.values():
        for ext, other in ((f.dst, f.src), (f.src, f.dst)):
            if is_public(ext) and is_internal(other):
                entry = by_ext.setdefault(ext, {
                    "ip": ext, "internal_hosts": set(), "subnets": set(),
                    "bytes": 0, "packets": 0, "ports": set(), "beaconing": False,
                })
                entry["internal_hosts"].add(other)
                entry["bytes"] += f.bytes
                entry["packets"] += f.packets
                if f.dst_port:
                    entry["ports"].add(f.dst_port)

    out = []
    for entry in by_ext.values():
        if len(entry["internal_hosts"]) >= SHARED_DEST_MIN_HOSTS:
            out.append(entry)
    out.sort(key=lambda e: len(e["internal_hosts"]), reverse=True)
    return out


# ---- DNS / lateral-movement thresholds & lookups ----------------------------
DGA_MIN_LABEL_LEN = 12
DGA_MIN_ENTROPY = 3.8
DGA_MIN_PER_CLIENT = 5          # distinct DGA-like domains before flagging a host
TUNNEL_MIN_QUERIES = 200
TUNNEL_MIN_AVG_QLEN = 40
TUNNEL_TXT_RATIO = 0.3
NXDOMAIN_MIN = 50
DNS_EXFIL_MIN_SUBDOMAINS = 40   # unique subdomains under one parent -> tunneling

# Volumetric exfil to a single destination, and ICMP tunneling.
EXFIL_DEST_MIN_BYTES = 2_000_000
EXFIL_DEST_RATIO = 3.0          # upload:download to one external host
ICMP_EXFIL_BYTES = 64_000       # ICMP payload volume to one host (normally tiny)

# Registrable domains that legitimately use random-looking subdomains (CDNs,
# telemetry) — excluded from DGA flagging to cut false positives.
DGA_SAFE_PARENTS = {
    "microsoft.com", "windows.com", "office.com", "office365.com", "live.com",
    "google.com", "googleapis.com", "gstatic.com", "googleusercontent.com",
    "akamai.net", "akamaiedge.net", "akadns.net", "cloudfront.net",
    "amazonaws.com", "azure.com", "azureedge.net", "apple.com", "icloud.com",
    "facebook.com", "fbcdn.net", "cloudflare.com", "edgekey.net", "trafficmanager.net",
}

# Lateral-movement service ports -> (label, ATT&CK technique)
LATERAL_PORTS = {
    445:  ("SMB", "T1021.002"),
    139:  ("SMB/NetBIOS", "T1021.002"),
    3389: ("RDP", "T1021.001"),
    5985: ("WinRM-HTTP", "T1021.006"),
    5986: ("WinRM-HTTPS", "T1021.006"),
    135:  ("RPC/WMI", "T1047"),
    22:   ("SSH", "T1021.004"),
    23:   ("Telnet", "T1021"),
    5900: ("VNC", "T1021.005"),
}

# ATT&CK technique per finding kind (for the headline tag)
ATTACK = {
    "shared_destination": "T1071", "beaconing": "T1071", "fanout": "T1046",
    "scan": "T1046", "exfil": "T1048", "dga": "T1568.002",
    "dns_tunnel": "T1071.004", "nxdomain": "T1568.002",
    "dns_exfil": "T1048.003", "exfil_dest": "T1048", "icmp_tunnel": "T1048.003",
    "intel_ip": "T1071", "intel_domain": "T1071", "bad_ja3": "T1071",
    "lateral": "T1021",
}


def _entropy(s):
    if not s:
        return 0.0
    freq = {}
    for ch in s:
        freq[ch] = freq.get(ch, 0) + 1
    n = len(s)
    return -sum((c / n) * math.log2(c / n) for c in freq.values())


def _registrable(domain):
    parts = domain.rstrip(".").split(".")
    return ".".join(parts[-2:]) if len(parts) >= 2 else domain


def is_dga(domain):
    """Heuristic: a high-entropy, long left-most label on a domain that isn't a
    known CDN/telemetry parent."""
    if not domain or _registrable(domain) in DGA_SAFE_PARENTS:
        return False
    label = domain.split(".")[0]
    if len(label) < DGA_MIN_LABEL_LEN:
        return False
    digits = sum(c.isdigit() for c in label)
    return _entropy(label) >= DGA_MIN_ENTROPY or digits / len(label) >= 0.35


def is_beaconing(flow) -> bool:
    """Regular cadence to a destination. Uses the captured timestamp sample for
    a jitter-tolerant check, falling back to the running CV."""
    times = getattr(flow, "times", None)
    if times and len(times) >= BEACON_MIN_EVENTS + 1:
        ts = sorted(times)
        gaps = [b - a for a, b in zip(ts, ts[1:]) if b - a >= 0]
        if len(gaps) >= BEACON_MIN_EVENTS:
            gaps.sort()
            median = gaps[len(gaps) // 2]
            if median >= BEACON_MIN_MEAN_INTERVAL:
                within = sum(1 for g in gaps if abs(g - median) <= 0.3 * median)
                if within / len(gaps) >= 0.7:        # mostly-regular, jitter ok
                    return True
    return (
        flow.interval_count >= BEACON_MIN_EVENTS
        and flow.mean_interval >= BEACON_MIN_MEAN_INTERVAL
        and flow.jitter_cv <= BEACON_MAX_CV
    )


def beacon_period(flow):
    """Best estimate of a beacon's period in seconds (median gap)."""
    times = sorted(getattr(flow, "times", []) or [])
    gaps = sorted(b - a for a, b in zip(times, times[1:]) if b - a >= 0)
    if gaps:
        return gaps[len(gaps) // 2]
    return flow.mean_interval


def dns_findings(result):
    out = []
    # DGA per client
    for client, c in result.dns_by_client.items():
        dga = [n for n in c["names"] if is_dga(n)]
        if len(dga) >= DGA_MIN_PER_CLIENT:
            out.append({
                "kind": "dga", "severity": "high", "ip": client,
                "title": f"{client} queried {len(dga)} algorithmic-looking domains",
                "detail": "possible DGA / C2 (review — may include random CDN subdomains): "
                          + ", ".join(sorted(dga)[:3]),
            })
        # DNS tunneling
        q = c["queries"]
        if q >= TUNNEL_MIN_QUERIES:
            avg_len = c["qlen_sum"] / max(1, q)
            txt_ratio = c["txt"] / max(1, q)
            if avg_len >= TUNNEL_MIN_AVG_QLEN or txt_ratio >= TUNNEL_TXT_RATIO:
                out.append({
                    "kind": "dns_tunnel", "severity": "high", "ip": client,
                    "title": f"{client} shows DNS-tunneling traits "
                             f"({q} queries, avg len {avg_len:.0f}, TXT {txt_ratio:.0%})",
                    "detail": "long/encoded qnames or heavy TXT use",
                })
        if c["nx"] >= NXDOMAIN_MIN:
            out.append({
                "kind": "nxdomain", "severity": "medium", "ip": client,
                "title": f"{client} got {c['nx']} NXDOMAIN responses",
                "detail": "domain-generation / failed-resolution spike",
            })
        # DNS exfil: many unique subdomains under one (non-CDN) parent domain
        by_parent = {}
        for n in c["names"]:
            by_parent.setdefault(_registrable(n), set()).add(n)
        for parent, subs in by_parent.items():
            if parent in DGA_SAFE_PARENTS or len(subs) < DNS_EXFIL_MIN_SUBDOMAINS:
                continue
            out.append({
                "kind": "dns_exfil", "severity": "high", "ip": client,
                "title": f"{client} queried {len(subs)} unique subdomains of {parent}",
                "detail": "encoded data in DNS labels — likely DNS tunneling / exfiltration",
            })
    return out


def lateral_findings(result):
    """Internal->internal connections on remote-service ports. Many targets on
    the same service from one host are aggregated into a single finding."""
    groups = {}   # (client, port) -> set(targets)
    for f in result.flows.values():
        if not (is_internal(f.client_ip) and is_internal(f.server_ip)):
            continue
        if f.server_port in LATERAL_PORTS:
            groups.setdefault((f.client_ip, f.server_port), set()).add(f.server_ip)

    out = []
    for (client, port), targets in groups.items():
        name, tech = LATERAL_PORTS[port]
        if len(targets) >= 5:
            out.append({
                "kind": "lateral", "severity": "high",
                "ip": sorted(targets)[0], "attack": tech,
                "title": f"{client} → {len(targets)} internal hosts on {name} ({port})",
                "detail": "lateral movement / spread: " + ", ".join(sorted(targets)[:4])
                          + ("…" if len(targets) > 4 else ""),
            })
        else:
            for t in sorted(targets):
                out.append({
                    "kind": "lateral", "severity": "high", "ip": t, "attack": tech,
                    "title": f"{client} → {t}:{port} ({name})",
                    "detail": "internal lateral-movement service",
                })
    return out


# ---- non-standard traffic detection ----------------------------------------
COMMON_TLS_PORTS = {443, 8443, 993, 995, 465, 990, 5223, 4433, 853}
COMMON_HTTP_PORTS = {80, 8080, 8000, 8888, 3000, 591, 8081}
# Broad set of ports considered "normal" for a host to talk to.
KNOWN_PORTS = {
    20, 21, 22, 25, 53, 67, 68, 80, 88, 110, 123, 135, 137, 138, 139, 143,
    161, 389, 443, 445, 465, 514, 515, 543, 587, 593, 636, 853, 873, 989, 990,
    993, 995, 1080, 1433, 1521, 1701, 1723, 1900, 2049, 3128, 3268, 3306, 3389,
    3478, 5060, 5061, 5222, 5223, 5228, 5353, 5355, 5985, 5986, 8000, 8008,
    8080, 8443, 8888, 9000, 9090, 9200, 11211, 27017, 500, 4500, 51820,
}
# Ports strongly associated with backdoors / C2 / abuse.
SUSPICIOUS_PORTS = {
    1337: "leet/backdoor", 4444: "Metasploit", 4445: "Metasploit",
    31337: "Back Orifice", 12345: "NetBus", 6666: "IRC/bot", 6667: "IRC",
    6697: "IRC", 9001: "Tor", 1234: "backdoor", 5555: "ADB/backdoor",
    2323: "Telnet-botnet", 7777: "backdoor", 8087: "C2",
}
# Cleartext / risky services -> (label, severity)
CLEARTEXT_PORTS = {
    21: ("FTP", "high"), 23: ("Telnet", "high"), 512: ("rexec", "high"),
    513: ("rlogin", "high"), 514: ("rsh", "high"), 25: ("SMTP", "medium"),
    110: ("POP3", "medium"), 143: ("IMAP", "medium"), 161: ("SNMP", "medium"),
    69: ("TFTP", "high"), 79: ("Finger", "medium"), 389: ("LDAP", "low"),
}
# Infra ports that are normally contacted by IP without a DNS lookup.
_DIRECT_IP_OK_PORTS = {53, 67, 68, 123, 137, 138, 1900, 5353, 5355, 3478}


def nonstandard(result):
    """Per internal host, find non-standard / suspicious traffic.

    Returns {internal_ip: [ {kind, severity, label, detail, attack}, ... ]}.
    """
    cached = getattr(result, "_ns_cache", None)
    if cached is not None:
        return cached
    from netutils import is_internal as _ii, is_public as _ip  # local import ok
    per_host = {}
    direct_ip = {}     # internal -> set(external IPs contacted with no DNS name)

    def _add(host, item):
        per_host.setdefault(host, []).append(item)

    for f in result.flows.values():
        ci, si = _ii(f.client_ip), _ii(f.server_ip)
        internal = f.client_ip if ci else (f.server_ip if si else None)
        if internal is None:
            continue
        external = f.server_ip if (ci and not si) else (f.client_ip if (si and not ci) else None)
        port = f.server_port

        # 1) port / protocol mismatch (positive protocol ID on the wrong port)
        if f.ja3 and port not in COMMON_TLS_PORTS:
            _add(internal, {"kind": "mismatch", "severity": "medium",
                            "label": f"TLS on port {port}",
                            "detail": f"{f.client_ip}→{f.server_ip} (SNI {f.sni or '?'})",
                            "attack": "T1571"})
        if f.http_host and port not in COMMON_HTTP_PORTS:
            _add(internal, {"kind": "mismatch", "severity": "medium",
                            "label": f"HTTP on port {port}",
                            "detail": f"{f.client_ip}→{f.server_ip} (Host {f.http_host})",
                            "attack": "T1571"})

        # 2) suspicious / rare ports & odd protocols
        if port in SUSPICIOUS_PORTS:
            _add(internal, {"kind": "suspicious_port", "severity": "high",
                            "label": f"port {port} ({SUSPICIOUS_PORTS[port]})",
                            "detail": f"{f.client_ip}→{f.server_ip}", "attack": "T1571"})
        elif f.proto == "OTHER":
            _add(internal, {"kind": "odd_proto", "severity": "medium",
                            "label": "non-TCP/UDP/ICMP protocol",
                            "detail": f"{f.client_ip}↔{f.server_ip}", "attack": "T1095"})

        # 3) cleartext / risky services
        if port in CLEARTEXT_PORTS:
            name, sev = CLEARTEXT_PORTS[port]
            _add(internal, {"kind": "cleartext", "severity": sev,
                            "label": f"{name} (port {port})",
                            "detail": f"{f.client_ip}→{f.server_ip}", "attack": "T1071"})

        # 4) direct-to-IP (external never resolved via DNS, no SNI/Host)
        if external and _ip(external) and port not in _DIRECT_IP_OK_PORTS \
                and external not in result.ip_to_domain \
                and not f.sni and not f.http_host:
            direct_ip.setdefault(internal, set()).add(external)

    for host, ips in direct_ip.items():
        _add(host, {"kind": "direct_ip", "severity": "medium",
                    "label": f"{len(ips)} direct-to-IP destinations (no DNS)",
                    "detail": ", ".join(sorted(ips)[:6]) + ("…" if len(ips) > 6 else ""),
                    "attack": "T1571", "ips": sorted(ips)})
    try:
        result._ns_cache = per_host
    except Exception:
        pass
    return per_host


def nonstandard_findings(per_host):
    """Summarise the per-host non-standard items into headline findings."""
    out = []
    for host, items in per_host.items():
        by_kind = {}
        for it in items:
            by_kind.setdefault(it["kind"], []).append(it)
        for kind, lst in by_kind.items():
            sev = "high" if any(i["severity"] == "high" for i in lst) else "medium"
            sample = lst[0]
            label = sample["label"] if len(lst) == 1 else f"{len(lst)}× {kind.replace('_', ' ')}"
            out.append({
                "kind": f"nonstd_{kind}", "severity": sev, "ip": host,
                "attack": sample.get("attack"),
                "title": f"{host}: {label}",
                "detail": sample["detail"][:120],
            })
    return out


def exfil_findings(result):
    """Volumetric exfil to a single external host, and ICMP tunneling."""
    pair = {}
    for f in result.flows.values():
        ci, si = is_internal(f.client_ip), is_internal(f.server_ip)
        if ci == si:
            continue
        host = f.client_ip if ci else f.server_ip
        dest = f.server_ip if ci else f.client_ip
        if not is_public(dest):
            continue
        up = f.c2s_bytes if host == f.client_ip else f.s2c_bytes
        down = f.s2c_bytes if host == f.client_ip else f.c2s_bytes
        e = pair.setdefault((host, dest), {"up": 0, "down": 0, "icmp": 0})
        e["up"] += up
        e["down"] += down
        if f.proto == "ICMP":
            e["icmp"] += f.bytes

    out = []
    for (host, dest), e in pair.items():
        if e["up"] >= EXFIL_DEST_MIN_BYTES and e["up"] / max(1, e["down"]) >= EXFIL_DEST_RATIO:
            out.append({
                "kind": "exfil_dest", "severity": "high", "ip": dest,
                "title": f"{host} uploaded {e['up'] / 1e6:.1f} MB to {dest} "
                         f"({e['up'] / max(1, e['down']):.0f}× download)",
                "detail": "upload-heavy transfer to one host — possible exfiltration",
            })
        if e["icmp"] >= ICMP_EXFIL_BYTES:
            out.append({
                "kind": "icmp_tunnel", "severity": "high", "ip": dest,
                "title": f"{host} sent {e['icmp'] / 1024:.0f} KB of ICMP to {dest}",
                "detail": "large ICMP volume — possible ICMP tunneling / exfiltration",
            })
    return out


def intel_findings(result):
    if not threatintel.available():
        return []
    out, seen_ip, seen_ja3, seen_dom = [], set(), set(), set()
    for f in result.flows.values():
        for ip in (f.client_ip, f.server_ip):
            if is_public(ip) and ip not in seen_ip and threatintel.match_ip(ip):
                seen_ip.add(ip)
                out.append({
                    "kind": "intel_ip", "severity": "high", "ip": ip,
                    "title": f"{ip} is on a threat-intel blocklist",
                    "detail": "known-bad IP",
                })
        dom = f.sni or f.http_host
        if dom and dom not in seen_dom:
            hit = threatintel.match_domain(dom)
            if hit:
                seen_dom.add(dom)
                out.append({
                    "kind": "intel_domain", "severity": "high", "ip": f.server_ip,
                    "title": f"{dom} matches blocklist ({hit})",
                    "detail": f"to {f.server_ip}",
                })
        if f.ja3_hash and f.ja3_hash not in seen_ja3 and threatintel.match_ja3(f.ja3_hash):
            seen_ja3.add(f.ja3_hash)
            out.append({
                "kind": "bad_ja3", "severity": "high", "ip": f.server_ip,
                "title": f"malicious JA3 {f.ja3_hash[:16]}… → {f.server_ip}",
                "detail": f"SNI {f.sni or '?'}",
            })
    return out


def findings(result, hosts=None):
    """Produce a flat list of human-readable findings for the triage panel."""
    if hosts is None:
        hosts = per_host_stats(result)
    out = []

    for entry in shared_external_destinations(result):
        n = len(entry["internal_hosts"])
        out.append({
            "kind": "shared_destination", "severity": "high", "ip": entry["ip"],
            "title": f"{entry['ip']} contacted by {n} internal hosts",
            "detail": f"ports {sorted(entry['ports'])[:8]}",
        })

    for f in result.flows.values():
        if is_beaconing(f) and is_public(f.dst):
            period = beacon_period(f)
            out.append({
                "kind": "beaconing", "severity": "high", "ip": f.dst,
                "title": f"{f.src} → {f.dst}:{f.dst_port} regular cadence (~{period:.1f}s)",
                "detail": f"{f.packets} packets over {f.duration:.0f}s"
                          + (f" · {f.sni}" if f.sni else ""),
            })

    for ip, s in hosts.items():
        if not s["internal"]:
            continue
        if len(s["ext_peers_nonweb"]) >= FANOUT_MIN_DESTS:
            out.append({
                "kind": "fanout", "severity": "medium", "ip": ip,
                "title": f"{ip} fanned out to {len(s['ext_peers_nonweb'])} external "
                         f"hosts on non-web ports",
                "detail": "possible scanning or spray (web/DNS traffic excluded)",
            })
        if len(s["unacked_targets"]) >= SCAN_MIN_TARGETS:
            out.append({
                "kind": "scan", "severity": "medium", "ip": ip,
                "title": f"{ip} has {len(s['unacked_targets'])} unanswered/RST targets",
                "detail": "possible port/host scan",
            })
        if s["bytes_out"] >= EXFIL_MIN_BYTES and s["bytes_in"] > 0 \
                and s["bytes_out"] / max(1, s["bytes_in"]) >= EXFIL_OUT_IN_RATIO:
            mb = s["bytes_out"] / 1e6
            out.append({
                "kind": "exfil", "severity": "medium", "ip": ip,
                "title": f"{ip} sent {mb:.1f} MB out "
                         f"({s['bytes_out'] / max(1, s['bytes_in']):.0f}x inbound)",
                "detail": "possible data exfiltration",
            })

    out.extend(dns_findings(result))
    out.extend(exfil_findings(result))
    out.extend(lateral_findings(result))
    out.extend(intel_findings(result))
    out.extend(nonstandard_findings(nonstandard(result)))

    # attach ATT&CK technique tags
    for fnd in out:
        fnd.setdefault("attack", ATTACK.get(fnd["kind"]))

    sev_rank = {"high": 0, "medium": 1, "low": 2}
    out.sort(key=lambda x: sev_rank.get(x["severity"], 9))
    return out