Replace Internal Database IDs with Natural Keys in API URLs (VCID and PURL)

vulnerabilities/api.py

@@ -13,6 +13,10 @@
 from cvss.exceptions import CVSS3MalformedError
 from cvss.exceptions import CVSS4MalformedError
 from django.db.models import Prefetch
+from django.db.models import Exists
+from django.db.models import OuterRef
+from django.http import Http404
+from django.shortcuts import get_object_or_404
 from django_filters import rest_framework as filters
 from drf_spectacular.utils import extend_schema
 from packageurl import PackageURL
@@ -22,6 +26,7 @@
 from rest_framework import viewsets
 from rest_framework.decorators import action
 from rest_framework.response import Response
+from rest_framework.reverse import reverse
 from rest_framework.throttling import AnonRateThrottle
 
 from vulnerabilities.models import Alias
@@ -114,7 +119,7 @@ class MinimalPackageSerializer(BaseResourceSerializer):
     """
 
     affected_by_vulnerabilities = VulnVulnIDSerializer(source="affecting_vulns", many=True)
-
+    url = serializers.SerializerMethodField()
     purl = serializers.CharField(source="package_url")
 
     is_vulnerable = serializers.BooleanField()
@@ -123,6 +128,16 @@ class Meta:
         model = Package
         fields = ["url", "purl", "is_vulnerable", "affected_by_vulnerabilities"]
 
+    def get_url(self, package):
+        request = self.context.get("request")
+        if request:
+            return request.build_absolute_uri(
+                reverse("package-detail", kwargs={'package_url': package.package_url})
+            )
+        return None
+
+    def get_qualifiers(self, package):
+        return normalize_qualifiers(package.qualifiers, encode=False)
 
 class MinimalVulnerabilitySerializer(BaseResourceSerializer):
     """
@@ -148,7 +163,16 @@ class VulnSerializerRefsAndSummary(BaseResourceSerializer):
     """
     Lookup vulnerabilities references by aliases (such as a CVE).
     """
-
+    url = serializers.SerializerMethodField()
+
+    def get_url(self, vulnerability):
+        request = self.context.get("request")
+        if request:
+            return request.build_absolute_uri(
+                reverse("vulnerability-detail", kwargs={'vulnerability_id': vulnerability.vulnerability_id})
+            )
+        return None
+
     fixed_packages = MinimalPackageSerializer(
         many=True, source="filtered_fixed_packages", read_only=True
     )
@@ -236,7 +260,16 @@ class VulnerabilitySerializer(BaseResourceSerializer):
     exploits = ExploitSerializer(many=True, read_only=True)
     weaknesses = WeaknessSerializer(many=True)
     severity_range_score = serializers.SerializerMethodField()
+    url = serializers.SerializerMethodField()
 
+    def get_url(self, vulnerability):
+        request = self.context.get("request")
+        if request:
+            return request.build_absolute_uri(
+                reverse("vulnerability-detail", kwargs={'vulnerability_id': vulnerability.vulnerability_id})
+            )
+        return None
+
     def to_representation(self, instance):
         data = super().to_representation(instance)
 
@@ -308,6 +341,8 @@ class PackageSerializer(BaseResourceSerializer):
     next_non_vulnerable_version = serializers.CharField(read_only=True)
     latest_non_vulnerable_version = serializers.CharField(read_only=True)
 
+    url = serializers.SerializerMethodField()
+
     purl = serializers.CharField(source="package_url")
 
     affected_by_vulnerabilities = serializers.SerializerMethodField("get_affected_vulnerabilities")
@@ -318,6 +353,10 @@ class PackageSerializer(BaseResourceSerializer):
 
     is_vulnerable = serializers.BooleanField()
 
+    def get_url(self, package):
+        request = self.context.get("request")
+        return reverse("package-detail", kwargs={'package_url': package.package_url}, request=request)
+
     def get_qualifiers(self, package):
         return normalize_qualifiers(package.qualifiers, encode=False)
 
@@ -373,24 +412,33 @@ def get_fixing_vulnerabilities(self, package) -> dict:
 
         return self.get_vulnerabilities_for_a_package(package=package, fix=True)
 
-    def get_affected_vulnerabilities(self, package) -> dict:
+    def get_affected_vulnerabilities(self, package) -> list:
         """
-        Return a mapping of vulnerabilities that affect the given `package` (including packages that
-        fix each vulnerability and whose version is greater than the `package` version).
+        Return vulnerabilities that affect the given `package`.
         """
         excluded_purls = []
         package_vulnerabilities = self.get_vulnerabilities_for_a_package(package=package, fix=False)
-
+
+        request = self.context.get("request")
+
+        # Process vulnerabilities and add the URL manually
         for vuln in package_vulnerabilities:
