performing-ai-driven-osint-correlation

执行 AI 驱动的 OSINT 关联分析

何时使用

• 你已从多个工具和来源收集了原始 OSINT 数据，但需要识别其中的关联、矛盾之处和模式。
• 你需要从碎片化数据中为目标实体（个人、组织或基础设施）构建统一的情报画像。
• 对于已收集的数据量而言，传统的人工关联分析太慢或容易出错。
• 你希望获得跨平台身份关联的置信度评分评估，而非简单的关键词匹配。

前提条件

• Python 3.10+，并安装 requests、json 和 csv 库
• 已安装 Sherlock（pip install sherlock-project）
• 已安装 theHarvester（pip install theHarvester）
• SpiderFoot 4.0+ 运行在 localhost:5001
• 可访问 LLM API（OpenAI、Anthropic，或通过 Ollama 使用的本地模型）
• 可选：Maltego CE，用于关联结果的图谱可视化
• 可选：Shodan、VirusTotal、HaveIBeenPwned、Hunter.io 的 API 密钥

工作流

法律与道德要求

• 在任何调查之前，获取有记录的书面授权
• 确立数据处理的合法依据（执法、企业政策等）
• 定义 PII 保留限制和数据处理程序
• 遵守当地隐私法规（GDPR、CCPA 等）

阶段 1 — 多源 OSINT 收集

0. 为所有 OSINT 输出创建工作目录：

   mkdir -p /tmp/osint
   

1. 使用 Sherlock 跨平台枚举用户名：

   sherlock "targetusername" --output /tmp/osint/sherlock-results.txt --csv
   

2. 使用 theHarvester 收集邮箱、子域名和主机：

   theHarvester -d targetdomain.com -b all -f /tmp/osint/harvester-results.json
   

3. 通过 REST API 运行 SpiderFoot 被动扫描：

   curl -s http://localhost:5001/api/scan/start \
     -d "scanname=target-recon&scantarget=targetdomain.com&usecase=passive" \
     | jq '.scanid'
   

4. 扫描完成后导出 SpiderFoot 结果：

   SCAN_ID="<scanid_from_step_3>"
   curl -s "http://localhost:5001/api/scan/${SCAN_ID}/results?type=all" \
     -o /tmp/osint/spiderfoot-results.json
   

5. 查询泄露数据库以检查邮箱暴露情况（以 HIBP API 为例）：

   curl -s -H "hibp-api-key: ${HIBP_KEY}" \
     -H "User-Agent: OSINT-Correlation-Skill" \
     "https://haveibeenpwned.com/api/v3/breachedaccount/[email protected]" \
     -o /tmp/osint/breach-results.json
   

阶段 2 — 数据规范化

6. 将所有收集的数据规范化为通用 schema。 创建一个统一的 JSON 结构，为每条发现标注其来源、时间戳和数据类型：

   cat > /tmp/osint/normalize.py << 'EOF'
   import json, csv, sys, os
   from datetime import datetime
   
   findings = []
   
   # Normalize Sherlock CSV results
   sherlock_path = "/tmp/osint/sherlock-results.txt"
   if os.path.exists(sherlock_path):
       with open(sherlock_path) as f:
           for row in csv.DictReader(f):
               findings.append({
                   "source": "sherlock",
                   "type": "social_profile",
                   "platform": row.get("name", ""),
                   "url": row.get("url_user", ""),
                   "username": row.get("username", ""),
                   "status": row.get("status", ""),
                   "collected_at": datetime.utcnow().isoformat()
               })
   
   # Normalize theHarvester JSON results
   harvester_path = "/tmp/osint/harvester-results.json"
   if os.path.exists(harvester_path):
       with open(harvester_path) as f:
           data = json.load(f)
           for email in data.get("emails", []):
               findings.append({
                   "source": "theHarvester",
                   "type": "email",
                   "value": email,
                   "collected_at": datetime.utcnow().isoformat()
               })
           for host in data.get("hosts", []):
               findings.append({
                   "source": "theHarvester",
                   "type": "hostname",
                   "value": host,
                   "collected_at": datetime.utcnow().isoformat()
               })
   
   # Normalize SpiderFoot results
   sf_path = "/tmp/osint/spiderfoot-results.json"
   if os.path.exists(sf_path):
       with open(sf_path) as f:
           for item in json.load(f):
               findings.append({
                   "source": "spiderfoot",
                   "type": item.get("type", "unknown"),
                   "value": item.get("data", ""),
                   "module": item.get("module", ""),
                   "collected_at": datetime.utcnow().isoformat()
               })
   
   with open("/tmp/osint/normalized-findings.json", "w") as f:
       json.dump(findings, f, indent=2)
   
   print(f"Normalized {len(findings)} findings from {len(set(f['source'] for f in findings))} sources")
   EOF
   python3 /tmp/osint/normalize.py
   

