building-ioc-enrichment-pipeline-with-opencti

Construindo Pipeline de Enriquecimento de IOC com OpenCTI

Visão Geral

O OpenCTI é uma plataforma de código aberto para gerenciar inteligência de ameaças cibernéticas, construída sobre o STIX 2.1 como modelo de dados nativo. Este skill cobre a construção de um pipeline automatizado de enriquecimento de IOC usando o ecossistema de conectores do OpenCTI para enriquecer indicadores com contexto de VirusTotal, Shodan, AbuseIPDB, GreyNoise e outras fontes. O pipeline enriquece automaticamente indicadores recém-ingestados, correlaciona-os com atores e campanhas de ameaças conhecidas e os pontua para priorização do analista.

Quando Usar

• Ao implantar ou configurar capacidades de pipeline de enriquecimento de IOC no seu ambiente
• Ao estabelecer controles de segurança alinhados a requisitos de conformidade
• Ao construir ou melhorar a arquitetura de segurança para este domínio
• Ao conduzir avaliações de segurança que exigem esta implementação

Pré-requisitos

• Docker e Docker Compose para implantação do OpenCTI
• Python 3.9+ com a biblioteca pycti
• Chaves de API para serviços de enriquecimento: VirusTotal, Shodan, AbuseIPDB, GreyNoise
• Compreensão do modelo de dados STIX 2.1 e seus relacionamentos
• ElasticSearch ou OpenSearch para o backend do OpenCTI
• RabbitMQ ou Redis para mensagens dos conectores

Conceitos Chave

Arquitetura do OpenCTI

O OpenCTI usa um frontend de API GraphQL com backend ElasticSearch para armazenamento e Redis/RabbitMQ para comunicação entre conectores. Os dados são armazenados nativamente como objetos STIX 2.1 com relacionamentos. Os conectores são categorizados como: Importação Externa (ingestão de feeds), Importação Interna (análise de arquivos), Enriquecimento Interno (adição de contexto) e Stream (exportação em tempo real).

Modelo de Conector de Enriquecimento

Conectores de enriquecimento interno são acionados automaticamente quando novos observáveis são criados ou manualmente por analistas. Cada conector recebe objetos STIX, consulta serviços externos e retorna pacotes STIX 2.1 que aumentam o observável original com contexto adicional, etiquetas e relacionamentos.

Pontuação de Confiança

O OpenCTI usa uma escala de confiança de 0 a 100 para indicadores. Conectores de enriquecimento podem atualizar pontuações de confiança com base em validação externa: taxas de detecção do VirusTotal, dados de exposição do Shodan, contagens de relatórios do AbuseIPDB e resultados de classificação do GreyNoise.

Fluxo de Trabalho

Passo 1: Implantar OpenCTI com Docker Compose

# docker-compose.yml (serviços principais)
version: '3'
services:
  opencti:
    image: opencti/platform:6.4.4
    environment:
      - APP__PORT=8080
      - [email protected]
      - APP__ADMIN__PASSWORD=ChangeMeNow
      - APP__ADMIN__TOKEN=your-admin-token-uuid
      - ELASTICSEARCH__URL=http://elasticsearch:9200
      - MINIO__ENDPOINT=minio
      - RABBITMQ__HOSTNAME=rabbitmq
    ports:
      - "8080:8080"
    depends_on:
      - elasticsearch
      - minio
      - rabbitmq
      - redis

  connector-virustotal:
    image: opencti/connector-virustotal:6.4.4
    environment:
      - OPENCTI_URL=http://opencti:8080
      - OPENCTI_TOKEN=your-admin-token-uuid
      - CONNECTOR_ID=connector-virustotal-id
      - CONNECTOR_NAME=VirusTotal
      - CONNECTOR_SCOPE=StixFile,Artifact,IPv4-Addr,Domain-Name,Url
      - CONNECTOR_AUTO=true
      - VIRUSTOTAL_TOKEN=your-vt-api-key
      - VIRUSTOTAL_MAX_TLP=TLP:AMBER

  connector-shodan:
    image: opencti/connector-shodan:6.4.4
    environment:
      - OPENCTI_URL=http://opencti:8080
      - OPENCTI_TOKEN=your-admin-token-uuid
      - CONNECTOR_ID=connector-shodan-id
      - CONNECTOR_NAME=Shodan
      - CONNECTOR_SCOPE=IPv4-Addr
      - CONNECTOR_AUTO=true
      - SHODAN_TOKEN=your-shodan-api-key
      - SHODAN_MAX_TLP=TLP:AMBER

  connector-abuseipdb:
    image: opencti/connector-abuseipdb:6.4.4
    environment:
      - OPENCTI_URL=http://opencti:8080
      - OPENCTI_TOKEN=your-admin-token-uuid
      - CONNECTOR_ID=connector-abuseipdb-id
      - CONNECTOR_NAME=AbuseIPDB
      - CONNECTOR_SCOPE=IPv4-Addr
      - CONNECTOR_AUTO=true
      - ABUSEIPDB_API_KEY=your-abuseipdb-key

