Module: PWN::AI::Agent::Extrospection

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

Overview

PWN::AI::Agent::Extrospection is the outward-facing counterpart to PWN::AI::Agent::Learning (introspection). Where Learning/Metrics look INWARD at the agent's own tool telemetry, task outcomes and session transcripts, Extrospection looks OUTWARD at the world the agent operates in: host state, toolchain versions, network posture, repo drift, and external threat-intel (CVE / Exploit-DB / ATT&CK).

Together they close BOTH halves of the pwn-ai feedback loop:

INTROSPECTIVE (self)        EXTROSPECTIVE (world)
----------------------      -------------------------------------
Metrics.record              Extrospection.snapshot   (host probe)
Learning.note_outcome       Extrospection.observe    (recon fact)
Learning.reflect            Extrospection.drift      (env delta)
Learning.stats              Extrospection.intel      (CVE/EDB)
                          Extrospection.correlate  (self x world)

PromptBuilder re-injects Extrospection.to_context on every turn so the model gains situational awareness of what changed on THIS host between sessions ("kernel upgraded", "nmap now missing", "port 8080 newly listening", "CVE-2026-XXXX matches installed openssl") and can correlate that drift against introspective failures.

Everything is file-backed under ~/.pwn/extrospection.json so it survives across REPL restarts and is shared by every future session.

Constant Summary collapse

EXTRO_FILE = 
File.join(Dir.home, '.pwn', 'extrospection.json')
MAX_OBSERVATIONS = 
500
PROBE_BINS = 
%w[nmap curl git ruby python3 gcc msfconsole sqlmap burpsuite zaproxy openssl docker].freeze

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com



589
590
591
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 589

public_class_method def self.authors
  "AUTHOR(S):\n  0day Inc. <support@0dayinc.com>\n"
end

.auto_extrospect(opts = {}) ⇒ Object

Supported Method Parameters

PWN::AI::Agent::Extrospection.auto_extrospect( session_id: 'optional - id of the just-completed session (for tagging)' )

Called by Learning.auto_reflect when PWN::Env[:agent][:auto_extrospect] is truthy. Captures a fresh snapshot and, if drift is non-trivial, records it as an observation and a PWN::Memory :env fact so the NEXT session's system prompt already knows the world moved. Never raises.



364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 364

public_class_method def self.auto_extrospect(opts = {})
  sid = opts[:session_id]
  return unless auto_extrospect_enabled?

  res   = snapshot(persist: true)
  delta = res[:drift]
  moved = Array(delta[:changed]).length + Array(delta[:added]).length + Array(delta[:removed]).length
  return res if moved.zero?

  summary = summarise_drift(delta: delta)
  observe(source: 'auto_extrospect', category: :env, data: summary, tags: ['drift', sid.to_s].reject(&:empty?))
  if defined?(PWN::Memory)
    key = :"extro_drift_#{res[:snapshot][:fingerprint]}"
    PWN::Memory.remember(key: key, value: "ENV DRIFT #{res[:snapshot][:captured_at]}: #{summary[0, 260]}", category: :env)
  end
  res
rescue StandardError => e
  warn "[pwn-ai/extrospection] auto_extrospect swallowed: #{e.class}: #{e.message}"
  nil
end

.correlate(opts = {}) ⇒ Object

Supported Method Parameters

findings = PWN::AI::Agent::Extrospection.correlate( limit: 'optional - max findings returned (default 10)' )

THE join between introspection and extrospection. Cross-references:

* Metrics tools with success_rate < 50 %  vs  toolchain drift / missing bins
* Learning failures                       vs  host / net drift on the same day
* Observations tagged :vuln / :intel      vs  installed package versions

Emits human-readable, actionable findings the model can reason on.



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

