实战：利用 NVD API 与 Python 脚本批量获取并分析 1000 个 CVE 的 CVSS 3.1 分数

实战基于NVD API与Python的CVE批量分析系统开发指南1. 环境准备与API基础在开始构建CVE分析系统前我们需要配置合适的开发环境。推荐使用Python 3.8版本这是目前大多数安全工具链兼容性最好的版本。以下是基础环境配置步骤# 创建虚拟环境 python -m venv cve_analysis source cve_analysis/bin/activate # Linux/macOS cve_analysis\Scripts\activate # Windows # 安装核心依赖 pip install requests pandas matplotlib seabornNVD官方提供了两种API访问方式免费版每小时限50次请求无需认证企业版无限制访问需要API密钥对于我们的批量分析需求建议注册获取免费API密钥以避免速率限制。以下是API关键端点端点类型URL格式说明CVE详情https://services.nvd.nist.gov/rest/json/cves/2.0?cveId{CVE-ID}获取单个CVE完整数据批量查询https://services.nvd.nist.gov/rest/json/cves/2.0分页获取CVE列表关键词搜索https://services.nvd.nist.gov/rest/json/cves/2.0?keywordSearch{query}按关键词过滤提示实际调用时需添加apiKey参数例如apiKey您的密钥2. 高效数据获取策略2.1 分页获取机制NVD API采用分页返回结果每页最多2000条记录。以下是优化后的分页获取代码import requests import time def fetch_cves(start_index0, results_per_page2000, api_keyNone): base_url https://services.nvd.nist.gov/rest/json/cves/2.0 headers {Accept: application/json} params { startIndex: start_index, resultsPerPage: results_per_page } if api_key: params[apiKey] api_key try: response requests.get(base_url, headersheaders, paramsparams) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f请求失败: {e}) return None2.2 请求限速处理为避免触发API限制需要实现智能节流控制from ratelimit import limits, sleep_and_retry # 限制为每分钟45次请求保留安全余量 sleep_and_retry limits(calls45, period60) def safe_fetch(start_index): return fetch_cves(start_indexstart_index)2.3 断点续传设计对于大规模数据获取建议实现持久化记录import json from pathlib import Path def batch_fetch(total1000, batch_size200, resume_fileprogress.json): progress {last_index: 0} if Path(resume_file).exists(): with open(resume_file) as f: progress json.load(f) results [] while len(results) total: data safe_fetch(progress[last_index]) if not data: break results.extend(data[vulnerabilities]) progress[last_index] len(data[vulnerabilities]) with open(resume_file, w) as f: json.dump(progress, f) print(f已获取 {len(results)}/{total} 条记录) if len(data[vulnerabilities]) batch_size: break return results[:total]3. CVSS 3.1深度解析3.1 向量字符串解码CVSS 3.1向量字符串示例CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H我们可以构建解析器将其转换为结构化数据def parse_cvss_vector(vector_string): if not vector_string or not vector_string.startswith(CVSS:3.1): return None metrics {} parts vector_string.split(/)[1:] # 去掉前缀 for part in parts: if : not in part: continue key, value part.split(:) metrics[key] value return metrics3.2 关键指标权重表以下是CVSS 3.1基础指标权重分配指标缩写权重可能值攻击向量AV0.20N(网络), A(相邻), L(本地), P(物理)攻击复杂度AC0.22L(低), H(高)权限要求PR0.27N(无), L(低), H(高)用户交互UI0.08N(无), R(需要)范围S0.33U(不变), C(改变)机密性影响C0.22H(高), L(低), N(无)完整性影响I0.22H(高), L(低), N(无)可用性影响A0.22H(高), L(低), N(无)3.3 分数计算实现根据官方公式实现分数计算def calculate_base_score(metrics): # 实现ISS和Impact计算 iss 1 - ((1 - metrics[C]) * (1 - metrics[I]) * (1 - metrics[A])) if metrics[S] U: impact 6.42 * iss else: impact 7.52 * (iss - 0.029) - 3.25 * (iss - 0.02)**15 # 实现Exploitability计算 exploitability 8.22 * metrics[AV] * metrics[AC] * metrics[PR] * metrics[UI] # 最终分数计算 if impact 0: return 0 if metrics[S] U: return round(min(impact exploitability, 10), 1) else: return round(min(1.08 * (impact exploitability), 10), 1)4. 数据分析与可视化4.1 高危漏洞筛选策略定义高危漏洞标准评分 ≥ 9.0严重漏洞7.0 ≤ 评分 9.0高危漏洞4.0 ≤ 评分 7.0中危漏洞评分 4.0低危漏洞使用Pandas进行快速筛选import pandas as pd def analyze_cves(cve_list): df pd.DataFrame([{ CVE_ID: item[cve][id], Published: item[cve][published], CVSS_Score: float(item[cve][metrics][cvssMetricV31][0][cvssData][baseScore]), Vector: item[cve][metrics][cvssMetricV31][0][cvssData][vectorString] } for item in cve_list if cvssMetricV31 in item[cve][metrics]]) df[Severity] pd.cut(df[CVSS_Score], bins[0, 4, 7, 9, 10], labels[Low, Medium, High, Critical], rightFalse) return df4.2 时间趋势分析使用Matplotlib生成漏洞时间分布图import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter def plot_trend(df): fig, ax plt.subplots(figsize(12, 6)) # 按月统计漏洞数量 monthly df.set_index(pd.to_datetime(df[Published])).resample(M).size() ax.plot(monthly.index, monthly.values, markero, linestyle-) ax.set_title(CVE Publication Trend) ax.set_ylabel(Number of CVEs) ax.xaxis.set_major_formatter(DateFormatter(%Y-%m)) plt.xticks(rotation45) plt.tight_layout() plt.savefig(cve_trend.png, dpi300) plt.close()4.3 攻击向量分布生成攻击向量环形图def plot_attack_vectors(df): vectors df[Vector].apply(lambda x: x.split(/)[1].split(:)[1]) counts vectors.value_counts() fig, ax plt.subplots(figsize(8, 8)) ax.pie(counts, labelscounts.index, autopct%1.1f%%, wedgepropsdict(width0.4)) ax.set_title(Attack Vector Distribution) plt.savefig(attack_vectors.png, dpi300) plt.close()5. 自动化报告生成5.1 报告模板设计使用Jinja2模板引擎生成HTML报告from jinja2 import Environment, FileSystemLoader def generate_report(df, template_dirtemplates): env Environment(loaderFileSystemLoader(template_dir)) template env.get_template(report.html) stats { total: len(df), critical: len(df[df[Severity] Critical]), high: len(df[df[Severity] High]), top_cves: df.nlargest(5, CVSS_Score).to_dict(records) } with open(cve_report.html, w) as f: f.write(template.render(statsstats, trend_imagecve_trend.png, vector_imageattack_vectors.png))5.2 关键指标表格在报告中嵌入动态生成的指标表格table classmetrics-table tr thSeverity/th thCount/th thPercentage/th /tr {% for level in [Critical, High, Medium, Low] %} tr td{{ level }}/td td{{ stats[level.lower()] }}/td td{{ (stats[level.lower()]/stats.total*100)|round(1) }}%/td /tr {% endfor %} /table6. 系统优化与扩展6.1 缓存机制实现使用Redis缓存API响应import redis import pickle class CVECache: def __init__(self, hostlocalhost, port6379, db0): self.redis redis.Redis(hosthost, portport, dbdb) def get(self, cve_id): data self.redis.get(cve_id) return pickle.loads(data) if data else None def set(self, cve_id, data, expire86400): self.redis.setex(cve_id, expire, pickle.dumps(data))6.2 异步处理架构使用Celery实现任务队列from celery import Celery app Celery(cve_tasks, brokerredis://localhost:6379/0) app.task(bindTrue) def fetch_cve_task(self, cve_id): try: data fetch_single_cve(cve_id) cache.set(cve_id, data) return data except Exception as e: self.retry(exce, countdown60)6.3 企业级部署建议对于生产环境推荐以下架构负载均衡层 → API网关 → 微服务集群 → 分布式缓存 → 数据库集群关键配置参数每个worker进程内存限制2GB最大并发请求数50/worker数据库连接池大小CPU核心数×2 17. 安全注意事项7.1 API密钥管理采用环境变量存储敏感信息import os from dotenv import load_dotenv load_dotenv() API_KEY os.getenv(NVD_API_KEY)7.2 数据存储加密使用AES加密存储本地数据from cryptography.fernet import Fernet def encrypt_data(data, key): fernet Fernet(key) return fernet.encrypt(pickle.dumps(data)) def decrypt_data(encrypted, key): fernet Fernet(key) return pickle.loads(fernet.decrypt(encrypted))7.3 异常处理策略实现全面的错误处理def robust_fetch(url, max_retries3): for attempt in range(max_retries): try: response requests.get(url, timeout10) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: if e.response.status_code 404: raise wait 2 ** attempt time.sleep(wait) except requests.exceptions.RequestException: time.sleep(1) raise Exception(fFailed after {max_retries} attempts)