diff --git a/sniffer/parse_hci.py b/sniffer/parse_hci.py new file mode 100644 index 0000000..66ae4a5 --- /dev/null +++ b/sniffer/parse_hci.py @@ -0,0 +1,229 @@ +""" +parse_hci.py - Parse Android btsnoop_hci.log to extract KAIYU BLE commands. + +Reads a btsnoop v1 binary log and extracts: + - BLE service / characteristic UUIDs + - Every ATT Write Command payload (the actual LED control bytes) + +Usage: + python sniffer/parse_hci.py + python sniffer/parse_hci.py D:\\Claude\\bugreport\\btsnoop_hci.log + python sniffer/parse_hci.py --out results.json +""" + +import struct +import sys +import json +import os +import argparse +from dataclasses import dataclass, field +from typing import Optional + +BTSNOOP_MAGIC = b"btsnoop\x00" +ATT_WRITE_CMD = 0x52 +ATT_WRITE_REQ = 0x12 +ATT_READ_BY_TYPE_RSP = 0x09 +ATT_CID = 0x0004 + + +@dataclass +class BLEWrite: + timestamp_us: int + direction: str + handle: int + data: bytes + data_hex: str = field(init=False) + + def __post_init__(self): + self.data_hex = self.data.hex() + + +@dataclass +class ParseResult: + writes: list = field(default_factory=list) + handles_seen: set = field(default_factory=set) + uuid_map: dict = field(default_factory=dict) + + +def parse_btsnoop(filepath: str) -> ParseResult: + result = ParseResult() + with open(filepath, "rb") as f: + magic = f.read(8) + if magic != BTSNOOP_MAGIC: + raise ValueError( + f"Not a btsnoop binary file (magic={magic!r}). " + "Need the raw btsnoop_hci.log binary, not the bugreport .txt." + ) + version, datalink = struct.unpack(">II", f.read(8)) + print(f"BTSnoop v{version}, datalink type={datalink}") + + packet_num = 0 + while True: + hdr = f.read(24) + if len(hdr) < 24: + break + orig_len, incl_len, flags, drops, timestamp_us = struct.unpack(">IIIIq", hdr) + data = f.read(incl_len) + packet_num += 1 + + direction = "RX" if (flags & 1) else "TX" + if len(data) < 5 or data[0] != 0x02: + continue + + payload = data[5:] + if len(payload) < 4: + continue + + l2cap_len, l2cap_cid = struct.unpack("= item_len and item_len >= 7: + item = items[:item_len] + items = items[item_len:] + value_handle = struct.unpack(" Optional[str]: + if not data: + return None + if len(data) >= 7 and data[0] == 0x7e and data[-1] == 0xef: + cmd = data[2] + if cmd == 0x04: + return f"7E -> Power {'ON' if data[7] != 0 else 'OFF'}" + elif cmd == 0x05 and len(data) >= 8: + return f"7E -> RGB({data[4]},{data[5]},{data[6]})" + elif cmd == 0x01: + return f"7E -> Brightness {data[3]}%" + elif cmd == 0x03: + return f"7E -> Effect 0x{data[3]:02x}" + return f"7E -> cmd=0x{cmd:02x}" + if data[0] == 0x56 and len(data) >= 7: + if data[5] == 0xf0 and data[6] == 0xaa: + return f"56 -> RGB({data[1]},{data[2]},{data[3]})" + elif data[5] == 0x0f and data[6] == 0xaa: + return f"56 -> White level={data[4]}" + if len(data) == 3 and data[0] == 0xcc: + return f"56 -> Power {'ON' if data[1] == 0x23 else 'OFF'}" + return None + + +def analyse(result: ParseResult): + print(f"\n{'='*60}") + print(f" PARSE RESULTS") + print(f"{'='*60}") + print(f"Write packets : {len(result.writes)}") + print(f"Handles used : {[f'0x{h:04x}' for h in sorted(result.handles_seen)]}") + + if result.uuid_map: + print(f"\nCharacteristic UUIDs:") + for handle, uuid in sorted(result.uuid_map.items()): + marker = " <- WRITE TARGET" if handle in result.handles_seen else "" + print(f" 0x{handle:04x} : {uuid}{marker}") + + if not result.writes: + print("\n[!] No ATT writes found.") + print(" Ensure HCI Snoop Log was enabled BEFORE using KAIYU app.") + print(" Steps: enable it -> toggle BT off/on -> use app -> re-pull log.") + return + + print(f"\n{'─'*60}") + print(f" ALL WRITE COMMANDS") + print(f"{'─'*60}") + prev_ts = result.writes[0].timestamp_us + for i, w in enumerate(result.writes): + delta_ms = (w.timestamp_us - prev_ts) / 1000 + prev_ts = w.timestamp_us + decoded = try_decode(w.data) or "" + print(f"[{i+1:>3}] +{delta_ms:>8.1f}ms 0x{w.handle:04x} {w.data_hex:<30} {decoded}") + + print(f"\n{'─'*60}") + print(f" UNIQUE PAYLOADS") + print(f"{'─'*60}") + seen = {} + for w in result.writes: + seen.setdefault(w.data_hex, w) + for hex_str, w in seen.items(): + print(f" 0x{w.handle:04x} {hex_str:<30} {try_decode(w.data) or 'unknown format'}") + + most_used = max(result.handles_seen, + key=lambda h: sum(1 for w in result.writes if w.handle == h)) + uuid = result.uuid_map.get(most_used, "UNKNOWN - note the handle and check in nRF Connect") + + print(f"\n{'='*60}") + print(f" IDENTIFIED WRITE CHARACTERISTIC") + print(f"{'='*60}") + print(f" Handle : 0x{most_used:04x}") + print(f" UUID : {uuid}") + print(f"\n Add to controller/config.py:") + print(f' WRITE_CHARACTERISTIC = "{uuid}"') + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("logfile", nargs="?") + parser.add_argument("--out", "-o") + args = parser.parse_args() + + candidates = [ + args.logfile, + r"D:\Claude\bugreport\btsnoop_hci.log", + r"D:\HereMyHope\btsnoop.log", + "btsnoop_hci.log", + ] + logfile = next((p for p in candidates if p and os.path.exists(p)), None) + if not logfile: + print("No btsnoop log found. Run pull_bt_log.bat first.") + sys.exit(1) + + print(f"Parsing : {logfile} ({os.path.getsize(logfile):,} bytes)") + try: + result = parse_btsnoop(logfile) + except ValueError as e: + print(f"\n[ERROR] {e}") + sys.exit(1) + + analyse(result) + + if args.out: + with open(args.out, "w") as f: + json.dump({ + "writes": [{"handle": f"0x{w.handle:04x}", "hex": w.data_hex, + "decoded": try_decode(w.data)} for w in result.writes], + "uuid_map": {f"0x{h:04x}": u for h, u in result.uuid_map.items()}, + }, f, indent=2) + print(f"\nSaved to {args.out}") + + +if __name__ == "__main__": + main()