Implementando Integração de Feed STIX/TAXII

Visão Geral

STIX (Structured Threat Information eXpression) e TAXII (Trusted Automated eXchange of Intelligence Information) são padrões abertos do OASIS para representar e transportar inteligência de ameaças cibernéticas. Esta habilidade abrange a implementação de um consumidor e produtor de feed STIX/TAXII 2.1 usando Python, configurando a descoberta de servidor TAXII, gerenciamento de coleções, sondagem de novas inteligências, análise de objetos STIX 2.1 e integração de feeds em plataformas SIEM e TIP.

Quando Usar

• Ao implantar ou configurar a implementação de capacidades de integração de feed stix taxii em seu ambiente
• Ao estabelecer controles de segurança alinhados aos requisitos de conformidade
• Ao construir ou aprimorar a arquitetura de segurança para este domínio
• Ao realizar avaliações de segurança que exigem esta implementação

Pré-requisitos

• Python 3.9+ com taxii2-client, stix2, cti-taxii-client bibliotecas
• Compreensão do modelo de dados STIX 2.1 (SDOs, SCOs, SROs)
• Compreensão do protocolo TAXII 2.1 (descoberta, raízes de API, coleções)
• Acesso à rede a servidores TAXII (MITRE ATT&CK TAXII, Anomali STAXX)
• Opcional: Medallion para executar um servidor TAXII 2.1 local

Conceitos Chave

Arquitetura TAXII 2.1

O TAXII define uma API RESTful com três tipos de serviço:

• Descoberta: Retorna informações sobre raízes de API disponíveis
• Raiz de API: Contém coleções e serve como o principal ponto de interação
• Coleção: Um agrupamento lógico de objetos STIX acessíveis via GET/POST

Modelo de Objeto STIX 2.1

Os objetos STIX são categorizados como:

• SDOs (STIX Domain Objects): Indicador, Malware, Ameaça Ativa, Campanha, Padrão de Ataque, Ferramenta, Infraestrutura, Vulnerabilidade, Identidade, Localização, Nota, Opinião, Relatório, Agrupamento
• SCOs (STIX Cyber Observables): IPv4-Addr, Domain-Name, URL, File, Email-Addr, Process, Network-Traffic, Artifact
• SROs (STIX Relationship Objects): Relationship, Sighting
• Meta Objects: Marking Definition (TLP), Language Content, Extension Definition

STIX Bundle

Um Bundle é uma coleção de objetos STIX transmitidos juntos. Bundles têm um ID único e contêm um array de objetos. Coleções TAXII servem bundles em resposta a requisições GET.

Fluxo de Trabalho

Etapa 1: Descoberta do Servidor TAXII

from taxii2client.v21 import Server, Collection, as_pages

# Connect to MITRE ATT&CK TAXII server
server = Server("https://cti-taxii.mitre.org/taxii2/", user="", password="")

print(f"Title: {server.title}")
print(f"Description: {server.description}")

# List API roots
for api_root in server.api_roots:
    print(f"\nAPI Root: {api_root.title}")
    print(f"  URL: {api_root.url}")

    # List collections
    for collection in api_root.collections:
        print(f"  Collection: {collection.title} (ID: {collection.id})")
        print(f"    Can Read: {collection.can_read}")
        print(f"    Can Write: {collection.can_write}")

Etapa 2: Buscar Objetos STIX da Coleção

from taxii2client.v21 import Collection, as_pages
import json

# Connect to Enterprise ATT&CK collection
ENTERPRISE_ATTACK_ID = "95ecc380-afe9-11e4-9b6c-751b66dd541e"
collection = Collection(
    f"https://cti-taxii.mitre.org/stix/collections/{ENTERPRISE_ATTACK_ID}/",
    user="",
    password="",
)

print(f"Collection: {collection.title}")

# Fetch all objects (paginated)
all_objects = []
for envelope in as_pages(collection.get_objects, per_request=50):
    objects = envelope.get("objects", [])
    all_objects.extend(objects)
    print(f"  Fetched {len(objects)} objects (total: {len(all_objects)})")

print(f"\nTotal objects retrieved: {len(all_objects)}")

# Categorize by type
type_counts = {}
for obj in all_objects:
    obj_type = obj.get("type", "unknown")
    type_counts[obj_type] = type_counts.get(obj_type, 0) + 1

for obj_type, count in sorted(type_counts.items()):
    print(f"  {obj_type}: {count}")

Etapa 3: Analisar Objetos STIX 2.1 com a Biblioteca stix2

from stix2 import parse, Filter, MemoryStore

# Carregar objetos em um MemoryStore para consulta
store = MemoryStore(stix_data=all_objects)

# Consultar todos os indicadores
indicators = store.query([Filter("type", "=", "indicator")])
print(f"Indicadores: {len(indicators)}")

for ind in indicators[:5]:
    print(f"  {ind.name}: {ind.pattern}")

# Consultar malware
malware_list = store.query([Filter("type", "=", "malware")])
print(f"\nFamílias de malware: {len(malware_list)}")

# Consultar ameaças ativas

…