sandbox_setup.py

#!/usr/bin/env python3"""Sandbox Setup for Trojan AnalysisThis script sets up an isolated sandbox environment for safely analyzingsuspicious files and potential trojans.""" import osimport sysimport jsonimport timeimport argparseimport subprocessimport platformimport shutilimport tempfileimport hashlibfrom datetime import datetimefrom pathlib import Pathfrom colorama import init, Fore, Style init(autoreset=True) class SandboxSetup:    """    Sandbox Environment Setup for Malware Analysis    """        def __init__(self, sandbox_dir="/sandbox", network_isolation=True,                 snapshot_enabled=True, log_file="sandbox.log"):        """        Initialize Sandbox Setup                Args:            sandbox_dir: Sandbox directory            network_isolation: Enable network isolation            snapshot_enabled: Enable VM snapshots            log_file: Log file path        """        self.sandbox_dir = sandbox_dir        self.network_isolation = network_isolation        self.snapshot_enabled = snapshot_enabled        self.log_file = log_file                self.system = platform.system()        self.sandbox_ready = False                # Create sandbox directory        os.makedirs(sandbox_dir, exist_ok=True)                # Sandbox components        self.components = {            'virtualization': None,            'network': None,            'monitoring': None,            'isolation': None        }                # Statistics        self.stats = {            'samples_analyzed': 0,            'suspicious_files': 0,            'network_connections': 0,            'start_time': time.time()        }                # Setup logging        self.setup_logging()        def setup_logging(self):        """Setup logging"""        import logging        logging.basicConfig(            level=logging.INFO,            format='%(asctime)s - %(levelname)s - %(message)s',            handlers=[                logging.FileHandler(self.log_file),                logging.StreamHandler()            ]        )        self.logger = logging.getLogger(__name__)        def check_prerequisites(self):        """Check system prerequisites"""        print(f"\n{Fore.CYAN}[*] Checking prerequisites...{Style.RESET_ALL}")                checks = []                # Check virtualization support        if self.system == 'Linux':            # Check KVM support            if os.path.exists('/dev/kvm'):                checks.append(('KVM', True))                self.components['virtualization'] = 'kvm'            else:                checks.append(('KVM', False))                # Check Docker        if shutil.which('docker'):            checks.append(('Docker', True))            if not self.components['virtualization']:                self.components['virtualization'] = 'docker'        else:            checks.append(('Docker', False))                # Check VirtualBox        if shutil.which('vboxmanage') or shutil.which('VBoxManage'):            checks.append(('VirtualBox', True))            self.components['virtualization'] = 'virtualbox'        else:            checks.append(('VirtualBox', False))                # Check network isolation tools        if shutil.which('iptables'):            checks.append(('iptables', True))            self.components['network'] = 'iptables'        else:            checks.append(('iptables', False))                # Check monitoring tools        monitoring_tools = ['strace', 'ltrace', 'tcpdump', 'wireshark']        available = []        for tool in monitoring_tools:            if shutil.which(tool):                available.append(tool)                if available:            checks.append(('Monitoring tools', True))            self.components['monitoring'] = available        else:            checks.append(('Monitoring tools', False))                # Display results        for check, status in checks:            status_str = f"{Fore.GREEN}✓{Style.RESET_ALL}" if status else f"{Fore.RED}✗{Style.RESET_ALL}"            print(f"  {status_str} {check}")                return all(status for _, status in checks if _ != 'Monitoring tools')  # Monitoring tools optional        def setup_network_isolation(self):        """Setup network isolation"""        if not self.network_isolation:            print(f"{Fore.YELLOW}[!] Network isolation disabled{Style.RESET_ALL}")            return True                print(f"\n{Fore.CYAN}[*] Setting up network isolation...{Style.RESET_ALL}")                if self.system == 'Linux':            try:                # Create network namespace                subprocess.run(['sudo', 'ip', 'netns', 'add', 'sandbox_ns'], check=True)                                # Create virtual Ethernet pair                subprocess.run(['sudo', 'ip', 'link', 'add', 'veth0', 'type', 'veth', 'peer', 'name', 'veth1'], check=True)                                # Move one end to namespace                subprocess.run(['sudo', 'ip', 'link', 'set', 'veth1', 'netns', 'sandbox_ns'], check=True)                                # Configure interfaces                subprocess.run(['sudo', 'ip', 'addr', 'add', '192.168.100.1/24', 'dev', 'veth0'], check=True)                subprocess.run(['sudo', 'ip', 'link', 'set', 'veth0', 'up'], check=True)                                subprocess.run(['sudo', 'ip', 'netns', 'exec', 'sandbox_ns', 'ip', 'addr', 'add', '192.168.100.2/24', 'dev', 'veth1'], check=True)                subprocess.run(['sudo', 'ip', 'netns', 'exec', 'sandbox_ns', 'ip', 'link', 'set', 'veth1', 'up'], check=True)                subprocess.run(['sudo', 'ip', 'netns', 'exec', 'sandbox_ns', 'ip', 'link', 'set', 'lo', 'up'], check=True)                                # Block external internet (allow only local network)                subprocess.run(['sudo', 'ip', 'netns', 'exec', 'sandbox_ns', 'iptables', '-P', 'INPUT', 'DROP'], check=True)                subprocess.run(['sudo', 'ip', 'netns', 'exec', 'sandbox_ns', 'iptables', '-P', 'FORWARD', 'DROP'], check=True)                subprocess.run(['sudo', 'ip', 'netns', 'exec', 'sandbox_ns', 'iptables', '-P', 'OUTPUT', 'DROP'], check=True)                                # Allow local network                subprocess.run(['sudo', 'ip', 'netns', 'exec', 'sandbox_ns', 'iptables', '-A', 'INPUT', '-s', '192.168.100.0/24', '-j', 'ACCEPT'], check=True)                subprocess.run(['sudo', 'ip', 'netns', 'exec', 'sandbox_ns', 'iptables', '-A', 'OUTPUT', '-d', '192.168.100.0/24', '-j', 'ACCEPT'], check=True)                                print(f"{Fore.GREEN}[✓] Network namespace created and isolated{Style.RESET_ALL}")                return True                            except subprocess.CalledProcessError as e:                print(f"{Fore.RED}[✗] Error setting up network isolation: {e}{Style.RESET_ALL}")                return False                elif self.system == 'Windows':            print(f"{Fore.YELLOW}[!] Network isolation on Windows requires Hyper-V or manual setup{Style.RESET_ALL}")            return False        def setup_docker_sandbox(self):        """Setup Docker-based sandbox"""        print(f"\n{Fore.CYAN}[*] Setting up Docker sandbox...{Style.RESET_ALL}")                try:            # Pull analysis image            subprocess.run(['docker', 'pull', 'remnux/remnux-distro'], check=True)                        # Create sandbox container            container_name = f"sandbox_{datetime.now().strftime('%Y%m%d_%H%M%S')}"                        docker_args = [                'docker', 'run', '-d',                '--name', container_name,                '--cap-drop=ALL',  # Drop all capabilities                '--security-opt=no-new-privileges',                '--read-only',  # Read-only root filesystem                '-v', f"{self.sandbox_dir}:/sandbox:ro",  # Mount samples read-only            ]                        # Add network isolation            if self.network_isolation:                docker_args.append('--network=none')                        docker_args.append('remnux/remnux-distro')            docker_args.append('tail')            docker_args.append('-f')            docker_args.append('/dev/null')                        subprocess.run(docker_args, check=True)                        print(f"{Fore.GREEN}[✓] Docker sandbox created: {container_name}{Style.RESET_ALL}")            self.components['isolation'] = f"docker:{container_name}"            return True                    except subprocess.CalledProcessError as e:            print(f"{Fore.RED}[✗] Error setting up Docker sandbox: {e}{Style.RESET_ALL}")            return False        def setup_virtualbox_sandbox(self):        """Setup VirtualBox sandbox"""        print(f"\n{Fore.CYAN}[*] Setting up VirtualBox sandbox...{Style.RESET_ALL}")                vboxmanage = shutil.which('VBoxManage') or shutil.which('vboxmanage')        if not vboxmanage:            print(f"{Fore.RED}[✗] VirtualBox not found{Style.RESET_ALL}")            return False                try:            # Create VM            vm_name = f"Sandbox_{datetime.now().strftime('%Y%m%d')}"                        # Create VM            subprocess.run([vboxmanage, 'createvm', '--name', vm_name, '--register'], check=True)                        # Configure VM            subprocess.run([vboxmanage, 'modifyvm', vm_name, '--memory', '2048'], check=True)            subprocess.run([vboxmanage, 'modifyvm', vm_name, '--cpus', '2'], check=True)                        # Network configuration            if self.network_isolation:                subprocess.run([vboxmanage, 'modifyvm', vm_name, '--nic1', 'hostonly'], check=True)                subprocess.run([vboxmanage, 'modifyvm', vm_name, '--hostonlyadapter1', 'vboxnet0'], check=True)            else:                subprocess.run([vboxmanage, 'modifyvm', vm_name, '--nic1', 'nat'], check=True)                        # Disable unnecessary features            subprocess.run([vboxmanage, 'modifyvm', vm_name, '--clipboard', 'disabled'], check=True)            subprocess.run([vboxmanage, 'modifyvm', vm_name, '--draganddrop', 'disabled'], check=True)            subprocess.run([vboxmanage, 'modifyvm', vm_name, '--audio', 'none'], check=True)            subprocess.run([vboxmanage, 'modifyvm', vm_name, '--usb', 'off'], check=True)                        # Create snapshot for easy revert            if self.snapshot_enabled:                subprocess.run([vboxmanage, 'snapshot', vm_name, 'take', 'clean_state'], check=True)                        print(f"{Fore.GREEN}[✓] VirtualBox sandbox created: {vm_name}{Style.RESET_ALL}")            self.components['isolation'] = f"virtualbox:{vm_name}"            return True                    except subprocess.CalledProcessError as e:            print(f"{Fore.RED}[✗] Error setting up VirtualBox sandbox: {e}{Style.RESET_ALL}")            return False        def setup_monitoring(self):        """Setup monitoring tools"""        print(f"\n{Fore.CYAN}[*] Setting up monitoring...{Style.RESET_ALL}")                monitor_dir = os.path.join(self.sandbox_dir, 'monitoring')        os.makedirs(monitor_dir, exist_ok=True)                # Create monitoring script        monitor_script = os.path.join(monitor_dir, 'monitor.sh')        with open(monitor_script, 'w') as f:            f.write("""#!/bin/bash# Sandbox Monitoring ScriptMONITOR_DIR="$(dirname "$0")"PID=$1# Monitor system callsif command -v strace &> /dev/null; then    sudo strace -p $PID -o "$MONITOR_DIR/strace.log" &fi# Monitor network trafficif command -v tcpdump &> /dev/null; then    sudo tcpdump -i any -w "$MONITOR_DIR/network.pcap" &fi# Monitor file changesinotifywait -m -r --format '%w%f' / 2>/dev/null | while read file; do    echo "$(date): $file" >> "$MONITOR_DIR/file_changes.log"done &wait""")                os.chmod(monitor_script, 0o755)                # Create analysis script        analysis_script = os.path.join(monitor_dir, 'analyze.sh')        with open(analysis_script, 'w') as f:            f.write("""#!/bin/bash# Sandbox Analysis ScriptANALYSIS_DIR="$(dirname "$0")"echo "=== Sandbox Analysis Report ==="echo "Date: $(date)"echo# Check for suspicious system callsif [ -f "$ANALYSIS_DIR/strace.log" ]; then    echo "=== Suspicious System Calls ==="    grep -E "socket|connect|execve|ptrace" "$ANALYSIS_DIR/strace.log"    echofi# Check for network connectionsif [ -f "$ANALYSIS_DIR/network.pcap" ]; then    echo "=== Network Connections ==="    tcpdump -r "$ANALYSIS_DIR/network.pcap" -n 2>/dev/null | head -20    echofi# Check for file changesif [ -f "$ANALYSIS_DIR/file_changes.log" ]; then    echo "=== File System Changes ==="    cat "$ANALYSIS_DIR/file_changes.log" | head -20    echofiecho "=== Analysis Complete ==="""")                os.chmod(analysis_script, 0o755)                print(f"{Fore.GREEN}[✓] Monitoring setup complete in {monitor_dir}{Style.RESET_ALL}")        self.components['monitoring'] = monitor_dir                return True        def prepare_sample_directory(self):        """Prepare directory for samples"""        samples_dir = os.path.join(self.sandbox_dir, 'samples')        results_dir = os.path.join(self.sandbox_dir, 'results')                os.makedirs(samples_dir, exist_ok=True)        os.makedirs(results_dir, exist_ok=True)                print(f"{Fore.GREEN}[✓] Sample directory: {samples_dir}{Style.RESET_ALL}")        print(f"{Fore.GREEN}[✓] Results directory: {results_dir}{Style.RESET_ALL}")                return samples_dir, results_dir        def create_sandbox_config(self):        """Create sandbox configuration file"""        config = {            'sandbox_id': datetime.now().strftime('%Y%m%d_%H%M%S'),            'created': datetime.now().isoformat(),            'components': self.components,            'settings': {                'network_isolation': self.network_isolation,                'snapshot_enabled': self.snapshot_enabled,                'sandbox_dir': self.sandbox_dir            },            'stats': self.stats        }                config_file = os.path.join(self.sandbox_dir, 'sandbox_config.json')        with open(config_file, 'w') as f:            json.dump(config, f, indent=2)                print(f"{Fore.GREEN}[✓] Sandbox configuration saved to {config_file}{Style.RESET_ALL}")                return config_file        def analyze_sample(self, sample_path):        """Analyze a sample in the sandbox"""        if not os.path.exists(sample_path):            print(f"{Fore.RED}[✗] Sample not found: {sample_path}{Style.RESET_ALL}")            return None                # Copy sample to sandbox        sample_name = os.path.basename(sample_path)        sandbox_sample = os.path.join(self.sandbox_dir, 'samples',                                       f"{int(time.time())}_{sample_name}")        shutil.copy2(sample_path, sandbox_sample)                # Calculate hash        with open(sandbox_sample, 'rb') as f:            file_hash = hashlib.sha256(f.read()).hexdigest()                print(f"\n{Fore.CYAN}[*] Analyzing sample: {sample_name}{Style.RESET_ALL}")        print(f"{Fore.CYAN}    SHA256: {file_hash}{Style.RESET_ALL}")                self.stats['samples_analyzed'] += 1                # Analysis depends on virtualization type        if 'docker:' in str(self.components.get('isolation', '')):            container = self.components['isolation'].split(':')[1]                        try:                # Copy sample to container                subprocess.run([                    'docker', 'cp', sandbox_sample,                    f"{container}:/tmp/{sample_name}"                ], check=True)                                # Run monitoring                monitor_dir = self.components.get('monitoring')                if monitor_dir:                    # This is simplified - real analysis would be more complex                    print(f"{Fore.YELLOW}[!] Sample copied to container for analysis{Style.RESET_ALL}")                    print(f"{Fore.YELLOW}[!] Manual analysis required in container{Style.RESET_ALL}")                                return {                    'sample': sample_name,                    'hash': file_hash,                    'sandbox_path': sandbox_sample,                    'container': container                }                            except Exception as e:                print(f"{Fore.RED}[✗] Error analyzing sample: {e}{Style.RESET_ALL}")                return None                else:            print(f"{Fore.YELLOW}[!] Virtualization not available - manual analysis required{Style.RESET_ALL}")            return {                'sample': sample_name,                'hash': file_hash,                'sandbox_path': sandbox_sample            }        def cleanup(self):        """Clean up sandbox resources"""        print(f"\n{Fore.CYAN}[*] Cleaning up sandbox...{Style.RESET_ALL}")                # Remove network namespace        if self.network_isolation and self.system == 'Linux':            try:                subprocess.run(['sudo', 'ip', 'netns', 'delete', 'sandbox_ns'], check=True)                subprocess.run(['sudo', 'ip', 'link', 'delete', 'veth0'], check=True)                print(f"{Fore.GREEN}[✓] Network namespace removed{Style.RESET_ALL}")            except:                pass                # Remove Docker containers        if 'docker:' in str(self.components.get('isolation', '')):            container = self.components['isolation'].split(':')[1]            try:                subprocess.run(['docker', 'stop', container], check=True)                subprocess.run(['docker', 'rm', container], check=True)                print(f"{Fore.GREEN}[✓] Docker container removed{Style.RESET_ALL}")            except:                pass                print(f"{Fore.GREEN}[✓] Cleanup complete{Style.RESET_ALL}")        def setup(self):        """Complete sandbox setup"""        print(f"""        {Fore.CYAN}╔═══════════════════════════════════════╗        ║        Sandbox Setup Tool           ║        ║      Trojan Analysis Environment    ║        ╚═══════════════════════════════════════╝{Style.RESET_ALL}        """)                print(f"{Fore.YELLOW}[!] WARNING: This sets up an isolated environment{Style.RESET_ALL}")        print(f"{Fore.YELLOW}[!] Only use for legitimate security research{Style.RESET_ALL}\n")                # Check prerequisites        if not self.check_prerequisites():            print(f"{Fore.RED}[✗] Prerequisites not met{Style.RESET_ALL}")            return False                # Setup network isolation        if self.network_isolation:            self.setup_network_isolation()                # Setup virtualization        if self.components['virtualization'] == 'docker':            self.setup_docker_sandbox()        elif self.components['virtualization'] == 'virtualbox':            self.setup_virtualbox_sandbox()        else:            print(f"{Fore.YELLOW}[!] No virtualization available - using host with restrictions{Style.RESET_ALL}")                # Setup monitoring        self.setup_monitoring()                # Prepare directories        self.prepare_sample_directory()                # Create config        self.create_sandbox_config()                self.sandbox_ready = True                print(f"\n{Fore.GREEN}[✓] Sandbox setup complete!{Style.RESET_ALL}")        print(f"{Fore.GREEN}    Sandbox directory: {self.sandbox_dir}{Style.RESET_ALL}")                return True        def interactive_menu(self):        """Interactive sandbox menu"""        if not self.sandbox_ready:            print(f"{Fore.YELLOW}[!] Sandbox not set up. Run setup first.{Style.RESET_ALL}")            return                while True:            print(f"\n{Fore.CYAN}{'='*50}{Style.RESET_ALL}")            print(f"{Fore.CYAN}SANDBOX MENU{Style.RESET_ALL}")            print(f"{Fore.CYAN}{'='*50}{Style.RESET_ALL}")            print("1. Analyze sample")            print("2. View monitoring logs")            print("3. Run analysis report")            print("4. Clean up sandbox")            print("5. View statistics")            print("0. Exit")                        choice = input(f"\n{Fore.YELLOW}Enter choice: {Style.RESET_ALL}")                        if choice == '1':                sample = input("Enter path to sample file: ")                result = self.analyze_sample(sample)                if result:                    print(f"{Fore.GREEN}[✓] Sample prepared for analysis{Style.RESET_ALL}")                        elif choice == '2':                monitor_dir = self.components.get('monitoring')                if monitor_dir and os.path.exists(monitor_dir):                    subprocess.run(['ls', '-la', monitor_dir])                else:                    print(f"{Fore.YELLOW}[!] No monitoring logs found{Style.RESET_ALL}")                        elif choice == '3':                analysis_script = os.path.join(self.components.get('monitoring', ''), 'analyze.sh')                if os.path.exists(analysis_script):                    subprocess.run(['bash', analysis_script])                else:                    print(f"{Fore.YELLOW}[!] Analysis script not found{Style.RESET_ALL}")                        elif choice == '4':                self.cleanup()                self.sandbox_ready = False                break                        elif choice == '5':                print(f"\n{Fore.CYAN}📊 Statistics:{Style.RESET_ALL}")                print(f"  Samples Analyzed: {self.stats['samples_analyzed']}")                print(f"  Suspicious Files: {self.stats['suspicious_files']}")                print(f"  Network Connections: {self.stats['network_connections']}")                print(f"  Uptime: {int(time.time() - self.stats['start_time'])} seconds")                        elif choice == '0':                break def main():    parser = argparse.ArgumentParser(description='Sandbox Setup for Trojan Analysis')    parser.add_argument('--dir', default='/sandbox', help='Sandbox directory')    parser.add_argument('--no-network', action='store_true', help='Disable network isolation')    parser.add_argument('--no-snapshot', action='store_true', help='Disable snapshots')    parser.add_argument('--setup', action='store_true', help='Run sandbox setup')    parser.add_argument('--interactive', action='store_true', help='Interactive mode')    parser.add_argument('--analyze', help='Analyze a sample file')        args = parser.parse_args()        sandbox = SandboxSetup(        sandbox_dir=args.dir,        network_isolation=not args.no_network,        snapshot_enabled=not args.no_snapshot    )        if args.setup:        sandbox.setup()        if args.analyze:        if not sandbox.sandbox_ready:            sandbox.setup()        sandbox.analyze_sample(args.analyze)        if args.interactive:        if not sandbox.sandbox_ready:            sandbox.setup()        sandbox.interactive_menu()        if not any([args.setup, args.analyze, args.interactive]):        print("Use --setup to create sandbox or --interactive for menu") if __name__ == "__main__":    main()