Add support for ref cpes & fix the tests

Ignore affected packages

Signed-off-by: ziadhany <[email protected]>

Author: ziadhany

vulnerabilities/importers/vulnrichment.py

@@ -1,56 +1,21 @@
 import json
 import logging
 import re
-from datetime import datetime
 from pathlib import Path
 from typing import Iterable
-from typing import List
-from typing import Optional
 
 import dateparser
-from packageurl import PackageURL
-from univers.version_constraint import VersionConstraint
-from univers.version_range import RANGE_CLASS_BY_SCHEMES
-from univers.version_range import VersionRange
-from univers.versions import AlpineLinuxVersion
-from univers.versions import ArchLinuxVersion
-from univers.versions import ComposerVersion
-from univers.versions import DebianVersion
-from univers.versions import GenericVersion
-from univers.versions import GentooVersion
-from univers.versions import GolangVersion
-from univers.versions import InvalidVersion
-from univers.versions import LegacyOpensslVersion
-from univers.versions import MavenVersion
-from univers.versions import NginxVersion
-from univers.versions import NugetVersion
-from univers.versions import OpensslVersion
-from univers.versions import PypiVersion
-from univers.versions import RpmVersion
-from univers.versions import SemverVersion
-from univers.versions import Version
 
 from vulnerabilities.importer import AdvisoryData
-from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
-from vulnerabilities.utils import build_description
-from vulnerabilities.utils import dedupe
 from vulnerabilities.utils import get_advisory_url
 from vulnerabilities.utils import get_cwe_id
 
 logger = logging.getLogger(__name__)
 
-VULNRICH_VERSION_CLASS_SCHEMES = {
-    "semver": SemverVersion,
-    "python": PypiVersion,
-    "custom": GenericVersion,
-    "rpm": RpmVersion,
-    "maven": MavenVersion,
-}
-
 
 class VulnrichImporter(Importer):
     spdx_license_expression = "CC0-1.0"
@@ -88,76 +53,14 @@ def parse_cve_advisory(raw_data, advisory_url):
     state = cve_metadata.get("state")
 
     date_published = cve_metadata.get("datePublished")
-    date_published = dateparser.parse(date_published)
+    if date_published:
+        date_published = dateparser.parse(date_published)
 
     # Extract containers
     containers = raw_data.get("containers", {})
     cna_data = containers.get("cna", {})
     adp_data = containers.get("adp", {})
 
-    # Extract affected products
-    affected_packages = []
-    for affected_product in cna_data.get("affected", []):
-        if type(affected_product) != dict:
-            continue
-        cpes = affected_product.get("cpes")  # TODO Add references cpes
-
-        vendor = affected_product.get("vendor") or ""
-        collection_url = affected_product.get("collectionURL") or ""
-        product = affected_product.get("product") or ""
-        package_name = affected_product.get("packageName") or ""
-
-        platforms = affected_product.get("platforms", [])
-        default_status = affected_product.get("defaultStatus")
-
-        affected_packages = []
-        # purl (vendor, collection_url, product, package_name, platforms)
-        purl = PackageURL(
-            type=vendor,
-            name=product,
-            namespace=package_name,
-        )
-
-        versions = affected_product.get("versions", [])
-        for version_data in versions:
-            # version ≤ V ≤ (lessThanOrEqual/lessThan)
-            # right_version ≤ V ≤ left_version
-            version_constraints = []
-            r_version = version_data.get("version")
-            version_type = version_data.get("versionType")
-            version_class = VULNRICH_VERSION_CLASS_SCHEMES.get(version_type)
-            if not version_class:
-                logger.error(f"Invalid version_class type: {version_type}")
-                continue
-
-            l_version, l_comparator = None, ""
-            if "lessThan" in version_data:
-                l_version = version_data.get("lessThan")
-                l_comparator = "<"
-            elif "lessThanOrEqual" in version_data:
-                l_version = version_data.get("lessThanOrEqual")
-                l_comparator = "<="
-            try:
-                if l_version and l_comparator:
-                    version_constraints.append(
-                        VersionConstraint(comparator=l_comparator, version=version_class(l_version))
-                    )
-                if r_version:
-                    version_constraints.append(
-                        VersionConstraint(comparator=">", version=version_class(r_version))
-                    )
-            except InvalidVersion:
-                logger.error(f"InvalidVersion: {l_version}-{r_version}")
-                continue
-
-            affected_packages.append(
-                AffectedPackage(
-                    purl,
-                    affected_version_range=VersionRange(constraints=version_constraints),
-                )
-            )
-            status = version_data.get("status")
-
     # Extract descriptions
     summary = ""
     description_list = cna_data.get("descriptions", [])
@@ -204,35 +107,49 @@ def parse_cve_advisory(raw_data, advisory_url):
                 )
                 severities.append(severity)
 
-    # Extract references
+    # 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 [])  # TODO Add references cpes
+
     # TODO ADD reference type
     references = [
         Reference(url=ref.get("url"), severities=severities)
         for ref in cna_data.get("references", [])
     ]
 
-    weaknesses = []
+    cpes_ref = [
+        Reference(
+            reference_id=cpe,
+            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.append(get_cwe_id(cwe_id))
+                weaknesses.add(get_cwe_id(cwe_id))
 
             description_text = description.get("description")
             if description_text:
-                pattern = r"CWE-(\d{3})"
+                pattern = r"CWE-(\d+)"
                 match = re.search(pattern, description_text)
                 if match:
-                    weaknesses.append(match.group(1))
+                    weaknesses.add(int(match.group(1)))
 
     return AdvisoryData(
         aliases=[cve_id],
         summary=summary,
-        affected_packages=affected_packages,
         references=references,
-        # date_published=dateparser.parse(self.cve_item.get("publishedDate")),
-        weaknesses=weaknesses,
+        date_published=date_published,
+        weaknesses=list(weaknesses),
         url=advisory_url,
     )
 

vulnerabilities/tests/test_data/vulnrichment/vulnrichment-data1-expected.json

@@ -38,7 +38,7 @@
       ]
     }
   ],
-  "date_published": null,
+  "date_published": "2024-03-30T11:17:25.675000+00:00",
   "weaknesses": [
     "502"
   ],