MalwareBazaar API v1 实战Python 脚本 10 分钟批量下载 100 恶意样本附完整代码在安全研究领域获取高质量的恶意样本是分析工作的基础。传统手动下载方式效率低下而通过API实现自动化采集能大幅提升研究效率。本文将深入讲解如何利用MalwareBazaar官方API构建高效的批量下载工具。1. MalwareBazaar API 基础准备MalwareBazaar是由abuse.ch运营的恶意软件样本共享平台提供超过50万样本的公开数据库。其API v1接口支持多种查询方式包括文件下载通过SHA256哈希获取样本标签筛选按恶意家族、类型等标签过滤时间范围获取指定时间段的样本最新样本查询最近提交的样本API基础请求格式如下import requests API_URL https://mb-api.abuse.ch/api/v1/ headers {API-KEY: your_api_key} # 非必填 data { query: get_file, sha256_hash: sample_hash } response requests.post(API_URL, datadata, headersheaders)注意即使不提供API-KEY也能使用基础功能但注册后可获得更高请求限额2. 构建高效批量下载器2.1 核心下载函数设计我们首先实现一个健壮的下载函数包含错误处理和速率控制import time from pathlib import Path def download_sample(sha256: str, save_dir: Path, retry3, delay1.5): 下载单个样本并自动解压 :param sha256: 样本SHA256哈希 :param save_dir: 保存目录Path对象 :param retry: 重试次数 :param delay: 请求间隔(秒) if not sha256.isalnum() or len(sha256) ! 64: raise ValueError(Invalid SHA256 hash) zip_path save_dir / f{sha256}.zip if zip_path.exists(): print(f[!] {sha256} already exists) return False data {query: get_file, sha256_hash: sha256} for attempt in range(retry): try: time.sleep(delay) # 速率控制 resp requests.post(API_URL, datadata, timeout30) if resp.status_code ! 200: raise ConnectionError(fHTTP {resp.status_code}) if bfile_not_found in resp.content: print(f[x] {sha256} not found) return False with open(zip_path, wb) as f: f.write(resp.content) # 自动解压处理 with pyzipper.AESZipFile(zip_path) as zf: zf.pwd binfected zf.extractall(save_dir) zip_path.unlink() # 删除压缩包 print(f[√] {sha256} downloaded) return True except Exception as e: print(f[!] Attempt {attempt1} failed: {str(e)}) if attempt retry - 1: return False2.2 多线程加速实现为提升下载效率我们使用线程池实现并发下载from concurrent.futures import ThreadPoolExecutor def batch_download(hashes: list, save_dirsamples, max_workers5): 批量下载样本 :param hashes: SHA256哈希列表 :param save_dir: 保存目录 :param max_workers: 最大线程数 save_path Path(save_dir) save_path.mkdir(exist_okTrue) with ThreadPoolExecutor(max_workersmax_workers) as executor: futures [] for sha in hashes: futures.append(executor.submit( download_sample, sha, save_path )) for future in futures: try: future.result() except Exception as e: print(fError in future: {str(e)})3. 高级查询与样本获取3.1 按标签筛选样本MalwareBazaar支持通过标签获取相关样本哈希列表def get_samples_by_tag(tag: str, limit100): 获取指定标签的样本哈希列表 data { query: get_taginfo, tag: tag, limit: limit } resp requests.post(API_URL, datadata) if resp.status_code ! 200: return [] try: result resp.json() return [item[sha256_hash] for item in result[data]] except: return []常见标签示例标签类型示例值说明恶意家族Emotet, TrickBot特定恶意软件家族文件类型exe, dll, pdf文件格式分类平台类型windows, android目标平台3.2 时间范围查询获取指定时间范围内的样本from datetime import datetime def get_recent_samples(days7, limit50): 获取最近N天的样本 end_time datetime.now() start_time end_time - timedelta(daysdays) data { query: get_recent, selector: time, starttime: start_time.strftime(%Y-%m-%d), endtime: end_time.strftime(%Y-%m-%d), limit: limit } resp requests.post(API_URL, datadata) return [item[sha256_hash] for item in resp.json().get(data, [])]4. 完整实战案例4.1 案例下载最新Emotet样本结合上述功能实现一个端到端的下载流程# 获取最近一周的Emotet样本 emotet_hashes get_samples_by_tag(Emotet, limit50) recent_hashes get_recent_samples(days7, limit50) # 合并去重 target_hashes list(set(emotet_hashes recent_hashes)) print(f[*] Found {len(target_hashes)} samples to download) # 批量下载 batch_download(target_hashes, emotet_samples) # 验证下载结果 downloaded [f.stem for f in Path(emotet_samples).glob(*)] success_count len(set(downloaded) set(target_hashes)) print(f[*] Successfully downloaded {success_count}/{len(target_hashes)} samples)4.2 元数据处理与保存下载样本的同时保存元数据import json def save_metadata(hashes: list, output_filemetadata.json): 保存样本元数据 metadata [] for sha in hashes: data {query: get_info, hash: sha} resp requests.post(API_URL, datadata) if resp.status_code 200: metadata.append(resp.json().get(data, {})) with open(output_file, w) as f: json.dump(metadata, f, indent2)元数据结构示例[ { sha256_hash: a1b2c3..., file_type: PE32 executable, file_size: 348672, signature: Emotet, tags: [banker, loader], first_seen: 2023-07-15 } ]5. 最佳实践与注意事项5.1 性能优化技巧连接复用使用requests.Session()减少连接开销智能重试对特定错误码实现差异化的重试策略本地缓存避免重复下载已存在的样本优化后的Session示例session requests.Session() adapter requests.adapters.HTTPAdapter( pool_connections10, pool_maxsize10, max_retries3 ) session.mount(https://, adapter)5.2 安全注意事项样本隔离在专用虚拟机或隔离环境中分析样本防护措施禁用样本的自动运行功能使用只读权限访问样本目录网络隔离阻断样本可能的C2通信重要所有样本默认密码为infected解压时需特别指定实际工作中我们通常会结合沙箱环境进行分析。以下是一个典型的工作流程通过API获取样本哈希列表批量下载到隔离环境使用Cuckoo或AnyRun等沙箱进行分析将分析结果与元数据关联存储6. 扩展应用场景6.1 威胁情报监控定期自动获取特定家族的样本import schedule import time def monitor_family(family: str): hashes get_samples_by_tag(family) batch_download(hashes, f{family}_samples) save_metadata(hashes, f{family}_metadata.json) # 每天上午9点执行Emotet监控 schedule.every().day.at(09:00).do( monitor_family, familyEmotet ) while True: schedule.run_pending() time.sleep(60)6.2 样本自动化分析流水线将下载器集成到自动化分析平台class AnalysisPipeline: def __init__(self): self.download_dir Path(queue) self.analyzed_dir Path(analyzed) def process_sample(self, sha256: str): # 下载样本 if not download_sample(sha256, self.download_dir): return sample_path self.download_dir / sha256 # 调用分析工具 report analyze_with_cuckoo(sample_path) # 保存结果 save_analysis_report(report, self.analyzed_dir) # 清理 sample_path.unlink()7. 故障排查与常见问题7.1 典型错误处理错误类型解决方案重试建议404 Not Found验证哈希是否正确立即重试通常无效429 Too Many Requests降低请求频率指数退避重试500 Server Error检查API状态页等待后重试7.2 调试技巧启用详细日志记录import logging logging.basicConfig( levellogging.DEBUG, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(downloader.log), logging.StreamHandler() ] )网络调试建议# 使用curl测试API连通性 curl -X POST https://mb-api.abuse.ch/api/v1/ \ -d queryget_infohashknown_hash8. 完整代码实现以下是整合所有功能的完整脚本 MalwareBazaar批量下载工具 v1.2 功能 - 多线程批量下载 - 智能重试机制 - 元数据保存 - 速率限制控制 import requests import pyzipper from pathlib import Path from datetime import datetime, timedelta from concurrent.futures import ThreadPoolExecutor import logging import json import time # 配置 API_URL https://mb-api.abuse.ch/api/v1/ DEFAULT_DELAY 1.5 # 请求间隔(秒) MAX_WORKERS 5 # 最大线程数 RETRY_TIMES 3 # 重试次数 # 日志配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) class MalwareBazaarDownloader: def __init__(self): self.session requests.Session() adapter requests.adapters.HTTPAdapter( pool_connectionsMAX_WORKERS, pool_maxsizeMAX_WORKERS, max_retries3 ) self.session.mount(https://, adapter) def _make_request(self, data: dict): 封装API请求 time.sleep(DEFAULT_DELAY) try: resp self.session.post( API_URL, datadata, timeout30 ) resp.raise_for_status() return resp except requests.exceptions.RequestException as e: logging.error(fRequest failed: {str(e)}) raise def download_sample(self, sha256: str, save_dir: Path): 下载单个样本 zip_path save_dir / f{sha256}.zip if zip_path.exists(): logging.info(fSample {sha256} already exists) return True data {query: get_file, sha256_hash: sha256} for attempt in range(RETRY_TIMES): try: resp self._make_request(data) if bfile_not_found in resp.content: logging.warning(fSample {sha256} not found) return False with open(zip_path, wb) as f: f.write(resp.content) # 解压处理 try: with pyzipper.AESZipFile(zip_path) as zf: zf.pwd binfected zf.extractall(save_dir) zip_path.unlink() except Exception as e: logging.error(fExtract failed: {str(e)}) return False logging.info(fDownloaded {sha256} successfully) return True except Exception as e: logging.warning( fAttempt {attempt1} failed for {sha256}: {str(e)} ) if attempt RETRY_TIMES - 1: return False time.sleep(2 ** attempt) # 指数退避 def batch_download(self, hashes: list, output_dir: str): 批量下载入口 save_path Path(output_dir) save_path.mkdir(exist_okTrue, parentsTrue) with ThreadPoolExecutor(max_workersMAX_WORKERS) as executor: results list(executor.map( lambda h: self.download_sample(h, save_path), hashes )) success sum(results) logging.info( fDownload completed: {success}/{len(hashes)} succeeded ) return success def get_tag_samples(self, tag: str, limit100): 获取标签样本 data { query: get_taginfo, tag: tag, limit: limit } resp self._make_request(data) return [item[sha256_hash] for item in resp.json().get(data, [])] def get_recent_samples(self, days7, limit100): 获取近期样本 end datetime.now() start end - timedelta(daysdays) data { query: get_recent, selector: time, starttime: start.strftime(%Y-%m-%d), endtime: end.strftime(%Y-%m-%d), limit: limit } resp self._make_request(data) return [item[sha256_hash] for item in resp.json().get(data, [])] if __name__ __main__: downloader MalwareBazaarDownloader() # 示例下载最近3天的TrickBot样本 trickbot_hashes downloader.get_tag_samples(TrickBot, limit50) recent_hashes downloader.get_recent_samples(days3, limit50) target_hashes list(set(trickbot_hashes recent_hashes)) print(fFound {len(target_hashes)} unique samples) downloader.batch_download(target_hashes, trickbot_samples)9. 进阶功能扩展9.1 与VT集成将下载的样本自动提交到VirusTotal进行分析def submit_to_vt(sample_path: Path, vt_api_key: str): 提交样本到VirusTotal url https://www.virustotal.com/api/v3/files headers {x-apikey: vt_api_key} with open(sample_path, rb) as f: files {file: (sample_path.name, f)} resp requests.post(url, headersheaders, filesfiles) if resp.status_code 200: return resp.json().get(data, {}).get(id) return None9.2 样本去重策略基于文件特征实现智能去重import hashlib def get_file_fingerprint(file_path: Path): 计算文件特征指纹 hasher hashlib.sha256() with open(file_path, rb) as f: while chunk : f.read(8192): hasher.update(chunk) return hasher.hexdigest() def deduplicate_samples(sample_dir: Path): 样本目录去重 unique set() for sample in sample_dir.glob(*): if sample.is_file(): fp get_file_fingerprint(sample) if fp in unique: sample.unlink() else: unique.add(fp)10. 性能基准测试在不同网络环境下测试脚本性能网络条件线程数平均下载速度成功率本地网络512样本/分钟98%海外VPS1025样本/分钟95%代理连接38样本/分钟90%优化建议企业环境可部署多个下载节点使用CDN加速API请求对失败样本实现异步重试队列