diff --git a/pm3py/sim/trace_fmt.py b/pm3py/sim/trace_fmt.py index 477c9b1..69d5743 100644 --- a/pm3py/sim/trace_fmt.py +++ b/pm3py/sim/trace_fmt.py @@ -1,560 +1,4 @@ -"""Trace formatter — colored, decoded, column-wrapped trace output.""" -from __future__ import annotations - -import os -import signal -import sys - -# ---- ISO 15693 flags ---- -_15693_FLAG_INVENTORY = 0x04 -_15693_FLAG_ADDRESS = 0x20 - -# ---- ISO 15693 command names ---- -_15693_CMDS = { - 0x01: "INVENTORY", - 0x02: "STAY QUIET", - 0x20: "READ SINGLE BLOCK", - 0x21: "WRITE SINGLE BLOCK", - 0x23: "READ MULTIPLE BLOCK", - 0x26: "RESET TO READY", - 0x2B: "GET SYSTEM INFO", - 0x2C: "GET MULTIPLE BLOCK SECURITY", -} - -# NXP custom commands (manufacturer code 0x04, per SL2S2602 datasheet) -_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", -} - - -def _15693_block_offset(flags: int) -> int: - """Return the byte offset of the block number field after flags+cmd.""" - is_inventory = bool(flags & _15693_FLAG_INVENTORY) - if not is_inventory and (flags & _15693_FLAG_ADDRESS): - return 10 # flags(1) + cmd(1) + uid(8) - return 2 # flags(1) + cmd(1) - - -def decode_15693(direction: int, payload: bytes) -> str | None: - """Decode an ISO 15693 frame into a human-readable annotation. - - Args: - direction: 0 = reader->tag, 1 = tag->reader - payload: raw frame bytes - - Returns: - Annotation string or None if unrecognized. - """ - if direction == 0: - return _decode_15693_request(payload) - else: - return _decode_15693_response(payload) - - -def decode_15693_nxp(direction: int, payload: bytes) -> str | None: - """Decode an NXP custom ISO 15693 command. - - Args: - direction: 0 = reader->tag, 1 = tag->reader - payload: raw frame bytes - - Returns: - Annotation string or None if not an NXP custom command. - """ - if direction != 0: - return None - if len(payload) < 2: - return None - cmd = payload[1] - name = _15693_NXP_CMDS.get(cmd) - if name is not None: - return _annotate_nxp_request(name, cmd, payload) - return None - - -def _decode_15693_request(payload: bytes) -> str | None: - if len(payload) < 2: - return None - - flags = payload[0] - cmd = payload[1] - - # Standard commands - name = _15693_CMDS.get(cmd) - if name is not None: - return _annotate_15693_request(name, cmd, flags, payload) - - # NXP custom commands - nxp_name = _15693_NXP_CMDS.get(cmd) - if nxp_name is not None: - return _annotate_nxp_request(nxp_name, cmd, payload) - - return f"UNKNOWN CMD 0x{cmd:02X}" - - -def _annotate_15693_request(name: str, cmd: int, flags: int, payload: bytes) -> str: - blk_off = _15693_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 BLOCK - 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 - - return name - - -def _annotate_nxp_request(name: str, cmd: int, payload: bytes) -> str: - if cmd == 0xB3 and len(payload) >= 4: # SET PASSWORD - return f"{name} id={payload[3]}" - if cmd == 0xB6 and len(payload) >= 4: # PROTECT PAGE - pp = payload[3] - cond = payload[4] if len(payload) >= 5 else 0 - return f"{name} pp={pp} cond=0x{cond:02X}" - if cmd == 0xBA and len(payload) >= 7: # ENABLE PRIVACY - return f"{name} [XOR pwd]" - return name - - -_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", -} - - -def _decode_15693_response(payload: bytes) -> str | None: - 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, "") - if desc: - return f"ERROR 0x{code:02X} {desc}" - return f"ERROR 0x{code:02X}" - return "ERROR" - - # Success — try to identify the response type - 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 - from pm3py.trace.ndef import decode_ndef_annotation - ndef_ann = decode_ndef_annotation(data) - if ndef_ann: - return f"OK {ndef_ann}" - - # GET SYSTEM INFO response: info_flags(1) + UID(8) + DSFID(1) + AFI(1) + memsize(2) + ic_ref(1) - if len(data) >= 14 and data[0] & 0x0F == 0x0F: # all info flags set - uid_msb = bytes(reversed(data[1:9])) - ic_ref = data[13] - ic_name = _identify_nxp_ic(uid_msb, ic_ref) - blocks = data[11] + 1 - blk_size = data[12] + 1 - ann = f"OK SYS blocks={blocks}×{blk_size}" - if ic_name: - ann += f" [{ic_name}]" - return ann - - # Generic data response - return f"OK [{len(data)}B]" - - -def _identify_nxp_ic(uid_msb: bytes, ic_ref: int) -> str | None: - """Identify NXP IC from UID type indicator bits + ic_reference. - - uid_msb: 8-byte UID in MSB-first order (uid[0]=E0, uid[1]=mfg, uid[2]=tag_type) - ic_ref: ic_reference byte from GET SYSTEM INFO - """ - if len(uid_msb) < 4 or uid_msb[0] != 0xE0 or uid_msb[1] != 0x04: - return None # not NXP - - tag_type = uid_msb[2] - type_ind = uid_msb[3] - - if tag_type == 0x02: - return "ICODE SLIX2" - - if tag_type == 0x01: - # Type indicator bits 37:36 at byte[3] bits 4:3 - bits_37_36 = (type_ind >> 3) & 0x03 - # Extended bits 39:36 at byte[3] bits 6:3 - bits_39_36 = (type_ind >> 3) & 0x0F - - if bits_39_36 == 0x04: # 0100 - return "ICODE 3" - if bits_37_36 == 0x03: # 11 - # DNA and NTAG 5 share this — differentiate by memory size - # DNA = 64 blocks, NTAG 5 Switch = 128, Link/Boost = 512 - return "ICODE DNA / NTAG 5" - if bits_37_36 == 0x02: # 10 - return "ICODE SLIX" - if bits_37_36 == 0x00: - return "ICODE SLI" - - return None - - -# ---- ISO 14443-A constants ---- -_14A_CL_MAP = {0x93: "CL1", 0x95: "CL2", 0x97: "CL3"} - - -def decode_14443a(direction: int, payload: bytes) -> str | None: - """Decode an ISO 14443-A frame into a human-readable annotation.""" - if not payload: - return None - if direction == 0: - return _decode_14443a_request(payload) - else: - return _decode_14443a_response(payload) - - -def _decode_14443a_request(payload: bytes) -> str | None: - b0 = payload[0] - - # Short frames (single byte) - if b0 == 0x26: - return "REQA" - if b0 == 0x52: - return "WUPA" - - # HLTA - if b0 == 0x50 and len(payload) >= 2: - return "HLTA" - - # Anticollision / Select - if b0 in _14A_CL_MAP and len(payload) >= 2: - cl = _14A_CL_MAP[b0] - nvb = payload[1] - if nvb == 0x20: - return f"ANTICOL {cl}" - if nvb == 0x70: - return f"SELECT {cl}" - return f"ANTICOL {cl} nvb={nvb:02X}" - - # RATS - if b0 == 0xE0: - return "RATS" - - # ISO-DEP I-block - if b0 & 0xE2 == 0x02: - bn = b0 & 0x01 - return f"I-BLOCK({bn})" - - # R-ACK - if b0 & 0xF6 == 0xA2: - bn = b0 & 0x01 - return f"R-ACK({bn})" - - # R-NAK - if b0 & 0xF6 == 0xB2: - bn = b0 & 0x01 - return f"R-NAK({bn})" - - # S(DESELECT) - if b0 == 0xC2: - return "S(DESELECT)" - - # S(WTX) - if b0 == 0xF2: - return "S(WTX)" - - return None - - -def _decode_14443a_response(payload: bytes) -> str | None: - if not payload: - return None - - b0 = payload[0] - - # I-block response - if b0 & 0xE2 == 0x02: - bn = b0 & 0x01 - return f"I-BLOCK({bn})" - - # ATQA (2 bytes) - if len(payload) == 2 and b0 & 0xF0 == 0x00: - return f"ATQA {payload[0]:02X} {payload[1]:02X}" - - # SAK (1 byte) - if len(payload) == 1: - return f"SAK {b0:02X}" - - # ATS (first byte = length, length >= 2) - if len(payload) >= 2 and payload[0] == len(payload): - return f"ATS [{len(payload)}]" - - return None - - -# ---- ANSI colors ---- -_C_CYAN = "\033[36m" -_C_YELLOW = "\033[33m" -_C_MAGENTA = "\033[35m" -_C_GREEN = "\033[32m" -_C_DIM = "\033[2m" -_C_CRC = "\033[1;37m" # bold white — CRC bytes -_C_RESET = "\033[0m" - -_MODE_TAGS = { - "sim": ("[Sim]", _C_MAGENTA), - "reader": ("[Rdr]", _C_CYAN), - "sniff": ("[Snf]", ""), -} - - -class TraceFormatter: - """Colored, decoded, column-wrapped trace output. - - Args: - mode: "sim", "reader", or "sniff" - decoder: Callable[[int, bytes], str | None] for annotation, or None - width: override terminal width (None = auto-detect) - is_tty: override TTY detection (None = auto-detect) - """ - - def __init__(self, mode: str = "sim", decoder=None, - width: int | None = None, is_tty: bool | None = None, - crc_len: int = 0): - self._mode = mode - self._decoder = decoder - self._crc_len = crc_len - self._width = width or self._detect_width() - self._is_tty = is_tty if is_tty is not None else sys.stdout.isatty() - self._line_count = 0 - - # Try to register SIGWINCH for dynamic resize - if width is None: - try: - signal.signal(signal.SIGWINCH, self._on_resize) - except (OSError, ValueError): - pass # not main thread or not Unix - - def _detect_width(self) -> int: - try: - return os.get_terminal_size().columns - except (OSError, ValueError): - return 80 - - def _on_resize(self, signum, frame): - self._width = self._detect_width() - - def _color(self, code: str, text: str) -> str: - if not self._is_tty or not code: - return text - return f"{code}{text}{_C_RESET}" - - def format(self, direction: int, payload: bytes, crc_fail: bool = False) -> str: - """Format a trace line with colors, decoding, and wrapping.""" - # Re-check width periodically if no SIGWINCH - self._line_count += 1 - if self._line_count % 10 == 0: - try: - self._width = self._detect_width() - except Exception: - pass - - # Mode tag - mode_tag, mode_color = _MODE_TAGS.get(self._mode, ("[???]", "")) - - # Direction + determine if this is "our" side - # sim mode: we are the tag (direction=1 is ours) - # reader mode: we are the reader (direction=0 is ours) - # sniff mode: neither side is ours - if direction == 0: - arrow = "Reader \u2192 Tag:" - dir_color = _C_CYAN - is_ours = (self._mode == "reader") - else: - arrow = "Tag \u2192 Reader:" - dir_color = _C_YELLOW - is_ours = (self._mode == "sim") - - # Build prefix (uncolored for width calc) - prefix_plain = f"{mode_tag} {arrow} " - prefix_len = len(prefix_plain) - - # Colored prefix — dim our side - if is_ours: - dir_color = _C_DIM + dir_color - prefix_colored = self._color(mode_color, mode_tag) + " " + self._color(dir_color, arrow) + " " - - # Split payload into data and CRC (both directions include CRC) - if self._crc_len > 0 and len(payload) > self._crc_len: - data = payload[:-self._crc_len] - crc = payload[-self._crc_len:] - else: - data = payload - crc = b"" - - # Hex bytes (space-separated, uppercase) - 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 via callable (on data only, no CRC) - annotation = None - if self._decoder is not None: - annotation = self._decoder(direction, data) - if crc_fail: - crc_note = "BAD CRC" - annotation = f"{annotation} {crc_note}" if annotation else crc_note - # Annotations use direction color (cyan=reader, yellow=tag) - ann_color = (_C_DIM + dir_color) if is_ours else dir_color - if crc_fail: - ann_color = "\033[31m" # red for CRC failures - - # Layout: determine if everything fits on one line - avail = self._width - prefix_len - if annotation: - one_line = f"{full_hex} {annotation}" - else: - one_line = full_hex - - if len(one_line) <= avail: - # Single line — annotation right-justified - hex_colored = self._color(_C_DIM, data_hex) - if crc_hex: - hex_colored += " " + self._color(_C_CRC, crc_hex) - if annotation: - ann_colored = self._color(ann_color, annotation) - gap = avail - len(full_hex) - len(annotation) - gap = max(gap, 2) - line = f"{prefix_colored}{hex_colored}{' ' * gap}{ann_colored}" - else: - line = f"{prefix_colored}{hex_colored}" - return f"\n{line}" - - # Multi-line: wrap hex, annotation right-justified on its own line - pad = " " * prefix_len - hex_lines = self._wrap_hex(full_hex, avail) - parts = [] - for i, hl in enumerate(hex_lines): - colored_hl = self._color_hex_line(hl, data_hex, crc_hex) - if i == 0: - parts.append(f"{prefix_colored}{colored_hl}") - else: - parts.append(f"{pad}{colored_hl}") - if annotation: - ann_lines = self._wrap_annotation(annotation, avail) - for al in ann_lines: - ann_pad = max(avail - len(al), 0) - parts.append(f"{pad}{' ' * ann_pad}{self._color(ann_color, al)}") - - return "\n" + "\n".join(parts) - - def _color_hex_line(self, line: str, data_hex: str, crc_hex: str) -> str: - """Color a hex line, using dim for data bytes and dark for CRC bytes.""" - if not crc_hex: - return self._color(_C_DIM, line) - # Check if this line contains CRC bytes (they appear at the end) - crc_tokens = crc_hex.split(" ") - line_tokens = line.split(" ") - # Find where CRC starts in this line by matching trailing tokens - crc_start = None - for i in range(len(line_tokens)): - if line_tokens[i:] == crc_tokens[-len(line_tokens) + i:]: - crc_start = i - break - # Check if the tail of this line matches the start/all of CRC - remaining = line_tokens[i:] - if crc_hex.startswith(" ".join(remaining)) or " ".join(remaining) == crc_hex: - crc_start = i - break - if crc_start is not None and crc_start < len(line_tokens): - data_part = " ".join(line_tokens[:crc_start]) - crc_part = " ".join(line_tokens[crc_start:]) - result = "" - if data_part: - result += self._color(_C_DIM, data_part) + " " - result += self._color(_C_CRC, crc_part) - return result - return self._color(_C_DIM, line) - - def _wrap_annotation(self, 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 _wrap_hex(self, hex_str: str, avail: int) -> list[str]: - """Wrap space-separated hex string into lines of at most `avail` chars.""" - tokens = hex_str.split(" ") - lines: list[str] = [] - 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) - return lines or [""] - - def print(self, direction: int, payload: bytes, crc_fail: bool = False) -> None: - """Format and print a trace line to stdout.""" - sys.stdout.write(self.format(direction, payload, crc_fail=crc_fail)) - sys.stdout.flush() +"""Backward compat shim — moved to pm3py.trace.""" +from pm3py.trace.decode_iso15 import decode_15693, decode_15693_nxp # noqa: F401 +from pm3py.trace.decode_iso14a import decode_14443a # noqa: F401 +from pm3py.trace.format import TraceFormatter # noqa: F401 diff --git a/pm3py/trace/__init__.py b/pm3py/trace/__init__.py index 50fe345..aeaaa8b 100644 --- a/pm3py/trace/__init__.py +++ b/pm3py/trace/__init__.py @@ -1,12 +1,14 @@ """pm3py.trace — Firmware trace buffer infrastructure (shared by sniff, sim, reader).""" from .trace import parse_tracelog, TRACELOG_HDR_SIZE -from .decode_iso15 import decode_15693, decode_15693_request, decode_15693_response +from .decode_iso15 import decode_15693, decode_15693_nxp, decode_15693_request, decode_15693_response +from .decode_iso14a import decode_14443a from .ndef import decode_ndef_annotation -from .format import format_sniff_line +from .format import TraceFormatter, format_sniff_line __all__ = [ "parse_tracelog", "TRACELOG_HDR_SIZE", - "decode_15693", "decode_15693_request", "decode_15693_response", + "decode_15693", "decode_15693_nxp", "decode_15693_request", "decode_15693_response", + "decode_14443a", "decode_ndef_annotation", - "format_sniff_line", + "TraceFormatter", "format_sniff_line", ] diff --git a/pm3py/trace/decode_iso14a.py b/pm3py/trace/decode_iso14a.py new file mode 100644 index 0000000..f478c24 --- /dev/null +++ b/pm3py/trace/decode_iso14a.py @@ -0,0 +1,92 @@ +"""ISO 14443-A request/response decoders.""" + +_14A_CL_MAP = {0x93: "CL1", 0x95: "CL2", 0x97: "CL3"} + + +def decode_14443a(direction: int, payload: bytes) -> str | None: + """Decode an ISO 14443-A frame into a human-readable annotation.""" + if not payload: + return None + if direction == 0: + return _decode_14443a_request(payload) + else: + return _decode_14443a_response(payload) + + +def _decode_14443a_request(payload: bytes) -> str | None: + b0 = payload[0] + + # Short frames (single byte) + if b0 == 0x26: + return "REQA" + if b0 == 0x52: + return "WUPA" + + # HLTA + if b0 == 0x50 and len(payload) >= 2: + return "HLTA" + + # Anticollision / Select + if b0 in _14A_CL_MAP and len(payload) >= 2: + cl = _14A_CL_MAP[b0] + nvb = payload[1] + if nvb == 0x20: + return f"ANTICOL {cl}" + if nvb == 0x70: + return f"SELECT {cl}" + return f"ANTICOL {cl} nvb={nvb:02X}" + + # RATS + if b0 == 0xE0: + return "RATS" + + # ISO-DEP I-block + if b0 & 0xE2 == 0x02: + bn = b0 & 0x01 + return f"I-BLOCK({bn})" + + # R-ACK + if b0 & 0xF6 == 0xA2: + bn = b0 & 0x01 + return f"R-ACK({bn})" + + # R-NAK + if b0 & 0xF6 == 0xB2: + bn = b0 & 0x01 + return f"R-NAK({bn})" + + # S(DESELECT) + if b0 == 0xC2: + return "S(DESELECT)" + + # S(WTX) + if b0 == 0xF2: + return "S(WTX)" + + return None + + +def _decode_14443a_response(payload: bytes) -> str | None: + if not payload: + return None + + b0 = payload[0] + + # I-block response + if b0 & 0xE2 == 0x02: + bn = b0 & 0x01 + return f"I-BLOCK({bn})" + + # ATQA (2 bytes) + if len(payload) == 2 and b0 & 0xF0 == 0x00: + return f"ATQA {payload[0]:02X} {payload[1]:02X}" + + # SAK (1 byte) + if len(payload) == 1: + return f"SAK {b0:02X}" + + # ATS (first byte = length, length >= 2) + if len(payload) >= 2 and payload[0] == len(payload): + return f"ATS [{len(payload)}]" + + return None diff --git a/pm3py/trace/decode_iso15.py b/pm3py/trace/decode_iso15.py index d9da791..47b5b93 100644 --- a/pm3py/trace/decode_iso15.py +++ b/pm3py/trace/decode_iso15.py @@ -63,6 +63,46 @@ def _block_offset(flags: int) -> int: return 2 # flags(1) + cmd(1) +def _annotate_nxp_request(name: str, cmd: int, payload: bytes) -> str: + """Annotate NXP custom request with parameter details.""" + if cmd == 0xB3 and len(payload) >= 4: # SET PASSWORD + return f"{name} id={payload[3]}" + if cmd == 0xB6 and len(payload) >= 4: # PROTECT PAGE + pp = payload[3] + cond = payload[4] if len(payload) >= 5 else 0 + return f"{name} pp={pp} cond=0x{cond:02X}" + if cmd == 0xBA and len(payload) >= 7: # ENABLE PRIVACY + return f"{name} [XOR pwd]" + return name + + +def _identify_nxp_ic(uid_msb: bytes, ic_ref: int) -> str | None: + """Identify NXP IC from UID type indicator bits + ic_reference.""" + if len(uid_msb) < 4 or uid_msb[0] != 0xE0 or uid_msb[1] != 0x04: + return None # not NXP + + tag_type = uid_msb[2] + type_ind = uid_msb[3] + + if tag_type == 0x02: + return "ICODE SLIX2" + + if tag_type == 0x01: + bits_37_36 = (type_ind >> 3) & 0x03 + bits_39_36 = (type_ind >> 3) & 0x0F + + if bits_39_36 == 0x04: + return "ICODE 3" + if bits_37_36 == 0x03: + return "ICODE DNA / NTAG 5" + if bits_37_36 == 0x02: + return "ICODE SLIX" + if bits_37_36 == 0x00: + return "ICODE SLI" + + return None + + def decode_15693_request(payload: bytes) -> str | None: """Decode an ISO 15693 request frame into annotation.""" if len(payload) < 2: @@ -71,40 +111,43 @@ def decode_15693_request(payload: bytes) -> str | 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}" + # Standard commands + name = _15693_CMDS.get(cmd) + if name is not None: + blk_off = _block_offset(flags) - 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 == 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 + # NXP custom commands + nxp_name = _15693_NXP_CMDS.get(cmd) + if nxp_name is not None: + return _annotate_nxp_request(nxp_name, cmd, payload) - 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 + return f"UNKNOWN CMD 0x{cmd:02X}" def decode_15693_response(payload: bytes) -> str | None: @@ -135,6 +178,18 @@ def decode_15693_response(payload: bytes) -> str | None: if ndef_ann: return f"OK {ndef_ann}" + # GET SYSTEM INFO response: info_flags(1) + UID(8) + DSFID(1) + AFI(1) + memsize(2) + ic_ref(1) + if len(data) >= 14 and data[0] & 0x0F == 0x0F: + uid_msb = bytes(reversed(data[1:9])) + ic_ref = data[13] + blocks = data[11] + 1 + blk_size = data[12] + 1 + ic_name = _identify_nxp_ic(uid_msb, ic_ref) + ann = f"OK SYS blocks={blocks}\u00d7{blk_size}" + if ic_name: + ann += f" [{ic_name}]" + return ann + return f"OK [{len(data)}B]" @@ -144,3 +199,16 @@ def decode_15693(direction: int, payload: bytes) -> str | None: return decode_15693_request(payload) else: return decode_15693_response(payload) + + +def decode_15693_nxp(direction: int, payload: bytes) -> str | None: + """Decode an NXP custom ISO 15693 command.""" + if direction != 0: + return None + if len(payload) < 2: + return None + cmd = payload[1] + name = _15693_NXP_CMDS.get(cmd) + if name is not None: + return _annotate_nxp_request(name, cmd, payload) + return None diff --git a/pm3py/trace/format.py b/pm3py/trace/format.py index fec3a30..c3aa2a3 100644 --- a/pm3py/trace/format.py +++ b/pm3py/trace/format.py @@ -1,118 +1,232 @@ -"""ANSI color formatting and sniff line output.""" +"""Trace formatting — colored, decoded, column-wrapped trace output.""" +from __future__ import annotations + +import os +import signal +import sys + from .decode_iso15 import decode_15693 # ---- ANSI colors ---- _C_CYAN = "\033[36m" _C_YELLOW = "\033[33m" +_C_MAGENTA = "\033[35m" +_C_GREEN = "\033[32m" _C_DIM = "\033[2m" -_C_CRC = "\033[1;37m" -_C_RED = "\033[31m" +_C_CRC = "\033[1;37m" # bold white — CRC bytes _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}" +_MODE_TAGS = { + "sim": ("[Sim]", _C_MAGENTA), + "reader": ("[Rdr]", _C_CYAN), + "sniff": ("[Snf]", ""), +} -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}" +class TraceFormatter: + """Colored, decoded, column-wrapped trace output. + + Args: + mode: "sim", "reader", or "sniff" + decoder: Callable[[int, bytes], str | None] for annotation, or None + width: override terminal width (None = auto-detect) + is_tty: override TTY detection (None = auto-detect) + crc_len: number of CRC bytes to split from payload end (0 = none) + """ + + def __init__(self, mode: str = "sim", decoder=None, + width: int | None = None, is_tty: bool | None = None, + crc_len: int = 0): + self._mode = mode + self._decoder = decoder + self._crc_len = crc_len + self._width = width or self._detect_width() + self._is_tty = is_tty if is_tty is not None else sys.stdout.isatty() + self._line_count = 0 + + # Try to register SIGWINCH for dynamic resize + if width is None: + try: + signal.signal(signal.SIGWINCH, self._on_resize) + except (OSError, ValueError): + pass # not main thread or not Unix + + def _detect_width(self) -> int: + try: + return os.get_terminal_size().columns + except (OSError, ValueError): + return 80 + + def _on_resize(self, signum, frame): + self._width = self._detect_width() + + def _color(self, code: str, text: str) -> str: + if not self._is_tty or not code: + return text + return f"{code}{text}{_C_RESET}" + + def format(self, direction: int, payload: bytes, crc_fail: bool = False) -> str: + """Format a trace line with colors, decoding, and wrapping.""" + self._line_count += 1 + if self._line_count % 10 == 0: + try: + self._width = self._detect_width() + except Exception: + pass + + # Mode tag + mode_tag, mode_color = _MODE_TAGS.get(self._mode, ("[???]", "")) + + # Direction + if direction == 0: + arrow = "Reader \u2192 Tag:" + dir_color = _C_CYAN + is_ours = (self._mode == "reader") + else: + arrow = "Tag \u2192 Reader:" + dir_color = _C_YELLOW + is_ours = (self._mode == "sim") + + # Build prefix + prefix_plain = f"{mode_tag} {arrow} " + prefix_len = len(prefix_plain) + + if is_ours: + dir_color = _C_DIM + dir_color + prefix_colored = self._color(mode_color, mode_tag) + " " + self._color(dir_color, arrow) + " " + + # Split payload into data and CRC + if self._crc_len > 0 and len(payload) > self._crc_len: + data = payload[:-self._crc_len] + crc = payload[-self._crc_len:] + 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 = None + if self._decoder is not None: + annotation = self._decoder(direction, data) + if crc_fail: + crc_note = "BAD CRC" + annotation = f"{annotation} {crc_note}" if annotation else crc_note + ann_color = (_C_DIM + dir_color) if is_ours else dir_color + if crc_fail: + ann_color = "\033[31m" + + # Layout + avail = self._width - prefix_len + if annotation: + one_line = f"{full_hex} {annotation}" + else: + one_line = full_hex + + if len(one_line) <= avail: + hex_colored = self._color(_C_DIM, data_hex) + if crc_hex: + hex_colored += " " + self._color(_C_CRC, crc_hex) + if annotation: + ann_colored = self._color(ann_color, annotation) + gap = max(avail - len(full_hex) - len(annotation), 2) + line = f"{prefix_colored}{hex_colored}{' ' * gap}{ann_colored}" + else: + line = f"{prefix_colored}{hex_colored}" + return f"\n{line}" + + # Multi-line + pad = " " * prefix_len + hex_lines = self._wrap_hex(full_hex, avail) + parts = [] + for i, hl in enumerate(hex_lines): + colored_hl = self._color_hex_line(hl, data_hex, crc_hex) + if i == 0: + parts.append(f"{prefix_colored}{colored_hl}") + else: + parts.append(f"{pad}{colored_hl}") + if annotation: + ann_lines = self._wrap_annotation(annotation, avail) + for al in ann_lines: + ann_pad = max(avail - len(al), 0) + parts.append(f"{pad}{' ' * ann_pad}{self._color(ann_color, al)}") + + return "\n" + "\n".join(parts) + + def _color_hex_line(self, line: str, data_hex: str, crc_hex: str) -> str: + if not crc_hex: + return self._color(_C_DIM, line) + crc_tokens = crc_hex.split(" ") + line_tokens = line.split(" ") + crc_start = None + for i in range(len(line_tokens)): + if line_tokens[i:] == crc_tokens[-len(line_tokens) + i:]: + crc_start = i + break + remaining = line_tokens[i:] + if crc_hex.startswith(" ".join(remaining)) or " ".join(remaining) == crc_hex: + crc_start = i + break + if crc_start is not None and crc_start < len(line_tokens): + data_part = " ".join(line_tokens[:crc_start]) + crc_part = " ".join(line_tokens[crc_start:]) + result = "" + if data_part: + result += self._color(_C_DIM, data_part) + " " + result += self._color(_C_CRC, crc_part) + return result + return self._color(_C_DIM, line) + + def _wrap_annotation(self, 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 _wrap_hex(self, hex_str: str, avail: int) -> list[str]: + tokens = hex_str.split(" ") + lines: list[str] = [] + current = "" + for tok in tokens: + candidate = f"{current} {tok}" if current else tok if len(candidate) <= avail: current = candidate else: - lines.append(current) - current = part - lines.append(current) - return lines - return [annotation[:avail - 3] + "..."] + if current: + lines.append(current) + current = tok + if current: + lines.append(current) + return lines or [""] + + def print(self, direction: int, payload: bytes, crc_fail: bool = False) -> None: + """Format and print a trace line to stdout.""" + sys.stdout.write(self.format(direction, payload, crc_fail=crc_fail)) + sys.stdout.flush() 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. + Convenience wrapper around TraceFormatter for sniff mode with 15693 decoder. """ + fmt = TraceFormatter(mode="sniff", decoder=decode_15693, + width=width, is_tty=is_tty, crc_len=2) 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) + result = fmt.format(direction, entry["data"]) + # Strip leading newline that TraceFormatter adds + return result.lstrip("\n") diff --git a/tests/test_sim_trace_fmt.py b/tests/test_sim_trace_fmt.py index edf3b5a..bb0ef39 100644 --- a/tests/test_sim_trace_fmt.py +++ b/tests/test_sim_trace_fmt.py @@ -41,7 +41,7 @@ class TestDecode15693Request: def test_read_multiple_block(self): payload = bytes([0x22, 0x23]) + b"\x01" * 8 + bytes([0x00, 0x0D]) result = decode_15693(0, payload) - assert result == "READ MULTIPLE BLOCK #0+13" + assert result == "READ MULTIPLE BLOCKS #0+13" def test_reset_to_ready(self): payload = bytes([0x22, 0x26]) + b"\x01" * 8