graph_builder.py

"""Turn the parsed conversation table into a vis.js graph.

Aggregation (see DESIGN.md §4). An external IP is drawn as an individual node
when it's *interesting*, and folded into a cloud super-node when it's not:

  * internal hosts                          -> individual nodes (full detail)
  * internal <-> internal                   -> individual edges (lateral movement)
  * external shared by >= N internal hosts  -> red 'hub' node (fan-in / C2)
  * external that is a top talker by bytes  -> individual node (endpoint view)
  * external showing beaconing cadence      -> individual node
  * everything else (one-off contacts)      -> folded into a '🌐' cloud node

The top-talker promotion is what makes a single-host / endpoint capture useful:
its biggest external destinations stay visible instead of vanishing into a blob.
"""

from __future__ import annotations

import ipaddress
from collections import Counter, defaultdict

import enrichment
import threatintel
import virustotal
from netutils import is_internal, is_public, subnet_of
from detect import (
    SHARED_DEST_MIN_HOSTS,
    LATERAL_PORTS,
    per_host_stats,
    shared_external_destinations,
    is_beaconing,
    is_dga,
    beacon_period,
    nonstandard,
    findings,
)

MAX_CLOUD_MEMBERS = 200
MAX_EXTERNAL_NODES = 500     # external nodes drawn before the rest go to a cloud
TOP_N_EDGES = 1500           # honest top-N; reported in stats if exceeded
MAX_CONNECTIONS_PER_NODE = 150  # per-node connection rows shipped to the browser
TOP_LIST = 100               # entries per per-host breakdown list (UI shows 10 + more)

# Friendly service names for common server ports (for cluster labels / breakdowns)
SERVICE_NAMES = {
    20: "FTP-data", 21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP", 53: "DNS",
    67: "DHCP", 68: "DHCP", 80: "HTTP", 88: "Kerberos", 110: "POP3",
    123: "NTP", 135: "RPC", 137: "NetBIOS", 138: "NetBIOS", 139: "NetBIOS",
    143: "IMAP", 161: "SNMP", 389: "LDAP", 443: "HTTPS/QUIC", 445: "SMB",
    465: "SMTPS", 514: "Syslog", 587: "SMTP", 636: "LDAPS", 853: "DoT",
    993: "IMAPS", 995: "POP3S", 1433: "MSSQL", 1900: "SSDP", 3128: "Proxy",
    3306: "MySQL", 3389: "RDP", 5060: "SIP", 5222: "XMPP", 5228: "FCM",
    5353: "mDNS", 5985: "WinRM", 5986: "WinRM", 8080: "HTTP-alt",
    8443: "HTTPS-alt", 11211: "memcached",
}


def _service_label(port):
    if not port:
        return "other"
    name = SERVICE_NAMES.get(port)
    return f"{name} ({port})" if name else f"port {port}"


# Application/protocol recognition: conversation -> (key, display label).
def _app_of(port, proto, sni, http_host, ja3):
    if proto == "ICMP":
        return ("icmp", "ICMP")
    if port in (53, 5353, 5355):
        return ("dns", "DNS")
    if sni or http_host or ja3 or port in (80, 443, 8080, 8443, 8000, 8888, 8081):
        return ("web", "Web (HTTP/S)")
    if port == 22:
        return ("ssh", "SSH")
    if port in (25, 465, 587, 110, 143, 993, 995):
        return ("mail", "Mail")
    if port == 123:
        return ("ntp", "NTP")
    if port in (445, 139, 135):
        return ("smb", "SMB/RPC")
    if port == 3389:
        return ("rdp", "RDP")
    if port in (5985, 5986):
        return ("winrm", "WinRM")
    if port == 5228:
        return ("push", "Push/FCM")
    if port in (67, 68):
        return ("dhcp", "DHCP")
    if port == 1900:
        return ("ssdp", "SSDP")
    if port == 161:
        return ("snmp", "SNMP")
    if port in (5060, 5061):
        return ("sip", "SIP/VoIP")
    if port in SERVICE_NAMES:                 # a recognised named service
        return (f"svc{port}", _service_label(port))
    # everything else is an unrecognised application
    return ("unknown", "Unknown")


