Поддержка MacOS (проверено в Tahoe)

misc/analyze_pcap.py

@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+"""Конвертер дампа USBPcap (pcapng) → лог control-трансферов SET/GET_REPORT.
+
+Использование: ./analyze_pcap.py <capture.pcapng> [out.txt]
+"""
+import struct
+import sys
+
+# тип блока Enhanced Packet в pcapng
+BLOCK_ENHANCED_PACKET = 6
+# тип transfer'а control в USBPcap
+XFER_CONTROL = 2
+# биты направления в endpoint
+EP_DIR_IN = 0x80  # device → host; иначе host → device
+
+
+def read_packet_blocks(data: bytes) -> list[bytes]:
+    """Извлечь данные всех Enhanced Packet-блоков из pcapng-файла."""
+    records = []
+    off = 0
+    while off + 12 <= len(data):
+        block_type, block_len = struct.unpack_from("<II", data, off)
+        if block_len < 12 or off + block_len > len(data):
+            break
+        if block_type == BLOCK_ENHANCED_PACKET:
+            caplen = struct.unpack_from("<I", data, off + 20)[0]
+            records.append(data[off + 28:off + 28 + caplen])
+        off += block_len
+    return records
+
+
+def parse_usb_header(packet: bytes):
+    """Разобрать заголовок USBPcap: → (irpid, endpoint, transfer, payload)."""
+    header_len = struct.unpack_from("<H", packet, 0)[0]
+    irpid = struct.unpack_from("<Q", packet, 2)[0]
+    endpoint = packet[21]
+    transfer = packet[22]
+    data_len = struct.unpack_from("<I", packet, 23)[0]
+    payload = packet[header_len:header_len + data_len]
+    return irpid, endpoint, transfer, payload
+
+
+def parse_events(records: list[bytes]) -> list[tuple]:
+    """Собрать события SET/GET_REPORT, склеивая setup+data GET-ов по irpid."""
+    events = []
+    pending_get = {}  # irpid → (iface, request_type, report_id)
+    for packet in records:
+        irpid, endpoint, transfer, data = parse_usb_header(packet)
+        if transfer != XFER_CONTROL or not data:
+            continue
+        if (not endpoint & EP_DIR_IN and len(data) >= 8
+                and data[0] == 0x21 and data[1] == 0x09):
+            # merged SET_REPORT: setup(8 байт) + payload
+            wValue = struct.unpack("<H", data[2:4])[0]
+            wIndex = struct.unpack("<H", data[4:6])[0]
+            events.append(("SET", wIndex, wValue >> 8, wValue & 0xFF, data[8:]))
+        elif (endpoint & EP_DIR_IN and len(data) == 8
+                and data[0] == 0xA1 and data[1] == 0x01):
+            # setup-часть GET_REPORT: данные придут отдельным пакетом с тем же irpid
+            wValue = struct.unpack("<H", data[2:4])[0]
+            wIndex = struct.unpack("<H", data[4:6])[0]
+            pending_get[irpid] = (wIndex, wValue >> 8, wValue & 0xFF)
+        elif endpoint & EP_DIR_IN and irpid in pending_get:
+            iface, request_type, report_id = pending_get.pop(irpid)
+            events.append(("GET", iface, request_type, report_id, data))
+    return events
+
+
+def main():
+    if len(sys.argv) < 2:
+        raise SystemExit(__doc__)
+    src = sys.argv[1]
+    dst = sys.argv[2] if len(sys.argv) > 2 else "/tmp/tx_log.txt"
+
+    with open(src, "rb") as f:
+        data = f.read()
+    events = parse_events(read_packet_blocks(data))
+
+    with open(dst, "w") as f:
+        for num, (kind, iface, rtype, rid, payload) in enumerate(events, 1):
+            f.write(f"{kind} #{num:4d} if={iface} t={rtype} rid={rid:02x} "
+                    f"len={len(payload)}: {payload.hex(' ')}\n")
+    print("events:", len(events))
+
+
+if __name__ == "__main__":
+    main()