Merges the package refactor (core/ sub-package, normalized naming, scaffolded sub-packages) while preserving all in-progress sniff infrastructure. Git rename detection carried WIP changes to new paths. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
289 lines
9.9 KiB
Python
289 lines
9.9 KiB
Python
import asyncio
|
|
import struct
|
|
from unittest.mock import AsyncMock
|
|
from pm3py.core.protocol import Cmd
|
|
from pm3py.core.transport import PM3Response
|
|
from pm3py.core.hf_iso15 import (
|
|
HF15Commands, parse_tracelog, TRACELOG_HDR_SIZE,
|
|
decode_15693_request, decode_15693_response, decode_15693,
|
|
format_sniff_line, decode_ndef_annotation,
|
|
)
|
|
|
|
def test_15_scan():
|
|
t = AsyncMock()
|
|
iso15 = HF15Commands(t)
|
|
# Fake response: flags(1) + DSFID(1) + UID(8)
|
|
resp_data = bytes([0x00, 0x00]) + bytes([0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0xE0])
|
|
t.send_ng.return_value = PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=0,
|
|
reason=0, ng=True, data=resp_data)
|
|
result = asyncio.get_event_loop().run_until_complete(iso15.scan())
|
|
assert "uid" in result
|
|
|
|
|
|
# ---- Tracelog parsing ----
|
|
|
|
def _make_tracelog_entry(timestamp, duration, is_response, data):
|
|
"""Build a raw tracelog entry matching firmware's tracelog_hdr_t."""
|
|
data_len_flags = len(data) | (0x8000 if is_response else 0)
|
|
hdr = struct.pack("<IHH", timestamp, duration, data_len_flags)
|
|
parity_len = (len(data) + 7) // 8 if data else 0
|
|
parity = bytes(parity_len)
|
|
return hdr + data + parity
|
|
|
|
|
|
def test_parse_tracelog_single_entry():
|
|
raw = _make_tracelog_entry(1000, 50, False, b"\x26\x01\x00\xAA\xBB")
|
|
entries = parse_tracelog(raw)
|
|
assert len(entries) == 1
|
|
assert entries[0]["timestamp"] == 1000
|
|
assert entries[0]["duration"] == 50
|
|
assert entries[0]["is_response"] is False
|
|
assert entries[0]["data"] == b"\x26\x01\x00\xAA\xBB"
|
|
|
|
|
|
def test_parse_tracelog_two_entries():
|
|
raw = (
|
|
_make_tracelog_entry(1000, 50, False, b"\x26\x01\x00\xAA\xBB")
|
|
+ _make_tracelog_entry(2000, 100, True, b"\x00\x00\x01\x02\x03\x04\x05\x06\x07\xE0\xCC\xDD")
|
|
)
|
|
entries = parse_tracelog(raw)
|
|
assert len(entries) == 2
|
|
assert entries[0]["is_response"] is False
|
|
assert entries[1]["is_response"] is True
|
|
|
|
|
|
def test_parse_tracelog_empty():
|
|
entries = parse_tracelog(b"")
|
|
assert entries == []
|
|
|
|
|
|
def test_parse_tracelog_zero_sentinel():
|
|
raw = _make_tracelog_entry(1000, 50, False, b"\x26\x01\x00\xAA\xBB")
|
|
raw += b"\x00" * TRACELOG_HDR_SIZE # zero sentinel
|
|
raw += _make_tracelog_entry(9999, 50, False, b"\xFF") # should not parse
|
|
entries = parse_tracelog(raw)
|
|
assert len(entries) == 1
|
|
|
|
|
|
# ---- 15693 decode ----
|
|
|
|
def test_decode_inventory_request():
|
|
ann = decode_15693_request(b"\x26\x01\x00")
|
|
assert ann == "INVENTORY"
|
|
|
|
|
|
def test_decode_inventory_with_mask():
|
|
ann = decode_15693_request(b"\x26\x01\x08\xFF")
|
|
assert "mask=8" in ann
|
|
|
|
|
|
def test_decode_read_single_unaddressed():
|
|
ann = decode_15693_request(b"\x02\x20\x05")
|
|
assert ann == "READ SINGLE BLOCK #5"
|
|
|
|
|
|
def test_decode_read_single_addressed():
|
|
# flags=0x22 (address + high data rate), cmd=0x20, uid(8), block=3
|
|
payload = bytes([0x22, 0x20]) + bytes(8) + bytes([3])
|
|
ann = decode_15693_request(payload)
|
|
assert ann == "READ SINGLE BLOCK #3"
|
|
|
|
|
|
def test_decode_read_multiple():
|
|
ann = decode_15693_request(b"\x02\x23\x00\x0F")
|
|
assert "READ MULTIPLE BLOCKS #0+15" in ann
|
|
|
|
|
|
def test_decode_get_system_info():
|
|
ann = decode_15693_request(b"\x02\x2B")
|
|
assert ann == "GET SYSTEM INFO"
|
|
|
|
|
|
def test_decode_nxp_set_password():
|
|
ann = decode_15693_request(bytes([0x02, 0xB3, 0x04, 0x04]))
|
|
assert "NXP SET PASSWORD" in ann
|
|
assert "id=4" in ann
|
|
|
|
|
|
def test_decode_unknown_command():
|
|
ann = decode_15693_request(bytes([0x02, 0xFE]))
|
|
assert "UNKNOWN CMD 0xFE" in ann
|
|
|
|
|
|
def test_decode_response_ok():
|
|
ann = decode_15693_response(b"\x00")
|
|
assert ann == "OK"
|
|
|
|
|
|
def test_decode_response_ok_data():
|
|
ann = decode_15693_response(b"\x00\xDE\xAD\xBE\xEF")
|
|
assert "OK [4B]" in ann
|
|
|
|
|
|
def test_decode_response_inventory():
|
|
# flags(1) + dsfid(1) + uid(8) = 10 bytes total, 9 after flags
|
|
payload = bytes([0x00, 0x00]) + bytes([0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0xE0])
|
|
ann = decode_15693_response(payload)
|
|
assert "INVENTORY" in ann
|
|
assert "UID=" in ann
|
|
|
|
|
|
def test_decode_response_error():
|
|
ann = decode_15693_response(bytes([0x01, 0x10]))
|
|
assert "ERROR" in ann
|
|
assert "block not available" in ann
|
|
|
|
|
|
def test_decode_15693_dispatches():
|
|
assert decode_15693(0, b"\x02\x2B") == "GET SYSTEM INFO"
|
|
assert "OK" in decode_15693(1, b"\x00")
|
|
|
|
|
|
# ---- format_sniff_line ----
|
|
|
|
def test_format_sniff_line_reader():
|
|
entry = {"is_response": False, "data": b"\x26\x01\x00\xAA\xBB",
|
|
"timestamp": 0, "duration": 0, "parity": b""}
|
|
line = format_sniff_line(entry, width=120, is_tty=False)
|
|
assert "Reader" in line
|
|
assert "Tag" in line
|
|
assert "INVENTORY" in line
|
|
|
|
|
|
def test_format_sniff_line_tag():
|
|
payload = bytes([0x00, 0x00]) + bytes([0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0xE0]) + bytes([0xCC, 0xDD])
|
|
entry = {"is_response": True, "data": payload,
|
|
"timestamp": 0, "duration": 0, "parity": b""}
|
|
line = format_sniff_line(entry, width=120, is_tty=False)
|
|
assert "Tag" in line
|
|
assert "INVENTORY" in line
|
|
|
|
|
|
# ---- NDEF decode ----
|
|
|
|
class TestDecodeNdefAnnotation:
|
|
def test_cc_block(self):
|
|
data = bytes([0xE1, 0x40, 0x0D, 0x01])
|
|
ann = decode_ndef_annotation(data)
|
|
assert ann is not None
|
|
assert "CC" in ann
|
|
assert "v1.0" in ann
|
|
assert "MBREAD" in ann
|
|
|
|
def test_cc_no_mbread(self):
|
|
data = bytes([0xE1, 0x40, 0x10, 0x00])
|
|
ann = decode_ndef_annotation(data)
|
|
assert "CC" in ann
|
|
assert "MBREAD" not in ann
|
|
|
|
def test_ndef_text_record(self):
|
|
msg = bytes([0xD1, 0x01, 0x07, 0x54, 0x02, 0x65, 0x6E, 0x74, 0x65, 0x73, 0x74])
|
|
tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
|
|
ann = decode_ndef_annotation(tlv)
|
|
assert ann is not None
|
|
assert '"test"' in ann
|
|
|
|
def test_ndef_uri_record(self):
|
|
uri_payload = bytes([0x04]) + b"example.com"
|
|
rec = bytes([0xD1, 0x01, len(uri_payload), 0x55]) + uri_payload
|
|
tlv = bytes([0x03, len(rec)]) + rec + bytes([0xFE])
|
|
ann = decode_ndef_annotation(tlv)
|
|
assert "https://example.com" in ann
|
|
|
|
def test_ndef_mime_record(self):
|
|
mime_type = b"text/plain"
|
|
payload = b"hello"
|
|
rec = bytes([0xD2, len(mime_type), len(payload)]) + mime_type + payload
|
|
tlv = bytes([0x03, len(rec)]) + rec + bytes([0xFE])
|
|
ann = decode_ndef_annotation(tlv)
|
|
assert "text/plain" in ann
|
|
|
|
def test_no_ndef_magic(self):
|
|
ann = decode_ndef_annotation(bytes([0x00, 0x00, 0x00, 0x00]))
|
|
assert ann is None
|
|
|
|
def test_cc_plus_ndef(self):
|
|
cc = bytes([0xE1, 0x40, 0x0D, 0x01])
|
|
msg = bytes([0xD1, 0x01, 0x07, 0x54, 0x02, 0x65, 0x6E, 0x74, 0x65, 0x73, 0x74])
|
|
tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
|
|
data = cc + tlv + bytes(50)
|
|
ann = decode_ndef_annotation(data)
|
|
assert "CC" in ann
|
|
assert '"test"' in ann
|
|
|
|
def test_empty_ndef(self):
|
|
tlv = bytes([0x03, 0x00, 0xFE])
|
|
ann = decode_ndef_annotation(tlv)
|
|
assert ann is not None
|
|
assert "empty" in ann.lower() or "NDEF" in ann
|
|
|
|
def test_long_text_truncated(self):
|
|
# Build a long NDEF text record manually (no sim.type5 dependency)
|
|
long_text = "A" * 200
|
|
lang = b"en"
|
|
payload = bytes([len(lang)]) + lang + long_text.encode("utf-8")
|
|
rec = bytes([0xD1, 0x01, len(payload), 0x54]) + payload
|
|
tlv = bytes([0x03, 0xFF, (len(rec) >> 8) & 0xFF, len(rec) & 0xFF]) + rec + bytes([0xFE])
|
|
ann = decode_ndef_annotation(tlv)
|
|
assert ann is not None
|
|
assert "..." in ann
|
|
|
|
|
|
class TestResponseNdefDecode:
|
|
def test_read_single_block_0_cc(self):
|
|
data = bytes([0x00, 0xE1, 0x40, 0x0D, 0x01])
|
|
ann = decode_15693_response(data)
|
|
assert "CC" in ann
|
|
assert "MBREAD" in ann
|
|
|
|
def test_read_multiple_blocks_ndef(self):
|
|
msg = bytes([0xD1, 0x01, 0x07, 0x54, 0x02, 0x65, 0x6E, 0x74, 0x65, 0x73, 0x74])
|
|
tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
|
|
data = bytes([0x00]) + tlv + bytes(50)
|
|
ann = decode_15693_response(data)
|
|
assert '"test"' in ann
|
|
|
|
def test_read_multiple_with_cc_and_ndef(self):
|
|
cc = bytes([0xE1, 0x40, 0x0D, 0x01])
|
|
msg = bytes([0xD1, 0x01, 0x07, 0x54, 0x02, 0x65, 0x6E, 0x74, 0x65, 0x73, 0x74])
|
|
tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
|
|
data = bytes([0x00]) + cc + tlv + bytes(40)
|
|
ann = decode_15693_response(data)
|
|
assert "CC" in ann
|
|
assert '"test"' in ann
|
|
|
|
def test_generic_data_no_ndef(self):
|
|
data = bytes([0x00, 0xDE, 0xAD, 0xBE, 0xEF])
|
|
ann = decode_15693_response(data)
|
|
assert "OK [4B]" in ann
|
|
|
|
def test_inventory_not_affected(self):
|
|
data = bytes([0x00, 0x00]) + bytes(8)
|
|
ann = decode_15693_response(data)
|
|
assert "INVENTORY" in ann
|
|
|
|
|
|
class TestSniffLineAnnotationOverflow:
|
|
def test_long_annotation_wraps(self):
|
|
cc = bytes([0xE1, 0x40, 0x0D, 0x01])
|
|
# Build NDEF text record manually (no sim.type5 dependency)
|
|
text = "A" * 30
|
|
lang = b"en"
|
|
payload = bytes([len(lang)]) + lang + text.encode("utf-8")
|
|
msg = bytes([0xD1, 0x01, len(payload), 0x54]) + payload
|
|
tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
|
|
payload = bytes([0x00]) + cc + tlv + bytes([0xAA, 0xBB])
|
|
entry = {"is_response": True, "data": payload,
|
|
"timestamp": 0, "duration": 0, "parity": b""}
|
|
line = format_sniff_line(entry, width=80, is_tty=False)
|
|
lines = line.split("\n")
|
|
ann_lines = [l for l in lines if "NDEF" in l]
|
|
assert len(ann_lines) >= 1
|
|
|
|
def test_short_annotation_no_wrap(self):
|
|
data = bytes([0x00, 0xE1, 0x40, 0x0D, 0x01, 0xAA, 0xBB])
|
|
entry = {"is_response": True, "data": data,
|
|
"timestamp": 0, "duration": 0, "parity": b""}
|
|
line = format_sniff_line(entry, width=120, is_tty=False)
|
|
assert "CC" in line
|