Configuration d'un Bac à Sable pour l'Analyse de Malware

malware-analysis-sandbox/
├── setup/
│   ├── virtualization-setup/
│   ├── network-isolation/
│   ├── tools-installation/
│   └── security-hardening/
├── samples/
│   ├── benign/
│   ├── suspicious/
│   └── malware/
├── analysis-scripts/
│   ├── static-analyzer.py
│   ├── dynamic-monitor.py
│   ├── ioc-extractor.py
│   └── report-generator.py
├── tools/
│   ├── yara-rules/
│   ├── config-files/
│   └── monitoring-scripts/
├── documentation/
│   ├── setup-guide.md
│   ├── safety-procedures.md
│   └── analysis-workflows.md
├── reports/
│   └── malware-analysis-template.md
└── README.md

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Machine Hôte  │    │  VM de Surveillance │    │ VM d'Analyse    │
│                 │    │  (Ubuntu)       │    │  (Windows)      │
│  VMware/VBox    │────│  - Wireshark    │────│  - Cible         │
│  - Contrôle     │    │  - tcpdump      │    │  - Malware       │
│    Réseau       │    │  - Serveur logs │    │  - Outils        │
│  - Snapshot     │    │  - Stockage IOC │    │    d'Analyse     │
└─────────────────┘    └─────────────────┘    └─────────────────┘
        │                       │                       │
        └───────────────────────┼───────────────────────┘
                                │
                        ┌─────────────────┐
                        │ Réseau Isolé    │
                        │  - Pas d'Internet│
                        │  - Services Fictifs│
                        │  - DNS Sinkhole  │
                        └─────────────────┘

# Créer une VM d'Analyse Windows 10
# - 4GB RAM minimum
# - 80GB d'espace disque
# - 2 cœurs CPU
# - Adaptateur réseau host-only

# Installer Windows 10/11
# - Installer toutes les mises à jour
# - Désactiver Windows Defender (temporairement)
# - Installer .NET Framework 3.5

# Windows PowerShell - Installer les outils communs

# Télécharger et installer Sysinternals Suite
Invoke-WebRequest -Uri "https://download.sysinternals.com/files/SysinternalsSuite.zip" -OutFile "SysinternalsSuite.zip"
Expand-Archive -Path "SysinternalsSuite.zip" -DestinationPath "C:\Tools\Sysinternals"

# Installer Sysmon
Invoke-WebRequest -Uri "https://download.sysinternals.com/files/Sysmon.zip" -OutFile "Sysmon.zip"
Expand-Archive -Path "Sysmon.zip" -DestinationPath "C:\Tools\Sysmon"
C:\Tools\Sysmon\Sysmon.exe -i -accepteula -n C:\Tools\Sysmon\sysmon.xml

# Configurer Sysmon pour la journalisation
C:\Tools\Sysmon\Sysmon.exe -c C:\Tools\Sysmon\malware-analysis-config.xml

# Configuration de la VM de surveillance Ubuntu
sudo apt update && sudo apt install -y tcpdump wireshark-common nginx

# Configurer les interfaces réseau
sudo ip link add br0 type bridge
sudo ip link set br0 up

# Configurer les services fictifs
# Serveur HTTP
python3 -m http.server 80

# Serveur FTP
sudo apt install vsftpd
sudo systemctl start vsftpd

import hashlib
import peutils
import pefile
import os
import subprocess
from pathlib import Path

class StaticAnalyzer:
    def __init__(self, file_path):
        self.file_path = file_path
        self.file_name = os.path.basename(file_path)
        self.results = {}
    
    def calculate_hashes(self):
        """Calculer plusieurs hachages de fichiers"""
        hashes = {}
        
        # MD5
        md5_hash = hashlib.md5()
        # SHA256
        sha256_hash = hashlib.sha256()
        # SHA1
        sha1_hash = hashlib.sha1()
        
        with open(self.file_path, 'rb') as f:
            chunk = f.read(8192)
            while chunk:
                md5_hash.update(chunk)
                sha256_hash.update(chunk)
                sha1_hash.update(chunk)
                chunk = f.read(8192)
        
        hashes['md5'] = md5_hash.hexdigest()
        hashes['sha256'] = sha256_hash.hexdigest()
        hashes['sha1'] = sha1_hash.hexdigest()
        
        return hashes
    
    def extract_strings(self, min_length=4):
        """Extraire les chaînes imprimables du binaire"""
        strings = []
        try:
            result = subprocess.run(['strings', '-n', str(min_length), self.file_path], 
                                  capture_output=True, text=True)
            strings = result.stdout.split('\n')
        except Exception as e:
            print(f"Erreur lors de l'extraction des chaînes: {e}")
        
        return [s.strip() for s in strings if s.strip()]
    
    def analyze_pe_headers(self):
        """Analyser les en-têtes du fichier PE"""
        try:
            pe = pefile.PE(self.file_path)
            
            pe_info = {
                'machine': pe.FILE_HEADER.Machine,
                'timestamp': pe.FILE_HEADER.TimeDateStamp,
                'characteristics': pe.FILE_HEADER.Characteristics,
                'entry_point': pe.OPTIONAL_HEADER.AddressOfEntryPoint,
                'image_base': pe.OPTIONAL_HEADER.ImageBase,
                'sections': [],
                'imports': [],
                'exports': []
            }
            
            # Informations sur les sections
            for section in pe.sections:
                pe_info['sections'].append({
                    'name': section.Name.decode('utf-8').strip('\x00'),
                    'virtual_address': section.VirtualAddress,
                    'size_of_raw_data': section.SizeOfRawData,
                    'characteristics': section.Characteristics
                })
            
            # Informations sur les imports
            if hasattr(pe, 'DIRECTORY_ENTRY_IMPORT'):
                for entry in pe.DIRECTORY_ENTRY_IMPORT:
                    dll_name = entry.dll.decode('utf-8')
                    imports = []
                    for imp in entry.imports:
                        if imp.name:
                            imports.append(imp.name.decode('utf-8'))
                    pe_info['imports'].append({'dll': dll_name, 'functions': imports})
            
            return pe_info
            
        except Exception as e:
            print(f"Erreur lors de l'analyse des en-têtes PE: {e}")
            return None
    
    def scan_with_yara(self, rules_path):
        """Scanner le fichier avec les règles YARA"""
        try:
            import yara
            rules = yara.compile(filepath=rules_path)
            matches = rules.match(self.file_path)
            
            return [{
                'rule': match.rule,
                'meta': match.meta,
                'strings': match.strings
            } for match in matches]
            
        except Exception as e:
            print(f"Erreur lors du scan avec YARA: {e}")
            return []
    
    def full_analysis(self):
        """Effectuer une analyse statique complète"""
        self.results = {
            'file_name': self.file_name,
            'file_size': os.path.getsize(self.file_path),
            'hashes': self.calculate_hashes(),
            'strings': self.extract_strings(),
            'pe_analysis': self.analyze_pe_headers(),
            'yara_matches': []
        }
        
        return self.results

