Integração de Feed de Inteligência de Ameaças

Quando Usar

Use esta habilidade quando:

• Equipes de SOC precisam de ingestão automatizada de feeds de inteligência de ameaças em plataformas SIEM
• Múltiplas fontes de TI precisam ser normalizadas em um formato comum (STIX 2.1)
• Sistemas de detecção precisam de correspondência de IOC em tempo real contra telemetria de rede e de endpoint
• Processos de avaliação de qualidade e deduplicação de feeds de TI precisam ser estabelecidos

Não use para busca manual de IOC — use ferramentas de enriquecimento dedicadas (VirusTotal, AbuseIPDB) para consultas ad-hoc.

Pré-requisitos

• Instância MISP ou Plataforma de Inteligência de Ameaças (TIP) para agregação de feeds
• Biblioteca cliente STIX/TAXII (pacotes Python taxii2-client, stix2)
• Plataforma SIEM (Splunk ES, Elastic Security ou Sentinel) com framework de TI configurado
• Chaves de API para feeds comerciais e de código aberto (AlienVault OTX, Abuse.ch, CISA AIS)
• Python 3.8+ para automação de processamento de feeds

Fluxo de Trabalho

Passo 1: Identificar e Catalogar Fontes de Inteligência

Mapeie os feeds disponíveis por tipo, formato e frequência de atualização:

Passo 2: Ingerir Feeds STIX/TAXII

Conecte-se a um servidor TAXII 2.1 e baixe os indicadores:

from taxii2client.v21 import Server, Collection
from stix2 import parse

# Connect to TAXII server (example: CISA AIS)
server = Server(
    "https://taxii.cisa.gov/taxii2/",
    user="your_username",
    password="your_password"
)

# List available collections
for api_root in server.api_roots:
    print(f"API Root: {api_root.title}")
    for collection in api_root.collections:
        print(f"  Collection: {collection.title} (ID: {collection.id})")

# Fetch indicators from a collection
collection = Collection(
    "https://taxii.cisa.gov/taxii2/collections/COLLECTION_ID/",
    user="your_username",
    password="your_password"
)

# Get indicators added in last 24 hours
from datetime import datetime, timedelta
added_after = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S.000Z")

response = collection.get_objects(added_after=added_after, type=["indicator"])
for obj in response.get("objects", []):
    indicator = parse(obj)
    print(f"Type: {indicator.type}")
    print(f"Pattern: {indicator.pattern}")
    print(f"Valid Until: {indicator.valid_until}")
    print(f"Confidence: {indicator.confidence}")
    print("---")

Passo 3: Ingerir Feeds de Código Aberto

Feed Abuse.ch URLhaus:

import requests
import csv
from io import StringIO

# Download URLhaus recent URLs
response = requests.get("https://urlhaus.abuse.ch/downloads/csv_recent/")
reader = csv.reader(StringIO(response.text), delimiter=',')

indicators = []
for row in reader:
    if row[0].startswith("#"):
        continue
    indicators.append({
        "id": row[0],
        "dateadded": row[1],
        "url": row[2],
        "url_status": row[3],
        "threat": row[5],
        "tags": row[6]
    })

print(f"Ingested {len(indicators)} URLs from URLhaus")

# Filter for active threats only
active = [i for i in indicators if i["url_status"] == "online"]
print(f"Active threats: {len(active)}")

Feed AlienVault OTX Pulse:

from OTXv2 import OTXv2, IndicatorTypes

otx = OTXv2("YOUR_OTX_API_KEY")

# Get subscribed pulses (last 24 hours)
pulses = otx.getall(modified_since="2024-03-14T00:00:00")

for pulse in pulses:
    print(f"Pulse: {pulse['name']}")
    print(f"Tags: {pulse['tags']}")
    for indicator in pulse["indicators"]:
        print(f"  IOC: {indicator['indicator']} ({indicator['type']})")

Abuse.ch Feodo Tracker (IPs C2):

response = requests.get("https://feodotracker.abuse.ch/downloads/ipblocklist_recommended.json")
c2_data = response.json()

for entry in c2_data:
    print(f"IP: {entry['ip_address']}:{entry['port']}")
    print(f"Malware: {entry['malware']}")
    print(f"First Seen: {entry['first_seen']}")
    print(f"Last Online: {entry['last_online']}")

Passo 4: Normalizar e Deduplicar

…