refactor: move modules into core/ sub-package, normalize naming
Phase 1 of package refactor. Moves all source modules into pm3py/core/ with file renames (hf_14a→hf_iso14a, hf_15→hf_iso15, hf_mf→hf_mfc) and client attribute normalization (hf.a14→hf.iso14a, hf.mf→hf.mfc). pm3py/__init__.py re-exports from core for backward compat. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
michael
177
pm3py/core/client.py
Normal file
177
pm3py/core/client.py
Normal file
@@ -0,0 +1,177 @@
|
||||
"""Main Proxmark3 client with nested API."""
|
||||
import asyncio
|
||||
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
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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.mfc = HFMFCommands(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 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)):
|
||||
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):
|
||||
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})"
|
||||
Reference in New Issue
Block a user