FOFA API v2.0 资产采集实战：Python 脚本 3 步自动化处理与 Rad 爬虫对接

FOFA API v2.0 资产采集与自动化处理实战指南1. 现代资产采集的技术演进在网络安全领域资产采集已经从传统的手工枚举发展为智能化的自动化流程。作为安全研究员我们需要掌握如何高效获取目标资产信息并将其转化为可操作的扫描目标。FOFA作为领先的网络空间测绘引擎其API v2.0版本提供了更强大的查询能力和更灵活的数据返回格式。资产采集的核心价值在于精准定位通过语法组合锁定特定技术栈的资产批量获取一次性收集数百甚至上千条相关资产数据持续监控定期执行采集任务发现新增暴露面# FOFA基础查询示例 import requests FOFA_EMAIL your_emaildomain.com FOFA_KEY your_api_key_here base_url https://fofa.info/api/v2/search query domainexample.com status_code200 fields host,title,ip,port,server size 100 # 每页结果数 page 1 # 页码 params { email: FOFA_EMAIL, key: FOFA_KEY, qbase64: base64.b64encode(query.encode()).decode(), fields: fields, size: size, page: page } response requests.get(base_url, paramsparams) if response.status_code 200: data response.json() print(f获取到 {len(data[results])} 条结果)2. FOFA API v2.0 深度解析2.1 API核心功能对比功能特性v1.0版本v2.0增强点查询语法基础支持支持更复杂的逻辑组合结果字段固定字段可自定义返回字段分页机制简单分页支持大结果集高效遍历速率限制较严格按账户等级动态调整数据新鲜度延迟更新近实时数据返回2.2 高效查询构建技巧构建优质FOFA查询需要关注三个维度目标特征识别目标的独特标识备案号、特定标题等技术指纹通过组件特征缩小范围中间件、框架等业务场景结合业务逻辑推测可能的资产分布# 高级查询构建示例 def build_fofa_query(target_domain, tech_stackNone): base_query fdomain{target_domain} if tech_stack: if tech_stack Spring: base_query headerX-Application-Context elif tech_stack ThinkPHP: base_query bodyThinkPHP # 排除常见干扰项 base_query !body403 Forbidden return base_query3. Python自动化处理流水线3.1 资产标准化处理采集到的原始数据往往需要经过清洗才能用于后续工具链。典型处理流程包括协议补充为IP或域名添加http/https前缀端口归一化根据服务类型标准化URL格式去重处理消除因CDN等导致的重复资产# URL标准化处理脚本 import re from urllib.parse import urlparse def normalize_assets(raw_assets): processed set() for asset in raw_assets: # 处理IP形式 if re.match(r^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(:\d)?$, asset): if : in asset: ip, port asset.split(:) processed.add(fhttp://{ip}:{port}) processed.add(fhttps://{ip}:{port}) else: processed.add(fhttp://{asset}) processed.add(fhttps://{asset}) # 处理域名形式 else: if not asset.startswith((http://, https://)): processed.add(fhttp://{asset}) processed.add(fhttps://{asset}) else: parsed urlparse(asset) if not parsed.path or parsed.path /: processed.add(asset) return sorted(processed)3.2 与Rad爬虫的高效对接Rad作为现代爬虫工具对输入格式有特定要求。我们需要确保每个URL独立一行支持HTTP/HTTPS协议保留端口号信息非标准端口时# Rad输入文件生成器 def generate_rad_input(fofa_results, output_filerad_input.txt): with open(output_file, w) as f: for result in fofa_results: host result.get(host, ) port str(result.get(port, 80)) if not host: continue if port in [80, 443]: scheme https if port 443 else http f.write(f{scheme}://{host}\n) else: f.write(fhttp://{host}:{port}\n) f.write(fhttps://{host}:{port}\n) print(f生成Rad输入文件{output_file})4. 实战构建端到端自动化流程4.1 完整工作流设计资产采集层FOFA API定时任务数据处理层自动清洗和标准化爬虫调度层并行启动Rad实例结果聚合层合并爬取结果供后续分析# 端到端自动化脚本框架 import json from concurrent.futures import ThreadPoolExecutor class AssetPipeline: def __init__(self, config): self.config config self.raw_assets [] self.processed_assets [] def fetch_from_fofa(self, query): # 实现FOFA API调用 pass def process_assets(self): # 实现资产标准化 pass def run_rad_crawler(self, input_file): # 实现Rad调用 pass def execute(self): print(启动资产采集流水线) self.fetch_from_fofa(self.config[fofa_query]) self.process_assets() self.run_rad_crawler(rad_input.txt) print(流水线执行完成) # 示例配置 config { fofa_query: domainexample.com, concurrency: 5 } pipeline AssetPipeline(config) pipeline.execute()4.2 性能优化技巧异步请求使用aiohttp提升API调用效率结果缓存避免重复查询相同条件智能重试对失败请求实现指数退避重试资源限制根据硬件条件调整并发度# 带重试机制的API调用 import time from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry def setup_session(retries3, backoff_factor0.3): session requests.Session() retry Retry( totalretries, readretries, connectretries, backoff_factorbackoff_factor, status_forcelist(500, 502, 504) ) adapter HTTPAdapter(max_retriesretry) session.mount(http://, adapter) session.mount(https://, adapter) return session # 使用示例 session setup_session() response session.get(https://fofa.info/api/v2/search, paramsparams)5. 高级应用场景5.1 持续监控系统构建通过将FOFA采集与自动化处理结合可以构建资产监控系统基线建立首次全量采集目标资产增量检测定期执行差异分析告警触发发现新暴露面时实时通知# 简易监控系统核心逻辑 class AssetMonitor: def __init__(self, baseline_file): self.baseline self.load_baseline(baseline_file) self.current set() def load_baseline(self, filepath): with open(filepath) as f: return set(line.strip() for line in f) def detect_changes(self, new_assets): new_assets set(new_assets) added new_assets - self.baseline removed self.baseline - new_assets if added: print(f发现 {len(added)} 个新增资产) with open(new_assets.txt, w) as f: f.write(\n.join(added)) if removed: print(f发现 {len(removed)} 个资产下线) return added, removed5.2 多工具链集成现代安全研究需要多种工具协同工作典型集成模式包括FOFA → Rad → Xray全自动化漏洞发现FOFA → Nuclei针对性漏洞检测FOFA → Custom Scanner定制化扫描方案# 工具链集成示例 def run_full_scan(target_domain): # 步骤1FOFA资产采集 fofa_results fetch_from_fofa(fdomain{target_domain}) # 步骤2数据处理 normalized normalize_assets([r[host] for r in fofa_results]) # 步骤3Rad爬虫 generate_rad_input(normalized, rad_input.txt) os.system(rad -t 10 -f rad_input.txt -o rad_output) # 步骤4Xray扫描 os.system(xray webscan --listen 127.0.0.1:7777 --html-output xray_report.html) print(全流程扫描完成结果保存在xray_report.html)在实际项目中这套方法帮助我们将资产收集效率提升了3-5倍同时减少了人为错误。关键在于保持脚本的模块化设计便于针对不同场景调整流程。