Fix grouping

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

Author: TG1999

vulnerabilities/api_v3.py

@@ -7,6 +7,7 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
+from typing import List
 from urllib.parse import urlencode
 
 from django.db.models import Exists
@@ -25,6 +26,8 @@
 from vulnerabilities.models import AdvisorySeverity
 from vulnerabilities.models import AdvisoryV2
 from vulnerabilities.models import AdvisoryWeakness
+from vulnerabilities.models import Group
+from vulnerabilities.models import GroupedAdvisory
 from vulnerabilities.models import ImpactedPackageAffecting
 from vulnerabilities.models import PackageV2
 from vulnerabilities.throttling import PermissionBasedUserRateThrottle
@@ -273,15 +276,15 @@ def get_affected_by_vulnerabilities(self, package):
             )
 
             affected_groups = [
-                (
-                    list(adv.aliases.all()),
-                    adv.primary_advisory,
-                    [member.advisory for member in adv.secondary_members],
+                Group(
+                    aliases=list(adv.aliases.all()),
+                    primary_advisory=adv.primary_advisory,
+                    secondaries=[member.advisory for member in adv.secondary_members],
                 )
                 for adv in affected_by_advisories_qs
             ]
 
-            advisories = get_advisories_from_groups(affected_groups)
+            advisories: List[GroupedAdvisory] = get_advisories_from_groups(affected_groups)
             return self.return_advisories_data(package, advisories_qs, advisories)
 
         if package.type in TYPES_WITH_MULTIPLE_IMPORTERS:
@@ -290,7 +293,9 @@ def get_affected_by_vulnerabilities(self, package):
                 "impacted_packages__affecting_packages",
                 "impacted_packages__fixed_by_packages",
             )
-            advisories = merge_and_save_grouped_advisories(package, advisories_qs, "affecting")
+            advisories: List[GroupedAdvisory] = merge_and_save_grouped_advisories(
+                package, advisories_qs, "affecting"
+            )
             return self.return_advisories_data(package, advisories_qs, advisories)
 
     def get_fixing_vulnerabilities(self, package):
@@ -333,15 +338,15 @@ def get_fixing_vulnerabilities(self, package):
             )
 
             fixing_groups = [
-                (
-                    list(adv.aliases.all()),
-                    adv.primary_advisory,
-                    [member.advisory for member in adv.secondary_members],
+                Group(
+                    aliases=list(adv.aliases.all()),
+                    primary_advisory=adv.primary_advisory,
+                    secondaries=[member.advisory for member in adv.secondary_members],
                 )
                 for adv in fixing_advisories_qs
             ]
 
-            advisories = get_advisories_from_groups(fixing_groups)
+            advisories: List[GroupedAdvisory] = get_advisories_from_groups(fixing_groups)
             return self.return_fixing_advisories_data(advisories)
 
         if package.type in TYPES_WITH_MULTIPLE_IMPORTERS:
@@ -350,15 +355,18 @@ def get_fixing_vulnerabilities(self, package):
                 "impacted_packages__affecting_packages",
                 "impacted_packages__fixed_by_packages",
             )
-            advisories = merge_and_save_grouped_advisories(package, advisories_qs, "fixing")
+            advisories: List[GroupedAdvisory] = merge_and_save_grouped_advisories(
+                package, advisories_qs, "fixing"
+            )
             return self.return_fixing_advisories_data(advisories)
 
     def return_fixing_advisories_data(self, advisories):
         result = []
         for advisory in advisories:
+            assert isinstance(advisory, GroupedAdvisory)
             result.append(
                 {
-                    "advisory_id": advisory["identifier"],
+                    "advisory_id": advisory.identifier,
                 }
             )
 
@@ -378,18 +386,19 @@ def return_advisories_data(self, package, advisories_qs, advisories):
 
         result = []
         for advisory in advisories:
