feat(rawcli): deep, traced identify — exact model via GET_VERSION
`identify` now probes and shows its work instead of a coarse SAK guess: - Streams every probe exchange to the trace (proxmark-style annotated): the reconstructed 14a activation (WUPA/ATQA/anticollision CL1+CL2/ SELECT/SAK) plus a REAL GET_VERSION (0x60 +CRC) sent over the persistent connection. - Names the exact IC from GET_VERSION: an NTAG216 now reports "NTAG216", not "MIFARE UL/NTAG". Exact map mirrors the transponder models' _version_bytes (static, to dodge an import-order circular; a test guards against drift). Unknown chips fall back to product family + an honest memory-size *range* (AN10833 storage-byte encoding), not a bogus size. - SAK fallback table + GET_VERSION product-nibble map aligned to NXP AN10833 (MIFARE type identification procedure). 11 tests (exact/structural decode, map-model sync, traced probe, SAK names). GET_VERSION is a real device exchange — the user's hardware test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
michael
@@ -73,7 +73,8 @@ def dispatch(state: RawSession, line: str) -> bool:
|
||||
if cmd.name == "help":
|
||||
_handle_help(state, cmd.args)
|
||||
elif cmd.name == "identify":
|
||||
print(format_summary(identify(state)))
|
||||
result = identify(state, emit=print) # stream the probe exchanges to the trace
|
||||
print(format_summary(result))
|
||||
elif cmd.name == "transponder":
|
||||
print(state.transponder or "no transponder identified — run 'identify'")
|
||||
elif cmd.name == "close":
|
||||
|
||||
@@ -1,26 +1,75 @@
|
||||
"""Tag identification for rawcli.
|
||||
|
||||
``identify`` probes HF (ISO 14443-A, then 15693) and then LF, records the RF field, the wire
|
||||
protocol, and a display name on the session, and — because the 14a scan uses NO_DISCONNECT — the
|
||||
tag stays selected (the persistent connection). Device methods are called synchronously (the app
|
||||
connects via ``Proxmark3.sync()``).
|
||||
``identify`` doesn't just select — it *probes*, and every raw exchange it uses to figure the tag
|
||||
out is streamed to the trace (reconstructed 14a activation + a real GET_VERSION). The GET_VERSION
|
||||
response is reverse-mapped against the transponder models' own ``_version_bytes`` to name the
|
||||
exact chip (NTAG216, not "MIFARE UL/NTAG"); the version's storage-size byte is what separates
|
||||
NTAG213/215/216 and the Ultralight EV1 variants. The 14a scan uses NO_DISCONNECT, so the tag stays
|
||||
selected for the follow-up probes — the persistent connection.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
import sys
|
||||
|
||||
# Coarse SAK → 14a family, for the breadcrumb name (GET_VERSION-level precision is a later add).
|
||||
from pm3py.trace import TraceFormatter, decode_15693
|
||||
|
||||
ISO14A_APPEND_CRC = 0x20 # pm3_cmd.h: append CRC to a raw 14a frame
|
||||
|
||||
# SAK → 14a family for tags without GET_VERSION (NXP AN10833 "MIFARE type identification
|
||||
# procedure", Table 4). Used as the fallback when GET_VERSION isn't available/applicable.
|
||||
_SAK_NAMES = {
|
||||
0x00: "MIFARE UL/NTAG",
|
||||
0x00: "MIFARE Ultralight / NTAG", # -> GET_VERSION for the exact IC
|
||||
0x04: "UID incomplete (cascade)",
|
||||
0x08: "MIFARE Classic 1K",
|
||||
0x09: "MIFARE Mini",
|
||||
0x10: "MIFARE Plus 2K (SL2)",
|
||||
0x11: "MIFARE Plus 4K (SL2)",
|
||||
0x18: "MIFARE Classic 4K",
|
||||
0x10: "MIFARE Plus 2K",
|
||||
0x11: "MIFARE Plus 4K",
|
||||
0x20: "ISO14443-4",
|
||||
0x28: "JCOP",
|
||||
0x20: "ISO14443-4 (DESFire / Plus SL3 / …)",
|
||||
0x28: "MIFARE Classic 1K + NFC-P2P",
|
||||
0x38: "MIFARE Classic 4K + NFC-P2P",
|
||||
}
|
||||
|
||||
# GET_VERSION byte-2 lower nibble → product family (AN10833 Table 1). Upper nibble is the HW
|
||||
# type (0x0 native, 0x8 implementation, 0x9 Java Card applet, 0xA MIFARE 2GO / DUOX / X DNA).
|
||||
_GETVER_PRODUCT = {
|
||||
0x1: "MIFARE DESFire", 0x2: "MIFARE Plus", 0x3: "MIFARE Ultralight",
|
||||
0x4: "NTAG", 0x7: "NTAG I2C", 0x8: "MIFARE DESFire Light",
|
||||
}
|
||||
|
||||
# GET_VERSION 8-byte response → exact model. Mirrors the transponder models' `_version_bytes`
|
||||
# (kept static here to avoid an import-order circular; a test asserts they stay in sync).
|
||||
_VERSION_MODELS = {
|
||||
"0004040101000b03": "NTAG210",
|
||||
"0004040101000e03": "NTAG212",
|
||||
"0004040201000f03": "NTAG213",
|
||||
"0004040201001103": "NTAG215",
|
||||
"0004040201001303": "NTAG216",
|
||||
"0004030101000b03": "MIFARE Ultralight EV1 (MF0UL11)",
|
||||
"0004030101000e03": "MIFARE Ultralight EV1 (MF0UL21)",
|
||||
"0004040502021303": "NTAG I2C plus 1K (NT3H2111)",
|
||||
"0004040502021503": "NTAG I2C plus 2K (NT3H2211)",
|
||||
}
|
||||
|
||||
|
||||
def _storage_range(byte: int) -> str:
|
||||
"""GET_VERSION storage-size byte → memory-size string. The 7 MSBs code 2^n; the LSB set
|
||||
means the size is *between* 2^n and 2^(n+1) (AN10833)."""
|
||||
lo = 1 << (byte >> 1)
|
||||
return f"{lo}–{lo * 2} B" if byte & 1 else f"{lo} B"
|
||||
|
||||
|
||||
def decode_version(version: bytes) -> str | None:
|
||||
"""Name the chip from an 8-byte GET_VERSION response: exact match against the known models,
|
||||
else a structural best-effort (product family + memory-size range, AN10833)."""
|
||||
exact = _VERSION_MODELS.get(bytes(version[:8]).hex())
|
||||
if exact:
|
||||
return exact
|
||||
if len(version) >= 7 and version[1] == 0x04: # NXP vendor
|
||||
family = _GETVER_PRODUCT.get(version[2] & 0x0F, f"NXP type {version[2]:#04x}")
|
||||
return f"{family} ({_storage_range(version[6])})"
|
||||
return None
|
||||
|
||||
|
||||
def name_14a(scan: dict) -> str:
|
||||
sak = scan.get("sak")
|
||||
@@ -29,6 +78,57 @@ def name_14a(scan: dict) -> str:
|
||||
return "ISO14443-A"
|
||||
|
||||
|
||||
def _ident_decode(direction: int, payload: bytes) -> str | None:
|
||||
"""Annotator for the identify probe frames (activation + GET_VERSION)."""
|
||||
if not payload:
|
||||
return None
|
||||
if direction == 0: # reader → tag
|
||||
head = payload[:2]
|
||||
if payload[:1] == b"\x52":
|
||||
return "WUPA"
|
||||
if payload[:1] == b"\x26":
|
||||
return "REQA"
|
||||
if payload[0] in (0x93, 0x95):
|
||||
cl = 1 if payload[0] == 0x93 else 2
|
||||
if len(payload) >= 2 and payload[1] == 0x20:
|
||||
return f"ANTICOLLISION CL{cl}"
|
||||
if len(payload) >= 2 and payload[1] == 0x70:
|
||||
return f"SELECT CL{cl}"
|
||||
if payload[0] == 0x60:
|
||||
return "GET_VERSION"
|
||||
if payload[0] == 0x50:
|
||||
return "HALT"
|
||||
else: # tag → reader
|
||||
if len(payload) >= 8 and payload[0] == 0x00 and payload[1] == 0x04:
|
||||
return "VERSION"
|
||||
return None
|
||||
|
||||
|
||||
def _activation_frames(scan: dict) -> list[tuple[int, bytes]]:
|
||||
"""Reconstruct the ISO 14443-3A activation (WUPA/ATQA/anticollision/SELECT/SAK) from the scan
|
||||
result, so the user sees the select the reader performed."""
|
||||
try:
|
||||
uid = bytes.fromhex(scan.get("uid") or "")
|
||||
atqa = bytes.fromhex(scan.get("atqa") or "")
|
||||
except ValueError:
|
||||
return []
|
||||
sak = scan.get("sak", 0)
|
||||
frames: list[tuple[int, bytes]] = [(0, b"\x52"), (1, atqa)]
|
||||
if len(uid) == 4:
|
||||
bcc = uid[0] ^ uid[1] ^ uid[2] ^ uid[3]
|
||||
frames += [(0, b"\x93\x20"), (1, uid + bytes([bcc])),
|
||||
(0, b"\x93\x70" + uid + bytes([bcc])), (1, bytes([sak]))]
|
||||
elif len(uid) == 7:
|
||||
ct = 0x88
|
||||
b1 = ct ^ uid[0] ^ uid[1] ^ uid[2]
|
||||
b2 = uid[3] ^ uid[4] ^ uid[5] ^ uid[6]
|
||||
frames += [(0, b"\x93\x20"), (1, bytes([ct]) + uid[0:3] + bytes([b1])),
|
||||
(0, b"\x93\x70" + bytes([ct]) + uid[0:3] + bytes([b1])), (1, b"\x04"),
|
||||
(0, b"\x95\x20"), (1, uid[3:7] + bytes([b2])),
|
||||
(0, b"\x95\x70" + uid[3:7] + bytes([b2])), (1, bytes([sak]))]
|
||||
return frames
|
||||
|
||||
|
||||
def _try(fn, default):
|
||||
try:
|
||||
return fn()
|
||||
@@ -36,9 +136,10 @@ def _try(fn, default):
|
||||
return default
|
||||
|
||||
|
||||
def identify(session) -> dict:
|
||||
"""Probe the field and update ``session`` (field / protocol / transponder). Returns a
|
||||
summary dict (``found`` + details)."""
|
||||
def identify(session, emit=None) -> dict:
|
||||
"""Probe the field and update ``session`` (field / protocol / transponder). ``emit`` receives
|
||||
each annotated trace line (default: no trace). Returns a summary dict."""
|
||||
emit = emit or (lambda _line: None)
|
||||
dev = session.device
|
||||
if dev is None:
|
||||
return {"found": False, "error": "no device connected"}
|
||||
@@ -46,16 +147,23 @@ def identify(session) -> dict:
|
||||
scan = _try(lambda: dev.hf.iso14a.scan(), {"found": False})
|
||||
if scan.get("found"):
|
||||
session.field, session.protocol = "hf", "hf14a"
|
||||
session.transponder = name_14a(scan)
|
||||
fmt = TraceFormatter(mode="reader", decoder=_ident_decode, crc_len=0)
|
||||
for direction, data in _activation_frames(scan):
|
||||
emit(fmt.format(direction, data).lstrip("\n"))
|
||||
model, version = None, None
|
||||
if scan.get("sak") == 0x00: # UL/NTAG family — ask GET_VERSION
|
||||
model, version = _probe_version(dev, fmt, emit)
|
||||
session.transponder = model or name_14a(scan)
|
||||
return {"found": True, "field": "hf", "protocol": "14a",
|
||||
"transponder": session.transponder, "uid": scan.get("uid"), "raw": scan}
|
||||
"transponder": session.transponder, "uid": scan.get("uid"),
|
||||
"version": version.hex() if version else None}
|
||||
|
||||
scan15 = _try(lambda: dev.hf.iso15.scan(), {"found": False})
|
||||
if scan15.get("found") or scan15.get("uid"):
|
||||
session.field, session.protocol = "hf", "hf15"
|
||||
session.transponder = "ISO15693"
|
||||
return {"found": True, "field": "hf", "protocol": "15693",
|
||||
"transponder": "ISO15693", "uid": scan15.get("uid"), "raw": scan15}
|
||||
"transponder": "ISO15693", "uid": scan15.get("uid")}
|
||||
|
||||
lf = _try(lambda: dev.lf.search(), None)
|
||||
if lf:
|
||||
@@ -68,6 +176,22 @@ def identify(session) -> dict:
|
||||
return {"found": False}
|
||||
|
||||
|
||||
def _probe_version(dev, fmt, emit) -> tuple[str | None, bytes | None]:
|
||||
"""Send a real GET_VERSION (0x60 +CRC), trace the exchange, decode the exact model."""
|
||||
cmd = b"\x60"
|
||||
emit(fmt.format(0, cmd).lstrip("\n"))
|
||||
resp = _try(lambda: dev.hf.iso14a.raw(cmd, flags=ISO14A_APPEND_CRC), None)
|
||||
raw = resp.get("raw") if isinstance(resp, dict) else None
|
||||
if not raw:
|
||||
return None, None
|
||||
emit(fmt.format(1, bytes(raw)).lstrip("\n"))
|
||||
version = bytes(raw[:8])
|
||||
model = decode_version(version)
|
||||
if model:
|
||||
emit(f" → {model}")
|
||||
return model, version
|
||||
|
||||
|
||||
def clear(session) -> None:
|
||||
"""Forget the identified tag (drop the logical connection); the device stays open."""
|
||||
session.field = None
|
||||
|
||||
@@ -146,11 +146,15 @@ class TestDispatch:
|
||||
assert "whole number of bytes" in capsys.readouterr().out
|
||||
|
||||
|
||||
def _mock_device(scan14=None, scan15=None):
|
||||
NTAG216_VERSION = bytes.fromhex("0004040201001303")
|
||||
|
||||
|
||||
def _mock_device(scan14=None, scan15=None, version=None):
|
||||
dev = MagicMock()
|
||||
dev.hf.iso14a.scan.return_value = scan14 or {"found": False}
|
||||
dev.hf.iso15.scan.return_value = scan15 or {"found": False}
|
||||
dev.lf.search.return_value = None
|
||||
dev.hf.iso14a.raw.return_value = {"raw": (version + b"\x99\x88") if version else None}
|
||||
return dev
|
||||
|
||||
|
||||
@@ -160,13 +164,22 @@ class TestIdentify:
|
||||
r = identify(s)
|
||||
assert r["found"] is False and s.field is None
|
||||
|
||||
def test_14a_sets_field_and_name(self):
|
||||
def test_14a_sak_only_name(self):
|
||||
# SAK != 0 -> named from AN10833 Table 4, no GET_VERSION
|
||||
s = RawSession(device=_mock_device(
|
||||
scan14={"found": True, "sak": 0x08, "uid": b"\x01\x02\x03\x04"}))
|
||||
scan14={"found": True, "sak": 0x08, "atqa": "0400", "uid": "01020304"}))
|
||||
r = identify(s)
|
||||
assert r["found"] and s.field == "hf" and s.protocol == "hf14a"
|
||||
assert s.transponder == "MIFARE Classic 1K"
|
||||
assert r["uid"] == b"\x01\x02\x03\x04"
|
||||
|
||||
def test_14a_getversion_exact(self):
|
||||
s = RawSession(device=_mock_device(
|
||||
scan14={"found": True, "sak": 0x00, "atqa": "4400", "uid": "04a1b2c3d4e5f6"},
|
||||
version=NTAG216_VERSION))
|
||||
r = identify(s)
|
||||
assert s.transponder == "NTAG216" # exact, not "MIFARE UL/NTAG"
|
||||
assert r["version"] == NTAG216_VERSION.hex()
|
||||
s.device.hf.iso14a.raw.assert_called_with(b"\x60", flags=0x20) # real GET_VERSION+CRC
|
||||
|
||||
def test_15693_fallback(self):
|
||||
s = RawSession(device=_mock_device(scan15={"found": True, "uid": b"\xE0\x04"}))
|
||||
@@ -174,24 +187,52 @@ class TestIdentify:
|
||||
assert r["found"] and s.field == "hf" and s.protocol == "hf15"
|
||||
assert s.transponder == "ISO15693"
|
||||
|
||||
def test_name_unknown_sak(self):
|
||||
def test_sak_names_from_an10833(self):
|
||||
assert name_14a({"sak": 0x18}) == "MIFARE Classic 4K"
|
||||
assert name_14a({"sak": 0x20}).startswith("ISO14443-4")
|
||||
assert name_14a({"sak": 0x99}) == "ISO14443-A (SAK 99)"
|
||||
|
||||
def test_clear_forgets_tag(self):
|
||||
s = RawSession(device=_mock_device(scan14={"found": True, "sak": 0x08}))
|
||||
s = RawSession(device=_mock_device(scan14={"found": True, "sak": 0x08, "uid": "01020304"}))
|
||||
identify(s)
|
||||
clear(s)
|
||||
assert s.field is None and s.transponder is None and s.protocol == "hf14a"
|
||||
|
||||
|
||||
class TestVersionId:
|
||||
def test_exact_models(self):
|
||||
from pm3py.cli.rawcli.identify import decode_version
|
||||
assert decode_version(bytes.fromhex("0004040201001303")) == "NTAG216"
|
||||
assert decode_version(bytes.fromhex("0004040201000f03")) == "NTAG213"
|
||||
assert "Ultralight EV1" in decode_version(bytes.fromhex("0004030101000b03"))
|
||||
|
||||
def test_structural_range_for_unknown(self):
|
||||
from pm3py.cli.rawcli.identify import decode_version
|
||||
# unknown NTAG-family version -> product + honest size range (AN10833)
|
||||
out = decode_version(bytes.fromhex("0004040201001503"))
|
||||
assert out.startswith("NTAG (") and "B)" in out
|
||||
|
||||
def test_static_map_matches_models(self):
|
||||
# guard against drift between the static version map and the transponder models
|
||||
from pm3py.cli.rawcli.identify import _VERSION_MODELS
|
||||
from pm3py.sim import (NTAG210, NTAG212, NTAG213, NTAG215, NTAG216,
|
||||
MF0UL11, MF0UL21, NT3H2111, NT3H2211)
|
||||
model_vbytes = {c._version_bytes.hex() for c in
|
||||
(NTAG210, NTAG212, NTAG213, NTAG215, NTAG216,
|
||||
MF0UL11, MF0UL21, NT3H2111, NT3H2211)}
|
||||
assert set(_VERSION_MODELS) == model_vbytes
|
||||
|
||||
|
||||
class TestControlCommands:
|
||||
def test_identify_command(self, capsys):
|
||||
s = RawSession(device=_mock_device(
|
||||
scan14={"found": True, "sak": 0x00, "uid": b"\x04\x11\x22\x33"}))
|
||||
scan14={"found": True, "sak": 0x00, "atqa": "4400", "uid": "04a1b2c3d4e5f6"},
|
||||
version=NTAG216_VERSION))
|
||||
dispatch(s, "identify")
|
||||
out = capsys.readouterr().out
|
||||
assert "MIFARE UL/NTAG" in out and "04112233" in out
|
||||
assert s.breadcrumb() == "[raw / hf / MIFARE UL/NTAG / 0x]"
|
||||
out = _plain(capsys.readouterr().out)
|
||||
assert "NTAG216" in out # exact model in the summary
|
||||
assert "WUPA" in out and "GET_VERSION" in out # probe exchanges streamed to the trace
|
||||
assert s.breadcrumb() == "[raw / hf / NTAG216 / 0x]"
|
||||
|
||||
def test_transponder_when_none(self, capsys):
|
||||
dispatch(RawSession(), "transponder")
|
||||
@@ -231,7 +272,7 @@ class TestCatalog:
|
||||
class TestFunctionCalls:
|
||||
def _id(self, sak=0x00):
|
||||
return RawSession(device=_mock_device(
|
||||
scan14={"found": True, "sak": sak, "uid": b"\x04\x01\x02\x03"}))
|
||||
scan14={"found": True, "sak": sak, "atqa": "4400", "uid": "04010203"}))
|
||||
|
||||
def test_call_read_sends_correct_bytes(self, capsys):
|
||||
s = self._id()
|
||||
|
||||
Reference in New Issue
Block a user