cve-tracking-system

CVE 跟踪系统专家

漏洞跟踪与安全管理专家。

核心能力

CVE 管理

• CVE 数据库监控
• CVSS 评分解读
• 漏洞优先级排序
• NVD/MITRE 集成

漏洞工作流

• 分诊与评估
• 修复规划
• 补丁管理
• 验证测试

合规与报告

• 安全公告
• 审计报告
• SLA 跟踪
• 高管仪表板

CVE 数据标准

CVE ID 格式

CVE-YYYY-NNNNN
│    │    │
│    │    └─ Sequential number (4-7 digits)
│    └────── Year of assignment
└─────────── Prefix

CVSS v3.1 指标

Base Score (0-10):
  Attack Vector (AV):
    - Network (N): 0.85
    - Adjacent (A): 0.62
    - Local (L): 0.55
    - Physical (P): 0.20

  Attack Complexity (AC):
    - Low (L): 0.77
    - High (H): 0.44

  Privileges Required (PR):
    - None (N): 0.85
    - Low (L): 0.62/0.68
    - High (H): 0.27/0.50

  User Interaction (UI):
    - None (N): 0.85
    - Required (R): 0.62

  Scope (S):
    - Unchanged (U)
    - Changed (C)

  Impact (CIA):
    - Confidentiality: None/Low/High
    - Integrity: None/Low/High
    - Availability: None/Low/High

严重性级别

分数	评级	颜色
9.0-10.0	严重	红色
7.0-8.9	高	橙色
4.0-6.9	中	黄色
0.1-3.9	低	绿色
0.0	无	灰色

NVD API 集成

import requests
from datetime import datetime, timedelta

class NVDClient:
    BASE_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"

    def __init__(self, api_key: str = None):
        self.api_key = api_key
        self.headers = {"apiKey": api_key} if api_key else {}

    def get_recent_cves(self, days: int = 7) -> list:
        """Fetch CVEs from the last N days."""
        end_date = datetime.utcnow()
        start_date = end_date - timedelta(days=days)

        params = {
            "pubStartDate": start_date.strftime("%Y-%m-%dT%H:%M:%S.000"),
            "pubEndDate": end_date.strftime("%Y-%m-%dT%H:%M:%S.000")
        }

        response = requests.get(
            self.BASE_URL,
            params=params,
            headers=self.headers
        )
        response.raise_for_status()

        return self._parse_cves(response.json())

    def search_by_keyword(self, keyword: str) -> list:
        """Search CVEs by keyword."""
        params = {"keywordSearch": keyword}
        response = requests.get(
            self.BASE_URL,
            params=params,
            headers=self.headers
        )
        return self._parse_cves(response.json())

    def get_by_cpe(self, cpe: str) -> list:
        """Get CVEs affecting a specific CPE."""
        params = {"cpeName": cpe}
        response = requests.get(
            self.BASE_URL,
            params=params,
            headers=self.headers
        )
        return self._parse_cves(response.json())

    def _parse_cves(self, data: dict) -> list:
        """Parse NVD response into structured CVE objects."""
        cves = []
        for item in data.get("vulnerabilities", []):
            cve = item.get("cve", {})
            cves.append({
                "id": cve.get("id"),
                "published": cve.get("published"),
                "description": self._get_description(cve),
                "cvss": self._get_cvss(cve),
                "affected_products": self._get_cpe(cve)
            })
        return cves

    def _get_description(self, cve: dict) -> str:
        descriptions = cve.get("descriptions", [])
        for desc in descriptions:
            if desc.get("lang") == "en":
                return desc.get("value", "")
        return ""

    def _get_cvss(self, cve: dict) -> dict:
        metrics = cve.get("metrics", {})
        if "cvssMetricV31" in metrics:
            return metrics["cvssMetricV31"][0]["cvssData"]
        if "cvssMetricV30" in metrics:
            return metrics["cvssMetricV30"][0]["cvssData"]
        return {}

CVSS 计算器