-            for pkg in vuln["fixed_packages"]:
+            if request and "vulnerability_id" in vuln:
+                vuln_id = vuln["vulnerability_id"]
+                vuln["url"] = request.build_absolute_uri(
+                    reverse("vulnerability-detail", kwargs={"vulnerability_id": vuln_id})
+                )
+
+            # Process fixed packages as before
+            for pkg in vuln.get("fixed_packages", []):
                 real_purl = PackageURL.from_string(pkg["purl"])
                 if package.version_class(real_purl.version) <= package.current_version:
                     excluded_purls.append(pkg)
-
+                    
             vuln["fixed_packages"] = [
                 pkg for pkg in vuln["fixed_packages"] if pkg not in excluded_purls
             ]
-
+            
         return package_vulnerabilities
 
     class Meta:
@@ -469,6 +517,7 @@ class PackageViewSet(viewsets.ReadOnlyModelViewSet):
 
     queryset = Package.objects.all()
     serializer_class = PackageSerializer
+    lookup_field = "package_url"
     filter_backends = (filters.DjangoFilterBackend,)
     filterset_class = PackageFilterSet
     throttle_classes = [StaffUserRateThrottle, AnonRateThrottle]
@@ -689,6 +738,7 @@ def get_queryset(self):
     filter_backends = (filters.DjangoFilterBackend,)
     filterset_class = VulnerabilityFilterSet
     throttle_classes = [StaffUserRateThrottle, AnonRateThrottle]
+    lookup_field = "vulnerability_id"
 
 
 class CPEFilterSet(filters.FilterSet):

vulnerabilities/importers/apache_httpd.py

@@ -8,6 +8,7 @@
 #
 
 import logging
+import re
 import urllib
 
 import requests
@@ -23,6 +24,8 @@
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.severity_systems import APACHE_HTTPD
+from vulnerabilities.utils import create_weaknesses_list
+from vulnerabilities.utils import cwe_regex
 from vulnerabilities.utils import get_item
 
 logger = logging.getLogger(__name__)
@@ -102,11 +105,14 @@ def to_advisory(self, data):
                 )
             )
 
+        weaknesses = get_weaknesses(data)
+
         return AdvisoryData(
             aliases=[alias],
             summary=description or "",
             affected_packages=affected_packages,
             references=[reference],
+            weaknesses=weaknesses,
             url=reference.url,
         )
 
@@ -152,3 +158,97 @@ def fetch_links(url):
             continue
         links.append(urllib.parse.urljoin(url, link))
     return links
+
+
+def get_weaknesses(cve_data):
+    """
+    Extract CWE IDs from CVE data.
+
+    Args:
+        cve_data (dict): The CVE data in a dictionary format.
+
+    Returns:
+        List[int]: A list of unique CWE IDs.
+
+    Examples:
+        >>> mock_cve_data1 = {
+        ...     "containers": {
+        ...         "cna": {
+        ...             "providerMetadata": {
+        ...                 "orgId": "f0158376-9dc2-43b6-827c-5f631a4d8d09"
+        ...             },
+        ...             "title": "mod_macro buffer over-read",
+        ...             "problemTypes": [
+        ...                 {
+        ...                     "descriptions": [
+        ...                         {
+        ...                             "description": "CWE-125 Out-of-bounds Read",
+        ...                             "lang": "en",
+        ...                             "cweId": "CWE-125",
+        ...                             "type": "CWE"
+        ...                         }
+        ...                     ]
+        ...                 }
+        ...             ]
+        ...         }
+        ...     }
+        ... }
+        >>> mock_cve_data2 = {
+        ...     "data_type": "CVE",
+        ...     "data_format": "MITRE",
+        ...     "data_version": "4.0",
+        ...     "generator": {
+        ...         "engine": "Vulnogram 0.0.9"
+        ...     },
+        ...     "CVE_data_meta": {
+        ...         "ID": "CVE-2022-28614",
+        ...         "ASSIGNER": "[email protected]",
+        ...         "TITLE": "read beyond bounds via ap_rwrite() ",
+        ...         "STATE": "PUBLIC"
+        ...     },
+        ...     "problemtype": {
+        ...         "problemtype_data": [
+        ...             {
+        ...                 "description": [
+        ...                     {
+        ...                         "lang": "eng",
+        ...                         "value": "CWE-190 Integer Overflow or Wraparound"
+        ...                     }
+        ...                 ]
+        ...             },
+        ...             {
+        ...                 "description": [
+        ...                     {
+        ...                         "lang": "eng",
+        ...                         "value": "CWE-200 Exposure of Sensitive Information to an Unauthorized Actor"
+        ...                     }
+        ...                 ]
+        ...             }
+        ...         ]
+        ...     }
+        ... }
+
+        >>> get_weaknesses(mock_cve_data1)
+        [125]
+
+        >>> get_weaknesses(mock_cve_data2)
+        [190, 200]
+    """
+    alias = get_item(cve_data, "CVE_data_meta", "ID")
+    cwe_strings = []
+    if alias:
+        problemtype_data = get_item(cve_data, "problemtype", "problemtype_data") or []
+        for problem in problemtype_data:
+            for desc in problem.get("description", []):
+                value = desc.get("value", "")
+                cwe_id_string_list = re.findall(cwe_regex, value)
+                cwe_strings.extend(cwe_id_string_list)
+    else:
+        problemTypes = cve_data.get("containers", {}).get("cna", {}).get("problemTypes", [])
+        descriptions = problemTypes[0].get("descriptions", []) if len(problemTypes) > 0 else []
+        for description in descriptions:
+            cwe_id_string = description.get("cweId", "")
+            cwe_strings.append(cwe_id_string)
+
+    weaknesses = create_weaknesses_list(cwe_strings)
+    return weaknesses

