Python自动化安全测试：从Fofa资产收集到POC批量验证实战

1. 项目概述与核心价值

在安全测试和漏洞挖掘的日常工作中，信息收集的效率直接决定了后续渗透测试的广度和深度。手动在Fofa、Shodan这类网络空间测绘引擎上一个个搜索、复制、整理资产信息，不仅耗时耗力，还容易遗漏关键目标。同时，对于SRC（安全应急响应中心）项目或内部资产梳理，如何从海量数据中快速定位可能存在漏洞的资产，并自动化验证其脆弱性，是每个安全工程师都想解决的痛点。

这个项目，就是利用Python将这几个环节串联起来，形成一个自动化的工作流。核心思路是：通过Python脚本自动化地从Fofa批量提取资产信息，然后根据SRC项目或自定义规则筛选出目标，最后调用对应的POC（概念验证代码）进行批量验证。它不是一个单一的工具，而是一个可定制、可扩展的自动化框架。对于安全研究人员、渗透测试工程师和SRC漏洞挖掘者来说，掌握这套方法，意味着能将大量重复性劳动交给机器，自己则专注于更核心的逻辑分析和漏洞利用上。

我最初搭建这套脚本也是为了应对公司内部大规模的资产梳理和周期性漏洞扫描需求。手动操作了几天后，我意识到必须用自动化来解放双手。经过多次迭代，现在的脚本已经稳定运行了很长时间，帮我发现了不少中低危漏洞，也极大地提升了应急响应时的信息收集速度。下面，我就把这套实战中打磨出来的思路、代码和踩过的坑，毫无保留地分享出来。

2. 整体架构设计与工具选型

在动手写代码之前，我们先要理清整个流程需要哪些模块，以及为什么选择这些工具库。一个健壮的自动化脚本，其架构应该清晰、模块解耦，方便后续维护和扩展。

2.1 核心流程拆解

整个项目可以分解为三个核心阶段，每个阶段对应一个Python模块：

1. 信息收集模块：负责与Fofa API交互，根据用户提供的搜索语法，批量获取资产信息（如IP、端口、协议、标题、Banner等）。
2. 资产处理与筛选模块：负责解析和清洗从Fofa获取的原始数据。更重要的是，要根据SRC目标范围（例如特定域名、IP段）或自定义规则（如特定标题、特定端口的服务）对资产进行过滤和整理，生成最终的目标列表。
3. POC验证模块：负责加载POC脚本，并发或顺序地对目标列表中的资产进行漏洞验证，并输出结构化的结果报告。

2.2 关键工具与库选型理由

• Requests：这是进行HTTP网络请求的不二之选。相比于Python内置的urllib，Requests的API更加人性化，会话保持、代理设置、超时处理都非常方便。我们将用它来调用Fofa API和进行POC验证时的网络探测。

  注意：务必使用Session对象来保持连接，特别是在进行大量POC验证时，可以显著提升效率并减少TCP握手开销。

• Pandas：虽然对于小规模数据，用Python内置的列表和字典也能处理，但一旦数据量上千，进行筛选、去重、合并等操作就会非常繁琐。Pandas的DataFrame数据结构就像一个内存中的电子表格，能让你用一行代码完成复杂的查询和数据处理，是处理Fofa返回的CSV/JSON数据的利器。

• Concurrent.futures / ThreadPoolExecutor：POC验证通常是I/O密集型任务（等待网络响应），非常适合使用多线程来提升速度。ThreadPoolExecutor提供了一个高级别的异步执行接口，让我们能轻松实现批量目标的并发测试，将数小时的验证任务压缩到几分钟内完成。

  实操心得：线程数并非越多越好。一般设置为20-50个左右比较合适，过多会导致网络连接数暴涨，可能触发目标防火墙的规则或被封IP。建议根据网络环境和目标承受能力动态调整。

• Argparse：为了让脚本更易用，我们需要一个命令行参数解析器。Argparse是Python标准库的一部分，功能强大，可以轻松定义必选参数、可选参数、帮助信息等，让我们的脚本看起来更专业。

