feat(pm3py): client shell and hw commands (ping, version, status, led, tune, etc)

pm3py/__init__.py

@@ -1 +1,6 @@
 """pm3py - Python wire protocol library for Proxmark3."""
+from .client import Proxmark3
+from .transport import PM3Error, PM3Response
+from .protocol import PM3Status, Cmd
+
+__all__ = ["Proxmark3", "PM3Error", "PM3Response", "PM3Status", "Cmd"]

pm3py/client.py

@@ -0,0 +1,44 @@
+"""Main Proxmark3 client with nested API."""
+from .transport import PM3Transport, PM3Response, ProgressCallback
+from .hw import HWCommands
+
+
+class Proxmark3:
+    """Proxmark3 device client with structured Python API.
+
+    Usage:
+        async with Proxmark3("/dev/ttyACM0") as pm3:
+            status = await pm3.hw.version()
+            print(status)
+
+            # Raw access:
+            resp = await pm3.send_ng(0x0108)
+    """
+
+    def __init__(self, port: str, baudrate: int = 115200):
+        self._transport = PM3Transport(port, baudrate)
+        self.hw = HWCommands(self._transport)
+
+    async def connect(self) -> None:
+        await self._transport.connect()
+
+    async def disconnect(self) -> None:
+        await self._transport.disconnect()
+
+    async def __aenter__(self) -> "Proxmark3":
+        await self.connect()
+        return self
+
+    async def __aexit__(self, *args) -> None:
+        await self.disconnect()
+
+    async def send_ng(self, cmd: int, payload: bytes = b"",
+                      timeout: float = 2.0) -> PM3Response:
+        """Raw NG command -- escape hatch for unwrapped commands."""
+        return await self._transport.send_ng(cmd, payload, timeout)
+
+    async def send_mix(self, cmd: int, arg0: int = 0, arg1: int = 0,
+                       arg2: int = 0, payload: bytes = b"",
+                       timeout: float = 2.0) -> PM3Response:
+        """Raw MIX command -- escape hatch for unwrapped commands."""
+        return await self._transport.send_mix(cmd, arg0, arg1, arg2, payload, timeout)

pm3py/hw.py

