PCAP Flow Summary

#!/usr/bin/env python3
"""
PCAP Flow Summary

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/pcap-flow-summary

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    # No extra packages — uses Python stdlib only
"""

import argparse
import ipaddress
import json
import os
import struct
import sys
from datetime import datetime, timezone

DASH = "-"

# ---------------------------------------------------------------------------
# Capture file readers.  A pcap is UNTRUSTED input: every length field in it
# is attacker- or bug-controlled, so nothing here trusts one without checking
# it against the bytes actually present.
# ---------------------------------------------------------------------------

LINKTYPE_NULL = 0
LINKTYPE_ETHERNET = 1
LINKTYPE_RAW = 101
LINKTYPE_LINUX_SLL = 113
LINKTYPE_IPV4 = 228
LINKTYPE_IPV6 = 229
LINKTYPE_LINUX_SLL2 = 276


class CaptureError(Exception):
    """The file is not a capture we can read at all."""


class Packet:
    __slots__ = ("ts", "caplen", "origlen", "data", "linktype")

    def __init__(self, ts, caplen, origlen, data, linktype):
        self.ts = ts
        self.caplen = caplen
        self.origlen = origlen
        self.data = data
        self.linktype = linktype


def _read_exact(fh, n):
    buf = fh.read(n)
    if len(buf) != n:
        return None
    return buf


def read_pcap(fh, head):
    """Classic libpcap.  Yields Packet, then a dict of file-level facts."""
    if head in (b"\xa1\xb2\xc3\xd4", b"\xa1\xb2\x3c\x4d"):
        endian = ">"
    elif head in (b"\xd4\xc3\xb2\xa1", b"\x4d\x3c\xb2\xa1"):
        endian = "<"
    else:
        raise CaptureError("not a pcap file")
    nanos = head in (b"\xa1\xb2\x3c\x4d", b"\x4d\x3c\xb2\xa1")
    rest = _read_exact(fh, 20)
    if rest is None:
        raise CaptureError("pcap header truncated at byte 24")
    _vmaj, _vmin, _tz, _sig, snaplen, linktype = struct.unpack(endian + "HHiIII", rest)
    meta = {"format": "pcap", "snaplen": snaplen, "linktypes": {linktype}, "truncated_file": False}
    divisor = 1_000_000_000.0 if nanos else 1_000_000.0
    while True:
        rh = fh.read(16)
        if not rh:
            break
        if len(rh) < 16:
            meta["truncated_file"] = True
            break
        secs, frac, caplen, origlen = struct.unpack(endian + "IIII", rh)
        if caplen > 0x400000:  # 4 MB: a length field that cannot be believed
            meta["truncated_file"] = True
            break
        data = fh.read(caplen)
        if len(data) < caplen:
            meta["truncated_file"] = True
            break
        yield Packet(secs + frac / divisor, caplen, origlen, data, linktype)
    yield meta


def read_pcapng(fh):
    """pcapng.  Only the blocks that carry packets matter here."""
    fh.seek(0)
    meta = {"format": "pcapng", "snaplen": 0, "linktypes": set(), "truncated_file": False}
    endian = "<"
    ifaces = []  # (linktype, tsresol_divisor)
    while True:
        bh = fh.read(8)
        if not bh:
            break
        if len(bh) < 8:
            meta["truncated_file"] = True
            break
        btype = struct.unpack(endian + "I", bh[:4])[0]
        if btype == 0x0A0D0D0A:  # Section Header: byte order magic decides endianness
            bom = fh.read(4)
            if len(bom) < 4:
                meta["truncated_file"] = True
                break
            # 0x1A2B3C4D stored least-significant byte first reads as
            # 4d 3c 2b 1a on the wire. Getting this pair the wrong way round
            # makes every block length absurd.
            if bom == b"\x4d\x3c\x2b\x1a":
                endian = "<"
            elif bom == b"\x1a\x2b\x3c\x4d":
                endian = ">"
            else:
                raise CaptureError("pcapng section header has no valid byte-order magic")
            blen = struct.unpack(endian + "I", bh[4:])[0]
            if blen < 16 or blen > 0x1000000:
                raise CaptureError("pcapng block length is not believable")
            # 8 header bytes and the 4-byte magic are already consumed.
            body = fh.read(blen - 12)
            if len(body) < blen - 12:
                meta["truncated_file"] = True
                break
            ifaces = []
            continue
        blen = struct.unpack(endian + "I", bh[4:])[0]
        if blen < 12 or blen > 0x1000000:
            meta["truncated_file"] = True
            break
        body = fh.read(blen - 8)
        if len(body) < blen - 8:
            meta["truncated_file"] = True
            break
        body = body[:-4]  # drop the trailing copy of the block length
        if btype == 0x00000001 and len(body) >= 8:  # Interface Description
            linktype, _res, snaplen = struct.unpack(endian + "HHI", body[:8])
            divisor = 1_000_000.0
            opts = body[8:]
            off = 0
            while off + 4 <= len(opts):
                ocode, olen = struct.unpack(endian + "HH", opts[off:off + 4])
                oval = opts[off + 4:off + 4 + olen]
                if ocode == 0:
                    break
                if ocode == 9 and len(oval) == 1:  # if_tsresol
                    raw = oval[0]
                    if raw & 0x80:
                        divisor = float(2 ** (raw & 0x7F))
                    else:
                        divisor = float(10 ** raw)
                off += 4 + olen + ((4 - olen % 4) % 4)
            ifaces.append((linktype, divisor))
            meta["linktypes"].add(linktype)
            if snaplen and (meta["snaplen"] == 0 or snaplen < meta["snaplen"]):
                meta["snaplen"] = snaplen
        elif btype == 0x00000006 and len(body) >= 20:  # Enhanced Packet
            iid, tsh, tsl, caplen, origlen = struct.unpack(endian + "IIIII", body[:20])
            if iid >= len(ifaces):
                continue
            linktype, divisor = ifaces[iid]
            if caplen > len(body) - 20:
                meta["truncated_file"] = True
                continue
            ts = ((tsh << 32) | tsl) / divisor
            yield Packet(ts, caplen, origlen, body[20:20 + caplen], linktype)
        elif btype == 0x00000003 and len(body) >= 4:  # Simple Packet: no timestamp
            origlen = struct.unpack(endian + "I", body[:4])[0]
            linktype, _d = ifaces[0] if ifaces else (LINKTYPE_ETHERNET, 1e6)
            data = body[4:4 + min(origlen, len(body) - 4)]
            yield Packet(None, len(data), origlen, data, linktype)
    yield meta


