pm3py/pm3py/lf/capture.py

"""Live-device LF capture + identify.

``demod_samples`` is the hardware-free heart: an envelope in, a decoded credential out — every
protocol decoder tried in turn. ``identify_lf`` drives a real device: it captures what the tag
*emits* (a plain read) and, separately, probes for a T55xx by reading config block 0, then
combines the two into "is it a T5577, and what is it (emulating)?".
"""
from __future__ import annotations

import inspect

from . import ask
from . import protocols

# how many envelope samples to pull from BigBuf after a capture (a T55xx/EM frame is a few
# thousand samples at RF/64; this spans several repeats for a clean lock)
_SAMPLE_COUNT = 15000

# every emitted-credential decoder, tried in order (headered/most-specific first)
_DECODERS = (
    protocols.decode_em4100,
    protocols.decode_hid,
    protocols.decode_awid,
    protocols.decode_fdxb,
)


def credential_label(c: dict) -> str:
    """One-line name for a decoded credential, per protocol (each decoder returns its own fields)."""
    p = c.get("protocol", "?")
    if p == "EM4100":
        return f"EM4100 {c['id_hex']}"
    if p == "HID Prox":
        return f"HID Prox {c['format']} FC {c['facility_code']} Card {c['card_number']}"
    if p == "AWID":
        return f"{c['format']} FC {c['facility_code']} Card {c['card_number']}"
    if p == "FDX-B":
        return f"FDX-B {c['id']}" + ("  (animal)" if c.get("animal") else "")
    return p


def demod_samples(samples) -> dict | None:
    """Decode a raw LF envelope to a credential by trying each protocol demod in turn.

    Hardware-free and the single point every decoder funnels through — feed it a synthetic
    envelope from a transponder encoder in tests, or a BigBuf download at runtime."""
    if not samples:
        return None
    for decode in _DECODERS:
        result = decode(samples)
        if result is not None:
            return result
    return None


# 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""
    except Exception:
        return b""


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."""
    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."""
    data = _capture(device, lambda: _try(lambda: device.lf.t55.readbl(0), None))
    return protocols.decode_t55xx_config(data) if data else None


def read_t55xx_block(device, block: int) -> 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))
    if not data:
        return None
    for bits in ask.manchester_bits(data):
        clean = [b for b in bits if b is not None]
        for word in protocols._repeating_words(clean, 32):
            return word
    return None


def identify_lf(device, emit=None) -> dict | None:
    """Full LF identify against a live device. Returns a result dict (``found``/``field``/
    ``label``/``chip``/``config``/``emulating``/``emitted``) or ``None`` if the device is
    unusable. ``emit`` streams human-readable progress lines to the caller's trace."""
    emit = emit or (lambda _l: None)
    if device is None:
        return None

    # capture drives the device *synchronously* (as Proxmark3.sync()'s proxy exposes it). A raw
    # async Proxmark3() returns un-awaited coroutines here — fail with an actionable message
    # instead of a cryptic "coroutine has no len()" deep in the demod.
    if inspect.iscoroutinefunction(getattr(getattr(device, "lf", None), "read", None)):
        raise TypeError(
            "identify_lf needs a synchronous device — use Proxmark3.sync(), not Proxmark3(). "
            "(A bare Proxmark3() is the async API and isn't connected.)"
        )

    # Force the device debug level to NONE for the duration: on debug-heavy firmware a raised
    # g_dbglevel makes LF acquisition emit Dbprintf frames over USB, which interleave with the
    # sample download and desync it. This is the device-side knob (not the C client's g_debugMode
    # spam), and pm3py's own reads are what matter here.
    _try(lambda: device.hw.dbg(0), 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:
        return None

    chip = None
    emulating = None
    if config is not None:
        chip = "T5577"
        emulating = config["emulating"]
        emit(f"  T55xx block 0 = 0x{config['config_hex']}  "
             f"({config['modulation']}, RF/{config['data_bit_rate']}, {config['max_block']} blocks)"
             f"  -> {emulating}")
    if emitted is not None:
        emit(f"  emitted: {credential_label(emitted)}")

    # 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.
    if chip and emitted:
        label = f"{chip} ({credential_label(emitted)})"
    elif chip:
        label = f"{chip} ({emulating})"
    else:
        label = credential_label(emitted)

    return {
        "found": True,
        "field": "lf",
        "protocol": "lf",
        "chip": chip,
        "config": config["config_hex"] if config else None,
        "emulating": emulating,
        "emitted": emitted,
        "label": label,
    }