building-ioc-defanging-and-sharing-pipeline

name	building-ioc-defanging-and-sharing-pipeline
description	构建自动化流水线，对失陷指标（URL、IP、域名、邮件）进行去危化处理以便安全共享，并通过 TAXII 推送和威胁情报平台以 STIX 格式分发。
domain	cybersecurity
subdomain	threat-intelligence
tags	["ioc","defanging","threat-sharing","stix","pipeline","indicator","automation","threat-intelligence"]
version	1.0
author	mahipal
license	Apache-2.0

构建 IOC 去危化与共享流水线

概述

IOC 去危化（Defanging）将潜在恶意指标（URL、IP 地址、域名、电子邮件地址）进行修改，防止意外点击或执行，同时保持可读性以便分析和共享。本技能涵盖构建自动化流水线：从多个来源摄取原始 IOC、规范化和去重、对人工查阅进行去危化处理、转换为 STIX 2.1 格式以便机器处理，并通过 TAXII 服务器、MISP 实例和邮件报告进行分发。

前置条件

• Python 3.9+，安装 defang、ioc-fanger、stix2、requests、validators 库
• MISP 实例或 TAXII 服务器用于自动化共享
• 了解 IOC 类型：IPv4/IPv6、域名、URL、电子邮件地址、文件哈希
• 熟悉 STIX 2.1 指标模式和 TLP 标记定义
• 访问威胁情报推送以摄取 IOC

核心概念

IOC 去危化标准

去危化替换活跃协议和域名组件以防止执行：http:// 变为 hxxp://，https:// 变为 hxxps://，域名/IP 中的点变为 [.]，邮件中的 @ 变为 [@]。这对于在报告、邮件、Slack 频道和粘贴站点中共享 IOC 至关重要——在这些场景下自动链接可能触发对恶意基础设施的网络连接。

IOC 规范化

来自不同来源的原始 IOC 格式不一致。规范化包括：转换为小写、去除尾部斜杠和空白、从 URL 中提取域名、解析 URL 编码、验证格式正确性，以及跨来源去重。

STIX 2.1 指标模式

STIX 模式以标准化格式表达 IOC：[ipv4-addr:value = '203.0.113.1']、[domain-name:value = 'malicious.example.com']、[url:value = 'http://evil.com/payload']、[file:hashes.'SHA-256' = 'abc123...']。每个指标包含 valid_from、indicator_types、confidence 和可选的 TLP 标记。

实践步骤

步骤 1：构建 IOC 提取与规范化

import re
import hashlib
from urllib.parse import urlparse, unquote
from datetime import datetime

class IOCExtractor:
    """从文本中提取并规范化 IOC。"""

    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):
        """从自由文本中提取所有类型的 IOC。"""
        # 先对已去危化的指标进行还原
        text = self._refang(text)
        iocs = {"ipv4": set(), "domain": set(), "url": set(),
                "email": set(), "md5": (), : (), : ()}

         ioc_type, pattern  .PATTERNS.items():
            matches = re.findall(pattern, text)
               matches:
                normalized = ._normalize(, ioc_type)
                 normalized   ._is_whitelisted(normalized, ioc_type):
                    iocs[ioc_type].add(normalized)

        
        url_domains = ()
         url  iocs[]:
            parsed = urlparse(url)
            url_domains.add(parsed.netloc)
        iocs[] -= url_domains

        total = ((v)  v  iocs.values())
        ()
         {k: (v)  k, v  iocs.items()}

     ():
        
        text = text.replace(, ).replace(, )
        text = text.replace(, ).replace(, )
        text = text.replace(, ).replace(, )
         text

     ():
        
        value = value.strip().lower()
         ioc_type == :
            value = unquote(value).rstrip()
         ioc_type == :
            value = value.rstrip()
         value

     ():
        
         ioc_type == :
             value  .WHITELIST_DOMAINS
         ioc_type == :
            parsed = urlparse(value)
             parsed.netloc  .WHITELIST_DOMAINS
         

