Module: PWN::SDR::Decoder::GPS

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

Overview

GPS L1 C/A acquisition (.detect) and optional native IQ/LNAV decoding. LNAV framing/parity does not imply ephemeris assembly or PVT.

Parallel-code-phase search: 1 ms of I/Q resampled to 2.046 Msps (2 samp/chip), FFT-correlated (via PWN::FFI::FFTW.cfft) against each PRN 1..32 Gold code across ±5 kHz Doppler in 500 Hz steps. Emits doppler_hz:, code_phase_chips:, cn0_db_hz: for every satellite whose peak/next-peak ratio clears threshold — the same cold-start acquisition every GNSS receiver runs, no gnss-sdr binary.

Defined Under Namespace

Classes: DemodIQ, IQTracker

Constant Summary collapse

CHIP_RATE = 
1_023_000
ACQ_RATE =

2 samp/chip → 2046-point FFT

2_046_000
DOPP_RANGE = 
5000
DOPP_STEP = 
500
ACQ_THRESH =

peak / mean ratio

2.5

[0xbb1f3480, 0x5d8f9a40, 0xaec7cd00, 0x5763e680, 0x6bb1f340, 0x8b7a89c0].freeze

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. [email protected]



409
410
411
 
# File 'lib/pwn/sdr/decoder/gps.rb', line 409

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

.decode(opts = {}) ⇒ Object

IQ mode supervises an optional native GNSS-SDR file receiver. It never selects hardware. Callbacks/output are synchronous; duration/stop are checked between callbacks. Detection alone uses Base.run_iq.

Raises:

  • (ArgumentError) 


287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
 
# File 'lib/pwn/sdr/decoder/gps.rb', line 287

public_class_method def self.decode(opts = {})
  mode = opts.fetch(:mode, :iq)
  raise ArgumentError, 'GPS mode must be :iq or :lnav_bits' unless i[iq lnav_bits].include?(mode)
  return IQTracker.new(opts).run if mode == :iq

  chunks = opts[:bit_chunks]
  raise ArgumentError, ':lnav_bits requires enumerable bit_chunks of tracked 50 bit/s NAV data' unless chunks.respond_to?(:each)

  buffer = []
  frames = []
  chunks.each do |chunk|
    raise ArgumentError, 'bit_chunks must contain arrays of binary bits' unless chunk.is_a?(Array) && chunk.all? { |b| b.is_a?(Integer) && [0, 1].include?(b) }

    buffer.concat(chunk)
    while buffer.length >= 300
      frame = decode_subframe(bits: buffer.first(300))
      buffer.shift(frame ? 300 : 1)
      next unless frame

      frames << frame
      opts[:on_frame]&.call(frame)
    end
  end
  frames
end

.decode_monitor(opts = {}) ⇒ Object

GNSS-SDR monitor words have data D30* de-inversion already applied. Restore the transmitted representation; never bypass our parity checks.



363
364
365
366
367
368
369
370
371
372
373
374
375
376
 
# File 'lib/pwn/sdr/decoder/gps.rb', line 363

public_class_method def self.decode_monitor(opts = {})
  message = opts[:nav_message]
  return nil unless message.is_a?(String) && message.match?(/\A[01]{300}\z/) && opts[:prn].is_a?(Integer) && opts[:prn].between?(1, 32)

  previous = 0 # LNAV word 10 terminates with D29*=D30*=0.
  bits = message.scan(/.{30}/).flat_map do |encoded|
    word = encoded.to_i(2)
    raw = previous.odd? ? word ^ 0x3fffffc0 : word
    previous = word & 3
    format('%030b', raw).chars.map(&:to_i)
  end
  frame = decode_subframe(bits: bits)
  frame&.merge(prn: opts[:prn], capability: 'iq-lnav-decode', backend: 'gnss-sdr')
end

.decode_subframe(opts = {}) ⇒ Object

Raises:

  • (ArgumentError) 


317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
 
# File 'lib/pwn/sdr/decoder/gps.rb', line 317