-            impact = impact_by_avid.get(advisory["advisory"].avid)
+            assert isinstance(advisory, GroupedAdvisory)
+            impact = impact_by_avid.get(advisory.advisory.avid)
             if not impact:
                 continue
 
             result.append(
                 {
-                    "advisory_id": advisory["identifier"],
-                    "aliases": [alias.alias for alias in advisory["aliases"]],
-                    "weighted_severity": advisory["weighted_severity"],
-                    "exploitability": advisory["exploitability"],
-                    "risk_score": advisory["risk_score"],
-                    "summary": advisory["advisory"].summary,
+                    "advisory_id": advisory.identifier,
+                    "aliases": [alias.alias for alias in advisory.aliases],
+                    "weighted_severity": advisory.weighted_severity,
+                    "exploitability": advisory.exploitability,
+                    "risk_score": advisory.risk_score,
+                    "summary": advisory.advisory.summary,
                     "fixed_by_packages": list(
                         set([pkg.purl for pkg in impact.fixed_by_packages.all()])
                     ),

vulnerabilities/models.py

@@ -20,6 +20,9 @@
 from operator import attrgetter
 from traceback import format_exc as traceback_format_exc
 from typing import List
+from typing import NamedTuple
+from typing import Optional
+from typing import Set
 from typing import Union
 from urllib.parse import urljoin
 
@@ -3714,3 +3717,26 @@ def __str__(self):
 
     class Meta:
         unique_together = ("vector", "source_advisory")
+
+
+class Group(NamedTuple):
+    """
+    A Group of advisories that have been merged together based on their content and identifiers.
+    """
+
+    aliases: Set[AdvisoryAlias]
+    primary: AdvisoryV2
+    secondaries: List[AdvisoryV2]
+
+
+class GroupedAdvisory(NamedTuple):
+    """
+    A GroupedAdvisory represents a single advisory that has been grouped with its aliases and related advisories.
+    """
+
+    aliases: Set[AdvisoryAlias]
+    advisory: AdvisoryV2
+    identifier: str
+    weighted_severity: Optional[float]
+    exploitability: Optional[float]
+    risk_score: Optional[float]

vulnerabilities/pipelines/v2_improvers/group_advisories_for_packages.py

@@ -7,7 +7,10 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
+from typing import List
+
 from vulnerabilities.models import AdvisoryV2
+from vulnerabilities.models import Group
 from vulnerabilities.models import PackageV2
 from vulnerabilities.pipelines import VulnerableCodePipeline
 from vulnerabilities.pipes.group_advisories import delete_and_save_advisory_set
@@ -48,8 +51,8 @@ def group_advisoris_for_packages(logger=None):
         )
 
         try:
-            affected_groups = merge_advisories(affecting_advisories, package)
-            fixed_by_groups = merge_advisories(fixed_by_advisories, package)
+            affected_groups: List[Group] = merge_advisories(affecting_advisories, package)
+            fixed_by_groups: List[Group] = merge_advisories(fixed_by_advisories, package)
             delete_and_save_advisory_set(affected_groups, package, relation="affecting")
             delete_and_save_advisory_set(fixed_by_groups, package, relation="fixing")
         except Exception as e:

vulnerabilities/pipes/group_advisories.py

@@ -14,31 +14,33 @@
 def delete_and_save_advisory_set(groups, package, relation=None):
     from vulnerabilities.models import AdvisorySet
     from vulnerabilities.models import AdvisorySetMember
+    from vulnerabilities.models import Group
 
     AdvisorySet.objects.filter(package=package, relation_type=relation).delete()
 
     membership_to_create = []
 
-    for identifiers, primary, secondary in groups:
+    for group in groups:
 
+        assert isinstance(group, Group)
         advisory_set = AdvisorySet.objects.create(
             package=package,
             relation_type=relation,
-            primary_advisory=primary,
+            primary_advisory=group.primary,
         )
 
-        advisory_set.aliases.add(*identifiers)
+        advisory_set.aliases.add(*group.aliases)
         advisory_set.save()
 
         membership_to_create.append(
             AdvisorySetMember(
                 advisory_set=advisory_set,
-                advisory=primary,
+                advisory=group.primary,
                 is_primary=True,
             )
         )
 
-        for adv in secondary:
+        for adv in group.secondaries:
             membership_to_create.append(
                 AdvisorySetMember(
                     advisory_set=advisory_set,

vulnerabilities/tests/test_advisory_merge.py

@@ -15,6 +15,7 @@
 from vulnerabilities.models import AdvisorySet
 from vulnerabilities.models import AdvisorySetMember
 from vulnerabilities.models import AdvisoryV2
+from vulnerabilities.models import Group
 from vulnerabilities.models import ImpactedPackage
 from vulnerabilities.models import PackageV2
 from vulnerabilities.utils import compute_advisory_content_hash
@@ -136,8 +137,8 @@ def test_get_advisories_from_groups(self):
         groups = get_merged_identifier_groups([adv])
         result = get_advisories_from_groups(groups)
 
-        assert result[0]["identifier"] == "GHSA-ABC-123"
-        assert len(result[0]["aliases"]) == 1
+        assert result[0].identifier == "GHSA-ABC-123"
+        assert len(result[0].aliases) == 1
 
     def test_delete_and_save_advisory_set(self):
         package = PackageV2.objects.from_purl("pkg:pypi/sample@1.0.0")
@@ -147,7 +148,7 @@ def test_delete_and_save_advisory_set(self):
 
         adv1.aliases.create(alias="CVE-1")
 
-        groups = [(set(adv1.aliases.all()), adv1, [adv2])]
+        groups = [Group(aliases=set(adv1.aliases.all()), primary=adv1, secondaries=[adv2])]
 
         delete_and_save_advisory_set(groups, package, relation="affecting")
 

vulnerabilities/utils.py

@@ -20,7 +20,9 @@
 from functools import total_ordering
 from http import HTTPStatus
 from typing import List
+from typing import NamedTuple
 from typing import Optional
+from typing import Set
 from typing import Tuple
 from typing import Union
 from unittest.mock import MagicMock
@@ -850,6 +852,7 @@ def merge_advisories(advisories, package):
     """
     Merge advisories based on their content hash and identifiers.
     """
+    from vulnerabilities.models import Group
 
     advisories = list(advisories)
 
@@ -859,7 +862,7 @@ def merge_advisories(advisories, package):
         content_hash = compute_advisory_content_hash(adv, package)
         content_hash_map[content_hash].append(adv)
 
-    final_groups = []
+    final_groups: List[Group] = []
 
     for group in content_hash_map.values():
         groups = get_merged_identifier_groups(group)
@@ -901,6 +904,7 @@ def get_merged_identifier_groups(advisories):
     Merge advisories based on their identifiers (advisory_id and aliases).
     Example: If two advisories share ``advisory_id`` or share an alias, they will be merged together.
     """
+    from vulnerabilities.models import Group
 
     identifier_groups = defaultdict(set)
 
@@ -938,7 +942,7 @@ def get_merged_identifier_groups(advisories):
         if adv not in all_grouped:
             merged.append({adv})
 
-    final_groups = []
+    final_groups: List[Group] = []
 
     for group in merged:
         identifiers = set()
@@ -950,7 +954,7 @@ def get_merged_identifier_groups(advisories):
 
         secondary = [a for a in group if a != primary]
 
-        final_groups.append((identifiers, primary, secondary))
+        final_groups.append(Group(aliases=identifiers, primary=primary, secondaries=secondary))
 
     return final_groups
 
@@ -959,35 +963,48 @@ def get_advisories_from_groups(groups):
     """
     Return a list of advisories from the merged groups of advisories.
     """
+    from vulnerabilities.models import Group
+    from vulnerabilities.models import GroupedAdvisory
+
     advisories = []
-    weighted_severity = None
-    exploitability = None
-    risk_score = None
-    for aliases, primary, secondaries in groups:
+
+    for group in groups:
+
+        assert isinstance(group, Group)
+        weighted_severity = None
+        exploitability = None
+        risk_score = None
+
         severity_scores = []
-        exploitability_scores = []
-        identifier = primary.advisory_id.split("/")[-1]
-        filtered_aliases = [alias for alias in aliases if alias.alias != identifier]
-        severity_scores.extend([adv.weighted_severity for adv in secondaries])
-        exploitability_scores.extend([adv.exploitability for adv in secondaries])
-        severity_scores.append(primary.weighted_severity)
-        exploitability_scores.append(primary.exploitability)
+        severity_scores.append(group.primary.weighted_severity or 0.0)
+        severity_scores.extend([adv.weighted_severity or 0.0 for adv in group.secondaries])
+
         if severity_scores:
             weighted_severity = round(max(severity_scores), 1)
+
+        exploitability_scores = []
+        exploitability_scores.append(group.primary.exploitability or 0.0)
+        exploitability_scores.extend([adv.exploitability or 0.0 for adv in group.secondaries])
+
         if exploitability_scores:
             exploitability = max(exploitability_scores)
+
         if exploitability and weighted_severity:
             risk_score = min(float(exploitability * weighted_severity), 10.0)
             risk_score = round(risk_score, 1)
+
+        identifier = group.primary.advisory_id.split("/")[-1]
+        filtered_aliases = [alias for alias in group.aliases if alias.alias != identifier]
+
         advisories.append(
-            {
-                "aliases": filtered_aliases,
-                "advisory": primary,
-                "identifier": identifier,
-                "weighted_severity": weighted_severity,
-                "exploitability": exploitability,
-                "risk_score": risk_score,
-            }
+            GroupedAdvisory(
+                aliases=filtered_aliases,
+                advisory=group.primary,
+                identifier=identifier,
+                weighted_severity=weighted_severity,
+                exploitability=exploitability,
+                risk_score=risk_score,
+            )
         )
 
     return advisories