From d1b6909ac5bd32b23f31900f9143cb1cd78f10af Mon Sep 17 00:00:00 2001
From: tbthatgirl <tbthatgirl@gmail.com>
Date: Sun, 14 Jun 2026 04:45:57 +0000
Subject: [PATCH] Add host-side fake ESP32 CSI UDP sender for testing --source
 esp32

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 CHANGELOG.md                     |  3 +
 scripts/fake_esp32_csi_sender.py | 94 ++++++++++++++++++++++++++++++++
 2 files changed, 97 insertions(+)
 create mode 100644 scripts/fake_esp32_csi_sender.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a7a9b3cb15..6ad0c07ab7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **`scripts/fake_esp32_csi_sender.py` — host-side CSI UDP test sender.** Emits ADR-018 CSI frames (magic `0xC511_0001`, 20-byte header + I/Q pairs) to the sensing-server's UDP port, so the `--source esp32` ingestion path (node registration, `/api/v1/nodes`, the Sensing-tab `LIVE — ESP32 HARDWARE` state) can be exercised without physical hardware. Configurable node count, subcarriers, frame rate, and synthesized breathing/motion.
+
 ### Security
 - **ADR-157 Milestone-1 B4 - constant-time HMAC sync-beacon tag compare (`wifi-densepose-hardware`).** `AuthenticatedBeacon::verify` compared the 8-byte HMAC-SHA256 tag with `self.hmac_tag == expected`, which short-circuits on the first differing byte and leaks, through verification latency, how many leading bytes an attacker's forged tag matched - a byte-by-byte tag-recovery oracle (~256*N trials instead of 256^N). Replaced with a hand-rolled branch-free `constant_time_tag_eq` (XOR-accumulate every byte difference into a single `u8`, no early exit, `#[inline(never)]` + `core::hint::black_box` to stop the optimizer reintroducing a short-circuit or a non-constant-time `memcmp`). **No new dependency** - ADR-157 had deferred this only to avoid adding the `subtle` crate; a fixed 8-byte compare needs none. Grade MEASURED (constant-time *construction*; micro-timing on a noisy host is a smoke check only, gated `#[ignore]`). Pinned by `tag_compare_is_constant_time_shape` (equal/first-differ/last-differ/all-differ/length-mismatch + an end-to-end `verify()` last-byte tamper), proven to fail on a last-byte-skipping constant-time bug. ADR-157 §8 B4 -> RESOLVED.
 - **ADR-080 open HIGH findings closed on the Rust `wifi-densepose-sensing-server` boundary (ADR-164 G11).** The QE sweep's three HIGH findings — XFF-spoofing bypass, leaked stack traces, JWT-in-URL (CWE-598) — were logged against the Python v1 API and never re-verified against the shipped Rust sensing-server; the HOMECORE/M7 sweep (ADR-161) covered `homecore-server`, not this crate.
diff --git a/scripts/fake_esp32_csi_sender.py b/scripts/fake_esp32_csi_sender.py
new file mode 100644
index 0000000000..0409f0c5fa
--- /dev/null
+++ b/scripts/fake_esp32_csi_sender.py
@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""Fake ESP32 CSI UDP sender — emits ADR-018 binary frames to the sensing-server.
+
+Wire format (matches parse_esp32_frame in wifi-densepose-sensing-server/src/main.rs
+and firmware csi_collector.c):
+
+    [0..3]   magic = 0xC511_0001 (u32 LE)
+    [4]      node_id (u8)
+    [5]      n_antennas (u8)
+    [6..7]   n_subcarriers (u16 LE)
+    [8..11]  freq_mhz (u32 LE)
+    [12..15] sequence (u32 LE)
+    [16]     rssi (i8)
+    [17]     noise_floor (i8)
+    [18]     ppdu_type (u8, 0 = HT/legacy)
+    [19]     reserved
+    [20..]   I/Q data: n_antennas * n_subcarriers pairs of (i8 I, i8 Q)
+
+Use this to exercise the --source esp32 ingestion path without real hardware.
+"""
+import argparse
+import math
+import socket
+import struct
+import time
+
+MAGIC = 0xC511_0001
+
+
+def build_frame(node_id, n_antennas, n_subcarriers, freq_mhz, sequence,
+                rssi, noise_floor, t, breath_hz, motion):
+    hdr = struct.pack(
+        "<IBBHIIbbBB",
+        MAGIC, node_id, n_antennas, n_subcarriers, freq_mhz, sequence,
+        rssi, noise_floor, 0, 0,
+    )
+    # Synthesize I/Q: a steady carrier + a slow breathing oscillation + motion noise.
+    body = bytearray()
+    breath = math.sin(2 * math.pi * breath_hz * t)
+    for a in range(n_antennas):
+        for k in range(n_subcarriers):
+            base = 18 + (k % 30)
+            amp = base + 8 * breath + motion * math.sin(k * 0.4 + t * 6.0)
+            phase = 0.3 * k + t * 1.5
+            i = int(max(-127, min(127, amp * math.cos(phase))))
+            q = int(max(-127, min(127, amp * math.sin(phase))))
+            body += struct.pack("bb", i, q)
+    return hdr + bytes(body)
+
+
+def main():
+    p = argparse.ArgumentParser()
+    p.add_argument("--host", default="127.0.0.1")
+    p.add_argument("--port", type=int, default=5005)
+    p.add_argument("--nodes", type=int, default=2, help="number of nodes (node_id 1..N)")
+    p.add_argument("--antennas", type=int, default=1)
+    p.add_argument("--subcarriers", type=int, default=64)
+    p.add_argument("--freq-mhz", type=int, default=2462)
+    p.add_argument("--rate", type=float, default=20.0, help="frames/sec per node")
+    p.add_argument("--rssi", type=int, default=-47)
+    p.add_argument("--noise", type=int, default=-90)
+    p.add_argument("--breath-hz", type=float, default=0.25, help="~15 BPM")
+    p.add_argument("--motion", type=float, default=4.0, help="0 = still, higher = more motion")
+    p.add_argument("--duration", type=float, default=0.0, help="seconds; 0 = forever")
+    args = p.parse_args()
+
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    period = 1.0 / args.rate
+    seq = 0
+    start = time.time()
+    print(f"Sending CSI frames -> {args.host}:{args.port} | "
+          f"{args.nodes} node(s), {args.subcarriers} subcarriers, {args.rate} Hz")
+    try:
+        while True:
+            t = time.time() - start
+            for node_id in range(1, args.nodes + 1):
+                frame = build_frame(
+                    node_id, args.antennas, args.subcarriers, args.freq_mhz,
+                    seq, args.rssi, args.noise, t, args.breath_hz, args.motion,
+                )
+                sock.sendto(frame, (args.host, args.port))
+            seq += 1
+            if seq % int(args.rate) == 0:
+                print(f"  t={t:5.1f}s  sent seq={seq} ({len(frame)} B/frame)")
+            if args.duration and t >= args.duration:
+                break
+            time.sleep(period)
+    except KeyboardInterrupt:
+        pass
+    print(f"Done. Sent {seq} sequences to {args.nodes} node(s).")
+
+
+if __name__ == "__main__":
+    main()