From 7c43a142e6032ab47722f3a9b462e6c544d68408 Mon Sep 17 00:00:00 2001 From: michael Date: Wed, 18 Mar 2026 19:44:54 -0700 Subject: [PATCH] refactor: extract sniff infrastructure into pm3py/sniff/ sub-package MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2 of package refactor. Extracts trace parsing, ISO 15693 decoders, NDEF annotation, and ANSI formatting from core/hf_iso15.py into dedicated sniff/ modules: - sniff/trace.py — parse_tracelog (protocol-agnostic) - sniff/ndef.py — NDEF TLV/record decode - sniff/decode_iso15.py — 15693 command/response decoders - sniff/format.py — ANSI color formatting - sniff/session.py — SniffSession with iso15() method core/hf_iso15.py re-exports sniff symbols for backward compat. Also fixes NTAG 5 placement (iso15693, not iso14443a4) in design doc. Co-Authored-By: Claude Opus 4.6 (1M context) --- CLAUDE.md | 45 +- docs/plans/2026-03-18-refactor-design.md | 231 +++++++++ docs/plans/2026-03-18-refactor-progress.md | 4 +- pm3py/core/hf_iso15.py | 523 +-------------------- pm3py/sniff/__init__.py | 13 + pm3py/sniff/decode_iso15.py | 146 ++++++ pm3py/sniff/format.py | 118 +++++ pm3py/sniff/ndef.py | 152 ++++++ pm3py/sniff/session.py | 75 +++ pm3py/sniff/trace.py | 41 ++ 10 files changed, 830 insertions(+), 518 deletions(-) create mode 100644 docs/plans/2026-03-18-refactor-design.md create mode 100644 pm3py/sniff/decode_iso15.py create mode 100644 pm3py/sniff/format.py create mode 100644 pm3py/sniff/ndef.py create mode 100644 pm3py/sniff/session.py create mode 100644 pm3py/sniff/trace.py diff --git a/CLAUDE.md b/CLAUDE.md index 46bd451..c44da53 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,17 +15,38 @@ python -m pytest tests/ -v pip install -e . ``` -## Architecture +## Package structure -- `protocol.py` — Wire constants, CRC-16/A, Cmd enum, PM3Status enum. Source of truth for command IDs and frame formats. -- `transport.py` — Frame encode/decode (`encode_ng_frame`, `encode_mix_frame`, `decode_response_frame`), `PM3Transport` async serial class. USB skips CRC (uses magic `0x3361`), FPC UART uses CRC with byte-swapped wire format (`(low << 8) | high`). -- `client.py` — `Proxmark3` class (async-first, `Proxmark3.sync()` for REPL), `_SyncProxy`, `FirmwareInfo`, port auto-detection. -- `hw.py` — Hardware commands. LED API is platform-aware (Easy vs RDV4 have different color-to-pin mappings). PWM validation fetches capabilities on first use. -- `lf.py` — LF commands + `T55xxCommands` + `LFSearchResult`. Full tag demod (EM410x, HID, etc.) is NOT implemented — that requires the C client's DSP stack. -- `hf.py` — HF core (tune, search, sniff, dropfield). -- `hf_14a.py` — ISO 14443-A (scan, raw, sniff, sim). Uses MIX frames. -- `hf_15.py` — ISO 15693 (scan, rdbl, wrbl, sniff). Uses NG frames with ISO command payloads. -- `hf_mf.py` — MIFARE Classic (rdbl, wrbl, rdsc, chk, sniff, sim, nested, cident). Read uses NG, write uses MIX. +``` +pm3py/ + __init__.py # re-exports Proxmark3, PM3Error, PM3Response, Cmd, PM3Status + core/ # wire protocol and device commands + protocol.py # wire constants, CRC-16/A, Cmd enum, PM3Status + transport.py # frame encode/decode, PM3Transport async serial + client.py # Proxmark3 class, _SyncProxy, FirmwareInfo + hw.py # hardware commands, LED API (platform-aware) + hf.py # HF core — tune, search, sniff, dropfield + hf_iso14a.py # ISO 14443-A — scan, raw. Uses MIX frames + hf_iso15.py # ISO 15693 — scan, rdbl, wrbl, sniff + trace decoders + hf_mfc.py # MIFARE Classic — rdbl, wrbl, rdsc, chk, nested, cident + lf.py # LF commands + T55xxCommands + LFSearchResult + sniff/ # sniff sessions, trace parsing, protocol decoders (scaffold) + sim/ # card simulation sessions, table compiler, relay (scaffold) + reader/ # higher-level reader modes (scaffold) + transponders/ # tag/transponder models independent of hardware (scaffold) +``` + +## Client API + +```python +pm3.hw.ping() # hardware +pm3.hf.iso14a.scan() # ISO 14443-A +pm3.hf.iso15.rdbl(4) # ISO 15693 +pm3.hf.mfc.rdbl(0) # MIFARE Classic +pm3.lf.t55.readbl(0) # T55xx +pm3.hf.tune() # HF antenna tune +pm3.hf.dropfield() # drop field +``` ## Wire protocol essentials @@ -37,7 +58,7 @@ pip install -e . ## Key patterns -- All command methods are `async`. The sync wrapper in `_SyncProxy` intercepts via `__getattr__` and calls `loop.run_until_complete()`. +- All command methods are `async`. The sync wrapper in `_SyncProxy` (in `core/client.py`) intercepts via `__getattr__` and calls `loop.run_until_complete()`. - Command classes hold a `self._t` reference to `PM3Transport` (or mock in tests). - Tests use `AsyncMock` for transport. Set `hw._is_rdv4 = False` to skip capabilities fetch in LED tests. - `capabilities()` response parsed at known byte/bit offsets from the C struct (version=7 format). @@ -84,7 +105,7 @@ Firmware maintenance: atomic single-file commits for easy rebase against upstrea ## Adding new commands -1. Find the `CMD_*` constant in `include/pm3_cmd.h` and add to `Cmd` enum in `protocol.py` +1. Find the `CMD_*` constant in `include/pm3_cmd.h` and add to `Cmd` enum in `core/protocol.py` 2. Check if the C client uses `SendCommandNG` (→ `send_ng`) or `SendCommandMIX` (→ `send_mix`) 3. Check the payload struct in `pm3_cmd.h` and use `struct.pack` to build it 4. Parse the response using `struct.unpack_from` on `resp.data` diff --git a/docs/plans/2026-03-18-refactor-design.md b/docs/plans/2026-03-18-refactor-design.md new file mode 100644 index 0000000..73ee0d4 --- /dev/null +++ b/docs/plans/2026-03-18-refactor-design.md @@ -0,0 +1,231 @@ +# pm3py Package Refactor Design + +**Date:** 2026-03-18 +**Status:** Design complete, ready for implementation + +## Motivation + +pm3py has grown from a flat wire-protocol library into a full ecosystem with +transponder models, sim sessions, sniff infrastructure, MCU bridges, and AES +crypto. The current flat layout (`pm3py/hf_14a.py`, `pm3py/hf_15.py`, etc.) +doesn't scale — sniff infrastructure is tangled with reader commands, naming is +inconsistent (`hf.a14` vs `hf.iso15`), and there's no clear home for +transponder models or higher-level workflows. + +## Design Decisions + +### 1. Single package, sub-packages (not namespace packages) + +Everything stays under `pm3py/`. One `pip install pm3py`. No separate +installable packages — this is one tool with one hardware target. Split later +if a genuine need arises. + +### 2. Hybrid client model + +- **Core commands** stay on the `Proxmark3` client — `pm3.hf.iso15.rdbl(4)`. + Feels like the PM3 CLI. +- **Sim and Sniff** are standalone session classes that take the transport (not + the client). They have lifecycle (start/stop/download) that doesn't fit the + stateless command pattern. +- **Reader modes** are higher-level classes that compose core commands. They + take a `Proxmark3` instance. + +### 3. Core preserves PM3 CLI feel + +Low-level commands (`scan`, `rdbl`, `wrbl`, `sniff`) stay in `core/`. Reader +modes are a layer on top. Someone who just wants `pm3.hf.iso15.rdbl(4)` +shouldn't have to think about "modes." + +### 4. Naming follows PM3 conventions + +- `hf_14a.py` → `hf_iso14a.py` +- `hf_15.py` → `hf_iso15.py` +- `hf_mf.py` → `hf_mfc.py` +- Client attributes: `hf.a14` → `hf.iso14a`, `hf.mf` → `hf.mfc`, `hf.iso15` stays + +### 5. Sniff extracted from core + +`hf_15.py` is currently ~60% sniff infrastructure (trace parsing, 15693 +decoders, NDEF annotation, ANSI formatting) and ~40% reader commands. The +sniff code becomes its own sub-package. Core keeps only thin `.sniff()` firmware +commands. + +### 6. Transponder hierarchy: frequency → standard → manufacturer → transponder + +For HF, the ISO standard is the second level. For LF, where there's typically +no standard, skip straight to manufacturer. + +### 7. Reader hierarchy: frequency → manufacturer + +Reader modes organized by frequency, then by reader IC manufacturer. + +## Package Structure + +``` +pm3py/ + __init__.py # re-exports from core for backward compat + + core/ + __init__.py # re-exports Proxmark3, PM3Error, PM3Response, Cmd, PM3Status + protocol.py # wire constants, CRC, Cmd enum, PM3Status + transport.py # frame encode/decode, PM3Transport + client.py # Proxmark3 class, _SyncProxy, FirmwareInfo + hw.py # HWCommands + hf.py # HFCommands — tune, search, dropfield + hf_iso14a.py # HF14ACommands — scan, raw + hf_iso15.py # HF15Commands — scan, rdbl, wrbl (sniff removed) + hf_mfc.py # HFMFCommands — rdbl, wrbl, rdsc, chk, nested, cident + lf.py # LFCommands, T55xxCommands, LFSearchResult + + sniff/ + __init__.py + session.py # SniffSession — start/stop/download per protocol + trace.py # parse_tracelog() — protocol-agnostic trace buffer parsing + decode_iso15.py # ISO 15693 request/response decoders, command tables + decode_iso14a.py # ISO 14443-A decoder (stub/future) + ndef.py # NDEF TLV/record decode for trace annotation + format.py # ANSI color formatting, format_sniff_line, wrapping + + sim/ + __init__.py + # sim session, table compiler, relay — moved from worktree + + reader/ + __init__.py + hf/ + __init__.py + nxp/ # NXP reader ICs (CLRC663, PN5xx) + st/ # ST reader ICs + lf/ + __init__.py + modes/ + __init__.py + inventory.py # multi-tag scan, anti-collision workflows + programming.py # bulk read/write/clone + access.py # Wiegand/OSDP credential operations + + transponders/ + __init__.py + hf/ + __init__.py + iso14443a3/ + __init__.py + nxp/ + __init__.py + mifare_classic.py + mifare_ultralight.py + ntag.py # NTAG 213/215/216 + iso14443a4/ + __init__.py + nxp/ + __init__.py + desfire.py + iso15693/ + __init__.py + common.py # base ISO 15693 system info, block layout + ndef.py # NFC Forum Type 5 NDEF models + nxp/ + __init__.py + icode_slix2.py + ntag5.py # NTAG 5 (ISO 15693 / NFC Type 5) + st/ + __init__.py + st25tv.py + lf/ + __init__.py + atmel/ + __init__.py + t55xx.py + em/ + __init__.py + em4100.py + em4x05.py +``` + +## Import Paths + +### Core commands (PM3 CLI feel, unchanged) + +```python +from pm3py import Proxmark3 + +async with Proxmark3() as pm3: + await pm3.hw.ping() + await pm3.hf.iso14a.scan() + await pm3.hf.iso15.rdbl(4) + await pm3.hf.mfc.rdbl(0) + await pm3.lf.t55.readbl(0) + await pm3.hf.tune() + await pm3.hf.dropfield() +``` + +### Sniff and Sim (standalone session classes) + +```python +from pm3py.sniff import SniffSession +from pm3py.sim import SimSession + +sniff = SniffSession(pm3._transport) +entries = await sniff.iso15(timeout=60.0) + +sim = SimSession(pm3._transport) +await sim.start(tag_model) +``` + +### Reader modes (compose core commands) + +```python +from pm3py.reader import InventoryMode + +inv = InventoryMode(pm3) +tags = await inv.scan_all_15() +``` + +### Transponder models (independent of hardware) + +```python +from pm3py.transponders.hf.iso14443a3.nxp import MifareClassic1K +from pm3py.transponders.hf.iso15693.nxp import IcodeSlix2 +from pm3py.transponders.hf.iso15693 import ndef +``` + +## Backward Compatibility + +- `pm3py/__init__.py` keeps re-exporting `Proxmark3`, `PM3Error`, + `PM3Response`, `Cmd`, `PM3Status` +- Existing `from pm3py import Proxmark3` continues to work +- **Breaking changes:** attribute renames on client only: + `hf.a14` → `hf.iso14a`, `hf.mf` → `hf.mfc` + +## Migration Phases + +### Phase 1 — Create `core/`, normalize names, fix imports + +- Create `pm3py/core/` and move protocol, transport, client, hw, hf, lf +- Rename during move: `hf_14a.py` → `hf_iso14a.py`, `hf_15.py` → + `hf_iso15.py`, `hf_mf.py` → `hf_mfc.py` +- Normalize client attributes: `hf.a14` → `hf.iso14a`, `hf.mf` → `hf.mfc` +- `pm3py/__init__.py` re-exports from `core` +- Move and update tests. All green before proceeding. + +### Phase 2 — Extract `sniff/` + +- Pull trace parsing, 15693 decoders, NDEF annotation, ANSI formatting out of + `hf_iso15.py` into `sniff/` +- Core `hf_iso15.py` keeps only thin `.sniff()` firmware command +- `sniff_decoded()` → `SniffSession.iso15()` + +### Phase 3 — Scaffold `transponders/` + +- Create the `hf/iso14443a3/`, `hf/iso14443a4/`, `hf/iso15693/`, `lf/atmel/`, + `lf/em/` tree +- Start with `hf/iso15693/nxp/icode_slix2.py` and + `hf/iso14443a3/nxp/mifare_classic.py` — these have existing models in the + sim worktree + +### Phase 4 — Scaffold `reader/` and `sim/` + +- `reader/` with `hf/`, `lf/`, `modes/` stubs +- `sim/` — migrate from worktree + +Each phase is one PR, tests green before merge. diff --git a/docs/plans/2026-03-18-refactor-progress.md b/docs/plans/2026-03-18-refactor-progress.md index 9ca6678..fb93adf 100644 --- a/docs/plans/2026-03-18-refactor-progress.md +++ b/docs/plans/2026-03-18-refactor-progress.md @@ -45,8 +45,8 @@ Created `pm3py/sniff/__init__.py` placeholder. Will extract into sub-modules whe Created full directory tree per design: - `transponders/hf/iso14443a3/nxp/` — MIFARE Classic, Ultralight, NTAG -- `transponders/hf/iso14443a4/nxp/` — DESFire, NTAG 5 -- `transponders/hf/iso15693/nxp/` — ICODE SLIX2 +- `transponders/hf/iso14443a4/nxp/` — DESFire +- `transponders/hf/iso15693/nxp/` — ICODE SLIX2, NTAG 5 - `transponders/hf/iso15693/st/` — ST25TV - `transponders/lf/atmel/` — T55xx - `transponders/lf/em/` — EM4100, EM4x05 diff --git a/pm3py/core/hf_iso15.py b/pm3py/core/hf_iso15.py index 4ee45cb..0da8dff 100644 --- a/pm3py/core/hf_iso15.py +++ b/pm3py/core/hf_iso15.py @@ -1,9 +1,16 @@ -"""ISO 15693 commands: hf.15.*""" +"""ISO 15693 commands: hf.iso15.*""" import struct -import sys -from .protocol import Cmd, PM3_CMD_DATA_SIZE +from .protocol import Cmd from .transport import PM3Transport, PM3Error +# Re-export sniff symbols for backward compat with existing test imports +from ..sniff.trace import parse_tracelog, TRACELOG_HDR_SIZE +from ..sniff.decode_iso15 import ( + decode_15693_request, decode_15693_response, decode_15693, +) +from ..sniff.ndef import decode_ndef_annotation +from ..sniff.format import format_sniff_line + # ISO15 PM3 flags ISO15_CONNECT = 0x01 ISO15_NO_DISCONNECT = 0x02 @@ -11,461 +18,6 @@ ISO15_RAW = 0x04 ISO15_APPEND_CRC = 0x08 ISO15_READ_RESPONSE = 0x10 -# tracelog_hdr_t size: timestamp(4) + duration(2) + data_len_flags(2) = 8 -TRACELOG_HDR_SIZE = 8 - - -def parse_tracelog(raw: bytes) -> list[dict]: - """Parse PM3 trace buffer into a list of trace entries. - - Each entry: {timestamp, duration, is_response, data, parity} - """ - entries = [] - offset = 0 - while offset + TRACELOG_HDR_SIZE <= len(raw): - timestamp, duration, data_len_flags = struct.unpack_from(" len(raw): - break - - data = raw[offset:offset + data_len] - offset += data_len - - parity_len = (data_len + 7) // 8 if data_len > 0 else 0 - parity = raw[offset:offset + parity_len] if parity_len else b"" - offset += parity_len - - entries.append({ - "timestamp": timestamp, - "duration": duration, - "is_response": is_response, - "data": data, - "parity": parity, - }) - return entries - - -# ---- NDEF decode (for trace annotation) ---- - -_NDEF_MAGIC = 0xE1 - -_URI_PREFIXES = { - 0x00: "", 0x01: "http://www.", 0x02: "https://www.", - 0x03: "http://", 0x04: "https://", 0x05: "tel:", 0x06: "mailto:", -} - -_TNF_EMPTY = 0x00 -_TNF_WELL_KNOWN = 0x01 -_TNF_MEDIA = 0x02 -_TNF_URI = 0x03 -_TNF_EXTERNAL = 0x04 -_MAX_ANN_TEXT = 40 - - -def decode_ndef_annotation(data: bytes) -> str | None: - """Decode NDEF TLV data into a human-readable annotation string. - - Handles CC (capability container) blocks and NDEF message TLVs. - Returns None if the data doesn't contain recognizable NDEF content. - """ - if len(data) < 3: - return None - - parts = [] - offset = 0 - - if data[0] == _NDEF_MAGIC and len(data) >= 4: - ver_major = (data[1] >> 6) & 0x03 - ver_minor = (data[1] >> 4) & 0x03 - mlen = data[2] - features = data[3] - cc_str = f"CC v{ver_major}.{ver_minor} MLEN={mlen}" - if features & 0x01: - cc_str += " MBREAD" - parts.append(cc_str) - offset = 4 - if offset >= len(data): - return cc_str - - while offset < len(data): - tlv_type = data[offset] - offset += 1 - if tlv_type == 0x00: - continue - if tlv_type == 0xFE: - break - if tlv_type != 0x03: - break - - if offset >= len(data): - break - tlv_len = data[offset] - offset += 1 - if tlv_len == 0xFF: - if offset + 2 > len(data): - break - tlv_len = (data[offset] << 8) | data[offset + 1] - offset += 2 - - if tlv_len == 0: - parts.append("NDEF empty") - continue - - msg_end = min(offset + tlv_len, len(data)) - records = _parse_ndef_records(data[offset:msg_end]) - for rec in records: - parts.append(rec) - offset = msg_end - - return " | ".join(parts) if parts else None - - -def _parse_ndef_records(msg: bytes) -> list[str]: - """Parse NDEF records from a message payload.""" - records = [] - offset = 0 - while offset < len(msg): - if offset + 3 > len(msg): - break - flags = msg[offset] - tnf = flags & 0x07 - sr = bool(flags & 0x10) - il = bool(flags & 0x08) - type_len = msg[offset + 1] - offset += 2 - - if sr: - if offset >= len(msg): - break - payload_len = msg[offset] - offset += 1 - else: - if offset + 4 > len(msg): - break - payload_len = int.from_bytes(msg[offset:offset + 4], "big") - offset += 4 - - id_len = 0 - if il: - if offset >= len(msg): - break - id_len = msg[offset] - offset += 1 - - if offset + type_len > len(msg): - break - rec_type = msg[offset:offset + type_len] - offset += type_len - offset += id_len - - if offset + payload_len > len(msg): - payload = msg[offset:] - else: - payload = msg[offset:offset + payload_len] - offset += payload_len - - records.append(_decode_ndef_record(tnf, rec_type, payload)) - return records - - -def _decode_ndef_record(tnf: int, rec_type: bytes, payload: bytes) -> str: - """Decode a single NDEF record into an annotation string.""" - if tnf == _TNF_WELL_KNOWN: - if rec_type == b"T" and len(payload) >= 1: - lang_len = payload[0] & 0x3F - lang = payload[1:1 + lang_len].decode("ascii", errors="replace") - text = payload[1 + lang_len:].decode("utf-8", errors="replace") - if len(text) > _MAX_ANN_TEXT: - text = text[:_MAX_ANN_TEXT] + "..." - return f'NDEF Text "{lang}" "{text}"' - if rec_type == b"U" and len(payload) >= 1: - prefix = _URI_PREFIXES.get(payload[0], "") - suffix = payload[1:].decode("utf-8", errors="replace") - uri = prefix + suffix - if len(uri) > _MAX_ANN_TEXT: - uri = uri[:_MAX_ANN_TEXT] + "..." - return f"NDEF URI {uri}" - return f"NDEF WK type={rec_type!r} [{len(payload)}B]" - if tnf == _TNF_MEDIA: - mime = rec_type.decode("ascii", errors="replace") - if len(mime) > 30: - mime = mime[:30] + "..." - return f"NDEF MIME {mime} [{len(payload)}B]" - if tnf == _TNF_EXTERNAL: - ext_type = rec_type.decode("ascii", errors="replace") - return f"NDEF EXT {ext_type} [{len(payload)}B]" - if tnf == _TNF_EMPTY: - return "NDEF empty" - return f"NDEF TNF={tnf} [{len(payload)}B]" - - -# ---- ISO 15693 command names (for sniff annotation) ---- -_15693_CMDS = { - 0x01: "INVENTORY", - 0x02: "STAY QUIET", - 0x20: "READ SINGLE BLOCK", - 0x21: "WRITE SINGLE BLOCK", - 0x22: "LOCK BLOCK", - 0x23: "READ MULTIPLE BLOCKS", - 0x24: "WRITE MULTIPLE BLOCKS", - 0x25: "SELECT", - 0x26: "RESET TO READY", - 0x27: "WRITE AFI", - 0x28: "LOCK AFI", - 0x29: "WRITE DSFID", - 0x2A: "LOCK DSFID", - 0x2B: "GET SYSTEM INFO", - 0x2C: "GET MULTIPLE BLOCK SECURITY", -} - -_15693_NXP_CMDS = { - 0xA0: "NXP INVENTORY READ", - 0xA1: "NXP FAST INVENTORY READ", - 0xA2: "NXP SET EAS", - 0xA3: "NXP RESET EAS", - 0xA4: "NXP LOCK EAS", - 0xA5: "NXP EAS ALARM", - 0xA6: "NXP PASSWORD PROTECT EAS/AFI", - 0xA7: "NXP WRITE EAS ID", - 0xAB: "NXP GET SYSTEM INFO", - 0xB2: "NXP GET RANDOM", - 0xB3: "NXP SET PASSWORD", - 0xB4: "NXP WRITE PASSWORD", - 0xB5: "NXP LOCK PASSWORD", - 0xB6: "NXP PROTECT PAGE", - 0xB7: "NXP LOCK PAGE PROTECTION", - 0xB9: "NXP DESTROY", - 0xBA: "NXP ENABLE PRIVACY", - 0xBB: "NXP 64-BIT PASSWORD PROTECTION", - 0xBD: "NXP READ SIGNATURE", -} - -_15693_ERRORS = { - 0x01: "not supported", - 0x02: "not recognized", - 0x10: "block not available", - 0x11: "block already locked", - 0x12: "block locked", - 0x13: "block not written", - 0x14: "block not locked", -} - -_FLAG_INVENTORY = 0x04 -_FLAG_ADDRESS = 0x20 - - -def _block_offset(flags: int) -> int: - """Byte offset of block number after flags+cmd.""" - if not (flags & _FLAG_INVENTORY) and (flags & _FLAG_ADDRESS): - return 10 # flags(1) + cmd(1) + uid(8) - return 2 # flags(1) + cmd(1) - - -def decode_15693_request(payload: bytes) -> str | None: - """Decode an ISO 15693 request frame into annotation.""" - if len(payload) < 2: - return None - - flags = payload[0] - cmd = payload[1] - - name = _15693_CMDS.get(cmd) or _15693_NXP_CMDS.get(cmd) - if name is None: - return f"UNKNOWN CMD 0x{cmd:02X}" - - blk_off = _block_offset(flags) - - if cmd == 0x01: # INVENTORY - if len(payload) > 2 and payload[2] > 0: - return f"{name} mask={payload[2]}" - return name - - if cmd in (0x20, 0x21): # READ/WRITE SINGLE - if len(payload) > blk_off: - block = payload[blk_off] - if cmd == 0x21: - data_len = len(payload) - blk_off - 1 - return f"{name} #{block} [{data_len}B]" - return f"{name} #{block}" - return name - - if cmd in (0x23, 0x2C): # READ MULTIPLE / GET MULTIPLE BLOCK SECURITY - if len(payload) > blk_off + 1: - start = payload[blk_off] - count = payload[blk_off + 1] - return f"{name} #{start}+{count}" - return name - - if cmd == 0x2B: # GET SYSTEM INFO - return name - - if cmd == 0xB3 and len(payload) >= 4: # NXP SET PASSWORD - return f"{name} id={payload[3]}" - - return name - - -def decode_15693_response(payload: bytes) -> str | None: - """Decode an ISO 15693 response frame into annotation.""" - if len(payload) < 1: - return None - - flags = payload[0] - - if flags & 0x01: # Error - if len(payload) >= 2: - code = payload[1] - desc = _15693_ERRORS.get(code, "") - return f"ERROR 0x{code:02X} {desc}" if desc else f"ERROR 0x{code:02X}" - return "ERROR" - - data = payload[1:] - if len(data) == 0: - return "OK" - - # Inventory response: dsfid(1) + uid(8) = 9 bytes - if len(data) == 9: - uid_msb = bytes(reversed(data[1:9])) - return f"OK INVENTORY UID={uid_msb.hex().upper()}" - - # Try NDEF decode on response data - ndef_ann = decode_ndef_annotation(data) - if ndef_ann: - return f"OK {ndef_ann}" - - return f"OK [{len(data)}B]" - - -def decode_15693(direction: int, payload: bytes) -> str | None: - """Decode an ISO 15693 frame for sniff trace.""" - if direction == 0: - return decode_15693_request(payload) - else: - return decode_15693_response(payload) - - -# ---- ANSI colors ---- -_C_CYAN = "\033[36m" -_C_YELLOW = "\033[33m" -_C_DIM = "\033[2m" -_C_CRC = "\033[1;37m" -_C_RED = "\033[31m" -_C_RESET = "\033[0m" - - -def _color(code: str, text: str, is_tty: bool = True) -> str: - if not is_tty or not code: - return text - return f"{code}{text}{_C_RESET}" - - -def _wrap_annotation(annotation: str, avail: int) -> list[str]: - if len(annotation) <= avail: - return [annotation] - if " | " in annotation: - parts = annotation.split(" | ") - lines = [] - current = parts[0] - for part in parts[1:]: - candidate = f"{current} | {part}" - if len(candidate) <= avail: - current = candidate - else: - lines.append(current) - current = part - lines.append(current) - return lines - return [annotation[:avail - 3] + "..."] - - -def format_sniff_line(entry: dict, width: int = 120, is_tty: bool = True) -> str: - """Format a single tracelog entry as a colored sniff line. - - Uses cyan for reader→tag, yellow for tag→reader, both hex and annotations. - """ - direction = 1 if entry["is_response"] else 0 - payload = entry["data"] - - if direction == 0: - arrow = "Reader → Tag:" - dir_color = _C_CYAN - else: - arrow = "Tag → Reader:" - dir_color = _C_YELLOW - - prefix_plain = f"[Snf] {arrow} " - prefix_len = len(prefix_plain) - prefix_colored = _color(_C_DIM, "[Snf]", is_tty) + " " + _color(dir_color, arrow, is_tty) + " " - - # Split payload into data + CRC (last 2 bytes) - if len(payload) > 2: - data = payload[:-2] - crc = payload[-2:] - else: - data = payload - crc = b"" - - data_hex = " ".join(f"{b:02X}" for b in data) - crc_hex = " ".join(f"{b:02X}" for b in crc) - full_hex = f"{data_hex} {crc_hex}" if crc_hex else data_hex - - # Decode annotation - annotation = decode_15693(direction, data) - ann_color = dir_color - - avail = width - prefix_len - if annotation: - one_line = f"{full_hex} {annotation}" - else: - one_line = full_hex - - if len(one_line) <= avail: - hex_colored = _color(_C_DIM, data_hex, is_tty) - if crc_hex: - hex_colored += " " + _color(_C_CRC, crc_hex, is_tty) - if annotation: - ann_colored = _color(ann_color, annotation, is_tty) - gap = max(avail - len(full_hex) - len(annotation), 2) - return f"{prefix_colored}{hex_colored}{' ' * gap}{ann_colored}" - return f"{prefix_colored}{hex_colored}" - - # Multi-line: wrap hex - lines = [] - tokens = full_hex.split(" ") - current = "" - for tok in tokens: - candidate = f"{current} {tok}" if current else tok - if len(candidate) <= avail: - current = candidate - else: - if current: - lines.append(current) - current = tok - if current: - lines.append(current) - - pad = " " * prefix_len - parts = [] - for i, hl in enumerate(lines): - colored_hl = _color(_C_DIM, hl, is_tty) - if i == 0: - parts.append(f"{prefix_colored}{colored_hl}") - else: - parts.append(f"{pad}{colored_hl}") - if annotation: - ann_lines = _wrap_annotation(annotation, avail) - for al in ann_lines: - ann_pad = max(avail - len(al), 0) - parts.append(f"{pad}{' ' * ann_pad}{_color(ann_color, al, is_tty)}") - - return "\n".join(parts) - class HF15Commands: """ISO 15693 commands.""" @@ -562,54 +114,17 @@ class HF15Commands: # Debug print or other message — discard and keep waiting async def download_trace(self) -> list[dict]: - """Download and parse the trace buffer from the device. - - Returns list of trace entries, each with: - timestamp, duration, is_response, data, parity - """ - # First download to discover trace length - raw, trace_len = await self._t.download_bigbuf( - offset=0, length=PM3_CMD_DATA_SIZE, timeout=4.0) - - if trace_len == 0: - return [] - - # If trace is larger than first chunk, download the full thing - if trace_len > PM3_CMD_DATA_SIZE: - raw, trace_len = await self._t.download_bigbuf( - offset=0, length=trace_len, timeout=10.0) - - return parse_tracelog(raw[:trace_len]) + """Download and parse the trace buffer from the device.""" + from ..sniff.session import SniffSession + session = SniffSession(self._t) + return await session._download_trace() async def sniff_decoded(self, timeout: float = 60.0) -> list[dict]: """Sniff ISO15693 traffic, download trace, decode and print. - Blocks until button press or timeout. Then downloads the trace, - decodes each frame, and prints formatted output. - - Returns the parsed trace entries. + Convenience method — delegates to SniffSession.iso15(). + Prefer using SniffSession directly for new code. """ - import os - try: - width = os.get_terminal_size().columns - except (OSError, ValueError): - width = 120 - is_tty = sys.stdout.isatty() - - print("[Snf] Sniffing ISO 15693... press PM3 button to stop.") - result = await self.sniff(timeout=timeout) - print(f"[Snf] Sniff ended (status={result['status']})") - - print("[Snf] Downloading trace...") - entries = await self.download_trace() - - if not entries: - print("[Snf] No trace data captured.") - return [] - - print(f"[Snf] {len(entries)} frames captured:\n") - for entry in entries: - line = format_sniff_line(entry, width=width, is_tty=is_tty) - print(line) - - return entries + from ..sniff.session import SniffSession + session = SniffSession(self._t) + return await session.iso15(timeout=timeout) diff --git a/pm3py/sniff/__init__.py b/pm3py/sniff/__init__.py index 68db8db..4e62135 100644 --- a/pm3py/sniff/__init__.py +++ b/pm3py/sniff/__init__.py @@ -1 +1,14 @@ """pm3py.sniff — Sniff sessions, trace parsing, protocol decoders, formatting.""" +from .session import SniffSession +from .trace import parse_tracelog, TRACELOG_HDR_SIZE +from .decode_iso15 import decode_15693, decode_15693_request, decode_15693_response +from .ndef import decode_ndef_annotation +from .format import format_sniff_line + +__all__ = [ + "SniffSession", + "parse_tracelog", "TRACELOG_HDR_SIZE", + "decode_15693", "decode_15693_request", "decode_15693_response", + "decode_ndef_annotation", + "format_sniff_line", +] diff --git a/pm3py/sniff/decode_iso15.py b/pm3py/sniff/decode_iso15.py new file mode 100644 index 0000000..d9da791 --- /dev/null +++ b/pm3py/sniff/decode_iso15.py @@ -0,0 +1,146 @@ +"""ISO 15693 request/response decoders and command tables.""" +from .ndef import decode_ndef_annotation + +# ---- ISO 15693 command names (for sniff annotation) ---- +_15693_CMDS = { + 0x01: "INVENTORY", + 0x02: "STAY QUIET", + 0x20: "READ SINGLE BLOCK", + 0x21: "WRITE SINGLE BLOCK", + 0x22: "LOCK BLOCK", + 0x23: "READ MULTIPLE BLOCKS", + 0x24: "WRITE MULTIPLE BLOCKS", + 0x25: "SELECT", + 0x26: "RESET TO READY", + 0x27: "WRITE AFI", + 0x28: "LOCK AFI", + 0x29: "WRITE DSFID", + 0x2A: "LOCK DSFID", + 0x2B: "GET SYSTEM INFO", + 0x2C: "GET MULTIPLE BLOCK SECURITY", +} + +_15693_NXP_CMDS = { + 0xA0: "NXP INVENTORY READ", + 0xA1: "NXP FAST INVENTORY READ", + 0xA2: "NXP SET EAS", + 0xA3: "NXP RESET EAS", + 0xA4: "NXP LOCK EAS", + 0xA5: "NXP EAS ALARM", + 0xA6: "NXP PASSWORD PROTECT EAS/AFI", + 0xA7: "NXP WRITE EAS ID", + 0xAB: "NXP GET SYSTEM INFO", + 0xB2: "NXP GET RANDOM", + 0xB3: "NXP SET PASSWORD", + 0xB4: "NXP WRITE PASSWORD", + 0xB5: "NXP LOCK PASSWORD", + 0xB6: "NXP PROTECT PAGE", + 0xB7: "NXP LOCK PAGE PROTECTION", + 0xB9: "NXP DESTROY", + 0xBA: "NXP ENABLE PRIVACY", + 0xBB: "NXP 64-BIT PASSWORD PROTECTION", + 0xBD: "NXP READ SIGNATURE", +} + +_15693_ERRORS = { + 0x01: "not supported", + 0x02: "not recognized", + 0x10: "block not available", + 0x11: "block already locked", + 0x12: "block locked", + 0x13: "block not written", + 0x14: "block not locked", +} + +_FLAG_INVENTORY = 0x04 +_FLAG_ADDRESS = 0x20 + + +def _block_offset(flags: int) -> int: + """Byte offset of block number after flags+cmd.""" + if not (flags & _FLAG_INVENTORY) and (flags & _FLAG_ADDRESS): + return 10 # flags(1) + cmd(1) + uid(8) + return 2 # flags(1) + cmd(1) + + +def decode_15693_request(payload: bytes) -> str | None: + """Decode an ISO 15693 request frame into annotation.""" + if len(payload) < 2: + return None + + flags = payload[0] + cmd = payload[1] + + name = _15693_CMDS.get(cmd) or _15693_NXP_CMDS.get(cmd) + if name is None: + return f"UNKNOWN CMD 0x{cmd:02X}" + + blk_off = _block_offset(flags) + + if cmd == 0x01: # INVENTORY + if len(payload) > 2 and payload[2] > 0: + return f"{name} mask={payload[2]}" + return name + + if cmd in (0x20, 0x21): # READ/WRITE SINGLE + if len(payload) > blk_off: + block = payload[blk_off] + if cmd == 0x21: + data_len = len(payload) - blk_off - 1 + return f"{name} #{block} [{data_len}B]" + return f"{name} #{block}" + return name + + if cmd in (0x23, 0x2C): # READ MULTIPLE / GET MULTIPLE BLOCK SECURITY + if len(payload) > blk_off + 1: + start = payload[blk_off] + count = payload[blk_off + 1] + return f"{name} #{start}+{count}" + return name + + if cmd == 0x2B: # GET SYSTEM INFO + return name + + if cmd == 0xB3 and len(payload) >= 4: # NXP SET PASSWORD + return f"{name} id={payload[3]}" + + return name + + +def decode_15693_response(payload: bytes) -> str | None: + """Decode an ISO 15693 response frame into annotation.""" + if len(payload) < 1: + return None + + flags = payload[0] + + if flags & 0x01: # Error + if len(payload) >= 2: + code = payload[1] + desc = _15693_ERRORS.get(code, "") + return f"ERROR 0x{code:02X} {desc}" if desc else f"ERROR 0x{code:02X}" + return "ERROR" + + data = payload[1:] + if len(data) == 0: + return "OK" + + # Inventory response: dsfid(1) + uid(8) = 9 bytes + if len(data) == 9: + uid_msb = bytes(reversed(data[1:9])) + return f"OK INVENTORY UID={uid_msb.hex().upper()}" + + # Try NDEF decode on response data + ndef_ann = decode_ndef_annotation(data) + if ndef_ann: + return f"OK {ndef_ann}" + + return f"OK [{len(data)}B]" + + +def decode_15693(direction: int, payload: bytes) -> str | None: + """Decode an ISO 15693 frame for sniff trace.""" + if direction == 0: + return decode_15693_request(payload) + else: + return decode_15693_response(payload) diff --git a/pm3py/sniff/format.py b/pm3py/sniff/format.py new file mode 100644 index 0000000..fec3a30 --- /dev/null +++ b/pm3py/sniff/format.py @@ -0,0 +1,118 @@ +"""ANSI color formatting and sniff line output.""" +from .decode_iso15 import decode_15693 + +# ---- ANSI colors ---- +_C_CYAN = "\033[36m" +_C_YELLOW = "\033[33m" +_C_DIM = "\033[2m" +_C_CRC = "\033[1;37m" +_C_RED = "\033[31m" +_C_RESET = "\033[0m" + + +def _color(code: str, text: str, is_tty: bool = True) -> str: + if not is_tty or not code: + return text + return f"{code}{text}{_C_RESET}" + + +def _wrap_annotation(annotation: str, avail: int) -> list[str]: + if len(annotation) <= avail: + return [annotation] + if " | " in annotation: + parts = annotation.split(" | ") + lines = [] + current = parts[0] + for part in parts[1:]: + candidate = f"{current} | {part}" + if len(candidate) <= avail: + current = candidate + else: + lines.append(current) + current = part + lines.append(current) + return lines + return [annotation[:avail - 3] + "..."] + + +def format_sniff_line(entry: dict, width: int = 120, is_tty: bool = True) -> str: + """Format a single tracelog entry as a colored sniff line. + + Uses cyan for reader->tag, yellow for tag->reader, both hex and annotations. + """ + direction = 1 if entry["is_response"] else 0 + payload = entry["data"] + + if direction == 0: + arrow = "Reader → Tag:" + dir_color = _C_CYAN + else: + arrow = "Tag → Reader:" + dir_color = _C_YELLOW + + prefix_plain = f"[Snf] {arrow} " + prefix_len = len(prefix_plain) + prefix_colored = _color(_C_DIM, "[Snf]", is_tty) + " " + _color(dir_color, arrow, is_tty) + " " + + # Split payload into data + CRC (last 2 bytes) + if len(payload) > 2: + data = payload[:-2] + crc = payload[-2:] + else: + data = payload + crc = b"" + + data_hex = " ".join(f"{b:02X}" for b in data) + crc_hex = " ".join(f"{b:02X}" for b in crc) + full_hex = f"{data_hex} {crc_hex}" if crc_hex else data_hex + + # Decode annotation + annotation = decode_15693(direction, data) + ann_color = dir_color + + avail = width - prefix_len + if annotation: + one_line = f"{full_hex} {annotation}" + else: + one_line = full_hex + + if len(one_line) <= avail: + hex_colored = _color(_C_DIM, data_hex, is_tty) + if crc_hex: + hex_colored += " " + _color(_C_CRC, crc_hex, is_tty) + if annotation: + ann_colored = _color(ann_color, annotation, is_tty) + gap = max(avail - len(full_hex) - len(annotation), 2) + return f"{prefix_colored}{hex_colored}{' ' * gap}{ann_colored}" + return f"{prefix_colored}{hex_colored}" + + # Multi-line: wrap hex + lines = [] + tokens = full_hex.split(" ") + current = "" + for tok in tokens: + candidate = f"{current} {tok}" if current else tok + if len(candidate) <= avail: + current = candidate + else: + if current: + lines.append(current) + current = tok + if current: + lines.append(current) + + pad = " " * prefix_len + parts = [] + for i, hl in enumerate(lines): + colored_hl = _color(_C_DIM, hl, is_tty) + if i == 0: + parts.append(f"{prefix_colored}{colored_hl}") + else: + parts.append(f"{pad}{colored_hl}") + if annotation: + ann_lines = _wrap_annotation(annotation, avail) + for al in ann_lines: + ann_pad = max(avail - len(al), 0) + parts.append(f"{pad}{' ' * ann_pad}{_color(ann_color, al, is_tty)}") + + return "\n".join(parts) diff --git a/pm3py/sniff/ndef.py b/pm3py/sniff/ndef.py new file mode 100644 index 0000000..be06410 --- /dev/null +++ b/pm3py/sniff/ndef.py @@ -0,0 +1,152 @@ +"""NDEF TLV/record decode for trace annotation.""" + +_NDEF_MAGIC = 0xE1 + +_URI_PREFIXES = { + 0x00: "", 0x01: "http://www.", 0x02: "https://www.", + 0x03: "http://", 0x04: "https://", 0x05: "tel:", 0x06: "mailto:", +} + +_TNF_EMPTY = 0x00 +_TNF_WELL_KNOWN = 0x01 +_TNF_MEDIA = 0x02 +_TNF_URI = 0x03 +_TNF_EXTERNAL = 0x04 +_MAX_ANN_TEXT = 40 + + +def decode_ndef_annotation(data: bytes) -> str | None: + """Decode NDEF TLV data into a human-readable annotation string. + + Handles CC (capability container) blocks and NDEF message TLVs. + Returns None if the data doesn't contain recognizable NDEF content. + """ + if len(data) < 3: + return None + + parts = [] + offset = 0 + + if data[0] == _NDEF_MAGIC and len(data) >= 4: + ver_major = (data[1] >> 6) & 0x03 + ver_minor = (data[1] >> 4) & 0x03 + mlen = data[2] + features = data[3] + cc_str = f"CC v{ver_major}.{ver_minor} MLEN={mlen}" + if features & 0x01: + cc_str += " MBREAD" + parts.append(cc_str) + offset = 4 + if offset >= len(data): + return cc_str + + while offset < len(data): + tlv_type = data[offset] + offset += 1 + if tlv_type == 0x00: + continue + if tlv_type == 0xFE: + break + if tlv_type != 0x03: + break + + if offset >= len(data): + break + tlv_len = data[offset] + offset += 1 + if tlv_len == 0xFF: + if offset + 2 > len(data): + break + tlv_len = (data[offset] << 8) | data[offset + 1] + offset += 2 + + if tlv_len == 0: + parts.append("NDEF empty") + continue + + msg_end = min(offset + tlv_len, len(data)) + records = _parse_ndef_records(data[offset:msg_end]) + for rec in records: + parts.append(rec) + offset = msg_end + + return " | ".join(parts) if parts else None + + +def _parse_ndef_records(msg: bytes) -> list[str]: + """Parse NDEF records from a message payload.""" + records = [] + offset = 0 + while offset < len(msg): + if offset + 3 > len(msg): + break + flags = msg[offset] + tnf = flags & 0x07 + sr = bool(flags & 0x10) + il = bool(flags & 0x08) + type_len = msg[offset + 1] + offset += 2 + + if sr: + if offset >= len(msg): + break + payload_len = msg[offset] + offset += 1 + else: + if offset + 4 > len(msg): + break + payload_len = int.from_bytes(msg[offset:offset + 4], "big") + offset += 4 + + id_len = 0 + if il: + if offset >= len(msg): + break + id_len = msg[offset] + offset += 1 + + if offset + type_len > len(msg): + break + rec_type = msg[offset:offset + type_len] + offset += type_len + offset += id_len + + if offset + payload_len > len(msg): + payload = msg[offset:] + else: + payload = msg[offset:offset + payload_len] + offset += payload_len + + records.append(_decode_ndef_record(tnf, rec_type, payload)) + return records + + +def _decode_ndef_record(tnf: int, rec_type: bytes, payload: bytes) -> str: + """Decode a single NDEF record into an annotation string.""" + if tnf == _TNF_WELL_KNOWN: + if rec_type == b"T" and len(payload) >= 1: + lang_len = payload[0] & 0x3F + lang = payload[1:1 + lang_len].decode("ascii", errors="replace") + text = payload[1 + lang_len:].decode("utf-8", errors="replace") + if len(text) > _MAX_ANN_TEXT: + text = text[:_MAX_ANN_TEXT] + "..." + return f'NDEF Text "{lang}" "{text}"' + if rec_type == b"U" and len(payload) >= 1: + prefix = _URI_PREFIXES.get(payload[0], "") + suffix = payload[1:].decode("utf-8", errors="replace") + uri = prefix + suffix + if len(uri) > _MAX_ANN_TEXT: + uri = uri[:_MAX_ANN_TEXT] + "..." + return f"NDEF URI {uri}" + return f"NDEF WK type={rec_type!r} [{len(payload)}B]" + if tnf == _TNF_MEDIA: + mime = rec_type.decode("ascii", errors="replace") + if len(mime) > 30: + mime = mime[:30] + "..." + return f"NDEF MIME {mime} [{len(payload)}B]" + if tnf == _TNF_EXTERNAL: + ext_type = rec_type.decode("ascii", errors="replace") + return f"NDEF EXT {ext_type} [{len(payload)}B]" + if tnf == _TNF_EMPTY: + return "NDEF empty" + return f"NDEF TNF={tnf} [{len(payload)}B]" diff --git a/pm3py/sniff/session.py b/pm3py/sniff/session.py new file mode 100644 index 0000000..0d1d086 --- /dev/null +++ b/pm3py/sniff/session.py @@ -0,0 +1,75 @@ +"""Sniff sessions — start/stop/download per protocol.""" +import sys + +from ..core.protocol import Cmd, PM3_CMD_DATA_SIZE +from ..core.transport import PM3Transport, encode_ng_frame +from .trace import parse_tracelog +from .format import format_sniff_line + + +class SniffSession: + """Sniff session that manages start/download/decode lifecycle. + + Takes a PM3Transport (not the client), since sniff has its own + lifecycle that doesn't fit the stateless command pattern. + """ + + def __init__(self, transport: PM3Transport): + self._t = transport + + async def _sniff_15(self, timeout: float = 60.0) -> dict: + """Start ISO 15693 sniff. Blocks until button press or timeout.""" + async with self._t._lock: + raw = encode_ng_frame(Cmd.HF_ISO15693_SNIFF) + await self._t.send_frame(raw) + while True: + resp = await self._t.read_response(timeout=timeout) + if resp.cmd == Cmd.HF_ISO15693_SNIFF: + return {"status": resp.status} + + async def _download_trace(self) -> list[dict]: + """Download and parse the trace buffer from the device.""" + raw, trace_len = await self._t.download_bigbuf( + offset=0, length=PM3_CMD_DATA_SIZE, timeout=4.0) + + if trace_len == 0: + return [] + + if trace_len > PM3_CMD_DATA_SIZE: + raw, trace_len = await self._t.download_bigbuf( + offset=0, length=trace_len, timeout=10.0) + + return parse_tracelog(raw[:trace_len]) + + async def iso15(self, timeout: float = 60.0) -> list[dict]: + """Sniff ISO 15693 traffic, download trace, decode and print. + + Blocks until button press or timeout. Then downloads the trace, + decodes each frame, and prints formatted output. + + Returns the parsed trace entries. + """ + import os + try: + width = os.get_terminal_size().columns + except (OSError, ValueError): + width = 120 + is_tty = sys.stdout.isatty() + + print("[Snf] Sniffing ISO 15693... press PM3 button to stop.") + result = await self._sniff_15(timeout=timeout) + print(f"[Snf] Sniff ended (status={result['status']})") + + print("[Snf] Downloading trace...") + entries = await self._download_trace() + + if not entries: + print("[Snf] No trace data captured.") + return [] + + print(f"[Snf] {len(entries)} frames captured:\n") + for entry in entries: + line = format_sniff_line(entry, width=width, is_tty=is_tty) + print(line) + + return entries diff --git a/pm3py/sniff/trace.py b/pm3py/sniff/trace.py new file mode 100644 index 0000000..a5bbc80 --- /dev/null +++ b/pm3py/sniff/trace.py @@ -0,0 +1,41 @@ +"""Protocol-agnostic trace buffer parsing.""" +import struct + +# tracelog_hdr_t size: timestamp(4) + duration(2) + data_len_flags(2) = 8 +TRACELOG_HDR_SIZE = 8 + + +def parse_tracelog(raw: bytes) -> list[dict]: + """Parse PM3 trace buffer into a list of trace entries. + + Each entry: {timestamp, duration, is_response, data, parity} + """ + entries = [] + offset = 0 + while offset + TRACELOG_HDR_SIZE <= len(raw): + timestamp, duration, data_len_flags = struct.unpack_from(" len(raw): + break + + data = raw[offset:offset + data_len] + offset += data_len + + parity_len = (data_len + 7) // 8 if data_len > 0 else 0 + parity = raw[offset:offset + parity_len] if parity_len else b"" + offset += parity_len + + entries.append({ + "timestamp": timestamp, + "duration": duration, + "is_response": is_response, + "data": data, + "parity": parity, + }) + return entries