From 137ec1e721c2e7b0e9ce04f59cd688637e364fbd Mon Sep 17 00:00:00 2001 From: michael Date: Sun, 5 Jul 2026 00:34:15 -0700 Subject: [PATCH] feat(iso15): add inventory() with 1/16-slot parsing + AFI/DSFID filtering inventory() parses both response formats: slots=1 -> single tag dict (or None); slots=16 -> list of tags from the iso15_inventory_response_t anti-collision struct, each CRC15-validated. Optional request-level AFI filter and client-side DSFID filter. One 16-slot round drops colliding slots (full tree-walk resolution is a follow-up); scan() stays the single-tag convenience. Co-Authored-By: Claude Opus 4.8 (1M context) --- pm3py/core/hf_iso15.py | 76 +++++++++++++++++++++++++++++++++++++++++- tests/test_hf_15.py | 40 ++++++++++++++++++++++ 2 files changed, 115 insertions(+), 1 deletion(-) diff --git a/pm3py/core/hf_iso15.py b/pm3py/core/hf_iso15.py index bb10f56..9a51ce0 100644 --- a/pm3py/core/hf_iso15.py +++ b/pm3py/core/hf_iso15.py @@ -1,6 +1,6 @@ """ISO 15693 commands: hf.iso15.*""" import struct -from .protocol import Cmd, crc15_bytes +from .protocol import Cmd, crc15, crc15_bytes from .transport import PM3Transport, PM3Error # Re-export trace symbols for backward compat with existing test imports @@ -40,6 +40,24 @@ def _iso15_payload(iso_cmd: bytes, flags: int = _READ_FLAGS) -> bytes: return struct.pack(" dict | None: + """Parse one inventory response ``[resp_flags][dsfid][uid(8)][crc(2)]``. + + Returns ``{'uid', 'dsfid', 'flags'}`` (uid as display-order hex), or None if + the record is too short or the CRC is bad (i.e. a collision/garbled slot). + """ + if len(rec) < 12 or crc15(rec[:-2]) != (rec[-2] | (rec[-1] << 8)): + return None + return {"uid": rec[2:10][::-1].hex(), "dsfid": rec[1], "flags": rec[0]} + + class HF15Commands: """ISO 15693 commands.""" @@ -82,6 +100,62 @@ class HF15Commands: return {"inv_flags": hex(inv_flags), "status": resp.status, "len": len(resp.data), "data": resp.data.hex()} + async def inventory(self, slots: int = 16, afi: int | None = None, + dsfid: int | None = None): + """Perform an ISO15693 inventory (anti-collision). + + slots=16 (default): one 16-slot anti-collision round → **list** of tag + dicts (empty list if none). Handles up to 16 non-colliding tags per + round; slots that collide are dropped (CRC fail). + slots=1: single-slot → **one** tag dict, or None if no/garbled response. + afi: request-level Application Family Identifier — only tags with this AFI + answer (sent in the inventory command). + dsfid: client-side filter — keep only tags whose DSFID equals this. + + Each tag: ``{'uid': str, 'dsfid': int, 'flags': int[, 'slot': int]}``. + (`scan()` remains the quick "is one tag present" convenience.) + """ + req = REQ_DATARATE_HIGH | REQ_INVENTORY + if slots == 1: + req |= REQINV_SLOT1 + if afi is not None: + req |= REQINV_AFI + iso_cmd = bytes([req, 0x01]) # request flags, INVENTORY command + if afi is not None: + iso_cmd += bytes([afi & 0xFF]) # AFI byte precedes the mask + iso_cmd += bytes([0x00]) # mask length = 0 + resp = await self._t.send_ng( + Cmd.HF_ISO15693_COMMAND, _iso15_payload(iso_cmd), timeout=5.0) + + def _keep(tag): + return tag is not None and (dsfid is None or tag["dsfid"] == dsfid) + + if slots == 1: + if resp.status != 0: + return None + tag = _parse_inv_record(bytes(resp.data)) + return tag if _keep(tag) else None + + # 16-slot: iso15_inventory_response_t = [slot_count][16x{status,len}][data] + tags = [] + data = bytes(resp.data) + if resp.status != 0 or len(data) < 1 + 16 * 2: + return tags + slot_count = data[0] + off = 1 + slot_count * 2 # concatenated slot data starts here + for s in range(slot_count): + status = data[1 + s * 2] + rlen = data[1 + s * 2 + 1] + rec = data[off:off + rlen] + off += rlen + if status != 1 or rlen == 0: # 0=empty, 2=collision + continue + tag = _parse_inv_record(rec) + if _keep(tag): + tag["slot"] = s + tags.append(tag) + return tags + async def rdbl(self, block: int, uid: str | None = None) -> dict: """Read a single block.""" # OPTION (0x40) makes the tag return the block-security (lock) byte this parser diff --git a/tests/test_hf_15.py b/tests/test_hf_15.py index 896a2e6..07b7670 100644 --- a/tests/test_hf_15.py +++ b/tests/test_hf_15.py @@ -286,3 +286,43 @@ class TestSniffLineAnnotationOverflow: "timestamp": 0, "duration": 0, "parity": b""} line = format_sniff_line(entry, width=120, is_tty=False) assert "CC" in line + + +def _inv_record(uid_display_hex, dsfid=0, flags=0): + from pm3py.core.protocol import crc15_bytes + body = bytes([flags, dsfid]) + bytes.fromhex(uid_display_hex)[::-1] + return body + crc15_bytes(body) + + +def test_15_inventory_single_slot(): + t = AsyncMock() + iso = HF15Commands(t) + t.send_ng.return_value = PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=0, + reason=0, ng=True, data=_inv_record("e004a2110b37c433")) + tag = asyncio.get_event_loop().run_until_complete(iso.inventory(slots=1)) + assert tag == {"uid": "e004a2110b37c433", "dsfid": 0, "flags": 0} + # SLOT1 flag set in the request (raw[0] bit 0x20) + payload = t.send_ng.call_args.args[1] + assert payload[3] & 0x20 # raw[0] after 3-byte flags+len header + + +def test_15_inventory_16_slot_list_and_dsfid_filter(): + t = AsyncMock() + iso = HF15Commands(t) + r0 = _inv_record("e004a2110b37c433") + r3 = _inv_record("e004a2110b3746e0", dsfid=5) + meta = bytearray(16 * 2) + meta[0:2] = bytes([1, len(r0)]) # slot 0 valid + meta[6:8] = bytes([1, len(r3)]) # slot 3 valid + meta[10:12] = bytes([2, 0]) # slot 5 collision + data = bytes([16]) + bytes(meta) + r0 + r3 + t.send_ng.return_value = PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=0, + reason=0, ng=True, data=data) + tags = asyncio.get_event_loop().run_until_complete(iso.inventory(slots=16)) + assert [x["uid"] for x in tags] == ["e004a2110b37c433", "e004a2110b3746e0"] + assert tags[0]["slot"] == 0 and tags[1]["slot"] == 3 + # client-side DSFID filter + t.send_ng.return_value = PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=0, + reason=0, ng=True, data=data) + only5 = asyncio.get_event_loop().run_until_complete(iso.inventory(slots=16, dsfid=5)) + assert [x["uid"] for x in only5] == ["e004a2110b3746e0"]