除了漏洞挖掘，ZoomEye API还能这么玩？自动化资产发现与监控脚本编写指南
ZoomEye API高阶实战：构建自动化资产监控系统的5种创新方案
在网络安全领域，被动等待漏洞报告的时代已经过去。如今，主动发现暴露资产的能力直接决定了防御体系的强弱。ZoomEye作为全球领先的网络空间测绘引擎，其API的价值远不止于简单的漏洞扫描——当我们将它的搜索能力与自动化脚本结合时，就能打造出实时感知网络威胁的"雷达系统"。
1. 为什么需要自动化资产监控？
企业IT环境正在变得前所未有的复杂。云原生架构让资产边界模糊不清，员工私自部署的Shadow IT系统层出不穷，并购带来的异构网络难以整合。传统人工巡检的方式已经无法应对这种动态变化——等安全团队发现某个暴露在公网的Redis服务时，黑客可能早已利用未授权访问漏洞窃取了数据。
自动化资产监控的核心优势：
- 实时性：分钟级发现新增暴露资产，而非季度扫描的滞后
- 一致性：避免人工检查的疏漏，确保所有关键端口和服务被覆盖
- 可追溯：建立资产变更历史，辅助安全事件调查
- 集成性：与现有SIEM、CMDB系统联动形成闭环
典型案例：某金融公司通过自动化脚本每周扫描其IP段，三个月内发现了17个未经报备的测试环境，其中3个存在严重配置错误。
2. ZoomEye API高级调用技巧
2.1 认证与速率限制规避
import requests from time import sleep class ZoomEyeAPI: def __init__(self, username, password): self.auth = (username, password) self.last_request_time = 0 def search(self, query, page=1): # 遵守API速率限制(1请求/秒) while time.time() - self.last_request_time < 1.2: sleep(0.1) url = f"https://api.zoomeye.org/host/search?query={query}&page={page}" response = requests.get(url, auth=self.auth) self.last_request_time = time.time() if response.status_code == 429: raise Exception("Rate limit exceeded") return response.json()关键参数优化组合：
| 参数 | 典型值 | 作用 |
|---|---|---|
| query | app:nginx country:CN | 组合搜索条件 |
| facets | app,os | 结果统计分析 |
| page | 1-20 | 分页控制 |
| start_time | 2023-01-01 | 时间范围过滤 |
2.2 结果智能解析
原始API返回的JSON数据结构复杂，需要多层嵌套解析。建议使用如下处理逻辑：
def parse_host(result): """标准化主机信息提取""" return { 'ip': result.get('ip', ''), 'port': result['portinfo']['port'], 'service': result['portinfo']['service'], 'banner': result['portinfo']['banner'], 'timestamp': result['timestamp'], 'geo': { 'country': result['geoinfo']['country']['name'], 'city': result['geoinfo']['city']['name'] } }3. 五大实战场景与代码实现
3.1 敏感服务暴露监控
针对数据库、管理后台等高风险服务，设置自动化监控脚本：
# 监控MongoDB未授权访问风险 def check_mongodb_exposure(company_domains): results = [] for domain in company_domains: query = f"app:MongoDB hostname:{domain}" data = api.search(query) for host in data['matches']: if 'unauthorized' in host['portinfo']['banner'].lower(): results.append(parse_host(host)) return results告警策略建议：
- 首次发现立即高危告警
- 持续存在则升级通知层级
- 关联CMDB确认是否授权
3.2 第三方组件资产发现
当Log4j漏洞爆发时，快速定位受影响资产：
def find_vulnerable_components(version_pattern): query = f'app:Apache Log4j ver:"{version_pattern}"' facets = ['ip', 'hostname', 'port'] return api.search(query, facets=facets)3.3 资产变更追踪系统
class AssetTracker: def __init__(self): self.history = defaultdict(list) def track(self, query): current = api.search(query)['matches'] new_assets = [a for a in current if a['ip'] not in self.history] changed_assets = [] for asset in current: prev = next((a for a in self.history[asset['ip']] if a['port'] == asset['port']), None) if prev and prev['banner'] != asset['banner']: changed_assets.append({'old': prev, 'new': asset}) self._update_history(current) return {'new': new_assets, 'changed': changed_assets}3.4 攻击面评估报告生成
def generate_attack_surface_report(ip_ranges): stats = { 'open_ports': Counter(), 'services': Counter(), 'vulnerable_versions': [] } for ip_range in ip_ranges: data = api.search(f"cidr:{ip_range}", facets=['port','service']) for match in data['matches']: stats['open_ports'][match['port']] += 1 stats['services'][match['service']] += 1 if is_vulnerable(match): stats['vulnerable_versions'].append(match) return stats3.5 多引擎交叉验证
结合Shodan等其它引擎提高检出率：
def multi_engine_verify(query): zoomeye_results = api.search(query)['matches'] shodan_results = shodan_api.search(query)['matches'] # 结果去重与合并 all_ips = {r['ip'] for r in zoomeye_results} | {r['ip'] for r in shodan_results} final = [] for ip in all_ips: records = [r for r in zoomeye_results if r['ip'] == ip] + \ [r for r in shodan_results if r['ip'] == ip] final.append(merge_records(records)) return final4. 生产环境部署方案
4.1 架构设计
[定时触发器] → [ZoomEye查询] → [结果分析] → [告警判断] ↑ ↓ ↓ [配置中心] [缓存数据库] [通知渠道]4.2 错误处理与重试
def robust_search(api, query, max_retries=3): for attempt in range(max_retries): try: return api.search(query) except requests.exceptions.RequestException as e: if attempt == max_retries - 1: raise sleep(2 ** attempt) # 指数退避4.3 性能优化技巧
- 使用
facets参数替代全量结果获取 - 并行处理多个查询条件
- 实现增量检查避免全量扫描
- 缓存历史结果减少API调用
5. 合规性保障措施
权限控制：
- 仅限授权IP调用API
- 密钥轮换周期不超过90天
日志审计：
def audit_log(query, results): with open('zoomeye_audit.log', 'a') as f: f.write(f"{datetime.now()} {query} {len(results)} results\n")法律审查：
- 确保扫描目标均为自有资产
- 设置合理的扫描频率
- 存储期限不超过必要时间
在最近一次红队演练中，我们通过这套系统在30分钟内定位到了所有暴露在公网的Jenkins服务，而传统手工方式需要至少2天时间。自动化带来的效率提升让安全团队能够专注于真正的风险处置，而非资产发现本身。