Module: PWN::Plugins::Radare2

Defined in:
lib/pwn/plugins/radare2.rb

Overview

Persistent r2pipe-style session: open/cmd/cmdj/close plus helpers.

Class Method Summary collapse

Class Method Details

.analyze_all(opts = {}) ⇒ Object

Normalized read-only entrypoint. Legacy session APIs retain their return shapes.



170
171
172
173
174
175
176
177
178
179
180
181
182
 
# File 'lib/pwn/plugins/radare2.rb', line 170

public_class_method def self.analyze_all(opts = {})
  path = File.realpath(File.expand_path(opts[:path].to_s))
  return BinaryAnalysis.analyze(opts.merge(path: path)) if opts[:backend] == 'binutils' || !BinaryAnalysis.available?(name: 'r2')

  data = {}
  { functions: 'aflj', strings: 'izzj', imports: 'iij' }.each do |key, command|
    raw = BinaryAnalysis.run(argv: ['r2', '-2', '-NN', '-q', '-e', 'scr.color=0', '-c', "aaa;#{command}", path], timeout: opts.fetch(:timeout, 60))
    data[key] = JSON.parse(raw)
  end
  data.merge(backend: 'radare2', status: 'ok', risk_level: 'low', path: path, sha256: Digest::SHA256.file(path).hexdigest, warnings: [])
rescue StandardError => e
  BinaryAnalysis.analyze(opts).tap { |result| result[:warnings] << "radare2: #{e.message}" }
end

.authors ⇒ Object 



221
222
223
 
# File 'lib/pwn/plugins/radare2.rb', line 221

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

