MicrOBU/obu-firmware/test/pcap_gn_tally.py

#!/usr/bin/env python3
"""Tally GeoNetworking header fields per sending station across .pcap captures.

Written to check the GN lifetime byte on air (see TODO.md), and it answers the general question
"what do real stations put in this header" too: one row per source MAC, packet type, BTP port and
lifetime byte, with a frame count.

    python obu-firmware/test/pcap_gn_tally.py its-g5-receiver-firmware/recordings/*.pcap

Handles linktype 127 (radiotap, what its-g5-receiver-firmware records) and 105 (bare 802.11).
Standard library only. Pseudonym MACs rotate, so one vehicle can appear as several rows. The
pcap-over-serial dump path corrupts roughly 0.3% of frames, so a stray odd row is tooling noise.
"""
import collections
import glob
import struct
import sys

LLC_SNAP_GN = b"\xaa\xaa\x03\x00\x00\x00\x89\x47"

# (HeaderType, HeaderSubtype) from the GN Common Header -> name, EN 302 636-4-1 table 9.
HEADER_TYPES = {
    (1, 0): "beacon",
    (4, 0): "GBC-circle",
    (4, 1): "GBC-rect",
    (4, 2): "GBC-ellipse",
    (5, 0): "SHB",
}
# Extended header length, i.e. the distance from the end of the Common Header to BTP-B.
EXT_LEN = {"SHB": 28, "GBC-circle": 44, "GBC-rect": 44, "GBC-ellipse": 44}


def lifetime_seconds(raw):
    # Multiplier in the upper 6 bits, base in the lower 2: 50 ms, 1 s, 10 s, 100 s.
    return (raw >> 2) * (0.05, 1, 10, 100)[raw & 3]


def frames(path):
    with open(path, "rb") as f:
        data = f.read()
    if len(data) < 24:
        return
    magic = struct.unpack("<I", data[:4])[0]
    endian = "<" if magic in (0xA1B2C3D4, 0xA1B23C4D) else ">"
    linktype = struct.unpack(endian + "I", data[20:24])[0]
    off = 24
    while off + 16 <= len(data):
        incl = struct.unpack(endian + "I", data[off + 8:off + 12])[0]
        pkt = data[off + 16:off + 16 + incl]
        off += 16 + incl
        if linktype == 127:
            if len(pkt) < 4:
                continue
            pkt = pkt[struct.unpack("<H", pkt[2:4])[0]:]
        elif linktype != 105:
            continue
        yield pkt


def gn_fields(f):
    if len(f) < 24 or (f[0] >> 2) & 3 != 2:
        return None  # Data frames only
    o = 24 + (2 if f[0] & 0x80 else 0)  # QoS Data carries a 2-byte QoS Control field
    if f[o:o + 8] != LLC_SNAP_GN or len(f) < o + 12:
        return None
    o += 8
    src = f[10:16].hex(":")
    version, next_header, lifetime = f[o] >> 4, f[o] & 0x0F, f[o + 2]
    port = "-"
    if next_header == 2:
        kind = "secured"  # Common Header is inside the security envelope
    elif next_header == 1 and len(f) >= o + 12:
        c = o + 4
        ht = (f[c + 1] >> 4, f[c + 1] & 0x0F)
        kind = HEADER_TYPES.get(ht, "type %d/%d" % ht)
        ext = EXT_LEN.get(kind)
        btp = c + 8 + (ext or 0)
        if ext and len(f) >= btp + 2:
            port = str(struct.unpack(">H", f[btp:btp + 2])[0])
    else:
        kind = "nh=%d" % next_header
    return src, version, kind, port, lifetime


def main(argv):
    # PowerShell does not expand wildcards itself, so do it here.
    paths = [p for arg in argv for p in (glob.glob(arg) or [arg])]
    if not paths:
        sys.exit(__doc__)
    tally = collections.Counter()
    for path in paths:
        for frame in frames(path):
            fields = gn_fields(frame)
            if fields:
                tally[fields] += 1
    print("%-17s %3s %-11s %5s %8s %8s %7s"
          % ("source", "ver", "packet", "port", "lifetime", "seconds", "frames"))
    for (src, ver, kind, port, lt), n in sorted(tally.items()):
        print("%-17s %3d %-11s %5s %8s %8g %7d"
              % (src, ver, kind, port, "0x%02x" % lt, lifetime_seconds(lt), n))


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