1. 项目概述：为什么需要一个双源CVE监控系统？

在安全运维和漏洞管理的日常工作中，我们常常面临一个核心痛点：信息滞后。一个高危漏洞（CVE）被披露后，从公开到被我们团队知晓，中间可能已经过去了数小时甚至一两天。这段时间差，就是攻击者最活跃的窗口期。传统的做法是手动刷新NVD（国家漏洞数据库）网站，或者订阅一些安全邮件列表，但这不仅效率低下，还容易遗漏。尤其是在面对像Log4Shell、Spring4Shell这类影响范围极广的漏洞时，早一分钟获取信息，就意味着早一分钟开始应急响应和修复工作。

“Medusa CVE监控系统”正是为了解决这个痛点而设计的。它的核心目标很简单： 实现CVE漏洞信息的实时、自动化监控与预警 。这个名字“Medusa”（美杜莎）很有意思，在希腊神话里，美杜莎的目光能让一切凝固；在安全领域，我们希望这个系统能“凝视”漏洞情报源，第一时间发现威胁并将其“定住”——也就是发出预警。

这个项目的独特之处在于其“双源”设计： GitHub 和 NIST 。为什么要选择这两个源？因为它们代表了漏洞情报的两种关键维度。NIST维护的NVD数据库是官方、权威的漏洞信息标准源，数据规范、描述准确，但更新有时存在延迟。而GitHub，特别是安全研究社区（如发布PoC的仓库、安全公司的研究博客同步等），往往是漏洞情报的第一现场，信息更前沿、更动态，甚至包含未分配正式CVE编号的漏洞讨论和概念验证代码。将两者结合，相当于既订阅了“官方新闻联播”，又监听了“前沿技术沙龙”，能最大程度避免漏报和误报。

这套系统适合谁？如果你是安全工程师、运维工程师、DevSecOps团队的成员，或者任何需要为自己负责的系统、应用保持漏洞感知的人，那么手动或半自动的监控方式迟早会让你疲于奔命。Medusa的目标就是帮你把这个过程自动化，把人力从重复的监控劳动中解放出来，聚焦于更重要的漏洞分析和修复工作。接下来，我将详细拆解这个系统的设计思路、核心实现以及我在搭建过程中踩过的坑和总结的经验。

2. 系统核心架构与设计思路拆解

一个监控系统，听起来简单，但要做到稳定、准确、易用，背后的设计考量不少。Medusa系统的设计遵循了“数据采集 -> 数据处理 -> 消息通知”的经典流水线模型，但在每个环节都针对CVE监控的特性做了优化。

2.1 整体架构与组件选型

系统的整体架构可以概括为以下几个核心组件：

  1. 数据采集器（Fetcher） ：负责定时从GitHub和NIST API拉取原始数据。
  2. 数据解析与去重引擎（Parser & Deduplicator） ：将原始JSON或HTML数据解析成结构化的漏洞信息，并基于CVE ID进行去重和关联。
  3. 存储与状态管理（Storage） ：存储已处理的漏洞信息，并记录其状态（如：新发现、已通知、已处理）。
  4. 规则引擎与过滤（Filter） ：允许用户自定义规则，例如只监控特定产品（如Apache, nginx）、特定严重等级（CRITICAL, HIGH）或包含特定关键词（如“RCE”, “Privilege Escalation”）的漏洞。
  5. 通知发送器（Notifier） ：将过滤后的漏洞警报通过多种渠道（如钉钉、企业微信、Slack、电子邮件）发送给相关人员。

在技术选型上，我倾向于使用轻量级、易于维护的方案：

  • 编程语言 ： Python 。生态丰富，有大量成熟的库用于HTTP请求、数据解析（JSON, HTML）、定时任务和消息发送，开发效率高。
  • 任务调度 ： APScheduler 或 Celery 。对于单机轻量级部署，APScheduler足够简单易用；如果需要分布式调度和更复杂的任务队列，Celery是更强大的选择。
  • 数据存储 ： SQLite 或 PostgreSQL 。初期或个人使用，SQLite零配置，单文件管理非常方便。团队协作或需要更复杂查询时，PostgreSQL更合适。核心表结构至少需要包含CVE ID、描述、严重性评分（CVSS）、受影响产品、来源、发现时间、通知状态等字段。
  • 部署方式 ： Docker容器化 。将整个应用及其依赖打包成Docker镜像，可以做到一键部署，环境隔离，也便于后续的扩展和迁移。