def read_capture(path):
    """Returns (list_of_packets, meta). Raises CaptureError on an unreadable file."""
    try:
        fh = open(path, "rb")
    except OSError as exc:
        raise CaptureError(f"cannot open {path}: {exc}") from exc
    with fh:
        head = fh.read(4)
        if len(head) < 4:
            raise CaptureError("file is shorter than a capture file header")
        if head == b"\x0a\x0d\x0d\x0a":
            gen = read_pcapng(fh)
        elif head[:2] == b"\x1f\x8b":
            raise CaptureError(
                "this is a gzip file. Decompress it first: gunzip capture.pcap.gz"
            )
        else:
            gen = read_pcap(fh, head)
        packets = []
        meta = {}
        for item in gen:
            if isinstance(item, dict):
                meta = item
            else:
                packets.append(item)
    return packets, meta


# ---------------------------------------------------------------------------
# Decoding.  Every step returns None rather than raising on a short buffer:
# a truncated packet is normal, not an error.
# ---------------------------------------------------------------------------

ETH_IPV4, ETH_IPV6, ETH_VLAN, ETH_QINQ, ETH_ARP = 0x0800, 0x86DD, 0x8100, 0x88A8, 0x0806
ETH_MPLS = 0x8847

PROTO_NAMES = {1: "ICMP", 6: "TCP", 17: "UDP", 47: "GRE", 50: "ESP", 51: "AH", 58: "ICMPv6", 89: "OSPF", 112: "VRRP", 132: "SCTP"}

WELL_KNOWN = {
    20: "ftp-data", 21: "ftp", 22: "ssh", 23: "telnet", 25: "smtp", 53: "dns",
    67: "dhcp", 68: "dhcp", 69: "tftp", 80: "http", 88: "kerberos", 110: "pop3",
    123: "ntp", 135: "msrpc", 137: "netbios", 139: "netbios", 143: "imap",
    161: "snmp", 179: "bgp", 389: "ldap", 443: "https", 445: "smb", 514: "syslog",
    520: "rip", 587: "smtp-sub", 636: "ldaps", 993: "imaps", 995: "pop3s",
    1433: "mssql", 1521: "oracle", 1701: "l2tp", 1812: "radius", 3306: "mysql",
    3389: "rdp", 4500: "ipsec-nat-t", 5060: "sip", 5432: "postgres",
    5671: "amqps", 6379: "redis", 8080: "http-alt", 8443: "https-alt", 9200: "elastic",
}


class Decoded:
    __slots__ = ("src", "dst", "proto", "sport", "dport", "ip_len", "payload_len",
                 "tcp", "l4_truncated", "vlan")

    def __init__(self):
        self.src = self.dst = None
        self.proto = None
        self.sport = self.dport = None
        self.ip_len = None
        self.payload_len = None
        self.tcp = None
        self.l4_truncated = False
        self.vlan = None


class TcpBits:
    __slots__ = ("seq", "ack", "flags", "window", "wscale_opt", "plen", "sack_blocks")

    def __init__(self, seq, ack, flags, window, plen, wscale_opt, sack_blocks):
        self.seq = seq
        self.ack = ack
        self.flags = flags
        self.window = window
        self.plen = plen
        self.wscale_opt = wscale_opt
        self.sack_blocks = sack_blocks


def strip_link(data, linktype):
    """Returns (ethertype_or_None, payload) or (None, None) if undecodable."""
    if linktype == LINKTYPE_ETHERNET:
        if len(data) < 14:
            return None, None, None
        etype = struct.unpack("!H", data[12:14])[0]
        off = 14
        vlan = None
        hops = 0
        while etype in (ETH_VLAN, ETH_QINQ) and hops < 4:
            if len(data) < off + 4:
                return None, None, None
            tci = struct.unpack("!H", data[off:off + 2])[0]
            if vlan is None:
                vlan = tci & 0x0FFF
            etype = struct.unpack("!H", data[off + 2:off + 4])[0]
            off += 4
            hops += 1
        return (etype, data[off:], vlan)
    if linktype in (LINKTYPE_RAW, LINKTYPE_IPV4, LINKTYPE_IPV6):
        if not data:
            return None, None, None
        ver = data[0] >> 4
        return (ETH_IPV4 if ver == 4 else ETH_IPV6 if ver == 6 else None, data, None)
    if linktype == LINKTYPE_NULL:
        if len(data) < 4:
            return None, None, None
        fam = struct.unpack("<I", data[:4])[0]
        if fam not in (2, 24, 28, 30):
            fam = struct.unpack(">I", data[:4])[0]
        return (ETH_IPV4 if fam == 2 else ETH_IPV6, data[4:], None)
    if linktype == LINKTYPE_LINUX_SLL:
        if len(data) < 16:
            return None, None, None
        return (struct.unpack("!H", data[14:16])[0], data[16:], None)
    if linktype == LINKTYPE_LINUX_SLL2:
        if len(data) < 20:
            return None, None, None
        return (struct.unpack("!H", data[:2])[0], data[20:], None)
    return None, None, None


