gen_beacon_iq.py

#!/usr/bin/env python3
"""Generate a HackRF-transmittable 1 Mbps DSSS 802.11b beacon as a signed-int8 .c8.

This builds a real, on-air-valid IEEE 802.11 Beacon (management) frame at 1 Mbps
DBPSK DSSS (long preamble, 11-chip Barker) with a caller-chosen BSSID and ESSID,
modulates it exactly the way the opendsss transmitter does, and writes a raw
HackRF sample file you can transmit with hackrf_transfer.

BSSID and ESSID are set on the command line (--bssid / --ssid), so you can beacon
any network name/address you like for bench bring-up and replay: point a HackRF at
your board (or capture with any 802.11b receiver) and the frames appear as a normal
1 Mbps DSSS beacon that can be grepped by SSID/BSSID.

Modulation (identical model to the opendsss dsss_tx core):
  * frame bits  = [128 SYNC ones][SFD 0xF3A0][PLCP: SIGNAL/SERVICE/LENGTH/CRC-16]
                  [MPDU octets][CRC-32 FCS]        (built by scripts/dsss_lib.py)
  * scrambler   = self-synchronizing x^7 + x^4 + 1  (C(n) = D(n) ^ C(n-4) ^ C(n-7))
  * DBPSK       = differential phase; per-symbol sign +1/-1
  * spreading   = true 11-chip Barker [+ - + + - + + + - - -] point-sampled at
                  40 MSPS (40 samples/symbol, ~3.64 samples/chip), real baseband
                  (Q = 0), full +/-11 MHz main lobe a commercial 11b chip despreads
  * output      = int16 pulse (peak 5040) rescaled by a single divisor to signed
                  int8, interleaved I,Q,I,Q,... (HackRF native, +/-127 full scale)

This is a self-contained beacon builder that depends only on scripts/dsss_lib.py
(the bit-exact framer/CRC model) and the Python standard library.

Usage
-----
    # 1 Mbps DSSS beacon, SSID "openwifi-dsss", channel 6, one frame
    python3 scripts/gen_beacon_iq.py --out /tmp/beacon

    # custom name/address, channel 11, 20 tiled beacons (each numbered by seq)
    python3 scripts/gen_beacon_iq.py \
        --ssid my-dsss --bssid 02:00:dd:55:55:01 --channel 11 --repeats 20 \
        --out /tmp/beacon

The script prints the exact hackrf_transfer command to transmit the file (looped).
"""
import argparse
import os
import struct
import sys

# Import the bit-exact framer/CRC model from the same directory, regardless of cwd.
_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)
import dsss_lib as L  # noqa: E402  build_descrambled_frame + CRC engines

