Resolve merge conflict

Signed-off-by: ziad hany <ziadhany2016@gmail.com>

Author: ziadhany

vulnerabilities/importers/__init__.py

@@ -47,6 +47,7 @@
 from vulnerabilities.pipelines.v2_importers import archlinux_importer as archlinux_importer_v2
 from vulnerabilities.pipelines.v2_importers import collect_fix_commits as collect_fix_commits_v2
 from vulnerabilities.pipelines.v2_importers import curl_importer as curl_importer_v2
+from vulnerabilities.pipelines.v2_importers import cvelistv5_importer as cvelistv5_importer_v2
 from vulnerabilities.pipelines.v2_importers import debian_importer as debian_importer_v2
 from vulnerabilities.pipelines.v2_importers import (
     elixir_security_importer as elixir_security_importer_v2,
@@ -92,6 +93,7 @@
         elixir_security_importer_v2.ElixirSecurityImporterPipeline,
         npm_importer_v2.NpmImporterPipeline,
         vulnrichment_importer_v2.VulnrichImporterPipeline,
+        cvelistv5_importer_v2.CVEListV5ImporterPipeline,
         apache_httpd_v2.ApacheHTTPDImporterPipeline,
         pypa_importer_v2.PyPaImporterPipeline,
         gitlab_importer_v2.GitLabImporterPipeline,

vulnerabilities/importers/cve_schema.py …ies/pipelines/v2_importers/cve_schema.pyvulnerabilities/importers/cve_schema.py renamed to vulnerabilities/pipelines/v2_importers/cve_schema.py vulnerabilities/importers/cve_schema.py renamed to vulnerabilities/pipelines/v2_importers/cve_schema.py

@@ -11,7 +11,7 @@
 
 import dateparser
 
-from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AdvisoryDataV2
 from vulnerabilities.importer import ReferenceV2
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.models import VulnerabilityReference
@@ -21,9 +21,10 @@
 from vulnerabilities.utils import ssvc_calculator
 
 
-def parse_cve_v5_advisory(raw_data, advisory_url):
+def parse_cve_advisory(raw_data, advisory_url):
     cve_metadata = raw_data.get("cveMetadata", {})
     cve_id = cve_metadata.get("cveId")
+    state = cve_metadata.get("state")
 
     date_published = cve_metadata.get("datePublished")
     if date_published:
@@ -55,7 +56,7 @@ def parse_cve_v5_advisory(raw_data, advisory_url):
         adp_metrics for data in adp_data for adp_metrics in data.get("metrics", [])
     ]
 
