Merge branch 'main' into add-almalinux-advisories

Author: ambuj-1211

CHANGELOG.rst

@@ -1,6 +1,20 @@
 Release notes
 =============
 
+Version v34.0.0rc5
+-------------------
+
+- Add safetydb importer.
+- Add missing width setting for the table in the vulnerability details UI.
+- Add KEV support.
+- Add UI template for API.
+- Use VersionRange.normalize to compare advisory.
+- Use integer column to display score.
+- Add support for CVSSv4 & SSVC and import the data using vulnrichment.
+- Add support for reference_type in the API.
+- Add API improvements for the package endpoint.
+
+
 Version v34.0.0rc4
 -------------------
 

requirements.txt

@@ -106,7 +106,7 @@ toml==0.10.2
 tomli==2.0.1
 traitlets==5.1.1
 typing_extensions==4.1.1
-univers==30.11.0
+univers==30.12.0
 urllib3==1.26.19
 wcwidth==0.2.5
 websocket-client==0.59.0

setup.cfg

@@ -71,7 +71,7 @@ install_requires =
 
     #essentials
     packageurl-python>=0.10.5rc1
-    univers>=30.11.0
+    univers>=30.12.0
     license-expression>=21.6.14
 
     # file and data formats
@@ -118,6 +118,7 @@ dev =
     commoncode
     # debug
     django-debug-toolbar
+    pyinstrument
 
 [options.entry_points]
 console_scripts =

vulnerabilities/api.py

@@ -38,7 +38,14 @@
 class VulnerabilitySeveritySerializer(serializers.ModelSerializer):
     class Meta:
         model = VulnerabilitySeverity
-        fields = ["value", "scoring_system", "scoring_elements"]
+        fields = ["value", "scoring_system", "scoring_elements", "published_at"]
+
+    def to_representation(self, instance):
+        data = super().to_representation(instance)
+        published_at = data.get("published_at", None)
+        if not published_at:
+            data.pop("published_at")
+        return data
 
 
 class VulnerabilityReferenceSerializer(serializers.ModelSerializer):
@@ -47,7 +54,7 @@ class VulnerabilityReferenceSerializer(serializers.ModelSerializer):
 
     class Meta:
         model = VulnerabilityReference
-        fields = ["reference_url", "reference_id", "scores", "url"]
+        fields = ["reference_url", "reference_id", "reference_type", "scores", "url"]
 
 
 class BaseResourceSerializer(serializers.HyperlinkedModelSerializer):
@@ -101,6 +108,8 @@ def get_vulnerability(self, vuln):
 
     purl = serializers.CharField(source="package_url")
 
+    is_vulnerable = serializers.BooleanField()
+
     class Meta:
         model = Package
         fields = ["url", "purl", "is_vulnerable", "affected_by_vulnerabilities"]
@@ -241,21 +250,27 @@ def get_latest_non_vulnerable(self, package):
 
     affected_by_vulnerabilities = serializers.SerializerMethodField("get_affected_vulnerabilities")
 
-    fixing_vulnerabilities = serializers.SerializerMethodField("get_fixed_vulnerabilities")
+    fixing_vulnerabilities = serializers.SerializerMethodField("get_fixing_vulnerabilities")
+
+    is_vulnerable = serializers.BooleanField()
 
     def get_fixed_packages(self, package):
         """
         Return a queryset of all packages that fix a vulnerability with
         same type, namespace, name, subpath and qualifiers of the `package`
         """
