拓冰网站定制 
简介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 查询语法拼接批量关键词怎么组装成合法 queryFOFA 的查询语法支持titlexxx、bodyxxx、domainxxx、ipx.x.x.x/24、port8080等字段多个条件用连接。批量搜索的关键是把一个关键词列表映射成一组合法 query。常见做法是给每个关键词套一个字段模板比如统一用title或body再按需叠加countryCN、protocolhttps之类的限定。这里有个容易忽略的点关键词里如果含有引号、反斜杠、等特殊字符直接拼接会导致语法错误或语义偏移。必须先做转义或过滤。我一般会写一个白名单清洗函数只保留字母、数字、中文、点、横线、下划线其余字符剔除或转义。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, countryCN) 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设太小会导致请求次数暴涨、配额消耗快设太大则单次响应体变大、超时风险上升。我的经验值是size100配合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, paramsparams, timeout15) 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 Noneqbase64是 query 的 base64 编码这是 FOFA API 的传参要求。fields指定返回字段按需裁剪能减小响应体。重试逻辑里对 429 做了指数退避普通错误则短等待后重试。注意timeout不要设太长批量任务里一个卡住的请求会拖慢整批进度。3. 把结果落成可用数据去重、入库与增量更新3.1 结果去重的三个维度host、ipport、指纹组合批量搜索最容易出的问题是重复数据。同一个资产可能被多个关键词命中也可能因为分页边界重复返回。去重不能只看 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(finsert error: {e}) conn.commit() return insertedUNIQUE(ip, port)是去重的核心约束INSERT OR IGNORE让重复记录静默跳过。keyword字段记录命中的关键词方便后续按来源筛选。实际生产环境可以换成 PostgreSQL 或 MySQL加上索引后万级数据写入很快。3.2 增量更新怎么只抓新增资产批量任务跑第二遍时全量重抓既浪费配额又没必要。增量更新的思路是记录每个 query 上次抓取的时间或游标下次只取新增部分。FOFA API 本身不直接提供「自上次以来新增」的参数但可以用after或时间范围限定来近似实现。常见做法是记录上次任务结束时间下次查询时叠加时间条件。另一种更稳的做法是本地比对抓回来的结果先和库里已有记录比对只入库新出现的 ipport 组合。这样即使 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, encodingutf-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, encodingutf-8) as f: json.dump(rows, f, ensure_asciiFalse, indent2)导出时注意编码中文 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 字符串是 unicodebase64 编码时用encode(utf-8)。如果搜title没结果试试body或cert。有些系统的标题是图片或 JS 渲染的FOFA 抓不到这时候换字段或换关键词更有效。4.4 现象请求被限速或 IP 被封 → 原因并发过高或频率太快 → 解决加延迟、降并发、用退避重试批量任务不要开几十个线程猛冲。我的习惯是单线程加time.sleep(1)或者最多 3 到 5 个并发每个请求间隔 0.5 到 1 秒。遇到 429 就指数退避。被封 IP 的代价远大于多等几分钟。4.5 现象入库后发现大量无效资产 → 原因关键词太宽泛或语法有误 → 解决先小批量验证 querytitle登录这种宽泛关键词会命中海量无关资产。正式跑批量前先用size10小批量验证每个 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(), keylambda 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 簇放最后。这样能把有限的精力花在影响面最大的资产上。希望帮到你。本文还有配套的精品资源点击获取 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 查询语法拼接批量关键词怎么组装成合法 queryFOFA 的查询语法支持titlexxx、bodyxxx、domainxxx、ipx.x.x.x/24、port8080等字段多个条件用连接。批量搜索的关键是把一个关键词列表映射成一组合法 query。常见做法是给每个关键词套一个字段模板比如统一用title或body再按需叠加countryCN、protocolhttps之类的限定。这里有个容易忽略的点关键词里如果含有引号、反斜杠、等特殊字符直接拼接会导致语法错误或语义偏移。必须先做转义或过滤。我一般会写一个白名单清洗函数只保留字母、数字、中文、点、横线、下划线其余字符剔除或转义。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, countryCN) 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设太小会导致请求次数暴涨、配额消耗快设太大则单次响应体变大、超时风险上升。我的经验值是size100配合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, paramsparams, timeout15) 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 Noneqbase64是 query 的 base64 编码这是 FOFA API 的传参要求。fields指定返回字段按需裁剪能减小响应体。重试逻辑里对 429 做了指数退避普通错误则短等待后重试。注意timeout不要设太长批量任务里一个卡住的请求会拖慢整批进度。3. 把结果落成可用数据去重、入库与增量更新3.1 结果去重的三个维度host、ipport、指纹组合批量搜索最容易出的问题是重复数据。同一个资产可能被多个关键词命中也可能因为分页边界重复返回。去重不能只看 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(finsert error: {e}) conn.commit() return insertedUNIQUE(ip, port)是去重的核心约束INSERT OR IGNORE让重复记录静默跳过。keyword字段记录命中的关键词方便后续按来源筛选。实际生产环境可以换成 PostgreSQL 或 MySQL加上索引后万级数据写入很快。3.2 增量更新怎么只抓新增资产批量任务跑第二遍时全量重抓既浪费配额又没必要。增量更新的思路是记录每个 query 上次抓取的时间或游标下次只取新增部分。FOFA API 本身不直接提供「自上次以来新增」的参数但可以用after或时间范围限定来近似实现。常见做法是记录上次任务结束时间下次查询时叠加时间条件。另一种更稳的做法是本地比对抓回来的结果先和库里已有记录比对只入库新出现的 ipport 组合。这样即使 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, encodingutf-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, encodingutf-8) as f: json.dump(rows, f, ensure_asciiFalse, indent2)导出时注意编码中文 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 字符串是 unicodebase64 编码时用encode(utf-8)。如果搜title没结果试试body或cert。有些系统的标题是图片或 JS 渲染的FOFA 抓不到这时候换字段或换关键词更有效。4.4 现象请求被限速或 IP 被封 → 原因并发过高或频率太快 → 解决加延迟、降并发、用退避重试批量任务不要开几十个线程猛冲。我的习惯是单线程加time.sleep(1)或者最多 3 到 5 个并发每个请求间隔 0.5 到 1 秒。遇到 429 就指数退避。被封 IP 的代价远大于多等几分钟。4.5 现象入库后发现大量无效资产 → 原因关键词太宽泛或语法有误 → 解决先小批量验证 querytitle登录这种宽泛关键词会命中海量无关资产。正式跑批量前先用size10小批量验证每个 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(), keylambda 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 簇放最后。这样能把有限的精力花在影响面最大的资产上。希望帮到你。本文还有配套的精品资源点击获取