def parse_tcp(buf, payload_available):
    if len(buf) < 20:
        return None, True
    sport, dport, seq, ack, offres, flags, window = struct.unpack("!HHIIBBH", buf[:16])
    doff = (offres >> 4) * 4
    if doff < 20:
        return None, True
    wscale = None
    sack_blocks = []
    if len(buf) >= doff:
        opts = buf[20:doff]
        i = 0
        while i < len(opts):
            kind = opts[i]
            if kind == 0:
                break
            if kind == 1:
                i += 1
                continue
            if i + 1 >= len(opts):
                break
            olen = opts[i + 1]
            if olen < 2 or i + olen > len(opts):
                break
            if kind == 3 and olen == 3:
                wscale = opts[i + 2]
            elif kind == 5 and olen >= 10:
                body = opts[i + 2:i + olen]
                for j in range(0, len(body) - 7, 8):
                    sack_blocks.append(struct.unpack("!II", body[j:j + 8]))
            i += olen
    truncated = len(buf) < doff
    plen = None
    if payload_available is not None and not truncated:
        plen = max(payload_available - doff, 0)
    return (sport, dport,
            TcpBits(seq, ack, flags, window, plen, wscale, sack_blocks)), truncated


def decode(pkt):
    etype, buf, vlan = strip_link(pkt.data, pkt.linktype)
    if etype is None or buf is None:
        return None
    d = Decoded()
    d.vlan = vlan
    if etype == ETH_IPV4:
        if len(buf) < 20:
            return None
        ihl = (buf[0] & 0x0F) * 4
        if ihl < 20:
            return None
        total_len = struct.unpack("!H", buf[2:4])[0]
        proto = buf[9]
        frag = struct.unpack("!H", buf[6:8])[0]
        d.src = str(ipaddress.IPv4Address(buf[12:16]))
        d.dst = str(ipaddress.IPv4Address(buf[16:20]))
        d.proto = proto
        d.ip_len = total_len if total_len else None
        # The bytes of L4 that WOULD be there if the capture were complete.
        l4_claimed = (total_len - ihl) if total_len >= ihl else None
        l4_present = len(buf) - ihl
        if l4_present < 0:
            d.l4_truncated = True
            return d
        l4 = buf[ihl:]
        first_frag = (frag & 0x1FFF) == 0
    elif etype == ETH_IPV6:
        if len(buf) < 40:
            return None
        plen = struct.unpack("!H", buf[4:6])[0]
        nxt = buf[6]
        d.src = str(ipaddress.IPv6Address(buf[8:24]))
        d.dst = str(ipaddress.IPv6Address(buf[24:40]))
        off = 40
        hops = 0
        while nxt in (0, 43, 60) and hops < 4:
            if len(buf) < off + 8:
                d.l4_truncated = True
                d.proto = nxt
                return d
            ext_len = (buf[off + 1] + 1) * 8
            nxt = buf[off]
            off += ext_len
            hops += 1
        if nxt == 44:  # fragment header
            d.proto = nxt
            d.ip_len = 40 + plen
            return d
        d.proto = nxt
        d.ip_len = 40 + plen if plen else None
        l4_claimed = plen if plen else None
        l4 = buf[off:]
        first_frag = True
    else:
        return None

    if d.proto == 6 and first_frag:
        parsed, trunc = parse_tcp(l4, l4_claimed)
        d.l4_truncated = trunc
        if parsed:
            d.sport, d.dport, d.tcp = parsed
    elif d.proto == 17 and first_frag:
        if len(l4) >= 8:
            d.sport, d.dport = struct.unpack("!HH", l4[:4])
            ulen = struct.unpack("!H", l4[4:6])[0]
            d.payload_len = max(ulen - 8, 0) if ulen >= 8 else None
        else:
            d.l4_truncated = True
    elif not first_frag:
        d.l4_truncated = True
    return d


# ---------------------------------------------------------------------------
# Flow accumulation
# ---------------------------------------------------------------------------

FIN, SYN, RST, PSH, ACK = 0x01, 0x02, 0x04, 0x08, 0x10


class Dir:
    __slots__ = ("packets", "bytes", "payload_bytes", "retrans", "retrans_known",
                 "zero_window", "dup_acks", "max_seq", "seen_ranges", "syn", "synack",
                 "fin", "rst", "last_ack", "ack_run", "min_window", "wscale",
                 "first_ts", "last_ts", "keepalives", "syn_retries", "l4_seen")

    def __init__(self):
        self.packets = 0
        self.bytes = 0
        self.payload_bytes = 0
        self.retrans = 0
        self.retrans_known = True
        self.zero_window = 0
        self.dup_acks = 0
        self.max_seq = None
        self.seen_ranges = set()
        self.syn = self.synack = self.fin = self.rst = False
        self.last_ack = None
        self.ack_run = 0
        self.min_window = None
        self.wscale = None
        self.first_ts = None
        self.last_ts = None
        self.keepalives = 0
        self.syn_retries = 0
        self.l4_seen = 0


