refactor: extract sniff infrastructure into pm3py/sniff/ sub-package
Phase 2 of package refactor. Extracts trace parsing, ISO 15693 decoders, NDEF annotation, and ANSI formatting from core/hf_iso15.py into dedicated sniff/ modules: - sniff/trace.py — parse_tracelog (protocol-agnostic) - sniff/ndef.py — NDEF TLV/record decode - sniff/decode_iso15.py — 15693 command/response decoders - sniff/format.py — ANSI color formatting - sniff/session.py — SniffSession with iso15() method core/hf_iso15.py re-exports sniff symbols for backward compat. Also fixes NTAG 5 placement (iso15693, not iso14443a4) in design doc. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
michael
@@ -1,9 +1,16 @@
|
||||
"""ISO 15693 commands: hf.15.*"""
|
||||
"""ISO 15693 commands: hf.iso15.*"""
|
||||
import struct
|
||||
import sys
|
||||
from .protocol import Cmd, PM3_CMD_DATA_SIZE
|
||||
from .protocol import Cmd
|
||||
from .transport import PM3Transport, PM3Error
|
||||
|
||||
# Re-export sniff symbols for backward compat with existing test imports
|
||||
from ..sniff.trace import parse_tracelog, TRACELOG_HDR_SIZE
|
||||
from ..sniff.decode_iso15 import (
|
||||
decode_15693_request, decode_15693_response, decode_15693,
|
||||
)
|
||||
from ..sniff.ndef import decode_ndef_annotation
|
||||
from ..sniff.format import format_sniff_line
|
||||
|
||||
# ISO15 PM3 flags
|
||||
ISO15_CONNECT = 0x01
|
||||
ISO15_NO_DISCONNECT = 0x02
|
||||
@@ -11,461 +18,6 @@ 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("<IHH", raw, offset)
|
||||
is_response = bool(data_len_flags & 0x8000)
|
||||
data_len = data_len_flags & 0x7FFF
|
||||
offset += TRACELOG_HDR_SIZE
|
||||
|
||||
if data_len == 0:
|
||||
break # end of trace / sentinel
|
||||
|
||||
if offset + data_len > 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."""
|
||||
@@ -562,54 +114,17 @@ class HF15Commands:
|
||||
# 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])
|
||||
"""Download and parse the trace buffer from the device."""
|
||||
from ..sniff.session import SniffSession
|
||||
session = SniffSession(self._t)
|
||||
return await session._download_trace()
|
||||
|
||||
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.
|
||||
Convenience method — delegates to SniffSession.iso15().
|
||||
Prefer using SniffSession directly for new code.
|
||||
"""
|
||||
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
|
||||
from ..sniff.session import SniffSession
|
||||
session = SniffSession(self._t)
|
||||
return await session.iso15(timeout=timeout)
|
||||
|
||||
@@ -1 +1,14 @@
|
||||
"""pm3py.sniff — Sniff sessions, trace parsing, protocol decoders, formatting."""
|
||||
from .session import SniffSession
|
||||
from .trace import parse_tracelog, TRACELOG_HDR_SIZE
|
||||
from .decode_iso15 import decode_15693, decode_15693_request, decode_15693_response
|
||||
from .ndef import decode_ndef_annotation
|
||||
from .format import format_sniff_line
|
||||
|
||||
__all__ = [
|
||||
"SniffSession",
|
||||
"parse_tracelog", "TRACELOG_HDR_SIZE",
|
||||
"decode_15693", "decode_15693_request", "decode_15693_response",
|
||||
"decode_ndef_annotation",
|
||||
"format_sniff_line",
|
||||
]
|
||||
|
||||
146
pm3py/sniff/decode_iso15.py
Normal file
146
pm3py/sniff/decode_iso15.py
Normal file
@@ -0,0 +1,146 @@
|
||||
"""ISO 15693 request/response decoders and command tables."""
|
||||
from .ndef import decode_ndef_annotation
|
||||
|
||||
# ---- 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)
|
||||
118
pm3py/sniff/format.py
Normal file
118
pm3py/sniff/format.py
Normal file
@@ -0,0 +1,118 @@
|
||||
"""ANSI color formatting and sniff line output."""
|
||||
from .decode_iso15 import decode_15693
|
||||
|
||||
# ---- 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)
|
||||
152
pm3py/sniff/ndef.py
Normal file
152
pm3py/sniff/ndef.py
Normal file
@@ -0,0 +1,152 @@
|
||||
"""NDEF TLV/record 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]"
|
||||
75
pm3py/sniff/session.py
Normal file
75
pm3py/sniff/session.py
Normal file
@@ -0,0 +1,75 @@
|
||||
"""Sniff sessions — start/stop/download per protocol."""
|
||||
import sys
|
||||
|
||||
from ..core.protocol import Cmd, PM3_CMD_DATA_SIZE
|
||||
from ..core.transport import PM3Transport, encode_ng_frame
|
||||
from .trace import parse_tracelog
|
||||
from .format import format_sniff_line
|
||||
|
||||
|
||||
class SniffSession:
|
||||
"""Sniff session that manages start/download/decode lifecycle.
|
||||
|
||||
Takes a PM3Transport (not the client), since sniff has its own
|
||||
lifecycle that doesn't fit the stateless command pattern.
|
||||
"""
|
||||
|
||||
def __init__(self, transport: PM3Transport):
|
||||
self._t = transport
|
||||
|
||||
async def _sniff_15(self, timeout: float = 60.0) -> dict:
|
||||
"""Start ISO 15693 sniff. Blocks until button press or timeout."""
|
||||
async with self._t._lock:
|
||||
raw = encode_ng_frame(Cmd.HF_ISO15693_SNIFF)
|
||||
await self._t.send_frame(raw)
|
||||
while True:
|
||||
resp = await self._t.read_response(timeout=timeout)
|
||||
if resp.cmd == Cmd.HF_ISO15693_SNIFF:
|
||||
return {"status": resp.status}
|
||||
|
||||
async def _download_trace(self) -> list[dict]:
|
||||
"""Download and parse the trace buffer from the device."""
|
||||
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_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 iso15(self, timeout: float = 60.0) -> list[dict]:
|
||||
"""Sniff ISO 15693 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_15(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
|
||||
41
pm3py/sniff/trace.py
Normal file
41
pm3py/sniff/trace.py
Normal file
@@ -0,0 +1,41 @@
|
||||
"""Protocol-agnostic trace buffer parsing."""
|
||||
import struct
|
||||
|
||||
# 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("<IHH", raw, offset)
|
||||
is_response = bool(data_len_flags & 0x8000)
|
||||
data_len = data_len_flags & 0x7FFF
|
||||
offset += TRACELOG_HDR_SIZE
|
||||
|
||||
if data_len == 0:
|
||||
break # end of trace / sentinel
|
||||
|
||||
if offset + data_len > 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
|
||||
Reference in New Issue
Block a user