From ce932d2e8a59c8fa98adee95f9acc1db6122e3bf Mon Sep 17 00:00:00 2001 From: Niel Nielsen Date: Thu, 2 Apr 2026 07:43:16 +0200 Subject: [PATCH] feat(data): add LF capture analysis commands --- firmware/application/Makefile | 8 +- firmware/application/src/app_cmd.c | 115 +++ firmware/application/src/app_status.h | 1 + firmware/application/src/data_cmd.h | 5 + .../application/src/rfid/nfctag/hf/nfc_14a.c | 16 + .../application/src/rfid/nfctag/hf/nfc_14a.h | 8 + .../src/rfid/nfctag/tag_emulation.h | 1 + .../src/rfid/reader/lf/lf_125khz_radio.h | 5 + .../src/rfid/reader/lf/lf_reader_generic.c | 22 +- .../src/rfid/reader/lf/lf_reader_generic.h | 23 + .../src/rfid/reader/lf/lf_reader_main.h | 3 + software/script/chameleon_cli_unit.py | 966 ++++++++++++++++++ software/script/chameleon_cmd.py | 58 ++ software/script/chameleon_enum.py | 4 + 14 files changed, 1227 insertions(+), 8 deletions(-) create mode 100644 firmware/application/src/rfid/reader/lf/lf_reader_generic.h diff --git a/firmware/application/Makefile b/firmware/application/Makefile index 0e2b231..44cb505 100644 --- a/firmware/application/Makefile +++ b/firmware/application/Makefile @@ -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/ \ diff --git a/firmware/application/src/app_cmd.c b/firmware/application/src/app_cmd.c index a8ada47..168808b 100644 --- a/firmware/application/src/app_cmd.c +++ b/firmware/application/src/app_cmd.c @@ -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 diff --git a/firmware/application/src/app_status.h b/firmware/application/src/app_status.h index 653336d..71762b3 100644 --- a/firmware/application/src/app_status.h +++ b/firmware/application/src/app_status.h @@ -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 diff --git a/firmware/application/src/data_cmd.h b/firmware/application/src/data_cmd.h index 51ac5fc..7b67468 100644 --- a/firmware/application/src/data_cmd.h +++ b/firmware/application/src/data_cmd.h @@ -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 diff --git a/firmware/application/src/rfid/nfctag/hf/nfc_14a.c b/firmware/application/src/rfid/nfctag/hf/nfc_14a.c index 7f208ac..d6fd0f6 100644 --- a/firmware/application/src/rfid/nfctag/hf/nfc_14a.c +++ b/firmware/application/src/rfid/nfctag/hf/nfc_14a.c @@ -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) { diff --git a/firmware/application/src/rfid/nfctag/hf/nfc_14a.h b/firmware/application/src/rfid/nfctag/hf/nfc_14a.h index e38b250..0d92022 100644 --- a/firmware/application/src/rfid/nfctag/hf/nfc_14a.h +++ b/firmware/application/src/rfid/nfctag/hf/nfc_14a.h @@ -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); diff --git a/firmware/application/src/rfid/nfctag/tag_emulation.h b/firmware/application/src/rfid/nfctag/tag_emulation.h index 9b3173a..f1e88c2 100644 --- a/firmware/application/src/rfid/nfctag/tag_emulation.h +++ b/firmware/application/src/rfid/nfctag/tag_emulation.h @@ -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); diff --git a/firmware/application/src/rfid/reader/lf/lf_125khz_radio.h b/firmware/application/src/rfid/reader/lf/lf_125khz_radio.h index f452ebf..f48f473 100644 --- a/firmware/application/src/rfid/reader/lf/lf_125khz_radio.h +++ b/firmware/application/src/rfid/reader/lf/lf_125khz_radio.h @@ -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); diff --git a/firmware/application/src/rfid/reader/lf/lf_reader_generic.c b/firmware/application/src/rfid/reader/lf/lf_reader_generic.c index 4fdbc42..d6da018 100644 --- a/firmware/application/src/rfid/reader/lf/lf_reader_generic.c +++ b/firmware/application/src/rfid/reader/lf/lf_reader_generic.c @@ -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); diff --git a/firmware/application/src/rfid/reader/lf/lf_reader_generic.h b/firmware/application/src/rfid/reader/lf/lf_reader_generic.h new file mode 100644 index 0000000..f7c3340 --- /dev/null +++ b/firmware/application/src/rfid/reader/lf/lf_reader_generic.h @@ -0,0 +1,23 @@ +#pragma once + +#include +#include +#include + +/* + * 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); diff --git a/firmware/application/src/rfid/reader/lf/lf_reader_main.h b/firmware/application/src/rfid/reader/lf/lf_reader_main.h index c975d13..ea1844d 100644 --- a/firmware/application/src/rfid/reader/lf/lf_reader_main.h +++ b/firmware/application/src/rfid/reader/lf/lf_reader_main.h @@ -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); diff --git a/software/script/chameleon_cli_unit.py b/software/script/chameleon_cli_unit.py index c9818a5..c2926b8 100644 --- a/software/script/chameleon_cli_unit.py +++ b/software/script/chameleon_cli_unit.py @@ -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 = ( + '' + ' LF field (ADC)  ' + '- - Mean  ' + '··· Gap threshold (mean÷2)  ' + '   ' + ' Field gap (below threshold)  ' + 'Ringing = exponential rise on field restore' + '' + ) + 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}") + diff --git a/software/script/chameleon_cmd.py b/software/script/chameleon_cmd.py index f541715..3de02da 100644 --- a/software/script/chameleon_cmd.py +++ b/software/script/chameleon_cmd.py @@ -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): """ diff --git a/software/script/chameleon_enum.py b/software/script/chameleon_enum.py index c159762..a381346 100644 --- a/software/script/chameleon_enum.py +++ b/software/script/chameleon_enum.py @@ -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