Threat Hunting Avançado: Guia de Técnicas de Detecção de Ameaças

O threat hunting é uma abordagem proativa de cibersegurança que visa identificar ameaças antes que elas causem danos. Este guia explora as técnicas avançadas utilizadas por especialistas em threat hunting.

1. Preparação do Ambiente

Configuração das Ferramentas

# Instalação do Zeek (anteriormente Bro)
sudo apt install zeek

# Configuração 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

Implantação do Monitoramento

# Script de monitoramento do 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 Detecção

Análise de Processos

# Detecção de processos 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"Processo suspeito detectado: {proc.info['pid']} - {proc.info['name']}")
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

Análise do Tráfego de Rede

# Análise de tráfego com 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álise dos resultados
    for ip, count in packet_counts.most_common():
        if count > 1000:  # Limite de detecção
            print(f"Tráfego suspeito detectado a partir de {ip}: {count} pacotes")

3. Análise de Logs

Análise dos Logs do Sistema

# Análise de logs com 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"Evento suspeito detectado: {pattern_name} - {line.strip()}")

Análise dos Logs de DNS

# Análise de 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:  # Limite de detecção
            suspicious_domains.add(domain)
    
    return suspicious_domains

4. Detecção de Malware

Análise Estática

# Análise de arquivos com 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álise Dinâmica

# Análise comportamental
import ctypes
import sys

def monitor_process_behavior(pid):
    suspicious_apis = [
        'CreateRemoteThread',
        'VirtualAlloc',
        'WriteProcessMemory',
        'ShellExecute'
    ]
    
    def hook_api(api_name):
        # Implementação do hook
        pass
    
    for api in suspicious_apis:
        hook_api(api)

5. Investigação Forense

Análise de Memória

# Script de análise de memória
import volatility.plugins.common as common
import volatility.plugins.taskmods as taskmods

def analyze_memory(memory_file):
    # Configuração do Volatility
    config = common.Config()
    config.add_option('FILE', short_option='f', default=memory_file)
    
    # Análise dos processos
    pslist = taskmods.PSList(config)
    for proc in pslist.calculate():
        print(f"Processo encontrado: {proc.pid} - {proc.name}")

Análise de Disco

# Script de análise 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 detectado: {file_path}")
            except (PermissionError, FileNotFoundError):
                continue

6. Automação e Orquestração

Pipeline de Detecção

# 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

Integração com SIEM

# Integração com o 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. Visualização e Relatórios

Dashboard de Monitoramento

# Criação de dashboard com 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='Alertas por Hora'
            )
        )
    ])

Geração de Relatórios

# Geração de relatórios
from jinja2 import Template
import pdfkit

def generate_report(alerts, findings):
    template = Template('''
        <html>
            <head>
                <title>Relatório de Threat Hunting</title>
                <style>
                    body { font-family: Arial; }
                    .alert { color: red; }
                    .finding { color: orange; }
                </style>
            </head>
            <body>
                <h1>Relatório de Threat Hunting</h1>
                <h2>Alertas</h2>
                {% for alert in alerts %}
                <div class="alert">
                    {{ alert }}
                </div>
                {% endfor %}
                <h2>Descobertas</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')

Conclusão

O threat hunting é uma disciplina essencial que exige:

• Um conhecimento aprofundado dos sistemas
• Ferramentas especializadas
• Uma metodologia rigorosa
• Uma documentação precisa
• Uma atualização constante sobre o cenário de ameaças

Pontos-chave a reter:

• Automatizar as tarefas repetitivas
• Manter uma base de conhecimento
• Colaborar com as equipes
• Documentar as descobertas
• Atualizar as ferramentas regularmente