Module: PWN::SDR::Decoder::LTE

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

Overview

LTE PSS observations (.detect) and optional native PBCH MIB (.decode). PBCH: acquired or caller-aligned SF0, 1.92Msps, FDD, normal CP, 2 ports. No SIB/traffic decoding or live-RF validation.

Defined Under Namespace

Classes: AcquiredPBCHIQ, DemodIQ, PBCHIQ

Constant Summary collapse

FS_BASE = 
1_920_000
NFFT = 
128
CP_NORM =

samples @ 1.92 Msps for symbols 1..6 (10 for symbol 0)

9
PSS_ROOTS = 
{ 0 => 25, 1 => 29, 2 => 34 }.freeze

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. [email protected]



356
357
358
 
# File 'lib/pwn/sdr/decoder/lte.rb', line 356

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

.cseq(opts = {}) ⇒ Object

Scrambling sequence c0 (x^5+x^3+1) tied to N_ID_2, ±1.



287
288
289
290
291
292
293
294
295
296
297
298
299
 
# File 'lib/pwn/sdr/decoder/lte.rb', line 287

public_class_method def self.cseq(opts = {})
  @cseq_base ||= begin
    reg = [0, 0, 0, 0, 1]
    Array.new(31) do
      o = reg[0]
      fb = reg[0] ^ reg[2]
      reg = reg[1..] + [fb]
      1 - (2 * o)
    end
  end
  sh = opts[:nid2].to_i
  Array.new(31) { |n| @cseq_base[(n + sh) % 31] }
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:

  • (NotImplementedError) 


309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
 
# File 'lib/pwn/sdr/decoder/lte.rb', line 309

public_class_method def self.decode(opts = {})
  raise NotImplementedError, "IQ mode #{opts[:mode] || :iq} is unsupported; use :pbch_iq for offline acquisition, :pbch_sf0_iq for aligned SF0, or .detect for observations" unless %i[pbch_sf0_iq pbch_iq].include?(opts[:mode])

  raise NotImplementedError, 'PBCH bridge supports only FDD, normal CP, two transmit ports' unless opts.fetch(:duplex, :fdd) == :fdd && opts.fetch(:cyclic_prefix, :normal) == :normal && opts.fetch(:tx_ports, 2) == 2

  rate = opts.fetch(:sample_rate, FS_BASE)
  raise ArgumentError, 'PBCH requires sample_rate: 1920000' unless rate == FS_BASE
  raise ArgumentError, 'PBCH requires integer PCI in 0..503 (caller supplied, not acquired)' if opts[:mode] == :pbch_sf0_iq && !(opts[:pci].is_a?(Integer) && opts[:pci].between?(0, 503))

  source = opts[:source]
  offline = source.respond_to?(:read) || (source.is_a?(Hash) && source[:kind] == :io && source[:io].respond_to?(:read)) || (opts[:file] && [nil, :file].include?(source))
  raise ArgumentError, 'PBCH requires an explicit offline file or IO; hardware acquisition is not supported' unless offline

  Base.run_iq(opts.merge(freq_obj: opts[:freq_obj] || {}, protocol: 'LTE', sample_rate: rate,
                         demod: opts[:mode] == :pbch_iq ? AcquiredPBCHIQ.new : PBCHIQ.new(pci: opts[:pci]),
                         note: 'Native PBCH MIB only: 1.92Msps FDD normal CP, two TX ports. :pbch_iq acquires PSS/SSS; :pbch_sf0_iq requires aligned SF0 and PCI. No SIB/traffic.'))
end

.detect(opts = {}) ⇒ Object 



327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
 
# File 'lib/pwn/sdr/decoder/lte.rb', line 327

public_class_method def self.detect(opts = {})
  freq_obj = opts[:freq_obj] || {}
  rate = (opts[:sample_rate] || freq_obj[:iq_rate] || 1_920_000).to_i
  PWN::SDR::Decoder::Base.run_iq(
    **opts,
    freq_obj: freq_obj,
    protocol: 'LTE',
    sample_rate: rate,
    source: opts[:source],
    file: opts[:file],
    demod: DemodIQ.new(rate: rate),
    note: 'OFDMA PSS correlation observations only; SSS/PCI, PBCH/MIB and traffic decoding are unsupported.',
    describe: proc { |_b| { modulation: 'OFDMA', subcarrier_khz: 15 } }
  )
end

.help ⇒ Object 



