b0b46b3919
Adds the 'is the device running the right firmware, and flash it if not' layer on top of the flasher. - _firmware.py: FirmwarePin (pinned to the firmware/ submodule commit) and matches() — a device version-string test against the pinned git SHA (or a release tag, once tagged releases exist). - FirmwareInfo gains expected_firmware + matches_pin, computed on connect, so pm3.firmware.matches_pin tells you whether the fork build is present. - pm3flash console-script: auto-detect, compare to the pin, resolve the image (--image / $PM3PY_FIRMWARE / local build), confirm, flash with progress, then reopen and verify the new version. Flashes only on mismatch unless --force; fullimage-only unless --allow-bootrom. Detection + CLI are unit-tested; the auto-download layer (fetch the pinned release asset) and hardware validation are still to come. Suite 1034 pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
198 lines
7.2 KiB
Python
198 lines
7.2 KiB
Python
"""Main Proxmark3 client with nested API."""
|
|
import asyncio
|
|
import functools
|
|
import glob
|
|
import inspect
|
|
import logging
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
|
|
from .transport import PM3Transport, PM3Response, PM3Error, ProgressCallback
|
|
from .hw import HWCommands
|
|
from .lf import LFCommands
|
|
from .hf import HFCommands
|
|
from .hf_iso14a import HF14ACommands
|
|
from .hf_iso15 import HF15Commands
|
|
from .hf_mfc import HFMFCommands
|
|
from .hf_mfu import HFMFUCommands
|
|
from .flash import Flasher
|
|
from .._firmware import FIRMWARE_PIN, matches as _fw_matches
|
|
|
|
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
|
|
expected_firmware: str = "" # the fork build this pm3py is pinned to
|
|
matches_pin: bool = False # device firmware == the pinned build?
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
|
|
def find_proxmark_port() -> str | None:
|
|
"""Auto-detect the first available Proxmark3 serial port."""
|
|
if sys.platform == "darwin":
|
|
patterns = ["/dev/tty.usbmodem*"]
|
|
elif sys.platform == "win32":
|
|
# On Windows, we can't glob — caller should specify port
|
|
return None
|
|
else:
|
|
patterns = ["/dev/ttyACM*", "/dev/ttyUSB*"]
|
|
|
|
for pattern in patterns:
|
|
ports = sorted(glob.glob(pattern))
|
|
if ports:
|
|
return ports[0]
|
|
return None
|
|
|
|
|
|
class Proxmark3:
|
|
"""Proxmark3 device client with structured Python API.
|
|
|
|
Usage:
|
|
async with Proxmark3() as pm3: # auto-detect port
|
|
print(await pm3.hw.ping())
|
|
|
|
async with Proxmark3("/dev/ttyACM0") as pm3: # explicit port
|
|
resp = await pm3.send_ng(0x0108)
|
|
|
|
pm3 = Proxmark3.sync() # sync REPL, auto-detect
|
|
"""
|
|
|
|
def __init__(self, port: str | None = None, baudrate: int = 115200):
|
|
if port is None:
|
|
port = find_proxmark_port()
|
|
if port is None:
|
|
raise PM3Error("No Proxmark3 device found")
|
|
log.info("pm3py: auto-detected %s", port)
|
|
self._transport = PM3Transport(port, baudrate)
|
|
self.hw = HWCommands(self._transport)
|
|
self.lf = LFCommands(self._transport)
|
|
self.hf = HFCommands(self._transport)
|
|
self.hf.iso14a = HF14ACommands(self._transport)
|
|
self.hf.iso15 = HF15Commands(self._transport)
|
|
self.hf.mf = HFMFCommands(self._transport)
|
|
self.hf.mfu = HFMFUCommands(self._transport)
|
|
self.flasher = Flasher(self._transport)
|
|
self.firmware = FirmwareInfo()
|
|
|
|
async def connect(self) -> None:
|
|
await self._transport.connect()
|
|
await self._probe_firmware()
|
|
|
|
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 _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 2.5: Compare against the pinned fork firmware (for pm3flash / live-sim)
|
|
fw.expected_firmware = FIRMWARE_PIN.tag or FIRMWARE_PIN.sha
|
|
fw.matches_pin = _fw_matches(fw.version_string)
|
|
|
|
# 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"",
|
|
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)
|
|
|
|
@classmethod
|
|
def sync(cls, port: str | None = None, baudrate: int = 115200) -> "Proxmark3":
|
|
"""Create a synchronous client for REPL / script usage.
|
|
|
|
Usage:
|
|
pm3 = Proxmark3.sync() # auto-detect
|
|
pm3 = Proxmark3.sync("/dev/ttyACM0") # explicit
|
|
print(pm3.firmware)
|
|
print(pm3.hw.ping())
|
|
pm3.close()
|
|
"""
|
|
instance = cls(port, baudrate)
|
|
loop = asyncio.new_event_loop()
|
|
loop.run_until_complete(instance.connect())
|
|
return _SyncProxy(instance, loop)
|
|
|
|
|
|
class _SyncProxy:
|
|
"""Wraps an async Proxmark3 instance so every async method call is synchronous."""
|
|
|
|
def __init__(self, target, loop: asyncio.AbstractEventLoop):
|
|
object.__setattr__(self, '_target', target)
|
|
object.__setattr__(self, '_loop', loop)
|
|
|
|
def __getattr__(self, name):
|
|
attr = getattr(self._target, name)
|
|
if isinstance(attr, (HWCommands, LFCommands, HFCommands,
|
|
HF14ACommands, HF15Commands, HFMFCommands,
|
|
HFMFUCommands)):
|
|
return _SyncProxy(attr, self._loop)
|
|
# Sub-modules attached dynamically (e.g. hf.iso14a, hf.mfc)
|
|
if hasattr(attr, '_t'):
|
|
return _SyncProxy(attr, self._loop)
|
|
if callable(attr) and inspect.iscoroutinefunction(attr):
|
|
# functools.wraps copies __doc__/__name__ AND sets __wrapped__, so
|
|
# inspect.signature() and help() report the real params (block, uid=None)
|
|
# instead of the wrapper's (*args, **kwargs).
|
|
@functools.wraps(attr)
|
|
def sync_wrapper(*args, **kwargs):
|
|
return self._loop.run_until_complete(attr(*args, **kwargs))
|
|
return sync_wrapper
|
|
return attr
|
|
|
|
def __dir__(self):
|
|
# Surface the wrapped target's command methods and sub-modules so
|
|
# dir()/REPL tab-completion and help() discovery work through the proxy.
|
|
return sorted(set(object.__dir__(self)) | set(dir(self._target)))
|
|
|
|
def close(self):
|
|
"""Disconnect and close the event loop."""
|
|
self._loop.run_until_complete(self._target.disconnect())
|
|
self._loop.close()
|
|
|
|
def __repr__(self):
|
|
return f"SyncProxy({self._target!r})"
|