Module: PWN::AI::Anthropic

Defined in:
lib/pwn/ai/anthropic.rb

Overview

This plugin interacts with Anthropic's Claude API. It provides methods to list models, generate completions, and chat. API documentation: https://docs.anthropic.com/en/api Obtain an API key from https://console.anthropic.com/

Constant Summary collapse

ANTHROPIC_OAUTH_CLIENT_ID =

Anthropic / Claude OAuth (Claude Pro/Max) -- public-client PKCE.

Same identity path Claude Code / community tools use:

* public client_id 9d1c250a-e61b-44d9-88ed-5944d1962f5e (no secret)
* authorization_code + S256 PKCE against claude.ai
* token endpoint at platform.claude.com
* redirect_uri is the official hosted callback that returns a
pasteable code (code=true) -- SSH / headless friendly, no
localhost listener required.

Access tokens are short-lived; refresh_token is durable. Persist via pwn-vault under ai.anthropic.oauth.refresh_token.

Wire format for OAuth bearers is different from API keys: Authorization: Bearer <access_token> anthropic-beta: (NO x-api-key)

'9d1c250a-e61b-44d9-88ed-5944d1962f5e'
ANTHROPIC_OAUTH_AUTHORIZE_URI = 
'https://claude.ai/oauth/authorize'
ANTHROPIC_OAUTH_TOKEN_URI = 
'https://platform.claude.com/v1/oauth/token'
ANTHROPIC_OAUTH_REDIRECT_URI = 
'https://platform.claude.com/oauth/code/callback'
ANTHROPIC_OAUTH_SCOPE = 
'user:profile user:inference user:sessions:claude_code user:mcp_servers user:file_upload'
ANTHROPIC_OAUTH_USER_AGENT = 
'claude-cli/2.1.81 (external, cli)'
ANTHROPIC_OAUTH_BETA_FLAGS = 
'claude-code-20250219,oauth-2025-04-20,interleaved-thinking-2025-05-14,prompt-caching-scope-2026-01-05'

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com



724
725
726
727
728
 
# File 'lib/pwn/ai/anthropic.rb', line 724

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

.chat(opts = {}) ⇒ Object

Supported Method Parameters

response = PWN::AI::Anthropic.chat( request: 'required - message to Anthropic', model: 'optional - model to use for text generation (defaults to PWN::Env[:anthropic][:model])', temp: 'optional - creative response float (defaults to PWN::Env[:anthropic][:temp])', system_role_content: 'optional - context to set up the model behavior for conversation (Default: PWN::Env[:anthropic][:system_role_content])', response_history: 'optional - pass response back in to have a conversation', speak_answer: 'optional speak answer using PWN::Plugins::Voice.text_to_speech (Default: nil)', timeout: 'optional timeout in seconds (defaults to 900)', spinner: 'optional - display spinner (defaults to false)' )



617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
 
# File 'lib/pwn/ai/anthropic.rb', line 617

