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: https://docs.x.ai/docs Obtain an API key from https://x.ai/api

Constant Summary collapse

XAI_OAUTH_ISSUER =

xAI Grok OAuth (SuperGrok) -- public-client OIDC, no client_secret.

auth.x.ai is a full OIDC provider (/.well-known/openid-configuration). xAI ships a PUBLIC client for the Grok CLI; the same client_id is used by NousResearch/hermes-agent (hermes auth add xai-oauth) and is reused here so PWN can obtain a Bearer for https://api.x.ai/v1 without an API key. token_endpoint_auth_methods_supported includes 'none', so no client_secret is required -- only PKCE / device_code.

Two grants are implemented:

* RFC 8628 device_authorization  -> default; headless / SSH friendly.
* refresh_token                  -> silent renewal once enrolled.

Access tokens are short-lived JWTs (ES256, exp claim). The refresh_token is long-lived -- persist it via pwn-vault under ai.grok.oauth.refresh_token and PWN::AI::Grok refreshes transparently on every run.

'https://auth.x.ai'
XAI_OAUTH_DEVICE_URI = 
"#{XAI_OAUTH_ISSUER}/oauth2/device/code".freeze
XAI_OAUTH_TOKEN_URI = 
"#{XAI_OAUTH_ISSUER}/oauth2/token".freeze
XAI_OAUTH_CLIENT_ID =

Public Grok-CLI client_id (same one hermes-agent uses).

'b1a00492-073a-47ea-816f-4c329264a828'
XAI_OAUTH_SCOPE = 
'openid profile email offline_access grok-cli:access api:access'

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com



643
644
645
646
647
 