class Flow:
    __slots__ = ("key", "proto", "a", "b", "fwd", "rev", "first_ts", "last_ts",
                 "truncated_l4", "icmp_types", "vlan")

    def __init__(self, key, proto, a, b):
        self.key = key
        self.proto = proto
        self.a = a  # (ip, port) -- the side that sent the first packet we saw
        self.b = b
        self.fwd = Dir()
        self.rev = Dir()
        self.first_ts = None
        self.last_ts = None
        self.truncated_l4 = 0
        self.icmp_types = set()
        self.vlan = None

    @property
    def packets(self):
        return self.fwd.packets + self.rev.packets

    @property
    def bytes(self):
        return self.fwd.bytes + self.rev.bytes

    @property
    def retrans(self):
        # Unknown the moment ANY direction that carried packets had its TCP
        # header cut off. "0 retransmissions" and "I could not look" are
        # opposite findings; this returns None and the report prints a dash.
        for d in (self.fwd, self.rev):
            if d.packets and not d.retrans_known:
                return None
        if self.proto != 6:
            return None
        return self.fwd.retrans + self.rev.retrans

    @property
    def zwin(self):
        for d in (self.fwd, self.rev):
            if d.packets and not d.retrans_known:
                return None
        if self.proto != 6:
            return None
        return self.fwd.zero_window + self.rev.zero_window

    @property
    def headers_readable(self):
        return all(d.l4_seen for d in (self.fwd, self.rev) if d.packets)

    @property
    def syn_retries(self):
        return self.fwd.syn_retries + self.rev.syn_retries

    @property
    def duration(self):
        if self.first_ts is None or self.last_ts is None:
            return None
        return self.last_ts - self.first_ts


def flow_key(d):
    pa = d.sport if d.sport is not None else -1
    pb = d.dport if d.dport is not None else -1
    one = (d.src, pa)
    two = (d.dst, pb)
    if one <= two:
        return (d.proto, one, two), True
    return (d.proto, two, one), False


def accumulate(packets, meta, want):
    flows = {}
    stats = {
        "total": len(packets), "decoded": 0, "undecodable": 0, "non_ip": 0,
        "truncated_caplen": 0, "at_snaplen": 0, "no_timestamp": 0,
        "first_ts": None, "last_ts": None, "l4_truncated": 0,
        "proto_bytes": {}, "proto_packets": {}, "fragments": 0,
    }
    snaplen = meta.get("snaplen") or 0
    for pkt in packets:
        if pkt.ts is None:
            stats["no_timestamp"] += 1
        else:
            if stats["first_ts"] is None or pkt.ts < stats["first_ts"]:
                stats["first_ts"] = pkt.ts
            if stats["last_ts"] is None or pkt.ts > stats["last_ts"]:
                stats["last_ts"] = pkt.ts
        if pkt.origlen and pkt.caplen < pkt.origlen:
            stats["truncated_caplen"] += 1
        if snaplen and pkt.caplen >= snaplen:
            stats["at_snaplen"] += 1
        d = decode(pkt)
        if d is None or d.src is None:
            stats["non_ip"] += 1
            continue
        stats["decoded"] += 1
        name = PROTO_NAMES.get(d.proto, f"IP-{d.proto}")
        wire = pkt.origlen or pkt.caplen
        stats["proto_bytes"][name] = stats["proto_bytes"].get(name, 0) + wire
        stats["proto_packets"][name] = stats["proto_packets"].get(name, 0) + 1
        if d.l4_truncated:
            stats["l4_truncated"] += 1
        if want and not want(d):
            continue
        key, forward = flow_key(d)
        fl = flows.get(key)
        if fl is None:
            fl = Flow(key, d.proto, key[1], key[2])
            fl.vlan = d.vlan
            flows[key] = fl
        side = fl.fwd if forward else fl.rev
        other = fl.rev if forward else fl.fwd
        side.packets += 1
        side.bytes += wire
        if pkt.ts is not None:
            if fl.first_ts is None or pkt.ts < fl.first_ts:
                fl.first_ts = pkt.ts
            if fl.last_ts is None or pkt.ts > fl.last_ts:
                fl.last_ts = pkt.ts
            if side.first_ts is None:
                side.first_ts = pkt.ts
            side.last_ts = pkt.ts
        if d.l4_truncated:
            fl.truncated_l4 += 1
            side.retrans_known = False
            continue
        if d.proto == 6 and d.tcp:
            side.l4_seen += 1
            tcp_update(side, other, d.tcp)
        elif d.proto == 17 and d.payload_len is not None:
            side.l4_seen += 1
            side.payload_bytes += d.payload_len
        elif d.proto in (1, 58):
            if len(pkt.data) >= 1:
                fl.icmp_types.add(d.proto)
    return flows, stats