选择双源，意味着要处理两种不同结构、不同频率的数据流，这是设计上的一个挑战，也是价值所在。

2.2 双源数据融合策略解析

GitHub和NIST的数据特点截然不同，融合它们需要清晰的策略。

NIST NVD数据源 ：

  • 接口 ：官方提供免费的JSON格式API，例如获取最近2小时更新的CVE列表： https://services.nvd.nist.gov/rest/json/cves/2.0?lastModStartDate=...&lastModEndDate=...
  • 特点 ：数据结构化程度极高，字段规范，包含完整的CVE描述、CVSS评分向量、受影响的CPE（通用平台枚举）列表、参考链接等。这是我们的“基准真相”。
  • 挑战 ：API有速率限制（默认每小时50次请求，需申请API Key提升限制），且数据更新并非完全实时，可能存在数小时的延迟。

GitHub数据源 ：

  • 接口 ：主要通过GitHub Search API来搜索相关的仓库、Issue或代码。例如，搜索标题或描述中含有“CVE-2024-”且包含“PoC”或“exploit”的仓库。
  • 特点 ：信息极其前沿和动态。安全研究员经常在GitHub上第一时间发布漏洞分析、PoC代码、利用工具。这是早期预警的关键。
  • 挑战 ：噪音极大。搜索关键词可能匹配到无关项目、复现教程、甚至是虚假信息。数据非结构化，需要更复杂的文本解析和过滤逻辑。

融合策略 ：

  1. 以CVE ID为唯一锚点 ：无论来自哪个源，都尝试从文本中提取出标准的CVE ID（如CVE-2024-12345）。这是关联两条数据的最可靠凭证。
  2. 优先级与补全 ：当同一个CVE ID从两个源都被捕获时，优先采用NIST的权威结构化信息作为主记录。GitHub源的信息则作为“补充情报”，例如，将GitHub上找到的PoC代码链接、技术分析文章链接，附加到该CVE记录的参考信息中。
  3. 时间窗口关联 ：对于GitHub上先出现但尚未分配正式CVE ID的漏洞讨论，系统可以暂时用一个内部ID标记，并持续监控。一旦NIST源出现了描述匹配的新CVE，则进行关联合并。

这个策略确保了警报既快又准。GitHub给我们“快”的优势，NIST给我们“准”的保障。

3. 核心模块实现与实操要点

理论讲完了，我们进入实战环节。我会分模块介绍如何实现Medusa系统的核心功能，并附上关键的代码片段和配置说明。

3.1 NIST数据采集器的实现

与NVD API交互，核心是构造正确的请求参数和处理分页。这里的关键是控制好请求频率，避免触发速率限制。 

import requests
import time
from datetime import datetime, timedelta
import sqlite3

class NISTFetcher:
    def __init__(self, api_key=None, base_url="https://services.nvd.nist.gov/rest/json/cves/2.0"):
        self.base_url = base_url
        self.api_key = api_key
        self.headers = {"apiKey": self.api_key} if api_key else {}

    def fetch_recent_cves(self, hours=2):
        """获取最近N小时内更新的CVE"""
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(hours=hours)
        # NVD API要求的时间格式
        start_str = start_time.isoformat(timespec='seconds') + "Z"
        end_str = end_time.isoformat(timespec='seconds') + "Z"

        params = {
            "lastModStartDate": start_str,
            "lastModEndDate": end_str,
            "startIndex": 0
        }
        all_cves = []
        while True:
            try:
                resp = requests.get(self.base_url, params=params, headers=self.headers, timeout=30)
                resp.raise_for_status()
                data = resp.json()
                vulnerabilities = data.get('vulnerabilities', [])
                all_cves.extend([v['cve'] for v in vulnerabilities])

                total_results = data.get('totalResults', 0)
                results_per_page = data.get('resultsPerPage', 0)
                current_index = params['startIndex']

                if current_index + results_per_page >= total_results:
                    break
                params['startIndex'] += results_per_page
                time.sleep(1) # 礼貌性延迟，避免请求过快
            except requests.exceptions.RequestException as e:
                print(f"请求NVD API失败: {e}")
                break
        return all_cves