@@ -0,0 +1,147 @@
+"""Hardware commands: hw.*"""
+import struct
+import asyncio
+from .protocol import Cmd, PM3Status
+from .transport import PM3Transport, PM3Response, PM3Error, ProgressCallback
+
+
+class HWCommands:
+    """Proxmark3 hardware commands."""
+
+    def __init__(self, transport: PM3Transport):
+        self._t = transport
+
+    async def ping(self, length: int = 32) -> dict:
+        """Send ping and verify echo. Returns {success, length, data}."""
+        data = bytes(i & 0xFF for i in range(length))
+        resp = await self._t.send_ng(Cmd.PING, data)
+        match = resp.data == data
+        return {"success": match, "length": len(resp.data), "data": resp.data.hex()}
+
+    async def version(self) -> dict:
+        """Get firmware version. Returns {chip_id, section_size, version_string}."""
+        resp = await self._t.send_ng(Cmd.VERSION, timeout=3.0)
+        if len(resp.data) < 12:
+            raise PM3Error("Version response too short")
+        chip_id, section_size, vstr_len = struct.unpack_from("<III", resp.data, 0)
+        vstr = resp.data[12:12 + vstr_len].decode("utf-8", errors="replace").rstrip("\x00")
+        return {
+            "chip_id": f"0x{chip_id:08X}",
+            "section_size": section_size,
+            "version_string": vstr,
+        }
+
+    async def status(self, speed_test_timeout: int = 0) -> dict:
+        """Get device status. Returns {status}."""
+        if speed_test_timeout > 0:
+            payload = struct.pack("<i", speed_test_timeout)
+        else:
+            payload = b""
+        resp = await self._t.send_ng(Cmd.STATUS, payload, timeout=2.0 + speed_test_timeout / 1000)
+        return {"status": resp.status}
+
+    async def led(self, led: int, action: str = "on", brightness: int = 100,
+                  speed: int = 500, count: int = 5) -> dict:
+        """Control LEDs. led: bitmask (1=A,2=B,4=C,8=D). action: on/off/toggle/pwm/pulse/fade/blink."""
+        actions = {"off": 0, "on": 1, "toggle": 2, "pwm": 3, "pulse": 4, "fade": 5, "blink": 6}
+        if action not in actions:
+            raise ValueError(f"Unknown action: {action}. Use one of: {list(actions.keys())}")
+        payload = struct.pack("<BBBHH", led, actions[action], brightness, speed, count)
+        resp = await self._t.send_ng(Cmd.LED_CONTROL, payload)
+        return {"status": resp.status}
+
+    async def dbg(self, level: int | None = None) -> dict:
+        """Get or set debug level. None=get, 0-4=set."""
+        if level is None:
+            resp = await self._t.send_ng(Cmd.GET_DBGMODE)
+            if len(resp.data) >= 1:
+                return {"level": resp.data[0]}
+            return {"level": 0}
+        else:
+            payload = struct.pack("<B", level)
+            resp = await self._t.send_ng(Cmd.SET_DBGMODE, payload)
+            return {"level": level, "status": resp.status}
+
+    async def tearoff(self, delay_us: int = 0, on: bool = True) -> dict:
+        """Set tearoff parameters."""
+        payload = struct.pack("<HbBB", delay_us, 0, int(on), int(not on))
+        resp = await self._t.send_ng(Cmd.SET_TEAROFF, payload)
+        return {"status": resp.status}
+
+    async def break_loop(self) -> None:
+        """Send break to stop long-running operations."""
+        await self._t.send_ng_no_response(Cmd.BREAK_LOOP)
+
+    async def standalone(self) -> dict:
+        """Start standalone mode."""
+        resp = await self._t.send_ng(Cmd.STANDALONE, timeout=5.0)
+        return {"status": resp.status}
+
+    async def reset(self) -> None:
+        """Reset the device."""
+        await self._t.send_ng_no_response(Cmd.HARDWARE_RESET)
+
+    async def capabilities(self) -> dict:
+        """Get device capabilities."""
+        resp = await self._t.send_ng(Cmd.CAPABILITIES)
+        if len(resp.data) < 9:
+            raise PM3Error("Capabilities response too short")
+        version = resp.data[0]
+        baudrate = struct.unpack_from("<I", resp.data, 1)[0]
+        bigbuf_size = struct.unpack_from("<I", resp.data, 5)[0]
+        # Bit flags start at byte 9
+        flags = 0
+        if len(resp.data) > 9:
+            flags = struct.unpack_from("<I", resp.data, 9)[0] if len(resp.data) >= 13 else resp.data[9]
+        return {
+            "version": version,
+            "baudrate": baudrate,
+            "bigbuf_size": bigbuf_size,
+            "raw_flags": flags,
+        }
+
+    async def tune(self, on_progress: ProgressCallback = None) -> dict:
+        """Run full antenna tuning. Returns LF/HF voltages, peak, Q factor."""
+        resp = await self._t.send_ng(Cmd.MEASURE_ANTENNA_TUNING, timeout=20.0)
+        if resp.status != 0:
+            raise PM3Error("Antenna tuning failed", status=resp.status)
+        if len(resp.data) < 28:
+            raise PM3Error("Tune response too short")
+
+        v_lf134, v_lf125, v_lfconf, v_hf, peak_v, peak_f, divisor = struct.unpack_from("<IIIIIIi", resp.data, 0)
+        results = list(resp.data[28:28 + 256]) if len(resp.data) >= 284 else []
+
+        ANTENNA_ERROR = 1.09
+        LF_DIVISOR_125 = 95
+        LF_DIVISOR_134 = 88
+
+        def div2freq(d):
+            return 12000.0 / (d + 1) if d > 0 else 0
+
+        result = {
+            "lf": {
+                "125kHz_mV": v_lf125,
+                "125kHz_V": round(v_lf125 * ANTENNA_ERROR / 1000, 2),
+                "134kHz_mV": v_lf134,
+                "134kHz_V": round(v_lf134 * ANTENNA_ERROR / 1000, 2),
+                "conf_mV": v_lfconf,
+                "peak_mV": peak_v,
+                "peak_V": round(peak_v * ANTENNA_ERROR / 1000, 2),
+                "peak_freq_kHz": round(div2freq(peak_f), 2),
+                "peak_divisor": peak_f,
+                "divisor": divisor,
+            },
+            "hf": {
+                "13.56MHz_mV": v_hf,
+                "13.56MHz_V": round(v_hf * ANTENNA_ERROR / 1000, 2),
+            },
+            "scan_results": results,
+        }
+
+        if on_progress:
+            event = {"type": "tune_complete", "result": result}
+            ret = on_progress(event)
+            if asyncio.iscoroutine(ret):
+                await ret
+
+        return result

tests/test_hw.py

@@ -0,0 +1,47 @@
+import struct
+import asyncio
+from unittest.mock import AsyncMock
+from pm3py.protocol import Cmd
+from pm3py.transport import PM3Response
+from pm3py.hw import HWCommands
+
+def make_ng_response(cmd: int, status: int, payload: bytes) -> PM3Response:
+    return PM3Response(cmd=cmd, status=status, reason=0, ng=True, data=payload)
+
+def make_mock_transport():
+    transport = AsyncMock()
+    return transport
+
+def test_ping():
+    transport = make_mock_transport()
+    hw = HWCommands(transport)
+    ping_data = bytes(range(32))
+    transport.send_ng.return_value = make_ng_response(Cmd.PING, 0, ping_data)
+    result = asyncio.get_event_loop().run_until_complete(hw.ping(32))
+    assert result["success"] is True
+    assert result["length"] == 32
+
+def test_version():
+    transport = make_mock_transport()
+    hw = HWCommands(transport)
+    # Build version response: id(4) + section_size(4) + versionstr_len(4) + versionstr
+    vstr = b"Proxmark3 OS\x00"
+    payload = struct.pack("<III", 0x270B0A40, 512*1024, len(vstr)) + vstr
+    transport.send_ng.return_value = make_ng_response(Cmd.VERSION, 0, payload)
+    result = asyncio.get_event_loop().run_until_complete(hw.version())
+    assert "chip_id" in result
+    assert "version_string" in result
+
+def test_status():
+    transport = make_mock_transport()
+    hw = HWCommands(transport)
+    transport.send_ng.return_value = make_ng_response(Cmd.STATUS, 0, b"")
+    result = asyncio.get_event_loop().run_until_complete(hw.status())
+    assert result["status"] == 0
+
+def test_led():
+    transport = make_mock_transport()
+    hw = HWCommands(transport)
+    transport.send_ng.return_value = make_ng_response(Cmd.LED_CONTROL, 0, b"")
+    result = asyncio.get_event_loop().run_until_complete(hw.led(1, "on"))
+    transport.send_ng.assert_called_once()