Add SSVC trees, resource URL and max_advisories

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

Author: TG1999

vulnerabilities/api_v3.py

@@ -23,20 +23,19 @@
 from rest_framework.reverse import reverse
 from rest_framework.throttling import AnonRateThrottle
 
+from vulnerabilities.models import SSVC
 from vulnerabilities.models import AdvisoryAlias
 from vulnerabilities.models import AdvisoryReference
 from vulnerabilities.models import AdvisorySet
 from vulnerabilities.models import AdvisorySetMember
 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
 from vulnerabilities.utils import TYPES_WITH_MULTIPLE_IMPORTERS
-from vulnerabilities.utils import get_advisories_from_groups
 from vulnerabilities.utils import merge_and_save_grouped_advisories
 
 
@@ -48,6 +47,7 @@ class PackageQuerySerializer(serializers.Serializer):
     )
     details = serializers.BooleanField(default=False)
     ignore_qualifiers_subpath = serializers.BooleanField(default=False)
+    max_advisories = serializers.IntegerField(default=100, min_value=1, max_value=10000)
 
     def validate(self, data):
         if not data["purls"]:
@@ -229,11 +229,17 @@ def get_affected_by_vulnerabilities(self, package):
             for adv in advisories:
                 fixed = impact_map.get(adv["avid"])
                 adv.pop("avid", None)
+                resource_url = None
+
+                if request := self.context.get("request", None):
+                    resource_url = adv.pop("resource_url", None)
+                    resource_url = request.build_absolute_uri(location=resource_url)
 
                 result.append(
                     {
                         **adv,
                         "fixed_by_packages": fixed,
+                        "resource_url": resource_url,
                     }
                 )
 
@@ -247,9 +253,20 @@ def get_affected_by_vulnerabilities(self, package):
             advisories_ids = advisories_qs.only("id")
 
             advisories_ids = list(advisories_ids[:101])
-            if len(advisories_ids) > 100:
+            if len(advisories_ids) > self.context.get("max_advisories", 100):
                 return None
 
+            advisories_qs = advisories_qs.prefetch_related(
+                "aliases",
+                Prefetch(
+                    "related_ssvcs",
+                    queryset=SSVC.objects.select_related("source_advisory").only(
+                        "id", "decision", "options", "vector", "source_advisory__url"
+                    ),
+                    to_attr="prefetched_ssvc_trees",
+                ),
+            )
+
             advisory_by_avid = {adv.avid: adv for adv in advisories_qs}
             avids = advisory_by_avid.keys()
 
@@ -265,8 +282,14 @@ def get_affected_by_vulnerabilities(self, package):
 
             for advisory in advisories_qs:
                 impact = impact_by_avid.get(advisory.avid)
-                if not impact:
-                    continue
+                fixed_by_packages = []
+                if impact:
+                    fixed_by_packages = [pkg.purl for pkg in impact.fixed_by_packages.all()]
+
+                resource_url = None
+
+                if request := self.context.get("request", None):
+                    resource_url = request.build_absolute_uri(location=advisory.get_absolute_url())
 
                 result.append(
                     {
@@ -277,7 +300,17 @@ def get_affected_by_vulnerabilities(self, package):
                         "severity": advisory.weighted_severity,
                         "exploitability": advisory.exploitability,
                         "risk_score": advisory.risk_score,
-                        "fixed_by_packages": [pkg.purl for pkg in impact.fixed_by_packages.all()],
+                        "fixed_by_packages": fixed_by_packages,
+                        "resource_url": resource_url,
+                        "ssvc_trees": [
+                            {
+                                "vector": ssvc.vector,
+                                "decision": ssvc.decision,
+                                "options": ssvc.options,
+                                "source_url": ssvc.source_advisory.url,
+                            }
+                            for ssvc in advisory.prefetched_ssvc_trees
+                        ],
                     }
                 )
 
@@ -297,25 +330,38 @@ def get_affected_by_vulnerabilities(self, package):
 
     def get_fixing_vulnerabilities(self, package):
         advisories = self.context["fixing_advisory_map"].get(package.id, [])
-        if advisories:
-            return advisories
+        results = []
+        for advisory in advisories:
+            if request := self.context.get("request", None):
+                resource_url = request.build_absolute_uri(location=advisory["resource_url"])
+            results.append(
+                {
+                    "advisory_id": advisory["advisory_id"],
+                    "resource_url": resource_url,
+                }
+            )
+        if results:
+            return results
 
         advisories_qs = AdvisoryV2.objects.latest_fixed_by_advisories_for_purl(package.package_url)
 
         if not package.type in TYPES_WITH_MULTIPLE_IMPORTERS:
             advisories_ids = advisories_qs.only("id")
 
             advisories_ids = list(advisories_ids[:101])