import psutil
import time
import subprocess
import threading
from collections import defaultdict

class DynamicMonitor:
    def __init__(self, sample_path, analysis_duration=300):
        self.sample_path = sample_path
        self.analysis_duration = analysis_duration
        self.monitoring_active = False
        self.events = []
        
    def start_monitoring(self):
        """Démarrer la surveillance système avant l'exécution"""
        self.monitoring_active = True
        
        # Démarrer les threads de surveillance
        threads = [
            threading.Thread(target=self.monitor_processes),
            threading.Thread(target=self.monitor_network),
            threading.Thread(target=self.monitor_file_system),
            threading.Thread(target=self.monitor_registry)
        ]
        
        for thread in threads:
            thread.daemon = True
            thread.start()
        
        return threads
    
    def execute_sample(self):
        """Exécuter l'échantillon de malware"""
        try:
            # Démarrer la surveillance
            self.start_monitoring()
            time.sleep(2)  # Laisser la surveillance se stabiliser
            
            # Exécuter l'échantillon
            subprocess.Popen([self.sample_path], shell=True)
            
            # Surveiller pendant la durée spécifiée
            time.sleep(self.analysis_duration)
            
            self.monitoring_active = False
            
        except Exception as e:
            print(f"Erreur lors de l'exécution de l'échantillon: {e}")
    
    def monitor_processes(self):
        """Surveiller la création et la terminaison des processus"""
        initial_processes = {p.pid: p.info for p in psutil.process_iter(['name', 'pid', 'cmdline'])}
        
        while self.monitoring_active:
            current_processes = {p.pid: p.info for p in psutil.process_iter(['name', 'pid', 'cmdline'])}
            
            # Nouveaux processus
            for pid, info in current_processes.items():
                if pid not in initial_processes:
                    self.events.append({
                        'timestamp': time.time(),
                        'type': 'process_created',
                        'data': info
                    })
            
            # Processus terminés
            for pid in initial_processes:
                if pid not in current_processes:
                    self.events.append({
                        'timestamp': time.time(),
                        'type': 'process_terminated',
                        'data': initial_processes[pid]
                    })
            
            time.sleep(1)
    
    def monitor_network(self):
        """Surveiller les connexions réseau"""
        while self.monitoring_active:
            try:
                connections = psutil.net_connections()
                for conn in connections:
                    if conn.status == 'ESTABLISHED' and conn.raddr:
                        self.events.append({
                            'timestamp': time.time(),
                            'type': 'network_connection',
                            'data': {
                                'local_address': f"{conn.laddr.ip}:{conn.laddr.port}",
                                'remote_address': f"{conn.raddr.ip}:{conn.raddr.port}",
                                'status': conn.status,
                                'pid': conn.pid
                            }
                        })
            except Exception:
                pass
            
            time.sleep(2)
    
    def extract_iocs(self):
        """Extraire les IOC des données de surveillance"""
        iocs = {
            'processes': set(),
            'network_connections': set(),
            'file_operations': set(),
            'registry_operations': set()
        }
        
        for event in self.events:
            if event['type'] == 'process_created':
                iocs['processes'].add(event['data']['name'])
            elif event['type'] == 'network_connection':
                iocs['network_connections'].add(event['data']['remote_address'])
        
        # Convertir les ensembles en listes
        for key in iocs:
            iocs[key] = list(iocs[key])
        
        return iocs

rule Suspicious_PE_Characteristics
{
    meta:
        description = "Détecte les caractéristiques suspectes de fichiers PE"
        author = "Analyste SOC"
        date = "2026-02-12"
    
    strings:
        $suspicious_imports = {
            "CreateRemoteThread" and
            "WriteProcessMemory" and
            "VirtualAllocEx"
        }
        
        $anti_debug = "IsDebuggerPresent"
        $packing_section = ".upx"
    
    condition:
        uint16(0) == 0x5A4D and // Signature PE
        ($suspicious_imports or $anti_debug or $packing_section)
}

rule Network_Beaconing
{
    meta:
        description = "Détecte le balisage C2 potentiel dans les chaînes"
        author = "Analyste SOC"
        date = "2026-02-12"
    
    strings:
        $http_methods = /(GET|POST|PUT)\s+\/api\//
        $user_agents = /User-Agent:\s+(curl|wget|python)/
        $domains = /[a-z0-9]{20,}\.com/
    
    condition:
        any of them
}