threat intelligence

Building IOC Defanging and Sharing Pipeline

Build an automated pipeline to defang indicators of compromise (URLs, IPs, domains, emails) for safe sharing and distribute them in STIX format through TAXII feeds and threat intelligence platforms.

automationdefangingindicatoriocpipelinestixthreat-intelligencethreat-sharing
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

Overview

IOC defanging modifies potentially malicious indicators (URLs, IP addresses, domains, email addresses) to prevent accidental clicks or execution while preserving readability for analysis and sharing. This skill covers building an automated pipeline that ingests raw IOCs from multiple sources, normalizes and deduplicates them, applies defanging for safe human consumption, converts them to STIX 2.1 format for machine consumption, and distributes through TAXII servers, MISP instances, and email reports.

When to Use

  • When deploying or configuring building ioc defanging and sharing pipeline 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 defang, ioc-fanger, stix2, requests, validators libraries
  • MISP instance or TAXII server for automated sharing
  • Understanding of IOC types: IPv4/IPv6, domains, URLs, email addresses, file hashes
  • Familiarity with STIX 2.1 Indicator patterns and TLP marking definitions
  • Access to threat intelligence feeds for IOC ingestion

Key Concepts

IOC Defanging Standards

Defanging replaces active protocol and domain components to prevent execution: http:// becomes hxxp://, https:// becomes hxxps://, dots in domains/IPs become [.], @ in emails becomes [@]. This is critical for sharing IOCs in reports, emails, Slack channels, and paste sites where auto-linking could trigger network connections to malicious infrastructure.

IOC Normalization

Raw IOCs from different sources come in inconsistent formats. Normalization involves converting to lowercase, removing trailing slashes and whitespace, extracting domains from URLs, resolving URL encoding, validating format correctness, and deduplicating across sources.

STIX 2.1 Indicator Patterns

STIX patterns express IOCs in a standardized format: [ipv4-addr:value = '203.0.113.1'], [domain-name:value = 'malicious.example.com'], [url:value = 'http://evil.com/payload'], [file:hashes.'SHA-256' = 'abc123...']. Each indicator includes valid_from, indicator_types, confidence, and optional TLP markings.

Workflow

Step 1: Build IOC Extraction and Normalization

import re
import hashlib
from urllib.parse import urlparse, unquote
from datetime import datetime
 
class IOCExtractor:
    """Extract and normalize IOCs from text."""
 
    PATTERNS = {
        "ipv4": r'\b(?:(?:25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)\.){3}(?:25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)\b',
        "domain": r'\b(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}\b',
        "url": r'https?://[^\s<>"{}|\\^`\[\]]+',
        "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        "md5": r'\b[a-fA-F0-9]{32}\b',
        "sha1": r'\b[a-fA-F0-9]{40}\b',
        "sha256": r'\b[a-fA-F0-9]{64}\b',
    }
 
    WHITELIST_DOMAINS = {
        "google.com", "microsoft.com", "amazon.com", "github.com",
        "cloudflare.com", "akamai.com", "example.com",
    }
 
    def extract_from_text(self, text):
        """Extract all IOC types from free text."""
        # Refang any already-defanged indicators first
        text = self._refang(text)
        iocs = {"ipv4": set(), "domain": set(), "url": set(),
                "email": set(), "md5": set(), "sha1": set(), "sha256": set()}
 
        for ioc_type, pattern in self.PATTERNS.items():
            matches = re.findall(pattern, text)
            for match in matches:
                normalized = self._normalize(match, ioc_type)
                if normalized and not self._is_whitelisted(normalized, ioc_type):
                    iocs[ioc_type].add(normalized)
 
        # Remove domains that are part of URLs
        url_domains = set()
        for url in iocs["url"]:
            parsed = urlparse(url)
            url_domains.add(parsed.netloc)
        iocs["domain"] -= url_domains
 
        total = sum(len(v) for v in iocs.values())
        print(f"[+] Extracted {total} unique IOCs from text")
        return {k: sorted(v) for k, v in iocs.items()}
 
    def _refang(self, text):
        """Convert defanged indicators back to active form."""
        text = text.replace("hxxp://", "http://").replace("hxxps://", "https://")
        text = text.replace("[.]", ".").replace("[@]", "@")
        text = text.replace("[://]", "://").replace("(.)", ".")
        return text
 
    def _normalize(self, value, ioc_type):
        """Normalize an IOC value."""
        value = value.strip().lower()
        if ioc_type == "url":
            value = unquote(value).rstrip("/")
        elif ioc_type == "domain":
            value = value.rstrip(".")
        return value
 
    def _is_whitelisted(self, value, ioc_type):
        """Check if IOC is in whitelist."""
        if ioc_type == "domain":
            return value in self.WHITELIST_DOMAINS
        if ioc_type == "url":
            parsed = urlparse(value)
            return parsed.netloc in self.WHITELIST_DOMAINS
        return False
 