Passo 2: Construir Conector de Enriquecimento Personalizado

import os
from pycti import OpenCTIConnectorHelper, get_config_variable
from stix2 import (
    Bundle, Indicator, Note, Relationship,
    IPv4Address, DomainName
)
import requests


class CustomEnrichmentConnector:
    def __init__(self):
        config = {
            "opencti": {
                "url": os.environ.get("OPENCTI_URL"),
                "token": os.environ.get("OPENCTI_TOKEN"),
            },
            "connector": {
                "id": os.environ.get("CONNECTOR_ID"),
                "name": "CustomEnrichment",
                "scope": "IPv4-Addr,Domain-Name,Url",
                "auto": True,
                "type": "INTERNAL_ENRICHMENT",
            },
        }
        self.helper = OpenCTIConnectorHelper(config)
        self.helper.listen(self._process_message)

    def _process_message(self, data):
        entity_id = data["entity_id"]
        stix_object = self.helper.api.stix_cyber_observable.read(id=entity_id)

        if not stix_object:
            return "Observável não encontrado"

        observable_type = stix_object["entity_type"]
        observable_value = stix_object.get("value", "")

        enrichment_results = []

        if observable_type == "IPv4-Addr":
            enrichment_results = self._enrich_ip(observable_value, entity_id)
        elif observable_type == "Domain-Name":
            enrichment_results = self._enrich_domain(observable_value, entity_id)

        if enrichment_results:
            bundle = Bundle(objects=enrichment_results, allow_custom=True)
            self.helper.send_stix2_bundle(bundle.serialize())

        return "Enriquecimento concluído"

    def _enrich_ip(self, ip_address, entity_id):
        """Enriquecer endereço IP com contexto GreyNoise e AbuseIPDB."""
        objects = []

        # API GreyNoise Community
        try:
            gn_response = requests.get(
                f"https://api.greynoise.io/v3/community/{ip_address}",
                headers={"key": os.environ.get("GREYNOISE_API_KEY")},
                timeout=30,
            )
            if gn_response.status_code == 200:
                gn_data = gn_response.json()
                classification = gn_data.get("classification", "unknown")
                noise = gn_data.get("noise", False)
                riot = gn_data.get("riot", False)

                note_content = (
                    f"## Enriquecimento GreyNoise\n"
                    f"- Classificação: {classification}\n"
                    f"- Ruído na Internet: {noise}\n"
                    f"- RIOT (Serviço Benigno): {riot}\n"
                    f"- Nome: {gn_data.get('name', 'N/A')}\n"
                    f"- Última Visualização: {gn_data.get('last_seen', 'N/A')}"
                )

                note = Note(
                    content=note_content,
                    object_refs=[entity_id],
                    abstract=f"GreyNoise: {classification}",
                    allow_custom=True,
                )
                objects.append(note)

                # Adicionar etiquetas com base na classificação
                if classification == "malicious":
                    self.helper.api.stix_cyber_observable.add_label(
                        id=entity_id, label_name="greynoise:malicioso"
                    )
                elif riot:
                    self.helper.api.stix_cyber_observable.add_label(
                        id=entity_id, label_name="greynoise:servico-benigno"
                    )

        except Exception as e:
            self.helper.log_error(f"Falha no enriquecimento GreyNoise: {e}")

        return objects

    def _enrich_domain(self, domain, entity_id):
        """Enriquecer domínio com contexto WHOIS e DNS."""
        objects = []

        try:
            # Usar API SecurityTrails para enriquecimento de domínio
            st_response = requests.get(
                f"https://api.securitytrails.com/v1/domain/{domain}",
                headers={"APIKEY": os.environ.get("SECURITYTRAILS_API_KEY")},
                timeout=30,
            )
            if st_response.status_code == 200:
                st_data = st_response.json()
                current_dns = st_data.get("current_dns", {})

                a_records = [
                    r.get("ip") for r in current_dns.get("a", {}).get("values", [])
                ]

                note_content = (
                    f"## Enriquecimento SecurityTrails\n"
                    f"- Registros A: {', '.join(a_records)}\n"
                    f"- Alexa Rank: {st_data.get('alexa_rank', 'N/A')}\n"
                    f"- Hostname: {st_data.get('hostname', 'N/A')}"
                )

                note = Note(
                    content=note_content,
                    object_refs=[entity_id],
                    abstract=f"SecurityTrails: {domain}",
                    allow_custom=True,
                )
                objects.append(note)

        except Exception as e:
            self.helper.log_error(f"Falha no enriquecimento SecurityTrails: {e}")

        return objects


if __name__ == "__main__":
    connector = CustomEnrichmentConnector()