From 829f254194a72db6c24fe7232e3c3450c68a9ff0 Mon Sep 17 00:00:00 2001 From: michael Date: Mon, 6 Jul 2026 07:56:13 -0700 Subject: [PATCH] Add pyws workspace plugin (P1: reader resource + namespace + effects) Exposes pm3py as a pyws workspace, discovered via the pyws.plugins entry point: - ReaderResource: connects/disconnects a Proxmark3 as `reader`; a missing device is non-fatal (reports disconnected) - namespace injection of the transponder/sim classes, SimSession and Proxmark3 - effect declarations so autoreplay/replay withhold writes/field/sim/destructive while reads and pure model edits replay Hardware-free tests via AsyncMock. The live `sim` resource is P2. See docs/plans/2026-07-05-pm3py-pyws-plugin-plan.md. Co-Authored-By: Claude Opus 4.8 --- .../2026-07-05-pm3py-pyws-plugin-plan.md | 158 ++++++++++++++++++ pm3py/pyws_plugin.py | 108 ++++++++++++ pyproject.toml | 4 + tests/test_pyws_plugin.py | 85 ++++++++++ 4 files changed, 355 insertions(+) create mode 100644 docs/plans/2026-07-05-pm3py-pyws-plugin-plan.md create mode 100644 pm3py/pyws_plugin.py create mode 100644 tests/test_pyws_plugin.py diff --git a/docs/plans/2026-07-05-pm3py-pyws-plugin-plan.md b/docs/plans/2026-07-05-pm3py-pyws-plugin-plan.md new file mode 100644 index 0000000..863702b --- /dev/null +++ b/docs/plans/2026-07-05-pm3py-pyws-plugin-plan.md @@ -0,0 +1,158 @@ +# pm3py ↔ pyws plugin — implementation plan (2026-07-05) + +A pyws plugin, living in **this** repo, that turns pm3py into a first-class pyws workspace: +open `pyws pm3` and you land in a shell with the reader connected (or the sim ready), the +transponder classes in scope, your tags reconstructed from history, and replay/save that +understand which pm3py operations touch hardware. + +pyws itself stays domain-agnostic; all pm3py knowledge lives here and is discovered through the +`pyws.plugins` entry point. + +## Decisions (locked with the user) + +- **Sim arming is explicit (option b).** On load the connection opens and the tag model is + reconstructed, but the sim goes live only when *you* call `sim.start(tag)`. No auto-arm. +- **One connection, one mode per workspace** — sim-mode (`SimSession` owns the serial port) or + reader-mode (`Proxmark3` owns it), never both live. They can't share `/dev/ttyACM*`. +- **`sim` is a lifecycle wrapper around `SimSession`** (which already runs its own trace thread) — + response-table sims only for now, since upstream's WTX relay (`sim_session.py:_relay_loop`) is a + stub for future 14a Layer-4. +- **14a / NTAG first**, ISO-15693 a fast-follow. +- `SimSession.open()` → `connect` effect (replayable); it raises loudly if no PM3 is attached — the + resource surfaces that as a *disconnected* status on load (non-fatal) and as a clear error on an + explicit call. + +## Wiring + +- Module: `pm3py/pyws_plugin.py`, class `Pm3pyPlugin(WorkspacePlugin)`. +- Entry point in pm3py's `pyproject.toml`: + ```toml + [project.entry-points."pyws.plugins"] + pm3py = "pm3py.pyws_plugin:Pm3pyPlugin" + + [project.optional-dependencies] + pyws = ["pyws"] # installs the engine; dependency points pm3py -> pyws, never the reverse + ``` +- Install for use: `pip install -e ".[pyws]"` into the pm3py venv (which already has pm3py + deps). + +## What the plugin provides + +### Namespace (`Pm3pyPlugin.namespace(session)`) +Inject what a pm3py workspace user reaches for, so `startup.py` needs no `from pm3py… import …`: +- `SimSession` and the common tag classes (`NTAG213/215/216`, `MifareUltralight*`, + `MifareClassicTag`, `NfcType2Tag`, `Tag15693`/`NfcType5Tag`, implant factories `xNT/xM1/NExT/…`). + These are re-exported from `pm3py.sim` (`pm3py/sim/__init__.py:71`). +- `reader` and/or `sim` (the resource objects, see below), bound by the resource layer. +- NDEF helpers (`ndef_uri`, `ndef_text`, `ndef_mime` from `pm3py.sim.type5`). + +Keep it a curated set, not a wildcard dump — anything else is a normal `from pm3py.sim import X`. + +### `sim` resource (sim-mode) — `create_resource("sim", cfg, session)` +A pyws `Resource` wrapping a `SimSession`. +- `start()` (resource lifecycle, at session start): `SimSession.open(port, baudrate)` + (`sim_session.py:73`). Port from cfg (`sim: {port: /dev/ttyACM0}`), else pm3py auto-detect. + On failure (no device) → catch, `status = disconnected`, log a notice; **do not crash the + session** (you can still edit tags offline). +- Bound object `sim` exposes a thin facade over the session: + - `sim.start(tag, *, tagtype=…, trace=False)` → dispatch to `sess.start_14a(tag, …)` + (`sim_session.py:116`) or `sess.start_15693(tag, …)` (`:226`) by tag base class. **This is the + explicit arm.** Effect: `sim`. + - `sim.stop()` → `sess.stop()` (`:515`). Effect: `sim`. + - `sim.push(tag=None)` → `tag.sync()` (or `sess.sync()`), the "push my edits to the running sim" + step. Effect: `write`. + - `sim.frames` → `sess.entries` (decoded reader↔tag trace). +- `stop()` (session exit): `sess.stop(); sess.close()` (`:531`) — port released, tag unbound. + +### `reader` resource (reader-mode) — `create_resource("reader", cfg, session)` +A pyws `Resource` wrapping `Proxmark3` (`core/client.py:50`). +- `start()`: build `Proxmark3(port, baudrate)` and connect. Since pyws resources start on the + ResourceManager's async loop, `await pm3.connect()` fits naturally; the bound `reader` object is + the connected client (`.hw/.hf/.lf`, `pm3.hf.iso14a.scan()`, `pm3.hf.mfu.rdbl()`, …). No-device → + disconnected status, non-fatal. +- `stop()`: `await pm3.disconnect()`. + +### One mode per workspace +A workspace declares exactly one of `reader:` / `sim:` under `resources:`. If both are declared, +the plugin refuses the second with a clear message (they contend on the port). In-session switching +(`sim.stop()` then connect a reader) is possible but not the default path. + +### `tag` — user-created, **not** a resource +The tag model is created by you (`authorized_tag = NTAG213(uid=…, password=…, pack=…)`) in +`startup.py` or interactively, and reconstructed on load by autoreplay. It is *not* a plugin +resource. (Optional future convenience: a `tag:` resource that builds a default tag from yaml +config — deferred; yaml is a poor place for uid/keys/memory.) + +## Effect declarations (replay safety) + +`Pm3pyPlugin.effects` — pattern policy consumed by pyws's classifier. Safe-by-default deny-list: +anything classified `write/field/sim/destructive` is withheld on autoreplay; everything else +(incl. `connect`, `read`, pure model edits) replays. + +```python +effects = { + "connect": ["sim.open", "sim.close", "reader.connect", "reader.disconnect"], + "read": ["reader.hf.*.scan", "reader.hf.*.rdbl", "reader.hf.*.rdsc", + "reader.hf.mfu.rdbl", "reader.hf.*.inventory", "*.read_block*"], + "field": ["reader.hf.tune", "reader.hf.dropfield", "reader.lf.tune"], + "write": ["reader.hf.*.wrbl", "reader.hf.*.writebl", "reader.lf.*.writebl", + "*.sync", "*.write_block*"], + "sim": ["sim.start", "sim.stop", "sim.start_14a", "sim.start_15693", + "reader.hf.*.sim"], + "destructive": ["*.format*", "*.lock*"], +} +``` + +Consequences (the important ones): +- `authorized_tag = NTAG213(…)` and `tag.set_page(4, …)` / `tag.set_ndef(…)` → no match → `unknown` + → **replay** (pure-Python model edits, no device). So the tag rebuilds on load. +- `tag.sync()` → `write` → **withheld** on autoreplay. Reconstruction rebuilds the *model*, never + re-pushes to firmware. +- `sim.start(tag)` → `sim` → **withheld** → the sim is armed explicitly (decision b). +- `reader.hf.iso14a.scan()` → `read` → replays; `reader.hf.mfu.wrbl(…)` → `write` → withheld. + +Patterns are refined against real method names during P3; treat the above as the shape. + +## How it rides pyws's features + +- **Autoreplay + `tag.sync()` ordering** — the model rebuilds from history (`set_page` etc. replay), + but `sync()` is withheld, so after load the *model* is current while firmware is not. Since arming + is explicit, your `sim.start(tag)` (or `sim.push()`) is what pushes the final model — no stale-sync + hazard, and no arming before the tag exists. +- **save() / state.py** — the tag is a live object → not written to `state.py` (only round-trippable + data is). It comes back via history replay of its construction + edits. Keys/uids that you typed as + literals persist fine; the object is rebuilt, not serialized. +- **Events** — the plugin may subscribe to `WorkspaceReloaded` later for niceties (e.g. warn if the + live sim is now serving a tag you've edited but not re-pushed). Not required for v1. +- **Serial arbitration** — enforced by one-mode-per-workspace; the resource owns the single port. + +## Phases + +- **P1 — reader + namespace.** `Pm3pyPlugin` + entry point + `[pyws]` extra; `reader` resource + (connect/disconnect) + namespace injection. `pyws` in a reader-mode workspace → connected `reader`. + Tested against pm3py's existing `AsyncMock` transport (no hardware). +- **P2 — sim resource.** `sim` resource wrapping `SimSession` (open on start, explicit + `sim.start(tag)` / `sim.stop`, `sim.push`, `sim.frames`, clean teardown). Mock-tested where the + serial layer allows; otherwise a thin fake `SimSession`. +- **P3 — effects + docs.** Effect patterns above, verified against real method names; add the pyws + integration section to pm3py's `CLAUDE.md`; a documented (non-scaffolded) `pm3` workspace example. +- **P4 — hardware acceptance.** `pyws pm3` in sim-mode → create/reconstruct a tag → `sim.start(tag)` + → read it back with the ACR1552U. The real end-to-end. + +## Testing (hardware-free) + +- Reuse pm3py's `AsyncMock` transport pattern for the `reader` resource. +- For `sim`, inject a fake `SimSession` (or patch `SimSession.open`) so the resource lifecycle, + arming, effect classification, and teardown are all testable without a PM3. +- pyws-side already covers resource lifecycle, effect gating, autoreplay — the plugin tests only its + own wiring. + +## Open / future +- ISO-15693 sim path (`start_15693`, response-table compile/upload). +- Auto-arm option (opt-in `sim.live: true`) if the explicit model gets tedious — the `WorkspaceLoaded` + hook is already there. +- Dual-interface (NTAG5, PM3 RF + MCU I2C via `DualInterfaceSession`) as a distinct resource. +- WTX relay for 14a Layer-4 once upstream `_relay_loop` is real. + +## pm3py docs to update (P3) +- `CLAUDE.md`: a short "pyws integration" section — entry point, `[pyws]` extra, the reader/sim + resources, and the effect/replay semantics. diff --git a/pm3py/pyws_plugin.py b/pm3py/pyws_plugin.py new file mode 100644 index 0000000..3840e71 --- /dev/null +++ b/pm3py/pyws_plugin.py @@ -0,0 +1,108 @@ +"""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. + +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. +""" +from __future__ import annotations + +from typing import Any + +from pyws import Resource, ResourceStatus, WorkspacePlugin + +#: names injected into the workspace namespace from pm3py.sim (guarded by hasattr) +_SIM_EXPORTS = [ + "SimSession", + "NTAG210", "NTAG212", "NTAG213", "NTAG215", "NTAG216", + "MifareUltralight", "MifareUltralightC", "MifareUltralightEV1", + "MifareClassicTag", "NfcType2Tag", "NfcType4Tag", + "Tag15693", "NfcType5Tag", "IcodeSlix2Tag", + "xNT", "xM1", "NExT", "FlexDF", + "ndef_uri", "ndef_text", "ndef_mime", +] + + +class ReaderResource(Resource): + """A connected ``Proxmark3`` client, exposed as ``reader`` in the namespace. + + Opens the device on session start and drops it on stop. A missing / unreachable device is + non-fatal — the resource reports ``disconnected`` so you can still work offline. ``client`` + is injectable for tests. + """ + + kind = "reader" + + def __init__(self, name: str = "reader", config: dict | None = None, *, client: Any = None): + super().__init__(name, obj=None, config=config or {}) + self._client = client + + async def start(self) -> None: + self.status = ResourceStatus.CONNECTING + try: + if self._client is None: + from pm3py import Proxmark3 + + self._client = Proxmark3( + self.config.get("port"), self.config.get("baudrate", 115200) + ) + await self._client.connect() + except Exception as exc: # no device / connect failure — non-fatal + self.status = ResourceStatus.DISCONNECTED + self.detail = f"no device ({type(exc).__name__}: {exc})" + self._client = None + return + self.obj = self._client + self.status = ResourceStatus.CONNECTED + fw = getattr(self._client, "firmware", None) + self.detail = str(getattr(fw, "version_string", "") or "connected") + + async def stop(self) -> None: + if self._client is not None: + try: + await self._client.disconnect() + except Exception: + pass + self.status = ResourceStatus.DISCONNECTED + + +class Pm3pyPlugin(WorkspacePlugin): + name = "pm3py" + version = "0.1.0" + + #: 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. + #: Patterns are matched against the FULL dotted call path (e.g. reader.hf.mfu.wrbl). + effects = { + "connect": ["reader.connect", "reader.disconnect", "sim.open", "sim.close"], + "read": [ + "reader.hf.*.scan", "reader.hf.*.rdbl", "reader.hf.*.rdsc", + "reader.hf.mfu.rdbl", "reader.hf.*.inventory", "reader.lf.*.read", "*.read_block*", + ], + "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*", + ], + "sim": ["sim.start", "sim.stop", "sim.start_14a", "sim.start_15693", "reader.hf.*.sim"], + "destructive": ["*.format*", "*.lock*"], + } + + def namespace(self, session) -> dict[str, Any]: + import pm3py + import pm3py.sim as sim + + ns: dict[str, Any] = {"Proxmark3": pm3py.Proxmark3} + for name in _SIM_EXPORTS: + obj = getattr(sim, name, None) + if obj is not None: + ns[name] = obj + return ns + + def create_resource(self, name: str, config: dict, session): + if name == "reader": + return ReaderResource(name, config) + # the live ``sim`` resource lands in P2 + return None diff --git a/pyproject.toml b/pyproject.toml index a365082..5e251c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,10 @@ dev = ["pytest"] [project.scripts] pm3flash = "pm3py.flash_cli:main" +# pyws workspace plugin — discovered by the pyws engine (installed alongside) via this group. +[project.entry-points."pyws.plugins"] +pm3py = "pm3py.pyws_plugin:Pm3pyPlugin" + [tool.setuptools.packages.find] include = ["pm3py*"] diff --git a/tests/test_pyws_plugin.py b/tests/test_pyws_plugin.py new file mode 100644 index 0000000..11669a9 --- /dev/null +++ b/tests/test_pyws_plugin.py @@ -0,0 +1,85 @@ +"""Tests for the pm3py pyws plugin (P1: reader resource + namespace + effects). Hardware-free.""" +import asyncio +from unittest.mock import AsyncMock + +from pyws import ResourceStatus, WorkspacePlugin + +from pm3py.pyws_plugin import Pm3pyPlugin, ReaderResource + + +def _run(coro): + # match pm3py's suite convention — a shared loop that isn't closed between tests + # (asyncio.run() closes the loop and poisons every test that runs after this file) + return asyncio.get_event_loop().run_until_complete(coro) + + +def test_plugin_is_a_workspace_plugin(): + assert issubclass(Pm3pyPlugin, WorkspacePlugin) + assert Pm3pyPlugin.name == "pm3py" + + +def test_entry_point_registered(): + # pyws discovers plugins from this group; requires an (editable) install of pm3py + from importlib.metadata import entry_points + + eps = entry_points() + group = eps.select(group="pyws.plugins") if hasattr(eps, "select") else eps.get("pyws.plugins", []) + assert "pm3py" in {ep.name for ep in group} + + +def test_namespace_injects_tag_classes(): + ns = Pm3pyPlugin().namespace(session=None) + assert "SimSession" in ns + assert "NTAG213" in ns and callable(ns["NTAG213"]) + assert "Proxmark3" in ns + # a constructed tag really is a pm3py transponder + tag = ns["NTAG213"]() + assert hasattr(tag, "uid") + + +def test_effects_cover_reader_and_sim(): + eff = Pm3pyPlugin.effects + assert any("wrbl" in p for p in eff["write"]) + assert any("scan" in p for p in eff["read"]) + assert "sim.start" in eff["sim"] + assert any(p == "*.sync" for p in eff["write"]) # tag.sync() is a write + + +def test_effects_classify_nested_pm3py_calls(): + # end-to-end: the declared patterns must actually classify pm3py's nested API + from pyws.effects import EffectResolver, classify_statement + + r = EffectResolver() + r.add_plugin(Pm3pyPlugin()) + assert classify_statement("reader.hf.mfu.wrbl(4, b'aa')", {}, r) == "write" + assert classify_statement("reader.hf.iso14a.scan()", {}, r) == "read" + assert classify_statement("authorized_tag.sync()", {}, r) == "write" + # a pure model edit is unmatched -> unknown -> replays under safe-by-default + assert classify_statement("authorized_tag.set_page(4, b'aa')", {}, r) == "unknown" + + +def test_reader_resource_connects_and_disconnects(): + mock = AsyncMock() + r = ReaderResource("reader", client=mock) + _run(r.start()) + assert r.status is ResourceStatus.CONNECTED + assert r.obj is mock + mock.connect.assert_awaited_once() + _run(r.stop()) + mock.disconnect.assert_awaited_once() + assert r.status is ResourceStatus.DISCONNECTED + + +def test_reader_resource_no_device_is_nonfatal(monkeypatch): + import pm3py.core.client as client + + monkeypatch.setattr(client, "find_proxmark_port", lambda: None) + r = ReaderResource("reader") # no client, no port -> Proxmark3(None) raises -> disconnected + _run(r.start()) + assert r.status is ResourceStatus.DISCONNECTED + assert r.obj is None + + +def test_create_resource_builds_reader(): + assert isinstance(Pm3pyPlugin().create_resource("reader", {}, session=None), ReaderResource) + assert Pm3pyPlugin().create_resource("sim", {}, session=None) is None # P2