Module: PWN::AI::Agent::Engagement

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

Overview

First-class engagement scope, RoE, and findings pointer.

Constant Summary collapse

ROOT = 
File.join(Dir.home, '.pwn', 'engagements')
ACTIVE_FILE = 
File.join(ROOT, 'active')

Class Method Summary collapse

Class Method Details

.authors ⇒ Object 



159
160
161
 
# File 'lib/pwn/ai/agent/engagement.rb', line 159

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

.close(opts = {}) ⇒ Object 



35
36
37
38
 
# File 'lib/pwn/ai/agent/engagement.rb', line 35

public_class_method def self.close(opts = {})
  FileUtils.rm_f(ACTIVE_FILE) if opts.is_a?(Hash)
  { active: nil }
end

.current_name(opts = {}) ⇒ Object 



50
51
52
53
54
55
 
# File 'lib/pwn/ai/agent/engagement.rb', line 50

public_class_method def self.current_name(opts = {})
  return nil unless opts.is_a?(Hash)
  return nil unless File.file?(ACTIVE_FILE)

  File.read(ACTIVE_FILE).strip
end

.deny_if_out_of_scope(opts = {}) ⇒ Object 



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/pwn/ai/agent/engagement.rb', line 116

public_class_method def self.deny_if_out_of_scope(opts = {})
  args = opts[:args] || opts[:command] || opts[:text]
  blob = args.is_a?(Hash) ? args.inspect : args.to_s
  roe = load_roe
  tech_hit = Array(roe[:techniques_deny]).find { |t| blob.downcase.include?(t.to_s.downcase) }
  if tech_hit
    return {
      success: false,
      error: "roe_deny: technique #{tech_hit}",
      code: 'ROE_DENY',
      violating: [tech_hit]
    }
  end
  tokens = blob.scan(/(?:\d{1,3}\.){3}\d{1,3}|[A-Za-z0-9.-]+\.[A-Za-z]{2,}/).uniq
  bad = tokens.reject { |tok| in_scope?(host: tok) }
  return nil if bad.empty?

  {
    success: false,
    error: "out_of_scope: #{bad.first}",
    code: 'SCOPE_DENY',
    violating: bad
  }
end

.help ⇒ Object 



163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
 
# File 'lib/pwn/ai/agent/engagement.rb', line 163

public_class_method def self.help
  puts "USAGE:
    # Open or update an engagement and mark it active.
    #{self}.open(
      name: 'required - engagement name',
      engagement: 'optional - alias for name',
      scope_cidrs: 'optional - Array of CIDR strings',
      scope_domains: 'optional - Array of DNS suffixes',
      excluded: 'optional - Array of excluded hosts',
      roe: 'optional - rules of engagement text',
      window: 'optional - time window string'
    )

    # Clear the active engagement pointer.
    #{self}.close(
      name: 'optional - unused; closing always clears the active pointer'
    )

    # Return the active engagement document.
    #{self}.status(
      name: 'optional - engagement name (defaults to active)'
    )

    # Return the active engagement name or nil.
    #{self}.current_name(
      unused: 'optional - reserved'
    )

    # True when host/ip/url is inside the active engagement scope.
    #{self}.in_scope?(
      host: 'optional - DNS hostname to test against scope_domains',
      ip: 'optional - IPv4 or IPv6 address to test against scope_cidrs',
      target: 'optional - host or IP alias when the caller has one field',
      url: 'optional - URL whose host is extracted and tested'
    )

    # Structured denial when args mention an out-of-scope host.
    #{self}.deny_if_out_of_scope(
      args: 'optional - Hash of tool args',
      command: 'optional - command string',
      text: 'optional - free-form blob to scan'
    )

    # Load the RoE policy file when present for allow and deny lists.
    #{self}.load_roe(
      path: 'optional - filesystem path of roe.yaml (defaults to ~/.pwn/roe.yaml)'
    )

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

.in_scope?(opts = {}) ⇒ Boolean 



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

