What it does. Build out a full CTI program around the six-phase threat intelligence Part of mukul975/Anthropic-Cybersecurity-Skills (817 security skills) (mukul975/Anthropic-Cybersecurity-Skills).

Install

• npx skills add mukul975/Anthropic-Cybersecurity-Skills --skill implementing-threat-intelligence-lifecycle-management, or copy the skill folder into ~/.claude/skills/implementing-threat-intelligence-lifecycle-management/.
• Raw file: curl -sL https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/implementing-threat-intelligence-lifecycle-management/SKILL.md

SKILL.md (verbatim)

name: implementing-threat-intelligence-lifecycle-management
description: Build out a full CTI program around the six-phase threat intelligence
  lifecycle (direction, collection, processing, analysis, dissemination, feedback),
  including defining intelligence requirements, building a collection pipeline, normalizing
  data, and tracking dissemination feedback. Use when standing up or maturing a threat
  intelligence program, defining intelligence requirements, or designing collection-to-dissemination
  workflows for a CTI team.
domain: cybersecurity
subdomain: threat-intelligence
tags:
- threat-intelligence
- lifecycle
- intelligence-cycle
- collection
- analysis
- dissemination
- strategic-intelligence
- cti-program
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:
- T1591
- T1592
- T1593
- T1589

Implementing Threat Intelligence Lifecycle Management

Overview

The threat intelligence lifecycle is a structured, iterative process for transforming raw data into actionable intelligence. Based on the intelligence cycle used by military and government agencies, it comprises six phases: Direction (requirements gathering), Collection (data acquisition), Processing (normalization and deduplication), Analysis (contextualization and assessment), Dissemination (distribution to stakeholders), and Feedback (evaluation and refinement). This skill covers building each phase with tooling, metrics, and integration points for a mature CTI program.

When to Use

• When deploying or configuring implementing threat intelligence lifecycle management 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

• Python 3.9+ with pymisp, stix2, requests, pandas libraries
• MISP or OpenCTI as threat intelligence platform
• Ticketing system (Jira, ServiceNow) for requirements management
• SIEM integration (Splunk, Elastic) for indicator operationalization
• Understanding of intelligence analysis techniques (ACH, Diamond Model)

Key Concepts

Intelligence Requirements (IR)

Priority Intelligence Requirements (PIRs) define what the organization needs to know. Examples: Which threat actors target our sector? What vulnerabilities are being actively exploited? Are our brand or credentials being traded on dark web? PIRs drive collection planning and ensure intelligence production is relevant.

Collection Management Framework

A collection management framework maps intelligence requirements to collection sources, tracks collection gaps, and ensures coverage across the threat landscape. Sources include OSINT, commercial feeds, ISAC sharing, internal telemetry, and human intelligence from industry contacts.

Intelligence Levels

Strategic intelligence informs executive decision-making (threat landscape, risk trends, geopolitical context). Operational intelligence supports security operations (campaign tracking, actor TTPs, attack timing). Tactical intelligence enables immediate defense (IOCs, detection rules, blocklists).

Workflow

Step 1: Define Intelligence Requirements

import json
from datetime import datetime
from enum import Enum

class Priority(Enum):
    CRITICAL = 1
    HIGH = 2
    MEDIUM = 3
    LOW = 4

class IntelligenceRequirement:
    def __init__(self, requirement_id, question, priority, stakeholder,
                 intelligence_level, collection_sources=None):
        self.id = requirement_id
        self.question = question
        self.priority = priority
        self.stakeholder = stakeholder
        self.level = intelligence_level
        self.sources = collection_sources or []
        self.created = datetime.now().isoformat()
        self.status = "active"
        self.last_answered = None

    def to_dict(self):
        return {
            "id": self.id,
            "question": self.question,
            "priority": self.priority.name,
            "stakeholder": self.stakeholder,
            "intelligence_level": self.level,
            "collection_sources": self.sources,
            "created": self.created,
            "status": self.status,
            "last_answered": self.last_answered,
        }

class RequirementsManager:
    def __init__(self):
        self.requirements = []

    def add_requirement(self, requirement):
        self.requirements.append(requirement)
        print(f"[+] Added IR-{requirement.id}: {requirement.question[:60]}...")

    def get_active_requirements(self, priority=None, level=None):
        filtered = [r for r in self.requirements if r.status == "active"]
        if priority:
            filtered = [r for r in filtered if r.priority == priority]
        if level:
            filtered = [r for r in filtered if r.level == level]
        return filtered

    def export_requirements(self, output_file="intelligence_requirements.json"):
        data = [r.to_dict() for r in self.requirements]
        with open(output_file, "w") as f:
            json.dump(data, f, indent=2)
        print(f"[+] Exported {len(data)} requirements to {output_file}")