阶段 3 — AI 驱动的关联分析

7. 将规范化后的发现发送给 LLM，进行跨来源关联分析：

   cat > /tmp/osint/correlate.py << 'PYEOF'
   import json, os
   from openai import OpenAI  # or anthropic, ollama, etc.
   
   client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
   
   with open("/tmp/osint/normalized-findings.json") as f:
       findings = json.load(f)
   
   correlation_prompt = f"""You are an OSINT analyst. Analyze these findings collected
   from multiple sources and produce a correlation report.
   
   For each identity or entity you detect:
   1. List all linked accounts/profiles with the evidence connecting them.
   2. Assign a confidence score (0.0-1.0) for each linkage based on:
      - Exact username match across platforms (high)
      - Similar usernames with shared metadata (medium)
      - Same email in breach data and registration (high)
      - Co-occurring infrastructure (IP, domain) (medium)
      - Temporal correlation of account creation dates (low-medium)
   3. Identify contradictions or potential false positives.
   4. Flag high-risk exposures (breached credentials, PII leaks, infrastructure overlaps).
   5. Produce a structured JSON report.
   
   Raw findings:
   {json.dumps(findings[:500], indent=2)}
   """
   
   response = client.chat.completions.create(
       model="gpt-4o",
       messages=[
           {"role": "system", "content": "You are an expert OSINT analyst specializing in identity correlation and link analysis."},
           {"role": "user", "content": correlation_prompt}
       ],
       temperature=0.1,
       response_format={"type": "json_object"}
   )
   
   report = json.loads(response.choices[0].message.content)
   
   with open("/tmp/osint/correlation-report.json", "w") as f:
       json.dump(report, f, indent=2)
   
   print(json.dumps(report, indent=2))
   PYEOF
   python3 /tmp/osint/correlate.py
   

8. 执行实体解析 — 对相关身份进行去重和合并：

   cat > /tmp/osint/resolve.py << 'PYEOF'
   import json
   
   with open("/tmp/osint/correlation-report.json") as f:
       report = json.load(f)
   
   # Extract entities and build a link graph
   entities = report.get("entities", [])
   print(f"Identified {len(entities)} distinct entities")
   for entity in entities:
       name = entity.get("identifier", "unknown")
       confidence = entity.get("confidence", 0)
       links = entity.get("linked_accounts", [])
       risk = entity.get("risk_level", "unknown")
       print(f"  [{confidence:.0%}] {name} — {len(links)} linked accounts — risk: {risk}")
   PYEOF
   python3 /tmp/osint/resolve.py
   

阶段 4 — 报告与可视化

9. 以 Markdown 生成最终情报画像：

   cat > /tmp/osint/report.py << 'PYEOF'
   import json
   from datetime import datetime
   
   with open("/tmp/osint/correlation-report.json") as f:
       report = json.load(f)
   
   md = f"# OSINT Correlation Report\n\n"
   md += f"**Generated:** {datetime.utcnow().isoformat()}Z\n\n"
   md += "## Entity Profiles\n\n"
   
   for entity in report.get("entities", []):
       eid = entity.get("identifier", "Unknown")
       conf = entity.get("confidence", 0)
       md += f"### {eid} (Confidence: {conf:.0%})\n\n"
       md += "| Source | Platform | Evidence |\n|--------|----------|----------|\n"
       for link in entity.get("linked_accounts", []):
           md += f"| {link.get('source','')} | {link.get('platform','')} | {link.get('evidence','')} |\n"
       md += f"\n**Risk Level:** {entity.get('risk_level', 'N/A')}\n\n"
       for flag in entity.get("flags", []):
           md += f"- ⚠️ {flag}\n"
       md += "\n"
   
   with open("/tmp/osint/intelligence-profile.md", "w") as f:
       f.write(md)
   
   print("Report written to /tmp/osint/intelligence-profile.md")
   PYEOF
   python3 /tmp/osint/report.py
   

