building-threat-intelligence-platform

Building a Threat Intelligence Platform (TIP) involves deploying and integrating multiple CTI tools into a unified system for collecting, analyzing, enriching, and disseminating threat intelligence. T

Jump to install

Repository: yanacuti1121/Yana-AI
Last source activity: July 2, 2026 at 23:25
Detected SKILL.md language: English
Stars: 1
Forks: 0

Install options

The review-first prompt is selected by default. You can switch to a direct command or download a local copy.

Review the source files

Read SKILL.md and any companion files shown by SkillsMP before deciding whether to install.

Back to search Back to installation

name	building-threat-intelligence-platform
description	Building a Threat Intelligence Platform (TIP) involves deploying and integrating multiple CTI tools into a unified system for collecting, analyzing, enriching, and disseminating threat intelligence. T
domain	cybersecurity
subdomain	threat-intelligence
tags	["threat-intelligence","cti","ioc","mitre-attack","stix","platform-building","misp","opencti"]
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","T1588.001","T1591"]
source	https://github.com/mukul975/Anthropic-Cybersecurity-Skills
source_commit	04450304b12645cb2b974ab96d28c0664758a88d
note	Vendored verbatim from an external Apache-2.0 security-skill library, pinned by commit. Exceeds the internal 300-line skill guideline (agent-code-constraints.md) -- kept as-is because this is vendored reference material (forensics/threat-intel procedure), not Yana AI-authored content, and trimming would damage technical accuracy.

Building Threat Intelligence Platform

Overview

Building a Threat Intelligence Platform (TIP) involves deploying and integrating multiple CTI tools into a unified system for collecting, analyzing, enriching, and disseminating threat intelligence. This skill covers designing TIP architecture using open-source tools (MISP, OpenCTI, TheHive, Cortex), configuring feed ingestion pipelines, establishing enrichment workflows, implementing STIX/TAXII interoperability, and building analyst dashboards for CTI operations.

When to Use

When deploying or configuring building threat intelligence platform 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 deploying platform components
Python 3.9+ with pymisp, pycti, thehive4py libraries
Elasticsearch/OpenSearch cluster for data storage
Redis and RabbitMQ for message queuing
Understanding of STIX 2.1 data model and TAXII 2.1 transport
API keys for enrichment services (VirusTotal, Shodan, AbuseIPDB)

Key Concepts

TIP Architecture Components

Collection Layer: Feed ingestion from OSINT, commercial, and internal sources
Storage Layer: Elasticsearch/OpenSearch for indexed CTI data with STIX 2.1 schema
Analysis Layer: OpenCTI for knowledge graph analysis and MISP for IOC correlation
Enrichment Layer: Cortex analyzers for automated IOC enrichment
Response Layer: TheHive for case management and incident response integration
Sharing Layer: TAXII server for outbound intelligence sharing

Platform Integration Points

MISP <-> OpenCTI: Bidirectional sync via OpenCTI MISP connector
OpenCTI <-> TheHive: Alert/case creation from high-confidence indicators
TheHive <-> Cortex: Automated analysis and enrichment of case observables
All <-> SIEM: Real-time IOC push to Splunk/Elastic via API or Kafka

Workflow

Step 1: Deploy Platform with Docker Compose

version:

from pymisp import PyMISP
from pycti import OpenCTIApiClient
import json

class TIPFeedManager:
    """Manage threat intelligence feed ingestion across platform components."""

    def __init__(self, misp_url, misp_key, opencti_url, opencti_token):
        self.misp = PyMISP(misp_url, misp_key, ssl=False)
        self.opencti = OpenCTIApiClient(opencti_url, opencti_token)

    def configure_osint_feeds(self):
        """Enable default OSINT feeds in MISP."""
        osint_feeds = [
            {"name": "CIRCL OSINT", "id": 1},
            {"name": "Botvrij.eu", "id": 2},
            {"name": "abuse.ch URLhaus", "id": 5},
            {"name": "abuse.ch Feodo Tracker", "id": 6},
        ]
        for feed in osint_feeds:
            try:
                self.misp.enable_feed(feed["id"])
                self.misp.fetch_feed(feed["id"])
                print(f"[+] Enabled feed: {feed['name']}")
            except Exception as e:
                print(f"[-] Failed: {feed['name']}: {e}")

    def configure_opencti_connectors(self):
        """List and verify OpenCTI connector status."""
        connectors = self.opencti.connector.list()
        for conn in connectors:
            print(
                f"  Connector: {conn['name']} - "
                f"Active: {conn['active']} - "
                f"Type: {conn['connector_type']}"
            )

    def sync_misp_to_opencti(self):
        """Verify MISP-OpenCTI sync is operational."""
        # OpenCTI MISP connector handles this automatically
        # Check connector status
        connectors = self.opencti.connector.list()
        misp_connector = [
            c for c in connectors if "misp" in c["name"].lower()
        ]
        if misp_connector:
            print(f"[+] MISP connector active: {misp_connector[0]['active']}")
        else:
            print("[-] MISP connector not found - configure in Docker Compose")

import requests

class CortexEnrichment:
    """Integrate Cortex analyzers for automated enrichment."""

    def __init__(self, cortex_url, cortex_key):
        self.url = cortex_url
        self.headers = {"Authorization": f"Bearer {cortex_key}"}

    def list_analyzers(self):
        """List available Cortex analyzers."""
        resp = requests.get(
            f"{self.url}/api/analyzer",
            headers=self.headers,
            timeout=30,
        )
        if resp.status_code == 200:
            analyzers = resp.json()
            for a in analyzers:
                print(f"  {a['name']}: {a.get('description', '')[:60]}")
            return analyzers
        return []

    def analyze_observable(self, observable_type, observable_value, analyzer_id):
        """Submit an observable for analysis."""
        job = {
            "data": observable_value,
            "dataType": observable_type,
            "tlp": 2,
            "message": "TIP automated enrichment",
        }
        resp = requests.post(
            f"{self.url}/api/analyzer/{analyzer_id}/run",
            json=job,
            headers=self.headers,
            timeout=30,
        )
        if resp.status_code == 200:
            return resp.json()
        return None

    def get_job_report(self, job_id):
        """Get the report for a completed analysis job."""
        resp = requests.get(
            f"{self.url}/api/job/{job_id}/report",
            headers=self.headers,
            timeout=60,
        )
        if resp.status_code == 200:
            return resp.json()
        return None

class TIPMetrics:
    """Collect platform metrics for analyst dashboards."""

    def __init__(self, misp, opencti):
        self.misp = misp
        self.opencti = opencti

    def get_platform_stats(self):
        """Collect statistics across all platform components."""
        stats = {}

        # MISP stats
        misp_stats = self.misp.get_server_statistics()
        stats["misp"] = {
            "total_events": misp_stats.get("event_count", 0),
            "total_attributes": misp_stats.get("attribute_count", 0),
            "active_feeds": len([
                f for f in self.misp.feeds()
                if f.get("Feed", {}).get("enabled")
            ]),
        }

        # OpenCTI stats via GraphQL
        stats["opencti"] = {
            "total_indicators": self.opencti.indicator.list(
                first=0, withPagination=True
            ).get("pagination", {}).get("globalCount", 0),
            "total_reports": self.opencti.report.list(
                first=0, withPagination=True
            ).get("pagination", {}).get("globalCount", 0),
        }

        return stats