الصيد الاستباقي للتهديدات المتقدم: دليل تقنيات اكتشاف التهديدات

يُعدّ الصيد الاستباقي للتهديدات (threat hunting) نهجًا استباقيًا في الأمن السيبراني يهدف إلى تحديد التهديدات قبل أن تُلحق الضرر. يستعرض هذا الدليل التقنيات المتقدمة التي يستخدمها خبراء الصيد الاستباقي للتهديدات.

1. تحضير البيئة

إعداد الأدوات

# Installation de Zeek (anciennement Bro)
sudo apt install zeek

# Configuration de base
cat > /usr/local/zeek/share/zeek/site/local.zeek << EOF
@load frameworks/files/extract-all-files
@load frameworks/communication/listen
@load frameworks/communication/seen
@load frameworks/communication/notice
@load frameworks/communication/notice/actions/email
EOF

إعداد المراقبة

# Script de monitoring système
import psutil
import time
from datetime import datetime

def monitor_system():
    while True:
        metrics = {
            'cpu_percent': psutil.cpu_percent(),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_usage': psutil.disk_usage('/').percent,
            'network_connections': len(psutil.net_connections()),
            'timestamp': datetime.now().isoformat()
        }
        print(metrics)
        time.sleep(60)

2. تقنيات الاكتشاف

تحليل العمليات

# Détection de processus malveillants
import psutil
import re

def detect_malicious_processes():
    suspicious_patterns = [
        r'cryptominer',
        r'miner',
        r'botnet',
        r'backdoor'
    ]
    
    for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
        try:
            cmdline = ' '.join(proc.info['cmdline'])
            for pattern in suspicious_patterns:
                if re.search(pattern, cmdline, re.I):
                    print(f"Processus suspect détecté: {proc.info['pid']} - {proc.info['name']}")
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

تحليل حركة مرور الشبكة

# Analyse du trafic avec Scapy
from scapy.all import *
import collections

def analyze_network_traffic():
    packet_counts = collections.Counter()
    
    def packet_callback(packet):
        if IP in packet:
            packet_counts[packet[IP].src] += 1
    
    sniff(prn=packet_callback, store=0, timeout=60)
    
    # Analyse des résultats
    for ip, count in packet_counts.most_common():
        if count > 1000:  # Seuil de détection
            print(f"Trafic suspect détecté depuis {ip}: {count} paquets")

3. تحليل السجلات

تحليل سجلات النظام

# Analyse des logs avec Python
import re
from datetime import datetime, timedelta

def analyze_system_logs():
    suspicious_patterns = {
        'failed_login': r'Failed password for',
        'sudo_attempt': r'sudo:.*authentication failure',
        'port_scan': r'port scan detected'
    }
    
    with open('/var/log/auth.log', 'r') as f:
        for line in f:
            for pattern_name, pattern in suspicious_patterns.items():
                if re.search(pattern, line):
                    print(f"Événement suspect détecté: {pattern_name} - {line.strip()}")

تحليل سجلات DNS

# Analyse des requêtes DNS
import dnslib
from collections import defaultdict

def analyze_dns_queries():
    query_counts = defaultdict(int)
    suspicious_domains = set()
    
    def process_dns_query(query):
        domain = str(query.qname)
        query_counts[domain] += 1
        
        if query_counts[domain] > 1000:  # Seuil de détection
            suspicious_domains.add(domain)
    
    return suspicious_domains

4. اكتشاف البرمجيات الخبيثة

التحليل الساكن

# Analyse de fichiers avec YARA
import yara

def analyze_file(file_path):
    rules = yara.compile(source='''
        rule MaliciousFile {
            strings:
                $a = "CreateRemoteThread" nocase
                $b = "VirtualAlloc" nocase
                $c = "WriteProcessMemory" nocase
            condition:
                any of them
        }
    ''')
    
    matches = rules.match(file_path)
    return matches

التحليل الديناميكي

# Analyse comportementale
import ctypes
import sys

def monitor_process_behavior(pid):
    suspicious_apis = [
        'CreateRemoteThread',
        'VirtualAlloc',
        'WriteProcessMemory',
        'ShellExecute'
    ]
    
    def hook_api(api_name):
        # Implémentation du hook
        pass
    
    for api in suspicious_apis:
        hook_api(api)

