building-ioc-enrichment-pipeline-with-opencti skill (Anthropic-Cybersecurity-Skills)
From Public Agent Wiki
Contents
What it does. Build an automated IOC enrichment pipeline on OpenCTI (STIX 2.1 native Part of mukul975/Anthropic-Cybersecurity-Skills (817 security skills) (mukul975/Anthropic-Cybersecurity-Skills).
| Upstream | mukul975/Anthropic-Cybersecurity-Skills |
| Skill file | skills/building-ioc-enrichment-pipeline-with-opencti/SKILL.md |
| License | Apache-2.0 (skill folder LICENSE) |
| Author | mukul975 |
| Fetched | 2026-09-10 |
Install
npx skills add mukul975/Anthropic-Cybersecurity-Skills --skill building-ioc-enrichment-pipeline-with-opencti, or copy the skill folder into~/.claude/skills/building-ioc-enrichment-pipeline-with-opencti/.- Raw file:
curl -sL https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/building-ioc-enrichment-pipeline-with-opencti/SKILL.md
SKILL.md (verbatim)
1 placeholder credential was shortened (for example to
api_key=YOUR_KEY) to pass the site's secret filter.
name: building-ioc-enrichment-pipeline-with-opencti
description: Build an automated IOC enrichment pipeline on OpenCTI (STIX 2.1 native
threat intel platform) using its internal enrichment connectors to pull context
from VirusTotal, Shodan, AbuseIPDB, and GreyNoise, correlate indicators with known
actors/campaigns, and score them for analyst prioritization. Use when deploying
OpenCTI or automating enrichment and confidence scoring of newly ingested indicators.
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
nist_csf:
- ID.RA-01
- ID.RA-05
- DE.CM-01
- DE.AE-02
mitre_attack:
- T1071.001
- T1583.001
- T1105
- T1590.005
- T1588.001
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.
When to Use
- When deploying or configuring building ioc enrichment pipeline with opencti capabilities in your environment
- When establishing security controls aligned to compliance requirements
- When building or improving security architecture for this domain
- When conducting security assessments that require this implementation
Prerequisites
- Docker and Docker Compose for OpenCTI deployment
- Python 3.9+ with
pyctilibrary - 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.
Workflow
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=admin@opencti.io
- 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_KEY
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 = []
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 "Enrichment completed"
def _enrich_ip(self, ip_address, entity_id):
"""Enrich IP address with GreyNoise, AbuseIPDB context."""
objects = []
# GreyNoise Community API
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"## GreyNoise Enrichment\n"
f"- Classification: {classification}\n"
f"- Internet Noise: {noise}\n"
f"- RIOT (Benign Service): {riot}\n"
f"- Name: {gn_data.get('name', 'N/A')}\n"
f"- Last Seen: {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)
# Add labels based on 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"GreyNoise enrichment failed: {e}")
return objects
def _enrich_domain(self, domain, entity_id):
"""Enrich domain with WHOIS and DNS context."""
objects = []
try:
# Use SecurityTrails API for domain enrichment
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"## SecurityTrails Enrichment\n"
f"- A Records: {', '.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"SecurityTrails enrichment failed: {e}")
return objects
if __name__ == "__main__":
connector = CustomEnrichmentConnector()
## Other files in this skill
- [LICENSE](https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/building-ioc-enrichment-pipeline-with-opencti/LICENSE)
- [assets/template.md](https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/building-ioc-enrichment-pipeline-with-opencti/assets/template.md)
- [references/api-reference.md](https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/building-ioc-enrichment-pipeline-with-opencti/references/api-reference.md)
- [references/standards.md](https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/building-ioc-enrichment-pipeline-with-opencti/references/standards.md)
- [references/workflows.md](https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/building-ioc-enrichment-pipeline-with-opencti/references/workflows.md)
- [scripts/agent.py](https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/building-ioc-enrichment-pipeline-with-opencti/scripts/agent.py)
- [scripts/process.py](https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/building-ioc-enrichment-pipeline-with-opencti/scripts/process.py)
## assets/template.md (verbatim)
# IOC Enrichment Report Template
## Report Metadata
| Field | Value |
|-------|-------|
| Report ID | ENRICH-YYYY-NNNN |
| Date | YYYY-MM-DD HH:MM UTC |
| Platform | OpenCTI v6.x |
| Analyst | [Analyst Name] |
| Classification | TLP:AMBER |
## Observable Summary
| Observable | Type | Initial Score | Enriched Score | Priority |
|-----------|------|---------------|----------------|----------|
| x.x.x.x | IPv4-Addr | 0 | 85 | Critical |
| evil.com | Domain-Name | 0 | 62 | High |
## Enrichment Results
### Observable: [Value]
**Type**: IPv4-Addr / Domain-Name / StixFile
#### VirusTotal
| Metric | Value |
|--------|-------|
| Malicious Detections | X / Y engines |
| Suspicious | X |
| Reputation Score | X |
| AS Owner | |
| Country | |
#### Shodan
| Metric | Value |
|--------|-------|
| Open Ports | |
| Known Vulnerabilities | |
| ISP | |
| Organization | |
| Operating System | |
#### AbuseIPDB
| Metric | Value |
|--------|-------|
| Abuse Confidence Score | X% |
| Total Reports | X |
| Distinct Reporters | X |
| Is Tor Exit Node | Yes/No |
| Usage Type | |
#### GreyNoise
| Metric | Value |
|--------|-------|
| Classification | malicious/benign/unknown |
| Internet Noise | Yes/No |
| RIOT (Benign Service) | Yes/No |
| Name | |
| Last Seen | |
## Confidence Scoring Breakdown
| Source | Weight | Score Contribution |
|--------|--------|--------------------|
| VirusTotal | 30% | X points |
| AbuseIPDB | 30% | X points |
| GreyNoise | 20% | X points |
| Shodan | 20% | X points |
| **Total** | **100%** | **X / 100** |
## Recommended Actions
| Priority | Action | Observable | Reason |
|----------|--------|-----------|--------|
| Critical | Block immediately | | Score > 80 |
| High | Add to watchlist | | Score 50-79 |
| Medium | Monitor | | Score 20-49 |
| Low | No action needed | | Score < 20 |
## STIX Relationships Created
| Source | Relationship | Target |
|--------|-------------|--------|
| [Observable] | indicates | [Malware/Campaign] |
| [Observable] | related-to | [Infrastructure] |
| [Threat Actor] | uses | [Observable] |
## references/api-reference.md (verbatim)
# API Reference: IOC Enrichment Pipeline with OpenCTI
## pycti — OpenCTI Python Client
### Installation
```bash
pip install pycti
Client Initialization
from pycti import OpenCTIApiClient
client = OpenCTIApiClient(
url="http://localhost:8080",
token=os.environ.get("OPENCTI_TOKEN", "")
)
Indicator Operations
# List indicators with filter
filters = {
"mode": "and",
"filters": [{"key": "value", "values": ["198.51.100.42"]}],
"filterGroups": []
}
indicators = client.indicator.list(filters=filters)
# Create indicator
client.indicator.create(
name="Malicious IP",
pattern="[ipv4-addr:value = '198.51.100.42']",
pattern_type="stix",
x_opencti_score=80,
valid_from="2025-01-01T00:00:00Z"
)
Observable Operations
# Search observables
obs = client.stix_cyber_observable.list(filters=filters)
# Create observable
client.stix_cyber_observable.create(
observableData={
"type": "ipv4-addr",
"value": "198.51.100.42"
}
)
Relationship Queries
# Get relationships from entity
rels = client.stix_core_relationship.list(
filters={
"mode": "and",
"filters": [{"key": "fromId", "values": [entity_id]}],
"filterGroups": []
}
)
OpenCTI GraphQL API
Endpoint
POST /graphql
Authorization: Bearer <token>
Content-Type: application/json
Example Query
query {
indicators(filters: {
mode: and
filters: [{ key: "value", values: ["198.51.100.42"] }]
filterGroups: []
}) {
edges {
node {
id
pattern
x_opencti_score
createdBy { name }
objectLabel { value }
}
}
}
}
STIX Indicator Patterns
| Type | STIX Pattern |
|---|---|
| IPv4 | |
| Domain | |
| URL | |
| SHA-256 | |
| MD5 | |
references/standards.md (verbatim)
Standards and Frameworks Reference
STIX 2.1 (Native Data Model for OpenCTI)
STIX Domain Objects (SDOs)
- Indicator: Contains detection patterns (STIX patterning, YARA, Sigma)
- Malware: Represents malware families and variants
- Threat Actor: Describes adversary groups and individuals
- Campaign: Groups related intrusion activity
- Attack Pattern: Maps to MITRE ATT&CK techniques
- Infrastructure: Represents adversary-owned systems (C2, exploit kits)
- Tool: Legitimate software used by adversaries
STIX Cyber Observables (SCOs)
- IPv4-Addr / IPv6-Addr: Network addresses
- Domain-Name: DNS domain names
- URL: Full URL indicators
- StixFile: File hashes (MD5, SHA-1, SHA-256)
- Email-Addr: Email addresses
- Artifact: Binary content (malware samples)
- Process: Running process information
- Network-Traffic: Network flow data
STIX Relationship Objects (SROs)
- Relationship: Connects two SDOs (e.g., Threat Actor "uses" Malware)
- Sighting: Records observation of an indicator or malware
OpenCTI Connector Standards
Connector Types
- EXTERNAL_IMPORT: Ingest data from external sources (MISP, TAXII feeds)
- INTERNAL_IMPORT_FILE: Parse uploaded files (PDF reports, STIX bundles)
- INTERNAL_ENRICHMENT: Enrich existing observables with external data
- INTERNAL_ANALYSIS: Analyze content for indicators
- STREAM: Real-time export to external systems (SIEM, SOAR)
Connector Communication Protocol
- Connectors communicate via RabbitMQ message queues
- Messages contain STIX 2.1 bundles in JSON format
- Enrichment connectors receive entity_id and return STIX bundles
- Rate limiting and retry logic handled by connector framework
Enrichment Service APIs
VirusTotal v3 API
- Endpoint:
https://www.virustotal.com/api/v3/ - Resources: files, urls, domains, ip_addresses
- Rate limits: 4 requests/minute (free), 1000/minute (premium)
- Returns: detection ratios, behavioral analysis, relationships
Shodan API
- Endpoint:
https://api.shodan.io/ - Resources: host/{ip}, dns/resolve, search
- Returns: open ports, services, banners, vulnerabilities, ASN info
AbuseIPDB v2 API
- Endpoint:
https://api.abuseipdb.com/api/v2/ - Resources: check, reports, blacklist
- Returns: abuse confidence score, total reports, categories, country
GreyNoise v3 API
- Endpoint:
https://api.greynoise.io/v3/ - Resources: community/{ip}, noise/context/{ip}
- Returns: classification (benign/malicious/unknown), RIOT status, tags
MITRE ATT&CK Framework
- OpenCTI maps Attack Patterns to ATT&CK techniques
- Supports Enterprise, Mobile, and ICS matrices
- Technique relationships enable campaign-level analysis
- Sub-technique granularity (e.g., T1059.001 - PowerShell)
References
- OpenCTI Documentation
- STIX 2.1 Specification
- OpenCTI GitHub
- OpenCTI Connectors Ecosystem
- VirusTotal API v3
references/workflows.md (verbatim)
OpenCTI IOC Enrichment Workflows
Workflow 1: Automatic Enrichment Pipeline
[New Observable Created] --> [RabbitMQ Queue] --> [Enrichment Connectors]
|
+-----------+-----------+
| | |
v v v
[VirusTotal] [Shodan] [AbuseIPDB]
| | |
v v v
[STIX Bundle] [STIX Bundle] [STIX Bundle]
| | |
+-----------+-----------+
|
v
[Merged into OpenCTI]
|
v
[Confidence Updated]
Steps:
- Observable Ingestion: New IP/domain/hash created via feed import or manual entry
- Queue Distribution: OpenCTI sends observable to enrichment connector queues
- Parallel Enrichment: Each connector queries its respective external API
- STIX Bundle Generation: Connectors produce STIX 2.1 bundles with notes, labels, relationships
- Merge: Enrichment results merged into the observable's knowledge graph
- Scoring: Confidence score updated based on aggregated enrichment data
Workflow 2: Analyst-Triggered Enrichment
[Analyst Selects Observable] --> [Manual Enrichment Request] --> [Selected Connectors]
| |
v v
[Review Results] <-- [Enrichment Dashboard] <-- [Results Returned]
|
v
[Update Tags/Labels] --> [Add to Investigation]
Steps:
- Selection: Analyst identifies observable requiring additional context
- Connector Choice: Select specific enrichment connectors to run
- Execution: Connectors query external services with observable value
- Review: Analyst reviews enrichment results in observable detail view
- Curation: Analyst updates labels, confidence, and adds notes
- Investigation: Link enriched observable to ongoing investigation case
Workflow 3: Bulk Enrichment Pipeline
[STIX Import] --> [Observable Extraction] --> [Batch Queue] --> [Rate-Limited Enrichment]
|
v
[Progress Tracking]
|
v
[Enrichment Report]
Steps:
- Bulk Import: Import STIX bundle with hundreds of observables
- Extraction: OpenCTI extracts unique observables from imported data
- Queue Management: Observables queued for enrichment with rate limiting
- Progressive Enrichment: Connectors process queue respecting API rate limits
- Monitoring: Track enrichment progress via connector status dashboard
- Reporting: Generate enrichment summary with coverage statistics
Workflow 4: Enrichment-Driven Scoring
[Raw IOC (Score: 0)] --> [VirusTotal] --> [Score += VT_detections/total * 30]
|
v
[AbuseIPDB] --> [Score += abuse_confidence * 0.3]
|
v
[GreyNoise] --> [Score += classification_weight]
|
v
[Shodan] --> [Score += open_ports_risk]
|
v
[Final Score (0-100)] --> [Priority Classification]
|
+---------+---------+
| | |
v v v
[Critical] [High] [Low]
(80-100) (50-79) (0-49)
Steps:
- Baseline: Observable starts with confidence score of 0
- VT Score: VirusTotal detection ratio contributes up to 30 points
- Abuse Score: AbuseIPDB confidence contributes up to 30 points
- Classification: GreyNoise malicious/benign classification adds/subtracts points
- Exposure: Shodan data on open ports and known vulnerabilities adds risk points
- Final Priority: Aggregated score determines analyst priority queue placement
Back to mukul975/Anthropic-Cybersecurity-Skills (817 security skills) or Agent skills.