FofaViewer批量搜索爬虫：从单条查询到万级资产测绘的落地路径-平芜编程栈

简介：FofaViewer（俗称“佛法”）是一款面向网络安全研究人员与渗透测试人员的批量资产搜索爬虫工具，基于FOFA搜索引擎接口，帮助用户快速定位、检索与分析互联网公开资产，适用于资产梳理、漏洞挖掘与风险评估等场景。资源包共2个文件，压缩后约32.69MB，包含可执行的jar主程序与properties配置文件，前者负责界面交互与搜索逻辑，后者用于填写API Key、搜索限制及输出参数，便于按需调整。目前已有2015人学习下载。通过该工具，读者可掌握批量搜索条件的配置方法，将关键字、域名、IP范围等组合成检索策略，并把结果导出为CSV或Excel进行后续分析；同时可借助定时任务持续监控网络资产变化，配合图形界面降低上手门槛，形成从检索、筛选到导出的完整工作流，为安全评估与风险防范提供高效的数据支撑。

1. FofaViewer 批量搜索爬虫：从单条查询到万级资产测绘的落地路径

做资产测绘的同行大概率都经历过这个场景：手里攥着几十个关键词，想在 FOFA 上一条条查，查完还要手动翻页、复制、去重、导出，一个下午就耗在浏览器标签页里了。FofaViewer 这类批量搜索爬虫工具解决的正是这个问题——把「查询语法 → 分页抓取 → 结果落库 → 去重导出」这条链路自动化，让一个人也能跑出接近小型测绘团队的产出。它适合安全研究员、红队打点前期的资产梳理、企业外部攻击面管理，以及做漏洞影响面排查的运维。但批量爬取不是把请求循环一遍就完事，配额、限速、语法拼接、结果去重，每一环都有翻车的可能。这篇笔记按我实际搭这套流程的顺序，把选型理由、可复现的代码、参数怎么调、坑在哪讲清楚。

2. 批量搜索的请求模型：FOFA API 与页面爬取的取舍

2.1 为什么优先走 API 而不是硬爬页面

FOFA 提供官方 API 接口，返回结构化 JSON，字段包括 host、ip、port、title、domain、server、protocol 等，直接可入库。硬爬搜索结果页则要面对动态渲染、翻页参数混淆、验证码拦截三座大山。我一般会先确认账号的 API 配额：免费账号额度很低，会员账号按等级给不同的查询次数和结果条数上限。批量场景下，API 的稳定性远高于页面爬取，除非你只是偶尔查几条、不想消耗 API 额度，才考虑页面方式。

选 API 的核心理由是「可预期」：每次请求消耗多少配额、返回多少条、翻页游标怎么传，都是确定的。页面爬取的翻页逻辑会随前端改版失效，维护成本高。批量任务动辄几百次请求，任何一次解析失败都可能导致整批数据缺口，所以结构化接口是首选。

2.2 查询语法拼接：批量关键词怎么组装成合法 query

FOFA 的查询语法支持title="xxx"、body="xxx"、domain="xxx"、ip="x.x.x.x/24"、port="8080"等字段，多个条件用&&连接。批量搜索的关键是把一个关键词列表映射成一组合法 query。常见做法是给每个关键词套一个字段模板，比如统一用title或body，再按需叠加country="CN"、protocol="https"之类的限定。

这里有个容易忽略的点：关键词里如果含有引号、反斜杠、&&等特殊字符，直接拼接会导致语法错误或语义偏移。必须先做转义或过滤。我一般会写一个白名单清洗函数，只保留字母、数字、中文、点、横线、下划线，其余字符剔除或转义。

import re def sanitize_keyword(kw: str) -> str: # 只保留常见安全字符，剔除可能破坏 FOFA 语法的符号 cleaned = re.sub(r'[^\w\u4e00-\u9fa5.\-]', '', kw) return cleaned.strip() def build_query(keyword: str, field: str = "title", extra: str = "") -> str: kw = sanitize_keyword(keyword) if not kw: return "" base = f'{field}="{kw}"' if extra: base += f' && {extra}' return base # 示例 keywords = ['后台管理', '登录系统', 'api gateway'] queries = [build_query(k, "title", 'country="CN"') for k in keywords] for q in queries: print(q)

这段代码做了两件事：sanitize_keyword过滤掉可能干扰语法的字符，build_query按字段模板拼出完整 query。field参数决定搜标题还是搜正文，extra用来追加国家、协议等限定条件。实际使用时，extra建议单独维护一个配置，不要写死在循环里，方便按任务切换。

2.3 分页与配额控制：page 和 size 怎么设才不浪费

