Threat Hunting Avanzado: Guía de Técnicas de Detección de Amenazas

El threat hunting es un enfoque proactivo de la ciberseguridad que busca identificar las amenazas antes de que causen daños. Esta guía explora las técnicas avanzadas que utilizan los expertos en threat hunting.

1. Preparación del Entorno

Configuración de las Herramientas

# Instalación de Zeek (anteriormente Bro)
sudo apt install zeek

# Configuración básica
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

Puesta en Marcha del Monitoring

# Script de monitorización del sistema
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. Técnicas de Detección

Análisis de Procesos

# Detección de procesos maliciosos
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

Análisis del Tráfico de Red

# Análisis del tráfico con 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)
    
    # Análisis de los resultados
    for ip, count in packet_counts.most_common():
        if count > 1000:  # Umbral de detección
            print(f"Trafic suspect détecté depuis {ip}: {count} paquets")

3. Análisis de Logs

Análisis de los Logs del Sistema

# Análisis de logs con 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()}")

Análisis de los Logs DNS

# Análisis de las consultas 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:  # Umbral de detección
            suspicious_domains.add(domain)
    
    return suspicious_domains

4. Detección de Malware

Análisis Estático

# Análisis de archivos con 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

Análisis Dinámico

# Análisis de comportamiento
import ctypes
import sys

def monitor_process_behavior(pid):
    suspicious_apis = [
        'CreateRemoteThread',
        'VirtualAlloc',
        'WriteProcessMemory',
        'ShellExecute'
    ]
    
    def hook_api(api_name):
        # Implementación del hook
        pass
    
    for api in suspicious_apis:
        hook_api(api)

5. Investigación Forense

Análisis de la Memoria

# Script de análisis de memoria
import volatility.plugins.common as common
import volatility.plugins.taskmods as taskmods

def analyze_memory(memory_file):
    # Configuración de Volatility
    config = common.Config()
    config.add_option('FILE', short_option='f', default=memory_file)
    
    # Análisis de los procesos
    pslist = taskmods.PSList(config)
    for proc in pslist.calculate():
        print(f"Processus trouvé: {proc.pid} - {proc.name}")

Análisis del Disco

# Script de análisis de disco
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. Automatización y Orquestación

Pipeline de Detección

# 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

Integración con SIEM

# Integración con 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. Visualización y Reporting

Dashboard de Monitoring

# Creación de un dashboard con 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'
            )
        )
    ])

Generación de Informes

# Generación de informes
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')

Conclusión

El threat hunting es una disciplina esencial que requiere:

• Un conocimiento profundo de los sistemas
• Herramientas especializadas
• Una metodología rigurosa
• Una documentación precisa
• Una vigilancia constante

Puntos clave que recordar:

• Automatizar las tareas repetitivas
• Mantener una base de conocimientos
• Colaborar con los equipos
• Documentar los hallazgos
• Actualizar las herramientas con regularidad