public_class_method def self.correlate(opts = {})
  limit    = opts[:limit] || 10
  findings = []
  delta    = drift(live: false)
  snap     = load[:snapshot] || {}

  # 1) failing tools whose backing binary vanished / changed version
  if defined?(Metrics)
    Metrics.summary(limit: 50).each do |m|
      next unless m[:success_rate] < 0.5 && m[:calls] > 2

      bin = m[:name].to_s.split('_').first
      tc  = Array(delta[:changed]).find { |c| c[:path].to_s.include?("toolchain.#{bin}") }
      miss = snap.dig(:toolchain, bin.to_sym).to_s.empty?
      next unless tc || miss

      findings << { kind: :tool_env_mismatch, tool: m[:name], success_rate: m[:success_rate], evidence: tc || "binary '#{bin}' not found in PATH", advice: "Re-verify `which #{bin}` / reinstall before relying on #{m[:name]}." }
    end
  end

  # 2) introspective failures coinciding with extrospective drift
  if defined?(Learning)
    Learning.outcomes(limit: 30, success: false).each do |o|
      day = o[:timestamp].to_s[0, 10]
      hit = Array(delta[:changed]).find { |c| c[:after].to_s.include?(day) || c[:path].to_s.match?(/kernel|repo|net/) }
      findings << { kind: :failure_during_drift, task: o[:task], on: day, drift: hit, advice: 'Environment changed around this failure — re-test under current snapshot before trusting the negative result.' } if hit
    end
  end

  # 3) intel observations matching installed components
  pkgs = snap[:toolchain] || {}
  observations(category: 'intel', limit: 100).each do |ob|
    blob = ob[:data].to_s.downcase
    pkgs.each do |bin, ver|
      next if ver.to_s.empty?
      next unless blob.include?(bin.to_s.downcase)

      findings << { kind: :intel_matches_host, component: bin, installed: ver, intel: ob[:data], source: ob[:source], advice: "Review #{ob[:source]} advisory for #{bin} #{ver} on this host." }
    end
  end

  # 4) raw drift as low-priority findings when nothing else matched
  Array(delta[:added]).first(5).each   { |c| findings << { kind: :env_added,   detail: c } } if findings.empty?
  Array(delta[:removed]).first(5).each { |c| findings << { kind: :env_removed, detail: c } } if findings.empty?

  findings.uniq.first(limit)
end

.drift(opts = {}) ⇒ Object

Supported Method Parameters

delta = PWN::AI::Agent::Extrospection.drift( live: 'optional - probe host NOW and diff vs stored snapshot (default true). When false, diff stored :snapshot vs stored :previous.' )



186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 186

public_class_method def self.drift(opts = {})
  live  = if opts.key?(:live)
            opts[:live] ? true : false
          else
            true
          end
  store = load
  if live
    after  = snapshot(persist: false)[:snapshot]
    before = store[:snapshot] || {}
  else
    after  = store[:snapshot] || {}
    before = store[:previous] || {}
  end
  compute_drift(before: before, after: after)
end

.help ⇒ Object

Display Usage for this Module



595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 595

public_class_method def self.help
  puts <<~USAGE
    USAGE:
      PWN::AI::Agent::Extrospection.snapshot                          # probe host, persist, return {snapshot:, drift:}
      PWN::AI::Agent::Extrospection.drift(live: true)                 # what changed vs last snapshot
      PWN::AI::Agent::Extrospection.observe(source: 'nmap', category: :recon, target: '10.0.0.5', data: '22/tcp open ssh 9.6')
      PWN::AI::Agent::Extrospection.observations(category: 'recon', target: '10.0.0.5')
      PWN::AI::Agent::Extrospection.intel(query: 'openssl 3.0', record: true)
      PWN::AI::Agent::Extrospection.correlate                         # introspection x extrospection findings
      PWN::AI::Agent::Extrospection.to_context                        # injected by PromptBuilder
      PWN::AI::Agent::Extrospection.stats
      PWN::AI::Agent::Extrospection.auto_extrospect(session_id: sid)  # called from Learning.auto_reflect
      PWN::AI::Agent::Extrospection.reset

      Enable end-of-run auto-extrospection with:
        PWN::Env[:ai][:agent][:auto_extrospect] = true

      #{self}.authors
  USAGE
end

.intel(opts = {}) ⇒ Object

Supported Method Parameters

hits = PWN::AI::Agent::Extrospection.intel( query: 'required - keyword / product / CVE id to search external threat-intel for', feeds: 'optional - Array subset of [:nvd, :circl, :exploitdb] (default all)', limit: 'optional - max results per feed (default 5)', record: 'optional - also persist each hit as an observation (default false)' )

Best-effort external lookups. Network / API failures degrade to an empty result for that feed rather than raising, so learning never breaks the primary loop when offline.



215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 215

public_class_method def self.intel(opts = {})
  query = opts[:query].to_s.strip
  raise 'ERROR: query is required' if query.empty?

  feeds  = Array(opts[:feeds]).map(&:to_sym)
  feeds  = %i[nvd circl exploitdb] if feeds.empty?
  limit  = opts[:limit] || 5
  record = opts[:record] ? true : false

  results = {}
  results[:nvd]       = intel_nvd(query: query, limit: limit)       if feeds.include?(:nvd)
  results[:circl]     = intel_circl(query: query, limit: limit)     if feeds.include?(:circl)
  results[:exploitdb] = intel_exploitdb(query: query, limit: limit) if feeds.include?(:exploitdb)

  if record
    results.each do |feed, hits|
      Array(hits).each do |h|
        observe(source: feed.to_s, category: :intel, data: h, tags: ['intel', query], target: h.is_a?(Hash) ? h[:id].to_s : nil)
      end
    end
  end
  { query: query, feeds: feeds, results: results, total: results.values.flatten.compact.length }
