mirror of
https://github.com/RfidResearchGroup/ChameleonUltra.git
synced 2026-08-28 02:14:06 +00:00
feat(data): add LF capture analysis commands
This commit is contained in:
Niel Nielsen
@@ -35,8 +35,8 @@ SRC_FILES += \
|
||||
$(PROJ_DIR)/rfid/nfctag/lf/utils/circular_buffer.c \
|
||||
$(PROJ_DIR)/rfid/nfctag/lf/utils/manchester.c \
|
||||
$(PROJ_DIR)/rfid/nfctag/lf/protocols/em410x.c \
|
||||
$(PROJ_DIR)/rfid/nfctag/lf/protocols/ioprox.c \
|
||||
$(PROJ_DIR)/rfid/nfctag/lf/protocols/hidprox.c \
|
||||
$(PROJ_DIR)/rfid/nfctag/lf/protocols/ioprox.c \
|
||||
$(PROJ_DIR)/rfid/nfctag/lf/protocols/viking.c \
|
||||
$(PROJ_DIR)/rfid/nfctag/lf/protocols/wiegand.c \
|
||||
$(PROJ_DIR)/utils/dataframe.c \
|
||||
@@ -341,13 +341,15 @@ ifeq (${CURRENT_DEVICE_TYPE}, ${CHAMELEON_ULTRA})
|
||||
$(PROJ_DIR)/rfid/reader/hf/rc522.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_125khz_radio.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_em410x_data.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_em4x05_data.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_gap.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_reader_generic.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_reader_data.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_reader_main.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_t55xx_data.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_ioprox_data.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_hidprox_data.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_ioprox_data.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_viking_data.c \
|
||||
$(PROJ_DIR)/rfid/reader/lf/lf_reader_generic.c \
|
||||
|
||||
INC_FOLDERS +=\
|
||||
${PROJ_DIR}/rfid/reader/ \
|
||||
|
||||
@@ -14,6 +14,10 @@
|
||||
#include "settings.h"
|
||||
#include "delayed_reset.h"
|
||||
#include "netdata.h"
|
||||
#include "bsp_wdt.h"
|
||||
#include "lf_reader_generic.h"
|
||||
#include "lf_em4x05_data.h"
|
||||
#include "nfc_14a.h"
|
||||
|
||||
|
||||
#define NRF_LOG_MODULE_NAME app_cmd
|
||||
@@ -1729,6 +1733,114 @@ static data_frame_tx_t *cmd_processor_mf0_get_emulator_config(uint16_t cmd, uint
|
||||
* (cmd -> processor) function map, the map struct is:
|
||||
* cmd code before process cmd processor after process
|
||||
*/
|
||||
static data_frame_tx_t *cmd_processor_em4x05_scan(uint16_t cmd, uint16_t status, uint16_t length, uint8_t *data) {
|
||||
em4x05_data_t tag = {0};
|
||||
status = scan_em4x05(&tag);
|
||||
if (status != STATUS_LF_TAG_OK) {
|
||||
return data_frame_make(cmd, status, 0, NULL);
|
||||
}
|
||||
struct {
|
||||
uint32_t config;
|
||||
uint32_t uid;
|
||||
uint32_t uid_hi;
|
||||
uint8_t is_em4x69;
|
||||
} PACKED payload;
|
||||
payload.config = U32HTONL(tag.config);
|
||||
payload.uid = U32HTONL(tag.uid);
|
||||
payload.uid_hi = U32HTONL(tag.uid_hi);
|
||||
payload.is_em4x69 = tag.is_em4x69 ? 1 : 0;
|
||||
return data_frame_make(cmd, STATUS_LF_TAG_OK, sizeof(payload), (uint8_t *)&payload);
|
||||
}
|
||||
|
||||
static data_frame_tx_t *cmd_processor_lf_sniff(uint16_t cmd, uint16_t status, uint16_t length, uint8_t *data) {
|
||||
/* Optional 2-byte big-endian timeout in ms from host (default 2000ms) */
|
||||
uint32_t timeout_ms = 2000;
|
||||
if (length >= 2) {
|
||||
timeout_ms = ((uint32_t)data[0] << 8) | data[1];
|
||||
if (timeout_ms == 0 || timeout_ms > 10000) timeout_ms = 2000;
|
||||
}
|
||||
|
||||
static uint8_t sniff_buf[LF_SNIFF_MAX_SAMPLES];
|
||||
size_t outlen = 0;
|
||||
raw_read_to_buffer(sniff_buf, LF_SNIFF_MAX_SAMPLES, timeout_ms, &outlen);
|
||||
|
||||
if (outlen == 0) {
|
||||
return data_frame_make(cmd, STATUS_LF_TAG_NO_FOUND, 0, NULL);
|
||||
}
|
||||
return data_frame_make(cmd, STATUS_LF_TAG_OK, (uint16_t)outlen, sniff_buf);
|
||||
}
|
||||
#define HF_SNIFF_BUF_SIZE 3800 /* leave room for USB framing */
|
||||
#define HF_SNIFF_MAX_FRAMES 200
|
||||
|
||||
static uint8_t m_sniff_buf[HF_SNIFF_BUF_SIZE];
|
||||
static uint16_t m_sniff_buf_len = 0;
|
||||
static bool m_sniff_active = false;
|
||||
static uint16_t m_sniff_cb_count = 0; /* debug: total callback invocations */
|
||||
|
||||
static void hf14a_sniff_frame_cb(const uint8_t *data, uint16_t szBits) {
|
||||
m_sniff_cb_count++; /* count even if buffer full or inactive */
|
||||
if (!m_sniff_active) return;
|
||||
uint16_t szBytes = (szBits + 7) / 8;
|
||||
/* Check space: 2 bytes header + data */
|
||||
if (m_sniff_buf_len + 2 + szBytes > HF_SNIFF_BUF_SIZE) return;
|
||||
/* Write bit count big-endian */
|
||||
m_sniff_buf[m_sniff_buf_len++] = (szBits >> 8) & 0xFF;
|
||||
m_sniff_buf[m_sniff_buf_len++] = szBits & 0xFF;
|
||||
/* Write frame bytes */
|
||||
memcpy(&m_sniff_buf[m_sniff_buf_len], data, szBytes);
|
||||
m_sniff_buf_len += szBytes;
|
||||
}
|
||||
|
||||
static data_frame_tx_t *cmd_processor_hf14a_sniff(uint16_t cmd, uint16_t status, uint16_t length, uint8_t *data) {
|
||||
/* Optional 2-byte big-endian timeout in ms (default 5000ms) */
|
||||
uint32_t timeout_ms = 5000;
|
||||
if (length >= 2) {
|
||||
timeout_ms = ((uint32_t)data[0] << 8) | data[1];
|
||||
if (timeout_ms == 0 || timeout_ms > 30000) timeout_ms = 5000;
|
||||
}
|
||||
|
||||
/* Reload active slot data before sniffing.
|
||||
* The NFCT anti-collision response is built from m_tag_information which
|
||||
* points into the shared tag data buffer. After a slot switch the buffer
|
||||
* may still contain the previous slot's UID if the FDS async load has not
|
||||
* completed. A forced reload here ensures the correct UID is presented
|
||||
* during the sniff session.
|
||||
* A short settle delay follows to allow the reload to complete before
|
||||
* the first field detection can trigger the anti-collision path. */
|
||||
tag_emulation_load_data();
|
||||
bsp_delay_ms(100);
|
||||
|
||||
/* Install sniff callback into the already-running tag emulation stack.
|
||||
* Do NOT call tag_mode_enter() or sense_switch() here — those reinit
|
||||
* NFCT and wipe the anti-collision data, breaking the emulation.
|
||||
* The device must already be in emulator mode (hw mode --emulator)
|
||||
* with a slot active before running this command. */
|
||||
m_sniff_buf_len = 0;
|
||||
m_sniff_cb_count = 0;
|
||||
m_sniff_active = true;
|
||||
nfc_tag_14a_set_sniff_cb(hf14a_sniff_frame_cb);
|
||||
|
||||
/* Wait for duration, yielding each ms so USB stack stays alive.
|
||||
* Feed watchdog every iteration — WDT timeout is 5000ms and the
|
||||
* main loop cannot feed it while we are blocking here. */
|
||||
autotimer *p_at = bsp_obtain_timer(0);
|
||||
while (NO_TIMEOUT_1MS(p_at, timeout_ms)) {
|
||||
bsp_delay_ms(1);
|
||||
bsp_wdt_feed();
|
||||
}
|
||||
bsp_return_timer(p_at);
|
||||
|
||||
/* Remove callback and restore normal sense state */
|
||||
m_sniff_active = false;
|
||||
nfc_tag_14a_clear_sniff_cb();
|
||||
tag_emulation_sense_run(); /* restore slot-based sense state */
|
||||
|
||||
if (m_sniff_buf_len == 0) {
|
||||
return data_frame_make(cmd, STATUS_HF_TAG_NO, 0, NULL);
|
||||
}
|
||||
return data_frame_make(cmd, STATUS_SUCCESS, m_sniff_buf_len, m_sniff_buf);
|
||||
}
|
||||
|
||||
static cmd_data_map_t m_data_cmd_map[] = {
|
||||
{ DATA_CMD_GET_APP_VERSION, NULL, cmd_processor_get_app_version, NULL },
|
||||
{ DATA_CMD_CHANGE_DEVICE_MODE, NULL, cmd_processor_change_device_mode, NULL },
|
||||
@@ -1808,6 +1920,9 @@ static cmd_data_map_t m_data_cmd_map[] = {
|
||||
|
||||
{ DATA_CMD_IOPROX_DECODE_RAW, NULL, cmd_processor_ioprox_decode_raw, NULL },
|
||||
{ DATA_CMD_IOPROX_COMPOSE_ID, NULL, cmd_processor_ioprox_compose_id, NULL },
|
||||
{ DATA_CMD_EM4X05_SCAN, before_reader_run, cmd_processor_em4x05_scan, NULL },
|
||||
{ DATA_CMD_LF_SNIFF, before_reader_run, cmd_processor_lf_sniff, NULL },
|
||||
{ DATA_CMD_HF14A_SNIFF, NULL, cmd_processor_hf14a_sniff, NULL },
|
||||
|
||||
#endif
|
||||
|
||||
|
||||
@@ -19,6 +19,7 @@
|
||||
/////////////////////////////////////////////////////////////////////
|
||||
#define STATUS_LF_TAG_OK (0x40) // Some of the low -frequency cards are successful!
|
||||
#define STATUS_LF_TAG_NO_FOUND (0x41) // Can't search for valid LF tags
|
||||
#define STATUS_LF_TAG_LOGIN_REQUIRED (0x42) // Tag requires LOGIN before read
|
||||
|
||||
/////////////////////////////////////////////////////////////////////
|
||||
// other status
|
||||
|
||||
@@ -78,6 +78,7 @@
|
||||
|
||||
#define DATA_CMD_HF14A_GET_CONFIG (2200)
|
||||
#define DATA_CMD_HF14A_SET_CONFIG (2201)
|
||||
#define DATA_CMD_HF14A_SNIFF (2020)
|
||||
|
||||
//
|
||||
// ******************************************************************
|
||||
@@ -173,4 +174,8 @@
|
||||
#define DATA_CMD_IOPROX_SET_EMU_ID (5008)
|
||||
#define DATA_CMD_IOPROX_GET_EMU_ID (5009)
|
||||
|
||||
#define DATA_CMD_EM4X05_SCAN (3030)
|
||||
#define DATA_CMD_EM4X05_READSNIFF (3032)
|
||||
#define DATA_CMD_LF_SNIFF (3031)
|
||||
|
||||
#endif
|
||||
|
||||
@@ -59,6 +59,17 @@ const uint16_t ats_fsdi_table[] = {
|
||||
static volatile bool m_is_responded = false;
|
||||
// Receiving buffer
|
||||
static uint8_t m_nfc_rx_buffer[MAX_NFC_RX_BUFFER_SIZE] = { 0x00 };
|
||||
|
||||
/* Optional sniff callback — fires for every received frame */
|
||||
static nfc_tag_14a_sniff_cb_t m_sniff_cb = NULL;
|
||||
|
||||
void nfc_tag_14a_set_sniff_cb(nfc_tag_14a_sniff_cb_t cb) {
|
||||
m_sniff_cb = cb;
|
||||
}
|
||||
|
||||
void nfc_tag_14a_clear_sniff_cb(void) {
|
||||
m_sniff_cb = NULL;
|
||||
}
|
||||
static uint8_t m_nfc_tx_buffer[MAX_NFC_TX_BUFFER_SIZE] = { 0x00 };
|
||||
// The N -secondary connection needs to use SAK, when the "third 'bit' in SAK is 1 is 1, the logo UID is incomplete
|
||||
static uint8_t m_uid_incomplete_sak[] = { 0x04, 0xda, 0x17 };
|
||||
@@ -326,6 +337,11 @@ void nfc_tag_14a_data_process(uint8_t *p_data) {
|
||||
// Because of this error receiving event caused by this possible interference
|
||||
return;
|
||||
}
|
||||
|
||||
/* Sniff hook — fire before any tag response logic */
|
||||
if (m_sniff_cb != NULL) {
|
||||
m_sniff_cb(p_data, szDataBits);
|
||||
}
|
||||
// Manually draw frame, separate data and strange school inspection
|
||||
#if !NFC_TAG_14A_RX_PARITY_AUTO_DEL_ENABLE
|
||||
if (szDataBits >= 9) {
|
||||
|
||||
@@ -82,6 +82,14 @@ typedef struct {
|
||||
|
||||
// Communication reception function that needs to be implemented
|
||||
typedef void (*nfc_tag_14a_reset_handler_t)(void);
|
||||
|
||||
/* Sniff callback — called for every received frame before the tag handler.
|
||||
* data : raw frame bytes (after parity strip)
|
||||
* szBits : number of bits received */
|
||||
typedef void (*nfc_tag_14a_sniff_cb_t)(const uint8_t *data, uint16_t szBits);
|
||||
|
||||
void nfc_tag_14a_set_sniff_cb(nfc_tag_14a_sniff_cb_t cb);
|
||||
void nfc_tag_14a_clear_sniff_cb(void);
|
||||
typedef void (*nfc_tag_14a_state_handler_t)(uint8_t *data, uint16_t szBits);
|
||||
typedef nfc_tag_14a_coll_res_reference_t *(*nfc_tag_14a_coll_handler_t)(void);
|
||||
|
||||
|
||||
@@ -80,6 +80,7 @@ void tag_emulation_init(void);
|
||||
void tag_emulation_save(void);
|
||||
|
||||
// Starting and ending of the emulation card
|
||||
void tag_emulation_load_data(void);
|
||||
void tag_emulation_sense_run(void);
|
||||
void tag_emulation_sense_end(void);
|
||||
|
||||
|
||||
@@ -1,6 +1,11 @@
|
||||
#pragma once
|
||||
|
||||
#include "ble_main.h"
|
||||
#include "nrfx_pwm.h"
|
||||
|
||||
/* Exposed so lf_gap.c can stop the PWM and drive LF_ANT_DRIVER directly
|
||||
* to create clean field gaps without relying on PWM pin release state. */
|
||||
extern nrfx_pwm_t m_pwm;
|
||||
|
||||
void lf_125khz_radio_init(void);
|
||||
void lf_125khz_radio_uninit(void);
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
#include "lf_reader_generic.h"
|
||||
#include "lf_reader_data.h"
|
||||
|
||||
#include "bsp_delay.h"
|
||||
#include "bsp_wdt.h"
|
||||
#include "bsp_time.h"
|
||||
#include "circular_buffer.h"
|
||||
#include "lf_125khz_radio.h"
|
||||
@@ -13,15 +15,19 @@
|
||||
#include "nrf_log_default_backends.h"
|
||||
NRF_LOG_MODULE_REGISTER();
|
||||
|
||||
#define CIRCULAR_BUFFER_SIZE (128)
|
||||
/*
|
||||
* Circular buffer for SAADC samples.
|
||||
* Increased from 128 to 512 to reduce overrun risk during USB transfer.
|
||||
* The main loop drains it as fast as possible into the output buffer.
|
||||
*/
|
||||
#define CIRCULAR_BUFFER_SIZE (512)
|
||||
static circular_buffer cb;
|
||||
|
||||
// saadc irq is used to sample ANT GPIO.
|
||||
static void saadc_cb(nrf_saadc_value_t *vals, size_t size) {
|
||||
for (int i = 0; i < size; i++) {
|
||||
nrf_saadc_value_t val = vals[i];
|
||||
if (!cb_push_back(&cb, &val)) {
|
||||
return;
|
||||
return; /* buffer full — oldest samples dropped */
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -41,14 +47,20 @@ bool raw_read_to_buffer(uint8_t *data, size_t maxlen, uint32_t timeout_ms, size_
|
||||
init_saadc_hw();
|
||||
start_lf_125khz_radio();
|
||||
|
||||
/* Wait for antenna to settle before capturing.
|
||||
* The LC circuit rings for ~400µs on field startup, then takes
|
||||
* another ~800µs to reach steady state. Skip 2ms to be safe. */
|
||||
bsp_delay_ms(2);
|
||||
|
||||
autotimer *p_at = bsp_obtain_timer(0);
|
||||
while (NO_TIMEOUT_1MS(p_at, timeout_ms) && *outlen < maxlen) {
|
||||
uint16_t val = 0;
|
||||
while (cb_pop_front(&cb, &val) && *outlen < maxlen) {
|
||||
val = val >> 5; // 14 bit ADC to 8 bit value and /2 range
|
||||
data[*outlen] = val > 0xff ? 0xff : val;
|
||||
val = val >> 5; /* 14-bit ADC → 9-bit, then >>5 gives 8-bit */
|
||||
data[*outlen] = val > 0xff ? 0xff : (uint8_t)val;
|
||||
++(*outlen);
|
||||
}
|
||||
bsp_wdt_feed(); /* prevent watchdog reset during long captures */
|
||||
}
|
||||
|
||||
bsp_return_timer(p_at);
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
#pragma once
|
||||
|
||||
#include <stdbool.h>
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
|
||||
/*
|
||||
* Capture raw ADC samples from the LF antenna field.
|
||||
*
|
||||
* The SAADC samples at the PWM period rate (125kHz = 8µs/sample).
|
||||
* Each sample is an 8-bit value (14-bit ADC >> 5, clamped to 0xFF).
|
||||
* A steady carrier reads ~0x80-0x82; a gap reads noticeably lower.
|
||||
*
|
||||
* @param data Output buffer for raw samples
|
||||
* @param maxlen Max bytes to capture (max 4000 for USB frame limit)
|
||||
* @param timeout_ms Stop after this many ms even if buffer not full
|
||||
* @param outlen Actual number of bytes written
|
||||
* @return true on success
|
||||
*/
|
||||
/** Maximum bytes a single raw capture can return (USB frame limit). */
|
||||
#define LF_SNIFF_MAX_SAMPLES 4000
|
||||
|
||||
bool raw_read_to_buffer(uint8_t *data, size_t maxlen, uint32_t timeout_ms, size_t *outlen);
|
||||
@@ -5,6 +5,9 @@
|
||||
|
||||
#include "app_status.h"
|
||||
#include "lf_125khz_radio.h"
|
||||
#if defined(PROJECT_CHAMELEON_ULTRA)
|
||||
#include "lf_em4x05_data.h"
|
||||
#endif
|
||||
#include "lf_reader_data.h"
|
||||
|
||||
void set_scan_tag_timeout(uint32_t ms);
|
||||
|
||||
@@ -753,6 +753,10 @@ hf_mfu = hf.subgroup("mfu", "MIFARE Ultralight / NTAG commands")
|
||||
|
||||
lf = root.subgroup("lf", "Low Frequency commands")
|
||||
lf_em = lf.subgroup("em", "EM commands")
|
||||
lf_em_4x05 = lf_em.subgroup("4x05", "EM4x05/EM4x69 commands")
|
||||
data = root.subgroup('data', 'Data analysis and visualization commands')
|
||||
|
||||
|
||||
lf_em_410x = lf_em.subgroup("410x", "EM410x commands")
|
||||
lf_hid = lf.subgroup("hid", "HID commands")
|
||||
lf_hid_prox = lf_hid.subgroup("prox", "HID Prox commands")
|
||||
@@ -6734,3 +6738,965 @@ examples/notes:
|
||||
)
|
||||
else:
|
||||
print(f" [*] {color_string((CY, 'No response'))}")
|
||||
|
||||
|
||||
@lf_em_4x05.command("read")
|
||||
class LFEm4x05Read(ReaderRequiredUnit):
|
||||
def args_parser(self) -> ArgumentParserNoExit:
|
||||
parser = ArgumentParserNoExit()
|
||||
parser.description = (
|
||||
"Scan EM4x05 or EM4x69 tag (reader-talk-first) and print config, UID"
|
||||
)
|
||||
return parser
|
||||
|
||||
def on_exec(self, args: argparse.Namespace):
|
||||
try:
|
||||
pwd = int(args.pwd, 16) if hasattr(args, 'pwd') and args.pwd else 0
|
||||
except ValueError:
|
||||
print(f"{CR}Invalid password, expected hex{C0}")
|
||||
return
|
||||
(config, uid, uid_hi, is_em4x69, uid_block) = self.cmd.em4x05_scan(pwd=pwd)
|
||||
tag_label = "EM4x69" if is_em4x69 else "EM4x05"
|
||||
rl = bool((config >> 6) & 1)
|
||||
print(f" Tag type : {CG}{tag_label}{C0}")
|
||||
print(f" Config : {CG}{config:#010x}{C0}")
|
||||
print(f" UID block: {CG}{uid_block}{C0}")
|
||||
if rl:
|
||||
print(f" Auth : {CG}LOGIN used (pwd={args.pwd.upper() if hasattr(args, 'pwd') and args.pwd else '00000000'}){C0}")
|
||||
if is_em4x69:
|
||||
uid64 = (uid_hi << 32) | uid
|
||||
print(f" UID (64) : {CG}{uid64:016x}{C0}")
|
||||
else:
|
||||
print(f" UID : {CG}{uid:08x}{C0}")
|
||||
|
||||
|
||||
@lf.command('sniff')
|
||||
class LFSniff(ReaderRequiredUnit):
|
||||
def args_parser(self) -> ArgumentParserNoExit:
|
||||
parser = ArgumentParserNoExit()
|
||||
parser.description = (
|
||||
"Capture raw LF field ADC samples (125kHz, 8µs/sample). "
|
||||
"~0x80 = field on, lower values = gap or no field."
|
||||
)
|
||||
parser.add_argument(
|
||||
'--timeout', type=int, default=2000, metavar='MS',
|
||||
help='Capture duration in milliseconds (default: 2000, max: 10000, firmware blocks for full duration)'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--out', type=str, default=None, metavar='FILE',
|
||||
help='Save raw samples to binary file (for offline analysis)'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--hex', action='store_true',
|
||||
help='Print hex dump of samples to screen'
|
||||
)
|
||||
return parser
|
||||
|
||||
def on_exec(self, args: argparse.Namespace):
|
||||
timeout = max(1, min(10000, args.timeout))
|
||||
print(f" Capturing LF field for {timeout}ms at 125kHz (8µs/sample)...")
|
||||
resp = self.cmd.lf_sniff(timeout_ms=timeout)
|
||||
|
||||
if resp.status != Status.LF_TAG_OK or not resp.data:
|
||||
print(f"{CR}No samples captured{C0}")
|
||||
return
|
||||
|
||||
import chameleon_cli_unit as _self_mod
|
||||
data = bytes(resp.data)
|
||||
_self_mod._last_capture = data
|
||||
|
||||
n = len(data)
|
||||
duration_ms = n * 8 / 1000
|
||||
print(f" Captured : {CG}{n}{C0} bytes ({duration_ms:.1f}ms)")
|
||||
|
||||
mn = min(data)
|
||||
mx = max(data)
|
||||
mean = sum(data) // len(data)
|
||||
print(f" Range : {CG}0x{mn:02x}{C0} – {CG}0x{mx:02x}{C0} mean: {CG}0x{mean:02x}{C0}")
|
||||
|
||||
# Detect real field gaps — they drop to near zero (0x00-0x40),
|
||||
# well below the steady carrier (~0xb0). Use half of mean as threshold
|
||||
# to avoid false positives from the antenna startup transient.
|
||||
gap_threshold = mean // 2
|
||||
# Skip first 200 samples (1.6ms) to ignore startup ringing
|
||||
steady_data = data[200:]
|
||||
gap_count = sum(1 for b in steady_data if b < gap_threshold)
|
||||
if gap_count > 0:
|
||||
print(f" Gaps : {CG}{gap_count}{C0} samples below 0x{gap_threshold:02x} (real field drops)")
|
||||
else:
|
||||
print(f" Gaps : {CR}none detected — flat carrier (no gap commands sent){C0}")
|
||||
|
||||
if args.hex:
|
||||
print()
|
||||
print(f" addr {'hex bytes':47s} level")
|
||||
print(f" ---- {'-'*47} ----------------")
|
||||
for i in range(0, min(n, 256), 16):
|
||||
row = data[i:i+16]
|
||||
hex_part = ' '.join(f'{b:02x}' for b in row)
|
||||
bar = ''
|
||||
for b in row:
|
||||
if b < 0x10:
|
||||
bar += '_' # gap / field off
|
||||
elif b < 0x40:
|
||||
bar += '.' # ringing decay
|
||||
elif b < 0x80:
|
||||
bar += '-' # low
|
||||
elif b < 0xa0:
|
||||
bar += '+' # mid
|
||||
elif b < 0xc0:
|
||||
bar += 'o' # steady carrier
|
||||
elif b < 0xe0:
|
||||
bar += 'O' # high
|
||||
else:
|
||||
bar += '#' # clipped 0xff
|
||||
print(f" {i:04x} {hex_part:<47s} {bar}")
|
||||
if n > 256:
|
||||
print(f" ... ({n - 256} more bytes, use --out to save all)")
|
||||
print()
|
||||
print(" _ gap . ringing - low + mid o carrier O high # clipped")
|
||||
|
||||
if args.out:
|
||||
try:
|
||||
with open(args.out, 'wb') as f:
|
||||
f.write(data)
|
||||
print(f" Saved : {CG}{args.out}{C0} ({n} bytes)")
|
||||
except Exception as e:
|
||||
print(f"{CR}Failed to save: {e}{C0}")
|
||||
|
||||
|
||||
@hf_14a.command("info")
|
||||
@hf_14a.command('sniff')
|
||||
class HF14ASniff(BaseCLIUnit):
|
||||
def args_parser(self) -> ArgumentParserNoExit:
|
||||
parser = ArgumentParserNoExit()
|
||||
parser.description = (
|
||||
"Capture ISO14443A reader frames while CU acts as a tag. "
|
||||
"Place CU near a reader — all commands the reader sends are logged. "
|
||||
"Useful for understanding what a reader expects before configuring emulation."
|
||||
)
|
||||
parser.add_argument(
|
||||
'--timeout', type=int, default=5000, metavar='MS',
|
||||
help='Listen duration in milliseconds (default: 5000, max: 30000, firmware blocks for full duration)'
|
||||
)
|
||||
return parser
|
||||
|
||||
def on_exec(self, args: argparse.Namespace):
|
||||
timeout = max(1, min(30000, args.timeout))
|
||||
print(f" Listening for reader frames for {timeout}ms...")
|
||||
print(" Place CU near a reader now.")
|
||||
print()
|
||||
|
||||
try:
|
||||
resp = self.cmd.hf14a_sniff(timeout_ms=timeout)
|
||||
except Exception as e:
|
||||
if 'CMDInvalid' in type(e).__name__ or '2020' in str(e):
|
||||
print(f"{CR}Command not supported — reflash firmware to enable hf 14a sniff{C0}")
|
||||
else:
|
||||
print(f"{CR}{e}{C0}")
|
||||
return
|
||||
|
||||
if resp.status not in (Status.HF_TAG_OK, Status.SUCCESS):
|
||||
cb_count = 0
|
||||
if resp.data and len(resp.data) >= 2:
|
||||
cb_count = (resp.data[0] << 8) | resp.data[1]
|
||||
if cb_count > 0:
|
||||
print(f"{CY} Callback fired {cb_count}x but no valid frames buffered{C0}")
|
||||
else:
|
||||
print(" No frames captured — no reader detected")
|
||||
return
|
||||
|
||||
# Parse packed frame buffer: [2 bytes bits BE][N bytes data] ...
|
||||
buf = bytes(resp.data)
|
||||
frames = []
|
||||
i = 0
|
||||
while i + 2 <= len(buf):
|
||||
szBits = (buf[i] << 8) | buf[i+1]
|
||||
i += 2
|
||||
if szBits == 0:
|
||||
break
|
||||
szBytes = (szBits + 7) // 8
|
||||
if i + szBytes > len(buf):
|
||||
break
|
||||
raw = buf[i:i+szBytes]
|
||||
i += szBytes
|
||||
|
||||
# ISO14443-A frames include one parity bit per byte.
|
||||
# Short frames (< 8 bits, e.g. REQA=7 bits) have no parity.
|
||||
# All other frames: szBits = data_bytes * 9, strip every 9th bit.
|
||||
if szBits >= 8 and szBits % 9 == 0:
|
||||
n_bytes = szBits // 9
|
||||
all_bits = []
|
||||
for byte in raw:
|
||||
for b in range(8):
|
||||
all_bits.append((byte >> b) & 1)
|
||||
stripped = []
|
||||
for nb in range(n_bytes):
|
||||
val = 0
|
||||
for b in range(8):
|
||||
val |= all_bits[nb * 9 + b] << b
|
||||
stripped.append(val)
|
||||
data = bytes(stripped)
|
||||
szBits = n_bytes * 8
|
||||
else:
|
||||
data = raw
|
||||
|
||||
frames.append((szBits, data))
|
||||
|
||||
if not frames:
|
||||
print(f"{CR}No frames decoded{C0}")
|
||||
return
|
||||
|
||||
print(f" Captured : {CG}{len(frames)}{C0} frame(s)")
|
||||
print()
|
||||
print(f" {'#':>3} {'bits':>4} {'hex data':<42} decoded")
|
||||
print(f" {'---':>3} {'----':>4} {'-'*42} {'-'*35}")
|
||||
|
||||
for n, (szBits, data) in enumerate(frames):
|
||||
hex_str = ' '.join(f'{b:02x}' for b in data)
|
||||
decoded, col = _decode_14a_frame_col(data, szBits)
|
||||
print(f" {CY}{n+1:>3}{C0} {szBits:>4} {hex_str:<42} {col}{decoded}{C0}")
|
||||
|
||||
# Summary block
|
||||
print()
|
||||
_print_14a_sniff_summary(frames)
|
||||
|
||||
|
||||
|
||||
|
||||
def _decode_14a_frame_col(data: bytes, szBits: int):
|
||||
"""Return (description, colour) for a 14A frame."""
|
||||
if not data:
|
||||
return '', C0
|
||||
b0 = data[0]
|
||||
|
||||
# Short frames (7-bit)
|
||||
if szBits == 7:
|
||||
if b0 == 0x26:
|
||||
return 'REQA', CG
|
||||
if b0 == 0x52:
|
||||
return 'WUPA', CG
|
||||
return f'short(0x{b0:02x})', CC
|
||||
|
||||
# Anti-collision / Select
|
||||
if b0 == 0x93:
|
||||
if len(data) > 1 and data[1] == 0x70:
|
||||
uid = ' '.join(f'{b:02x}' for b in data[2:6]) if len(data) >= 6 else ''
|
||||
return f'SELECT CL1 UID={uid}', CB
|
||||
nvb = f'NVB={data[1]:02x}' if len(data) > 1 else ''
|
||||
return f'ANTICOLL CL1 {nvb}', CB
|
||||
if b0 == 0x95:
|
||||
if len(data) > 1 and data[1] == 0x70:
|
||||
uid = ' '.join(f'{b:02x}' for b in data[2:6]) if len(data) >= 6 else ''
|
||||
return f'SELECT CL2 UID={uid}', CB
|
||||
nvb = f'NVB={data[1]:02x}' if len(data) > 1 else ''
|
||||
return f'ANTICOLL CL2 {nvb}', CB
|
||||
if b0 == 0x97:
|
||||
if len(data) > 1 and data[1] == 0x70:
|
||||
uid = ' '.join(f'{b:02x}' for b in data[2:6]) if len(data) >= 6 else ''
|
||||
return f'SELECT CL3 UID={uid}', CB
|
||||
nvb = f'NVB={data[1]:02x}' if len(data) > 1 else ''
|
||||
return f'ANTICOLL CL3 {nvb}', CB
|
||||
|
||||
# HALT (0x50 0x00 + CRC — b1 may vary after parity strip)
|
||||
if b0 == 0x50:
|
||||
return 'HALT', CC
|
||||
|
||||
# S-DESELECT (ISO14443-4 block)
|
||||
if b0 == 0xc2:
|
||||
return 'S-DESELECT', CC
|
||||
|
||||
# PPS
|
||||
if b0 == 0xd0:
|
||||
return f'PPS PPS1={data[1]:02x}' if len(data) > 1 else 'PPS', CC
|
||||
|
||||
# RATS
|
||||
if b0 == 0xe0:
|
||||
fsdi = (data[1] >> 4) if len(data) > 1 else 0
|
||||
cid = (data[1] & 0xf) if len(data) > 1 else 0
|
||||
return f'RATS FSDI={fsdi} CID={cid}', CC
|
||||
|
||||
# MIFARE Classic commands
|
||||
if b0 == 0x60:
|
||||
return f'AUTH KeyA block={data[1]}' if len(data) > 1 else 'AUTH KeyA', CR
|
||||
if b0 == 0x61:
|
||||
return f'AUTH KeyB block={data[1]}' if len(data) > 1 else 'AUTH KeyB', CR
|
||||
# Encrypted nonce / auth response (follows AUTH, first byte varies)
|
||||
if szBits == 72:
|
||||
return '(encrypted nonce — auth challenge/response)', CC
|
||||
|
||||
if b0 == 0x30:
|
||||
return f'READ block={data[1]}' if len(data) > 1 else 'READ', CC
|
||||
if b0 == 0xa0:
|
||||
return f'WRITE block={data[1]}' if len(data) > 1 else 'WRITE', CY
|
||||
if b0 == 0x40:
|
||||
return 'MAGIC WUPC1', CY
|
||||
if b0 == 0x43:
|
||||
return 'MAGIC WUPC2', CY
|
||||
if b0 == 0x41:
|
||||
return 'MAGIC WIPE', CR
|
||||
|
||||
# ISO 7816-4 APDUs
|
||||
if len(data) >= 2 and b0 in (0x00, 0x80, 0x90, 0xa0):
|
||||
cla, ins = data[0], data[1]
|
||||
p1 = data[2] if len(data) > 2 else 0
|
||||
p2 = data[3] if len(data) > 3 else 0
|
||||
# SELECT FILE / AID
|
||||
if cla == 0x00 and ins == 0xa4:
|
||||
if len(data) > 5:
|
||||
aid = ' '.join(f'{b:02x}' for b in data[5:5+data[4]])
|
||||
# Identify known AIDs
|
||||
aid_raw = bytes(data[5:5+data[4]])
|
||||
name = _known_aid(aid_raw)
|
||||
label = f'SELECT AID {aid.upper()}'
|
||||
if name:
|
||||
label += f' ({name})'
|
||||
return label, CY
|
||||
return 'SELECT', CY
|
||||
# READ BINARY
|
||||
if cla == 0x00 and ins == 0xb0:
|
||||
return f'READ BINARY off={p1<<8|p2} len={data[4] if len(data)>4 else 0}', CC
|
||||
# READ RECORD
|
||||
if cla == 0x00 and ins == 0xb2:
|
||||
sfi = p2 >> 3
|
||||
return f'READ RECORD SFI={sfi} rec={p1}', CC
|
||||
# GET DATA
|
||||
if cla == 0x80 and ins == 0xca:
|
||||
tag = (p1 << 8) | p2
|
||||
name = _known_bertag(tag)
|
||||
return f'GET DATA {p1:02x}{p2:02x}' + (f' ({name})' if name else ''), CC
|
||||
# GET PROCESSING OPTIONS
|
||||
if cla == 0x80 and ins == 0xa8:
|
||||
return 'GPO (Get Processing Options)', CY
|
||||
# GENERATE AC
|
||||
if cla == 0x80 and ins == 0xae:
|
||||
actype = {0x00:'AAC', 0x40:'TC', 0x80:'ARQC'}.get(p1 & 0xc0, f'AC/{p1:02x}')
|
||||
return f'GENERATE AC requesting {actype}', CR
|
||||
# VERIFY
|
||||
if cla == 0x00 and ins == 0x20:
|
||||
return 'VERIFY PIN', CY
|
||||
# INTERNAL AUTHENTICATE
|
||||
if cla == 0x00 and ins == 0x88:
|
||||
return 'INTERNAL AUTH', CR
|
||||
# EXTERNAL AUTHENTICATE
|
||||
if cla == 0x00 and ins == 0x82:
|
||||
return 'EXTERNAL AUTH', CR
|
||||
# MANAGE CHANNEL
|
||||
if cla == 0x00 and ins == 0x70:
|
||||
return 'MANAGE CHANNEL', CC
|
||||
return f'APDU CLA={cla:02x} INS={ins:02x} P1={p1:02x} P2={p2:02x}', CY
|
||||
|
||||
# Unknown — show first byte
|
||||
return f'unknown (0x{b0:02x})', CC
|
||||
|
||||
|
||||
|
||||
|
||||
def _known_aid(aid: bytes) -> str:
|
||||
table = {
|
||||
bytes.fromhex('a0000000031010'): 'Visa Credit/Debit',
|
||||
bytes.fromhex('a0000000032010'): 'Visa Electron',
|
||||
bytes.fromhex('a0000000033010'): 'Visa Classic',
|
||||
bytes.fromhex('a0000000038010'): 'Visa Plus',
|
||||
bytes.fromhex('a0000000041010'): 'Mastercard',
|
||||
bytes.fromhex('a0000000043060'): 'Maestro',
|
||||
bytes.fromhex('a000000025010801'): 'AmEx',
|
||||
bytes.fromhex('a0000000181002'): 'Mastercard Debit',
|
||||
bytes.fromhex('d2760000850101'): 'NDEF (NFC Forum)',
|
||||
bytes.fromhex('d27600002545'): 'NDEF Type 4',
|
||||
bytes.fromhex('315041592e5359532e4444463031'): 'PPSE (2PAY.SYS.DDF01)',
|
||||
}
|
||||
return table.get(aid, '')
|
||||
|
||||
|
||||
|
||||
|
||||
def _known_bertag(tag: int) -> str:
|
||||
table = {
|
||||
0x9f36: 'ATC',
|
||||
0x9f13: 'Last Online ATC',
|
||||
0x9f17: 'PIN Try Counter',
|
||||
0x9f4f: 'Log Format',
|
||||
0x9f4e: 'Merchant Name',
|
||||
}
|
||||
return table.get(tag, '')
|
||||
|
||||
|
||||
|
||||
|
||||
def _print_14a_sniff_summary(frames):
|
||||
"""Print a decoded summary of the sniff session."""
|
||||
uid_cl1 = None
|
||||
uid_cl2 = None
|
||||
uid_cl3 = None
|
||||
aids = []
|
||||
auth_blocks = [] # (key_type, block)
|
||||
auth_seen = False
|
||||
arqc_seen = False
|
||||
tc_seen = False
|
||||
halted = False
|
||||
rats_seen = False
|
||||
atc_tag = None
|
||||
amount = None
|
||||
|
||||
for szBits, data in frames:
|
||||
if not data:
|
||||
continue
|
||||
b0 = data[0]
|
||||
|
||||
# Extract UID from anticoll frames (NVB != 70 = anticoll, NVB = 70 = select)
|
||||
# Anticoll frame with NVB=41 means we're requesting UID bytes
|
||||
# The *response* from the tag contains the UID — but we only see reader frames
|
||||
# So extract from SELECT (NVB=70) which contains the full UID
|
||||
if b0 in (0x93, 0x95, 0x97) and len(data) >= 5 and data[1] == 0x70:
|
||||
uid_bytes = bytes(data[2:6])
|
||||
if b0 == 0x93:
|
||||
if data[2] == 0x88:
|
||||
uid_cl1 = None # cascade tag, UID continues in CL2
|
||||
else:
|
||||
uid_cl1 = uid_bytes
|
||||
elif b0 == 0x95:
|
||||
uid_cl2 = uid_bytes
|
||||
elif b0 == 0x97:
|
||||
uid_cl3 = uid_bytes
|
||||
|
||||
# Extract partial UID from anticoll frames — the tag sends back UID bytes
|
||||
# We capture the reader's anticoll command which may contain partial UID
|
||||
# NVB high nibble = number of full bytes sent, low nibble = bits
|
||||
# NVB=41 means reader sent 4 bits, so tag should respond with rest
|
||||
# NVB=e1 (225) is unusual — may be tag response captured by NFCT
|
||||
|
||||
# RATS
|
||||
if b0 == 0xe0:
|
||||
rats_seen = True
|
||||
|
||||
# SELECT AID
|
||||
if b0 == 0x00 and len(data) > 5 and data[1] == 0xa4:
|
||||
aid = bytes(data[5:5+data[4]])
|
||||
name = _known_aid(aid)
|
||||
entry = aid.hex().upper()
|
||||
if name:
|
||||
entry += f' ({name})'
|
||||
if entry not in aids:
|
||||
aids.append(entry)
|
||||
|
||||
# MIFARE Classic auth
|
||||
if b0 in (0x60, 0x61) and len(data) > 1:
|
||||
auth_seen = True
|
||||
key_type = 'KeyA' if b0 == 0x60 else 'KeyB'
|
||||
block = data[1]
|
||||
if (key_type, block) not in auth_blocks:
|
||||
auth_blocks.append((key_type, block))
|
||||
|
||||
# GENERATE AC — check AC type
|
||||
if b0 == 0x80 and len(data) > 2 and data[1] == 0xae:
|
||||
if (data[2] & 0xc0) == 0x80:
|
||||
arqc_seen = True
|
||||
if (data[2] & 0xc0) == 0x40:
|
||||
tc_seen = True
|
||||
|
||||
# GET DATA — ATC
|
||||
if b0 == 0x80 and len(data) > 2 and data[1] == 0xca:
|
||||
tag = (data[2] << 8) | data[3]
|
||||
atc_tag = _known_bertag(tag) or f'{data[2]:02x}{data[3]:02x}'
|
||||
|
||||
# GPO — extract amount if PDOL present
|
||||
if b0 == 0x80 and len(data) > 4 and data[1] == 0xa8:
|
||||
# Amount is usually first 6 bytes of PDOL data at offset 4+
|
||||
if len(data) >= 11:
|
||||
amt_bytes = data[5:11]
|
||||
amt = int.from_bytes(amt_bytes, 'big')
|
||||
if amt > 0:
|
||||
amount = amt
|
||||
|
||||
# HALT / DESELECT
|
||||
if b0 == 0x50 or b0 == 0xc2:
|
||||
halted = True
|
||||
|
||||
# Build UID from cascade levels
|
||||
uid_bytes = None
|
||||
if uid_cl1 and uid_cl2:
|
||||
uid_bytes = uid_cl1 + uid_cl2
|
||||
if uid_cl3:
|
||||
uid_bytes = uid_bytes + uid_cl3
|
||||
elif uid_cl1:
|
||||
uid_bytes = uid_cl1
|
||||
|
||||
# Do NOT attempt to extract UID from anticoll frames (81-bit NVB=e1):
|
||||
# those frames are the CU's own emulated tag responding, so the UID
|
||||
# would always be the active slot's UID — not useful information.
|
||||
# Only report UID when we see a completed SELECT (NVB=70).
|
||||
|
||||
print(f" {'─'*55}")
|
||||
if uid_bytes:
|
||||
uid_str = ' '.join(f'{b:02X}' for b in uid_bytes)
|
||||
cascade = f' ({len(uid_bytes)}-byte UID)' if uid_bytes else ''
|
||||
print(f" {CC}UID :{C0} {CG}{uid_str}{cascade}{C0}")
|
||||
if rats_seen:
|
||||
print(f" {CC}Protocol :{C0} ISO14443-4 (RATS seen)")
|
||||
for aid in aids:
|
||||
print(f" {CC}AID :{C0} {CY}{aid}{C0}")
|
||||
if amount is not None:
|
||||
major = amount // 100
|
||||
minor = amount % 100
|
||||
print(f" {CC}Amount :{C0} {CG}{major}.{minor:02d}{C0} (raw={amount})")
|
||||
if auth_blocks:
|
||||
for key_type, block in auth_blocks:
|
||||
print(f" {CC}Auth :{C0} {CR}MIFARE Classic {key_type} block={block}{C0}")
|
||||
elif auth_seen:
|
||||
print(f" {CC}Auth :{C0} {CR}MIFARE Classic auth detected{C0}")
|
||||
if arqc_seen:
|
||||
print(f" {CC}Auth type:{C0} {CR}ARQC — online authorisation requested{C0}")
|
||||
if tc_seen:
|
||||
print(f" {CC}Auth type:{C0} {CG}TC — approved offline{C0}")
|
||||
if atc_tag:
|
||||
print(f" {CC}ATC :{C0} tag {atc_tag} (transaction counter)")
|
||||
if halted:
|
||||
print(f" {CC}End :{C0} HALT / DESELECT")
|
||||
if not uid_bytes and not aids and not auth_seen and not rats_seen:
|
||||
print(f" {CC}Note :{C0} anti-collision incomplete — no SELECT seen (reader could not complete exchange)")
|
||||
|
||||
|
||||
def _get_capture():
|
||||
"""Return last capture buffer or print error."""
|
||||
import chameleon_cli_unit as _m
|
||||
if not _m._last_capture:
|
||||
return None
|
||||
return _m._last_capture
|
||||
|
||||
|
||||
|
||||
|
||||
@data.command('hexsamples')
|
||||
class DataHexsamples(BaseCLIUnit):
|
||||
def args_parser(self) -> ArgumentParserNoExit:
|
||||
parser = ArgumentParserNoExit()
|
||||
parser.description = 'Dump last LF sniff capture as hex bytes (PM3 style)'
|
||||
parser.add_argument('-n', '--num', type=int, default=512, metavar='N',
|
||||
help='Number of bytes to display (default: 512)')
|
||||
return parser
|
||||
|
||||
def on_exec(self, args: argparse.Namespace):
|
||||
buf = _get_capture()
|
||||
if buf is None:
|
||||
print(f"{CR}No capture in buffer — run lf sniff first{C0}")
|
||||
return
|
||||
n = min(args.num, len(buf))
|
||||
print(f" Buffer: {CG}{len(buf)}{C0} bytes total, showing {n}")
|
||||
print()
|
||||
for row in range(0, n, 16):
|
||||
chunk = buf[row:row+16]
|
||||
hex_part = ' '.join(f'{b:02x}' for b in chunk)
|
||||
bar = ''
|
||||
for b in chunk:
|
||||
if b < 0x10:
|
||||
bar += '_'
|
||||
elif b < 0x40:
|
||||
bar += '.'
|
||||
elif b < 0x80:
|
||||
bar += '-'
|
||||
elif b < 0xa0:
|
||||
bar += '+'
|
||||
elif b < 0xc0:
|
||||
bar += 'o'
|
||||
elif b < 0xe0:
|
||||
bar += 'O'
|
||||
else:
|
||||
bar += '#'
|
||||
print(f" {row // 16 :02d} | {hex_part:<47s} | {bar}")
|
||||
print()
|
||||
print(" _ gap . ringing - low + mid o carrier O high # clipped")
|
||||
|
||||
|
||||
|
||||
@data.command('plot')
|
||||
class DataPlot(BaseCLIUnit):
|
||||
def args_parser(self) -> ArgumentParserNoExit:
|
||||
parser = ArgumentParserNoExit()
|
||||
parser.description = 'Graphical waveform plot of last LF sniff capture (PyQt5 or matplotlib)'
|
||||
parser.add_argument('--start', type=int, default=0, metavar='N',
|
||||
help='Start sample (default: 0)')
|
||||
parser.add_argument('--len', type=int, default=4000, metavar='N',
|
||||
help='Number of samples to plot (default: all)')
|
||||
parser.add_argument('--ascii', action='store_true',
|
||||
help='Force ASCII plot even if GUI is available')
|
||||
return parser
|
||||
|
||||
def on_exec(self, args: argparse.Namespace):
|
||||
buf = _get_capture()
|
||||
if buf is None:
|
||||
print(f"{CR}No capture in buffer — run lf sniff first{C0}")
|
||||
return
|
||||
|
||||
start = max(0, args.start)
|
||||
end = min(len(buf), start + args.len)
|
||||
view = list(buf[start:end])
|
||||
n = len(view)
|
||||
|
||||
# X axis: time in µs (1 sample = 8µs)
|
||||
xs = [((start + i) * 8) for i in range(n)]
|
||||
|
||||
mean = sum(view) // n
|
||||
threshold = mean // 2
|
||||
|
||||
if not args.ascii:
|
||||
# Try PyQt5 first, then matplotlib
|
||||
try:
|
||||
from PyQt5.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QWidget
|
||||
from PyQt5.QtCore import Qt
|
||||
import pyqtgraph as pg
|
||||
_plot_pyqtgraph(xs, view, mean, threshold, start, end)
|
||||
return
|
||||
except ImportError:
|
||||
pass
|
||||
try:
|
||||
import matplotlib
|
||||
matplotlib.use('Qt5Agg')
|
||||
import matplotlib.pyplot as plt
|
||||
_plot_matplotlib(xs, view, mean, threshold, start, end)
|
||||
return
|
||||
except ImportError:
|
||||
pass
|
||||
try:
|
||||
import matplotlib.pyplot as plt
|
||||
_plot_matplotlib(xs, view, mean, threshold, start, end)
|
||||
return
|
||||
except ImportError:
|
||||
print(" No GUI library found (install PyQt5+pyqtgraph or matplotlib)")
|
||||
print(" Falling back to ASCII plot...")
|
||||
|
||||
# ASCII fallback
|
||||
w = 64
|
||||
bsize = max(1, n // w)
|
||||
buckets = []
|
||||
for i in range(0, n, bsize):
|
||||
chunk = view[i:i+bsize]
|
||||
buckets.append(sum(chunk) // len(chunk))
|
||||
buckets = buckets[:w]
|
||||
mn, mx = min(view), max(view)
|
||||
print(f" Samples {start}–{end} range 0x{mn:02x}–0x{mx:02x} mean 0x{mean:02x}")
|
||||
print()
|
||||
levels = [0xe0, 0xc0, 0xa0, 0x80, 0x60, 0x40, 0x20, 0x00]
|
||||
labels = ['0xff','0xc0','0xa0','0x80','0x60','0x40','0x20','0x00']
|
||||
for thresh, lbl in zip(levels, labels):
|
||||
row = ''.join('#' if v >= thresh else ' ' for v in buckets)
|
||||
print(f" {lbl} |{row}|")
|
||||
print(f" +{'-'*len(buckets)}+")
|
||||
|
||||
|
||||
|
||||
|
||||
def _plot_matplotlib(xs, ys, mean, threshold, start, end):
|
||||
import matplotlib.pyplot as plt
|
||||
import matplotlib.patches as mpatches
|
||||
|
||||
fig, ax = plt.subplots(figsize=(14, 5))
|
||||
fig.patch.set_facecolor('#1a1a2e')
|
||||
ax.set_facecolor('#0d1117')
|
||||
|
||||
# Main waveform
|
||||
ax.plot(xs, ys, color='#00e5ff', linewidth=0.8, label='LF field')
|
||||
|
||||
# Mean and gap threshold lines
|
||||
ax.axhline(mean, color='#ffb020', linewidth=0.8, linestyle='--', label=f'mean 0x{mean:02x}')
|
||||
ax.axhline(threshold, color='#ff3d57', linewidth=0.8, linestyle=':', label=f'gap threshold 0x{threshold:02x}')
|
||||
|
||||
# Shade gap regions
|
||||
in_gap = False
|
||||
gap_start = 0
|
||||
for i, v in enumerate(ys):
|
||||
if not in_gap and v < threshold:
|
||||
in_gap = True
|
||||
gap_start = xs[i]
|
||||
elif in_gap and v >= threshold:
|
||||
ax.axvspan(gap_start, xs[i], alpha=0.25, color='#ff3d57', linewidth=0)
|
||||
in_gap = False
|
||||
if in_gap:
|
||||
ax.axvspan(gap_start, xs[-1], alpha=0.25, color='#ff3d57', linewidth=0)
|
||||
|
||||
ax.set_xlabel('Time (µs)', color='#8899b4')
|
||||
ax.set_ylabel('ADC value', color='#8899b4')
|
||||
ax.set_title(f'LF Sniff — samples {start}–{end} ({(end-start)*8}µs)',
|
||||
color='#dde8f5', fontsize=11)
|
||||
ax.set_ylim(0, 270)
|
||||
ax.set_xlim(xs[0], xs[-1])
|
||||
ax.tick_params(colors='#8899b4')
|
||||
for spine in ax.spines.values():
|
||||
spine.set_edgecolor('#21262d')
|
||||
ax.legend(facecolor='#161b22', edgecolor='#30363d', labelcolor='#c9d1d9',
|
||||
fontsize=8, loc='upper right')
|
||||
ax.grid(True, color='#21262d', linewidth=0.5)
|
||||
|
||||
gap_patch = mpatches.Patch(color='#ff3d57', alpha=0.4, label='field gap')
|
||||
handles, labels = ax.get_legend_handles_labels()
|
||||
ax.legend(handles + [gap_patch], labels + ['field gap'],
|
||||
facecolor='#161b22', edgecolor='#30363d',
|
||||
labelcolor='#c9d1d9', fontsize=8, loc='upper right')
|
||||
|
||||
plt.tight_layout()
|
||||
plt.show()
|
||||
|
||||
|
||||
|
||||
|
||||
def _plot_pyqtgraph(xs, ys, mean, threshold, start, end):
|
||||
import sys
|
||||
from PyQt5.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QWidget, QLabel
|
||||
from PyQt5.QtCore import Qt
|
||||
from PyQt5.QtGui import QFont
|
||||
import pyqtgraph as pg
|
||||
|
||||
pg.setConfigOption('background', '#0d1117')
|
||||
pg.setConfigOption('foreground', '#8899b4')
|
||||
|
||||
app = QApplication.instance() or QApplication(sys.argv)
|
||||
|
||||
win = pg.GraphicsLayoutWidget(title='ChameleonUltra — LF Sniff')
|
||||
win.resize(1200, 400)
|
||||
win.setWindowTitle(f'LF Sniff — samples {start}–{end} ({(end-start)*8}µs)')
|
||||
|
||||
plot = win.addPlot()
|
||||
plot.setLabel('bottom', 'Time (µs)')
|
||||
plot.setLabel('left', 'ADC value')
|
||||
plot.showGrid(x=True, y=True, alpha=0.2)
|
||||
plot.setYRange(0, 270)
|
||||
|
||||
# Waveform
|
||||
plot.plot(xs, ys, pen=pg.mkPen('#00e5ff', width=1))
|
||||
|
||||
# Mean line
|
||||
plot.addLine(y=mean, pen=pg.mkPen('#ffb020', width=1, style=pg.QtCore.Qt.DashLine))
|
||||
# Gap threshold line
|
||||
plot.addLine(y=threshold, pen=pg.mkPen('#ff3d57', width=1, style=pg.QtCore.Qt.DotLine))
|
||||
|
||||
# Shade gaps
|
||||
for i in range(len(ys)-1):
|
||||
if ys[i] < threshold:
|
||||
r = pg.LinearRegionItem([xs[i], xs[i+1]],
|
||||
brush=pg.mkBrush(255, 61, 87, 40),
|
||||
pen=pg.mkPen(None), movable=False)
|
||||
plot.addItem(r)
|
||||
|
||||
# Legend / info panel
|
||||
legend_text = (
|
||||
'<span style="color:#8899b4; font-size:11px;">'
|
||||
'<span style="color:#00e5ff;">━</span> LF field (ADC) '
|
||||
'<span style="color:#ffb020;">- -</span> Mean '
|
||||
'<span style="color:#ff3d57;">···</span> Gap threshold (mean÷2) '
|
||||
'<span style="background:#ff3d57; opacity:0.3;"> </span>'
|
||||
' Field gap (below threshold) '
|
||||
'<span style="color:#8899b4;">Ringing = exponential rise on field restore</span>'
|
||||
'</span>'
|
||||
)
|
||||
legend = pg.LabelItem(legend_text, justify='left')
|
||||
win.addItem(legend, row=1, col=0)
|
||||
|
||||
win.show()
|
||||
app.exec_()
|
||||
|
||||
|
||||
|
||||
|
||||
@data.command('manrawdecode')
|
||||
class DataManrawdecode(BaseCLIUnit):
|
||||
def args_parser(self) -> ArgumentParserNoExit:
|
||||
parser = ArgumentParserNoExit()
|
||||
parser.description = 'Manchester decode the last LF sniff capture'
|
||||
parser.add_argument('--clock', type=int, default=64, metavar='N',
|
||||
help='Clock divisor in Tc (default: 64 = RF/64)')
|
||||
parser.add_argument('--invert', action='store_true',
|
||||
help='Invert logic (high=0, low=1)')
|
||||
return parser
|
||||
|
||||
def on_exec(self, args: argparse.Namespace):
|
||||
buf = _get_capture()
|
||||
if buf is None:
|
||||
print(f"{CR}No capture in buffer — run lf sniff first{C0}")
|
||||
return
|
||||
|
||||
# Binarise: above mean = 1 (carrier), below = 0 (gap)
|
||||
mean = sum(buf) // len(buf)
|
||||
threshold = mean // 2
|
||||
bits_raw = [1 if b > threshold else 0 for b in buf]
|
||||
if args.invert:
|
||||
bits_raw = [1 - b for b in bits_raw]
|
||||
|
||||
# Find transitions and measure run lengths
|
||||
runs = []
|
||||
cur = bits_raw[0]
|
||||
count = 1
|
||||
for b in bits_raw[1:]:
|
||||
if b == cur:
|
||||
count += 1
|
||||
else:
|
||||
runs.append((cur, count))
|
||||
cur = b
|
||||
count = 1
|
||||
runs.append((cur, count))
|
||||
|
||||
# Clock period in samples (1 sample = 8µs)
|
||||
half_clk = args.clock // 2 # samples per half-bit
|
||||
|
||||
# Decode Manchester: half-bit transitions
|
||||
# Low->High = 0, High->Low = 1 (standard Manchester)
|
||||
decoded_bits = []
|
||||
tol = max(2, half_clk // 3)
|
||||
|
||||
i = 0
|
||||
while i < len(runs):
|
||||
val, cnt = runs[i]
|
||||
# Short run = half period, long run = full period
|
||||
half = abs(cnt - half_clk) <= tol
|
||||
full = abs(cnt - args.clock) <= tol
|
||||
if half:
|
||||
# need next run to complete bit
|
||||
if i + 1 < len(runs):
|
||||
nval, ncnt = runs[i+1]
|
||||
nhalf = abs(ncnt - half_clk) <= tol
|
||||
if nhalf:
|
||||
# two halves: transition val->nval
|
||||
if val == 0 and nval == 1:
|
||||
decoded_bits.append(0)
|
||||
elif val == 1 and nval == 0:
|
||||
decoded_bits.append(1)
|
||||
i += 2
|
||||
continue
|
||||
elif full:
|
||||
# biphase / stay same level for full period = repeated bit
|
||||
decoded_bits.append(val)
|
||||
i += 1
|
||||
|
||||
if not decoded_bits:
|
||||
print(f"{CR}No bits decoded — check clock rate or signal quality{C0}")
|
||||
print(f" Mean threshold: 0x{threshold:02x} Clock: RF/{args.clock}")
|
||||
return
|
||||
|
||||
bits_str = ''.join(str(b) for b in decoded_bits)
|
||||
hex_str = hex(int(bits_str, 2))[2:] if decoded_bits else ''
|
||||
|
||||
print(f" Clock : RF/{args.clock} ({args.clock} Tc = {args.clock*8}µs/bit)")
|
||||
print(f" Threshold: 0x{threshold:02x} Inverted: {args.invert}")
|
||||
print(f" Bits : {CG}{len(decoded_bits)}{C0}")
|
||||
print()
|
||||
# Print in rows of 64
|
||||
for i in range(0, len(bits_str), 64):
|
||||
print(f" {bits_str[i:i+64]}")
|
||||
if hex_str:
|
||||
print()
|
||||
print(f" Hex: {CG}{hex_str[:64]}{C0}{'...' if len(hex_str) > 64 else ''}")
|
||||
|
||||
|
||||
|
||||
|
||||
@data.command('modulation')
|
||||
class DataModulation(BaseCLIUnit):
|
||||
def args_parser(self) -> ArgumentParserNoExit:
|
||||
parser = ArgumentParserNoExit()
|
||||
parser.description = 'Detect clock rate and modulation type in last LF capture'
|
||||
return parser
|
||||
|
||||
def on_exec(self, args: argparse.Namespace):
|
||||
buf = _get_capture()
|
||||
if buf is None:
|
||||
print(f"{CR}No capture in buffer — run lf sniff first{C0}")
|
||||
return
|
||||
|
||||
n = len(buf)
|
||||
mean = sum(buf) // n
|
||||
mn = min(buf)
|
||||
mx = max(buf)
|
||||
threshold = mean // 2
|
||||
|
||||
print(f" Samples : {CG}{n}{C0} ({n*8}µs)")
|
||||
print(f" Range : 0x{mn:02x} – 0x{mx:02x} mean: 0x{mean:02x}")
|
||||
print()
|
||||
|
||||
# Check if there is any modulation at all
|
||||
dynamic_range = mx - mn
|
||||
if dynamic_range < 0x20:
|
||||
print(f" Modulation: {CR}none — flat carrier (no signal){C0}")
|
||||
return
|
||||
|
||||
# Binarise
|
||||
bits = [1 if b > threshold else 0 for b in buf]
|
||||
|
||||
# Measure run lengths (periods between transitions)
|
||||
runs = []
|
||||
cur = bits[0]
|
||||
count = 1
|
||||
for b in bits[1:]:
|
||||
if b == cur:
|
||||
count += 1
|
||||
else:
|
||||
runs.append(count)
|
||||
cur = b
|
||||
count = 1
|
||||
|
||||
runs.append(count)
|
||||
|
||||
if len(runs) < 4:
|
||||
print(f" Modulation: {CR}insufficient transitions{C0}")
|
||||
return
|
||||
|
||||
runs_sorted = sorted(runs)
|
||||
# Remove outliers (top/bottom 10%)
|
||||
trim = max(1, len(runs) // 10)
|
||||
|
||||
# Estimate clock: most common run length = half-period
|
||||
from collections import Counter
|
||||
run_counts = Counter(runs)
|
||||
most_common_run = run_counts.most_common(1)[0][0]
|
||||
|
||||
# Map to nearest standard RF divider
|
||||
half_samples = most_common_run
|
||||
full_period_us = half_samples * 2 * 8 # us
|
||||
|
||||
rf_dividers = [8, 16, 32, 40, 50, 64, 100, 128]
|
||||
tc_us = 8 # 1 Tc = 8µs at 125kHz
|
||||
best_div = min(rf_dividers, key=lambda d: abs(d*tc_us - full_period_us))
|
||||
|
||||
print(f" Half-period : ~{most_common_run} samples = {most_common_run*8}µs")
|
||||
print(f" Full period : ~{full_period_us}µs")
|
||||
print(f" Nearest RF : {CG}RF/{best_div}{C0} ({best_div*tc_us}µs/bit)")
|
||||
print()
|
||||
|
||||
# Modulation type heuristic
|
||||
# Manchester: runs cluster around 1 value (half period) and 2x that (full period)
|
||||
# ASK/NRZ: long runs of same value
|
||||
# FSK: two distinct run lengths alternating
|
||||
|
||||
unique_runs = set(runs)
|
||||
long_runs = [r for r in runs if r > most_common_run * 3]
|
||||
|
||||
# Manchester has runs clustering at N and 2N (half and full period)
|
||||
# Check if second most common run is ~2x the most common
|
||||
tol = max(2, most_common_run // 3)
|
||||
top2 = run_counts.most_common(2)
|
||||
is_manchester = (len(top2) >= 2 and
|
||||
abs(top2[1][0] - most_common_run * 2) <= tol)
|
||||
|
||||
if len(long_runs) > len(runs) * 0.3:
|
||||
mod = "ASK / NRZ (long steady periods)"
|
||||
col = CG
|
||||
elif is_manchester:
|
||||
mod = f"Manchester (RF/{best_div})"
|
||||
col = CG
|
||||
elif len(unique_runs) <= 4:
|
||||
mod = f"Biphase (RF/{best_div})"
|
||||
col = CG
|
||||
else:
|
||||
mod = "FSK or mixed (multiple run lengths)"
|
||||
col = CG
|
||||
|
||||
print(f" Modulation : {col}{mod}{C0}")
|
||||
|
||||
# Gap detection
|
||||
gap_threshold = mean // 2
|
||||
gaps = [i for i, b in enumerate(buf[200:]) if b < gap_threshold]
|
||||
|
||||
|
||||
if gaps:
|
||||
print(f" RTF gaps : {CG}{len(gaps)}{C0} samples below 0x{gap_threshold:02x}"
|
||||
f" ^`^t gap commands present")
|
||||
else:
|
||||
print(f" RTF gaps : {CR}none ^`^t no gap commands detected{C0}")
|
||||
|
||||
|
||||
@@ -425,6 +425,22 @@ class ChameleonCMD:
|
||||
i += 14
|
||||
return resp
|
||||
|
||||
def hf14a_sniff(self, timeout_ms: int = 5000):
|
||||
"""
|
||||
Capture ISO14443A reader frames while CU acts as a tag emulator.
|
||||
|
||||
The firmware installs a sniff callback into the HF14A stack for the
|
||||
requested duration, then returns all captured frames packed as:
|
||||
[2 bytes: bit count, big-endian] [N bytes: frame data, ceil(bits/8)] ...
|
||||
|
||||
:param timeout_ms: Listen duration in ms (1-30000, default 5000)
|
||||
:return: Raw response — check .status and .data
|
||||
"""
|
||||
timeout_ms = max(1, min(30000, timeout_ms))
|
||||
payload = bytes([(timeout_ms >> 8) & 0xFF, timeout_ms & 0xFF])
|
||||
timeout_s = (timeout_ms // 1000) + 5
|
||||
return self.device.send_cmd_sync(Command.HF14A_SNIFF, payload, timeout=timeout_s)
|
||||
|
||||
@expect_response(Status.SUCCESS)
|
||||
def hf14a_get_config(self):
|
||||
"""
|
||||
@@ -555,6 +571,48 @@ class ChameleonCMD:
|
||||
resp.parsed = struct.unpack(">BBH8sBBBB", resp.data[:16])
|
||||
return resp
|
||||
|
||||
|
||||
|
||||
def lf_sniff(self, timeout_ms: int = 2000):
|
||||
"""
|
||||
Capture raw LF field ADC samples.
|
||||
|
||||
The ChameleonUltra samples the LF antenna at 125kHz (8µs/sample).
|
||||
Each byte is an 8-bit ADC value: ~0x80 = field on, lower = gap/no field.
|
||||
|
||||
:param timeout_ms: Capture duration in ms (1-10000, default 2000)
|
||||
:return: Raw response object — check .status and .data
|
||||
"""
|
||||
timeout_ms = max(1, min(10000, timeout_ms))
|
||||
payload = bytes([(timeout_ms >> 8) & 0xFF, timeout_ms & 0xFF])
|
||||
timeout_s = (timeout_ms // 1000) + 2
|
||||
return self.device.send_cmd_sync(Command.LF_SNIFF, payload, timeout=timeout_s)
|
||||
|
||||
|
||||
|
||||
@expect_response(Status.LF_TAG_OK)
|
||||
def em4x05_scan(self, pwd: int = 0):
|
||||
"""
|
||||
Read an EM4x05 or EM4x69 tag (reader-talk-first).
|
||||
|
||||
Response payload (14 bytes, big-endian):
|
||||
config 4 bytes — block 0 configuration word
|
||||
uid 4 bytes — EM4x05 UID
|
||||
uid_hi 4 bytes — EM4x69 uid_hi (zero for plain EM4x05)
|
||||
is_em4x69 1 byte — 1 if a 64-bit EM4x69 UID was read
|
||||
uid_block 1 byte — block number UID was read from
|
||||
|
||||
:param pwd: 32-bit password for LOGIN (default 0x00000000)
|
||||
:return: parsed tuple (config, uid, uid_hi, is_em4x69, uid_block)
|
||||
"""
|
||||
pwd_bytes = struct.pack('!I', pwd & 0xFFFFFFFF)
|
||||
resp = self.device.send_cmd_sync(Command.EM4X05_SCAN, pwd_bytes)
|
||||
if resp.status == Status.LF_TAG_OK:
|
||||
resp.parsed = struct.unpack('!IIIBB', resp.data[:14])
|
||||
return resp
|
||||
|
||||
|
||||
|
||||
@expect_response(Status.LF_TAG_OK)
|
||||
def viking_scan(self):
|
||||
"""
|
||||
|
||||
@@ -75,6 +75,7 @@ class Command(enum.IntEnum):
|
||||
MF1_CHECK_KEYS_ON_BLOCK = 2015
|
||||
HF14A_GET_CONFIG = 2200
|
||||
HF14A_SET_CONFIG = 2201
|
||||
HF14A_SNIFF = 2020
|
||||
|
||||
EM410X_SCAN = 3000
|
||||
EM410X_WRITE_TO_T55XX = 3001
|
||||
@@ -143,6 +144,9 @@ class Command(enum.IntEnum):
|
||||
VIKING_GET_EMU_ID = 5005
|
||||
IOPROX_SET_EMU_ID = 5008
|
||||
IOPROX_GET_EMU_ID = 5009
|
||||
EM4X05_SCAN = 3030
|
||||
EM4X05_READSNIFF = 3032
|
||||
LF_SNIFF = 3031
|
||||
|
||||
|
||||
@enum.unique
|
||||
|
||||
Reference in New Issue
Block a user