注意 ：务必申请一个NVD API Key（免费），这能将每小时请求限制从50次提升到100次。在生产环境中，建议将API Key存储在环境变量中，而不是硬编码在代码里。

3.2 GitHub情报监控的实现

监控GitHub更侧重于“搜索”而非“订阅”。我们需要精心设计搜索查询语句，以平衡查全率和查准率。 

import requests
from urllib.parse import quote

class GitHubFetcher:
    def __init__(self, token=None):
        self.base_url = "https://api.github.com/search/repositories"
        self.headers = {"Accept": "application/vnd.github.v3+json"}
        if token:
            self.headers["Authorization"] = f"token {token}"

    def search_cve_related(self, keyword="CVE-2024", sort="updated", order="desc"):
        """搜索与CVE相关的仓库"""
        # 构造查询语句，可以组合多种条件
        # 例如：搜索名称或描述中包含CVE-2024，并且包含‘poc’或‘exploit’的仓库
        query = f"{keyword} in:name,description poc OR exploit"
        encoded_query = quote(query)
        url = f"{self.base_url}?q={encoded_query}&sort={sort}&order={order}&per_page=30"

        try:
            resp = requests.get(url, headers=self.headers, timeout=30)
            resp.raise_for_status()
            return resp.json().get('items', [])
        except requests.exceptions.RequestException as e:
            print(f"搜索GitHub失败: {e}")
            return []

实操心得 ：GitHub搜索语法非常强大。除了 in:name,description ，还可以用 language:python 限定语言，用 pushed:>2024-01-01 限定更新时间。多尝试不同的关键词组合，并定期评估搜索结果的质量，动态调整你的搜索策略。使用GitHub Token可以大幅提高API的速率限制（从60次/小时到5000次/小时）。

3.3 数据解析、去重与存储

从两个源获取的数据需要清洗、解析并存入数据库。这里以SQLite为例，展示核心的数据处理逻辑。 

import re
import sqlite3
from datetime import datetime

