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

FOFA API v2.0 资产采集与自动化处理实战指南

1. 现代资产采集的技术演进

在网络安全领域，资产采集已经从传统的手工枚举发展为智能化的自动化流程。作为安全研究员，我们需要掌握如何高效获取目标资产信息，并将其转化为可操作的扫描目标。FOFA作为领先的网络空间测绘引擎，其API v2.0版本提供了更强大的查询能力和更灵活的数据返回格式。

资产采集的核心价值在于：

• 精准定位：通过语法组合锁定特定技术栈的资产
• 批量获取：一次性收集数百甚至上千条相关资产数据
• 持续监控：定期执行采集任务，发现新增暴露面

# FOFA基础查询示例 import requests FOFA_EMAIL = "your_email@domain.com" FOFA_KEY = "your_api_key_here" base_url = "https://fofa.info/api/v2/search" query = 'domain="example.com" && status_code="200"' 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, params=params) if response.status_code == 200: data = response.json() print(f"获取到 {len(data['results'])} 条结果")

2. FOFA API v2.0 深度解析

2.1 API核心功能对比

2.2 高效查询构建技巧

构建优质FOFA查询需要关注三个维度：

1. 目标特征：识别目标的独特标识（备案号、特定标题等）
2. 技术指纹：通过组件特征缩小范围（中间件、框架等）
3. 业务场景：结合业务逻辑推测可能的资产分布

# 高级查询构建示例 def build_fofa_query(target_domain, tech_stack=None): base_query = f'domain="{target_domain}"' if tech_stack: if tech_stack == "Spring": base_query += ' && header="X-Application-Context"' elif tech_stack == "ThinkPHP": base_query += ' && body="ThinkPHP"' # 排除常见干扰项 base_query += ' && !body="403 Forbidden"' return base_query

3. Python自动化处理流水线

3.1 资产标准化处理

采集到的原始数据往往需要经过清洗才能用于后续工具链。典型处理流程包括：

1. 协议补充：为IP或域名添加http/https前缀
2. 端口归一化：根据服务类型标准化URL格式
3. 去重处理：消除因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(f"http://{ip}:{port}") processed.add(f"https://{ip}:{port}") else: processed.add(f"http://{asset}") processed.add(f"https://{asset}") # 处理域名形式 else: if not asset.startswith(('http://', 'https://')): processed.add(f"http://{asset}") processed.add(f"https://{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_file="rad_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(f"http://{host}:{port}\n") f.write(f"https://{host}:{port}\n") print(f"生成Rad输入文件：{output_file}")

4. 实战：构建端到端自动化流程

4.1 完整工作流设计

1. 资产采集层：FOFA API定时任务
2. 数据处理层：自动清洗和标准化
3. 爬虫调度层：并行启动Rad实例
4. 结果聚合层：合并爬取结果供后续分析

# 端到端自动化脚本框架 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": 'domain="example.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(retries=3, backoff_factor=0.3): session = requests.Session() retry = Retry( total=retries, read=retries, connect=retries, backoff_factor=backoff_factor, status_forcelist=(500, 502, 504) ) adapter = HTTPAdapter(max_retries=retry) session.mount('http://', adapter) session.mount('https://', adapter) return session # 使用示例 session = setup_session() response = session.get('https://fofa.info/api/v2/search', params=params)

5. 高级应用场景

5.1 持续监控系统构建

通过将FOFA采集与自动化处理结合，可以构建资产监控系统：

1. 基线建立：首次全量采集目标资产
2. 增量检测：定期执行差异分析
3. 告警触发：发现新暴露面时实时通知

# 简易监控系统核心逻辑 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, removed

5.2 多工具链集成

现代安全研究需要多种工具协同工作，典型集成模式包括：

• FOFA → Rad → Xray：全自动化漏洞发现
• FOFA → Nuclei：针对性漏洞检测
• FOFA → Custom Scanner：定制化扫描方案

# 工具链集成示例 def run_full_scan(target_domain): # 步骤1：FOFA资产采集 fofa_results = fetch_from_fofa(f'domain="{target_domain}"') # 步骤2：数据处理 normalized = normalize_assets([r['host'] for r in fofa_results]) # 步骤3：Rad爬虫 generate_rad_input(normalized, "rad_input.txt") os.system("rad -t 10 -f rad_input.txt -o rad_output") # 步骤4：Xray扫描 os.system("xray webscan --listen 127.0.0.1:7777 --html-output xray_report.html") print("全流程扫描完成，结果保存在xray_report.html")

在实际项目中，这套方法帮助我们将资产收集效率提升了3-5倍，同时减少了人为错误。关键在于保持脚本的模块化设计，便于针对不同场景调整流程。