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
RF_BINS = 
%w[rtl_sdr rtl_test rtl_433 hackrf_info gqrx dump1090 multimon-ng SoapySDRUtil].freeze
PROBE_BINS = 
(%w[nmap curl git ruby python3 gcc msfconsole sqlmap burpsuite zaproxy openssl docker] + RF_BINS).freeze
WEB_SHOT_DIR = 
File.join(Dir.home, '.pwn', 'extrospection', 'web')
DEFAULT_WEB_ANCHORS = 
%w[
  https://services.nvd.nist.gov/rest/json/cves/2.0
  https://www.exploit-db.com/
  https://raw.githubusercontent.com/0dayinc/pwn/master/lib/pwn/version.rb
].freeze

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com



1068
1069
1070
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 1068

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_introspect 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.



606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 606

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.



443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 443

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) :rf observations vs missing SDR hardware / binaries
  rf = snap[:rf] || {}
  hw_present = %i[rtl_sdr hackrf flipper gqrx_sock].any? { |k| rf_present?(val: rf[k]) }
  observations(category: 'rf', limit: 50).each do |ob|
    miss = RF_BINS.select { |b| pkgs[b.to_sym].to_s.empty? }
    if !hw_present
      findings << { kind: :rf_no_hardware, observation: ob[:data], source: ob[:source], target: ob[:target], advice: 'RF observation recorded but no SDR hardware detected in snapshot — plug in RTL-SDR/HackRF/Flipper or start gqrx (`-r`) before trusting RF results.' }
    elsif rf[:gqrx_sock] == false && ob[:source].to_s == 'gqrx'
      findings << { kind: :rf_gqrx_down, observation: ob[:data], target: ob[:target], advice: 'gqrx remote-control socket (127.0.0.1:7356) is closed — start gqrx with remote control enabled before re-running the scan.' }
    elsif !miss.empty?
      findings << { kind: :rf_toolchain_gap, missing: miss, observation: ob[:data], advice: "SDR toolchain gap: install #{miss.join(', ')} to decode/act on this RF observation." }
    end
  end

  # 5) :web drift on a target the same day a Learning failure references it —
  #    "your exploit stopped working because the TARGET changed, not your approach"
  if defined?(Learning)
    observations(category: 'web', limit: 50).each do |ob|
      tgt  = ob[:target].to_s
      host = safe_host(url: tgt)
      Learning.outcomes(limit: 30, success: false).each do |o|
        next unless o[:task].to_s.include?(host) || o[:details].to_s.include?(host)

        findings << { kind: :target_web_drift, target: tgt, dom_sha: ob.dig(:data, :dom_sha), task: o[:task], advice: "Target #{host} DOM changed (#{ob[:timestamp]}) around this failure — re-recon before assuming your technique is wrong." }
      end
    end
  end

  # 6) extro_verify refutations whose claim substring appears in a PWN::Memory :fact -> stale memory
  if defined?(PWN::Memory)
    observations(category: 'web', tag: 'stale', limit: 50).each do |ob|
      findings << { kind: :stale_memory_fact, key: ob[:target], evidence: ob[:data], advice: "PWN::Memory[:#{ob[:target]}] failed browser re-verification — audit or memory_forget it before it poisons future prompts." }
    end
  end

  # 7) :intel observations whose source anchor is currently probe_web-unreachable -> downgrade
  web = snap[:web] || {}
  web.each do |host, fp|
    next unless fp.is_a?(Hash) && (fp[:reachable] == false || fp[:status].to_i >= 500)

    observations(category: 'intel', limit: 100).each do |ob|
      next unless ob[:source].to_s.include?(host.to_s) || ob[:data].to_s.include?(host.to_s)

      findings << { kind: :intel_source_unreachable, source: host, status: fp[:status], intel: ob[:data].to_s[0, 120], advice: "Feed anchor #{host} is currently unreachable (#{fp[:status] || 'down'}) — treat this :intel as stale until probe_web sees it 2xx again." }
    end
  end

  # 8) 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.' )



196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 196

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



1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 1074

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.observe(source: 'gqrx', category: :rf, target: '433.920MHz', data: 'peak -34.2 dBFS bw=200k')
      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_introspect
      PWN::AI::Agent::Extrospection.snapshot(sections: %i[web])        # opt-in browser probe of web_anchors
      PWN::AI::Agent::Extrospection.watch(url: 'https://target/api/version')
      PWN::AI::Agent::Extrospection.verify(claim: 'CVE-2026-12345 affects OpenSSL 3.2.1')
      PWN::AI::Agent::Extrospection.revalidate_memory                  # cron: GC stale PWN::Memory :fact entries
      PWN::AI::Agent::Extrospection.reset

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

      Configure browser-backed :web probe / verify / watch with:
        PWN::Env[:ai][:agent][:extrospection][:web] =
          { anchors: [...], proxy: 'tor', max_anchors: 8, per_page_timeout: 15, screenshot: false, allow_targets: false }

      #{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.



225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 225

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



57
58
59
60
61
62
63
64
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 57

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)' )