vulnerabilities/importers/debian.py

@@ -8,12 +8,14 @@
 #
 
 import logging
+import re
 from typing import Any
 from typing import Iterable
 from typing import List
 from typing import Mapping
 
 import requests
+from cwe2.database import Database
 from packageurl import PackageURL
 from univers.version_range import DebianVersionRange
 from univers.versions import DebianVersion
@@ -22,6 +24,7 @@
 from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
+from vulnerabilities.utils import create_weaknesses_list
 from vulnerabilities.utils import dedupe
 from vulnerabilities.utils import get_item
 
@@ -93,6 +96,7 @@ def advisory_data(self) -> Iterable[AdvisoryData]:
             yield from self.parse(pkg_name, records)
 
     def parse(self, pkg_name: str, records: Mapping[str, Any]) -> Iterable[AdvisoryData]:
+
         for cve_id, record in records.items():
             affected_versions = []
             fixed_versions = []
@@ -150,10 +154,29 @@ def parse(self, pkg_name: str, records: Mapping[str, Any]) -> Iterable[AdvisoryD
                         fixed_version=DebianVersion(fixed_version),
                     )
                 )
+            weaknesses = get_cwe_from_debian_advisory(record)
+
             yield AdvisoryData(
                 aliases=[cve_id],
                 summary=record.get("description", ""),
                 affected_packages=affected_packages,
                 references=references,
+                weaknesses=weaknesses,
                 url=self.api_url,
             )
+
+
+def get_cwe_from_debian_advisory(record):
+    """
+    Extracts CWE ID strings from the given raw_data and returns a list of CWE IDs.
+
+        >>> get_cwe_from_debian_advisory({"description":"PEAR HTML_QuickForm version 3.2.14 contains an eval injection (CWE-95) vulnerability in HTML_QuickForm's getSubmitValue method, HTML_QuickForm's validate method, HTML_QuickForm_hierselect's _setOptions method, HTML_QuickForm_element's _findValue method, HTML_QuickForm_element's _prepareValue method. that can result in Possible information disclosure, possible impact on data integrity and execution of arbitrary code. This attack appear to be exploitable via A specially crafted query string could be utilised, e.g. http://www.example.com/admin/add_practice_type_id[1]=fubar%27])%20OR%20die(%27OOK!%27);%20//&mode=live. This vulnerability appears to have been fixed in 3.2.15."})
+        [95]
+        >>> get_cwe_from_debian_advisory({"description":"There is no WEAKNESS DATA"})
+        []
+    """
+    description = record.get("description") or ""
+    pattern = r"CWE-\d+"
+    cwe_strings = re.findall(pattern, description)
+    weaknesses = create_weaknesses_list(cwe_strings)
+    return weaknesses

vulnerabilities/importers/fireeye.py

@@ -16,6 +16,8 @@
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
 from vulnerabilities.utils import build_description
+from vulnerabilities.utils import create_weaknesses_list
+from vulnerabilities.utils import cwe_regex
 from vulnerabilities.utils import dedupe
 
 logger = logging.getLogger(__name__)
@@ -77,10 +79,13 @@ def parse_advisory_data(raw_data, file, base_path) -> AdvisoryData:
     disc_credits = md_dict.get("## Discovery Credits")  # not used
     disc_timeline = md_dict.get("## Disclosure Timeline")  # not used
     references = md_dict.get("## References") or []
+    cwe_data = md_dict.get("## Common Weakness Enumeration") or []
+
     return AdvisoryData(
         aliases=get_aliases(database_id, cve_ref),
         summary=build_description(" ".join(summary), " ".join(description)),
         references=get_references(references),
+        weaknesses=get_weaknesses(cwe_data),
         url=advisory_url,
     )
 
@@ -140,3 +145,22 @@ def md_list_to_dict(md_list):
         else:
             md_dict[md_key].append(md_line)
     return md_dict
+
+
+def get_weaknesses(cwe_data):
+    """
+    Return the list of CWE IDs as integers from a list of weakness summaries, e.g., [379].
+
+        >>> get_weaknesses([
+        ... "CWE-379: Creation of Temporary File in Directory with Insecure Permissions",
+        ... "CWE-362: Concurrent Execution using Shared Resource with Improper Synchronization ('Race Condition')"
+        ... ])
+        [379, 362]
+    """
+    cwe_list = []
+    for line in cwe_data:
+        cwe_ids = re.findall(cwe_regex, line)
+        cwe_list.extend(cwe_ids)
+
+    weaknesses = create_weaknesses_list(cwe_list)
+    return weaknesses