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. [email protected]



187
188
189
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 187

public_class_method def self.authors
  "AUTHOR(S):\n  0day Inc. <[email protected]>\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



193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 193

public_class_method def self.help
  puts "USAGE:
    # Run available and return its result
    #{self}.available?

    # Run list devices and return its result
    #{self}.list_devices

    # Open a session or connection and return a handle.
    #{self}.open(
      index: 'optional - index value consumed by #open'
    )

    # Close a session previously returned by #open.
    #{self}.close(
      device: 'optional - device value consumed by #close'
    )

    # Run configure and return its result
    #{self}.configure(
      device: 'required - pointer from .open',
      freq_hz: 'required - center frequency Hz',
      rate_hz: 'optional - sample rate (default 2.048e6)',
      gain_db: 'optional - tenths of dB (e.g. 496 = 49.6 dB). nil = auto',
      ppm: 'optional - freq correction ppm (default 0)'
    )

    # Run read sync and return its result
    #{self}.read_sync(
      device: 'required - pointer from .open',
      bytes: 'optional - number of raw bytes to read (default 262144)'
    )

    # Run tuner gains and return its result
    #{self}.tuner_gains(
      device: 'required - device value consumed by #tuner_gains'
    )

    # Print the AUTHOR(S) string for this module.
    #{self}.authors
  "
  constants.sort
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. Short, complete IQ reads are returned intact. Native errors and odd byte counts raise. librtlsdr sync reads provide no dropped-sample count; successful reads alone do not establish continuous hardware capture. Callers must finish an in-flight read before closing the device.



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 153

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
  raise "ERROR: rtlsdr_read_sync invalid IQ length #{got}/#{nbytes}" if got.negative? || got > nbytes || got.odd?

  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.



173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/pwn/ffi/rtl_sdr.rb', line 173

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