diff --git a/pm3py/cli/rawcli/app.py b/pm3py/cli/rawcli/app.py index cd79239..005604b 100644 --- a/pm3py/cli/rawcli/app.py +++ b/pm3py/cli/rawcli/app.py @@ -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": diff --git a/pm3py/cli/rawcli/identify.py b/pm3py/cli/rawcli/identify.py index 8357f93..12f0e9f 100644 --- a/pm3py/cli/rawcli/identify.py +++ b/pm3py/cli/rawcli/identify.py @@ -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 diff --git a/tests/test_rawcli.py b/tests/test_rawcli.py index 3dbdf02..a9dbdda 100644 --- a/tests/test_rawcli.py +++ b/tests/test_rawcli.py @@ -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()