feat(pm3py): add Proxmark3.sync() for REPL/script usage without async boilerplate
This commit is contained in:
@@ -1,4 +1,6 @@
|
||||
"""Main Proxmark3 client with nested API."""
|
||||
import asyncio
|
||||
import inspect
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
@@ -102,3 +104,51 @@ class Proxmark3:
|
||||
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, baudrate: int = 115200) -> "Proxmark3":
|
||||
"""Create a synchronous client for REPL / script usage.
|
||||
|
||||
Usage:
|
||||
pm3 = Proxmark3.sync("/dev/ttyACM0")
|
||||
print(pm3.firmware)
|
||||
print(pm3.hw.ping())
|
||||
print(pm3.hf.mf.rdbl(block=0))
|
||||
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)):
|
||||
return _SyncProxy(attr, self._loop)
|
||||
# Sub-modules attached dynamically (e.g. hf.a14, hf.mf)
|
||||
if hasattr(attr, '_t'):
|
||||
return _SyncProxy(attr, self._loop)
|
||||
if callable(attr) and inspect.iscoroutinefunction(attr):
|
||||
def sync_wrapper(*args, **kwargs):
|
||||
return self._loop.run_until_complete(attr(*args, **kwargs))
|
||||
sync_wrapper.__name__ = name
|
||||
sync_wrapper.__doc__ = attr.__doc__
|
||||
return sync_wrapper
|
||||
return attr
|
||||
|
||||
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})"
|
||||
|
||||
@@ -52,6 +52,30 @@ def test_mf_rdbl_returns_hex():
|
||||
assert result["raw"] == block
|
||||
|
||||
|
||||
def test_sync_proxy():
|
||||
"""Verify sync proxy wraps async calls."""
|
||||
from pm3py.transport import PM3Response
|
||||
from pm3py.client import _SyncProxy
|
||||
|
||||
pm3 = Proxmark3("/dev/null")
|
||||
loop = asyncio.new_event_loop()
|
||||
proxy = _SyncProxy(pm3, loop)
|
||||
|
||||
# Mock the transport on the command object
|
||||
mock_transport = AsyncMock()
|
||||
proxy._target.hw._t = mock_transport
|
||||
ping_data = bytes(range(32))
|
||||
mock_transport.send_ng.return_value = PM3Response(
|
||||
cmd=Cmd.PING, status=0, reason=0, ng=True, data=ping_data)
|
||||
|
||||
# Call synchronously — no await needed
|
||||
result = proxy.hw.ping(32)
|
||||
assert result["success"] is True
|
||||
assert result["length"] == 32
|
||||
|
||||
loop.close()
|
||||
|
||||
|
||||
def test_firmware_probe_compatible():
|
||||
"""Verify firmware probe populates FirmwareInfo on successful connect."""
|
||||
pm3 = Proxmark3("/dev/null")
|
||||
|
||||
Reference in New Issue
Block a user