Module: PWN::FFI::RTLSdr

Extended by:
Library
Defined in:
lib/pwn/ffi/rtl_sdr.rb

Overview

Thin librtlsdr binding (no -dev headers required — symbols resolved from the installed shared object). Control-plane + blocking read_sync so PWN::SDR::Decoder::* / Extrospection probe_rf can pull raw u8 I/Q without shelling out to rtl_sdr.

If librtlsdr is missing .available? is false and callers fall back.

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.load_error ⇒ Object (readonly)

Returns the value of attribute load_error.



26
27
28
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 26

def load_error
  @load_error
end

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com



181
182
183
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 181

public_class_method def self.authors
  "AUTHOR(S):\n  0day Inc. <support@0dayinc.com>\n"
end

.available? ⇒ Boolean

Supported Method Parameters

PWN::FFI::RTLSdr.available?

Returns:

  • (Boolean) 


55
56
57
58
59
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 55

public_class_method def self.available?
  !@load_error && respond_to?(:rtlsdr_get_device_count, true)
rescue StandardError
  false
end

.close(opts = {}) ⇒ Object

Supported Method Parameters

PWN::FFI::RTLSdr.close(device: pointer)



102
103
104
105
106
107
108
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 102

public_class_method def self.close(opts = {})
  dev = opts[:device]
  return unless dev && !dev.null?

  rtlsdr_close(dev)
  nil
end

.configure(opts = {}) ⇒ Object

rubocop:disable Naming/PredicateMethod



119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 119

public_class_method def self.configure(opts = {}) # rubocop:disable Naming/PredicateMethod
  dev = opts[:device]
  raise 'ERROR: :device required' if dev.nil? || dev.null?

  freq = opts[:freq_hz].to_i
  rate = (opts[:rate_hz] || 2_048_000).to_i
  ppm  = (opts[:ppm] || 0).to_i

  check!(rtlsdr_set_sample_rate(dev, rate), 'set_sample_rate')
  check!(rtlsdr_set_center_freq(dev, freq), 'set_center_freq')
  check!(rtlsdr_set_freq_correction(dev, ppm), 'set_freq_correction') if ppm != 0

  if opts.key?(:gain_db) && !opts[:gain_db].nil?
    check!(rtlsdr_set_tuner_gain_mode(dev, 1), 'gain_mode manual')
    check!(rtlsdr_set_tuner_gain(dev, opts[:gain_db].to_i), 'set_tuner_gain')
  else
    check!(rtlsdr_set_tuner_gain_mode(dev, 0), 'gain_mode auto')
    rtlsdr_set_agc_mode(dev, 1)
  end
  check!(rtlsdr_reset_buffer(dev), 'reset_buffer')
  true
end

.help ⇒ Object

Display Usage for this Module



187
188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 187

public_class_method def self.help
  puts "USAGE:
    #{self}.available?
    #{self}.list_devices
    dev = #{self}.open(index: 0)
    #{self}.configure(device: dev, freq_hz:, rate_hz: 2_048_000, gain_db: nil, ppm: 0)
    #{self}.tuner_gains(device: dev)
    iq = #{self}.read_sync(device: dev, bytes: 262_144)  # u8 I/Q String
    #{self}.close(device: dev)

    #{self}.authors
  "
end

.list_devices ⇒ Object

Supported Method Parameters

devices = PWN::FFI::RTLSdr.list_devices Returns Array of { index:, name:, manufacturer:, product:, serial: }.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 65

public_class_method def self.list_devices
  raise 'ERROR: librtlsdr not available' unless available?

  count = rtlsdr_get_device_count
  Array.new(count) do |i|
    mfr  = PubFFI::MemoryPointer.new(:char, 256)
    prod = PubFFI::MemoryPointer.new(:char, 256)
    ser  = PubFFI::MemoryPointer.new(:char, 256)
    rtlsdr_get_device_usb_strings(i, mfr, prod, ser)
    {
      index: i,
      name: (rtlsdr_get_device_name(i) || '').to_s,
      manufacturer: mfr.read_string,
      product: prod.read_string,
      serial: ser.read_string
    }
  end
end

.open(opts = {}) ⇒ Object

Supported Method Parameters

dev = PWN::FFI::RTLSdr.open(index: 0) Returns opaque FFI::Pointer (rtlsdr_dev_t*).



88
89
90
91
92
93
94
95
96
97
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 88

public_class_method def self.open(opts = {})
  raise 'ERROR: librtlsdr not available' unless available?

  idx = (opts[:index] || 0).to_i
  dev_ptr = PubFFI::MemoryPointer.new(:pointer)
  rc = rtlsdr_open(dev_ptr, idx)
  raise "ERROR: rtlsdr_open rc=#{rc}" unless rc.zero?

  dev_ptr.read_pointer
end

.read_sync(opts = {}) ⇒ Object

Supported Method Parameters

iq_u8 = PWN::FFI::RTLSdr.read_sync( device: 'required - pointer from .open', bytes: 'optional - number of raw bytes to read (default 262144)' ) Returns a binary String of unsigned 8-bit interleaved I/Q samples.



149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 149

public_class_method def self.read_sync(opts = {})
  dev = opts[:device]
  raise 'ERROR: :device required' if dev.nil? || dev.null?

  nbytes = (opts[:bytes] || 262_144).to_i
  buf = PubFFI::MemoryPointer.new(:uint8, nbytes)
  n_read = PubFFI::MemoryPointer.new(:int)
  rc = rtlsdr_read_sync(dev, buf, nbytes, n_read)
  raise "ERROR: rtlsdr_read_sync rc=#{rc}" unless rc.zero?

  got = n_read.read_int
  buf.read_string(got)
end

.tuner_gains(opts = {}) ⇒ Object

Supported Method Parameters

gains = PWN::FFI::RTLSdr.tuner_gains(device: pointer) Returns Array of supported gains in tenths of a dB.



167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 167

public_class_method def self.tuner_gains(opts = {})
  dev = opts[:device]
  raise 'ERROR: :device required' if dev.nil? || dev.null?

  n = rtlsdr_get_tuner_gains(dev, nil)
  return [] if n <= 0

  ptr = PubFFI::MemoryPointer.new(:int, n)
  rtlsdr_get_tuner_gains(dev, ptr)
  ptr.read_array_of_int(n)
end