udpscope/tools/capture_to_pcap.py

#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: 2026 Martino Ferrari

"""Converts a udpscope packet capture to a classic pcap file.

$ tools/capture_to_pcap.py ~/.cache/udpscope/192.168.1.40-5000/capture.bin out.pcap
1000 slots, 412 stored, 0 torn, 0 truncated -> 412 packets, 26.1 MiB
$ wireshark out.pcap

The capture is udpscope's ring of the last N datagrams exactly as recv() returned
them (see docs/superpowers/specs/2026-10-09-udp-packet-capture-design.md): a 64-byte
header, then fixed-size slots whose trailing `seq` word is written last, so a slot
carrying a non-zero `seq` is complete even after a crash. This script walks the slots,
orders them by `seq`, skips the incomplete ones, and wraps every record in a
synthesised Ethernet/IPv4/UDP header built from the capture header's peer address and
ports -- pure standard library, no numpy, so it runs wherever the self-check runs.

The counts it prints to stderr are a diagnostic in their own right: `torn` or
`truncated` above zero is a finding about the capture before anything about the stream.
"""

import struct
import sys

MAGIC = b"UDPSCAP\x00"
HEADER_BYTES = 64
RECORD_FIXED = 16  # t_arrival f64, len u32, orig_len u32
SEQ_BYTES = 8

PCAP_MAGIC = 0xA1B23C4D  # microsecond-resolution, writing the non-nanosecond classic
LINKTYPE_ETHERNET = 1


def fail(msg):
    print(f"capture_to_pcap: {msg}", file=sys.stderr)
    sys.exit(2)


def main(argv):
    if len(argv) != 3:
        fail(f"usage: {argv[0]} capture.bin out.pcap")
    src, dst = argv[1], argv[2]

    with open(src, "rb") as f:
        blob = f.read()

    if len(blob) < HEADER_BYTES:
        fail(f"{src}: {len(blob)} bytes, smaller than a capture header")
    magic, version, slot_count, slot_bytes, cursor = struct.unpack_from("<8sIIII", blob, 0)
    (t0,) = struct.unpack_from("<d", blob, 24)
    peer_addr, peer_port, local_port = struct.unpack_from("<IHH", blob, 32)
    if magic != MAGIC:
        fail(f"{src}: bad magic {magic!r}")
    if version != 1:
        fail(f"{src}: unsupported capture version {version}")
    payload_bytes = slot_bytes - RECORD_FIXED - SEQ_BYTES
    if slot_count == 0 or payload_bytes <= 0:
        fail(f"{src}: implausible geometry: {slot_count} slots of {slot_bytes} bytes")
    if len(blob) < HEADER_BYTES + slot_count * slot_bytes:
        fail(f"{src}: truncated file: {len(blob)} bytes for {slot_count} slots")

    records = []
    stored = torn = truncated = empty = 0
    payload_sum = 0
    for i in range(slot_count):
        off = HEADER_BYTES + i * slot_bytes
        slot = blob[off : off + slot_bytes]
        (t_arrival, length, orig_len) = struct.unpack_from("<dII", slot, 0)
        (seq,) = struct.unpack_from("<Q", slot, slot_bytes - SEQ_BYTES)
        if seq == 0 and length == 0 and orig_len == 0:
            empty += 1
            continue
        if seq == 0:
            torn += 1
            continue
        length = min(length, payload_bytes)
        if orig_len > payload_bytes:
            truncated += 1
        records.append((seq, t_arrival, slot[RECORD_FIXED : RECORD_FIXED + length]))
        payload_sum += length
        stored += 1

    records.sort(key=lambda r: r[0])

    ip = bytes([(peer_addr >> 24) & 0xFF, (peer_addr >> 16) & 0xFF, (peer_addr >> 8) & 0xFF,
                peer_addr & 0xFF])
    zero_ip = b"\x00\x00\x00\x00"

    with open(dst, "wb") as out:
        out.write(struct.pack("<IHHiIII", PCAP_MAGIC, 0, 0, 0, 0, 0, LINKTYPE_ETHERNET))
        for seq, t_arrival, payload in records:
            # The client stamps arrivals with CLOCK_REALTIME, so a `t_arrival` is
            # already an epoch; `t0_realtime` is the epoch a *relative* stamp would be
            # measured from. Take whichever the file carries.
            ts = t_arrival if t_arrival > 1e6 else t0 + t_arrival
            sec = int(ts)
            usec = int(round((ts - sec) * 1e6))
            if usec >= 1_000_000:
                sec, usec = sec + 1, usec - 1_000_000

            udp_len = 8 + len(payload)
            udp = struct.pack(">HHHH", peer_port, local_port, udp_len, 0) + payload

            total_len = 20 + udp_len
            ip_hdr = struct.pack(
                ">BBHHHBBH4s4s",
                0x45,          # version 4, IHL 5
                0,             # DSCP
                total_len,
                seq & 0xFFFF,  # id: the capture's own ordinal, which is what a
                               # duplicate datagram would show as a repeated id
                0,             # flags/offset
                64, 17,        # TTL, proto UDP
                0,             # checksum: wireshark recomputes and says "incorrect",
                               # which is honest for a synthesised header
                zero_ip,       # source: the capture is on the receiver, so the local
                               # address is unknown to it
                ip,            # destination: the source we captured from
            )
            frame = b"\x00" * 6 + b"\x01" * 6 + b"\x08\x00" + ip_hdr + udp
            out.write(struct.pack("<IIII", sec, usec, len(frame), len(frame)))
            out.write(frame)

    mib = len(blob) / 1048576.0
    print(
        f"{slot_count} slots, {stored} stored, {torn} torn, {truncated} truncated, "
        f"{empty} empty -> {len(records)} packets, {mib:.1f} MiB, "
        f"payload {payload_sum} bytes",
        file=sys.stderr,
    )


if __name__ == "__main__":
    main(sys.argv)