168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 168

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, :env, :rf, :web, :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.



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 134

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



630
631
632
633
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 630

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

.revalidate_memory(opts = {}) ⇒ Object

Supported Method Parameters

report = PWN::AI::Agent::Extrospection.revalidate_memory( limit: 'optional - max :fact entries to check (default 25)', proxy: 'optional - upstream proxy for TransparentBrowser' )

The browser as garbage-collector for PWN::Memory. Walks :fact entries containing a CVE id, version string or URL, runs verify() on each and prefixes stale ones with [UNVERIFIED yyyy-mm-dd] so the injected MEMORY block stops calcifying into confidently-wrong priors. Designed to be scheduled: cron_create(schedule:'0 4 * * 0', ruby:'PWN::AI::Agent::Extrospection.revalidate_memory')



407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 407

public_class_method def self.revalidate_memory(opts = {})
  return { checked: 0, refuted: [], confirmed: [], unknown: [] } unless defined?(PWN::Memory)

  lim   = opts[:limit] || 25
  proxy = opts[:proxy] || web_config[:proxy]
  rx    = %r{(CVE-\d{4}-\d{4,}|v?\d+\.\d+\.\d+|https?://\S+)}i
  out   = { checked: 0, refuted: [], confirmed: [], unknown: [] }

  PWN::Memory.load.each do |key, v|
    break if out[:checked] >= lim
    next unless v.is_a?(Hash) && v[:category].to_s == 'fact' && v[:value].to_s =~ rx
    next if v[:value].to_s.start_with?('[UNVERIFIED')

    res = verify(claim: v[:value].to_s[0, 400], commit: false, proxy: proxy)
    out[:checked] += 1
    out[res[:verdict]] << key
    next unless res[:verdict] == :refuted && res[:confidence] >= 0.6

    stamp = Time.now.utc.strftime('%Y-%m-%d')
    PWN::Memory.remember(key: key, value: "[UNVERIFIED #{stamp}] #{v[:value]}", category: :fact)
    observe(source: 'extro_verify', category: :web, target: key.to_s, data: "Memory :fact '#{key}' failed re-validation (#{res[:confidence]})", tags: %w[stale memory])
  end
  out
end

.save(opts = {}) ⇒ Object

Supported Method Parameters

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



71
72
73
74
75
76
77
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 71

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, :rf, :web] (default all except :web)' )

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



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

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 rf] 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[:rf]        = probe_rf        if sections.include?(:rf)
  snap[:web]       = probe_web       if sections.include?(:web)
  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



577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 577

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,
    rf_devices: (snap[:rf] || {}).values_at(:rtl_sdr, :hackrf, :flipper, :gqrx_sock).count { |v| rf_present?(val: v) },
    web_anchors: (snap[:web] || {}).count { |_, v| v.is_a?(Hash) && v[:reachable] }
  }
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)' )



544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 544

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

.verify(opts = {}) ⇒ Object

Supported Method Parameters

verdict = PWN::AI::Agent::Extrospection.verify( claim: 'required - factual claim to fact-check against the live web', kind: 'optional - :cve | :version | :doc | :generic (default auto-detect)', url: 'optional - explicit URL the claim cites (forces kind: :doc)', commit: 'optional - write Mistakes/Memory/observe on refute/confirm (default true)', proxy: 'optional - upstream proxy for TransparentBrowser (e.g. "tor")' )

Browser-backed self fact-checking. Drives PWN::Plugins::TransparentBrowser in :headless mode against a canonical source for the claim class, renders the DOM (JS executed), and returns:

{ claim:, kind:, verdict: :confirmed|:refuted|:unknown,
confidence: 0.0..1.0,
evidence: [{url:, title:, excerpt:, dom_sha:, screenshot:}],
action_taken: :mistakes_record | :extro_observe | :learning_note | nil }

On :refuted -> Mistakes.record(tool:'assumption', error:claim) so the KNOWN MISTAKES block warns every future run. On :confirmed-> observe(category::intel, ttl:30d) reinforces w/ freshness. On :unknown -> Learning.note_outcome(success:false, tags:['needs_human']).

