Module: PWN::AI::Agent::Confirmation

Defined in:
lib/pwn/ai/agent/confirmation.rb

Overview

Per-engagement ACK for exploit/destructive tool calls (PWN-AI-005 tiers).

Constant Summary collapse

TIERS = 
%w[read_only active_scan exploit destructive].freeze
DEFAULTS = 
{ 'read_only' => 'auto', 'active_scan' => 'auto', 'exploit' => 'prompt', 'destructive' => 'prompt' }.freeze
AUTONOMOUS = 
{ 'read_only' => 'auto', 'active_scan' => 'auto', 'exploit' => 'auto', 'destructive' => 'auto' }.freeze

Class Method Summary collapse

Class Method Details

.authors ⇒ Object 



68
69
70
 
# File 'lib/pwn/ai/agent/confirmation.rb', line 68

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

.gate(opts = {}) ⇒ Object

Return an ACK pause hash, or nil when the call may proceed.



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
 
# File 'lib/pwn/ai/agent/confirmation.rb', line 38

public_class_method def self.gate(opts = {})
  name = opts[:name].to_s
  args = opts[:args]
  args = {} unless args.is_a?(Hash)
  klass = tier(name: name, args: args)
  policy = load_policy(opts)
  action = (policy[klass] || DEFAULTS[klass] || 'auto').to_s
  return { success: false, error: 'unattended mission refuses exploit and destructive calls without scope.yaml confirmation', code: 'ACK_DENY', tier: klass } if opts[:unattended] && policy.equal?(AUTONOMOUS) && %w[exploit destructive].include?(klass)
  return { success: false, error: "confirmation deny for #{klass}", code: 'ACK_DENY', tier: klass } if action == 'deny'
  return nil unless action == 'prompt'

  eng = engagement_id(opts.merge(args: args))
  return scoped_ack(opts.merge(name: name, args: args, engagement_id: eng, tier: klass)) if opts[:ack_scope]
  return nil if acked?(engagement_id: eng)

  if truthy?(value: opts[:operator_ack] || args[:operator_ack] || args['operator_ack'])
    persist_ack(engagement_id: eng)
    return nil
  end
  {
    success: false,
    needs_ack: true,
    code: 'ACK_REQUIRED',
    tier: klass,
    engagement_id: eng,
    diff: intended_diff(name: name, args: args, tier: klass),
    error: "operator ACK required for #{klass}-tier tool call"
  }
end

.help ⇒ Object 



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
 
# File 'lib/pwn/ai/agent/confirmation.rb', line 72

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

    # Classify a tool call into a PWN-AI-005 side-effect tier.
    #{self}.tier(
      name: 'required - registry tool name',
      args: 'optional - Hash of tool arguments'
    )

    # Return an ACK pause hash, or nil when the call may proceed.
    #{self}.gate(
      name: 'required - registry tool name',
      args: 'optional - Hash of tool arguments',
      engagement_id: 'optional - engagement id used to cache the ACK',
      operator_ack: 'optional - true records a one-time ACK for this engagement',
      scope_path: 'optional - scope.yaml path; defaults to ~/.pwn/scope.yaml',
      unattended: 'optional - true denies exploit and destructive when confirmation is absent',
      ack_scope: 'optional - true scopes the ACK to tool, host, and arguments',
      ack_ttl: 'optional - seconds before a scoped ACK expires'
    )

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

.required_bins ⇒ Object 



18
19
20
 
# File 'lib/pwn/ai/agent/confirmation.rb', line 18

public_class_method def self.required_bins
  []
end

.tier(opts = {}) ⇒ Object

Classify a tool call into a PWN-AI-005 side-effect tier.



23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/pwn/ai/agent/confirmation.rb', line 23

public_class_method def self.tier(opts = {})
  name = opts[:name].to_s
  args = opts[:args]
  args = {} unless args.is_a?(Hash)
  tagged = (args[:side_effect] || args['side_effect']).to_s
  return tagged if TIERS.include?(tagged)
  return 'exploit' if name.match?(/exploit|gdb_run_to_crash|fuzz_campaign/)
  return 'active_scan' if name.match?(/nmap_scan|nuclei_scan|sbom_scan|http_proxy/)
  return payload_tier(payload: args[:command] || args['command']) if name == 'shell'
  return payload_tier(payload: args[:code] || args['code']) if name == 'pwn_eval'

  'read_only'
end