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. 継続的に監視する — 定期スキャンだけにとどめないでください