extractor = IOCExtractor()
sample_text = """
Malware C2: hxxps://evil-domain[.]com/beacon
Drops payload from 192.168.1.100 and contacts 10[.]0[.]0[.]1
SHA256: 275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f
Phishing email from attacker[@]phishing-domain[.]com
"""
iocs = extractor.extract_from_text(sample_text)

Step 2: Defanging Engine

class IOCDefanger:
    """Defang IOCs for safe sharing in reports and communications."""
 
    def defang_url(self, url):
        return url.replace("http://", "hxxp://").replace("https://", "hxxps://").replace(".", "[.]")
 
    def defang_domain(self, domain):
        return domain.replace(".", "[.]")
 
    def defang_ip(self, ip):
        return ip.replace(".", "[.]")
 
    def defang_email(self, email):
        return email.replace("@", "[@]").replace(".", "[.]")
 
    def defang_all(self, iocs):
        """Defang all IOCs in a dictionary."""
        defanged = {}
        for ioc_type, values in iocs.items():
            if ioc_type == "url":
                defanged[ioc_type] = [self.defang_url(v) for v in values]
            elif ioc_type == "domain":
                defanged[ioc_type] = [self.defang_domain(v) for v in values]
            elif ioc_type == "ipv4":
                defanged[ioc_type] = [self.defang_ip(v) for v in values]
            elif ioc_type == "email":
                defanged[ioc_type] = [self.defang_email(v) for v in values]
            else:
                defanged[ioc_type] = values  # Hashes don't need defanging
        return defanged
 
    def generate_sharing_report(self, iocs, defanged, report_name="IOC Report"):
        """Generate a human-readable defanged IOC report."""
        report = f"# {report_name}\n"
        report += f"Generated: {datetime.now().isoformat()}\n\n"
 
        for ioc_type in ["url", "domain", "ipv4", "email", "sha256", "sha1", "md5"]:
            values = defanged.get(ioc_type, [])
            if values:
                report += f"## {ioc_type.upper()} ({len(values)})\n"
                for v in values:
                    report += f"- `{v}`\n"
                report += "\n"
 
        return report
 
defanger = IOCDefanger()
defanged = defanger.defang_all(iocs)
report = defanger.generate_sharing_report(iocs, defanged, "Malware Campaign IOCs")
print(report)

Step 3: Convert to STIX 2.1 Format

from stix2 import Indicator, Bundle, TLP_WHITE, TLP_GREEN, TLP_AMBER
from datetime import datetime
 
class STIXConverter:
    """Convert raw IOCs to STIX 2.1 Indicator objects."""
 
    TLP_MAP = {"white": TLP_WHITE, "green": TLP_GREEN, "amber": TLP_AMBER}
 
    def iocs_to_stix(self, iocs, tlp="green", confidence=75):
        """Convert IOC dictionary to STIX 2.1 bundle."""
        stix_objects = []
        marking = self.TLP_MAP.get(tlp, TLP_GREEN)
 
        for ip in iocs.get("ipv4", []):
            stix_objects.append(Indicator(
                name=f"Malicious IP: {ip}",
                pattern=f"[ipv4-addr:value = '{ip}']",
                pattern_type="stix",
                valid_from=datetime.now(),
                indicator_types=["malicious-activity"],
                confidence=confidence,
                object_marking_refs=[marking],
            ))
 
        for domain in iocs.get("domain", []):
            stix_objects.append(Indicator(
                name=f"Malicious Domain: {domain}",
                pattern=f"[domain-name:value = '{domain}']",
                pattern_type="stix",
                valid_from=datetime.now(),
                indicator_types=["malicious-activity"],
                confidence=confidence,
                object_marking_refs=[marking],
            ))
 
        for url in iocs.get("url", []):
            escaped = url.replace("'", "\\'")
            stix_objects.append(Indicator(
                name=f"Malicious URL: {url[:60]}",
                pattern=f"[url:value = '{escaped}']",
                pattern_type="stix",
                valid_from=datetime.now(),
                indicator_types=["malicious-activity"],
                confidence=confidence,
                object_marking_refs=[marking],
            ))
 
        for sha256 in iocs.get("sha256", []):
            stix_objects.append(Indicator(
                name=f"Malicious File Hash: {sha256[:16]}...",
                pattern=f"[file:hashes.'SHA-256' = '{sha256}']",
                pattern_type="stix",
                valid_from=datetime.now(),
                indicator_types=["malicious-activity"],
                confidence=confidence,
                object_marking_refs=[marking],
            ))
 
        bundle = Bundle(objects=stix_objects)
        print(f"[+] Created STIX bundle with {len(stix_objects)} indicators")
        return bundle
 