def tcp_update(side, other, t):
    f = t.flags
    if f & SYN and not (f & ACK):
        side.syn = True
    if (f & SYN) and (f & ACK):
        side.synack = True
    if f & FIN:
        side.fin = True
    if f & RST:
        side.rst = True
    if t.wscale_opt is not None:
        side.wscale = t.wscale_opt
    plen = t.plen or 0
    side.payload_bytes += plen

    # Zero window: the RECEIVER telling the sender to stop. It is a statement
    # by this host's own stack, which is why it is the strongest evidence in
    # the whole summary.
    if not (f & RST):
        scaled = t.window
        if side.wscale is not None and not (f & SYN):
            scaled = t.window << side.wscale
        if side.min_window is None or scaled < side.min_window:
            side.min_window = scaled
        if t.window == 0 and not (f & (SYN | FIN)):
            side.zero_window += 1

    # Retransmission: a segment that occupies sequence space we have already
    # seen from this direction. Pure ACKs carry no sequence space and are
    # never retransmissions.
    if f & SYN:
        # A repeated SYN or SYN-ACK is a handshake retry, not a data
        # retransmission. Counting it as one sends the reader hunting for
        # packet loss when the real finding is "nobody answered".
        side.syn_retries += 1
        if side.syn_retries > 1:
            return
    consumes = plen > 0 or (f & FIN)
    if consumes:
        span = plen + (1 if f & SYN else 0) + (1 if f & FIN else 0)
        end = (t.seq + span) & 0xFFFFFFFF
        if side.max_seq is not None and _seq_le(end, side.max_seq) and (t.seq, end) in side.seen_ranges:
            side.retrans += 1
        elif side.max_seq is not None and _seq_lt(t.seq, side.max_seq):
            side.retrans += 1
        side.seen_ranges.add((t.seq, end))
        if side.max_seq is None or _seq_lt(side.max_seq, end):
            side.max_seq = end
    elif f & ACK and plen == 0 and not (f & (SYN | FIN | RST)):
        # TCP keepalive: one byte below the expected sequence number, or a
        # zero-length segment at seq-1. Counting it as a retransmission is a
        # classic false positive.
        if side.max_seq is not None and t.seq == ((side.max_seq - 1) & 0xFFFFFFFF):
            side.keepalives += 1
        if side.last_ack is not None and t.ack == side.last_ack:
            side.ack_run += 1
            if side.ack_run >= 2:
                side.dup_acks += 1
        else:
            side.ack_run = 0
        side.last_ack = t.ack
    if f & ACK and plen == 0 and side.last_ack is None:
        side.last_ack = t.ack


def _seq_lt(a, b):
    return ((a - b) & 0xFFFFFFFF) > 0x80000000


def _seq_le(a, b):
    return a == b or _seq_lt(a, b)


# ---------------------------------------------------------------------------
# Judgement.  This is the part that is not "parse a pcap".
# ---------------------------------------------------------------------------

SEV_ORDER = {"HIGH": 0, "MEDIUM": 1, "LOW": 2, "INFO": 3}


def classify(fl, capture_is_partial, capture_end=None):
    """Returns (severity, label, explanation) or None when nothing stands out.

    Every branch answers the same question: WHAT DOES THIS CAPTURE ACTUALLY
    PROVE? A check that could not run returns an INFO saying so, never a
    finding.
    """
    if fl.proto != 6:
        return None
    if not fl.headers_readable:
        return ("INFO", "TCP headers not captured",
                "the snaplen cut this flow's TCP headers, so handshake, "
                "retransmissions and window could not be read. Nothing about "
                "this flow is being claimed - recapture with -s 0")

    a, b = fl.fwd, fl.rev
    unidirectional = a.packets == 0 or b.packets == 0
    initiator = a if a.syn else (b if b.syn else None)
    saw_syn = a.syn or b.syn
    saw_synack = a.synack or b.synack

    # Did the capture simply stop? You cannot conclude "no reply" from a flow
    # whose last packet is in the last second of the file.
    ends_at_edge = (capture_end is not None and fl.last_ts is not None
                    and (capture_end - fl.last_ts) < 1.0)

    if saw_syn and not saw_synack:
        if ends_at_edge and fl.syn_retries <= 1:
            return ("INFO", "SYN at the very end of the capture",
                    "no reply, but the capture stops here - there was no time "
                    "for one. Recapture with a longer window before concluding")
        extra = " after %d attempt(s)" % fl.syn_retries if fl.syn_retries > 1 else ""
        if unidirectional:
            return ("HIGH", "SYN, no reply seen" + extra,
                    "nothing came back - but only one direction is in this "
                    "capture, so this may be the capture point and not the host")
        return ("HIGH", "SYN unanswered" + extra,
                "the destination never sent SYN-ACK: filtered in the path, or "
                "nothing listening on that port. Look at the FAR END")
    if saw_syn and saw_synack:
        completed = initiator is not None and initiator.packets > 1
        if not completed and not ends_at_edge:
            return ("HIGH", "half-open: SYN-ACK never acknowledged",
                    "the server answered and the client never completed. Look "
                    "at the CLIENT or at the return path, not at the server")

    zw = fl.zwin or 0
    rt = fl.retrans

    if zw > 0:
        who = "both ends" if (a.zero_window and b.zero_window) else (
            fmt_ep(fl.a if a.zero_window else fl.b))
        tail = ""
        if rt:
            tail = (" The %d resend(s) here are window probes and stalled "
                    "data, not evidence of loss." % rt)
        return ("HIGH", "zero window advertised by %s (%dx)" % (who, zw),
                "the receiver's own stack said its buffer is full. The "
                "application is not reading fast enough - this is the strongest "
                "evidence in the summary that the NETWORK is innocent." + tail)

    if rt is None:
        return ("INFO", "retransmissions unknown",
                "TCP headers were cut in this flow, so retransmissions could "
                "not be counted. Shown as a dash, never as zero")

    data_pkts = max(a.packets + b.packets, 1)
    if rt > 0:
        pct = 100.0 * rt / data_pkts
        if pct >= 2.0:
            return ("MEDIUM", "%d retransmission(s), %.1f%% of packets" % (rt, pct),
                    "segments were resent. One capture point cannot say WHERE "
                    "they were lost, and the heaviest talker is as often the "
                    "VICTIM of congestion as its cause. Capture at both ends "
                    "before blaming this host")
        return ("LOW", "%d retransmission(s), %.1f%% of packets" % (rt, pct),
                "below the level worth chasing on a WAN path on its own")

    if a.rst or b.rst:
        side = fmt_ep(fl.a if a.rst else fl.b)
        return ("MEDIUM", "reset by " + side,
                "torn down with RST rather than FIN. A RST whose TTL differs "
                "from the rest of the flow was forged by a middlebox - this "
                "script does not compare TTLs, tshark does")

    if unidirectional and fl.packets > 2:
        return ("MEDIUM", "one direction only",
                "no packets came back. This almost never means one-way traffic; "
                "it means the capture point or the capture filter missed the "
                "return path")

    if not (a.fin or b.fin or a.rst or b.rst) and not capture_is_partial:
        return ("LOW", "no close seen",
                "neither FIN nor RST. The flow was still open when the capture "
                "stopped, or the close is outside the capture window")
    return None


