pm3py/pm3py/core/client.py

"""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})"