Module: PWN::AI::MCP::ComboNation

Defined in:
lib/pwn/ai/mcp/combo_nation.rb

Overview

JSON-RPC 2.0 / MCP 2024-11-05 stdio client for combo.nation. Speaks the native menu-oriented tools over newline-delimited JSON; it does not scrape a terminal or enable hardware unless requested.

Constant Summary collapse

PROTOCOL_VERSION = 
'2024-11-05'
DEFAULT_COMMAND = 
'/opt/combo.nation/combo.nation'
TOOLS = 
%w[
  menu_catalog menu_main menu_dial_calibration menu_guess
  hardware_connect operation_status operation_answer operation_cancel servo_pause
].freeze

Class Method Summary collapse

Class Method Details

.argv(opts = {}) ⇒ Object

Build the combo.nation MCP argv. Hardware remains off unless requested.

Raises:

  • (ArgumentError) 


27
28
29
30
31
32
33
34
35
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 27

public_class_method def self.argv(opts = {})
  command = (opts[:command] || DEFAULT_COMMAND).to_s
  raise ArgumentError, 'command is required' if command.empty?

  argv = [command, '--mcp']
  Array(opts[:args]).each { |arg| argv << arg.to_s }
  argv << '--mcp-allow-hardware' if opts[:allow_hardware]
  argv
end

.authors ⇒ Object

Author(s)

0day Inc. [email protected]



275
276
277
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 275

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

.call_tool(opts = {}) ⇒ Object

MCP tools/call with JSON text payloads parsed when possible.

Raises:

  • (ArgumentError) 


173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 173

public_class_method def self.call_tool(opts = {})
  name = opts[:name].to_s
  raise ArgumentError, 'name is required' if name.empty?

  arguments = opts[:arguments]
  arguments = {} if arguments.nil?
  raise ArgumentError, 'arguments must be a Hash' unless arguments.is_a?(Hash)

  result = request(
    session: opts[:session],
    method: 'tools/call',
    params: { 'name' => name, 'arguments' => stringify(value: arguments) }
  )
  parse_tool_result(result: result)
end

.connect(opts = {}) ⇒ Object

Spawn or attach to combo.nation --mcp and complete initialize.



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
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 39

public_class_method def self.connect(opts = {})
  timeout = (opts[:timeout] || 5).to_f
  allow_hardware = opts[:allow_hardware] ? true : false
  reader = opts[:reader]
  writer = opts[:writer]
  stdin = stdout = stderr = wait_thr = nil
  unless reader && writer
    command = argv(opts)
    begin
      stdin, stdout, stderr, wait_thr = Open3.popen3(*command)
    rescue Errno::ENOENT
      raise IOError, "combo.nation MCP executable not found: #{command.first}. Build combo.nation or pass command: to #{self}.connect."
    end
    reader = stdout
    writer = stdin
  end
  session = {
    reader: reader,
    writer: writer,
    stdin: stdin,
    stdout: stdout,
    stderr: stderr,
    wait_thr: wait_thr,
    mutex: Mutex.new,
    next_id: 1,
    timeout: timeout,
    allow_hardware: allow_hardware,
    stderr_buf: +''
  }
  drain_stderr(session: session)
  begin
    result = request(
      session: session,
      method: 'initialize',
      params: {
        'protocolVersion' => PROTOCOL_VERSION,
        'capabilities' => {},
        'clientInfo' => { 'name' => 'pwn', 'version' => PWN::VERSION.to_s }
      }
    )
    notify(session: session, method: 'notifications/initialized')
    session[:protocol_version] = result['protocolVersion']
    session[:server_info] = result['serverInfo']
    session
  rescue StandardError
    disconnect(session: session)
    raise
  end
end

.disconnect(opts = {}) ⇒ Object

Close the MCP stdio session and reap a spawned combo.nation process.



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
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 91

public_class_method def self.disconnect(opts = {})
  session = opts[:session]
  return unless session.is_a?(Hash)

  close_io(io: session[:writer])
  if session[:wait_thr]
    begin
      Timeout.timeout(session[:timeout] || 5) { session[:wait_thr].value }
    rescue Timeout::Error
      begin
        Process.kill('TERM', session[:wait_thr].pid)
      rescue StandardError
        nil
      end
      begin
        session[:wait_thr].value
      rescue StandardError
        nil
      end
    end
  end
  close_io(io: session[:reader])
  close_io(io: session[:stderr])
  session[:stderr_thread]&.join(1)
  session
end

.hardware_connect(opts = {}) ⇒ Object

Start hardware initialization; requires --mcp-allow-hardware on the server.



215
216
217
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 215

public_class_method def self.hardware_connect(opts = {})
  call_tool(session: opts[:session], name: 'hardware_connect')
