Module: PWN::AI::Agent::Manifest

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

Overview

Trusted, on-disk tool declarations. Never load a manifest from tool args.

Constant Summary collapse

RISKS = 
%w[info low med high crit].freeze
GATES = 
%w[auto prompt deny].freeze
DIRECTORY = 
File.expand_path('../tools', __dir__).freeze

Class Method Summary collapse

Class Method Details

.authors ⇒ Object 



144
145
146
 
# File 'lib/pwn/ai/agent/manifest.rb', line 144

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

.check(opts = {}) ⇒ Object

Missing policy preserves the legacy unrestricted operator workflow.



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

public_class_method def self.check(opts = {})
  path = opts[:scope_path] || File.expand_path('~/.pwn/scope.yaml')
  policy = opts[:scope_policy]
  return nil if policy.nil? && !File.exist?(path)

  policy = YAML.safe_load_file(path, permitted_classes: [], aliases: false) if policy.nil?

  policy = JSON.parse(JSON.generate(policy))
  raise ArgumentError, 'scope policy must be an object' unless policy.is_a?(Hash)
  return nil if policy['enabled'] == false

  entries = load(directory: opts[:manifest_directory] || DIRECTORY)
  declaration = entries[opts[:name]] || {}
  args = JSON.parse(JSON.generate(opts[:args]))
  fields = declaration['target_params'] || { 'hosts' => %w[target targets host hosts domain domains cidr cidrs url urls], 'ports' => %w[port ports] }
  hosts = Array(fields['hosts']).flat_map { |key| Array(args[key]) }
  ports = Array(fields['ports']).flat_map { |key| Array(args[key]) }
  outside = hosts.any? do |target|
    host = target.to_s
    if host.include?('://')
      uri = URI.parse(host)
      ports << uri.port if uri.respond_to?(:port) && uri.port
      host = uri.host.to_s
    end
    !host_allowed?(host: host, policy: policy)
  end
  outside ||= ports.any? { |port| Array(policy['allowed_ports']).none? { |allowed| port_range(value: allowed).cover?(port_range(value: port)) } }
  return deny(opts.merge(reason: 'out_of_scope')) if outside

  risk = declaration['risk_level'] || 'crit'
  gate = (policy['risk_gates'] || {}).fetch(risk, 'deny')
  return deny(opts.merge(reason: 'risk_denied')) unless GATES.include?(gate) && gate != 'deny'

  if gate == 'prompt'
    callback = opts[:approval_callback]
    request = { name: opts[:name], risk_level: risk, args: JSON.parse(JSON.generate(args)) }
    return deny(opts.merge(reason: 'approval_required')) unless callback.respond_to?(:call) && callback.call(request) == true
  end

  nil
rescue StandardError
  deny(opts.merge(reason: 'invalid_scope_policy'))
end

.help ⇒ Object 



148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/pwn/ai/agent/manifest.rb', line 148

public_class_method def self.help
  puts "USAGE:
    # Load trusted YAML tool declarations.
    #{self}.load(
      directory: 'optional - trusted manifest directory; default lib/pwn/ai/tools'
    )

    # Register whitelisted plugin methods as tools.
    #{self}.register(
      directory: 'optional - trusted manifest directory; default lib/pwn/ai/tools'
    )

    # Check declared targets and risk gates before dispatch.
    #{self}.check(
      name: 'required - registered tool name',
      args: 'required - parsed tool argument Hash',
      scope_policy: 'optional - trusted policy Hash instead of the policy file',
      scope_path: 'optional - trusted scope YAML path; default ~/.pwn/scope.yaml',
      manifest_directory: 'optional - trusted YAML manifest directory',
      approval_callback: 'optional - trusted callback returning literal true for prompt approval',
      audit_path: 'optional - trusted audit JSONL destination'
    )

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

.load(opts = {}) ⇒ Object

Raises:

  • (ArgumentError) 


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

public_class_method def self.load(opts = {})
  directory = opts[:directory] || DIRECTORY
  entries = Dir[File.join(directory, '*.yaml')].flat_map do |path|
    data = YAML.safe_load_file(path, permitted_classes: [], aliases: false)
    data.is_a?(Array) ? data : [data]
  end
  entries.each do |entry|
    raise ArgumentError, 'invalid manifest entry' unless entry.is_a?(Hash) && entry['name'].is_a?(String) && entry['description'].is_a?(String) && RISKS.include?(entry['risk_level']) && entry['params'].is_a?(Hash)
    raise ArgumentError, 'invalid manifest schema' unless JSONSchemer.valid_schema?(entry['params'])
  end
  raise ArgumentError, 'duplicate manifest tool' unless entries.map { |entry| entry['name'] }.uniq.length == entries.length

  entries.to_h { |entry| [entry['name'], entry] }
end

.register(opts = {}) ⇒ Object 



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/pwn/ai/agent/manifest.rb', line 35

public_class_method def self.register(opts = {})
  entries = load(directory: opts[:directory] || DIRECTORY)
  entries.each_value do |entry|
    next unless entry['plugin']
    raise ArgumentError, 'plugin must be a PWN::Plugins constant' unless entry['plugin'].match?(/\APWN::Plugins::[A-Z]\w*\z/)
    raise ArgumentError, 'invalid plugin method' unless entry['method'].to_s.match?(/\A[a-z]\w*[!?]?\z/)

    declaration = entry
    Registry.register(
      name: entry['name'], toolset: 'manifest',
      schema: { name: entry['name'], description: entry['description'], parameters: JSON.parse(JSON.generate(entry['params']), symbolize_names: true) },
      handler: lambda { |args|
        plugin = declaration['plugin'].split('::').inject(Object) { |mod, name| mod.const_get(name, false) }
        method = plugin.public_method(declaration['method'])
        method.arity.zero? ? method.call : method.call(args)
      }
    )
  end
  entries
end