1
2
3
4
5
6
7

graph LR
    A[취약점 발견] --> B[CVE 번호 할당]
    B --> C[NVD 분석 대기]
    C --> D[CVE 상세 분석]
    D --> E[CVSS 점수 산정]
    E --> F[NVD 데이터베이스 등록]
    F --> G[보안 도구 업데이트]

1
2
3
4
5
6
7
8

graph TD
    A[CVE 접수] --> B{자동화된 1차 평가}
    B -->|고위험| C[즉시 심층 분석]
    B -->|중위험| D[표준 분석 큐]
    B -->|저위험| E[최소 분석 또는 보류]
    C --> F[24시간 내 CVSS 산정]
    D --> G[30일 내 CVSS 산정]
    E --> H[필요시 재평가]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

import requests
from datetime import datetime, timedelta

class CVETriage:
    """NVD 처리 지연 대응을 위한 자체 위험 평가 클래스"""
    
    def __init__(self, cve_id):
        self.cve_id = cve_id
        self.cve_data = self._fetch_cve_data()
        
    def _fetch_cve_data(self):
        """NVD API에서 CVE 데이터 조회"""
        url = f"https://services.nvd.nist.gov/rest/json/cves/2.0?cveId={self.cve_id}"
        headers = {"apiKey": "YOUR_API_KEY"}
        
        try:
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"[ERROR] API 조회 실패: {e}")
            return None
    
    def assess_risk_level(self):
        """다양한 요소를 종합한 위험도 평가"""
        if not self.cve_data:
            return "UNKNOWN", "데이터 조회 실패"
        
        risk_score = 0
        reasons = []
        
        # 1. CVSS 점수 확인 (NVD에서 이미 산정한 경우)
        metrics = self.cve_data.get("vulnerabilities", [{}])[0].get("cve", {}).get("metrics", {})
        
        if "cvssMetricV31" in metrics:
            cvss_score = metrics["cvssMetricV31"][0]["cvssData"]["baseScore"]
            if cvss_score >= 9.0:
                risk_score += 40
                reasons.append(f"Critical CVSS: {cvss_score}")
            elif cvss_score >= 7.0:
                risk_score += 25
                reasons.append(f"High CVSS: {cvss_score}")
        else:
            # CVSS 점수가 없으면 공지된 심각도 확인
            risk_score += 15
            reasons.append("CVSS 미산정 - 수동 평가 필요")
        
        # 2. 공개된 지 얼마나 되었는지 (최근일수록 위험)
        published = self.cve_data.get("vulnerabilities", [{}])[0].get("cve", {}).get("published", "")
        if published:
            pub_date = datetime.fromisoformat(published.replace("Z", "+00:00"))
            days_old = (datetime.now(pub_date.tzinfo) - pub_date).days
            
            if days_old <= 7:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

                risk_score += 20
                reasons.append(f"최근 공개: {days_old}일 전")
            elif days_old <= 30:
                risk_score += 10
        
        # 3. 영향받는 제품 수 (CPE 정보 기반)
        configurations = self.cve_data.get("vulnerabilities", [{}])[0].get("cve", {}).get("configurations", [])
        affected_products = set()
        
        for config in configurations:
            for node in config.get("nodes", []):
                for cpe in node.get("cpeMatch", []):
                    criteria = cpe.get("criteria", "")
                    if "apache" in criteria.lower() or "nginx" in criteria.lower():
                        risk_score += 15
                        reasons.append("주요 인프라 소프트웨어 영향")
                        break
        
        # 4. 익스플로잇 가능성 지표 (참고 정보)
        references = self.cve_data.get("vulnerabilities", [{}])[0].get("cve", {}).get("references", [])
        for ref in references:
            url = ref.get("url", "").lower()
            if "exploit" in url or "poc" in url or "github.com" in url:
                risk_score += 25
                reasons.append("공개 익스플로잇/POC 확인됨")
                break
        
        # 위험 등급 산정
        if risk_score >= 70:
            return "CRITICAL", "; ".join(reasons)
        elif risk_score >= 40:
            return "HIGH", "; ".join(reasons)
        elif risk_score >= 20:
            return "MEDIUM", "; ".join(reasons)
        else:
            return "LOW", "; ".join(reasons)

# 실행 예시
if __name__ == "__main__":
    triage = CVETriage("CVE-2024-3094")  # XZ Utils 백도어 취약점
    level, detail = triage.assess_risk_level()
    print(f"[{level}] {triage.cve_id}")
    print(f"평가 근거: {detail}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 다중 소스 취약점 정보 통합 예시
SOURCES = {
    "nvd": "https://services.nvd.nist.gov/rest/json/cves/2.0",
    "github": "https://api.github.com/advisories",
    "cisa_kev": "https://www.cisa.gov/known-exploited-vulnerabilities-catalog"
}

def aggregate_vulnerability_data(cve_id):
    """다중 소스에서 취약점 정보를 취합하는 함수"""
    aggregated = {"cve_id": cve_id, "sources": {}}
    
    for source_name, base_url in SOURCES.items():
        try:
            data = fetch_from_source(source_name, base_url, cve_id)
            aggregated["sources"][source_name] = data
        except Exception:
            aggregated["sources"][source_name] = {"status": "unavailable"}
    
    # 가장 높은 위험 평가를 기준으로 채택
    return aggregated