Module: PWN::Plugins::REPL::Mesh

Defined in:
lib/pwn/plugins/repl/mesh.rb

Overview

pwn-mesh REPL mode.

Constant Summary collapse

PWN_MESH_TRANSPORTS = 
i[serial bluetooth tcp mqtt].freeze
EMPTY_DATA_DECODER =

meshtastic 0.0.184 Data#to_h omits empty payload bytes, and an unknown PortNum stays an integer (51 is not in the enum). That is not an undecodable application payload. Bytes on an unsupported port still warn.

Module.new do
  define_method(:decode_payload) do |opts = {}|
    payload = opts[:payload]
    return nil if payload.nil? || payload == ''

    super(opts)
  end
end

Class Method Summary collapse

Class Method Details

.add_commands ⇒ Object 



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/pwn/plugins/repl/mesh.rb', line 17

def self.add_commands
  Pry::Commands.create_command 'pwn-mesh' do
    description 'Communicate with Meshtastic network within pwn REPL.'

    def process
      pi = pry_instance
      # meshtastic is a *setup-managed* gem (see pwn.gemspec / PWN::Setup):
      # its rubygems.org releases carry `required_ruby_version >= 4.0`, so
      # it cannot be a hard runtime dependency while pwn supports ruby 3.3+.
      begin
        require 'meshtastic'
      rescue LoadError => e
        output.puts "pwn-mesh unavailable: #{e.message}"
        output.puts "  meshtastic requires ruby >= 4.0 (running #{RUBY_VERSION})." if Gem::Version.new(RUBY_VERSION) < Gem::Version.new('4.0.0')
        output.puts '  Run: `pwn setup --profile full` (or `gem install meshtastic`) on ruby >= 4.0.'
        return
      end

      pi.config.pwn_mesh = true
      pi.config.pwn_ai = false
      pi.config.pwn_asm = false
      meshtastic_env = PWN::Env[:plugins][:meshtastic] || {}

      PWN.send(:remove_const, :MeshTxEchoThread) if PWN.const_defined?(:MeshTxEchoThread)
      PWN.send(:remove_const, :MqttObj) if PWN.const_defined?(:MqttObj)
      PWN.send(:remove_const, :MeshObj) if PWN.const_defined?(:MeshObj)
      PWN.send(:remove_const, :MeshRxHeaderWin) if PWN.const_defined?(:MeshRxHeaderWin)
      PWN.send(:remove_const, :MeshRxBodyWin) if PWN.const_defined?(:MeshRxBodyWin)
      PWN.send(:remove_const, :MeshTxWin) if PWN.const_defined?(:MeshTxWin)
      PWN.send(:remove_const, :MeshMutex) if PWN.const_defined?(:MeshMutex)
      PWN.send(:remove_const, :MqttSubThread) if PWN.const_defined?(:MqttSubThread)
      PWN.send(:remove_const, :MeshSubThread) if PWN.const_defined?(:MeshSubThread)
      PWN.send(:remove_const, :MeshTransport) if PWN.const_defined?(:MeshTransport)
      PWN.send(:remove_const, :MeshTxPrompt) if PWN.const_defined?(:MeshTxPrompt)
      PWN.send(:remove_const, :MeshTxEpoch) if PWN.const_defined?(:MeshTxEpoch)
      PWN.send(:remove_const, :MeshTxBlank) if PWN.const_defined?(:MeshTxBlank)
      PWN.send(:remove_const, :MeshLastSubmit) if PWN.const_defined?(:MeshLastSubmit)
      PWN.send(:remove_const, :MeshRxState) if PWN.const_defined?(:MeshRxState)

      begin
        mesh_obj = PWN::Plugins::REPL.send(:mesh_connect, env: meshtastic_env)
      rescue StandardError => e
        output.puts "pwn-mesh unavailable: #{e.class}: #{e.message}"
        pi.config.pwn_mesh = false
        return
      end
      PWN.const_set(:MeshObj, mesh_obj)
      PWN.const_set(:MqttObj, mesh_obj)

      active_channel = meshtastic_env[:channel][:active].to_s.to_sym
      channel_env = meshtastic_env[:channel][active_channel]
      psk = channel_env[:psk]
      region = channel_env[:region]
      topic = channel_env[:topic]
      channel_num = channel_env[:channel_num]
      link = PWN::Plugins::REPL.send(:mesh_link_label, env: meshtastic_env)

      # Init ncurses UI (idempotent) with separate RX (top) and TX (bottom) panes
      Curses.init_screen
      Curses.curs_set(0)
      Curses.noecho
      Curses.cbreak
      Curses.crmode
      Curses.ESCDELAY = 0
      Curses.start_color
      Curses.use_default_colors
      Curses.init_pair(20, Curses::COLOR_CYAN, -1)
      Curses.init_pair(21, Curses::COLOR_YELLOW, -1)
      Curses.init_pair(22, Curses::COLOR_BLACK, Curses::COLOR_CYAN)
      Curses.init_pair(23, Curses::COLOR_GREEN, -1)
      Curses.init_pair(24, Curses::COLOR_WHITE, -1)

      PWN.const_set(:MeshColors, [20, 23, 21])
      PWN.const_set(:MeshLastColor, 20)

      layout = PWN::Plugins::REPL.send(:mesh_layout, lines: Curses.lines, cols: Curses.cols)
      mesh_tx_rows = layout[:tx]
      mesh_header_rows = layout[:header]
      body_height = layout[:body]
      rx_header_win = Curses::Window.new(mesh_header_rows, Curses.cols, 0, 0)
      rx_header_win.scrollok(false)
      rx_header_win.nodelay = true
      PWN::Plugins::REPL.send(:mesh_box!, win: rx_header_win)
      rx_header_win.attron(Curses.color_pair(20) | Curses::A_BOLD)
      rx_header = " pwn.mesh  #{link}  #{region}/#{topic}  ch #{channel_num} "
      inner_cols = [Curses.cols - 2, 1].max
      rx_header_win.setpos(1, 1)
      rx_header_win.addstr(rx_header[0, inner_cols].to_s.ljust(inner_cols))
      rx_header_win.attroff(Curses.color_pair(20) | Curses::A_BOLD)
      rx_header_win.refresh
      PWN.const_set(:MeshRxHeaderWin, rx_header_win)

      body_start_row = mesh_header_rows
      frame = Curses::Window.new(body_height, Curses.cols, body_start_row, 0)
      PWN::Plugins::REPL.send(:mesh_box!, win: frame)
      frame.setpos(0, 2)
      frame.addstr(' CONVERSATION ')
      frame.refresh
      PWN.const_set(:MeshRxFrameWin, frame)
      rx_body_win = Curses::Window.new(
        [body_height - 2, 1].max,
        [Curses.cols - 4, 1].max,
        body_start_row + (body_height > 2 ? 1 : 0),
        Curses.cols > 4 ? 2 : 0
      )
      rx_body_win.scrollok(true)
      rx_body_win.nodelay = true
      rx_body_win.refresh
      PWN.const_set(:MeshRxBodyWin, rx_body_win)

      tx_win = Curses::Window.new(mesh_tx_rows, Curses.cols, layout[:tx_top], 0)
      tx_win.scrollok(false)
      tx_win.nodelay = true
      PWN::Plugins::REPL.send(:mesh_box!, win: tx_win)
      tx_win.refresh

      PWN.const_set(:MeshTxWin, tx_win)
      PWN.const_set(:MeshMutex, Mutex.new)

      # Curses owns input and rendering until this command returns to Pry.
      PWN.const_set(:MeshEvents, Queue.new)
      PWN::Plugins::REPL.send(:mesh_capture_output, run: proc {
        PWN::Plugins::REPL.send(:mesh_start_rx!, env: meshtastic_env, obj: mesh_obj)
        PWN::Plugins::REPL.send(:mesh_refresh_ui!, env: meshtastic_env)
        PWN::Plugins::REPL.send(:mesh_ui_puts, text: 'Ready. /menu opens settings. Incoming channel messages appear here.')
        PWN::Plugins::REPL.send(:mesh_console_loop, pry: pi)
      })
    rescue StandardError => e
      raise e
    ensure
      PWN::Plugins::REPL.leave_special_mode!(pry: pi) if pi&.config&.pwn_mesh
    end
  end
