简介：FofaViewer（俗称“佛法”）是一款面向网络安全研究人员与渗透测试人员的批量搜索爬虫工具，基于FOFA网络资产搜索引擎接口，帮助用户快速定位和分析互联网公开资产，适用于资产梳理、漏洞挖掘与风险评估等场景。资源包共2个文件，包含1个jar可执行程序与1个properties配置文件，压缩包约32.69MB，前者为工具主程序，后者用于设置API Key、搜索限制与输出参数，实现个性化检索。工具支持按关键字、域名、IP范围等条件批量查询，结果可导出为CSV或Excel，并提供图形界面与定时监控能力，降低使用门槛。目前已有2015人学习下载，读者可借此掌握批量资产检索、结果导出与持续监控的完整流程，提升安全评估效率，同时需注意遵守法律法规与合规使用要求。
1. FofaViewer 批量搜索爬虫：从单条查询到万级资产测绘的落地路径
做资产测绘的同行大概率都经历过这种场景：手工在 Fofa 网页端一条一条敲语法，查完一个 C 段复制粘贴导出，再换下一个目标，一天下来眼睛发直，真正有价值的资产没捞到几条。FofaViewer 这类批量搜索爬虫工具解决的正是这个痛点——把查询语法、分页翻页、结果去重、字段导出这几件事串成一条自动化流水线，让原本需要几小时的重复劳动压缩到几分钟。它适合做红队前期信息收集、企业外部攻击面梳理、SRC 资产排查的从业者，也适合想把 Fofa 查询能力接进自己工作流的安全工程师。核心逻辑不复杂：模拟登录态、按语法批量请求、解析返回、落地存储，但真正跑通并跑稳，坑比想象中多。
2. FofaViewer 批量搜索的请求链路与登录态维持
2.1 为什么不能直接裸请求 API
很多人第一反应是直接调 Fofa 的公开 API，但免费账号的 API 额度和返回字段都受限，批量场景下很快触顶。FofaViewer 这类工具走的是网页端查询接口，本质是复用浏览器登录后的 Cookie 会话。这里有个关键认知：网页端接口返回的是 JSON 结构，字段比 API 更全，但请求头里必须带上有效的登录凭证，否则返回的是登录页 HTML 而不是数据。
请求链路大致是：构造查询 URL → 携带 Cookie 和 UA → 服务端校验会话 → 返回 JSON 数据 → 解析字段。其中会话校验是黑匣子，一旦 Cookie 过期，返回体结构会变，解析逻辑直接翻车。所以第一步不是写查询，而是把登录态拿到并验证。
2.2 获取并验证登录 Cookie 的最小步骤
常见做法是浏览器登录后从开发者工具复制 Cookie，但手工复制不适合长期跑。更稳的方式是用脚本模拟登录拿 Cookie，或者用持久化会话文件。下面这段 Python 演示如何用 requests 维持会话并验证 Cookie 是否有效：
import requests # 从浏览器复制或登录接口获取的 Cookie 字符串 COOKIE = "cookie_here" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Cookie": COOKIE, "Referer": "https://fofa.info/", } def check_session(): """请求一个需要登录的接口，判断会话是否有效""" url = "https://fofa.info/api/v1/info/my" resp = requests.get(url, headers=HEADERS, timeout=10) data = resp.json() # 有效会话会返回用户信息，无效则返回错误码 if data.get("error"): print("会话失效，需要重新登录:", data.get("errmsg")) return False print("会话有效，当前账号:", data.get("email")) return True if __name__ == "__main__": check_session()逻辑说明：先请求一个轻量的用户信息接口，用返回的 error 字段判断会话状态，避免拿着失效 Cookie 去跑批量查询浪费请求。参数上，timeout 建议设 10 秒，Referer 必须带，否则部分接口会拒绝。Cookie 里的关键字段是 session 标识，过期时间通常几小时到几天不等，批量任务前先跑一次 check_session 是血泪经验。
2.3 查询接口的参数拆解
网页端查询接口的核心参数有这几个：qbase64（Base64 编码后的查询语法）、page（页码）、size（每页条数）、fields（返回字段）。其中 qbase64 是必须的，直接把语法明文拼进 URL 会被拒。fields 决定返回哪些列，默认只返回 host 和 ip，想要 port、title、domain 得显式指定。
| 参数 | 含义 | 常用值 | 注意 |
|---|---|---|---|
| qbase64 | Base64 编码的查询语法 | base64.b64encode(b'domain="example.com"') | 必须编码，明文会报错 |
| page | 页码 | 1, 2, 3... | 从 1 开始，不是 0 |
| size | 每页条数 | 20 / 100 | 免费账号上限通常 100 |
| fields | 返回字段 | host,ip,port,title,domain | 逗号分隔，字段名要准确 |
编码这一步新手最容易踩坑：语法里的引号和空格编码后不能有换行，否则服务端解析失败。建议统一用base64.b64encode(query.encode()).decode()处理。
3. 批量查询的并发控制与结果去重落地
3.1 并发不是越高越好
批量查询的核心矛盾是速度与封禁风险。单线程顺序请求稳但慢，一个 C 段几百页要跑很久；并发太高又容易触发风控，返回 429 或直接封 IP。我一般会把并发控制在 3 到 5 之间，配合每请求间隔 1 到 2 秒的随机延迟。这个参数没有标准答案，取决于账号等级和目标量级，熟手可以根据返回状态码动态调整。
下面是一个带并发控制和重试的查询函数：
import base64 import time import random import requests from concurrent.futures import ThreadPoolExecutor, as_completed HEADERS = {"User-Agent": "Mozilla/5.0", "Cookie": "cookie_here", "Referer": "https://fofa.info/"} def build_query(query, page, size=100, fields="host,ip,port,title"): """构造查询 URL，qbase64 必须编码""" qbase64 = base64.b64encode(query.encode()).decode() return f"https://fofa.info/api/v1/search/all?qbase64={qbase64}&page={page}&size={size}&fields={fields}" def fetch_page(query, page, retry=3): """单页查询，带重试和随机延迟""" url = build_query(query, page) for attempt in range(retry): try: resp = requests.get(url, headers=HEADERS, timeout=15) if resp.status_code == 429: # 触发限流，退避后重试 time.sleep(5 * (attempt + 1)) continue data = resp.json() if data.get("error"): print(f"第 {page} 页返回错误: {data.get('errmsg')}") return [] return data.get("results", []) except Exception as e: print(f"第 {page} 页请求异常: {e}") time.sleep(2) return [] def batch_search(query, max_page=10, workers=4): """并发批量查询，收集所有结果""" all_results = [] with ThreadPoolExecutor(max_workers=workers) as executor: futures = {executor.submit(fetch_page, query, p): p for p in range(1, max_page + 1)} for future in as_completed(futures): page = futures[future] results = future.result() all_results.extend(results) # 随机延迟，降低风控概率 time.sleep(random.uniform(1, 2)) return all_results逻辑说明：build_query 负责编码和拼 URL，fetch_page 处理单页请求并针对 429 做退避重试，batch_search 用线程池控制并发。参数上，workers 建议 3 到 5，max_page 根据目标量级设，retry 设 3 次足够。关键点是 429 退避时间要递增，固定延迟容易被持续限流。
3.2 结果去重与字段清洗
批量查询返回的结果里重复项很常见，尤其是跨页查询时同一资产可能出现在不同页。去重不能只按 host，因为同一 host 不同端口是不同资产。稳妥的做法是用 host + port 组合作为唯一键。字段清洗方面，title 里常带 HTML 标签和多余空格，导出前要处理掉。
import re def clean_title(title): """去掉 HTML 标签和多余空白""" if not title: return "" title = re.sub(r"<[^>]+>", "", title) return title.strip() def dedup_results(results): """按 host+port 去重，保留首次出现""" seen = set() unique = [] for item in results: # item 结构通常是 [host, ip, port, title] key = f"{item[0]}:{item[2]}" if key not in seen: seen.add(key) item[3] = clean_title(item[3]) unique.append(item) return unique逻辑说明：dedup_results 用集合记录已出现的 host:port，重复的直接跳过。clean_title 用正则去标签，这一步在导出 CSV 前做，避免 Excel 打开时格式错乱。参数上，item 的索引取决于 fields 的顺序，fields 变了索引也要跟着改，这是容易忽略的坑。
3.3 落地存储的格式选择
结果落地常见三种格式：CSV、JSON、SQLite。CSV 适合直接给 Excel 看，JSON 适合接后续脚本，SQLite 适合反复查询和增量更新。批量场景我一般先落 SQLite，再按需导出 CSV。SQLite 建表时把 host 和 port 设联合主键，天然去重。
import sqlite3 def save_to_sqlite(results, db_path="fofa_results.db"): """结果写入 SQLite，host+port 联合主键去重""" conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS assets ( host TEXT, ip TEXT, port TEXT, title TEXT, PRIMARY KEY (host, port) ) """) for item in results: cursor.execute( "INSERT OR IGNORE INTO assets (host, ip, port, title) VALUES (?, ?, ?, ?)", (item[0], item[1], item[2], item[3]) ) conn.commit() conn.close()逻辑说明：INSERT OR IGNORE 配合联合主键实现幂等写入，重复跑同一查询不会产生脏数据。参数上，db_path 建议按任务命名，避免多个任务混在一个库里。这一步做完，批量查询的完整链路就闭环了。
4. FofaViewer 批量搜索的避坑与排查清单
4.1 返回结果为空但语法没问题
现象：查询语法在网页端能出结果，脚本跑出来是空列表。原因通常是 qbase64 编码时带了换行符，或者 fields 字段名拼错导致服务端返回空。解决：打印编码后的字符串检查有无\n，fields 严格对照接口文档的字段名，别自己造。
4.2 跑几十页后突然全部失败
现象：前几十页正常，之后每页都返回错误或超时。原因是触发了频率限制，账号被临时限流。解决：降低并发到 2，加大请求间隔到 3 秒以上，或者换账号继续。后悔药是提前把 max_page 分批跑，别一次性拉太多。
4.3 Cookie 中途失效导致解析报错
现象：脚本跑一半抛 JSON 解析异常。原因是会话过期，返回体变成了登录页 HTML。解决：在 fetch_page 里加一层判断，resp.text 以<!DOCTYPE开头就说明会话失效，直接终止任务并提示重新登录，别硬解析。
4.4 去重后数量对不上预期
现象：明明查了几百条，去重后只剩几十条。原因是 host+port 组合重复率高，或者 fields 顺序和代码索引不一致导致 key 取错。解决：先打印原始结果的前几条，确认字段顺序，再核对去重逻辑。字段顺序这个坑很隐蔽，fields 一改索引全乱。
4.5 导出 CSV 中文乱码
现象：CSV 用 Excel 打开 title 列全是乱码。原因是编码没指定 utf-8-sig。解决：写文件时用encoding="utf-8-sig"，这个 BOM 头能让 Excel 正确识别中文。小细节但每次都能坑到人。
5. 把批量查询接进日常 workflow 的进阶技巧
跑通单次批量查询只是起点，真正提升效率的是把它变成可复用的工作流。我一般会做三件事：查询语法模板化、结果增量对比、定时任务化。
查询语法模板化是指把常用语法存成配置文件，比如按行业、按地区、按组件分类，跑的时候直接读配置。这样不用每次手敲语法，也方便团队共享。下面是一个简单的配置结构：
# queries.yaml 示例结构（用 dict 演示） QUERY_TEMPLATES = { "nginx_servers": 'app="nginx" && country="CN"', "spring_boot": 'app="Spring Boot" && country="CN"', "specific_domain": 'domain="example.com"', }结果增量对比是指每次跑完把结果入库，下次跑完和上次对比，只关注新增资产。这对持续监控外部攻击面特别有用，新增的资产往往是最需要关注的。实现上就是在 SQLite 里加一个 first_seen 时间戳字段，查询时按时间过滤。
定时任务化是指用 cron 或任务计划把批量查询挂起来，每天或每周自动跑一次。注意定时任务里要处理好 Cookie 过期问题，常见做法是单独写一个登录刷新脚本，定时任务先跑刷新再跑查询。
验证方法上，我习惯用一个小目标先跑通全链路，比如查一个已知的小域名，确认返回、去重、入库、导出都正常，再放大到生产目标。这个习惯帮我省了很多次大规模翻车。
最后说个我自己的教训：早期我图快，把并发开到 20，结果账号被限流三天，那三天的任务全泡汤。后来老老实实把并发压到 3，配合随机延迟，反而整体效率更高，因为不用反复处理失败重试。批量搜索这件事，稳比快重要。希望帮到你。
本文还有配套的精品资源，点击获取
