基于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平台积累的海量资产指纹数据，但不再需要通过网页点击，而是用代码发送请求、获取结构化的结果。这对于自动化来说是天作之合。

它的工作原理可以类比为“定制化的搜索引擎”。你通过编写特定的查询语法（FQL，FOFA Query Language），告诉API：“请帮我找所有符合这些特征的资产”。比如，domain="example.com"是找该域名下的所有子域名，ip="192.168.1.0/24"是找这个网段的所有IP，port="6379" && protocol="redis"是找所有开放了Redis服务的资产。

API返回的数据通常是JSON格式，包含了资产的IP、端口、协议、服务、组件、标题、甚至部分响应头或证书信息。这些结构化的数据，正是我们程序能够理解和处理的“原料”。

注意：使用FOFA API需要账户具备相应的API查询权限（通常是会员功能）。请确保你的FOFA账户有可用的API积分（F币），并且妥善保管你的email（登录邮箱）和key（API密钥），这两个是调用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或PowerShell，Mac/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/simple

3.3 准备FOFA API凭证

登录你的FOFA账户，在个人中心找到API接口相关页面，你会看到你的email和key。请将它们记录下来，我们将用它们来初始化脚本。

安全警告：绝对不要将真实的email和key硬编码在脚本里，更不要上传到GitHub等公开代码仓库！一旦泄露，攻击者可以用你的凭证疯狂查询，消耗你的积分，甚至获取你公司的资产信息。正确的做法是使用环境变量或单独的配置文件。

我推荐创建一个名为config.ini的配置文件来存储敏感信息：

[fofa] email = your_fofa_email@example.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_file='config.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, encoding='utf-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(f"FOFA客户端初始化成功，用户：{self.email}") def search(self, query, fields='ip,port,protocol,host,domain', page=1, size=100): """ 执行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, params=params, timeout=30) response.raise_for_status() # 如果状态码不是200，抛出HTTPError异常 result = response.json() # 检查API返回的错误 if result.get('error'): print(f"FOFA 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(f"JSON解析失败：{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, fields='ip,port,protocol,host,domain', max_pages=10): """ 自动翻页获取所有结果（直到没有更多结果或达到最大页数限制） :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 = 'domain="example.com"' assets = client.get_all_results(test_query) if assets: print("前5条资产示例：") for i, asset in enumerate(assets[:5]): print(f" {i+1}. {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_file='fofa_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', encoding='utf-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', encoding='utf-8') as f: json.dump(baseline, f, ensure_ascii=False, indent=2) # 替换原文件 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_keys=True) current_json = json.dumps(current_dict[key], sort_keys=True) 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_file='asset_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', encoding='utf-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, title="FOFA资产监控告警"): """发送钉钉机器人告警（示例）""" import json import requests headers = {'Content-Type': 'application/json'} data = { "msgtype": "markdown", "markdown": { "title": title, "text": content } } try: response = requests.post(webhook_url, headers=headers, data=json.dumps(data), timeout=10) 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 域名下的所有资产 'domain="example.com"', # 示例：监控特定IP段 # 'ip="192.168.1.0/24"', # 示例：监控公司常用的云服务商IP段（需替换为实际IP段） # 'ip="203.0.113.0/24"', # 示例：监控所有包含公司名称关键词的主机 # 'host="yourcompany"', ] all_current_assets = [] for query in monitor_queries: print(f"\n>>> 开始执行查询: {query}") assets = client.get_all_results(query, fields='ip,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 = f"FOFA资产监控于 {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_token=YOUR_TOKEN" # Notifier.send_dingtalk(dingtalk_webhook, alert_content, alert_subject) # 方式三：发送邮件（需配置SMTP） # mail_to = ["security-team@example.com", "ops@example.com"] # Notifier.send_mail(alert_subject, alert_content, mail_to, # smtp_server="smtp.example.com", smtp_port=465, # username="alert@example.com", password="your_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信息。
日常监控：此后，脚本每天会自动运行，将扫描结果与基线比对，一旦发现新增资产、高风险资产等异常，就会通过你配置的渠道发送告警。安全运维人员只需关注告警信息，并跟进处理即可。