class CVEProcessor:
    def __init__(self, db_path='cve_monitor.db'):
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        """初始化数据库表"""
        cursor = self.conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS cves (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                cve_id TEXT UNIQUE NOT NULL,
                description TEXT,
                severity TEXT,
                cvss_score REAL,
                affected_products TEXT, -- 可存储为JSON字符串
                source TEXT, -- 'nist' 或 'github'
                source_url TEXT,
                poc_url TEXT, -- 来自GitHub的PoC链接
                published_date TEXT,
                last_modified_date TEXT,
                notified BOOLEAN DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.conn.commit()

    def parse_and_store_nist(self, cve_data_list):
        """解析并存储NIST CVE数据"""
        for cve_item in cve_data_list:
            cve_id = cve_item.get('id')
            descriptions = cve_item.get('descriptions', [])
            # 优先取英文描述
            description = next((d['value'] for d in descriptions if d['lang'] == 'en'), '')
            metrics = cve_item.get('metrics', {})
            cvss_v3 = metrics.get('cvssMetricV31', [{}])[0]
            cvss_data = cvss_v3.get('cvssData', {})
            severity = cvss_data.get('baseSeverity', 'N/A')
            score = cvss_data.get('baseScore', 0.0)

            # 提取受影响产品（简化处理，取第一个配置）
            affected = []
            configurations = cve_item.get('configurations', [])
            for config in configurations:
                for node in config.get('nodes', []):
                    for cpe_match in node.get('cpeMatch', []):
                        affected.append(cpe_match.get('criteria', ''))
            affected_str = ','.join(affected[:5]) # 只存前几个

            cursor = self.conn.cursor()
            try:
                cursor.execute('''
                    INSERT OR IGNORE INTO cves (cve_id, description, severity, cvss_score, affected_products, source, source_url, published_date)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ''', (cve_id, description, severity, score, affected_str, 'nist', f"https://nvd.nist.gov/vuln/detail/{cve_id}", cve_item.get('published')))
            except sqlite3.IntegrityError:
                # 如果已存在，则更新部分信息（如描述可能更详细）
                pass
        self.conn.commit()

    def extract_cve_id_from_text(self, text):
        """从文本中提取CVE ID，使用正则表达式"""
        pattern = r'CVE-\d{4}-\d{4,}'
        matches = re.findall(pattern, text, re.IGNORECASE)
        return list(set(matches)) # 去重

注意事项 ：数据库设计要考虑扩展性。 affected_products 字段存储为文本可能不利于后续的精确过滤。在实际项目中，可以考虑将其拆分为单独的表，或者使用支持JSON查询的数据库（如PostgreSQL）。 OR IGNORE 或 ON CONFLICT UPDATE 子句能很好地处理重复CVE ID的问题。

3.4 规则引擎与智能过滤

不是每一个CVE都需要报警。我们需要一个灵活的过滤规则系统。这里实现一个基于简单规则引擎的过滤器。 

class CVEFilter:
    def __init__(self, rules):
        """
        rules: 规则列表，每条规则是一个字典。
        例如: {'field': 'severity', 'op': 'in', 'value': ['CRITICAL', 'HIGH']}
             {'field': 'description', 'op': 'contains', 'value': 'Apache Tomcat'}
        """
        self.rules = rules

    def apply(self, cve_record):
        """应用所有规则，只有全部通过才返回True"""
        for rule in self.rules:
            field_value = cve_record.get(rule['field'], '')
            op = rule['op']
            rule_value = rule['value']

            if op == 'equals' and field_value != rule_value:
                return False
            elif op == 'contains' and rule_value not in str(field_value):
                return False
            elif op == 'in' and field_value not in rule_value:
                return False
            elif op == 'gt' and not (isinstance(field_value, (int, float)) and field_value > rule_value):
                return False
            # 可以扩展更多操作符...
        return True

# 示例规则：只关注严重或高危，且影响Apache或nginx的漏洞
my_rules = [
    {'field': 'severity', 'op': 'in', 'value': ['CRITICAL', 'HIGH']},
    {'field': 'affected_products', 'op': 'contains', 'value': ['apache', 'nginx']}
]
filter_engine = CVEFilter(my_rules)

你可以将规则配置在YAML或JSON文件中，这样无需修改代码就能调整监控策略。例如，在应急响应期间，可以临时添加一条规则，监控所有包含特定产品名称的CVE，无论其严重等级。

3.5 多渠道通知发送集成

警报必须触达责任人。集成常见的消息平台是关键一步。以钉钉机器人为例： 

import requests
import json

class DingTalkNotifier:
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url

    def send(self, cve_list):
        """发送CVE警报到钉钉群"""
        if not cve_list:
            return
        # 构造Markdown格式消息
        title = "🚨 发现新的高危CVE漏洞"
        text = f"### {title}\n\n"
        for cve in cve_list[:5]: # 一次最多发送5条，避免消息过长
            text += f"**{cve['cve_id']} - {cve['severity']} ({cve['cvss_score']})**\n"
            text += f"> {cve['description'][:150]}...\n"
            text += f"影响产品: {cve['affected_products'][:100]}...\n"
            text += f"[详情]({cve['source_url']})\n\n"
            if cve.get('poc_url'):
                text += f"[PoC链接]({cve['poc_url']})\n\n"

        message = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text
            },
            "at": {
                "isAtAll": False # 根据需要@特定人或所有人
            }
        }
        try:
            resp = requests.post(self.webhook_url, json=message, timeout=10)
            resp.raise_for_status()
            print("钉钉通知发送成功")
        except Exception as e:
            print(f"钉钉通知发送失败: {e}")

