Handle large number of advisories case

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/api_v3.py

@@ -215,6 +215,45 @@ def get_affected_by_vulnerabilities(self, package):
 
         advisories = []
 
+        if package.type not in TYPES_WITH_MULTIPLE_IMPORTERS:
+            advisories_ids = advisories_qs.only("id")
+
+            advisories_ids = list(advisories_ids[:101])
+            if len(advisories_ids) > 100:
+                return None
+
+            advisory_by_avid = {adv.avid: adv for adv in advisories_qs}
+            avids = advisory_by_avid.keys()
+
+            impacts = (
+                package.affected_in_impacts.filter(advisory__avid__in=avids)
+                .select_related("advisory")
+                .prefetch_related("fixed_by_packages")
+            )
+
+            impact_by_avid = {impact.advisory.avid: impact for impact in impacts}
+
+            result = []
+
+            for advisory in advisories_qs:
+                impact = impact_by_avid.get(advisory.avid)
+                if not impact:
+                    continue
+
+                result.append(
+                    {
+                        "advisory_id": advisory.advisory_id.split("/")[-1],
+                        "aliases": [alias.alias for alias in advisory.aliases.all()],
+                        "summary": advisory.summary,
+                        "fixed_by_packages": [pkg.purl for pkg in impact.fixed_by_packages.all()],
+                        "severity": advisory.weighted_severity,
+                        "exploitability": advisory.exploitability,
+                        "risk_score": advisory.risk_score,
+                    }
+                )
+
+            return result
+
         is_grouped = AdvisorySet.objects.filter(package=package, relation_type="affecting").exists()
 
         if is_grouped:
@@ -239,43 +278,25 @@ def get_affected_by_vulnerabilities(self, package):
             advisories = merge_and_save_grouped_advisories(package, advisories_qs, "affecting")
             return self.return_advisories_data(package, advisories_qs, advisories)
 
-        advisories_ids = advisories_qs.only("id")
-
-        advisories_ids = list(advisories_ids[:101])
-        if len(advisories_ids) > 100:
-            return None
-
-        advisory_by_avid = {adv.avid: adv for adv in advisories_qs}
-        avids = advisory_by_avid.keys()
-
-        impacts = (
-            package.affected_in_impacts.filter(advisory__avid__in=avids)
-            .select_related("advisory")
-            .prefetch_related("fixed_by_packages")
-        )
-
-        impact_by_avid = {impact.advisory.avid: impact for impact in impacts}
-
-        result = []
+    def get_fixing_vulnerabilities(self, package):
+        advisories_qs = AdvisoryV2.objects.latest_fixed_by_advisories_for_purl(package.package_url)
 
-        for advisory in advisories_qs:
-            impact = impact_by_avid.get(advisory.avid)
-            if not impact:
-                continue
+        if not package.type in TYPES_WITH_MULTIPLE_IMPORTERS:
+            advisories_ids = advisories_qs.only("id")
 
-            result.append(
-                {
-                    "advisory_id": advisory.advisory_id.split("/")[-1],
-                    "aliases": [alias.alias for alias in advisory.aliases.all()],
-                    "summary": advisory.summary,
-                    "fixed_by_packages": [pkg.purl for pkg in impact.fixed_by_packages.all()],
-                }
-            )
+            advisories_ids = list(advisories_ids[:101])
+            if len(advisories_ids) > 100:
+                return None
 
-        return result
+            results = []
 
-    def get_fixing_vulnerabilities(self, package):
-        advisories_qs = AdvisoryV2.objects.latest_fixed_by_advisories_for_purl(package.package_url)
+            for advisory in advisories_qs:
+                results.append(
+                    {
+                        "advisory_id": advisory.advisory_id.split("/")[-1],
+                    }
+                )
+            return results
 
         advisories = []
 
@@ -302,22 +323,6 @@ def get_fixing_vulnerabilities(self, package):
             advisories = merge_and_save_grouped_advisories(package, advisories_qs, "fixing")
             return self.return_fixing_advisories_data(advisories)
 
