Add API/ UI support for Patch/PackageCommitPatch

vulnerabilities/api_v2.py

@@ -9,6 +9,7 @@
 
 
 from django.db.models import Prefetch
+from django.db.models import Q
 from django_filters import rest_framework as filters
 from drf_spectacular.utils import OpenApiParameter
 from drf_spectacular.utils import extend_schema
@@ -33,14 +34,17 @@
 from vulnerabilities.models import CodeFixV2
 from vulnerabilities.models import ImpactedPackage
 from vulnerabilities.models import Package
+from vulnerabilities.models import PackageCommitPatch
 from vulnerabilities.models import PackageV2
+from vulnerabilities.models import Patch
 from vulnerabilities.models import PipelineRun
 from vulnerabilities.models import PipelineSchedule
 from vulnerabilities.models import Vulnerability
 from vulnerabilities.models import VulnerabilityReference
 from vulnerabilities.models import VulnerabilitySeverity
 from vulnerabilities.models import Weakness
 from vulnerabilities.throttling import PermissionBasedUserRateThrottle
+from vulnerabilities.utils import generate_patch_url
 from vulnerabilities.utils import group_advisories_by_content
 
 
@@ -333,6 +337,48 @@ def get_fixing_vulnerabilities(self, obj):
         return [vuln.vulnerability_id for vuln in obj.fixing_vulnerabilities.all()]
 
 
+class PackageCommitPatchSerializer(serializers.ModelSerializer):
+    introduced_in_advisories = serializers.SerializerMethodField()
+    fixed_in_advisories = serializers.SerializerMethodField()
+
+    class Meta:
+        model = PackageCommitPatch
+        fields = [
+            "id",
+            "commit_hash",
+            "vcs_url",
+            "patch_url",
+            "introduced_in_advisories",
+            "fixed_in_advisories",
+        ]
+
+    def get_introduced_in_advisories(self, obj):
+        impacts = obj.introduced_in_impacts.all()
+        return self.serialize_impacts(impacts)
+
+    def get_fixed_in_advisories(self, obj):
+        impacts = obj.fixed_in_impacts.all()
+        return self.serialize_impacts(impacts)
+
+    @staticmethod
+    def serialize_impacts(impacts):
+        unique_pairs = set()
+        for impact in impacts:
+            unique_pairs.add((impact.base_purl, impact.advisory.avid))
+        return [{"package": base_purl, "avid": avid} for base_purl, avid in unique_pairs]
+
+
+class PatchSerializer(serializers.ModelSerializer):
+    in_advisories = serializers.SerializerMethodField()
+
+    class Meta:
+        model = Patch
+        fields = ["id", "patch_url", "in_advisories"]
+
+    def get_in_advisories(self, obj):
+        return [{"avid": advisory.avid} for advisory in obj.advisories.all()]
+
+
 class PackageV3Serializer(serializers.ModelSerializer):
     purl = serializers.CharField(source="package_url")
     risk_score = serializers.FloatField(read_only=True)
@@ -889,6 +935,65 @@ def get_queryset(self):
         return queryset
 
 
