Construction d'un pipeline d'enrichissement des IOC avec OpenCTI
Aperçu
OpenCTI est une plateforme open source de gestion des connaissances en cyber-renseignement (CTI), dont le modèle de données natif est STIX 2.1. Cette compétence couvre la construction d'un pipeline automatisé d'enrichissement des IOC exploitant l'écosystème de connecteurs d'OpenCTI pour enrichir les indicateurs avec du contexte provenant de VirusTotal, Shodan, AbuseIPDB, GreyNoise et d'autres sources. Le pipeline enrichit automatiquement les indicateurs nouvellement ingérés, les corrèle avec des acteurs malveillants et des campagnes connus, et leur attribue un score pour la priorisation par les analystes.
Quand l'utiliser
- Lors du déploiement ou de la configuration des capacités de pipeline d'enrichissement des IOC dans votre environnement
- Lors de la mise en place de contrôles de sécurité alignés sur les exigences de conformité
- Lors de la construction ou de l'amélioration de l'architecture de sécurité pour ce domaine
- Lors de la réalisation d'évaluations de sécurité nécessitant cette mise en œuvre
Prérequis
- Docker et Docker Compose pour le déploiement d'OpenCTI
- Python 3.9+ avec la bibliothèque
pycti - Clés API pour les services d'enrichissement : VirusTotal, Shodan, AbuseIPDB, GreyNoise
- Compréhension du modèle de données STIX 2.1 et de ses relations
- ElasticSearch ou OpenSearch pour le backend d'OpenCTI
- RabbitMQ ou Redis pour la messagerie des connecteurs
Concepts clés
Architecture d'OpenCTI
OpenCTI utilise une API frontale GraphQL, ElasticSearch pour le stockage et Redis/RabbitMQ pour la communication entre connecteurs. Les données sont stockées en natif sous forme d'objets STIX 2.1 avec relations. Les connecteurs sont catégorisés ainsi : Importation externe (ingestion de flux), Importation interne (analyse de fichiers), Enrichissement interne (ajout de contexte) et Flux (export en temps réel).
Modèle des connecteurs d'enrichissement
Les connecteurs d'enrichissement interne sont déclenchés automatiquement lors de la création de nouveaux observables, ou manuellement par les analystes. Chaque connecteur reçoit des objets STIX, interroge des services externes et renvoie des bundles STIX 2.1 qui enrichissent l'observable original avec du contexte supplémentaire, des étiquettes et des relations.
Score de confiance
OpenCTI utilise une échelle de confiance de 0 à 100 pour les indicateurs. Les connecteurs d'enrichissement peuvent mettre à jour les scores de confiance en fonction de la validation externe : ratios de détection VirusTotal, données d'exposition Shodan, nombre de signalements AbuseIPDB et résultats de classification GreyNoise.
Workflow
Étape 1 : Déployer OpenCTI avec Docker Compose
# docker-compose.yml (services clés)
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
Étape 2 : Créer un connecteur d'enrichissement personnalisé
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 "Observable non trouvé"
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 "Enrichissement terminé"
def _enrich_ip(self, ip_address, entity_id):
"""Enrichir une adresse IP avec le contexte GreyNoise et 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"## Enrichissement GreyNoise\n"
f"- Classification : {classification}\n"
f"- Bruit internet : {noise}\n"
f"- RIOT (service légitime) : {riot}\n"
f"- Nom : {gn_data.get('name', 'N/A')}\n"
f"- Dernière vue : {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)
# Ajouter des étiquettes basées sur la classification
if classification == "malicious":
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:malicious"
)
elif riot:
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:benign-service"
)
except Exception as e:
self.helper.log_error(f"Échec de l'enrichissement GreyNoise : {e}")
return objects
def _enrich_domain(self, domain, entity_id):
"""Enrichir un domaine avec le contexte WHOIS et DNS."""
objects = []
try:
# Utiliser l'API SecurityTrails pour l'enrichissement de domaine
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"## Enrichissement SecurityTrails\n"
f"- Enregistrements A : {', '.join(a_records)}\n"
f"- Rang Alexa : {st_data.get('alexa_rank', 'N/A')}\n"
f"- Nom d'hôte : {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"Échec de l'enrichissement SecurityTrails : {e}")
return objects
if __name__ == "__main__":
connector = CustomEnrichmentConnector()