diff --git a/pm3py/lf/capture.py b/pm3py/lf/capture.py index 4d12941..5e8c651 100644 --- a/pm3py/lf/capture.py +++ b/pm3py/lf/capture.py @@ -50,7 +50,34 @@ def demod_samples(samples) -> dict | None: return None -def _download(device, count: int = _SAMPLE_COUNT) -> bytes: +# 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. +_DOWNLOAD_BYTES = 12000 + +# markers that a "sample" buffer is actually desynced BigBuf memory, not an envelope +_POISON = b"\xef\xbe\xad\xde" # 0xDEADBEEF — firmware stack canary +_RESP_MAGIC = b"\x50\x4d\x33\x62" # 0x62334d50 — a PM3 response frame sitting in the buffer + + +def _try(fn, default): + try: + return fn() + except Exception: + return default + + +def _looks_like_samples(data: bytes) -> bool: + """Reject a download that is clearly desynced BigBuf memory rather than an LF envelope: the + DEADBEEF stack canary, an embedded PM3 response magic, or a mostly-zero (unwritten) buffer.""" + if len(data) < 256: + return False + if _POISON in data or _RESP_MAGIC in data: + return False + return data.count(0) < len(data) // 2 + + +def _download(device, count: int = _DOWNLOAD_BYTES) -> bytes: """Pull the last capture's envelope out of the device BigBuf (best-effort -> b'').""" try: return device.lf.download_samples(count=count) or b"" @@ -58,38 +85,31 @@ def _download(device, count: int = _SAMPLE_COUNT) -> bytes: return b"" -def _acquired_bytes(read_result, default: int) -> int: - """How many BigBuf bytes hold valid samples, from an ``lf.read()`` result. The firmware - reports the sample size in BITS (count * bits_per_sample); at 8 bits/sample that is one byte - per sample. Downloading exactly this avoids the BigBuf memory *past* the acquired samples - (heap / the 0xDEADBEEF stack canary) that otherwise poisons the demod.""" - if isinstance(read_result, dict): - bits = read_result.get("samples", 0) - if isinstance(bits, int) and bits > 0: - return max(512, min(bits // 8, 60000)) - return default +def _capture(device, do_read, count: int = _DOWNLOAD_BYTES, tries: int = 3) -> bytes: + """Read + download a clean LF envelope, retrying past the intermittent desyncs a flaky NG + link produces. Returns validated sample bytes, or ``b''`` if every attempt came back as + memory (device likely needs a power-cycle / lower debug level).""" + for _ in range(tries): + if do_read() is None: + continue + data = _download(device, count=count) + if _looks_like_samples(data): + return data + return b"" def read_emitted(device) -> dict | None: """Capture and decode whatever the tag is emitting in the field (no downlink) — the EM4100 ID a tag (native or T5577-emulated) streams continuously.""" - try: - result = device.lf.read(samples=_SAMPLE_COUNT) - except Exception: - return None - return demod_samples(_download(device, count=_acquired_bytes(result, _SAMPLE_COUNT))) + data = _capture(device, lambda: _try(lambda: device.lf.read(samples=15000), None)) + return demod_samples(data) if data else None def read_config(device) -> 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. - The block-0 read (DoPartialAcquisition) captures ~12000 samples and returns no count, so we - download that fixed span — enough for the short block response, without a memory tail.""" - try: - device.lf.t55.readbl(0) # fills BigBuf with the block-0 response envelope - except Exception: - return None - return protocols.decode_t55xx_config(_download(device, count=12000)) + 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)) + return protocols.decode_t55xx_config(data) if data else None def identify_lf(device, emit=None) -> dict | None: diff --git a/tests/test_lf_demod.py b/tests/test_lf_demod.py index 503abed..a0d7729 100644 --- a/tests/test_lf_demod.py +++ b/tests/test_lf_demod.py @@ -364,10 +364,29 @@ class TestIdentifyLF: assert identify_lf(None) is None -class TestAcquiredBytes: - def test_bit_count_to_bytes(self): - from pm3py.lf.capture import _acquired_bytes - assert _acquired_bytes({"samples": 98304}, 15000) == 12288 # 98304 bits / 8 = 12288 samples - assert _acquired_bytes({"samples": 0}, 15000) == 15000 # no count -> fallback - assert _acquired_bytes(None, 15000) == 15000 - assert _acquired_bytes({"samples": 100}, 15000) == 512 # clamped up to a floor +class TestCaptureValidation: + def test_looks_like_samples_rejects_memory(self): + from pm3py.lf.capture import _looks_like_samples + env = _em4100_env(0x0102030405) + assert _looks_like_samples(env) # a real envelope passes + assert not _looks_like_samples(b"\xef\xbe\xad\xde" * 500) # DEADBEEF stack canary + assert not _looks_like_samples(b"\x50\x4d\x33\x62" + bytes(500)) # PM3 response magic + assert not _looks_like_samples(bytes(4000)) # all-zero (unwritten) buffer + assert not _looks_like_samples(b"\x80" * 100) # too short + + def test_capture_retries_past_garbage(self): + from pm3py.lf.capture import _capture + good = _em4100_env(0x0102030405) + dev = MagicMock() + # first download is desynced memory, second is a real envelope -> retry returns the good one + dev.lf.download_samples.side_effect = [b"\xef\xbe\xad\xde" * 500, good] + dev.lf.read.return_value = {"status": 0} + data = _capture(dev, lambda: dev.lf.read(samples=15000)) + assert data == good + + def test_capture_gives_up_all_garbage(self): + from pm3py.lf.capture import _capture + dev = MagicMock() + dev.lf.download_samples.return_value = b"\xef\xbe\xad\xde" * 500 + dev.lf.read.return_value = {"status": 0} + assert _capture(dev, lambda: dev.lf.read(samples=15000)) == b""