MikeFaith/pm3py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2b087a302a9b525d085abeffcdcc9155c5ee4dcb
pm3py/docs/plans/2026-03-19-live-sniff-plan.md
michael 615d4a6115 docs: roadmap index + custom-14a-command plans + backfill plan docs 
- Add docs/plans/README.md as the roadmap index cataloguing all design/plan docs.
- Add the custom ISO14443-A command handling design + plan (L3 NTAG I2C
  SECTOR_SELECT / cross-sector reads + native auth; L4 static APDUs + WTX relay),
  produced from a multi-agent design workflow.
- Backfill previously-uncommitted plan docs (trace-formatter, ndef-trace-decode,
  live-sniff, firmware-upstream-rebase, 14a-live-trace) and NTAG5_SECURITY.md.
- Sync CLAUDE.md package structure with the committed transponder models.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-05 12:58:48 -07:00

1109 lines
37 KiB
Markdown
Raw Blame History

# Live Sniff Streaming Implementation Plan
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** Rewrite sniff from batch download to live-streaming with button-toggle pause/resume, auto-loaded in the REPL.
**Architecture:** Firmware sends trace entries as fire-and-forget `reply_ng()` packets on RF idle (reusing sim's ring buffer pattern). Python `SniffSession` receives in a background thread (mirroring `SimSession._trace_reader`). `.pythonstartup.py` auto-creates `session` in sniff mode.
**Tech Stack:** C (firmware ARM), Python 3.12 (pyserial, threading), pytest with mocking
**Design doc:** `docs/plans/2026-03-19-live-sniff-design.md`
---
### Task 1: Firmware — Add command IDs and structs
**Files:**
- Modify: `firmware/include/pm3_cmd.h:734` (after CMD_HF_ISO15693_SIM_TRACE)
- Modify: `pm3py/core/protocol.py:203` (after HF_ISO15693_EML_SETMEM)
**Step 1: Add firmware command definitions**
In `firmware/include/pm3_cmd.h`, after line 734 (`CMD_HF_ISO15693_SIM_TRACE`), add:
```c
#define CMD_HF_SNIFF_STREAM 0x0337 // FW->Host: live sniff trace entry
#define CMD_HF_SNIFF_STATUS 0x0338 // FW->Host: sniff state change
// Sniff streaming flags (sent in sniff command payload)
#define SNIFF_FLAG_STREAMING 0x01
#define SNIFF_FLAG_BUTTON_TOGGLE 0x02
// Sniff state values (in CMD_HF_SNIFF_STATUS payload)
#define SNIFF_STATE_STARTED 0
#define SNIFF_STATE_PAUSED 1
#define SNIFF_STATE_RESUMED 2
#define SNIFF_STATE_STOPPED 3
```
**Step 2: Add Python Cmd enum entries**
In `pm3py/core/protocol.py`, after `HF_ISO15693_EML_SETMEM = 0x0331` (line 203), add:
```python
HF_SNIFF_STREAM = 0x0337
HF_SNIFF_STATUS = 0x0338
```
Also add constants after the `Cmd` class:
```python
# Sniff streaming flags
SNIFF_FLAG_STREAMING = 0x01
SNIFF_FLAG_BUTTON_TOGGLE = 0x02
# Sniff states
SNIFF_STATE_STARTED = 0
SNIFF_STATE_PAUSED = 1
SNIFF_STATE_RESUMED = 2
SNIFF_STATE_STOPPED = 3
```
**Step 3: Commit**
```bash
git add firmware/include/pm3_cmd.h pm3py/core/protocol.py
git commit --no-gpg-sign -m "feat: add CMD_HF_SNIFF_STREAM/STATUS command IDs and constants"
```
---
### Task 2: Firmware — Streaming sniff loop
**Files:**
- Modify: `firmware/armsrc/iso15693.c:1654-1875` (SniffIso15693)
- Modify: `firmware/armsrc/appmain.c:1500-1503` (CMD_HF_ISO15693_SNIFF handler)
**Step 1: Modify appmain.c to pass payload to SniffIso15693**
Currently (line 1500-1503):
```c
case CMD_HF_ISO15693_SNIFF: {
SniffIso15693(0, NULL, false);
reply_ng(CMD_HF_ISO15693_SNIFF, PM3_SUCCESS, NULL, 0);
break;
}
```
Change to:
```c
case CMD_HF_ISO15693_SNIFF: {
uint16_t idle_timeout_ms = 0;
uint8_t sniff_flags = 0;
if (packet->length >= 3) {
idle_timeout_ms = packet->data.asBytes[0] | (packet->data.asBytes[1] << 8);
sniff_flags = packet->data.asBytes[2];
}
SniffIso15693(0, NULL, false, idle_timeout_ms, sniff_flags);
if (!(sniff_flags & SNIFF_FLAG_STREAMING)) {
// Legacy mode: send completion response (streaming mode sends STATUS)
reply_ng(CMD_HF_ISO15693_SNIFF, PM3_SUCCESS, NULL, 0);
}
break;
}
```
**Step 2: Update SniffIso15693 signature and add streaming mode**
In `firmware/armsrc/iso15693.c`, change the function signature (line 1654):
```c
void SniffIso15693(uint8_t jam_search_len, uint8_t *jam_search_string, bool iclass,
uint16_t idle_timeout_ms, uint8_t sniff_flags) {
```
After the existing init code (after line 1698, the DMA setup), add the streaming mode branch. The key change: when `sniff_flags & SNIFF_FLAG_STREAMING`, use a ring buffer + idle flush loop instead of LogTrace to BigBuf.
Insert after the DMA setup succeeds (after line 1697), before the state variables (line 1699):
```c
// --- Streaming mode ---
if (sniff_flags & SNIFF_FLAG_STREAMING) {
// Ring buffer (same pattern as SimTagIso15693)
#define SNIFF_TRACE_ENTRIES 8
#define SNIFF_TRACE_ENTRY_MAX 72 // 8-byte header + 64 data
static uint8_t sniff_ring[SNIFF_TRACE_ENTRIES][SNIFF_TRACE_ENTRY_MAX];
static uint16_t sniff_ring_len[SNIFF_TRACE_ENTRIES];
uint8_t sniff_wr = 0, sniff_rd = 0;
uint16_t frame_count = 0;
uint32_t last_activity = 0;
bool sniff_paused = false;
// Macro: push trace entry with full metadata
// Layout: [protocol(1) | direction(1) | duration(2) | timestamp(4) | data(*)]
#define SNIFF_PUSH(dir, data_ptr, data_len, ts, dur) do { \
if (((sniff_wr + 1) % SNIFF_TRACE_ENTRIES) != sniff_rd) { \
uint8_t *_e = sniff_ring[sniff_wr]; \
_e[0] = 0; /* protocol: 0=15693 */ \
_e[1] = (dir); \
_e[2] = (dur) & 0xFF; _e[3] = ((dur) >> 8) & 0xFF; \
_e[4] = (ts) & 0xFF; _e[5] = ((ts) >> 8) & 0xFF; \
_e[6] = ((ts) >> 16) & 0xFF; _e[7] = ((ts) >> 24) & 0xFF; \
uint16_t _dl = (data_len); \
if (_dl > SNIFF_TRACE_ENTRY_MAX - 8) _dl = SNIFF_TRACE_ENTRY_MAX - 8; \
memcpy(_e + 8, (data_ptr), _dl); \
sniff_ring_len[sniff_wr] = 8 + _dl; \
sniff_wr = (sniff_wr + 1) % SNIFF_TRACE_ENTRIES; \
} \
} while(0)
#define SNIFF_FLUSH() do { \
while (sniff_rd != sniff_wr) { \
reply_ng(CMD_HF_SNIFF_STREAM, PM3_SUCCESS, \
sniff_ring[sniff_rd], sniff_ring_len[sniff_rd]); \
sniff_rd = (sniff_rd + 1) % SNIFF_TRACE_ENTRIES; \
} \
} while(0)
// Send started status
uint8_t status_buf[3] = { SNIFF_STATE_STARTED, 0, 0 };
reply_ng(CMD_HF_SNIFF_STATUS, PM3_SUCCESS, status_buf, 3);
LED_A_ON(); // green = sniffing active
// --- Streaming sniff main loop ---
// (This duplicates the DMA decode logic from the legacy path below,
// but replaces LogTrace calls with SNIFF_PUSH and adds idle flush +
// button toggle + BREAK_LOOP check.)
bool tag_is_active = false;
bool reader_is_active = false;
bool expect_tag_answer = false;
bool expect_fsk_answer = false;
bool expect_fast_answer = true;
int dma_start_time = 0;
int samples = 0;
const uint16_t *upTo = dma->buf;
bool exit_sniff = false;
while (!exit_sniff) {
volatile int behind_by = ((uint16_t *)AT91C_BASE_PDC_SSC->PDC_RPR - upTo) & (DMA_BUFFER_SIZE - 1);
if (behind_by < 1) {
// No new DMA data — check idle flush, button, break
if (!sniff_paused && sniff_wr != sniff_rd && last_activity > 0) {
uint32_t now = GetTickCount();
if ((now - last_activity) > idle_timeout_ms) {
SNIFF_FLUSH();
}
}
// Check for Python BREAK_LOOP
if (data_available()) {
PacketCommandNG rx;
if (receive_ng(&rx) == PM3_SUCCESS && rx.cmd == CMD_BREAK_LOOP) {
SNIFF_FLUSH();
exit_sniff = true;
}
}
continue;
}
samples++;
if (samples == 1) {
dma_start_time = GetCountSspClk() & 0xfffffff0;
}
volatile uint16_t sniffdata = 0;
volatile uint16_t sniffdata_prev = sniffdata;
sniffdata = *upTo++;
if (upTo >= dma->buf + DMA_BUFFER_SIZE) {
upTo = dma->buf;
if (AT91C_BASE_SSC->SSC_SR & (AT91C_SSC_ENDRX)) {
if (AT91C_BASE_PDC_SSC->PDC_RCR == 0) {
AT91C_BASE_PDC_SSC->PDC_RPR = (uint32_t) dma->buf;
AT91C_BASE_PDC_SSC->PDC_RCR = DMA_BUFFER_SIZE;
}
if (AT91C_BASE_PDC_SSC->PDC_RNCR == 0) {
AT91C_BASE_PDC_SSC->PDC_RNPR = (uint32_t) dma->buf;
AT91C_BASE_PDC_SSC->PDC_RNCR = DMA_BUFFER_SIZE;
}
WDT_HIT();
// Button toggle
if (BUTTON_PRESS()) {
SpinDelay(200); // debounce
while (BUTTON_PRESS()) { WDT_HIT(); } // wait for release
SpinDelay(200); // debounce release
if (sniff_paused) {
sniff_paused = false;
LED_A_ON();
status_buf[0] = SNIFF_STATE_RESUMED;
status_buf[1] = frame_count & 0xFF;
status_buf[2] = (frame_count >> 8) & 0xFF;
reply_ng(CMD_HF_SNIFF_STATUS, PM3_SUCCESS, status_buf, 3);
} else {
SNIFF_FLUSH();
sniff_paused = true;
LED_A_OFF();
status_buf[0] = SNIFF_STATE_PAUSED;
status_buf[1] = frame_count & 0xFF;
status_buf[2] = (frame_count >> 8) & 0xFF;
reply_ng(CMD_HF_SNIFF_STATUS, PM3_SUCCESS, status_buf, 3);
}
}
// Check for Python BREAK_LOOP
if (data_available()) {
PacketCommandNG rx;
if (receive_ng(&rx) == PM3_SUCCESS && rx.cmd == CMD_BREAK_LOOP) {
SNIFF_FLUSH();
exit_sniff = true;
}
}
}
}
if (sniff_paused) continue;
// --- Reader decode (same as legacy) ---
if (tag_is_active == false) {
int extra_8s = 1;
if (Handle15693SampleFromReader((sniffdata & 0x02) >> 1, &dreader) ||
(++extra_8s && Handle15693SampleFromReader(sniffdata & 0x01, &dreader))) {
if (dreader.byteCount > 0) {
uint32_t eof_time = dma_start_time + (samples * 16) + (extra_8s * 8) - DELAY_READER_TO_ARM_SNIFF;
uint32_t sof_time = eof_time
- dreader.byteCount * (dreader.Coding == CODING_1_OUT_OF_4 ? 1024 : 16384)
- 256 - 128;
uint32_t ts_us = (sof_time * 4) / 14; // ssp_clk*4 -> ~us
uint16_t dur = (uint16_t)(((eof_time - sof_time) * 4) / 14);
SNIFF_PUSH(0, dreader.output, dreader.byteCount, ts_us, dur);
frame_count++;
last_activity = GetTickCount();
if (iclass == false) {
expect_fsk_answer = dreader.output[0] & ISO15_REQ_SUBCARRIER_TWO;
expect_fast_answer = dreader.output[0] & ISO15_REQ_DATARATE_HIGH;
}
}
DecodeTagReset(&dtag);
DecodeTagFSKReset(&dtagfsk);
reader_is_active = false;
expect_tag_answer = true;
} else {
reader_is_active = (dreader.state >= STATE_READER_RECEIVE_DATA_1_OUT_OF_4);
}
}
// --- Tag decode: single subcarrier (same as legacy) ---
if ((reader_is_active == false) && expect_tag_answer) {
if (expect_fsk_answer == false) {
if (Handle15693SamplesFromTag((sniffdata >> 4) << 2, &dtag, expect_fast_answer)) {
uint32_t eof_time = dma_start_time + (samples * 16) - DELAY_TAG_TO_ARM_SNIFF;
if (dtag.lastBit == SOF_PART2) {
eof_time -= (8 * 16);
}
uint32_t sof_time = eof_time
- dtag.len * 1024 - 512
- (dtag.lastBit != SOF_PART2 ? 512 : 0);
uint32_t ts_us = (sof_time * 4) / 14;
uint16_t dur = (uint16_t)(((eof_time - sof_time) * 4) / 14);
SNIFF_PUSH(1, dtag.output, dtag.len, ts_us, dur);
frame_count++;
last_activity = GetTickCount();
DecodeTagReset(&dtag);
DecodeTagFSKReset(&dtagfsk);
DecodeReaderReset(&dreader);
expect_tag_answer = false;
tag_is_active = false;
} else {
tag_is_active = (dtag.state >= STATE_TAG_RECEIVING_DATA);
}
} else {
// --- Tag decode: dual subcarrier FSK (same as legacy) ---
if (FREQ_IS_0((sniffdata >> 2) & 0x3)) {
sniffdata = sniffdata_prev;
}
if (Handle15693FSKSamplesFromTag((sniffdata >> 2) & 0x3, &dtagfsk, expect_fast_answer)) {
if (dtagfsk.len > 0) {
uint32_t eof_time = dma_start_time + (samples * 16) - DELAY_TAG_TO_ARM_SNIFF;
if (dtagfsk.lastBit == SOF) {
eof_time -= (8 * 16);
}
uint32_t sof_time = eof_time
- dtagfsk.len * 1016 - 512
- (dtagfsk.lastBit != SOF ? 512 : 0);
uint32_t ts_us = (sof_time * 4) / 14;
uint16_t dur = (uint16_t)(((eof_time - sof_time) * 4) / 14);
SNIFF_PUSH(1, dtagfsk.output, dtagfsk.len, ts_us, dur);
frame_count++;
last_activity = GetTickCount();
}
DecodeTagFSKReset(&dtagfsk);
DecodeReaderReset(&dreader);
expect_tag_answer = false;
tag_is_active = false;
expect_fsk_answer = false;
} else {
tag_is_active = (dtagfsk.state >= STATE_FSK_RECEIVING_DATA_484);
}
}
}
}
// Cleanup and send stopped status
FpgaDisableTracing();
switch_off();
status_buf[0] = SNIFF_STATE_STOPPED;
status_buf[1] = frame_count & 0xFF;
status_buf[2] = (frame_count >> 8) & 0xFF;
reply_ng(CMD_HF_SNIFF_STATUS, PM3_SUCCESS, status_buf, 3);
return;
}
// --- Legacy mode (existing code, unchanged from here) ---
```
Also update the function declaration in `iso15693.h` (if it exists) or any forward declarations.
**Step 3: Update any forward declarations**
Search for `SniffIso15693` declarations and update the signature to include the two new parameters.
**Step 4: Commit**
```bash
git add firmware/armsrc/iso15693.c firmware/armsrc/appmain.c
# Also add any header files modified
git commit --no-gpg-sign -m "feat(fw): streaming sniff with ring buffer, button-toggle, idle flush"
```
---
### Task 3: Python — SniffSession rewrite with streaming
**Files:**
- Modify: `pm3py/sniff/session.py` (full rewrite)
- Reference: `pm3py/sim/sim_session.py:54-83` (open pattern), `pm3py/sim/sim_session.py:237-290` (trace reader pattern)
**Step 1: Write tests for the new SniffSession**
Create `tests/test_sniff_session.py`:
```python
"""Tests for SniffSession — live streaming sniff with button-toggle."""
import struct
import threading
import time
from unittest.mock import MagicMock, patch, PropertyMock
import pytest
from pm3py.core.protocol import (
Cmd, SNIFF_FLAG_STREAMING, SNIFF_FLAG_BUTTON_TOGGLE,
SNIFF_STATE_STARTED, SNIFF_STATE_PAUSED,
SNIFF_STATE_RESUMED, SNIFF_STATE_STOPPED,
)
from pm3py.core.transport import (
encode_ng_frame, RESP_PREAMBLE_MAGIC, RESP_POSTAMBLE_NOCRC,
)
from pm3py.sniff.session import SniffSession
def _make_resp_frame(cmd: int, status: int, data: bytes) -> bytes:
"""Build a raw NG response frame (USB, no CRC)."""
payload_len = len(data)
preamble = struct.pack("<IHBBH",
RESP_PREAMBLE_MAGIC,
payload_len | 0x8000, # NG bit set
status & 0xFF,
0, # reason
cmd,
)
postamble = struct.pack("<H", RESP_POSTAMBLE_NOCRC)
return preamble + data + postamble
def _make_sniff_stream_frame(protocol: int, direction: int,
duration: int, timestamp: int,
data: bytes) -> bytes:
"""Build a CMD_HF_SNIFF_STREAM response frame."""
payload = struct.pack("<BBHI", protocol, direction, duration, timestamp) + data
return _make_resp_frame(Cmd.HF_SNIFF_STREAM.value, 0, payload)
def _make_status_frame(state: int, count: int) -> bytes:
"""Build a CMD_HF_SNIFF_STATUS response frame."""
payload = struct.pack("<BH", state, count)
return _make_resp_frame(Cmd.HF_SNIFF_STATUS.value, 0, payload)
class FakeSerial:
"""Mock serial port that feeds pre-built response frames."""
def __init__(self, frames: list[bytes]):
self._buf = bytearray()
for f in frames:
self._buf.extend(f)
self._pos = 0
self._closed = False
@property
def in_waiting(self):
return len(self._buf) - self._pos
def read(self, size=1):
if self._closed or self._pos >= len(self._buf):
time.sleep(0.01) # avoid busy spin
return b""
end = min(self._pos + size, len(self._buf))
data = bytes(self._buf[self._pos:end])
self._pos = end
return data
def write(self, data):
return len(data)
def reset_input_buffer(self):
pass
def reset_output_buffer(self):
pass
def close(self):
self._closed = True
class TestSniffSessionStreaming:
"""Tests for live streaming mode."""
def test_start_15693_sends_streaming_flags(self):
"""start_15693(live=True) sends correct payload."""
ser = FakeSerial([
_make_status_frame(SNIFF_STATE_STARTED, 0),
_make_status_frame(SNIFF_STATE_STOPPED, 0),
])
sent = bytearray()
orig_write = ser.write
def capture_write(data):
sent.extend(data)
return orig_write(data)
ser.write = capture_write
session = SniffSession(ser)
session.start_15693(live=True, idle_timeout_ms=400)
time.sleep(0.1)
session.stop()
# Verify payload contains flags
assert len(sent) > 0
# The payload should contain idle_timeout_ms (400=0x0190) and flags (0x03)
assert b"\x90\x01\x03" in bytes(sent)
def test_start_15693_legacy_mode(self):
"""start_15693(live=False) sends zero flags."""
ser = FakeSerial([
_make_resp_frame(Cmd.HF_ISO15693_SNIFF.value, 0, b""),
])
sent = bytearray()
orig_write = ser.write
def capture_write(data):
sent.extend(data)
return orig_write(data)
ser.write = capture_write
session = SniffSession(ser)
session.start_15693(live=False)
# Legacy should block and return
def test_receives_trace_entries(self):
"""Background thread decodes CMD_HF_SNIFF_STREAM frames."""
reader_cmd = bytes([0x26, 0x01, 0x00]) # INVENTORY
tag_resp = bytes([0x00, 0xE0, 0x04])
ser = FakeSerial([
_make_status_frame(SNIFF_STATE_STARTED, 0),
_make_sniff_stream_frame(0, 0, 100, 1000, reader_cmd),
_make_sniff_stream_frame(0, 1, 200, 2000, tag_resp),
_make_status_frame(SNIFF_STATE_STOPPED, 2),
])
session = SniffSession(ser)
session.start_15693(live=True)
time.sleep(0.3) # let background thread process
assert len(session.entries) == 2
assert session.entries[0]["direction"] == 0
assert session.entries[0]["data"] == reader_cmd
assert session.entries[0]["timestamp"] == 1000
assert session.entries[1]["direction"] == 1
assert session.entries[1]["data"] == tag_resp
def test_status_updates(self):
"""Status frames update session state."""
ser = FakeSerial([
_make_status_frame(SNIFF_STATE_STARTED, 0),
_make_sniff_stream_frame(0, 0, 100, 1000, b"\x26\x01\x00"),
_make_status_frame(SNIFF_STATE_PAUSED, 1),
_make_status_frame(SNIFF_STATE_RESUMED, 1),
_make_status_frame(SNIFF_STATE_STOPPED, 1),
])
session = SniffSession(ser)
session.start_15693(live=True)
time.sleep(0.3)
assert session.state == "stopped"
def test_stop_sends_break_loop(self):
"""stop() sends CMD_BREAK_LOOP to firmware."""
ser = FakeSerial([
_make_status_frame(SNIFF_STATE_STARTED, 0),
_make_status_frame(SNIFF_STATE_STOPPED, 0),
])
sent = bytearray()
orig_write = ser.write
def capture_write(data):
sent.extend(data)
return orig_write(data)
ser.write = capture_write
session = SniffSession(ser)
session.start_15693(live=True)
time.sleep(0.1)
session.stop()
# BREAK_LOOP = 0x0118
assert b"\x18\x01" in bytes(sent)
def test_entries_accumulate_across_cycles(self):
"""Entries persist across pause/resume cycles."""
ser = FakeSerial([
_make_status_frame(SNIFF_STATE_STARTED, 0),
_make_sniff_stream_frame(0, 0, 100, 1000, b"\x26\x01\x00"),
_make_status_frame(SNIFF_STATE_PAUSED, 1),
_make_status_frame(SNIFF_STATE_RESUMED, 1),
_make_sniff_stream_frame(0, 0, 100, 3000, b"\x26\x01\x00"),
_make_status_frame(SNIFF_STATE_STOPPED, 2),
])
session = SniffSession(ser)
session.start_15693(live=True)
time.sleep(0.3)
assert len(session.entries) == 2
def test_clear_entries(self):
"""clear() resets accumulated entries."""
ser = FakeSerial([
_make_status_frame(SNIFF_STATE_STARTED, 0),
_make_sniff_stream_frame(0, 0, 100, 1000, b"\x26\x01\x00"),
_make_status_frame(SNIFF_STATE_STOPPED, 1),
])
session = SniffSession(ser)
session.start_15693(live=True)
time.sleep(0.3)
assert len(session.entries) == 1
session.clear()
assert len(session.entries) == 0
```
**Step 2: Run tests to verify they fail**
```bash
cd /home/work/pm3py && python -m pytest tests/test_sniff_session.py -v
```
Expected: ImportError / AttributeError — new constants and SniffSession API don't exist yet.
**Step 3: Implement SniffSession**
Rewrite `pm3py/sniff/session.py`:
```python
"""Sniff sessions — live streaming with button-toggle, or legacy batch."""
import struct
import sys
import threading
import time
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.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:
"""Persistent sniff session with live streaming and button-toggle.
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, 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"
@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()
ser.reset_input_buffer()
ser.reset_output_buffer()
time.sleep(0.5)
ser.reset_input_buffer()
return cls(ser)
def start_15693(self, live: bool = True, idle_timeout_ms: int = 400) -> None:
"""Start 15693 sniff session.
live=True (default): streaming with button-toggle pause/resume.
live=False: legacy BigBuf batch mode, blocks until button exits.
"""
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:
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:
# Final response (legacy path shouldn't hit here, but safety)
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."""
from ..core.transport import PM3Transport
print("[Snf] Sniffing ISO 15693... press PM3 button to stop.")
# 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)
idx = buf.find(struct.pack("<I", RESP_PREAMBLE_MAGIC))
if idx < 0:
continue
if idx > 0:
buf = buf[idx:]
if len(buf) < RESP_PREAMBLE_SIZE:
continue
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...")
# Download BigBuf (synchronous, blocking reads)
# Send DOWNLOAD_BIGBUF command
from ..core.protocol import PM3_CMD_DATA_SIZE as CHUNK
dl_frame = encode_ng_frame(Cmd.DOWNLOAD_BIGBUF,
struct.pack("<QQQ", 0, CHUNK, 0)[:24])
self._ser.write(dl_frame)
# This is simplified — for legacy mode, we recommend using
# the async PM3Transport path via pm3.hf.iso15.sniff_decoded()
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()
```
**Step 4: Run tests**
```bash
cd /home/work/pm3py && python -m pytest tests/test_sniff_session.py -v
```
Expected: All pass.
**Step 5: Commit**
```bash
git add pm3py/sniff/session.py tests/test_sniff_session.py pm3py/core/protocol.py
git commit --no-gpg-sign -m "feat: rewrite SniffSession with live streaming and button-toggle"
```
---
### Task 4: Python — Update .pythonstartup.py and sniff/__init__.py
**Files:**
- Modify: `.pythonstartup.py:57-63` (SNIFF imports), `.pythonstartup.py:98-105` (auto-connect)
- Modify: `pm3py/sniff/__init__.py`
**Step 1: Update .pythonstartup.py**
In the SNIFF import group (lines 57-63), the SniffSession import stays. The key change is in the auto-connect section.
After the existing `if _MODE == "core":` block (lines 98-105), add sniff mode auto-session:
```python
if _MODE == "sniff":
try:
session = SniffSession.open()
print("Sniff session ready -- session.start_15693()")
except Exception as e:
print(f" (Sniff session failed: {e})")
```
**Step 2: Update sniff/__init__.py exports**
No changes needed — `SniffSession` is already exported. The new constants are in `core/protocol.py`.
**Step 3: Run existing tests to verify nothing broke**
```bash
cd /home/work/pm3py && python -m pytest tests/ -v
```
Expected: All existing tests pass.
**Step 4: Commit**
```bash
git add .pythonstartup.py
git commit --no-gpg-sign -m "feat: auto-create sniff session in REPL sniff mode"
```
---
### Task 5: Verify end-to-end with mocked serial
**Files:**
- Modify: `tests/test_sniff_session.py` (add integration-style test)
**Step 1: Add end-to-end test simulating full REPL workflow**
Add to `tests/test_sniff_session.py`:
```python
class TestSniffSessionEndToEnd:
"""Integration-style test: full start → stream → pause → resume → stop."""
def test_full_workflow(self):
"""Simulate complete sniff session lifecycle."""
inv_cmd = bytes([0x26, 0x01, 0x00])
inv_resp = bytes([0x00, 0xE0, 0x04, 0x12, 0x34, 0x56, 0x78])
rdbl_cmd = bytes([0x22, 0x20, 0x04])
rdbl_resp = bytes([0x00, 0xE1, 0x10, 0x06, 0x01])
frames = [
_make_status_frame(SNIFF_STATE_STARTED, 0),
# First burst
_make_sniff_stream_frame(0, 0, 100, 1000, inv_cmd),
_make_sniff_stream_frame(0, 1, 200, 1500, inv_resp),
_make_sniff_stream_frame(0, 0, 100, 2000, rdbl_cmd),
_make_sniff_stream_frame(0, 1, 200, 2500, rdbl_resp),
# Pause
_make_status_frame(SNIFF_STATE_PAUSED, 4),
# Resume
_make_status_frame(SNIFF_STATE_RESUMED, 4),
# Second burst
_make_sniff_stream_frame(0, 0, 100, 5000, inv_cmd),
_make_sniff_stream_frame(0, 1, 200, 5500, inv_resp),
# Stop
_make_status_frame(SNIFF_STATE_STOPPED, 6),
]
ser = FakeSerial(frames)
session = SniffSession(ser)
session.start_15693(live=True)
time.sleep(0.5)
assert session.state == "stopped"
assert len(session.entries) == 6
# Verify timestamps are monotonically increasing
timestamps = [e["timestamp"] for e in session.entries]
assert timestamps == sorted(timestamps)
# Verify directions alternate (reader/tag pairs)
directions = [e["direction"] for e in session.entries]
assert directions == [0, 1, 0, 1, 0, 1]
def test_close_cleans_up(self):
"""close() stops session and closes serial port."""
ser = FakeSerial([
_make_status_frame(SNIFF_STATE_STARTED, 0),
_make_status_frame(SNIFF_STATE_STOPPED, 0),
])
session = SniffSession(ser)
session.start_15693(live=True)
time.sleep(0.1)
session.close()
assert session.state == "stopped"
assert ser._closed
```
**Step 2: Run all tests**
```bash
cd /home/work/pm3py && python -m pytest tests/test_sniff_session.py -v
```
Expected: All pass.
**Step 3: Run full test suite**
```bash
cd /home/work/pm3py && python -m pytest tests/ -v
```
Expected: All pass, no regressions.
**Step 4: Commit**
```bash
git add tests/test_sniff_session.py
git commit --no-gpg-sign -m "test: add end-to-end sniff session workflow tests"
```
---
### Task 6: Firmware build verification
**Files:**
- No new files — verify firmware compiles
**Step 1: Build firmware**
```bash
cd /home/work/pm3py/firmware && make armsrc/obj/iso15693.o
```
Expected: Compiles without errors or warnings. If there are compile errors, fix them (likely missing includes or function signature mismatches).
**Step 2: Check for any forward declaration mismatches**
```bash
grep -rn "SniffIso15693" /home/work/pm3py/firmware/armsrc/*.h /home/work/pm3py/firmware/include/
```
Update any declarations to match the new 5-parameter signature.
**Step 3: Commit any fixes**
```bash
git add firmware/
git commit --no-gpg-sign -m "fix(fw): update SniffIso15693 forward declarations"
```
Reference in New Issue View Git Blame Copy Permalink