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



820
821
822
823
824
 
# File 'lib/pwn/ai/grok.rb', line 820

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



734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
 
# File 'lib/pwn/ai/grok.rb', line 734

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: ,
    quiet: opts[:quiet]
  )
  return nil if response.nil? || response.to_s.strip.empty?

  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.



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/grok.rb', line 682

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?

  # xAI rejects Hash function.arguments / Hash content (422 map → string).
  messages = PWN::AI::Agent::Loop.openai_wire_messages(messages: messages) if defined?(PWN::AI::Agent::Loop) && PWN::AI::Agent::Loop.respond_to?(:openai_wire_messages)

  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],
    quiet: opts[:quiet]
  )
  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



502
503
504
505
506
507
508
 
# File 'lib/pwn/ai/grok.rb', line 502

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

.get_plan_usage(opts = {}) ⇒ Object

Supported Method Parameters

usage = PWN::AI::Grok.get_plan_usage( timeout: 'optional - seconds (default 8)' )

Subscription / team usage for the PS1 percent suffix. GET /v1/api-key yields team_id; then billing preview / spending-limits / prepaid balance compute used vs limit. GET /v1/me is also tried (OAuth personal-team payload).



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

public_class_method def self.get_plan_usage(opts = {})
  timeout = opts[:timeout] || 8
  return { available: false, engine: :grok } unless plan_usage_credentials?

  me = parse_plan_usage_json(
    raw: grok_rest_call(rest_call: 'me', timeout: timeout, spinner: false, quiet: true, non_interactive: true)
  )
  if me.is_a?(Hash)
    used = first_numeric(src: me, keys: %i[used usage spend spent used_usd usage_usd])
    limit = first_numeric(src: me, keys: %i[limit quota allowance hard_limit spending_limit])
    normalized = PWN::AI.normalize_plan_usage(
      used: used,
      limit: limit,
      source: 'me',
      engine: :grok
    )
    return normalized if normalized[:available]
  end

  key_info = parse_plan_usage_json(
    raw: grok_rest_call(rest_call: 'api-key', timeout: timeout, spinner: false, quiet: true, non_interactive: true)
  )
  team_id = key_info.is_a?(Hash) ? key_info[:team_id] : nil
  team_id ||= me.is_a?(Hash) ? (me[:team_id] || me.dig(:team, :id)) : nil

  if team_id.to_s.strip != ''
    preview = parse_plan_usage_json(
      raw: grok_rest_call(
        rest_call: "billing/teams/#{team_id}/postpaid/invoice/preview",
        timeout: timeout,
        spinner: false,
        quiet: true,
        non_interactive: true
      )
    )
    limits = parse_plan_usage_json(
      raw: grok_rest_call(
        rest_call: "billing/teams/#{team_id}/postpaid/spending-limits",
        timeout: timeout,
        spinner: false,
        quiet: true,
        non_interactive: true
      )
    )
    balance = parse_plan_usage_json(
      raw: grok_rest_call(
        rest_call: "billing/teams/#{team_id}/prepaid/balance",
        timeout: timeout,
        spinner: false,
        quiet: true,
        non_interactive: true
      )
    )

    used = nil
    limit = nil
    if preview.is_a?(Hash)
      core = preview[:coreInvoice] || preview[:core_invoice] || {}
      used_cents = cents_val(src: core, key: :prepaidCreditsUsed) ||
                   cents_val(src: core, key: :prepaid_credits_used) ||
                   core[:amountAfterVat] ||
                   core[:amount_after_vat]
      prepaid_cents = cents_val(src: core, key: :prepaidCredits) ||
                      cents_val(src: core, key: :prepaid_credits)
      spend_limit_cents = preview[:effectiveSpendingLimit] || preview[:effective_spending_limit]
      used = to_f_or_nil(value: used_cents)
      limit = to_f_or_nil(value: spend_limit_cents)
      limit = to_f_or_nil(value: prepaid_cents).abs if (limit.nil? || limit <= 0) && prepaid_cents
    end
    used = to_f_or_nil(value: cents_val(src: balance, key: :total)).to_f.abs if (used.nil? || used <= 0) && balance.is_a?(Hash)
    if (limit.nil? || limit <= 0) && limits.is_a?(Hash)
      sl = limits[:spendingLimits] || limits[:spending_limits] || {}
      limit = to_f_or_nil(value: cents_val(src: sl, key: :effectiveSl)) ||
              to_f_or_nil(value: cents_val(src: sl, key: :effectiveHardSl)) ||
              to_f_or_nil(value: cents_val(src: sl, key: :softSl))
    end

    normalized = PWN::AI.normalize_plan_usage(
      used: used,
      limit: limit,
      source: 'billing/teams',
      engine: :grok
    )
    return normalized if normalized[:available]
  end

  { available: false, engine: :grok }
rescue StandardError
  { available: false, engine: :grok }
end

.help ⇒ Object

Display Usage for this Module



828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
 
# File 'lib/pwn/ai/grok.rb', line 828

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

    usage = #{self}.get_plan_usage

    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