-            if len(advisories_ids) > 100:
+            if len(advisories_ids) > self.context.get("max_advisories", 100):
                 return None
 
             results = []
 
             for advisory in advisories_qs:
+                if request := self.context.get("request", None):
+                    resource_url = request.build_absolute_uri(location=advisory.get_absolute_url())
                 results.append(
                     {
                         "advisory_id": advisory.advisory_id.split("/")[-1],
                         "advisory_uid": advisory.avid,
+                        "resource_url": resource_url,
                     }
                 )
             return results
@@ -337,10 +383,16 @@ def return_fixing_advisories_data(self, advisories):
         result = []
         for advisory in advisories:
             assert isinstance(advisory, GroupedAdvisory)
+            resource_url = None
+            if request := self.context.get("request", None):
+                resource_url = request.build_absolute_uri(
+                    location=advisory.advisory.get_absolute_url()
+                )
             result.append(
                 {
                     "advisory_id": advisory.identifier,
                     "advisory_uid": advisory.advisory.avid,
+                    "resource_url": resource_url,
                 }
             )
 
@@ -361,9 +413,15 @@ def return_advisories_data(self, package, advisories_qs, advisories):
         result = []
         for advisory in advisories:
             assert isinstance(advisory, GroupedAdvisory)
+            resource_url = None
+            fixed_by_packages = []
+            if request := self.context.get("request", None):
+                resource_url = request.build_absolute_uri(
+                    location=advisory.advisory.get_absolute_url()
+                )
             impact = impact_by_avid.get(advisory.advisory.avid)
             if not impact:
-                continue
+                fixed_by_packages = list(set([pkg.purl for pkg in impact.fixed_by_packages.all()]))
 
             result.append(
                 {
@@ -374,9 +432,9 @@ def return_advisories_data(self, package, advisories_qs, advisories):
                     "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()])
-                    ),
+                    "fixed_by_packages": fixed_by_packages,
+                    "resource_url": resource_url,
+                    "ssvc_trees": advisory.ssvc_trees,
                 }
             )
 
@@ -405,6 +463,7 @@ def create(self, request, *args, **kwargs):
         purls = serializer.validated_data["purls"]
         details = serializer.validated_data["details"]
         ignore_qualifiers_subpath = serializer.validated_data["ignore_qualifiers_subpath"]
+        max_advisories = serializer.validated_data["max_advisories"]
 
         if not purls:
             impacted = ImpactedPackageAffecting.objects.filter(package_id=OuterRef("id"))
@@ -469,6 +528,7 @@ def create(self, request, *args, **kwargs):
                 "advisory_map": affected_advisory_map,
                 "impact_map": impact_map,
                 "fixing_advisory_map": fixing_advisory_map,
+                "max_advisories": max_advisories,
             },
         )
         return self.get_paginated_response(serializer.data)
@@ -583,7 +643,25 @@ def get_affected_advisories_bulk(packages):
             relation_type="affecting",
         )
         .select_related("primary_advisory")