类似地，可以封装 EmailNotifier 、 WeComNotifier （企业微信）、 SlackNotifier 等。建议实现一个通知管理器，支持同时向多个渠道发送，并具备失败重试机制。

4. 部署、调度与运维实践

系统开发完成后，如何让它7x24小时稳定运行？这里涉及到部署、调度和日常维护。

4.1 使用Docker容器化部署

编写一个 Dockerfile 和 docker-compose.yml ，能让部署变得极其简单。

Dockerfile : 

FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "main_scheduler.py"]

docker-compose.yml : 

version: '3.8'
services:
  medusa:
    build: .
    container_name: cve-monitor-medusa
    restart: unless-stopped
    volumes:
      - ./data:/app/data  # 挂载数据库文件
      - ./config:/app/config  # 挂载配置文件
    environment:
      - TZ=Asia/Shanghai
      - NVD_API_KEY=${NVD_API_KEY}
      - GITHUB_TOKEN=${GITHUB_TOKEN}
      - DINGTALK_WEBHOOK=${DINGTALK_WEBHOOK}

使用 docker-compose up -d 即可后台启动。 restart: unless-stopped 保证了容器意外退出时会自动重启，增强了可靠性。

4.2 任务调度与频率设置

调度策略直接影响系统的实时性和对API的友好程度。不建议设置得过密。 

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

scheduler = BlockingScheduler(timezone="Asia/Shanghai")

# 每30分钟抓取一次NIST数据（NVD更新频率本身不高）
@scheduler.scheduled_job('interval', minutes=30)
def fetch_nist_job():
    print(f"{datetime.now()} - 开始执行NIST数据抓取")
    fetcher = NISTFetcher(api_key=os.getenv('NVD_API_KEY'))
    cves = fetcher.fetch_recent_cves(hours=1) # 抓取过去1小时更新的
    if cves:
        processor.parse_and_store_nist(cves)
        # 触发后续处理和通知
        check_and_notify_new_cves()
    print(f"{datetime.now()} - NIST数据抓取完成")

# 每20分钟搜索一次GitHub（信息更动态，频率可稍高）
@scheduler.scheduled_job('interval', minutes=20)
def fetch_github_job():
    print(f"{datetime.now()} - 开始执行GitHub情报搜索")
    fetcher = GitHubFetcher(token=os.getenv('GITHUB_TOKEN'))
    repos = fetcher.search_cve_related(keyword="CVE-2024")
    for repo in repos:
        # 提取CVE ID，解析，存储（标记来源为github）
        # ...
        pass
    print(f"{datetime.now()} - GitHub情报搜索完成")

if __name__ == '__main__':
    scheduler.start()

重要提醒 ：务必遵守各平台的API速率限制。过于频繁的请求可能导致IP或API Key被临时封禁。合理的间隔（如20-30分钟）既能满足实时性要求，又显得友好。可以将调度时间配置化，便于调整。

4.3 日志记录与监控

一个后台运行的系统，必须有清晰的日志，方便排查问题。 

import logging
import sys

def setup_logger():
    logger = logging.getLogger('medusa')
    logger.setLevel(logging.INFO)
    # 控制台输出
    console_handler = logging.StreamHandler(sys.stdout)
    console_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    console_handler.setFormatter(console_format)
    logger.addHandler(console_handler)
    # 文件输出
    file_handler = logging.FileHandler('medusa.log', encoding='utf-8')
    file_handler.setLevel(logging.WARNING) # 文件只记录警告及以上
    file_format = logging.Formatter('%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s')
    file_handler.setFormatter(file_format)
    logger.addHandler(file_handler)
    return logger

logger = setup_logger()
# 在代码中使用 logger.info(), logger.warning(), logger.error()

