feat(pm3py): soft firmware compatibility check on connect (ping + version probe)
This commit is contained in:
@@ -1,5 +1,8 @@
|
|||||||
"""Main Proxmark3 client with nested API."""
|
"""Main Proxmark3 client with nested API."""
|
||||||
from .transport import PM3Transport, PM3Response, ProgressCallback
|
import logging
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
|
||||||
|
from .transport import PM3Transport, PM3Response, PM3Error, ProgressCallback
|
||||||
from .hw import HWCommands
|
from .hw import HWCommands
|
||||||
from .lf import LFCommands
|
from .lf import LFCommands
|
||||||
from .hf import HFCommands
|
from .hf import HFCommands
|
||||||
@@ -7,12 +10,27 @@ from .hf_14a import HF14ACommands
|
|||||||
from .hf_15 import HF15Commands
|
from .hf_15 import HF15Commands
|
||||||
from .hf_mf import HFMFCommands
|
from .hf_mf import HFMFCommands
|
||||||
|
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class FirmwareInfo:
|
||||||
|
"""Firmware version and compatibility information populated on connect."""
|
||||||
|
version_string: str = ""
|
||||||
|
chip_id: str = ""
|
||||||
|
section_size: int = 0
|
||||||
|
ng_protocol: bool = False
|
||||||
|
compatible: bool = False
|
||||||
|
warnings: list[str] = field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
class Proxmark3:
|
class Proxmark3:
|
||||||
"""Proxmark3 device client with structured Python API.
|
"""Proxmark3 device client with structured Python API.
|
||||||
|
|
||||||
Usage:
|
Usage:
|
||||||
async with Proxmark3("/dev/ttyACM0") as pm3:
|
async with Proxmark3("/dev/ttyACM0") as pm3:
|
||||||
|
if not pm3.firmware.compatible:
|
||||||
|
print("Warnings:", pm3.firmware.warnings)
|
||||||
status = await pm3.hw.version()
|
status = await pm3.hw.version()
|
||||||
print(status)
|
print(status)
|
||||||
|
|
||||||
@@ -28,9 +46,11 @@ class Proxmark3:
|
|||||||
self.hf.a14 = HF14ACommands(self._transport)
|
self.hf.a14 = HF14ACommands(self._transport)
|
||||||
self.hf.iso15 = HF15Commands(self._transport)
|
self.hf.iso15 = HF15Commands(self._transport)
|
||||||
self.hf.mf = HFMFCommands(self._transport)
|
self.hf.mf = HFMFCommands(self._transport)
|
||||||
|
self.firmware = FirmwareInfo()
|
||||||
|
|
||||||
async def connect(self) -> None:
|
async def connect(self) -> None:
|
||||||
await self._transport.connect()
|
await self._transport.connect()
|
||||||
|
await self._probe_firmware()
|
||||||
|
|
||||||
async def disconnect(self) -> None:
|
async def disconnect(self) -> None:
|
||||||
await self._transport.disconnect()
|
await self._transport.disconnect()
|
||||||
@@ -42,6 +62,36 @@ class Proxmark3:
|
|||||||
async def __aexit__(self, *args) -> None:
|
async def __aexit__(self, *args) -> None:
|
||||||
await self.disconnect()
|
await self.disconnect()
|
||||||
|
|
||||||
|
async def _probe_firmware(self) -> None:
|
||||||
|
"""Ping device and fetch firmware version after connecting."""
|
||||||
|
fw = self.firmware
|
||||||
|
fw.warnings = []
|
||||||
|
|
||||||
|
# Step 1: Ping to verify NG protocol support
|
||||||
|
try:
|
||||||
|
ping_result = await self.hw.ping(32)
|
||||||
|
fw.ng_protocol = ping_result.get("success", False)
|
||||||
|
if not fw.ng_protocol:
|
||||||
|
fw.warnings.append("Ping echo mismatch — device may not support NG protocol")
|
||||||
|
except PM3Error:
|
||||||
|
fw.ng_protocol = False
|
||||||
|
fw.warnings.append("Ping failed — device may not support NG protocol or may be in bootloader mode")
|
||||||
|
|
||||||
|
# Step 2: Fetch firmware version
|
||||||
|
try:
|
||||||
|
ver = await self.hw.version()
|
||||||
|
fw.version_string = ver.get("version_string", "")
|
||||||
|
fw.chip_id = ver.get("chip_id", "")
|
||||||
|
fw.section_size = ver.get("section_size", 0)
|
||||||
|
except PM3Error:
|
||||||
|
fw.warnings.append("Could not read firmware version — device may be in bootloader mode")
|
||||||
|
|
||||||
|
# Step 3: Assess compatibility
|
||||||
|
fw.compatible = fw.ng_protocol and len(fw.warnings) == 0
|
||||||
|
|
||||||
|
for w in fw.warnings:
|
||||||
|
log.warning("pm3py: %s", w)
|
||||||
|
|
||||||
async def send_ng(self, cmd: int, payload: bytes = b"",
|
async def send_ng(self, cmd: int, payload: bytes = b"",
|
||||||
timeout: float = 2.0) -> PM3Response:
|
timeout: float = 2.0) -> PM3Response:
|
||||||
"""Raw NG command -- escape hatch for unwrapped commands."""
|
"""Raw NG command -- escape hatch for unwrapped commands."""
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ import struct
|
|||||||
import asyncio
|
import asyncio
|
||||||
from unittest.mock import AsyncMock, MagicMock
|
from unittest.mock import AsyncMock, MagicMock
|
||||||
from pm3py import Proxmark3, Cmd
|
from pm3py import Proxmark3, Cmd
|
||||||
|
from pm3py.client import FirmwareInfo
|
||||||
|
|
||||||
|
|
||||||
def test_full_api_surface():
|
def test_full_api_surface():
|
||||||
@@ -18,6 +19,8 @@ def test_full_api_surface():
|
|||||||
assert hasattr(pm3.hf, "mf")
|
assert hasattr(pm3.hf, "mf")
|
||||||
assert hasattr(pm3, "send_ng")
|
assert hasattr(pm3, "send_ng")
|
||||||
assert hasattr(pm3, "send_mix")
|
assert hasattr(pm3, "send_mix")
|
||||||
|
assert hasattr(pm3, "firmware")
|
||||||
|
assert isinstance(pm3.firmware, FirmwareInfo)
|
||||||
|
|
||||||
|
|
||||||
def test_hw_ping_roundtrip():
|
def test_hw_ping_roundtrip():
|
||||||
@@ -47,3 +50,52 @@ def test_mf_rdbl_returns_hex():
|
|||||||
pm3.hf.mf.rdbl(block=4, key="FFFFFFFFFFFF"))
|
pm3.hf.mf.rdbl(block=4, key="FFFFFFFFFFFF"))
|
||||||
assert result["data"] == "000102030405060708090a0b0c0d0e0f"
|
assert result["data"] == "000102030405060708090a0b0c0d0e0f"
|
||||||
assert result["raw"] == block
|
assert result["raw"] == block
|
||||||
|
|
||||||
|
|
||||||
|
def test_firmware_probe_compatible():
|
||||||
|
"""Verify firmware probe populates FirmwareInfo on successful connect."""
|
||||||
|
pm3 = Proxmark3("/dev/null")
|
||||||
|
mock_transport = AsyncMock()
|
||||||
|
pm3._transport = mock_transport
|
||||||
|
pm3.hw._t = mock_transport
|
||||||
|
|
||||||
|
from pm3py.transport import PM3Response
|
||||||
|
|
||||||
|
ping_data = bytes(range(32))
|
||||||
|
vstr = b"Proxmark3 RDV4.0 FW v4.18000\x00"
|
||||||
|
version_payload = struct.pack("<III", 0x270B0A40, 512*1024, len(vstr)) + vstr
|
||||||
|
|
||||||
|
# send_ng called twice: once for ping, once for version
|
||||||
|
mock_transport.send_ng.side_effect = [
|
||||||
|
PM3Response(cmd=Cmd.PING, status=0, reason=0, ng=True, data=ping_data),
|
||||||
|
PM3Response(cmd=Cmd.VERSION, status=0, reason=0, ng=True, data=version_payload),
|
||||||
|
]
|
||||||
|
|
||||||
|
asyncio.get_event_loop().run_until_complete(pm3._probe_firmware())
|
||||||
|
assert pm3.firmware.compatible is True
|
||||||
|
assert pm3.firmware.ng_protocol is True
|
||||||
|
assert "Proxmark3 RDV4.0" in pm3.firmware.version_string
|
||||||
|
assert pm3.firmware.warnings == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_firmware_probe_ping_fails():
|
||||||
|
"""Verify firmware probe marks incompatible when ping fails."""
|
||||||
|
pm3 = Proxmark3("/dev/null")
|
||||||
|
mock_transport = AsyncMock()
|
||||||
|
pm3._transport = mock_transport
|
||||||
|
pm3.hw._t = mock_transport
|
||||||
|
|
||||||
|
from pm3py.transport import PM3Response, PM3Error
|
||||||
|
|
||||||
|
vstr = b"Proxmark3 OLD\x00"
|
||||||
|
version_payload = struct.pack("<III", 0x270B0A40, 256*1024, len(vstr)) + vstr
|
||||||
|
|
||||||
|
mock_transport.send_ng.side_effect = [
|
||||||
|
PM3Error("Ping failed"),
|
||||||
|
PM3Response(cmd=Cmd.VERSION, status=0, reason=0, ng=True, data=version_payload),
|
||||||
|
]
|
||||||
|
|
||||||
|
asyncio.get_event_loop().run_until_complete(pm3._probe_firmware())
|
||||||
|
assert pm3.firmware.compatible is False
|
||||||
|
assert pm3.firmware.ng_protocol is False
|
||||||
|
assert len(pm3.firmware.warnings) > 0
|
||||||
|
|||||||
Reference in New Issue
Block a user