implementing-threat-intelligence-lifecycle-management

Implement a structured threat intelligence lifecycle encompassing planning, collection, processing, analysis, dissemination, and feedback stages to produce actionable intelligence for organizational decision-making.

Aller à l'installation

Dépôt: xalgord/xalgorix
Dernière activité de la source: 6 juin 2026 à 16:41
Langue détectée de SKILL.md: anglais
Étoiles: 813
Forks: 146

Options d'installation

Le prompt qui vérifie d'abord la source est sélectionné par défaut. Vous pouvez passer à une commande directe ou télécharger une copie locale.

Vérifiez les fichiers source

Lisez SKILL.md et les fichiers associés affichés par SkillsMP avant de décider de l'installer.

Retour à la recherche Retour à l’installation

name	implementing-threat-intelligence-lifecycle-management
description	Implement a structured threat intelligence lifecycle encompassing planning, collection, processing, analysis, dissemination, and feedback stages to produce actionable intelligence for organizational decision-making.
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"]

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

Common Misconfigurations & Verification

Collection without PIRs: wiring up CISA KEV, OTX, and MalwareBazaar feeds before defining Priority Intelligence Requirements produces data overload and no answers. Map every collection source to a PIR or drop it.
Processing stage skipped: feeding raw collection straight to analysis without the normalize → dedup → enrich → score pipeline floods analysts with duplicates. Confirm the dedup hash keys on type:value:source so distinct-context IOCs are not over-merged.
Confidence/TLP defaults unset: _normalize() defaulting confidence to 50 and TLP to green means unscored, freely-shareable indicators leak downstream. Set these explicitly per source.
No feedback loop: dissemination without the feedback stage means products are never validated against PIRs - wire collect_feedback() to a defined SLA.
Indicator expiry ignored: tagging with a collection date but no TTL/expiration lets stale IOCs persist in the SIEM. Add decay handling.
Verification: confirm each active PIR maps to >=1 collection source, the pipeline reports duplicates removed, analysis output references a specific requirement_id, and distribution metrics show products reaching each stakeholder channel.

Prerequisites

Python 3.9+ with pymisp, stix2, requests, pandas libraries
MISP or OpenCTI as threat intelligence platform
Ticketing system (Jira, ServiceNow) for requirements management

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()

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))

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()

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

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()