end

.help ⇒ Object 



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
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 279

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

    # Build combo.nation MCP argv without launching a process.
    #{self}.argv(
      command: 'optional - combo.nation executable path (default #{DEFAULT_COMMAND})',
      args: 'optional - extra argv after --mcp',
      allow_hardware: 'optional - append --mcp-allow-hardware (default false)'
    )

    # Spawn or attach to combo.nation --mcp and complete MCP initialize.
    #{self}.connect(
      command: 'optional - combo.nation executable path (default #{DEFAULT_COMMAND})',
      args: 'optional - extra argv after --mcp',
      allow_hardware: 'optional - pass --mcp-allow-hardware (default false; never implied)',
      timeout: 'optional - seconds for handshake and subsequent reads (default 5)',
      reader: 'optional - IO providing server stdout for tests',
      writer: 'optional - IO consuming server stdin for tests'
    )

    # Close pipes and reap a spawned combo.nation process.
    #{self}.disconnect(
      session: 'required - Hash returned by connect'
    )

    # Send a JSON-RPC 2.0 request and return its result object.
    #{self}.request(
      session: 'required - Hash returned by connect',
      method: 'required - JSON-RPC method name such as ping or tools/list',
      params: 'optional - Hash or Array params object',
      timeout: 'optional - seconds to wait for this response'
    )

    # Send a JSON-RPC notification that expects no response.
    #{self}.notify(
      session: 'required - Hash returned by connect',
      method: 'required - notification method such as notifications/initialized',
      params: 'optional - Hash params object'
    )

    # Send an MCP ping request.
    #{self}.ping(
      session: 'required - Hash returned by connect'
    )

    # List tools advertised by the combo.nation MCP server.
    #{self}.list_tools(
      session: 'required - Hash returned by connect'
    )

    # MCP tools/call; JSON text content is parsed when valid.
    #{self}.call_tool(
      session: 'required - Hash returned by connect',
      name: 'required - tool name advertised by tools/list',
      arguments: 'optional - Hash of tool arguments (string menu IDs stay strings)'
    )

    # List actual application menus and exact string option IDs.
    #{self}.menu_catalog(
      session: 'required - Hash returned by connect'
    )

    # Select a main-menu option by exact string ID.
    #{self}.menu_main(
      session: 'required - Hash returned by connect',
      option: 'required - exact menu ID string such as 0.3'
    )

    # Select a dial-calibration option by exact string ID.
    #{self}.menu_dial_calibration(
      session: 'required - Hash returned by connect',
      option: 'required - exact menu ID string such as 3.99'
    )

    # Select a guess-menu option by exact string ID.
    #{self}.menu_guess(
      session: 'required - Hash returned by connect',
      option: 'required - exact menu ID string such as 5.10'
    )

    # Start hardware initialization; the server still requires --mcp-allow-hardware.
    #{self}.hardware_connect(
      session: 'required - Hash returned by connect'
    )

    # Read the current operation, pending prompt, updates, and hardware flags.
    #{self}.operation_status(
      session: 'required - Hash returned by connect'
    )

    # Answer the pending prompt by id.
    #{self}.operation_answer(
      session: 'required - Hash returned by connect',
      prompt_id: 'required - integer prompt id from operation_status',
      value: 'required - number, exact option string, rotation plan, or true for ack'
    )

    # Request cooperative cancellation of the active operation.
    #{self}.operation_cancel(
      session: 'required - Hash returned by connect'
    )

    # Pause or resume the current servo move.
    #{self}.servo_pause(
      session: 'required - Hash returned by connect',
      action: 'required - pause or resume'
    )

    # Poll operation_status until a requested state; the first reply is not completion.
    #{self}.poll_until(
      session: 'required - Hash returned by connect',
      states: 'optional - Array of state strings (default completed failed cancelled)',
      interval: 'optional - seconds between polls (default 1)',
      timeout: 'optional - seconds before Timeout::Error (default 30)'
    )

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

.list_tools(opts = {}) ⇒ Object

MCP tools/list.



166
167
168
169
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 166

public_class_method def self.list_tools(opts = {})
  result = request(session: opts[:session], method: 'tools/list')
  Array(result && result['tools'])
end

.menu_catalog(opts = {}) ⇒ Object

List actual main, dial calibration, and guess menus with string IDs.



191
192
193
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 191

public_class_method def self.menu_catalog(opts = {})
  call_tool(session: opts[:session], name: 'menu_catalog')
end

.menu_dial_calibration(opts = {}) ⇒ Object

Select a dial-calibration option by exact string ID.



203
204
205
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 203

public_class_method def self.menu_dial_calibration(opts = {})
  menu_call(session: opts[:session], name: 'menu_dial_calibration', option: opts[:option])
