MikeFaith/pm3py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sim): ISO14443-A live sim trace consumer (Y1-Y3 + T1)

Browse Source 
Y1: Cmd.HF_ISO14443A_SIM_TRACE (0x0339).
Y2: generalize SimSession._trace_reader dispatch to a {cmd: protocol} map so
    14a (proto 1) and 15693 (proto 0) share one reader; the ADC field-strength
    branch is now guarded to 15693.
Y3: rewrite start_14a as a synchronous trace path mirroring start_15693 (raw
    pyserial + the _trace_reader thread). Sends the correct HF_ISO14443A_SIMULATE
    NG payload (same layout as hf.iso14a.sim()), sets FLAG_SIM_TRACE when
    streaming, and takes trace/on_frame/quiet/tagtype/exit_after kwargs. Loading
    tag memory into emulator RAM stays a separate (hardware) step; the trace
    streams every reader<->tag frame regardless. The WTX _relay_loop is retained
    for the future 14a-4 path.
T1: mocked-transport unit tests for the 14a trace path (tests/test_sim_trace_14a.py).

Requires the bumped firmware submodule. 996 tests pass.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
michael
2026-07-05 08:36:39 -07:00
parent 64cc6794d9
commit 950628ef4e
3 changed files with 219 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pm3py/core/protocol.py
View File

@@ -222,6 +222,7 @@ class Cmd(IntEnum):
HF_SNIFF_STREAM = 0x0337
HF_SNIFF_STATUS = 0x0338
HF_ISO14443A_SIM_TRACE = 0x0339 # FW->Host: live 14a sim trace (fire-and-forget)
# Sim table commands (firmware response table)
SIM_TABLE_UPLOAD = 0x0900

125
pm3py/sim/sim_session.py
View File

