feat(pm3py): HF core + ISO14443A commands

pm3py/client.py

@@ -2,6 +2,8 @@
 from .transport import PM3Transport, PM3Response, ProgressCallback
 from .hw import HWCommands
 from .lf import LFCommands
+from .hf import HFCommands
+from .hf_14a import HF14ACommands
 
 
 class Proxmark3:
@@ -20,6 +22,8 @@ class Proxmark3:
         self._transport = PM3Transport(port, baudrate)
         self.hw = HWCommands(self._transport)
         self.lf = LFCommands(self._transport)
+        self.hf = HFCommands(self._transport)
+        self.hf.a14 = HF14ACommands(self._transport)
 
     async def connect(self) -> None:
         await self._transport.connect()

pm3py/hf.py

@@ -0,0 +1,75 @@
+"""High frequency commands: hf.*"""
+import struct
+import asyncio
+from .protocol import Cmd
+from .transport import PM3Transport, PM3Error, ProgressCallback
+
+
+class HFCommands:
+    """High frequency commands."""
+
+    def __init__(self, transport: PM3Transport):
+        self._t = transport
+        # Sub-modules attached by client.py
+        self.a14 = None  # type: ignore
+        self.iso15 = None  # type: ignore
+        self.mf = None  # type: ignore
+
+    async def tune(self, on_progress: ProgressCallback = None) -> dict:
+        """Measure HF antenna voltage."""
+        resp = await self._t.send_ng(Cmd.MEASURE_ANTENNA_TUNING_HF, timeout=10.0)
+        voltage = struct.unpack_from("<I", resp.data, 0)[0] if len(resp.data) >= 4 else 0
+        result = {
+            "voltage_mV": voltage,
+            "voltage_V": round(voltage / 1000.0, 3),
+        }
+        if on_progress:
+            ret = on_progress({"type": "hf_tune", "result": result})
+            if asyncio.iscoroutine(ret):
+                await ret
+        return result
+
+    async def search(self, on_progress: ProgressCallback = None) -> dict:
+        """Search for HF tags. Tries 14a, then 15."""
+        results = {}
+
+        if on_progress:
+            ret = on_progress({"type": "hf_search", "phase": "14a"})
+            if asyncio.iscoroutine(ret):
+                await ret
+
+        if self.a14:
+            try:
+                card = await self.a14.scan()
+                if card.get("uid"):
+                    results["iso14443a"] = card
+            except PM3Error:
+                pass
+
+        if on_progress:
+            ret = on_progress({"type": "hf_search", "phase": "15693"})
+            if asyncio.iscoroutine(ret):
+                await ret
+
+        if self.iso15:
+            try:
+                tag = await self.iso15.scan()
+                if tag.get("uid"):
+                    results["iso15693"] = tag
+            except PM3Error:
+                pass
+
+        results["found"] = len(results) > 0
+        return results
+
+    async def sniff(self, samples: int = 0, skip: int = 0,
+                    on_progress: ProgressCallback = None) -> dict:
+        """Sniff HF traffic."""
+        payload = struct.pack("<II", samples, skip) if samples else b""
+        resp = await self._t.send_ng(Cmd.HF_SNIFF, payload, timeout=30.0)
+        return {"status": resp.status, "data": resp.data.hex()}
+
+    async def dropfield(self) -> dict:
+        """Turn off HF field."""
+        resp = await self._t.send_ng(Cmd.HF_DROPFIELD)
+        return {"status": resp.status}

pm3py/hf_14a.py

@@ -0,0 +1,96 @@
+"""ISO 14443-A commands: hf.14a.*"""
+import struct
+from .protocol import Cmd
+from .transport import PM3Transport, PM3Error
+
+# ISO14A command flags (from pm3_cmd.h)
+ISO14A_CONNECT = 0x0001
+ISO14A_NO_DISCONNECT = 0x0002
+ISO14A_RAW = 0x0008
+ISO14A_TOPAZMODE = 0x0100
+ISO14A_NO_RATS = 0x0200
+
+
+class HF14ACommands:
+    """ISO 14443-A commands."""
+
+    def __init__(self, transport: PM3Transport):
+        self._t = transport
+
+    async def scan(self) -> dict:
+        """Scan for ISO14443A card. Returns UID, ATQA, SAK, ATS."""
+        resp = await self._t.send_mix(
+            Cmd.HF_ISO14443A_READER,
+            arg0=ISO14A_CONNECT | ISO14A_NO_DISCONNECT,
+            timeout=5.0,
+        )
+
+        select_status = resp.oldarg[0] if resp.oldarg else 0
+        if select_status == 0:
+            return {"found": False, "uid": None}
+
+        # Parse iso14a_card_select_t from response data
+        data = resp.data
+        if len(data) < 13:  # uid(10) + uidlen(1) + atqa(2)
+            return {"found": False, "uid": None}
+
+        uid_raw = data[0:10]
+        uid_len = data[10]
+        atqa = data[11:13]
+        sak = data[13] if len(data) > 13 else 0
+        ats_len = data[14] if len(data) > 14 else 0
+        ats = data[15:15 + ats_len] if ats_len > 0 and len(data) > 15 else b""
+
+        uid = uid_raw[:uid_len]
+
+        return {
+            "found": True,
+            "uid": uid.hex(),
+            "uid_len": uid_len,
+            "atqa": atqa.hex(),
+            "sak": sak,
+            "ats": ats.hex() if ats else None,
+            "select_status": select_status,  # 1=OK+ATS, 2=OK no ATS, 3=proprietary
+        }
+
+    async def raw(self, data: bytes, flags: int = 0, timeout_14a: int = 0) -> dict:
+        """Send raw ISO14443A command."""
+        resp = await self._t.send_mix(
+            Cmd.HF_ISO14443A_READER,
+            arg0=flags | ISO14A_RAW | ISO14A_NO_DISCONNECT,
+            arg1=len(data),
+            arg2=timeout_14a,
+            payload=data,
+            timeout=5.0,
+        )
+        return {
+            "status": resp.status,
+            "data": resp.data.hex() if resp.data else None,
+            "raw": resp.data,
+        }
+
+    async def sniff(self, param: int = 0) -> dict:
+        """Sniff ISO14443A traffic."""
+        resp = await self._t.send_mix(
+            Cmd.HF_ISO14443A_SNIFF, arg0=param, timeout=30.0,
+        )
+        return {"status": resp.status}
+
+    async def sim(self, uid: bytes, sak: int = 0x08, atqa: bytes = b"\x04\x00",
+                  flags: int = 0) -> dict:
+        """Simulate ISO14443A card."""
+        if len(uid) == 4:
+            uid_flag = 0x0010
+        elif len(uid) == 7:
+            uid_flag = 0x0020
+        elif len(uid) == 10:
+            uid_flag = 0x0030
+        else:
+            raise ValueError("UID must be 4, 7, or 10 bytes")
+
+        flag_val = flags | uid_flag | 0x0001  # FLAG_INTERACTIVE
+        payload = atqa + bytes([sak]) + uid
+        resp = await self._t.send_mix(
+            Cmd.HF_ISO14443A_SIMULATE, arg0=flag_val, timeout=60.0, payload=payload,
+        )
+        return {"status": resp.status}