public_class_method def self.decode_subframe(opts = {})
  bits = opts[:bits]
  raise ArgumentError, 'LNAV subframe requires exactly 300 binary bits' unless bits.is_a?(Array) && bits.length == 300 && bits.all? { |b| b.is_a?(Integer) && [0, 1].include?(b) }

  preamble = bits.first(8).join
  return nil unless %w[10001011 01110100].include?(preamble)

  raw_words = bits.each_slice(30).map { |word| word.reduce(0) { |a, b| (a << 1) | b } }
  [0, 0x3fffffff].each do |polarity|
    4.times do |initial|
      previous = initial
      data = []
      raw_words.each do |raw|
        raw ^= polarity
        word = (previous << 30) | raw
        word ^= 0x3fffffc0 if previous.odd?
        parity = LNAV_PARITY_MASKS.reduce(0) { |a, mask| (a << 1) | ((word & mask).digits(2).sum & 1) }
        break unless parity == (raw & 0x3f)

        data << ((word >> 6) & 0xffffff)
        previous = raw & 3
      end
      next unless data.length == 10 && (data[0] >> 16) == 0x8b

      subframe_id = (data[1] >> 2) & 7
      tow = data[1] >> 7
      next unless subframe_id.between?(1, 5) && tow < 100_800
      next unless (raw_words[1] ^ polarity).nobits?(3) && (raw_words[9] ^ polarity).nobits?(3)

      frame = {
        protocol: 'GPS', event: 'lnav-subframe', capability: 'lnav-symbol-decode', decoded: true,
        checksum_verified: true, integrity: { algorithm: 'LNAV-word-parity', valid: true, words_verified: 10 },
        subframe_id: subframe_id, tow_next_seconds: tow * 6,
        alert: (data[1] >> 6).allbits?(1), anti_spoof: (data[1] >> 5).allbits?(1),
        payload_hex: data.map { |word| format('%06x', word) }.join, payload_bits: 240,
        summary: "GPS LNAV subframe #{subframe_id} parity verified TOW(next)=#{tow * 6}s"
      }
      frame[:week_mod1024] = data[2] >> 14 if subframe_id == 1
      return frame
    end
  end
  nil
end

.detect(opts = {}) ⇒ Object 



378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
 
# File 'lib/pwn/sdr/decoder/gps.rb', line 378

public_class_method def self.detect(opts = {})
  freq_obj = opts[:freq_obj] || {}
  rate = (opts[:sample_rate] || freq_obj[:iq_rate] || 2_048_000).to_i
  PWN::SDR::Decoder::Base.run_iq(
    **opts,
    freq_obj: freq_obj,
    protocol: 'GPS-L1CA',
    sample_rate: rate,
    source: opts[:source],
    file: opts[:file],
    demod: DemodIQ.new(rate: rate),
    note: 'BPSK/DSSS 1.023 Mcps — I/Q→FFT parallel-code-phase acquisition (PWN::FFI::FFTW) → PRN/Doppler/C-N0.',
    describe: proc { |_b| { modulation: 'BPSK/DSSS', chip_rate: CHIP_RATE } }
  )
end

.help ⇒ Object 



413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
 
# File 'lib/pwn/sdr/decoder/gps.rb', line 413

public_class_method def self.help
  puts "USAGE:
    # Parse a bounded GNSS-SDR monitor protobuf datagram.
    #{self}::IQTracker.parse_packet(packet: 'required - binary String containing one monitor datagram')

    # Track L1 C/A IQ and emit independently parity-verified LNAV frames.
    # Requires optional native gnss-sdr; never opens RF hardware.
    #{self}.decode(
      mode: 'optional - :iq (default) or :lnav_bits',
      source: 'required - :file for IQ mode',
      file: 'required - regular interleaved IQ file for IQ mode',
      sample_rate: 'optional - 2000000..25000000 Hz, default 4000000',
      iq_format: 'optional - :cs16 (default), :cs8 or :cf32, native little-endian',
      gnss_sdr: 'optional - native receiver executable, default gnss-sdr',
      duration: 'optional - positive finite timeout, default 300 seconds',
      stop: 'optional - cooperative callable checked between callbacks',
      output: 'optional - writable JSONL IO',
      bit_chunks: 'required - Enumerable of Array<0|1> for :lnav_bits only',
      on_frame: 'optional - synchronous callback before source EOF'
    )
    # Restore GNSS-SDR monitor D30 inversion and verify all word parities.
    #{self}.decode_monitor(
      nav_message: 'required - 300-character monitor bit string',
      prn: 'required - GPS PRN integer 1..32'
    )
    # Verify all ten LNAV word parities and extract subframe header fields.
    #{self}.decode_subframe(bits: 'required - exactly 300 binary bits; nil on integrity failure')

    # Capture observations only, never decoded payloads.
    #{self}.detect(
      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',
      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 



394
395
396
397
398
399
400
401
402
403
404
405
 
# File 'lib/pwn/sdr/decoder/gps.rb', line 394

public_class_method def self.parse_line(opts = {})
  line = opts[:line].to_s
  out  = { protocol: 'GPS' }
  out[:prn]  = ::Regexp.last_match(1) if line =~ /PRN[ =]?(\d{1,2})/
  out[:cn0]  = ::Regexp.last_match(1) if line =~ /CN0[ =]?([\d.]+)/i
  out[:lat]  = ::Regexp.last_match(1) if line =~ /Lat(?:itude)?\s*=\s*(-?[\d.]+)/i
  out[:lon]  = ::Regexp.last_match(1) if line =~ /Long(?:itude)?\s*=\s*(-?[\d.]+)/i
  out[:alt]  = ::Regexp.last_match(1) if line =~ /Height\s*=\s*(-?[\d.]+)/i
  out[:nmea] = line if line.start_with?('$G')
  out[:summary] = out[:lat] ? "GPS FIX #{out[:lat]},#{out[:lon]}" : "GPS PRN#{out[:prn]} CN0=#{out[:cn0]}"
  out.compact
end