@@ -11,6 +11,7 @@ import serial
from .frame import RFFrame
from .table_compiler import ResponseTable, TableCompiler
from pm3py.trace.format import TraceFormatter
from pm3py.trace.decode_iso14a import decode_14443a
from .transponder import Transponder
from ..core.protocol import Cmd
from ..core.transport import encode_ng_frame, decode_response_frame, RESP_PREAMBLE_MAGIC, RESP_PREAMBLE_SIZE, RESP_POSTAMBLE_SIZE
@@ -20,6 +21,20 @@ CMD_SIM_TABLE_UPLOAD = 0x0950
CMD_SIM_TABLE_CLEAR = 0x0951
CMD_SIM_TABLE_UPDATE = 0x0952
CMD_HF_ISO15693_SIM_TRACE = 0x0336
CMD_HF_ISO14443A_SIM_TRACE = 0x0339
# 14a sim flag bit (pm3_cmd.h FLAG_SIM_TRACE): stream live reader<->tag trace.
FLAG_SIM_TRACE = 0x2000
# Sim-trace command id -> protocol tag used in trace entries (0 = ISO 15693,
# 1 = ISO 14443-A). The firmware ring format is identical across protocols:
# byte 0 = dir (low nibble) | crc_fail (0x80) | 0x03 ADC report (15693 only),
# followed by the raw frame bytes. The active protocol's decoder lives in the
# session formatter, so this only needs the protocol tag.
_TRACE_CMDS = {
CMD_HF_ISO15693_SIM_TRACE: 0,
CMD_HF_ISO14443A_SIM_TRACE: 1,
}
# Max NG payload per frame
MAX_PAYLOAD = 512
@@ -98,33 +113,94 @@ class SimSession:
"""Clear the firmware response table."""
await self._t.send_ng(CMD_SIM_TABLE_CLEAR, b"")
async def start_14a(self, tag, compile: bool = True) -> None:
"""Compile table, upload, start 14443-A sim with WTX relay loop."""
self._tag_model = tag
if compile:
if hasattr(tag, '_keys_a'):
table = TableCompiler.compile_mifare(tag)
else:
table = TableCompiler.compile_14a(tag)
await self.upload_table(table)
def start_14a(self, tag, tagtype: int = 7, trace: bool = False,
on_frame=None, quiet: bool = False, exit_after: int = 0) -> None:
"""Start an ISO 14443-A (Layer 3) sim, optionally streaming a live trace.
# Start sim (fire-and-forget — response comes when sim ends)
Synchronous, mirroring :meth:`start_15693`: the firmware runs
anticollision and the tagtype's built-in command set autonomously; this
sends the sim-start frame over raw serial and, when tracing, spawns the
reader thread that decodes each reader↔tag frame.
Args:
tagtype: firmware emulation type (see ``hf.iso14a.sim``; 7 = MFU
EV1/NTAG215, 1 = MIFARE Classic 1k, 2 = Ultralight, ...).
trace: if True, print live reader↔tag communication to the console.
on_frame: optional callback(frame: dict) invoked per decoded frame
(see the ``entries`` property for the dict shape). Passing it
auto-enables trace streaming.
quiet: if True, don't print frames (callback + ``entries`` still fill).
exit_after: stop the sim after N reader reads (0 = run until stopped).
Note: this starts anticollision + protocol handling from the tagtype's
firmware model. Loading the tag's memory contents into emulator RAM (so
reads return model data) is a separate step and is not performed here;
the trace still shows every reader↔tag frame regardless.
14a-4 (ISO-DEP / WTX relay) reuses the same firmware trace but routes it
through the async relay loop — see the plan's I/O model decision.
"""
self._tag_model = tag
self._on_frame = on_frame
self._quiet = quiet
self._print_frames = trace and not quiet
stream = trace or (on_frame is not None)
# Resolve the raw serial port
if self._port:
ser = self._port
elif self._t and self._t._writer:
ser = self._t._writer.transport.serial
else:
raise RuntimeError("No serial port available")
tag._serial = ser
# Fold the UID length into flags per FLAG_SET_UID_IN_DATA (pm3_cmd.h).
uid = tag._uid
flag_val = 0x0001 # FLAG_INTERACTIVE
if len(uid) == 4:
flag_val |= 0x0010
uid_flag = 0x0010 # FLAG_4B_UID_IN_DATA
elif len(uid) == 7:
flag_val |= 0x0020
uid_flag = 0x0020 # FLAG_7B_UID_IN_DATA
elif len(uid) == 10:
flag_val |= 0x0030
atqa = tag.atqa if hasattr(tag, 'atqa') else b"\x04\x00"
sak = tag.sak if hasattr(tag, 'sak') else 0x08
payload = atqa + bytes([sak]) + uid
await self._t.send_ng_no_response(Cmd.HF_ISO14443A_SIMULATE)
# TODO: use send_mix for 14a sim (MIX frame format)
uid_flag = 0x0030 # FLAG_10B_UID_IN_DATA
else:
raise ValueError("UID must be 4, 7, or 10 bytes")
flag_val = uid_flag
if stream:
flag_val |= FLAG_SIM_TRACE
# HF_ISO14443A_SIMULATE NG payload — same layout as hf.iso14a.sim():
# struct { u8 tagtype; u16 flags; u8 uid[10]; u8 exitAfter;
# u8 rats[20]; u8 ulauth_1a1_len; u8 ulauth_1a2_len;
# u8 ulauth_1a1[16]; u8 ulauth_1a2[16]; bool mirror; } PACKED
payload = struct.pack(
"<BH10sB20sBB16s16sB",
tagtype & 0xFF,
flag_val & 0xFFFF,
uid.ljust(10, b"\x00")[:10],
exit_after & 0xFF,
b"\x00" * 20,
0, 0,
b"\x00" * 16, b"\x00" * 16,
0,
)
frame = encode_ng_frame(Cmd.HF_ISO14443A_SIMULATE, payload)
ser.write(frame)
time.sleep(0.2) # let firmware start the sim before streaming
self._active = True
self._relay_task = asyncio.create_task(self._relay_loop(tag))
if stream:
decoder = getattr(tag, "decode_trace", None) or decode_14443a
self._formatter = TraceFormatter(mode="sim", decoder=decoder, crc_len=2)
self._trace_thread = threading.Thread(
target=self._trace_reader, args=(ser,), daemon=True
)
self._trace_thread.start()
print(f"[SimSession] 14a sim started, tagtype={tagtype}, UID={uid.hex()}"
+ (" (trace ON)" if stream else ""))
def start_15693(self, tag, compile: bool = True, trace: bool = False,
on_frame=None, quiet: bool = False) -> None:
@@ -281,12 +357,13 @@ class SimSession:
resp = decode_response_frame(bytes(buf[:frame_len]))
buf = buf[frame_len:]
if resp["cmd"] == CMD_HF_ISO15693_SIM_TRACE:
if resp["cmd"] in _TRACE_CMDS:
protocol = _TRACE_CMDS[resp["cmd"]]
data = resp["data"]
if len(data) >= 2:
direction = data[0] & 0x0F
if direction == 0x03 and len(data) >= 3:
# ADC field strength report
if protocol == 0 and direction == 0x03 and len(data) >= 3:
# ADC field strength report (ISO 15693 only)
adc_mv = (data[1] << 8) | data[2]
if self.on_field_strength:
self.on_field_strength(adc_mv)
@@ -294,7 +371,7 @@ class SimSession:
crc_fail = bool(data[0] & 0x80)
payload = data[1:]
entry = {
"protocol": 0, # ISO 15693
"protocol": protocol,
"direction": direction,
"timestamp": None,
"duration": None,

117
tests/test_sim_trace_14a.py Normal file
View File

@@ -0,0 +1,117 @@
"""ISO 14443-A sim live trace: firmware CMD_HF_ISO14443A_SIM_TRACE -> entries.
Mirrors the SimSession portion of test_trace_scripting.py but for the 14a sim
trace path (protocol tag 1, decode_14443a). Uses synthetic response frames fed
through SimSession._trace_reader with a mocked serial port — no hardware.
"""
import struct
from unittest.mock import MagicMock
from pm3py.core.protocol import RESP_PREAMBLE_MAGIC
from pm3py.core.transport import RESP_POSTAMBLE_NOCRC
from pm3py.sim.sim_session import SimSession, CMD_HF_ISO14443A_SIM_TRACE
from pm3py.trace.decode_iso14a import decode_14443a
from pm3py.trace.format import TraceFormatter
REQA = bytes.fromhex("26") # reader -> tag, short frame
ATQA = bytes.fromhex("0400") # tag -> reader, no CRC
def _fmt():
return TraceFormatter(mode="sim", decoder=decode_14443a, crc_len=2, is_tty=False)
def _trace_frame(direction, payload):
"""Build one CMD_HF_ISO14443A_SIM_TRACE response frame (USB, no CRC)."""
data = bytes([direction]) + payload
header = struct.pack("<IHbbH", RESP_PREAMBLE_MAGIC, len(data) | 0x8000,
0, 0, CMD_HF_ISO14443A_SIM_TRACE)
return header + data + struct.pack("<H", RESP_POSTAMBLE_NOCRC)
def _run_sim_reader(session, buf):
"""Drive _trace_reader once over `buf`, then stop it."""
ser = MagicMock()
ser.in_waiting = len(buf)
n = 0
def side(_):
nonlocal n
n += 1
if n == 1:
return buf
session._active = False
return b""
ser.read.side_effect = side
session._trace_reader(ser)
def _session():
s = SimSession()
s._formatter = _fmt()
s._active = True
return s
def test_14a_reader_frame_decodes():
s = _session()
frames = []
s._on_frame = frames.append
s._print_frames = False
_run_sim_reader(s, _trace_frame(0, REQA))
assert len(frames) == 1 and len(s.entries) == 1
f = frames[0]
assert f["protocol"] == 1 # ISO 14443-A
assert f["direction"] == 0
assert f["data"] == REQA and f["data_hex"] == REQA.hex()
assert f["decoded"] == "REQA"
assert f["crc_fail"] is False
def test_14a_tag_frame_decodes():
s = _session()
_run_sim_reader(s, _trace_frame(1, ATQA))
assert len(s.entries) == 1
f = s.entries[0]
assert f["protocol"] == 1
assert f["direction"] == 1
assert f["decoded"] == "ATQA 04 00"
def test_14a_crc_fail_bit():
s = _session()
# High bit of the dir byte flags a bad-CRC frame (dir 0 | 0x80).
_run_sim_reader(s, _trace_frame(0x80, REQA))
assert len(s.entries) == 1
f = s.entries[0]
assert f["direction"] == 0
assert f["crc_fail"] is True
assert "BAD CRC" in f["decoded"]
def test_14a_quiet_suppresses_console(capsys):
s = _session()
frames = []
s._on_frame = frames.append
s._print_frames = False # quiet
_run_sim_reader(s, _trace_frame(0, REQA))
assert len(frames) == 1
assert capsys.readouterr().out == ""
def test_14a_entries_accumulate_and_clear():
s = _session()
buf = _trace_frame(0, REQA) + _trace_frame(1, ATQA)
_run_sim_reader(s, buf)
assert len(s.entries) == 2
assert [e["direction"] for e in s.entries] == [0, 1]
s.clear()
assert s.entries == []