# File 'lib/pwn/ai/grok.rb', line 643

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[:grok][:model])', temp: 'optional - creative response float (deafults to PWN::Env[:grok][:temp])', system_role_content: 'optional - context to set up the model behavior for conversation (Default: PWN::Env[: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)' )



559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
 
# File 'lib/pwn/ai/grok.rb', line 559

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 [function:{...}]', tool_choice: 'optional - "auto" | "none" | "required" | function:{name:..}', model: 'optional - overrides PWN::Env[:grok][:model]', temp: 'optional - temperature (defaults to PWN::Env[: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.



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
537
538
539
540
541
542
543
544
545
 
# File 'lib/pwn/ai/grok.rb', line 511

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



487
488
489
490
491
492
493
 
# File 'lib/pwn/ai/grok.rb', line 487

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



651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
 
# File 'lib/pwn/ai/grok.rb', line 651

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: 'optional - xAI OAuth Client ID (defaults to public Grok-CLI client)', scope: 'optional - space-delimited scopes (defaults to XAI_OAUTH_SCOPE)', timeout: 'optional - seconds to wait for user consent (default 300)' )

Runs the RFC 8628 OAuth 2.0 Device Authorization Grant against auth.x.ai using xAI's public Grok-CLI client (no client_secret). This is the same identity path hermes auth add xai-oauth uses -- a SuperGrok / X Premium+ subscription on the account is what grants api:access at consent time.

1. POST /oauth2/device/code    -> device_code, user_code, verification_uri
2. User opens verification_uri_complete in a browser and approves.
3. Poll POST /oauth2/token (grant_type=device_code) until access_token.

On success the access_token + refresh_token are written back into the passed opts/oauth Hash (so PWN::Env[:grok][:oauth] is live-cached) and the operator is told exactly what to persist via pwn-vault.



235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
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
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
 
# File 'lib/pwn/ai/grok.rb', line 235

public_class_method def self.obtain_oauth_bearer_token(opts = {})
  client_id = real_config_value?(value: opts[:client_id]) ? opts[:client_id] : XAI_OAUTH_CLIENT_ID
  scope     = real_config_value?(value: opts[:scope])     ? opts[:scope]     : XAI_OAUTH_SCOPE
  token_uri = real_config_value?(value: opts[:token_uri]) ? opts[:token_uri] : XAI_OAUTH_TOKEN_URI
  timeout   = (opts[:timeout] || 300).to_i

  # -- Step 1: request device + user code -------------------------------
  dev = JSON.parse(
    RestClient.post(
      XAI_OAUTH_DEVICE_URI,
      { client_id: client_id, scope: scope },
      content_type: 'application/x-www-form-urlencoded',
      accept: 'application/json'
    ).body
  )
  raise "xAI device_code error: #{dev['error']} - #{dev['error_description']}" if dev['error']

  device_code   = dev['device_code']
  user_code     = dev['user_code']
  verify_uri    = dev['verification_uri_complete'] || dev['verification_uri']
  interval      = (dev['interval'] || 5).to_i
  expires_in    = (dev['expires_in'] || timeout).to_i
  deadline      = Time.now.to_i + [expires_in, timeout].min

  puts "\n[*] xAI Grok OAuth -- Device Authorization (RFC 8628, public client, no secret)"
  puts '    A SuperGrok / X Premium+ subscription on the approving account is required.'
  puts ''
  puts '    Step 1: In a browser (any device), open:'
  puts "            #{verify_uri}"
  puts "    Step 2: Confirm the code matches:  #{user_code}"
  puts '    Step 3: Approve access for "Grok CLI".'
  puts ''
  puts "    Waiting for approval (polling every #{interval}s, timeout #{deadline - Time.now.to_i}s)..."

  # -- Step 2: poll the token endpoint ---------------------------------
  data = nil
  loop do
    sleep interval
    begin
      tok = RestClient.post(
        token_uri,
        {
          grant_type: 'urn:ietf:params:oauth:grant-type:device_code',
          device_code: device_code,
          client_id: client_id
        },
        content_type: 'application/x-www-form-urlencoded',
        accept: 'application/json'
      )
      data = JSON.parse(tok.body)
    rescue RestClient::ExceptionWithResponse => e
      data = begin
        JSON.parse(e.response.body)
      rescue StandardError
        { 'error' => "http_#{e.http_code}", 'error_description' => e.response&.body }
      end
    end

    case data['error']
    when nil
      break # success
    when 'authorization_pending'
      raise 'xAI OAuth device flow timed out waiting for user approval.' if Time.now.to_i >= deadline

      next
    when 'slow_down'
      interval += 5
      next
    when 'access_denied'
      raise 'xAI OAuth device flow: user denied the authorization request.'
    when 'expired_token'
      raise 'xAI OAuth device flow: device_code expired before approval; re-run enrollment.'
    else
      raise "xAI OAuth device flow error: #{data['error']} - #{data['error_description']}"
    end
  end

  access_token  = data['access_token']
  refresh_token = data['refresh_token']
  raise 'xAI 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: xAI Grok OAuth bearer obtained via device_code grant."
  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.grok.oauth.refresh_token = #{refresh_token}" if refresh_token
  puts "      ai.grok.oauth.bearer_token  = #{access_token}"
  puts '    On future runs the refresh_token alone is enough -- PWN::AI::Grok 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 Grok OAuth bearer token (HTTP #{e.http_code}): #{e.response&.body}"
rescue StandardError => e
  raise "Failed to obtain Grok OAuth bearer token: #{e.message}"
end

.refresh_oauth_bearer_token(opts = {}) ⇒ Object

Supported Method Parameters

access_token = PWN::AI::Grok.refresh_oauth_bearer_token( refresh_token: 'required - xAI OAuth refresh_token', client_id: 'optional - defaults to public Grok-CLI client', token_uri: 'optional - defaults to https://auth.x.ai/oauth2/token' )

Exchanges a refresh_token for a fresh access_token at auth.x.ai. On success, writes :bearer_token (and a rotated :refresh_token if returned) back into the passed opts/oauth Hash so the live PWN::Env stays warm for the rest of the process. Also mirrors those values onto PWN::Env[:grok][:oauth] when that Hash is available, then attempts to re-encrypt the updated tokens into ~/.pwn/pwn.yaml when the matching decryptor (key + iv) is present.



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
 
# File 'lib/pwn/ai/grok.rb', line 94

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] : XAI_OAUTH_CLIENT_ID
  token_uri = real_config_value?(value: opts[:token_uri]) ? opts[:token_uri] : XAI_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'
  )
  data = JSON.parse(resp.body)
  raise "xAI OAuth refresh error: #{data['error']} - #{data['error_description']}" 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']

  # Always keep the live session Env warm, even when +opts+ is a copy
  # rather than the object identity of PWN::Env[:ai][:grok][:oauth].
  sync_oauth_into_env(oauth: opts)
  persist_oauth_to_vault(oauth: opts)

  data['access_token']
rescue RestClient::ExceptionWithResponse => e
  raise "xAI OAuth refresh failed (HTTP #{e.http_code}): #{e.response&.body}"
end