Compare commits
20 Commits
3b4ec4541b
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
907522f0ae | ||
|
|
a990d725ef | ||
|
|
e3a394859c | ||
|
|
4226c9a11c | ||
|
|
8833753c45 | ||
|
|
8f136e7dc4 | ||
|
|
218e6dfc2a | ||
|
|
4684d0cc83 | ||
|
|
9e18a10274 | ||
|
|
d0339bceb5 | ||
|
|
72815e34eb | ||
|
|
251e276fbb | ||
|
|
7b028f4ddc | ||
|
|
a61e9724a6 | ||
|
|
2ba570a7ef | ||
|
|
4b5df12a76 | ||
|
|
6e46916dd9 | ||
|
|
c675f6b278 | ||
|
|
6ab8582308 | ||
|
|
9c54015735 |
93
docs/plans/2026-07-16-c-client-oracle-and-lf-demod-parity.md
Normal file
93
docs/plans/2026-07-16-c-client-oracle-and-lf-demod-parity.md
Normal file
@@ -0,0 +1,93 @@
|
||||
# C client oracle + LF demod parity — future-work notes
|
||||
|
||||
*2026-07-16. Written while finishing the rawcli T5577 catalog and validating on real
|
||||
hardware (a T5577 emulating **EM4100 `00FFFFFFFF`**, config word **`00148040`**).*
|
||||
|
||||
I wanted the stock proxmark3 C client as a ground-truth **oracle** for LF demod — to
|
||||
cross-check pm3py's decode of the live tag. It wouldn't run in this environment, and digging
|
||||
into why surfaced the notes below. **Net: there is no C-client _code_ bug to PR from what we
|
||||
found** — the two real items are (1) an environment blocker and (2) a pm3py-side demod-parity
|
||||
gap. Captured here so a future effort doesn't re-derive it.
|
||||
|
||||
## 1. The C client won't run inside the VS Code snap sandbox (oracle blocker)
|
||||
|
||||
- **Symptom:** `./pm3 -p /dev/ttyACM0 -c 'lf search'` →
|
||||
`symbol lookup error: /snap/core20/current/lib/x86_64-linux-gnu/libpthread.so.0:
|
||||
undefined symbol: __libc_pthread_init, version GLIBC_PRIVATE`
|
||||
- **Root cause:** this shell runs *inside* the VS Code snap (`SNAP_REVISION`, `SNAP_REAL_HOME`,
|
||||
`/snap/code/250/...`). The snap runtime injects core20's older glibc / `libpthread.so.0`
|
||||
ahead of the system libs the client was built against, so a `GLIBC_PRIVATE` symbol the
|
||||
system libc no longer exports gets looked up in the old snap libpthread and fails.
|
||||
- **Confirmed it is NOT a client defect:**
|
||||
- The binary has **no RPATH/RUNPATH**; its `NEEDED` libs (`libpython3.12`, `Qt5*`,
|
||||
`readline`, …) all resolve to system `/lib/x86_64-linux-gnu` under `ldd`.
|
||||
- `LD_LIBRARY_PATH` is **unset**; clearing `LD_LIBRARY_PATH`/`GTK_PATH` did not help — the
|
||||
injection is at the snap-runtime level, not via a variable we can scrub.
|
||||
- **To use the client as an oracle:** run it from a **non-snap login shell / real terminal**
|
||||
(outside VS Code's snap), or build+run it inside a matched (non-snap) environment. No
|
||||
proxmark3 change fixes a sandbox library injection.
|
||||
- **Marginal upstream idea (only if it ever matters):** the `pm3` launcher could sanitize
|
||||
obviously-contaminating library paths before `exec`, but that is a weak, environment-specific
|
||||
band-aid — not worth a PR on its own.
|
||||
- Client checkout here: `dangerous-tac0s/proxmark3` `v4.20728-1264-g273777b21`.
|
||||
|
||||
## 2. LF clock detection — pm3py's port was weaker than the C client (pm3py-side parity)
|
||||
|
||||
We found and fixed a real bug in pm3py `pm3py/lf/dsp.py::detect_clock`: it estimated the bit
|
||||
clock from the **minimum** edge interval, so a handful of stray sub-bit edges on a
|
||||
strongly-coupled, rail-clipped tag collapsed the estimate (true half-bit 32 → ~3), snapping the
|
||||
data clock to 8 and producing an all-phase-error ASK demod (nothing decoded). Fixed to key off
|
||||
the **shortest _well-supported_ interval** (≥20% of the modal count) instead of the raw minimum.
|
||||
|
||||
The C client's `DetectASKClock` (`common/lfdemod.c`) is already robust to this — worth porting
|
||||
its approach for full parity:
|
||||
|
||||
- It first runs `DetectCleanAskWave` and, for clean/strong/**clipped** peaks, routes to a
|
||||
dedicated `DetectStrongAskClock` — exactly the case that broke our port.
|
||||
- Otherwise it **error-scores every candidate clock** `{8,16,32,40,50,64,100,128,272}` against
|
||||
the wave and picks the best fit — not a single-interval heuristic.
|
||||
|
||||
**Action (pm3py, not a C-client PR):** port the candidate-clock error-scoring + a
|
||||
strong/clean-ASK fast path into `pm3py/lf`. Our support-based `detect_clock` fix is a partial,
|
||||
targeted step; the C client's scoring is the fuller solution.
|
||||
|
||||
## 3. Data-block reads — investigated to ground truth (no bug), one real improvement
|
||||
|
||||
Post-fix, pm3py decodes the live tag as:
|
||||
|
||||
- Emitted credential: **EM4100 `00FFFFFFFF`**
|
||||
- T5577 config block 0: **`00148040`** (Manchester, RF/64, MAXBLK 2, `ST=0`, non-inverted →
|
||||
EM4100/EM4102) — decoded rotation-resolved against presets (reliable).
|
||||
- Data blocks: `READ(1)=A108421F`, `READ(2)=007FE108`, which at first looked *wrong* (not a
|
||||
rotation of the encoder's `FF801EF7`/`BDEF7BC0`).
|
||||
|
||||
Chased to ground truth against the **ATA5577C datasheet + proxmark3 C source + an on-hardware
|
||||
write-read-back** — and there is **no read/addressing bug**:
|
||||
|
||||
- Datasheet §5.9: the direct-access command (`opcode 10` + `0` + 3-bit addr — exactly what the
|
||||
firmware sends) reads *only the addressed block*, repetitively. A clean 32-bit repeat in the
|
||||
capture proves block-read mode engaged (the 64-bit emulation stream can't produce one).
|
||||
- `T55xx_SetBits` clocks the address MSB-first; pm3py's `readbl` payload matches the firmware
|
||||
struct byte-for-byte. Addressing is stock-proxmark, verified end-to-end.
|
||||
- A block-3 write-read-back was **not** decisive on its own: `writebl` and `readbl` share the
|
||||
same address path, so they round-trip regardless of whether addressing is correct (it only
|
||||
proved the demod inverts). Proxmark, Flipper, and pm3py all write the header into block 1, so
|
||||
the alternative "pm3py reads 1↔2 swapped" had to be excluded with an **addressing-free** read.
|
||||
- **Decisive test (MAXBLK=1):** temporarily set the config MAXBLK field to 1 (`00148040 →
|
||||
00148020`) so regular-read streams *only physical block 1* (datasheet §5.11.1), read it off
|
||||
the air (no read-command addressing), then restored `00148040`. Physical block 1 came back as
|
||||
`~BDEF7BC0` — the **non-header** half — matching `READ(1)`. So `READ(1)` == physical block 1.
|
||||
|
||||
Conclusion: this tag was simply **cloned with the two EM4100 halves in reverse order**
|
||||
(block 1 = `BDEF7BC0`, block 2 = `FF801EF7`) by a non-proxmark/Flipper/pm3py tool; both orders
|
||||
stream the same cyclic waveform and read as `00FFFFFFFF`. pm3py reads them correctly. The "wrong"
|
||||
appearance was polarity (inversion) + rotation, i.e. the documented "best-effort — bit alignment
|
||||
not verified".
|
||||
|
||||
**Real improvement (open):** `read_t55xx_block` should *resolve* data-block polarity/rotation
|
||||
instead of leaving it best-effort. Polarity is determinable — read block 0 (config), resolve its
|
||||
polarity against the known preset, and carry that polarity to the data blocks (they share the
|
||||
tag's modulation). Rotation can be anchored off the block-read response start. The C client's
|
||||
`cmdlft55xx` demod already does this alignment — worth porting alongside §2. Still not
|
||||
cross-checked against `lf t55 dump` (client blocked by §1), but the write-read-back makes that
|
||||
non-blocking.
|
||||
@@ -11,17 +11,22 @@ dropdown** — never on the bottom bar, never in command output.
|
||||
- **Must work in both hex AND binary entry modes**, rendering in the active base.
|
||||
- Function-call args are base-10 (`READ(4)`); raw bytes are hex, `0b` to intermix binary.
|
||||
|
||||
## BUG — binary-mode autocomplete does not fire in the live prompt
|
||||
## FIXED — raw byte-line autocomplete now fires in hex AND binary
|
||||
|
||||
The completer logic is correct when fed the text directly (hex and binary), but in the running
|
||||
REPL the completion menu doesn't reliably appear while entering **binary**.
|
||||
- Prime suspect: `entry._regroup_after` sets `buf.document = Document(...)` directly on
|
||||
auto-spacing, which resets prompt_toolkit's `complete_state` and cancels the menu. Binary
|
||||
regroups every 8 digits, so it bites almost immediately.
|
||||
- Fix direction: regroup via a completion-preserving path (don't replace `buf.document`
|
||||
wholesale); re-arm `complete_while_typing` after regroup. Also confirm the Ctrl-t / Ctrl-_
|
||||
toggle actually flips the breadcrumb to `0b`.
|
||||
- Verify on hardware in binary mode (needs a tag on the antenna).
|
||||
Root cause: `entry._digit` applied each keystroke as `buf.document = Document(...)`, which resets
|
||||
prompt_toolkit's `complete_state` and never re-fires completion — so the live menu never appeared
|
||||
during raw byte entry. This hit **both** hex and binary (the earlier "hex works, binary doesn't"
|
||||
was inaccurate; only command-name completion worked, because that path already used `insert_text`).
|
||||
Fixed by appending the per-keystroke delta via `buf.insert_text` instead — `type_digit` only ever
|
||||
appends (optionally after an auto-space) or drops, so the change is always a pure suffix, and
|
||||
`insert_text` triggers completion exactly like an ordinary keystroke. Verified by driving the real
|
||||
key handler: the menu now populates on every digit in both modes. Regression:
|
||||
`TestKeyBindings::test_digit_binding_fires_completion_trigger_for_raw_bytes`.
|
||||
|
||||
The Ctrl-t / Ctrl-_ toggle correctly flips the breadcrumb `0x` <-> `0b` and requests a redraw
|
||||
(verified, guarded by `test_toggle_updates_breadcrumb_prefix`). An end-to-end live-session test
|
||||
(`TestRawcliLiveSession`) drives the real PromptSession via a pipe input. A human eyeball of the
|
||||
rendered dropdown on a real terminal is the only thing left unautomated.
|
||||
|
||||
## Priority 1 — MIFARE Classic (do this first)
|
||||
|
||||
@@ -36,16 +41,55 @@ REPL the completion menu doesn't reliably appear while entering **binary**.
|
||||
authenticate step is the reader's job before these.
|
||||
- Autocomplete must work in hex + binary (see the bug above).
|
||||
|
||||
## After MFC — 15693 vicinity chips: ICODE SLIX, ICODE SLIX2, NTAG5
|
||||
## After MFC — 15693 vicinity chips: ICODE SLIX ✅, SLIX2 ✅, NTAG5 ✅, ICODE DNA (next)
|
||||
|
||||
1. **15693 header builder** (shared foundation). Replace the ad-hoc `bytes([0x02, cmd, …])` with a
|
||||
real request-header builder: flags (data rate, addressed-vs-nonaddressed + 8-byte UID, option,
|
||||
protocol-extension) + command + NXP **manufacturer code 0x04** for ICODE/NTAG5 custom commands.
|
||||
2. **Per-chip identify.** Today identify stamps `"ISO15693"` generically; resolve the actual IC
|
||||
(UID mfg byte `E0 04…` + GET_SYSTEM_INFO / IC-reference) so catalog + map key off SLIX2 / NTAG5.
|
||||
3. **Catalogs + memory maps** per chip (blocks, config, counters, EAS/AFI, password/AES) — sourced
|
||||
from the models in `transponders/hf/iso15693/nxp/` (`icode_slix`, `icode_slix2`, `ntag5_*`).
|
||||
4. Same autocomplete UX (hex + binary).
|
||||
1. **15693 header builder** ✅ — `iso15_frame()` (flags + command + NXP mfg 0x04), shared by all
|
||||
15693 catalogs. The **full request-flags byte** is settable: sub-carrier (0x01), data-rate,
|
||||
inventory, protocol-ext (0x08), and the mode-dependent bits 5-6 (Select/Address vs AFI byte +
|
||||
Nb_slots 1/16), plus option. `iso15_flags_decode` decodes every bit for the byte-0 hint.
|
||||
2. **Per-chip identify** ✅ — `identify.name_iso15` now splits the whole E0 04 family with a
|
||||
READ_CONFIG (0xC0) probe (addressed + retried), **hardware-verified against a real NTAG5 and a
|
||||
real DNA**: session register 0xA0 succeeds only on NTAG5 (`00 01400000`); DNA errors on 0xA0
|
||||
(`01 0F`) but its config block 0x00 succeeds; SLIX2 has no READ_CONFIG (falls to the 0xBD
|
||||
signature step); plain SLIX answers neither. So NTAG5 → "NTAG 5", DNA → "ICODE DNA" (was
|
||||
mislabeled SLIX2), SLIX2/SLIX as before. Real UIDs: NTAG5 `E0 04 01 58`, DNA `E0 04 01 18`
|
||||
(the model's NTAG5 prefix `…18` was copied from DNA; real NTAG5 is `…58`).
|
||||
3. **Catalogs** — `_SLIX`, `_SLIX2`, and **`_NTAG5`** ✅ all done, dispatched by `catalog_for`,
|
||||
verified live. `_NTAG5` = SLIX2 minus STAY_QUIET_PERSISTENT (NTAG5 rejects it) + READ_CONFIG
|
||||
(0xC0) / WRITE_CONFIG (0xC1) / PICK_RANDOM_UID (0xC2) / READ_SRAM (0xD2) / WRITE_SRAM (0xD3).
|
||||
**Follow-up:** the AES / ISO 29167-10 commands AUTHENTICATE (0x35) / CHALLENGE (0x39) / READBUFFER
|
||||
(0x3A) for Link/Boost (AES-capable) variants — a crypto sub-protocol with its own param format.
|
||||
4. Autocomplete UX ✅ (hex + binary, command names + opcode-by-value; the byte-line completion bug
|
||||
is fixed).
|
||||
|
||||
`INVENTORY_READ` (0xA0) / `FAST_INVENTORY_READ` (0xA1) ✅ — in `_SLIX` (inherited by `_SLIX2`),
|
||||
**full request body** via `_inv_read_frame`: `[AFI] | mask_len(bits) | mask value | first_block |
|
||||
count-1`, with the inventory flags set by `iso15_frame(inventory=…, afi=…)`. Mask-less (default),
|
||||
mask-based selective anticollision, and AFI filter all supported. **15693 is now datasheet-complete
|
||||
end to end** (flags byte + both SLIX/SLIX2 command sets + inventory-read body).
|
||||
|
||||
### NEXT: ICODE DNA catalog (`_DNA`)
|
||||
|
||||
DNA is now identified ("ICODE DNA") but borrows `_SLIX2`. Build a dedicated `_DNA` catalog. Sourcing
|
||||
(datasheets are local in the repo root — see [[reference_datasheets]] in memory):
|
||||
- **Command set + semantics + memory map:** `SL2S6002_SDS.pdf` (the DNA short data sheet). It names
|
||||
the commands but — being the *short* sheet — omits exact opcodes (defers to the NDA full sheet).
|
||||
- **Opcodes for the shared commands:** `SL2S2602.pdf` (SLIX2) — DNA extends SLIX2, so EAS/AFI,
|
||||
password (0xB2-B5), READ_SIGNATURE (0xBD), persistent-quiet, etc. carry over. Start from
|
||||
`_SLIX2.commands` like `_NTAG5` did.
|
||||
- **DNA-specific:** READ_CONFIG (0xC0) / WRITE_CONFIG (0xC1) — hardware-confirmed on the real DNA;
|
||||
CID (default 0xC000); AES CHALLENGE (0x39) / AUTHENTICATE (0x35) / READBUFFER (0x3A) from
|
||||
`docs/NTAG5_SECURITY.md` (verified protocol) — same crypto sub-protocol as the NTAG5 AES follow-up.
|
||||
- **Memory map** (for `memory.py`): user = **63 blocks (0-62)**, **block 63 = 16-bit counter**;
|
||||
**config memory = 48 blocks** via READ_CONFIG (holds keys, CID, originality signature, privileges).
|
||||
- Route `"DNA"` → `_DNA` in `catalog_for` (currently → `_SLIX2`). Hardware-verify against the real
|
||||
DNA (finicky: 16-slot inventory + addressed + retries, per [[project_ntag5_dna_id]]).
|
||||
|
||||
The NTAG5 AES commands (0x35/0x39/0x3A) follow-up shares this crypto work — source from `NTA5332.pdf`
|
||||
(the real NTAG5 Link/Boost datasheet, also local) + `docs/NTAG5_SECURITY.md`.
|
||||
|
||||
**Hardware verify pending:** SLIX2 catalog was verified only against unit tests + a live-session
|
||||
drive (no SLIX2 tag on the reader). Confirm on a real SLIX2 when one is available.
|
||||
|
||||
## Also queued
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""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."""
|
||||
with command-tree autocompletion, live hints, and right-arrow help shown in the completion
|
||||
dropdown. Mirrors the :mod:`pm3py.cli.rawcli.app` skeleton; the testable logic lives in the
|
||||
sibling modules."""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
@@ -10,9 +11,9 @@ 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
|
||||
from .hints import one_line_hint
|
||||
|
||||
_DEFAULT_HINT = "type a pm3 command · Tab to complete · → toggles help on a selection · 'quit'"
|
||||
_DEFAULT_HINT = "type a pm3 command · Tab to complete · → help on a selection · 'quit'"
|
||||
|
||||
|
||||
def dispatch(state, line, out=print) -> bool:
|
||||
@@ -60,15 +61,9 @@ def run(port=None, commands=None, client="proxmark3", driver="auto") -> int:
|
||||
state = ClientSession(dev, tree)
|
||||
|
||||
def bottom_toolbar():
|
||||
# live relevance for the current input; the expanded help panel when the toggle is open
|
||||
# only ever the one-line relevance hint — the full help lives in the completion dropdown
|
||||
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)
|
||||
hint = one_line_hint(tree, get_app().current_buffer.text)
|
||||
except Exception:
|
||||
hint = None
|
||||
return ANSI(hint) if hint else _DEFAULT_HINT
|
||||
@@ -92,6 +87,7 @@ def run(port=None, commands=None, client="proxmark3", driver="auto") -> int:
|
||||
with patch_stdout():
|
||||
while True:
|
||||
state.help_open = False # each new line starts with help collapsed
|
||||
state.help_target = None
|
||||
try:
|
||||
line = session.prompt(message)
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
|
||||
from prompt_toolkit.completion import Completer, Completion
|
||||
|
||||
from .hints import help_rows
|
||||
from .tree import parse_option
|
||||
|
||||
|
||||
@@ -17,6 +18,13 @@ class ClientCompleter(Completer):
|
||||
self._session = session
|
||||
|
||||
def get_completions(self, document, complete_event):
|
||||
# help mode (RIGHT on a selection): the dropdown becomes the command's help — the same
|
||||
# two-column rows as normal completions, but non-inserting (LEFT restores completion).
|
||||
if self._session.help_open:
|
||||
for disp, meta in help_rows(self._session.help_target):
|
||||
yield Completion("", start_position=0, display=disp, display_meta=meta)
|
||||
return
|
||||
|
||||
tree = self._session.tree
|
||||
text = document.text_before_cursor
|
||||
tokens = text.split()
|
||||
|
||||
@@ -1,21 +1,14 @@
|
||||
"""Bottom-toolbar hints for clientcli.
|
||||
"""Hints for clientcli, all driven by the command tree.
|
||||
|
||||
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:`one_line_hint` — the one-line relevance hint on the bottom bar for the current input.
|
||||
- :func:`help_rows` — the full help of a command as ``(display, meta)`` rows for the *completion
|
||||
dropdown* (never the bottom bar — the dropdown is where all hints live, per rawcli 65be53d);
|
||||
: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
|
||||
@@ -38,41 +31,31 @@ def one_line_hint(tree, text):
|
||||
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."""
|
||||
def help_rows(node):
|
||||
"""The full help of a highlighted ``node`` as ``(display, meta)`` rows for the completion
|
||||
dropdown — a leaf's name/usage/options/notes, or an interior node's subcommands. Presented in
|
||||
the same two-column form the normal completions use. ``[]`` when there is nothing to show."""
|
||||
if node is None:
|
||||
return None
|
||||
return []
|
||||
if node.is_leaf:
|
||||
return help_panel(node.entry)
|
||||
e = node.entry
|
||||
rows = [(e["command"], e.get("description", ""))]
|
||||
if e.get("usage"):
|
||||
rows.append((e["usage"], "usage"))
|
||||
for opt in e.get("options", []):
|
||||
flags, meta, desc = parse_option(opt)
|
||||
if not flags or _skip(flags):
|
||||
continue
|
||||
rows.append(((", ".join(flags) + (f" {meta}" if meta else "")).strip(), desc))
|
||||
rows += [(n, "example") for n in e.get("notes", [])]
|
||||
return rows
|
||||
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}"
|
||||
rows = [(node.token or "(commands)", f"{len(kids)} subcommands")]
|
||||
for tok in kids:
|
||||
child = node.children[tok]
|
||||
meta = child.entry.get("description", "") if child.is_leaf else f"{len(child.children)} subcommands"
|
||||
rows.append((tok, meta))
|
||||
return rows
|
||||
|
||||
|
||||
def resolve_target(tree, buffer_text, current_completion):
|
||||
@@ -82,12 +65,17 @@ def resolve_target(tree, buffer_text, current_completion):
|
||||
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 ""
|
||||
# subcommand completions carry a trailing space (the Tab-descend mechanism) — strip it or
|
||||
# node_at's exact-token lookup misses and the help panel silently never shows
|
||||
ctext = (getattr(current_completion, "text", "") if current_completion else "").strip()
|
||||
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)
|
||||
# menu navigation has already inserted the highlighted completion into the buffer
|
||||
# (Buffer.go_to_completion), so the path may already end with ctext — don't append it twice
|
||||
if ctext and (not path or path[-1] != ctext):
|
||||
path = path + [ctext]
|
||||
return tree.node_at(path)
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
"""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.
|
||||
When a completion is highlighted in the menu, RIGHT swaps the *dropdown* to the full help of that
|
||||
command (name, usage, options, notes as rows); LEFT (primary) or ESCAPE / Ctrl-g (secondary)
|
||||
restores normal completion. The help never touches the bottom bar — the dropdown is where all hints
|
||||
live. The highlighted node is captured into ``session.help_target`` at the moment RIGHT is pressed,
|
||||
then the completer renders its help while ``help_open`` is set.
|
||||
|
||||
The pure ``open_help`` / ``close_help`` helpers carry the state transitions for unit tests; the key
|
||||
wiring is exercised in the live REPL.
|
||||
@@ -11,15 +12,18 @@ 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:
|
||||
def open_help(session, node) -> bool:
|
||||
"""Open help iff ``node`` resolved to something. Captures it as ``help_target`` so the completer
|
||||
can render its help. Returns the resulting ``help_open`` state."""
|
||||
if node is not None:
|
||||
session.help_target = node
|
||||
session.help_open = True
|
||||
return session.help_open
|
||||
|
||||
|
||||
def close_help(session) -> bool:
|
||||
session.help_open = False
|
||||
session.help_target = None
|
||||
return session.help_open
|
||||
|
||||
|
||||
@@ -51,24 +55,72 @@ 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
|
||||
from prompt_toolkit.key_binding.bindings.named_commands import get_by_name
|
||||
|
||||
from .hints import resolve_target
|
||||
|
||||
help_is_open = Condition(lambda: session.help_open)
|
||||
|
||||
# The exact buffer document + completion menu captured when RIGHT was pressed, so LEFT can put
|
||||
# the user back on the completion they had highlighted (not a freshly regenerated menu).
|
||||
saved = {}
|
||||
|
||||
# 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
|
||||
# complete_while_typing only re-runs completion on INSERT, so deleting characters leaves the
|
||||
# dropdown stale (or blank). Re-run the completer after each deletion so the menu tracks the
|
||||
# now-shorter input. Wrap the default readline command, then refresh — but not while the help
|
||||
# panel is up (editing exits help first, handled by the LEFT/close path).
|
||||
def _delete_then_complete(command_name):
|
||||
binding = get_by_name(command_name)
|
||||
|
||||
def _handler(event):
|
||||
binding.handler(event) # the default deletion
|
||||
if not session.help_open:
|
||||
event.current_buffer.start_completion(select_first=False) # refresh the dropdown
|
||||
return _handler
|
||||
|
||||
for _key, _cmd in (("backspace", "backward-delete-char"),
|
||||
("delete", "delete-char"),
|
||||
("c-w", "unix-word-rubout"),
|
||||
("c-u", "unix-line-discard")):
|
||||
_bind(kb, _key, _delete_then_complete(_cmd))
|
||||
|
||||
def _open(event):
|
||||
buff = event.current_buffer
|
||||
state = buff.complete_state
|
||||
cur = state.current_completion if state else None
|
||||
node = resolve_target(session.tree, buff.text, cur)
|
||||
if not open_help(session, node): # only if it resolved to a real command
|
||||
return
|
||||
# Remember exactly where we are. DETACH the menu (complete_state = None) rather than
|
||||
# cancel_completion() — the latter calls go_to_completion(None), which would mutate this
|
||||
# very state object and wipe the highlighted index we need to restore.
|
||||
saved["document"] = buff.document
|
||||
saved["state"] = state
|
||||
buff.complete_state = None
|
||||
buff.start_completion(select_first=False) # regenerate → help rows (completer sees help_open)
|
||||
event.app.invalidate()
|
||||
|
||||
def _close(event):
|
||||
session.help_open = False
|
||||
buff = event.current_buffer
|
||||
close_help(session)
|
||||
if saved.get("state") is not None:
|
||||
# Restore the pre-RIGHT selection exactly. Set the document first (this detaches
|
||||
# complete_state without firing completion), then re-attach the saved menu — the same
|
||||
# order Buffer.go_to_completion uses.
|
||||
buff.document = saved["document"]
|
||||
buff.complete_state = saved["state"]
|
||||
saved.clear()
|
||||
else:
|
||||
buff.cancel_completion()
|
||||
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)
|
||||
# RIGHT opens help for the highlighted completion. With nothing selected (or help already open)
|
||||
# the filter is false, so the binding is inert and the default cursor-right still works.
|
||||
_bind(kb, "right", _open, filter=completion_is_selected & ~help_is_open)
|
||||
# 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
|
||||
|
||||
@@ -14,12 +14,14 @@ _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."""
|
||||
trie. ``help_open`` swaps the completion dropdown to the help of ``help_target`` (the node
|
||||
captured when RIGHT was pressed); LEFT clears both back to normal completion."""
|
||||
|
||||
def __init__(self, driver: Any, tree: Any):
|
||||
self.driver = driver
|
||||
self.tree = tree
|
||||
self.help_open: bool = False
|
||||
self.help_target: Any = None
|
||||
|
||||
@property
|
||||
def prompt_label(self) -> str:
|
||||
|
||||
@@ -20,13 +20,18 @@ class TagCommand:
|
||||
build: Callable[..., bytes] | None = None # HF: string args -> raw payload bytes to exchange
|
||||
help: str = ""
|
||||
run: Callable[..., object] | None = None # LF: (device, *args) -> a human result line
|
||||
opcode_byte: int | None = None # explicit wire opcode. ISO15693 sets this to the
|
||||
# COMMAND byte (frame byte 1), since byte 0 is flags
|
||||
|
||||
def usage(self) -> str:
|
||||
return f"{self.name}({', '.join(self.params)})"
|
||||
|
||||
def opcode(self) -> int | None:
|
||||
"""The command's first sent byte, by building it with placeholder args — used to match raw
|
||||
input back to a command. None when there's no build (LF run-only commands)."""
|
||||
"""The command's wire opcode — an explicit ``opcode_byte`` if given (ISO15693: the command
|
||||
byte, not the flags-first frame byte), else the first built byte. None for run-only LF
|
||||
commands with no build."""
|
||||
if self.opcode_byte is not None:
|
||||
return self.opcode_byte
|
||||
if self.build is None:
|
||||
return None
|
||||
try:
|
||||
@@ -71,8 +76,8 @@ def _catalog(name: str, protocol: str, *cmds: TagCommand) -> Catalog:
|
||||
return Catalog(name, protocol, {c.name.upper(): c for c in cmds})
|
||||
|
||||
|
||||
def _c(name, params, build=None, help="", run=None) -> TagCommand:
|
||||
return TagCommand(name, tuple(params), build, help, run)
|
||||
def _c(name, params, build=None, help="", run=None, opcode=None) -> TagCommand:
|
||||
return TagCommand(name, tuple(params), build, help, run, opcode)
|
||||
|
||||
|
||||
# ---- ISO 14443-A Type 2 (NTAG / Ultralight) ----
|
||||
@@ -220,56 +225,299 @@ MFC = _catalog(
|
||||
_c("HALT", [], lambda: bytes([0x50, 0x00]), "halt the card (0x50 00)"),
|
||||
)
|
||||
|
||||
# ---- ISO 15693 (flags byte 0x02 = high data rate, single tag) ----
|
||||
# ---- ISO 15693 request framing --------------------------------------------------------------
|
||||
# A 15693 request is flags | command | [mfg] | [UID LSB-first] | params | CRC. The first byte is
|
||||
# the request-FLAGS bitfield, NOT the command — so 15693 TagCommands carry an explicit opcode=
|
||||
# (the command byte) for completion, and iso15_frame() assembles the wire frame (CRC added by
|
||||
# hf.iso15.raw()).
|
||||
# Request-flags byte (byte 0). Bits 5-6 (0x10/0x20) have two meanings, chosen by the Inventory
|
||||
# flag (0x04): non-inventory -> Select/Address; inventory -> AFI/Nb_slots (ISO 15693-3 §7.3.1).
|
||||
F15_SUBCARRIER = 0x01 # two subcarriers (else one)
|
||||
F15_HIGH_RATE = 0x02 # data_rate: high (else low)
|
||||
F15_INVENTORY = 0x04
|
||||
F15_PROTO_EXT = 0x08
|
||||
F15_SELECT = 0x10 # non-inventory: Select flag
|
||||
F15_AFI = 0x10 # inventory: AFI flag (same bit, inventory mode)
|
||||
F15_ADDRESSED = 0x20 # non-inventory: Address flag
|
||||
F15_1SLOT = 0x20 # inventory: Nb_slots = 1 (else 16)
|
||||
F15_OPTION = 0x40
|
||||
NXP_MFG = 0x04
|
||||
|
||||
|
||||
def iso15_frame(command: int, params: bytes = b"", *, uid=None, addressed: bool = False,
|
||||
high_rate: bool = True, option: bool = False, inventory: bool = False,
|
||||
select: bool = False, proto_ext: bool = False, two_subcarrier: bool = False,
|
||||
slots16: bool = False, afi: int | None = None, mfg: int | None = None) -> bytes:
|
||||
"""Assemble an ISO 15693 request frame (pre-CRC): flags | command | [mfg] | [UID|AFI] | params.
|
||||
|
||||
Exposes the full request-flags byte (byte 0). Bits 5-6 differ by mode (ISO 15693-3 §7.3.1):
|
||||
non-inventory -> Select(0x10)/Address(0x20); inventory -> AFI(0x10)/Nb_slots(0x20). Set
|
||||
``inventory=True`` for the INVENTORY family — 1 slot unless ``slots16``, and the AFI byte + flag
|
||||
when ``afi`` is given; otherwise ``select``/``addressed`` apply. ``two_subcarrier`` (0x01),
|
||||
``proto_ext`` (0x08) and ``option`` (0x40) apply in both modes. An addressed UID is transmitted
|
||||
LSByte-first; the inventory AFI byte precedes ``params`` (i.e. the mask length/value)."""
|
||||
flags = 0
|
||||
if two_subcarrier:
|
||||
flags |= F15_SUBCARRIER
|
||||
if high_rate:
|
||||
flags |= F15_HIGH_RATE
|
||||
if proto_ext:
|
||||
flags |= F15_PROTO_EXT
|
||||
if option:
|
||||
flags |= F15_OPTION
|
||||
if inventory:
|
||||
flags |= F15_INVENTORY
|
||||
if afi is not None:
|
||||
flags |= F15_AFI # inventory: AFI flag (bit 5)
|
||||
if not slots16:
|
||||
flags |= F15_1SLOT # inventory: Nb_slots = 1 (bit 6)
|
||||
else:
|
||||
if select:
|
||||
flags |= F15_SELECT # non-inventory: Select (bit 5)
|
||||
if addressed and uid:
|
||||
flags |= F15_ADDRESSED # non-inventory: Address (bit 6)
|
||||
out = bytearray([flags, command])
|
||||
if mfg is not None:
|
||||
out.append(mfg)
|
||||
if inventory:
|
||||
if afi is not None:
|
||||
out.append(afi)
|
||||
elif addressed and uid:
|
||||
u = bytes.fromhex(uid) if isinstance(uid, str) else bytes(uid)
|
||||
out += u[::-1] # UID is transmitted LSByte-first
|
||||
return bytes(out) + bytes(params)
|
||||
|
||||
|
||||
def iso15_flags_decode(flags: int) -> str:
|
||||
"""Human decode of a 15693 request-flags byte (byte 0), for the raw-entry hint. Covers every
|
||||
bit; bits 5-6 are mode-dependent (Select/Address vs AFI/Nb_slots, see iso15_frame)."""
|
||||
if flags & F15_INVENTORY:
|
||||
parts = ["inventory", "high rate" if flags & F15_HIGH_RATE else "low rate"]
|
||||
if flags & F15_SUBCARRIER:
|
||||
parts.append("two subcarrier")
|
||||
parts.append("1 slot" if flags & F15_1SLOT else "16 slots")
|
||||
if flags & F15_AFI:
|
||||
parts.append("AFI")
|
||||
if flags & F15_PROTO_EXT:
|
||||
parts.append("protocol-ext")
|
||||
else:
|
||||
parts = ["high rate" if flags & F15_HIGH_RATE else "low rate"]
|
||||
if flags & F15_SUBCARRIER:
|
||||
parts.append("two subcarrier")
|
||||
parts.append("addressed" if flags & F15_ADDRESSED else "non-addressed")
|
||||
if flags & F15_SELECT:
|
||||
parts.append("select")
|
||||
if flags & F15_PROTO_EXT:
|
||||
parts.append("protocol-ext")
|
||||
if flags & F15_OPTION:
|
||||
parts.append("option")
|
||||
return " · ".join(parts)
|
||||
|
||||
|
||||
# common flags bytes offered at frame byte 0 (the decode hint covers any value you type)
|
||||
ISO15_FLAG_LANDMARKS = [0x02, 0x22, 0x12, 0x42, 0x62, 0x0A, 0x26, 0x06, 0x36, 0x00]
|
||||
|
||||
|
||||
def _inv_read_frame(cmd: int, first_block, count, mask_len="0", mask="", afi="") -> bytes:
|
||||
"""INVENTORY READ / FAST INVENTORY READ request body:
|
||||
``[AFI] | mask_len(bits) | mask_value | first_block | (count-1)``.
|
||||
|
||||
``mask_len`` 0 (default) reads the one tag in the field. A non-zero ``mask_len`` (bits) plus a
|
||||
``mask`` value (hex, LSByte-first on the wire, padded/truncated to ceil(bits/8) bytes) does
|
||||
selective anticollision; ``afi`` (when given) adds the AFI filter byte + flag. This is the full
|
||||
ICODE inventory-read request per the datasheet."""
|
||||
ml = _int(mask_len)
|
||||
nbytes = (ml + 7) // 8
|
||||
mask_bytes = ((_hex(mask) if mask else b"") + b"\x00" * nbytes)[:nbytes]
|
||||
body = bytes([ml]) + mask_bytes + bytes([_int(first_block), _int(count) - 1])
|
||||
return iso15_frame(cmd, body, inventory=True,
|
||||
afi=(_int(afi) if str(afi) != "" else None), mfg=NXP_MFG)
|
||||
|
||||
|
||||
# ---- ISO 15693 (generic) ----
|
||||
ISO15 = _catalog(
|
||||
"ISO 15693", "hf15",
|
||||
_c("INVENTORY", [], lambda: bytes([0x26, 0x01, 0x00]), "anticollision inventory (0x01)"),
|
||||
_c("READ_BLOCK", ["block"], lambda block: bytes([0x02, 0x20, _int(block)]),
|
||||
"read a single block (0x20)"),
|
||||
_c("READ_BLOCK", ["block"], lambda block: iso15_frame(0x20, bytes([_int(block)])),
|
||||
"read a single block (0x20)", opcode=0x20),
|
||||
_c("WRITE_BLOCK", ["block", "data"],
|
||||
lambda block, data: bytes([0x02, 0x21, _int(block)]) + _hex(data),
|
||||
"write a single block (0x21)"),
|
||||
_c("GET_SYSTEM_INFO", [], lambda: bytes([0x02, 0x2B]), "get system information (0x2B)"),
|
||||
lambda block, data: iso15_frame(0x21, bytes([_int(block)]) + _hex(data)),
|
||||
"write a single 4-byte block (0x21)", opcode=0x21),
|
||||
_c("GET_SYSTEM_INFO", [], lambda: iso15_frame(0x2B), "get system information (0x2B)", opcode=0x2B),
|
||||
)
|
||||
|
||||
|
||||
# ---- ICODE SLIX (NXP SL2S2002) — datasheet command set via the header builder ----
|
||||
_SLIX = _catalog(
|
||||
"ICODE SLIX", "hf15",
|
||||
# standard ISO 15693
|
||||
_c("READ_BLOCK", ["block"], lambda block: iso15_frame(0x20, bytes([_int(block)])),
|
||||
"read a single 4-byte block (0x20)", opcode=0x20),
|
||||
_c("WRITE_BLOCK", ["block", "data"],
|
||||
lambda block, data: iso15_frame(0x21, bytes([_int(block)]) + _hex(data).ljust(4, b"\x00")[:4]),
|
||||
"write a single 4-byte block (0x21)", opcode=0x21),
|
||||
_c("LOCK_BLOCK", ["block"], lambda block: iso15_frame(0x22, bytes([_int(block)])),
|
||||
"permanently lock a block (0x22)", opcode=0x22),
|
||||
_c("READ_MULTIPLE", ["first", "count"],
|
||||
lambda first, count: iso15_frame(0x23, bytes([_int(first), _int(count) - 1])),
|
||||
"read <count> blocks from <first> (0x23)", opcode=0x23),
|
||||
_c("SELECT", [], lambda: iso15_frame(0x25), "put the tag in Selected state (0x25)", opcode=0x25),
|
||||
_c("RESET_TO_READY", [], lambda: iso15_frame(0x26), "reset the tag to Ready (0x26)", opcode=0x26),
|
||||
_c("STAY_QUIET", [], lambda: iso15_frame(0x02), "silence the tag until reset (0x02)", opcode=0x02),
|
||||
_c("WRITE_AFI", ["afi"], lambda afi: iso15_frame(0x27, bytes([_int(afi)])),
|
||||
"write the AFI byte (0x27)", opcode=0x27),
|
||||
_c("LOCK_AFI", [], lambda: iso15_frame(0x28), "lock the AFI (0x28)", opcode=0x28),
|
||||
_c("WRITE_DSFID", ["dsfid"], lambda dsfid: iso15_frame(0x29, bytes([_int(dsfid)])),
|
||||
"write the DSFID byte (0x29)", opcode=0x29),
|
||||
_c("LOCK_DSFID", [], lambda: iso15_frame(0x2A), "lock the DSFID (0x2A)", opcode=0x2A),
|
||||
_c("GET_SYSTEM_INFO", [], lambda: iso15_frame(0x2B), "get system information (0x2B)", opcode=0x2B),
|
||||
_c("GET_SECURITY", ["first", "count"],
|
||||
lambda first, count: iso15_frame(0x2C, bytes([_int(first), _int(count) - 1])),
|
||||
"get block security (lock) status (0x2C)", opcode=0x2C),
|
||||
# NXP custom (manufacturer code 0x04 inserted by the builder)
|
||||
_c("GET_NXP_SYSTEM_INFO", [], lambda: iso15_frame(0xAB, mfg=NXP_MFG),
|
||||
"NXP system info — protection pointer, lock, features (0xAB)", opcode=0xAB),
|
||||
_c("GET_RANDOM", [], lambda: iso15_frame(0xB2, mfg=NXP_MFG),
|
||||
"get a 16-bit challenge to XOR the password with (0xB2)", opcode=0xB2),
|
||||
_c("SET_PASSWORD", ["pwd_id", "xor_pwd"],
|
||||
lambda pwd_id, xor_pwd: iso15_frame(0xB3, bytes([_int(pwd_id)]) + _hex(xor_pwd), mfg=NXP_MFG),
|
||||
"present a password (4 bytes XOR'd with GET_RANDOM) (0xB3)", opcode=0xB3),
|
||||
_c("WRITE_PASSWORD", ["pwd_id", "pwd"],
|
||||
lambda pwd_id, pwd: iso15_frame(0xB4, bytes([_int(pwd_id)]) + _hex(pwd), mfg=NXP_MFG),
|
||||
"write a new password (0xB4)", opcode=0xB4),
|
||||
_c("LOCK_PASSWORD", ["pwd_id"],
|
||||
lambda pwd_id: iso15_frame(0xB5, bytes([_int(pwd_id)]), mfg=NXP_MFG),
|
||||
"lock a password (0xB5)", opcode=0xB5),
|
||||
_c("PROTECT_PAGE", ["page", "prot"],
|
||||
lambda page, prot: iso15_frame(0xB6, bytes([_int(page), _int(prot)]), mfg=NXP_MFG),
|
||||
"set a page's read/write protection (0xB6)", opcode=0xB6),
|
||||
_c("LOCK_PAGE_PROTECTION", ["page"],
|
||||
lambda page: iso15_frame(0xB7, bytes([_int(page)]), mfg=NXP_MFG),
|
||||
"lock a page's protection condition (0xB7)", opcode=0xB7),
|
||||
_c("SET_EAS", [], lambda: iso15_frame(0xA2, mfg=NXP_MFG), "set the EAS bit (0xA2)", opcode=0xA2),
|
||||
_c("RESET_EAS", [], lambda: iso15_frame(0xA3, mfg=NXP_MFG), "clear the EAS bit (0xA3)", opcode=0xA3),
|
||||
_c("LOCK_EAS", [], lambda: iso15_frame(0xA4, mfg=NXP_MFG), "lock the EAS bit (0xA4)", opcode=0xA4),
|
||||
_c("EAS_ALARM", [], lambda: iso15_frame(0xA5, mfg=NXP_MFG), "read the EAS alarm sequence (0xA5)", opcode=0xA5),
|
||||
_c("PROTECT_EAS_AFI", [], lambda: iso15_frame(0xA6, mfg=NXP_MFG),
|
||||
"password-protect EAS/AFI (0xA6)", opcode=0xA6),
|
||||
_c("WRITE_EAS_ID", ["eas_id"], lambda eas_id: iso15_frame(0xA7, _hex(eas_id), mfg=NXP_MFG),
|
||||
"write the 16-bit EAS ID (0xA7)", opcode=0xA7),
|
||||
_c("ENABLE_PRIVACY", ["xor_pwd"], lambda xor_pwd: iso15_frame(0xBA, _hex(xor_pwd), mfg=NXP_MFG),
|
||||
"enter privacy mode with the privacy password (0xBA)", opcode=0xBA),
|
||||
_c("DESTROY", ["xor_pwd"], lambda xor_pwd: iso15_frame(0xB9, _hex(xor_pwd), mfg=NXP_MFG),
|
||||
"permanently destroy the tag — IRREVERSIBLE (0xB9)", opcode=0xB9),
|
||||
# INVENTORY READ family: inventory + block read in one exchange. Full body via _inv_read_frame
|
||||
# ([AFI] | mask_len | mask | first_block | count-1); mask-less reads the one tag in the field.
|
||||
_c("INVENTORY_READ", ["first_block", "count", "mask_len", "mask", "afi"],
|
||||
lambda first_block, count, mask_len="0", mask="", afi="":
|
||||
_inv_read_frame(0xA0, first_block, count, mask_len, mask, afi),
|
||||
"inventory + read <count> blocks from <first_block>; optional <mask_len> bits + <mask> value "
|
||||
"(selective anticollision) and <afi> filter (0xA0)", opcode=0xA0),
|
||||
_c("FAST_INVENTORY_READ", ["first_block", "count", "mask_len", "mask", "afi"],
|
||||
lambda first_block, count, mask_len="0", mask="", afi="":
|
||||
_inv_read_frame(0xA1, first_block, count, mask_len, mask, afi),
|
||||
"INVENTORY_READ with the response at the higher data rate (0xA1)", opcode=0xA1),
|
||||
)
|
||||
|
||||
|
||||
# ICODE SLIX2 (SL2S2602): everything the SLIX has, plus three SLIX2-only commands and a larger
|
||||
# 80-block memory. READ_SIGNATURE (0xBD) is the one identify probes to tell SLIX2 from SLIX.
|
||||
_SLIX2 = _catalog(
|
||||
"ICODE SLIX2", "hf15", *_SLIX.commands.values(),
|
||||
_c("READ_SIGNATURE", [], lambda: iso15_frame(0xBD, mfg=NXP_MFG),
|
||||
"read the 32-byte ECC originality signature — SLIX2 only (0xBD)", opcode=0xBD),
|
||||
_c("PASSWORD_PROTECTION_64BIT", [], lambda: iso15_frame(0xBB, mfg=NXP_MFG),
|
||||
"combine the read+write passwords into 64-bit protection — SLIX2 only (0xBB)", opcode=0xBB),
|
||||
_c("STAY_QUIET_PERSISTENT", [], lambda: iso15_frame(0xBC, mfg=NXP_MFG),
|
||||
"persistent quiet that survives power cycles — SLIX2 only (0xBC)", opcode=0xBC),
|
||||
)
|
||||
|
||||
|
||||
# NTAG 5 (NTP5210 Switch / NTA5332 Link/Boost): the SLIX2 command set minus STAY_QUIET_PERSISTENT
|
||||
# (NTAG5 rejects it) plus config memory + session registers, PICK RANDOM UID, and — on Link/Boost —
|
||||
# SRAM. NTAG5 shares the DNA UID (E0 04 01 18), so routing here needs a functional identify probe.
|
||||
_NTAG5 = _catalog(
|
||||
"NTAG 5", "hf15",
|
||||
*[c for c in _SLIX2.commands.values() if c.name != "STAY_QUIET_PERSISTENT"],
|
||||
_c("READ_CONFIG", ["block", "count"],
|
||||
lambda block, count: iso15_frame(0xC0, bytes([_int(block), _int(count) - 1]), mfg=NXP_MFG),
|
||||
"read config / session-register blocks (0xA0 STATUS_REG, 0xA1 CONFIG_REG) (0xC0)", opcode=0xC0),
|
||||
_c("WRITE_CONFIG", ["block", "data"],
|
||||
lambda block, data: iso15_frame(
|
||||
0xC1, bytes([_int(block)]) + _hex(data).ljust(4, b"\x00")[:4], mfg=NXP_MFG),
|
||||
"write a 4-byte config / session-register block (0xC1)", opcode=0xC1),
|
||||
_c("PICK_RANDOM_UID", [], lambda: iso15_frame(0xC2, mfg=NXP_MFG),
|
||||
"generate a random UID for privacy mode — AES variants only (0xC2)", opcode=0xC2),
|
||||
_c("READ_SRAM", ["block", "count"],
|
||||
lambda block, count: iso15_frame(0xD2, bytes([_int(block), _int(count) - 1]), mfg=NXP_MFG),
|
||||
"read SRAM blocks — Link/Boost only (0xD2)", opcode=0xD2),
|
||||
_c("WRITE_SRAM", ["block", "data"],
|
||||
lambda block, data: iso15_frame(
|
||||
0xD3, bytes([_int(block)]) + _hex(data).ljust(4, b"\x00")[:4], mfg=NXP_MFG),
|
||||
"write a 4-byte SRAM block — Link/Boost only (0xD3)", opcode=0xD3),
|
||||
)
|
||||
|
||||
|
||||
# ---- LF T5577 / ATA5577 (block read/write over the T55xx downlink; executes via lf.t55) ----
|
||||
# build() exists only to expose the downlink opcode for hex/binary completion; execution goes
|
||||
# through run() (the device method), since LF isn't a raw-byte exchange like 14a.
|
||||
def _t55_read(dev, block):
|
||||
b = _int(block)
|
||||
if b == 0: # config block — the reliable one
|
||||
def _word(s) -> int:
|
||||
"""A 32-bit T55xx data/password word — parsed as hex (0x/spaces tolerated), matching the pm3
|
||||
client's ``-d``/``-p`` and every other rawcli WRITE. Blocks/pages stay decimal (``_int``)."""
|
||||
h = str(s).lower().replace("0x", "").replace(" ", "") or "0"
|
||||
return int(h, 16) & 0xFFFFFFFF
|
||||
|
||||
|
||||
def _optpwd(password) -> int | None:
|
||||
"""A T55xx password argument (hex word), or None when omitted (the empty-string default)."""
|
||||
return _word(password) if str(password) != "" else None
|
||||
|
||||
|
||||
def _pwd_note(password) -> str:
|
||||
return f" (pwd {_word(password):08X})" if str(password) != "" else ""
|
||||
|
||||
|
||||
def _t55_read(dev, block, page="0", password=""):
|
||||
b, pg, pwd = _int(block), _int(page), _optpwd(password)
|
||||
if b == 0 and pg == 0: # config block — the reliable one
|
||||
from pm3py.lf.capture import read_config
|
||||
r = read_config(dev)
|
||||
r = read_config(dev, password=pwd)
|
||||
if r:
|
||||
return (f"block 0 = {r['config_hex']} ({r['modulation']}, RF/{r['data_bit_rate']}, "
|
||||
f"{r['max_block']} blocks) -> {r['emulating']}")
|
||||
return "READ block 0: no clean config recovered"
|
||||
f"{r['max_block']} blocks) -> {r['emulating']}{_pwd_note(password)}")
|
||||
return f"READ block 0: no clean config recovered{_pwd_note(password)}"
|
||||
from pm3py.lf.capture import read_t55xx_block
|
||||
word = read_t55xx_block(dev, b)
|
||||
word = read_t55xx_block(dev, b, page=pg, password=pwd)
|
||||
where = f"block {b}" + (" (page 1)" if pg else "")
|
||||
if word is None:
|
||||
return f"READ block {b}: no clean response (LF block reads are demod-dependent)"
|
||||
return f"block {b} = {word:08X} (best-effort — bit alignment not verified)"
|
||||
return f"READ {where}: no clean response (LF block reads are demod-dependent){_pwd_note(password)}"
|
||||
return f"{where} = {word:08X} (best-effort — bit alignment not verified){_pwd_note(password)}"
|
||||
|
||||
|
||||
def _t55_write(dev, block, data):
|
||||
r = dev.lf.t55.writebl(_int(block), _int(data))
|
||||
def _t55_write(dev, block, data, page="0", password=""):
|
||||
w = _word(data)
|
||||
r = dev.lf.t55.writebl(_int(block), w, page=_int(page), password=_optpwd(password))
|
||||
ok = isinstance(r, dict) and r.get("status") == 0
|
||||
return f"WRITE block {_int(block)} <- {_int(data):08X}: {'ok' if ok else 'failed'}"
|
||||
where = f"block {_int(block)}" + (" (page 1)" if _int(page) else "")
|
||||
return f"WRITE {where} <- {w:08X}: {'ok' if ok else 'failed'}{_pwd_note(password)}"
|
||||
|
||||
|
||||
def _t55_wake(dev, password):
|
||||
r = dev.lf.t55.wakeup(_int(password))
|
||||
r = dev.lf.t55.wakeup(_word(password))
|
||||
return f"WAKE: {'ok' if isinstance(r, dict) and r.get('status') == 0 else 'failed'}"
|
||||
|
||||
|
||||
def _t55_detect(dev):
|
||||
def _t55_reset(dev):
|
||||
r = dev.lf.t55.reset()
|
||||
return f"RESET: {'ok' if isinstance(r, dict) and r.get('status') == 0 else 'failed'}"
|
||||
|
||||
|
||||
def _t55_detect(dev, password=""):
|
||||
from pm3py.lf.capture import read_config
|
||||
r = read_config(dev)
|
||||
r = read_config(dev, password=_optpwd(password))
|
||||
if not r:
|
||||
return "DETECT: no T55xx config recovered"
|
||||
return f"DETECT: no T55xx config recovered{_pwd_note(password)}"
|
||||
return (f"config {r['config_hex']} ({r['modulation']}, RF/{r['data_bit_rate']}, "
|
||||
f"{r['max_block']} blocks) -> {r['emulating']}")
|
||||
f"{r['max_block']} blocks) -> {r['emulating']}{_pwd_note(password)}")
|
||||
|
||||
|
||||
def _lf_reread(dev):
|
||||
@@ -278,16 +526,26 @@ def _lf_reread(dev):
|
||||
return r["label"] if r else "no LF credential decoded"
|
||||
|
||||
|
||||
# The ATA5577C downlink command set. The wire opcode is a 2-bit field (10/11 = read/write page
|
||||
# 0/1, 00 = reset) followed by optional 32-bit password, lock bit, data and 3-bit address — a
|
||||
# bit-level frame, not a byte exchange, so build() returns a synthetic opcode byte only for the
|
||||
# hex/binary completion; execution goes through run() (lf.t55). READ/WRITE take an optional <page>
|
||||
# (1 = traceability data) and <password> for password-mode access.
|
||||
T5577 = _catalog(
|
||||
"T5577 / ATA5577", "lf",
|
||||
_c("READ", ["block"], build=lambda block: bytes([0x01, _int(block)]), run=_t55_read,
|
||||
help="read a 32-bit block (0=config, 7=password, 1-6=data) — best-effort demod"),
|
||||
_c("WRITE", ["block", "data"], build=lambda block, data: bytes([0x02, _int(block)]),
|
||||
run=_t55_write, help="write 32-bit <data> to <block> (0x02)"),
|
||||
_c("READ", ["block", "page", "password"],
|
||||
build=lambda block, page="0", password="": bytes([0x01, _int(block)]), run=_t55_read,
|
||||
help="read a 32-bit block (0=config, 7=password, 1-6=data); <page> 1 = traceability, "
|
||||
"optional <password> — best-effort demod"),
|
||||
_c("WRITE", ["block", "data", "page", "password"],
|
||||
build=lambda block, data, page="0", password="": bytes([0x02, _int(block)]),
|
||||
run=_t55_write, help="write 32-bit <data> to <block>; optional <page> / <password> (0x02)"),
|
||||
_c("WAKE", ["password"], build=lambda password: bytes([0x03]), run=_t55_wake,
|
||||
help="wake a password-protected tag with the block-7 password (0x03)"),
|
||||
_c("DETECT", [], build=lambda: bytes([0x01, 0x00]), run=_t55_detect,
|
||||
help="read + decode the config block (block 0)"),
|
||||
help="wake a password-protected tag with the block-7 password — AOR (0x03)"),
|
||||
_c("RESET", [], build=lambda: bytes([0x00]), run=_t55_reset,
|
||||
help="send the reset command (opcode 00) — realign the tag's bitstream"),
|
||||
_c("DETECT", ["password"], build=lambda password="": bytes([0x01, 0x00]), run=_t55_detect,
|
||||
help="read + decode the config block (block 0); optional <password>"),
|
||||
)
|
||||
|
||||
# ---- LF read-only credentials (EM4100/HID/AWID/FDX-B): broadcast tags with no command set ----
|
||||
@@ -301,7 +559,11 @@ def catalog_for(session) -> Catalog | None:
|
||||
"""Resolve the command catalog for the session's identified transponder, or None."""
|
||||
name = (session.transponder or "").upper()
|
||||
if session.protocol == "hf15":
|
||||
return ISO15
|
||||
if "NTAG5" in name or "NTAG 5" in name:
|
||||
return _NTAG5
|
||||
if "SLIX2" in name or "DNA" in name: # DNA: closest existing set; own catalog is a follow-up
|
||||
return _SLIX2
|
||||
return _SLIX if "SLIX" in name or "ICODE" in name else ISO15
|
||||
if session.protocol == "hf14a":
|
||||
if not session.transponder:
|
||||
return None
|
||||
|
||||
@@ -19,6 +19,7 @@ from prompt_toolkit.completion import Completer, Completion
|
||||
from .parser import CONTROL_VERBS, parse_token
|
||||
from .catalog import catalog_for
|
||||
from .memory import landmark_pages
|
||||
from .entry import is_byte_line
|
||||
|
||||
_HEXDIGITS = set("0123456789abcdefABCDEF")
|
||||
_BINDIGITS = set("01")
|
||||
@@ -50,6 +51,12 @@ class RawCompleter(Completer):
|
||||
tokens = stripped.split()
|
||||
at_word = bool(tokens) and not stripped.endswith(" ")
|
||||
|
||||
# ISO 15693 raw frame is position-aware: byte 0 = flags, byte 1 = command, byte 2 = mfg
|
||||
# (a bitfield-first frame, unlike 14a where byte 0 is the opcode).
|
||||
if getattr(self._session, "protocol", None) == "hf15" and is_byte_line(stripped):
|
||||
yield from self._iso15_raw(tokens, at_word)
|
||||
return
|
||||
|
||||
# raw byte entry, on the byte after a page-command opcode (30 …): offer the memory map as
|
||||
# raw bytes, so `30 04` gets the same suggestions as READ( — inserting the raw value
|
||||
on_second = (len(tokens) == 2 and at_word) or (len(tokens) == 1 and not at_word)
|
||||
@@ -108,6 +115,59 @@ class RawCompleter(Completer):
|
||||
yield Completion(val, start_position=-len(partial),
|
||||
display=f"{val} {role}", display_meta=role)
|
||||
|
||||
def _raw_base(self, partial: str) -> tuple[str, str]:
|
||||
"""(base, body) for a raw-frame byte token — a 0x/0b prefix wins, else the session mode."""
|
||||
low = partial.lower()
|
||||
if low.startswith("0x"):
|
||||
return "hex", low[2:]
|
||||
if low.startswith("0b"):
|
||||
return "bin", low[2:]
|
||||
return ("bin" if getattr(self._session, "entry_mode", "hex") == "bin" else "hex"), low
|
||||
|
||||
def _iso15_byte(self, value: int, partial: str, label: str, meta: str):
|
||||
"""A single raw-byte completion rendered in the entry base, filtered by ``partial``."""
|
||||
base, body = self._raw_base(partial)
|
||||
val = f"{value:08b}" if base == "bin" else f"{value:02X}"
|
||||
if body and not val.lower().startswith(body):
|
||||
return None
|
||||
return Completion(val, start_position=-len(partial), display=f"{val} {label}", display_meta=meta)
|
||||
|
||||
def _iso15_raw(self, tokens: list[str], at_word: bool):
|
||||
"""Position-aware completion for an ISO 15693 raw frame: byte 0 = the request-flags menu
|
||||
(each value decoded), byte 1 = the command, byte 2 = the NXP mfg code on custom commands."""
|
||||
from .catalog import ISO15_FLAG_LANDMARKS, iso15_flags_decode, NXP_MFG
|
||||
pos = (len(tokens) - 1) if at_word else len(tokens)
|
||||
partial = tokens[pos] if (at_word and pos < len(tokens)) else ""
|
||||
if pos == 0: # flags byte — the header menu + decode
|
||||
for f in ISO15_FLAG_LANDMARKS:
|
||||
c = self._iso15_byte(f, partial, iso15_flags_decode(f), iso15_flags_decode(f))
|
||||
if c:
|
||||
yield c
|
||||
elif pos == 1: # command byte
|
||||
catalog = catalog_for(self._session)
|
||||
if catalog is None:
|
||||
return
|
||||
for name in catalog.names():
|
||||
op = catalog.get(name).opcode()
|
||||
if op is None:
|
||||
continue
|
||||
c = self._iso15_byte(op, partial, name, catalog.get(name).help)
|
||||
if c:
|
||||
yield c
|
||||
elif pos == 2: # mfg byte, only after a custom command
|
||||
cmd = self._token_byte(tokens[1]) if len(tokens) > 1 else None
|
||||
if cmd is not None and cmd >= 0xA0:
|
||||
c = self._iso15_byte(NXP_MFG, partial, "NXP manufacturer code", "NXP custom (0x04)")
|
||||
if c:
|
||||
yield c
|
||||
|
||||
def _token_byte(self, tok: str) -> int | None:
|
||||
try:
|
||||
b = parse_token(tok, getattr(self._session, "entry_mode", "hex"))
|
||||
except ValueError:
|
||||
return None
|
||||
return b[0] if len(b) == 1 else None
|
||||
|
||||
def _page_arg(self, cmd_name: str, partial: str):
|
||||
"""Complete a page/block argument with the identified tag's memory landmarks, rendered in
|
||||
the active entry mode — ``0x04`` in hex, ``0b00000100`` in binary (every bit visible,
|
||||
|
||||
@@ -75,7 +75,6 @@ def install_key_bindings(kb, session) -> None:
|
||||
"""Wire the entry-mode toggle (Ctrl-/ via Ctrl-_, plus Ctrl-t) and per-byte base-aware digit
|
||||
entry onto ``kb``. ``session`` is the :class:`RawSession` whose ``entry_mode`` is toggled/read
|
||||
and whose ``byte_bases`` records each raw byte's base. Never raises on an unknown key name."""
|
||||
from prompt_toolkit.document import Document
|
||||
|
||||
def _toggle(event):
|
||||
session.toggle_entry_mode() # only flips the mode — the next byte you type uses it
|
||||
@@ -93,7 +92,13 @@ def install_key_bindings(kb, session) -> None:
|
||||
buf.insert_text(d)
|
||||
return
|
||||
new_text, session.byte_bases = type_digit(text, session.byte_bases, session.entry_mode, d)
|
||||
buf.document = Document(new_text, len(new_text))
|
||||
# Apply the change as an append via insert_text, NOT `buf.document = …`: assigning the
|
||||
# document resets prompt_toolkit's complete_state and never re-fires completion, which
|
||||
# killed the live autocomplete menu during raw byte entry. type_digit only ever appends
|
||||
# (optionally after an auto-space) or drops, so the delta is always a pure suffix.
|
||||
suffix = new_text[len(text):]
|
||||
if suffix:
|
||||
buf.insert_text(suffix)
|
||||
|
||||
for ch in sorted(_DIGITS[HEX] | _DIGITS[BIN]):
|
||||
_bind(kb, ch, _digit)
|
||||
|
||||
@@ -78,6 +78,68 @@ def name_14a(scan: dict) -> str:
|
||||
return "ISO14443-A"
|
||||
|
||||
|
||||
def _iso15_uid(dev) -> str | None:
|
||||
"""Get a 15693 tag's UID robustly. The anticollision inventory misses often on vicinity tags,
|
||||
and this link also returns the odd garbage frame — so retry both the inventory and a direct
|
||||
GET_SYSTEM_INFO (0x2B, which carries the UID), validating each response. Returns display-order
|
||||
UID hex, or None."""
|
||||
for _ in range(3):
|
||||
scan = _try(lambda: dev.hf.iso15.scan(), {"found": False})
|
||||
uid = scan.get("uid")
|
||||
if scan.get("found") and isinstance(uid, str) and uid.lower().startswith("e0"):
|
||||
return uid
|
||||
for _ in range(4): # GET_SYSTEM_INFO: [flags][info][UID(8)]..
|
||||
resp = _try(lambda: dev.hf.iso15.raw(b"\x02\x2b"), None)
|
||||
raw = resp.get("raw") if isinstance(resp, dict) else None
|
||||
if raw and len(raw) >= 10 and raw[0] == 0x00 and raw[9] == 0xE0: # UID LSB-first ends in E0
|
||||
return bytes(raw[2:10])[::-1].hex()
|
||||
return None
|
||||
|
||||
|
||||
def name_iso15(dev, uid15: str) -> str:
|
||||
"""Name a 15693 tag from its UID + probes. ``E0 04`` = NXP ICODE. NTAG5 shares ``E0 04 01`` with
|
||||
ICODE DNA, so it is told apart by a functional probe (see below); SLIX2 is distinguished from
|
||||
plain SLIX by READ_SIGNATURE (0xBD). Non-NXP falls back to generic ISO15693."""
|
||||
if not uid15.lower().startswith("e004"):
|
||||
return "ISO15693"
|
||||
# NTAG5 / DNA / SLIX2 all share E0 04 and answer READ_SIGNATURE, so split them with READ_CONFIG
|
||||
# (0xC0), addressed + retried (these tags are finicky). Session register 0xA0 (STATUS_REG)
|
||||
# succeeds only on NTAG5 (its I2C-side registers). DNA has no session registers (0xA0 errors)
|
||||
# but does have config memory (block 0x00 succeeds); SLIX2 has no READ_CONFIG command at all.
|
||||
# Hardware-verified: NTAG5 (E0 04 01 58) 0xA0 -> 00 01400000; DNA (E0 04 01 18) 0xA0 -> 01 0F,
|
||||
# 0x00 -> 00 ...
|
||||
try:
|
||||
uid_wire = bytes.fromhex(uid15)[::-1] # UID LSByte-first for addressing
|
||||
|
||||
def _read_config(block: int):
|
||||
frame = bytes([0x22, 0xC0, 0x04]) + uid_wire + bytes([block, 0x00])
|
||||
for _ in range(6):
|
||||
r = _try(lambda: dev.hf.iso15.raw(frame), None)
|
||||
raw = r.get("raw") if isinstance(r, dict) else None
|
||||
if raw and raw[0] in (0x00, 0x01): # a real ok/error response (not a miss)
|
||||
return raw
|
||||
return None
|
||||
|
||||
a0 = _read_config(0xA0)
|
||||
if a0 and a0[0] == 0x00 and len(a0) >= 5: # valid STATUS_REG -> NTAG5
|
||||
return "NTAG 5"
|
||||
if a0 and a0[0] == 0x01: # 0xA0 errored -> not NTAG5; DNA has
|
||||
cfg0 = _read_config(0x00) # config memory, SLIX2 has no 0xC0
|
||||
if cfg0 and cfg0[0] == 0x00 and len(cfg0) >= 5:
|
||||
return "ICODE DNA"
|
||||
except ValueError:
|
||||
pass
|
||||
# READ_SIGNATURE (0xBD): SLIX2 returns a 32-byte ECC signature (response >= 33 bytes with the
|
||||
# flags byte); plain SLIX ignores the command. Require the full length so a flaky/stale frame
|
||||
# can't false-positive; probe twice in case the real signature frame is dropped.
|
||||
for _ in range(2):
|
||||
sig = _try(lambda: dev.hf.iso15.raw(b"\x02\xbd\x04"), None)
|
||||
raw = sig.get("raw") if isinstance(sig, dict) else None
|
||||
if raw is not None and len(raw) >= 33 and raw[0] == 0x00:
|
||||
return "ICODE SLIX2"
|
||||
return "ICODE SLIX"
|
||||
|
||||
|
||||
def _ident_decode(direction: int, payload: bytes) -> str | None:
|
||||
"""Annotator for the identify probe frames (activation + GET_VERSION)."""
|
||||
if not payload:
|
||||
@@ -169,14 +231,14 @@ def identify(session, emit=None) -> dict:
|
||||
"transponder": session.transponder, "uid": scan.get("uid"),
|
||||
"version": version.hex() if version else None}
|
||||
|
||||
# ISO 15693 — require found AND a real E0-prefixed UID, else a 14a miss reads as a bogus tag.
|
||||
scan15 = _try(lambda: dev.hf.iso15.scan(), {"found": False})
|
||||
uid15 = scan15.get("uid")
|
||||
if scan15.get("found") and isinstance(uid15, str) and uid15.lower().startswith("e0"):
|
||||
# ISO 15693 — require a real E0-prefixed UID (inventory, or a direct GET_SYSTEM_INFO fallback
|
||||
# for flaky vicinity anticollision), else a 14a miss reads as a bogus tag.
|
||||
uid15 = _iso15_uid(dev)
|
||||
if isinstance(uid15, str) and uid15.lower().startswith("e0"):
|
||||
session.field, session.protocol = "hf", "hf15"
|
||||
session.transponder = "ISO15693"
|
||||
session.transponder = name_iso15(dev, uid15)
|
||||
return {"found": True, "field": "hf", "protocol": "15693",
|
||||
"transponder": "ISO15693", "uid": uid15}
|
||||
"transponder": session.transponder, "uid": uid15}
|
||||
|
||||
# LF: no HF tag, so demodulate the field. pm3py.lf captures the envelope and decodes it —
|
||||
# what the tag emits (EM4100 ID, …) and, via a block-0 read, whether it's actually a T5577
|
||||
|
||||
@@ -133,14 +133,29 @@ def _t5577_block_role(block: int | None) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def _is_iso15(transponder: str | None) -> bool:
|
||||
up = (transponder or "").upper()
|
||||
return "SLIX" in up or "ICODE" in up
|
||||
|
||||
|
||||
def _iso15_block_role(block: int | None) -> str | None:
|
||||
"""ISO 15693 / ICODE blocks are uniform 4-byte user-memory blocks; AFI / DSFID / EAS / config
|
||||
live behind commands, not block addresses."""
|
||||
if block is None or not (0 <= block <= 255):
|
||||
return None
|
||||
return "user memory (4-byte block)"
|
||||
|
||||
|
||||
def page_role(transponder: str | None, page: int | None) -> str | None:
|
||||
"""The role of ``page``/``block`` on ``transponder`` — the full Type 2 header (UID / static
|
||||
lock / CC-or-OTP), user memory, dynamic lock, CFG0/CFG1 (family-accurate fields), PWD, PACK —
|
||||
or a T5577 block role, or None when unknown."""
|
||||
or a T5577 block role, or an ISO15693 block, or None when unknown."""
|
||||
if transponder and "T5577" in transponder.upper():
|
||||
return _t5577_block_role(page)
|
||||
if _mfc_size(transponder):
|
||||
return _mfc_block_role(transponder, page)
|
||||
if _is_iso15(transponder):
|
||||
return _iso15_block_role(page)
|
||||
layout = _resolve_layout(transponder)
|
||||
if layout is None or page is None:
|
||||
return None
|
||||
@@ -174,6 +189,8 @@ def landmark_pages(transponder: str | None) -> list[tuple[int, str]]:
|
||||
return [(b, _t5577_block_role(b)) for b in range(8)]
|
||||
if _mfc_size(transponder):
|
||||
return _mfc_landmarks(transponder)
|
||||
if _is_iso15(transponder):
|
||||
return [(0, "user memory (4-byte block)")] # uniform blocks; AFI/DSFID/EAS via commands
|
||||
layout = _resolve_layout(transponder)
|
||||
if layout is None:
|
||||
return []
|
||||
|
||||
@@ -27,6 +27,12 @@ _READ_FLAGS = ISO15_CONNECT | ISO15_HIGH_SPEED | ISO15_READ_RESPONSE # 0x31
|
||||
# (mirrors the C client's hf_15_write_blk).
|
||||
_WRITE_FLAGS = ISO15_CONNECT | ISO15_READ_RESPONSE | ISO15_LONG_WAIT # 0x61
|
||||
|
||||
# Command bytes that program the tag and so need the long tag-ack window (raw() auto-detects these
|
||||
# from iso_cmd[1]): std write/lock, EAS writes, password/protect/destroy/privacy.
|
||||
_WRITE_CMDS = frozenset({0x21, 0x22, 0x24, 0x27, 0x28, 0x29, 0x2A,
|
||||
0xA2, 0xA3, 0xA4, 0xA6, 0xA7,
|
||||
0xB3, 0xB4, 0xB5, 0xB6, 0xB7, 0xB9, 0xBA})
|
||||
|
||||
|
||||
def _iso15_payload(iso_cmd: bytes, flags: int = _READ_FLAGS) -> bytes:
|
||||
"""Build a HF_ISO15693_COMMAND payload from a raw ISO15693 command.
|
||||
@@ -204,6 +210,24 @@ class HF15Commands:
|
||||
success = resp.status == 0 and len(resp.data) > 0 and resp.data[0] == 0
|
||||
return {"success": success, "block": block}
|
||||
|
||||
async def raw(self, iso_cmd: bytes | str, long_wait: bool | None = None,
|
||||
timeout: float = 5.0) -> dict:
|
||||
"""Send a raw ISO15693 request frame and return the tag's response.
|
||||
|
||||
``iso_cmd`` is the request **without** the CRC (flags | command | [mfg] | [UID] | params);
|
||||
the ISO15693 CRC is appended host-side. ``long_wait`` uses the write-length tag-ack timeout
|
||||
(needed for WRITE/LOCK-class commands); ``None`` (default) auto-selects it from the command
|
||||
byte. The returned ``raw`` is the firmware-relayed tag response (response-flags + data +
|
||||
its CRC)."""
|
||||
if isinstance(iso_cmd, str):
|
||||
iso_cmd = bytes.fromhex(iso_cmd.replace(" ", ""))
|
||||
if long_wait is None:
|
||||
long_wait = (len(iso_cmd) > 1 and iso_cmd[1] in _WRITE_CMDS)
|
||||
payload = _iso15_payload(iso_cmd, flags=_WRITE_FLAGS if long_wait else _READ_FLAGS)
|
||||
resp = await self._t.send_ng(Cmd.HF_ISO15693_COMMAND, payload, timeout=timeout)
|
||||
data = bytes(resp.data)
|
||||
return {"status": resp.status, "raw": data, "data": data.hex()}
|
||||
|
||||
async def sniff(self, timeout: float = 60.0) -> dict:
|
||||
"""Start ISO15693 sniff. Blocks until button press or timeout.
|
||||
|
||||
|
||||
@@ -25,8 +25,15 @@ class T55xxCommands:
|
||||
}
|
||||
|
||||
async def writebl(self, block: int, data: int, page: int = 0,
|
||||
password: int | None = None, downlink_mode: int = 0) -> dict:
|
||||
"""Write a T55xx block."""
|
||||
password: int | None = None, downlink_mode: int = 0,
|
||||
test: bool = False) -> dict:
|
||||
"""Write a T55xx block.
|
||||
|
||||
``test`` uses the test-mode write (downlink opcode 01) instead of the standard opcode 10.
|
||||
Test mode is the datasheet's reconfiguration path (§5.10.3) — it can rewrite a tag that a
|
||||
normal write can't (a corrupted/locked config), provided the master key still permits it
|
||||
(allowed for key 9, denied once key 6 is set). Use it for recovery, not routine writes.
|
||||
"""
|
||||
pwd = password if password is not None else 0
|
||||
# Firmware t55xx_write_block_t: data(4)+pwd(4)+blockno(1)+flags(1).
|
||||
# flags: 0x01 PwdMode, 0x02 Page, 0x04 test, 0x18 downlink_mode<<3.
|
||||
@@ -35,6 +42,8 @@ class T55xxCommands:
|
||||
flags |= 0x01
|
||||
if page:
|
||||
flags |= 0x02
|
||||
if test:
|
||||
flags |= 0x04
|
||||
flags |= (downlink_mode & 0x03) << 3
|
||||
payload = struct.pack("<IIBB", data, pwd, block, flags)
|
||||
resp = await self._t.send_ng(Cmd.LF_T55XX_WRITEBL, payload, timeout=5.0)
|
||||
@@ -48,6 +57,18 @@ class T55xxCommands:
|
||||
resp = await self._t.send_ng(Cmd.LF_T55XX_WAKEUP, payload)
|
||||
return {"status": resp.status}
|
||||
|
||||
async def reset(self, downlink_mode: int = 0) -> dict:
|
||||
"""Send the T55xx reset command (downlink opcode 00) and read the stream.
|
||||
|
||||
Resets the tag's modem to the start of its bitstream so a subsequent read starts
|
||||
aligned. The firmware (T55xxResetRead) takes a single flags byte carrying the
|
||||
downlink mode; the captured stream lands in BigBuf.
|
||||
"""
|
||||
flags = (downlink_mode & 0x03) << 3
|
||||
payload = struct.pack("<B", flags)
|
||||
resp = await self._t.send_ng(Cmd.LF_T55XX_RESET_READ, payload)
|
||||
return {"status": resp.status}
|
||||
|
||||
# Default T55xx downlink-mode timing table (generic/PM3 Easy build), stored
|
||||
# as raw struct values (field-clocks x 8). Mirrors T55xx_Timing in
|
||||
# armsrc/lfops.c. Firmware exposes no GET, and CMD_LF_T55XX_SET_CONFIG is
|
||||
|
||||
@@ -23,6 +23,20 @@ _ASYNC_FRAME_CMDS = frozenset({
|
||||
_MAX_ASYNC_SKIP = 64 # bound the skip loop so a chatty/looping device can't hang a read
|
||||
|
||||
|
||||
def _wtx_ms(resp) -> int | None:
|
||||
"""Waiting-Time-eXtension value (ms) if ``resp`` is a CMD_WTX control frame, else ``None``.
|
||||
|
||||
The firmware emits CMD_WTX to ask the host to *keep waiting* — notably on FPGA bitstream
|
||||
reload (``send_wtx(FPGA_LOAD_WAIT_TIME)``, 1500ms), which the first HF command after connect
|
||||
triggers. It is NOT a command reply. The C client extends its response timeout by this value
|
||||
and reads on (comms.c). pm3py must do the same: otherwise the WTX frame is returned as the
|
||||
reply, and — because every later read then grabs the *previous* command's response — one WTX
|
||||
permanently desyncs the whole session (the bug behind the '15693 garbage/shift' reports)."""
|
||||
if resp.cmd == Cmd.WTX and len(resp.data) >= 2:
|
||||
return struct.unpack_from("<H", resp.data)[0]
|
||||
return None
|
||||
|
||||
|
||||
def _crc_to_wire(crc: int) -> int:
|
||||
"""Byte-swap CRC to match PM3 wire format: (low << 8) | high."""
|
||||
return ((crc & 0xFF) << 8) | ((crc >> 8) & 0xFF)
|
||||
@@ -211,9 +225,10 @@ class PM3Transport:
|
||||
as the reply — the cause of garbage sample counts and desynced downloads."""
|
||||
if not self._reader:
|
||||
raise PM3Error("Not connected")
|
||||
extra = 0.0 # accumulated WTX wait-extension (seconds)
|
||||
for _ in range(_MAX_ASYNC_SKIP):
|
||||
try:
|
||||
raw = await asyncio.wait_for(self._read_response_frame(), timeout=timeout)
|
||||
raw = await asyncio.wait_for(self._read_response_frame(), timeout=timeout + extra)
|
||||
except asyncio.TimeoutError:
|
||||
self._resync_needed = True # late response would desync the next read — drain it
|
||||
raise PM3Error("Response timeout", status=-4)
|
||||
@@ -221,6 +236,10 @@ class PM3Transport:
|
||||
resp = PM3Response.from_dict(decode_response_frame(raw))
|
||||
except ValueError:
|
||||
continue # malformed frame — resync on the next magic
|
||||
wtx = _wtx_ms(resp)
|
||||
if wtx is not None:
|
||||
extra += wtx / 1000.0 # device asked to wait longer — keep reading
|
||||
continue
|
||||
if resp.cmd in _ASYNC_FRAME_CMDS:
|
||||
continue # device debug output — not our reply
|
||||
return resp
|
||||
@@ -364,9 +383,10 @@ class PM3Transport:
|
||||
"""Read one NG-or-OLD frame, skipping async debug-print frames (as read_response does)."""
|
||||
if not self._reader:
|
||||
raise PM3Error("Not connected")
|
||||
extra = 0.0 # accumulated WTX wait-extension (seconds)
|
||||
for _ in range(_MAX_ASYNC_SKIP):
|
||||
try:
|
||||
raw = await asyncio.wait_for(self._read_any_frame(), timeout=timeout)
|
||||
raw = await asyncio.wait_for(self._read_any_frame(), timeout=timeout + extra)
|
||||
except asyncio.TimeoutError:
|
||||
self._resync_needed = True # late response would desync the next read — drain it
|
||||
raise PM3Error("Response timeout", status=-4)
|
||||
@@ -380,6 +400,10 @@ class PM3Transport:
|
||||
data=d["data"], oldarg=d["oldarg"])
|
||||
except ValueError:
|
||||
continue
|
||||
wtx = _wtx_ms(resp)
|
||||
if wtx is not None:
|
||||
extra += wtx / 1000.0 # device asked to wait longer — keep reading
|
||||
continue
|
||||
if resp.cmd in _ASYNC_FRAME_CMDS:
|
||||
continue # device debug output — not our reply
|
||||
return resp
|
||||
|
||||
@@ -17,6 +17,6 @@ Layers:
|
||||
Everything is testable without hardware: the LF transponder models already *encode* these
|
||||
protocols, so a synthetic envelope built from an encoder round-trips through the demod.
|
||||
"""
|
||||
from .capture import demod_samples, read_config, identify_lf
|
||||
from .capture import demod_samples, demod_raw, read_config, identify_lf
|
||||
|
||||
__all__ = ["demod_samples", "read_config", "identify_lf"]
|
||||
__all__ = ["demod_samples", "demod_raw", "read_config", "identify_lf"]
|
||||
|
||||
@@ -127,3 +127,41 @@ def manchester_bits(samples):
|
||||
for phase in (0, 1):
|
||||
for minv in (False, True):
|
||||
yield dsp.manchester_decode(raw, phase, minv)
|
||||
|
||||
|
||||
def nrz_bits(samples):
|
||||
"""Yield candidate raw NRZ / direct bit streams. In NRZ (the T5577 "direct" line code) the
|
||||
data bit is output with no coding — the envelope sampled once per bit *is* the data — so the
|
||||
bit clock is the shortest run itself, not twice a half-bit as in Manchester. Yields both
|
||||
polarities over a few phase offsets; the caller keeps whichever repeats cleanly."""
|
||||
core = trim_to_signal(samples)
|
||||
if len(core) < 64 or dsp.is_flat(core):
|
||||
return
|
||||
binary = dsp.binarize(core)
|
||||
half = dsp.detect_clock(binary)
|
||||
if not half or half < 8:
|
||||
return
|
||||
clk = min(_CLOCKS, key=lambda c: abs(c - half)) # NRZ: one bit = the shortest run
|
||||
for offset in (0.5, 0.0, 0.25, 0.75):
|
||||
raw = dsp.resample(binary, clk, offset)
|
||||
if len(raw) < 32:
|
||||
continue
|
||||
yield raw
|
||||
yield [b ^ 1 for b in raw]
|
||||
|
||||
|
||||
def biphase_bits(samples):
|
||||
"""Yield candidate raw bit streams for a biphase / differential-Manchester ASK envelope — the
|
||||
FDX-B line code (also Indala, Gallagher, …). Resample to half-bit resolution and run the
|
||||
biphase decoder for both polarities; phase-error markers are dropped."""
|
||||
core = trim_to_signal(samples)
|
||||
if len(core) < 64 or dsp.is_flat(core):
|
||||
return
|
||||
binary = dsp.binarize(core)
|
||||
half = dsp.detect_clock(binary)
|
||||
if not half or half < 4:
|
||||
return
|
||||
raw = dsp.resample(binary, half)
|
||||
for invert in (0, 1):
|
||||
bits, _err = dsp.biphase_raw_decode(raw, 0, invert)
|
||||
yield [b for b in bits if b != 7]
|
||||
|
||||
@@ -10,6 +10,7 @@ from __future__ import annotations
|
||||
import inspect
|
||||
|
||||
from . import ask
|
||||
from . import dsp
|
||||
from . import protocols
|
||||
|
||||
# how many envelope samples to pull from BigBuf after a capture (a T55xx/EM frame is a few
|
||||
@@ -53,6 +54,70 @@ def demod_samples(samples) -> dict | None:
|
||||
return None
|
||||
|
||||
|
||||
def _transition_density(bits) -> float:
|
||||
return sum(1 for i in range(1, len(bits)) if bits[i] != bits[i - 1]) / max(1, len(bits) - 1)
|
||||
|
||||
|
||||
def _frame_period(bits, min_p: int = 16) -> int | None:
|
||||
"""The smallest bit-period the stream repeats at (its frame length), or None.
|
||||
|
||||
Guards against the failure mode where the wrong line code decodes a signal into a mostly
|
||||
*constant* run (e.g. biphase-decoding a non-biphase tag emits a short prefix then all-1s):
|
||||
such a stream trivially "repeats" at any period. So we reject a near-constant stream up front,
|
||||
require room for a few real repeats, and reject a frame that is itself near-constant or
|
||||
trivially sub-periodic (an alternating 0101 fill)."""
|
||||
n = len(bits)
|
||||
if n < 3 * min_p:
|
||||
return None
|
||||
if _transition_density(bits) < 0.10: # mostly constant -> not a real frame
|
||||
return None
|
||||
# find the *smallest true* period, from 2 up — so a trivial fill (0101 -> period 2, 0011 ->
|
||||
# period 4) is caught as tiny and rejected, instead of being reported at the first p >= min_p.
|
||||
for p in range(2, n // 3 + 1):
|
||||
span = n - p
|
||||
if sum(1 for i in range(span) if bits[i] == bits[i + p]) >= span * 0.97:
|
||||
return p if p >= min_p else None # real tag frame is >= min_p bits
|
||||
return None
|
||||
|
||||
|
||||
def demod_raw(samples) -> dict | None:
|
||||
"""Physical-layer fallback for an ASK tag no named decoder recognises: recover the raw
|
||||
repeating frame under the NRZ and biphase line codes so the tag still yields its bits.
|
||||
|
||||
Returns whichever encodings produced a clean repeating frame (bit string + hex + period in
|
||||
bits) plus the recovered bit clock, or None if nothing repeats cleanly. This is deliberately
|
||||
*below* the credential decoders — it names no format, it just hands back the bits an expert
|
||||
(or a future format decoder) can read."""
|
||||
core = ask.trim_to_signal(samples)
|
||||
if len(core) < 64 or dsp.is_flat(core):
|
||||
return None
|
||||
half = dsp.detect_clock(dsp.binarize(core))
|
||||
if not half:
|
||||
return None
|
||||
encs: dict = {}
|
||||
for name, gen in (("nrz", ask.nrz_bits), ("biphase", ask.biphase_bits)):
|
||||
best = None
|
||||
for bits in gen(samples):
|
||||
p = _frame_period(bits)
|
||||
if p and (best is None or p < best[0]):
|
||||
best = (p, bits[:p])
|
||||
if best:
|
||||
p, frame = best
|
||||
s = "".join(str(b) for b in frame)
|
||||
encs[name] = {"period": p, "bits": s, "hex": f"{int(s, 2):0{-(-p // 4)}X}"}
|
||||
if not encs:
|
||||
return None
|
||||
clk = min(ask._CLOCKS, key=lambda c: abs(c - half))
|
||||
return {"protocol": "raw", "shortest_run": round(half, 1), "rf": f"RF/{clk}",
|
||||
"encodings": encs}
|
||||
|
||||
|
||||
def raw_label(raw: dict) -> str:
|
||||
"""One-line name for a raw physical-layer dump (no credential decoded)."""
|
||||
parts = [f"{k} {v['hex']} ({v['period']}b)" for k, v in raw["encodings"].items()]
|
||||
return f"unknown LF ({raw['rf']}) — " + "; ".join(parts)
|
||||
|
||||
|
||||
# download a fixed, safe span rather than trusting lf.read()'s reported count — on a flaky NG
|
||||
# link that count is sometimes garbage, and over-reading spills into BigBuf memory past the
|
||||
# samples. 12000 is under a normal acquisition, enough frames for any LF frame to repeat.
|
||||
@@ -108,19 +173,22 @@ def read_emitted(device) -> dict | None:
|
||||
return demod_samples(data) if data else None
|
||||
|
||||
|
||||
def read_config(device) -> dict | None:
|
||||
def read_config(device, password: int | None = None) -> dict | None:
|
||||
"""Probe for a T55xx by reading config block 0 and demodulating the response. A clean,
|
||||
sane config both proves the chip is a T5577 and says what it's programmed to emulate."""
|
||||
data = _capture(device, lambda: _try(lambda: device.lf.t55.readbl(0), None))
|
||||
sane config both proves the chip is a T5577 and says what it's programmed to emulate.
|
||||
``password`` (when set) reads a password-protected tag in password mode."""
|
||||
data = _capture(device, lambda: _try(lambda: device.lf.t55.readbl(0, password=password), None))
|
||||
return protocols.decode_t55xx_config(data) if data else None
|
||||
|
||||
|
||||
def read_t55xx_block(device, block: int) -> int | None:
|
||||
def read_t55xx_block(device, block: int, page: int = 0, password: int | None = None) -> int | None:
|
||||
"""Best-effort read of a T55xx 32-bit data block: send the block read, demod the first
|
||||
repeating 32-bit word. A bare block has no header/CRC, so the bit alignment (rotation) can't
|
||||
be verified — this is inherently best-effort. Block 0 (config) is the reliable one; read it
|
||||
via read_config, which resolves the rotation against known presets."""
|
||||
data = _capture(device, lambda: _try(lambda: device.lf.t55.readbl(int(block)), None))
|
||||
via read_config, which resolves the rotation against known presets. ``page`` 1 reaches the
|
||||
traceability data; ``password`` reads a password-protected tag in password mode."""
|
||||
data = _capture(device, lambda: _try(
|
||||
lambda: device.lf.t55.readbl(int(block), page=int(page), password=password), None))
|
||||
if not data:
|
||||
return None
|
||||
for bits in ask.manchester_bits(data):
|
||||
@@ -156,7 +224,14 @@ def identify_lf(device, emit=None) -> dict | None:
|
||||
config = read_config(device) # T55xx? -> chip + emulation
|
||||
emitted = read_emitted(device) # what's on the air -> EM4100 ID (etc.)
|
||||
|
||||
if config is None and emitted is None:
|
||||
# nothing named decoded off the air -> try the physical-layer fallback (NRZ/biphase) so a tag
|
||||
# no credential decoder handles still yields its raw bits rather than reading as "nothing".
|
||||
raw = None
|
||||
if emitted is None:
|
||||
data = _capture(device, lambda: _try(lambda: device.lf.read(samples=_SAMPLE_COUNT), None))
|
||||
raw = demod_raw(data) if data else None
|
||||
|
||||
if config is None and emitted is None and raw is None:
|
||||
return None
|
||||
|
||||
chip = None
|
||||
@@ -169,15 +244,21 @@ def identify_lf(device, emit=None) -> dict | None:
|
||||
f" -> {emulating}")
|
||||
if emitted is not None:
|
||||
emit(f" emitted: {credential_label(emitted)}")
|
||||
elif raw is not None:
|
||||
emit(f" emitted (raw, no format decoded): {raw_label(raw)}")
|
||||
|
||||
# label: name the chip and what it's actually emitting (the reliable, decoded credential);
|
||||
# fall back to the config's emulation guess, or just the emitted credential.
|
||||
# fall back to the raw physical dump, then the config's emulation guess.
|
||||
if chip and emitted:
|
||||
label = f"{chip} ({credential_label(emitted)})"
|
||||
elif chip and raw:
|
||||
label = f"{chip} ({raw_label(raw)})"
|
||||
elif chip:
|
||||
label = f"{chip} ({emulating})"
|
||||
else:
|
||||
elif emitted:
|
||||
label = credential_label(emitted)
|
||||
else:
|
||||
label = raw_label(raw)
|
||||
|
||||
return {
|
||||
"found": True,
|
||||
@@ -187,5 +268,6 @@ def identify_lf(device, emit=None) -> dict | None:
|
||||
"config": config["config_hex"] if config else None,
|
||||
"emulating": emulating,
|
||||
"emitted": emitted,
|
||||
"raw": raw,
|
||||
"label": label,
|
||||
}
|
||||
|
||||
@@ -55,23 +55,32 @@ def _edges(binary: list[int]) -> list[int]:
|
||||
return [i for i in range(1, len(binary)) if binary[i] != binary[i - 1]]
|
||||
|
||||
|
||||
def detect_clock(binary: list[int], min_run: int = 3) -> float | None:
|
||||
def detect_clock(binary: list[int], min_run: int = 3, support: float = 0.2) -> float | None:
|
||||
"""Estimate the shortest symbol width (samples) from the edge-interval distribution.
|
||||
|
||||
For Manchester this is the *half-bit* (there is always a mid-bit transition, so runs are
|
||||
one or two half-bits); for plain OOK it is the bit itself. Runs shorter than ``min_run``
|
||||
are treated as noise/jitter and ignored. Returns the mean of the shortest interval cluster,
|
||||
or ``None`` when there aren't enough clean edges to decide."""
|
||||
or ``None`` when there aren't enough clean edges to decide.
|
||||
|
||||
The base symbol is the shortest interval with *real support* — a count at least ``support``×
|
||||
the most-common interval's count — not the raw minimum. A real ADC capture always carries a
|
||||
few stray sub-bit edges; keying off the bare minimum let a single 3-sample outlier collapse a
|
||||
clean RF/64 tag's estimate from 32 to ~3 (hardware-observed on a strongly-coupled EM4100),
|
||||
snapping the data clock to 8 and killing the demod. The dominant interval is robust to that."""
|
||||
edges = _edges(binary)
|
||||
if len(edges) < 4:
|
||||
return None
|
||||
intervals = [edges[i] - edges[i - 1] for i in range(1, len(edges)) if edges[i] - edges[i - 1] >= min_run]
|
||||
if len(intervals) < 3:
|
||||
return None
|
||||
intervals.sort()
|
||||
base = intervals[0]
|
||||
# average the cluster within 1.5x of the shortest interval -> the fundamental symbol width
|
||||
cluster = [d for d in intervals if d <= base * 1.5]
|
||||
counts: dict[int, int] = {}
|
||||
for d in intervals:
|
||||
counts[d] = counts.get(d, 0) + 1
|
||||
floor = max(counts.values()) * support
|
||||
base = min(d for d, n in counts.items() if n >= floor) # shortest well-supported interval
|
||||
# average the cluster around the base (0.75x–1.5x) -> the fundamental symbol width
|
||||
cluster = [d for d in intervals if base * 0.75 <= d <= base * 1.5]
|
||||
return sum(cluster) / len(cluster)
|
||||
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@ 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.hints import one_line_hint, help_rows, 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,
|
||||
@@ -189,24 +189,36 @@ class TestHints:
|
||||
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_help_rows_leaf(self):
|
||||
# a leaf's help as dropdown rows: (display, meta) — name, usage, each option, each note
|
||||
rows = help_rows(_tree().node_at(["hf", "mf", "rdbl"]))
|
||||
displays = [d for d, _ in rows]
|
||||
metas = [m for _, m in rows]
|
||||
assert ("hf mf rdbl", "Read MIFARE Classic block") == rows[0]
|
||||
assert any("--blk" in d for d in displays)
|
||||
assert any("--key <hex>" in d for d in displays)
|
||||
assert not any("This help" in d for d in displays) # -h/--help is skipped
|
||||
assert "usage" in metas # the usage row
|
||||
assert "hf mf rdbl --blk 0" in displays # a note row
|
||||
|
||||
def test_panel_for_leaf(self):
|
||||
node = _tree().node_at(["hf", "mf", "rdbl"])
|
||||
assert panel_for(node) == help_panel(node.entry)
|
||||
def test_help_rows_non_inserting_and_readable(self):
|
||||
# every row carries display + display_meta (the two-column dropdown form)
|
||||
rows = help_rows(_tree().node_at(["hf", "mf", "rdbl"]))
|
||||
assert all(isinstance(d, str) and d for d, _ in rows)
|
||||
assert all(isinstance(m, str) for _, m in rows)
|
||||
|
||||
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_help_rows_interior(self):
|
||||
rows = help_rows(_tree().node_at(["hf", "mf"]))
|
||||
displays = [d for d, _ in rows]
|
||||
assert displays[0] == "mf" # header row
|
||||
assert "rdbl" in displays and "wrbl" in displays
|
||||
|
||||
def test_help_rows_none(self):
|
||||
assert help_rows(None) == []
|
||||
|
||||
def test_resolve_target_subcommand(self):
|
||||
cur = SimpleNamespace(text="rdbl")
|
||||
# real subcommand completions carry a trailing space (Tab-descend mechanism)
|
||||
cur = SimpleNamespace(text="rdbl ")
|
||||
node = resolve_target(_tree(), "hf mf ", cur)
|
||||
assert node is not None and node.token == "rdbl" and node.is_leaf
|
||||
|
||||
@@ -216,30 +228,62 @@ class TestHints:
|
||||
assert node is not None and node.token == "rdbl"
|
||||
|
||||
def test_resolve_target_interior(self):
|
||||
cur = SimpleNamespace(text="mf")
|
||||
cur = SimpleNamespace(text="mf ")
|
||||
node = resolve_target(_tree(), "hf ", cur)
|
||||
assert node is not None and node.token == "mf" and not node.is_leaf
|
||||
|
||||
def test_resolve_target_after_menu_insertion(self):
|
||||
# arrowing onto a completion inserts it into the buffer (go_to_completion), so the
|
||||
# buffer already ends with the highlighted token — it must not be appended twice
|
||||
cur = SimpleNamespace(text="rdbl ")
|
||||
node = resolve_target(_tree(), "hf mf rdbl ", cur)
|
||||
assert node is not None and node.token == "rdbl" and node.is_leaf
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- keys
|
||||
|
||||
class TestKeys:
|
||||
def test_open_help_requires_selection(self):
|
||||
def test_open_help_requires_a_resolved_node(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
|
||||
assert open_help(s, None) is False # nothing highlighted → no help
|
||||
assert s.help_open is False and s.help_target is None
|
||||
node = s.tree.node_at(["hf", "mf", "rdbl"])
|
||||
assert open_help(s, node) is True
|
||||
assert s.help_open is True and s.help_target is node
|
||||
|
||||
def test_close_help(self):
|
||||
s = _session()
|
||||
s.help_open = True
|
||||
s.help_target = object()
|
||||
assert close_help(s) is False
|
||||
assert s.help_open is False
|
||||
assert s.help_open is False and s.help_target is None
|
||||
|
||||
def test_completer_yields_help_rows_when_open(self):
|
||||
# while help_open, the dropdown renders the target's help as non-inserting completions
|
||||
from prompt_toolkit.completion import CompleteEvent
|
||||
from prompt_toolkit.document import Document
|
||||
s = _session()
|
||||
s.help_open = True
|
||||
s.help_target = s.tree.node_at(["hf", "mf", "rdbl"])
|
||||
comps = list(ClientCompleter(s).get_completions(Document("hf mf rdbl ", 11), CompleteEvent()))
|
||||
assert comps and all(c.text == "" for c in comps) # selecting one inserts nothing
|
||||
shown = [_plain(c.display[0][1]) for c in comps]
|
||||
assert any("--blk" in d for d in shown)
|
||||
|
||||
def test_install_does_not_raise(self):
|
||||
install_key_bindings(KeyBindings(), _session())
|
||||
|
||||
def test_delete_keys_rebound_to_refresh_dropdown(self):
|
||||
# complete_while_typing only completes on insert, so the deletion keys must be rebound to
|
||||
# re-run completion — else the dropdown goes stale/blank when you backspace.
|
||||
kb = KeyBindings()
|
||||
install_key_bindings(kb, _session())
|
||||
bound = {str(k) for b in kb.bindings for k in b.keys}
|
||||
assert "Keys.ControlH" in bound # backspace
|
||||
assert "Keys.Delete" in bound # delete
|
||||
assert "Keys.ControlW" in bound # delete word
|
||||
assert "Keys.ControlU" in bound # delete to line start
|
||||
|
||||
def _buffer_with_menu(self, text):
|
||||
from prompt_toolkit.buffer import Buffer, CompletionState
|
||||
from prompt_toolkit.completion import CompleteEvent
|
||||
@@ -252,6 +296,32 @@ class TestKeys:
|
||||
completions=comps, complete_index=None)
|
||||
return buff
|
||||
|
||||
def test_left_restore_preserves_highlighted_index(self):
|
||||
# The RIGHT→LEFT round-trip must return the user to the completion they had highlighted.
|
||||
# Mechanism: help mode DETACHES the menu (complete_state = None) and re-attaches the same
|
||||
# object on LEFT. It must NOT use cancel_completion(), which calls go_to_completion(None)
|
||||
# and wipes the highlighted index — the bug that made Tab commit the wrong command.
|
||||
buff = self._buffer_with_menu("hf mf ")
|
||||
buff.go_to_completion(1) # DOWN onto the 2nd child (wrbl)
|
||||
doc, state, idx = buff.document, buff.complete_state, buff.complete_state.complete_index
|
||||
assert idx == 1 and buff.text == "hf mf wrbl "
|
||||
|
||||
buff.complete_state = None # RIGHT: detach without mutating `state`
|
||||
assert state.complete_index == idx # ← the invariant; cancel_completion breaks it
|
||||
|
||||
buff.document = doc # LEFT: restore exactly
|
||||
buff.complete_state = state
|
||||
assert buff.text == "hf mf wrbl " and buff.complete_state.complete_index == 1
|
||||
|
||||
def test_cancel_completion_would_lose_the_index(self):
|
||||
# Guard the contrast: prove cancel_completion() is the wrong tool here, so nobody swaps it
|
||||
# back in "to simplify".
|
||||
buff = self._buffer_with_menu("hf mf ")
|
||||
buff.go_to_completion(1)
|
||||
state = buff.complete_state
|
||||
buff.cancel_completion()
|
||||
assert state.complete_index is None # cancel wiped it — must not be used for RIGHT
|
||||
|
||||
def test_tab_commits_first_completion(self):
|
||||
# with nothing highlighted, Tab commits the FIRST completion (+ its trailing space)
|
||||
buff = self._buffer_with_menu("hf ")
|
||||
|
||||
@@ -20,6 +20,24 @@ def test_15_scan():
|
||||
assert "uid" in result
|
||||
|
||||
|
||||
def test_15_raw():
|
||||
from pm3py.core.protocol import crc15_bytes
|
||||
t = AsyncMock()
|
||||
iso15 = HF15Commands(t)
|
||||
resp_data = bytes([0x00, 0x00, 0x11, 0x22, 0x33, 0x44, 0x96, 0x3A]) # flags + data + crc
|
||||
t.send_ng.return_value = PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=0,
|
||||
reason=0, ng=True, data=resp_data)
|
||||
# send a raw request frame (no CRC); it should be CRC'd + wrapped, and the tag reply returned
|
||||
r = asyncio.get_event_loop().run_until_complete(iso15.raw(b"\x02\x20\x00"))
|
||||
assert r["status"] == 0 and r["raw"] == resp_data and r["data"] == resp_data.hex()
|
||||
payload = t.send_ng.call_args.args[1] # [flags(1)][len(2)][iso_cmd][crc15(2)]
|
||||
assert payload[3:6] == b"\x02\x20\x00" # the frame, verbatim
|
||||
assert payload[6:8] == crc15_bytes(b"\x02\x20\x00") # ISO15693 CRC appended host-side
|
||||
# a hex string is accepted too
|
||||
asyncio.get_event_loop().run_until_complete(iso15.raw("02 B2 04"))
|
||||
assert t.send_ng.call_args.args[1][3:6] == b"\x02\xB2\x04"
|
||||
|
||||
|
||||
# ---- Tracelog parsing ----
|
||||
|
||||
def _make_tracelog_entry(timestamp, duration, is_response, data):
|
||||
|
||||
@@ -57,6 +57,17 @@ def test_lf_t55_writebl():
|
||||
# PwdMode(0x01) | Page(0x02) | downlink 2<<3 (0x10) = 0x13
|
||||
assert flags == 0x13
|
||||
|
||||
def test_lf_t55_writebl_testmode():
|
||||
t = AsyncMock()
|
||||
lf = LFCommands(t)
|
||||
t.send_ng.return_value = make_response(Cmd.LF_T55XX_WRITEBL, 0, b"")
|
||||
asyncio.get_event_loop().run_until_complete(
|
||||
lf.t55.writebl(block=0, data=0x00148040, test=True))
|
||||
_, _, blockno, flags = struct.unpack("<IIBB", t.send_ng.call_args.args[1])
|
||||
assert blockno == 0
|
||||
assert flags & 0x04 # test-mode bit set (downlink opcode 01)
|
||||
assert not (flags & 0x01) # no password mode by default
|
||||
|
||||
def test_lf_t55_wakeup():
|
||||
t = AsyncMock()
|
||||
lf = LFCommands(t)
|
||||
@@ -69,6 +80,16 @@ def test_lf_t55_wakeup():
|
||||
assert pwd == 0x11223344
|
||||
assert flags == (3 << 3)
|
||||
|
||||
def test_lf_t55_reset():
|
||||
t = AsyncMock()
|
||||
lf = LFCommands(t)
|
||||
t.send_ng.return_value = make_response(Cmd.LF_T55XX_RESET_READ, 0, b"")
|
||||
asyncio.get_event_loop().run_until_complete(lf.t55.reset(downlink_mode=2))
|
||||
cmd, payload = t.send_ng.call_args.args[0], t.send_ng.call_args.args[1]
|
||||
assert cmd == Cmd.LF_T55XX_RESET_READ
|
||||
# single flags byte carrying downlink_mode<<3
|
||||
assert payload == struct.pack("<B", 2 << 3)
|
||||
|
||||
def test_lf_t55_config_client_side():
|
||||
t = AsyncMock()
|
||||
lf = LFCommands(t)
|
||||
|
||||
@@ -8,6 +8,7 @@ from unittest.mock import MagicMock
|
||||
|
||||
from pm3py.lf import dsp, fsk, demod_samples, read_config, identify_lf
|
||||
from pm3py.lf import protocols
|
||||
from pm3py.lf import capture
|
||||
from pm3py.transponders.lf.em.em4100 import EM4100Tag
|
||||
from pm3py.transponders.lf.hid.hid import HIDProxTag
|
||||
from pm3py.transponders.lf.atmel.t5577 import T5577Config, CONFIG_EM4100
|
||||
@@ -88,6 +89,19 @@ class TestDSP:
|
||||
half = dsp.detect_clock(dsp.binarize(env))
|
||||
assert half is not None and abs(half - 32) < 4 # RF/64 -> 32-sample half-bit
|
||||
|
||||
def test_clock_survives_sparse_stray_edges(self):
|
||||
# hardware regression: a strongly-coupled EM4100 gives a clean RF/64 signal (intervals
|
||||
# dominated by 32/64) with a few sub-bit stray edges from ADC ripple. The old min-based
|
||||
# estimator keyed off the bare shortest interval and collapsed to ~3 (-> data clock 8 ->
|
||||
# dead demod). The dominant interval is still 32, so detect_clock must ride through it.
|
||||
env = bytearray(_em4100_env(0x0102030405, rf=64, high=255, low=0))
|
||||
for i in range(200, len(env), 900): # sparse single-sample spikes
|
||||
env[i] = 0 if env[i] > 128 else 255
|
||||
half = dsp.detect_clock(dsp.binarize(bytes(env)))
|
||||
assert half is not None and abs(half - 32) < 4 # not dragged down to the ~3 outlier
|
||||
r = protocols.decode_em4100(bytes(env))
|
||||
assert r and r["id_hex"] == "0102030405" # end-to-end decode survives
|
||||
|
||||
def test_manchester_roundtrip(self):
|
||||
# (1,0)->1, (0,1)->0 at phase 0
|
||||
assert dsp.manchester_decode([1, 0, 0, 1, 1, 0]) == [1, 0, 1]
|
||||
@@ -259,6 +273,71 @@ def _fdxb_env(country, national, animal=0, **kw):
|
||||
return _biphase_env(_fdxb_bits(country, national, animal), **kw)
|
||||
|
||||
|
||||
# --- NRZ / raw physical-layer fallback --------------------------------------------------------
|
||||
|
||||
def _nrz_env(databits, rf=32, high=200, low=60, repeats=8):
|
||||
"""Direct/NRZ ASK envelope: each data bit is output as-is, held for one bit-width (rf
|
||||
samples). The inverse of ask.nrz_bits."""
|
||||
env = []
|
||||
for d in databits * repeats:
|
||||
env += [high if d else low] * rf
|
||||
return bytes(env)
|
||||
|
||||
|
||||
def _cyclic_match(bitstr, frame) -> bool:
|
||||
"""Is ``bitstr`` a rotation of ``frame`` or its bitwise inverse? Raw demod carries polarity
|
||||
and start-of-frame ambiguity, so cyclic-plus-invert is the right equivalence."""
|
||||
f = "".join(str(b) for b in frame)
|
||||
if len(bitstr) != len(f):
|
||||
return False
|
||||
inv = "".join("1" if c == "0" else "0" for c in f)
|
||||
return bitstr in (f + f) or bitstr in (inv + inv)
|
||||
|
||||
|
||||
class TestRawDemod:
|
||||
# a 32-bit frame with no short internal period (so the recovered frame length is 32)
|
||||
FRAME = [1, 0, 1, 1, 0, 0, 1, 0, 1, 1, 1, 0, 0, 0, 1, 0,
|
||||
0, 1, 1, 0, 1, 0, 0, 1, 1, 1, 0, 1, 0, 0, 0, 1]
|
||||
|
||||
def test_nrz_round_trips(self):
|
||||
r = capture.demod_raw(_nrz_env(self.FRAME, rf=32))
|
||||
assert r and "nrz" in r["encodings"]
|
||||
enc = r["encodings"]["nrz"]
|
||||
assert enc["period"] == 32
|
||||
assert _cyclic_match(enc["bits"], self.FRAME)
|
||||
assert r["rf"] == "RF/32"
|
||||
|
||||
def test_nrz_various_rates(self):
|
||||
for rf in (32, 64):
|
||||
r = capture.demod_raw(_nrz_env(self.FRAME, rf=rf))
|
||||
assert r and _cyclic_match(r["encodings"]["nrz"]["bits"], self.FRAME)
|
||||
|
||||
def test_biphase_round_trips(self):
|
||||
r = capture.demod_raw(_biphase_env(self.FRAME, rf=32))
|
||||
assert r and "biphase" in r["encodings"]
|
||||
assert _cyclic_match(r["encodings"]["biphase"]["bits"], self.FRAME)
|
||||
|
||||
def test_noise_returns_none(self):
|
||||
assert capture.demod_raw(bytes([128] * 4000)) is None
|
||||
assert capture.demod_raw(b"") is None
|
||||
|
||||
def test_raw_is_below_credentials(self):
|
||||
# a genuine EM4100 must decode as a credential; the raw fallback is only for the unnamed
|
||||
assert demod_samples(_em4100_env(0x0102030405))["protocol"] == "EM4100"
|
||||
|
||||
def test_generators_yield_bits(self):
|
||||
from pm3py.lf import ask
|
||||
assert any(len(b) >= 32 for b in ask.nrz_bits(_nrz_env(self.FRAME, rf=32)))
|
||||
assert any(len(b) >= 32 for b in ask.biphase_bits(_biphase_env(self.FRAME, rf=32)))
|
||||
|
||||
def test_rejects_constant_tail(self):
|
||||
# hardware regression: biphase-decoding a non-biphase tag emits a short prefix then a long
|
||||
# constant run, which trivially "repeats" at any period. That must NOT report a frame.
|
||||
assert capture._frame_period([0, 0, 0, 1, 0, 1, 0, 1] + [1] * 200) is None
|
||||
assert capture._frame_period([1] * 200) is None # fully constant
|
||||
assert capture._frame_period(([0, 1] * 100)) is None # alternating fill, sub-periodic
|
||||
|
||||
|
||||
class TestCRC16:
|
||||
def test_known_vectors(self):
|
||||
# "123456789" -> CRC-16/XMODEM 0x31C3, CRC-16/KERMIT 0x2189 (anchors the CRC impl)
|
||||
|
||||
@@ -11,7 +11,8 @@ from pm3py.cli.rawcli.trace_view import render_exchange
|
||||
from pm3py.cli.rawcli.parser import parse_token, parse_bytes, parse_line
|
||||
from pm3py.cli.rawcli.entry import type_digit, is_byte_line, reconcile_bases
|
||||
from pm3py.cli.rawcli.identify import identify, clear, name_14a, format_summary
|
||||
from pm3py.cli.rawcli.catalog import TYPE2, ISO15, T5577, LF_READONLY, MFC, catalog_for, catalog_for as _cf
|
||||
from pm3py.cli.rawcli.catalog import (TYPE2, ISO15, T5577, LF_READONLY, MFC, _SLIX, _SLIX2, _NTAG5,
|
||||
iso15_frame, iso15_flags_decode, catalog_for, catalog_for as _cf)
|
||||
from pm3py.cli.rawcli.tlv import parse_tlv, format_tlv
|
||||
from pm3py.cli.rawcli.completer import RawCompleter
|
||||
from pm3py.cli.rawcli.app import dispatch
|
||||
@@ -249,9 +250,16 @@ class TestIdentify:
|
||||
assert page_role(s.transponder, 4) == "user memory"
|
||||
|
||||
def test_15693_when_e0_uid(self):
|
||||
# E0 04 = NXP ICODE; no READ_SIGNATURE (mock) -> named ICODE SLIX, not generic ISO15693
|
||||
s = RawSession(device=_mock_device(scan15={"found": True, "uid": "e004010203040506"}))
|
||||
r = identify(s)
|
||||
assert r["found"] and s.field == "hf" and s.protocol == "hf15"
|
||||
assert s.transponder == "ICODE SLIX"
|
||||
|
||||
def test_15693_non_nxp_generic(self):
|
||||
# a non-NXP E0 UID stays generic ISO15693
|
||||
s = RawSession(device=_mock_device(scan15={"found": True, "uid": "e007010203040506"}))
|
||||
identify(s)
|
||||
assert s.transponder == "ISO15693"
|
||||
|
||||
def test_15693_rejects_bogus_uid(self):
|
||||
@@ -260,6 +268,34 @@ class TestIdentify:
|
||||
r = identify(s)
|
||||
assert r["found"] is False and s.transponder is None
|
||||
|
||||
def test_name_iso15_ntag5_dna_slix2_slix(self):
|
||||
# split the E0 04 family by READ_CONFIG (0xC0) probe, using the real hardware captures:
|
||||
# NTAG5 -> session reg 0xA0 succeeds; DNA -> 0xA0 errors but config 0x00 succeeds;
|
||||
# SLIX2 -> no READ_CONFIG (timeout) but READ_SIGNATURE answers; SLIX -> neither.
|
||||
from pm3py.cli.rawcli.identify import name_iso15
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
def dev_with(responses):
|
||||
dev = MagicMock()
|
||||
def _raw(frame):
|
||||
f = bytes(frame)
|
||||
if f[:2] == b"\x22\xc0": # READ_CONFIG addressed: key by block
|
||||
return responses.get(("cfg", f[-2]), {"raw": None})
|
||||
if f[:3] == b"\x02\xbd\x04": # READ_SIGNATURE
|
||||
return responses.get("sig", {"raw": None})
|
||||
return {"raw": None}
|
||||
dev.hf.iso15.raw.side_effect = _raw
|
||||
return dev
|
||||
|
||||
ntag5 = dev_with({("cfg", 0xA0): {"raw": bytes.fromhex("0001400000bad5")}})
|
||||
assert name_iso15(ntag5, "e0040158108ee802") == "NTAG 5"
|
||||
dna = dev_with({("cfg", 0xA0): {"raw": bytes.fromhex("010f68ee")},
|
||||
("cfg", 0x00): {"raw": bytes.fromhex("001e4d3235a1b8")}})
|
||||
assert name_iso15(dna, "e0040118009b4ccb") == "ICODE DNA"
|
||||
slix2 = dev_with({"sig": {"raw": bytes([0x00]) + bytes(34)}}) # 35-byte signature reply
|
||||
assert name_iso15(slix2, "e0040203040506") == "ICODE SLIX2"
|
||||
assert name_iso15(dev_with({}), "e004a2110b37c06b") == "ICODE SLIX"
|
||||
|
||||
def test_no_flat_lf(self):
|
||||
# nothing on HF and a flat LF envelope must report not-found, not a bogus "LF tag"
|
||||
assert identify(RawSession(device=_mock_device()))["found"] is False
|
||||
@@ -418,8 +454,90 @@ class TestCatalog:
|
||||
assert "not yet wired" in MFC.get("PERSONALIZE_UID").run(dev, "2") # EV1 opcode-only
|
||||
|
||||
def test_iso15_builds(self):
|
||||
# 15693 frames are flags | command | [mfg] | params, assembled by iso15_frame
|
||||
assert ISO15.get("READ_BLOCK").build("4") == b"\x02\x20\x04"
|
||||
assert ISO15.get("INVENTORY").build() == b"\x26\x01\x00"
|
||||
assert ISO15.get("GET_SYSTEM_INFO").build() == b"\x02\x2B"
|
||||
assert ISO15.get("READ_BLOCK").opcode() == 0x20 # opcode = the command byte, not flags
|
||||
|
||||
def test_iso15_frame_builder(self):
|
||||
assert iso15_frame(0x20, bytes([4])) == b"\x02\x20\x04" # non-addressed, high rate
|
||||
assert iso15_frame(0xB2, mfg=0x04) == b"\x02\xB2\x04" # NXP custom -> mfg byte
|
||||
# addressed: flags 0x22, UID appended LSByte-first
|
||||
f = iso15_frame(0x20, bytes([4]), uid="e004010203040506", addressed=True)
|
||||
assert f == bytes.fromhex("2220") + bytes.fromhex("e004010203040506")[::-1] + bytes([4])
|
||||
assert iso15_frame(0x20, bytes([4]), option=True) == b"\x42\x20\x04"
|
||||
# full flags coverage: every request-flag bit is settable
|
||||
assert iso15_frame(0x20, bytes([4]), select=True) == b"\x12\x20\x04" # 0x10 select
|
||||
assert iso15_frame(0x20, bytes([4]), proto_ext=True) == b"\x0A\x20\x04" # 0x08 proto-ext
|
||||
assert iso15_frame(0x20, bytes([4]), two_subcarrier=True) == b"\x03\x20\x04" # 0x01
|
||||
assert iso15_frame(0x20, bytes([4]), high_rate=False) == b"\x00\x20\x04" # low rate
|
||||
assert iso15_frame(0x01, inventory=True, slots16=True) == b"\x06\x01" # 16-slot inventory
|
||||
# inventory AFI: flag 0x10 set + AFI byte inserted before the params (mask/blocks)
|
||||
assert iso15_frame(0x01, bytes([0]), inventory=True, afi=0x9A) == bytes.fromhex("36019a00")
|
||||
|
||||
def test_iso15_flags_decode(self):
|
||||
assert iso15_flags_decode(0x02) == "high rate · non-addressed"
|
||||
assert iso15_flags_decode(0x22) == "high rate · addressed"
|
||||
assert "option" in iso15_flags_decode(0x42)
|
||||
assert "inventory" in iso15_flags_decode(0x26)
|
||||
# every bit decodes, in both modes
|
||||
assert "select" in iso15_flags_decode(0x12)
|
||||
assert "protocol-ext" in iso15_flags_decode(0x0A)
|
||||
assert "two subcarrier" in iso15_flags_decode(0x03)
|
||||
assert "16 slots" in iso15_flags_decode(0x06)
|
||||
assert "AFI" in iso15_flags_decode(0x36)
|
||||
assert iso15_flags_decode(0x00).startswith("low rate")
|
||||
|
||||
def test_slix_catalog(self):
|
||||
# ICODE SLIX datasheet command set: standard + NXP custom, frames via the header builder
|
||||
assert _SLIX.get("READ_BLOCK").build("4") == b"\x02\x20\x04"
|
||||
assert _SLIX.get("GET_RANDOM").build() == b"\x02\xB2\x04" # NXP mfg inserted
|
||||
assert _SLIX.get("SET_PASSWORD").build("4", "DEADBEEF") == bytes.fromhex("02b30404deadbeef")
|
||||
assert _SLIX.get("READ_MULTIPLE").build("0", "4") == b"\x02\x23\x00\x03" # count-1 on wire
|
||||
# opcode = the command byte (frame byte 1), so by_opcode maps correctly
|
||||
assert _SLIX.get("GET_RANDOM").opcode() == 0xB2
|
||||
assert _SLIX.by_opcode(0xB2).name == "GET_RANDOM" and _SLIX.by_opcode(0x20).name == "READ_BLOCK"
|
||||
# INVENTORY READ family: inventory flags (0x26) | 0xA0/0xA1 | mfg | mask_len 0 | first | count-1
|
||||
assert _SLIX.get("INVENTORY_READ").build("0", "4") == bytes.fromhex("26a004000003")
|
||||
assert _SLIX.get("FAST_INVENTORY_READ").build("2", "1") == bytes.fromhex("26a104000200")
|
||||
assert _SLIX.by_opcode(0xA0).name == "INVENTORY_READ"
|
||||
assert _SLIX.by_opcode(0xA1).name == "FAST_INVENTORY_READ"
|
||||
# full body: 16-bit mask (selective anticollision), and AFI filter (flag 0x10 -> flags 0x36)
|
||||
assert _SLIX.get("INVENTORY_READ").build("0", "4", "16", "abcd") == bytes.fromhex("26a00410abcd0003")
|
||||
assert _SLIX.get("INVENTORY_READ").build("0", "4", afi="0x9a") == bytes.fromhex("36a0049a000003")
|
||||
# dispatch: an identified SLIX (or generic ICODE) resolves to this catalog
|
||||
s = RawSession(); s.protocol, s.transponder = "hf15", "ICODE SLIX"
|
||||
assert catalog_for(s) is _SLIX
|
||||
|
||||
def test_slix2_catalog(self):
|
||||
# SLIX2 = the full SLIX datasheet set plus three SLIX2-only commands (0xBD/0xBB/0xBC)
|
||||
assert set(_SLIX.commands) <= set(_SLIX2.commands) # superset of SLIX
|
||||
for name, op in [("READ_SIGNATURE", 0xBD), ("PASSWORD_PROTECTION_64BIT", 0xBB),
|
||||
("STAY_QUIET_PERSISTENT", 0xBC)]:
|
||||
assert name not in _SLIX.commands # SLIX2-only
|
||||
assert _SLIX2.get(name).opcode() == op
|
||||
assert _SLIX2.by_opcode(op).name == name # raw opcode maps back
|
||||
assert _SLIX2.get("READ_SIGNATURE").build() == b"\x02\xBD\x04" # NXP mfg inserted
|
||||
assert _SLIX2.by_opcode(0xA0).name == "INVENTORY_READ" # inherited from SLIX
|
||||
# dispatch: SLIX2 -> _SLIX2, plain SLIX still -> _SLIX (SLIX2 test comes first)
|
||||
s = RawSession(); s.protocol = "hf15"
|
||||
s.transponder = "ICODE SLIX2"; assert catalog_for(s) is _SLIX2
|
||||
s.transponder = "ICODE SLIX"; assert catalog_for(s) is _SLIX
|
||||
|
||||
def test_ntag5_catalog(self):
|
||||
# NTAG5 = SLIX2 minus STAY_QUIET_PERSISTENT (rejected), plus config / SRAM / pick-random-UID
|
||||
assert "STAY_QUIET_PERSISTENT" not in _NTAG5.commands
|
||||
assert "READ_SIGNATURE" in _NTAG5.commands # inherited from SLIX2
|
||||
for name, op in [("READ_CONFIG", 0xC0), ("WRITE_CONFIG", 0xC1), ("PICK_RANDOM_UID", 0xC2),
|
||||
("READ_SRAM", 0xD2), ("WRITE_SRAM", 0xD3)]:
|
||||
assert _NTAG5.get(name).opcode() == op
|
||||
assert _NTAG5.by_opcode(op).name == name
|
||||
assert _NTAG5.get("READ_CONFIG").build("0xA1", "1") == bytes.fromhex("02c004a100") # CONFIG_REG
|
||||
assert _NTAG5.get("WRITE_CONFIG").build("0xA1", "20000000") == bytes.fromhex("02c104a120000000")
|
||||
assert _NTAG5.get("PICK_RANDOM_UID").build() == b"\x02\xC2\x04"
|
||||
# dispatch (checked before SLIX2/SLIX): an identified NTAG5 resolves here
|
||||
s = RawSession(); s.protocol, s.transponder = "hf15", "NTAG 5"
|
||||
assert catalog_for(s) is _NTAG5
|
||||
|
||||
def test_resolver(self):
|
||||
s = RawSession()
|
||||
@@ -446,8 +564,12 @@ class TestCatalog:
|
||||
assert T5577.get("READ").build("0")[0] == 0x01
|
||||
assert T5577.get("WRITE").build("0", "0")[0] == 0x02
|
||||
assert T5577.get("WAKE").build("0")[0] == 0x03
|
||||
assert {c.name for c in T5577.commands.values()} == {"READ", "WRITE", "WAKE", "DETECT"}
|
||||
assert T5577.get("RESET").build()[0] == 0x00 # reset command = downlink opcode 00
|
||||
assert {c.name for c in T5577.commands.values()} == {"READ", "WRITE", "WAKE", "RESET", "DETECT"}
|
||||
assert T5577.get("READ").run is not None and LF_READONLY.get("INFO").run is not None
|
||||
# READ/WRITE take optional page + password; build() tolerates the extra args
|
||||
assert T5577.get("READ").build("4", "1", "0xAABBCCDD")[0] == 0x01
|
||||
assert T5577.get("WRITE").build("4", "12345678", "1", "0xAABBCCDD")[0] == 0x02
|
||||
|
||||
def test_opcode_covers_two_phase_write(self):
|
||||
# opcode() tolerates data args (which need valid hex) and multi-frame builds
|
||||
@@ -470,6 +592,52 @@ class TestCatalog:
|
||||
assert TYPE2.by_opcode(0x99) is None
|
||||
|
||||
|
||||
class TestT5577Run:
|
||||
"""The T5577 run() functions thread page/password through to the lf.t55 device methods."""
|
||||
|
||||
def test_write_threads_page_and_password(self):
|
||||
dev = MagicMock()
|
||||
dev.lf.t55.writebl.return_value = {"status": 0}
|
||||
out = T5577.get("WRITE").run(dev, "4", "0x12345678", "1", "0xAABBCCDD")
|
||||
dev.lf.t55.writebl.assert_called_once_with(4, 0x12345678, page=1, password=0xAABBCCDD)
|
||||
assert "block 4 (page 1)" in out and "ok" in out and "AABBCCDD" in out
|
||||
|
||||
def test_write_defaults_are_page0_no_password(self):
|
||||
dev = MagicMock()
|
||||
dev.lf.t55.writebl.return_value = {"status": 0}
|
||||
T5577.get("WRITE").run(dev, "4", "0x12345678")
|
||||
dev.lf.t55.writebl.assert_called_once_with(4, 0x12345678, page=0, password=None)
|
||||
|
||||
def test_write_data_and_password_are_hex(self):
|
||||
# data/password are hex words (like pm3 `lf t55 write -d/-p`), no 0x needed
|
||||
dev = MagicMock()
|
||||
dev.lf.t55.writebl.return_value = {"status": 0}
|
||||
T5577.get("WRITE").run(dev, "0", "00088040", "0", "50524F58")
|
||||
dev.lf.t55.writebl.assert_called_once_with(0, 0x00088040, page=0, password=0x50524F58)
|
||||
|
||||
def test_reset_sends_reset(self):
|
||||
dev = MagicMock()
|
||||
dev.lf.t55.reset.return_value = {"status": 0}
|
||||
assert "ok" in T5577.get("RESET").run(dev)
|
||||
dev.lf.t55.reset.assert_called_once_with()
|
||||
|
||||
def test_read_config_forwards_password(self):
|
||||
# block 0 -> read_config; a password reads in password mode (readbl(0, password=...))
|
||||
dev = MagicMock()
|
||||
dev.lf.t55.readbl.return_value = None # no envelope -> "no clean config"
|
||||
dev.lf.download_samples.return_value = b""
|
||||
out = T5577.get("READ").run(dev, "0", "0", "0x11223344")
|
||||
dev.lf.t55.readbl.assert_called_with(0, password=0x11223344)
|
||||
assert "11223344" in out
|
||||
|
||||
def test_read_block_forwards_page_and_password(self):
|
||||
dev = MagicMock()
|
||||
dev.lf.t55.readbl.return_value = None
|
||||
dev.lf.download_samples.return_value = b""
|
||||
T5577.get("READ").run(dev, "2", "1", "0x11223344")
|
||||
dev.lf.t55.readbl.assert_called_with(2, page=1, password=0x11223344)
|
||||
|
||||
|
||||
class TestFunctionCalls:
|
||||
def _id(self, sak=0x00):
|
||||
return RawSession(device=_mock_device(
|
||||
@@ -615,6 +783,24 @@ class TestCompleter:
|
||||
gv = next(c for c in cs if "GET_VERSION" in self._disp(c))
|
||||
assert gv.text == "01100000" # raw binary byte, not the command name
|
||||
|
||||
def test_iso15_position_aware_completion(self):
|
||||
# 15693 raw frame: byte 0 = flags menu (decoded), byte 1 = command, byte 2 = NXP mfg
|
||||
s = RawSession()
|
||||
s.protocol, s.transponder = "hf15", "ICODE SLIX"
|
||||
flags = {c.text: c.display_meta_text for c in self._complete(s, "")}
|
||||
assert flags["02"] == "high rate · non-addressed"
|
||||
assert flags["22"] == "high rate · addressed"
|
||||
cmds = {c.text: self._disp(c) for c in self._complete(s, "02 ")} # after flags -> command
|
||||
assert "READ_BLOCK" in cmds["20"] and "GET_RANDOM" in cmds["B2"]
|
||||
assert "04" in [c.text for c in self._complete(s, "02 B2 ")] # custom -> mfg byte
|
||||
assert self._complete(s, "02 20 ") == [] # standard cmd -> no mfg
|
||||
# function-call form still gets the block map, command names still complete by letters
|
||||
assert any("user memory" in (c.display_meta_text or "") for c in self._complete(s, "READ_BLOCK("))
|
||||
assert any("GET_RANDOM" in self._disp(c) for c in self._complete(s, "get_r"))
|
||||
# binary flags render as 8 bits
|
||||
s.entry_mode = "bin"
|
||||
assert any(c.text == "00000010" for c in self._complete(s, ""))
|
||||
|
||||
def test_mfc_opcode_completion(self):
|
||||
# the new MIFARE Classic datasheet opcodes surface by hex AND binary value, inserting the byte
|
||||
s = RawSession()
|
||||
@@ -831,6 +1017,26 @@ class TestKeyBindings:
|
||||
toggle.handler(event)
|
||||
assert s.entry_mode == "bin" and buf.text == "30" # mode flipped, buffer intact
|
||||
|
||||
def test_toggle_updates_breadcrumb_prefix(self):
|
||||
# the toggle flips the breadcrumb entry prefix 0x <-> 0b and requests a redraw
|
||||
from types import SimpleNamespace
|
||||
from prompt_toolkit.key_binding import KeyBindings
|
||||
from prompt_toolkit.buffer import Buffer
|
||||
from pm3py.cli.rawcli.entry import install_key_bindings
|
||||
s = RawSession()
|
||||
kb = KeyBindings()
|
||||
install_key_bindings(kb, s)
|
||||
toggle = next(b for b in kb.bindings
|
||||
if getattr(b.keys[0], "value", b.keys[0]) in ("c-t", "c-_"))
|
||||
redraws = []
|
||||
ev = SimpleNamespace(current_buffer=Buffer(), data="",
|
||||
app=SimpleNamespace(invalidate=lambda: redraws.append(1)))
|
||||
assert "0x" in s.breadcrumb() and "0b" not in s.breadcrumb()
|
||||
toggle.handler(ev)
|
||||
assert "0b" in s.breadcrumb() and s.entry_prefix == "0b" and redraws # flipped + redraw
|
||||
toggle.handler(ev)
|
||||
assert "0x" in s.breadcrumb() and "0b" not in s.breadcrumb() # and back
|
||||
|
||||
def test_digit_binding_tracks_per_byte_base(self):
|
||||
# replay keystrokes through the real digit binding: 30 (hex) then binary -> 30 00000100
|
||||
from types import SimpleNamespace
|
||||
@@ -853,3 +1059,88 @@ class TestKeyBindings:
|
||||
for ch in "00000100":
|
||||
press(ch)
|
||||
assert buf.text == "30 00000100" and s.byte_bases == ["hex", "bin"]
|
||||
|
||||
def test_digit_binding_fires_completion_trigger_for_raw_bytes(self):
|
||||
# regression: raw byte entry must append via insert_text (which fires completion), not
|
||||
# replace buf.document (which resets complete_state and silently killed the live
|
||||
# autocomplete menu during hex/binary byte entry). on_text_insert fires from insert_text,
|
||||
# never from a document= swap, so it is a faithful proxy for "the menu would refresh".
|
||||
from types import SimpleNamespace
|
||||
from prompt_toolkit.key_binding import KeyBindings
|
||||
from prompt_toolkit.buffer import Buffer
|
||||
from pm3py.cli.rawcli.entry import install_key_bindings
|
||||
s = RawSession()
|
||||
s.toggle_entry_mode() # binary: regroups every 8 digits (worst case)
|
||||
kb = KeyBindings()
|
||||
install_key_bindings(kb, s)
|
||||
digit = {getattr(b.keys[0], "value", b.keys[0]): b for b in kb.bindings}
|
||||
buf = Buffer()
|
||||
fired = []
|
||||
buf.on_text_insert += lambda _b: fired.append(1)
|
||||
app = SimpleNamespace(invalidate=lambda: None)
|
||||
for ch in "00000100":
|
||||
digit[ch].handler(SimpleNamespace(current_buffer=buf, app=app, data=ch))
|
||||
assert buf.text == "00000100"
|
||||
assert len(fired) == 8 # one insert_text per digit (0 with document=)
|
||||
|
||||
|
||||
class TestRawcliLiveSession:
|
||||
"""End-to-end through the REAL PromptSession run() builds (real key bindings + RawCompleter +
|
||||
complete_while_typing), driven by a prompt_toolkit pipe input that simulates TTY keystrokes.
|
||||
Exercises the actual rawcli input stack; no device needed for the input layer."""
|
||||
|
||||
@staticmethod
|
||||
def _run_keys(keys):
|
||||
import asyncio
|
||||
from prompt_toolkit import PromptSession
|
||||
from prompt_toolkit.input import create_pipe_input
|
||||
from prompt_toolkit.output import DummyOutput
|
||||
from prompt_toolkit.key_binding import KeyBindings
|
||||
from pm3py.cli.rawcli.entry import install_key_bindings
|
||||
from pm3py.cli.rawcli.completer import RawCompleter
|
||||
state = RawSession()
|
||||
comp = RawCompleter(state)
|
||||
queried = []
|
||||
_orig = comp.get_completions
|
||||
def _rec(document, event):
|
||||
queried.append(document.text)
|
||||
yield from _orig(document, event)
|
||||
comp.get_completions = _rec
|
||||
kb = KeyBindings()
|
||||
install_key_bindings(kb, state)
|
||||
# PromptSession.prompt() runs asyncio.run() internally, which nulls the current-loop
|
||||
# pointer; save/restore it so the shared-loop `_run` convention in other test files isn't
|
||||
# poisoned (see tests/test_pyws_plugin.py).
|
||||
try:
|
||||
prev_loop = asyncio.get_event_loop()
|
||||
except RuntimeError:
|
||||
prev_loop = None
|
||||
try:
|
||||
with create_pipe_input() as inp:
|
||||
session = PromptSession(key_bindings=kb, completer=comp, complete_while_typing=True,
|
||||
input=inp, output=DummyOutput())
|
||||
inp.send_text(keys + "\r")
|
||||
line = session.prompt("> ")
|
||||
finally:
|
||||
if prev_loop is not None and not prev_loop.is_closed():
|
||||
asyncio.set_event_loop(prev_loop)
|
||||
return line, queried
|
||||
|
||||
def test_accepted_line_across_input_paths(self):
|
||||
# the change touched the digit handler; confirm every input path still yields the right line
|
||||
toggle = "\x14" # Ctrl-t entry-mode toggle
|
||||
for keys, expect in [("3004", "30 04"),
|
||||
(toggle + "00000100", "00000100"),
|
||||
("30" + toggle + "00000100", "30 00000100"),
|
||||
("read(4)", "read(4)"),
|
||||
("identify", "identify")]:
|
||||
line, _ = self._run_keys(keys)
|
||||
assert line == expect, f"{keys!r} -> {line!r} != {expect!r}"
|
||||
|
||||
def test_live_completion_fires_during_binary_entry(self):
|
||||
# the fix, in the real session: the completer is queried with the byte-line text while
|
||||
# typing binary (0 queries before the fix, since buf.document= reset complete_state)
|
||||
line, queried = self._run_keys("\x14" + "00000100")
|
||||
byte_q = [q for q in queried if q and q.replace(" ", "").strip("01") == ""]
|
||||
assert line == "00000100"
|
||||
assert byte_q, "completer never queried for byte-line text — live menu not firing"
|
||||
|
||||
@@ -186,6 +186,21 @@ def test_read_response_skips_multiple_debug_then_reply():
|
||||
assert resp.cmd == int(Cmd.PING) and resp.data == b"pong"
|
||||
|
||||
|
||||
def test_read_response_skips_wtx_frame_not_returns_it():
|
||||
# a CMD_WTX "wait N ms" control frame (firmware sends it on FPGA bitstream reload) must be
|
||||
# skipped like a debug frame — returning it as the reply is what permanently desynced every
|
||||
# subsequent 15693 read (each command then grabbed the previous command's response)
|
||||
from unittest.mock import AsyncMock as _AM
|
||||
t = PM3Transport("/dev/null")
|
||||
t._reader = object()
|
||||
wtx = _resp_frame(int(Cmd.WTX), struct.pack("<H", 1500)) # FPGA_LOAD_WAIT_TIME
|
||||
reply = _resp_frame(int(Cmd.HF_ISO15693_COMMAND), b"\x00\x0f\x6b\xc0")
|
||||
t._read_response_frame = _AM(side_effect=[wtx, reply])
|
||||
resp = _run(t.read_response())
|
||||
assert resp.cmd == int(Cmd.HF_ISO15693_COMMAND)
|
||||
assert resp.data == b"\x00\x0f\x6b\xc0" # the real tag reply, not the WTX frame
|
||||
|
||||
|
||||
def test_read_timeout_sets_resync_next_send_drains():
|
||||
# a read timeout flags a resync; the next send drains the late response before writing
|
||||
import asyncio as _a
|
||||
|
||||
Reference in New Issue
Block a user