FOFA API 的分页靠page参数，每页条数由size控制，常见上限是 100 条。批量任务里，size设太小会导致请求次数暴涨、配额消耗快；设太大则单次响应体变大、超时风险上升。我的经验值是size=100，配合page从 1 递增，直到返回结果为空或达到该 query 的结果上限。

配额控制要单独做一层计数。每次请求前检查剩余额度，低于阈值就暂停或切换任务。不要等 API 返回 429 或配额耗尽错误才处理，那时候已经浪费了请求。下面是一个带配额检查和重试的请求骨架。

import time import requests API_URL = "https://fofa.info/api/v1/search/all" def fofa_search(query: str, email: str, key: str, page: int = 1, size: int = 100): params = { "email": email, "key": key, "qbase64": __import__("base64").b64encode(query.encode()).decode(), "page": page, "size": size, "fields": "host,ip,port,title,domain,server,protocol" } for attempt in range(3): try: resp = requests.get(API_URL, params=params, timeout=15) if resp.status_code == 200: return resp.json() elif resp.status_code == 429: time.sleep(5 * (attempt + 1)) # 退避等待 else: time.sleep(2) except requests.RequestException: time.sleep(3) return None

qbase64是 query 的 base64 编码，这是 FOFA API 的传参要求。fields指定返回字段，按需裁剪能减小响应体。重试逻辑里对 429 做了指数退避，普通错误则短等待后重试。注意timeout不要设太长，批量任务里一个卡住的请求会拖慢整批进度。

3. 把结果落成可用数据：去重、入库与增量更新

3.1 结果去重的三个维度：host、ip+port、指纹组合

批量搜索最容易出的问题是重复数据。同一个资产可能被多个关键词命中，也可能因为分页边界重复返回。去重不能只看 host，因为同一 IP 的不同端口是不同资产，同一域名的不同子路径也可能需要分开记录。我一般按「ip + port」作为主键去重，host 作为辅助字段保留。如果做的是 Web 资产梳理，再叠加 title 或 server 做指纹级去重。

去重逻辑放在入库前，用集合或数据库唯一索引实现。内存集合适合单次任务，跨任务去重要靠数据库。下面是一个基于 SQLite 的落库示例，用唯一索引兜底。

import sqlite3 def init_db(path: str = "assets.db"): conn = sqlite3.connect(path) conn.execute(""" CREATE TABLE IF NOT EXISTS assets ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip TEXT, port TEXT, host TEXT, title TEXT, domain TEXT, server TEXT, protocol TEXT, keyword TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(ip, port) ) """) conn.commit() return conn def save_results(conn, results: list, keyword: str): inserted = 0 for item in results: try: conn.execute( "INSERT OR IGNORE INTO assets (ip, port, host, title, domain, server, protocol, keyword) VALUES (?,?,?,?,?,?,?,?)", (item.get("ip"), item.get("port"), item.get("host"), item.get("title"), item.get("domain"), item.get("server"), item.get("protocol"), keyword) ) inserted += 1 except sqlite3.Error as e: print(f"insert error: {e}") conn.commit() return inserted

UNIQUE(ip, port)是去重的核心约束，INSERT OR IGNORE让重复记录静默跳过。keyword字段记录命中的关键词，方便后续按来源筛选。实际生产环境可以换成 PostgreSQL 或 MySQL，加上索引后万级数据写入很快。

3.2 增量更新：怎么只抓新增资产

批量任务跑第二遍时，全量重抓既浪费配额又没必要。增量更新的思路是记录每个 query 上次抓取的时间或游标，下次只取新增部分。FOFA API 本身不直接提供「自上次以来新增」的参数，但可以用after或时间范围限定来近似实现。常见做法是记录上次任务结束时间，下次查询时叠加时间条件。

另一种更稳的做法是本地比对：抓回来的结果先和库里已有记录比对，只入库新出现的 ip+port 组合。这样即使 API 返回了旧数据，也不会污染库。代价是仍然消耗了配额去抓旧数据。如果配额紧张，优先用时间条件减少返回量；如果配额充足，本地比对更简单可靠。

3.3 导出格式：CSV、JSON 与对接下游工具

落库之后，导出环节决定数据能不能被下游用起来。CSV 适合给 Excel 或人工看，JSON 适合喂给自动化脚本，对接扫描器时可能还需要特定格式。我一般会写一个通用导出函数，按需切换。

