Processamento de Feeds STIX/TAXII

Quando Usar

Use esta habilidade quando:

• Estiver integrando uma nova coleção TAXII 2.1 de um feed governamental (CISA AIS, FS-ISAC) ou provedor comercial
• Estiver validando se os pacotes STIX ingeridos estão em conformidade com a especificação OASIS STIX 2.1 antes da importação
• Estiver construindo pipelines automatizados que analisam objetos de relacionamento STIX para reconstruir o contexto de campanhas

Não use esta habilidade para formatos de feed de fornecedores proprietários (JSON do Recorded Future, listas de IOCs do CrowdStrike) que exigem analisadores específicos do fornecedor em vez de processamento STIX.

Pré-requisitos

• Python 3.9+ com a biblioteca stix2 (pip install stix2) e a biblioteca taxii2-client
• Acesso à rede ao endpoint do servidor TAXII 2.1 com credenciais válidas
• TIP ou SIEM de destino com API de importação (MISP, OpenCTI ou Splunk ES)

Fluxo de Trabalho

Etapa 1: Descobrir Coleções do Servidor TAXII

from taxii2client.v21 import Server, as_pages

server = Server("https://cti.example.com/taxii/",
                user="apiuser", password="apikey")
api_root = server.api_roots[0]
for collection in api_root.collections:
    print(collection.id, collection.title, collection.can_read)

Selecione coleções relevantes para o seu perfil de ameaças. O CISA AIS fornece coleções segmentadas por setor (financeiro, energia, saúde).

Etapa 2: Buscar Pacotes STIX com Paginação

from taxii2client.v21 import Collection
from datetime import datetime, timedelta, timezone

collection = Collection(
    "https://cti.example.com/taxii/api1/collections/<id>/objects/",
    user="apiuser", password="apikey")

# Fetch only objects added in the last 24 hours
added_after = datetime.now(timezone.utc) - timedelta(hours=24)
for bundle_page in as_pages(collection.get_objects,
                             added_after=added_after, per_request=100):
    process_bundle(bundle_page)

Etapa 3: Analisar e Validar Objetos STIX

import stix2

def process_bundle(bundle_dict):
    bundle = stix2.parse(bundle_dict, allow_custom=True)
    for obj in bundle.objects:
        if obj.type == "indicator":
            validate_indicator(obj)
        elif obj.type == "threat-actor":
            upsert_threat_actor(obj)
        elif obj.type == "relationship":
            link_objects(obj)

def validate_indicator(indicator):
    required = ["id", "type", "spec_version", "created",
                "modified", "pattern", "pattern_type", "valid_from"]
    for field in required:
        if not hasattr(indicator, field):
            raise ValueError(f"Missing required field: {field}")
    # Check confidence range
    if hasattr(indicator, "confidence"):
        assert 0 <= indicator.confidence <= 100

Etapa 4: Roteamento de Objetos para Plataformas Consumidoras

Mapeie os tipos de objetos STIX para os sistemas de destino:

• Objetos indicator → Tabelas de consulta do SIEM e listas de bloqueio de firewall
• Objetos malware → Biblioteca de inteligência de ameaças do EDR
• Objetos threat-actor / campaign → TIP para contexto de analista
• Objetos course-of-action → Wiki da equipe de segurança ou gatilhos de playbook SOAR

Use definições de marcação TLP para impor restrições de compartilhamento:

for marking in obj.get("object_marking_refs", []):
    if "tlp-red" in marking:
        route_to_restricted_platform_only(obj)

Etapa 5: Publicar de Volta para TAXII (Compartilhamento Bidirecional)

# Add validated local intelligence back to shared collection
new_indicator = stix2.Indicator(
    name="Malicious C2 Domain",
    pattern="[domain-name:value = 'evil-c2.example.com']",
    pattern_type="stix",
    valid_from="2025-01-15T00:00:00Z",
    confidence=80,
    labels=["malicious-activity"],
    object_marking_refs=["marking-definition--34098fce-860f-479c-ae..."]  # TLP:GREEN
)
collection.add_objects(stix2.Bundle(new_indicator))

Conceitos Chave

…