vt1200/lat/probe/announce_replay.py

#!/usr/bin/env python3
"""Re-send the LAT *service announcements* (message type 0xA) from a capture.

Why: the VT1200's LAT receive path handles type 0xA in the dispatcher at byte
0x11634 -> branch 0x118C0 -> LAD update (funcs 0x10660 / 0x103CA), which does
memcpy()s with announcement-derived lengths and no confirmed upper bound
(see ../notes/lat-receive-audit.md, Finding 3).  The host fully controls the
announcement, so it is a real attack surface.

In the capture where the terminal died (`filter-test2.pcap`) the host sent 91 of
these multicast announcements, one every ~10 s, for the whole session.  A replay
of just the session *data* (no announcements) did not crash the terminal, so the
announcements are the one protocol-level difference left to test.

The frames are re-sent **verbatim** (identical bytes to the original run), so the
terminal cannot tell them from latd's own.

    # just show what would go out
    sudo python3 announce_replay.py ../local/filter-test2.pcap --dry-run

    # send the whole set, at the captured ~10 s cadence
    sudo python3 announce_replay.py ../local/filter-test2.pcap --interval 10

    # keep sending until interrupted (for a long soak test)
    sudo python3 announce_replay.py ../local/filter-test2.pcap --interval 10 --forever
"""
import argparse
import os
import struct
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import extract_stream as ex          # noqa: E402
import lat_probe                     # noqa: E402

ETHERTYPE_LAT = 0x6004
LAT_MULTICAST = b"\x09\x00\x2b\x00\x00\x0f"
MSG_TYPE_SERVICE = 0xA


def load_announcements(pcap, mcast=LAT_MULTICAST, only_type=MSG_TYPE_SERVICE):
    """Return the host -> LAT-multicast frames from `pcap`, in order."""
    out = []
    for _ts, f in ex.read_pcap(pcap):
        if len(f) < 22 or struct.unpack_from(">H", f, 12)[0] != ETHERTYPE_LAT:
            continue
        if f[0:6] != mcast:
            continue
        if only_type is not None and ((f[14] >> 2) & 0x3F) != only_type:
            continue
        out.append(f)
    return out


def replay(frames, send_fn, interval=10.0, loops=1, forever=False,
           sleeper=time.sleep, log=None):
    """Send `frames` `loops` times (or until `forever` stops being signalled).

    Returns the number of frames sent.
    """
    sent = 0
    n = 0
    while forever or n < loops:
        for f in frames:
            send_fn(f)
            sent += 1
            if log:
                log(sent, f)
            if interval:
                sleeper(interval)
        n += 1
        if not frames:
            break
    return sent


def summarize(frame):
    """Human-readable one-liner for an announcement frame."""
    lat = frame[14:]
    nbr = lat[1]
    off = 8
    bits = []
    for _ in range(nbr):
        if off + 4 > len(lat):
            break
        bc, ct = lat[off + 2], lat[off + 3]
        data = lat[off + 4:off + 4 + bc]
        if bc:
            bits.append(f"ty={ct & 0xF:#x} {bytes(data)[:44]!r}")
        off += 4 + bc
    return " | ".join(bits)


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("pcap")
    ap.add_argument("-i", "--iface", default="enp197s0f3u1")
    ap.add_argument("--interval", type=float, default=10.0,
                    help="seconds between announcements (capture used ~10)")
    ap.add_argument("--loops", type=int, default=1)
    ap.add_argument("--forever", action="store_true")
    ap.add_argument("--limit", type=int, default=0,
                    help="cap the number of frames per loop (0 = all)")
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args(argv)

    frames = load_announcements(args.pcap)
    if args.limit:
        frames = frames[:args.limit]
    if not frames:
        print(f"{args.pcap}: no type-0x{A:x} announcements found "
              f"(is this a capture that includes the LAT multicast address?)")
        return 1

    uniq = len({bytes(f) for f in frames})
    print(f"{args.pcap}: {len(frames)} announcements ({uniq} distinct), "
          f"interval {args.interval}s")
    print("first:", summarize(frames[0]))
    if len(frames) > 1:
        print("last :", summarize(frames[-1]))

    if args.dry_run:
        for i, f in enumerate(frames):
            if i < 3 or i >= len(frames) - 2:
                print(f"  #{i:3} len={len(f)} {summarize(f)}")
        print(f"dry-run: would send {len(frames)} frames, "
              f"{len(frames) * args.interval:.0f}s per loop")
        return 0

    def send(f):
        lat_probe.send(f, args.iface)

    def log(n, f):
        if n % 10 == 1:
            print(f"  sent #{n}  {summarize(f)[:70]}", flush=True)

    try:
        sent = replay(frames, send, args.interval, args.loops, args.forever,
                      log=log)
    except KeyboardInterrupt:
        print("\ninterrupted")
        return 0
    print(f"done: sent {sent} announcements")
    return 0


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