import csv import json def export_csv(conn, path: str = "assets.csv"): cursor = conn.execute("SELECT ip, port, host, title, domain, server, protocol, keyword FROM assets") with open(path, "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(["ip", "port", "host", "title", "domain", "server", "protocol", "keyword"]) writer.writerows(cursor.fetchall()) def export_json(conn, path: str = "assets.json"): cursor = conn.execute("SELECT ip, port, host, title, domain, server, protocol, keyword FROM assets") cols = [d[0] for d in cursor.description] rows = [dict(zip(cols, row)) for row in cursor.fetchall()] with open(path, "w", encoding="utf-8") as f: json.dump(rows, f, ensure_ascii=False, indent=2)

导出时注意编码，中文 title 用utf-8避免乱码。CSV 的newline=""参数在 Windows 上能防止多余空行。如果下游是扫描器，通常需要host:port格式的列表，可以再加一个纯文本导出。

4. 避坑与排查：批量爬取中最容易翻车的五个点

4.1 现象：请求全部返回 401 或配额错误 → 原因：key 编码或额度耗尽 → 解决：检查 base64 与配额计数

401 最常见的原因是qbase64编码不对，或者 email/key 拼写有误。FOFA 要求 query 先 base64 再作为参数传递，直接传明文会报错。另一个原因是配额耗尽但错误信息不直观。解决方法是单独写一个配额查询接口调用，任务开始前先确认剩余额度，任务中每次请求后递减计数，低于阈值就停。

4.2 现象：翻页到后面结果重复 → 原因：结果集在翻页期间发生变化 → 解决：固定时间窗口或按 id 排序

FOFA 的结果集不是快照，翻页期间如果有新资产入库，分页边界会偏移，导致重复或遗漏。缓解办法是给 query 叠加时间范围，缩小结果集变动窗口；或者改用基于游标的方式，如果 API 支持的话。实操中更常见的是接受少量重复，靠入库去重兜底。

4.3 现象：中文关键词搜不到结果 → 原因：编码或字段不匹配 → 解决：确认 base64 编码与字段选择

中文关键词要先确保 Python 字符串是 unicode，base64 编码时用encode("utf-8")。如果搜title没结果，试试body或cert。有些系统的标题是图片或 JS 渲染的，FOFA 抓不到，这时候换字段或换关键词更有效。

4.4 现象：请求被限速或 IP 被封 → 原因：并发过高或频率太快 → 解决：加延迟、降并发、用退避重试

批量任务不要开几十个线程猛冲。我的习惯是单线程加time.sleep(1)，或者最多 3 到 5 个并发，每个请求间隔 0.5 到 1 秒。遇到 429 就指数退避。被封 IP 的代价远大于多等几分钟。

4.5 现象：入库后发现大量无效资产 → 原因：关键词太宽泛或语法有误 → 解决：先小批量验证 query

title="登录"这种宽泛关键词会命中海量无关资产。正式跑批量前，先用size=10小批量验证每个 query 的命中质量，确认无误再放大。语法错误也会导致命中偏移，比如&&写成&或漏了引号。

5. 进阶技巧：用指纹聚类把万级结果压缩成可处置清单

批量搜索跑完，库里躺着几千上万条记录，直接交给下游是灾难。真正有价值的一步是聚类：把 title、server、body 特征相似的资产归为一类，每类挑代表样本，优先处置高价值簇。我一般用 title 的归一化文本做聚类，去掉数字和随机串，再按 server 和 protocol 分组。

import re from collections import defaultdict def normalize_title(title: str) -> str: if not title: return "unknown" # 去掉数字、日期、随机串，保留主体词 t = re.sub(r'\d+', '', title) t = re.sub(r'[^\w\u4e00-\u9fa5]', '', t) return t.strip().lower()[:50] def cluster_assets(conn): cursor = conn.execute("SELECT ip, port, title, server, protocol FROM assets") clusters = defaultdict(list) for ip, port, title, server, protocol in cursor.fetchall(): key = (normalize_title(title), server or "unknown", protocol or "unknown") clusters[key].append(f"{ip}:{port}") return clusters # 输出每簇数量和代表样本 conn = init_db() clusters = cluster_assets(conn) for key, members in sorted(clusters.items(), key=lambda x: -len(x[1]))[:20]: print(f"簇: {key} | 数量: {len(members)} | 样本: {members[:3]}")

normalize_title把标题里的数字和符号去掉，保留语义主体，这样「后台管理 v2.1」和「后台管理 v3.0」会归到同一簇。cluster_assets按归一化标题、server、protocol 三元组分组，返回每簇的资产列表。实际使用时，可以按簇大小排序，优先看大簇，因为大簇往往意味着通用组件或批量部署的系统，影响面更大。

验证聚类效果的办法是抽样：从每个大簇里随机抽 3 到 5 个资产，人工确认它们是否真的同类。如果发现误聚，调整normalize_title的截断长度或过滤规则。这个环节没有银弹，靠的是对目标系统的理解加上几轮迭代。

我自己的习惯是每次批量任务结束后，先看聚类结果再决定下一步：大簇优先做指纹识别和漏洞匹配，小簇和 unknown 簇放最后。这样能把有限的精力花在影响面最大的资产上。希望帮到你。

本文还有配套的精品资源，点击获取