def build_apps(result, ext_domain, beacon_ext, bad_set):
    """Per internal host: traffic grouped by application/protocol, each with the
    destination hosts that had sessions of it. Powers the drill-down graph
    (host -> app box -> destination hosts)."""
    apps = defaultdict(lambda: defaultdict(
        lambda: {"label": "", "sessions": 0, "bytes": 0, "dests": {}, "ports": set()}))
    for f in result.flows.values():
        ci, si = is_internal(f.client_ip), is_internal(f.server_ip)
        if not (ci or si):
            continue
        host = f.client_ip if ci else f.server_ip
        peer = f.server_ip if host == f.client_ip else f.client_ip
        key, label = _app_of(f.server_port, f.proto, f.sni, f.http_host, f.ja3_hash)
        a = apps[host][key]
        a["label"] = label
        a["sessions"] += 1
        a["bytes"] += f.bytes
        if f.server_port:
            a["ports"].add(f.server_port)
        d = a["dests"].get(peer)
        if d is None:
            d = {"ip": peer, "domain": ext_domain.get(peer),
                 "sessions": 0, "bytes": 0, "pkts_out": 0, "pkts_in": 0,
                 "ports": set(), "beacon": peer in beacon_ext,
                 "threat": peer in bad_set, "internal": is_internal(peer)}
            a["dests"][peer] = d
        d["sessions"] += 1
        d["bytes"] += f.bytes
        if host == f.client_ip:
            d["pkts_out"] += f.c2s_pkts; d["pkts_in"] += f.s2c_pkts
        else:
            d["pkts_out"] += f.s2c_pkts; d["pkts_in"] += f.c2s_pkts
        if f.server_port:
            d["ports"].add(f.server_port)

    out = {}
    for host, cats in apps.items():
        lst = []
        for key, a in cats.items():
            dests = sorted(a["dests"].values(), key=lambda x: -x["bytes"])[:300]
            for d in dests:
                d["ports"] = sorted(d["ports"])[:8]
                if key == "dns":   # attach the queries this host sent to this resolver
                    pair = result.dns_by_pair.get((host, d["ip"]))
                    if pair:
                        d["queries"] = [{"domain": n, "count": c} for n, c in
                                        sorted(pair["name_counts"].items(), key=lambda kv: -kv[1])[:300]]
                        d["query_total"] = len(pair["name_counts"])
            lst.append({
                "key": key, "label": a["label"], "sessions": a["sessions"],
                "bytes": a["bytes"], "dest_count": len(a["dests"]),
                "ports": sorted(a["ports"])[:16],
                "threat": any(x["threat"] for x in a["dests"].values()),
                "beacon": any(x["beacon"] for x in a["dests"].values()),
                "dests": dests,
            })
        lst.sort(key=lambda x: -x["bytes"])
        out[host] = lst
    return out


def scan_indicators(result, cap):
    """Collect the indicators a VirusTotal scan would check: public external IPs
    (ranked by bytes, most significant first) plus every domain observed for
    them. Returns (indicator_list, n_ips, n_domains); the list is capped."""
    ip_bytes = defaultdict(int)
    for f in result.flows.values():
        ci, si = is_internal(f.client_ip), is_internal(f.server_ip)
        if ci == si:
            continue
        ext = f.server_ip if ci else f.client_ip
        if is_public(ext):
            ip_bytes[ext] += f.bytes
    ips = [ip for ip, _ in sorted(ip_bytes.items(), key=lambda kv: -kv[1])]
    domains = []
    seen_dom = set()
    for ip in ips:
        for d in result.ip_to_domain.get(ip, ()):
            if d not in seen_dom:
                seen_dom.add(d)
                domains.append(d)
    # interleave by significance: IPs first (already ranked), then domains
    inds = [("ip", ip) for ip in ips] + [("domain", d) for d in domains]
    return inds[:cap], len(ips), len(domains)


