Compare commits
7 Commits
4684d0cc83
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
907522f0ae | ||
|
|
a990d725ef | ||
|
|
e3a394859c | ||
|
|
4226c9a11c | ||
|
|
8833753c45 | ||
|
|
8f136e7dc4 | ||
|
|
218e6dfc2a |
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.
|
||||
@@ -41,7 +41,7 @@ rendered dropdown on a real terminal is the only thing left unautomated.
|
||||
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 (next)
|
||||
## After MFC — 15693 vicinity chips: ICODE SLIX ✅, SLIX2 ✅, NTAG5 ✅, ICODE DNA (next)
|
||||
|
||||
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,
|
||||
@@ -68,6 +68,26 @@ count-1`, with the inventory flags set by `iso15_frame(inventory=…, afi=…)`.
|
||||
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.
|
||||
|
||||
|
||||
@@ -460,40 +460,64 @@ _NTAG5 = _catalog(
|
||||
# ---- 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):
|
||||
@@ -502,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 ----
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -564,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
|
||||
@@ -588,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(
|
||||
|
||||
Reference in New Issue
Block a user