class CVSSCalculator:
    """CVSS v3.1 Base Score Calculator."""

    WEIGHTS = {
        "AV": {"N": 0.85, "A": 0.62, "L": 0.55, "P": 0.20},
        "AC": {"L": 0.77, "H": 0.44},
        "PR": {
            "U": {"N": 0.85, "L": 0.62, "H": 0.27},
            "C": {"N": 0.85, "L": 0.68, "H": 0.50}
        },
        "UI": {"N": 0.85, "R": 0.62},
        "C": {"H": 0.56, "L": 0.22, "N": 0},
        "I": {"H": 0.56, "L": 0.22, "N": 0},
        "A": {"H": 0.56, "L": 0.22, "N": 0}
    }

    def calculate(self, vector: dict) -> dict:
        """Calculate CVSS score from vector components."""
        scope = vector.get("S", "U")

        # Impact Sub Score
        isc_base = 1 - (
            (1 - self.WEIGHTS["C"][vector["C"]]) *
            (1 - self.WEIGHTS["I"][vector["I"]]) *
            (1 - self.WEIGHTS["A"][vector["A"]])
        )

        if scope == "U":
            impact = 6.42 * isc_base
        else:
            impact = 7.52 * (isc_base - 0.029) - 3.25 * pow(isc_base - 0.02, 15)

        # Exploitability Sub Score
        exploitability = (
            8.22 *
            self.WEIGHTS["AV"][vector["AV"]] *
            self.WEIGHTS["AC"][vector["AC"]] *
            self.WEIGHTS["PR"][scope][vector["PR"]] *
            self.WEIGHTS["UI"][vector["UI"]]
        )

        # Base Score
        if impact <= 0:
            base_score = 0
        elif scope == "U":
            base_score = min(impact + exploitability, 10)
        else:
            base_score = min(1.08 * (impact + exploitability), 10)

        return {
            "base_score": round(base_score, 1),
            "impact": round(impact, 1),
            "exploitability": round(exploitability, 1),
            "severity": self._get_severity(base_score)
        }

    def _get_severity(self, score: float) -> str:
        if score >= 9.0: return "CRITICAL"
        if score >= 7.0: return "HIGH"
        if score >= 4.0: return "MEDIUM"
        if score > 0: return "LOW"
        return "NONE"

资产清单与 CPE 匹配

class AssetInventory:
    """Track software assets and match against CVEs."""

    def __init__(self):
        self.assets = []

    def add_asset(self, asset: dict):
        """Add software asset to inventory."""
        self.assets.append({
            "name": asset["name"],
            "vendor": asset["vendor"],
            "version": asset["version"],
            "cpe": self._generate_cpe(asset),
            "criticality": asset.get("criticality", "medium"),
            "environment": asset.get("environment", "production")
        })

    def _generate_cpe(self, asset: dict) -> str:
        """Generate CPE 2.3 string for asset."""
        return f"cpe:2.3:a:{asset['vendor']}:{asset['name']}:{asset['version']}:*:*:*:*:*:*:*"

    def find_affected(self, cve_cpes: list) -> list:
        """Find assets affected by CVE."""
        affected = []
        for asset in self.assets:
            for cve_cpe in cve_cpes:
                if self._cpe_matches(asset["cpe"], cve_cpe):
                    affected.append(asset)
                    break
        return affected

    def _cpe_matches(self, asset_cpe: str, cve_cpe: str) -> bool:
        """Check if asset CPE matches CVE CPE pattern."""
        asset_parts = asset_cpe.split(":")
        cve_parts = cve_cpe.split(":")

        for i, (a, c) in enumerate(zip(asset_parts, cve_parts)):
            if c == "*":
                continue
            if a != c:
                return False
        return True

漏洞工作流

自动分诊系统