• Logging：任何严肃的脚本都需要完善的日志系统。Python标准库的logging模块可以分级（DEBUG, INFO, WARNING, ERROR）输出日志到控制台和文件，在排查复杂问题时能救命。

  踩坑记录：初期我直接用print打印信息，当脚本在后台运行时，一旦出错几乎无法追溯。引入logging后，配合日志轮转，可以清晰看到每个任务的执行状态和错误上下文。

• POC框架的考虑：你可以自己编写简单的POC验证函数，但对于维护大量POC，建议借鉴或集成成熟的框架思路，如Pocsuite3、Nuclei（通过子进程调用）或Xray的被动扫描模式。本项目为了讲解原理，我们会从编写一个简单的POC模板开始。

3. 核心模块一：Fofa资产批量提取

这是整个流程的源头，稳定、高效地从Fofa获取数据是关键。

3.1 Fofa API的申请与使用要点

首先，你需要一个Fofa会员账号来获取API密钥（email和key）。免费账户有调用频率和数量限制，但对于个人学习和小规模使用足够了。

API的核心调用参数是：

• qbase64: 经过Base64编码的搜索语法（如domain="example.com"）。
• fields: 指定返回的字段，如host,title,ip,domain,port,protocol。
• size: 每次查询返回的最大条数（有上限，通常为10000）。
• page: 翻页参数。

我们的脚本需要实现：1. 对查询语法进行Base64编码；2. 构造带有认证头的请求；3. 处理分页，直到获取所有数据或达到数量上限。

import base64 import requests import pandas as pd import time from typing import List, Dict class FofaClient: def __init__(self, email: str, key: str): self.email = email self.key = key self.base_url = "https://fofa.info/api/v1/search/all" self.session = requests.Session() # 设置公共请求头，模拟浏览器 self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' }) def search(self, query: str, fields: str = 'host,title,ip,port,domain', max_count: int = 1000) -> List[Dict]: """ 执行FOFA搜索 :param query: FOFA搜索语法 :param fields: 返回字段 :param max_count: 最大获取数量 :return: 资产信息列表 """ all_results = [] page = 1 size = 100 # 单页最大数量，根据API权限调整 # 编码查询语句 query_base64 = base64.b64encode(query.encode()).decode() while len(all_results) < max_count: params = { 'qbase64': query_base64, 'fields': fields, 'page': page, 'size': size, 'full': 'false' # 不返回完整数据，节省流量 } # FOFA API认证方式：通过email和key参数 auth_params = {'email': self.email, 'key': self.key} params.update(auth_params) try: resp = self.session.get(self.base_url, params=params, timeout=30) resp.raise_for_status() # 检查HTTP错误 data = resp.json() if data['error']: print(f"FOFA API错误: {data['errmsg']}") break results = data['results'] if not results: break # 没有更多数据了 # 将结果转换为字典列表 for item in results: # fields 字符串分割为字段名列表 field_list = fields.split(',') result_dict = {field: item[i] for i, field in enumerate(field_list)} all_results.append(result_dict) print(f"已获取第 {page} 页，累计 {len(all_results)} 条记录") page += 1 # 礼貌性延迟，避免请求过快 time.sleep(1) # 如果返回数量小于请求数量，说明已是最后一页 if len(results) < size: break except requests.exceptions.RequestException as e: print(f"网络请求失败: {e}") break except ValueError as e: print(f"解析JSON响应失败: {e}") break # 限制返回数量 return all_results[:max_count] # 使用示例 if __name__ == '__main__': client = FofaClient(email='your_email@example.com', key='your_api_key') # 搜索所有标题包含“OA”且开放80端口的资产 assets = client.search(query='title="OA" && port="80"', max_count=500) print(f"共获取到 {len(assets)} 条资产信息")

3.2 数据处理与去重策略

从API拿到的原始数据往往存在重复（同一IP不同端口、同一域名不同路径）或字段不全的情况。直接用于后续验证会浪费资源。