converter = STIXConverter()
stix_bundle = converter.iocs_to_stix(iocs, tlp="amber", confidence=80)
with open("iocs_stix_bundle.json", "w") as f:
    f.write(stix_bundle.serialize(pretty=True))

Step 4: Distribute Through MISP and TAXII

import requests
import json
 
class IOCDistributor:
    """Distribute IOCs through various channels."""
 
    def push_to_misp(self, iocs, misp_url, misp_key, event_info):
        """Push IOCs to MISP as a new event."""
        headers = {
            "Authorization": misp_key,
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        event = {
            "Event": {
                "info": event_info,
                "distribution": "1",  # This community only
                "threat_level_id": "2",  # Medium
                "analysis": "2",  # Completed
                "Attribute": [],
            }
        }
 
        type_mapping = {
            "ipv4": "ip-dst",
            "domain": "domain",
            "url": "url",
            "email": "email-src",
            "md5": "md5",
            "sha1": "sha1",
            "sha256": "sha256",
        }
 
        for ioc_type, values in iocs.items():
            misp_type = type_mapping.get(ioc_type)
            if misp_type:
                for value in values:
                    event["Event"]["Attribute"].append({
                        "type": misp_type,
                        "value": value,
                        "category": "Network activity" if ioc_type in ("ipv4", "domain", "url") else "Payload delivery",
                        "to_ids": True,
                    })
 
        resp = requests.post(
            f"{misp_url}/events",
            headers=headers,
            json=event,
            verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true",  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
        )
        if resp.status_code == 200:
            event_id = resp.json().get("Event", {}).get("id", "")
            print(f"[+] MISP event created: {event_id}")
            return event_id
        else:
            print(f"[-] MISP error: {resp.status_code} - {resp.text[:200]}")
            return None
 
    def push_to_taxii(self, stix_bundle, taxii_url, collection_id, username, password):
        """Push STIX bundle to TAXII 2.1 collection."""
        from taxii2client.v21 import Collection
        collection = Collection(
            f"{taxii_url}/collections/{collection_id}/",
            user=username, password=password,
        )
        response = collection.add_objects(stix_bundle.serialize())
        print(f"[+] TAXII: Published bundle, status: {response.status}")
        return response
 
distributor = IOCDistributor()
distributor.push_to_misp(
    iocs,
    misp_url="https://misp.organization.com",
    misp_key="YOUR_MISP_API_KEY",
    event_info="Malware Campaign IOCs - 2025",
)

Validation Criteria

  • IOCs extracted correctly from free text with refanging support
  • Defanging produces safe, non-clickable indicators
  • STIX 2.1 bundle contains valid indicator patterns
  • IOCs distributed to MISP and TAXII successfully
  • Deduplication prevents duplicate indicators
  • Whitelisting prevents false positives on known-good domains

References

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.5 KB

API Reference: IOC Defanging and Sharing Pipeline

IOC Defanging Rules

Type Original Defanged
IPv4 192.168.1.1 192[.]168[.]1[.]1
Domain evil.com evil[.]com
URL https://evil.com/payload hxxps://evil[.]com/payload
Email attacker@evil.com attacker@evil[.]com

IOC Extraction Patterns

Type Regex
IPv4 \b(?:\d{1,3}\.){3}\d{1,3}\b
Domain \b(?:[a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}\b
URL https?://[^\s"'<>]+
MD5 \b[a-fA-F0-9]{32}\b
SHA256 \b[a-fA-F0-9]{64}\b

STIX 2.1 Indicator Format

{
  "type": "indicator",
  "spec_version": "2.1",
  "pattern_type": "stix",
  "pattern": "[ipv4-addr:value = '1.2.3.4']",
  "valid_from": "2024-01-01T00:00:00Z",
  "labels": ["malicious-activity"]
}

VirusTotal API v3

GET https://www.virustotal.com/api/v3/files/{hash}
x-apikey: YOUR_KEY

AbuseIPDB API v2

GET https://api.abuseipdb.com/api/v2/check
Key: YOUR_KEY
Params: ipAddress, maxAgeInDays

STIX Pattern Examples

IOC Type STIX Pattern
IPv4 [ipv4-addr:value = '1.2.3.4']
Domain [domain-name:value = 'evil.com']
URL [url:value = 'https://evil.com']
MD5 [file:hashes.'MD5' = 'abc123...']
SHA256 [file:hashes.'SHA-256' = 'def456...']

TAXII 2.1 Sharing

POST https://taxii.server/api/collections/{id}/objects/
Authorization: Basic BASE64
Content-Type: application/taxii+json;version=2.1
Body: STIX Bundle JSON

Scripts 1

agent.py6.9 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""IOC Defanging and Sharing Pipeline Agent - Defangs, enriches, and shares IOCs in STIX format."""

import json
import logging
import argparse
import os
import re
from datetime import datetime

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

IOC_PATTERNS = {
    "ipv4": re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"),
    "domain": re.compile(r"\b(?:[a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}\b"),
    "url": re.compile(r"https?://[^\s\"'<>]+"),
    "md5": re.compile(r"\b[a-fA-F0-9]{32}\b"),
    "sha256": re.compile(r"\b[a-fA-F0-9]{64}\b"),
    "email": re.compile(r"\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b"),
}

EXCLUSION_DOMAINS = {"example.com", "example.org", "example.net", "localhost", "schema.org",
                     "w3.org", "microsoft.com", "google.com", "github.com"}


def extract_iocs(text):
    """Extract IOCs from raw text."""
    results = {}
    for ioc_type, pattern in IOC_PATTERNS.items():
        matches = set(pattern.findall(text))
        if ioc_type == "domain":
            matches = {m for m in matches if m.split(".")[-1] not in {"py", "js", "json", "xml", "yml"}
                       and m not in EXCLUSION_DOMAINS and not any(excl in m for excl in EXCLUSION_DOMAINS)}
        results[ioc_type] = list(matches)
    logger.info("Extracted IOCs: %s", {k: len(v) for k, v in results.items()})
    return results


def defang_ioc(value, ioc_type):
    """Defang an IOC for safe sharing."""
    if ioc_type == "ipv4":
        return value.replace(".", "[.]")
    elif ioc_type in ("domain", "email"):
        return value.replace(".", "[.]")
    elif ioc_type == "url":
        return value.replace("http://", "hxxp://").replace("https://", "hxxps://").replace(".", "[.]")
    return value


def refang_ioc(value, ioc_type):
    """Refang a defanged IOC back to original form."""
    if ioc_type in ("ipv4", "domain", "email"):
        return value.replace("[.]", ".")
    elif ioc_type == "url":
        return value.replace("hxxp://", "http://").replace("hxxps://", "https://").replace("[.]", ".")
    return value


def enrich_ioc(value, ioc_type, vt_key=None, abuseipdb_key=None):
    """Enrich IOC with threat intelligence from VirusTotal and AbuseIPDB."""
    vt_key = vt_key or os.environ.get("VT_API_KEY", "")
    abuseipdb_key = abuseipdb_key or os.environ.get("ABUSEIPDB_KEY", "")
    enrichment = {"value": value, "type": ioc_type, "sources": []}
    if ioc_type in ("md5", "sha256") and vt_key:
        try:
            resp = requests.get(f"https://www.virustotal.com/api/v3/files/{value}",
                                headers={"x-apikey": vt_key}, timeout=10)
            if resp.status_code == 200:
                data = resp.json().get("data", {}).get("attributes", {}).get("last_analysis_stats", {})
                enrichment["vt_malicious"] = data.get("malicious", 0)
                enrichment["sources"].append("virustotal")
        except requests.RequestException:
            pass
    elif ioc_type == "ipv4" and abuseipdb_key:
        try:
            resp = requests.get("https://api.abuseipdb.com/api/v2/check",
                                params={"ipAddress": value, "maxAgeInDays": 90},
                                headers={"Key": abuseipdb_key, "Accept": "application/json"}, timeout=10)
            if resp.status_code == 200:
                data = resp.json().get("data", {})
                enrichment["abuse_score"] = data.get("abuseConfidenceScore", 0)
                enrichment["sources"].append("abuseipdb")
        except requests.RequestException:
            pass
    return enrichment


def to_stix_bundle(iocs):
    """Convert IOCs to STIX 2.1 bundle for sharing."""
    objects = []
    stix_type_map = {"ipv4": "ipv4-addr", "domain": "domain-name", "url": "url",
                     "md5": "file", "sha256": "file", "email": "email-addr"}
    for ioc_type, values in iocs.items():
        for value in values:
            stype = stix_type_map.get(ioc_type)
            if not stype:
                continue
            indicator = {
                "type": "indicator",
                "spec_version": "2.1",
                "id": f"indicator--{hash(value) % (10**12):012d}",
                "created": datetime.utcnow().isoformat() + "Z",
                "pattern_type": "stix",
                "pattern": f"[{stype}:value = '{value}']" if stype != "file"
                           else f"[file:hashes.'{ioc_type.upper()}' = '{value}']",
                "valid_from": datetime.utcnow().isoformat() + "Z",
                "labels": ["malicious-activity"],
            }
            objects.append(indicator)
    bundle = {"type": "bundle", "id": f"bundle--{hash(str(iocs)) % (10**12):012d}",
              "objects": objects}
    logger.info("Created STIX bundle with %d indicators", len(objects))
    return bundle


def defang_all(iocs):
    """Defang all extracted IOCs."""
    defanged = {}
    for ioc_type, values in iocs.items():
        defanged[ioc_type] = [{"original": v, "defanged": defang_ioc(v, ioc_type)} for v in values]
    return defanged


def generate_report(iocs, defanged, stix_bundle):
    """Generate IOC sharing pipeline report."""
    total = sum(len(v) for v in iocs.values())
    report = {
        "timestamp": datetime.utcnow().isoformat(),
        "total_iocs": total,
        "by_type": {k: len(v) for k, v in iocs.items()},
        "defanged_iocs": defanged,
        "stix_indicator_count": len(stix_bundle.get("objects", [])),
        "stix_bundle": stix_bundle,
    }
    print(f"IOC PIPELINE REPORT: {total} IOCs extracted, {len(stix_bundle.get('objects', []))} STIX indicators")
    return report


def main():
    parser = argparse.ArgumentParser(description="IOC Defanging and Sharing Pipeline")
    parser.add_argument("--input-file", required=True, help="Text file containing IOCs")
    parser.add_argument("--enrich", action="store_true", help="Enrich IOCs with threat intel")
    parser.add_argument("--vt-key", default=os.environ.get("VT_API_KEY", ""), help="VirusTotal API key")
    parser.add_argument("--abuseipdb-key", default=os.environ.get("ABUSEIPDB_KEY", ""), help="AbuseIPDB API key")
    parser.add_argument("--output", default="ioc_pipeline_report.json")
    args = parser.parse_args()

    with open(args.input_file) as f:
        text = f.read()

    iocs = extract_iocs(text)
    defanged = defang_all(iocs)
    stix_bundle = to_stix_bundle(iocs)

    if args.enrich:
        enrichments = []
        for ioc_type, values in iocs.items():
            for value in values[:10]:
                enrichments.append(enrich_ioc(value, ioc_type, args.vt_key, args.abuseipdb_key))
        logger.info("Enriched %d IOCs", len(enrichments))

    report = generate_report(iocs, defanged, stix_bundle)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)
    logger.info("Report saved to %s", args.output)


if __name__ == "__main__":
    main()
Keep exploring