pcap_parser.py

"""Streaming pcap/pcapng parser for PCAPGraph (IPv4 only, for now).

Packets are streamed one at a time with Scapy's ``PcapReader`` and folded into
a **conversation** table. A conversation is a single bidirectional exchange
between two endpoints — both directions of a TCP/UDP session collapse into one
entry — so we don't split each session into a forward flow (to :443) and a
return flow (to the ephemeral port). Memory scales with the number of unique
conversations, not with file size or packet count.

Each conversation tracks per-direction byte/packet counts and running
client→server inter-arrival statistics (so beaconing can be detected later
without storing every timestamp).
"""

from __future__ import annotations

import math
import socket
import struct
import time
from dataclasses import dataclass, field

from scapy.utils import RawPcapReader

from netutils import is_noise
import payload as payload_parser

# struct unpackers (compiled once)
_u16 = struct.Struct(">H").unpack_from

# link-layer types we parse natively
DLT_NULL = 0
DLT_EN10MB = 1
DLT_RAW1 = 12
DLT_RAW2 = 14
DLT_LINUX_SLL = 113
DLT_RAW = 101
DLT_IPV4 = 228


@dataclass
class Conversation:
    """One bidirectional exchange. ``client`` initiated; ``server`` is the
    well-known/lower-port side."""

    proto: str
    client_ip: str
    client_port: int
    server_ip: str
    server_port: int

    packets: int = 0
    bytes: int = 0
    c2s_pkts: int = 0
    c2s_bytes: int = 0
    s2c_pkts: int = 0
    s2c_bytes: int = 0

    first_ts: float = 0.0
    last_ts: float = 0.0

    syn: int = 0
    rst: int = 0
    fin: int = 0
    ack: int = 0

    # application-layer enrichment (first value seen)
    sni: str = None
    http_host: str = None
    ja3: str = None
    ja3_hash: str = None
    user_agent: str = None

    # client->server cadence (Welford-ish) + capped timestamp sample for the
    # jitter-tolerant beaconing check and the sparkline
    _prev_c2s_ts: float = 0.0
    interval_count: int = 0
    interval_sum: float = 0.0
    interval_sumsq: float = 0.0
    times: list = field(default_factory=list)

    def add(self, ts, length, is_c2s, flags):
        self.packets += 1
        self.bytes += length
        if is_c2s:
            self.c2s_pkts += 1
            self.c2s_bytes += length
            if self._prev_c2s_ts and ts >= self._prev_c2s_ts:
                gap = ts - self._prev_c2s_ts
                self.interval_count += 1
                self.interval_sum += gap
                self.interval_sumsq += gap * gap
            self._prev_c2s_ts = ts
            if len(self.times) < 512:
                self.times.append(ts)
        else:
            self.s2c_pkts += 1
            self.s2c_bytes += length

        if self.first_ts == 0.0 or ts < self.first_ts:
            self.first_ts = ts
        if ts > self.last_ts:
            self.last_ts = ts

        if flags is not None:
            if "S" in flags:
                self.syn += 1
            if "R" in flags:
                self.rst += 1
            if "F" in flags:
                self.fin += 1
            if "A" in flags:
                self.ack += 1

    # ---- compatibility aliases (graph_builder / detect read these) ----------
    @property
    def src(self):
        return self.client_ip

    @property
    def dst(self):
        return self.server_ip

    @property
    def dst_port(self):
        return self.server_port

    @property
    def bytes_out(self):  # client -> server
        return self.c2s_bytes

    @property
    def bytes_in(self):   # server -> client
        return self.s2c_bytes

    @property
    def duration(self):
        return max(0.0, self.last_ts - self.first_ts)

    @property
    def mean_interval(self):
        return self.interval_sum / self.interval_count if self.interval_count else 0.0

    @property
    def interval_stddev(self):
        n = self.interval_count
        if n < 2:
            return 0.0
        mean = self.interval_sum / n
        var = max(0.0, (self.interval_sumsq / n) - (mean * mean))
        return math.sqrt(var)

    @property
    def jitter_cv(self):
        m = self.mean_interval
        return (self.interval_stddev / m) if m > 0 else 0.0


