- Add docs/plans/README.md as the roadmap index cataloguing all design/plan docs. - Add the custom ISO14443-A command handling design + plan (L3 NTAG I2C SECTOR_SELECT / cross-sector reads + native auth; L4 static APDUs + WTX relay), produced from a multi-agent design workflow. - Backfill previously-uncommitted plan docs (trace-formatter, ndef-trace-decode, live-sniff, firmware-upstream-rebase, 14a-live-trace) and NTAG5_SECURITY.md. - Sync CLAUDE.md package structure with the committed transponder models. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
999 lines
30 KiB
Markdown
999 lines
30 KiB
Markdown
# Trace Formatter Implementation Plan
|
|
|
|
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
|
|
|
|
**Goal:** Replace raw `print()` trace output with colored, decoded, column-wrapped trace rendering for sim sessions.
|
|
|
|
**Architecture:** New `trace_fmt.py` module with `TraceFormatter` class + protocol decode functions. `sim_session.py` delegates to it. Terminal width detected dynamically with SIGWINCH support.
|
|
|
|
**Tech Stack:** Python stdlib only (os, signal, sys, struct, shutil). No external dependencies.
|
|
|
|
---
|
|
|
|
### Task 1: ISO 15693 request decoder
|
|
|
|
**Files:**
|
|
- Create: `pm3py/sim/trace_fmt.py`
|
|
- Test: `tests/test_sim_trace_fmt.py`
|
|
|
|
**Step 1: Write failing tests for 15693 request decoding**
|
|
|
|
```python
|
|
"""Tests for pm3py.sim.trace_fmt — trace formatting and protocol decoding."""
|
|
import os
|
|
import pytest
|
|
|
|
from pm3py.sim.trace_fmt import decode_15693
|
|
|
|
|
|
class TestDecode15693Request:
|
|
"""Decode reader->tag (direction=0) 15693 commands."""
|
|
|
|
def test_inventory(self):
|
|
# flags=0x26, cmd=0x01, mask_len=0
|
|
payload = bytes([0x26, 0x01, 0x00])
|
|
result = decode_15693(0, payload)
|
|
assert result == "INVENTORY"
|
|
|
|
def test_inventory_with_mask(self):
|
|
payload = bytes([0x26, 0x01, 0x08, 0xAB])
|
|
result = decode_15693(0, payload)
|
|
assert result == "INVENTORY mask=8"
|
|
|
|
def test_stay_quiet(self):
|
|
payload = bytes([0x22, 0x02]) + b"\x01" * 8
|
|
result = decode_15693(0, payload)
|
|
assert result == "STAY QUIET"
|
|
|
|
def test_read_single_block_addressed(self):
|
|
# flags=0x22 (addressed), cmd=0x20, uid(8), block=0x03
|
|
payload = bytes([0x22, 0x20]) + b"\x01" * 8 + bytes([0x03])
|
|
result = decode_15693(0, payload)
|
|
assert result == "READ SINGLE BLOCK #3"
|
|
|
|
def test_read_single_block_unaddressed(self):
|
|
# flags=0x02 (no address flag), cmd=0x20, block=0x00
|
|
payload = bytes([0x02, 0x20, 0x00])
|
|
result = decode_15693(0, payload)
|
|
assert result == "READ SINGLE BLOCK #0"
|
|
|
|
def test_write_single_block(self):
|
|
payload = bytes([0x22, 0x21]) + b"\x01" * 8 + bytes([0x05]) + b"\xDE\xAD\xBE\xEF"
|
|
result = decode_15693(0, payload)
|
|
assert result == "WRITE SINGLE BLOCK #5 [4B]"
|
|
|
|
def test_read_multiple_block(self):
|
|
# flags=0x22, cmd=0x23, uid(8), start=0x00, count=0x0D
|
|
payload = bytes([0x22, 0x23]) + b"\x01" * 8 + bytes([0x00, 0x0D])
|
|
result = decode_15693(0, payload)
|
|
assert result == "READ MULTIPLE BLOCK #0+13"
|
|
|
|
def test_reset_to_ready(self):
|
|
payload = bytes([0x22, 0x26]) + b"\x01" * 8
|
|
result = decode_15693(0, payload)
|
|
assert result == "RESET TO READY"
|
|
|
|
def test_get_system_info(self):
|
|
payload = bytes([0x22, 0x2B]) + b"\x01" * 8
|
|
result = decode_15693(0, payload)
|
|
assert result == "GET SYSTEM INFO"
|
|
|
|
def test_get_multiple_block_security(self):
|
|
payload = bytes([0x22, 0x2C]) + b"\x01" * 8 + bytes([0x00, 0x3F])
|
|
result = decode_15693(0, payload)
|
|
assert result == "GET MULTIPLE BLOCK SECURITY #0+63"
|
|
|
|
def test_nxp_read_config(self):
|
|
# flags=0x22, cmd=0xA1, mfg=0x04, reg=0x02
|
|
payload = bytes([0x22, 0xA1, 0x04, 0x02])
|
|
result = decode_15693(0, payload)
|
|
assert result == "NXP READ CONFIG reg=2"
|
|
|
|
def test_nxp_write_config(self):
|
|
payload = bytes([0x22, 0xA2, 0x04, 0x03, 0xFF])
|
|
result = decode_15693(0, payload)
|
|
assert result == "NXP WRITE CONFIG reg=3 val=0xFF"
|
|
|
|
def test_nxp_get_random(self):
|
|
payload = bytes([0x22, 0xB2, 0x04])
|
|
result = decode_15693(0, payload)
|
|
assert result == "NXP GET RANDOM"
|
|
|
|
def test_nxp_set_password(self):
|
|
payload = bytes([0x22, 0xB3, 0x04, 0x01]) + b"\x00" * 4
|
|
result = decode_15693(0, payload)
|
|
assert result == "NXP SET PASSWORD id=1"
|
|
|
|
def test_unknown_command(self):
|
|
payload = bytes([0x22, 0xFF])
|
|
result = decode_15693(0, payload)
|
|
assert result is None
|
|
|
|
def test_too_short(self):
|
|
result = decode_15693(0, bytes([0x26]))
|
|
assert result is None
|
|
```
|
|
|
|
**Step 2: Run tests to verify they fail**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode15693Request -v`
|
|
Expected: FAIL — `ModuleNotFoundError: No module named 'pm3py.sim.trace_fmt'`
|
|
|
|
**Step 3: Implement 15693 request decoder**
|
|
|
|
Create `pm3py/sim/trace_fmt.py`:
|
|
|
|
```python
|
|
"""Trace formatter — colored, decoded, column-wrapped trace output."""
|
|
from __future__ import annotations
|
|
|
|
# ---- 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)
|
|
_15693_NXP_CMDS = {
|
|
0xA1: "NXP READ CONFIG",
|
|
0xA2: "NXP WRITE CONFIG",
|
|
0xB2: "NXP GET RANDOM",
|
|
0xB3: "NXP SET PASSWORD",
|
|
}
|
|
|
|
|
|
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_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 (check manufacturer code)
|
|
name = _15693_NXP_CMDS.get(cmd)
|
|
if name is not None:
|
|
return _annotate_nxp_request(name, cmd, payload)
|
|
|
|
return None
|
|
|
|
|
|
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 == 0xA1 and len(payload) >= 4: # READ CONFIG
|
|
return f"{name} reg={payload[3]}"
|
|
if cmd == 0xA2 and len(payload) >= 5: # WRITE CONFIG
|
|
return f"{name} reg={payload[3]} val=0x{payload[4]:02X}"
|
|
if cmd == 0xB3 and len(payload) >= 4: # SET PASSWORD
|
|
return f"{name} id={payload[3]}"
|
|
return name
|
|
|
|
|
|
def _decode_15693_response(payload: bytes) -> str | None:
|
|
# Placeholder — implemented in Task 2
|
|
return None
|
|
```
|
|
|
|
**Step 4: Run tests to verify they pass**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode15693Request -v`
|
|
Expected: all PASS
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
cd /home/work/pm3py/.worktrees/sim-framework
|
|
git add pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py
|
|
git commit --no-gpg-sign -m "feat(trace): add ISO 15693 request decoder"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 2: ISO 15693 response decoder
|
|
|
|
**Files:**
|
|
- Modify: `pm3py/sim/trace_fmt.py` — replace `_decode_15693_response` placeholder
|
|
- Test: `tests/test_sim_trace_fmt.py`
|
|
|
|
**Step 1: Write failing tests**
|
|
|
|
Append to `tests/test_sim_trace_fmt.py`:
|
|
|
|
```python
|
|
class TestDecode15693Response:
|
|
"""Decode tag->reader (direction=1) 15693 responses."""
|
|
|
|
def test_ok_empty(self):
|
|
# Just flags=0x00, no data (e.g., write ack)
|
|
result = decode_15693(1, bytes([0x00]))
|
|
assert result == "OK"
|
|
|
|
def test_inventory_response(self):
|
|
# flags=0x00, dsfid=0x00, uid (8 bytes LSB-first)
|
|
uid_lsb = bytes([0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x04, 0xE0])
|
|
payload = bytes([0x00, 0x00]) + uid_lsb
|
|
result = decode_15693(1, payload)
|
|
assert result == "OK INVENTORY UID=E004010101010101"
|
|
|
|
def test_error_response(self):
|
|
payload = bytes([0x01, 0x0F])
|
|
result = decode_15693(1, payload)
|
|
assert result == "ERROR 0x0F"
|
|
|
|
def test_error_block_not_available(self):
|
|
payload = bytes([0x01, 0x10])
|
|
result = decode_15693(1, payload)
|
|
assert result == "ERROR 0x10 block not available"
|
|
|
|
def test_data_response(self):
|
|
# flags=0x00 + 4 data bytes (read single block response)
|
|
payload = bytes([0x00, 0xDE, 0xAD, 0xBE, 0xEF])
|
|
result = decode_15693(1, payload)
|
|
assert result == "OK [4B]"
|
|
|
|
def test_too_short(self):
|
|
result = decode_15693(1, b"")
|
|
assert result is None
|
|
```
|
|
|
|
**Step 2: Run to verify failure**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode15693Response -v`
|
|
Expected: FAIL — responses decode as `None`
|
|
|
|
**Step 3: Implement response decoder**
|
|
|
|
Replace `_decode_15693_response` in `trace_fmt.py`:
|
|
|
|
```python
|
|
_15693_ERRORS = {
|
|
0x01: "not supported",
|
|
0x02: "not recognized",
|
|
0x0F: "unknown error",
|
|
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()}"
|
|
|
|
# Generic data response
|
|
return f"OK [{len(data)}B]"
|
|
```
|
|
|
|
**Step 4: Run tests**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode15693Response -v`
|
|
Expected: all PASS
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
cd /home/work/pm3py/.worktrees/sim-framework
|
|
git add pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py
|
|
git commit --no-gpg-sign -m "feat(trace): add ISO 15693 response decoder"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 3: ISO 14443-A decoder
|
|
|
|
**Files:**
|
|
- Modify: `pm3py/sim/trace_fmt.py`
|
|
- Test: `tests/test_sim_trace_fmt.py`
|
|
|
|
**Step 1: Write failing tests**
|
|
|
|
Append to test file:
|
|
|
|
```python
|
|
from pm3py.sim.trace_fmt import decode_14443a
|
|
|
|
|
|
class TestDecode14443aRequest:
|
|
"""Decode reader->tag 14443-A commands."""
|
|
|
|
def test_reqa(self):
|
|
assert decode_14443a(0, bytes([0x26])) == "REQA"
|
|
|
|
def test_wupa(self):
|
|
assert decode_14443a(0, bytes([0x52])) == "WUPA"
|
|
|
|
def test_hlta(self):
|
|
assert decode_14443a(0, bytes([0x50, 0x00])) == "HLTA"
|
|
|
|
def test_anticol_cl1(self):
|
|
assert decode_14443a(0, bytes([0x93, 0x20])) == "ANTICOL CL1"
|
|
|
|
def test_select_cl1(self):
|
|
payload = bytes([0x93, 0x70]) + b"\x01\x02\x03\x04\x04"
|
|
assert decode_14443a(0, payload) == "SELECT CL1"
|
|
|
|
def test_anticol_cl2(self):
|
|
assert decode_14443a(0, bytes([0x95, 0x20])) == "ANTICOL CL2"
|
|
|
|
def test_select_cl2(self):
|
|
payload = bytes([0x95, 0x70]) + b"\x01\x02\x03\x04\x04"
|
|
assert decode_14443a(0, payload) == "SELECT CL2"
|
|
|
|
def test_anticol_cl3(self):
|
|
assert decode_14443a(0, bytes([0x97, 0x20])) == "ANTICOL CL3"
|
|
|
|
def test_select_cl3(self):
|
|
payload = bytes([0x97, 0x70]) + b"\x01\x02\x03\x04\x04"
|
|
assert decode_14443a(0, payload) == "SELECT CL3"
|
|
|
|
def test_rats(self):
|
|
assert decode_14443a(0, bytes([0xE0, 0x50])) == "RATS"
|
|
|
|
def test_iblock_even(self):
|
|
assert decode_14443a(0, bytes([0x02, 0x00, 0xA4])) == "I-BLOCK(0)"
|
|
|
|
def test_iblock_odd(self):
|
|
assert decode_14443a(0, bytes([0x03, 0x00, 0xA4])) == "I-BLOCK(1)"
|
|
|
|
def test_rack(self):
|
|
assert decode_14443a(0, bytes([0xA2])) == "R-ACK(0)"
|
|
assert decode_14443a(0, bytes([0xA3])) == "R-ACK(1)"
|
|
|
|
def test_rnak(self):
|
|
assert decode_14443a(0, bytes([0xB2])) == "R-NAK(0)"
|
|
assert decode_14443a(0, bytes([0xB3])) == "R-NAK(1)"
|
|
|
|
def test_deselect(self):
|
|
assert decode_14443a(0, bytes([0xC2])) == "S(DESELECT)"
|
|
|
|
def test_wtx(self):
|
|
assert decode_14443a(0, bytes([0xF2, 0x01])) == "S(WTX)"
|
|
|
|
def test_unknown(self):
|
|
assert decode_14443a(0, bytes([0xFF])) is None
|
|
|
|
def test_empty(self):
|
|
assert decode_14443a(0, b"") is None
|
|
|
|
|
|
class TestDecode14443aResponse:
|
|
"""Decode tag->reader 14443-A responses."""
|
|
|
|
def test_atqa(self):
|
|
# 2-byte response to REQA/WUPA
|
|
assert decode_14443a(1, bytes([0x04, 0x00])) == "ATQA 04 00"
|
|
|
|
def test_sak(self):
|
|
# 1-byte response to SELECT
|
|
assert decode_14443a(1, bytes([0x20])) == "SAK 20"
|
|
|
|
def test_ats(self):
|
|
# ATS: first byte is length
|
|
ats = bytes([0x05, 0x78, 0x80, 0x70, 0x02])
|
|
assert decode_14443a(1, ats) == "ATS [5]"
|
|
|
|
def test_iblock_response(self):
|
|
assert decode_14443a(1, bytes([0x02, 0x90, 0x00])) == "I-BLOCK(0)"
|
|
|
|
def test_empty(self):
|
|
assert decode_14443a(1, b"") is None
|
|
```
|
|
|
|
**Step 2: Run to verify failure**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode14443aRequest tests/test_sim_trace_fmt.py::TestDecode14443aResponse -v`
|
|
Expected: FAIL — `ImportError: cannot import name 'decode_14443a'`
|
|
|
|
**Step 3: Implement 14443-A decoder**
|
|
|
|
Add to `trace_fmt.py`:
|
|
|
|
```python
|
|
# ---- 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
|
|
```
|
|
|
|
**Step 4: Run tests**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py -v`
|
|
Expected: all PASS (15693 + 14443-A)
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
cd /home/work/pm3py/.worktrees/sim-framework
|
|
git add pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py
|
|
git commit --no-gpg-sign -m "feat(trace): add ISO 14443-A decoder"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 4: TraceFormatter — color, wrapping, terminal width
|
|
|
|
**Files:**
|
|
- Modify: `pm3py/sim/trace_fmt.py`
|
|
- Test: `tests/test_sim_trace_fmt.py`
|
|
|
|
**Step 1: Write failing tests**
|
|
|
|
Append to test file:
|
|
|
|
```python
|
|
from pm3py.sim.trace_fmt import TraceFormatter
|
|
|
|
|
|
class TestTraceFormatterBasic:
|
|
"""Test formatting output (colors stripped for assertion)."""
|
|
|
|
def _strip_ansi(self, s: str) -> str:
|
|
import re
|
|
return re.sub(r'\033\[[0-9;]*m', '', s)
|
|
|
|
def test_starts_with_newline(self):
|
|
fmt = TraceFormatter(mode="sim")
|
|
result = fmt.format(0, bytes([0x26, 0x01, 0x00]))
|
|
assert result.startswith("\n")
|
|
|
|
def test_sim_mode_tag(self):
|
|
fmt = TraceFormatter(mode="sim")
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "[Sim]" in result
|
|
|
|
def test_reader_mode_tag(self):
|
|
fmt = TraceFormatter(mode="reader")
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "[Rdr]" in result
|
|
|
|
def test_sniff_mode_tag(self):
|
|
fmt = TraceFormatter(mode="sniff")
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "[Snf]" in result
|
|
|
|
def test_reader_to_tag_arrow(self):
|
|
fmt = TraceFormatter(mode="sim")
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "Reader \u2192 Tag:" in result
|
|
|
|
def test_tag_to_reader_arrow(self):
|
|
fmt = TraceFormatter(mode="sim")
|
|
result = self._strip_ansi(fmt.format(1, bytes([0x00])))
|
|
assert "Tag \u2192 Reader:" in result
|
|
|
|
def test_hex_space_separated(self):
|
|
fmt = TraceFormatter(mode="sim")
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x22, 0x20, 0x03])))
|
|
assert "22 20 03" in result
|
|
|
|
def test_annotation_present(self):
|
|
fmt = TraceFormatter(mode="sim")
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "INVENTORY" in result
|
|
|
|
def test_no_annotation_for_unknown(self):
|
|
fmt = TraceFormatter(mode="sim")
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x22, 0xFF])))
|
|
assert "22 FF" in result or "22 ff" in result.lower()
|
|
|
|
|
|
class TestTraceFormatterWrapping:
|
|
"""Test column-aligned wrapping for long payloads."""
|
|
|
|
def _strip_ansi(self, s: str) -> str:
|
|
import re
|
|
return re.sub(r'\033\[[0-9;]*m', '', s)
|
|
|
|
def test_short_payload_inline_annotation(self):
|
|
"""Short payload: annotation on same line as hex."""
|
|
fmt = TraceFormatter(mode="sim", width=80)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
lines = result.strip().split("\n")
|
|
assert len(lines) == 1
|
|
assert "INVENTORY" in lines[0]
|
|
assert "26 01 00" in lines[0]
|
|
|
|
def test_long_payload_wraps(self):
|
|
"""Long payload wraps at prefix column."""
|
|
fmt = TraceFormatter(mode="sim", width=60)
|
|
long_payload = bytes([0x00]) + bytes(40)
|
|
result = self._strip_ansi(fmt.format(1, long_payload))
|
|
lines = result.strip().split("\n")
|
|
assert len(lines) > 1
|
|
# Continuation lines should be indented to prefix column
|
|
first_hex_col = lines[0].index("00")
|
|
for line in lines[1:]:
|
|
if line.strip():
|
|
leading = len(line) - len(line.lstrip())
|
|
assert leading >= first_hex_col - 1
|
|
|
|
def test_wrapped_annotation_on_own_line(self):
|
|
"""When hex wraps, annotation goes on its own line."""
|
|
fmt = TraceFormatter(mode="sim", width=50)
|
|
# Inventory response with UID — long enough to wrap at width=50
|
|
uid_lsb = bytes([0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x04, 0xE0])
|
|
payload = bytes([0x00, 0x00]) + uid_lsb
|
|
result = self._strip_ansi(fmt.format(1, payload))
|
|
lines = result.strip().split("\n")
|
|
# Annotation should be on a separate line
|
|
annotation_lines = [l for l in lines if "INVENTORY" in l]
|
|
hex_lines = [l for l in lines if "00 01" in l.lower() or "04 e0" in l.lower()]
|
|
if len(hex_lines) > 1:
|
|
# Wrapped — annotation must be on its own line
|
|
assert len(annotation_lines) == 1
|
|
assert annotation_lines[0] not in hex_lines
|
|
|
|
|
|
class TestTraceFormatterNoColor:
|
|
"""Colors suppressed when is_tty=False."""
|
|
|
|
def test_no_ansi_when_not_tty(self):
|
|
fmt = TraceFormatter(mode="sim", is_tty=False)
|
|
result = fmt.format(0, bytes([0x26, 0x01, 0x00]))
|
|
assert "\033[" not in result
|
|
```
|
|
|
|
**Step 2: Run to verify failure**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestTraceFormatterBasic tests/test_sim_trace_fmt.py::TestTraceFormatterWrapping tests/test_sim_trace_fmt.py::TestTraceFormatterNoColor -v`
|
|
Expected: FAIL — `ImportError: cannot import name 'TraceFormatter'`
|
|
|
|
**Step 3: Implement TraceFormatter**
|
|
|
|
Add to `trace_fmt.py`:
|
|
|
|
```python
|
|
import os
|
|
import signal
|
|
import sys
|
|
|
|
|
|
# ---- ANSI colors ----
|
|
_C_CYAN = "\033[36m"
|
|
_C_YELLOW = "\033[33m"
|
|
_C_MAGENTA = "\033[35m"
|
|
_C_GREEN = "\033[32m"
|
|
_C_DIM = "\033[2m"
|
|
_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"
|
|
protocol: "15693" or "14443a" (selects decoder)
|
|
width: override terminal width (None = auto-detect)
|
|
is_tty: override TTY detection (None = auto-detect)
|
|
"""
|
|
|
|
def __init__(self, mode: str = "sim", protocol: str = "15693",
|
|
width: int | None = None, is_tty: bool | None = None):
|
|
self._mode = mode
|
|
self._protocol = protocol
|
|
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) -> 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
|
|
if direction == 0:
|
|
arrow = "Reader \u2192 Tag:"
|
|
dir_color = _C_CYAN
|
|
else:
|
|
arrow = "Tag \u2192 Reader:"
|
|
dir_color = _C_YELLOW
|
|
|
|
# Build prefix (uncolored for width calc)
|
|
prefix_plain = f"{mode_tag} {arrow} "
|
|
prefix_len = len(prefix_plain)
|
|
|
|
# Colored prefix
|
|
prefix_colored = self._color(mode_color, mode_tag) + " " + self._color(dir_color, arrow) + " "
|
|
|
|
# Hex bytes (space-separated, uppercase)
|
|
hex_str = " ".join(f"{b:02X}" for b in payload)
|
|
|
|
# Decode annotation
|
|
if self._protocol == "15693":
|
|
annotation = decode_15693(direction, payload)
|
|
elif self._protocol == "14443a":
|
|
annotation = decode_14443a(direction, payload)
|
|
else:
|
|
annotation = None
|
|
|
|
# Layout: determine if everything fits on one line
|
|
avail = self._width - prefix_len
|
|
if annotation:
|
|
one_line = f"{hex_str} {annotation}"
|
|
else:
|
|
one_line = hex_str
|
|
|
|
if len(one_line) <= avail:
|
|
# Single line
|
|
hex_colored = self._color(_C_DIM, hex_str)
|
|
if annotation:
|
|
ann_colored = self._color(_C_GREEN, annotation)
|
|
line = f"{prefix_colored}{hex_colored} {ann_colored}"
|
|
else:
|
|
line = f"{prefix_colored}{hex_colored}"
|
|
return f"\n{line}"
|
|
|
|
# Multi-line: wrap hex, annotation on its own line
|
|
pad = " " * prefix_len
|
|
hex_lines = self._wrap_hex(hex_str, avail)
|
|
parts = [f"{prefix_colored}{self._color(_C_DIM, hex_lines[0])}"]
|
|
for hl in hex_lines[1:]:
|
|
parts.append(f"{pad}{self._color(_C_DIM, hl)}")
|
|
if annotation:
|
|
parts.append(f"{pad}{self._color(_C_GREEN, annotation)}")
|
|
|
|
return "\n" + "\n".join(parts)
|
|
|
|
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 = []
|
|
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) -> None:
|
|
"""Format and print a trace line to stdout."""
|
|
sys.stdout.write(self.format(direction, payload))
|
|
sys.stdout.flush()
|
|
```
|
|
|
|
**Step 4: Run tests**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py -v`
|
|
Expected: all PASS
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
cd /home/work/pm3py/.worktrees/sim-framework
|
|
git add pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py
|
|
git commit --no-gpg-sign -m "feat(trace): add TraceFormatter with color, wrapping, terminal resize"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 5: Wire TraceFormatter into SimSession
|
|
|
|
**Files:**
|
|
- Modify: `pm3py/sim/sim_session.py:1-2,126-175,177-224`
|
|
|
|
**Step 1: Write failing test**
|
|
|
|
Append to `tests/test_sim_trace_fmt.py`:
|
|
|
|
```python
|
|
from unittest.mock import MagicMock, patch
|
|
import struct
|
|
|
|
from pm3py.sim.sim_session import SimSession, CMD_HF_ISO15693_SIM_TRACE
|
|
from pm3py.transport import encode_ng_frame, RESP_PREAMBLE_MAGIC, RESP_PREAMBLE_SIZE, RESP_POSTAMBLE_SIZE
|
|
|
|
|
|
class TestSimSessionTrace:
|
|
"""Verify SimSession uses TraceFormatter instead of raw print."""
|
|
|
|
def _make_trace_frame(self, direction: int, payload: bytes) -> bytes:
|
|
"""Build a fake firmware trace response frame."""
|
|
from pm3py.transport import RESP_POSTAMBLE_MAGIC
|
|
data = bytes([direction]) + payload
|
|
length = len(data) | 0x8000 # NG bit set
|
|
header = struct.pack("<IHHI",
|
|
RESP_PREAMBLE_MAGIC,
|
|
length,
|
|
0, # status
|
|
CMD_HF_ISO15693_SIM_TRACE)
|
|
postamble = struct.pack("<H", RESP_POSTAMBLE_MAGIC)
|
|
return header + data + postamble
|
|
|
|
@patch("pm3py.sim.sim_session.TraceFormatter")
|
|
def test_trace_reader_uses_formatter(self, MockFormatter):
|
|
"""_trace_reader should call formatter.print() not raw print()."""
|
|
mock_fmt = MockFormatter.return_value
|
|
session = SimSession()
|
|
session._formatter = mock_fmt
|
|
session._active = True
|
|
|
|
# Build a trace frame for: reader->tag inventory
|
|
frame = self._make_trace_frame(0, bytes([0x26, 0x01, 0x00]))
|
|
|
|
# Mock serial that returns frame then stops
|
|
mock_serial = MagicMock()
|
|
mock_serial.in_waiting = len(frame)
|
|
call_count = 0
|
|
|
|
def read_side_effect(n):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
return frame
|
|
session._active = False
|
|
return b""
|
|
|
|
mock_serial.read.side_effect = read_side_effect
|
|
|
|
session._trace_reader(mock_serial)
|
|
mock_fmt.print.assert_called_once_with(0, bytes([0x26, 0x01, 0x00]))
|
|
```
|
|
|
|
**Step 2: Run to verify failure**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestSimSessionTrace -v`
|
|
Expected: FAIL — `ImportError` or `AttributeError` (no `TraceFormatter` import in sim_session)
|
|
|
|
**Step 3: Modify sim_session.py**
|
|
|
|
Add import at top of `sim_session.py`:
|
|
|
|
```python
|
|
from .trace_fmt import TraceFormatter
|
|
```
|
|
|
|
In `start_15693()`, replace the trace thread creation (around line 167-171) to instantiate the formatter:
|
|
|
|
```python
|
|
if trace:
|
|
self._formatter = TraceFormatter(mode="sim", protocol="15693")
|
|
self._trace_thread = threading.Thread(
|
|
target=self._trace_reader, args=(ser,), daemon=True
|
|
)
|
|
self._trace_thread.start()
|
|
```
|
|
|
|
In `_trace_reader()`, replace line 216-217:
|
|
|
|
```python
|
|
arrow = "Reader → Tag" if direction == 0 else "Tag → Reader"
|
|
print(f"[Trace] {arrow}: {payload.hex()}")
|
|
```
|
|
|
|
With:
|
|
|
|
```python
|
|
self._formatter.print(direction, payload)
|
|
```
|
|
|
|
**Step 4: Run all tests**
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py -v`
|
|
Expected: all PASS
|
|
|
|
Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/ -v`
|
|
Expected: all PASS (no regressions)
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
cd /home/work/pm3py/.worktrees/sim-framework
|
|
git add pm3py/sim/sim_session.py pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py
|
|
git commit --no-gpg-sign -m "feat(trace): wire TraceFormatter into SimSession"
|
|
```
|