当前位置：首页 > news >正文

Python实战：利用fofa API高效搜索网络资产

news 2026/4/16 8:59:28

1. 为什么需要fofa API自动化搜索

第一次接触网络安全资产搜索时，你可能和我一样，都是手动在fofa官网输入查询语句。但很快就发现三个痛点：一是重复查询效率低下，二是结果导出步骤繁琐，三是无法与其他工具联动。这些问题在批量资产收集时尤为明显。

fofa提供的API接口完美解决了这些问题。通过Python脚本调用API，可以实现：

定时自动执行：设置计划任务定期收集目标资产
结果自动处理：直接输出结构化数据供其他工具使用
复杂查询组合：动态生成查询语句比手动输入更灵活

我去年参与某企业安全评估项目时，需要监控300多个关键组件的暴露情况。如果手动操作，至少需要2人天的工作量。改用API自动化后，只需1小时就能完成全量扫描和报告生成。

2. 准备工作与环境配置

2.1 获取API认证信息

使用fofa API需要两个关键凭证：

注册邮箱：个人资料页显示的联系邮箱
API Key：在[个人中心]-[API管理]处获取

这两个参数相当于访问凭证，需要像保护密码一样妥善保管。建议：

不要直接写在脚本里
使用环境变量或配置文件存储
设置访问权限限制

2.2 安装Python依赖库

核心需要三个库，用pip一键安装：

pip install requests base64

各库的作用：

requests：发送HTTP请求获取API数据
base64：对查询语句进行编码转换
sys（Python内置）：处理命令行参数

我习惯用virtualenv创建隔离环境：

python -m venv fofa_env source fofa_env/bin/activate # Linux/Mac fofa_env\Scripts\activate.bat # Windows

3. 核心代码实现详解

3.1 API请求构造

fofa API的基础请求格式如下：

api_url = 'https://fofa.so/api/v1/search/all?email={}&key={}&qbase64={}&size={}'

关键参数说明：

参数名	是否必填	示例值	说明
email	是	user@example.com	注册邮箱
key	是	32位字符串	API密钥
qbase64	是	dGl0bGU9ImxvZ2luIg==	Base64编码的查询语句
size	否	100	返回结果数量(1-10000)

3.2 查询语句编码处理

fofa要求查询语句必须经过Base64编码。这里有个坑要注意：需要先转为bytes类型再编码。

错误写法：

query = "title=\"admin\"" base64.b64encode(query) # 报错！

正确做法：

query = "title=\"admin\"" encoded = base64.b64encode(query.encode('utf-8')).decode('utf-8')

3.3 完整脚本示例

这是我优化后的增强版脚本，新增了错误处理和进度显示：

import requests import base64 import sys import time def search_fofa(query, email, key, max_results=1000): api_template = "https://fofa.so/api/v1/search/all?email={}&key={}&qbase64={}&size={}" try: encoded_query = base64.b64encode(query.encode()).decode() url = api_template.format(email, key, encoded_query, max_results) print(f"[*] 正在查询: {query}") start = time.time() response = requests.get(url, timeout=30) if response.status_code != 200: raise Exception(f"API请求失败: {response.status_code}") data = response.json() if not data.get("error"): results = data["results"] print(f"[+] 找到 {len(results)} 条结果 (耗时:{time.time()-start:.2f}s)") return results else: raise Exception(f"API返回错误: {data.get('errmsg')}") except Exception as e: print(f"[-] 发生错误: {str(e)}") return None if __name__ == "__main__": if len(sys.argv) < 2: print("用法: python fofa_search.py '查询语句' [输出文件]") sys.exit(1) query = sys.argv[1] output_file = sys.argv[2] if len(sys.argv) > 2 else f"{query.replace(' ', '_')}.txt" # 从环境变量获取凭证 import os email = os.getenv("FOFA_EMAIL") key = os.getenv("FOFA_KEY") if not email or not key: print("请设置FOFA_EMAIL和FOFA_KEY环境变量") sys.exit(1) results = search_fofa(query, email, key) if results: with open(output_file, "w") as f: for item in results: f.write(f"{item[0]}\n") print(f"结果已保存到 {output_file}")

4. 高级使用技巧

4.1 批量查询实现

实际工作中经常需要执行多个关联查询。我们可以改造脚本支持批量操作：

queries = [ 'title="管理后台"', 'body="admin"', 'port="8080"' ] for q in queries: results = search_fofa(q, email, key) if results: with open(f"{q.replace(' ', '_')}.txt", "w") as f: f.write("\n".join(r[0] for r in results))

4.2 结果去重处理

fofa可能返回重复结果，建议添加去重逻辑：

seen = set() unique_results = [] for item in results: if item[0] not in seen: seen.add(item[0]) unique_results.append(item)

4.3 异常处理与重试

网络请求不稳定时，建议实现自动重试机制：

max_retries = 3 retry_delay = 5 for attempt in range(max_retries): try: response = requests.get(url, timeout=30) break except requests.exceptions.RequestException as e: if attempt == max_retries - 1: raise print(f"请求失败，{retry_delay}秒后重试...") time.sleep(retry_delay)

5. 实战案例：监控暴露的Web服务

去年我帮某客户发现他们有一批测试环境的Jenkins服务暴露在公网。用这个脚本可以持续监控：

import schedule import time def job(): results = search_fofa('title="Jenkins" && port="8080"', email, key) if results and len(results) > 10: # 超过10个实例暴露 alert_admin(results) schedule.every(6).hours.do(job) while True: schedule.run_pending() time.sleep(60)

关键改进点：