-        .prefetch_related(Prefetch("aliases", queryset=AdvisoryAlias.objects.only("alias")))
+        .prefetch_related(
+            Prefetch("aliases", queryset=AdvisoryAlias.objects.only("alias")),
+            Prefetch(
+                "members",
+                queryset=AdvisorySetMember.objects.select_related("advisory").prefetch_related(
+                    Prefetch(
+                        "advisory__related_ssvcs",
+                        queryset=SSVC.objects.select_related("source_advisory").only(
+                            "id",
+                            "options",
+                            "decision",
+                            "vector",
+                            "source_advisory__url",
+                        ),
+                        to_attr="prefetched_ssvc_trees",
+                    )
+                ),
+            ),
+        )
         .annotate(
             max_severity=Max(
                 "members__advisory__weighted_severity",
@@ -627,6 +705,20 @@ def get_affected_advisories_bulk(packages):
             identifier = primary.advisory_id.split("/")[-1]
 
             aliases = [a for a in adv._aliases_cache if a != identifier]
+            all_ssvc = []
+
+            for member in adv.members.all():
+                all_ssvc.extend(member.advisory.prefetched_ssvc_trees)
+
+            for ssvc in all_ssvc:
+                all_ssvc.append(
+                    {
+                        "vector": ssvc.vector,
+                        "decision": ssvc.decision,
+                        "options": ssvc.options,
+                        "source_url": ssvc.source_advisory.url,
+                    }
+                )
 
             grouped.append(
                 {
@@ -637,6 +729,8 @@ def get_affected_advisories_bulk(packages):
                     "exploitability": exploitability,
                     "risk_score": risk_score,
                     "summary": primary.summary,
+                    "resource_url": primary.get_absolute_url(),
+                    "ssvc_trees": all_ssvc,
                 }
             )
 
@@ -697,17 +791,21 @@ def get_fixing_advisories_bulk(packages):
     package_map = defaultdict(list)
 
     for adv in advisory_sets:
-        package_map[adv.package_id].append(adv.primary_advisory.advisory_id)
+        package_map[adv.package_id].append(adv.primary_advisory)
 
     result = {}
 
     for package in packages:
         groups = package_map.get(package.id, [])
         grouped = []
 
-        for adv_id in groups:
+        for advisory in groups:
             grouped.append(
-                {"advisory_id": adv_id.split("/")[-1], "advisory_uid": adv_id.split("/")[-1]}
+                {
+                    "advisory_id": advisory.advisory_id.split("/")[-1],
+                    "resource_url": advisory.get_absolute_url(),
+                    "advisory_uid": advisory.avid,
+                }
             )
 
         result[package.id] = grouped

vulnerabilities/models.py

@@ -19,7 +19,7 @@
 from itertools import groupby
 from operator import attrgetter
 from traceback import format_exc as traceback_format_exc
-from typing import List
+from typing import Dict, List
 from typing import NamedTuple
 from typing import Optional
 from typing import Set
@@ -3872,6 +3872,7 @@ class GroupedAdvisory(NamedTuple):
     weighted_severity: Optional[float]
     exploitability: Optional[float]
     risk_score: Optional[float]
+    ssvc_trees: List[Dict]
 
 
 class AdvisoryPOC(models.Model):

vulnerabilities/utils.py

@@ -35,6 +35,7 @@
 import urllib3
 from cwe2.database import Database
 from cwe2.database import InvalidCWEError
+from django.db.models import Prefetch
 from packageurl import PackageURL
 from packageurl.contrib.django.utils import without_empty_values
 from univers.version_range import RANGE_CLASS_BY_SCHEMES
@@ -979,10 +980,12 @@ def get_merged_identifier_groups(advisories):
     return final_groups
 
 
-def get_advisories_from_groups(groups):
+def get_advisories_from_groups(groups, include_ssvc_trees=False):
     """
     Return a list of advisories from the merged groups of advisories.
     """
+    from vulnerabilities.models import SSVC
+    from vulnerabilities.models import AdvisoryV2
     from vulnerabilities.models import Group
     from vulnerabilities.models import GroupedAdvisory
 
@@ -1016,6 +1019,35 @@ def get_advisories_from_groups(groups):
         identifier = group.primary.advisory_id.split("/")[-1]
         filtered_aliases = [alias for alias in group.aliases if alias.alias != identifier]
 
+        ssvc_trees = []
+
+        if include_ssvc_trees:
+
+            all_advs = [group.primary] + list(group.secondaries)
+
+            advisories_qs = AdvisoryV2.objects.filter(
+                id__in=[adv.id for adv in all_advs]
+            ).prefetch_related(
+                Prefetch(
+                    "related_ssvcs",
+                    queryset=SSVC.objects.select_related("source_advisory")
+                    .only("id", "vector", "decision", "options", "source_advisory__url")
+                    .distinct(),
+                    to_attr="ssvc_trees",
+                )
+            )
+
+            ssvc_trees = [
+                {
+                    "vector": ssvc.vector,
+                    "decision": ssvc.decision,
+                    "options": ssvc.options,
+                    "url": ssvc.source_advisory.url if ssvc.source_advisory else None,
+                }
+                for adv in advisories_qs
+                for ssvc in adv.ssvc_trees
+            ]
+
         advisories.append(
             GroupedAdvisory(
                 aliases=filtered_aliases,
@@ -1024,6 +1056,7 @@ def get_advisories_from_groups(groups):
                 weighted_severity=weighted_severity,
                 exploitability=exploitability,
                 risk_score=risk_score,
+                ssvc_trees=ssvc_trees or [],
             )
         )