building-ioc-enrichment-pipeline-with-opencti

name	building-ioc-enrichment-pipeline-with-opencti
description	OpenCTI is an open-source platform for managing cyber threat intelligence knowledge, built on STIX 2.1 as its native data model. This skill covers building an automated IOC enrichment pipeline using O
domain	cybersecurity
subdomain	threat-intelligence
tags	["threat-intelligence","cti","ioc","mitre-attack","stix","opencti","enrichment","virustotal"]
version	1.0
author	mahipal
license	Apache-2.0

Building IOC Enrichment Pipeline with OpenCTI

Overview

OpenCTI is an open-source platform for managing cyber threat intelligence knowledge, built on STIX 2.1 as its native data model. This skill covers building an automated IOC enrichment pipeline using OpenCTI's connector ecosystem to enrich indicators with context from VirusTotal, Shodan, AbuseIPDB, GreyNoise, and other sources. The pipeline automatically enriches newly ingested indicators, correlates them with known threat actors and campaigns, and scores them for analyst prioritization.

Prerequisites

• Docker and Docker Compose for OpenCTI deployment
• Python 3.9+ with pycti library
• API keys for enrichment services: VirusTotal, Shodan, AbuseIPDB, GreyNoise
• Understanding of STIX 2.1 data model and relationships
• ElasticSearch or OpenSearch for OpenCTI backend
• RabbitMQ or Redis for connector messaging

Key Concepts

OpenCTI Architecture

OpenCTI uses a GraphQL API frontend backed by ElasticSearch for storage and Redis/RabbitMQ for connector communication. Data is natively stored as STIX 2.1 objects with relationships. Connectors are categorized as: External Import (feed ingestion), Internal Import (file parsing), Internal Enrichment (context addition), and Stream (real-time export).

Enrichment Connector Model

Internal enrichment connectors are triggered automatically when new observables are created or manually by analysts. Each connector receives STIX objects, queries external services, and returns STIX 2.1 bundles that augment the original observable with additional context, labels, and relationships.

Confidence Scoring

OpenCTI uses a 0-100 confidence scale for indicators. Enrichment connectors can update confidence scores based on external validation: VirusTotal detection ratios, Shodan exposure data, AbuseIPDB report counts, and GreyNoise classification results.

Practical Steps

Step 1: Deploy OpenCTI with Docker Compose

# docker-compose.yml (key services)
version: '3'
services:
  opencti:
    image: opencti/platform:6.4.4
    environment:
      - APP__PORT=8080
      - APP__ADMIN__EMAIL=[e-mail]
      - 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
       
       

  
     
    
       
       
       
       
       
       
       
       

  
     
    
       
       
       
       
       
       
       

Step 2: Build Custom Enrichment Connector

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 not found"

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

        enrichment_results = []

         observable_type == :
            enrichment_results = ._enrich_ip(observable_value, entity_id)
         observable_type == :
            enrichment_results = ._enrich_domain(observable_value, entity_id)

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

         

     ():
        
        objects = []

        
        :
            gn_response = requests.get(
                ,
                headers={: os.environ.get()},
                timeout=,
            )
             gn_response.status_code == :
                gn_data = gn_response.json()
                classification = gn_data.get(, )
                noise = gn_data.get(, )
                riot = gn_data.get(, )

                note_content = (
                    
                    
                    
                    
                    
                    
                )

                note = Note(
                    content=note_content,
                    object_refs=[entity_id],
                    abstract=,
                    allow_custom=,
                )
                objects.append(note)

                
                 classification == :
                    .helper.api.stix_cyber_observable.add_label(
                        =entity_id, label_name=
                    )
                 riot:
                    .helper.api.stix_cyber_observable.add_label(
                        =entity_id, label_name=
                    )

         Exception  e:
            .helper.log_error()

         objects

     ():
        
        objects = []

        :
            
            st_response = requests.get(
                ,
                headers={: os.environ.get()},
                timeout=,
            )
             st_response.status_code == :
                st_data = st_response.json()
                current_dns = st_data.get(, {})

                a_records = [
                    r.get()  r  current_dns.get(, {}).get(, [])
                ]

                note_content = (
                    
                    
                    
                    
                )

                note = Note(
                    content=note_content,
                    object_refs=[entity_id],
                    abstract=,
                    allow_custom=,
                )
                objects.append(note)

         Exception  e:
            .helper.log_error()

         objects


 __name__ == :
    connector = CustomEnrichmentConnector()