.binary_info(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Plugins::Radare2.binary_info( session: 'required - session id returned by #open' )



153
154
155
 
# File 'lib/pwn/plugins/radare2.rb', line 153

public_class_method def self.binary_info(opts = {})
  cmdj(opts.merge(cmd: 'iIj'))
end

.close(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Plugins::Radare2.close( session: 'required - session id returned by #open' )



74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/pwn/plugins/radare2.rb', line 74

public_class_method def self.close(opts = {})
  sess = session!(opts)
  sess[:stdin].write("q\n")
  sess[:stdin].close
  sess[:stdout].close
  @sessions.delete(opts[:session].to_s)
  true
rescue StandardError
  @sessions.delete(opts[:session].to_s)
  false
end

.cmd(opts = {}) ⇒ Object

Supported Method Parameters

text = PWN::Plugins::Radare2.cmd( session: 'required - session id returned by #open', cmd: 'required - r2 command to run; returns raw text (e.g. pdf @ main)' )



48
49
50
51
52
53
54
55
56
 
# File 'lib/pwn/plugins/radare2.rb', line 48

public_class_method def self.cmd(opts = {})
  sess = session!(opts)
  line = opts[:cmd].to_s
  raise 'ERROR: cmd is required' if line.empty?

  sess[:stdin].write("#{line}\n")
  sess[:stdin].flush
  read_until_null(io: sess[:stdout])
end

.cmdj(opts = {}) ⇒ Object

Supported Method Parameters

json = PWN::Plugins::Radare2.cmdj( session: 'required - session id returned by #open', cmd: 'required - r2 command; trailing j is added if missing and stdout is JSON.parse (e.g. afl or aflj)' )



63
64
65
66
67
68
 
# File 'lib/pwn/plugins/radare2.rb', line 63

public_class_method def self.cmdj(opts = {})
  raw = cmd(opts.merge(cmd: opts[:cmd].to_s.sub(/j?\z/, 'j')))
  JSON.parse(raw)
rescue JSON::ParserError
  raw
end

.decompile(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Plugins::Radare2.decompile( session: 'required - session id returned by #open', addr: 'required - address or flag to decompile (needs r2ghidra)' )



162
163
164
165
166
167
 
# File 'lib/pwn/plugins/radare2.rb', line 162

public_class_method def self.decompile(opts = {})
  addr = opts[:addr].to_s
  cmd(opts.merge(cmd: "pdg @ #{addr}"))
rescue StandardError => e
  { error: "#{e.class}: #{e.message}", hint: 'r2ghidra plugin may be absent' }
end

.disasm(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Plugins::Radare2.disasm( session: 'required - session id returned by #open', addr: 'required - address or flag to disassemble from', n: 'optional - instruction count (defaults to 32)' )



115
116
117
118
119
 
# File 'lib/pwn/plugins/radare2.rb', line 115

public_class_method def self.disasm(opts = {})
  addr = opts[:addr].to_s
  n = (opts[:n] || opts[:len] || 32).to_i
  cmd(opts.merge(cmd: "pd #{n} @ #{addr}"))
end

.disasm_function(opts = {}) ⇒ Object 



188
189
190
 
# File 'lib/pwn/plugins/radare2.rb', line 188

public_class_method def self.disasm_function(opts = {})
  normalized_query(opts.merge(command: 'pdfj'))
end

.functions(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Plugins::Radare2.functions( session: 'required - session id returned by #open' )



90
91
92
 
# File 'lib/pwn/plugins/radare2.rb', line 90

public_class_method def self.functions(opts = {})
  cmdj(opts.merge(cmd: 'aflj'))
end

.help ⇒ Object 



225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
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
312
313
314
315
316
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
360
361
362
363
364
365
366
367
368
369
370
371
 
# File 'lib/pwn/plugins/radare2.rb', line 225

public_class_method def self.help
  puts "USAGE:
    # List host binaries this module expects to be installed.
    #{self}.required_bins

    # Open the binary in r2 (-q0) and return a session id.
    #{self}.open(
      path: 'required - filesystem path to the binary r2 should open'
    )

    # Run a command and return raw text output.
    #{self}.cmd(
      session: 'required - session id returned by #open',
      cmd: 'required - r2 command to run; returns raw text (e.g. pdf @ main)'
    )

    # Run a command, request JSON, and parse the result.
    #{self}.cmdj(
      session: 'required - session id returned by #open',
      cmd: 'required - r2 command; trailing j is added if missing and stdout is JSON.parse (e.g. afl or aflj)'
    )

    # Close a session previously returned by #open.
    #{self}.close(
      session: 'required - session id returned by #open'
    )

    # List functions after analysis (aflj JSON).
    #{self}.functions(
      session: 'required - session id returned by #open'
    )

    # List xrefs to an address or flag (axtj JSON).
    #{self}.xrefs_to(
      session: 'required - session id returned by #open',
      addr: 'required - address or flag to list xrefs to (e.g. main or 0x401000)'
    )

    # List xrefs from an address or flag (axfj JSON).
    #{self}.xrefs_from(
      session: 'required - session id returned by #open',
      addr: 'required - address or flag to list xrefs from'
    )

    # Disassemble n instructions at addr (pd text).
    #{self}.disasm(
      session: 'required - session id returned by #open',
      addr: 'required - address or flag to disassemble from',
      n: 'optional - instruction count (defaults to 32)',
      len: 'optional - alias for n, number of instructions to disassemble'
    )

    # List strings in the binary (izj JSON).
    #{self}.strings(
      session: 'required - session id returned by #open'
    )

    # List imported symbols (iij JSON).
    #{self}.imports(
      session: 'required - session id returned by #open'
    )

    # List sections / segments (iSj JSON).
    #{self}.sections(
      session: 'required - session id returned by #open'
    )

    # Print binary header info (iIj JSON).
    #{self}.binary_info(
      session: 'required - session id returned by #open'
    )

    # Decompile a function at addr via r2ghidra (pdg text).
    #{self}.decompile(
      session: 'required - session id returned by #open',
      addr: 'required - address or flag to decompile (needs r2ghidra)'
    )

    # Print the AUTHOR(S) string for this module.
    #{self}.authors
    # Invoke required_bins with the documented options; normalized path APIs report their backend.
    #{self}.required_bins
    # Invoke open with the documented options; normalized path APIs report their backend.
    #{self}.open(
      path: 'optional - filesystem path to the local artifact or binary'
    )
    # Invoke cmd with the documented options; normalized path APIs report their backend.
    #{self}.cmd(
      cmd: 'optional - raw radare2 command for the legacy session interface'
    )
    # Invoke cmdj with the documented options; normalized path APIs report their backend.
    #{self}.cmdj(
      cmd: 'optional - raw radare2 command for the legacy session interface'
    )
    # Invoke close with the documented options; normalized path APIs report their backend.
    #{self}.close(
      session: 'optional - session identifier returned by Radare2.open'
    )
    # Invoke functions with the documented options; normalized path APIs report their backend.
    #{self}.functions
    # Invoke xrefs_to with the documented options; normalized path APIs report their backend.
    #{self}.xrefs_to(
      addr: 'optional - hexadecimal address or binary symbol name'
    )
    # Invoke xrefs_from with the documented options; normalized path APIs report their backend.
    #{self}.xrefs_from(
      addr: 'optional - hexadecimal address or binary symbol name'
    )
    # Invoke disasm with the documented options; normalized path APIs report their backend.
    #{self}.disasm(
      addr: 'optional - hexadecimal address or binary symbol name',
      len: 'optional - alternative maximum disassembly instruction count',
      n: 'optional - maximum disassembly instruction count; defaults to 32'
    )
    # Invoke strings with the documented options; normalized path APIs report their backend.
    #{self}.strings(
      path: 'optional - filesystem path to the local artifact or binary'
    )
    # Invoke imports with the documented options; normalized path APIs report their backend.
    #{self}.imports(
      path: 'optional - filesystem path to the local artifact or binary'
    )
    # Invoke sections with the documented options; normalized path APIs report their backend.
    #{self}.sections
    # Invoke binary_info with the documented options; normalized path APIs report their backend.
    #{self}.binary_info
    # Invoke decompile with the documented options; normalized path APIs report their backend.
    #{self}.decompile(
      addr: 'optional - hexadecimal address or binary symbol name'
    )
    # Invoke analyze_all with the documented options; normalized path APIs report their backend.
    #{self}.analyze_all(
      backend: 'optional - analysis backend name; binutils forces lightweight fallback',
      path: 'optional - filesystem path to the local artifact or binary',
      timeout: 'optional - positive subprocess or HTTP deadline in seconds'
    )
    # Invoke list_functions with the documented options; normalized path APIs report their backend.
    #{self}.list_functions
    # Invoke disasm_function with the documented options; normalized path APIs report their backend.
    #{self}.disasm_function
    # Invoke xrefs with the documented options; normalized path APIs report their backend.
    #{self}.xrefs
    # Invoke authors with the documented options; normalized path APIs report their backend.
    #{self}.authors
  "
  constants.sort
