pm3py/pm3py/cli/rawcli/catalog.py

"""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] | None = None   # HF: string args -> raw payload bytes to exchange
    help: str = ""
    run: Callable[..., object] | None = None    # LF: (device, *args) -> a human result line

    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=None, help="", run=None) -> TagCommand:
    return TagCommand(name, tuple(params), build, help, run)


# ---- 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("PWD_AUTH", ["password"],
       lambda password: bytes([0x1B]) + _hex(password).ljust(4, b"\x00")[:4],
       "authenticate with the 4-byte password → PACK (0x1B)"),
    _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)"),
    _c("COMPAT_WRITE", ["page", "data"],
       lambda page, data: [bytes([0xA0, _int(page)]), _hex(data).ljust(16, b"\x00")[:16]],
       "compatibility write: 16-byte <data> to <page>, two-phase (0xA0)"),
    _c("HALT", [], lambda: bytes([0x50, 0x00]), "halt the tag (0x50 00)"),
)

# ---- 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", "data"],
       lambda block, data: [bytes([0xA0, _int(block)]), _hex(data).ljust(16, b"\x00")[:16]],
       "write a 16-byte block, two-phase (0xA0) — needs prior auth"),
    _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)"),
)


# ---- LF T5577 / ATA5577 (block read/write over the T55xx downlink; executes via lf.t55) ----
# build() exists only to expose the downlink opcode for hex/binary completion; execution goes
# through run() (the device method), since LF isn't a raw-byte exchange like 14a.
def _t55_read(dev, block):
    b = _int(block)
    if b == 0:                                          # config block — the reliable one
        from pm3py.lf.capture import read_config
        r = read_config(dev)
        if r:
            return (f"block 0 = {r['config_hex']}  ({r['modulation']}, RF/{r['data_bit_rate']}, "
                    f"{r['max_block']} blocks)  ->  {r['emulating']}")
        return "READ block 0: no clean config recovered"
    from pm3py.lf.capture import read_t55xx_block
    word = read_t55xx_block(dev, b)
    if word is None:
        return f"READ block {b}: no clean response (LF block reads are demod-dependent)"
    return f"block {b} = {word:08X}  (best-effort — bit alignment not verified)"


def _t55_write(dev, block, data):
    r = dev.lf.t55.writebl(_int(block), _int(data))
    ok = isinstance(r, dict) and r.get("status") == 0
    return f"WRITE block {_int(block)} <- {_int(data):08X}: {'ok' if ok else 'failed'}"


def _t55_wake(dev, password):
    r = dev.lf.t55.wakeup(_int(password))
    return f"WAKE: {'ok' if isinstance(r, dict) and r.get('status') == 0 else 'failed'}"


def _t55_detect(dev):
    from pm3py.lf.capture import read_config
    r = read_config(dev)
    if not r:
        return "DETECT: no T55xx config recovered"
    return (f"config {r['config_hex']}  ({r['modulation']}, RF/{r['data_bit_rate']}, "
            f"{r['max_block']} blocks)  ->  {r['emulating']}")


def _lf_reread(dev):
    from pm3py.lf import identify_lf
    r = identify_lf(dev)
    return r["label"] if r else "no LF credential decoded"


T5577 = _catalog(
    "T5577 / ATA5577", "lf",
    _c("READ", ["block"], build=lambda block: bytes([0x01, _int(block)]), run=_t55_read,
       help="read a 32-bit block (0=config, 7=password, 1-6=data) — best-effort demod"),
    _c("WRITE", ["block", "data"], build=lambda block, data: bytes([0x02, _int(block)]),
       run=_t55_write, help="write 32-bit <data> to <block> (0x02)"),
    _c("WAKE", ["password"], build=lambda password: bytes([0x03]), run=_t55_wake,
       help="wake a password-protected tag with the block-7 password (0x03)"),
    _c("DETECT", [], build=lambda: bytes([0x01, 0x00]), run=_t55_detect,
       help="read + decode the config block (block 0)"),
)

# ---- LF read-only credentials (EM4100/HID/AWID/FDX-B): broadcast tags with no command set ----
LF_READONLY = _catalog(
    "LF credential (read-only)", "lf",
    _c("INFO", [], run=_lf_reread, help="re-read and decode the broadcast credential"),
)


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
    if session.protocol == "lf":
        if not session.transponder:
            return None
        return T5577 if "T5577" in name else LF_READONLY
    return None