feat(cli): add pm3py client — autocompleting console over the stock proxmark3 client
Wraps the standard Proxmark3 C client (so it works with any Proxmark/firmware) in a prompt_toolkit REPL: full command-tree completion sourced from the client's commands.json, option-flag completion, Tab-to-descend, a press-right help toggle, and the client's own coloured output. Two swappable backends behind ClientDriver: in-process SWIG (_pm3) by default, a pexpect-around-the-binary fallback otherwise. Both auto-detect /dev/ttyACM* and discover the checkout via its doc/commands.json marker (walk-up + side-by-side), so no path is hardcoded. `pm3py client --pack OUT` spins off a self-contained drop-in — a readable .py, or a .pyz with prompt_toolkit/pexpect bundled — amalgamated from the package (the tested source of truth) so it can't drift. Adds pexpect to deps; hardware-free tests in tests/test_clientcli.py. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
8
pm3py/cli/clientcli/__init__.py
Normal file
8
pm3py/cli/clientcli/__init__.py
Normal file
@@ -0,0 +1,8 @@
|
|||||||
|
"""``pm3py client`` — an interactive prompt_toolkit REPL that wraps the stock Proxmark3 C client.
|
||||||
|
|
||||||
|
Unlike :mod:`pm3py.cli.rawcli` (which speaks pm3py's pure-Python protocol stack), this drives the
|
||||||
|
real ``proxmark3`` binary over pexpect, so it works with any Proxmark3 / firmware. It offers
|
||||||
|
autocompletion and help hints for the full pm3 command tree — sourced from the client's
|
||||||
|
``commands.json`` help dump — plus a right-arrow help toggle: highlight a completion, press RIGHT to
|
||||||
|
expand its full help, LEFT / ESCAPE to back out.
|
||||||
|
"""
|
||||||
105
pm3py/cli/clientcli/app.py
Normal file
105
pm3py/cli/clientcli/app.py
Normal file
@@ -0,0 +1,105 @@
|
|||||||
|
"""clientcli interactive loop — a prompt_toolkit REPL that drives the stock ``proxmark3`` client
|
||||||
|
with command-tree autocompletion, live hints, and a right-arrow help toggle. Mirrors the
|
||||||
|
:mod:`pm3py.cli.rawcli.app` skeleton; the testable logic lives in the sibling modules."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import sys
|
||||||
|
|
||||||
|
from .tree import load_command_tree
|
||||||
|
from .driver import make_driver, DriverError
|
||||||
|
from .session import ClientSession
|
||||||
|
from .completer import ClientCompleter
|
||||||
|
from .keys import install_key_bindings
|
||||||
|
from .hints import one_line_hint, resolve_target, panel_for
|
||||||
|
|
||||||
|
_DEFAULT_HINT = "type a pm3 command · Tab to complete · → toggles help on a selection · 'quit'"
|
||||||
|
|
||||||
|
|
||||||
|
def dispatch(state, line, out=print) -> bool:
|
||||||
|
"""Handle one input line. Returns False to exit. ``quit``/``exit``/``q`` leave the wrapper;
|
||||||
|
everything else is sent to the client and its output printed. A driver failure ends the loop."""
|
||||||
|
cmd = line.strip()
|
||||||
|
if cmd.lower() in ("quit", "exit", "q"):
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
output = state.driver.execute(cmd)
|
||||||
|
except DriverError as exc:
|
||||||
|
out(f"client error: {exc}")
|
||||||
|
return False
|
||||||
|
if output:
|
||||||
|
out(output)
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def run(port=None, commands=None, client="proxmark3", driver="auto") -> int:
|
||||||
|
from prompt_toolkit import PromptSession, print_formatted_text
|
||||||
|
from prompt_toolkit.application import get_app
|
||||||
|
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
|
||||||
|
from prompt_toolkit.formatted_text import ANSI
|
||||||
|
from prompt_toolkit.key_binding import KeyBindings
|
||||||
|
from prompt_toolkit.patch_stdout import patch_stdout
|
||||||
|
from prompt_toolkit.shortcuts import CompleteStyle
|
||||||
|
|
||||||
|
def out(text):
|
||||||
|
# render the ANSI our client/formatters emit (colour on a TTY, stripped when piped)
|
||||||
|
print_formatted_text(ANSI(str(text)))
|
||||||
|
|
||||||
|
try:
|
||||||
|
tree, meta = load_command_tree(commands)
|
||||||
|
except FileNotFoundError as exc:
|
||||||
|
print(f"pm3py client: {exc}", file=sys.stderr)
|
||||||
|
return 2
|
||||||
|
|
||||||
|
dev = make_driver(port=port, client=client, driver=driver)
|
||||||
|
try:
|
||||||
|
dev.start()
|
||||||
|
except DriverError as exc:
|
||||||
|
print(f"pm3py client: {exc}", file=sys.stderr)
|
||||||
|
return 1
|
||||||
|
|
||||||
|
state = ClientSession(dev, tree)
|
||||||
|
|
||||||
|
def bottom_toolbar():
|
||||||
|
# live relevance for the current input; the expanded help panel when the toggle is open
|
||||||
|
try:
|
||||||
|
buf = get_app().current_buffer
|
||||||
|
cur = buf.complete_state.current_completion if buf.complete_state else None
|
||||||
|
if state.help_open and cur is not None:
|
||||||
|
panel = panel_for(resolve_target(tree, buf.text, cur))
|
||||||
|
if panel:
|
||||||
|
return ANSI(panel)
|
||||||
|
hint = one_line_hint(tree, buf.text)
|
||||||
|
except Exception:
|
||||||
|
hint = None
|
||||||
|
return ANSI(hint) if hint else _DEFAULT_HINT
|
||||||
|
|
||||||
|
kb = KeyBindings()
|
||||||
|
install_key_bindings(kb, state)
|
||||||
|
session = PromptSession(
|
||||||
|
completer=ClientCompleter(state),
|
||||||
|
complete_while_typing=True, # ipython-style live menu
|
||||||
|
complete_style=CompleteStyle.COLUMN, # single column keeps ←/→ free for the toggle
|
||||||
|
auto_suggest=AutoSuggestFromHistory(), # inline suggestion hints
|
||||||
|
bottom_toolbar=bottom_toolbar, # live relevance / help panel
|
||||||
|
key_bindings=kb,
|
||||||
|
)
|
||||||
|
|
||||||
|
def message():
|
||||||
|
return ANSI(state.colored_breadcrumb() + " ")
|
||||||
|
|
||||||
|
n = meta.get("commands_extracted", "?")
|
||||||
|
out(f"pm3py client — {n} commands · {dev.backend} backend · → help · 'quit' to exit")
|
||||||
|
with patch_stdout():
|
||||||
|
while True:
|
||||||
|
state.help_open = False # each new line starts with help collapsed
|
||||||
|
try:
|
||||||
|
line = session.prompt(message)
|
||||||
|
except (EOFError, KeyboardInterrupt):
|
||||||
|
break
|
||||||
|
if not line.strip():
|
||||||
|
continue
|
||||||
|
if not dispatch(state, line, out):
|
||||||
|
break
|
||||||
|
dev.close()
|
||||||
|
out("bye")
|
||||||
|
return 0
|
||||||
74
pm3py/cli/clientcli/completer.py
Normal file
74
pm3py/cli/clientcli/completer.py
Normal file
@@ -0,0 +1,74 @@
|
|||||||
|
"""Completion for ``pm3py client``: walk the command trie by cursor position, offering subcommands
|
||||||
|
at each level and option flags on a leaf. Models :class:`pm3py.cli.rawcli.completer.RawCompleter`
|
||||||
|
— route by position, yield ``Completion`` with ``display_meta`` as the tooltip."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from prompt_toolkit.completion import Completer, Completion
|
||||||
|
|
||||||
|
from .tree import parse_option
|
||||||
|
|
||||||
|
|
||||||
|
def _skip(flags) -> bool:
|
||||||
|
return "--help" in flags or "-h" in flags
|
||||||
|
|
||||||
|
|
||||||
|
class ClientCompleter(Completer):
|
||||||
|
def __init__(self, session):
|
||||||
|
self._session = session
|
||||||
|
|
||||||
|
def get_completions(self, document, complete_event):
|
||||||
|
tree = self._session.tree
|
||||||
|
text = document.text_before_cursor
|
||||||
|
tokens = text.split()
|
||||||
|
at_word = bool(tokens) and not text.endswith(" ")
|
||||||
|
partial = tokens[-1] if at_word else ""
|
||||||
|
path = tokens[:-1] if at_word else tokens
|
||||||
|
|
||||||
|
node = tree.node_at(path)
|
||||||
|
if node is not None:
|
||||||
|
# inside the tree: a leaf with no more subcommands (or a flag being typed) → options;
|
||||||
|
# otherwise the child subcommands at this level.
|
||||||
|
if node.is_leaf and (partial.startswith("-") or not node.children):
|
||||||
|
yield from self._options(node.entry, partial)
|
||||||
|
else:
|
||||||
|
yield from self._subcommands(node, partial)
|
||||||
|
return
|
||||||
|
|
||||||
|
# a path token fell off the tree — we're in a leaf's argument territory; offer flags only.
|
||||||
|
leaf = self._deepest_leaf(tree, path)
|
||||||
|
if leaf is not None and partial.startswith("-"):
|
||||||
|
yield from self._options(leaf, partial)
|
||||||
|
|
||||||
|
def _subcommands(self, node, partial):
|
||||||
|
for tok in sorted(node.children):
|
||||||
|
if not tok.startswith(partial):
|
||||||
|
continue
|
||||||
|
child = node.children[tok]
|
||||||
|
if child.is_leaf:
|
||||||
|
meta = child.entry.get("description", "")
|
||||||
|
else:
|
||||||
|
meta = f"{len(child.children)} subcommands"
|
||||||
|
# insert a trailing space so accepting the completion commits the token and immediately
|
||||||
|
# re-triggers completion at the next level ("hf " + <tab> → "hf 14a " → 14a's children)
|
||||||
|
yield Completion(tok + " ", start_position=-len(partial), display=tok, display_meta=meta)
|
||||||
|
|
||||||
|
def _options(self, entry, partial):
|
||||||
|
for opt in entry.get("options", []):
|
||||||
|
flags, meta, desc = parse_option(opt)
|
||||||
|
if not flags or _skip(flags):
|
||||||
|
continue
|
||||||
|
match = next((f for f in flags if f.startswith(partial)), None)
|
||||||
|
if match is None:
|
||||||
|
continue
|
||||||
|
disp = (", ".join(flags) + (f" {meta}" if meta else "")).strip()
|
||||||
|
yield Completion(match, start_position=-len(partial), display=disp, display_meta=desc)
|
||||||
|
|
||||||
|
def _deepest_leaf(self, tree, path):
|
||||||
|
node, leaf = tree.root, None
|
||||||
|
for tok in path:
|
||||||
|
node = node.children.get(tok)
|
||||||
|
if node is None:
|
||||||
|
break
|
||||||
|
if node.is_leaf:
|
||||||
|
leaf = node.entry
|
||||||
|
return leaf
|
||||||
281
pm3py/cli/clientcli/driver.py
Normal file
281
pm3py/cli/clientcli/driver.py
Normal file
@@ -0,0 +1,281 @@
|
|||||||
|
"""Drivers for the stock Proxmark3 C client.
|
||||||
|
|
||||||
|
``pm3py client`` does not reimplement the wire protocol — it drives the real client, so it works
|
||||||
|
with any Proxmark3 / firmware. Two interchangeable backends behind :class:`ClientDriver`:
|
||||||
|
|
||||||
|
- :class:`SwigDriver` (default) — the in-process ``_pm3`` SWIG bindings: ``pm3.pm3(port).console()``
|
||||||
|
+ ``grabbed_output``. Cleaner (no prompt-matching) but needs the built ``_pm3`` extension and a
|
||||||
|
connected device (the C ``pm3_open`` calls ``exit()`` on a real port with no responding PM3).
|
||||||
|
- :class:`PexpectDriver` — spawns the ``proxmark3`` binary and reads output up to the next
|
||||||
|
``pm3 -->`` prompt. Zero build; the fallback when ``_pm3`` isn't importable. The client's ANSI
|
||||||
|
colour is preserved (we ANSI-strip only for prompt/echo matching), and we never mutate ``prefs``.
|
||||||
|
|
||||||
|
:func:`make_driver` picks SWIG when ``_pm3`` is importable, else pexpect.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import glob
|
||||||
|
import importlib.machinery
|
||||||
|
import importlib.util
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import shutil
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from .tree import find_repo_root
|
||||||
|
|
||||||
|
#: full prompt matcher for clean text: ``"[<dev><net><ctx>] pm3 --> "``. Used to parse a label the
|
||||||
|
#: caller already has in hand (and in tests); pexpect matches on the tail instead — see below.
|
||||||
|
PROMPT_RE = re.compile(r"\[([^\]]*)\] pm3 -->\s*$")
|
||||||
|
#: what pexpect waits for. The ``pm3 --> `` tail is emitted uncoloured, so this matches regardless
|
||||||
|
#: of colour or the bracketed-paste sequences readline emits around the ``[...]`` label — matching
|
||||||
|
#: the label directly would let a stray ``[`` (from ``\x1b[?2004h``) capture the wrong bracket.
|
||||||
|
_EXPECT_RE = re.compile(r"pm3 -->\s*$")
|
||||||
|
#: the ``[label]`` at the end of the ANSI-stripped text just before the tail.
|
||||||
|
_LABEL_RE = re.compile(r"\[([^\]]*)\]\s*$")
|
||||||
|
_ANSI = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]")
|
||||||
|
#: every CSI sequence except SGR colour (final byte ``m``) — strips readline's bracketed-paste /
|
||||||
|
#: cursor / erase controls from passed-through output while keeping the client's colour intact.
|
||||||
|
_NONSGR = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-ln-~]")
|
||||||
|
|
||||||
|
#: fallback locations searched when a bare client name isn't on PATH (typical fork build layouts).
|
||||||
|
_CLIENT_FALLBACKS = (
|
||||||
|
"~/proxmark3/client/proxmark3",
|
||||||
|
"/usr/local/bin/proxmark3",
|
||||||
|
"/usr/bin/proxmark3",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_client(client: str) -> str:
|
||||||
|
"""Resolve the client binary. An explicit path is used verbatim; a bare name is looked up on
|
||||||
|
PATH. Only the *default* ``proxmark3`` name falls back to known fork build locations — an
|
||||||
|
explicitly-named client that isn't found is returned unchanged, so the spawn fails clearly
|
||||||
|
rather than silently running a different binary. """
|
||||||
|
if os.sep in client or (os.altsep and os.altsep in client):
|
||||||
|
return client
|
||||||
|
found = shutil.which(client)
|
||||||
|
if found:
|
||||||
|
return found
|
||||||
|
if client == "proxmark3": # auto-locate only the default binary
|
||||||
|
candidates = []
|
||||||
|
root = find_repo_root()
|
||||||
|
if root is not None: # the checkout this file was dropped beside
|
||||||
|
candidates.append(root / "client" / "proxmark3")
|
||||||
|
candidates += [Path(c).expanduser() for c in _CLIENT_FALLBACKS]
|
||||||
|
for p in candidates:
|
||||||
|
if p.is_file() and os.access(p, os.X_OK):
|
||||||
|
return str(p)
|
||||||
|
return client
|
||||||
|
|
||||||
|
|
||||||
|
class DriverError(RuntimeError):
|
||||||
|
"""The underlying client could not be launched, timed out, or exited."""
|
||||||
|
|
||||||
|
|
||||||
|
def strip_ansi(text: str) -> str:
|
||||||
|
return _ANSI.sub("", text or "")
|
||||||
|
|
||||||
|
|
||||||
|
def _scrubbed_env() -> dict:
|
||||||
|
"""A minimal environment for spawning the client. Snap / GTK / ``LD_*`` vars leaking in from a
|
||||||
|
sandboxed terminal (e.g. VSCode's snap) make the Qt-linked client crash on a libpthread symbol
|
||||||
|
clash, so we hand it only what it needs."""
|
||||||
|
env = {"TERM": os.environ.get("TERM", "xterm")}
|
||||||
|
for key in ("HOME", "PATH", "USER", "LANG", "LC_ALL"):
|
||||||
|
if key in os.environ:
|
||||||
|
env[key] = os.environ[key]
|
||||||
|
env.setdefault("PATH", "/usr/local/bin:/usr/bin:/bin")
|
||||||
|
return env
|
||||||
|
|
||||||
|
|
||||||
|
def _extract_output(raw: str, sent: str) -> str:
|
||||||
|
"""A command's output: ``child.before`` minus the echoed command line and framing blank lines.
|
||||||
|
The client's ANSI colour is preserved; readline's non-colour control sequences (bracketed
|
||||||
|
paste, cursor moves) are stripped, and the echo comparison is fully ANSI-stripped."""
|
||||||
|
lines = _NONSGR.sub("", raw or "").replace("\r", "").split("\n")
|
||||||
|
while lines and not strip_ansi(lines[0]).strip():
|
||||||
|
lines.pop(0)
|
||||||
|
if lines and strip_ansi(lines[0]).strip() == sent.strip():
|
||||||
|
lines.pop(0)
|
||||||
|
return "\n".join(lines).strip("\n")
|
||||||
|
|
||||||
|
|
||||||
|
class ClientDriver:
|
||||||
|
"""Driver interface. ``prompt`` is the last-seen context label (``usb`` / ``offline`` / …);
|
||||||
|
``backend`` names the mechanism (``swig`` / ``pexpect``)."""
|
||||||
|
|
||||||
|
prompt = "offline"
|
||||||
|
backend = "?"
|
||||||
|
|
||||||
|
def start(self) -> None: ...
|
||||||
|
def execute(self, line: str) -> str: ...
|
||||||
|
def close(self) -> None: ...
|
||||||
|
|
||||||
|
|
||||||
|
class PexpectDriver(ClientDriver):
|
||||||
|
backend = "pexpect"
|
||||||
|
|
||||||
|
def __init__(self, port=None, client="proxmark3", timeout=30):
|
||||||
|
self._port = port
|
||||||
|
self._client = resolve_client(client)
|
||||||
|
self._timeout = timeout
|
||||||
|
self._child = None
|
||||||
|
self.prompt = "offline"
|
||||||
|
|
||||||
|
def start(self) -> None:
|
||||||
|
import pexpect
|
||||||
|
|
||||||
|
# the stock binary does NOT auto-detect a port (that's the `pm3` wrapper's job) — without
|
||||||
|
# one it drops to offline mode, so resolve a port ourselves when none was given
|
||||||
|
port = self._port or _autodetect_port()
|
||||||
|
args = [port] if port else []
|
||||||
|
try:
|
||||||
|
self._child = pexpect.spawn(
|
||||||
|
self._client, args, env=_scrubbed_env(),
|
||||||
|
encoding="utf-8", codec_errors="replace", timeout=self._timeout,
|
||||||
|
)
|
||||||
|
self._child.expect(_EXPECT_RE)
|
||||||
|
except pexpect.ExceptionPexpect as exc:
|
||||||
|
raise DriverError(
|
||||||
|
f"could not start the proxmark3 client ({self._client!r}): {exc}\n"
|
||||||
|
" pass --client /path/to/proxmark3, or add it to your PATH"
|
||||||
|
) from exc
|
||||||
|
self._sync_prompt()
|
||||||
|
|
||||||
|
def execute(self, line: str) -> str:
|
||||||
|
import pexpect
|
||||||
|
|
||||||
|
if self._child is None:
|
||||||
|
raise DriverError("client not started")
|
||||||
|
try:
|
||||||
|
self._child.sendline(line)
|
||||||
|
self._child.expect(_EXPECT_RE)
|
||||||
|
except pexpect.EOF as exc:
|
||||||
|
raise DriverError("client exited") from exc
|
||||||
|
except pexpect.TIMEOUT as exc:
|
||||||
|
raise DriverError(f"client did not respond within {self._timeout}s") from exc
|
||||||
|
out = _extract_output(self._child.before, line)
|
||||||
|
self._sync_prompt()
|
||||||
|
return out
|
||||||
|
|
||||||
|
def _sync_prompt(self) -> None:
|
||||||
|
# the "[label]" sits at the end of the text just before the matched "pm3 -->" tail
|
||||||
|
m = _LABEL_RE.search(strip_ansi(self._child.before or ""))
|
||||||
|
if m:
|
||||||
|
self.prompt = m.group(1).strip() or "offline"
|
||||||
|
|
||||||
|
def close(self) -> None:
|
||||||
|
if self._child is None:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
self._child.sendline("quit")
|
||||||
|
self._child.close()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
self._child = None
|
||||||
|
|
||||||
|
|
||||||
|
#: extra checkout roots probed for the SWIG build (the discovered repo is tried first).
|
||||||
|
_SWIG_ROOTS = ("~/proxmark3",)
|
||||||
|
|
||||||
|
|
||||||
|
def _load_pm3_module():
|
||||||
|
"""Import and return the SWIG ``pm3`` module. If it isn't already importable, locate the built
|
||||||
|
library (``libpm3rrg_rdv4.so``, which exports ``PyInit__pm3``) and the ``pm3.py`` stub from the
|
||||||
|
discovered checkout and load the extension under the name ``_pm3`` — no symlink or filesystem
|
||||||
|
write. Raises :class:`ImportError` if the build can't be found."""
|
||||||
|
try:
|
||||||
|
import pm3
|
||||||
|
return pm3
|
||||||
|
except ImportError:
|
||||||
|
pass
|
||||||
|
roots = []
|
||||||
|
if os.environ.get("PM3_SWIG_PATH"):
|
||||||
|
roots.append(os.environ["PM3_SWIG_PATH"])
|
||||||
|
repo = find_repo_root()
|
||||||
|
if repo is not None:
|
||||||
|
roots.append(str(repo))
|
||||||
|
roots += [os.path.expanduser(r) for r in _SWIG_ROOTS]
|
||||||
|
for root in roots:
|
||||||
|
build = os.path.join(root, "client", "experimental_lib", "build")
|
||||||
|
pyscripts = os.path.join(root, "client", "pyscripts")
|
||||||
|
lib = os.path.join(build, "libpm3rrg_rdv4.so")
|
||||||
|
stub = os.path.join(pyscripts, "pm3.py")
|
||||||
|
if not (os.path.isfile(lib) and os.path.isfile(stub)):
|
||||||
|
# PM3_SWIG_PATH may itself be the dir holding both files
|
||||||
|
lib, stub, pyscripts = (os.path.join(root, "libpm3rrg_rdv4.so"),
|
||||||
|
os.path.join(root, "pm3.py"), root)
|
||||||
|
if not (os.path.isfile(lib) and os.path.isfile(stub)):
|
||||||
|
continue
|
||||||
|
if "_pm3" not in sys.modules:
|
||||||
|
loader = importlib.machinery.ExtensionFileLoader("_pm3", lib)
|
||||||
|
spec = importlib.util.spec_from_loader("_pm3", loader)
|
||||||
|
mod = importlib.util.module_from_spec(spec)
|
||||||
|
loader.exec_module(mod)
|
||||||
|
sys.modules["_pm3"] = mod
|
||||||
|
if pyscripts not in sys.path:
|
||||||
|
sys.path.insert(0, pyscripts)
|
||||||
|
import pm3
|
||||||
|
return pm3
|
||||||
|
raise ImportError(
|
||||||
|
"the pm3 SWIG extension (_pm3) isn't importable. Build it (in client/experimental_lib: "
|
||||||
|
"cmake .. && make -j), set $PM3_SWIG_PATH, or use --driver pexpect"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _autodetect_port():
|
||||||
|
ports = sorted(glob.glob("/dev/ttyACM*")) + sorted(glob.glob("/dev/ttyUSB*"))
|
||||||
|
return ports[0] if ports else None
|
||||||
|
|
||||||
|
|
||||||
|
class SwigDriver(ClientDriver):
|
||||||
|
"""In-process driver over the ``_pm3`` SWIG bindings. Needs a connected device: the C
|
||||||
|
``pm3_open`` calls ``exit()`` on a real port with no responding PM3, so we can't recover from a
|
||||||
|
bad connection — we only open an auto-detected or explicitly given port."""
|
||||||
|
|
||||||
|
backend = "swig"
|
||||||
|
|
||||||
|
def __init__(self, port=None, pm3_module=None):
|
||||||
|
self._port = port
|
||||||
|
self._mod = pm3_module
|
||||||
|
self._p = None
|
||||||
|
self.prompt = "offline"
|
||||||
|
|
||||||
|
def start(self) -> None:
|
||||||
|
mod = self._mod or _load_pm3_module()
|
||||||
|
port = self._port or _autodetect_port()
|
||||||
|
if port is None:
|
||||||
|
raise DriverError(
|
||||||
|
"no serial device found for the SWIG driver (no /dev/ttyACM*). plug in a "
|
||||||
|
"Proxmark, pass --port, or use --driver pexpect"
|
||||||
|
)
|
||||||
|
self._p = mod.pm3(port) # exits the process if this port has no PM3
|
||||||
|
self.prompt = "usb"
|
||||||
|
|
||||||
|
def execute(self, line: str) -> str:
|
||||||
|
if self._p is None:
|
||||||
|
raise DriverError("client not started")
|
||||||
|
# capture=False, quiet=False: let the client print its OWN coloured output straight to
|
||||||
|
# stdout. The grab buffer always strips ANSI (ui.c force-filters it), so capturing would
|
||||||
|
# lose the Proxmark's colours — the whole point of the PRINT path. Returns "" (already out).
|
||||||
|
self._p.console(line, False, False)
|
||||||
|
return ""
|
||||||
|
|
||||||
|
def close(self) -> None:
|
||||||
|
self._p = None
|
||||||
|
|
||||||
|
|
||||||
|
def make_driver(port=None, client="proxmark3", driver="auto"):
|
||||||
|
"""The driver for ``pm3py client``. ``driver`` is ``auto`` (SWIG if ``_pm3`` importable, else
|
||||||
|
pexpect), ``swig`` (error if unavailable), or ``pexpect``."""
|
||||||
|
if driver in ("auto", "swig"):
|
||||||
|
try:
|
||||||
|
mod = _load_pm3_module()
|
||||||
|
except ImportError as exc:
|
||||||
|
if driver == "swig":
|
||||||
|
raise DriverError(str(exc)) from exc
|
||||||
|
mod = None # auto → fall back to pexpect
|
||||||
|
if mod is not None:
|
||||||
|
return SwigDriver(port=port, pm3_module=mod)
|
||||||
|
return PexpectDriver(port=port, client=client)
|
||||||
93
pm3py/cli/clientcli/hints.py
Normal file
93
pm3py/cli/clientcli/hints.py
Normal file
@@ -0,0 +1,93 @@
|
|||||||
|
"""Bottom-toolbar hints for clientcli.
|
||||||
|
|
||||||
|
Two channels, both driven by the command tree (models :func:`pm3py.cli.rawcli.memory.input_hint`):
|
||||||
|
|
||||||
|
- :func:`one_line_hint` — a one-line relevance hint for the current input (always on).
|
||||||
|
- :func:`help_panel` / :func:`panel_for` — the multi-line help shown when the help toggle is open;
|
||||||
|
:func:`resolve_target` maps the *highlighted* completion back to the node whose help to show.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from .tree import parse_option
|
||||||
|
|
||||||
|
# ANSI (same palette as the trace / breadcrumb)
|
||||||
|
_DIM = "\033[2m"
|
||||||
|
_BOLD = "\033[1m"
|
||||||
|
_CYAN = "\033[36m"
|
||||||
|
_RESET = "\033[0m"
|
||||||
|
|
||||||
|
|
||||||
|
def _skip(flags) -> bool:
|
||||||
|
return "--help" in flags or "-h" in flags
|
||||||
|
|
||||||
|
|
||||||
|
def one_line_hint(tree, text):
|
||||||
|
"""A one-line hint for ``text``: a leaf's description + usage, or an interior node's subcommand
|
||||||
|
count. Resolves on the completed prefix when the buffer ends mid-word. ``None`` if unknown."""
|
||||||
|
tokens = (text or "").split()
|
||||||
|
if not tokens:
|
||||||
|
return None
|
||||||
|
node, used = tree.node_at(tokens), tokens
|
||||||
|
if node is None:
|
||||||
|
node, used = tree.node_at(tokens[:-1]), tokens[:-1]
|
||||||
|
if node is None or node is tree.root:
|
||||||
|
return None
|
||||||
|
if node.is_leaf:
|
||||||
|
e = node.entry
|
||||||
|
return f"{e['command']} — {e['description']} · {e['usage']}"
|
||||||
|
return f"{' '.join(used)} — {len(node.children)} subcommands"
|
||||||
|
|
||||||
|
|
||||||
|
def help_panel(entry) -> str:
|
||||||
|
"""The full multi-line help for a leaf command: description, usage, options, and notes."""
|
||||||
|
if not entry:
|
||||||
|
return ""
|
||||||
|
out = [f"{_BOLD}{entry['command']}{_RESET} — {entry.get('description', '')}",
|
||||||
|
f"{_DIM}usage:{_RESET} {entry.get('usage', '')}"]
|
||||||
|
rows = []
|
||||||
|
for opt in entry.get("options", []):
|
||||||
|
flags, meta, desc = parse_option(opt)
|
||||||
|
if not flags or _skip(flags):
|
||||||
|
continue
|
||||||
|
rows.append((", ".join(flags) + (f" {meta}" if meta else ""), desc))
|
||||||
|
if rows:
|
||||||
|
w = min(max(len(h) for h, _ in rows), 28)
|
||||||
|
out.append(f"{_DIM}options:{_RESET}")
|
||||||
|
out += [f" {_CYAN}{h:<{w}}{_RESET} {d}" for h, d in rows]
|
||||||
|
notes = entry.get("notes", [])
|
||||||
|
if notes:
|
||||||
|
out.append(f"{_DIM}notes:{_RESET}")
|
||||||
|
out += [f" {_DIM}{n}{_RESET}" for n in notes]
|
||||||
|
return "\n".join(out)
|
||||||
|
|
||||||
|
|
||||||
|
def panel_for(node) -> str | None:
|
||||||
|
"""The help panel for a highlighted node — a leaf's full help, or an interior node's child
|
||||||
|
list. ``None`` when there is nothing to show."""
|
||||||
|
if node is None:
|
||||||
|
return None
|
||||||
|
if node.is_leaf:
|
||||||
|
return help_panel(node.entry)
|
||||||
|
kids = sorted(node.children)
|
||||||
|
if not kids:
|
||||||
|
return None
|
||||||
|
head = f"{_BOLD}{node.token or '(commands)'}{_RESET} — {len(kids)} subcommands"
|
||||||
|
return head + "\n" + f"{_DIM}{', '.join(kids)}{_RESET}"
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_target(tree, buffer_text, current_completion):
|
||||||
|
"""The tree node whose help to show for the highlighted completion. A subcommand completion
|
||||||
|
extends the current path; a flag completion targets the leaf under the cursor. ``None`` if
|
||||||
|
nothing resolves."""
|
||||||
|
tokens = (buffer_text or "").split()
|
||||||
|
at_word = bool(tokens) and not (buffer_text or "").endswith(" ")
|
||||||
|
path = tokens[:-1] if at_word else tokens
|
||||||
|
ctext = getattr(current_completion, "text", "") if current_completion else ""
|
||||||
|
if ctext.startswith("-"):
|
||||||
|
for i in range(len(path), -1, -1):
|
||||||
|
node = tree.node_at(path[:i])
|
||||||
|
if node is not None and node.is_leaf:
|
||||||
|
return node
|
||||||
|
return None
|
||||||
|
full = path + [ctext] if ctext else path
|
||||||
|
return tree.node_at(full)
|
||||||
78
pm3py/cli/clientcli/keys.py
Normal file
78
pm3py/cli/clientcli/keys.py
Normal file
@@ -0,0 +1,78 @@
|
|||||||
|
"""The help-toggle key bindings for clientcli — the centrepiece interaction.
|
||||||
|
|
||||||
|
When a completion is highlighted in the menu, RIGHT toggles a full help panel for it in the bottom
|
||||||
|
toolbar; LEFT (primary) or ESCAPE / Ctrl-g (secondary) collapses it. Only a single ``help_open``
|
||||||
|
flag lives on the session — the toolbar recomputes the *target* from the live highlighted completion
|
||||||
|
each render, so arrowing through the menu re-targets the panel automatically.
|
||||||
|
|
||||||
|
The pure ``open_help`` / ``close_help`` helpers carry the state transitions for unit tests; the key
|
||||||
|
wiring is exercised in the live REPL.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
|
||||||
|
def open_help(session, has_selection: bool) -> bool:
|
||||||
|
"""Open help iff a completion is highlighted. Returns the resulting ``help_open`` state."""
|
||||||
|
if has_selection:
|
||||||
|
session.help_open = True
|
||||||
|
return session.help_open
|
||||||
|
|
||||||
|
|
||||||
|
def close_help(session) -> bool:
|
||||||
|
session.help_open = False
|
||||||
|
return session.help_open
|
||||||
|
|
||||||
|
|
||||||
|
def apply_tab(buff) -> None:
|
||||||
|
"""Commit the highlighted completion (or the first, if none is highlighted) into ``buff``.
|
||||||
|
|
||||||
|
The default Tab only *previews* under ``complete_while_typing`` — completions regenerate on
|
||||||
|
every buffer change, so the preview never sticks. ``apply_completion`` commits it; our
|
||||||
|
trailing-space completions then re-trigger completion at the next level. When no menu is open
|
||||||
|
yet, start one (selecting the first)."""
|
||||||
|
state = buff.complete_state
|
||||||
|
if state is None:
|
||||||
|
buff.start_completion(select_first=True)
|
||||||
|
return
|
||||||
|
completion = state.current_completion or (state.completions[0] if state.completions else None)
|
||||||
|
if completion is not None:
|
||||||
|
buff.apply_completion(completion)
|
||||||
|
|
||||||
|
|
||||||
|
def _bind(kb, key, handler, **kw) -> None:
|
||||||
|
"""Add a binding, ignoring key names a given prompt_toolkit build doesn't accept."""
|
||||||
|
try:
|
||||||
|
kb.add(key, **kw)(handler)
|
||||||
|
except ValueError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def install_key_bindings(kb, session) -> None:
|
||||||
|
"""Wire TAB completion and RIGHT/LEFT/ESCAPE/Ctrl-g help toggling onto ``kb``. Never raises on
|
||||||
|
an unknown key."""
|
||||||
|
from prompt_toolkit.filters import Condition, completion_is_selected
|
||||||
|
|
||||||
|
help_is_open = Condition(lambda: session.help_open)
|
||||||
|
|
||||||
|
# Tab commits the highlighted (or first) completion so it sticks and descends a level;
|
||||||
|
# Shift-Tab keeps prompt_toolkit's default (previous completion).
|
||||||
|
_bind(kb, "tab", lambda event: apply_tab(event.current_buffer))
|
||||||
|
|
||||||
|
def _toggle(event):
|
||||||
|
session.help_open = not session.help_open
|
||||||
|
event.app.invalidate()
|
||||||
|
|
||||||
|
def _close(event):
|
||||||
|
session.help_open = False
|
||||||
|
event.app.invalidate()
|
||||||
|
|
||||||
|
# RIGHT toggles help for the highlighted completion. With nothing selected the filter is false,
|
||||||
|
# so the binding doesn't apply and the default cursor-right still works.
|
||||||
|
_bind(kb, "right", _toggle, filter=completion_is_selected)
|
||||||
|
# LEFT is the instant back-out; active only while help is open, else default cursor-left.
|
||||||
|
_bind(kb, "left", _close, filter=help_is_open)
|
||||||
|
# ESCAPE also backs out, but it is prompt_toolkit's meta prefix, so a bare escape fires only
|
||||||
|
# after ~0.5s (timeoutlen); Ctrl-g is the instant alternative. Do NOT use eager=True — it would
|
||||||
|
# fix the delay at the cost of breaking Alt-key sequences.
|
||||||
|
_bind(kb, "escape", _close, filter=help_is_open)
|
||||||
|
_bind(kb, "c-g", _close, filter=help_is_open)
|
||||||
142
pm3py/cli/clientcli/pack.py
Normal file
142
pm3py/cli/clientcli/pack.py
Normal file
@@ -0,0 +1,142 @@
|
|||||||
|
"""Generate a standalone, dependency-light single-file ``pm3cli`` from this package.
|
||||||
|
|
||||||
|
The runtime modules (:mod:`tree`, :mod:`driver`, :mod:`session`, :mod:`hints`, :mod:`completer`,
|
||||||
|
:mod:`keys`, :mod:`app`) are concatenated in dependency order with their intra-package imports
|
||||||
|
stripped, then a self-contained ``main()`` is appended. The result is one file that needs no pm3py
|
||||||
|
install — drop it inside a stock Iceman/RRG Proxmark3 checkout and run it. Two outputs:
|
||||||
|
|
||||||
|
- ``pm3cli.py`` — a readable single file; the target runs ``pip install prompt_toolkit`` (and
|
||||||
|
``pexpect`` for the non-SWIG backend), then ``python3 pm3cli.py``.
|
||||||
|
- ``pm3cli.pyz`` — a zipapp with prompt_toolkit/pexpect bundled inside; ``python3 pm3cli.pyz`` with
|
||||||
|
nothing to install.
|
||||||
|
|
||||||
|
The package (with its tests) stays the source of truth — this only re-emits it, so the standalone
|
||||||
|
can't drift.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import importlib
|
||||||
|
import os
|
||||||
|
import shutil
|
||||||
|
import tempfile
|
||||||
|
import zipapp
|
||||||
|
|
||||||
|
#: runtime modules, in definition-before-use order (tree first, app last).
|
||||||
|
_MODULES = ("tree", "driver", "session", "hints", "completer", "keys", "app")
|
||||||
|
|
||||||
|
#: bundled into the .pyz so it runs with zero pip installs (all pure-Python).
|
||||||
|
_PYZ_DEPS = ("prompt_toolkit", "pexpect", "ptyprocess", "wcwidth")
|
||||||
|
|
||||||
|
_HEADER = '''\
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
"""pm3cli — an autocompleting REPL for the stock Proxmark3 client.
|
||||||
|
|
||||||
|
GENERATED from pm3py.cli.clientcli — do not edit by hand; regenerate with `pm3py client --pack`.
|
||||||
|
|
||||||
|
Drop this beside (or inside) a Proxmark3 checkout and run it: python3 pm3cli.py
|
||||||
|
It finds the checkout's doc/commands.json + client automatically, auto-detects /dev/ttyACM*,
|
||||||
|
and drives the stock client (SWIG _pm3 if built, else the proxmark3 binary via pexpect).
|
||||||
|
|
||||||
|
Needs prompt_toolkit (pip install prompt_toolkit); the pexpect backend also needs pexpect.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
'''
|
||||||
|
|
||||||
|
_MAIN = '''
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- standalone entrypoint
|
||||||
|
def main(argv=None) -> int:
|
||||||
|
import argparse
|
||||||
|
import importlib.util
|
||||||
|
import sys
|
||||||
|
|
||||||
|
p = argparse.ArgumentParser(
|
||||||
|
prog="pm3cli", description="autocompleting REPL for the stock Proxmark3 client")
|
||||||
|
p.add_argument("--port", default=None, help="serial port (default: auto-detect /dev/ttyACM*)")
|
||||||
|
p.add_argument("--commands", default=None,
|
||||||
|
help="path to commands.json (default: auto-locate in the checkout)")
|
||||||
|
p.add_argument("--client", default="proxmark3",
|
||||||
|
help="proxmark3 binary to wrap (default: PATH, then the checkout)")
|
||||||
|
p.add_argument("--driver", choices=["auto", "swig", "pexpect"], default="auto",
|
||||||
|
help="backend: auto (SWIG if built, else pexpect), swig, or pexpect")
|
||||||
|
args = p.parse_args(argv)
|
||||||
|
|
||||||
|
if importlib.util.find_spec("prompt_toolkit") is None:
|
||||||
|
print("pm3cli needs prompt_toolkit: pip install prompt_toolkit", file=sys.stderr)
|
||||||
|
return 2
|
||||||
|
if args.driver != "swig" and importlib.util.find_spec("pexpect") is None:
|
||||||
|
print("pm3cli's pexpect backend needs pexpect: pip install pexpect\\n"
|
||||||
|
" (or build the SWIG _pm3 lib and pass --driver swig)", file=sys.stderr)
|
||||||
|
return 2
|
||||||
|
return run(port=args.port, commands=args.commands, client=args.client, driver=args.driver)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
|
'''
|
||||||
|
|
||||||
|
|
||||||
|
def _drop(line: str) -> bool:
|
||||||
|
"""True for intra-package / future imports that must not survive amalgamation."""
|
||||||
|
s = line.lstrip()
|
||||||
|
return (s.startswith("from __future__")
|
||||||
|
or s.startswith("from .")
|
||||||
|
or s.startswith("from pm3py")
|
||||||
|
or s.startswith("import pm3py"))
|
||||||
|
|
||||||
|
|
||||||
|
def amalgamate() -> str:
|
||||||
|
"""Return the full standalone source as a string."""
|
||||||
|
here = os.path.dirname(os.path.abspath(__file__))
|
||||||
|
parts = [_HEADER]
|
||||||
|
for mod in _MODULES:
|
||||||
|
src = open(os.path.join(here, mod + ".py"), encoding="utf-8").read()
|
||||||
|
body = "\n".join(ln for ln in src.splitlines() if not _drop(ln))
|
||||||
|
parts.append(f"\n\n# ===================================================== clientcli.{mod}\n"
|
||||||
|
+ body.strip() + "\n")
|
||||||
|
parts.append(_MAIN)
|
||||||
|
return "".join(parts)
|
||||||
|
|
||||||
|
|
||||||
|
def _build_pyz(code: str, out_path: str) -> None:
|
||||||
|
import importlib.metadata as _im
|
||||||
|
|
||||||
|
with tempfile.TemporaryDirectory() as d:
|
||||||
|
with open(os.path.join(d, "__main__.py"), "w", encoding="utf-8") as f:
|
||||||
|
f.write(code)
|
||||||
|
for pkg in _PYZ_DEPS:
|
||||||
|
try:
|
||||||
|
mod = importlib.import_module(pkg)
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
src = mod.__file__
|
||||||
|
if src.endswith("__init__.py"): # a package → copy the whole dir
|
||||||
|
shutil.copytree(os.path.dirname(src), os.path.join(d, pkg),
|
||||||
|
ignore=shutil.ignore_patterns("__pycache__", "*.pyc",
|
||||||
|
"tests", "test"))
|
||||||
|
else: # a single-module dep
|
||||||
|
shutil.copy2(src, os.path.join(d, os.path.basename(src)))
|
||||||
|
# bundle the .dist-info too — some packages read their own version via
|
||||||
|
# importlib.metadata at import time (e.g. prompt_toolkit), which reads it from the zip.
|
||||||
|
try:
|
||||||
|
info = getattr(_im.distribution(pkg), "_path", None)
|
||||||
|
if info is not None and os.path.isdir(info):
|
||||||
|
shutil.copytree(info, os.path.join(d, os.path.basename(str(info))))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
zipapp.create_archive(d, out_path, interpreter="/usr/bin/env python3")
|
||||||
|
os.chmod(out_path, 0o755)
|
||||||
|
|
||||||
|
|
||||||
|
def pack(out_path: str) -> str:
|
||||||
|
"""Write the standalone to ``out_path``. A ``.pyz`` extension builds a dependency-bundled
|
||||||
|
zipapp; anything else writes the readable single ``.py``. Returns ``out_path``."""
|
||||||
|
code = amalgamate()
|
||||||
|
compile(code, out_path, "exec") # fail loudly if the amalgamation is invalid
|
||||||
|
if out_path.endswith(".pyz"):
|
||||||
|
_build_pyz(code, out_path)
|
||||||
|
else:
|
||||||
|
with open(out_path, "w", encoding="utf-8") as f:
|
||||||
|
f.write(code)
|
||||||
|
os.chmod(out_path, 0o755)
|
||||||
|
return out_path
|
||||||
40
pm3py/cli/clientcli/session.py
Normal file
40
pm3py/cli/clientcli/session.py
Normal file
@@ -0,0 +1,40 @@
|
|||||||
|
"""clientcli session state: the driver, the loaded command tree, and the help-toggle flag. Kept
|
||||||
|
free of prompt_toolkit so it stays unit-testable (mirrors :mod:`pm3py.cli.rawcli.session`)."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
# ANSI palette aligned with pm3py/trace/format.py (same as rawcli's breadcrumb)
|
||||||
|
_DIM = "\033[2m"
|
||||||
|
_BGREEN = "\033[1;32m"
|
||||||
|
_BRED = "\033[1;31m"
|
||||||
|
_BOLD = "\033[1m"
|
||||||
|
_RESET = "\033[0m"
|
||||||
|
|
||||||
|
|
||||||
|
class ClientSession:
|
||||||
|
"""Mutable clientcli state. ``driver`` runs the stock client, ``tree`` is the loaded command
|
||||||
|
trie, and ``help_open`` toggles the bottom-toolbar help panel."""
|
||||||
|
|
||||||
|
def __init__(self, driver: Any, tree: Any):
|
||||||
|
self.driver = driver
|
||||||
|
self.tree = tree
|
||||||
|
self.help_open: bool = False
|
||||||
|
|
||||||
|
@property
|
||||||
|
def prompt_label(self) -> str:
|
||||||
|
"""The client's current context label (``usb`` / ``fpc`` / ``offline`` / ``usb|tcp`` …)."""
|
||||||
|
return getattr(self.driver, "prompt", "offline") or "offline"
|
||||||
|
|
||||||
|
def breadcrumb(self) -> str:
|
||||||
|
return f"[client / {self.prompt_label}] pm3py -->"
|
||||||
|
|
||||||
|
def colored_breadcrumb(self) -> str:
|
||||||
|
"""The breadcrumb with ANSI colour — bold ``client``, green when a device is attached, red
|
||||||
|
when offline. Stripping the ANSI yields :meth:`breadcrumb`."""
|
||||||
|
label = self.prompt_label
|
||||||
|
color = _BRED if "offline" in label else _BGREEN
|
||||||
|
sep = f"{_DIM} / {_RESET}"
|
||||||
|
segs = [f"{_BOLD}client{_RESET}", f"{color}{label}{_RESET}"]
|
||||||
|
return (f"{_DIM}[{_RESET}" + sep.join(segs) + f"{_DIM}]{_RESET} "
|
||||||
|
f"{_BOLD}pm3py -->{_RESET}")
|
||||||
158
pm3py/cli/clientcli/tree.py
Normal file
158
pm3py/cli/clientcli/tree.py
Normal file
@@ -0,0 +1,158 @@
|
|||||||
|
"""Command-tree model for ``pm3py client``.
|
||||||
|
|
||||||
|
Loads the stock Proxmark3 client's machine-readable help dump (``doc/commands.json``) and builds a
|
||||||
|
prefix trie over the full command paths (``"hf mf rdbl"``). The JSON keys are **leaf** paths only —
|
||||||
|
interior nodes (``"hf"``, ``"hf mf"``) are synthesised from the path segments (no leaf path is a
|
||||||
|
prefix of another, so the trie is unambiguous). Option strings are pre-rendered by the client
|
||||||
|
(``"-k, --key <hex> key, 6 hex bytes"``); :func:`parse_option` splits them into
|
||||||
|
``(flags, metavar, description)`` for option-flag completion.
|
||||||
|
|
||||||
|
Regenerate the JSON for another client build with::
|
||||||
|
|
||||||
|
proxmark3 --fulltext | client/pyscripts/pm3_help2json.py - -
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
#: env override for the commands.json path (highest priority in :func:`load_command_tree`).
|
||||||
|
_ENV_VAR = "PM3_COMMANDS_JSON"
|
||||||
|
|
||||||
|
#: a single option flag, e.g. ``-a``, ``-1``, ``--key`` (numeric short flags like ``-1`` exist).
|
||||||
|
_FLAG = re.compile(r"^--?[A-Za-z0-9][\w-]*$")
|
||||||
|
|
||||||
|
|
||||||
|
def find_repo_root(start=None):
|
||||||
|
"""Locate a Proxmark3 client checkout (stock Iceman/RRG or a fork) by its ``doc/commands.json``
|
||||||
|
marker, walking up from ``$PM3_REPO``/``$PM3_ROOT``, ``start``, the current directory, and this
|
||||||
|
file's location. Returns a :class:`~pathlib.Path` to the repo root, or ``None``. This is what
|
||||||
|
lets a dropped-in single file find whatever checkout it lives beside."""
|
||||||
|
seeds = []
|
||||||
|
for env in ("PM3_REPO", "PM3_ROOT"):
|
||||||
|
if os.environ.get(env):
|
||||||
|
seeds.append(Path(os.environ[env]))
|
||||||
|
if start:
|
||||||
|
seeds.append(Path(start))
|
||||||
|
seeds.append(Path.cwd())
|
||||||
|
try:
|
||||||
|
seeds.append(Path(__file__).resolve().parent)
|
||||||
|
except NameError: # __file__ absent (e.g. frozen/REPL)
|
||||||
|
pass
|
||||||
|
seen = set()
|
||||||
|
for seed in seeds:
|
||||||
|
base = seed.resolve() if seed.exists() else seed
|
||||||
|
for cand in (base, *base.parents):
|
||||||
|
if cand in seen:
|
||||||
|
continue
|
||||||
|
seen.add(cand)
|
||||||
|
if (cand / "doc" / "commands.json").is_file():
|
||||||
|
return cand # dropped inside the checkout
|
||||||
|
sub = cand / "proxmark3" # a side-by-side checkout (dev layout)
|
||||||
|
if (sub / "doc" / "commands.json").is_file():
|
||||||
|
return sub
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class CommandNode:
|
||||||
|
"""A node in the command trie. ``entry`` is the ``commands.json`` leaf dict for a command, or
|
||||||
|
``None`` for a synthesised interior (category) node."""
|
||||||
|
|
||||||
|
token: str # last path segment; "" for the root
|
||||||
|
children: dict[str, "CommandNode"] = field(default_factory=dict)
|
||||||
|
entry: dict | None = None
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_leaf(self) -> bool:
|
||||||
|
return self.entry is not None
|
||||||
|
|
||||||
|
|
||||||
|
class CommandTree:
|
||||||
|
def __init__(self, commands: dict):
|
||||||
|
self.root = CommandNode("")
|
||||||
|
for path, entry in commands.items():
|
||||||
|
node = self.root
|
||||||
|
for seg in path.split():
|
||||||
|
node = node.children.setdefault(seg, CommandNode(seg))
|
||||||
|
node.entry = entry
|
||||||
|
|
||||||
|
def node_at(self, tokens) -> CommandNode | None:
|
||||||
|
"""The node reached by walking ``tokens`` from the root, or ``None`` if any is unknown."""
|
||||||
|
node = self.root
|
||||||
|
for tok in tokens:
|
||||||
|
node = node.children.get(tok)
|
||||||
|
if node is None:
|
||||||
|
return None
|
||||||
|
return node
|
||||||
|
|
||||||
|
def children_of(self, tokens) -> list[CommandNode]:
|
||||||
|
"""Completion candidates below ``tokens`` (alphabetical), or ``[]`` for an unknown path."""
|
||||||
|
node = self.node_at(tokens)
|
||||||
|
if node is None:
|
||||||
|
return []
|
||||||
|
return [node.children[k] for k in sorted(node.children)]
|
||||||
|
|
||||||
|
def entry_of(self, tokens) -> dict | None:
|
||||||
|
"""The leaf metadata at ``tokens`` (``None`` for an interior or unknown path)."""
|
||||||
|
node = self.node_at(tokens)
|
||||||
|
return node.entry if node else None
|
||||||
|
|
||||||
|
|
||||||
|
def parse_option(line: str):
|
||||||
|
"""Split a rendered option string into ``(flags, metavar, description)``.
|
||||||
|
|
||||||
|
``"-k, --key <hex> key, 6 hex bytes"`` -> ``(["-k", "--key"], "<hex>", "key, 6 hex bytes")``
|
||||||
|
``"--blk <dec> block number"`` -> ``(["--blk"], "<dec>", "block number")``
|
||||||
|
``"-a input key type is key A (def)"`` -> ``(["-a"], "", "input key type is key A (def)")``
|
||||||
|
|
||||||
|
The leading run of flag tokens is delimited by trailing commas (``-k,`` promises another flag);
|
||||||
|
a following ``<...>`` token is the metavar; the rest is the description.
|
||||||
|
"""
|
||||||
|
toks = line.split()
|
||||||
|
flags: list[str] = []
|
||||||
|
i, more = 0, True
|
||||||
|
while i < len(toks) and more:
|
||||||
|
raw = toks[i]
|
||||||
|
core = raw[:-1] if raw.endswith(",") else raw
|
||||||
|
if not _FLAG.match(core):
|
||||||
|
break
|
||||||
|
flags.append(core)
|
||||||
|
more = raw.endswith(",") # a trailing comma promises another flag
|
||||||
|
i += 1
|
||||||
|
meta = ""
|
||||||
|
if i < len(toks) and toks[i].startswith("<"):
|
||||||
|
meta = toks[i]
|
||||||
|
i += 1
|
||||||
|
return flags, meta, " ".join(toks[i:])
|
||||||
|
|
||||||
|
|
||||||
|
def load_command_tree(path=None):
|
||||||
|
"""Return ``(CommandTree, metadata)``. Resolution order: an explicit ``path`` (``--commands``),
|
||||||
|
then ``$PM3_COMMANDS_JSON``, then the discovered repo's ``doc/commands.json`` (see
|
||||||
|
:func:`find_repo_root`), then ``./doc/commands.json``. Raises :class:`FileNotFoundError` (with
|
||||||
|
the regen hint) when nothing is found."""
|
||||||
|
candidates = []
|
||||||
|
if path:
|
||||||
|
candidates.append(Path(path))
|
||||||
|
elif os.environ.get(_ENV_VAR):
|
||||||
|
candidates.append(Path(os.environ[_ENV_VAR]))
|
||||||
|
else:
|
||||||
|
root = find_repo_root()
|
||||||
|
if root is not None:
|
||||||
|
candidates.append(root / "doc" / "commands.json")
|
||||||
|
candidates.append(Path("doc/commands.json"))
|
||||||
|
for cand in candidates:
|
||||||
|
p = cand.expanduser()
|
||||||
|
if p.is_file():
|
||||||
|
data = json.loads(p.read_text())
|
||||||
|
return CommandTree(data["commands"]), data.get("metadata", {})
|
||||||
|
raise FileNotFoundError(
|
||||||
|
"no Proxmark3 commands.json found (looked in: "
|
||||||
|
+ ", ".join(str(c) for c in candidates) + ").\n"
|
||||||
|
" run this from inside a Proxmark3 checkout, pass --commands PATH, set $PM3_REPO, or\n"
|
||||||
|
" regenerate it: proxmark3 --fulltext | client/pyscripts/pm3_help2json.py - -"
|
||||||
|
)
|
||||||
@@ -1,7 +1,8 @@
|
|||||||
"""`pm3py` top-level CLI dispatcher.
|
"""`pm3py` top-level CLI dispatcher.
|
||||||
|
|
||||||
Subcommands live in submodules; ``rawcli`` is the first (the interactive raw-command TUI). Room
|
Subcommands live in submodules: ``rawcli`` (the interactive raw-command TUI over pm3py's own
|
||||||
is left for ``flash`` (wrapping pm3flash) and a future ``client`` command CLI.
|
protocol stack) and ``client`` (an autocompleting REPL wrapping the stock ``proxmark3`` binary).
|
||||||
|
Room is left for ``flash`` (wrapping pm3flash).
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
@@ -14,6 +15,16 @@ def build_parser() -> argparse.ArgumentParser:
|
|||||||
sub = parser.add_subparsers(dest="command")
|
sub = parser.add_subparsers(dest="command")
|
||||||
p = sub.add_parser("rawcli", help="interactive raw-command TUI for a connected Proxmark3")
|
p = sub.add_parser("rawcli", help="interactive raw-command TUI for a connected Proxmark3")
|
||||||
p.add_argument("--port", default=None, help="serial port (default: auto-detect)")
|
p.add_argument("--port", default=None, help="serial port (default: auto-detect)")
|
||||||
|
|
||||||
|
c = sub.add_parser("client", help="autocompleting REPL wrapping the stock proxmark3 client")
|
||||||
|
c.add_argument("--port", default=None, help="serial port passed to proxmark3 (default: auto)")
|
||||||
|
c.add_argument("--commands", default=None, help="path to commands.json (default: auto-locate)")
|
||||||
|
c.add_argument("--client", default="proxmark3", help="proxmark3 binary to wrap (default: PATH)")
|
||||||
|
c.add_argument("--driver", choices=["auto", "swig", "pexpect"], default="auto",
|
||||||
|
help="backend: auto (SWIG if built, else pexpect), swig, or pexpect")
|
||||||
|
c.add_argument("--pack", metavar="OUT", default=None,
|
||||||
|
help="spin off a standalone drop-in instead of launching: OUT.py (single file) "
|
||||||
|
"or OUT.pyz (deps bundled). Distributable to stock Proxmark3 users.")
|
||||||
return parser
|
return parser
|
||||||
|
|
||||||
|
|
||||||
@@ -32,5 +43,30 @@ def main(argv: list[str] | None = None) -> int:
|
|||||||
return 2
|
return 2
|
||||||
from pm3py.cli.rawcli.app import run
|
from pm3py.cli.rawcli.app import run
|
||||||
return run(port=args.port)
|
return run(port=args.port)
|
||||||
|
if args.command == "client":
|
||||||
|
if args.pack:
|
||||||
|
from pm3py.cli.clientcli.pack import pack
|
||||||
|
out = pack(args.pack)
|
||||||
|
print(f"wrote standalone drop-in: {out}")
|
||||||
|
return 0
|
||||||
|
# prompt_toolkit is always needed; pexpect only for the pexpect backend (SWIG doesn't use it)
|
||||||
|
needs = ["prompt_toolkit"] + (["pexpect"] if args.driver != "swig" else [])
|
||||||
|
missing = [m for m in needs if not _importable(m)]
|
||||||
|
if missing:
|
||||||
|
print(
|
||||||
|
f"pm3py client needs {' and '.join(missing)} but "
|
||||||
|
f"{'they are' if len(missing) > 1 else 'it is'} not importable.\n"
|
||||||
|
f" reinstall pm3py, or: pip install {' '.join(missing)}",
|
||||||
|
file=sys.stderr,
|
||||||
|
)
|
||||||
|
return 2
|
||||||
|
from pm3py.cli.clientcli.app import run
|
||||||
|
return run(port=args.port, commands=args.commands, client=args.client, driver=args.driver)
|
||||||
parser.print_help()
|
parser.print_help()
|
||||||
return 1
|
return 1
|
||||||
|
|
||||||
|
|
||||||
|
def _importable(name: str) -> bool:
|
||||||
|
import importlib.util
|
||||||
|
|
||||||
|
return importlib.util.find_spec(name) is not None
|
||||||
|
|||||||
@@ -39,6 +39,7 @@ dependencies = [
|
|||||||
"bitarray>=2.0",
|
"bitarray>=2.0",
|
||||||
"pycryptodome>=3.10",
|
"pycryptodome>=3.10",
|
||||||
"prompt_toolkit>=3.0",
|
"prompt_toolkit>=3.0",
|
||||||
|
"pexpect>=4.8",
|
||||||
]
|
]
|
||||||
|
|
||||||
[project.optional-dependencies]
|
[project.optional-dependencies]
|
||||||
|
|||||||
55
tests/fixtures/commands_min.json
vendored
Normal file
55
tests/fixtures/commands_min.json
vendored
Normal file
@@ -0,0 +1,55 @@
|
|||||||
|
{
|
||||||
|
"metadata": {
|
||||||
|
"commands_extracted": 5,
|
||||||
|
"extracted_by": "test-fixture",
|
||||||
|
"extracted_on": "2026-01-01T00:00:00+00:00"
|
||||||
|
},
|
||||||
|
"commands": {
|
||||||
|
"clear": {
|
||||||
|
"command": "clear",
|
||||||
|
"description": "Clear screen",
|
||||||
|
"notes": [],
|
||||||
|
"offline": true,
|
||||||
|
"options": ["-h, --help This help"],
|
||||||
|
"usage": "clear [-h]"
|
||||||
|
},
|
||||||
|
"hf 14a info": {
|
||||||
|
"command": "hf 14a info",
|
||||||
|
"description": "Get info about an ISO14443-A tag",
|
||||||
|
"notes": ["hf 14a info -v"],
|
||||||
|
"offline": false,
|
||||||
|
"options": ["-h, --help This help", "-v, --verbose verbose output"],
|
||||||
|
"usage": "hf 14a info [-hv]"
|
||||||
|
},
|
||||||
|
"hf mf rdbl": {
|
||||||
|
"command": "hf mf rdbl",
|
||||||
|
"description": "Read MIFARE Classic block",
|
||||||
|
"notes": ["hf mf rdbl --blk 0", "hf mf rdbl --blk 0 -k A0A1A2A3A4A5"],
|
||||||
|
"offline": false,
|
||||||
|
"options": [
|
||||||
|
"-h, --help This help",
|
||||||
|
"--blk <dec> block number",
|
||||||
|
"-a input key type is key A (def)",
|
||||||
|
"-k, --key <hex> key, 6 hex bytes",
|
||||||
|
"-v, --verbose verbose output"
|
||||||
|
],
|
||||||
|
"usage": "hf mf rdbl [-habv] --blk <dec> [-k <hex>]"
|
||||||
|
},
|
||||||
|
"hf mf wrbl": {
|
||||||
|
"command": "hf mf wrbl",
|
||||||
|
"description": "Write MIFARE Classic block",
|
||||||
|
"notes": [],
|
||||||
|
"offline": false,
|
||||||
|
"options": ["-h, --help This help", "--blk <dec> block number", "-k, --key <hex> key, 6 hex bytes"],
|
||||||
|
"usage": "hf mf wrbl [-h] --blk <dec> [-k <hex>]"
|
||||||
|
},
|
||||||
|
"lf search": {
|
||||||
|
"command": "lf search",
|
||||||
|
"description": "Read and identify a LF tag",
|
||||||
|
"notes": ["lf search -1"],
|
||||||
|
"offline": false,
|
||||||
|
"options": ["-h, --help This help", "-1 Use data from Graphbuffer (offline mode)", "-c Continue searching"],
|
||||||
|
"usage": "lf search [-1ch]"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
549
tests/test_clientcli.py
Normal file
549
tests/test_clientcli.py
Normal file
@@ -0,0 +1,549 @@
|
|||||||
|
"""clientcli — command tree, completer, hints, help-toggle, driver helpers, dispatch. Hardware-free
|
||||||
|
(the pexpect path is not driven here; a FakeDriver stands in for the client)."""
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
from pathlib import Path
|
||||||
|
from types import SimpleNamespace
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from prompt_toolkit.document import Document
|
||||||
|
from prompt_toolkit.key_binding import KeyBindings
|
||||||
|
|
||||||
|
from pm3py.cli.main import build_parser, main
|
||||||
|
from pm3py.cli.clientcli.tree import CommandTree, load_command_tree, parse_option
|
||||||
|
from pm3py.cli.clientcli.session import ClientSession
|
||||||
|
from pm3py.cli.clientcli.completer import ClientCompleter
|
||||||
|
from pm3py.cli.clientcli.hints import one_line_hint, help_panel, panel_for, resolve_target
|
||||||
|
from pm3py.cli.clientcli.keys import open_help, close_help, install_key_bindings, apply_tab
|
||||||
|
from pm3py.cli.clientcli.driver import (
|
||||||
|
strip_ansi, _extract_output, PROMPT_RE, resolve_client,
|
||||||
|
make_driver, SwigDriver, PexpectDriver, DriverError,
|
||||||
|
)
|
||||||
|
from pm3py.cli.clientcli.app import dispatch
|
||||||
|
|
||||||
|
FIXTURE = Path(__file__).parent / "fixtures" / "commands_min.json"
|
||||||
|
|
||||||
|
_ANSI = re.compile(r"\x1b\[[0-9;]*m")
|
||||||
|
|
||||||
|
|
||||||
|
def _plain(s: str) -> str:
|
||||||
|
return _ANSI.sub("", s)
|
||||||
|
|
||||||
|
|
||||||
|
def _tree() -> CommandTree:
|
||||||
|
return CommandTree(json.loads(FIXTURE.read_text())["commands"])
|
||||||
|
|
||||||
|
|
||||||
|
class FakeDriver:
|
||||||
|
"""Records executed lines and returns canned output; stands in for PexpectDriver."""
|
||||||
|
|
||||||
|
def __init__(self, output="", prompt="offline"):
|
||||||
|
self.prompt = prompt
|
||||||
|
self.lines = []
|
||||||
|
self._output = output
|
||||||
|
self.closed = False
|
||||||
|
|
||||||
|
def start(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def execute(self, line):
|
||||||
|
self.lines.append(line)
|
||||||
|
return self._output
|
||||||
|
|
||||||
|
def close(self):
|
||||||
|
self.closed = True
|
||||||
|
|
||||||
|
|
||||||
|
def _session(driver=None, tree=None):
|
||||||
|
return ClientSession(driver or FakeDriver(), tree or _tree())
|
||||||
|
|
||||||
|
|
||||||
|
def _complete(session, text):
|
||||||
|
return list(ClientCompleter(session).get_completions(Document(text), None))
|
||||||
|
|
||||||
|
|
||||||
|
def _texts(comps):
|
||||||
|
return [c.text for c in comps]
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- tree
|
||||||
|
|
||||||
|
class TestTree:
|
||||||
|
def test_top_level_mixes_leaves_and_categories(self):
|
||||||
|
toks = [n.token for n in _tree().children_of([])]
|
||||||
|
assert toks == ["clear", "hf", "lf"]
|
||||||
|
|
||||||
|
def test_interior_children_synthesised(self):
|
||||||
|
assert [n.token for n in _tree().children_of(["hf"])] == ["14a", "mf"]
|
||||||
|
assert [n.token for n in _tree().children_of(["hf", "mf"])] == ["rdbl", "wrbl"]
|
||||||
|
|
||||||
|
def test_entry_of_leaf(self):
|
||||||
|
e = _tree().entry_of(["hf", "mf", "rdbl"])
|
||||||
|
assert e is not None and e["description"] == "Read MIFARE Classic block"
|
||||||
|
|
||||||
|
def test_interior_has_no_entry(self):
|
||||||
|
node = _tree().node_at(["hf", "mf"])
|
||||||
|
assert node is not None and node.entry is None and not node.is_leaf
|
||||||
|
|
||||||
|
def test_leaf_flags_as_leaf(self):
|
||||||
|
node = _tree().node_at(["clear"])
|
||||||
|
assert node.is_leaf and node.children == {}
|
||||||
|
|
||||||
|
def test_unknown_path(self):
|
||||||
|
assert _tree().node_at(["hf", "zz"]) is None
|
||||||
|
assert _tree().children_of(["hf", "zz"]) == []
|
||||||
|
|
||||||
|
|
||||||
|
class TestParseOption:
|
||||||
|
def test_short_and_long_with_meta(self):
|
||||||
|
assert parse_option("-k, --key <hex> key, 6 hex bytes") == (
|
||||||
|
["-k", "--key"], "<hex>", "key, 6 hex bytes")
|
||||||
|
|
||||||
|
def test_long_with_meta(self):
|
||||||
|
assert parse_option("--blk <dec> block number") == (["--blk"], "<dec>", "block number")
|
||||||
|
|
||||||
|
def test_short_no_meta(self):
|
||||||
|
assert parse_option("-a input key type is key A (def)") == (
|
||||||
|
["-a"], "", "input key type is key A (def)")
|
||||||
|
|
||||||
|
def test_numeric_short_flag(self):
|
||||||
|
assert parse_option("-1 Use data from Graphbuffer (offline mode)") == (
|
||||||
|
["-1"], "", "Use data from Graphbuffer (offline mode)")
|
||||||
|
|
||||||
|
def test_help(self):
|
||||||
|
assert parse_option("-h, --help This help") == (["-h", "--help"], "", "This help")
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- completer
|
||||||
|
|
||||||
|
class TestCompleter:
|
||||||
|
def test_top_level(self):
|
||||||
|
assert _texts(_complete(_session(), "")) == ["clear ", "hf ", "lf "]
|
||||||
|
|
||||||
|
def test_partial_top_level(self):
|
||||||
|
assert _texts(_complete(_session(), "h")) == ["hf "]
|
||||||
|
|
||||||
|
def test_subcommand_partial(self):
|
||||||
|
assert _texts(_complete(_session(), "hf m")) == ["mf "]
|
||||||
|
|
||||||
|
def test_subcommands_after_space(self):
|
||||||
|
assert _texts(_complete(_session(), "hf mf ")) == ["rdbl ", "wrbl "]
|
||||||
|
|
||||||
|
def test_subcommand_completion_has_trailing_space(self):
|
||||||
|
# the trailing space makes accepting a completion commit + descend a level
|
||||||
|
assert all(t.endswith(" ") for t in _texts(_complete(_session(), "hf mf ")))
|
||||||
|
|
||||||
|
def test_accepting_subcommand_descends(self):
|
||||||
|
# "hf " → accept "14a " → "hf 14a " must offer 14a's children, not re-offer 14a
|
||||||
|
assert "14a " in _texts(_complete(_session(), "hf "))
|
||||||
|
assert _texts(_complete(_session(), "hf 14a ")) == ["info "]
|
||||||
|
|
||||||
|
def test_subcommand_display_omits_trailing_space(self):
|
||||||
|
disp = {c.display_text: c.text for c in _complete(_session(), "hf mf ")}
|
||||||
|
assert "rdbl" in disp and disp["rdbl"] == "rdbl "
|
||||||
|
|
||||||
|
def test_subcommand_meta_is_description(self):
|
||||||
|
comps = {c.text: c.display_meta_text for c in _complete(_session(), "hf mf ")}
|
||||||
|
assert comps["rdbl "] == "Read MIFARE Classic block"
|
||||||
|
|
||||||
|
def test_interior_meta_is_count(self):
|
||||||
|
comps = {c.text: c.display_meta_text for c in _complete(_session(), "hf ")}
|
||||||
|
assert comps["mf "] == "2 subcommands"
|
||||||
|
|
||||||
|
def test_option_flags_on_leaf(self):
|
||||||
|
texts = _texts(_complete(_session(), "hf mf rdbl --"))
|
||||||
|
assert "--blk" in texts and "--key" in texts and "--verbose" in texts
|
||||||
|
|
||||||
|
def test_option_meta_is_description(self):
|
||||||
|
comps = {c.text: c.display_meta_text for c in _complete(_session(), "hf mf rdbl --")}
|
||||||
|
assert comps["--blk"] == "block number"
|
||||||
|
|
||||||
|
def test_help_flag_excluded(self):
|
||||||
|
assert "--help" not in _texts(_complete(_session(), "hf mf rdbl --"))
|
||||||
|
|
||||||
|
def test_leaf_bare_lists_all_options(self):
|
||||||
|
texts = _texts(_complete(_session(), "hf mf rdbl "))
|
||||||
|
assert "--blk" in texts and "-a" in texts and "-k" in texts
|
||||||
|
|
||||||
|
def test_argument_value_slot_is_empty(self):
|
||||||
|
assert _complete(_session(), "hf mf rdbl -k ") == []
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- hints
|
||||||
|
|
||||||
|
class TestHints:
|
||||||
|
def test_leaf_one_line(self):
|
||||||
|
h = one_line_hint(_tree(), "hf mf rdbl")
|
||||||
|
assert "Read MIFARE Classic block" in h and "usage" not in h.lower()
|
||||||
|
assert "hf mf rdbl [-habv]" in h
|
||||||
|
|
||||||
|
def test_interior_one_line(self):
|
||||||
|
assert "2 subcommands" in one_line_hint(_tree(), "hf mf")
|
||||||
|
|
||||||
|
def test_partial_word_resolves_prefix(self):
|
||||||
|
assert "2 subcommands" in one_line_hint(_tree(), "hf mf rd")
|
||||||
|
|
||||||
|
def test_unknown_is_none(self):
|
||||||
|
assert one_line_hint(_tree(), "zz") is None
|
||||||
|
assert one_line_hint(_tree(), "") is None
|
||||||
|
|
||||||
|
def test_help_panel_multiline(self):
|
||||||
|
panel = _plain(help_panel(_tree().entry_of(["hf", "mf", "rdbl"])))
|
||||||
|
assert "hf mf rdbl" in panel
|
||||||
|
assert "usage:" in panel
|
||||||
|
assert "--blk" in panel and "--key <hex>" in panel
|
||||||
|
assert "This help" not in panel # -h/--help is skipped
|
||||||
|
assert "hf mf rdbl --blk 0" in panel # a note
|
||||||
|
|
||||||
|
def test_panel_for_leaf(self):
|
||||||
|
node = _tree().node_at(["hf", "mf", "rdbl"])
|
||||||
|
assert panel_for(node) == help_panel(node.entry)
|
||||||
|
|
||||||
|
def test_panel_for_interior(self):
|
||||||
|
panel = _plain(panel_for(_tree().node_at(["hf", "mf"])))
|
||||||
|
assert "2 subcommands" in panel and "rdbl" in panel and "wrbl" in panel
|
||||||
|
|
||||||
|
def test_resolve_target_subcommand(self):
|
||||||
|
cur = SimpleNamespace(text="rdbl")
|
||||||
|
node = resolve_target(_tree(), "hf mf ", cur)
|
||||||
|
assert node is not None and node.token == "rdbl" and node.is_leaf
|
||||||
|
|
||||||
|
def test_resolve_target_flag_points_at_leaf(self):
|
||||||
|
cur = SimpleNamespace(text="--blk")
|
||||||
|
node = resolve_target(_tree(), "hf mf rdbl --", cur)
|
||||||
|
assert node is not None and node.token == "rdbl"
|
||||||
|
|
||||||
|
def test_resolve_target_interior(self):
|
||||||
|
cur = SimpleNamespace(text="mf")
|
||||||
|
node = resolve_target(_tree(), "hf ", cur)
|
||||||
|
assert node is not None and node.token == "mf" and not node.is_leaf
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- keys
|
||||||
|
|
||||||
|
class TestKeys:
|
||||||
|
def test_open_help_requires_selection(self):
|
||||||
|
s = _session()
|
||||||
|
assert open_help(s, has_selection=False) is False
|
||||||
|
assert s.help_open is False
|
||||||
|
assert open_help(s, has_selection=True) is True
|
||||||
|
assert s.help_open is True
|
||||||
|
|
||||||
|
def test_close_help(self):
|
||||||
|
s = _session()
|
||||||
|
s.help_open = True
|
||||||
|
assert close_help(s) is False
|
||||||
|
assert s.help_open is False
|
||||||
|
|
||||||
|
def test_install_does_not_raise(self):
|
||||||
|
install_key_bindings(KeyBindings(), _session())
|
||||||
|
|
||||||
|
def _buffer_with_menu(self, text):
|
||||||
|
from prompt_toolkit.buffer import Buffer, CompletionState
|
||||||
|
from prompt_toolkit.completion import CompleteEvent
|
||||||
|
from prompt_toolkit.document import Document
|
||||||
|
buff = Buffer()
|
||||||
|
buff.set_document(Document(text, len(text)))
|
||||||
|
comps = list(ClientCompleter(_session()).get_completions(Document(text, len(text)),
|
||||||
|
CompleteEvent()))
|
||||||
|
buff.complete_state = CompletionState(original_document=buff.document,
|
||||||
|
completions=comps, complete_index=None)
|
||||||
|
return buff
|
||||||
|
|
||||||
|
def test_tab_commits_first_completion(self):
|
||||||
|
# with nothing highlighted, Tab commits the FIRST completion (+ its trailing space)
|
||||||
|
buff = self._buffer_with_menu("hf ")
|
||||||
|
apply_tab(buff)
|
||||||
|
assert buff.text == "hf 14a " # descends: next Tab would show 14a's kids
|
||||||
|
|
||||||
|
def test_tab_commits_highlighted_completion(self):
|
||||||
|
buff = self._buffer_with_menu("hf ")
|
||||||
|
buff.complete_state.complete_index = 1 # highlight the 2nd (mf)
|
||||||
|
apply_tab(buff)
|
||||||
|
assert buff.text == "hf mf "
|
||||||
|
|
||||||
|
def test_tab_completes_partial(self):
|
||||||
|
buff = self._buffer_with_menu("hf m")
|
||||||
|
apply_tab(buff)
|
||||||
|
assert buff.text == "hf mf "
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- driver helpers
|
||||||
|
|
||||||
|
class TestDriverHelpers:
|
||||||
|
def test_strip_ansi(self):
|
||||||
|
assert strip_ansi("\x1b[1;32musb\x1b[0m") == "usb"
|
||||||
|
|
||||||
|
def test_prompt_re_variants(self):
|
||||||
|
cases = {
|
||||||
|
"[usb] pm3 --> ": "usb",
|
||||||
|
"[offline] pm3 --> ": "offline",
|
||||||
|
"[usb|tcp] pm3 --> ": "usb|tcp",
|
||||||
|
"[fpc|script] pm3 --> ": "fpc|script",
|
||||||
|
}
|
||||||
|
for text, label in cases.items():
|
||||||
|
m = PROMPT_RE.search(text)
|
||||||
|
assert m is not None and m.group(1) == label
|
||||||
|
|
||||||
|
def test_prompt_re_captures_through_ansi(self):
|
||||||
|
colored = "[\x1b[1;32musb\x1b[0m] pm3 --> "
|
||||||
|
m = PROMPT_RE.search(colored)
|
||||||
|
assert m is not None and strip_ansi(m.group(1)) == "usb"
|
||||||
|
|
||||||
|
def test_extract_output_drops_echo(self):
|
||||||
|
raw = "hf 14a info\r\n[+] UID: 04 11 22 33\r\n"
|
||||||
|
assert _extract_output(raw, "hf 14a info") == "[+] UID: 04 11 22 33"
|
||||||
|
|
||||||
|
def test_extract_output_preserves_color(self):
|
||||||
|
raw = "hw version\r\n\x1b[32m[+] ok\x1b[0m\r\n"
|
||||||
|
assert _extract_output(raw, "hw version") == "\x1b[32m[+] ok\x1b[0m"
|
||||||
|
|
||||||
|
def test_extract_output_no_echo(self):
|
||||||
|
assert _extract_output("just output\r\n", "cmd") == "just output"
|
||||||
|
|
||||||
|
def test_extract_output_strips_bracketed_paste(self):
|
||||||
|
raw = "hw status\x1b[?2004l\r\n\x1b[?2004h[+] ok\r\n"
|
||||||
|
assert _extract_output(raw, "hw status") == "[+] ok"
|
||||||
|
|
||||||
|
def test_resolve_client_explicit_path_verbatim(self):
|
||||||
|
assert resolve_client("/opt/pm3/proxmark3") == "/opt/pm3/proxmark3"
|
||||||
|
assert resolve_client("./proxmark3") == "./proxmark3"
|
||||||
|
|
||||||
|
def test_resolve_client_named_binary_not_substituted(self):
|
||||||
|
# an explicitly-named client that isn't found is returned unchanged (never swapped for a
|
||||||
|
# fallback), so the spawn fails clearly instead of running a different binary
|
||||||
|
assert resolve_client("definitely-not-a-real-pm3-binary") == "definitely-not-a-real-pm3-binary"
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- dispatch
|
||||||
|
|
||||||
|
class _FakePm3:
|
||||||
|
def __init__(self, port):
|
||||||
|
self.name = port
|
||||||
|
self.calls = []
|
||||||
|
|
||||||
|
def console(self, cmd, capture=True, quiet=True):
|
||||||
|
self.calls.append((cmd, capture, quiet))
|
||||||
|
return 0
|
||||||
|
|
||||||
|
@property
|
||||||
|
def grabbed_output(self):
|
||||||
|
return ""
|
||||||
|
|
||||||
|
|
||||||
|
class _FakePm3Module:
|
||||||
|
pm3 = _FakePm3
|
||||||
|
|
||||||
|
|
||||||
|
class TestSwigDriver:
|
||||||
|
def test_execute_prints_via_client_not_grab(self):
|
||||||
|
# the client prints its own coloured output (capture=False, quiet=False); execute returns ""
|
||||||
|
d = SwigDriver(port="/dev/ttyACM0", pm3_module=_FakePm3Module())
|
||||||
|
d.start()
|
||||||
|
assert d.backend == "swig" and d.prompt == "usb"
|
||||||
|
assert d.execute("hw version") == ""
|
||||||
|
assert d._p.calls == [("hw version", False, False)]
|
||||||
|
|
||||||
|
def test_execute_before_start(self):
|
||||||
|
d = SwigDriver(port="/dev/ttyACM0", pm3_module=_FakePm3Module())
|
||||||
|
with pytest.raises(DriverError):
|
||||||
|
d.execute("hw status")
|
||||||
|
|
||||||
|
def test_no_device_errors_without_calling_pm3(self, monkeypatch):
|
||||||
|
import pm3py.cli.clientcli.driver as drv
|
||||||
|
monkeypatch.setattr(drv, "_autodetect_port", lambda: None)
|
||||||
|
d = SwigDriver(port=None, pm3_module=_FakePm3Module())
|
||||||
|
with pytest.raises(DriverError, match="no serial device"):
|
||||||
|
d.start()
|
||||||
|
assert d._p is None # never constructed → no exit() risk
|
||||||
|
|
||||||
|
|
||||||
|
class TestPexpectPort:
|
||||||
|
def test_autodetects_port_when_none_given(self, monkeypatch):
|
||||||
|
# the stock binary needs a port arg (no auto-detect); the driver must resolve one itself
|
||||||
|
import pexpect
|
||||||
|
import pm3py.cli.clientcli.driver as drv
|
||||||
|
|
||||||
|
class _FakeChild:
|
||||||
|
before = "[usb] "
|
||||||
|
def expect(self, pat):
|
||||||
|
return 0
|
||||||
|
|
||||||
|
captured = {}
|
||||||
|
def fake_spawn(command, args, **kw):
|
||||||
|
captured["command"], captured["args"] = command, args
|
||||||
|
return _FakeChild()
|
||||||
|
|
||||||
|
monkeypatch.setattr(pexpect, "spawn", fake_spawn)
|
||||||
|
monkeypatch.setattr(drv, "_autodetect_port", lambda: "/dev/ttyACM9")
|
||||||
|
d = drv.PexpectDriver(port=None, client="proxmark3")
|
||||||
|
d.start()
|
||||||
|
assert captured["args"] == ["/dev/ttyACM9"] # resolved, not empty (would be offline)
|
||||||
|
assert d.prompt == "usb" # label parsed from the prompt tail
|
||||||
|
|
||||||
|
def test_explicit_port_used_verbatim(self, monkeypatch):
|
||||||
|
import pexpect
|
||||||
|
import pm3py.cli.clientcli.driver as drv
|
||||||
|
|
||||||
|
class _FakeChild:
|
||||||
|
before = "[usb] "
|
||||||
|
def expect(self, pat):
|
||||||
|
return 0
|
||||||
|
|
||||||
|
captured = {}
|
||||||
|
def fake_spawn(command, args, **kw):
|
||||||
|
captured["args"] = args
|
||||||
|
return _FakeChild()
|
||||||
|
monkeypatch.setattr(pexpect, "spawn", fake_spawn)
|
||||||
|
monkeypatch.setattr(drv, "_autodetect_port", lambda: "/dev/ttyACM9")
|
||||||
|
drv.PexpectDriver(port="/dev/ttyACM0").start()
|
||||||
|
assert captured["args"] == ["/dev/ttyACM0"] # explicit --port wins over auto-detect
|
||||||
|
|
||||||
|
|
||||||
|
class TestMakeDriver:
|
||||||
|
def test_explicit_pexpect(self):
|
||||||
|
assert isinstance(make_driver(driver="pexpect"), PexpectDriver)
|
||||||
|
|
||||||
|
def test_auto_uses_swig_when_available(self, monkeypatch):
|
||||||
|
import pm3py.cli.clientcli.driver as drv
|
||||||
|
monkeypatch.setattr(drv, "_load_pm3_module", lambda: _FakePm3Module())
|
||||||
|
d = make_driver(port="/dev/ttyACM0", driver="auto")
|
||||||
|
assert isinstance(d, SwigDriver)
|
||||||
|
|
||||||
|
def test_auto_falls_back_to_pexpect(self, monkeypatch):
|
||||||
|
import pm3py.cli.clientcli.driver as drv
|
||||||
|
def boom():
|
||||||
|
raise ImportError("no _pm3")
|
||||||
|
monkeypatch.setattr(drv, "_load_pm3_module", boom)
|
||||||
|
assert isinstance(make_driver(driver="auto"), PexpectDriver)
|
||||||
|
|
||||||
|
def test_swig_explicit_unavailable_errors(self, monkeypatch):
|
||||||
|
import pm3py.cli.clientcli.driver as drv
|
||||||
|
def boom():
|
||||||
|
raise ImportError("no _pm3")
|
||||||
|
monkeypatch.setattr(drv, "_load_pm3_module", boom)
|
||||||
|
with pytest.raises(DriverError, match="_pm3"):
|
||||||
|
make_driver(driver="swig")
|
||||||
|
|
||||||
|
|
||||||
|
class TestDispatch:
|
||||||
|
def test_runs_command(self):
|
||||||
|
drv = FakeDriver(output="[+] done")
|
||||||
|
out = []
|
||||||
|
assert dispatch(_session(drv), "hf 14a info", out.append) is True
|
||||||
|
assert drv.lines == ["hf 14a info"]
|
||||||
|
assert out == ["[+] done"]
|
||||||
|
|
||||||
|
def test_quit_exits(self):
|
||||||
|
drv = FakeDriver()
|
||||||
|
assert dispatch(_session(drv), "quit", lambda *_: None) is False
|
||||||
|
assert drv.lines == []
|
||||||
|
|
||||||
|
def test_empty_output_not_printed(self):
|
||||||
|
out = []
|
||||||
|
dispatch(_session(FakeDriver(output="")), "clear", out.append)
|
||||||
|
assert out == []
|
||||||
|
|
||||||
|
def test_driver_error_ends_loop(self):
|
||||||
|
from pm3py.cli.clientcli.driver import DriverError
|
||||||
|
|
||||||
|
class Boom(FakeDriver):
|
||||||
|
def execute(self, line):
|
||||||
|
raise DriverError("client exited")
|
||||||
|
|
||||||
|
out = []
|
||||||
|
assert dispatch(_session(Boom()), "hw status", out.append) is False
|
||||||
|
assert any("client exited" in m for m in out)
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- session + argparse + loader
|
||||||
|
|
||||||
|
class TestSession:
|
||||||
|
def test_breadcrumb_offline(self):
|
||||||
|
s = _session(FakeDriver(prompt="offline"))
|
||||||
|
assert s.breadcrumb() == "[client / offline] pm3py -->"
|
||||||
|
assert _plain(s.colored_breadcrumb()) == s.breadcrumb()
|
||||||
|
|
||||||
|
def test_breadcrumb_usb(self):
|
||||||
|
assert _session(FakeDriver(prompt="usb")).breadcrumb() == "[client / usb] pm3py -->"
|
||||||
|
|
||||||
|
|
||||||
|
class TestArgparse:
|
||||||
|
def test_client_subcommand(self):
|
||||||
|
args = build_parser().parse_args(["client", "--port", "X", "--client", "pm3"])
|
||||||
|
assert args.command == "client" and args.port == "X" and args.client == "pm3"
|
||||||
|
assert args.driver == "auto" # SWIG-first default
|
||||||
|
|
||||||
|
def test_client_driver_choice(self):
|
||||||
|
args = build_parser().parse_args(["client", "--driver", "pexpect"])
|
||||||
|
assert args.driver == "pexpect"
|
||||||
|
|
||||||
|
def test_no_subcommand_prints_help(self):
|
||||||
|
assert main([]) == 1
|
||||||
|
|
||||||
|
|
||||||
|
class TestLoader:
|
||||||
|
def test_loads_fixture(self):
|
||||||
|
tree, meta = load_command_tree(str(FIXTURE))
|
||||||
|
assert meta["commands_extracted"] == 5
|
||||||
|
assert tree.entry_of(["hf", "mf", "rdbl"])["command"] == "hf mf rdbl"
|
||||||
|
|
||||||
|
def test_missing_raises(self):
|
||||||
|
with pytest.raises(FileNotFoundError):
|
||||||
|
load_command_tree("/nonexistent/commands.json")
|
||||||
|
|
||||||
|
|
||||||
|
class TestRepoDiscovery:
|
||||||
|
def _make_repo(self, base):
|
||||||
|
repo = base / "proxmark3"
|
||||||
|
(repo / "doc").mkdir(parents=True)
|
||||||
|
(repo / "doc" / "commands.json").write_text('{"commands": {}}')
|
||||||
|
return repo
|
||||||
|
|
||||||
|
def test_via_env(self, tmp_path, monkeypatch):
|
||||||
|
from pm3py.cli.clientcli.tree import find_repo_root
|
||||||
|
repo = self._make_repo(tmp_path)
|
||||||
|
monkeypatch.setenv("PM3_REPO", str(repo))
|
||||||
|
assert find_repo_root() == repo.resolve()
|
||||||
|
|
||||||
|
def test_walks_up_from_subdir(self, tmp_path, monkeypatch):
|
||||||
|
from pm3py.cli.clientcli.tree import find_repo_root
|
||||||
|
monkeypatch.delenv("PM3_REPO", raising=False)
|
||||||
|
monkeypatch.delenv("PM3_ROOT", raising=False)
|
||||||
|
repo = self._make_repo(tmp_path)
|
||||||
|
deep = repo / "client" / "src"
|
||||||
|
deep.mkdir(parents=True)
|
||||||
|
assert find_repo_root(start=deep) == repo.resolve()
|
||||||
|
|
||||||
|
def test_finds_sibling_checkout(self, tmp_path, monkeypatch):
|
||||||
|
# dev layout: run from a sibling dir, discover ../proxmark3 (keeps `pm3py client` working)
|
||||||
|
from pm3py.cli.clientcli.tree import find_repo_root
|
||||||
|
monkeypatch.delenv("PM3_REPO", raising=False)
|
||||||
|
monkeypatch.delenv("PM3_ROOT", raising=False)
|
||||||
|
repo = self._make_repo(tmp_path)
|
||||||
|
sibling = tmp_path / "pm3py"
|
||||||
|
sibling.mkdir()
|
||||||
|
assert find_repo_root(start=sibling) == repo.resolve()
|
||||||
|
|
||||||
|
|
||||||
|
class TestPacker:
|
||||||
|
def test_amalgamate_is_self_contained(self):
|
||||||
|
from pm3py.cli.clientcli.pack import amalgamate
|
||||||
|
code = amalgamate()
|
||||||
|
compile(code, "pm3cli", "exec") # valid Python
|
||||||
|
assert code.count("from __future__") == 1 # exactly one, at the top
|
||||||
|
for ln in code.splitlines():
|
||||||
|
s = ln.lstrip()
|
||||||
|
assert not s.startswith("from ."), ln # no intra-package imports survive
|
||||||
|
assert not s.startswith(("from pm3py", "import pm3py")), ln
|
||||||
|
assert "def main(" in code and 'if __name__ == "__main__"' in code
|
||||||
|
assert "def run(" in code # the REPL entrypoint is present
|
||||||
|
|
||||||
|
def test_pack_writes_executable_py(self, tmp_path):
|
||||||
|
from pm3py.cli.clientcli.pack import pack
|
||||||
|
out = tmp_path / "pm3cli.py"
|
||||||
|
pack(str(out))
|
||||||
|
assert out.is_file() and os.access(out, os.X_OK)
|
||||||
|
compile(out.read_text(), str(out), "exec")
|
||||||
Reference in New Issue
Block a user