-        advisories_ids = advisories_qs.only("id")
-
-        advisories_ids = list(advisories_ids[:101])
-        if len(advisories_ids) > 100:
-            return None
-
-        results = []
-
-        for advisory in advisories_qs:
-            results.append(
-                {
-                    "advisory_id": advisory.advisory_id.split("/")[-1],
-                }
-            )
-        return results
-
     def return_fixing_advisories_data(self, advisories):
         result = []
         for advisory in advisories:

vulnerabilities/views.py

@@ -216,6 +216,41 @@ def get_context_data(self, **kwargs):
         context["latest_non_vulnerable"] = latest_non_vulnerable
         context["package_search_form"] = PackageSearchForm(self.request.GET)
 
+        if not package.type in TYPES_WITH_MULTIPLE_IMPORTERS:
+            context["grouped"] = False
+
+            affected_by_advisories_url = None
+            fixing_advisories_url = None
+
+            affected_by_advisories_qs_ids = affecting_advisories.only("id")
+            fixing_advisories_qs_ids = fixed_by_advisories.only("id")
+
+            affected_by_advisories = list(affected_by_advisories_qs_ids[:101])
+            if len(affected_by_advisories) > 101:
+                affected_by_advisories_url = reverse_lazy(
+                    "affected_by_advisories_v2", kwargs={"purl": package.package_url}
+                )
+                context["affected_by_advisories_v2_url"] = affected_by_advisories_url
+
+            else:
+                fixed_pkg_details = get_fixed_package_details(package)
+                context["fixed_package_details"] = fixed_pkg_details
+                context["affected_by_advisories_v2"] = affecting_advisories
+                context["affected_by_advisories_v2_url"] = None
+
+            fixing_advisories = list(fixing_advisories_qs_ids[:101])
+            if len(fixing_advisories) > 101:
+                fixing_advisories_url = reverse_lazy(
+                    "fixing_advisories_v2", kwargs={"purl": package.package_url}
+                )
+                context["fixing_advisories_v2_url"] = fixing_advisories_url
+                context["fixing_advisories_v2"] = []
+
+            else:
+                context["fixing_advisories_v2"] = fixed_by_advisories
+
+            return context
+
         is_grouped = models.AdvisorySet.objects.filter(package=package).exists()
 
         if is_grouped:
@@ -284,40 +319,6 @@ def get_context_data(self, **kwargs):
             context["fixing_advisories_v2"] = fixing_advisories
             return context
 
-        context["grouped"] = False
-
-        affected_by_advisories_url = None
-        fixing_advisories_url = None
-
-        affected_by_advisories_qs_ids = affecting_advisories.only("id")
-        fixing_advisories_qs_ids = fixed_by_advisories.only("id")
-
-        affected_by_advisories = list(affected_by_advisories_qs_ids[:101])
-        if len(affected_by_advisories) > 101:
-            affected_by_advisories_url = reverse_lazy(
-                "affected_by_advisories_v2", kwargs={"purl": package.package_url}
-            )
-            context["affected_by_advisories_v2_url"] = affected_by_advisories_url
-
-        else:
-            fixed_pkg_details = get_fixed_package_details(package)
-            context["fixed_package_details"] = fixed_pkg_details
-            context["affected_by_advisories_v2"] = affecting_advisories
-            context["affected_by_advisories_v2_url"] = None
-
-        fixing_advisories = list(fixing_advisories_qs_ids[:101])
-        if len(fixing_advisories) > 101:
-            fixing_advisories_url = reverse_lazy(
-                "fixing_advisories_v2", kwargs={"purl": package.package_url}
-            )
-            context["fixing_advisories_v2_url"] = fixing_advisories_url
-            context["fixing_advisories_v2"] = []
-
-        else:
-            context["fixing_advisories_v2"] = fixed_by_advisories
-
-        return context
-
     def get_object(self, queryset=None):
         if queryset is None:
             queryset = self.get_queryset()