除了记录到文件，还可以考虑将错误日志接入像Sentry这样的应用监控平台，实现异常报警。同时，可以编写一个简单的健康检查接口或脚本，定时检查主程序是否在运行，数据库是否可连接。

5. 常见问题排查与优化经验实录

在实际搭建和运行Medusa系统的过程中，我遇到了不少典型问题。这里把它们总结出来，希望能帮你避开这些坑。

5.1 数据源获取失败问题

问题1：NVD API返回403或速率限制错误。

  • 现象 ：日志中频繁出现 HTTP 403 Forbidden 或 API rate limit exceeded 。
  • 排查 ：
    1. 检查是否使用了API Key。没有API Key时，每小时限制50次请求，很容易超限。
    2. 检查代码中的请求间隔。即使有API Key，过于频繁的请求（如每秒一次）也会被限制。
    3. 确认API Key是否正确，是否已过期（虽然NVD的Key目前似乎不会过期）。
  • 解决 ：
    • 务必申请并使用API Key 。
    • 增加请求间隔 。对于定时抓取任务，30分钟或1小时一次完全足够。在代码中，每次请求后使用 time.sleep(1) 进行短暂暂停也是好习惯。
    • 实现重试机制 。对于偶发的网络错误或429（Too Many Requests）状态码，可以使用指数退避算法进行重试。

问题2：GitHub API返回认证失败或结果为空。

  • 现象 ： HTTP 401 Unauthorized 或搜索始终返回空列表。
  • 排查 ：
    1. 检查GitHub Token是否有 repo 和 read:packages 等必要的权限（对于公开仓库搜索，其实不需要特殊权限，但使用Token能提高速率限制）。
    2. 检查搜索关键词（ keyword ）是否太宽泛或太冷僻。可以先用GitHub网页版手动搜索测试一下。
    3. GitHub Search API有索引延迟，新创建的仓库可能不会立即出现在搜索结果中。
  • 解决 ：
    • 使用有效的GitHub Personal Access Token。
    • 优化搜索语法。例如， CVE-2024- in:name,description poc 比单纯的 CVE 更精确。可以结合 language: 、 pushed: 等过滤器。
    • 接受一定程度的延迟，这不是实时流式API。

5.2 数据解析与去重逻辑缺陷

问题3：同一个CVE被重复报警多次。

  • 现象 ：钉钉群里连续收到好几条一模一样的CVE警报。
  • 排查 ：
    1. 检查数据库的 INSERT 逻辑是否使用了 ON CONFLICT 或先查询后插入的方式来保证唯一性。
    2. 检查 notified 状态字段是否在成功发送通知后被正确更新为 True 。
    3. 检查调度任务是否在短时间内重复运行，导致在状态更新前又处理了同一批数据。
  • 解决 ：
    • 在数据库层面为 cve_id 字段设置 UNIQUE 约束，这是最根本的保障。
    • 在发送通知前，必须检查 notified 字段是否为 False 。发送成功后，立即在同一个数据库事务中将其更新为 True 。
    • 确保任务调度是“幂等”的，即多次执行同一任务不会产生副作用。

问题4：GitHub信息中提取的CVE ID不准确或遗漏。

  • 现象 ：有些明显的漏洞讨论没被监控到，或者错误地将非CVE编号（如内部编号）识别为CVE。
  • 排查 ：
    1. 正则表达式 r'CVE-\d{4}-\d{4,}' 可能无法匹配所有变体，比如全小写的 cve-2024-12345 ，或者编号位数不同的老CVE（如CVE-1999-0067）。
    2. 有些文章可能在正文深处才提到CVE ID，而你的搜索只针对标题和描述。
  • 解决 ：
    • 使用更健壮的正则表达式，例如 r'CVE-\d{4}-\d+ ，并添加 re.IGNORECASE 标志忽略大小写。
    • 考虑在获取仓库基本信息后，进一步爬取README文件或最近提交的代码/文档内容，进行更深度的文本分析来提取CVE ID。但这会显著增加复杂度和API调用。

5.3 通知发送与性能优化