1. 转换为DataFrame：利用Pandas可以轻松处理。

   import pandas as pd df = pd.DataFrame(assets)

2. 关键字段提取与构造：我们最终验证需要的是host（可能是IP:PORT或域名）和protocol（http/https）。需要从原始数据中智能提取。

   def normalize_host(row): """将host字段标准化为可用于请求的URL或IP:PORT格式""" host = row.get('host', '') ip = row.get('ip', '') port = row.get('port', '') protocol = row.get('protocol', 'http').lower() # 默认http if not host: if ip and port: return f"{ip}:{port}" else: return None # 如果host已经是域名或IP:PORT格式，直接使用 # 否则，尝试组合 if '://' not in host: # 假设是域名或IP if port and port not in ['80', '443']: host = f"{host}:{port}" # 为host添加协议头，便于requests直接使用 base_url = f"{protocol}://{host}" else: base_url = host return base_url df['target_url'] = df.apply(normalize_host, axis=1) df = df.dropna(subset=['target_url']) # 删除无效目标

3. 去重：根据ip或target_url进行去重，避免对同一目标重复测试。

   df_deduplicated = df.drop_duplicates(subset=['ip']) # 基于IP去重 # 或者基于归一化后的URL去重 df_deduplicated = df.drop_duplicates(subset=['target_url'])

4. 筛选：结合SRC范围进行筛选。例如，如果SRC目标域名为*.example.com，我们可以：

   src_domains = ['.example.com', '.test-example.com'] def is_in_scope(url): import urllib.parse try: parsed = urllib.parse.urlparse(url) netloc = parsed.netloc.split(':')[0] # 移除端口 for domain in src_domains: if netloc.endswith(domain): return True except: pass return False df_filtered = df_deduplicated[df_deduplicated['target_url'].apply(is_in_scope)]

经过以上步骤，我们就得到了一个干净、唯一且属于目标范围的目标列表target_list = df_filtered['target_url'].tolist()。

4. 核心模块二：POC验证引擎的设计与实现

这是项目的核心，一个灵活、健壮的POC验证引擎能极大提升漏洞发现的效率。

4.1 POC脚本的标准化模板

为了统一管理，我们要求每个POC都遵循一个固定的结构。这里设计一个简单的类模板：

# poc_template.py import requests import logging from urllib.parse import urljoin class BasePOC: """POC基类，所有具体POC都应继承此类""" # POC信息 vuln_id = "CVE-XXXX-XXXX" # 漏洞编号 name = "漏洞名称" version = "1.0" def __init__(self, target_url: str, timeout: int = 10): """ 初始化 :param target_url: 目标URL (e.g., http://example.com) :param timeout: 请求超时时间 """ self.target_url = target_url.rstrip('/') self.timeout = timeout self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Accept': '*/*', 'Connection': 'close' }) # 禁用SSL警告（仅用于测试环境） requests.packages.urllib3.disable_warnings() self.verify_ssl = False # 测试时可关闭SSL验证，生产环境慎用 self.logger = logging.getLogger(self.__class__.__name__) def verify(self) -> dict: """ 验证漏洞的主函数 返回格式: { 'success': True/False, 'message': '描述信息', 'evidence': '漏洞证据（如响应片段）', 'extra': {} # 额外信息，如回显内容 } """ raise NotImplementedError("子类必须实现 verify 方法") def _request(self, method, path='', **kwargs): """统一的请求方法，处理异常和日志""" url = urljoin(self.target_url, path) try: kwargs.setdefault('timeout', self.timeout) kwargs.setdefault('verify', self.verify_ssl) resp = self.session.request(method, url, **kwargs) self.logger.debug(f"{method} {url} - Status: {resp.status_code}") return resp except requests.exceptions.Timeout: self.logger.warning(f"请求超时: {url}") return None except requests.exceptions.RequestException as e: self.logger.error(f"请求失败: {url}, Error: {e}") return None # 一个具体的POC示例：检测一个简单的信息泄露 class ExampleInfoDisclosurePOC(BasePOC): vuln_id = "EXAMPLE-2023-001" name = "示例信息泄露漏洞" def verify(self): result_template = {'success': False, 'message': '', 'evidence': '', 'extra': {}} # 尝试访问一个常见的敏感路径 test_path = '/.git/config' resp = self._request('GET', test_path) if resp and resp.status_code == 200 and '[core]' in resp.text: result_template['success'] = True result_template['message'] = f"发现.git配置信息泄露" result_template['evidence'] = resp.text[:500] # 截取部分证据 return result_template result_template['message'] = '目标不存在此漏洞' return result_template