public_class_method def self.chat(opts = {})
  engine = PWN::Env[:ai][:anthropic]
  request = opts[:request]
  max_prompt_length = engine[:max_prompt_length] ||= 200_000
  request_trunc_idx = ((max_prompt_length - 1) / 3.36).floor
  request = request[0..request_trunc_idx]

  model = opts[:model] ||= engine[:model]
  raise 'ERROR: Model is required.  Call #get_models method for details' if model.nil?

  temp = opts[:temp].to_f ||= engine[:temp].to_f
  temp = 1 if temp.zero?

  rest_call = 'messages'

  response_history = opts[:response_history]

  system_role_content = opts[:system_role_content] ||= engine[:system_role_content]

  system_role = {
    role: 'system',
    content: system_role_content
  }

  user_role = {
    role: 'user',
    content: request
  }

  response_history ||= { choices: [system_role] }

  http_body = {
    model: model,
    max_tokens: engine[:max_tokens] || 128_000,
    temperature: temp,
    system: system_role_content,
    messages: []
  }

  if response_history[:choices].length > 1
    response_history[:choices][1..].each do |message|
      next if message[:role] == 'system'

      http_body[:messages].push(role: message[:role].to_s, content: message[:content].to_s)
    end
  end

  http_body[:messages].push(role: 'user', content: request)

  timeout = opts[:timeout]
   = opts[:spinner]

  response = anthropic_rest_call(
    http_method: :post,
    rest_call: rest_call,
    http_body: http_body,
    timeout: timeout,
    spinner: 
  )

  json_resp = JSON.parse(response, symbolize_names: true)
  raise "Anthropic API Error: #{json_resp[:error] || json_resp}" if json_resp[:error] || json_resp[:type] == 'error'

  assistant_content = if json_resp[:content] && json_resp[:content].is_a?(Array) && json_resp[:content].first
                        json_resp[:content].first[:text]
                      else
                        ''
                      end
  assistant_resp = {
    role: 'assistant',
    content: assistant_content
  }

  # Build choices for PWN compatibility: [system, ...history..., user, assistant]
  json_resp[:choices] = [system_role] + http_body[:messages]
  json_resp[:choices].push(assistant_resp)

  # Ensure compatibility fields
  json_resp[:id] ||= "msg_#{SecureRandom.hex(8)}"
  json_resp[:object] ||= 'message'
  json_resp[:model] ||= model

  if json_resp[:usage].is_a?(Hash)
    inp_tokens = json_resp[:usage][:input_tokens] || 0
    out_tokens = json_resp[:usage][:output_tokens] || 0
    json_resp[:usage][:total_tokens] = inp_tokens + out_tokens
  else
    json_resp[:usage] = { input_tokens: 0, output_tokens: 0, total_tokens: 0 }
  end

  speak_answer = true if opts[:speak_answer]

  if speak_answer
    answer = assistant_resp[:content]
    text_path = "/tmp/#{SecureRandom.hex}.pwn_voice"
    File.write(text_path, answer)
    PWN::Plugins::Voice.text_to_speech(text_path: text_path)
    File.unlink(text_path)
  end

  json_resp
rescue StandardError => e
  raise e
end

.chat_with_tools(opts = {}) ⇒ Object

Supported Method Parameters

response = PWN::AI::Anthropic.chat_with_tools( messages: 'required - OpenAI-format messages array (system/user/assistant/tool)', tools: 'optional - OpenAI tools array [function:{...}]', tool_choice: 'optional - "auto" | "none" | "required" | function:{name:..}', model: 'optional - overrides PWN::Env[:anthropic][:model]', temp: 'optional - temperature (defaults to PWN::Env[:anthropic][:temp] || 1)', max_tokens: 'optional - defaults to PWN::Env[:anthropic][:max_tokens] || 128_000', timeout: 'optional - seconds (default 900)', spinner: 'optional - display spinner (default false)' )



420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
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
 
# File 'lib/pwn/ai/anthropic.rb', line 420

public_class_method def self.chat_with_tools(opts = {})
  engine   = PWN::Env[:ai][:anthropic]
  messages = opts[:messages]
  raise 'ERROR: messages array is required' if messages.nil? || messages.empty?

  model = opts[:model] ||= engine[:model]
  raise 'ERROR: Model is required.  Call #get_models method for details' if model.nil?

  temp = opts[:temp].to_f
  temp = engine[:temp].to_f.nonzero? || 1 if temp.zero?

  system_str, anth_messages = oa_messages_to_anthropic(messages: messages)

  http_body = {
    model: model,
    max_tokens: opts[:max_tokens] || engine[:max_tokens] || 128_000,
    temperature: temp,
    messages: anth_messages
  }
  http_body[:system] = system_str if system_str && !system_str.empty?

  if opts[:tools] && !opts[:tools].empty?
    http_body[:tools] = opts[:tools].map do |t|
      fn = t[:function] || t['function'] || t
      {
        name: fn[:name] || fn['name'],
        description: fn[:description] || fn['description'],
        input_schema: fn[:parameters] || fn['parameters'] || { type: 'object', properties: {} }
      }
    end
    http_body[:tool_choice] = anth_tool_choice(choice: opts[:tool_choice]) if opts[:tool_choice]
  end

  response = anthropic_rest_call(
    http_method: :post,
    rest_call: 'messages',
    http_body: http_body,
    timeout: opts[:timeout],
    spinner: opts[:spinner]
  )
  return nil if response.nil?

  json_resp = JSON.parse(response, symbolize_names: true)
  raise "Anthropic API Error: #{json_resp[:error] || json_resp}" if json_resp[:error] || json_resp[:type] == 'error'

  anthropic_resp_to_oa(response: json_resp)
