building-threat-actor-profile-from-osint

从 OSINT 构建威胁行为者档案

概述

利用 OSINT 进行威胁行为者画像，系统性地收集和分析公开可用信息，以构建对手组织的全面档案。本技能涵盖从公开来源（安全厂商报告、粘贴站点、暗网论坛、社交媒体、代码仓库）收集情报、跨平台关联指标、使用 Maltego 和 SpiderFoot 等工具映射对手基础设施，以及生成指导防御策略和归因评估的结构化威胁行为者档案。

前置条件

• Python 3.9+，安装 shodan、requests、beautifulsoup4、maltego-trx、stix2 库
• SpiderFoot 或 SpiderFoot HX
• Maltego CE 或 Maltego XL 用于链接分析
• API 密钥：Shodan、VirusTotal、AlienVault OTX、PassiveTotal/RiskIQ
• MITRE ATT&CK 知识用于 TTP 映射
• 了解 STIX 2.1 入侵集合、威胁行为者和身份 SDO

核心概念

威胁行为者画像的 OSINT 来源

主要情报来源包括：厂商威胁报告（Mandiant、CrowdStrike、Recorded Future、Talos）、政府建议（CISA、NSA、FBI 联合建议）、学术研究论文、恶意软件仓库（VirusTotal、MalwareBazaar、Malpedia）、粘贴站点（Pastebin、GitHub Gists）、代码仓库、社交媒体账号、暗网论坛和证书透明度日志。

结构化分析技术

画像使用菱形模型（对手、基础设施、能力、受害者）、竞争假说分析（ACH）进行归因置信度评估，以及 MITRE ATT&CK 映射进行 TTP 记录。Maltego 等链接分析工具可视化指标、基础设施和行为者之间的关系。

档案组成要素

完整的威胁行为者档案包括：各厂商的别名和命名规范、可疑来源和赞助方、动机（间谍、财务、黑客主义、破坏），以及目标行业和地区、已知攻击活动和行动、映射到 ATT&CK 的 TTP、工具集和恶意软件家族、基础设施模式和历史时间线。

实践步骤

步骤 1：从多个来源收集情报

import requests
import json
from datetime import datetime

class OSINTCollector:
    def __init__(self, vt_key=None, otx_key=None, shodan_key=None):
        self.vt_key = vt_key
        self.otx_key = otx_key
        self.shodan_key = shodan_key
        self.collected_data = {"sources": [], "indicators": [], "reports": []}

    def search_alienvault_otx(self, actor_name):
        """在 AlienVault OTX 中搜索威胁行为者情报包。"""
        headers = {"X-OTX-API-KEY": self.otx_key}
        url = f"https://otx.alienvault.com/api/v1/search/pulses?q={actor_name}&limit=20"
        resp = requests.get(url, headers=headers)
        if resp.status_code == 200:
            data = resp.json()
            pulses = data.get("results", [])
            for pulse in pulses:
                self.collected_data["reports"].append({
                    "source": "AlienVault OTX",
                    "title": pulse.get("name", ""),
                    "created": pulse.get("created", ""),
                    "description": pulse.get("description", "")[:500],
                    "tags": pulse.get("tags", []),
                    "indicators_count": len(pulse.get("indicators", [])),
                    "pulse_id": pulse.get("id", ""),
                })
                for ioc in pulse.get("indicators", []):
                    self.collected_data["indicators"].append({
                        "type": ioc.get("type", ""),
                        "value": ioc.get("indicator", ""),
                        "source": "OTX",
                        "pulse": pulse.get("name", ""),
                    })
            print(f"[+] OTX: 为 '{actor_name}' 找到 {len(pulses)} 个情报包")
        return self.collected_data

    def search_virustotal_collections(self, actor_name):
        """在 VirusTotal 中搜索威胁行为者集合。"""
        headers = {"x-apikey": self.vt_key}
        url = "https://www.virustotal.com/api/v3/intelligence/search"
        params = {"query": f"tag:{actor_name.lower().replace(' ', '-')}"}
        resp = requests.get(url, headers=headers, params=params)
        if resp.status_code == 200:
            results = resp.json().get("data", [])
            print(f"[+] VT: 找到 {len(results)} 个标记为 '{actor_name}' 的样本")
            return results
        return []

    def query_shodan_infrastructure(self, indicators):
        """通过 Shodan 查询 IP 的基础设施详情。"""
        results = []
        for ip in indicators:
            url = f"https://api.shodan.io/shodan/host/{ip}?key={self.shodan_key}"
            resp = requests.get(url)
            if resp.status_code == 200:
                data = resp.json()
                results.append({
                    "ip": ip,
                    "org": data.get("org", ""),
                    "asn": data.get("asn", ""),
                    "country": data.get("country_code", ""),
                    "ports": data.get("ports", []),
                    "hostnames": data.get("hostnames", []),
                    "os": data.get("os", ""),
                    "last_update": data.get("last_update", ""),
                })
        print(f"[+] Shodan: 已富化 {len(results)} 个 IP")
        return results