end

.imports(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Plugins::Radare2.imports( session: 'required - session id returned by #open' )



135
136
137
138
139
 
# File 'lib/pwn/plugins/radare2.rb', line 135

public_class_method def self.imports(opts = {})
  return analyze_all(opts).then { |result| result.merge(data: result[:imports]) } if opts[:path]

  cmdj(opts.merge(cmd: 'iij'))
end

.list_functions(opts = {}) ⇒ Object 



184
185
186
 
# File 'lib/pwn/plugins/radare2.rb', line 184

public_class_method def self.list_functions(opts = {})
  analyze_all(opts).then { |result| result.merge(data: result[:functions]) }
end

.open(opts = {}) ⇒ Object

Supported Method Parameters

session = PWN::Plugins::Radare2.open( path: 'required - filesystem path to the binary r2 should open' )



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/pwn/plugins/radare2.rb', line 23

public_class_method def self.open(opts = {})
  PWN::Plugins::PreflightChecker.require_bin!(name: 'r2')
  path = opts[:path].to_s
  raise 'ERROR: path is required' if path.empty?
  raise "ERROR: binary not found: #{path}" unless File.file?(path)

  sha = Digest::SHA256.file(path).hexdigest
  hit = @sessions.find { |_id, sess| sess[:path] == path && sess[:sha256] == sha }
  return hit[0] if hit

  stdin, stdout, waiter = Open3.popen2('r2', '-q0', '-e', 'scr.color=0', path)
  sid = SecureRandom.hex(8)
  pid = waiter.pid if waiter.respond_to?(:pid)
  @sessions[sid] = { stdin: stdin, stdout: stdout, waiter: waiter, path: path, sha256: sha, pid: pid }
  read_until_null(io: stdout)
  cmd(session: sid, cmd: 'aaa')
  @sessions[sid][:aaa] = true
  sid
end

.required_bins ⇒ Object 



15
16
17
 
# File 'lib/pwn/plugins/radare2.rb', line 15

public_class_method def self.required_bins
  %w[r2]
end

.sections(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Plugins::Radare2.sections( session: 'required - session id returned by #open' )



145
146
147
 
# File 'lib/pwn/plugins/radare2.rb', line 145

public_class_method def self.sections(opts = {})
  cmdj(opts.merge(cmd: 'iSj'))
end

.strings(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Plugins::Radare2.strings( session: 'required - session id returned by #open' )



125
126
127
128
129
 
# File 'lib/pwn/plugins/radare2.rb', line 125

public_class_method def self.strings(opts = {})
  return analyze_all(opts).then { |result| result.merge(data: result[:strings]) } if opts[:path]

  cmdj(opts.merge(cmd: 'izj'))
end

.xrefs(opts = {}) ⇒ Object 



192
193
194
 
# File 'lib/pwn/plugins/radare2.rb', line 192

public_class_method def self.xrefs(opts = {})
  normalized_query(opts.merge(command: 'axtj'))
end

.xrefs_from(opts = {}) ⇒ Object 



104
105
106
107
 
# File 'lib/pwn/plugins/radare2.rb', line 104

public_class_method def self.xrefs_from(opts = {})
  addr = opts[:addr].to_s
  cmdj(opts.merge(cmd: "axfj #{addr}"))
end

.xrefs_to(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Plugins::Radare2.xrefs_to( session: 'required - session id returned by #open', addr: 'required - address or flag to list xrefs to (e.g. main or 0x401000)' )



99
100
101
102
 
# File 'lib/pwn/plugins/radare2.rb', line 99

public_class_method def self.xrefs_to(opts = {})
  addr = opts[:addr].to_s
  cmdj(opts.merge(cmd: "axtj #{addr}"))
end