+class PackageCommitPatchViewSet(viewsets.ReadOnlyModelViewSet):
+    """
+    API endpoint that allows viewing PackageCommitPatch entries.
+    """
+
+    serializer_class = PackageCommitPatchSerializer
+    throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
+
+    def get_queryset(self):
+        queryset = PackageCommitPatch.objects.prefetch_related(
+            "introduced_in_impacts__advisory", "fixed_in_impacts__advisory"
+        )
+
+        pk = self.request.query_params.get("id")
+        if pk:
+            queryset = queryset.filter(id=pk)
+
+        advisory_id = self.request.query_params.get("advisory_id")
+        if advisory_id:
+            queryset = queryset.filter(
+                Q(introduced_in_impacts__advisory__avid=advisory_id)
+                | Q(fixed_in_impacts__advisory__avid=advisory_id)
+            ).distinct()
+
+        purl = self.request.query_params.get("purl")
+        if purl:
+            queryset = queryset.filter(
+                Q(introduced_in_impacts__base_purl__icontains=purl)
+                | Q(fixed_in_impacts__base_purl__icontains=purl)
+            ).distinct()
+
+        return queryset
+
+
+class PatchViewSet(viewsets.ReadOnlyModelViewSet):
+    """
+    API endpoint that allows viewing PackageCommitPatch entries.
+    """
+
+    serializer_class = PatchSerializer
+    throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
+
+    def get_queryset(self):
+        queryset = Patch.objects.all()
+
+        pk = self.request.query_params.get("id")
+        if pk:
+            queryset = queryset.filter(id=pk)
+
+        advisory_id = self.request.query_params.get("advisory_id")
+        if advisory_id:
+            queryset = queryset.filter(advisory__advisory_id=advisory_id).distinct()
+
+        purl = self.request.query_params.get("purl")
+        if purl:
+            queryset = queryset.filter(package__package_url__icontains=purl).distinct()
+        return queryset
+
+
 class CodeFixV2ViewSet(viewsets.ReadOnlyModelViewSet):
     """
     API endpoint that allows viewing CodeFix entries.

vulnerabilities/models.py

@@ -70,6 +70,8 @@
 from vulnerabilities.severity_systems import EPSS
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
 from vulnerabilities.utils import compute_patch_checksum
+from vulnerabilities.utils import generate_commit_url
+from vulnerabilities.utils import generate_patch_url
 from vulnerabilities.utils import normalize_list
 from vulnerabilities.utils import normalize_purl
 from vulnerabilities.utils import normalize_text
@@ -2833,6 +2835,20 @@ class PackageCommitPatch(models.Model):
     patch_text = models.TextField(blank=True, null=True)
     patch_checksum = models.CharField(max_length=128, blank=True, null=True)
 
+    @property
+    def commit_url(self):
+        """
+        Generates Commit URL using the VCS URL and Commit Hash.
+        """
+        return generate_commit_url(self.vcs_url, self.commit_hash)
+
+    @property
+    def patch_url(self):
+        """
+        Generates Patch URL using the VCS URL and Commit Hash.
+        """
+        return generate_patch_url(self.vcs_url, self.commit_hash)
+
     def save(self, *args, **kwargs):
         if self.patch_text:
             self.patch_checksum = compute_patch_checksum(self.patch_text)

vulnerabilities/templates/advisory_detail.html

@@ -80,7 +80,16 @@
                         </a>
                     </li>
                 {% endif %}
-
+
+                <li data-tab="patch-url">
+                    <a>
+                        <span>
+                            {% with pcp_length=package_commit_patches|length %}
+                               Patches: ({{ advisory.patches.count|add:pcp_length }})
+                            {% endwith %}
+                        </span>
+                    </a>
+                </li>
                 <!-- <li data-tab="history">
                     <a>
                         <span>
@@ -413,7 +422,6 @@
                     </tr>
                 {% endfor %}
             </div>
-
 
             <div class="tab-div content" data-content="epss">
                 {% if epss_data %}
@@ -480,6 +488,28 @@
                 {% endif %}
             </div>
 
+            <div class="tab-div content" data-content="patch-url">
+                <table class="table is-bordered is-striped is-narrow is-hoverable is-fullwidth">
+                    <thead>
+                        <tr>
+                            <th style="width: 250px;"> Patch URL </th>
+                        </tr>
+                    </thead>
+                    {% for patch in patches %}
+                    <tr>
+                        <td class="wrap-strings"><a href="{{ patch.patch_url }}" target="_blank">{{ patch.patch_url }}<i
+                                    class="fa fa-external-link fa_link_custom"></i></a></td>
+                    </tr>
+                    {% empty %}
+                    <tr>
+                        <td colspan="2">
+                            There are no known patches.
+                        </td>
+                    </tr>
+                    {% endfor %}
+                </table>
+            </div>
+
             <div class="tab-div content" data-content="severities-vectors">
                 {% for severity_vector in severity_vectors %}
                     {% if severity_vector.vector.version == '2.0'  %}

vulnerabilities/templates/advisory_package_details.html

@@ -60,6 +60,67 @@
                 </div>
         </div>
     </section>
+    <section class="section pt-0">
+        <div class="details-container">
+            <article class="panel is-info panel-header-only">
+                <div class="panel-heading py-2 is-size-6">
+                    Vulnerable and Fixing Package Commit Patch details for Advisory:
+                    <span class="tag is-white custom">
+                        {{ advisoryv2.advisory_id }}
+                    </span>
+                </div>
+            </article>
+
+        <div id="tab-content">
+            <table class="table vcio-table width-100-pct mt-2">
+                <thead>
+                    <tr>
+                        <th style="width: 50%;">Introduced in</th>
+                        <th>Fixed by</th>
+                    </tr>
+                </thead>
+                <tbody>
+                    {% for impact in advisoryv2.impacted_packages.all %}
+
+                        {% for pkg_commit_patch in impact.introduced_by_package_commit_patches.all %}
+                            <tr>
+                                <td>
+                                    <a href="{{ pkg_commit_patch.patch_url }}" target="_self">
+                                        {{ pkg_commit_patch.vcs_url }}@{{ pkg_commit_patch.commit_hash }}
+                                    </a>
+                                    <br />
+                                    <a href="{{ pkg_commit_patch.patch_url }}" target="_self">{{ pkg_commit_patch.patch_url }}</a>
+                                </td>
+                                <td></td>
+                            </tr>
+                        {% endfor %}
+
+                        {% for pkg_commit_patch in impact.fixed_by_package_commit_patches.all %}
+                            <tr>
+                                <td></td>
+                                <td>
+                                    <a href="{{ pkg_commit_patch.commit_url }}" target="_self">
+                                        {{ impact.base_purl }}@{{ pkg_commit_patch.commit_hash }}
+                                    </a>
+                                    <br />
+                                    <a href="{{ pkg_commit_patch.patch_url }}" target="_self">{{ pkg_commit_patch.patch_url }}</a>
+                                </td>
+                            </tr>
+                        {% endfor %}
+
+                    {% empty %}
+                        <tr>
+                            <td colspan="2">
+                                This vulnerability is not known to affect any package commits.
+                            </td>
+                        </tr>
+                    {% endfor %}
+                </tbody>
+            </table>
+        </div>
+
+        </div>
+    </section>
 {% endif %}
 
 <script src="{% static 'js/main.js' %}" crossorigin="anonymous"></script>

vulnerabilities/utils.py

@@ -35,6 +35,8 @@
 from cwe2.database import InvalidCWEError
 from packageurl import PackageURL
 from packageurl.contrib.django.utils import without_empty_values
+from packageurl.contrib.purl2url import purl2url
+from packageurl.contrib.url2purl import url2purl
 from univers.version_range import RANGE_CLASS_BY_SCHEMES
 from univers.version_range import AlpineLinuxVersionRange
 from univers.version_range import NginxVersionRange
@@ -867,3 +869,51 @@ def group_advisories_by_content(advisories):
             entry["secondary"].add(advisory)
 
     return grouped
+
+
+def generate_patch_url(vcs_url, commit_hash):
+    """
+    Generate patch URL from VCS URL and commit hash.
+    """
+    if not vcs_url or not commit_hash:
+        return None
+
+    vcs_url = vcs_url.rstrip("/")
+
+    if vcs_url.startswith("https://github.com"):
+        return f"{vcs_url}/commit/{commit_hash}.patch"
+    elif vcs_url.startswith("https://gitlab.com"):
+        return f"{vcs_url}/-/commit/{commit_hash}.patch"
+    elif vcs_url.startswith("https://codeberg.org"):
+        return f"{vcs_url}/-/commit/{commit_hash}.patch"
+    elif vcs_url.startswith("https://android.googlesource.com"):
+        return f"{vcs_url}/+/{commit_hash}%5E%21?format=TEXT"
+    elif vcs_url.startswith("https://bitbucket.org"):
+        return f"{vcs_url}/-/commit/{commit_hash}/raw"
+    elif vcs_url.startswith("https://git.kernel.org"):
+        return f"{vcs_url}/patch/?id={commit_hash}"
+    return
+
+
+def generate_commit_url(vcs_url, commit_hash):
+    """
+    Generate commit URL from VCS URL and commit hash.
+    """
+    if not vcs_url or not commit_hash:
+        return None
+
+    vcs_url = vcs_url.rstrip("/")
+
+    if vcs_url.startswith("https://github.com"):
+        return f"{vcs_url}/commit/{commit_hash}"
+    elif vcs_url.startswith("https://gitlab.com"):
+        return f"{vcs_url}/-/commit/{commit_hash}"
+    elif vcs_url.startswith("https://codeberg.org"):
+        return f"{vcs_url}/-/commit/{commit_hash}"
+    elif vcs_url.startswith("https://android.googlesource.com"):
+        return f"{vcs_url}/+/{commit_hash}"
+    elif vcs_url.startswith("https://bitbucket.org"):
+        return f"{vcs_url}/-/commit/{commit_hash}"
+    elif vcs_url.startswith("https://git.kernel.org"):
+        return f"{vcs_url}/patch/?id={commit_hash}"
+    return