collector = OSINTCollector(
    vt_key="YOUR_VT_KEY",
    otx_key="YOUR_OTX_KEY",
    shodan_key="YOUR_SHODAN_KEY",
)
data = collector.search_alienvault_otx("APT29")

步骤 2：构建结构化威胁行为者档案

from stix2 import ThreatActor, IntrusionSet, Identity, Relationship, Bundle
from datetime import datetime

# 创建 STIX 2.1 威胁行为者档案
identity = Identity(
    name="网络安全分析师",
    identity_class="individual",
)

threat_actor = ThreatActor(
    name="APT29",
    description="APT29（也称为 Cozy Bear、Midnight Blizzard、NOBELIUM、The Dukes）"
                "是归因于俄罗斯对外情报局（SVR）的俄罗斯国家支持的威胁组织。"
                "该组织至少自 2008 年起活跃，主要针对 NATO 国家的政府、外交、"
                "智库、医疗和能源机构实施网络间谍活动。",
    aliases=["Cozy Bear", "Midnight Blizzard", "NOBELIUM", "The Dukes",
             "Dark Halo", "UNC2452", "YTTRIUM", "Blue Kitsune", "Iron Ritual"],
    roles=["agent"],
    sophistication="strategic",
    resource_level="government",
    primary_motivation="organizational-gain",
    secondary_motivations=["ideology"],
    threat_actor_types=["nation-state"],
    goals=["收集外国政府情报",
           "长期持续访问高价值目标",
           "供应链入侵以获得广泛访问"],
    created_by_ref=identity.id,
)

intrusion_set = IntrusionSet(
    name="APT29",
    description="归因于俄罗斯 SVR 的入侵集合 APT29。",
    aliases=["Cozy Bear", "Midnight Blizzard"],
    first_seen="2008-01-01T00:00:00Z",
    goals=["espionage"],
    resource_level="government",
    primary_motivation="organizational-gain",
)

relationship = Relationship(
    relationship_type="attributed-to",
    source_ref=intrusion_set.id,
    target_ref=threat_actor.id,
)

bundle = Bundle(objects=[identity, threat_actor, intrusion_set, relationship])
with open("apt29_profile.json", "w") as f:
    f.write(bundle.serialize(pretty=True))
print("[+] STIX 档案已保存: apt29_profile.json")

步骤 3：将 TTP 映射到 MITRE ATT&CK

from attackcti import attack_client

lift = attack_client()
apt29_techs = lift.get_techniques_used_by_group("G0016")

profile_ttps = {
    "initial_access": [],
    "execution": [],
    "persistence": [],
    "defense_evasion": [],
    "credential_access": [],
    "lateral_movement": [],
    "collection": [],
    "c2": [],
    "exfiltration": [],
}

tactic_mapping = {
    "initial-access": "initial_access",
    "execution": "execution",
    "persistence": "persistence",
    "defense-evasion": "defense_evasion",
    "credential-access": "credential_access",
    "lateral-movement": "lateral_movement",
    "collection": "collection",
    "command-and-control": "c2",
    "exfiltration": "exfiltration",
}

for tech in apt29_techs:
    tech_id = ""
    for ref in tech.get("external_references", []):
        if ref.get("source_name") == "mitre-attack":
            tech_id = ref.get("external_id", "")
            break
    for phase in tech.get("kill_chain_phases", []):
        tactic = phase.get("phase_name", "")
        key = tactic_mapping.get(tactic)
        if key:
            profile_ttps[key].append({
                "id": tech_id,
                "name": tech.get("name", ""),
                "description": tech.get("description", "")[:200],
            })

print("=== APT29 TTP 档案 ===")
for tactic, techs in profile_ttps.items():
    if techs:
        print(f"\n{tactic.upper()}（{len(techs)} 个技术）:")
        for t in techs[:5]:
            print(f"  {t['id']}: {t['name']}")

步骤 4：使用 SpiderFoot 关联基础设施

import subprocess
import json