end

.load ⇒ Object

Supported Method Parameters

store = PWN::AI::Agent::Extrospection.load



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

public_class_method def self.load
  FileUtils.mkdir_p(File.dirname(EXTRO_FILE))
  return { snapshot: {}, previous: {}, observations: [], updated_at: nil } unless File.exist?(EXTRO_FILE)

  JSON.parse(File.read(EXTRO_FILE), symbolize_names: true)
rescue StandardError
  { snapshot: {}, previous: {}, observations: [], updated_at: nil }
end

.observations(opts = {}) ⇒ Object

Supported Method Parameters

rows = PWN::AI::Agent::Extrospection.observations( limit: 'optional - max entries newest-first (default 50)', source: 'optional - filter by source substring', category: 'optional - filter by category', target: 'optional - filter by target substring', tag: 'optional - filter by tag substring', fresh_only: 'optional - drop entries whose ttl has expired (default false)' )



158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 158

public_class_method def self.observations(opts = {})
  limit  = opts[:limit] || 50
  src    = opts[:source].to_s.downcase
  cat    = opts[:category].to_s.downcase
  tgt    = opts[:target].to_s.downcase
  tag    = opts[:tag].to_s.downcase
  fresh  = opts[:fresh_only] ? true : false
  now    = Time.now.utc

  rows = Array(load[:observations])
  rows = rows.reject do |o|
    next true if fresh && o[:ttl] && (Time.parse(o[:timestamp].to_s) + o[:ttl].to_i) < now

    (src.empty? ? false : !o[:source].to_s.downcase.include?(src)) ||
      (cat.empty? ? false : o[:category].to_s.downcase != cat) ||
      (tgt.empty? ? false : !o[:target].to_s.downcase.include?(tgt)) ||
      (tag.empty? ? false : Array(o[:tags]).none? { |t| t.to_s.downcase.include?(tag) })
  rescue StandardError
    false
  end
  rows.reverse.first(limit)
end

.observe(opts = {}) ⇒ Object

Supported Method Parameters

obs = PWN::AI::Agent::Extrospection.observe( source: 'required - where the observation came from (nmap, shodan, burp, cve, human, ...)', data: 'required - the observation payload (String or Hash)', category: 'optional - :recon, :vuln, :intel, :target, :network, :misc (default :misc)', target: 'optional - host/ip/url/asset the observation is about', tags: 'optional - Array of String labels', ttl: 'optional - seconds until this observation is considered stale (default nil = forever)' )