def malicious_scan(result, cap=None):
    """Query VirusTotal for the capture's public IPs/domains and return the top-5
    malicious indicators, each with the internal hosts that contacted it (for
    graph highlighting). Heavy on the VT API — gated behind a UI confirmation."""
    if not virustotal.available():
        return {"available": False, "note": virustotal.NOT_CONFIGURED}
    cap = cap or virustotal.scan_max()
    inds, n_ips, n_domains = scan_indicators(result, cap)

    hits, errors = [], 0
    for kind, value in inds:
        r = virustotal.lookup(kind, value)
        if not r.get("available"):                 # token went away / rejected
            return {"available": False, "note": r.get("note")}
        if r.get("error"):
            errors += 1
            continue
        det = (r.get("malicious", 0) or 0) + (r.get("suspicious", 0) or 0)
        if r.get("found") and det > 0:
            hits.append({
                "kind": kind, "value": value,
                "malicious": r.get("malicious", 0), "suspicious": r.get("suspicious", 0),
                "total": r.get("total", 0), "reputation": r.get("reputation"),
                "permalink": r.get("permalink"),
            })
    hits.sort(key=lambda h: -(h["malicious"] + h["suspicious"]))
    top = hits[:5]

    # attach resolved IPs + connecting internal hosts so the UI can draw the path
    for h in top:
        if h["kind"] == "ip":
            h["ips"] = [h["value"]]
        else:
            h["ips"] = sorted({ip for ip, doms in result.ip_to_domain.items()
                               if h["value"] in doms})
        rl = reverse_lookup(result, h["value"])
        h["hosts"] = [x["host"] for x in rl["hosts"]]

    return {
        "available": True, "top": top,
        "scanned": len(inds), "ips": n_ips, "domains": n_domains,
        "cap": cap, "errors": errors, "malicious_total": len(hits),
    }


def reverse_lookup(result, query):
    """Reverse search: given a suspicious IP or domain, find every INTERNAL host
    that talked to it. Scans the full conversation table (not the drill-down
    cap), so nothing is missed.

    `query` is matched as an exact IP, or — if it isn't an IP literal — as a
    case-insensitive substring against each peer's observed domains (DNS answers,
    TLS SNI, HTTP Host). Returns hosts ranked by bytes, each with the matched
    targets broken down by protocol + server port."""
    qraw = (query or "").strip()
    q = qraw.lower()
    if not q:
        return {"query": query, "host_count": 0, "matched_ips": [], "hosts": []}

    try:
        ipaddress.ip_address(qraw)
        is_ip = True
    except ValueError:
        is_ip = False

    def _matches(ip, f):
        if is_ip:
            return ip == qraw
        for dom in result.ip_to_domain.get(ip, ()):  # DNS-observed names for this IP
            if q in dom.lower():
                return True
        if f.sni and q in f.sni.lower():
            return True
        if f.http_host and q in f.http_host.lower():
            return True
        return False

    def _domain_of(ip, f):
        doms = result.ip_to_domain.get(ip)
        if doms:
            return sorted(doms, key=len)[0]
        return f.sni or f.http_host

    hosts_agg = {}
    matched_targets = set()
    for f in result.flows.values():
        s_match = _matches(f.server_ip, f)
        c_match = _matches(f.client_ip, f) if not s_match else False
        if not (s_match or c_match):
            continue
        # the matched endpoint is the target; the other endpoint is the connector
        if s_match:
            target, conn = f.server_ip, f.client_ip
            out_p, in_p = f.c2s_pkts, f.s2c_pkts
            out_b, in_b = f.c2s_bytes, f.s2c_bytes
        else:
            target, conn = f.client_ip, f.server_ip
            out_p, in_p = f.s2c_pkts, f.c2s_pkts
            out_b, in_b = f.s2c_bytes, f.c2s_bytes
        if not is_internal(conn):
            continue  # we only report internal hosts as connectors
        matched_targets.add(target)
        ha = hosts_agg.get(conn)
        if ha is None:
            ha = {"host": conn, "subnet": subnet_of(conn),
                  "total_bytes": 0, "total_sessions": 0, "rows": {}}
            hosts_agg[conn] = ha
        rk = (target, f.proto, f.server_port)
        row = ha["rows"].get(rk)
        if row is None:
            row = {"target": target, "domain": _domain_of(target, f),
                   "proto": f.proto, "port": f.server_port,
                   "sessions": 0, "pkts_out": 0, "pkts_in": 0,
                   "bytes_out": 0, "bytes_in": 0,
                   "threat": bool(threatintel.match_ip(target)),
                   "internal": is_internal(target)}
            ha["rows"][rk] = row
        row["sessions"] += 1
        row["pkts_out"] += out_p
        row["pkts_in"] += in_p
        row["bytes_out"] += out_b
        row["bytes_in"] += in_b
        ha["total_sessions"] += 1
        ha["total_bytes"] += f.bytes

    hosts = []
    for ha in hosts_agg.values():
        rows = sorted(ha["rows"].values(),
                      key=lambda r: -(r["bytes_out"] + r["bytes_in"]))
        ha["rows"] = rows
        hosts.append(ha)
    hosts.sort(key=lambda h: -h["total_bytes"])
    return {
        "query": qraw,
        "is_ip": is_ip,
        "host_count": len(hosts),
        "matched_ips": sorted(matched_targets),
        "hosts": hosts,
    }