def run_spiderfoot_scan(target, scan_name="actor_recon"):
    """对目标域名或 IP 运行 SpiderFoot 扫描。"""
    cmd = [
        "python3", "-m", "spiderfoot", "-s", target,
        "-m", "sfp_dns,sfp_whois,sfp_shodan,sfp_virustotal,sfp_certspotter",
        "-o", "json", "-q",
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    if result.returncode == 0:
        findings = json.loads(result.stdout) if result.stdout else []
        print(f"[+] SpiderFoot: 为 {target} 找到 {len(findings)} 个发现")
        return findings
    return []

def correlate_infrastructure(indicators):
    """查找基础设施指标之间的关系。"""
    ip_to_domains = {}
    domain_to_ips = {}
    registrar_patterns = {}

    for indicator in indicators:
        ioc_type = indicator.get("type", "")
        value = indicator.get("value", "")

        if ioc_type == "IP_ADDRESS":
            if value not in ip_to_domains:
                ip_to_domains[value] = set()
        elif ioc_type == "INTERNET_NAME":
            if value not in domain_to_ips:
                domain_to_ips[value] = set()

    # 识别共享托管和注册模式
    shared_ips = {ip: domains for ip, domains in ip_to_domains.items() if len(domains) > 1}
    print(f"[+] 共享基础设施 IP: {len(shared_ips)} 个")
    return {"shared_ips": shared_ips, "registrar_patterns": registrar_patterns}

步骤 5：生成威胁行为者档案报告

def generate_dossier(actor_name, profile_data, ttp_data, infrastructure_data):
    dossier = f"""# 威胁行为者档案: {actor_name}
## 生成时间: {datetime.now().isoformat()}

## 摘要
{profile_data.get('description', '')}

## 归因
- **可疑来源**: {profile_data.get('origin', '未知')}
- **赞助方**: {profile_data.get('sponsorship', '未知')}
- **置信度**: {profile_data.get('confidence', '中等')}
- **首次发现**: {profile_data.get('first_seen', '未知')}

## 别名
{', '.join(profile_data.get('aliases', []))}

## 目标
- **行业**: {', '.join(profile_data.get('sectors', []))}
- **地区**: {', '.join(profile_data.get('regions', []))}
- **动机**: {profile_data.get('motivation', '未知')}

## TTP 摘要（MITRE ATT&CK）
"""
    for tactic, techs in ttp_data.items():
        if techs:
            dossier += f"\n### {tactic.replace('_', ' ').title()}\n"
            for t in techs:
                dossier += f"- **{t['id']}**: {t['name']}\n"

    dossier += f"""
## 基础设施模式
- 已知 C2 服务器: {len(infrastructure_data.get('c2_servers', []))} 个
- 域名模式: {', '.join(infrastructure_data.get('domain_patterns', []))}
- 托管偏好: {', '.join(infrastructure_data.get('hosting', []))}

## 建议措施
1. 在 EDR/SIEM 中监控已知 TTP
2. 封锁已知基础设施指标
3. 在网络流量中狩猎行为模式
4. 针对主要技术缺口实施检测
"""
    with open(f"{actor_name.lower().replace(' ', '_')}_dossier.md", "w") as f:
        f.write(dossier)
    print(f"[+] 已为 {actor_name} 保存档案报告")

generate_dossier("APT29", {
    "description": "归因于 SVR 的俄罗斯国家支持的间谍组织",
    "origin": "俄罗斯", "sponsorship": "SVR（对外情报局）",
    "confidence": "高", "first_seen": "2008",
    "aliases": ["Cozy Bear", "Midnight Blizzard", "NOBELIUM", "The Dukes"],
    "sectors": ["政府", "外交", "智库", "医疗", "能源"],
    "regions": ["北美", "欧洲", "NATO 国家"],
    "motivation": "间谍活动",
}, profile_ttps, {"c2_servers": [], "domain_patterns": [], "hosting": []})

验收标准

• 从至少 3 个 OSINT 来源收集情报
• 正确创建 STIX 2.1 威胁行为者和入侵集合对象
• TTP 已映射到 ATT&CK 并包含技术 ID 和过程示例
• 基础设施指标已跨来源关联
• 档案包含带有置信度级别的归因评估
• 档案可直接用于检测工程和威胁狩猎

参考资料

• Huntress: 威胁行为者画像
• CrowdStrike: 网络安全中的 OSINT
• SpiderFoot OSINT 工具
• MITRE ATT&CK 组织
• ShadowDragon: OSINT 技术
• ISACA: 构建以威胁为导向的网络安全项目