def fmt_ep(ep):
    ip, port = ep
    if ":" in ip and port >= 0:
        return f"[{ip}]:{port}"
    if port < 0:
        return ip
    return f"{ip}:{port}"


def service_of(fl):
    ports = [p for p in (fl.a[1], fl.b[1]) if p is not None and p >= 0]
    for p in sorted(ports):
        if p in WELL_KNOWN:
            return WELL_KNOWN[p]
    return None


# ---------------------------------------------------------------------------
# Presentation
# ---------------------------------------------------------------------------

def human_bytes(n):
    if n is None:
        return DASH
    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.0
    return DASH


def human_dur(s):
    if s is None:
        return DASH
    if s < 1:
        return f"{s * 1000:.0f}ms"
    if s < 60:
        return f"{s:.2f}s"
    m, sec = divmod(s, 60)
    if m < 60:
        return f"{int(m)}m{sec:04.1f}s"
    h, m = divmod(int(m), 60)
    return f"{h}h{m:02d}m"


def iso(ts):
    if ts is None:
        return DASH
    return datetime.fromtimestamp(ts, timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] + "Z"


def capture_quality(stats, meta, flows):
    """Facts about the CAPTURE, not about the network. Printed before any
    verdict, because they decide how much the verdicts are worth."""
    notes = []
    snaplen = meta.get("snaplen") or 0
    if stats["truncated_caplen"]:
        notes.append(
            f"{stats['truncated_caplen']} packet(s) were cut short by the capture "
            f"snaplen ({snaplen if snaplen else DASH} bytes). Payload byte counts "
            "below come from the IP header length field, not from what was stored.")
    if stats["l4_truncated"]:
        notes.append(
            f"{stats['l4_truncated']} packet(s) had their TCP/UDP header cut off. "
            "Retransmissions and window sizes cannot be counted for those flows "
            "and are shown as '-', never as zero.")
    uni = [f for f in flows.values() if f.fwd.packets == 0 or f.rev.packets == 0]
    if uni:
        notes.append(
            f"{len(uni)} of {len(flows)} flow(s) have traffic in ONE direction only. "
            "That is almost never one-way traffic - it means the capture point or "
            "the capture filter did not see the return path.")
    tcp_flows = [f for f in flows.values() if f.proto == 6]
    readable = [f for f in tcp_flows if f.headers_readable]
    midstream = [f for f in readable if not (f.fwd.syn or f.rev.syn)]
    if midstream:
        notes.append(
            f"{len(midstream)} of {len(readable)} readable TCP flow(s) have no SYN: "
            "capture started after they did. Their byte totals are partial and "
            "their handshake cannot be judged.")
    noports = [f for f in flows.values() if f.a[1] < 0 or f.b[1] < 0]
    if noports:
        notes.append(
            f"{len(noports)} flow(s) have no port numbers, because the L4 header "
            "was not captured. Everything between the same pair of addresses is "
            "merged into one row there - that row is an address pair, not a "
            "conversation.")
    if meta.get("truncated_file"):
        notes.append(
            "The capture FILE itself ends mid-record - it was cut off while being "
            "written, or copied while still open. Everything after that point is missing.")
    if stats["no_timestamp"]:
        notes.append(
            f"{stats['no_timestamp']} packet(s) carry no timestamp (pcapng Simple "
            "Packet Blocks). They are counted but cannot be placed in time.")
    if stats["non_ip"]:
        notes.append(
            f"{stats['non_ip']} packet(s) were not IPv4/IPv6 or could not be "
            "decoded (ARP, STP, LLDP, unsupported link type). They are excluded "
            "from flows but counted in the total.")
    return notes