end

.authors ⇒ Object

Author(s)

0day Inc. [email protected]



2246
2247
2248
2249
2250
 
# File 'lib/pwn/plugins/repl/mesh.rb', line 2246

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

.help ⇒ Object

Display usage for this module.



2254
2255
2256
2257
2258
2259
2260
2261
2262
2263
2264
2265
2266
2267
2268
2269
2270
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294
2295
2296
2297
2298
2299
2300
2301
2302
2303
2304
 
# File 'lib/pwn/plugins/repl/mesh.rb', line 2254

public_class_method def self.help
  puts "USAGE:
    # Register the pwn-mesh Pry command.
    #{self}.add_commands

    # Curses arrow-key picker for pwn-mesh channel, transport, and device lists.
    #{self}.mesh_menu_pick(
      title: 'optional - window title drawn on the boxed curses menu',
      items: 'required - Array of selectable strings such as channel names',
      current: 'optional - currently selected item to highlight',
      getch: 'optional - proc returning the next key so specs can drive the menu without a TTY'
    )

    # Clear the pwn-mesh TX buffer after a slash command or sent mesh line.
    #{self}.mesh_reset_input!(
      pry: 'optional - Pry instance whose Reline/line_buffer should be emptied',
      submitted: 'optional - the line that was just accepted so the TX pane can hide it'
    )

    # Curses overlay rows for the pwn-mesh slash menu while typing a leading slash.
    #{self}.pwn_mesh_menu_rows(
      line: 'optional - current TX buffer; leading slash lists matching mesh commands'
    )

    # TAB hits for pwn-mesh slash menus (commands, named channels, transports, devices).
    #{self}.pwn_mesh_complete(
      target: 'required - token Reline is completing in pwn-mesh',
      line: 'optional - full line buffer so /channel P<TAB> can list named channels'
    )

    # Install Reline dropdown for pwn-mesh slash commands.
    #{self}.install_pwn_mesh_completer!(
      pry: 'optional - Pry instance stored for mesh TAB completion'
    )

    # Run a leading-slash pwn-mesh command locally instead of sending it as mesh text.
    #{self}.pwn_mesh_dispatch_slash!(
      request: 'optional - full line such as /channel list or /transport serial',
      pry: 'optional - Pry instance used by /back to leave pwn-mesh'
    )

    # Write plugins.meshtastic from the live Env into the encrypted pwn.yaml vault.
    #{self}.persist_mesh_env(
      mesh: 'optional - meshtastic Hash to persist (defaults to PWN::Env plugins.meshtastic)'
    )

    # Print the AUTHOR(S) string for this module.
    #{self}.authors
  "
  constants.sort
end