f09946c4a3
A real download came back as BigBuf memory (0xDEADBEEF poison) with a PM3 response magic (0x62334d50) embedded — the classic signature of a desynced read: the stream had stale bytes (a prior partial transaction, or the C client having touched the port), and the bulk reader consumes fixed 544-byte OLD frames that carry no magic to resync on, so once misaligned it stays misaligned and reads memory as "samples". transport.download_bulk now flushes the input first (reset_input_buffer + drain the asyncio StreamReader until an 80ms quiet gap), so the transfer always begins byte- aligned. Drains no-op when disconnected. Test asserts the flush runs before the request. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
154 lines
6.1 KiB
Python
154 lines
6.1 KiB
Python
import struct
|
|
from pm3py.core.transport import encode_ng_frame, decode_response_frame, encode_mix_frame, _crc_to_wire
|
|
from pm3py.core.protocol import Cmd, CMD_PREAMBLE_MAGIC, CMD_POSTAMBLE_NOCRC, RESP_PREAMBLE_MAGIC, RESP_POSTAMBLE_NOCRC, crc16_a
|
|
|
|
def test_encode_ng_frame_ping_no_payload():
|
|
frame = encode_ng_frame(Cmd.PING, b"")
|
|
# 8 byte preamble + 0 payload + 2 postamble = 10 bytes
|
|
assert len(frame) == 10
|
|
magic, length_ng, cmd = struct.unpack_from("<IHH", frame, 0)
|
|
assert magic == CMD_PREAMBLE_MAGIC
|
|
assert length_ng & 0x8000 # ng bit set
|
|
assert (length_ng & 0x7FFF) == 0
|
|
assert cmd == Cmd.PING
|
|
# Default: no CRC (USB mode)
|
|
crc_actual = struct.unpack_from("<H", frame, 8)[0]
|
|
assert crc_actual == CMD_POSTAMBLE_NOCRC
|
|
|
|
def test_encode_ng_frame_with_payload_no_crc():
|
|
"""Default USB mode: no CRC, uses magic postamble."""
|
|
payload = bytes(range(32))
|
|
frame = encode_ng_frame(Cmd.PING, payload)
|
|
assert len(frame) == 10 + 32
|
|
_, length_ng, _ = struct.unpack_from("<IHH", frame, 0)
|
|
assert (length_ng & 0x7FFF) == 32
|
|
assert frame[8:40] == payload
|
|
crc_actual = struct.unpack_from("<H", frame, 40)[0]
|
|
assert crc_actual == CMD_POSTAMBLE_NOCRC
|
|
|
|
def test_encode_ng_frame_with_crc():
|
|
"""FPC UART mode: real CRC, byte-swapped."""
|
|
payload = bytes(range(32))
|
|
frame = encode_ng_frame(Cmd.PING, payload, use_crc=True)
|
|
crc_expected = _crc_to_wire(crc16_a(frame[:40]))
|
|
crc_actual = struct.unpack_from("<H", frame, 40)[0]
|
|
assert crc_actual == crc_expected
|
|
|
|
def test_encode_mix_frame():
|
|
frame = encode_mix_frame(Cmd.HF_ISO14443A_READER, 0x107, 0, 0, b"")
|
|
_, length_ng, _ = struct.unpack_from("<IHH", frame, 0)
|
|
assert not (length_ng & 0x8000) # ng bit NOT set for MIX
|
|
assert (length_ng & 0x7FFF) == 24 # 3x uint64 = 24 bytes
|
|
|
|
def test_decode_response_frame_no_crc():
|
|
"""USB mode: response uses no-CRC magic."""
|
|
payload = b"\x01\x02\x03\x04"
|
|
length_ng = len(payload) | 0x8000
|
|
preamble = struct.pack("<IHbbH", RESP_PREAMBLE_MAGIC, length_ng, 0, 0, Cmd.PING)
|
|
body = preamble + payload
|
|
postamble = struct.pack("<H", RESP_POSTAMBLE_NOCRC)
|
|
frame = body + postamble
|
|
|
|
resp = decode_response_frame(frame)
|
|
assert resp["cmd"] == Cmd.PING
|
|
assert resp["status"] == 0
|
|
assert resp["data"] == payload
|
|
assert resp["ng"] is True
|
|
|
|
def test_decode_response_frame_with_crc():
|
|
"""FPC UART mode: response has real CRC, byte-swapped."""
|
|
payload = b"\x01\x02\x03\x04"
|
|
length_ng = len(payload) | 0x8000
|
|
preamble = struct.pack("<IHbbH", RESP_PREAMBLE_MAGIC, length_ng, 0, 0, Cmd.PING)
|
|
body = preamble + payload
|
|
crc = _crc_to_wire(crc16_a(body))
|
|
postamble = struct.pack("<H", crc)
|
|
frame = body + postamble
|
|
|
|
resp = decode_response_frame(frame)
|
|
assert resp["cmd"] == Cmd.PING
|
|
assert resp["status"] == 0
|
|
assert resp["data"] == payload
|
|
|
|
def test_decode_response_frame_error_status():
|
|
length_ng = 0 | 0x8000
|
|
preamble = struct.pack("<IHbbH", RESP_PREAMBLE_MAGIC, length_ng, -4, -1, Cmd.STATUS)
|
|
postamble = struct.pack("<H", RESP_POSTAMBLE_NOCRC)
|
|
frame = preamble + postamble
|
|
|
|
resp = decode_response_frame(frame)
|
|
assert resp["status"] == -4 # ETIMEOUT
|
|
assert resp["data"] == b""
|
|
|
|
|
|
# --- bulk download (BigBuf): one request -> OLD data frames -> ACK -----------------------------
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock
|
|
from pm3py.core.transport import PM3Transport, PM3Response
|
|
from pm3py.core.protocol import PM3_CMD_DATA_SIZE
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.get_event_loop().run_until_complete(coro)
|
|
|
|
|
|
def _bigbuf_frame(offset, data):
|
|
# OLD-format CMD_DOWNLOADED_BIGBUF: oldarg=[offset, len, tracelen], data padded to 512
|
|
return PM3Response(cmd=int(Cmd.DOWNLOADED_BIGBUF), status=0, reason=0, ng=False,
|
|
data=data + b"\x00" * (PM3_CMD_DATA_SIZE - len(data)),
|
|
oldarg=[offset, len(data), 4242])
|
|
|
|
|
|
def _ack():
|
|
return PM3Response(cmd=int(Cmd.ACK), status=0, reason=0, ng=False, data=b"", oldarg=[1, 0, 0])
|
|
|
|
|
|
def _transport_with(frames):
|
|
t = PM3Transport("/dev/null")
|
|
t.send_frame = AsyncMock()
|
|
t._read_any_response = AsyncMock(side_effect=frames)
|
|
return t
|
|
|
|
|
|
def test_download_bulk_count_terminated():
|
|
payload = bytes((i * 7) % 256 for i in range(1224))
|
|
t = _transport_with([_bigbuf_frame(0, payload[0:512]), _bigbuf_frame(512, payload[512:1024]),
|
|
_bigbuf_frame(1024, payload[1024:1224]), _ack()])
|
|
got = _run(t.download_bulk(Cmd.DOWNLOAD_BIGBUF, 0, 1224, Cmd.DOWNLOADED_BIGBUF, Cmd.ACK))
|
|
assert got == payload # exactly count bytes, reassembled in order
|
|
assert t.send_frame.call_count == 1 # ONE request, not one per chunk
|
|
|
|
|
|
def test_download_bulk_ack_terminated_early():
|
|
# device has fewer bytes than requested -> stops at the ACK, returns what came
|
|
payload = bytes(range(200))
|
|
t = _transport_with([_bigbuf_frame(0, payload), _ack()])
|
|
got = _run(t.download_bulk(Cmd.DOWNLOAD_BIGBUF, 0, 99999, Cmd.DOWNLOADED_BIGBUF, Cmd.ACK))
|
|
assert got == payload
|
|
|
|
|
|
def test_download_bulk_uses_oldarg_length_not_padding():
|
|
# the OLD frame is always 512 bytes; only oldarg[1] bytes are valid (rest is padding)
|
|
t = _transport_with([_bigbuf_frame(0, b"\xAA\xBB\xCC"), _ack()])
|
|
got = _run(t.download_bulk(Cmd.DOWNLOAD_BIGBUF, 0, 3, Cmd.DOWNLOADED_BIGBUF, Cmd.ACK))
|
|
assert got == b"\xAA\xBB\xCC" # trimmed to oldarg length, not 512 of padding
|
|
|
|
|
|
def test_download_bulk_drains_stale_input_first():
|
|
# a bulk download must flush stale bytes before sending, so OLD frames start byte-aligned
|
|
payload = bytes(range(100))
|
|
t = _transport_with([_bigbuf_frame(0, payload), _ack()])
|
|
calls = []
|
|
async def fake_drain(quiet=0.08):
|
|
calls.append("drain")
|
|
return 7
|
|
orig_send = t.send_frame
|
|
async def tracking_send(frame):
|
|
calls.append("send")
|
|
return await orig_send(frame)
|
|
t._drain_input = fake_drain
|
|
t.send_frame = tracking_send
|
|
_run(t.download_bulk(Cmd.DOWNLOAD_BIGBUF, 0, 100, Cmd.DOWNLOADED_BIGBUF, Cmd.ACK))
|
|
assert calls[0] == "drain" and calls[1] == "send" # drained BEFORE the request went out
|