10. 可选 — 将关联图谱导入 Maltego 进行可视化：

    # 将实体导出为 Maltego 兼容的 CSV，以便手动导入
    cat > /tmp/osint/maltego_export.py << 'PYEOF'
    import json, csv
    
    with open("/tmp/osint/correlation-report.json") as f:
        report = json.load(f)
    
    with open("/tmp/osint/maltego-import.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["Entity Type", "Value", "Linked To", "Link Label", "Confidence"])
        for entity in report.get("entities", []):
            for link in entity.get("linked_accounts", []):
                writer.writerow([
                    link.get("type", "Alias"),
                    link.get("value", ""),
                    entity.get("identifier", ""),
                    link.get("evidence", ""),
                    link.get("confidence", "")
                ])
    
    print("Maltego CSV exported to /tmp/osint/maltego-import.csv")
    PYEOF
    python3 /tmp/osint/maltego_export.py
    

核心概念

概念	说明
跨源关联	在独立的 OSINT 来源之间匹配标识符（用户名、邮箱、IP），以建立实体关联
置信度评分	根据证据强度和佐证情况，为每条关联赋予概率性置信度（0.0–1.0）
实体解析	对分散数据集中指向同一现实实体的记录进行去重与合并
误报检测	利用 AI 推理区分偶然匹配与真实身份关联
多向量情报	将来自社交媒体、DNS、泄露数据和基础设施的发现整合为单一威胁图景
链接分析	基于图的方式检查实体、账户和基础设施之间的关系

工具与系统

工具	在工作流中的角色
Sherlock	跨 400+ 社交平台的用户名枚举
theHarvester	从公开来源发现邮箱、子域名和主机
SpiderFoot	跨 200+ 模块的自动化 OSINT 收集
Maltego	基于图的实体关系可视化
LLM API (GPT-4, Claude, Ollama)	跨源推理、模式检测和置信度评分
HaveIBeenPwned	泄露暴露和凭据泄露检测

常见场景

• 威胁行为者归因： 将网络钓鱼活动中发现的可疑用户名与社交媒体资料、域名注册和泄露数据相关联，以构建归因画像。
• 攻击面测绘： 将发现的子域名、邮箱和员工社交账户联系起来，以描绘组织的完整外部暴露面。
• 内部人员调查： 将员工的已知账户与暗网活动及泄露数据库进行交叉比对。
• 品牌仿冒检测： 通过关联注册模式、命名惯例和时间信号，识别各平台上模仿目标品牌的账户。

输出格式

最终输出为结构化的 JSON 关联报告和 Markdown 情报画像 ```json { "meta": { "target": "targetdomain.com", "sources_used": ["sherlock", "theHarvester", "spiderfoot", "hibp"], "total_findings": 247, "generated_at": "2025-01-15T14:30:00Z" }, "entities": [ { "identifier": "john.target", "confidence": 0.92, "linked_accounts": [ { "source": "sherlock", "platform": "GitHub", "value": "john.target", "evidence": "Exact username match, bio references targetdomain.com", "confidence": 0.95 } ], "risk_level": "high", "flags": [ "Credentials exposed in 2 breaches (2022, 2023)", "Admin email for targetdomain.com found in public WHOIS" ] } ], "contradictions": [], "recommendations": [] }

## 验证

- 在赋予置信度 > 0.8 之前，确认每个关联账户均已对照至少两个来源独立验证。
- 对随机样本（10–20%）交叉检查 AI 关联结果，以验证准确性。
- 验证来自常见名称（如 "admin"、"test"）的误报没有夸大实体画像。
- 确保泄露数据时间戳为最新，且来自信誉良好的聚合源。
- 验证最终报告不包含过时或已撤回的 OSINT 数据。