public_class_method def self.in_scope?(opts = {})
  token = (opts[:host] || opts[:ip] || opts[:target] || opts[:url]).to_s
  return true if token.empty?

  roe = load_roe
  host = token
  begin
    host = URI.parse(token).host || token if token.include?('://')
  rescue StandardError
    host = token
  end
  host = host.sub(%r{\Ahttps?://}i, '').split('/').first.to_s.split(':').first
  if roe[:present]
    deny = Array(roe[:targets_deny])
    return false if deny.any? { |ex| host == ex.to_s || host.end_with?(".#{ex}") || host.include?(ex.to_s) }

    allow = Array(roe[:targets_allow])
    unless allow.empty?
      ok = allow.any? do |a|
        a = a.to_s
        host == a || host.end_with?(".#{a}") || begin
          IPAddr.new(a).include?(IPAddr.new(host))
        rescue StandardError
          false
        end
      end
      return false unless ok
    end
  end

  row = status
  return true if row[:active].nil? || row[:missing]

  enforce = engagement_enforce
  return true if enforce == 'off'

  host = token
  begin
    host = URI.parse(token).host || token if token.include?('://')
  rescue StandardError
    host = token
  end
  host = host.sub(%r{\Ahttps?://}i, '').split('/').first.to_s.split(':').first
  return false if Array(row[:excluded]).any? { |ex| host.include?(ex.to_s) }

  cidrs = Array(row[:scope_cidrs]).map(&:to_s).reject(&:empty?)
  domains = Array(row[:scope_domains]).map(&:to_s).reject(&:empty?)
  return true if cidrs.empty? && domains.empty?

  ip_ok = begin
    addr = IPAddr.new(host)
    cidrs.any? { |c| IPAddr.new(c).include?(addr) }
  rescue StandardError
    false
  end
  dom_ok = domains.any? { |d| host == d || host.end_with?(".#{d}") }
  ip_ok || dom_ok
end

.load_roe(opts = {}) ⇒ Object 



141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/pwn/ai/agent/engagement.rb', line 141

public_class_method def self.load_roe(opts = {})
  path = (opts[:path] || File.join(Dir.home, '.pwn', 'roe.yaml')).to_s
  return { present: false, targets_allow: [], targets_deny: [], techniques_deny: [] } unless File.file?(path)

  require 'yaml'
  raw = YAML.safe_load_file(path, permitted_classes: [], symbolize_names: true) || {}
  raw = {} unless raw.is_a?(Hash)
  {
    present: true,
    targets_allow: Array(raw[:targets_allow] || raw['targets_allow']),
    targets_deny: Array(raw[:targets_deny] || raw['targets_deny']),
    techniques_deny: Array(raw[:techniques_deny] || raw['techniques_deny']),
    time_windows: raw[:time_windows] || raw['time_windows']
  }
rescue StandardError
  { present: false, targets_allow: [], targets_deny: [], techniques_deny: [] }
end

.open(opts = {}) ⇒ Object 



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/pwn/ai/agent/engagement.rb', line 16

public_class_method def self.open(opts = {})
  name = (opts[:name] || opts[:engagement] || 'default').to_s
  raise 'ERROR: name is required' if name.empty?

  FileUtils.mkdir_p(ROOT)
  path = File.join(ROOT, "#{name}.json")
  row = File.file?(path) ? JSON.parse(File.read(path), symbolize_names: true) : {}
  row[:name] = name
  row[:scope_cidrs] = Array(opts[:scope_cidrs] || row[:scope_cidrs])
  row[:scope_domains] = Array(opts[:scope_domains] || row[:scope_domains])
  row[:excluded] = Array(opts[:excluded] || row[:excluded])
  row[:roe] = (opts[:roe] || row[:roe]).to_s
  row[:window] = opts[:window] || row[:window]
  row[:findings] ||= []
  File.write(path, JSON.pretty_generate(row))
  File.write(ACTIVE_FILE, name)
  row.merge(path: path, active: true)
end

.status(opts = {}) ⇒ Object 



40
41
42
43
44
45
46
47
48
 
# File 'lib/pwn/ai/agent/engagement.rb', line 40

public_class_method def self.status(opts = {})
  name = (opts[:name] || current_name).to_s
  return { active: nil } if name.empty?

  path = File.join(ROOT, "#{name}.json")
  return { active: name, missing: true } unless File.file?(path)

  JSON.parse(File.read(path), symbolize_names: true).merge(active: name)
end