dandri/Flipper-ARF
1
0
Fork 0
mirror of https://github.com/D4C1-Labs/Flipper-ARF.git synced 2026-03-29 10:29:59 +00:00
Code Issues Actions Packages Projects Releases 38 Wiki Activity

Delete applications/main/KeylessGoSniffer directory

Browse Source
This commit is contained in:
D4rk$1d3
2026-03-18 19:17:43 -03:00
committed by GitHub
parent c6265ea29b
commit b93a970647
4 changed files with 0 additions and 1043 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
applications/main/KeylessGoSniffer/application.fam
View File

@@ -1,11 +0,0 @@
App(
appid="lf_sniffer",
name="KeylessGO Sniffer",
apptype=FlipperAppType.MENUEXTERNAL,
entry_point="lf_sniffer_app",
requires=["gui", "notification", "storage"],
stack_size=4 * 1024,
fap_category="GPIO",
fap_version=(1, 0),
fap_description="LF 125kHz challenge sniffer for KeylessGO analysis",
)

506
applications/main/KeylessGoSniffer/lf_analyze.py
View File

@@ -1,506 +0,0 @@
#!/usr/bin/env python3
"""
LF Analyzer -- Decodes captures from the LF Sniffer FAP
=========================================================
Analyzes edge timings captured by the Flipper Zero FAP and determines:
- Modulation type (ASK/FSK/PSK)
- Encoding (Manchester, Biphase, NRZ, PWM)
- Bit period and carrier frequency
- Decoded bit stream
- Likely protocol (Hitag2, Hitag S, EM4100, etc.)
- KeylessGO challenge candidate if applicable
Usage:
python3 lf_analyze.py capture_0000.csv
python3 lf_analyze.py capture_0000.csv --verbose
python3 lf_analyze.py capture_0000.csv --protocol hitag2
"""
import argparse
import csv
import sys
import math
from collections import Counter
from pathlib import Path
# -- CSV loader ----------------------------------------------------------------
def load_csv(path: str) -> tuple[list, dict]:
edges = []
meta = {}
with open(path, 'r') as f:
for line in f:
line = line.strip()
if line.startswith('#'):
# Parse key:value pairs from metadata comment lines
if 'edges:' in line:
for token in line.split():
if ':' in token:
k, v = token.split(':', 1)
try:
meta[k.lstrip('#')] = int(v)
except ValueError:
pass
continue
if line.startswith('index'):
continue # skip column header
parts = line.split(',')
if len(parts) >= 3:
try:
edges.append({
'idx': int(parts[0]),
'dur_us': int(parts[1]),
'level': int(parts[2]),
'note': parts[3].strip() if len(parts) > 3 else ''
})
except ValueError:
pass
print(f"[*] Loaded {len(edges)} edges")
if meta:
print(f" Metadata: {meta}")
return edges, meta
# -- Carrier frequency detection ----------------------------------------------
def analyze_carrier(edges: list, verbose: bool = False) -> dict:
"""
Detects the LF carrier by measuring the shortest pulses in the capture.
These correspond to half-periods of the unmodulated carrier.
"""
durations = [e['dur_us'] for e in edges if 1 < e['dur_us'] < 50]
if not durations:
return {'carrier_hz': 0, 'half_period_us': 0}
counter = Counter(durations)
most_common = counter.most_common(20)
if verbose:
print("\n[*] Most frequent edge durations (us):")
for dur, cnt in most_common[:10]:
print(f" {dur:4d} us x{cnt:5d}")
# Carrier half-period = most frequent pulse in the 1-20 us range
short = [(d, c) for d, c in most_common if 1 < d < 20]
if not short:
short = [(d, c) for d, c in most_common if d < 50]
if not short:
return {'carrier_hz': 0, 'half_period_us': 0}
half_period = short[0][0]
carrier_hz = int(1_000_000 / (2 * half_period)) if half_period > 0 else 0
print(f"\n[*] Detected carrier:")
print(f" Half-period : {half_period} us")
print(f" Frequency : {carrier_hz:,} Hz (~{carrier_hz/1000:.1f} kHz)")
if 100_000 < carrier_hz < 150_000:
print(f" [+] Matches 125 kHz LF band (KeylessGO / RFID)")
elif 110_000 < carrier_hz < 140_000:
print(f" [~] Close to 125 kHz")
return {
'carrier_hz': carrier_hz,
'half_period_us': half_period,
'period_us': half_period * 2,
}
# -- Packet segmentation ------------------------------------------------------
def find_packets(edges: list, gap_threshold_us: int = 5000) -> list:
"""Splits the edge list into packets separated by gaps."""
packets = []
start = 0
for i, e in enumerate(edges):
if e['dur_us'] > gap_threshold_us:
if i - start > 8:
packets.append(edges[start:i])
start = i + 1
if len(edges) - start > 8:
packets.append(edges[start:])
print(f"\n[*] Detected packets: {len(packets)}")
for i, pkt in enumerate(packets):
total = sum(e['dur_us'] for e in pkt)
print(f" PKT {i}: {len(pkt)} edges, {total} us ({total/1000:.1f} ms)")
return packets
# -- Bit period detection -----------------------------------------------------
def detect_bit_period(packet: list, carrier: dict, verbose: bool = False) -> dict:
"""
Estimates the bit period from modulated pulse widths.
KeylessGO uses Manchester at ~4 kbps over 125 kHz, so bit_period ~250 us.
EM4100 / Hitag use 125 kHz / 32 = ~256 us per bit.
"""
min_dur = carrier.get('half_period_us', 4)
durations = [e['dur_us'] for e in packet if e['dur_us'] > min_dur * 2]
if not durations:
return {'bit_period_us': 0, 'baud_rate': 0}
# Histogram with 10 us resolution
hist = Counter(round(d / 10) * 10 for d in durations)
peaks = sorted(hist.items(), key=lambda x: x[1], reverse=True)
if verbose:
print("\n Modulated pulse durations (top 15):")
for dur, cnt in peaks[:15]:
print(f" {dur:5d} us x{cnt:4d}")
if not peaks:
return {'bit_period_us': 0, 'baud_rate': 0}
# Bit period = smallest frequently-occurring modulated pulse
candidates = [d for d, c in peaks if c >= 3]
if not candidates:
candidates = [peaks[0][0]]
half_period = min(candidates)
# Check if this is a half-period (consecutive same-width pulses that
# alternate level = Manchester half-periods). If so, double it.
# Two equal half-periods make one Manchester bit period.
same_width_pairs = sum(
1 for i in range(len(packet)-1)
if abs(packet[i]['dur_us'] - half_period) < half_period*0.3
and abs(packet[i+1]['dur_us'] - half_period) < half_period*0.3
)
total_half_pulses = sum(
1 for e in packet
if abs(e['dur_us'] - half_period) < half_period*0.3
)
# If >80% of pulses are this width, they are half-periods
if total_half_pulses > len(packet) * 0.7:
bit_period = half_period * 2
is_half = True
else:
bit_period = half_period
is_half = False
baud_rate = int(1_000_000 / bit_period) if bit_period > 0 else 0
print(f"\n[*] Estimated bit period: {bit_period} us ({baud_rate} baud)")
if is_half:
print(f" (half-period detected: {half_period} us x2)")
if 3800 < baud_rate < 4200:
print(" -> Likely: Manchester 4 kbps (Hitag2 / Hitag S / EM4100 / Keyless)")
elif 7500 < baud_rate < 8500:
print(" -> Likely: Manchester 8 kbps")
elif 1900 < baud_rate < 2100:
print(" -> Likely: 2 kbps biphase (EM4100)")
return {
'bit_period_us': bit_period,
'baud_rate': baud_rate,
'half_bit_us': bit_period // 2,
}
# -- Manchester decoder -------------------------------------------------------
def decode_manchester(packet: list, bit_period_us: int, verbose: bool = False) -> list:
"""
Decodes Manchester-encoded bit stream from edge timings.
Both half-period and full-period pulses are handled.
Returns a list of integers (0 or 1).
"""
if bit_period_us == 0:
return []
half = bit_period_us // 2
tolerance = half // 2 # +/- 50% of the half-period
bits = []
i = 0
while i < len(packet):
dur = packet[i]['dur_us']
if abs(dur - half) < tolerance:
# Half-period pulse -- needs the next one to complete a bit
if i + 1 < len(packet):
dur2 = packet[i + 1]['dur_us']
if abs(dur2 - half) < tolerance:
level = packet[i]['level']
bits.append(1 if level else 0)
i += 2
continue
i += 1
elif abs(dur - bit_period_us) < tolerance:
# Full-period pulse -- encodes one bit by itself
level = packet[i]['level']
bits.append(1 if level else 0)
i += 1
else:
if verbose:
print(f" [!] Unexpected duration at edge {i}: {dur} us")
i += 1
return bits
# -- Bit / byte helpers -------------------------------------------------------
def bits_to_bytes(bits: list) -> bytes:
result = bytearray()
for i in range(0, len(bits) - len(bits) % 8, 8):
byte = 0
for j in range(8):
byte = (byte << 1) | bits[i + j]
result.append(byte)
return bytes(result)
def print_bits(bits: list, label: str = "Bits"):
print(f"\n[*] {label} ({len(bits)} bits):")
for i in range(0, len(bits), 64):
chunk = bits[i:i + 64]
groups = [chunk[j:j + 8] for j in range(0, len(chunk), 8)]
line = ' '.join(''.join(str(b) for b in g) for g in groups)
print(f" {i//8:4d}: {line}")
if len(bits) >= 8:
data = bits_to_bytes(bits)
hex_str = ' '.join(f'{b:02X}' for b in data)
print(f"\n HEX: {hex_str}")
# -- Protocol identification --------------------------------------------------
def identify_protocol(bits: list, baud_rate: int, carrier_hz: int) -> str:
n = len(bits)
print(f"\n[*] Protocol identification:")
print(f" Bits: {n} | Baud: {baud_rate} | Carrier: {carrier_hz:,} Hz")
# EM4100: 64-bit frame, 9-bit preamble of all-ones, Manchester
if 50 < n < 80 and baud_rate > 3000:
for start in range(n - 9):
if all(bits[start + j] == 1 for j in range(9)):
print(f" -> EM4100 candidate (preamble at bit {start})")
data_bits = bits[start + 9:]
if len(data_bits) >= 55:
_decode_em4100(data_bits[:55])
break
# Hitag2: ~96-bit frame, Manchester 4 kbps
if 80 < n < 120 and 3500 < baud_rate < 4500:
print(f" -> Hitag2 candidate ({n} bits)")
_analyze_hitag2(bits)
# Hitag S: variable length, Manchester 4 kbps
if n > 120 and 3500 < baud_rate < 4500:
print(f" -> Hitag S candidate ({n} bits)")
# KeylessGO challenge: typically 32-64 data bits
if 20 < n < 80 and 3500 < baud_rate < 4500:
print(f" -> KeylessGO challenge candidate")
_analyze_keylessgo_challenge(bits)
return "unknown"
def _decode_em4100(bits: list):
if len(bits) < 55:
return
# EM4100 data layout: [version 8b][data 32b][col_parity 4b][stop 1b]
# Each row: 4 data bits + 1 parity bit
print("\n EM4100 decode:")
customer = (bits[0] << 7) | (bits[1] << 6) | (bits[2] << 5) | \
(bits[3] << 4) | (bits[5] << 3) | (bits[6] << 2) | \
(bits[7] << 1) | bits[8]
print(f" Customer code: 0x{customer:02X} ({customer})")
def _analyze_hitag2(bits: list):
if len(bits) < 32:
return
data = bits_to_bytes(bits[:32])
print(f"\n Hitag2 first 32 bits: {data.hex().upper()}")
print(f" As uint32 BE : 0x{int.from_bytes(data, 'big'):08X}")
def _analyze_keylessgo_challenge(bits: list):
"""
Looks for a KeylessGO challenge structure.
The car transmits: [sync/preamble] [32-bit challenge] [checksum]
"""
print(f"\n Keyless challenge analysis:")
# Alternating preamble (010101...) is typically used as sync
for start in range(min(len(bits) - 8, 20)):
window = bits[start:start + 8]
alternating = all(window[j] != window[j + 1] for j in range(7))
if alternating:
print(f" Alternating sync at bit {start}: {''.join(str(b) for b in window)}")
payload_start = start + 8
if len(bits) - payload_start >= 32:
challenge_bits = bits[payload_start:payload_start + 32]
challenge_bytes = bits_to_bytes(challenge_bits)
challenge_val = int.from_bytes(challenge_bytes, 'big')
print(f" Challenge candidate (32b): 0x{challenge_val:08X}")
print(f" As bytes : {challenge_bytes.hex().upper()}")
break
# Also print every 32-bit aligned window as a candidate
print(f"\n All 32-bit blocks in capture:")
for offset in range(0, min(len(bits) - 32, 48), 4):
chunk = bits[offset:offset + 32]
val = 0
for b in chunk:
val = (val << 1) | b
preview = ''.join(str(b) for b in chunk[:16])
print(f" offset {offset:3d}: 0x{val:08X} [{preview}...]")
# -- Full analysis entry point ------------------------------------------------
def analyze_capture(path: str, verbose: bool = False, protocol_hint: str = None):
print(f"\n{'='*60}")
print(f"LF Capture Analyzer -- {Path(path).name}")
print('='*60)
edges, meta = load_csv(path)
if not edges:
print("ERROR: No data found in file.")
return
# Step 1: detect carrier
carrier = analyze_carrier(edges, verbose)
# Step 2: segment packets
packets = find_packets(edges)
if not packets:
print("\n[!] No packets detected. Check:")
print(" - Car is emitting 125 kHz LF field")
print(" - Coil is connected correctly to PB2")
print(" - LM393 is powered from 3.3V")
return
# Step 3: analyze each packet
for i, pkt in enumerate(packets):
print(f"\n{'─'*50}")
print(f"PACKET {i} -- {len(pkt)} edges")
print('─'*50)
bit_info = detect_bit_period(pkt, carrier, verbose)
bit_period = bit_info.get('bit_period_us', 0)
baud_rate = bit_info.get('baud_rate', 0)
if bit_period == 0:
print(" [!] Could not determine bit period")
continue
# Step 4: Manchester decode
bits = decode_manchester(pkt, bit_period, verbose)
if len(bits) < 8:
print(f" [!] Only {len(bits)} bits decoded -- likely noise or raw carrier")
if verbose:
raw = [e['dur_us'] for e in pkt[:40]]
print(f" RAW first durations: {raw}")
continue
print_bits(bits, f"Manchester decoded ({len(bits)} bits)")
# Step 5: identify protocol
identify_protocol(bits, baud_rate, carrier.get('carrier_hz', 0))
# Step 6: apply explicit decoder if requested
if protocol_hint == 'hitag2':
decode_hitag2_explicit(bits)
print(f"\n{'='*60}")
print("SUMMARY")
print('='*60)
print(f" Carrier : {carrier.get('carrier_hz', 0):,} Hz")
print(f" Packets : {len(packets)}")
print(f" Total edges: {len(edges)}")
print()
print("NEXT STEPS:")
print(" 1. If you see 'Keyless challenge candidate' -> use the 0xXXXXXXXX value")
print(" with your AUT64 implementation to compute the expected response.")
print(" 2. If Hitag2 is detected -> the protocol is HMAC-SHA1 / AUT64 depending")
print(" on the generation.")
print(" 3. If carrier is 125 kHz but no packets appear -> car is not emitting")
print(" a challenge (get closer, < 30 cm).")
def decode_hitag2_explicit(bits: list):
"""Explicit Hitag2 frame decode when --protocol hitag2 is specified."""
print("\n[*] Hitag2 frame decode:")
if len(bits) < 5:
return
print(f" Start of frame : {bits[0]}")
if len(bits) > 5:
cmd = 0
for b in bits[1:5]:
cmd = (cmd << 1) | b
cmds = {
0b0001: 'REQUEST',
0b0011: 'SELECT',
0b0101: 'READ',
0b1001: 'WRITE'
}
print(f" Command : 0b{cmd:04b} = {cmds.get(cmd, 'UNKNOWN')}")
if len(bits) >= 37:
uid_bits = bits[5:37]
uid_val = 0
for b in uid_bits:
uid_val = (uid_val << 1) | b
print(f" UID candidate : 0x{uid_val:08X}")
# -- CLI ----------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(
description='Analyze LF captures from the Flipper Zero LF Sniffer FAP',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
python3 lf_analyze.py capture_0000.csv
python3 lf_analyze.py capture_0000.csv --verbose
python3 lf_analyze.py capture_0000.csv --protocol hitag2
"""
)
parser.add_argument('capture', help='CSV file from LF Sniffer FAP')
parser.add_argument('--verbose', '-v', action='store_true')
parser.add_argument('--protocol',
choices=['hitag2', 'em4100', 'Keyless', 'auto'],
default='auto',
help='Force a specific protocol decoder (default: auto)')
args = parser.parse_args()
if not Path(args.capture).exists():
print(f"ERROR: {args.capture} not found")
sys.exit(1)
analyze_capture(
args.capture,
verbose=args.verbose,
protocol_hint=args.protocol if args.protocol != 'auto' else None
)
if __name__ == '__main__':
main()

446
applications/main/KeylessGoSniffer/lf_sniffer.c
View File

@@ -1,446 +0,0 @@
#include "lf_sniffer.h"
#include <furi.h>
#include <furi_hal_gpio.h>
#include <furi_hal_resources.h>
#include <furi_hal_cortex.h>
#include <gui/gui.h>
#include <notification/notification_messages.h>
#include <storage/storage.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#define TAG "LFSniffer"
#define DWT_CYCCNT (*(volatile uint32_t*)0xE0001004)
#define DWT_CTRL (*(volatile uint32_t*)0xE0001000)
#define DEMCR (*(volatile uint32_t*)0xE000EDFC)
static inline void dwt_init(void) {
DEMCR |= (1U << 24);
DWT_CYCCNT = 0;
DWT_CTRL |= 1U;
}
static inline uint32_t dwt_us(void) {
return DWT_CYCCNT / 64;
}
static volatile LFEdge* g_edges = NULL;
static volatile uint32_t g_edge_count = 0;
static volatile uint32_t g_last_time = 0;
static volatile bool g_active = false;
static volatile bool g_overflow = false;
static void lf_gpio_isr(void* ctx) {
UNUSED(ctx);
if(!g_active) return;
uint32_t now = dwt_us();
uint32_t delta = now - g_last_time;
g_last_time = now;
if(g_edge_count >= LF_MAX_EDGES) {
g_overflow = true;
g_active = false;
return;
}
bool current_level = furi_hal_gpio_read(LF_INPUT_PIN);
g_edges[g_edge_count].duration_us = delta;
g_edges[g_edge_count].level = !current_level;
g_edge_count++;
}
static void draw_callback(Canvas* canvas, void* ctx) {
LFSnifferApp* app = ctx;
furi_mutex_acquire(app->mutex, FuriWaitForever);
canvas_clear(canvas);
canvas_set_font(canvas, FontPrimary);
canvas_draw_str(canvas, 2, 12, "KeylessGO Sniffer");
canvas_set_font(canvas, FontSecondary);
switch(app->state) {
case LFStateIdle:
canvas_draw_str(canvas, 2, 26, "Pin: PB2 (header pin 6)");
canvas_draw_str(canvas, 2, 36, "OK = Start capture");
canvas_draw_str(canvas, 2, 46, "Approach car without key");
canvas_draw_str(canvas, 2, 58, "Back = Exit");
break;
case LFStateCapturing: {
char buf[32];
canvas_draw_str(canvas, 2, 26, "CAPTURING...");
snprintf(buf, sizeof(buf), "Edges: %lu", (unsigned long)app->edge_count);
canvas_draw_str(canvas, 2, 36, buf);
snprintf(buf, sizeof(buf), "Packets: %lu", (unsigned long)app->packet_count);
canvas_draw_str(canvas, 2, 46, buf);
canvas_draw_str(canvas, 2, 58, "OK/Back = Stop");
break;
}
case LFStateDone: {
char buf[48];
snprintf(buf, sizeof(buf), "Edges:%lu Pkts:%lu",
(unsigned long)app->edge_count,
(unsigned long)app->packet_count);
canvas_draw_str(canvas, 2, 26, buf);
snprintf(buf, sizeof(buf), "Min:%luus Max:%luus",
(unsigned long)app->min_pulse_us,
(unsigned long)app->max_pulse_us);
canvas_draw_str(canvas, 2, 36, buf);
canvas_draw_str(canvas, 2, 46, "OK = Save to SD");
canvas_draw_str(canvas, 2, 58, "Back = Discard");
break;
}
case LFStateSaving:
canvas_draw_str(canvas, 2, 26, "Saving...");
canvas_draw_str(canvas, 2, 36, app->filename);
break;
case LFStateSaved:
canvas_draw_str(canvas, 2, 26, "Saved OK:");
canvas_draw_str(canvas, 2, 36, app->filename);
canvas_draw_str(canvas, 2, 46, app->status_msg);
canvas_draw_str(canvas, 2, 58, "Back = New capture");
break;
case LFStateError:
canvas_draw_str(canvas, 2, 26, "ERROR:");
canvas_draw_str(canvas, 2, 36, app->status_msg);
canvas_draw_str(canvas, 2, 58, "Back = Retry");
break;
}
furi_mutex_release(app->mutex);
}
static void input_callback(InputEvent* event, void* ctx) {
LFSnifferApp* app = ctx;
furi_message_queue_put(app->queue, event, FuriWaitForever);
}
static void lf_analyze_packets(LFSnifferApp* app) {
app->packet_count = 0;
app->min_pulse_us = 0xFFFFFFFF;
app->max_pulse_us = 0;
app->carrier_pulses = 0;
app->data_pulses = 0;
if(app->edge_count < 4) return;
uint32_t pkt_start = 0;
bool in_packet = false;
for(uint32_t i = 1; i < app->edge_count; i++) {
uint32_t dur = app->edges[i].duration_us;
if(dur < app->min_pulse_us) app->min_pulse_us = dur;
if(dur > app->max_pulse_us) app->max_pulse_us = dur;
if(dur < 12) {
app->carrier_pulses++;
} else if(dur < 15) {
app->data_pulses++;
}
if(dur > LF_GAP_THRESHOLD_US) {
if(in_packet && i > pkt_start + 8) {
if(app->packet_count < LF_MAX_PACKETS) {
app->packets[app->packet_count].start_idx = pkt_start;
app->packets[app->packet_count].edge_count = i - pkt_start;
app->packets[app->packet_count].duration_us = 0;
for(uint32_t j = pkt_start; j < i; j++)
app->packets[app->packet_count].duration_us +=
app->edges[j].duration_us;
app->packet_count++;
}
}
in_packet = false;
pkt_start = i;
} else {
if(!in_packet) {
in_packet = true;
pkt_start = i;
}
}
}
if(in_packet && app->edge_count > pkt_start + 8) {
if(app->packet_count < LF_MAX_PACKETS) {
app->packets[app->packet_count].start_idx = pkt_start;
app->packets[app->packet_count].edge_count = app->edge_count - pkt_start;
app->packets[app->packet_count].duration_us = 0;
for(uint32_t j = pkt_start; j < app->edge_count; j++)
app->packets[app->packet_count].duration_us +=
app->edges[j].duration_us;
app->packet_count++;
}
}
}
static bool lf_save_csv(LFSnifferApp* app) {
Storage* storage = furi_record_open(RECORD_STORAGE);
storage_common_mkdir(storage, "/ext/keyless_sniffer");
snprintf(app->filename, sizeof(app->filename),
"/ext/keyless_sniffer/capture_%04lu.csv",
(unsigned long)app->file_index);
File* file = storage_file_alloc(storage);
bool ok = false;
if(storage_file_open(file, app->filename, FSAM_WRITE, FSOM_CREATE_ALWAYS)) {
const char* header = "index,duration_us,level,note\n";
storage_file_write(file, header, strlen(header));
char meta[128];
snprintf(meta, sizeof(meta),
"# Keyless Sniffer capture -- edges:%lu packets:%lu\n",
(unsigned long)app->edge_count,
(unsigned long)app->packet_count);
storage_file_write(file, meta, strlen(meta));
snprintf(meta, sizeof(meta),
"# min_us:%lu max_us:%lu carrier:%lu data:%lu\n",
(unsigned long)app->min_pulse_us,
(unsigned long)app->max_pulse_us,
(unsigned long)app->carrier_pulses,
(unsigned long)app->data_pulses);
storage_file_write(file, meta, strlen(meta));
uint32_t pkt_idx = 0;
char line[64];
for(uint32_t i = 0; i < app->edge_count; i++) {
const char* note = "";
if(pkt_idx < app->packet_count &&
app->packets[pkt_idx].start_idx == i) {
note = "PKT_START";
pkt_idx++;
}
snprintf(line, sizeof(line),
"%lu,%lu,%d,%s\n",
(unsigned long)i,
(unsigned long)app->edges[i].duration_us,
app->edges[i].level ? 1 : 0,
note);
storage_file_write(file, line, strlen(line));
}
storage_file_write(file, "# PACKETS\n", 10);
for(uint32_t p = 0; p < app->packet_count; p++) {
snprintf(meta, sizeof(meta),
"# PKT %lu: start=%lu edges=%lu dur=%luus\n",
(unsigned long)p,
(unsigned long)app->packets[p].start_idx,
(unsigned long)app->packets[p].edge_count,
(unsigned long)app->packets[p].duration_us);
storage_file_write(file, meta, strlen(meta));
}
storage_file_close(file);
ok = true;
snprintf(app->status_msg, sizeof(app->status_msg),
"%lu edges, %lu packets",
(unsigned long)app->edge_count,
(unsigned long)app->packet_count);
} else {
snprintf(app->status_msg, sizeof(app->status_msg), "Failed to open file");
}
storage_file_free(file);
furi_record_close(RECORD_STORAGE);
return ok;
}
static void lf_start_capture(LFSnifferApp* app) {
furi_mutex_acquire(app->mutex, FuriWaitForever);
app->edge_count = 0;
app->packet_count = 0;
app->total_time_us = 0;
app->min_pulse_us = 0xFFFFFFFF;
app->max_pulse_us = 0;
app->carrier_pulses = 0;
app->data_pulses = 0;
g_overflow = false;
g_edges = app->edges;
g_edge_count = 0;
g_last_time = dwt_us();
g_active = true;
furi_hal_gpio_init(LF_INPUT_PIN, GpioModeInterruptRiseFall,
GpioPullUp, GpioSpeedVeryHigh);
furi_hal_gpio_add_int_callback(LF_INPUT_PIN, lf_gpio_isr, NULL);
app->state = LFStateCapturing;
furi_mutex_release(app->mutex);
notification_message(app->notif, &sequence_blink_green_100);
FURI_LOG_I(TAG, "Capture started");
}
static void lf_stop_capture(LFSnifferApp* app) {
g_active = false;
furi_hal_gpio_remove_int_callback(LF_INPUT_PIN);
furi_hal_gpio_init(LF_INPUT_PIN, GpioModeInput, GpioPullUp, GpioSpeedLow);
furi_mutex_acquire(app->mutex, FuriWaitForever);
app->edge_count = g_edge_count;
lf_analyze_packets(app);
app->state = LFStateDone;
furi_mutex_release(app->mutex);
notification_message(app->notif, &sequence_blink_blue_100);
FURI_LOG_I(TAG, "Capture stopped: %lu edges, %lu packets",
(unsigned long)app->edge_count,
(unsigned long)app->packet_count);
}
int32_t lf_sniffer_app(void* p) {
UNUSED(p);
dwt_init();
LFSnifferApp* app = malloc(sizeof(LFSnifferApp));
furi_check(app != NULL);
memset(app, 0, sizeof(LFSnifferApp));
app->edges = malloc(LF_MAX_EDGES * sizeof(LFEdge));
furi_check(app->edges != NULL);
app->mutex = furi_mutex_alloc(FuriMutexTypeNormal);
app->queue = furi_message_queue_alloc(16, sizeof(InputEvent));
app->notif = furi_record_open(RECORD_NOTIFICATION);
app->view_port = view_port_alloc();
app->state = LFStateIdle;
app->file_index = 0;
view_port_draw_callback_set(app->view_port, draw_callback, app);
view_port_input_callback_set(app->view_port, input_callback, app);
app->gui = furi_record_open(RECORD_GUI);
gui_add_view_port(app->gui, app->view_port, GuiLayerFullscreen);
FURI_LOG_I(TAG, "Keyless Sniffer started");
InputEvent event;
bool running = true;
while(running) {
if(app->state == LFStateCapturing) {
furi_mutex_acquire(app->mutex, FuriWaitForever);
app->edge_count = g_edge_count;
if(g_overflow) {
furi_mutex_release(app->mutex);
lf_stop_capture(app);
furi_mutex_acquire(app->mutex, FuriWaitForever);
snprintf(app->status_msg, sizeof(app->status_msg),
"Buffer full (%d edges)", LF_MAX_EDGES);
}
furi_mutex_release(app->mutex);
view_port_update(app->view_port);
}
FuriStatus status = furi_message_queue_get(app->queue, &event, 100);
if(status != FuriStatusOk) continue;
if(event.type != InputTypeShort && event.type != InputTypeLong) continue;
switch(app->state) {
case LFStateIdle:
if(event.key == InputKeyOk) {
lf_start_capture(app);
view_port_update(app->view_port);
} else if(event.key == InputKeyBack) {
running = false;
}
break;
case LFStateCapturing:
if(event.key == InputKeyOk || event.key == InputKeyBack) {
lf_stop_capture(app);
view_port_update(app->view_port);
}
break;
case LFStateDone:
if(event.key == InputKeyOk) {
furi_mutex_acquire(app->mutex, FuriWaitForever);
app->state = LFStateSaving;
furi_mutex_release(app->mutex);
view_port_update(app->view_port);
bool saved = lf_save_csv(app);
furi_mutex_acquire(app->mutex, FuriWaitForever);
app->state = saved ? LFStateSaved : LFStateError;
app->file_index++;
furi_mutex_release(app->mutex);
view_port_update(app->view_port);
if(saved) {
notification_message(app->notif, &sequence_success);
FURI_LOG_I(TAG, "Saved: %s", app->filename);
}
} else if(event.key == InputKeyBack) {
furi_mutex_acquire(app->mutex, FuriWaitForever);
app->state = LFStateIdle;
app->edge_count = 0;
app->packet_count = 0;
furi_mutex_release(app->mutex);
view_port_update(app->view_port);
}
break;
case LFStateSaved:
case LFStateError:
if(event.key == InputKeyBack) {
furi_mutex_acquire(app->mutex, FuriWaitForever);
app->state = LFStateIdle;
app->edge_count = 0;
app->packet_count = 0;
furi_mutex_release(app->mutex);
view_port_update(app->view_port);
}
break;
default:
break;
}
}
if(app->state == LFStateCapturing) {
g_active = false;
furi_hal_gpio_remove_int_callback(LF_INPUT_PIN);
furi_hal_gpio_init(LF_INPUT_PIN, GpioModeAnalog, GpioPullNo, GpioSpeedLow);
}
view_port_enabled_set(app->view_port, false);
gui_remove_view_port(app->gui, app->view_port);
view_port_free(app->view_port);
furi_record_close(RECORD_GUI);
furi_record_close(RECORD_NOTIFICATION);
furi_mutex_free(app->mutex);
furi_message_queue_free(app->queue);
free(app->edges);
free(app);
return 0;
}

80
applications/main/KeylessGoSniffer/lf_sniffer.h
View File

@@ -1,80 +0,0 @@
#pragma once
#include <furi.h>
#include <gui/gui.h>
#include <gui/view_port.h>
#include <notification/notification.h>
#include <input/input.h>
#include <storage/storage.h>
#include <furi_hal_gpio.h>
#include <furi_hal_resources.h>
#include <stdbool.h>
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
#endif
#define LF_MAX_EDGES 4096
#define LF_GAP_THRESHOLD_US 5000
#define LF_CARRIER_HZ 125000
#define LF_BIT_PERIOD_US 250
#define LF_TIMEOUT_MS 10000
#define LF_MAX_PACKETS 16
#define LF_INPUT_PIN (&gpio_ext_pb2)
typedef struct {
uint32_t duration_us;
bool level;
} LFEdge;
typedef struct {
uint32_t start_idx;
uint32_t edge_count;
uint32_t duration_us;
} LFPacket;
typedef enum {
LFStateIdle,
LFStateCapturing,
LFStateDone,
LFStateSaving,
LFStateSaved,
LFStateError,
} LFState;
typedef struct {
Gui* gui;
ViewPort* view_port;
FuriMessageQueue* queue;
FuriMutex* mutex;
NotificationApp* notif;
LFState state;
char status_msg[64];
LFEdge* edges;
uint32_t edge_count;
uint32_t total_time_us;
LFPacket packets[LF_MAX_PACKETS];
uint32_t packet_count;
uint32_t last_edge_time_us;
bool capturing;
uint32_t carrier_pulses;
uint32_t data_pulses;
uint32_t min_pulse_us;
uint32_t max_pulse_us;
char filename[128];
uint32_t file_index;
} LFSnifferApp;
int32_t lf_sniffer_app(void* p);
#ifdef __cplusplus
}
#endif