extractor = IOCExtractor()
sample_text = 
iocs = extractor.extract_from_text(sample_text)

步骤 2：去危化引擎

class IOCDefanger:
    """对 IOC 进行去危化处理，用于报告和通信中的安全共享。"""

    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):
        """对字典中的所有 IOC 进行去危化处理。"""
        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]
             ioc_type == :
                defanged[ioc_type] = [.defang_ip(v)  v  values]
             ioc_type == :
                defanged[ioc_type] = [.defang_email(v)  v  values]
            :
                defanged[ioc_type] = values  
         defanged

     ():
        
        report = 
        report += 

         ioc_type  [, , , , , , ]:
            values = defanged.get(ioc_type, [])
             values:
                report += 
                 v  values:
                    report += 
                report += 

         report

defanger = IOCDefanger()
defanged = defanger.defang_all(iocs)
report = defanger.generate_sharing_report(iocs, defanged, )
(report)

步骤 3：转换为 STIX 2.1 格式

from stix2 import Indicator, Bundle, TLP_WHITE, TLP_GREEN, TLP_AMBER
from datetime import datetime

class STIXConverter:
    """将原始 IOC 转换为 STIX 2.1 指标对象。"""

    TLP_MAP = {"white": TLP_WHITE, "green": TLP_GREEN, "amber": TLP_AMBER}

    def iocs_to_stix(self, iocs, tlp="green", confidence=75):
        """将 IOC 字典转换为 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"恶意 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"恶意域名: {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=,
                pattern=,
                pattern_type=,
                valid_from=datetime.now(),
                indicator_types=[],
                confidence=confidence,
                object_marking_refs=[marking],
            ))

         sha256  iocs.get(, []):
            stix_objects.append(Indicator(
                name=,
                pattern=,
                pattern_type=,
                valid_from=datetime.now(),
                indicator_types=[],
                confidence=confidence,
                object_marking_refs=[marking],
            ))

        bundle = Bundle(objects=stix_objects)
        ()
         bundle

converter = STIXConverter()
stix_bundle = converter.iocs_to_stix(iocs, tlp=, confidence=)
 (, )  f:
    f.write(stix_bundle.serialize(pretty=))

步骤 4：通过 MISP 和 TAXII 分发

import requests
import json

class IOCDistributor:
    """通过各种渠道分发 IOC。"""

    def push_to_misp(self, iocs, misp_url, misp_key, event_info):
        """将 IOC 作为新事件推送到 MISP。"""
        headers = {
            "Authorization": misp_key,
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        event = {
            "Event": {
                "info": event_info,
                "distribution": "1",  # 仅限本社区
                "threat_level_id": "2",  # 中等
                "analysis": "2",  # 已完成
                "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[][].append({
                        : misp_type,
                        : value,
                        :   ioc_type  (, , )  ,
                        : ,
                    })

        resp = requests.post(
            ,
            headers=headers,
            json=event,
            verify=,
        )
         resp.status_code == :
            event_id = resp.json().get(, {}).get(, )
            ()
             event_id
        :
            ()
             

     ():
        
         taxii2client.v21  Collection
        collection = Collection(
            ,
            user=username, password=password,
        )
        response = collection.add_objects(stix_bundle.serialize())
        ()
         response

distributor = IOCDistributor()
distributor.push_to_misp(
    iocs,
    misp_url=,
    misp_key=,
    event_info=,
)

验收标准

• IOC 已从自由文本中正确提取并支持还原去危化
• 去危化产生安全、不可点击的指标
• STIX 2.1 bundle 包含有效的指标模式
• IOC 已成功分发到 MISP 和 TAXII
• 去重防止重复指标
• 白名单防止对已知良好域名产生误报

参考资料

• CISA: 网络安全自动化与威胁情报共享
• Defang Python 库
• ioc-fanger
• STIX 2.1 指标规范
• Grokipedia: Defanging
• Hunt.io: Best IOC Feeds