class ParseResult:
    def __init__(self):
        self.flows: dict[tuple, Conversation] = {}
        self.total_packets = 0
        self.parsed_packets = 0
        self.non_ipv4_packets = 0
        self.noise_packets = 0
        self.unsupported_link_packets = 0
        self.first_ts = 0.0
        self.last_ts = 0.0
        self.truncated_packets = False
        self.truncated_time = False
        self.elapsed = 0.0

        # DNS intelligence gathered from UDP/53 payloads
        self.dns_by_name = {}      # qname -> {count, qtypes, rcodes, clients, answers}
        self.dns_by_client = {}    # client -> {queries, nx, names, txt, qlen_sum}
        self.dns_by_pair = {}      # (client, resolver) -> {queries, nx, name_counts}
        self.dns_resolvers = set()
        self.ip_to_domain = {}     # answer IP -> set(domains) (for node labels)
        self.http_requests = []    # capped list of {client, host, method, path, ua}

    def record_dns(self, client, resolver, dns):
        self.dns_resolvers.add(resolver)
        c = self.dns_by_client.get(client)
        if c is None:
            c = {"queries": 0, "nx": 0, "names": set(), "txt": 0,
                 "qlen_sum": 0, "name_counts": {}, "resolvers": set()}
            self.dns_by_client[client] = c
        c["resolvers"].add(resolver)
        pair = self.dns_by_pair.get((client, resolver))
        if pair is None:
            pair = {"queries": 0, "nx": 0, "name_counts": {}}
            self.dns_by_pair[(client, resolver)] = pair
        name = dns["qname"]
        if dns["qr"] == 0:                       # query
            c["queries"] += 1
            pair["queries"] += 1
            if name:
                c["names"].add(name)
                c["qlen_sum"] += len(name)
                c["name_counts"][name] = c["name_counts"].get(name, 0) + 1
                pair["name_counts"][name] = pair["name_counts"].get(name, 0) + 1
            if dns["qtype"] == 16:               # TXT
                c["txt"] += 1
        elif dns["rcode"] == 3:                  # NXDOMAIN response
            c["nx"] += 1
            pair["nx"] += 1
        if name and len(self.dns_by_name) < 100_000:
            e = self.dns_by_name.get(name)
            if e is None:
                e = {"count": 0, "qtypes": set(), "rcodes": set(),
                     "clients": set(), "answers": set()}
                self.dns_by_name[name] = e
            e["count"] += 1
            e["qtypes"].add(dns["qtype"])
            e["rcodes"].add(dns["rcode"])
            e["clients"].add(client)
            for atype, adata in dns["answers"]:
                if atype in ("A", "AAAA"):
                    e["answers"].add(adata)
                    self.ip_to_domain.setdefault(adata, set()).add(name)

    # kept for the stats payload's existing key name
    @property
    def non_ip_packets(self):
        return self.non_ipv4_packets

    @property
    def capture_duration(self):
        return max(0.0, self.last_ts - self.first_ts)


def _classify_endpoints(src, sport, dst, dport, proto, flags):
    """Decide which side is client (initiator) and which is server.

    A pure SYN (TCP, S without A) nails it: its destination is the server.
    Otherwise the lower / well-known port is taken as the server side.
    """
    if proto == "TCP" and flags is not None and "S" in flags and "A" not in flags:
        return (src, sport), (dst, dport)  # client, server
    # heuristic: lower port == server (443 < ephemeral, 53 < ephemeral, ...)
    if dport == sport:
        return (src, sport), (dst, dport)
    if dport < sport:
        return (src, sport), (dst, dport)
    return (dst, dport), (src, sport)


def _ipv4_offset(linktype, data):
    """Byte offset where the IPv4 header starts, or None if not IPv4 / unknown
    link type. Parses just enough of the link layer to find it."""
    if linktype == DLT_EN10MB:
        if len(data) < 14:
            return None
        etype = _u16(data, 12)[0]
        if etype == 0x8100 or etype == 0x88A8:  # VLAN / QinQ tag
            if len(data) < 18:
                return None
            etype = _u16(data, 16)[0]
            return 18 if etype == 0x0800 else None
        return 14 if etype == 0x0800 else None
    if linktype == DLT_LINUX_SLL:
        if len(data) < 16:
            return None
        return 16 if _u16(data, 14)[0] == 0x0800 else None
    if linktype == DLT_NULL:
        if len(data) < 4:
            return None
        return 4 if data[:4] in (b"\x02\x00\x00\x00", b"\x00\x00\x00\x02") else None
    if linktype in (DLT_RAW, DLT_RAW1, DLT_RAW2, DLT_IPV4):
        return 0 if data and (data[0] >> 4) == 4 else None
    return -1  # unsupported link type


def _ts_of(meta):
    """Seconds (float) from a RawPcap(Ng) metadata record."""
    tsresol = getattr(meta, "tsresol", None)
    if tsresol:  # pcapng
        raw = (getattr(meta, "tshigh", 0) << 32) | getattr(meta, "tslow", 0)
        return raw / tsresol
    sec = getattr(meta, "sec", 0)
    usec = getattr(meta, "usec", 0)
    return sec + usec / 1_000_000.0


