威胁情报源与恶意样本批量获取：从情报源选型到API实操

做威胁情报或样本分析最怕的不是样本分析不出来而是连样本都找不到。很多时候一个可疑样本到你手上你要先回答三个问题这个文件有没有被公开标记过属于哪个家族跟哪个组织有关联要回答这些问题既靠经验也靠日常积累的情报源。这篇文章就当是把我这些年常用的恶意样本、APT样本、APT组织信息获取站点以及快速批量爬取信息、下载样本的方法整理成一份可以照着用的记录。适合刚入门的威胁分析师、蓝队同学也适合自己在做样本研究的独立安全工程师。先说清楚一件事这些站点和API都是公开的威胁情报资源日常安全研究、事件响应、样本分析完全够用。但“能够访问”不等于“可以乱来”任何爬取和批量拉取都必须遵守目标站点的API条款、robots协议和相关规则控制请求频率注意数据使用边界。下面直接进入正题。1. 情报源选型先搞清三类情报源再动手我见过不少新人一上来就抱着一个网站猛爬爬到被限流甚至封IP然后才开始思考“我到底想要什么数据”。做情报收集第一步不是写代码而是先想清楚你要解决什么问题。按照我自己的习惯情报源可以分成三类每一类解决一类具体问题。1.1 样本库源解决“有没有人看过这个文件”第一类是样本库和沙箱平台比如MalwareBazaar、VirusTotal、Hybrid Analysis、Any.run这些。判断一个文件是不是恶意样本、有没有被其他人提交过通常看检测结果、行为报告和社区标签就够了。这类平台的核心价值在于“去重”和“初筛”。拿到一个样本先查hash如果已经有几十个杀毒引擎报毒、有现成的行为分析报告那说明这份样本已经被研究过你可以直接把精力放到写报告、提炼IOC、分析关联关系上而不是从头开虚拟机做行为分析。用图书管理来类比样本库就是图书馆的检索目录你不需要把每本书从头读一遍先看卡片、看摘要确定值得精读再花时间。1.2 组织与战术情报源解决“谁在打你、用什么手法”第二类是组织情报和战术情报源比如MITRE ATTCK、AlienVault OTX以及各大安全厂商的APT报告。这类源回答的问题是这个样本背后的组织是谁攻击链怎么打的基础设施长什么样对应的检测点在哪。这里要注意APT组织命名很混乱。同一个组织卡巴斯基叫“Turla”赛门铁克叫“Venomous Bear”厂商内部还有不同的命名编号体系。所以做组织信息收集时不能只看一家报告要交叉验证把不同报告中对同一组织的描述映射到一起。这也是情报工作最容易踩的坑后面我会单独说。1.3 为什么优先用API而不是写爬虫很多人一听“批量爬取”第一反应就是写爬虫去抓网页。但我的经验是能走API就别爬网页除非完全没有API。原因有几个网页结构说变就变。你今天写好的CSS选择器明天网站改版就全废了维护成本极高。爬虫的请求特征很容易被识别触发反爬、验证码、IP封禁都是常有的事。API有正式的限速配额只要别超频稳定得多。数据源合规角度API明确规定了数据使用范围和频率正常使用一般没有争议爬虫在边界模糊时反而容易惹麻烦。所以整篇文章的核心思路就是优先梳理各平台的API能力用API去批量取数必须要兜底的时候再退回到带合理延时的页面爬取方案。2. 常用网站清单与实战场景2.1 样本与沙箱类平台对比下面这个表是我自己常用的样本相关站点重点看有没有样本下载能力、有没有API、免费额度怎么样。表格只能给一个大方向具体配额会随时间变化要以官方文档为准。平台定位是否能下载样本是否提供API适合场景MalwareBazaar恶意样本共享库是是快速下载已知样本、按标签/签名检索样本VirusTotal多引擎检测与关联分析需要授权是hash查询、关联分析、大规模批量检测Hybrid Analysis沙箱行为分析部分是查看行为报告、提取主机IOCAny.run在线交互式沙箱付费可下载是交互式分析、快速看行为链Triage沙箱平台是是自动化批量沙箱分析Joe Sandbox深度沙箱是是高级样本行为分析我实际用得最多的是MalwareBazaar和VirusTotal。MalwareBazaar最大的好处是下载样本几乎没门槛——公开的恶意样本你只要按hash请求就能拿非常适合搭一个本地样本库。VirusTotal则胜在关联数据丰富一个文件hash丢进去能看到谁提交过、哪些引擎报毒、关联的域名IP、对应的行为标签是做样本拓线的好帮手。2.2 组织情报与ATTCK战术库组织情报这块我主要看这几个方向MITRE ATTCK这不是样本库而是一套战术、技术和过程知识库。它的价值在于给你一套“通用语言”描述攻击行为时用ATTCK编号比如T1059命令解释器、T1566钓鱼附件大家都能看懂。另外ATTCK的Groups和Software条目会列出已知组织使用的工具和样本很多时候可以顺着组织名去查关联的样本hash。AlienVault OTXOTX很适合做威胁情报的“故事化”收集。它把IOC封装成Pulse每个Pulse对应一个攻击主题或事件里面包含相关的域名、IP、hash、规则片段。你可以订阅某些组织或行业相关的Pulse然后通过API持续拉取。厂商APT报告卡巴斯基、赛门铁克、Palo Alto Unit 42、ESET、Mandiant这些厂商会周期性发布APT报告里面的IOC域名、IP、hash、Yara规则是最新鲜的情报源。建议直接订阅厂商的RSS或威胁情报邮件列表报告出来后第一时间提取IOC入库。表格对比信息来源内容特点更新频率使用建议MITRE ATTCK战术、组织、软件知识库季度更新建立分析语言、关联组织AlienVault OTXPulse化IOC集合持续更新自动化拉取、导入SIEM/MISP厂商APT报告深度分析报告IOC不定期人工精读提取IOC手工入库ThreatFox恶意软件指标共享持续更新快速检索新型恶意软件相关IOC2.3 情报源使用的通用要点无论用哪个平台有几个通用习惯需要养成。第一API key不要硬编码在脚本里用环境变量或者配置文件管理。我见过有人在群里发截图把key漏出来的那真的是事故。第二本地要建一个“已拉取hash”的缓存表。情报拉取经常会重复做个缓存数据库按hash去重可以省大量配额。第三拉下来的数据一定要保留原始JSON。不要只抽几个字段存进自己的表里——原始响应里往往藏着当时没注意到的字段以后做关联分析会用到。第四执行批量下载前先确认目标文件确实是恶意样本并且后续操作在隔离环境中进行。样本下载不是闹着玩的后面第4章我会专门说安全注意事项。3. 快速批量爬取信息与下载样本的实操方法下面进入实操环节。我会给三个最常用的批量方法从MalwareBazaar批量下载样本、用VirusTotal批量查询文件属性、用OTX拉取组织威胁情报。代码以Python为主依赖requests库所有代码基于API公开文档编写。3.1 用MalwareBazaar API批量下载样本MalwareBazaar的API比较直接接口地址是https://mb-api.abuse.ch/api/v1/所有的操作都是POST请求通过query参数区分功能。我们这里需要两个操作get_info用来校验hash信息get_file用来下载样本。先说验证hash是否有效。如果手里有一批hash最优的做法是先做一次批量查询筛掉那些在库中不存在的hash避免白白下载。示例代码如下import requests import time import json API_URL https://mb-api.abuse.ch/api/v1/ HEADERS {User-Agent: threat-intel-research/0.1} def query_malwarebazaar(file_hash): data { query: get_info, hash_file: file_hash } resp requests.post(API_URL, datadata, headersHEADERS, timeout30) if resp.status_code 200: result resp.json() if result.get(query_status) ok: # 提取关键信息 entry result[data][0] return { hash: entry.get(sha256_hash), signature: entry.get(signature), tags: entry.get(tags), first_seen: entry.get(first_seen), file_name: entry.get(intelligence, {}).get(uploads, [{}])[0].get(file_name) } return None实际跑的时候建议先做一个get_info批量查询把不存在的hash直接过滤掉再进入下载环节。下载样本本身更简单把query字段改成get_file即可def download_sample(file_hash, save_dirsamples): import hashlib data { query: get_file, hash_file: file_hash } resp requests.post(API_URL, datadata, headersHEADERS, timeout60) # 如果请求返回的是JSON说明有错误如果是二进制流说明下载成功 content_type resp.headers.get(Content-Type, ) if resp.status_code 200 and json not in content_type: save_path f{save_dir}/{file_hash}.bin with open(save_path, wb) as f: f.write(resp.content) # 校验一致性确保下载的文件hash与目标hash一致 local_hash hashlib.sha256(resp.content).hexdigest() return local_hash.lower() file_hash.lower(), save_path else: return False, resp.text这段代码里有几个关键点。下载成功与否不能只看状态码因为MalwareBazaar在出错时也可能返回200但body是JSON格式错误信息所以要检查Content-Type。另外下载后必须做本地sha256校验防止文件损坏或者被中间过程污染。我一般会把待下载的hash放在一个文本文件里逐行读取每下载一个后sleep 2秒钟。这是对不同情报源的尊重也能避免触发限流。完整的批量循环可以这样写with open(hash_list.txt, r) as f: hashes [line.strip() for line in f if line.strip()] for h in hashes: success, result download_sample(h) if success: print(f[] {h} - {result}) else: print(f[-] {h} - {result}) time.sleep(2)实测下来MalwareBazaar的查询和下载接口对普通研究请求比较宽容但如果你一天拉几万个样本还是要小心一点。建议批量控制在几千个以内分多天完成。3.2 用VirusTotal API批量查询文件情报VirusTotal API v3比旧版v2规范了很多所有文件检索都是GET请求路径是/api/v3/files/{hash}。免费API key有配额所以策略上要更精细。一个经典的批量查询流程是读取一个txt文件里的hash列表逐个查询VirusTotal把检测结果写入CSV同时做本地去重。代码可以这么写import requests import time import csv import os VT_API_KEY os.environ.get(VT_API_KEY) API_URL https://www.virustotal.com/api/v3/files/{file_hash} HEADERS {x-apikey: VT_API_KEY} def vt_query(hash_file): url API_URL.format(file_hashhash_file) resp requests.get(url, headersHEADERS, timeout30) if resp.status_code 200: data resp.json() attrs data[data][attributes] stats attrs.get(last_analysis_stats, {}) return { hash: hash_file, malicious: stats.get(malicious, 0), suspicious: stats.get(suspicious, 0), harmless: stats.get(harmless, 0), undetected: stats.get(undetected, 0), meaningful_name: attrs.get(meaningful_name, ), type_description: attrs.get(type_description, ) } elif resp.status_code 404: # 文件hash没有被收录 return {hash: hash_file, status: not found} else: return {hash: hash_file, status: ferror {resp.status_code}}使用VT的时候有一个容易被忽略的细节同一个hash你查了很多次会占用很多配额。所以在循环里加一个内存缓存已经查过的hash就不重复请求了。cache set() results [] for h in hash_list: h h.strip() if h in cache: continue cache.add(h) result vt_query(h) results.append(result) time.sleep(1.5) # 控制频率避免触发限流关于限额VT免费key的配额非常紧张按我最近一次测试每分钟只有个位数次请求额度而且每天总量有限。如果你有几百个hash要查可以考虑把脚本分散到多个时间段跑或者直接升级到付费的Intelligence那样还能用到更高级的关联图和检索语法。3.3 用AlienVault OTX拉取APT组织威胁情报OTX的特色是Pulse机制。一个Pulse里包含一批IOC还有描述、标签、参考链接。对于APT组织信息收集你只要找到对应的Pulse ID就能把里面所有的IOC一次拉出来。API请求方式如下import requests import os OTX_KEY os.environ.get(OTX_API_KEY) PULSE_ID 566edbb4f6ca2e8e0e6c2e9a # 示例ID实际使用时替换 url fhttps://otx.alienvault.com/api/v1/pulses/{PULSE_ID} headers {X-OTX-API-KEY: OTX_KEY} resp requests.get(url, headersheaders, timeout30) data resp.json() print(Pulse名称:, data.get(name)) print(攻击目标行业:, data.get(attack_ids)) for indicator in data.get(indicators, []): print(indicator[type], indicator[indicator])OTX还支持按订阅拉取你可以关注特定用户的Pulse然后调/api/v1/pulses/subscribed来获取所有订阅的最新动态。这个接口适合做持续监控每天定时跑一次把新增IOC入库。我自己的做法是把OTX的IOC输出保存成CSV然后导入到本地MISP或者简单的SQLite库里这样就能和其他来源的数据做关联。OTX的IOC类型包含IPv4、域名、主机名、URL、文件hash、email、CVE编号字段比较全面。成本上OTX免费API key的配额比VT慷慨正常研究完全够用。3.4 没有API时的页面爬取兜底方案有些情报站没有开放API或者部分数据接口需要付费这时候只能退回到页面爬取。我建议的兜底方案是“静态页面 合理延时 遵守robots”能不做动态渲染抓取就不做否则你会陷入selenium的泥潭。以简单静态页面为例用requests加BeautifulSoup就够import requests from bs4 import BeautifulSoup import time headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 topic-research/0.1 } def fetch_and_parse(url): resp requests.get(url, headersheaders, timeout30) resp.raise_for_status() soup BeautifulSoup(resp.text, html.parser) # 提取目标数据比如表格行 rows soup.select(table tr) return rows for page in range(1, 5): url fhttps://example.com/list?page{page} rows fetch_and_parse(url) print(page, len(rows)) time.sleep(3)这条方案只适合目标站点结构简单、更新频率低的场景。遇到前端渲染的页面我宁愿去翻了开发者工具里的XHR接口找到背后的JSON数据接口再抓这通常比渲染整个页面更省资源、更不容易被识别为爬虫。4. 常见问题与情报分析技巧4.1 API限流与错误码排查情报API用多了总会遇到各种错误。我总结了一个速查表基本覆盖常见情况。错误码或表现常见原因解决方法401 UnauthorizedAPI key错误或未提供检查环境变量、key权限403 Forbiddenkey被禁用或没有该数据权限查看套餐权限联系平台404 Not Found目标数据不存在确认hash大小写、确认ID格式429 Too Many Requests请求频率超过配额增加sleep时间考虑排队分批跑503 Service Unavailable平台临时过载退避重试间隔拉长到分钟级超时/连接中断网络问题或单次请求量太大加超时控制重试3次后放弃一个排查思路供参考所有API请求先打印状态码和一个响应截断体不要直接丢进异常处理里。很多问题一眼就能从状态码看出来省得反复在代码里瞎试。4.2 下载恶意样本前必须做的安全措施这部分必须单独强调恶意样本不是普通文件下载和存储都有安全要求。下载前先做三件事。第一确认hash本身可信。优先用厂商报告或官方IOC来源提供的hash不要从不知名网站随便拿hash去下载以防被投毒。第二设置隔离存储目录。本地建一个专门的虚拟环境文件夹不给样本执行权限文件名用hash重命名不要用原始文件名。所有样本统一打包时用密码压缩社区常见约定口令是infected这个口令既不是秘密也不影响杀毒软件识别样本只是防止日常误打开。第三分析环境必须是快照虚拟机或专用沙箱。下载完成后不要在宿主机上直接双击。我的固定流程是下载到隔离目录把文件传到快照虚拟机里做行为分析虚拟机断外网使用伪造的网络服务或者只允许DNS解析到本地模拟器避免样本回连真实C2。对于没有行为分析条件的场景直接把样本提交到Any.run或Triage这类在线沙箱也比自己盲跑强得多。强调一句样本分析不是“跑一下看结果”那么轻松的事。恶意样本可能带有反调试、反虚拟机、反沙箱代码很多行为要结合静态分析、动态监控、流量捕捉三方面交叉验证才能下结论。4.3 用贝叶斯思路给样本做冷情报筛选最后分享一个我做情报批量处理时的小技巧正好可以回答“拿到一大批样本怎么快速排序优先级”的问题。核心不是用深度学习而是用朴素贝叶斯建模把冷冰冰的IOC元数据变成可以排序的风险概率。思路很简单。对每一个样本提取几个特征比如是否命中已知威胁标签、VT检测引擎报毒数量、文件类型是否属于常见恶意类型如exe、dll、js、lnk、首次提交时间距今的天数、是否匹配ATTCK的某个软件条目。这些特征都可以数值化。有了特征矩阵之后用历史已标注样本作为训练集把“已知家族”做标签训练一个朴素贝叶斯分类器。每个新样本进来计算它属于某个已知家族的后验概率按概率排序。概率高的优先深入分析概率低的先放一放。核心代码思路如下# 简化示例对样本特征做贝叶斯风险排序 import pandas as pd from sklearn.naive_bayes import GaussianNB from sklearn.model_selection import train_test_split # train.csv / test.csv 来自历史已标注情报数据 train_df pd.read_csv(train.csv) test_df pd.read_csv(test.csv) features [vt_malicious_count, tag_hit_count, is_exe, days_since_first_seen] x_train train_df[features] y_train train_df[family_label] model GaussianNB() model.fit(x_train, y_train) x_test test_df[features] test_df[risk_score] model.predict_proba(x_test)[:, 1] test_df test_df.sort_values(risk_score, ascendingFalse) print(test_df.head(20))这里要注意朴素贝叶斯假设特征相互独立现实中并不完全成立所以它更适合做初筛排序而不是作为最终判定依据。另一个陷阱是训练集和测试集的时间分布要尽量接近不要拿半年前的样本分布预测现在的新样本IOC更新非常快模型需要定期重训。这个思路对“批量获取样本后的整理环节”特别有用。爬虫拿回来几千个文件不可能挨个分析用模型先算一遍风险分值能让你把有限的人工精力投入在最关键的样本上。最后分享一个实战习惯说实话把情报源跑通只是第一步真正让你效率提升的反而是管理。我自己维护了一个简单的SQLite库里面存了所有拉取过的hash、来源平台、拉取时间、检测结果和对应Pulse编号。每次新样本进来先查库如果已经存在就直接看历史结论省掉的重复工作远比想象的要多。后来这套逻辑越用越顺手我又把OTX、厂商报告IOC都接进来做成了一个轻量级的情报汇总表。整体上恶意样本和研究情报的拉取过程如果养成“先缓存、再入库、最后分析”的习惯你会少走很多弯路。如果你打算长期做威胁情报后续可以考虑把整套流程接到MISP或OpenCTI这类开源平台里让情报源自动跑起来那样效果会更好。