Records a fact about the OUTSIDE world (as opposed to Learning.note_outcome which records a fact about the agent's own behaviour). Observations are re-injected via .to_context so recon findings and threat-intel persist across sessions.



124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 124

public_class_method def self.observe(opts = {})
  source = opts[:source].to_s
  data   = opts[:data]
  raise 'ERROR: source is required' if source.strip.empty?
  raise 'ERROR: data is required'   if data.to_s.strip.empty?

  entry = {
    id: Digest::SHA256.hexdigest("#{source}-#{data}-#{Time.now.to_f}")[0, 12],
    source: source,
    category: (opts[:category] || :misc).to_s,
    target: opts[:target].to_s.empty? ? nil : opts[:target].to_s,
    data: data.is_a?(String) ? data[0, 4_000] : data,
    tags: Array(opts[:tags]).map(&:to_s),
    ttl: opts[:ttl]&.to_i,
    timestamp: Time.now.utc.iso8601
  }
  store = load
  store[:observations] ||= []
  store[:observations] << entry
  store[:observations].shift(store[:observations].length - MAX_OBSERVATIONS) if store[:observations].length > MAX_OBSERVATIONS
  save(store: store)
  entry
end

.reset ⇒ Object

Supported Method Parameters

PWN::AI::Agent::Extrospection.reset



388
389
390
391
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 388

public_class_method def self.reset
  FileUtils.rm_f(EXTRO_FILE)
  { cleared: true }
end

.save(opts = {}) ⇒ Object

Supported Method Parameters

PWN::AI::Agent::Extrospection.save( store: 'required - Hash returned by .load / mutated in place' )



63
64
65
66
67
68
69
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 63

public_class_method def self.save(opts = {})
  store = opts[:store] ||= { snapshot: {}, previous: {}, observations: [] }
  store[:updated_at] = Time.now.utc.iso8601
  FileUtils.mkdir_p(File.dirname(EXTRO_FILE))
  File.write(EXTRO_FILE, JSON.pretty_generate(store))
  store
end

.snapshot(opts = {}) ⇒ Object

Supported Method Parameters

snap = PWN::AI::Agent::Extrospection.snapshot( persist: 'optional - Boolean, write snapshot to disk & rotate previous (default true)', sections: 'optional - Array subset of [:host, :net, :toolchain, :repo, :env] (default all)' )

Captures a fingerprint of the OUTSIDE world. When persist:true the prior snapshot is rotated into :previous so .drift can diff them.



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

public_class_method def self.snapshot(opts = {})
  persist  = if opts.key?(:persist)
               opts[:persist] ? true : false
             else
               true
             end
  sections = Array(opts[:sections]).map(&:to_sym)
  sections = %i[host net toolchain repo env] if sections.empty?

  snap = {}
  snap[:host]      = probe_host      if sections.include?(:host)
  snap[:net]       = probe_net       if sections.include?(:net)
  snap[:toolchain] = probe_toolchain if sections.include?(:toolchain)
  snap[:repo]      = probe_repo      if sections.include?(:repo)
  snap[:env]       = probe_env       if sections.include?(:env)
  snap[:captured_at] = Time.now.utc.iso8601
  snap[:fingerprint] = Digest::SHA256.hexdigest(JSON.generate(snap.except(:captured_at)))[0, 16]

  if persist
    store = load
    store[:previous] = store[:snapshot] || {}
    store[:snapshot] = snap
    save(store: store)
  end

  drift = compute_drift(before: load[:previous] || {}, after: snap)
  { snapshot: snap, drift: drift, persisted: persist }
end

.stats ⇒ Object

Supported Method Parameters

stats = PWN::AI::Agent::Extrospection.stats



337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 337

public_class_method def self.stats
  store = load
  snap  = store[:snapshot] || {}
  delta = compute_drift(before: store[:previous] || {}, after: snap)
  {
    snapshot_captured_at: snap[:captured_at],
    snapshot_fingerprint: snap[:fingerprint],
    observations: Array(store[:observations]).length,
    drift_changed: Array(delta[:changed]).length,
    drift_added: Array(delta[:added]).length,
    drift_removed: Array(delta[:removed]).length,
    toolchain_bins: (snap[:toolchain] || {}).count { |_, v| !v.to_s.empty? },
    listening_ports: Array(snap.dig(:net, :listening)).length
  }
end

.to_context(opts = {}) ⇒ Object

Supported Method Parameters

ctx = PWN::AI::Agent::Extrospection.to_context( drift_limit: 'optional - max drift lines (default 4)', obs_limit: 'optional - max observation lines (default 4)' )



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

public_class_method def self.to_context(opts = {})
  dlim = opts[:drift_limit] || 4
  olim = opts[:obs_limit]   || 4
  store = load
  return '' if (store[:snapshot] || {}).empty? && Array(store[:observations]).empty?

  snap  = store[:snapshot] || {}
  delta = compute_drift(before: store[:previous] || {}, after: snap)
  lines = []
  host  = snap[:host] || {}
  lines << "  host_fp   : #{snap[:fingerprint]}  (#{host[:os]} #{host[:kernel]}, #{host[:arch]})" if snap[:fingerprint]
  lines << "  captured  : #{snap[:captured_at]}" if snap[:captured_at]

  ch = Array(delta[:changed]).first(dlim).map { |c| "    ~ #{c[:path]}: #{c[:before].to_s[0, 40]} -> #{c[:after].to_s[0, 40]}" }
  ad = Array(delta[:added]).first(dlim).map   { |c| "    + #{c[:path]}: #{c[:after].to_s[0, 60]}" }
  rm = Array(delta[:removed]).first(dlim).map { |c| "    - #{c[:path]}: #{c[:before].to_s[0, 60]}" }
  unless ch.empty? && ad.empty? && rm.empty?
    lines << '  drift     :'
    lines.concat(ch).concat(ad).concat(rm)
  end

  obs = observations(limit: olim, fresh_only: true)
  unless obs.empty?
    lines << '  observed  :'
    obs.each { |o| lines << "    * [#{o[:category]}/#{o[:source]}] #{"#{o[:target]} — " if o[:target]}#{o[:data].to_s.gsub(/\s+/, ' ')[0, 120]}" }
  end

  "EXTROSPECTION (world-state; correlate with introspective failures)\n#{lines.join("\n")}\n\n"
end