diff --git a/pm3py/cli/rawcli/app.py b/pm3py/cli/rawcli/app.py index 7dc5861..6dfdab4 100644 --- a/pm3py/cli/rawcli/app.py +++ b/pm3py/cli/rawcli/app.py @@ -13,6 +13,7 @@ from .session import RawSession from .parser import parse_line from .entry import install_key_bindings from .identify import identify, clear, format_summary +from .catalog import catalog_for from .trace_view import render_exchange _HELP = """\ @@ -70,7 +71,7 @@ def dispatch(state: RawSession, line: str) -> bool: if cmd.name in ("quit", "exit", "q"): return False if cmd.name == "help": - print(_HELP) + _handle_help(state, cmd.args) elif cmd.name == "identify": print(format_summary(identify(state))) elif cmd.name == "transponder": @@ -81,8 +82,7 @@ def dispatch(state: RawSession, line: str) -> bool: return True if cmd.kind == "call": - # function-style transponder commands land in Phase 4 (the command catalog) - print(f"rawcli: {cmd.name}(...) needs an identified transponder + catalog (Phase 4)") + _handle_call(state, cmd) return True # raw payload @@ -94,6 +94,40 @@ def dispatch(state: RawSession, line: str) -> bool: return True +def _handle_call(state: RawSession, cmd) -> None: + catalog = catalog_for(state) + if catalog is None: + print("no transponder identified — run 'identify' first") + return + tc = catalog.get(cmd.name) + if tc is None: + print(f"unknown command {cmd.name!r} for {catalog.name}; try 'help'") + return + try: + payload = tc.build(*cmd.args) + except (TypeError, ValueError) as exc: + print(f"usage: {tc.usage()} ({exc})") + return + response = _raw_exchange(state.device, payload, protocol=catalog.protocol) + print(render_exchange(payload, response, protocol=catalog.protocol, + is_tty=sys.stdout.isatty())) + + +def _handle_help(state: RawSession, args) -> None: + catalog = catalog_for(state) + if args and catalog is not None: + tc = catalog.get(args[0]) + if tc is not None: + print(f"{tc.usage()} — {tc.help}") + return + print(_HELP) + if catalog is not None: + print(f"\n{catalog.name} commands:") + for name in catalog.names(): + tc = catalog.get(name) + print(f" {tc.usage():<28}{tc.help}") + + def run(port=None) -> int: from prompt_toolkit import PromptSession from prompt_toolkit.key_binding import KeyBindings diff --git a/pm3py/cli/rawcli/catalog.py b/pm3py/cli/rawcli/catalog.py new file mode 100644 index 0000000..99d0e63 --- /dev/null +++ b/pm3py/cli/rawcli/catalog.py @@ -0,0 +1,102 @@ +"""Per-transponder command catalogs. + +A declarative, enumerable command set for each tag family: each entry knows its parameters, how +to **build** the raw bytes to send, and its help text. rawcli surfaces these for ``help`` and for +function-style calls (``READ(4)`` → build → raw exchange → annotated trace). Homed here for now; +the richer per-vendor catalogs can migrate into the ``pm3py/reader/`` scaffold as they grow. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable + + +@dataclass +class TagCommand: + name: str + params: tuple[str, ...] + build: Callable[..., bytes] # string args -> raw payload bytes + help: str = "" + + def usage(self) -> str: + return f"{self.name}({', '.join(self.params)})" + + +@dataclass +class Catalog: + name: str + protocol: str + commands: dict[str, TagCommand] + + def get(self, name: str) -> TagCommand | None: + return self.commands.get(name.upper()) + + def names(self) -> list[str]: + return list(self.commands) + + +def _int(s: str) -> int: + return int(s, 0) # accepts decimal or 0x-prefixed + + +def _hex(s: str) -> bytes: + return bytes.fromhex(s.lower().replace("0x", "").replace(" ", "")) + + +def _catalog(name: str, protocol: str, *cmds: TagCommand) -> Catalog: + return Catalog(name, protocol, {c.name.upper(): c for c in cmds}) + + +def _c(name, params, build, help="") -> TagCommand: + return TagCommand(name, tuple(params), build, help) + + +# ---- ISO 14443-A Type 2 (NTAG / Ultralight) ---- +TYPE2 = _catalog( + "NTAG / Ultralight (Type 2)", "hf14a", + _c("READ", ["page"], lambda page: bytes([0x30, _int(page)]), + "read 4 pages starting at (0x30)"), + _c("FAST_READ", ["start", "end"], lambda start, end: bytes([0x3A, _int(start), _int(end)]), + "read pages .. in one frame (0x3A)"), + _c("GET_VERSION", [], lambda: bytes([0x60]), "product/version info (0x60)"), + _c("READ_SIG", [], lambda: bytes([0x3C, 0x00]), "read the ECC originality signature (0x3C)"), + _c("READ_CNT", ["counter"], lambda counter="0": bytes([0x39, _int(counter)]), + "read the 24-bit NFC counter (0x39)"), + _c("WRITE", ["page", "data"], + lambda page, data: bytes([0xA2, _int(page)]) + _hex(data).ljust(4, b"\x00")[:4], + "write 4 bytes to (0xA2)"), +) + +# ---- MIFARE Classic (block ops; auth is the reader's job before these) ---- +MFC = _catalog( + "MIFARE Classic", "hf14a", + _c("READ", ["block"], lambda block: bytes([0x30, _int(block)]), + "read a 16-byte block (needs prior auth) (0x30)"), + _c("WRITE", ["block"], lambda block: bytes([0xA0, _int(block)]), + "start a block write, two-phase (0xA0)"), + _c("HALT", [], lambda: bytes([0x50, 0x00]), "halt the card (0x50 00)"), +) + +# ---- ISO 15693 (flags byte 0x02 = high data rate, single tag) ---- +ISO15 = _catalog( + "ISO 15693", "hf15", + _c("INVENTORY", [], lambda: bytes([0x26, 0x01, 0x00]), "anticollision inventory (0x01)"), + _c("READ_BLOCK", ["block"], lambda block: bytes([0x02, 0x20, _int(block)]), + "read a single block (0x20)"), + _c("WRITE_BLOCK", ["block", "data"], + lambda block, data: bytes([0x02, 0x21, _int(block)]) + _hex(data), + "write a single block (0x21)"), + _c("GET_SYSTEM_INFO", [], lambda: bytes([0x02, 0x2B]), "get system information (0x2B)"), +) + + +def catalog_for(session) -> Catalog | None: + """Resolve the command catalog for the session's identified transponder, or None.""" + name = (session.transponder or "").upper() + if session.protocol == "hf15": + return ISO15 + if session.protocol == "hf14a": + if not session.transponder: + return None + return MFC if "CLASSIC" in name else TYPE2 + return None diff --git a/tests/test_rawcli.py b/tests/test_rawcli.py index 5c36147..be2110a 100644 --- a/tests/test_rawcli.py +++ b/tests/test_rawcli.py @@ -11,6 +11,7 @@ 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.catalog import TYPE2, ISO15, catalog_for, catalog_for as _cf from pm3py.cli.rawcli.app import dispatch _ANSI = re.compile(r"\x1b\[[0-9;]*m") @@ -200,3 +201,67 @@ class TestControlCommands: dispatch(s, "close") assert "closed" in capsys.readouterr().out assert s.transponder is None + + +class TestCatalog: + def test_type2_builds(self): + assert TYPE2.get("READ").build("4") == b"\x30\x04" + assert TYPE2.get("read").build("0x04") == b"\x30\x04" # case + hex arg + assert TYPE2.get("FAST_READ").build("0", "3") == b"\x3A\x00\x03" + assert TYPE2.get("GET_VERSION").build() == b"\x60" + assert TYPE2.get("WRITE").build("4", "01020304") == b"\xA2\x04\x01\x02\x03\x04" + + def test_iso15_builds(self): + assert ISO15.get("READ_BLOCK").build("4") == b"\x02\x20\x04" + assert ISO15.get("INVENTORY").build() == b"\x26\x01\x00" + + def test_resolver(self): + s = RawSession() + assert catalog_for(s) is None # nothing identified + s.protocol, s.transponder = "hf14a", "NTAG215" + assert catalog_for(s) is TYPE2 + s.transponder = "MIFARE Classic 1K" + assert catalog_for(s).name == "MIFARE Classic" + s.protocol, s.transponder = "hf15", "ISO15693" + assert catalog_for(s) is ISO15 + + +class TestFunctionCalls: + def _id(self, sak=0x00): + return RawSession(device=_mock_device( + scan14={"found": True, "sak": sak, "uid": b"\x04\x01\x02\x03"})) + + def test_call_read_sends_correct_bytes(self, capsys): + s = self._id() + dispatch(s, "identify") + s.device.hf.iso14a.raw.return_value = {"data": b"\xAA\xBB\xCC\xDD"} + capsys.readouterr() + dispatch(s, "READ(4)") + out = _plain(capsys.readouterr().out) + assert "30 04" in out # the built command hit the trace + s.device.hf.iso14a.raw.assert_called_with(b"\x30\x04") + + def test_call_without_identify(self, capsys): + dispatch(RawSession(), "READ(4)") + assert "identify" in capsys.readouterr().out + + def test_unknown_call(self, capsys): + s = self._id(); dispatch(s, "identify"); capsys.readouterr() + dispatch(s, "FLY(4)") + assert "unknown command" in capsys.readouterr().out + + def test_bad_args_shows_usage(self, capsys): + s = self._id(); dispatch(s, "identify"); capsys.readouterr() + dispatch(s, "READ()") # missing page arg + assert "usage: READ(page)" in capsys.readouterr().out + + def test_help_lists_catalog(self, capsys): + s = self._id(); dispatch(s, "identify"); capsys.readouterr() + dispatch(s, "help") + out = capsys.readouterr().out + assert "READ(page)" in out and "GET_VERSION" in out + + def test_help_command_detail(self, capsys): + s = self._id(); dispatch(s, "identify"); capsys.readouterr() + dispatch(s, "help READ") + assert "0x30" in capsys.readouterr().out