diff --git a/.gitignore b/.gitignore index cf268a5..16a80a5 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,7 @@ __pycache__/ .pytest_cache/ venv/ .worktrees/ +*.pdf + +# Capture data (sniffs, sims, taginfo scans) +data/ diff --git a/firmware b/firmware index 8f8b9c5..206c0f0 160000 --- a/firmware +++ b/firmware @@ -1 +1 @@ -Subproject commit 8f8b9c51a1a58d85f6e124d0dd0708b1566ecdae +Subproject commit 206c0f0fbc192b27cdc294173dcea1465b356ef5 diff --git a/pm3py/hf_15.py b/pm3py/hf_15.py index 348c32f..4ee45cb 100644 --- a/pm3py/hf_15.py +++ b/pm3py/hf_15.py @@ -1,6 +1,7 @@ """ISO 15693 commands: hf.15.*""" import struct -from .protocol import Cmd +import sys +from .protocol import Cmd, PM3_CMD_DATA_SIZE from .transport import PM3Transport, PM3Error # ISO15 PM3 flags @@ -10,6 +11,461 @@ ISO15_RAW = 0x04 ISO15_APPEND_CRC = 0x08 ISO15_READ_RESPONSE = 0x10 +# tracelog_hdr_t size: timestamp(4) + duration(2) + data_len_flags(2) = 8 +TRACELOG_HDR_SIZE = 8 + + +def parse_tracelog(raw: bytes) -> list[dict]: + """Parse PM3 trace buffer into a list of trace entries. + + Each entry: {timestamp, duration, is_response, data, parity} + """ + entries = [] + offset = 0 + while offset + TRACELOG_HDR_SIZE <= len(raw): + timestamp, duration, data_len_flags = struct.unpack_from(" len(raw): + break + + data = raw[offset:offset + data_len] + offset += data_len + + parity_len = (data_len + 7) // 8 if data_len > 0 else 0 + parity = raw[offset:offset + parity_len] if parity_len else b"" + offset += parity_len + + entries.append({ + "timestamp": timestamp, + "duration": duration, + "is_response": is_response, + "data": data, + "parity": parity, + }) + return entries + + +# ---- NDEF decode (for trace annotation) ---- + +_NDEF_MAGIC = 0xE1 + +_URI_PREFIXES = { + 0x00: "", 0x01: "http://www.", 0x02: "https://www.", + 0x03: "http://", 0x04: "https://", 0x05: "tel:", 0x06: "mailto:", +} + +_TNF_EMPTY = 0x00 +_TNF_WELL_KNOWN = 0x01 +_TNF_MEDIA = 0x02 +_TNF_URI = 0x03 +_TNF_EXTERNAL = 0x04 +_MAX_ANN_TEXT = 40 + + +def decode_ndef_annotation(data: bytes) -> str | None: + """Decode NDEF TLV data into a human-readable annotation string. + + Handles CC (capability container) blocks and NDEF message TLVs. + Returns None if the data doesn't contain recognizable NDEF content. + """ + if len(data) < 3: + return None + + parts = [] + offset = 0 + + if data[0] == _NDEF_MAGIC and len(data) >= 4: + ver_major = (data[1] >> 6) & 0x03 + ver_minor = (data[1] >> 4) & 0x03 + mlen = data[2] + features = data[3] + cc_str = f"CC v{ver_major}.{ver_minor} MLEN={mlen}" + if features & 0x01: + cc_str += " MBREAD" + parts.append(cc_str) + offset = 4 + if offset >= len(data): + return cc_str + + while offset < len(data): + tlv_type = data[offset] + offset += 1 + if tlv_type == 0x00: + continue + if tlv_type == 0xFE: + break + if tlv_type != 0x03: + break + + if offset >= len(data): + break + tlv_len = data[offset] + offset += 1 + if tlv_len == 0xFF: + if offset + 2 > len(data): + break + tlv_len = (data[offset] << 8) | data[offset + 1] + offset += 2 + + if tlv_len == 0: + parts.append("NDEF empty") + continue + + msg_end = min(offset + tlv_len, len(data)) + records = _parse_ndef_records(data[offset:msg_end]) + for rec in records: + parts.append(rec) + offset = msg_end + + return " | ".join(parts) if parts else None + + +def _parse_ndef_records(msg: bytes) -> list[str]: + """Parse NDEF records from a message payload.""" + records = [] + offset = 0 + while offset < len(msg): + if offset + 3 > len(msg): + break + flags = msg[offset] + tnf = flags & 0x07 + sr = bool(flags & 0x10) + il = bool(flags & 0x08) + type_len = msg[offset + 1] + offset += 2 + + if sr: + if offset >= len(msg): + break + payload_len = msg[offset] + offset += 1 + else: + if offset + 4 > len(msg): + break + payload_len = int.from_bytes(msg[offset:offset + 4], "big") + offset += 4 + + id_len = 0 + if il: + if offset >= len(msg): + break + id_len = msg[offset] + offset += 1 + + if offset + type_len > len(msg): + break + rec_type = msg[offset:offset + type_len] + offset += type_len + offset += id_len + + if offset + payload_len > len(msg): + payload = msg[offset:] + else: + payload = msg[offset:offset + payload_len] + offset += payload_len + + records.append(_decode_ndef_record(tnf, rec_type, payload)) + return records + + +def _decode_ndef_record(tnf: int, rec_type: bytes, payload: bytes) -> str: + """Decode a single NDEF record into an annotation string.""" + if tnf == _TNF_WELL_KNOWN: + if rec_type == b"T" and len(payload) >= 1: + lang_len = payload[0] & 0x3F + lang = payload[1:1 + lang_len].decode("ascii", errors="replace") + text = payload[1 + lang_len:].decode("utf-8", errors="replace") + if len(text) > _MAX_ANN_TEXT: + text = text[:_MAX_ANN_TEXT] + "..." + return f'NDEF Text "{lang}" "{text}"' + if rec_type == b"U" and len(payload) >= 1: + prefix = _URI_PREFIXES.get(payload[0], "") + suffix = payload[1:].decode("utf-8", errors="replace") + uri = prefix + suffix + if len(uri) > _MAX_ANN_TEXT: + uri = uri[:_MAX_ANN_TEXT] + "..." + return f"NDEF URI {uri}" + return f"NDEF WK type={rec_type!r} [{len(payload)}B]" + if tnf == _TNF_MEDIA: + mime = rec_type.decode("ascii", errors="replace") + if len(mime) > 30: + mime = mime[:30] + "..." + return f"NDEF MIME {mime} [{len(payload)}B]" + if tnf == _TNF_EXTERNAL: + ext_type = rec_type.decode("ascii", errors="replace") + return f"NDEF EXT {ext_type} [{len(payload)}B]" + if tnf == _TNF_EMPTY: + return "NDEF empty" + return f"NDEF TNF={tnf} [{len(payload)}B]" + + +# ---- ISO 15693 command names (for sniff annotation) ---- +_15693_CMDS = { + 0x01: "INVENTORY", + 0x02: "STAY QUIET", + 0x20: "READ SINGLE BLOCK", + 0x21: "WRITE SINGLE BLOCK", + 0x22: "LOCK BLOCK", + 0x23: "READ MULTIPLE BLOCKS", + 0x24: "WRITE MULTIPLE BLOCKS", + 0x25: "SELECT", + 0x26: "RESET TO READY", + 0x27: "WRITE AFI", + 0x28: "LOCK AFI", + 0x29: "WRITE DSFID", + 0x2A: "LOCK DSFID", + 0x2B: "GET SYSTEM INFO", + 0x2C: "GET MULTIPLE BLOCK SECURITY", +} + +_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", +} + +_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", +} + +_FLAG_INVENTORY = 0x04 +_FLAG_ADDRESS = 0x20 + + +def _block_offset(flags: int) -> int: + """Byte offset of block number after flags+cmd.""" + if not (flags & _FLAG_INVENTORY) and (flags & _FLAG_ADDRESS): + return 10 # flags(1) + cmd(1) + uid(8) + return 2 # flags(1) + cmd(1) + + +def decode_15693_request(payload: bytes) -> str | None: + """Decode an ISO 15693 request frame into annotation.""" + if len(payload) < 2: + return 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}" + + 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 == 0xB3 and len(payload) >= 4: # NXP SET PASSWORD + return f"{name} id={payload[3]}" + + return name + + +def decode_15693_response(payload: bytes) -> str | None: + """Decode an ISO 15693 response frame into annotation.""" + 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, "") + return f"ERROR 0x{code:02X} {desc}" if desc else f"ERROR 0x{code:02X}" + return "ERROR" + + 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 + ndef_ann = decode_ndef_annotation(data) + if ndef_ann: + return f"OK {ndef_ann}" + + return f"OK [{len(data)}B]" + + +def decode_15693(direction: int, payload: bytes) -> str | None: + """Decode an ISO 15693 frame for sniff trace.""" + if direction == 0: + return decode_15693_request(payload) + else: + return decode_15693_response(payload) + + +# ---- ANSI colors ---- +_C_CYAN = "\033[36m" +_C_YELLOW = "\033[33m" +_C_DIM = "\033[2m" +_C_CRC = "\033[1;37m" +_C_RED = "\033[31m" +_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}" + + +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}" + if len(candidate) <= avail: + current = candidate + else: + lines.append(current) + current = part + lines.append(current) + return lines + return [annotation[:avail - 3] + "..."] + + +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. + """ + 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) + class HF15Commands: """ISO 15693 commands.""" @@ -87,7 +543,73 @@ class HF15Commands: success = resp.status == 0 and len(resp.data) > 0 and resp.data[0] == 0 return {"success": success, "block": block} - async def sniff(self) -> dict: - """Sniff ISO15693 traffic.""" - resp = await self._t.send_ng(Cmd.HF_ISO15693_SNIFF, timeout=30.0) - return {"status": resp.status} + async def sniff(self, timeout: float = 60.0) -> dict: + """Start ISO15693 sniff. Blocks until button press or timeout. + + Returns {"status": int}. + """ + from .transport import encode_ng_frame + async with self._t._lock: + raw = encode_ng_frame(Cmd.HF_ISO15693_SNIFF) + await self._t.send_frame(raw) + + # Firmware sends debug prints before the sniff completion response. + # Keep reading until we get the actual sniff response. + while True: + resp = await self._t.read_response(timeout=timeout) + if resp.cmd == Cmd.HF_ISO15693_SNIFF: + return {"status": resp.status} + # Debug print or other message — discard and keep waiting + + async def download_trace(self) -> list[dict]: + """Download and parse the trace buffer from the device. + + Returns list of trace entries, each with: + timestamp, duration, is_response, data, parity + """ + # First download to discover trace length + raw, trace_len = await self._t.download_bigbuf( + offset=0, length=PM3_CMD_DATA_SIZE, timeout=4.0) + + if trace_len == 0: + return [] + + # If trace is larger than first chunk, download the full thing + if trace_len > PM3_CMD_DATA_SIZE: + raw, trace_len = await self._t.download_bigbuf( + offset=0, length=trace_len, timeout=10.0) + + return parse_tracelog(raw[:trace_len]) + + async def sniff_decoded(self, timeout: float = 60.0) -> list[dict]: + """Sniff ISO15693 traffic, download trace, decode and print. + + Blocks until button press or timeout. Then downloads the trace, + decodes each frame, and prints formatted output. + + Returns the parsed trace entries. + """ + import os + try: + width = os.get_terminal_size().columns + except (OSError, ValueError): + width = 120 + is_tty = sys.stdout.isatty() + + print("[Snf] Sniffing ISO 15693... press PM3 button to stop.") + result = await self.sniff(timeout=timeout) + print(f"[Snf] Sniff ended (status={result['status']})") + + print("[Snf] Downloading trace...") + entries = await self.download_trace() + + if not entries: + print("[Snf] No trace data captured.") + return [] + + print(f"[Snf] {len(entries)} frames captured:\n") + for entry in entries: + line = format_sniff_line(entry, width=width, is_tty=is_tty) + print(line) + + return entries diff --git a/pm3py/transport.py b/pm3py/transport.py index 341d428..3b91b50 100644 --- a/pm3py/transport.py +++ b/pm3py/transport.py @@ -233,3 +233,34 @@ class PM3Transport: async with self._lock: frame = encode_ng_frame(cmd, payload) await self.send_frame(frame) + + async def download_bigbuf(self, offset: int = 0, length: int = PM3_CMD_DATA_SIZE, + timeout: float = 4.0) -> tuple[bytes, int]: + """Download data from device BigBuf. + + Firmware responds with reply_mix (NG frame with MIX args) for both + data chunks and the final ACK. + + Returns (data, trace_len). + """ + async with self._lock: + frame = encode_mix_frame(Cmd.DOWNLOAD_BIGBUF, arg0=offset, arg1=length) + await self.send_frame(frame) + + buf = bytearray(length) + trace_len = 0 + + while True: + resp = await self.read_response(timeout=timeout) + + if resp.cmd == Cmd.ACK: + if resp.oldarg: + trace_len = resp.oldarg[2] + return bytes(buf[:trace_len or length]), trace_len + + if resp.cmd == Cmd.DOWNLOADED_BIGBUF and resp.oldarg: + chunk_offset = resp.oldarg[0] + chunk_len = min(resp.oldarg[1], len(resp.data)) + trace_len = resp.oldarg[2] + end = min(chunk_offset + chunk_len, length) + buf[chunk_offset:end] = resp.data[:end - chunk_offset] diff --git a/tests/test_hf_15.py b/tests/test_hf_15.py index fe687c0..72ff4ed 100644 --- a/tests/test_hf_15.py +++ b/tests/test_hf_15.py @@ -1,8 +1,14 @@ import asyncio +import struct from unittest.mock import AsyncMock from pm3py.protocol import Cmd from pm3py.transport import PM3Response -from pm3py.hf_15 import HF15Commands +from pm3py.hf_15 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() @@ -13,3 +19,271 @@ def test_15_scan(): 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("> 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