基于FOFA API与Python的自动化资产暴露面监控系统实践

1. 项目概述为什么自动化监控资产暴露面是安全运维的“生命线”最近和几个做安全的朋友聊天大家不约而同地提到了同一个痛点公司资产暴露面越来越大今天刚梳理完的IP、域名、端口明天可能就因为某个业务上线或配置变更又冒出来几个。靠人工去FOFA、Shodan这类网络空间测绘引擎上手动搜不仅效率低下还容易遗漏等真出了事往往就是“为什么这个服务暴露在公网没人知道”的灵魂拷问。这其实就是典型的资产暴露面管理失控。所谓“资产暴露面”简单说就是你的公司有哪些“家当”服务器、域名、API、摄像头、打印机…直接暴露在互联网上能被任何人包括攻击者扫描和访问到。它就像你家房子的门窗门窗越多、越不结实被小偷光顾的风险就越大。安全运维的核心工作之一就是时刻清楚自家有多少扇“门窗”它们是否牢固有没有多出不该有的“后门”。手动监控的局限性太大了。我经历过一次一个测试环境的Redis服务被运维同学不小心配置成了0.0.0.0监听并且没设密码这个信息在公网上挂了快一周我们才发现。攻击者利用这种未授权访问分分钟就能拿到服务器权限。这件事之后我就下决心要把资产监控自动化做起来。而FOFA作为国内领先的网络空间测绘引擎它就像一双24小时不休息的“天眼”持续不断地扫描和收录互联网上的资产信息。通过它的API我们可以编程化地、定期地去查询与我们公司相关的资产指纹一旦发现新的、未知的暴露资产或者已有资产的端口、服务、组件发生了非预期的变化就能第一时间告警。这相当于给你的资产暴露面装上了一套“7x24小时智能门磁报警系统”。所以这个项目的核心价值非常明确将被动、滞后、碎片化的人工资产梳理转变为主动、实时、系统化的自动化监控从根本上提升安全运维的主动防御能力和风险响应速度。接下来我就把自己用FOFA API Python搭建这套系统的完整思路、踩过的坑和最终可运行的脚本毫无保留地分享给你。2. 核心工具与原理FOFA API与Python如何联动2.1 FOFA API你的自动化“侦察兵”FOFA API本质上是一个功能强大的数据查询接口。它允许你使用FOFA平台积累的海量资产指纹数据但不再需要通过网页点击而是用代码发送请求、获取结构化的结果。这对于自动化来说是天作之合。它的工作原理可以类比为“定制化的搜索引擎”。你通过编写特定的查询语法FQLFOFA Query Language告诉API“请帮我找所有符合这些特征的资产”。比如domainexample.com是找该域名下的所有子域名ip192.168.1.0/24是找这个网段的所有IPport6379 protocolredis是找所有开放了Redis服务的资产。API返回的数据通常是JSON格式包含了资产的IP、端口、协议、服务、组件、标题、甚至部分响应头或证书信息。这些结构化的数据正是我们程序能够理解和处理的“原料”。注意使用FOFA API需要账户具备相应的API查询权限通常是会员功能。请确保你的FOFA账户有可用的API积分F币并且妥善保管你的email登录邮箱和keyAPI密钥这两个是调用API的凭证相当于账号密码绝对不能泄露到代码仓库或公开场合。2.2 Python自动化流程的“指挥官”Python在这里扮演着“大脑”和“流水线”的角色。它负责统筹整个监控流程主要做以下几件事构造请求根据监控需求拼装好要查询的FQL语句。调用API使用requests库携带你的认证信息email和key向FOFA API发送HTTP请求。解析数据接收API返回的JSON数据用json库解析成Python的字典或列表方便我们提取关键信息。数据处理与比对这是核心逻辑。程序需要有一个“记忆”功能记住上一次扫描到的资产清单基线。本次扫描到的新数据需要与基线进行比对找出“新增资产”、“资产变更”如端口变化和“减少资产”可能下线了。告警通知当发现异常如新增了高危端口3389、6379或出现了未知域名时通过邮件、钉钉、企业微信、飞书等渠道发送告警信息给相关人员。持久化存储将每次的扫描结果基线保存下来通常用简单的JSON文件、SQLite数据库或者更专业的MySQL/PostgreSQL。我推荐从JSON文件开始简单够用。整个流程可以封装成一个Python脚本然后用Linux的crontab或者Windows的“计划任务”定时比如每天凌晨2点执行从而实现无人值守的自动化监控。2.3 技术选型背后的考量为什么是Python而不是Shell或其他语言首先Python处理HTTP请求和JSON解析有天然优势requests和json都是标准库或事实标准简单几行代码就能搞定。其次后续的数据比对、逻辑判断、甚至连接数据库进行更复杂的分析Python的生态和可读性都更好。最后团队协作和维护成本低安全团队的同事即使不是专业开发看懂Python脚本的难度也相对较低。为什么不直接用FOFA的监控功能FOFA企业版确实有资产监控功能但一是成本可能较高二是自定义程度和与企业内部流程如CMDB对接、内部工单系统联动的集成度可能不如自己写的脚本灵活。自己动手丰俭由人还能根据自身业务特点做深度定制。3. 环境准备与核心依赖安装工欲善其事必先利其器。在写代码之前我们需要把运行环境搭建好。这个项目对环境要求极低几乎在任何有Python的电脑上都能跑。3.1 Python环境搭建如果你还没有安装Python建议直接安装Python 3.8或以上的版本。去Python官网下载安装包安装时务必勾选“Add Python to PATH”这样就能在命令行里直接使用python和pip命令了。安装完成后打开终端Windows是CMD或PowerShellMac/Linux是Terminal输入以下命令验证python --version pip --version如果能正确显示版本号说明安装成功。3.2 安装必要的Python库我们这个项目主要依赖两个库requests用于发送HTTP请求到FOFA API。pandas非必须但强烈推荐。它是一个强大的数据处理库当你的资产数量很多需要进行复杂的筛选、比对和导出为Excel时pandas能让你事半功倍。对于初学者如果觉得pandas稍复杂可以先用Python自带的列表和字典完成核心功能。使用pip一键安装pip install requests pandas如果安装速度慢可以使用国内镜像源例如清华源pip install requests pandas -i https://pypi.tuna.tsinghua.edu.cn/simple3.3 准备FOFA API凭证登录你的FOFA账户在个人中心找到API接口相关页面你会看到你的email和key。请将它们记录下来我们将用它们来初始化脚本。安全警告绝对不要将真实的email和key硬编码在脚本里更不要上传到GitHub等公开代码仓库一旦泄露攻击者可以用你的凭证疯狂查询消耗你的积分甚至获取你公司的资产信息。正确的做法是使用环境变量或单独的配置文件。我推荐创建一个名为config.ini的配置文件来存储敏感信息[fofa] email your_fofa_emailexample.com key your_fofa_api_key_here然后在Python脚本中读取这个配置文件。这样当你需要分享脚本时只需分享不包含config.ini的代码部分并提醒对方自行创建配置文件。4. 脚本核心模块拆解与实现下面我们来一步步构建这个自动化监控脚本。我会把代码分成几个功能模块并详细解释每一行代码的作用。4.1 模块一配置文件读取与API请求封装首先我们创建一个fofa_monitor.py文件并写入以下代码。这个模块负责安全地读取凭证并封装一个通用的函数用于向FOFA API发送请求。import configparser import requests import json import hashlib import time import sys import os from pathlib import Path class FofaClient: FOFA API客户端封装类 def __init__(self, config_fileconfig.ini): 初始化客户端从配置文件读取email和key :param config_file: 配置文件路径 self.config configparser.ConfigParser() # 检查配置文件是否存在 if not Path(config_file).exists(): print(f错误配置文件 {config_file} 不存在) print(请创建config.ini文件并填写[fofa]节下的email和key。) sys.exit(1) self.config.read(config_file, encodingutf-8) try: self.email self.config.get(fofa, email) self.key self.config.get(fofa, key) self.base_url https://fofa.info/api/v1/search/all except (configparser.NoSectionError, configparser.NoOptionError) as e: print(f配置文件读取失败{e}) print(请确保config.ini文件格式正确包含[fofa]节以及email和key选项。) sys.exit(1) # 简单的凭证非空校验 if not self.email or not self.key: print(错误email或key为空请检查config.ini配置。) sys.exit(1) print(fFOFA客户端初始化成功用户{self.email}) def search(self, query, fieldsip,port,protocol,host,domain, page1, size100): 执行FOFA搜索 :param query: FQL查询语句 :param fields: 需要返回的字段默认为ip,port,protocol,host,domain :param page: 页码默认为1 :param size: 每页大小最大10000默认为100 :return: 查询结果的JSON数据如果失败返回None # 构造请求参数 params { email: self.email, key: self.key, qbase64: self._encode_query(query), # FOFA API要求查询语句进行Base64编码 fields: fields, page: page, size: size } try: print(f正在查询: {query} (第{page}页每页{size}条)) response requests.get(self.base_url, paramsparams, timeout30) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 result response.json() # 检查API返回的错误 if result.get(error): print(fFOFA API返回错误{result[error]}) return None print(f查询成功本页获取到 {len(result.get(results, []))} 条资产记录。) return result except requests.exceptions.RequestException as e: print(f网络请求失败{e}) return None except json.JSONDecodeError as e: print(fJSON解析失败{e}) return None def _encode_query(self, query): 对查询语句进行Base64编码FOFA API要求 import base64 return base64.b64encode(query.encode(utf-8)).decode(utf-8) def get_all_results(self, query, fieldsip,port,protocol,host,domain, max_pages10): 自动翻页获取所有结果直到没有更多结果或达到最大页数限制 :param query: FQL查询语句 :param fields: 需要返回的字段 :param max_pages: 最大翻页数防止查询结果过多 :return: 所有资产的列表 all_results [] page 1 size 100 # FOFA API单页最大支持10000但为稳定起见建议先用100或1000 while page max_pages: result self.search(query, fields, page, size) if not result: break current_results result.get(results, []) if not current_results: break all_results.extend(current_results) # 判断是否还有更多数据 total_size result.get(size, 0) if len(all_results) total_size or len(current_results) size: break page 1 time.sleep(1) # 礼貌性延迟避免请求过快 print(f查询完成共获取 {len(all_results)} 条资产记录。) return all_results # 示例如何使用这个类 if __name__ __main__: # 初始化客户端 client FofaClient() # 示例查询查找example.com域名的所有资产 test_query domainexample.com assets client.get_all_results(test_query) if assets: print(前5条资产示例) for i, asset in enumerate(assets[:5]): print(f {i1}. {asset})代码解读与注意事项封装成类我们将FOFA API操作封装成一个FofaClient类这样代码更整洁也方便后续扩展和维护。安全的配置读取使用configparser从独立的config.ini文件读取敏感信息避免硬编码。Base64编码FOFA API要求查询语句q参数必须进行Base64编码我们在_encode_query私有方法中处理了这一点。错误处理代码中加入了网络请求异常、JSON解析异常、API返回错误等多种情况的处理增强了脚本的健壮性。自动翻页get_all_results方法实现了自动翻页逻辑可以一次性获取查询的所有结果受max_pages限制。这对于资产数量较多的情况非常有用。请求延迟在翻页请求间加入了time.sleep(1)这是一个好习惯可以避免对FOFA API服务器造成过大压力防止被限流。实操心得在实际使用中我发现FOFA API对查询频率有一定限制。如果短时间内发起大量请求可能会返回错误或暂时限制访问。因此在循环查询多个资产或翻页时务必加入适当的延迟如1-2秒。另外size参数虽然最大可设为10000但一次请求数据量过大会增加网络超时和解析失败的风险。我通常先设为100或1000进行测试稳定后再调整。4.2 模块二资产数据持久化与基线比对获取到资产数据只是第一步我们需要将其保存下来并与历史数据基线进行比对才能发现变化。这里我们使用JSON文件来存储基线数据因为它简单直观无需数据库服务。我们在FofaClient类的基础上继续添加资产管理的功能。为了清晰我们可以创建一个新的类AssetManager或者直接在原脚本中添加函数。import json from datetime import datetime from collections import defaultdict class AssetManager: 资产数据管理器负责存储、加载和比对资产 def __init__(self, baseline_filefofa_baseline.json): 初始化资产管理器 :param baseline_file: 存储基线数据的JSON文件路径 self.baseline_file baseline_file self.baseline_data self._load_baseline() def _load_baseline(self): 从JSON文件加载基线数据如果文件不存在则返回空字典 try: if Path(self.baseline_file).exists(): with open(self.baseline_file, r, encodingutf-8) as f: data json.load(f) print(f已加载基线数据包含 {len(data.get(assets, []))} 条资产记录。) return data else: print(基线文件不存在将创建新的基线。) return {last_updated: None, assets: []} except Exception as e: print(f加载基线文件失败{e}将使用空基线。) return {last_updated: None, assets: []} def save_baseline(self, assets, query): 将当前扫描到的资产保存为新的基线 :param assets: 当前资产列表 :param query: 本次查询使用的FQL语句 baseline { last_updated: datetime.now().strftime(%Y-%m-%d %H:%M:%S), query: query, assets: assets } try: # 先写入临时文件再重命名避免写入过程中程序崩溃导致文件损坏 temp_file self.baseline_file .tmp with open(temp_file, w, encodingutf-8) as f: json.dump(baseline, f, ensure_asciiFalse, indent2) # 替换原文件 Path(temp_file).replace(self.baseline_file) print(f基线数据已保存至 {self.baseline_file}) except Exception as e: print(f保存基线数据失败{e}) def compare_with_baseline(self, current_assets): 将当前资产与基线资产进行比对 :param current_assets: 当前扫描到的资产列表 :return: 包含新增、减少、变更资产的字典 old_assets self.baseline_data.get(assets, []) # 为快速比对我们为每个资产生成一个唯一标识符。 # 一个简单有效的标识符是ip:port 或 host:port def get_asset_key(asset): # asset是一个字典例如 {ip: 1.2.3.4, port: 80, host: www.example.com} # 优先使用host如果没有则使用ip host asset.get(host) or asset.get(ip, ) port asset.get(port, ) return f{host}:{port} # 将基线资产和当前资产转换为字典键为资产标识符值为资产详情 old_dict {get_asset_key(a): a for a in old_assets} current_dict {get_asset_key(a): a for a in current_assets} old_keys set(old_dict.keys()) current_keys set(current_dict.keys()) # 找出新增资产在当前中存在在基线中不存在 new_keys current_keys - old_keys new_assets [current_dict[k] for k in new_keys] # 找出减少资产在基线中存在在当前中不存在 removed_keys old_keys - current_keys removed_assets [old_dict[k] for k in removed_keys] # 找出可能变更的资产既在基线中也在当前中但某些属性可能变了 # 这里我们简单比较整个资产字典的JSON字符串是否一致不一致则认为有变更 changed_assets [] common_keys old_keys current_keys for key in common_keys: # 将资产字典排序后转换为JSON字符串进行比较避免因顺序不同导致的误判 import json old_json json.dumps(old_dict[key], sort_keysTrue) current_json json.dumps(current_dict[key], sort_keysTrue) if old_json ! current_json: changed_assets.append({ key: key, old: old_dict[key], new: current_dict[key] }) return { new: new_assets, removed: removed_assets, changed: changed_assets, summary: { total_current: len(current_assets), total_baseline: len(old_assets), new_count: len(new_assets), removed_count: len(removed_assets), changed_count: len(changed_assets) } } def generate_report(self, comparison_result, output_fileasset_report.md): 生成资产变更报告Markdown格式 :param comparison_result: compare_with_baseline方法返回的结果 :param output_file: 报告输出文件路径 new_assets comparison_result[new] removed_assets comparison_result[removed] changed_assets comparison_result[changed] summary comparison_result[summary] report_lines [] report_lines.append(f# 资产暴露面监控报告) report_lines.append(f生成时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}\n) report_lines.append(f## 扫描概览) report_lines.append(f- **当前资产总数**{summary[total_current]}) report_lines.append(f- **基线资产总数**{summary[total_baseline]}) report_lines.append(f- **新增资产**{summary[new_count]} 个) report_lines.append(f- **减少资产**{summary[removed_count]} 个) report_lines.append(f- **变更资产**{summary[changed_count]} 个\n) if new_assets: report_lines.append(f## 新增资产 ({len(new_assets)}个)) report_lines.append(| IP | 端口 | 协议 | 主机名 | 域名 |) report_lines.append(|----|------|------|--------|------|) for asset in new_assets: report_lines.append(f| {asset.get(ip, N/A)} | {asset.get(port, N/A)} | {asset.get(protocol, N/A)} | {asset.get(host, N/A)} | {asset.get(domain, N/A)} |) report_lines.append() if removed_assets: report_lines.append(f## ❌ 减少资产 ({len(removed_assets)}个)) report_lines.append(| IP | 端口 | 协议 | 主机名 | 域名 |) report_lines.append(|----|------|------|--------|------|) for asset in removed_assets: report_lines.append(f| {asset.get(ip, N/A)} | {asset.get(port, N/A)} | {asset.get(protocol, N/A)} | {asset.get(host, N/A)} | {asset.get(domain, N/A)} |) report_lines.append() if changed_assets: report_lines.append(f## 变更资产 ({len(changed_assets)}个)) for change in changed_assets: old change[old] new change[new] report_lines.append(f### 资产 {change[key]}) report_lines.append(| 属性 | 旧值 | 新值 |) report_lines.append(|------|------|------|) # 找出所有不同的属性 all_keys set(old.keys()) | set(new.keys()) for key in sorted(all_keys): old_val old.get(key, N/A) new_val new.get(key, N/A) if str(old_val) ! str(new_val): report_lines.append(f| {key} | {old_val} | {new_val} |) report_lines.append() if not (new_assets or removed_assets or changed_assets): report_lines.append(## ✅ 无变化) report_lines.append(本次扫描未发现资产变更。\n) report_content \n.join(report_lines) try: with open(output_file, w, encodingutf-8) as f: f.write(report_content) print(f报告已生成{output_file}) except Exception as e: print(f生成报告失败{e}) return report_content代码解读与注意事项资产标识符比对资产的关键是为每个资产生成一个唯一标识符Key。这里我们使用了host:port或ip:port的组合。这个选择很重要因为它决定了什么算作“同一个资产”。例如同一IP的不同端口会被视为不同资产而同一个域名解析到不同IP只要端口相同也可能被视为同一资产因为host相同。你需要根据你的监控重点来调整这个逻辑。变更检测代码中通过比较资产字典的JSON字符串来判断资产是否发生变更。这是一种简单有效的方法能检测出资产任何属性的变化如协议从http变成了https标题变了等。你也可以只关注特定关键字段的变化。报告生成我们生成了一个Markdown格式的报告内容清晰可以直接在支持Markdown的协作平台如GitLab、飞书文档中查看。你也可以很容易地修改为生成HTML或纯文本格式。文件操作安全在save_baseline方法中我们采用了“先写临时文件再原子替换”的策略。这是一个好习惯可以防止在写入JSON文件过程中脚本崩溃导致基线文件损坏。实操心得基线比对逻辑是监控系统的“大脑”。在实际使用中你可能会遇到一些特殊情况需要处理。例如有些临时性的测试环境IP可能频繁变化你不希望每次都告警。这时你可以在比对逻辑中加入“白名单”机制将已知的动态IP段或特定模式的资产过滤掉不纳入新增告警。同样对于减少的资产也可能是因为FOFA暂时没有扫描到而非真正下线你可以设置一个“连续N次扫描不到才告警”的机制来减少误报。4.3 模块三风险资产识别与告警通知仅仅发现资产变化还不够我们需要识别出其中的高风险资产并及时通知到人。这部分是安全监控的“价值升华点”。我们可以在比对结果的基础上增加风险规则引擎和告警模块。class RiskAnalyzer: 风险分析器识别高风险资产 # 定义高风险端口及其对应服务/风险说明 HIGH_RISK_PORTS { 22: SSH - 若弱口令或未授权访问可导致服务器沦陷, 23: Telnet - 明文传输易被嗅探, 135: Windows RPC - 可能存在永恒之蓝等漏洞, 139: NetBIOS - 信息泄露, 445: SMB - 可能传播勒索病毒, 1433: MSSQL - 数据库弱口令, 1521: Oracle - 数据库弱口令, 3306: MySQL - 数据库弱口令, 3389: RDP - 远程桌面暴力破解重灾区, 5432: PostgreSQL - 数据库弱口令, 5900: VNC - 远程控制若未授权可被直接控制, 6379: Redis - 未授权访问可导致服务器沦陷, 9200: Elasticsearch - 未授权访问可能导致数据泄露, 27017: MongoDB - 未授权访问, 11211: Memcached - 未授权访问可被用于DDoS放大攻击, } # 定义高风险服务或组件关键词 HIGH_RISK_SERVICES [ weblogic, jenkins, dubbo, zookeeper, docker, kibana, activemq, rabbitmq, hadoop, spark ] classmethod def analyze_assets(cls, assets): 分析资产列表识别高风险资产 :param assets: 资产字典列表 :return: 高风险资产列表每个元素包含资产信息和风险原因 high_risk_list [] for asset in assets: risks [] port asset.get(port, ) protocol asset.get(protocol, ).lower() host asset.get(host, ).lower() # 检查高风险端口 if port in cls.HIGH_RISK_PORTS: risks.append(f开放高风险端口 {port} ({cls.HIGH_RISK_PORTS[port]})) # 检查高风险服务通过协议或主机名判断 service_info f{protocol} {host} for risk_service in cls.HIGH_RISK_SERVICES: if risk_service in service_info: risks.append(f发现高风险服务/组件: {risk_service}) # 检查HTTP服务但未使用HTTPS如果端口是80且有对应443风险可降低 if port 80 and protocol http: # 这里可以更复杂比如检查同host是否有443端口 risks.append(HTTP服务未使用HTTPS存在流量窃听风险) if risks: high_risk_list.append({ asset: asset, risks: risks, risk_level: 高危 if 沦陷 in .join(risks) or 未授权 in .join(risks) else 中危 }) return high_risk_list classmethod def generate_risk_report(cls, high_risk_assets, title高风险资产告警): 生成高风险资产告警报告 if not high_risk_assets: return None report_lines [] report_lines.append(f# {title}) report_lines.append(f生成时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}\n) report_lines.append(f共发现 **{len(high_risk_assets)}** 个高风险资产。\n) for i, risk_item in enumerate(high_risk_assets, 1): asset risk_item[asset] risks risk_item[risks] risk_level risk_item[risk_level] report_lines.append(f## {i}. {risk_level}资产{asset.get(host, asset.get(ip, N/A))}:{asset.get(port, N/A)}) report_lines.append(f- **IP**: {asset.get(ip, N/A)}) report_lines.append(f- **端口**: {asset.get(port, N/A)}) report_lines.append(f- **协议**: {asset.get(protocol, N/A)}) report_lines.append(f- **主机名**: {asset.get(host, N/A)}) report_lines.append(f- **域名**: {asset.get(domain, N/A)}) report_lines.append(f- **风险详情**:) for risk in risks: report_lines.append(f - {risk}) report_lines.append() return \n.join(report_lines) class Notifier: 告警通知器基础版可扩展 staticmethod def send_mail(subject, content, to_addrs, smtp_server, smtp_port, username, password): 通过邮件发送告警示例需根据实际邮箱配置调整 # 这里省略具体的邮件发送代码可以使用smtplib和email库实现 # 由于涉及具体邮箱配置且各公司内部邮件系统不同这里只提供思路 print(f[邮件告警] 主题{subject}) print(f收件人{to_addrs}) print(f内容预览{content[:200]}...) # 实际使用时请取消注释并配置以下代码 import smtplib from email.mime.text import MIMEText from email.header import Header msg MIMEText(content, plain, utf-8) msg[Subject] Header(subject, utf-8) msg[From] username msg[To] , .join(to_addrs) if isinstance(to_addrs, list) else to_addrs try: server smtplib.SMTP_SSL(smtp_server, smtp_port) if smtp_port 465 else smtplib.SMTP(smtp_server, smtp_port) server.login(username, password) server.sendmail(username, to_addrs, msg.as_string()) server.quit() print(邮件发送成功) except Exception as e: print(f邮件发送失败{e}) staticmethod def send_dingtalk(webhook_url, content, titleFOFA资产监控告警): 发送钉钉机器人告警示例 import json import requests headers {Content-Type: application/json} data { msgtype: markdown, markdown: { title: title, text: content } } try: response requests.post(webhook_url, headersheaders, datajson.dumps(data), timeout10) if response.json().get(errcode) 0: print(钉钉机器人消息发送成功) else: print(f钉钉机器人消息发送失败{response.text}) except Exception as e: print(f钉钉机器人请求失败{e}) staticmethod def send_console(subject, content): 最简单的控制台输出告警用于测试 print(\n *60) print(f【告警】{subject}) print(*60) print(content) print(*60 \n)代码解读与注意事项风险规则RiskAnalyzer类定义了两类风险规则一是基于端口的如22(SSH)、3389(RDP)、6379(Redis)等常见高危端口二是基于服务指纹的如Weblogic、Jenkins等常出现漏洞的中间件。你可以根据你公司的实际情况不断丰富这个规则库。风险等级简单的规则可以根据风险描述关键词如“沦陷”、“未授权”来划分高、中危。在实际生产中你可以设计更复杂的评分模型。告警渠道Notifier类提供了邮件、钉钉和控制台三种告警方式。邮件适合正式报告钉钉适合即时提醒控制台用于调试。你完全可以在此基础上扩展企业微信、飞书、Slack等通知方式。告警收敛在实际运维中告警风暴是致命的。如果某个网段一次性上线了50台带Redis的测试服务器你会收到50条告警。因此一个成熟的系统还需要告警收敛功能比如将同一类风险如所有新增的6379端口合并为一条告警并注明数量和详情。踩坑提醒告警规则不是一成不变的。初期可以设置得严格一些宁可误报不可漏报。运行一段时间后根据告警的实际情况进行调整。例如如果公司规定所有Redis必须通过跳板机访问那么公网出现任何Redis端口都应是高危告警。但如果某些边缘业务确实需要公网访问则应将其IP加入白名单避免无效告警干扰。定期如每季度评审和更新风险规则库是保持监控系统有效性的关键。4.4 模块四主程序流程与完整脚本整合现在我们把所有模块像拼积木一样组合起来形成一个完整的、可定时执行的监控脚本。def main(): 主监控流程 print(*60) print(FOFA资产暴露面自动化监控脚本启动) print(*60) # 1. 初始化客户端和管理器 client FofaClient(config.ini) asset_manager AssetManager(fofa_baseline.json) # 2. 定义监控的资产范围这里用示例请替换为你的公司资产查询语句 # 你可以定义多个查询语句监控不同维度的资产 monitor_queries [ # 示例监控 example.com 域名下的所有资产 domainexample.com, # 示例监控特定IP段 # ip192.168.1.0/24, # 示例监控公司常用的云服务商IP段需替换为实际IP段 # ip203.0.113.0/24, # 示例监控所有包含公司名称关键词的主机 # hostyourcompany, ] all_current_assets [] for query in monitor_queries: print(f\n 开始执行查询: {query}) assets client.get_all_results(query, fieldsip,port,protocol,host,domain,title,server,country,city) if assets: all_current_assets.extend(assets) # 每个查询之间稍作停顿 time.sleep(2) if not all_current_assets: print(未获取到任何资产数据脚本退出。) return print(f\n 所有查询共获取 {len(all_current_assets)} 条资产记录。) # 3. 与基线比对 comparison asset_manager.compare_with_baseline(all_current_assets) # 4. 生成详细报告 report_content asset_manager.generate_report(comparison, asset_report.md) # 5. 风险分析 # 分析所有当前资产中的高风险项 all_high_risk RiskAnalyzer.analyze_assets(all_current_assets) # 特别关注新增资产中的高风险项 new_high_risk RiskAnalyzer.analyze_assets(comparison[new]) # 6. 告警判断与发送 # 如果有新增资产或者发现高风险资产则触发告警 should_alert False alert_messages [] summary comparison[summary] if summary[new_count] 0: alert_messages.append(f发现 {summary[new_count]} 个新增资产。) should_alert True if summary[removed_count] 0: alert_messages.append(f发现 {summary[removed_count]} 个资产减少。) # 资产减少通常风险较低可根据需要决定是否告警 # should_alert True if new_high_risk: alert_messages.append(f在新增资产中发现 {len(new_high_risk)} 个高风险资产) should_alert True if all_high_risk: alert_messages.append(f在所有资产中发现 {len(all_high_risk)} 个高风险资产。) # 已有资产的高风险项可能之前已知晓可根据需要决定是否每次扫描都告警 # should_alert True # 7. 发送告警 if should_alert: alert_subject f【安全告警】FOFA资产监控发现变更 - {datetime.now().strftime(%Y-%m-%d)} alert_content fFOFA资产监控于 {datetime.now().strftime(%Y-%m-%d %H:%M:%S)} 完成扫描。\n\n alert_content \n.join(alert_messages) \n\n if new_high_risk: risk_report RiskAnalyzer.generate_risk_report(new_high_risk, 新增高风险资产详情) alert_content risk_report \n\n alert_content f详细报告请查看asset_report.md\n alert_content f基线文件fofa_baseline.json # 选择一种或多种告警方式 # 方式一控制台输出测试用 Notifier.send_console(alert_subject, alert_content) # 方式二发送钉钉机器人需配置webhook # dingtalk_webhook https://oapi.dingtalk.com/robot/send?access_tokenYOUR_TOKEN # Notifier.send_dingtalk(dingtalk_webhook, alert_content, alert_subject) # 方式三发送邮件需配置SMTP # mail_to [security-teamexample.com, opsexample.com] # Notifier.send_mail(alert_subject, alert_content, mail_to, # smtp_serversmtp.example.com, smtp_port465, # usernamealertexample.com, passwordyour_password) else: print(\n 本次扫描未发现需告警的变更一切正常。) # 8. 更新基线无论是否有告警都更新基线为本次扫描结果 # 注意只有当你确认本次扫描结果准确且希望将其作为新的比对基准时才取消注释下一行。 # asset_manager.save_baseline(all_current_assets, ; .join(monitor_queries)) print(\n 提示本次扫描结果未自动保存为基线。) print( 请人工审核报告 asset_report.md确认无误后手动取消脚本中 save_baseline 行的注释并重新运行以更新基线。) print(\n *60) print(监控脚本执行完毕) print(*60) if __name__ __main__: main()完整脚本使用流程准备配置文件在与脚本同目录下创建config.ini填入你的FOFA邮箱和API Key。修改监控范围在main()函数的monitor_queries列表中将示例查询语句替换为你公司的真实资产查询语句。你可以添加多条语句监控不同维度。首次运行建立基线首次运行脚本前注释掉最后更新基线的代码即不执行save_baseline。运行脚本后查看生成的asset_report.md报告。因为此时没有基线报告会显示所有扫描到的资产为“新增”。人工审核这些资产确认它们都是你公司合法、已知的暴露资产。确认并保存基线确认无误后取消脚本末尾asset_manager.save_baseline(...)行的注释再次运行脚本。这次运行会将当前资产清单保存为基线文件fofa_baseline.json。配置定时任务与告警将脚本配置为定时任务如Linux的crontab每天凌晨2点执行。根据你的需求配置Notifier类中的钉钉或邮件告警填入正确的Webhook或SMTP信息。日常监控此后脚本每天会自动运行将扫描结果与基线比对一旦发现新增资产、高风险资产等异常就会通过你配置的渠道发送告警。安全运维人员只需关注告警信息并跟进处理即可。5. 进阶优化与生产级考量上面提供的脚本已经是一个可用的监控原型。但要投入到生产环境成为团队依赖的自动化工具还需要考虑更多细节。5.1 性能优化与大规模资产处理当公司资产规模很大数千甚至上万条时脚本可能需要处理更多数据。分批次查询与存储FOFA API单次查询最多返回10000条但处理大量数据的网络请求和内存占用可能成为瓶颈。可以考虑将大范围的查询拆分成多个小范围查询如按IP段拆分分批获取和处