def _net16(ip):
    try:
        addr = ipaddress.ip_address(ip)
    except ValueError:
        return "unknown"
    if addr.version == 4:
        return str(ipaddress.ip_network(f"{ip}/16", strict=False))
    return str(ipaddress.ip_network(f"{ip}/48", strict=False))


def _proto_mix(rows):
    """Total packets per protocol across a node's connection rows."""
    mix = defaultdict(int)
    for r in rows:
        mix[r["proto"]] += r["pkts_out"] + r["pkts_in"]
    return dict(mix)


def _connection_index(flows):
    """For each host, summarise its connections grouped by (peer, proto, server
    port) — packets in/out and how many sessions, from that host's viewpoint.
    Source/ephemeral ports are intentionally not part of the grouping key."""
    conn = defaultdict(lambda: defaultdict(
        lambda: {"pkts_out": 0, "pkts_in": 0, "bytes_out": 0, "bytes_in": 0,
                 "sessions": 0}))
    for f in flows:
        # client's view
        c = conn[f.client_ip][(f.server_ip, f.proto, f.server_port)]
        c["pkts_out"] += f.c2s_pkts; c["pkts_in"] += f.s2c_pkts
        c["bytes_out"] += f.c2s_bytes; c["bytes_in"] += f.s2c_bytes
        c["sessions"] += 1
        # server's view (mirror)
        srv = conn[f.server_ip][(f.client_ip, f.proto, f.server_port)]
        srv["pkts_out"] += f.s2c_pkts; srv["pkts_in"] += f.c2s_pkts
        srv["bytes_out"] += f.s2c_bytes; srv["bytes_in"] += f.c2s_bytes
        srv["sessions"] += 1
    return conn


def _fmt_bytes(n):
    n = float(n)
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if n < 1024 or unit == "TB":
            return f"{n:.0f} {unit}" if unit == "B" else f"{n:.1f} {unit}"
        n /= 1024
    return f"{n:.1f} TB"


