Module: PWN::AI::Grok

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

Overview

This plugin interacts with xAI’s Grok API, similar to the Grok plugin. It provides methods to list models, generate completions, and chat. API documentation: docs.x.ai/docs Obtain an API key from x.ai/api

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. <support@0dayinc.com>



404
405
406
407
408
 
# File 'lib/pwn/ai/grok.rb', line 404

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

.chat(opts = {}) ⇒ Object

Supported Method Parameters

response = PWN::AI::Grok.chat(

request: 'required - message to Grok'
model: 'optional - model to use for text generation (defaults to PWN::Env[:ai][:grok][:model])',
temp: 'optional - creative response float (deafults to PWN::Env[:ai][:grok][:temp])',
system_role_content: 'optional - context to set up the model behavior for conversation (Default: PWN::Env[:ai][:grok][: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)'

)



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
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
 
# File 'lib/pwn/ai/grok.rb', line 320

public_class_method def self.chat(opts = {})
  engine = PWN::Env[:ai][:grok]
  request = opts[:request]
  max_prompt_length = engine[:max_prompt_length] ||= 256_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 = 'chat/completions'

  response_history = opts[:response_history]

  max_tokens = response_history[:usage][:total_tokens] unless response_history.nil?

  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] }
  choices_len = response_history[:choices].length

  http_body = {
    model: model,
    messages: [system_role],
    temperature: temp,
    stream: false
  }

  if response_history[:choices].length > 1
    response_history[:choices][1..-1].each do |message|
      http_body[:messages].push(message)
    end
  end

  http_body[:messages].push(user_role)

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

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

  json_resp = JSON.parse(response, symbolize_names: true)
  assistant_resp = json_resp[:choices].first[:message]
  json_resp[:choices] = http_body[:messages]
  json_resp[:choices].push(assistant_resp)

  speak_answer = true if opts[:speak_answer]

  if speak_answer
    answer = assistant_resp[:content]
    text_path = "/tmp/#{SecureRandom.hex}.pwn_voice"
    # answer = json_resp[:choices].last[:text]
    # answer = json_resp[:choices].last[:content] if gpt
    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::Grok.chat_with_tools(

messages: 'required - full OpenAI-format messages array (system/user/assistant/tool)',
tools: 'optional - OpenAI tools array [{type:"function", function:{...}}]',
tool_choice: 'optional - "auto" | "none" | "required" | {type:"function", function:{name:..}}',
model: 'optional - overrides PWN::Env[:ai][:grok][:model]',
temp: 'optional - temperature (defaults to PWN::Env[:ai][:grok][:temp] || 1)',
timeout: 'optional - seconds (default 900)',
spinner: 'optional - display spinner (default false)'

)

Returns the raw chat/completions response Hash with :choices intact (including :message) — used by PWN::AI::Agent::Loop. xAI’s API is OpenAI-compatible for tool calling, so the request and response shapes are identical to PWN::AI::OpenAI.chat_with_tools.



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
297
298
299
300
301
302
303
304
305
306
 
# File 'lib/pwn/ai/grok.rb', line 272

public_class_method def self.chat_with_tools(opts = {})
  engine   = PWN::Env[:ai][:grok]
  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?

  http_body = {
    model: model,
    messages: messages,
    temperature: temp,
    stream: false
  }
  http_body[:tools]       = opts[:tools]       if opts[:tools] && !opts[:tools].empty?
  http_body[:tool_choice] = opts[:tool_choice] if opts[:tool_choice]

  response = grok_rest_call(
    http_method: :post,
    rest_call: 'chat/completions',
    http_body: http_body,
    timeout: opts[:timeout],
    spinner: opts[:spinner]
  )
  return nil if response.nil?

  json_resp = JSON.parse(response, symbolize_names: true)
  json_resp[:assistant_message] = json_resp.dig(:choices, 0, :message)
  json_resp
rescue StandardError => e
  raise e
end

.get_models ⇒ Object

Supported Method Parameters

models = PWN::AI::Grok.get_models



248
249
250
251
252
253
254
 
# File 'lib/pwn/ai/grok.rb', line 248

public_class_method def self.get_models
  models = grok_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



412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
 
# File 'lib/pwn/ai/grok.rb', line 412

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

    response = #{self}.chat(
      request: 'required - message to Grok',
      model: 'optional - model to use for text generation (defaults to PWN::Env[:ai][:grok][:model])',
      temp: 'optional - creative response float (defaults to PWN::Env[:ai][:grok][:temp])',
      system_role_content: 'optional - context to set up the model behavior for conversation (Default: PWN::Env[:ai][:grok][: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::Grok.obtain_oauth_bearer_token(

client_id: 'xAI OAuth Client ID',
client_secret: 'xAI OAuth Client Secret'
token_uri: 'optional - xAI OAuth token endpoint (defaults to https://auth.x.ai/oauth2/token)'

)