# --- DSSS TX modulation constants (must match the dsss_tx RTL core) ---------------
FS = 40_000_000                 # 40 MSPS DAC rate (true 11-Mcps Barker needs it)
N_SAMP_PER_SYM = 40             # 1 Msym/s at 40 MSPS -> 40 samples per DBPSK symbol
N_PREAMBLE_ONES = 128           # long preamble = 128 SYNC ones (spec long preamble)
PEAK = 5040                     # int16 pulse amplitude (unity bb_gain, < 2^15)
# TRUE 11-chip Barker point-sampled across the 40-sample symbol (40/11 = 3.64 samp/chip).
BARKER = [1, -1, 1, 1, -1, 1, 1, 1, -1, -1, -1]
PULSE40 = [BARKER[min(k * 11 // 40, 10)] * PEAK for k in range(N_SAMP_PER_SYM)]

# int16 (pulse peak 5040) -> signed int8 (HackRF +/-127). 5040/42 = 120 -> ~0.5 dB backoff.
INT8_DIV = 42
INT8_MAX, INT8_MIN = 127, -128

# idle (zero) samples padded before/after/between beacon frames so the receiver AGC
# and power estimator settle (100 us at 40 MSPS).
GAP_SAMPLES = 4000

# HackRF TX VGA gain used in the printed command (0..47 dB). Adjust to taste / range.
TX_VGA_GAIN = 40


def chan_to_freq_hz(channel):
    """2.4 GHz 802.11 channel -> center frequency in Hz (channel 14 special-cased)."""
    if channel == 14:
        mhz = 2484
    elif 1 <= channel <= 13:
        mhz = 2412 + (channel - 1) * 5
    else:
        raise ValueError("channel must be 1..14 (2.4 GHz)")
    return mhz * 1_000_000


def parse_bssid(s):
    """Parse a colon- or dash-separated hex MAC into 6 bytes."""
    parts = s.replace("-", ":").split(":")
    if len(parts) != 6:
        raise ValueError("BSSID must be 6 colon-hex octets, e.g. 02:00:dd:55:55:01")
    return bytes(int(p, 16) for p in parts)


def build_beacon_mpdu(ssid, bssid, seq, channel):
    """Return an IEEE 802.11 Beacon (management) MPDU as a list[int] of octets, WITHOUT
    the FCS (dsss_lib.build_descrambled_frame appends the standard CRC-32 FCS).

    Multi-byte fields are little-endian, per 802.11:
      FrameControl = 0x0080 (type MGMT, subtype Beacon, no flags) -> bytes 0x80, 0x00
      Duration     = 0x0000
      Address1 (DA)    = ff:ff:ff:ff:ff:ff (broadcast)
      Address2 (SA)    = bssid
      Address3 (BSSID) = bssid
      SequenceControl  = (seq << 4) little-endian (fragment 0, sequence seq)
      Frame body:
        Timestamp(8)        = seq (LE)   (a running counter, aids capture tracking)
        Beacon Interval     = 0x0064 (100 TU)
        Capability Info     = 0x0001 (ESS)
        SSID IE             : 0x00, len(ssid), ssid
        Supported Rates IE  : 0x01, 0x01, 0x82   (1 Mbps only, basic-rate bit set)
        DS Parameter Set IE : 0x03, 0x01, channel
    """
    if len(bssid) != 6:
        raise ValueError("bssid must be 6 bytes")
    sb = ssid.encode("ascii")
    if len(sb) > 32:
        raise ValueError("SSID must be <= 32 bytes")
    m = bytearray()
    m += bytes([0x80, 0x00])                     # Frame Control: MGMT / Beacon
    m += struct.pack("<H", 0x0000)               # Duration
    m += b"\xff\xff\xff\xff\xff\xff"             # Address1 (DA) = broadcast
    m += bssid                                   # Address2 (SA)  = bssid
    m += bssid                                   # Address3 (BSSID)
    m += struct.pack("<H", (seq << 4) & 0xFFFF)  # Sequence Control (frag 0, seq)
    # --- frame body ---
    m += struct.pack("<Q", seq & 0xFFFFFFFFFFFFFFFF)   # Timestamp (8 bytes)
    m += struct.pack("<H", 0x0064)               # Beacon Interval = 100 TU
    m += struct.pack("<H", 0x0001)               # Capability Info = ESS
    m += bytes([0x00, len(sb)]) + sb             # SSID IE
    m += bytes([0x01, 0x01, 0x82])               # Supported Rates IE: 1 Mbps only (basic)
    m += bytes([0x03, 0x01, channel & 0xFF])     # DS Parameter Set IE: channel
    return list(m)


def check_beacon_mpdu(mpdu, ssid, bssid, seq, channel):
    """Parse an MPDU from build_beacon_mpdu back into fields and verify them.
    Returns (ok, report_lines)."""
    rep, ok = [], True

    def need(cond, msg):
        nonlocal ok
        rep.append(("PASS " if cond else "FAIL ") + msg)
        ok = ok and cond

    b = bytes(mpdu)
    need(len(b) >= 24, "MPDU length %d >= 24 (MAC header)" % len(b))
    need(b[0:2] == b"\x80\x00", "FrameControl == 0x80 0x00 (MGMT/Beacon)")
    need(b[2:4] == b"\x00\x00", "Duration == 0x0000")
    need(b[4:10] == b"\xff\xff\xff\xff\xff\xff", "Address1 (DA) == broadcast")
    need(b[10:16] == bytes(bssid), "Address2 (SA) == %s" % bytes(bssid).hex(":"))
    need(b[16:22] == bytes(bssid), "Address3 (BSSID) == %s" % bytes(bssid).hex(":"))
    seqctl = struct.unpack_from("<H", b, 22)[0]
    need((seqctl >> 4) == (seq & 0x0FFF), "Sequence == %d" % (seq & 0x0FFF))
    need(struct.unpack_from("<H", b, 32)[0] == 0x0064, "Beacon Interval == 100 TU")
    need(struct.unpack_from("<H", b, 34)[0] & 0x0001, "Capability ESS bit set")
    ies, p = {}, 36
    while p + 2 <= len(b):
        eid, elen = b[p], b[p + 1]
        if p + 2 + elen > len(b):
            need(False, "IE %d overruns MPDU" % eid)
            break
        ies[eid] = b[p + 2:p + 2 + elen]
        p += 2 + elen
    need(ies.get(0, b"").decode("ascii", "replace") == ssid, "SSID IE == '%s'" % ssid)
    need(ies.get(1) == b"\x82", "Supported Rates IE == 1 Mbps only (basic)")
    need(ies.get(3) == bytes([channel & 0xFF]), "DS Parameter Set channel == %d" % channel)
    return ok, rep


def scramble(D):
    """Self-synchronizing scrambler x^7 + x^4 + 1: C(n) = D(n) ^ C(n-4) ^ C(n-7)."""
    C = []
    for n, d in enumerate(D):
        c4 = C[n - 4] if n >= 4 else 0
        c7 = C[n - 7] if n >= 7 else 0
        C.append(d ^ c4 ^ c7)
    return C


def modulate_frame(mpdu):
    """MPDU octets -> a list of (i16, q16) baseband samples for the frame body (no pad).
    Real baseband: Q is always 0. Returns (samples, n_frame_bits, n_symbols)."""
    D = L.build_descrambled_frame(mpdu, n_preamble_ones=N_PREAMBLE_ONES)
    C = scramble(D)
    # differential BPSK: phase(n) = phase(n-1) ^ C(n); sign = +1 if phase 0 else -1
    phase, signs = 0, []
    for c in C:
        phase ^= c
        signs.append(1 if phase == 0 else -1)
    samples = []
    for s in signs:
        for k in range(N_SAMP_PER_SYM):
            samples.append((s * PULSE40[k], 0))
    return samples, len(D), len(signs)


def int16_to_int8(samples, div):
    """Rescale (i16, q16) samples by one divisor and clip to signed int8; return
    (raw_bytes, n_clipped). Interleaved I,Q,I,Q,... one signed byte per rail."""
    out = bytearray()
    nclip = 0
    for ii, qq in samples:
        bi = int(round(ii / div))
        bq = int(round(qq / div))
        clipped = False
        if bi > INT8_MAX:
            bi, clipped = INT8_MAX, True
        elif bi < INT8_MIN:
            bi, clipped = INT8_MIN, True
        if bq > INT8_MAX:
            bq, clipped = INT8_MAX, True
        elif bq < INT8_MIN:
            bq, clipped = INT8_MIN, True
        nclip += 1 if clipped else 0
        out.append(bi & 0xFF)
        out.append(bq & 0xFF)
    return bytes(out), nclip


def main():
    ap = argparse.ArgumentParser(
        description="Generate a HackRF-transmittable 1 Mbps DSSS 802.11b beacon (.c8).")
    ap.add_argument("--ssid", default="openwifi-dsss",
                    help="beacon SSID / network name (default 'openwifi-dsss')")
    ap.add_argument("--bssid", default="02:00:dd:55:55:01",
                    help="beacon BSSID, colon-hex (default 02:00:dd:55:55:01, "
                         "locally-administered)")
    ap.add_argument("--channel", type=int, default=6,
                    help="2.4 GHz channel 1..14 (default 6); sets the DS Parameter Set "
                         "IE and the printed hackrf_transfer center frequency")
    ap.add_argument("--repeats", type=int, default=1,
                    help="number of beacon frames tiled into the file, each with an "
                         "incrementing Sequence Control (default 1)")
    ap.add_argument("--out", required=True,
                    help="output path; '.c8' is appended if absent")
    args = ap.parse_args()

    try:
        bssid = parse_bssid(args.bssid)
    except ValueError as e:
        ap.error(str(e))
    try:
        freq_hz = chan_to_freq_hz(args.channel)
    except ValueError as e:
        ap.error(str(e))
    if args.repeats < 1:
        ap.error("--repeats must be >= 1")

    out_c8 = args.out if args.out.endswith(".c8") else args.out + ".c8"
    out_dir = os.path.dirname(os.path.abspath(out_c8))
    os.makedirs(out_dir, exist_ok=True)

    gap = [(0, 0)] * GAP_SAMPLES
    stream = list(gap)                       # leading settle pad
    seqs = []
    mpdu0 = None
    n_bits = n_sym = frame_len = 0
    for k in range(args.repeats):
        seq = k                              # each tiled beacon is individually numbered
        seqs.append(seq)
        mpdu = build_beacon_mpdu(args.ssid, bssid, seq, args.channel)
        if mpdu0 is None:
            mpdu0 = mpdu
        body, n_bits, n_sym = modulate_frame(mpdu)
        frame_len = len(body)
        stream.extend(body)
        stream.extend(gap)                   # inter-frame / trailing idle

    raw8, nclip = int16_to_int8(stream, INT8_DIV)
    with open(out_c8, "wb") as f:
        f.write(raw8)

    total_samples = len(stream)
    dur_us = total_samples / FS * 1e6
    int8_peak = max((b - 256 if b > 127 else b) for b in raw8) if raw8 else 0
    int8_min = min((b - 256 if b > 127 else b) for b in raw8) if raw8 else 0

    # --- beacon self-check (parse the MPDU back + sanity-check the output size) ---
    ok, rep = check_beacon_mpdu(mpdu0, args.ssid, bssid, seqs[0], args.channel)

    print("wrote %s" % out_c8)
    print("  SSID='%s'  BSSID=%s  channel=%d (%d Hz)"
          % (args.ssid, bssid.hex(":"), args.channel, freq_hz))
    print("  MPDU=%d octets (+4 FCS)  frame=%d bits, %d symbols, %d samples/frame"
          % (len(mpdu0), n_bits, n_sym, frame_len))
    print("  repeats=%d  seq=%s  gap=%d samples"
          % (args.repeats, ("%d..%d" % (seqs[0], seqs[-1]) if len(seqs) > 1 else str(seqs[0])),
             GAP_SAMPLES))
    print("  %d samples = %d bytes int8 I/Q (%.1f us @ %d MSPS)  int8 range=[%d,%d]  clips=%d"
          % (total_samples, len(raw8), dur_us, FS // 1_000_000, int8_min, int8_peak, nclip))
    print("  --- beacon MPDU self-check (seq=%d) ---" % seqs[0])
    for line in rep:
        print("    " + line)
    print("  SELF-CHECK: %s" % ("PASS" if ok else "FAIL"))
    print("")
    print("Transmit it (looped) with:")
    print("  hackrf_transfer -t %s -f %d -s 40000000 -x %d -a 0 -R"
          % (out_c8, freq_hz, TX_VGA_GAIN))
    print("  (-x is the TX VGA gain in dB, 0..47; -a 0 keeps the extra RF amp off; "
          "-R loops the file.)")
    if not ok:
        sys.exit("SELF-CHECK FAILED -> beacon bytes are wrong; not safe to transmit.")


if __name__ == "__main__":
    main()