# Define organizational PIRs
mgr = RequirementsManager()
mgr.add_requirement(IntelligenceRequirement(
    "PIR-001", "Which threat actors are actively targeting our sector?",
    Priority.CRITICAL, "CISO", "strategic",
    ["MITRE ATT&CK", "ISAC feeds", "Vendor reports"],
))
mgr.add_requirement(IntelligenceRequirement(
    "PIR-002", "What vulnerabilities are being actively exploited in the wild?",
    Priority.CRITICAL, "Vulnerability Management", "operational",
    ["CISA KEV", "Exploit-DB", "VulnCheck", "Shodan"],
))
mgr.add_requirement(IntelligenceRequirement(
    "PIR-003", "Are any organization credentials or data exposed on dark web?",
    Priority.HIGH, "SOC Manager", "tactical",
    ["Dark web monitoring", "Paste site monitoring", "Breach databases"],
))
mgr.add_requirement(IntelligenceRequirement(
    "PIR-004", "What are the emerging attack techniques against cloud infrastructure?",
    Priority.HIGH, "Cloud Security", "operational",
    ["ATT&CK Cloud matrix", "Vendor advisories", "ISAC bulletins"],
))
mgr.export_requirements()

Step 2: Build Collection Pipeline

import requests
from datetime import datetime, timedelta

class CollectionPipeline:
    def __init__(self, config):
        self.config = config
        self.collected_data = []

    def collect_cisa_kev(self):
        """Collect CISA Known Exploited Vulnerabilities catalog."""
        url = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
        resp = requests.get(url, timeout=30)
        if resp.status_code == 200:
            data = resp.json()
            vulns = data.get("vulnerabilities", [])
            self.collected_data.append({
                "source": "CISA KEV",
                "type": "vulnerability",
                "count": len(vulns),
                "collected_at": datetime.now().isoformat(),
                "data": vulns,
            })
            print(f"[+] CISA KEV: {len(vulns)} known exploited vulnerabilities")
            return vulns
        return []

    def collect_otx_pulses(self, api_key, days=7):
        """Collect recent OTX pulses."""
        headers = {"X-OTX-API-KEY": api_key}
        since = (datetime.now() - timedelta(days=days)).isoformat()
        url = f"https://otx.alienvault.com/api/v1/pulses/subscribed?modified_since={since}"
        resp = requests.get(url, headers=headers, timeout=30)
        if resp.status_code == 200:
            pulses = resp.json().get("results", [])
            self.collected_data.append({
                "source": "AlienVault OTX",
                "type": "threat_intelligence",
                "count": len(pulses),
                "collected_at": datetime.now().isoformat(),
            })
            print(f"[+] OTX: {len(pulses)} pulses in last {days} days")
            return pulses
        return []

    def collect_abuse_ch(self):
        """Collect recent malware samples from MalwareBazaar."""
        url = "https://mb-api.abuse.ch/api/v1/"
        resp = requests.post(url, data={"query": "get_recent", "selector": "time"}, timeout=30)
        if resp.status_code == 200:
            data = resp.json().get("data", [])
            self.collected_data.append({
                "source": "MalwareBazaar",
                "type": "malware_samples",
                "count": len(data),
                "collected_at": datetime.now().isoformat(),
            })
            print(f"[+] MalwareBazaar: {len(data)} recent samples")
            return data
        return []

    def get_collection_summary(self):
        summary = {
            "total_sources": len(self.collected_data),
            "total_items": sum(d.get("count", 0) for d in self.collected_data),
            "sources": [
                {"name": d["source"], "type": d["type"], "count": d["count"]}
                for d in self.collected_data
            ],
        }
        return summary

pipeline = CollectionPipeline({})
pipeline.collect_cisa_kev()
pipeline.collect_abuse_ch()
print(json.dumps(pipeline.get_collection_summary(), indent=2))

Step 3: Process and Normalize Data