def print_report(flows, stats, meta, args, path):
    out = []
    w = out.append
    rule = "=" * 78

    w(rule)
    w(f"CAPTURE  {os.path.basename(path)}")
    w(rule)
    span = None
    if stats["first_ts"] is not None and stats["last_ts"] is not None:
        span = stats["last_ts"] - stats["first_ts"]
    w(f"  format         {meta.get('format', DASH)}   "
      f"snaplen {meta.get('snaplen') or DASH}   "
      f"linktype {','.join(str(x) for x in sorted(meta.get('linktypes') or [])) or DASH}")
    w(f"  packets        {stats['total']}  ({stats['decoded']} IP, {stats['non_ip']} other)")
    w(f"  first packet   {iso(stats['first_ts'])}")
    w(f"  last packet    {iso(stats['last_ts'])}")
    w(f"  span           {human_dur(span)}")
    w(f"  flows          {len(flows)}")
    w("")
    w("  All times are UTC and come from the CAPTURE POINT's clock, not from")
    w("  the sender. Two captures taken on two boxes do not share a clock.")
    w("")

    notes = capture_quality(stats, meta, flows)
    w(rule)
    w("CAPTURE QUALITY")
    w(rule)
    if notes:
        for n in notes:
            w("  [!] " + wrap(n, 72, "      "))
    else:
        w("  No truncation, no one-sided flows, every TCP flow has its handshake.")
        w("  This capture supports the conclusions below.")
    w("")

    if not flows:
        w("  No flows matched. Nothing to summarise.")
        print("\n".join(out))
        return

    ranked = []
    for fl in flows.values():
        verdict = classify(fl, bool(notes), stats["last_ts"])
        ranked.append((fl, verdict))

    anomalies = [(f, v) for f, v in ranked if v and v[0] in ("HIGH", "MEDIUM", "LOW")]
    anomalies.sort(key=lambda t: (SEV_ORDER[t[1][0]], -t[0].bytes))

    w(rule)
    w("FLOWS WORTH LOOKING AT")
    w(rule)
    unjudged = [f for f, v in ranked if v and v[0] == "INFO"]
    if not anomalies:
        if len(unjudged) == len([f for f in flows.values() if f.proto == 6]) and unjudged:
            w(f"  Nothing is being claimed: {len(unjudged)} TCP flow(s) could not be")
            w("  judged at all - see CAPTURE QUALITY. This is NOT a clean bill of")
            w("  health, it is an unreadable capture.")
        else:
            w("  Nothing stood out in the flows that could be judged. Handshakes")
            w("  completed, nothing retransmitted, no zero window advertised.")
            if unjudged:
                w(f"  {len(unjudged)} flow(s) could not be checked - see CAPTURE QUALITY.")
    else:
        for fl, (sev, label, why) in anomalies[:args.top]:
            svc = service_of(fl)
            w(f"  [{sev:<6}] {fmt_ep(fl.a)} <-> {fmt_ep(fl.b)}"
              f"  {PROTO_NAMES.get(fl.proto, fl.proto)}{'/' + svc if svc else ''}")
            w(f"            {label}")
            w(f"            {wrap(why, 64, '            ')}")
            w(f"            {fl.packets} pkt  {human_bytes(fl.bytes)}  "
              f"{human_dur(fl.duration)}  retrans "
              f"{fl.retrans if fl.retrans is not None else DASH}  zwin "
              f"{fl.zwin if fl.zwin is not None else DASH}")
            w("")
        if len(anomalies) > args.top:
            w(f"  ... and {len(anomalies) - args.top} more. Raise --top to see them.")
            w("")
        if unjudged:
            w(f"  {len(unjudged)} further flow(s) could not be judged at all - see")
            w("  CAPTURE QUALITY. They are not clean, they are unreadable.")
            w("")

    w("")
    w(rule)
    w("TOP FLOWS BY VOLUME")
    w(rule)
    w(table(sorted(flows.values(), key=lambda f: -f.bytes)[:args.top]))
    w("")

    timed = [f for f in flows.values() if f.duration is not None]
    w(rule)
    w("LONGEST FLOWS")
    w(rule)
    if timed:
        w(table(sorted(timed, key=lambda f: -f.duration)[:args.top]))
    else:
        w("  No flow carries timestamps on both ends.")
    w("")

    w(rule)
    w("PROTOCOL AND ENDPOINT BREAKDOWN")
    w(rule)
    if build_filter(args) is not None:
        w("  The protocol table counts the WHOLE capture. The endpoint list and")
        w("  everything above it count only the flows your filter kept.")
        w("")
    total_b = sum(stats["proto_bytes"].values()) or 1
    for name, b in sorted(stats["proto_bytes"].items(), key=lambda kv: -kv[1]):
        w(f"  {name:<8} {stats['proto_packets'][name]:>7} pkt  "
          f"{human_bytes(b):>9}  {100.0 * b / total_b:5.1f}%")
    w("")
    ep = {}
    for fl in flows.values():
        for side, d in ((fl.a, fl.fwd), (fl.b, fl.rev)):
            ep[side[0]] = ep.get(side[0], 0) + d.bytes
    w("  Busiest endpoints:")
    for ip, b in sorted(ep.items(), key=lambda kv: -kv[1])[:args.top]:
        w(f"    {ip:<40} {human_bytes(b):>9}")
    w("")

    sev_count = {}
    for _f, v in anomalies:
        sev_count[v[0]] = sev_count.get(v[0], 0) + 1
    w(rule)
    w("  " + ", ".join(f"{sev_count.get(s, 0)} {s}" for s in ("HIGH", "MEDIUM", "LOW"))
      + f" across {len(flows)} flow(s)."
      + ("  Read CAPTURE QUALITY above before acting on any of it."
         if notes else ""))
    w(rule)
    print("\n".join(out))