-        return Package.objects.filter(
-            name=package.name,
-            namespace=package.namespace,
-            type=package.type,
-            qualifiers=package.qualifiers,
-            subpath=package.subpath,
-            packagerelatedvulnerability__fix=True,
-        ).distinct()
+        return (
+            Package.objects.filter(
+                name=package.name,
+                namespace=package.namespace,
+                type=package.type,
+                qualifiers=package.qualifiers,
+                subpath=package.subpath,
+                packagerelatedvulnerability__fix=True,
+            )
+            .with_is_vulnerable()
+            .distinct()
+        )
 
     def get_vulnerabilities_for_a_package(self, package, fix) -> dict:
         """
@@ -278,7 +293,7 @@ def get_vulnerabilities_for_a_package(self, package, fix) -> dict:
             context={"request": self.context["request"]},
         ).data
 
-    def get_fixed_vulnerabilities(self, package) -> dict:
+    def get_fixing_vulnerabilities(self, package) -> dict:
         """
         Return a mapping of vulnerabilities fixed in the given `package`.
         """
@@ -315,12 +330,15 @@ class Meta:
             "version",
             "qualifiers",
             "subpath",
+            "is_vulnerable",
             "next_non_vulnerable_version",
             "latest_non_vulnerable_version",
             "affected_by_vulnerabilities",
             "fixing_vulnerabilities",
         ]
 
+    is_vulnerable = serializers.BooleanField()
+
 
 class PackageFilterSet(filters.FilterSet):
     purl = filters.CharFilter(method="filter_purl")
@@ -383,6 +401,9 @@ class PackageViewSet(viewsets.ReadOnlyModelViewSet):
     filterset_class = PackageFilterSet
     throttle_classes = [StaffUserRateThrottle, AnonRateThrottle]
 
+    def get_queryset(self):
+        return super().get_queryset().with_is_vulnerable()
+
     @extend_schema(
         request=PackageBulkSearchRequestSerializer,
         responses={200: PackageSerializer(many=True)},
@@ -429,6 +450,7 @@ def bulk_search(self, request):
                 Package.objects.filter(plain_package_url__in=plain_purls)
                 .order_by("plain_package_url")
                 .distinct("plain_package_url")
+                .with_is_vulnerable()
             )
 
             if not purl_only:
@@ -442,7 +464,7 @@ def bulk_search(self, request):
             vulnerable_purls = [str(package.plain_package_url) for package in vulnerable_purls]
             return Response(data=vulnerable_purls)
 
-        query = Package.objects.filter(package_url__in=purls).distinct()
+        query = Package.objects.filter(package_url__in=purls).distinct().with_is_vulnerable()
 
         if not purl_only:
             return Response(PackageSerializer(query, many=True, context={"request": request}).data)
@@ -456,7 +478,9 @@ def all(self, request):
         """
         Return the Package URLs of all packages known to be vulnerable.
         """
-        vulnerable_packages = Package.objects.vulnerable().only("package_url").distinct()
+        vulnerable_packages = (
+            Package.objects.vulnerable().only("package_url").distinct().with_is_vulnerable()
+        )
         vulnerable_purls = [str(package.package_url) for package in vulnerable_packages]
         return Response(vulnerable_purls)
 
@@ -487,11 +511,8 @@ def lookup(self, request):
         validated_data = serializer.validated_data
         purl = validated_data.get("purl")
 
-        return Response(
-            PackageSerializer(
-                Package.objects.for_purls([purl]), many=True, context={"request": request}
-            ).data
-        )
+        qs = self.get_queryset().for_purls([purl]).with_is_vulnerable()
+        return Response(PackageSerializer(qs, many=True, context={"request": request}).data)
 
     @extend_schema(
         request=PackageurlListSerializer,
@@ -522,7 +543,7 @@ def bulk_lookup(self, request):
 
         return Response(
             PackageSerializer(
-                Package.objects.for_purls(purls),
+                Package.objects.for_purls(purls).with_is_vulnerable(),
                 many=True,
                 context={"request": request},
             ).data
@@ -540,33 +561,47 @@ class VulnerabilityViewSet(viewsets.ReadOnlyModelViewSet):
     Lookup for vulnerabilities affecting packages.
     """
 
+    queryset = Vulnerability.objects.all()
+
     def get_fixed_packages_qs(self):
         """
         Filter the packages that fixes a vulnerability
         on fields like name, namespace and type.
         """
-        package_filter_data = {"packagerelatedvulnerability__fix": True}
+        return self.get_packages_qs().filter(packagerelatedvulnerability__fix=True)
 
+    def get_packages_qs(self):
+        """
+        Filter the packages on type, namespace and name.
+        """
         query_params = self.request.query_params
-        for field_name in ["name", "namespace", "type"]:
-            value = query_params.get(field_name)
-            if value:
+        package_filter_data = {}
+        for field_name in ("type", "namespace", "name"):
+            if value := query_params.get(field_name):
                 package_filter_data[field_name] = value
 
-        return PackageFilterSet(package_filter_data).qs
+        return PackageFilterSet(package_filter_data).qs.with_is_vulnerable()
 
     def get_queryset(self):
         """
         Assign filtered packages queryset from `get_fixed_packages_qs`
         to a custom attribute `filtered_fixed_packages`
         """