class IntelligenceProcessor:
    def __init__(self):
        self.processed_items = []
        self.dedup_hashes = set()

    def process_collection(self, raw_data, source_name):
        """Normalize and deduplicate collected intelligence."""
        processed = []
        duplicates = 0

        for item in raw_data:
            normalized = self._normalize(item, source_name)
            if normalized:
                item_hash = self._compute_hash(normalized)
                if item_hash not in self.dedup_hashes:
                    self.dedup_hashes.add(item_hash)
                    normalized["processed_at"] = datetime.now().isoformat()
                    processed.append(normalized)
                else:
                    duplicates += 1

        self.processed_items.extend(processed)
        print(f"[+] Processed {len(processed)} items from {source_name} "
              f"({duplicates} duplicates removed)")
        return processed

    def _normalize(self, item, source):
        """Normalize item to standard format."""
        return {
            "source": source,
            "type": item.get("type", "unknown"),
            "value": item.get("value", item.get("indicator", "")),
            "confidence": item.get("confidence", 50),
            "tlp": item.get("tlp", "green"),
            "tags": item.get("tags", []),
            "first_seen": item.get("first_seen", item.get("date_added", "")),
            "raw": item,
        }

    def _compute_hash(self, item):
        import hashlib
        key = f"{item['type']}:{item['value']}:{item['source']}"
        return hashlib.sha256(key.encode()).hexdigest()

processor = IntelligenceProcessor()

Step 4: Analyze and Produce Intelligence

class IntelligenceAnalyzer:
    def __init__(self, requirements, processed_data):
        self.requirements = requirements
        self.data = processed_data

    def answer_requirement(self, requirement_id):
        """Produce intelligence answering a specific requirement."""
        req = next((r for r in self.requirements if r.id == requirement_id), None)
        if not req:
            return None

        # Filter relevant data based on requirement type
        relevant = self.data  # In practice, filter by requirement topic
        analysis = {
            "requirement_id": requirement_id,
            "question": req.question,
            "intelligence_level": req.level,
            "data_points_analyzed": len(relevant),
            "produced_at": datetime.now().isoformat(),
            "key_findings": [],
            "confidence": "medium",
            "recommendations": [],
        }
        return analysis

    def produce_daily_brief(self):
        """Produce daily threat intelligence brief."""
        brief = {
            "date": datetime.now().strftime("%Y-%m-%d"),
            "total_items_processed": len(self.data),
            "highlights": [],
            "active_requirements_status": [
                {"id": r.id, "question": r.question[:80], "status": r.status}
                for r in self.requirements if r.status == "active"
            ],
        }
        return brief

Step 5: Disseminate and Track Feedback

class IntelligenceDisseminator:
    def __init__(self):
        self.distribution_log = []

    def distribute_report(self, report, channels, classification="TLP:GREEN"):
        """Distribute intelligence report to appropriate channels."""
        for channel in channels:
            entry = {
                "report_id": report.get("requirement_id", "daily"),
                "channel": channel,
                "classification": classification,
                "distributed_at": datetime.now().isoformat(),
                "status": "sent",
            }
            self.distribution_log.append(entry)
            print(f"  [+] Distributed to {channel}")

    def collect_feedback(self, report_id, stakeholder, rating, comments=""):
        """Collect stakeholder feedback on intelligence product."""
        feedback = {
            "report_id": report_id,
            "stakeholder": stakeholder,
            "rating": rating,  # 1-5
            "comments": comments,
            "received_at": datetime.now().isoformat(),
        }
        print(f"[+] Feedback received from {stakeholder}: {rating}/5")
        return feedback

    def calculate_metrics(self):
        """Calculate CTI program performance metrics."""
        metrics = {
            "total_products_distributed": len(self.distribution_log),
            "distribution_by_channel": {},
        }
        for entry in self.distribution_log:
            channel = entry["channel"]
            if channel not in metrics["distribution_by_channel"]:
                metrics["distribution_by_channel"][channel] = 0
            metrics["distribution_by_channel"][channel] += 1
        return metrics

disseminator = IntelligenceDisseminator()

Validation Criteria

• Intelligence requirements defined with priorities and stakeholders
• Collection pipeline gathering from multiple sources
• Processing deduplicates and normalizes data correctly
• Analysis produces intelligence answering specific requirements
• Dissemination reaches appropriate stakeholders through right channels
• Feedback mechanism captures and incorporates stakeholder input

References

• SANS: Cyber Threat Intelligence Lifecycle
• CISA: Cybersecurity Automation Best Practices
• CyCognito: Threat Intelligence Lifecycle
• MISP Project
• STIX/TAXII Documentation
• CISA Known Exploited Vulnerabilities

Other files in this skill

• LICENSE
• references/api-reference.md
• scripts/agent.py

references/api-reference.md (verbatim)

API Reference: Threat Intelligence Lifecycle Management