def table(flows):
    lines = [f"  {'FLOW':<47} {'PKT':>6} {'BYTES':>9} {'DURATION':>9} {'RETR':>5} {'ZWIN':>5}"]
    for fl in flows:
        name = f"{fmt_ep(fl.a)} <-> {fmt_ep(fl.b)}"
        if len(name) > 47:
            name = name[:44] + "..."
        lines.append(
            f"  {name:<47} {fl.packets:>6} {human_bytes(fl.bytes):>9} "
            f"{human_dur(fl.duration):>9} "
            f"{(str(fl.retrans) if fl.retrans is not None else DASH):>5} "
            f"{(str(fl.zwin) if fl.zwin is not None else DASH):>5}")
    return "\n".join(lines)


def wrap(text, width, indent):
    words = text.split()
    lines, cur = [], ""
    for word in words:
        if cur and len(cur) + 1 + len(word) > width:
            lines.append(cur)
            cur = word
        else:
            cur = f"{cur} {word}".strip()
    if cur:
        lines.append(cur)
    return ("\n" + indent).join(lines)


def to_json(flows, stats, meta, notes):
    return {
        "generated_utc": datetime.now(timezone.utc).isoformat(),
        "capture": {
            "format": meta.get("format"),
            "snaplen": meta.get("snaplen") or None,
            "packets": stats["total"],
            "ip_packets": stats["decoded"],
            "first_packet_utc": iso(stats["first_ts"]) if stats["first_ts"] else None,
            "last_packet_utc": iso(stats["last_ts"]) if stats["last_ts"] else None,
            "file_truncated": bool(meta.get("truncated_file")),
        },
        "capture_quality": notes,
        "flows": [
            {
                "proto": PROTO_NAMES.get(f.proto, f.proto),
                "a": fmt_ep(f.a), "b": fmt_ep(f.b),
                "packets": f.packets, "bytes": f.bytes,
                "duration_s": round(f.duration, 6) if f.duration is not None else None,
                "retransmissions": f.retrans,
                "zero_window": f.zwin,
                "verdict": (lambda v: {"severity": v[0], "label": v[1], "detail": v[2]}
                            if v else None)(classify(f, bool(notes), stats["last_ts"])),
            }
            for f in sorted(flows.values(), key=lambda x: -x.bytes)
        ],
    }


def build_filter(args):
    hosts = set(args.host or [])
    ports = set(args.port or [])
    proto = args.proto.lower() if args.proto else None
    if not hosts and not ports and not proto:
        return None
    pmap = {"tcp": 6, "udp": 17, "icmp": 1, "icmpv6": 58}

    def want(d):
        if hosts and d.src not in hosts and d.dst not in hosts:
            return False
        if ports and d.sport not in ports and d.dport not in ports:
            return False
        if proto and d.proto != pmap.get(proto, -1):
            return False
        return True
    return want


def main():
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    p = argparse.ArgumentParser(
        description="Summarise a packet capture into the flows worth looking at.")
    p.add_argument("capture", help="path to a .pcap or .pcapng file")
    p.add_argument("--top", type=int, default=10, help="rows per table (default 10)")
    p.add_argument("--host", action="append", help="only flows involving this IP (repeatable)")
    p.add_argument("--port", action="append", type=int, help="only flows on this port (repeatable)")
    p.add_argument("--proto", choices=["tcp", "udp", "icmp", "icmpv6"], help="only this protocol")
    p.add_argument("--min-bytes", type=int, default=0, help="ignore flows below this many bytes")
    p.add_argument("--json-out", help="write the full summary to a JSON file")
    p.add_argument("--fail-on", choices=["high", "medium", "low", "never"], default="high",
                   help="lowest severity that makes the script exit 1 (default high)")
    args = p.parse_args()

    if args.top < 1:
        print("--top must be at least 1", file=sys.stderr)
        return 2

    try:
        packets, meta = read_capture(args.capture)
    except CaptureError as exc:
        print(f"[!] {exc}", file=sys.stderr)
        return 2
    except MemoryError:
        print("[!] the capture does not fit in memory. Split it first:\n"
              "    editcap -c 200000 big.pcap part.pcap", file=sys.stderr)
        return 2

    if not packets:
        print("[!] the capture contains no packets.", file=sys.stderr)
        return 2

    flows, stats = accumulate(packets, meta, build_filter(args))
    if args.min_bytes:
        flows = {k: v for k, v in flows.items() if v.bytes >= args.min_bytes}

    print_report(flows, stats, meta, args, args.capture)

    notes = capture_quality(stats, meta, flows)
    if args.json_out:
        try:
            with open(args.json_out, "w", encoding="utf-8") as fh:
                json.dump(to_json(flows, stats, meta, notes), fh, indent=2)
            print(f"\n[+] JSON written to {args.json_out}")
        except OSError as exc:
            print(f"[!] could not write {args.json_out}: {exc}", file=sys.stderr)
            return 2

    if args.fail_on == "never":
        return 0
    threshold = SEV_ORDER[args.fail_on.upper()]
    for fl in flows.values():
        v = classify(fl, bool(notes), stats["last_ts"])
        if v and SEV_ORDER[v[0]] <= threshold:
            return 1
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\n[!] interrupted", file=sys.stderr)
        sys.exit(130)