This is the extrospective mirror of Mistakes: a PROACTIVE trigger that catches the model being wrong about the world before a human does.



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

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

  kind   = (opts[:kind] || detect_claim_kind(claim: claim, url: opts[:url])).to_sym
  commit = opts.key?(:commit) ? !opts[:commit].nil? && opts[:commit] != false : true
  proxy  = opts[:proxy] || web_config[:proxy]

  evidence = []
  verdict  = :unknown
  conf     = 0.0

  with_headless_browser(proxy: proxy) do |bo|
    case kind
    when :cve
      cve = claim[/CVE-\d{4}-\d{4,}/i].to_s.upcase
      %W[https://nvd.nist.gov/vuln/detail/#{cve} https://www.cve.org/CVERecord?id=#{cve}].each do |u|
        fp = fingerprint_page(browser_obj: bo, url: u)
        evidence << fp if fp
      end
      body = evidence.map { |e| e[:text].to_s }.join(' ')
      if evidence.empty? || evidence.all? { |e| e[:status].to_i >= 400 } || body =~ /could not be found|does not exist|reserved but/i
        verdict = :refuted
        conf = 0.8
      else
        prod = claim.sub(/CVE-\d{4}-\d{4,}/i, '').gsub(/affects?|is|in|the/i, ' ').split.select { |w| w.length > 2 }
        hit  = prod.count { |w| body.downcase.include?(w.downcase) }
        if prod.empty? || hit.to_f / prod.length >= 0.5
          verdict = :confirmed
          conf = prod.empty? ? 0.6 : [0.5 + (hit.to_f / prod.length * 0.5), 0.95].min
        else
          verdict = :refuted
          conf = 0.6
        end
      end
    when :doc
      u = opts[:url] || claim[URI::DEFAULT_PARSER.make_regexp(%w[http https])]
      fp = fingerprint_page(browser_obj: bo, url: u)
      evidence << fp if fp
      if fp.nil? || fp[:status].to_i >= 400 || fp[:reachable] == false
        verdict = :refuted
        conf = 0.7
      else
        snippet = claim.sub(u.to_s, '').strip
        overlap = fuzzy_overlap(needle: snippet, haystack: fp[:text].to_s)
        if fp[:title].to_s =~ /not found|404|error/i && overlap < 0.2
          verdict = :refuted
          conf = 0.7
        else
          verdict = overlap >= 0.4 ? :confirmed : :unknown
          conf    = overlap
        end
      end
    when :version
      proj = claim[/[A-Za-z][\w.+-]{2,}/].to_s
      ver  = claim[/v?\d+\.\d+(?:\.\d+)?/].to_s
      %W[https://rubygems.org/gems/#{proj} https://pypi.org/project/#{proj}/ https://github.com/search?q=#{URI.encode_www_form_component(proj)}&type=repositories].each do |u|
        fp = fingerprint_page(browser_obj: bo, url: u)
        evidence << fp if fp
      end
      body = evidence.map { |e| e[:text].to_s }.join(' ')
      if !ver.empty? && body.include?(ver)
        verdict = :confirmed
        conf = 0.7
      elsif body =~ /\d+\.\d+(?:\.\d+)?/
        verdict = :unknown
        conf = 0.3
      end
    else
      q  = URI.encode_www_form_component(claim)
      fp = fingerprint_page(browser_obj: bo, url: "https://html.duckduckgo.com/html/?q=#{q}")
      evidence << fp if fp
      overlap = fuzzy_overlap(needle: claim, haystack: fp ? fp[:text].to_s : '')
      verdict = overlap >= 0.5 ? :confirmed : :unknown
      conf    = overlap
    end
  end

  evidence.each { |e| e.delete(:text) }
  action = commit ? commit_verdict(claim: claim, kind: kind, verdict: verdict, confidence: conf, evidence: evidence) : nil
  { claim: claim, kind: kind, verdict: verdict, confidence: conf.round(2), evidence: evidence, action_taken: action }
rescue StandardError => e
  { claim: claim, kind: kind, verdict: :unknown, confidence: 0.0, evidence: evidence, error: "#{e.class}: #{e.message}", action_taken: nil }
end

.watch(opts = {}) ⇒ Object

Supported Method Parameters

obs = PWN::AI::Agent::Extrospection.watch( url: 'required - URL to render and fingerprint', selector: 'optional - CSS selector whose innerText to hash (default full body)', ttl: 'optional - seconds until stale (default 7 days)', proxy: 'optional - upstream proxy for TransparentBrowser' )

Passive change-detection on an external artefact you care about (target /api/version, vendor changelog, bounty scope page). Renders headless, hashes the DOM text, screenshots, and persists as observe(category: :web). On subsequent snapshot(sections:[:web]) or watch of the same URL, a hash mismatch surfaces in drift() as ~web..dom_sha exactly like ~toolchain.nmap today.



375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
 
# File 'lib/pwn/ai/agent/extrospection.rb', line 375

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

  ttl   = opts[:ttl] || (7 * 24 * 3600)
  proxy = opts[:proxy] || web_config[:proxy]
  fp    = nil
  with_headless_browser(proxy: proxy) do |bo|
    fp = fingerprint_page(browser_obj: bo, url: url, selector: opts[:selector], screenshot: true)
  end
  raise "unreachable: #{url}" unless fp

  prior = observations(category: 'web', target: url, limit: 1).first
  changed = prior && prior.dig(:data, :dom_sha) && prior.dig(:data, :dom_sha) != fp[:dom_sha]
  data = fp.slice(:status, :final_url, :title, :dom_sha, :cert_fp, :cert_not_after, :screenshot).merge(excerpt: fp[:text].to_s[0, 200])
  observe(source: 'transparent_browser', category: :web, target: url, data: data, tags: (['watch'] + Array(opts[:tags])).compact, ttl: ttl)
  { url: url, changed: changed, prior_sha: prior && prior.dig(:data, :dom_sha), current: data }
end