Resolve merge migration conflict

Signed-off-by: ziad hany <ziadhany2016@gmail.com>

Author: ziadhany

vulnerabilities/api_v3.py

@@ -15,6 +15,7 @@
 from django.db.models import Max
 from django.db.models import OuterRef
 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 extend_schema
 from packageurl import PackageURL
@@ -33,7 +34,9 @@
 from vulnerabilities.models import Group
 from vulnerabilities.models import GroupedAdvisory
 from vulnerabilities.models import ImpactedPackageAffecting
+from vulnerabilities.models import PackageCommitPatch
 from vulnerabilities.models import PackageV2
+from vulnerabilities.models import Patch
 from vulnerabilities.throttling import PermissionBasedUserRateThrottle
 from vulnerabilities.utils import TYPES_WITH_MULTIPLE_IMPORTERS
 from vulnerabilities.utils import get_advisories_from_groups
@@ -386,6 +389,48 @@ def get_latest_non_vulnerable_version(self, package):
             return latest_non_vulnerable.version
 
 
+class PackageCommitPatchSerializer(serializers.ModelSerializer):
+    introduced_in_advisories = serializers.SerializerMethodField()
+    fixed_in_advisories = serializers.SerializerMethodField()
+
+    class Meta:
+        model = PackageCommitPatch
+        fields = [
+            "commit_hash",
+            "vcs_url",
+            "commit_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 [{"purl": base_purl, "avid": avid} for base_purl, avid in unique_pairs]
+
+
+class PatchSerializer(serializers.ModelSerializer):
+    in_advisories = serializers.SerializerMethodField()
+
+    class Meta:
+        model = Patch
+        fields = ["patch_url", "in_advisories"]
+
+    def get_in_advisories(self, obj):
+        return [advisory.avid for advisory in obj.advisories.all()]
+
+
 class PackageV3ViewSet(viewsets.GenericViewSet):
     queryset = PackageV2.objects.all()
     serializer_class = PackageV3Serializer
@@ -558,6 +603,67 @@ def get_queryset(self):
         return AdvisoryV2.objects.filter(**{self.relation: purl}).latest_per_avid()
 
 
+class PackageCommitPatchFilter(filters.FilterSet):
+    advisory_avid = filters.CharFilter(method="filter_by_advisory", label="Advisory ID")
+    purl = filters.CharFilter(method="filter_by_purl", label="Purl")
+    commit_hash = filters.CharFilter(lookup_expr="exact", label="Commit Hash")
+    vcs_url = filters.CharFilter(lookup_expr="icontains", label="VCS URL")
+
+    class Meta:
+        model = PackageCommitPatch
+        fields = ["advisory_avid", "purl", "commit_hash", "vcs_url"]
+
+    def filter_by_advisory(self, queryset, name, value):
+        return queryset.filter(
+            Q(introduced_in_impacts__advisory__avid=value)
+            | Q(fixed_in_impacts__advisory__avid=value)
+        ).distinct()
+
+    def filter_by_purl(self, queryset, name, value):
+        return queryset.filter(
+            Q(introduced_in_impacts__base_purl__icontains=value)
+            | Q(fixed_in_impacts__base_purl__icontains=value)
+        ).distinct()
+
+
+class PackageCommitPatchViewSet(viewsets.ReadOnlyModelViewSet):
+    """
+    API endpoint that allows viewing PackageCommitPatch entries
+    """
+
+    serializer_class = PackageCommitPatchSerializer
+    throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
+    filter_backends = [filters.DjangoFilterBackend]
+    filterset_class = PackageCommitPatchFilter
+
+    def get_queryset(self):
+        return PackageCommitPatch.objects.prefetch_related(
+            "introduced_in_impacts__advisory", "fixed_in_impacts__advisory"
+        ).order_by("id")
+
+
+class PatchFilter(filters.FilterSet):
+    advisory_avid = filters.CharFilter(field_name="advisories__avid", label="Advisory ID")
+
+    class Meta:
+        model = Patch
+        fields = ["advisory_avid"]
+
+
+class PatchViewSet(viewsets.ReadOnlyModelViewSet):
+    """
+    API endpoint that allows viewing Patch entries
+    """
+
+    serializer_class = PatchSerializer
+    throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
+    filter_backends = [filters.DjangoFilterBackend]
+    filterset_class = PatchFilter
+
+    def get_queryset(self):
+        return Patch.objects.all()
+
+
 class FixingAdvisoriesViewSet(PackageAdvisoriesViewSet):
     relation = "impacted_packages__fixed_by_packages__package_url"
 

vulnerabilities/tests/test_api_v2.py

@@ -23,8 +23,11 @@
 from vulnerabilities.models import Alias
 from vulnerabilities.models import ApiUser
 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
@@ -834,3 +837,142 @@ def test_filter_codefix_by_advisory_id_not_found(self):
             response = self.client.get(self.url, {"advisory_id": "nonexistent/ADVISORY-ID"})
         assert response.status_code == status.HTTP_200_OK
         assert response.data["count"] == 0
+
+
+class PackageCommitPatchList(APITestCase):
+    def setUp(self):
+        self.advisory = AdvisoryV2.objects.create(
+            datasource_id="test_source",
+            advisory_id="TEST-2025-001",
+            avid="test_source/TEST-2025-001",
+            unique_content_id="a" * 64,
+            url="https://example.com/advisory",
+            date_collected="2025-07-01T00:00:00Z",
+        )
+
+        self.affected_package = PackageV2.objects.from_purl(purl="pkg:github/torvalds/linux@1.0.0")
+        self.fixed_package = PackageV2.objects.from_purl(purl="pkg:github/torvalds/linux@1.0.1")
+
+        self.pkg_commit_patch1 = PackageCommitPatch.objects.create(
+            commit_hash="2e1c42391ff2556387b3cb6308b24f6f65619feb",
+            vcs_url="https://github.com/torvalds/linux",
+            patch_text="From 2e1c42391ff2556387b3cb6308b24f6f65619feb Mon Sep 17 00:00:00 2001...",
+        )
+
+        self.pkg_commit_patch2 = PackageCommitPatch.objects.create(
+            commit_hash="99253eb750fda6a644d5188fb26c43bad8d5a745",
+            vcs_url="https://git.kernel.org/pub/scm/linux/kernel/git/stable/linux.git",
+            patch_text="From 99253eb750fda6a644d5188fb26c43bad8d5a745 Mon Sep 17 00:00:00 2001...",
+        )
+
+        self.pkg_commit_patch3 = PackageCommitPatch.objects.create(
+            commit_hash="f043bfc98c193c284e2cd768fefabe18ac2fed9b",
+            vcs_url="https://github.com/torvalds/linux",
+            patch_text="From f043bfc98c193c284e2cd768fefabe18ac2fed9b Mon Sep 17 00:00:00 2001...",
+        )
+
+        self.impacted_package1 = ImpactedPackage.objects.create(
+            base_purl="pkg:github/torvalds/linux",
+            advisory=self.advisory,
+        )
+
+        self.impacted_package2 = ImpactedPackage.objects.create(
+            base_purl="pkg:generic/git.kernel.org/pub/scm/linux/kernel",
+            advisory=self.advisory,
+        )
+
+        self.impacted_package1.fixed_by_package_commit_patches.add(self.pkg_commit_patch1)
+        self.impacted_package1.introduced_by_package_commit_patches.add(self.pkg_commit_patch3)
+        self.impacted_package2.fixed_by_package_commit_patches.add(self.pkg_commit_patch2)
+
+        self.user = ApiUser.objects.create_api_user(username="e@mail.com")
+        self.auth = f"Token {self.user.auth_token.key}"
+        self.client = APIClient(enforce_csrf_checks=True)
+        self.client.credentials(HTTP_AUTHORIZATION=self.auth)
+        self.url = reverse("package-commit-patch-list")
+
+    def test_package_commit_patches_list(self):
+        response = self.client.get(self.url)
+        assert response.status_code == 200
+        results = response.json().get("results", response.json())
+        assert len(results) == 3
+        patch_data = results[0]
+        assert patch_data["vcs_url"] == self.pkg_commit_patch1.vcs_url
+        assert patch_data["commit_hash"] == self.pkg_commit_patch1.commit_hash
+        assert patch_data["fixed_in_advisories"] == [
+            {"avid": self.advisory.avid, "purl": self.impacted_package1.base_purl}
+        ]
+        assert patch_data["introduced_in_advisories"] == []
+
+    def test_filter_by_commit_hash(self):
+        response = self.client.get(f"{self.url}?commit_hash={self.pkg_commit_patch1.commit_hash}")
+        results = response.json().get("results", response.json())
+        assert len(results) == 1
+
+        response = self.client.get(f"{self.url}?commit_hash=test")
+        results = response.json().get("results", response.json())
+        assert len(results) == 0
+
+    def test_filter_by_vcs_url(self):
+        response = self.client.get(f"{self.url}?vcs_url={self.pkg_commit_patch1.vcs_url}")
+        results = response.json().get("results", response.json())
+        assert len(results) == 2
+
+        response = self.client.get(f"{self.url}?vcs_url=test")
+        results = response.json().get("results", response.json())
+        assert len(results) == 0
+
+    def test_filter_by_advisory_avid(self):
+        response = self.client.get(f"{self.url}?advisory_avid={self.advisory.avid}")
+        results = response.json().get("results", response.json())
+        assert len(results) == 3
+
+        response = self.client.get(f"{self.url}?advisory_avid=test_source/DOES-NOT-EXIST")
+        results = response.json().get("results", response.json())
+        assert len(results) == 0
+
+    def test_filter_by_purl(self):
+        response = self.client.get(f"{self.url}?purl=pkg:github/torvalds/linux")
+        results = response.json().get("results", response.json())
+        assert len(results) == 2
+
+        response = self.client.get(f"{self.url}?purl=pkg:github/aboutcode-org")
+        results = response.json().get("results", response.json())
+        assert len(results) == 0
+
+
+class PatchList(APITestCase):
+    def setUp(self):
+        self.advisory = AdvisoryV2.objects.create(
+            datasource_id="test_source",
+            advisory_id="TEST-2025-001",
+            avid="test_source/TEST-2025-001",
+            unique_content_id="a" * 64,
+            url="https://example.com/advisory",
+            date_collected="2025-07-01T00:00:00Z",
+        )
+
+        self.patch = Patch.objects.create(
+            patch_url="https://lore.kernel.org/patchwork/patch/1086060/", patch_text="some text"
+        )
+
+        self.advisory.patches.add(self.patch)
+
+        self.user = ApiUser.objects.create_api_user(username="e@mail.com")
+        self.auth = f"Token {self.user.auth_token.key}"
+        self.client = APIClient(enforce_csrf_checks=True)
+        self.client.credentials(HTTP_AUTHORIZATION=self.auth)
+        self.url = reverse("patches-list")
+
+    def test_patch_list(self):
+        response = self.client.get(self.url)
+        assert response.status_code == 200
+        results = response.json().get("results", response.json())
+        assert len(results) == 1
+        assert results[0]["patch_url"] == self.patch.patch_url
+        assert results == [
+            {
+                "patch_url": "https://lore.kernel.org/patchwork/patch/1086060/",
+                "in_advisories": ["test_source/TEST-2025-001"],
+            }
+        ]

vulnerabilities/utils.py

@@ -37,6 +37,7 @@
 from cwe2.database import InvalidCWEError
 from packageurl import PackageURL
 from packageurl.contrib.django.utils import without_empty_values
+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
@@ -1033,3 +1034,51 @@ def merge_and_save_grouped_advisories(package, advisories, relation):
     "gem",
     "conan",
 ]
+
+
+def generate_commit_url(vcs_url, commit_hash):
+    """
+    Generate commit URL from VCS URL and commit hash.
+    """
+    from packageurl.contrib import purl2url
+
+    if not vcs_url or not commit_hash:
+        return
+
+    purl = url2purl(vcs_url)
+    if not purl:
+        return
+
+    new_purl = PackageURL(
+        type=purl.type,
+        namespace=purl.namespace,
+        name=purl.name,
+        version=commit_hash,
+        qualifiers=purl.qualifiers,
+    )
+
+    return purl2url.get_commit_url(str(new_purl))
+
+
+def generate_patch_url(vcs_url, commit_hash):
+    """
+    Generate patch URL from VCS URL and commit hash.
+    """
+    from packageurl.contrib import purl2url
+
+    if not vcs_url or not commit_hash:
+        return
+
+    purl = url2purl(vcs_url)
+    if not purl:
+        return
+
+    new_purl = PackageURL(
+        type=purl.type,
+        namespace=purl.namespace,
+        name=purl.name,
+        version=commit_hash,
+        qualifiers=purl.qualifiers,
+    )
+
+    return purl2url.get_patch_url(str(new_purl))