Libraries Used

Installation

pip install pymisp stix2 requests

Authentication

MISP Connection

from pymisp import PyMISP
import os

MISP_URL = os.environ["MISP_URL"]
MISP_KEY = os.environ["MISP_API_KEY"]
MISP_VERIFYCERT = os.environ.get("MISP_VERIFY", "True") == "True"

misp = PyMISP(MISP_URL, MISP_KEY, ssl=MISP_VERIFYCERT)

MISP API Operations

Search for Events

def search_events(tags=None, date_from=None, published=True):
    results = misp.search(
        controller="events",
        tags=tags,
        date_from=date_from,
        published=published,
        limit=100,
    )
    return results

Create a Threat Intelligence Event

from pymisp import MISPEvent, MISPAttribute

def create_ti_event(info, threat_level=2, analysis=1):
    event = MISPEvent()
    event.info = info
    event.threat_level_id = threat_level  # 1=High, 2=Medium, 3=Low, 4=Undefined
    event.analysis = analysis  # 0=Initial, 1=Ongoing, 2=Completed
    event.distribution = 1  # 1=This community
    created = misp.add_event(event)
    return created

Add Indicators to an Event

def add_indicators(event_id, indicators):
    for ioc in indicators:
        attr = MISPAttribute()
        attr.type = ioc["type"]  # "ip-dst", "domain", "sha256", "url"
        attr.value = ioc["value"]
        attr.category = ioc.get("category", "Network activity")
        attr.to_ids = ioc.get("to_ids", True)
        attr.comment = ioc.get("comment", "")
        misp.add_attribute(event_id, attr)

Search for Specific IOCs

def search_ioc(ioc_type, value):
    results = misp.search(
        controller="attributes",
        type_attribute=ioc_type,
        value=value,
    )
    return results

Tag Management

# Add TLP marking
misp.tag(event_id, "tlp:amber")

# Add threat actor tag
misp.tag(event_id, "mitre-attack-pattern:T1566.001")

# Add custom taxonomy
misp.tag(event_id, "adversary:APT29")

STIX 2.1 Intelligence Objects

Create STIX Indicator

import stix2

indicator = stix2.Indicator(
    name="Cobalt Strike C2 Domain",
    pattern="[domain-name:value = 'c2.evil.example.com']",
    pattern_type="stix",
    valid_from="2025-01-15T00:00:00Z",
    labels=["malicious-activity"],
    confidence=85,
    external_references=[
        stix2.ExternalReference(
            source_name="Internal IR",
            description="Observed during incident IR-2025-001",
        )
    ],
)

Create STIX Threat Actor

threat_actor = stix2.ThreatActor(
    name="APT29",
    aliases=["Cozy Bear", "The Dukes"],
    threat_actor_types=["nation-state"],
    roles=["agent"],
    sophistication="expert",
    resource_level="government",
    primary_motivation="espionage",
)

Create Relationships and Bundle

relationship = stix2.Relationship(
    relationship_type="indicates",
    source_ref=indicator.id,
    target_ref=threat_actor.id,
    confidence=80,
)

bundle = stix2.Bundle(objects=[indicator, threat_actor, relationship])

Convert MISP Event to STIX

def misp_to_stix(event):
    stix_objects = []
    for attr in event.get("Attribute", []):
        if attr["type"] == "ip-dst":
            stix_objects.append(stix2.Indicator(
                name=f"Malicious IP: {attr['value']}",
                pattern=f"[ipv4-addr:value = '{attr['value']}']",
                pattern_type="stix",
                valid_from=attr["timestamp"],
            ))
        elif attr["type"] == "domain":
            stix_objects.append(stix2.Indicator(
                name=f"Malicious Domain: {attr['value']}",
                pattern=f"[domain-name:value = '{attr['value']}']",
                pattern_type="stix",
                valid_from=attr["timestamp"],
            ))
    return stix2.Bundle(objects=stix_objects)

Intelligence Lifecycle Phases

Output Format

{
  "lifecycle_phase": "analysis",
  "events_processed": 42,
  "indicators_created": 156,
  "stix_objects": {
    "indicators": 120,
    "threat_actors": 5,
    "malware": 8,
    "relationships": 95,
    "attack_patterns": 23
  },
  "tlp_distribution": {
    "tlp:white": 30,
    "tlp:green": 45,
    "tlp:amber": 65,
    "tlp:red": 16
  }
}

Back to mukul975/Anthropic-Cybersecurity-Skills (817 security skills) or Agent skills.