Construindo Enriquecimento de Inteligência de Ameaças no Splunk

Visão Geral

O Framework de Inteligência de Ameaças do Splunk no Enterprise Security permite que equipes de SOC correlacionem automaticamente indicadores de comprometimento (IOCs) contra eventos de segurança. O framework ingere feeds de ameaças, normaliza indicadores em coleções do KV Store e usa buscas de correlação baseadas em lookup para sinalizar eventos correspondentes. O Splunk Threat Intelligence Management centraliza a coleta, normalização e enriquecimento de múltiplas fontes, reduzindo o tempo de triagem ao fornecer aos analistas contexto imediato.

Quando Usar

• Ao implantar ou configurar capacidades de construção de enriquecimento de inteligência de ameaças no Splunk em seu ambiente
• Ao estabelecer controles de segurança alinhados aos requisitos de conformidade
• Ao construir ou melhorar a arquitetura de segurança para este domínio
• Ao realizar avaliações de segurança que exigem esta implementação

Pré-requisitos

• Splunk Enterprise Security (ES) 7.x ou posterior
• Add-on Threat Intelligence Management ou Framework de Inteligência de Ameaças
• Chaves de API para feeds externos de inteligência de ameaças (MISP, OTX, VirusTotal, AbuseIPDB)
• KV Store habilitado e configurado corretamente
• Acesso de administrador para configuração de entrada modular

Arquitetura do Framework de Inteligência de Ameaças

External TI Sources (STIX/TAXII, CSV, API)
    |
    v
Modular Inputs (download and parse feeds)
    |
    v
KV Store Collections (normalized IOC storage)
    |-- ip_intel
    |-- domain_intel
    |-- file_intel
    |-- url_intel
    |-- email_intel
    |
    v
Threat Intelligence Lookups
    |
    v
Correlation Searches (match events against IOCs)
    |
    v
Notable Events (enriched with TI context)

Configurando Fontes de Inteligência de Ameaças

Integração de Feed STIX/TAXII

# inputs.conf - TAXII feed configuration
[threatlist://taxii_feed_example]
description = TAXII 2.1 Threat Feed
type = taxii
url = https://threatfeed.example.com/taxii2/
collection = threat-indicators-v21
polling_interval = 3600
api_key = <encrypted_api_key>
disabled = false

Lista de Ameaças Baseada em CSV

# inputs.conf - CSV threat list
[threatlist://custom_blocklist]
description = Internal threat blocklist
type = csv
url = https://internal.company.com/threat-feeds/blocklist.csv
polling_interval = 1800
disabled = false

Entrada Modular Personalizada para Feeds Baseados em API

# bin/threatfeed_otx.py - Coletor de feed OTX AlienVault
import json
import sys
import requests
from splunklib.modularinput import Script, Scheme, Argument, Event


class OTXFeedInput(Script):
    def get_scheme(self):
        scheme = Scheme("OTX AlienVault Feed")
        scheme.description = "Coleta IOCs do AlienVault OTX"
        scheme.use_external_validation = False
        scheme.streaming_mode = Scheme.streaming_mode_xml

        api_key_arg = Argument("api_key")
        api_key_arg.data_type = Argument.data_type_string
        api_key_arg.required_on_create = True
        scheme.add_argument(api_key_arg)

        pulse_days_arg = Argument("pulse_days")
        pulse_days_arg.data_type = Argument.data_type_number
        pulse_days_arg.required_on_create = False
        scheme.add_argument(pulse_days_arg)

        return scheme

    def stream_events(self, inputs, ew):
        for input_name, input_item in inputs.inputs.items():
            api_key = input_item["api_key"]
            pulse_days = int(input_item.get("pulse_days", 30))

            headers = {"X-OTX-API-KEY": api_key}
            url = f"https://otx.alienvault.com/api/v1/pulses/subscribed?modified_since={pulse_days}d"

            try:
                response = requests.get(url, headers=headers, timeout=60)
                response.raise_for_status()
                data = response.json()

                for pulse in data.get("results", []):
                    for indicator in pulse.get("indicators", []):
                        event = Event()
                        event.stanza = input_name
                        event.data = json.dumps({
                            "indicator": indicator["indicator"],
                            "type": indicator["type"],
                            "pulse_name": pulse["name"],
                            "pulse_id": pulse["id"],
                            "description": indicator.get("description", ""),
                            "created": indicator.get("created", ""),
                            "threat_source": "OTX",

…