4.2 并发验证引擎

有了POC模板，我们需要一个引擎来批量加载POC并对目标列表进行并发测试。

# poc_engine.py import importlib.util import sys import os from concurrent.futures import ThreadPoolExecutor, as_completed import logging from typing import List, Dict, Any class POCEngine: def __init__(self, max_workers: int = 20): self.max_workers = max_workers self.results = [] self.logger = logging.getLogger(__name__) def load_poc_from_file(self, file_path: str): """从单个文件动态加载POC类""" try: # 动态导入模块 module_name = os.path.splitext(os.path.basename(file_path))[0] spec = importlib.util.spec_from_file_location(module_name, file_path) module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) # 查找继承自BasePOC的类 for attr_name in dir(module): attr = getattr(module, attr_name) if (isinstance(attr, type) and issubclass(attr, BasePOC) and attr != BasePOC): self.logger.info(f"加载POC: {attr.name} ({attr.vuln_id})") return attr return None except Exception as e: self.logger.error(f"加载POC文件失败 {file_path}: {e}") return None def scan_target(self, target_url: str, poc_class): """对单个目标执行单个POC验证""" try: poc_instance = poc_class(target_url) result = poc_instance.verify() result['target'] = target_url result['poc_name'] = poc_class.name result['vuln_id'] = poc_class.vuln_id return result except Exception as e: self.logger.error(f"扫描目标 {target_url} 时发生异常: {e}") return {'target': target_url, 'success': False, 'message': f'扫描异常: {str(e)}'} def batch_scan(self, target_list: List[str], poc_file_path: str): """批量扫描：一个POC对多个目标""" poc_class = self.load_poc_from_file(poc_file_path) if not poc_class: self.logger.error("无法加载有效的POC类") return [] self.logger.info(f"开始批量扫描，目标数: {len(target_list)}, 线程数: {self.max_workers}") all_results = [] with ThreadPoolExecutor(max_workers=self.max_workers) as executor: # 提交所有任务 future_to_target = {executor.submit(self.scan_target, target, poc_class): target for target in target_list} # 处理完成的任务 for future in as_completed(future_to_target): target = future_to_target[future] try: result = future.result(timeout=30) # 每个任务超时 all_results.append(result) if result.get('success'): self.logger.warning(f"[+] 漏洞发现: {target} - {result.get('poc_name')}") else: self.logger.debug(f"[-] 未发现: {target}") except Exception as e: self.logger.error(f"处理目标 {target} 的结果时出错: {e}") all_results.append({'target': target, 'success': False, 'message': f'任务执行异常: {str(e)}'}) self.results = all_results return all_results

4.3 结果汇总与报告生成

扫描结束后，我们需要将结果清晰地呈现出来。可以生成控制台报告、文本报告或更结构化的JSON/CSV报告。