5. التحقيق الجنائي الرقمي

تحليل الذاكرة

# Script d'analyse mémoire
import volatility.plugins.common as common
import volatility.plugins.taskmods as taskmods

def analyze_memory(memory_file):
    # Configuration de Volatility
    config = common.Config()
    config.add_option('FILE', short_option='f', default=memory_file)
    
    # Analyse des processus
    pslist = taskmods.PSList(config)
    for proc in pslist.calculate():
        print(f"Processus trouvé: {proc.pid} - {proc.name}")

تحليل القرص

# Script d'analyse disque
import os
import hashlib
from datetime import datetime

def analyze_disk():
    known_malware_hashes = set()
    
    def calculate_file_hash(file_path):
        sha256_hash = hashlib.sha256()
        with open(file_path, "rb") as f:
            for byte_block in iter(lambda: f.read(4096), b""):
                sha256_hash.update(byte_block)
        return sha256_hash.hexdigest()
    
    for root, dirs, files in os.walk('/'):
        for file in files:
            try:
                file_path = os.path.join(root, file)
                file_hash = calculate_file_hash(file_path)
                if file_hash in known_malware_hashes:
                    print(f"Malware détecté: {file_path}")
            except (PermissionError, FileNotFoundError):
                continue

6. الأتمتة والتنسيق

خط أنابيب الاكتشاف

# Pipeline de threat hunting
from typing import List, Dict
import json

class ThreatHuntingPipeline:
    def __init__(self):
        self.detectors = []
        self.analyzers = []
    
    def add_detector(self, detector):
        self.detectors.append(detector)
    
    def add_analyzer(self, analyzer):
        self.analyzers.append(analyzer)
    
    def run(self, data: Dict) -> List[Dict]:
        alerts = []
        for detector in self.detectors:
            if detector.detect(data):
                alerts.append(detector.get_alert())
        
        for alert in alerts:
            for analyzer in self.analyzers:
                analyzer.analyze(alert)
        
        return alerts

التكامل مع SIEM

# Intégration avec Splunk
from splunklib.client import Service
import json

def send_to_siem(event_data):
    service = Service(
        host='splunk.example.com',
        port=8089,
        username='admin',
        password='password'
    )
    
    index = service.indexes['threat_hunting']
    index.submit(json.dumps(event_data))

7. التصور المرئي وإعداد التقارير

لوحة معلومات المراقبة

# Création de dashboard avec Dash
from dash import Dash, html, dcc
import plotly.express as px

app = Dash(__name__)

def create_dashboard():
    app.layout = html.Div([
        html.H1('Dashboard de Threat Hunting'),
        dcc.Graph(
            id='alerts-graph',
            figure=px.line(
                alerts_data,
                x='timestamp',
                y='count',
                title='Alertes par Heure'
            )
        )
    ])

إنشاء التقارير

# Génération de rapports
from jinja2 import Template
import pdfkit

def generate_report(alerts, findings):
    template = Template('''
        <html>
            <head>
                <title>Rapport de Threat Hunting</title>
                <style>
                    body { font-family: Arial; }
                    .alert { color: red; }
                    .finding { color: orange; }
                </style>
            </head>
            <body>
                <h1>Rapport de Threat Hunting</h1>
                <h2>Alertes</h2>
                {% for alert in alerts %}
                <div class="alert">
                    {{ alert }}
                </div>
                {% endfor %}
                <h2>Découvertes</h2>
                {% for finding in findings %}
                <div class="finding">
                    {{ finding }}
                </div>
                {% endfor %}
            </body>
        </html>
    ''')
    
    html_content = template.render(alerts=alerts, findings=findings)
    pdfkit.from_string(html_content, 'threat_hunting_report.pdf')

الخلاصة

يُعدّ الصيد الاستباقي للتهديدات (threat hunting) تخصصًا أساسيًا يتطلب:

• معرفة معمّقة بالأنظمة
• أدوات متخصصة
• منهجية صارمة
• توثيقًا دقيقًا
• يقظة مستمرة

أهم النقاط التي يجب تذكّرها:

• أتمتة المهام المتكررة
• الحفاظ على قاعدة معرفية
• التعاون مع الفرق
• توثيق الاكتشافات
• تحديث الأدوات بانتظام