diff --git a/pm3py/cli/rawcli/app.py b/pm3py/cli/rawcli/app.py index 33d3d2f..7dc5861 100644 --- a/pm3py/cli/rawcli/app.py +++ b/pm3py/cli/rawcli/app.py @@ -12,35 +12,45 @@ import sys from .session import RawSession from .parser import parse_line from .entry import install_key_bindings +from .identify import identify, clear, format_summary from .trace_view import render_exchange _HELP = """\ rawcli — raw command interface help show this help + identify probe the field; sets HF/LF + transponder, keeps the connection + transponder show the identified transponder + close drop the tag connection quit | exit | q leave rawcli send raw bytes to the tag; the annotated exchange prints above -Type hex as "30 04" or "3004". Entry-mode toggle (0x/0b), identify, and function-style -transponder commands (READ(4), GET_DATA) land in later phases.""" +Type hex as "30 04" or "3004"; Ctrl-/ toggles hex/binary entry (0x/0b). Function-style +transponder commands (READ(4), GET_DATA) land in the next phase.""" def _connect(port): """Best-effort device connection; returns the client or None (offline).""" try: from pm3py import Proxmark3 - return Proxmark3(port) + return Proxmark3.sync(port) # sync proxy: scan()/raw() are plain calls except Exception as exc: # no device / connect failure — run offline print(f"rawcli: no device ({type(exc).__name__}: {exc}); running offline", file=sys.stderr) return None -def _raw_exchange(device, payload: bytes) -> bytes | None: - """Issue one raw 14a exchange, returning the response bytes (or None).""" +def _raw_exchange(device, payload: bytes, protocol: str = "hf14a") -> bytes | None: + """Issue one raw exchange over the identified protocol, returning the response bytes.""" if device is None: return None try: - resp = device.hf.iso14a.raw(payload) + if protocol == "hf14a": + resp = device.hf.iso14a.raw(payload) + elif protocol == "hf15": + raw = getattr(device.hf.iso15, "raw", None) + resp = raw(payload) if raw else None + else: + return None # lf raw exchange lands in a later phase return resp.get("data") if isinstance(resp, dict) else None except Exception as exc: print(f"rawcli: exchange failed ({type(exc).__name__}: {exc})", file=sys.stderr) @@ -61,9 +71,13 @@ def dispatch(state: RawSession, line: str) -> bool: return False if cmd.name == "help": print(_HELP) - else: - # identify / transponder / close land in Phase 3 - print(f"rawcli: '{cmd.name}' is not wired yet (coming in a later phase)") + elif cmd.name == "identify": + print(format_summary(identify(state))) + elif cmd.name == "transponder": + print(state.transponder or "no transponder identified — run 'identify'") + elif cmd.name == "close": + clear(state) + print("connection closed") return True if cmd.kind == "call": @@ -74,8 +88,8 @@ def dispatch(state: RawSession, line: str) -> bool: # raw payload if not cmd.payload: return True - response = _raw_exchange(state.device, cmd.payload) - print(render_exchange(cmd.payload, response, protocol="hf14a", + response = _raw_exchange(state.device, cmd.payload, protocol=state.protocol) + print(render_exchange(cmd.payload, response, protocol=state.protocol, is_tty=sys.stdout.isatty())) return True diff --git a/pm3py/cli/rawcli/identify.py b/pm3py/cli/rawcli/identify.py new file mode 100644 index 0000000..8357f93 --- /dev/null +++ b/pm3py/cli/rawcli/identify.py @@ -0,0 +1,86 @@ +"""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()``). +""" +from __future__ import annotations + +from typing import Any + +# Coarse SAK → 14a family, for the breadcrumb name (GET_VERSION-level precision is a later add). +_SAK_NAMES = { + 0x00: "MIFARE UL/NTAG", + 0x08: "MIFARE Classic 1K", + 0x09: "MIFARE Mini", + 0x18: "MIFARE Classic 4K", + 0x10: "MIFARE Plus 2K", + 0x11: "MIFARE Plus 4K", + 0x20: "ISO14443-4", + 0x28: "JCOP", +} + + +def name_14a(scan: dict) -> str: + sak = scan.get("sak") + if isinstance(sak, int): + return _SAK_NAMES.get(sak, f"ISO14443-A (SAK {sak:02X})") + return "ISO14443-A" + + +def _try(fn, default): + try: + return fn() + except Exception: + return default + + +def identify(session) -> dict: + """Probe the field and update ``session`` (field / protocol / transponder). Returns a + summary dict (``found`` + details).""" + dev = session.device + if dev is None: + return {"found": False, "error": "no device connected"} + + scan = _try(lambda: dev.hf.iso14a.scan(), {"found": False}) + if scan.get("found"): + session.field, session.protocol = "hf", "hf14a" + session.transponder = name_14a(scan) + return {"found": True, "field": "hf", "protocol": "14a", + "transponder": session.transponder, "uid": scan.get("uid"), "raw": scan} + + 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} + + lf = _try(lambda: dev.lf.search(), None) + if lf: + session.field, session.protocol = "lf", "lf" + name = getattr(lf, "tag_type", None) or getattr(lf, "type", None) or "LF tag" + session.transponder = str(name) + return {"found": True, "field": "lf", "protocol": "lf", + "transponder": session.transponder, "raw": lf} + + return {"found": False} + + +def clear(session) -> None: + """Forget the identified tag (drop the logical connection); the device stays open.""" + session.field = None + session.transponder = None + session.protocol = "hf14a" + dev = session.device + if dev is not None: + _try(lambda: dev.hf.dropfield(), None) + + +def format_summary(result: dict) -> str: + if not result.get("found"): + return f"no tag found{': ' + result['error'] if result.get('error') else ''}" + uid = result.get("uid") + uid_hex = uid.hex().upper() if isinstance(uid, (bytes, bytearray)) else (uid or "?") + return f"{result['transponder']} ({result['field'].upper()} / {result['protocol']}) UID {uid_hex}" diff --git a/pm3py/cli/rawcli/session.py b/pm3py/cli/rawcli/session.py index c20515c..a92b728 100644 --- a/pm3py/cli/rawcli/session.py +++ b/pm3py/cli/rawcli/session.py @@ -21,6 +21,7 @@ class RawSession: self.device = device self.field: str | None = None self.transponder: str | None = None + self.protocol: str = "hf14a" # wire protocol for raw exchanges (set by identify) self.entry_mode: str = HEX self.connected: bool = device is not None diff --git a/tests/test_rawcli.py b/tests/test_rawcli.py index 09270d2..5c36147 100644 --- a/tests/test_rawcli.py +++ b/tests/test_rawcli.py @@ -1,6 +1,8 @@ """rawcli Phase 1 — session state, breadcrumb, trace rendering, CLI dispatch. Hardware-free.""" import re +from unittest.mock import MagicMock + import pytest from pm3py.cli.main import build_parser, main @@ -8,6 +10,7 @@ from pm3py.cli.rawcli.session import RawSession from pm3py.cli.rawcli.trace_view import render_exchange from pm3py.cli.rawcli.parser import parse_token, parse_bytes, parse_line from pm3py.cli.rawcli.entry import byte_space +from pm3py.cli.rawcli.identify import identify, clear, name_14a from pm3py.cli.rawcli.app import dispatch _ANSI = re.compile(r"\x1b\[[0-9;]*m") @@ -138,3 +141,62 @@ class TestDispatch: def test_bad_hex_reports(self, capsys): assert dispatch(RawSession(), "0b0011") is True assert "whole number of bytes" in capsys.readouterr().out + + +def _mock_device(scan14=None, scan15=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 + return dev + + +class TestIdentify: + def test_no_device(self): + s = RawSession() + r = identify(s) + assert r["found"] is False and s.field is None + + def test_14a_sets_field_and_name(self): + s = RawSession(device=_mock_device( + scan14={"found": True, "sak": 0x08, "uid": b"\x01\x02\x03\x04"})) + 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_15693_fallback(self): + s = RawSession(device=_mock_device(scan15={"found": True, "uid": b"\xE0\x04"})) + r = identify(s) + assert r["found"] and s.field == "hf" and s.protocol == "hf15" + assert s.transponder == "ISO15693" + + def test_name_unknown_sak(self): + 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})) + identify(s) + clear(s) + assert s.field is None and s.transponder is None and s.protocol == "hf14a" + + +class TestControlCommands: + def test_identify_command(self, capsys): + s = RawSession(device=_mock_device( + scan14={"found": True, "sak": 0x00, "uid": b"\x04\x11\x22\x33"})) + 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]" + + def test_transponder_when_none(self, capsys): + dispatch(RawSession(), "transponder") + assert "run 'identify'" in capsys.readouterr().out + + def test_close(self, capsys): + s = RawSession(device=_mock_device(scan14={"found": True, "sak": 0x08})) + dispatch(s, "identify") + dispatch(s, "close") + assert "closed" in capsys.readouterr().out + assert s.transponder is None