rescue StandardError => e
  raise e
end

.get_models ⇒ Object

Supported Method Parameters

models = PWN::AI::Anthropic.get_models



384
385
386
387
388
389
390
 
# File 'lib/pwn/ai/anthropic.rb', line 384

public_class_method def self.get_models
  models = anthropic_rest_call(rest_call: 'models')

  JSON.parse(models, symbolize_names: true)[:data]
rescue StandardError => e
  raise e
end

.help ⇒ Object

Display Usage for this Module



732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
 
# File 'lib/pwn/ai/anthropic.rb', line 732

public_class_method def self.help
  puts "USAGE:
    models = #{self}.get_models

    # One-time Claude Pro/Max OAuth enrollment (PKCE paste flow):
    bearer = #{self}.obtain_oauth_bearer_token
    # Subsequent runs silently refresh via ai.anthropic.oauth.refresh_token

    response = #{self}.chat(
      request: 'required - message to Anthropic',
      model: 'optional - model to use for text generation (defaults to PWN::Env[:ai][:anthropic][:model])',
      temp: 'optional - creative response float (defaults to PWN::Env[:ai][:anthropic][:temp])',
      system_role_content: 'optional - context to set up the model behavior for conversation (Default: PWN::Env[:ai][:anthropic][:system_role_content])',
      response_history: 'optional - pass response back in to have a conversation',
      speak_answer: 'optional speak answer using PWN::Plugins::Voice.text_to_speech (Default: nil)',
      timeout: 'optional - timeout in seconds (defaults to 900)',
      spinner: 'optional - display spinner (defaults to false)'
    )

    #{self}.authors
  "
end

.obtain_oauth_bearer_token(opts = {}) ⇒ Object

Supported Method Parameters

bearer = PWN::AI::Anthropic.obtain_oauth_bearer_token( client_id: 'optional - Claude Code public client id', scope: 'optional - space-delimited scopes', redirect_uri: 'optional - must match registered callback', authorize_uri:'optional - claude.ai authorize endpoint', token_uri: 'optional - platform.claude.com token endpoint' )

Runs the OAuth 2.0 Authorization Code + PKCE (S256) flow against Anthropic's public Claude Code client. The hosted redirect returns a pasteable code (code=true) so this works over SSH with no localhost listener -- same UX class as the Grok device flow.



139
140
141
142
143
144
145
146
147
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
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/anthropic.rb', line 139