Internal: only invoked when oauth config (client_id etc) is present and no valid bearer_token. Constructs the authorize URL (auth.x.ai/oauth2/authorize) for the user to complete consent in browser (standard for authorization_code flow). Prompts for the code returned after redirect (OOB), then exchanges at token_uri for the bearer_token. This fulfills calling the authorize endpoint (via URL) only when oauth configured. Uses xAI’s supported scopes like grok-cli:access. Stores result in the oauth hash for the session.

Supported Method Parameters

bearer = PWN::AI::Grok.obtain_oauth_bearer_token(

client_id: 'xAI OAuth Client ID',
client_secret: 'xAI OAuth Client Secret',
token_uri: 'optional - xAI OAuth token endpoint (defaults to https://auth.x.ai/oauth2/token)'

)

Public so users can manually trigger enrollment if desired. INTERNAL default path: only invoked from grok_rest_call when oauth client_id+secret present and no bearer_token yet in the loaded PWN::Env (from pwn-vault encrypted ~/.pwn/pwn.yaml).

This is a SINGULAR ENROLLMENT process (not per-call or per-session). The resulting bearer_token (and optional refresh_token) is long-lived for xAI SuperGrok subscriptions. Once you store it in your pwn-vault config, every future ‘pwn` / PWN::Env load will have it; the guard will skip this flow entirely and use “Authorization: Bearer …” directly. (No re-prompting every time you run pwn or call PWN::AI::Grok.chat.)



45
46
47
48
49
50
51
52
53
54
55
56
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
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/pwn/ai/grok.rb', line 45

public_class_method def self.obtain_oauth_bearer_token(opts = {})
  client_id = opts[:client_id]
  client_secret = opts[:client_secret]

  scope = 'grok-cli:access'
  redirect_uri = 'urn:ietf:wg:oauth:2.0:oob'
  auth_uri = 'https://auth.x.ai/oauth2/authorize'
  token_uri = opts[:token_uri] || 'https://auth.x.ai/oauth2/token'

  # Build authorize URL -- this is the "call" to the authorize endpoint (user opens to consent)
  params = {
    client_id: client_id,
    response_type: 'code',
    scope: scope,
    redirect_uri: redirect_uri
  }
  authorize_url = "#{auth_uri}?#{URI.encode_www_form(params)}"

  puts "\n[*] OAuth ENROLLMENT for Grok (xAI SuperGrok subscription)."
  puts '    This is a ONE-TIME / SINGULAR enrollment process.'
  puts '    The bearer_token you receive is LONG-LIVED (store it once; no re-obtain every call or run).'
  puts ''
  puts '    Step 1: Open this URL in your browser and complete the authorization/consent for the grok-cli app:'
  puts "            #{authorize_url}"
  puts ''
  puts '    Step 2: After consent you will see (or be redirected to) an authorization code. Copy it exactly.'
  puts ''

  code = PWN::Plugins::AuthenticationHelper.mask_password(prompt: 'Enter the authorization code from xAI OAuth')

  # Exchange code for bearer at token endpoint.
  # Use standard confidential client auth: Authorization: Basic base64(client_id:client_secret)
  # + client_id in body (secret NOT in body).
  basic = "Basic #{Base64.strict_encode64("#{client_id}:#{client_secret}")}"
  payload = {
    grant_type: 'authorization_code',
    code: code,
    redirect_uri: redirect_uri,
    client_id: client_id
  }

  response = RestClient.post(
    token_uri,
    payload,
    {
      content_type: 'application/x-www-form-urlencoded',
      authorization: basic
    }
  )

  data = JSON.parse(response.body)

  if data['error']
    desc = data['error_description'] || data['error']
    raise "xAI OAuth token endpoint error: #{data['error']} - #{desc}"
  end

  access_token = data['access_token']

  if access_token
    opts[:bearer_token] = access_token
    opts[:refresh_token] = data['refresh_token'] if data['refresh_token']
    puts "\n[*] SUCCESS: Bearer token obtained via authorize + token exchange."
    puts '    (Cached in-memory for this Ruby process so subsequent Grok calls in the same run skip re-enrollment.)'
    puts ''
    puts '    TO MAKE THIS PERMANENT (strongly recommended -- one-time only):'
    puts '    1. Copy the bearer_token below (and refresh_token if present).'
    puts '    2. Run your pwn-vault tool (or equivalent) and store under the ai.grok.oauth section:'
    puts "         ai.grok.oauth.bearer_token = #{access_token}"
    puts "         ai.grok.oauth.refresh_token = #{data['refresh_token']}" if data['refresh_token']
    puts '    3. (Optional) You may leave or remove client_id/client_secret after storing the bearer.'
    puts '    4. Next time PWN::Env loads (pwn -Y, pwn REPL, scripts, etc.) the bearer will be present'
    puts '       from your encrypted ~/.pwn/pwn.yaml -- the guard will skip this entire flow.'
    puts '       No more browser prompts or code pasting on future uses.'
    puts ''
    puts '    The token is long-lived for your SuperGrok subscription (xAI manages expiry/refresh as needed).'
    return access_token
  end

  raise 'No access_token received from xAI OAuth token endpoint (unexpected response)'
rescue StandardError => e
  raise "Failed to obtain Grok OAuth bearer token: #{e.message}"
end