end

.menu_guess(opts = {}) ⇒ Object

Select a guess-menu option by exact string ID such as 5.10.



209
210
211
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 209

public_class_method def self.menu_guess(opts = {})
  menu_call(session: opts[:session], name: 'menu_guess', option: opts[:option])
end

.menu_main(opts = {}) ⇒ Object

Select a main-menu option by exact string ID.



197
198
199
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 197

public_class_method def self.menu_main(opts = {})
  menu_call(session: opts[:session], name: 'menu_main', option: opts[:option])
end

.notify(opts = {}) ⇒ Object

Send a JSON-RPC notification with no response.

Raises:

  • (ArgumentError) 


147
148
149
150
151
152
153
154
155
156
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 147

public_class_method def self.notify(opts = {})
  session = session!(opts)
  method = opts[:method].to_s
  raise ArgumentError, 'method is required' if method.empty?

  message = { 'jsonrpc' => '2.0', 'method' => method }
  message['params'] = opts[:params] unless opts[:params].nil?
  session[:mutex].synchronize { write_line(session: session, message: message) }
  method
end

.operation_answer(opts = {}) ⇒ Object

Answer the pending prompt by id.

Raises:

  • (ArgumentError) 


227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 227

public_class_method def self.operation_answer(opts = {})
  prompt_id = opts[:prompt_id]
  raise ArgumentError, 'prompt_id is required' if prompt_id.nil?
  raise ArgumentError, 'value is required' unless opts.key?(:value)

  call_tool(
    session: opts[:session],
    name: 'operation_answer',
    arguments: { 'prompt_id' => prompt_id, 'value' => opts[:value] }
  )
end

.operation_cancel(opts = {}) ⇒ Object

Request cooperative cancellation of the active operation.



241
242
243
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 241

public_class_method def self.operation_cancel(opts = {})
  call_tool(session: opts[:session], name: 'operation_cancel')
end

.operation_status(opts = {}) ⇒ Object

Read the current operation, pending prompt, and hardware flags.



221
222
223
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 221

public_class_method def self.operation_status(opts = {})
  call_tool(session: opts[:session], name: 'operation_status')
end

.ping(opts = {}) ⇒ Object

MCP ping.



160
161
162
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 160

public_class_method def self.ping(opts = {})
  request(session: opts[:session], method: 'ping')
end

.poll_until(opts = {}) ⇒ Object

Poll operation_status until a terminal or requested state.



256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 256

public_class_method def self.poll_until(opts = {})
  session = opts[:session]
  wanted = Array(opts[:states] || %w[completed failed cancelled]).map(&:to_s)
  interval = (opts[:interval] || 1).to_f
  timeout = (opts[:timeout] || 30).to_f
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
  loop do
    status = operation_status(session: session)
    state = status.dig(:parsed, 'state').to_s
    return status if wanted.include?(state)

    raise Timeout::Error, "combo.nation operation still #{state.inspect}" if Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline

    sleep interval if interval.positive?
  end
end

.request(opts = {}) ⇒ Object

Send a JSON-RPC request and return its result object.

Raises:

  • (ArgumentError) 


120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 120

public_class_method def self.request(opts = {})
  session = session!(opts)
  method = opts[:method].to_s
  raise ArgumentError, 'method is required' if method.empty?

  params = opts[:params]
  timeout = (opts[:timeout] || session[:timeout] || 5).to_f
  session[:mutex].synchronize do
    ident = session[:next_id]
    session[:next_id] += 1
    message = { 'jsonrpc' => '2.0', 'id' => ident, 'method' => method }
    message['params'] = params unless params.nil?
    write_line(session: session, message: message)
    response = read_json(session: session, timeout: timeout)
    raise IOError, 'combo.nation MCP closed stdout' if response.nil?
    raise IOError, 'combo.nation MCP response id mismatch' unless response['id'] == ident

    if response['error']
      error = response['error']
      raise IOError, "combo.nation MCP error #{error['code']}: #{error['message']}"
    end
    response['result']
  end
end

.required_bins ⇒ Object 



21
22
23
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 21

public_class_method def self.required_bins
  %w[combo.nation]
end

.servo_pause(opts = {}) ⇒ Object

Pause or resume the current servo move.

Raises:

  • (ArgumentError) 


247
248
249
250
251
252
 
# File 'lib/pwn/ai/mcp/combo_nation.rb', line 247

public_class_method def self.servo_pause(opts = {})
  action = opts[:action].to_s
  raise ArgumentError, 'action is required' unless %w[pause resume].include?(action)

  call_tool(session: opts[:session], name: 'servo_pause', arguments: { 'action' => action })
end