88 lines
3.4 KiB
Python
Executable File
88 lines
3.4 KiB
Python
Executable File
#!/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()
|