def generate_report(self, output_format: str = 'console'): """生成扫描报告""" if not self.results: print("暂无扫描结果") return vuln_found = [r for r in self.results if r.get('success')] safe_count = len(self.results) - len(vuln_found) if output_format == 'console': print("\n" + "="*60) print("漏洞扫描报告") print("="*60) print(f"扫描目标总数: {len(self.results)}") print(f"发现漏洞数: {len(vuln_found)}") print(f"安全目标数: {safe_count}") print("-"*60) if vuln_found: print("[+] 漏洞列表:") for vuln in vuln_found: print(f" 目标: {vuln['target']}") print(f" 漏洞: {vuln['poc_name']} ({vuln['vuln_id']})") print(f" 证据: {vuln.get('evidence', 'N/A')[:200]}...") # 截断长证据 print(f" 信息: {vuln.get('message')}") print(" " + "-"*40) print("="*60) elif output_format == 'csv': import csv with open('scan_report.csv', 'w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=['target', 'vuln_id', 'poc_name', 'success', 'message', 'evidence']) writer.writeheader() for r in self.results: # 证据字段可能很长，可以只存路径或摘要 row = r.copy() if len(row.get('evidence', '')) > 500: row['evidence'] = row['evidence'][:500] + '...' writer.writerow(row) print(f"报告已生成至: scan_report.csv") # 可以扩展JSON、HTML等格式

5. 项目集成与实战演练

现在，我们将前几个模块组合起来，形成一个完整的命令行工具。

5.1 主程序逻辑与参数解析

# main.py import argparse import logging import sys from fofa_client import FofaClient from poc_engine import POCEngine import pandas as pd def setup_logging(level=logging.INFO): logging.basicConfig( level=level, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('scanner.log'), logging.StreamHandler(sys.stdout) ] ) def main(): parser = argparse.ArgumentParser(description='自动化FOFA资产收集与POC验证工具') parser.add_argument('--email', required=True, help='FOFA账号邮箱') parser.add_argument('--key', required=True, help='FOFA API Key') parser.add_argument('--query', required=True, help='FOFA搜索语法，如：title="OA" && country="CN"') parser.add_argument('--poc', required=True, help='POC脚本文件路径') parser.add_argument('--max-assets', type=int, default=1000, help='最大资产获取数量') parser.add_argument('--threads', type=int, default=20, help='并发线程数') parser.add_argument('--filter-domain', help='筛选特定域名范围内的资产，如：.example.com') parser.add_argument('--output', choices=['console', 'csv', 'json'], default='console', help='报告输出格式') args = parser.parse_args() setup_logging() logger = logging.getLogger(__name__) # 步骤1: 从FOFA获取资产 logger.info("步骤1: 正在从FOFA获取资产...") client = FofaClient(args.email, args.key) raw_assets = client.search(args.query, max_count=args.max_assets) if not raw_assets: logger.error("未从FOFA获取到任何资产，请检查查询语法或API权限。") return df = pd.DataFrame(raw_assets) logger.info(f"原始获取资产数: {len(df)}") # 步骤2: 数据处理与筛选 logger.info("步骤2: 正在处理与筛选资产...") # 归一化URL (此处调用之前定义的normalize_host函数，需导入或定义) from utils import normalize_host # 假设我们将数据处理函数放在utils.py df['target_url'] = df.apply(lambda row: normalize_host(row), axis=1) df = df.dropna(subset=['target_url']) # 域名筛选 if args.filter_domain: target_domain = args.filter_domain.lower() df = df[df['target_url'].str.contains(target_domain, case=False, na=False)] logger.info(f"按域名 '{args.filter_domain}' 筛选后资产数: {len(df)}") # 去重 df = df.drop_duplicates(subset=['target_url']) logger.info(f"去重后资产数: {len(df)}") target_list = df['target_url'].tolist() if not target_list: logger.error("经过筛选后无有效目标，程序退出。") return # 步骤3: POC批量验证 logger.info(f"步骤3: 开始POC批量验证，目标数: {len(target_list)}") engine = POCEngine(max_workers=args.threads) engine.batch_scan(target_list, args.poc) # 步骤4: 生成报告 logger.info("步骤4: 生成扫描报告...") engine.generate_report(output_format=args.output) if __name__ == '__main__': main()

5.2 一个完整的实战案例

假设某SRC项目范围是*.target-company.com，我们想找其暴露在公网的Jenkins服务是否存在未授权访问漏洞。

