From e90a9076309a7768fe6a1a04d10a6fe66bbc2b12 Mon Sep 17 00:00:00 2001 From: michael Date: Thu, 19 Mar 2026 12:34:05 -0700 Subject: [PATCH] feat: rewrite SniffSession with live streaming and REPL auto-load Replace lightweight async SniffSession with persistent synchronous session using raw pyserial (mirroring SimSession pattern). Background reader thread decodes and prints trace frames live as they arrive. API: session = SniffSession.open() session.start_15693() # live streaming (default) session.start_15693(live=False) # legacy batch mode session.stop() # or button press to pause/resume session.entries # all captured frames session.clear() .pythonstartup.py auto-creates session in sniff mode. --- .pythonstartup.py | 16 +++ pm3py/sniff/session.py | 318 ++++++++++++++++++++++++++++++++++------- 2 files changed, 281 insertions(+), 53 deletions(-) diff --git a/.pythonstartup.py b/.pythonstartup.py index b01ab7d..e2a8a41 100644 --- a/.pythonstartup.py +++ b/.pythonstartup.py @@ -95,6 +95,22 @@ for module_name, attr_name in imports: if loaded: print(f"[pm3py:{_MODE}] Loaded: {', '.join(loaded.keys())}") + if _MODE == "core": + try: + pm3 = Proxmark3.sync() + print() + print("Synchronous connection established") + print() + except Exception as e: + print(f" (Proxmark3 connection failed: {e})") + + if _MODE == "sniff": + try: + session = SniffSession.open() + print("Sniff session ready -- session.start_15693()") + except Exception as e: + print(f" (Sniff session failed: {e})") + if failed: print(f"\n[pm3py:{_MODE}] Failed:") for name, (mod, err) in failed.items(): diff --git a/pm3py/sniff/session.py b/pm3py/sniff/session.py index 0ab1166..57eb19d 100644 --- a/pm3py/sniff/session.py +++ b/pm3py/sniff/session.py @@ -1,75 +1,287 @@ -"""Sniff sessions — start/stop/download per protocol.""" +"""Sniff sessions — live streaming with button-toggle, or legacy batch.""" +import struct import sys +import threading +import time -from ..core.protocol import Cmd, PM3_CMD_DATA_SIZE -from ..core.transport import PM3Transport, encode_ng_frame +import serial + +from ..core.protocol import ( + Cmd, PM3_CMD_DATA_SIZE, + SNIFF_FLAG_STREAMING, SNIFF_FLAG_BUTTON_TOGGLE, + SNIFF_STATE_STARTED, SNIFF_STATE_PAUSED, + SNIFF_STATE_RESUMED, SNIFF_STATE_STOPPED, +) +from ..core.transport import ( + encode_ng_frame, decode_response_frame, + RESP_PREAMBLE_MAGIC, RESP_PREAMBLE_SIZE, RESP_POSTAMBLE_SIZE, +) from ..trace import parse_tracelog -from ..trace.format import format_sniff_line +from ..trace.decode_iso15 import decode_15693 +from ..trace.format import TraceFormatter, format_sniff_line + +# Default serial port (same as SimSession) +DEFAULT_PORT = "/dev/ttyACM0" + +_STATE_NAMES = { + SNIFF_STATE_STARTED: "sniffing", + SNIFF_STATE_PAUSED: "paused", + SNIFF_STATE_RESUMED: "sniffing", + SNIFF_STATE_STOPPED: "stopped", +} + +_STATE_ICONS = { + SNIFF_STATE_STARTED: ">>", + SNIFF_STATE_PAUSED: "||", + SNIFF_STATE_RESUMED: ">>", + SNIFF_STATE_STOPPED: "[]", +} + +_STATE_LABELS = { + SNIFF_STATE_STARTED: "Started", + SNIFF_STATE_PAUSED: "Paused", + SNIFF_STATE_RESUMED: "Resumed", + SNIFF_STATE_STOPPED: "Stopped", +} class SniffSession: - """Sniff session that manages start/download/decode lifecycle. + """Persistent sniff session with live streaming and button-toggle. - Takes a PM3Transport (not the client), since sniff has its own - lifecycle that doesn't fit the stateless command pattern. + For live mode (default): + session = SniffSession.open() + session.start_15693() # streams trace live, button toggles pause + session.stop() # or let button cycles accumulate + session.entries # all captured frames + + For legacy mode: + session.start_15693(live=False) # blocks until button, batch download """ - def __init__(self, transport: PM3Transport): - self._t = transport + def __init__(self, ser): + self._ser = ser + self._active = False + self._entries: list[dict] = [] + self._reader_thread: threading.Thread | None = None + self._formatter: TraceFormatter | None = None + self._state = "idle" - async def _sniff_15(self, timeout: float = 60.0) -> dict: - """Start ISO 15693 sniff. Blocks until button press or timeout.""" - async with self._t._lock: - raw = encode_ng_frame(Cmd.HF_ISO15693_SNIFF) - await self._t.send_frame(raw) - while True: - resp = await self._t.read_response(timeout=timeout) - if resp.cmd == Cmd.HF_ISO15693_SNIFF: - return {"status": resp.status} + @classmethod + def open(cls, port: str = DEFAULT_PORT, baudrate: int = 115200) -> "SniffSession": + """Open serial connection to PM3. Auto-detects port if default.""" + ser = serial.Serial() + ser.port = port + ser.baudrate = baudrate + ser.bytesize = 8 + ser.parity = "N" + ser.stopbits = 1 + ser.timeout = 0.1 + ser.xonxoff = False + ser.rtscts = False + ser.dsrdtr = False + ser.open() - async def _download_trace(self) -> list[dict]: - """Download and parse the trace buffer from the device.""" - raw, trace_len = await self._t.download_bigbuf( - offset=0, length=PM3_CMD_DATA_SIZE, timeout=4.0) + ser.reset_input_buffer() + ser.reset_output_buffer() + time.sleep(0.5) + ser.reset_input_buffer() - if trace_len == 0: - return [] + return cls(ser) - if trace_len > PM3_CMD_DATA_SIZE: - raw, trace_len = await self._t.download_bigbuf( - offset=0, length=trace_len, timeout=10.0) + def start_15693(self, live: bool = True, idle_timeout_ms: int = 400) -> None: + """Start 15693 sniff session. - return parse_tracelog(raw[:trace_len]) - - async def iso15(self, timeout: float = 60.0) -> list[dict]: - """Sniff ISO 15693 traffic, download trace, decode and print. - - Blocks until button press or timeout. Then downloads the trace, - decodes each frame, and prints formatted output. - - Returns the parsed trace entries. + live=True (default): streaming with button-toggle pause/resume. + live=False: legacy BigBuf batch mode, blocks until button exits. """ - import os + if self._active: + self.stop() + + self._protocol = 0 # 15693 + flags = 0 + if live: + flags = SNIFF_FLAG_STREAMING | SNIFF_FLAG_BUTTON_TOGGLE + + payload = struct.pack(" None: + """Background thread: read and print trace frames live.""" + buf = bytearray() + while self._active: + try: + chunk = self._ser.read(self._ser.in_waiting or 1) + if not chunk: + continue + buf.extend(chunk) + + while self._try_decode_frame(buf): + pass + + except serial.SerialException: + break + except Exception: + continue + + def _try_decode_frame(self, buf: bytearray) -> bool: + """Decode one response frame from buf. Returns True if consumed.""" + if len(buf) < RESP_PREAMBLE_SIZE + RESP_POSTAMBLE_SIZE: + return False + + idx = buf.find(struct.pack(" 0: + del buf[:idx] + + if len(buf) < RESP_PREAMBLE_SIZE: + return False + + length_ng = struct.unpack_from(" None: + """Parse and print a trace entry from CMD_HF_SNIFF_STREAM.""" + if len(data) < 8: + return + + protocol, direction, duration, timestamp = struct.unpack_from(" None: + """Handle a CMD_HF_SNIFF_STATUS state change.""" + if len(data) < 3: + return + + state, count = struct.unpack_from(" 0 else "" + + msg = f"\n[Snf] {icon} {label}{count_str}\n" + sys.stdout.write(msg) + sys.stdout.flush() + + if state == SNIFF_STATE_STOPPED: + self._active = False + + def _legacy_sniff(self) -> None: + """Legacy batch mode: block until button, download BigBuf, decode.""" print("[Snf] Sniffing ISO 15693... press PM3 button to stop.") - result = await self._sniff_15(timeout=timeout) - print(f"[Snf] Sniff ended (status={result['status']})") - print("[Snf] Downloading trace...") - entries = await self._download_trace() + # Read frames until we get the sniff completion response + buf = bytearray() + while True: + chunk = self._ser.read(self._ser.in_waiting or 1) + if chunk: + buf.extend(chunk) - if not entries: - print("[Snf] No trace data captured.") - return [] + idx = buf.find(struct.pack(" 0: + buf = buf[idx:] - print(f"[Snf] {len(entries)} frames captured:\n") - for entry in entries: - line = format_sniff_line(entry, width=width, is_tty=is_tty) - print(line) + if len(buf) < RESP_PREAMBLE_SIZE: + continue - return entries + length_ng = struct.unpack_from(" None: + """End sniff session from Python (sends BREAK_LOOP).""" + if self._active: + frame = encode_ng_frame(Cmd.BREAK_LOOP, b"") + self._ser.write(frame) + if self._reader_thread: + self._reader_thread.join(timeout=2.0) + self._active = False + self._state = "stopped" + + def close(self) -> None: + """Stop and close serial port.""" + if self._active: + self.stop() + self._ser.close() + + @property + def state(self) -> str: + """Current session state: idle, sniffing, paused, stopped.""" + return self._state + + @property + def entries(self) -> list[dict]: + """All captured trace entries across pause/resume cycles.""" + return list(self._entries) + + def clear(self) -> None: + """Clear accumulated entries.""" + self._entries.clear()