
Building Ioc Defanging And Sharing Pipeline
- 129 installs
- 27.3k repo stars
- Updated August 2, 2026
- mukul975/anthropic-cybersecurity-skills
Helps with ai & agent building tasks.
About
building-ioc-defanging-and-sharing-pipeline is a Claude Code skill in the AI & Agent Building category.
- building-ioc-defanging-and-sharing-pipeline
- AI & Agent Building
- AI-coding skill
Building Ioc Defanging And Sharing Pipeline by the numbers
- 129 all-time installs (skills.sh)
- +14 installs in the week ending Jul 26, 2026 (Skillselion tracking)
- Ranked #3,702 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/mukul975/anthropic-cybersecurity-skills --skill building-ioc-defanging-and-sharing-pipelineAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 129 |
|---|---|
| repo stars | ★ 27.3k |
| Last updated | August 2, 2026 |
| Repository | mukul975/anthropic-cybersecurity-skills ↗ |
What it does
Helps with ai & agent building tasks.
Files
Building IOC Defanging and Sharing Pipeline
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,validatorslibraries - 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
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to the Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by the Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding any notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. Please do not remove or change
the license header comment from a contributed file except when
necessary.
Copyright 2026 mukul975
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
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 |
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_KEYAbuseIPDB API v2
GET https://api.abuseipdb.com/api/v2/check
Key: YOUR_KEY
Params: ipAddress, maxAgeInDaysSTIX 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#!/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()