Module: PWN::SDR::Decoder::P25

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

Overview

APCO Project 25 Phase-1 (C4FM) true-air decoder.

I/Q → PWN::FFI::Liquid.freq_demod (or DSP.fm_demod_iq) → resample to 48 kHz → 4-level slice at 4800 sym/s → dibits → hunt the 24-symbol Frame Sync (0x5575F5FF77FF) → recover the 64-bit NID (12-bit NAC + 4-bit DUID + BCH(63,16,23) parity). Emits duid:, duid_name: from legacy acquisition (DemodIQ). The public .decode uses PacketIQ: BCH-checked one-to-three-block TSDU, trellis/CRC and group grants. Unconfirmed rate-1/2 packet data also supported; no voice or Phase 2.

Defined Under Namespace

Classes: DemodIQ, PacketData, PacketIQ

Constant Summary collapse

FS_DIBITS =

24-symbol / 48-bit Frame Sync (dibit MSB-first)

[
  1, 1, 1, 1, 3, 1, 1, 3, 3, 3, 3, 1, 1, 3, 3, 3,
  3, 3, 3, 3, 1, 3, 3, 3, 3, 3, 3, 3
].freeze
FS_DIBITS_24 =

→ 0x5575F5FF77FF (see TIA-102.BAAA)

[
  1, 1, 1, 1, 3, 1, 1, 3, 3, 3, 3, 1,
  3, 3, 1, 1, 3, 3, 3, 3, 3, 3, 3, 3
].freeze
FRAME_SYNC =

Correct 24-dibit Frame Sync per TIA-102 (+3 +3 +3 +3 −3 +3 …). Derived from bit pattern 5575F5FF77FF, MSB-first, 2 bits/sym, C4FM map: 01→+3, 00→+1, 10→−1, 11→−3 → dibits 1,0,2,3.

0x5575F5FF77FF
FS_BITS = 
Array.new(48) { |i| (FRAME_SYNC >> (47 - i)) & 1 }.freeze
FS_SYMS = 
FS_BITS.each_slice(2).map { |a, b| (a << 1) | b }.freeze
DUID_NAME = 
{
  0x0 => 'HDU', 0x3 => 'TDU', 0x5 => 'LDU1', 0x7 => 'TSBK',
  0xA => 'LDU2', 0xC => 'PDU', 0xF => 'TDULC'
}.freeze
BCH_GENERATOR = 
0xCD930BDD3B2B

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. [email protected]



378
379
380
 
# File 'lib/pwn/sdr/decoder/p25.rb', line 378

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) 


339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
 
# File 'lib/pwn/sdr/decoder/p25.rb', line 339

public_class_method def self.decode(opts = {})
  raise ArgumentError, 'only :tsbk, :pdu (unconfirmed), or :all are supported' unless %i[tsbk pdu all].include?(opts.fetch(:mode, :tsbk).to_sym)
  raise ArgumentError, 'only Phase 1 C4FM is supported' unless opts.fetch(:phase, 1) == 1 && opts.fetch(:modulation, :c4fm).to_sym == :c4fm

  freq_obj = opts[:freq_obj]
  rate = (opts[:sample_rate] || freq_obj[:iq_rate] || (48_000 * 20)).to_i
  PWN::SDR::Decoder::Base.run_iq(
    **opts,
    freq_obj: freq_obj,
    protocol: 'P25',
    sample_rate: rate,
    source: opts[:source],
    file: opts[:file],
    demod: PacketIQ.new(rate: rate, mode: opts.fetch(:mode, :tsbk)),
    fallback: :raise,
    note: 'C4FM TSDU/unconfirmed PDU: BCH NID, status removal, Viterbi, CRC16 and packet CRC32.',
    describe: proc { |b| { modulation: 'C4FM', classification: b[:duration_ms] > 180 ? 'voice-LDU' : 'TSBK/control' } }
  )
end

.detect(opts = {}) ⇒ Object

Energy observations only; never substituted for protocol decoding.



360
361
362
363
 
# File 'lib/pwn/sdr/decoder/p25.rb', line 360

public_class_method def self.detect(opts = {})
  Base.run_detector(opts.merge(protocol: 'P25',
                               note: 'Energy detector only; use .decode for supported protocol frames.'))
end

.help ⇒ Object 



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
419
420
421
422
423
424
 
# File 'lib/pwn/sdr/decoder/p25.rb', line 382

public_class_method def self.help
  puts "USAGE:
    # Parse and integrity-check an unconfirmed packet data unit.
    #{self}::PacketData.parse(
      data: 'required - Array of synchronized dibits without status symbols',
      info: 'required - Integer containing verified NAC and DUID bits',
      decoder: 'required - callable accepting a trellis block and returning bytes and corrected-bit count'
    )

    # Supported: centered Phase-1 C4FM TSDU with 1..3 TSBKs (mode: :tsbk).
    # mode: :pdu decodes unconfirmed format 0x15, rate-1/2 PDUs; :all enables both.
    # Integer samples/symbol, rate >= 48000. Status symbols are removed;
    # BCH and parity are checked (no NID correction), trellis Viterbi and CRC verified.
    # Protected payload is ciphertext, never plaintext. No voice, CQPSK,
    # Phase 2, confirmed PDU/AMBT or frequency-offset acquisition. No decryption.
    #{self}.detect(freq_obj: 'required', 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',
      sample_rate: 'optional - sample rate value consumed by #decode',
      mode: 'optional - :tsbk (default), :pdu (unconfirmed), :all',
      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 



365
366
367
368
369
370
371
372
373
374
 
# File 'lib/pwn/sdr/decoder/p25.rb', line 365

public_class_method def self.parse_line(opts = {})
  line = opts[:line].to_s
  out  = { protocol: 'P25' }
  out[:nac]  = ::Regexp.last_match(1) if line =~ /NAC[:= ]+([0-9A-Fa-f]+)/
  out[:tg]   = ::Regexp.last_match(1) if line =~ /(?:TG|talkgroup)[:= ]+(\d+)/i
  out[:rid]  = ::Regexp.last_match(1) if line =~ /(?:RID|src|source)[:= ]+(\d+)/i
  out[:duid] = ::Regexp.last_match(1) if line =~ /DUID[:= ]+(\w+)/i
  out[:summary] = "P25 NAC=#{out[:nac]} TG=#{out[:tg]} RID=#{out[:rid]}"
  out.compact
end