def parse(path, max_packets=5_000_000, timeout=120.0):
    """Stream-parse a capture into a conversation table using a lightweight
    manual header parser (no per-packet Scapy dissection — much faster)."""
    result = ParseResult()
    started = time.monotonic()
    flows = result.flows
    default_linktype = None  # classic pcap exposes one linktype on the reader

    reader = RawPcapReader(path)
    default_linktype = getattr(reader, "linktype", None)

    for data, meta in reader:
        result.total_packets += 1

        if result.total_packets % 50_000 == 0 and \
                time.monotonic() - started > timeout:
            result.truncated_time = True
            break
        if result.parsed_packets >= max_packets:
            result.truncated_packets = True
            break

        linktype = getattr(meta, "linktype", None)
        if linktype is None:
            linktype = default_linktype

        off = _ipv4_offset(linktype, data)
        if off is None:
            result.non_ipv4_packets += 1     # IPv6, ARP, non-IPv4
            continue
        if off == -1:
            result.unsupported_link_packets += 1
            continue
        if len(data) < off + 20:
            result.non_ipv4_packets += 1
            continue

        ihl = (data[off] & 0x0F) * 4
        proto_num = data[off + 9]
        src = socket.inet_ntoa(data[off + 12:off + 16])
        dst = socket.inet_ntoa(data[off + 16:off + 20])

        if is_noise(src) or is_noise(dst):
            result.noise_packets += 1        # multicast / broadcast / 0.0.0.0
            continue

        length = getattr(meta, "wirelen", None) or len(data)
        ts = _ts_of(meta)

        l4 = off + ihl
        proto, sport, dport, flags = "OTHER", 0, 0, None
        if proto_num == 6 and len(data) >= l4 + 14:        # TCP
            proto = "TCP"
            sport, dport = _u16(data, l4)[0], _u16(data, l4 + 2)[0]
            fb = data[l4 + 13]
            flags = ("F" if fb & 0x01 else "") + ("S" if fb & 0x02 else "") + \
                    ("R" if fb & 0x04 else "") + ("P" if fb & 0x08 else "") + \
                    ("A" if fb & 0x10 else "")
        elif proto_num == 17 and len(data) >= l4 + 4:      # UDP
            proto = "UDP"
            sport, dport = _u16(data, l4)[0], _u16(data, l4 + 2)[0]
        elif proto_num == 1:                               # ICMP
            proto = "ICMP"

        client, server = _classify_endpoints(src, sport, dst, dport, proto, flags)
        ep_a, ep_b = sorted((client, server))
        key = (proto, ep_a, ep_b)

        conv = flows.get(key)
        if conv is None:
            conv = Conversation(
                proto=proto,
                client_ip=client[0], client_port=client[1],
                server_ip=server[0], server_port=server[1],
            )
            flows[key] = conv

        is_c2s = (src == conv.client_ip and sport == conv.client_port)
        conv.add(ts, length, is_c2s, flags)

        # ---- application-layer payload inspection (DNS / TLS / HTTP) -------
        if proto == "UDP" and (sport == 53 or dport == 53):
            poff = l4 + 8
            if len(data) > poff:
                dns = payload_parser.parse_dns(data[poff:])
                if dns:
                    if dport == 53:
                        result.record_dns(src, dst, dns)
                    else:
                        result.record_dns(dst, src, dns)
        elif proto == "TCP" and (conv.ja3 is None or conv.http_host is None) \
                and len(data) > l4 + 13:
            thl = (data[l4 + 12] >> 4) * 4
            poff = l4 + thl
            if len(data) > poff:
                pl = data[poff:]
                if pl[:1] == b"\x16" and conv.ja3 is None:
                    th = payload_parser.parse_tls_client_hello(pl)
                    if th:
                        conv.sni = conv.sni or th["sni"]
                        conv.ja3, conv.ja3_hash = th["ja3"], th["ja3_hash"]
                elif conv.http_host is None:
                    ph = payload_parser.parse_http_request(pl)
                    if ph:
                        conv.http_host = ph["host"]
                        conv.user_agent = ph["user_agent"]
                        if len(result.http_requests) < 5000:
                            result.http_requests.append({"client": conv.client_ip, **ph})

        result.parsed_packets += 1
        if ts:
            if result.first_ts == 0.0 or ts < result.first_ts:
                result.first_ts = ts
            if ts > result.last_ts:
                result.last_ts = ts

    try:
        reader.close()
    except Exception:
        pass
    result.elapsed = time.monotonic() - started
    return result