1. 项目概述:从手动刷榜到自动化预警
每天上班第一件事,就是打开浏览器,手动刷新NVD(National Vulnerability Database)官网,在一堆CVE编号里寻找和自己负责的系统、组件相关的漏洞信息。这几乎是每个安全工程师、运维开发乃至关注安全的研发同学都经历过的日常。这个过程枯燥、低效,还容易遗漏关键信息,尤其是在漏洞爆发的高峰期,手动“刷榜”的滞后性可能会让整个团队陷入被动。
这个项目的核心,就是用Python脚本彻底解放你的双手,实现从NVD官方数据源定时抓取最新漏洞信息,并经过筛选、格式化后,自动推送到钉钉或飞书工作群。它解决的不仅仅是“手动”的问题,更是将漏洞情报的获取从被动接收转变为主动、实时、精准的推送。想象一下,当一个新的高危漏洞(比如某个广泛使用的日志组件或Web框架曝出RCE)发布时,你的手机能在几分钟内就收到一条结构清晰的告警消息,里面包含了CVE编号、严重等级、受影响产品和官方链接,整个团队可以立刻进入应急响应状态,而不是几小时甚至一天后才发现。
这个脚本适合所有需要关注安全动态的技术角色。对于安全团队,它是构建自动化安全运营中心(SOC)最基础的一环;对于运维和开发,它能帮你第一时间知悉所维护服务依赖的第三方库是否存在风险;对于技术负责人或项目经理,它提供了一种透明、高效的风险感知渠道。整个方案基于Python,利用其丰富的网络请求和数据处理库,配合成熟的即时通讯工具开放接口,技术栈平易近人,即使是Python新手,跟着步骤也能一步步搭建起来。接下来,我会详细拆解从设计思路到每一行代码的考量,以及我在多次迭代中踩过的坑和总结的经验。
2. 核心设计思路与架构选型
在动手写代码之前,我们需要先想清楚整个数据流和功能模块。一个健壮的自动化脚本,不能只是简单地把NVD页面HTML扒下来解析,那样既不稳定,也缺乏扩展性。我们的目标是构建一个可靠、可维护的“数据管道”。
2.1 为什么选择NVD的官方API而非爬取网页?
首先,数据源的选择至关重要。NVD官网提供了官方的RESTful API,这是最权威、最稳定的数据接口。相比于爬取网页,使用API有以下几个压倒性优势:
- 数据结构化:API返回的是标准的JSON格式数据,包含了CVE条目的所有元信息,如ID、描述、评分(CVSS)、受影响CPE(通用平台枚举)列表、参考链接等。这省去了我们解析复杂HTML结构的麻烦,也避免了因官网前端改版导致爬虫失效的风险。
- 稳定性与合规性:使用官方API是受支持的方式,只要遵循其使用条款(如请求频率限制),服务就是稳定的。而爬虫则可能因为触发反爬机制导致IP被封。
- 过滤与查询能力:NVD API支持丰富的查询参数,例如按发布时间范围、CVSS评分阈值、关键词等过滤漏洞。这让我们可以在获取数据的源头就进行初步筛选,只拉取我们关心的漏洞,极大地减少了后续处理的数据量。
因此,我们的脚本核心将围绕调用NVD API展开。你需要去NVD官网查看最新的API文档,了解其端点(Endpoint)和参数。通常,我们会使用/rest/json/cves/2.0这个端点,并通过lastModStartDate和lastModEndDate参数来获取特定时间段内更新或新增的漏洞。
2.2 消息推送渠道:钉钉与飞书机器人如何选?
输出渠道的选择取决于你的团队协作习惯。钉钉和飞书都提供了非常类似的“自定义机器人”功能,通过Webhook URL接收HTTP POST请求,并将格式化的消息发送到指定的群聊。
钉钉机器人:文档清晰,支持多种消息类型(文本、链接、Markdown、ActionCard等)。安全方面,支持签名校验和关键字/IP白名单,适合对安全有一定要求的内部环境。飞书机器人:功能上与钉钉类似,消息格式同样丰富。其优势在于与飞书套件的深度集成,如果团队主要使用飞书,选择它会更加无缝。
从技术实现角度看,两者大同小异,都是构造一个符合平台要求的JSON payload,然后发送HTTPS请求。我们的脚本可以设计成支持可配置的推送渠道,通过一个配置项轻松切换。考虑到热词中“钉钉”的出现频率更高,本文将主要以钉钉机器人为例进行详解,并在关键部分指出飞书的差异点。
2.3 整体脚本架构设计
基于以上,我们可以勾勒出脚本的四个核心模块:
- 数据获取模块:负责调用NVD API,处理分页,获取原始JSON数据。
- 数据处理与过滤模块:解析JSON,提取关键字段,并根据预定义的规则(如CVSS评分>7.0,或影响特定产品如“Apache Log4j”)进行过滤。
- 消息格式化模块:将过滤后的漏洞列表,按照钉钉/飞书机器人支持的格式(如Markdown),组装成易于阅读的消息体。
- 消息推送与调度模块:将格式化后的消息通过HTTP请求发送到Webhook URL,并集成定时任务调度(如使用
crontab或schedule库)。
此外,还需要考虑错误处理(API请求失败、网络异常、数据解析错误)、日志记录(方便问题排查)以及配置管理(将API URL、Webhook、过滤规则等外部化,避免硬编码)。一个简单的架构流程图在脑海中应该是:定时触发器 -> 获取数据 -> 解析过滤 -> 格式化成消息 -> 推送 -> 记录日志。
3. 分步实现与核心代码解析
接下来,我们进入实操环节。我会假设你有一个基本的Python3环境,并已经安装了requests库(如果没有,请先运行pip install requests)。
3.1 第一步:配置钉钉机器人并获取Webhook
- 在钉钉群中,点击右上角设置图标 ->
智能群助手。 - 在机器人管理页面,点击
添加机器人。 - 选择
自定义机器人,输入机器人名字(例如“安全漏洞监控”),选择需要发送消息的群。 - 安全设置非常重要:建议至少选择
自定义关键词,并设置一个关键词如“漏洞”。这意味着你发送的消息中必须包含这个词,机器人才会成功发送。你也可以设置IP白名单,进一步增强安全性。 - 完成设置后,钉钉会提供一个Webhook URL,格式类似:
https://oapi.dingtalk.com/robot/send?access_token=XXXXXX。请妥善保管这个URL,它就是脚本推送消息的地址。
注意:Webhook URL和访问令牌(access_token)是最高机密,绝对不能提交到公开的代码仓库(如GitHub)。务必将其放入配置文件或环境变量中。
3.2 第二步:编写NVD数据获取函数
我们首先编写一个函数,用于从NVD API获取数据。NVD API有公共版本,无需认证,但可能有速率限制(如每小时最多5次请求)。对于定时抓取(如每小时一次),这通常够用。
import requests import json from datetime import datetime, timedelta import time def fetch_cves_from_nvd(hours_back=24, max_results=50): """ 从NVD API获取最近指定小时内有更新的CVE数据。 参数: hours_back (int): 查询多少小时内的数据,默认24小时。 max_results (int): 最大返回结果数,默认50。API可能分页,这里简化处理。 返回: list: 包含CVE字典的列表,如果出错返回空列表。 """ base_url = "https://services.nvd.nist.gov/rest/json/cves/2.0" # 计算时间范围。NVD API 使用 ISO 8601 格式。 end_time = datetime.utcnow() start_time = end_time - timedelta(hours=hours_back) # 格式化时间字符串,精确到秒 start_str = start_time.strftime("%Y-%m-%dT%H:%M:%S.000") end_str = end_time.strftime("%Y-%m-%dT%H:%M:%S.000") params = { 'lastModStartDate': start_str, 'lastModEndDate': end_str, 'startIndex': 0, 'resultsPerPage': max_results # 每页结果数,最大可达5000 } try: response = requests.get(base_url, params=params, timeout=30) response.raise_for_status() # 如果状态码不是200,抛出HTTPError异常 data = response.json() # 解析返回的JSON,提取CVE列表 cve_items = data.get('vulnerabilities', []) cve_list = [] for item in cve_items: cve_data = item.get('cve', {}) cve_id = cve_data.get('id', 'N/A') descriptions = cve_data.get('descriptions', []) # 优先获取英文描述 description = next((desc['value'] for desc in descriptions if desc['lang'] == 'en'), 'No description available.') # 获取CVSS v3.1 或 v2.0 的严重性评分 metrics = cve_data.get('metrics', {}) cvss_v31 = metrics.get('cvssMetricV31', [{}])[0] if metrics.get('cvssMetricV31') else {} cvss_v30 = metrics.get('cvssMetricV30', [{}])[0] if metrics.get('cvssMetricV30') else {} cvss_v2 = metrics.get('cvssMetricV2', [{}])[0] if metrics.get('cvssMetricV2') else {} cvss_data = cvss_v31 or cvss_v30 or cvss_v2 base_score = cvss_data.get('cvssData', {}).get('baseScore', None) severity = cvss_data.get('cvssData', {}).get('baseSeverity') or cvss_data.get('baseSeverity', 'N/A') # 获取受影响的产品配置(CPE),这里取第一个匹配项简化展示 configurations = cve_data.get('configurations', []) affected_products = [] for config in configurations: nodes = config.get('nodes', []) for node in nodes: for cpe_match in node.get('cpeMatch', []): affected_products.append(cpe_match.get('criteria', 'N/A')) # 去重并只取前3个展示 affected_products = list(set(affected_products))[:3] cve_list.append({ 'id': cve_id, 'description': description[:200] + '...' if len(description) > 200 else description, # 截断长描述 'baseScore': base_score, 'severity': severity, 'affected_products': affected_products, 'references': [ref.get('url') for ref in cve_data.get('references', [])[:2]] # 取前两个参考链接 }) return cve_list except requests.exceptions.RequestException as e: print(f"请求NVD API时发生网络错误: {e}") return [] except json.JSONDecodeError as e: print(f"解析NVD API返回的JSON时发生错误: {e}") return [] except KeyError as e: print(f"解析NVD数据结构时,未找到预期键: {e}") return []代码解析与注意事项:
- 时间参数:我们使用
lastModStartDate和lastModEndDate来获取在指定时间范围内被修改过的CVE。这比按发布时间抓取更全面,因为NVD经常会更新已有CVE的信息(如补充评分、受影响范围)。 - 错误处理:网络请求和JSON解析都可能出错,必须用
try-except包裹,并返回空列表,避免脚本因单次失败而完全崩溃。 - 数据提取:NVD的JSON结构嵌套较深,需要小心地使用
.get()方法避免KeyError。描述可能有多种语言,我们优先取英文。CVSS评分有多个版本,我们按v3.1 -> v3.0 -> v2.0的优先级获取。 - 性能考虑:默认抓取24小时内更新的漏洞,最多50条。对于高频监控(如每小时一次),这个范围足够。如果你需要更全面的历史数据,需要处理API的分页逻辑(通过循环调整
startIndex参数)。
3.3 第三步:实现基于严重性与关键词的过滤逻辑
不是所有CVE都值得推送。我们需要一个过滤函数,只留下高风险或与我们技术栈相关的漏洞。
def filter_cves(cve_list, min_cvss_score=7.0, severity_filter=['HIGH', 'CRITICAL'], keyword_filters=None): """ 过滤CVE列表。 参数: cve_list (list): fetch_cves_from_nvd 返回的CVE字典列表。 min_cvss_score (float): CVSS基础分数的最低阈值。 severity_filter (list): 允许的严重性等级列表,如 ['HIGH', 'CRITICAL']。 keyword_filters (list): 关键词列表,用于在描述或影响产品中匹配。默认为None,不过滤。 返回: list: 过滤后的CVE列表。 """ if keyword_filters is None: keyword_filters = ['Apache', 'Spring', 'Log4j', 'nginx', 'MySQL', 'Redis'] # 示例关键词 filtered_list = [] for cve in cve_list: # 1. 按CVSS分数过滤 score = cve.get('baseScore') if score is not None and score < min_cvss_score: continue # 2. 按严重性等级过滤 if cve.get('severity') not in severity_filter: continue # 3. 按关键词过滤(描述或影响产品) matched = False search_text = cve['description'] + ' ' + ' '.join(cve['affected_products']) for keyword in keyword_filters: if keyword.lower() in search_text.lower(): matched = True break if not matched: continue filtered_list.append(cve) return filtered_list实操心得:
- 阈值设置:
min_cvss_score=7.0通常对应“高危”漏洞。你可以根据团队的风险承受能力调整,比如设为4.0来捕获更多中危漏洞。 - 关键词策略:
keyword_filters是脚本是否“智能”的关键。这里放的是你公司技术栈的核心组件名称(如Web框架、数据库、中间件)。更好的做法是将这个列表外置到一个配置文件里,方便不同团队或项目组自定义。 - 过滤顺序:先按分数和等级过滤,速度快,可以快速筛掉大量低风险条目,然后再进行相对耗时的关键词匹配,这是提高效率的小技巧。
3.4 第四步:构造并发送钉钉Markdown消息
钉钉机器人支持Markdown格式,可以很好地展示结构化信息。
def send_dingtalk_markdown(webhook_url, secret, title, cve_list): """ 发送Markdown格式的消息到钉钉群。 参数: webhook_url (str): 钉钉机器人的Webhook URL。 secret (str): 钉钉机器人的加签密钥(如果设置了加签)。 title (str): 消息标题。 cve_list (list): 过滤后的CVE字典列表。 """ if not cve_list: print("没有符合条件的CVE,本次不发送消息。") return # 构建Markdown正文 markdown_text = f"### {title}\n\n" markdown_text += f"**监控时段内发现 {len(cve_list)} 个需关注的漏洞**\n\n---\n" for idx, cve in enumerate(cve_list, 1): markdown_text += f"**{idx}. {cve['id']} - {cve['severity']} ({cve.get('baseScore', 'N/A')})**\n" markdown_text += f"- **描述**: {cve['description']}\n" if cve['affected_products']: markdown_text += f"- **影响产品**: {', '.join(cve['affected_products'])}\n" if cve['references']: # 钉钉Markdown中,链接格式为 [链接文本](链接地址) ref_links = ', '.join([f"[链接{i+1}]({url})" for i, url in enumerate(cve['references'])]) markdown_text += f"- **参考链接**: {ref_links}\n" markdown_text += "\n" # 钉钉要求消息中必须包含注册时设置的关键词,这里我们假设关键词是“漏洞” markdown_text += "> 本消息由安全监控机器人自动发送,详情请查阅NVD。\n" # 钉钉机器人消息体 data = { "msgtype": "markdown", "markdown": { "title": title, "text": markdown_text }, "at": { "isAtAll": False # 是否@所有人,谨慎使用 # 可以指定具体被@人的手机号: "atMobiles": ["138xxxx8822"] } } # 如果机器人设置了加签,需要计算签名 timestamp = str(round(time.time() * 1000)) if secret: import hmac import hashlib import base64 string_to_sign = f"{timestamp}\n{secret}" hmac_code = hmac.new(secret.encode('utf-8'), string_to_sign.encode('utf-8'), digestmod=hashlib.sha256).digest() sign = base64.b64encode(hmac_code).decode('utf-8') webhook_url = f"{webhook_url}×tamp={timestamp}&sign={sign}" headers = {'Content-Type': 'application/json'} try: response = requests.post(webhook_url, headers=headers, data=json.dumps(data), timeout=10) result = response.json() if result.get('errcode') == 0: print(f"钉钉消息发送成功!") else: print(f"钉钉消息发送失败: {result.get('errmsg')}") except Exception as e: print(f"发送钉钉消息时发生异常: {e}")飞书适配提示: 飞书机器人的消息结构略有不同。主要变化在于data的构造:
# 飞书机器人消息体示例 data = { "msg_type": "interactive", "card": { "elements": [{ "tag": "div", "text": { "content": markdown_text, # 飞书也支持Markdown "tag": "lark_md" } }], "header": { "title": { "content": title, "tag": "plain_text" } } } }Webhook URL的格式也不同,且签名计算方式有差异(飞书使用timestamp和sign作为请求头,而非URL参数)。你需要查阅飞书开放平台的最新文档进行调整。
3.5 第五步:组装主函数与配置管理
我们将所有模块组合起来,并引入配置文件。
config.yaml (配置文件):
dingtalk: webhook: "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN_HERE" secret: "YOUR_SECRET_HERE" # 如果未设置加签,留空 keyword: "漏洞" nvd: lookback_hours: 6 # 每次查询过去几小时的数据 max_results: 100 min_cvss_score: 7.0 severity_filter: - HIGH - CRITICAL filters: keywords: - Apache - Spring - Log4j - nginx - MySQL - Redis - Docker - Kubernetesmain.py (主脚本):
import yaml import logging from pathlib import Path # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def load_config(config_path='config.yaml'): with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) return config def main(): config = load_config() logger.info("开始抓取NVD漏洞数据...") cves_raw = fetch_cves_from_nvd( hours_back=config['nvd']['lookback_hours'], max_results=config['nvd']['max_results'] ) logger.info(f"共获取到 {len(cves_raw)} 个原始CVE条目。") logger.info("开始过滤CVE数据...") cves_filtered = filter_cves( cves_raw, min_cvss_score=config['nvd']['min_cvss_score'], severity_filter=config['nvd']['severity_filter'], keyword_filters=config['filters']['keywords'] ) logger.info(f"过滤后剩余 {len(cves_filtered)} 个需关注的CVE。") if cves_filtered: title = f"⚠️ 安全漏洞预警 ({len(cves_filtered)}个)" send_dingtalk_markdown( webhook_url=config['dingtalk']['webhook'], secret=config['dingtalk'].get('secret', ''), # 使用.get避免无secret时报错 title=title, cve_list=cves_filtered ) else: logger.info("没有需要推送的高危漏洞。") logger.info("本轮监控任务执行完毕。") if __name__ == "__main__": main()3.6 第六步:实现定时执行
脚本写好了,如何让它定时运行呢?有两种主流方式:
方案一:使用系统Crontab(推荐,简单可靠)在Linux/Unix系统或Mac上,使用crontab -e编辑定时任务。
# 例如,每天上午9点,下午3点各执行一次 0 9,15 * * * /usr/bin/python3 /path/to/your/main.py >> /path/to/log/monitor.log 2>&1方案二:使用Python的schedule库(适合在常驻进程内运行)
import schedule import time def job(): main() # 每4小时运行一次 schedule.every(4).hours.do(job) logger.info("漏洞监控调度器已启动...") while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次注意:使用
schedule库需要脚本一直运行在后台,可以用nohup或systemd托管。对于生产环境,crontab是更标准、更易于管理的方式。
4. 部署、优化与高级功能拓展
将脚本部署到一台稳定的服务器(如公司内网的Linux虚拟机)是使其可靠运行的关键。建议创建一个专门的目录,将脚本、配置文件、日志文件都放在里面。
4.1 部署与目录结构建议
/opt/cve-monitor/ ├── main.py # 主脚本 ├── config.yaml # 配置文件(务必加入.gitignore) ├── requirements.txt # Python依赖列表 ├── logs/ # 日志目录 │ └── monitor.log └── README.md # 部署说明在服务器上安装依赖:pip install -r requirements.txt(requirements.txt内容为requests和pyyaml)。 使用crontab设置定时任务,并将输出重定向到日志文件,便于后续排查。
4.2 性能优化与稳定性提升
- 处理API分页:当
resultsPerPage小于总结果数时,NVD API会返回totalResults和startIndex。我们需要一个循环来获取所有数据。修改fetch_cves_from_nvd函数:def fetch_cves_from_nvd(hours_back=24, results_per_page=2000): # ... 时间计算 ... all_cves = [] start_index = 0 total_results = float('inf') while start_index < total_results: params = {'lastModStartDate': start_str, 'lastModEndDate': end_str, 'startIndex': start_index, 'resultsPerPage': results_per_page} response = requests.get(base_url, params=params, timeout=60) data = response.json() total_results = data.get('totalResults', 0) cve_items = data.get('vulnerabilities', []) # ... 解析cve_items并添加到all_cves ... start_index += results_per_page time.sleep(2) # 礼貌性延迟,避免请求过快 return all_cves - 增加重试机制:网络请求可能偶尔失败。可以使用
tenacity或retrying库为requests.get添加重试逻辑,提高鲁棒性。 - 避免重复推送:如果脚本运行频率(如每小时)高于NVD数据更新频率,可能会重复推送相同的CVE。一个简单的去重方法是记录已推送的CVE ID(比如写入一个本地文件或小型数据库如SQLite),在发送前进行检查。
- 异常通知:脚本本身也可能出错。可以添加一个“守护”功能,当脚本因未捕获的异常退出时,或者连续多次未获取到数据时,通过另一个紧急通知渠道(如邮件或另一个钉钉机器人)告警。
4.3 高级功能拓展思路
这个基础脚本可以作为一个起点,扩展出很多有价值的功能:
- 多数据源聚合:除了NVD,还可以集成其他漏洞库,如CNVD、CNNVD,或者针对特定云服务商的漏洞公告(如AWS、Azure的安全公告),在消息中进行对比或去重。
- 资产关联匹配:将获取到的CVE与内部的CMDB(配置管理数据库)或资产清单进行自动化匹配。只有影响到你实际资产的漏洞才进行推送,实现精准告警。这需要你维护一份资产及其软件版本清单。
- 漏洞生命周期管理:推送消息后,可以自动在Jira、Trello等项目管理工具中创建一个跟踪工单,指派给相应的负责人,并关联漏洞信息。
- 数据持久化与仪表盘:将抓取到的漏洞数据存入数据库(如MySQL、PostgreSQL),然后利用Grafana等工具绘制仪表盘,展示漏洞趋势、各系统风险等级等,为安全决策提供数据支持。
- 分级推送:根据漏洞的CVSS评分和受影响资产的重要性,定义不同的推送策略。例如,超危漏洞(CVSS 9.0+)立即@相关责任人并发送短信,高危漏洞推送到核心安全群,中危漏洞推送到周报汇总。
5. 常见问题与排查技巧实录
在实际部署和运行过程中,你肯定会遇到各种问题。下面是我在多次实践中总结的“避坑指南”。
5.1 钉钉/飞书消息发送失败
- 问题:脚本运行无报错,但群里收不到消息。
- 排查步骤:
- 检查Webhook URL:确认URL完全正确,没有多余的空格或换行。最简单的方法是在命令行用
curl测试:curl -H "Content-Type: application/json" -X POST -d '{"msgtype":"text","text":{"content":"测试消息"}}' YOUR_WEBHOOK_URL。如果返回{"errcode":0,...}则URL正确。 - 检查安全设置:如果你设置了自定义关键词,请确保发送的消息体(
text或markdown.text字段)里包含这个词。在我们的代码中,Markdown正文最后一行加入了“漏洞”关键词。如果你设置了加签,请确保secret配置正确,并且签名计算代码无误。一个常见的错误是时间戳单位(钉钉是毫秒)或签名字符串的拼接格式不对。 - 检查网络连通性:确保运行脚本的服务器能够访问外网(
oapi.dingtalk.com或open.feishu.cn)。可以尝试ping一下域名。 - 查看机器人是否被禁用:在钉钉群机器人列表里,确认机器人是启用状态。
- 检查Webhook URL:确认URL完全正确,没有多余的空格或换行。最简单的方法是在命令行用
5.2 NVD API请求返回空数据或错误
- 问题:
fetch_cves_from_nvd函数返回的列表始终为空。 - 排查步骤:
- 检查时间参数:NVD API使用UTC时间。确保你代码中生成的时间字符串格式是
YYYY-MM-DDTHH:MM:SS.000。如果你的服务器是东八区时间,需要转换成UTC,或者直接使用datetime.utcnow()。 - 手动测试API:将代码中生成的URL(
base_url加上params)打印出来,直接在浏览器中访问,看看返回的JSON是否正常。例如:https://services.nvd.nist.gov/rest/json/cves/2.0?lastModStartDate=2024-01-01T00:00:00.000&lastModEndDate=2024-01-02T00:00:00.000。 - 关注速率限制:NVD对未认证的API调用有速率限制(通常每分钟5次,每小时50次)。如果你的脚本运行太频繁,可能会被暂时限制。添加
time.sleep()间隔,或者考虑申请一个API Key以获得更高的限额。 - 解析逻辑错误:API的JSON结构可能微调。打印出返回的
data变量,检查vulnerabilities、cve等关键字段的路径是否正确。使用.get()方法并提供默认值可以避免因字段缺失导致的程序中断。
- 检查时间参数:NVD API使用UTC时间。确保你代码中生成的时间字符串格式是
5.3 过滤效果不理想(漏报或误报)
- 问题:该报的漏洞没报,不该报的报了一大堆。
- 解决方案:
- 调整关键词:
keyword_filters是你的第一道防线。定期回顾和更新这个列表,加入你们新引入的技术组件。关键词不要太宽泛(如Java),尽量具体(如Spring Framework,Apache Tomcat)。 - 优化过滤逻辑:目前的过滤是“与”逻辑(必须同时满足分数、等级、关键词)。你可以考虑改为“或”逻辑,例如“CVSS > 9.0” 或 “包含关键词A或B”的漏洞都报警。这需要修改
filter_cves函数的逻辑。 - 引入CPE匹配:目前的关键词匹配在描述和产品字符串中搜索,不够精确。NVD数据中的
configurations字段包含了标准的CPE标识符。你可以编写更精确的CPE匹配逻辑,例如匹配cpe:2.3:a:apache:log4j:*:*:*:*:*:*:*:*这样的模式,来精准识别特定厂商和产品的所有版本。
- 调整关键词:
5.4 脚本在Crontab中运行异常
- 问题:手动执行
python3 main.py正常,但放在crontab里就不工作,也不生成日志。 - 排查技巧:
- 环境变量问题:
crontab执行的环境与用户shell环境不同,可能找不到python3命令或脚本依赖的库。在crontab中使用绝对路径:/usr/local/bin/python3。可以使用which python3命令查看绝对路径。 - 工作目录问题:脚本中使用的相对路径(如
config.yaml)在crontab执行时可能定位错误。在脚本开头使用os.chdir()切换到脚本所在目录,或者在crontab命令中先cd到目录:0 * * * * cd /opt/cve-monitor && /usr/bin/python3 main.py >> log.out 2>&1。 - 日志重定向:确保
crontab命令中的日志重定向路径是可写的。2>&1表示将标准错误也重定向到标准输出。检查生成的日志文件是否有权限错误。 - 依赖问题:如果使用了虚拟环境,需要在
crontab中激活它,或者直接使用虚拟环境内的python解释器:/path/to/venv/bin/python3 /path/to/main.py。
- 环境变量问题:
最后,这个脚本的价值在于它为你打开了一扇自动化安全运维的大门。从我个人的使用经验来看,最大的收获不是节省了每天刷CVE的那几分钟,而是建立了一种主动、持续的风险感知能力。它让安全信息像流水一样,自动、有序地流到需要它的人面前。你可以从这个小工具开始,逐步将它融入你们团队的DevSecOps流程,比如在CI/CD流水线中,增加一个检查环节,用类似的逻辑扫描项目依赖库的已知漏洞。自动化工具的意义,永远在于将人从重复、低价值的劳动中解放出来,去处理更复杂、更需要判断力的安全问题。