class VulnerabilityTriage:
    """Automated vulnerability triage based on risk factors."""

    SLA = {
        "CRITICAL": {"response": 24, "remediation": 72},
        "HIGH": {"response": 48, "remediation": 168},
        "MEDIUM": {"response": 168, "remediation": 720},
        "LOW": {"response": 720, "remediation": 2160}
    }

    def calculate_priority(self, vuln: dict, asset: dict) -> dict:
        """Calculate priority score for vulnerability."""
        base_score = vuln.get("cvss_score", 5.0)

        # Adjust for asset criticality
        criticality_multiplier = {
            "critical": 1.5,
            "high": 1.2,
            "medium": 1.0,
            "low": 0.8
        }
        adjusted_score = base_score * criticality_multiplier.get(
            asset.get("criticality", "medium"), 1.0
        )

        # Adjust for exploit availability
        if vuln.get("exploit_available"):
            adjusted_score *= 1.3

        # Adjust for internet exposure
        if asset.get("internet_facing"):
            adjusted_score *= 1.2

        # Cap at 10
        adjusted_score = min(adjusted_score, 10.0)

        severity = self._get_severity(adjusted_score)

        return {
            "priority_score": round(adjusted_score, 1),
            "severity": severity,
            "sla_response_hours": self.SLA[severity]["response"],
            "sla_remediation_hours": self.SLA[severity]["remediation"]
        }

    def _get_severity(self, score: float) -> str:
        if score >= 9.0: return "CRITICAL"
        if score >= 7.0: return "HIGH"
        if score >= 4.0: return "MEDIUM"
        return "LOW"

合规报告

高管仪表板数据

def generate_executive_summary(vulnerabilities: list) -> dict:
    """Generate executive summary metrics."""
    total = len(vulnerabilities)

    by_severity = {
        "CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0
    }

    sla_breaches = 0
    remediated_count = 0
    avg_remediation_time = 0

    for vuln in vulnerabilities:
        by_severity[vuln["severity"]] += 1

        if vuln.get("status") == "remediated":
            remediated_count += 1
            avg_remediation_time += vuln.get("remediation_hours", 0)

        if vuln.get("sla_breached"):
            sla_breaches += 1

    if remediated_count > 0:
        avg_remediation_time /= remediated_count

    return {
        "total_vulnerabilities": total,
        "by_severity": by_severity,
        "sla_compliance_rate": round((1 - sla_breaches / total) * 100, 1) if total > 0 else 100,
        "remediation_rate": round(remediated_count / total * 100, 1) if total > 0 else 0,
        "avg_remediation_hours": round(avg_remediation_time, 1),
        "critical_open": by_severity["CRITICAL"],
        "high_open": by_severity["HIGH"]
    }

合规框架映射

NIST Cybersecurity Framework:
  ID.RA-1: Asset vulnerabilities identified
  ID.RA-2: Threat intelligence received
  PR.IP-12: Vulnerability management plan
  DE.CM-8: Vulnerability scans performed
  RS.MI-3: Vulnerabilities mitigated

ISO 27001:
  A.12.6.1: Technical vulnerability management
  A.18.2.3: Technical compliance review

SOC 2:
  CC7.1: Vulnerability management procedures
  CC7.2: Security event monitoring

PCI DSS:
  6.1: Vulnerability identification process
  6.2: Security patches installation
  11.2: Quarterly vulnerability scans

集成模式

SIEM Integration:
  - Forward high/critical alerts
  - Enrich events with CVE data
  - Automated incident creation

Ticketing Systems:
  - Auto-create tickets for new vulns
  - Track remediation progress
  - SLA monitoring

CI/CD Pipeline:
  - Pre-deployment vulnerability check
  - Block deployments with critical CVEs
  - Automated dependency updates

Notification:
  - Slack: Critical alerts
  - Email: Daily/weekly summaries
  - PagerDuty: SLA breaches

最佳实践

Data Quality:
  - Deduplicate findings
  - Validate against authoritative sources
  - Enrich with threat intelligence
  - Track remediation history

Performance:
  - Incremental updates (not full sync)
  - Batch processing for analysis
  - Cache CVSS calculations
  - Index by CPE for fast matching

Security:
  - API rate limiting
  - Data encryption at rest
  - Role-based access control
  - Audit logging

最佳实践

1. 按风险排序 — 并非所有 CVE 都一样，应考虑上下文
2. 自动化分诊 — 自动执行日常分类工作
3. 跟踪 SLA — 监控响应时间
4. 全面集成 — 与 CI/CD、工单系统和 SIEM 连接
5. 向管理层报告 — 定期向管理层提交报告
6. 持续监控 — 不要只进行定期扫描