public_class_method def self.obtain_oauth_bearer_token(opts = {})
  client_id     = real_config_value?(value: opts[:client_id])     ? opts[:client_id]     : ANTHROPIC_OAUTH_CLIENT_ID
  scope         = real_config_value?(value: opts[:scope])         ? opts[:scope]         : ANTHROPIC_OAUTH_SCOPE
  redirect_uri  = real_config_value?(value: opts[:redirect_uri])  ? opts[:redirect_uri]  : ANTHROPIC_OAUTH_REDIRECT_URI
  authorize_uri = real_config_value?(value: opts[:authorize_uri]) ? opts[:authorize_uri] : ANTHROPIC_OAUTH_AUTHORIZE_URI
  token_uri     = real_config_value?(value: opts[:token_uri])     ? opts[:token_uri]     : ANTHROPIC_OAUTH_TOKEN_URI

  pkce = pkce_pair
  params = {
    'code' => 'true',
    'response_type' => 'code',
    'client_id' => client_id,
    'redirect_uri' => redirect_uri,
    'scope' => scope,
    'code_challenge' => pkce[:challenge],
    'code_challenge_method' => 'S256',
    'state' => pkce[:verifier]
  }
  auth_url = "#{authorize_uri}?#{URI.encode_www_form(params)}"

  puts "\n[*] Anthropic / Claude OAuth -- Authorization Code + PKCE (S256, public client, no secret)"
  puts '    A Claude Pro / Max subscription on the approving account is required for inference scope.'
  puts ''
  puts '    Step 1: In a browser (any device), open:'
  puts "            #{auth_url}"
  puts '    Step 2: Approve access for Claude Code / CLI.'
  puts '    Step 3: Copy the authorization code shown after redirect and paste it below.'
  puts ''
  print '    Authorization code> '
  raw = $stdin.gets
  raise 'Anthropic OAuth enrollment aborted: no code provided.' if raw.nil?

  # Hosted callback sometimes returns "code#state" — take the code segment only.
  code = raw.to_s.strip.split('#', 2).first
  raise 'Anthropic OAuth enrollment aborted: empty code.' if code.empty?

  resp = RestClient.post(
    token_uri,
    {
      grant_type: 'authorization_code',
      code: code,
      code_verifier: pkce[:verifier],
      client_id: client_id,
      redirect_uri: redirect_uri,
      state: pkce[:verifier]
    },
    content_type: 'application/x-www-form-urlencoded',
    accept: 'application/json',
    user_agent: ANTHROPIC_OAUTH_USER_AGENT
  )
  data = JSON.parse(resp.body)
  raise "Anthropic OAuth token error: #{data['error']} - #{data['error_description'] || data['message']}" if data['error']

  access_token  = data['access_token']
  refresh_token = data['refresh_token']
  raise 'Anthropic OAuth token endpoint returned no access_token.' unless access_token

  opts[:bearer_token]  = access_token
  opts[:refresh_token] = refresh_token if refresh_token
  opts[:expires_at]    = Time.now.to_i + data['expires_in'].to_i if data['expires_in']

  puts "\n[*] SUCCESS: Anthropic OAuth bearer obtained via authorization_code + PKCE."
  puts '    Cached in-memory for this pwn / pwn-ai process.'
  puts ''
  puts '    TO MAKE THIS PERMANENT (recommended -- one-time), store via pwn-vault:'
  puts "      ai.anthropic.oauth.refresh_token = #{refresh_token}" if refresh_token
  puts "      ai.anthropic.oauth.bearer_token  = #{access_token}"
  puts '    On future runs the refresh_token alone is enough -- PWN::AI::Anthropic will'
  puts '    silently exchange it for a fresh access_token (no browser, no prompt).'
  puts ''

  access_token
rescue RestClient::ExceptionWithResponse => e
  raise "Failed to obtain Anthropic OAuth bearer token (HTTP #{e.http_code}): #{e.response&.body}"
rescue StandardError => e
  raise "Failed to obtain Anthropic OAuth bearer token: #{e.message}"
end

.refresh_oauth_bearer_token(opts = {}) ⇒ Object

Supported Method Parameters

access_token = PWN::AI::Anthropic.refresh_oauth_bearer_token( refresh_token: 'required - Anthropic OAuth refresh_token', client_id: 'optional - defaults to Claude Code public client', token_uri: 'optional - defaults to platform.claude.com token endpoint' )



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/pwn/ai/anthropic.rb', line 97

public_class_method def self.refresh_oauth_bearer_token(opts = {})
  refresh_token = opts[:refresh_token]
  raise 'refresh_token is required' unless real_config_value?(value: refresh_token)

  client_id = real_config_value?(value: opts[:client_id]) ? opts[:client_id] : ANTHROPIC_OAUTH_CLIENT_ID
  token_uri = real_config_value?(value: opts[:token_uri]) ? opts[:token_uri] : ANTHROPIC_OAUTH_TOKEN_URI

  resp = RestClient.post(
    token_uri,
    {
      grant_type: 'refresh_token',
      refresh_token: refresh_token,
      client_id: client_id
    },
    content_type: 'application/x-www-form-urlencoded',
    accept: 'application/json',
    user_agent: ANTHROPIC_OAUTH_USER_AGENT
  )
  data = JSON.parse(resp.body)
  raise "Anthropic OAuth refresh error: #{data['error']} - #{data['error_description'] || data['message']}" if data['error']

  opts[:bearer_token]  = data['access_token']
  opts[:refresh_token] = data['refresh_token'] if data['refresh_token']
  opts[:expires_at]    = Time.now.to_i + data['expires_in'].to_i if data['expires_in']
  data['access_token']
rescue RestClient::ExceptionWithResponse => e
  raise "Anthropic OAuth refresh failed (HTTP #{e.http_code}): #{e.response&.body}"
end