问题5：通知发送失败，但程序没有记录或重试。

  • 现象 ：数据库里有新CVE，但没收到任何通知。
  • 排查 ：
    1. 网络问题导致HTTP请求失败。
    2. 钉钉/企业微信等Webhook地址配置错误或已失效。
    3. 消息内容过长或格式不符合机器人要求，被平台拒绝。
  • 解决 ：
    • 实现通知发送的失败重试机制 。例如，捕获请求异常，记录错误日志，并将该条通知放入一个重试队列，稍后再次尝试（例如5分钟后），最多重试3次。
    • 对消息内容进行裁剪 。钉钉/企业微信的Markdown消息有长度限制。对于描述过长的CVE，可以截断，并优先保证CVE ID、严重等级和详情链接的完整性。
    • 添加通知发送状态的日志 。每次尝试发送，无论成功失败，都记录详细的日志，方便溯源。

问题6：随着运行时间增长，数据库查询变慢。

  • 现象 ：系统运行几个月后，每次检查新漏洞或查询历史漏洞的响应时间明显变长。
  • 排查 ： cves 表数据量过大，且缺乏有效的索引。
  • 解决 ：
    • 为最常用的查询字段建立索引。例如， cve_id （唯一索引）、 severity 、 published_date 和 notified 。 
    CREATE INDEX idx_cves_severity ON cves (severity);
CREATE INDEX idx_cves_published ON cves (published_date);
CREATE INDEX idx_cves_notified ON cves (notified);
    • 考虑定期归档历史数据。例如，可以写一个脚本，将3个月前的、状态已是 notified=1 且严重等级非CRITICAL的CVE记录转移到历史表，或者直接导出为CSV备份后从主表删除，保持主表轻量。

5.4 系统扩展性思考

当监控的漏洞数量或规则非常多时，单机单线程的模式可能会遇到瓶颈。可以考虑以下优化方向：

  • 微服务化 ：将数据采集、解析、存储、通知拆分为独立的微服务，通过消息队列（如Redis, RabbitMQ）进行通信。这样每个部分可以独立扩展和部署。
  • 更智能的过滤 ：引入简单的机器学习模型，对漏洞描述进行文本分类，自动判断是否与自身业务相关，而不仅仅是关键词匹配。
  • 可视化仪表盘 ：集成Grafana，从数据库读取数据，展示漏洞趋势图、各产品漏洞分布、团队响应时间等指标，让安全状况一目了然。
  • 与CMDB/资产管理系统联动 ：将监控到的CVE与公司内部的资产库进行自动匹配，只对影响实际资产的漏洞发出警报，实现精准打击。

搭建Medusa系统的过程，是一个典型的“运维开发”实践。它没有多高深的技术，但非常实用，能切实提升安全运营的效率和主动性。从最简单的脚本开始，逐步迭代功能，最终形成一个可靠的小型系统，这种成就感是巨大的。最关键的是，通过这个过程，你不仅得到了一个工具，更深入理解了漏洞情报的生命周期和自动化运维的思维方式。

Logo
亚马逊云科技技术品牌专区

更多推荐

cover

基于大数据+Hadoop+深度学习的淘宝用户购物可视化与行为预测系统设计与开发

avatar 亚马逊云科技技术品牌专区
cover

基于大数据爬虫+Hadoop+深度学习的旅游推荐系统设计与开发

avatar 亚马逊云科技技术品牌专区

学Linux选学校：一线城市还是老家？

而二线城市很多机构的老师，自己都没进过大厂，技术停留在五年前的水平，甚至对着PPT照本宣科。而二线机构缺乏企业合作资源，项目案例永远是“搭建博客系统”“配置个LAMP环境”——这些在现在的云计算岗位面试中，连初筛都过不了。在一线城市学习，你接触的同学、老师、企业内推资源，都是二线城市没法比的。而老家机构的学员，最好的结果也就是去本地小公司做网管，月薪五六千，天花板肉眼可见。在IT培训这行干了快十年

avatar 亚马逊云科技技术品牌专区