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.
This commit is contained in:
michael
@@ -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():
|
||||
|
||||
@@ -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("<HB", idle_timeout_ms, flags)
|
||||
frame = encode_ng_frame(Cmd.HF_ISO15693_SNIFF, payload)
|
||||
self._ser.write(frame)
|
||||
|
||||
if live:
|
||||
self._formatter = TraceFormatter(
|
||||
mode="sniff", decoder=decode_15693, crc_len=2)
|
||||
self._active = True
|
||||
self._state = "starting"
|
||||
self._reader_thread = threading.Thread(
|
||||
target=self._stream_reader, daemon=True)
|
||||
self._reader_thread.start()
|
||||
else:
|
||||
self._legacy_sniff()
|
||||
|
||||
def _stream_reader(self) -> None:
|
||||
"""Background thread: read and print trace frames live."""
|
||||
buf = bytearray()
|
||||
while self._active:
|
||||
try:
|
||||
width = os.get_terminal_size().columns
|
||||
except (OSError, ValueError):
|
||||
width = 120
|
||||
is_tty = sys.stdout.isatty()
|
||||
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("<I", RESP_PREAMBLE_MAGIC))
|
||||
if idx < 0:
|
||||
buf[:] = buf[-3:]
|
||||
return False
|
||||
if idx > 0:
|
||||
del buf[:idx]
|
||||
|
||||
if len(buf) < RESP_PREAMBLE_SIZE:
|
||||
return False
|
||||
|
||||
length_ng = struct.unpack_from("<H", buf, 4)[0]
|
||||
payload_len = length_ng & 0x7FFF
|
||||
frame_len = RESP_PREAMBLE_SIZE + payload_len + RESP_POSTAMBLE_SIZE
|
||||
|
||||
if len(buf) < frame_len:
|
||||
return False
|
||||
|
||||
try:
|
||||
resp = decode_response_frame(bytes(buf[:frame_len]))
|
||||
del buf[:frame_len]
|
||||
except Exception:
|
||||
del buf[:4]
|
||||
return True
|
||||
|
||||
cmd = resp["cmd"]
|
||||
data = resp.get("data", b"")
|
||||
|
||||
if cmd == Cmd.HF_SNIFF_STREAM.value:
|
||||
self._handle_trace(data)
|
||||
elif cmd == Cmd.HF_SNIFF_STATUS.value:
|
||||
self._handle_status(data)
|
||||
elif cmd == Cmd.HF_ISO15693_SNIFF.value:
|
||||
self._active = False
|
||||
|
||||
return True
|
||||
|
||||
def _handle_trace(self, data: bytes) -> None:
|
||||
"""Parse and print a trace entry from CMD_HF_SNIFF_STREAM."""
|
||||
if len(data) < 8:
|
||||
return
|
||||
|
||||
protocol, direction, duration, timestamp = struct.unpack_from("<BBHI", data)
|
||||
payload = data[8:]
|
||||
|
||||
entry = {
|
||||
"protocol": protocol,
|
||||
"direction": direction,
|
||||
"duration": duration,
|
||||
"timestamp": timestamp,
|
||||
"data": payload,
|
||||
}
|
||||
self._entries.append(entry)
|
||||
|
||||
if self._formatter:
|
||||
self._formatter.print(direction, payload)
|
||||
|
||||
def _handle_status(self, data: bytes) -> None:
|
||||
"""Handle a CMD_HF_SNIFF_STATUS state change."""
|
||||
if len(data) < 3:
|
||||
return
|
||||
|
||||
state, count = struct.unpack_from("<BH", data)
|
||||
self._state = _STATE_NAMES.get(state, "unknown")
|
||||
|
||||
icon = _STATE_ICONS.get(state, "??")
|
||||
label = _STATE_LABELS.get(state, "Unknown")
|
||||
count_str = f" ({count} frames)" if count > 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("<I", RESP_PREAMBLE_MAGIC))
|
||||
if idx < 0:
|
||||
continue
|
||||
if idx > 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("<H", buf, 4)[0]
|
||||
payload_len = length_ng & 0x7FFF
|
||||
frame_len = RESP_PREAMBLE_SIZE + payload_len + RESP_POSTAMBLE_SIZE
|
||||
|
||||
if len(buf) < frame_len:
|
||||
continue
|
||||
|
||||
try:
|
||||
resp = decode_response_frame(bytes(buf[:frame_len]))
|
||||
buf = buf[frame_len:]
|
||||
if resp["cmd"] == Cmd.HF_ISO15693_SNIFF.value:
|
||||
break
|
||||
except Exception:
|
||||
buf = buf[4:]
|
||||
|
||||
print("[Snf] Sniff ended, downloading trace...")
|
||||
# Legacy download not fully implemented in SniffSession.
|
||||
# Use pm3.hf.iso15.sniff_decoded() for legacy batch mode.
|
||||
print("[Snf] Legacy download not fully implemented in SniffSession.")
|
||||
print("[Snf] Use pm3.hf.iso15.sniff_decoded() for legacy batch mode.")
|
||||
|
||||
def stop(self) -> 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()
|
||||
|
||||
Reference in New Issue
Block a user