1. 编写POC(poc_jenkins_unauth.py):

   from poc_template import BasePOC class JenkinsUnauthorizedPOC(BasePOC): vuln_id = "CVE-2017-1000353" # 示例，实际是Jenkins未授权访问通用漏洞 name = "Jenkins未授权访问" def verify(self): result = {'success': False, 'message': '', 'evidence': '', 'extra': {}} # 尝试访问Jenkins控制台 test_paths = ['/manage', '/script', '/asynchPeople/'] for path in test_paths: resp = self._request('GET', path) if resp and resp.status_code == 200: # 简单关键词判断，实际应更严谨 if 'Jenkins' in resp.text and ('管理' in resp.text or 'Dashboard' in resp.text): result['success'] = True result['message'] = f'目标 {self.target_url} 可能存在Jenkins未授权访问' result['evidence'] = f'访问路径 {path} 返回了Jenkins管理界面' return result result['message'] = '未发现Jenkins未授权访问漏洞' return result

2. 运行命令:

   python main.py \ --email your_fofa_email \ --key your_fofa_key \ --query 'app="Jenkins" && domain="target-company.com"' \ --poc ./pocs/poc_jenkins_unauth.py \ --filter-domain .target-company.com \ --threads 30 \ --output csv

3. 结果：脚本会自动从Fofa拉取所有归属于target-company.com且识别为Jenkins的资产，去重筛选后，用30个线程并发检测未授权访问漏洞，最后将结果输出到scan_report.csv。

6. 常见问题、优化思路与避坑指南

在实际使用中，你会遇到各种各样的问题。这里我总结了一些典型场景和解决方案。

6.1 常见问题排查表

6.2 高级优化与扩展思路

1. 异步化升级：将concurrent.futures替换为asyncio+aiohttp，可以更高效地管理成千上万的并发HTTP请求，尤其适合大规模资产验证。
2. POC仓库管理：建立一个POC目录，主程序自动遍历加载所有符合规范的POC脚本。可以给POC打上标签（如info,rce,sqli），实现按漏洞类型扫描。
3. 结果去重与聚合：对于同一个目标，多个POC可能发现不同漏洞。报告应支持按目标聚合漏洞，并自动合并相同类型的发现（如多个目录都泄露了源码）。
4. 与扫描器联动：可以将筛选后的资产列表导出为targets.txt，直接喂给Nuclei、Xray等更专业的扫描器进行深度检测，本脚本作为前置资产收集和初筛工具。
5. 定时任务与持续监控：结合crontab（Linux）或Task Scheduler（Windows），定期执行脚本，监控新增资产或漏洞状态变化，实现持续的安全监控。
6. 图形化界面（GUI）：使用PyQt或Tkinter为脚本制作一个简单的图形界面，方便不熟悉命令行的同事使用，通过界面输入API密钥、查询语法和选择POC。

6.3 最重要的安全与法律提醒

警告：能力越大，责任越大。请务必在合法授权的范围内使用此脚本。

• 仅用于授权测试：绝对不要对未授权的任何系统进行扫描或测试。
• 遵守SRC规则：在参与SRC项目时，严格遵循该平台规定的测试范围、测试方法和漏洞提交规范。
• 控制扫描力度：设置合理的并发数、请求延迟，避免对目标业务造成拒绝服务（DoS）影响。
• 妥善保管数据：扫描结果可能包含敏感信息，请妥善存储，切勿公开泄露。
• 本脚本仅供学习交流：使用者需自行承担因不当使用而产生的一切法律后果。

这套脚本的搭建过程，也是我个人对安全自动化理解的深化过程。从最初只能跑通单个功能，到现在能处理从信息收集到漏洞验证的完整链路，中间经历了无数次的调试和重构。最大的体会是，自动化不是为了替代思考，而是将人从重复劳动中解放出来，去关注更本质的安全问题。希望这个分享能为你打开一扇门，你可以基于这个框架，融入更多自己的想法和技巧，打造出更贴合自己工作流的利器。