diff --git a/pm3py/pyws_plugin.py b/pm3py/pyws_plugin.py index 3840e71..1f1b225 100644 --- a/pm3py/pyws_plugin.py +++ b/pm3py/pyws_plugin.py @@ -1,18 +1,23 @@ """pyws workspace plugin for pm3py. -Turns pm3py into a pyws workspace: connects the Proxmark3 as a ``reader`` resource and injects -the transponder / sim classes into the session namespace. Discovered by the pyws engine via the -``pyws.plugins`` entry point (see pyproject.toml). pyws must be installed alongside pm3py. +Turns pm3py into a pyws workspace: connects the Proxmark3 (reader-mode) or opens a live +``SimSession`` (sim-mode), and injects the transponder / sim classes into the session +namespace. Discovered by the pyws engine via the ``pyws.plugins`` entry point (see +pyproject.toml). pyws must be installed alongside pm3py. -P1 scope (this file): reader resource + namespace + effect declarations. The live ``sim`` -resource is P2. See docs/plans/2026-07-05-pm3py-pyws-plugin-plan.md. +A workspace uses one connection mode — ``reader`` OR ``sim`` — never both, since they contend +on the serial port. The sim is armed explicitly (``sim.start(tag)``); opening the workspace +only opens the port. See docs/plans/2026-07-05-pm3py-pyws-plugin-plan.md. """ from __future__ import annotations +import logging from typing import Any from pyws import Resource, ResourceStatus, WorkspacePlugin +log = logging.getLogger("pm3py.pyws") + #: names injected into the workspace namespace from pm3py.sim (guarded by hasattr) _SIM_EXPORTS = [ "SimSession", @@ -68,9 +73,92 @@ class ReaderResource(Resource): self.status = ResourceStatus.DISCONNECTED +class Sim: + """User-facing live-sim controls, bound as ``sim`` in the namespace. + + Arming is explicit: opening the workspace opens the port, but the sim goes live only when + you call ``sim.start(tag)``. Push model edits to a running sim with ``sim.push(tag)``. + """ + + def __init__(self, session: Any): + self._sess = session + + def start(self, tag: Any, **kwargs: Any): + """Arm the live sim for ``tag`` — dispatches 14a vs 15693 by tag type.""" + from pm3py.sim import Tag14443A, Tag15693 + + if isinstance(tag, Tag15693): + return self._sess.start_15693(tag, **kwargs) + if isinstance(tag, Tag14443A): + return self._sess.start_14a(tag, **kwargs) + raise TypeError(f"cannot simulate {type(tag).__name__}: not a 14a or 15693 tag") + + def stop(self): + return self._sess.stop() + + def push(self, tag: Any): + """Push model edits to the running sim (``tag.sync()``).""" + return tag.sync() + + @property + def frames(self): + """Decoded reader↔tag trace frames from the running sim.""" + return self._sess.entries + + @property + def session(self): + return self._sess + + def __repr__(self) -> str: + return f"" + + +class SimResource(Resource): + """A live ``SimSession``, exposed as ``sim`` in the namespace. + + Opens the port on session start; the sim is armed explicitly via ``sim.start(tag)``. A + missing device is non-fatal (disconnected). ``session`` is injectable for tests. + """ + + kind = "sim" + + def __init__(self, name: str = "sim", config: dict | None = None, *, session: Any = None): + super().__init__(name, obj=None, config=config or {}) + self._sess = session + + async def start(self) -> None: + self.status = ResourceStatus.CONNECTING + try: + if self._sess is None: + from pm3py.sim import SimSession + + self._sess = SimSession.open( + self.config.get("port", "/dev/ttyACM0"), + self.config.get("baudrate", 115200), + ) + except Exception as exc: # no device — non-fatal + self.status = ResourceStatus.DISCONNECTED + self.detail = f"no device ({type(exc).__name__}: {exc})" + self._sess = None + return + self.obj = Sim(self._sess) + self.status = ResourceStatus.CONNECTED + self.detail = "ready (not armed — call sim.start(tag))" + + async def stop(self) -> None: + if self._sess is not None: + try: + self._sess.close() # stop() + close serial + unbind tag + except Exception: + pass + self.status = ResourceStatus.DISCONNECTED + + class Pm3pyPlugin(WorkspacePlugin): name = "pm3py" version = "0.1.0" + #: the first connection resource created wins the serial port (reader OR sim, not both) + _mode: str | None = None #: replay-safety effect patterns (pyws safe-by-default deny-list): write / field / sim / #: destructive are withheld on autoreplay; connect / read and pure model edits replay. @@ -84,7 +172,7 @@ class Pm3pyPlugin(WorkspacePlugin): "field": ["reader.hf.tune", "reader.hf.dropfield", "reader.lf.tune", "reader.hf.search"], "write": [ "reader.hf.*.wrbl", "reader.hf.*.writebl", "reader.lf.*.writebl", - "*.sync", "*.write_block*", + "*.sync", "sim.push", "*.write_block*", ], "sim": ["sim.start", "sim.stop", "sim.start_14a", "sim.start_15693", "reader.hf.*.sim"], "destructive": ["*.format*", "*.lock*"], @@ -102,7 +190,17 @@ class Pm3pyPlugin(WorkspacePlugin): return ns def create_resource(self, name: str, config: dict, session): + if name in ("reader", "sim"): + if self._mode is not None and self._mode != name: + log.warning( + "pm3py: workspace declares both %r and %r, which contend on the serial " + "port; skipping %r", + self._mode, name, name, + ) + return None + self._mode = name if name == "reader": return ReaderResource(name, config) - # the live ``sim`` resource lands in P2 + if name == "sim": + return SimResource(name, config) return None diff --git a/tests/test_pyws_plugin.py b/tests/test_pyws_plugin.py index 11669a9..d3e078c 100644 --- a/tests/test_pyws_plugin.py +++ b/tests/test_pyws_plugin.py @@ -1,10 +1,10 @@ """Tests for the pm3py pyws plugin (P1: reader resource + namespace + effects). Hardware-free.""" import asyncio -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock from pyws import ResourceStatus, WorkspacePlugin -from pm3py.pyws_plugin import Pm3pyPlugin, ReaderResource +from pm3py.pyws_plugin import Pm3pyPlugin, ReaderResource, Sim, SimResource def _run(coro): @@ -80,6 +80,56 @@ def test_reader_resource_no_device_is_nonfatal(monkeypatch): assert r.obj is None -def test_create_resource_builds_reader(): +def test_create_resource_builds_reader_and_sim(): + # fresh plugin instances → each mode is available assert isinstance(Pm3pyPlugin().create_resource("reader", {}, session=None), ReaderResource) - assert Pm3pyPlugin().create_resource("sim", {}, session=None) is None # P2 + assert isinstance(Pm3pyPlugin().create_resource("sim", {}, session=None), SimResource) + + +def test_one_mode_per_workspace(): + plug = Pm3pyPlugin() + assert isinstance(plug.create_resource("reader", {}, None), ReaderResource) + # a second, different-mode connection resource is refused (serial contention) + assert plug.create_resource("sim", {}, None) is None + + +def test_sim_resource_opens_and_closes(): + fake = MagicMock() + r = SimResource("sim", session=fake) + _run(r.start()) + assert r.status is ResourceStatus.CONNECTED + assert isinstance(r.obj, Sim) and r.obj.session is fake + _run(r.stop()) + fake.close.assert_called_once() + + +def test_sim_resource_no_device_is_nonfatal(): + r = SimResource("sim", config={"port": "/dev/pyws-no-such-device"}) + _run(r.start()) + assert r.status is ResourceStatus.DISCONNECTED + assert r.obj is None + + +def test_sim_start_dispatches_by_protocol(): + from pm3py.sim import NTAG213, Tag15693 + + sess = MagicMock() + sim = Sim(sess) + sim.start(NTAG213()) # ISO 14443-A + sess.start_14a.assert_called_once() + sess.start_15693.assert_not_called() + sim.start(Tag15693()) # ISO 15693 + sess.start_15693.assert_called_once() + + +def test_sim_push_calls_tag_sync(): + sim = Sim(MagicMock()) + tag = MagicMock() + sim.push(tag) + tag.sync.assert_called_once() + + +def test_sim_effect_patterns(): + eff = Pm3pyPlugin.effects + assert "sim.start" in eff["sim"] and "sim.stop" in eff["sim"] + assert "sim.push" in eff["write"]