Module: PWN::SDR::Decoder::LoRa

Defined in:
lib/pwn/sdr/decoder/lora.rb

Overview

LoRa (Semtech CSS) raw PHY payload decoder and separate detector. .decode supports SF7-12, CR4/5..4/8, explicit/implicit headers, optional payload CRC, LDRO and normal/inverted IQ at sample_rate >= BW. Raw payload bytes only: no LoRaWAN decryption or clock-drift tracking. The following preamble-only description applies to .detect.

I/Q resampled so fs = BW (default 125 kHz), one complex sample per chirp step. For each SF ∈ 7..12: dechirp with a reference down-chirp (DSP.cmul + PWN::FFI::FFTW.cfft), find ≥6 consecutive symbols whose FFT-argmax bin is identical (preamble), then read the two sync-word symbols and two SFD down-chirps. Emits bw_hz:, sync_word:, preamble_len:, cfo_bins:. This is preamble/sync metadata only, not LoRa payload decoding.

Defined Under Namespace

Classes: DemodIQ, PayloadIQ

Constant Summary collapse

DEFAULT_BW = 
125_000
SF_RANGE = 
(7..12)
KNOWN_SYNC =

Public LoRaWAN sync = 0x34; private/Meshtastic default = 0x12.

{ 0x34 => 'LoRaWAN', 0x12 => 'private/RadioLib' }.freeze

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. [email protected]



489
490
491
 
# File 'lib/pwn/sdr/decoder/lora.rb', line 489

public_class_method def self.authors
  "AUTHOR(S):\n  0day Inc. <[email protected]>\n"
end

.decode(opts = {}) ⇒ Object

Realtime options forwarded to Base: on_frame (Hash callback), output (writable IO), interactive (default true), duration (seconds), stop (callable), queue_size (bounded chunks), log_file (path or false).

Raises:

  • (ArgumentError) 


435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
 
# File 'lib/pwn/sdr/decoder/lora.rb', line 435

public_class_method def self.decode(opts = {})
  mode = opts[:mode] || :css
  raise ArgumentError, 'LoRa supports CSS mode only' unless mode.to_sym == :css

  freq_obj = opts[:freq_obj] || {}
  bw = opts[:bw] || freq_obj[:lora_bw] || DEFAULT_BW
  rate = opts[:sample_rate] || freq_obj[:iq_rate] || bw
  sf = opts[:sf] || 7
  demod = PayloadIQ.new(rate: rate, bw: bw, sf: sf, sync_word: opts[:sync_word] || 0x12,
                        implicit_header: opts.fetch(:implicit_header, false), payload_length: opts[:payload_length],
                        cr: opts.fetch(:cr, 1), crc: opts.fetch(:crc, true), ldro: opts[:ldro], invert_iq: opts.fetch(:invert_iq, false))
  PWN::SDR::Decoder::Base.run_iq(
    **opts, freq_obj: freq_obj, protocol: 'LoRa', sample_rate: rate, demod: demod,
            note: 'Raw LoRa PHY: SF7-12, explicit/implicit headers, optional CRC, LDRO, normal/inverted IQ.',
            describe: proc { |frame| { modulation: 'CSS', sf: frame[:sf], crc_valid: frame[:crc_valid] } }
  )
end

.detect(opts = {}) ⇒ Object 



453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
 
# File 'lib/pwn/sdr/decoder/lora.rb', line 453

public_class_method def self.detect(opts = {})
  freq_obj = opts[:freq_obj]
  bw   = (opts[:bw] || freq_obj[:lora_bw] || DEFAULT_BW).to_i
  rate = (opts[:sample_rate] || freq_obj[:iq_rate] || (bw * 4)).to_i
  PWN::SDR::Decoder::Base.run_iq(
    **opts,
    freq_obj: freq_obj,
    protocol: 'LoRa',
    sample_rate: rate,
    source: opts[:source],
    file: opts[:file],
    demod: DemodIQ.new(rate: rate, bw: bw),
    note: 'CSS 125–500 kHz — I/Q→resample_iq(fs=BW)→dechirp+FFTW per SF→preamble/sync-word.',
    describe: proc { |b| { modulation: 'CSS', bw_khz_assumed: bw / 1000, sf_estimate: b[:sf] } }
  )
end

.help ⇒ Object 



493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
 
# File 'lib/pwn/sdr/decoder/lora.rb', line 493

public_class_method def self.help
  puts "USAGE:
    # .decode: raw PHY SF7-12, CR4/5..4/8, explicit/implicit headers.
    # Optional CRC, LDRO, inverted IQ, fs>=BW; no clock drift tracking.
    # Output is raw payload_hex, NOT decrypted LoRaWAN application data.
    # .detect retains CSS preamble/sync acquisition, never decoded payload.
    #{self}.detect(freq_obj: 'required', bw: 'optional bandwidth Hz',
                   source: 'optional IQ input', on_frame: 'optional callback')

    # Run decode and return its result
    #{self}.decode(
      freq_obj: 'required - freq_obj returned from PWN::SDR::GQRX.init_freq',
      on_frame: 'optional - callback receiving each emitted Hash',
      output: 'optional - writable IO (default stdout)',
      interactive: 'optional - false disables ENTER input',
      duration: 'optional - finite seconds to run',
      stop: 'optional - callable returning true to stop',
      queue_size: 'optional - bounded pending chunks (default 8)',
      log_file: 'optional - JSONL path or false to disable logging',
      mode: 'optional - :css only, the default supported PHY mode',
      implicit_header: 'optional - false (default) or true with payload_length/cr/crc configured',
      invert_iq: 'optional - conjugate received IQ (default false)',
      ldro: 'optional - true/false; default auto when symbol duration exceeds 16ms',
      crc: 'optional - implicit-mode CRC presence (default true); explicit header is authoritative',
      payload_length: 'optional - integer byte count 0..255; required for implicit mode',
      cr: 'optional - implicit-mode coding rate 1..4 for 4/5..4/8 (default 1)',
      sf: 'optional - integer 7..12 (default 7)',
      sync_word: 'optional - integer byte (default 0x12; LoRaWAN 0x34)',
      iq_format: 'optional - cu8 (default), cs8 or cs16 for Base source',
      bw: 'optional - 125000 (default), 250000 or 500000; sample_rate must be >= BW',
      sample_rate: 'optional - sample rate value consumed by #decode',
      source: 'optional - source value consumed by #decode',
      file: 'optional - filesystem path'
    )

    # Run parse line and return its result
    #{self}.parse_line(
      line: 'optional - line value consumed by #parse_line'
    )

    # Print the AUTHOR(S) string for this module.
    #{self}.authors
  "
  constants.sort
end

.parse_line(opts = {}) ⇒ Object 



470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
 
# File 'lib/pwn/sdr/decoder/lora.rb', line 470

public_class_method def self.parse_line(opts = {})
  line = opts[:line].to_s
  h = begin
    JSON.parse(line, symbolize_names: true)
  rescue StandardError
    { unparsed: line }
  end
  out = { protocol: 'LoRa' }.merge(h)
  bits = []
  bits << "SF#{out[:sf]}" if out[:sf]
  bits << "BW#{out[:bw]}" if out[:bw]
  bits << "sync=#{out[:sync_word]}" if out[:sync_word]
  bits << out[:payload].to_s[0, 40] if out[:payload]
  out[:summary] = bits.empty? ? line[0, 120] : bits.join(' ')
  out
end