-    cve_scoring_system = {
+    vulnrichment_scoring_system = {
         "cvssV4_0": SCORING_SYSTEMS["cvssv4"],
         "cvssV3_1": SCORING_SYSTEMS["cvssv3.1"],
         "cvssV3_0": SCORING_SYSTEMS["cvssv3"],
@@ -67,15 +68,15 @@ def parse_cve_v5_advisory(raw_data, advisory_url):
 
     for metric in metrics:
         for metric_type, metric_value in metric.items():
-            if metric_type not in cve_scoring_system:
+            if metric_type not in vulnrichment_scoring_system:
                 continue
 
             if metric_type == "other":
                 other_types = metric_value.get("type")
                 if other_types == "ssvc":
                     content = metric_value.get("content", {})
                     vector_string, decision = ssvc_calculator(content)
-                    scoring_system = cve_scoring_system[metric_type][other_types]
+                    scoring_system = vulnrichment_scoring_system[metric_type][other_types]
                     severity = VulnerabilitySeverity(
                         system=scoring_system, value=decision, scoring_elements=vector_string
                     )
@@ -84,7 +85,7 @@ def parse_cve_v5_advisory(raw_data, advisory_url):
             else:
                 vector_string = metric_value.get("vectorString")
                 base_score = metric_value.get("baseScore")
-                scoring_system = cve_scoring_system[metric_type]
+                scoring_system = vulnrichment_scoring_system[metric_type]
                 severity = VulnerabilitySeverity(
                     system=scoring_system, value=base_score, scoring_elements=vector_string
                 )
@@ -149,11 +150,11 @@ def parse_cve_v5_advisory(raw_data, advisory_url):
                 if match:
                     weaknesses.add(int(match.group(1)))
 
-    return AdvisoryData(
+    return AdvisoryDataV2(
         advisory_id=cve_id,
         aliases=[],
         summary=summary,
-        references_v2=references,
+        references=references,
         date_published=date_published,
         weaknesses=sorted(weaknesses),
         url=advisory_url,

vulnerabilities/pipelines/v2_importers/cvelistv5_importer.py

@@ -14,8 +14,8 @@
 from fetchcode.vcs import fetch_via_vcs
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importers.cve_schema import parse_cve_v5_advisory
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
+from vulnerabilities.pipelines.v2_importers.cve_schema import parse_cve_advisory
 from vulnerabilities.utils import get_advisory_url
 
 logger = logging.getLogger(__name__)
@@ -60,7 +60,7 @@ def collect_advisories(self) -> Iterable[AdvisoryData]:
 
             with open(file) as f:
                 raw_data = json.load(f)
-            yield parse_cve_v5_advisory(raw_data, advisory_url)
+            yield parse_cve_advisory(raw_data, advisory_url)
 
     def clean_downloads(self):
         if self.vcs_response:

vulnerabilities/pipelines/v2_importers/vulnrichment_importer.py

@@ -1,22 +1,14 @@
 import json
 import logging
-import re
 from pathlib import Path
 from typing import Iterable
 
-import dateparser
 from fetchcode.vcs import fetch_via_vcs
 
 from vulnerabilities.importer import AdvisoryDataV2
-from vulnerabilities.importer import ReferenceV2
-from vulnerabilities.importer import VulnerabilitySeverity
-from vulnerabilities.models import VulnerabilityReference
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
-from vulnerabilities.severity_systems import SCORING_SYSTEMS
+from vulnerabilities.pipelines.v2_importers.cve_schema import parse_cve_advisory
 from vulnerabilities.utils import get_advisory_url
-from vulnerabilities.utils import get_cwe_id
-from vulnerabilities.utils import get_reference_id
-from vulnerabilities.utils import ssvc_calculator
 
 logger = logging.getLogger(__name__)
 
@@ -63,148 +55,7 @@ def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
                 base_path=base_path,
                 url="https://github.com/cisagov/vulnrichment/blob/develop/",
             )
-            yield self.parse_cve_advisory(raw_data, advisory_url)
-
-    def parse_cve_advisory(self, raw_data, advisory_url):
-        cve_metadata = raw_data.get("cveMetadata", {})
-        cve_id = cve_metadata.get("cveId")
-        state = cve_metadata.get("state")
-
-        date_published = cve_metadata.get("datePublished")
-        if date_published:
-            date_published = dateparser.parse(
-                date_published,
-                settings={
-                    "TIMEZONE": "UTC",
-                    "RETURN_AS_TIMEZONE_AWARE": True,
-                    "TO_TIMEZONE": "UTC",
-                },
-            )
-
-        # Extract containers
-        containers = raw_data.get("containers", {})
-        cna_data = containers.get("cna", {})
-        adp_data = containers.get("adp", {})
-
-        # Extract descriptions
-        summary = ""
-        description_list = cna_data.get("descriptions", [])
-        for description_dict in description_list:
-            if not description_dict.get("lang") in ["en", "en-US"]:
-                continue
-            summary = description_dict.get("value")
-
-        # Extract metrics
-        severities = []
-        metrics = cna_data.get("metrics", []) + [
-            adp_metrics for data in adp_data for adp_metrics in data.get("metrics", [])
-        ]
-
-        vulnrichment_scoring_system = {
-            "cvssV4_0": SCORING_SYSTEMS["cvssv4"],
-            "cvssV3_1": SCORING_SYSTEMS["cvssv3.1"],
-            "cvssV3_0": SCORING_SYSTEMS["cvssv3"],
-            "cvssV2_0": SCORING_SYSTEMS["cvssv2"],
-            "other": {
-                "ssvc": SCORING_SYSTEMS["ssvc"],
-            },  # ignore kev
-        }
-
-        for metric in metrics:
-            for metric_type, metric_value in metric.items():
-                if metric_type not in vulnrichment_scoring_system:
-                    continue
-
-                if metric_type == "other":
-                    other_types = metric_value.get("type")
-                    if other_types == "ssvc":
-                        content = metric_value.get("content", {})
-                        vector_string, decision = ssvc_calculator(content)
-                        scoring_system = vulnrichment_scoring_system[metric_type][other_types]
-                        severity = VulnerabilitySeverity(
-                            system=scoring_system, value=decision, scoring_elements=vector_string
-                        )
-                        severities.append(severity)
-                    # ignore kev
-                else:
-                    vector_string = metric_value.get("vectorString")
-                    base_score = metric_value.get("baseScore")
-                    scoring_system = vulnrichment_scoring_system[metric_type]
-                    severity = VulnerabilitySeverity(
-                        system=scoring_system, value=base_score, scoring_elements=vector_string
-                    )
-                    severities.append(severity)
-
-        # Extract references cpes and ignore affected products
-        cpes = set()
-        for affected_product in cna_data.get("affected", []):
-            if type(affected_product) != dict:
-                continue
-            cpes.update(affected_product.get("cpes") or [])
-
-        references = []
-        for ref in cna_data.get("references", []):
-            # https://github.com/CVEProject/cve-schema/blob/main/schema/tags/reference-tags.json
-            # We removed all unwanted reference types and set the default reference type to 'OTHER'.
-            ref_type = VulnerabilityReference.OTHER
-            vul_ref_types = {
-                "exploit": VulnerabilityReference.EXPLOIT,
-                "issue-tracking": VulnerabilityReference.BUG,
-                "mailing-list": VulnerabilityReference.MAILING_LIST,
-                "third-party-advisory": VulnerabilityReference.ADVISORY,
-                "vendor-advisory": VulnerabilityReference.ADVISORY,
-                "vdb-entry": VulnerabilityReference.ADVISORY,
-            }
-
-            for tag_type in ref.get("tags", []):
-                if tag_type in vul_ref_types:
-                    ref_type = vul_ref_types.get(tag_type)
-
-            url = ref.get("url")
-            reference = ReferenceV2(
-                reference_id=get_reference_id(url),
-                url=url,
-                reference_type=ref_type,
-            )
-
-            references.append(reference)
-
-        cpes_ref = [
-            ReferenceV2(
-                reference_id=cpe,
-                reference_type=VulnerabilityReference.OTHER,
-                url=f"https://nvd.nist.gov/vuln/search/results?adv_search=true&isCpeNameSearch=true&query={cpe}",
-            )
-            for cpe in sorted(list(cpes))
-        ]
-        references.extend(cpes_ref)
-
-        weaknesses = set()
-        for problem_type in cna_data.get("problemTypes", []):
-            descriptions = problem_type.get("descriptions", [])
-            for description in descriptions:
-                cwe_id = description.get("cweId")
-                if cwe_id:
-                    weaknesses.add(get_cwe_id(cwe_id))
-
-                description_text = description.get("description")
-                if description_text:
-                    pattern = r"CWE-(\d+)"
-                    match = re.search(pattern, description_text)
-                    if match:
-                        weaknesses.add(int(match.group(1)))
-
-        return AdvisoryDataV2(
-            advisory_id=cve_id,
-            aliases=[],
-            summary=summary,
-            references=references,
-            date_published=date_published,
-            weaknesses=sorted(weaknesses),
-            url=advisory_url,
-            severities=severities,
-            original_advisory_text=json.dumps(raw_data, indent=2, ensure_ascii=False),
-        )
+            yield parse_cve_advisory(raw_data, advisory_url)
 
     def clean_downloads(self):
         if self.vcs_response:

vulnerabilities/tests/pipelines/v2_importers/test_cvelistv5_importer_v2.py

@@ -13,10 +13,12 @@
 
 import pytest
 
-from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AdvisoryDataV2
+from vulnerabilities.importer import ReferenceV2
 from vulnerabilities.importer import VulnerabilitySeverity
-from vulnerabilities.importers.cve_schema import parse_cve_v5_advisory
+from vulnerabilities.pipelines.v2_importers.cve_schema import parse_cve_advisory
 from vulnerabilities.pipelines.v2_importers.cvelistv5_importer import CVEListV5ImporterPipeline
+from vulnerabilities.severity_systems import Cvssv4ScoringSystem
 
 
 @pytest.fixture
@@ -98,17 +100,22 @@ def test_collect_advisories(mock_pathlib, mock_vcs_response, mock_fetch_via_vcs,
     with patch(
         "vulnerabilities.pipelines.v2_importers.cvelistv5_importer.CVEListV5ImporterPipeline"
     ) as mock_parse:
-        mock_parse.return_value = AdvisoryData(
+        mock_parse.return_value = AdvisoryDataV2(
             advisory_id="CVE-2021-1234",
             summary="Sample PyPI vulnerability",
-            references_v2=[{"url": "https://example.com"}],
+            references=[ReferenceV2(url="https://example.com")],
             affected_packages=[],
             weaknesses=[],
             url="https://github.com/CVEProject/cvelistV5/blob/cves/2021/1xxx/CVE-2021-1234.json",
             severities=[
                 VulnerabilitySeverity(
-                    system="cvssv4",
-                    value=7.5,
+                    system=Cvssv4ScoringSystem(
+                        identifier="cvssv4",
+                        name="CVSSv4 Base Score",
+                        url="https://www.first.org/cvss/v4-0/",
+                        notes="CVSSv4 base score and vector",
+                    ),
+                    value="7.5",
                     scoring_elements="AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H",
                 )
             ],
@@ -118,7 +125,6 @@ def test_collect_advisories(mock_pathlib, mock_vcs_response, mock_fetch_via_vcs,
         pipeline.clone()
         advisories = list(pipeline.collect_advisories())
 
-        # Ensure that advisories are parsed correctly
         assert len(advisories) == 1
         advisory = advisories[0]
         assert advisory.advisory_id == "CVE-2021-1234"
@@ -176,13 +182,13 @@ def test_parse_cve_advisory(mock_pathlib, mock_vcs_response, mock_fetch_via_vcs)
 
     pipeline = CVEListV5ImporterPipeline()
     pipeline.clone()
-    advisory = parse_cve_v5_advisory(raw_data, advisory_url)
+    advisory = parse_cve_advisory(raw_data, advisory_url)
 
     assert advisory.advisory_id == "CVE-2021-1234"
     assert advisory.summary == "Sample PyPI vulnerability"
     assert advisory.url == advisory_url
     assert len(advisory.severities) == 1
-    assert advisory.severities[0].value == 7.5
+    assert advisory.severities[0].value == "7.5"
 
 
 def test_collect_advisories_with_invalid_json(