feat(rawcli): command catalog + function-style commands (Phase 4)

- catalog.py: declarative per-transponder command sets (name, params,
  build(args)->bytes, help). Seeds: Type 2 (NTAG/UL: READ, FAST_READ,
  GET_VERSION, READ_SIG, READ_CNT, WRITE), MIFARE Classic (READ/WRITE/
  HALT), ISO 15693 (INVENTORY, READ_BLOCK, WRITE_BLOCK, GET_SYSTEM_INFO).
  catalog_for(session) resolves by protocol + identified transponder.
- app: function-style calls (READ(4) -> build -> raw exchange ->
  annotated trace); catalog-aware help (`help` lists the current tag's
  commands, `help READ` shows one). Bad args print usage.

9 new tests (catalog builds, resolver, call dispatch, help). Command
knowledge is declarative + hardware-free-testable; the exchange is the
user's device test.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
michael
2026-07-14 12:13:14 -07:00
parent e024294ec6
commit 9c84f9e7b0
3 changed files with 204 additions and 3 deletions

View File

@@ -13,6 +13,7 @@ from .session import RawSession
from .parser import parse_line from .parser import parse_line
from .entry import install_key_bindings from .entry import install_key_bindings
from .identify import identify, clear, format_summary from .identify import identify, clear, format_summary
from .catalog import catalog_for
from .trace_view import render_exchange from .trace_view import render_exchange
_HELP = """\ _HELP = """\
@@ -70,7 +71,7 @@ def dispatch(state: RawSession, line: str) -> bool:
if cmd.name in ("quit", "exit", "q"): if cmd.name in ("quit", "exit", "q"):
return False return False
if cmd.name == "help": if cmd.name == "help":
print(_HELP) _handle_help(state, cmd.args)
elif cmd.name == "identify": elif cmd.name == "identify":
print(format_summary(identify(state))) print(format_summary(identify(state)))
elif cmd.name == "transponder": elif cmd.name == "transponder":
@@ -81,8 +82,7 @@ def dispatch(state: RawSession, line: str) -> bool:
return True return True
if cmd.kind == "call": if cmd.kind == "call":
# function-style transponder commands land in Phase 4 (the command catalog) _handle_call(state, cmd)
print(f"rawcli: {cmd.name}(...) needs an identified transponder + catalog (Phase 4)")
return True return True
# raw payload # raw payload
@@ -94,6 +94,40 @@ def dispatch(state: RawSession, line: str) -> bool:
return True 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: def run(port=None) -> int:
from prompt_toolkit import PromptSession from prompt_toolkit import PromptSession
from prompt_toolkit.key_binding import KeyBindings from prompt_toolkit.key_binding import KeyBindings

102
pm3py/cli/rawcli/catalog.py Normal file
View File

@@ -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 <page> (0x30)"),
_c("FAST_READ", ["start", "end"], lambda start, end: bytes([0x3A, _int(start), _int(end)]),
"read pages <start>..<end> 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 <data> to <page> (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

View File

@@ -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.parser import parse_token, parse_bytes, parse_line
from pm3py.cli.rawcli.entry import byte_space from pm3py.cli.rawcli.entry import byte_space
from pm3py.cli.rawcli.identify import identify, clear, name_14a 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 from pm3py.cli.rawcli.app import dispatch
_ANSI = re.compile(r"\x1b\[[0-9;]*m") _ANSI = re.compile(r"\x1b\[[0-9;]*m")
@@ -200,3 +201,67 @@ class TestControlCommands:
dispatch(s, "close") dispatch(s, "close")
assert "closed" in capsys.readouterr().out assert "closed" in capsys.readouterr().out
assert s.transponder is None 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