def build(result):
    hosts = per_host_stats(result)
    shared = shared_external_destinations(result)
    shared_ips = {e["ip"] for e in shared}
    shared_by = {e["ip"]: len(e["internal_hosts"]) for e in shared}

    # --- pass 1: decide which external IPs get promoted to individual nodes --
    ext_pair_bytes = defaultdict(int)   # (internal, external) -> bytes
    ext_total_bytes = defaultdict(int)  # external -> bytes
    ext_total_pkts = defaultdict(int)
    ext_ports = defaultdict(set)
    ext_port_pkts = defaultdict(Counter)  # external -> Counter(server_port -> pkts)
    ext_domain = {}                     # external -> best domain label
    ext_sni = {}                        # external -> SNI seen
    ext_ja3 = {}                        # external -> JA3 hash seen
    ext_beacon_times = defaultdict(list)  # external -> relative c2s times
    ext_beacon_period = {}
    beacon_ext = set()

    def _label_domain(ip):
        if ip in ext_domain:
            return ext_domain[ip]
        doms = result.ip_to_domain.get(ip)
        return sorted(doms, key=len)[0] if doms else None

    for f in result.flows.values():
        ci, si = is_internal(f.client_ip), is_internal(f.server_ip)
        if ci == si:
            continue  # internal<->internal or external<->external handled later
        internal_ip = f.client_ip if ci else f.server_ip
        external_ip = f.server_ip if ci else f.client_ip
        if not is_public(external_ip):
            continue
        ext_pair_bytes[(internal_ip, external_ip)] += f.bytes
        ext_total_bytes[external_ip] += f.bytes
        ext_total_pkts[external_ip] += f.packets
        if f.server_port:
            ext_ports[external_ip].add(f.server_port)
            ext_port_pkts[external_ip][f.server_port] += f.packets
        if f.sni:
            ext_sni[external_ip] = f.sni
            ext_domain[external_ip] = f.sni
        elif f.http_host and external_ip not in ext_domain:
            ext_domain[external_ip] = f.http_host
        if f.ja3_hash:
            ext_ja3[external_ip] = f.ja3_hash
        if is_beaconing(f):
            beacon_ext.add(external_ip)
            ext_beacon_times[external_ip] = list(getattr(f, "times", []) or [])
            ext_beacon_period[external_ip] = beacon_period(f)

    # threat-intel-flagged externals (always promoted, never clustered away)
    bad_ext = set()
    for ip in ext_total_bytes:
        if threatintel.match_ip(ip) or threatintel.match_ja3(ext_ja3.get(ip)) \
                or (ext_domain.get(ip) and threatintel.match_domain(ext_domain[ip])):
            bad_ext.add(ip)

    # offline geo/ASN enrichment for promoted externals (for ASN clustering)
    geo_on = enrichment.geo_available()

    flagged = set(shared_ips) | set(beacon_ext) | bad_ext
    promoted = set(flagged)
    for ip, _b in sorted(ext_total_bytes.items(), key=lambda kv: -kv[1]):
        if len(promoted) >= MAX_EXTERNAL_NODES:
            break
        promoted.add(ip)

    # --- pass 2: build nodes and edges --------------------------------------
    nodes = {}
    pair_edges = defaultdict(
        lambda: {"bytes": 0, "packets": 0, "rst": 0, "syn": 0, "ack": 0,
                 "beacon": False, "ports": set()}
    )
    cloud = defaultdict(lambda: {"members": set(), "bytes": 0, "packets": 0})

    def ensure_host_node(ip):
        if ip in nodes:
            return
        s = hosts.get(ip, {})
        total = s.get("bytes_out", 0) + s.get("bytes_in", 0)
        nodes[ip] = {
            "id": ip, "label": ip, "type": "internal",
            "subnet": subnet_of(ip), "bytes": total,
            "packets": s.get("packets", 0),
            "fanout": len(s.get("ext_peers", set())),
            "value": max(1, total),
            "title": (f"<b>{ip}</b><br>internal · subnet {subnet_of(ip)}"
                      f"<br>{_fmt_bytes(total)} · {s.get('packets', 0)} pkts"
                      f"<br>peers: {len(s.get('peers', set()))}"),
        }

    def _threat(ip):
        """Return a short threat-intel reason for a node, or None."""
        if threatintel.match_ip(ip):
            return "blocklisted IP"
        dom = _label_domain(ip)
        if dom and threatintel.match_domain(dom):
            return f"blocklisted domain {dom}"
        if threatintel.match_ja3(ext_ja3.get(ip)):
            return "malicious JA3"
        return None

    def _common(ip, node):
        node["domain"] = _label_domain(ip)
        node["sni"] = ext_sni.get(ip)
        node["ja3"] = ext_ja3.get(ip)
        # cluster keys (service / subnet / ASN) used by the clustered graph view
        top_port = ext_port_pkts[ip].most_common(1)[0][0] if ext_port_pkts.get(ip) else 0
        node["service"] = _service_label(top_port)
        node["cluster_service"] = node["service"]
        node["cluster_subnet"] = _net16(ip)
        if geo_on:
            e = enrichment.enrich(ip)
            node["country"] = e.get("country_code")
            node["asn"] = e.get("asn")
            node["org"] = e.get("org")
            if e.get("country") or e.get("org"):
                node["title"] += f"<br>{e.get('country') or ''} {('· ' + e['org']) if e.get('org') else ''}"
        node["cluster_asn"] = (node.get("org") or
                               (f"AS{node['asn']}" if node.get("asn") else "unknown ASN"))
        if node["domain"]:
            node["label"] = node["domain"]
            node["title"] += f"<br>domain: {node['domain']}"
        if node["ja3"]:
            node["title"] += f"<br>JA3 {node['ja3'][:16]}…"
        if ip in beacon_ext:
            node["beacon"] = True
            node["beacon_period"] = round(ext_beacon_period.get(ip, 0), 1)
            ts = sorted(ext_beacon_times.get(ip, []))
            if ts:
                base = ts[0]
                node["beacon_times"] = [round(t - base, 2) for t in ts[:512]]
            node["title"] += f"<br>⏱ beaconing ~{node.get('beacon_period', 0)}s"
        threat = _threat(ip)
        if threat:
            node["threat"] = threat
            node["is_bad"] = True
            node["title"] += f"<br>☠ {threat}"

    def ensure_external_node(ip):
        if ip in nodes:
            return
        b = ext_total_bytes.get(ip, 0)
        nodes[ip] = {
            "id": ip, "label": ip,
            "type": "external", "beacon": ip in beacon_ext,
            "subnet": subnet_of(ip), "bytes": b,
            "value": max(2, b),
            "title": (f"<b>{ip}</b><br>external"
                      f"<br>{_fmt_bytes(b)} · {ext_total_pkts.get(ip, 0)} pkts"
                      f"<br>ports {sorted(ext_ports.get(ip, []))[:8]}"),
        }
        _common(ip, nodes[ip])

    def ensure_hub_node(ip):
        n = shared_by.get(ip, 0)
        nodes[ip] = {
            "id": ip, "label": ip, "type": "hub",
            "subnet": subnet_of(ip), "shared_by": n,
            "bytes": ext_total_bytes.get(ip, 0),
            "value": max(6, n * 6),
            "title": (f"<b>⚠ {ip}</b><br>shared by {n} internal hosts"
                      f"<br>{_fmt_bytes(ext_total_bytes.get(ip, 0))}"
                      f"<br>possible C2 / shared destination"),
        }
        _common(ip, nodes[ip])

    for f in result.flows.values():
        ci, si = is_internal(f.client_ip), is_internal(f.server_ip)

        if ci and si:
            ensure_host_node(f.client_ip)
            ensure_host_node(f.server_ip)
            e = pair_edges[(f.client_ip, f.server_ip)]
        elif ci != si:
            internal_ip = f.client_ip if ci else f.server_ip
            external_ip = f.server_ip if ci else f.client_ip
            if not is_public(external_ip):
                continue
            ensure_host_node(internal_ip)
            if external_ip in shared_ips:
                ensure_hub_node(external_ip)
                e = pair_edges[(f.client_ip, f.server_ip)]
            elif external_ip in promoted:
                ensure_external_node(external_ip)
                e = pair_edges[(f.client_ip, f.server_ip)]
            else:
                c = cloud[internal_ip]
                if len(c["members"]) < MAX_CLOUD_MEMBERS * 4:
                    c["members"].add(external_ip)
                c["bytes"] += f.bytes
                c["packets"] += f.packets
                continue
        else:
            continue  # external<->external

        e["bytes"] += f.bytes
        e["packets"] += f.packets
        e["rst"] += f.rst
        e["syn"] += f.syn
        e["ack"] += f.ack
        if f.server_port:
            e["ports"].add(f.server_port)
        if is_beaconing(f):
            e["beacon"] = True

    # --- materialise edges ---------------------------------------------------
    edges = []
    for (a, b), e in pair_edges.items():
        failed = e["rst"] > 0 or (e["syn"] > 0 and e["ack"] == 0)
        to_hub = nodes.get(b, {}).get("type") == "hub"
        lateral_ports = [p for p in e["ports"] if p in LATERAL_PORTS]
        lateral = is_internal(a) and is_internal(b) and bool(lateral_ports)
        if lateral:
            color = "#e0a526"          # amber for lateral movement
        elif failed or to_hub or e["beacon"]:
            color = "#e74c3c"
        else:
            color = "#5b8c5a"
        ports = sorted(e["ports"])[:6]
        svc = ", ".join(LATERAL_PORTS[p][0] for p in lateral_ports) if lateral else ""
        edges.append({
            "from": a, "to": b, "value": max(1, e["packets"]),
            "bytes": e["bytes"], "packets": e["packets"],
            "beacon": e["beacon"], "failed": failed, "lateral": lateral,
            "color": {"color": color},
            "label": ",".join(str(p) for p in ports) if ports else "",
            "title": (f"<b>{a} → {b}</b><br>{_fmt_bytes(e['bytes'])} · "
                      f"{e['packets']} pkts<br>ports {ports}"
                      + (f"<br>↔ lateral movement ({svc})" if lateral else "")
                      + ("<br>⚠ failed/RST connections" if failed else "")
                      + ("<br>⏱ regular cadence (beaconing)" if e["beacon"] else "")),
        })

    # --- materialise cloud super-nodes --------------------------------------
    for src, c in cloud.items():
        if not c["members"]:
            continue
        cid, count = f"cloud:{src}", len(c["members"])
        nodes[cid] = {
            "id": cid, "label": f"🌐 {count} ext host{'s' if count != 1 else ''}",
            "type": "cloud", "source": src, "count": count,
            "members": sorted(c["members"])[:MAX_CLOUD_MEMBERS],
            "members_truncated": count > MAX_CLOUD_MEMBERS,
            "value": max(3, count),
            "title": (f"<b>🌐 {count} external hosts</b><br>low-volume one-off "
                      f"destinations of {src}<br>{_fmt_bytes(c['bytes'])} · "
                      f"{c['packets']} pkts<br>click to list members"),
        }
        edges.append({
            "from": src, "to": cid, "value": max(1, c["packets"]),
            "bytes": c["bytes"], "packets": c["packets"],
            "color": {"color": "#7f8c8d"}, "dashes": True,
            "title": f"{src} → {count} external hosts · {_fmt_bytes(c['bytes'])}",
        })

    # --- attach per-node connection breakdowns ------------------------------
    conn = _connection_index(result.flows.values())
    for ip, node in nodes.items():
        if node["type"] not in ("internal", "external", "hub"):
            continue
        rows = [
            {"peer": peer, "proto": proto, "port": port,
             "sessions": v["sessions"],
             "pkts_out": v["pkts_out"], "pkts_in": v["pkts_in"],
             "bytes_out": v["bytes_out"], "bytes_in": v["bytes_in"]}
            for (peer, proto, port), v in conn.get(ip, {}).items()
        ]
        rows.sort(key=lambda r: -(r["pkts_out"] + r["pkts_in"]))
        node["connections_total"] = len(rows)
        node["sessions_total"] = sum(r["sessions"] for r in rows)
        node["proto_mix"] = _proto_mix(rows)
        # internal hosts use the drill-down panel instead of the connection table
        if node["type"] != "internal":
            node["connections"] = rows[:MAX_CONNECTIONS_PER_NODE]

    # --- per-host drill-down (DNS / ports / dests / non-standard) ------------
    ns_per_host = nonstandard(result)
    apps_map = build_apps(result, ext_domain, beacon_ext, bad_ext)
    host_ports = defaultdict(Counter)
    host_dests = defaultdict(Counter)
    dest_ports = defaultdict(set)                          # external IP -> server ports (global)
    host_dest_ports = defaultdict(lambda: defaultdict(set))  # host -> external IP -> server ports
    for f in result.flows.values():
        ci, si = is_internal(f.client_ip), is_internal(f.server_ip)
        internal_ip = f.client_ip if ci else (f.server_ip if si else None)
        if internal_ip is None:
            continue
        host_ports[internal_ip][f.server_port] += f.packets
        ext = f.server_ip if (ci and not si) else (f.client_ip if (si and not ci) else None)
        if ext and is_public(ext):
            host_dests[internal_ip][ext] += f.bytes
            if f.server_port:
                dest_ports[ext].add(f.server_port)
                host_dest_ports[internal_ip][ext].add(f.server_port)

    for ip, node in nodes.items():
        if node["type"] != "internal":
            continue
        c = result.dns_by_client.get(ip, {})
        ncounts = c.get("name_counts", {})
        dns_top = sorted(ncounts.items(), key=lambda kv: -kv[1])[:TOP_LIST]
        ports, dests = host_ports.get(ip, Counter()), host_dests.get(ip, Counter())
        node["apps"] = apps_map.get(ip, [])
        node["detail"] = {
            "dns": {
                "resolvers": sorted(c.get("resolvers", [])),
                "total": len(ncounts), "queries": c.get("queries", 0),
                "top": [{"domain": n, "count": cnt, "dga": is_dga(n)}
                        for n, cnt in dns_top],
            },
            "ports": {
                "total": len([p for p in ports if p]),
                "top": [{"port": p, "service": _service_label(p), "packets": pk}
                        for p, pk in ports.most_common(TOP_LIST) if p],
            },
            "dests": {
                "total": len(dests),
                "top": [{"ip": d, "domain": ext_domain.get(d), "bytes": b,
                         "ports": sorted(host_dest_ports[ip].get(d, set()))[:8]}
                        for d, b in dests.most_common(TOP_LIST)],
            },
            "nonstandard": ns_per_host.get(ip, []),
        }

    all_ports, all_dests = Counter(), Counter()
    for c in host_ports.values():
        all_ports.update(c)
    for c in host_dests.values():
        all_dests.update(c)
    overview = {
        "internal_hosts": sorted(n["id"] for n in nodes.values() if n["type"] == "internal"),
        "top_ports": [{"port": p, "service": _service_label(p), "packets": pk}
                      for p, pk in all_ports.most_common(TOP_LIST) if p],
        "top_dests": [{"ip": d, "domain": ext_domain.get(d), "bytes": b,
                       "ports": sorted(dest_ports.get(d, set()))[:8]}
                      for d, b in all_dests.most_common(TOP_LIST)],
        "nonstandard_hosts": len(ns_per_host),
    }

    # --- honest top-N on edges ----------------------------------------------
    edges_total = len(edges)
    edges_truncated = False
    if edges_total > TOP_N_EDGES:
        edges.sort(key=lambda x: x.get("packets", 0), reverse=True)
        edges = edges[:TOP_N_EDGES]
        kept = {e["from"] for e in edges} | {e["to"] for e in edges}
        nodes = {k: v for k, v in nodes.items() if k in kept}
        edges_truncated = True

    all_findings = findings(result, hosts)

    # --- DNS panel ----------------------------------------------------------
    dns_top = sorted(result.dns_by_name.items(), key=lambda kv: -kv[1]["count"])[:40]
    dns_panel = {
        "total_names": len(result.dns_by_name),
        "resolvers": sorted(result.dns_resolvers),
        "top": [{"domain": n, "count": d["count"], "clients": len(d["clients"]),
                 "dga": is_dga(n), "answers": sorted(d["answers"])[:4]}
                for n, d in dns_top],
        "dga": sorted({n for n in result.dns_by_name if is_dga(n)})[:60],
    }

    # --- IOC list (compiled from flagged nodes + findings) ------------------
    iocs = {}

    def _ioc(t, v, reason):
        if v and (t, v) not in iocs:
            iocs[(t, v)] = {"type": t, "value": v, "reason": reason}

    for ip in shared_ips:
        _ioc("ip", ip, "shared destination / possible C2")
    for ip in beacon_ext:
        _ioc("ip", ip, "beaconing")
        _ioc("domain", ext_domain.get(ip), "beaconing")
    for n in nodes.values():
        if n.get("is_bad"):
            _ioc("ip", n["id"], n.get("threat", "threat-intel hit"))
            _ioc("domain", n.get("domain"), n.get("threat", ""))
            _ioc("ja3", n.get("ja3"), "flagged host JA3")
    for dom in dns_panel["dga"]:
        _ioc("domain", dom, "algorithmic / DGA-like domain")
    for f in all_findings:
        if f.get("kind") in ("exfil_dest", "icmp_tunnel") and f.get("ip"):
            _ioc("ip", f["ip"], "exfiltration destination")

    # The graph starts with internal hosts only; app/destination nodes are
    # spawned client-side on click (drill-down). Keep internal<->internal
    # (lateral) edges so host-to-host movement is visible from the start.
    graph_nodes = [n for n in nodes.values() if n["type"] == "internal"]
    graph_ids = {n["id"] for n in graph_nodes}
    graph_edges = [e for e in edges if e["from"] in graph_ids and e["to"] in graph_ids]

    return {
        "nodes": graph_nodes,
        "edges": graph_edges,
        "shared_destinations": [
            {"ip": e["ip"], "internal_hosts": sorted(e["internal_hosts"]),
             "host_count": len(e["internal_hosts"]),
             "subnets": sorted({subnet_of(h) for h in e["internal_hosts"]}),
             "bytes": e["bytes"], "packets": e["packets"],
             "ports": sorted(e["ports"])[:12]}
            for e in shared
        ],
        "findings": all_findings,
        "dns": dns_panel,
        "overview": overview,
        "iocs": list(iocs.values()),
        "threatintel": {**threatintel.stats(), "available": threatintel.available()},
        "stats": {
            "total_packets": result.total_packets,
            "parsed_packets": result.parsed_packets,
            "non_ip_packets": result.non_ipv4_packets,
            "noise_packets": result.noise_packets,
            "flows": len(result.flows),
            "hosts": len([n for n in nodes.values()
                          if n["type"] in ("internal", "external", "hub")]),
            "internal_hosts": len([n for n in nodes.values() if n["type"] == "internal"]),
            "external_shown": len([n for n in nodes.values()
                                   if n["type"] in ("external", "hub")]),
            "shared_destinations": len(shared),
            "capture_duration": round(result.capture_duration, 1),
            "parse_seconds": round(result.elapsed, 2),
            "edges_total": edges_total,
            "edges_shown": len(edges),
            "edges_truncated": edges_truncated,
            "truncated_packets": result.truncated_packets,
            "truncated_time": result.truncated_time,
            "shared_dest_threshold": SHARED_DEST_MIN_HOSTS,
        },
    }