-        return Vulnerability.objects.prefetch_related(
-            "weaknesses",
-            Prefetch(
-                "packages",
-                queryset=self.get_fixed_packages_qs(),
-                to_attr="filtered_fixed_packages",
-            ),
+        return (
+            super()
+            .get_queryset()
+            .prefetch_related(
+                Prefetch(
+                    "packages",
+                    queryset=self.get_packages_qs(),
+                ),
+                "weaknesses",
+                Prefetch(
+                    "packages",
+                    queryset=self.get_fixed_packages_qs(),
+                    to_attr="filtered_fixed_packages",
+                ),
+            )
         )
 
     serializer_class = VulnerabilitySerializer

vulnerabilities/import_runner.py

@@ -70,17 +70,23 @@ def do_import(self, advisories) -> None:
             if advisory.date_imported:
                 continue
             logger.info(f"Processing advisory: {advisory!r}")
+            advisory_data = None
+            inferences = None
             try:
-                inferences = advisory_importer.get_inferences(
-                    advisory_data=advisory.to_advisory_data()
-                )
+                advisory_data = advisory.to_advisory_data()
+                inferences = advisory_importer.get_inferences(advisory_data=advisory_data)
                 process_inferences(
                     inferences=inferences,
                     advisory=advisory,
                     improver_name=importer_name,
                 )
-            except Exception as e:
-                logger.info(f"Failed to process advisory: {advisory!r} with error {e!r}")
+            except Exception:
+                from pprint import pformat
+
+                logger.warning(
+                    f"Failed to process advisory:\n{pformat(advisory_data.to_dict())}\n\n"
+                    f"with error:\n{traceback_format_exc()}\n\n"
+                )
         logger.info("Finished importing using %s.", advisory_importer.__class__.qualified_name)
 
     def process_advisories(
@@ -181,16 +187,23 @@ def process_inferences(inferences: List[Inference], advisory: Advisory, improver
                 reference=reference,
                 vulnerability=vulnerability,
             )
-
+            updated = False
             for severity in ref.severities:
-                _vs, updated = VulnerabilitySeverity.objects.update_or_create(
-                    scoring_system=severity.system.identifier,
-                    reference=reference,
-                    defaults={
-                        "value": str(severity.value),
-                        "scoring_elements": str(severity.scoring_elements),
-                    },
-                )
+                try:
+                    published_at = str(severity.published_at) if severity.published_at else None
+                    _vs, updated = VulnerabilitySeverity.objects.update_or_create(
+                        scoring_system=severity.system.identifier,
+                        reference=reference,
+                        defaults={
+                            "value": str(severity.value),
+                            "scoring_elements": str(severity.scoring_elements),
+                            "published_at": published_at,
+                        },
+                    )
+                except:
+                    logger.error(
+                        f"Failed to create VulnerabilitySeverity for: {severity} with error:\n{traceback_format_exc()}"
+                    )
                 if updated:
                     logger.info(
                         f"Severity updated for reference {ref!r} to value: {severity.value!r} "

vulnerabilities/importer.py

@@ -52,12 +52,17 @@ class VulnerabilitySeverity:
     system: ScoringSystem
     value: str
     scoring_elements: str = ""
+    published_at: Optional[datetime.datetime] = None
 
     def to_dict(self):
+        published_at_dict = (
+            {"published_at": self.published_at.isoformat()} if self.published_at else {}
+        )
         return {
             "system": self.system.identifier,
             "value": self.value,
             "scoring_elements": self.scoring_elements,
+            **published_at_dict,
         }
 
     @classmethod
@@ -70,12 +75,14 @@ def from_dict(cls, severity: dict):
             system=SCORING_SYSTEMS[severity["system"]],
             value=severity["value"],
             scoring_elements=severity.get("scoring_elements", ""),
+            published_at=severity.get("published_at"),
         )
 
 
 @dataclasses.dataclass(order=True)
 class Reference:
     reference_id: str = ""
+    reference_type: str = ""
     url: str = ""
     severities: List[VulnerabilitySeverity] = dataclasses.field(default_factory=list)
 
@@ -85,11 +92,17 @@ def __post_init__(self):
 
     def normalized(self):
         severities = sorted(self.severities)
-        return Reference(reference_id=self.reference_id, url=self.url, severities=severities)
+        return Reference(
+            reference_id=self.reference_id,
+            url=self.url,
+            severities=severities,
+            reference_type=self.reference_type,
+        )
 
     def to_dict(self):
         return {
             "reference_id": self.reference_id,
+            "reference_type": self.reference_type,
             "url": self.url,
             "severities": [severity.to_dict() for severity in self.severities],
         }
@@ -98,6 +111,7 @@ def to_dict(self):
     def from_dict(cls, ref: dict):
         return cls(
             reference_id=ref["reference_id"],
+            reference_type=ref["reference_type"],
             url=ref["url"],
             severities=[
                 VulnerabilitySeverity.from_dict(severity) for severity in ref["severities"]