360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
 
# File 'lib/pwn/sdr/decoder/lte.rb', line 360

public_class_method def self.help
  puts "USAGE:
    # SSS helper: (m0, m1) pair for a given N_ID_1 per TS 36.211 §6.11.2
    #{self}.sss_indices(
      nid1: 'optional - nid1 value consumed by #sss_indices'
    )

    # Length-31 m-sequence x^5+x^2+1, cyclic-shifted by `shift`, as ±1
    #{self}.mseq(
      shift: 'optional - shift value consumed by #mseq'
    )

    # Scrambling sequence c0 (x^5+x^3+1) tied to N_ID_2, ±1
    #{self}.cseq(
      nid2: 'optional - nid2 value consumed by #cseq'
    )

    # Optional native PBCH MIB: see ext/pwn_lte/README.md for build/provenance.
    # Input is consecutive, separately extracted 1ms subframe-zero blocks.
    # Not a continuous capture; caller supplies PCI/timing/frequency correction.
    # Only FDD, normal CP, two TX ports; no SIB or traffic.
    # For continuous central-six-PRB IQ, use mode: :pbch_iq and omit PCI.
    # This acquires PSS/SSS timing/PCI and estimates fractional CFO.
    #{self}.decode(
      mode: 'required - :pbch_sf0_iq for aligned subframe-zero IQ blocks',
      pci: 'required - physical cell identity integer from 0 through 503',
      sample_rate: 1920000,
      source: :file, file: 'aligned-sf0.cs16', iq_format: :cs16,
      duplex: :fdd, cyclic_prefix: :normal, tx_ports: 2,
      interactive: false, log_file: false,
      on_frame: proc { |frame| p frame }
    )
    # Other PHY modes remain unsupported; detection is always explicit.
    #{self}.detect(freq_obj: {}, source: :file, file: 'capture.cu8')
    # Run detection only.
    #{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

.mseq(opts = {}) ⇒ Object

Length-31 m-sequence x^5+x^2+1, cyclic-shifted by shift, as ±1.



272
273
274
275
276
277
278
279
280
281
282
283
284
 
# File 'lib/pwn/sdr/decoder/lte.rb', line 272

public_class_method def self.mseq(opts = {})
  @mseq_base ||= begin
    reg = [0, 0, 0, 0, 1]
    Array.new(31) do
      o = reg[0]
      fb = reg[0] ^ reg[3]
      reg = reg[1..] + [fb]
      1 - (2 * o)
    end
  end
  sh = opts[:shift].to_i
  Array.new(31) { |n| @mseq_base[(n + sh) % 31] }
end

.parse_line(opts = {}) ⇒ Object 



343
344
345
346
347
348
349
350
351
352
 
# File 'lib/pwn/sdr/decoder/lte.rb', line 343

public_class_method def self.parse_line(opts = {})
  line = opts[:line].to_s
  out  = { protocol: 'LTE' }
  out[:earfcn] = ::Regexp.last_match(1) if line =~ /EARFCN[:= ]+(\d+)/i
  out[:pci]    = ::Regexp.last_match(1) if line =~ /(?:PCI|N_id_cell|Id)[:= ]+(\d{1,3})/i
  out[:prb]    = (::Regexp.last_match(1) || ::Regexp.last_match(2)) if line =~ /(?:PRB[:= ]+(\d+)|(\d+)\s*PRB)/i
  out[:rsrp]   = ::Regexp.last_match(1) if line =~ /(-?\d+(?:\.\d+)?)\s*dBm/
  out[:summary] = "LTE PCI=#{out[:pci]} EARFCN=#{out[:earfcn]} PRB=#{out[:prb]} RSRP=#{out[:rsrp]}dBm"
  out.compact
end

.sss_indices(opts = {}) ⇒ Object

SSS helper: (m0, m1) pair for a given N_ID_1 per TS 36.211 §6.11.2.



261
262
263
264
265
266
267
268
269
 
# File 'lib/pwn/sdr/decoder/lte.rb', line 261

public_class_method def self.sss_indices(opts = {})
  nid1 = opts[:nid1].to_i
  qp = (nid1 / 30)
  q  = ((nid1 + (qp * (qp + 1) / 2)) / 30)
  mp = nid1 + (q * (q + 1) / 2)
  m0 = mp % 31
  m1 = (m0 + (mp / 31) + 1) % 31
  [m0, m1]
end