pm3py/pm3py/core/hf_iso15.py

"""ISO 15693 commands: hf.iso15.*"""
import struct
from .protocol import Cmd, crc15_bytes
from .transport import PM3Transport, PM3Error

# Re-export trace symbols for backward compat with existing test imports
from ..trace import (
    parse_tracelog, TRACELOG_HDR_SIZE,
    decode_15693_request, decode_15693_response, decode_15693,
    decode_ndef_annotation,
    format_sniff_line,
)

# ISO15 PM3 transport flags — values MUST match firmware include/iso15.h
ISO15_CONNECT = 0x01
ISO15_NO_DISCONNECT = 0x02
ISO15_RAW = 0x04
ISO15_APPEND_CRC = 0x08        # legacy: ARM firmware ignores it — CRC is added host-side
ISO15_HIGH_SPEED = 0x10
ISO15_READ_RESPONSE = 0x20
ISO15_LONG_WAIT = 0x40


def _iso15_payload(iso_cmd: bytes) -> bytes:
    """Build a HF_ISO15693_COMMAND payload from a raw ISO15693 command.

    Appends the ISO15693 CRC host-side (the ARM firmware ignores ISO15_APPEND_CRC —
    the client must add it, like the C client's AddCrc15) and prefixes the PM3 transport
    flags. HIGH_SPEED + READ_RESPONSE make the firmware receive at high data rate and
    actually listen for the tag's answer — without READ_RESPONSE (0x20) the firmware
    passes a NULL rx buffer and replies empty.
    """
    raw = iso_cmd + crc15_bytes(iso_cmd)
    flags = ISO15_CONNECT | ISO15_HIGH_SPEED | ISO15_READ_RESPONSE
    return struct.pack("<BH", flags, len(raw)) + raw


class HF15Commands:
    """ISO 15693 commands."""

    def __init__(self, transport: PM3Transport):
        self._t = transport

    async def scan(self) -> dict:
        """Scan for ISO15693 tags using inventory command."""
        # ISO15693 inventory: flags=0x26 (high data rate + inventory + 1-slot), cmd=0x01.
        # SLOT1 (0x20) is required: without it the firmware runs a 16-slot anti-collision
        # round and replies with an iso15_slot_result_t struct, not [flags][dsfid][uid],
        # so this parser would read the struct header as a zeroed UID. 1-slot gives the
        # simple single-tag response this method expects. Use hf 15 reader --all for many.
        iso_cmd = bytes([0x26, 0x01, 0x00])  # flags, inventory cmd, mask_len=0
        payload = _iso15_payload(iso_cmd)
        resp = await self._t.send_ng(Cmd.HF_ISO15693_COMMAND, payload, timeout=5.0)

        if resp.status != 0 or len(resp.data) < 10:
            return {"found": False, "uid": None}

        resp_flags = resp.data[0]
        dsfid = resp.data[1]
        uid = resp.data[2:10]

        return {
            "found": True,
            "uid": uid[::-1].hex(),  # ISO15693 UIDs are reversed
            "dsfid": dsfid,
            "response_flags": resp_flags,
        }

    async def raw_inventory(self, inv_flags: int = 0x26) -> dict:
        """Diagnostic: send a raw ISO15693 inventory with a caller-chosen request-flags
        byte and return the raw device response (no parsing). Use to isolate 1-slot vs
        16-slot behavior: inv_flags=0x26 → 1 slot (SLOT1 set), 0x06 → 16 slots.
        """
        iso_cmd = bytes([inv_flags, 0x01, 0x00])  # inv flags, INVENTORY cmd, mask_len=0
        payload = _iso15_payload(iso_cmd)
        resp = await self._t.send_ng(Cmd.HF_ISO15693_COMMAND, payload, timeout=5.0)
        return {"inv_flags": hex(inv_flags), "status": resp.status,
                "len": len(resp.data), "data": resp.data.hex()}

    async def rdbl(self, block: int, uid: str | None = None) -> dict:
        """Read a single block."""
        if uid:
            uid_bytes = bytes.fromhex(uid)[::-1]  # reverse for wire
            iso_cmd = bytes([0x22, 0x20]) + uid_bytes + bytes([block])
        else:
            iso_cmd = bytes([0x02, 0x20, block])  # addressed without UID

        payload = _iso15_payload(iso_cmd)
        resp = await self._t.send_ng(Cmd.HF_ISO15693_COMMAND, payload, timeout=5.0)

        if resp.status != 0 or len(resp.data) < 2:
            return {"success": False, "error": resp.status}

        resp_flags = resp.data[0]
        if resp_flags != 0:
            return {"success": False, "error_flag": resp_flags}

        lock_status = resp.data[1]
        block_data = resp.data[2:]

        return {
            "success": True,
            "block": block,
            "locked": bool(lock_status),
            "data": block_data.hex(),
            "raw": block_data,
        }

    async def wrbl(self, block: int, data: bytes | str, uid: str | None = None) -> dict:
        """Write a single block."""
        if isinstance(data, str):
            data = bytes.fromhex(data)

        if uid:
            uid_bytes = bytes.fromhex(uid)[::-1]
            iso_cmd = bytes([0x22, 0x21]) + uid_bytes + bytes([block]) + data
        else:
            iso_cmd = bytes([0x02, 0x21, block]) + data

        payload = _iso15_payload(iso_cmd)
        resp = await self._t.send_ng(Cmd.HF_ISO15693_COMMAND, payload, timeout=5.0)

        success = resp.status == 0 and len(resp.data) > 0 and resp.data[0] == 0
        return {"success": success, "block": block}

    async def sniff(self, timeout: float = 60.0) -> dict:
        """Start ISO15693 sniff. Blocks until button press or timeout.

        Returns {"status": int}.
        """
        from .transport import encode_ng_frame
        async with self._t._lock:
            raw = encode_ng_frame(Cmd.HF_ISO15693_SNIFF)
            await self._t.send_frame(raw)

            # Firmware sends debug prints before the sniff completion response.
            # Keep reading until we get the actual sniff response.
            while True:
                resp = await self._t.read_response(timeout=timeout)
                if resp.cmd == Cmd.HF_ISO15693_SNIFF:
                    return {"status": resp.status}
                # Debug print or other message — discard and keep waiting

    async def download_trace(self) -> list[dict]:
        """Download and parse the trace buffer from the device."""
        from ..sniff.session import SniffSession
        session = SniffSession(self._t)
        return await session._download_trace()

    async def sniff_decoded(self, timeout: float = 60.0) -> list[dict]:
        """Sniff ISO15693 traffic, download trace, decode and print.

        Convenience method — delegates to SniffSession.iso15().
        Prefer using SniffSession directly for new code.
        """
        from ..sniff.session import SniffSession
        session = SniffSession(self._t)
        return await session.iso15(timeout=timeout)