diff --git a/pm3py/client.py b/pm3py/client.py index 8c16bff..0d78e17 100644 --- a/pm3py/client.py +++ b/pm3py/client.py @@ -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})" diff --git a/tests/test_integration.py b/tests/test_integration.py index 9537270..777b6cd 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -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")