Add advisory codefix V2 URL

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

Author: TG1999

CHANGELOG.rst

@@ -1,6 +1,15 @@
 Release notes
 =============
 
+Version v37.0.0
+---------------------
+
+- This is a major version, this version introduces Advisory level details.
+- We have added new models AdvisoryV2, AdvisoryAlias, AdvisoryReference, AdvisorySeverity, AdvisoryWeakness, PackageV2 and CodeFixV2.
+- We are using ``avid`` as an internal advisory ID for uniquely identifying advisories.
+- We have a new route ``/v2`` which only support package search which has information on packages that are reported to be affected or fixing by advisories.
+- This version introduces ``/api/v2/advisories-packages`` which has information on packages that are reported to be affected or fixing by advisories.
+
 Version v36.1.3
 ---------------------
 

vulnerabilities/api_v2.py

@@ -331,7 +331,7 @@ def get_affected_by_vulnerabilities(self, obj):
             # Get code fixed for a vulnerability
             code_fixes = CodeFixV2.objects.filter(advisory=adv).distinct()
             code_fix_urls = [
-                reverse("codefix-detail", args=[code_fix.id], request=request)
+                reverse("advisory-codefix-detail", args=[code_fix.id], request=request)
                 for code_fix in code_fixes
             ]
 
@@ -714,6 +714,58 @@ class Meta:
         read_only_fields = ["created_at", "updated_at"]
 
 
+class CodeFixV2Serializer(serializers.ModelSerializer):
+    """
+    Serializer for the CodeFix model.
+    Provides detailed information about a code fix.
+    """
+
+    affected_advisory_id = serializers.CharField(
+        source="advisory.avid",
+        read_only=True,
+        help_text="ID of the advisory affecting the package.",
+    )
+    affected_package_purl = serializers.CharField(
+        source="affected_package.package_url",
+        read_only=True,
+        help_text="PURL of the affected package.",
+    )
+    fixed_package_purl = serializers.CharField(
+        source="fixed_package.package_url",
+        read_only=True,
+        help_text="PURL of the fixing package (if available).",
+    )
+    created_at = serializers.DateTimeField(
+        format="%Y-%m-%dT%H:%M:%SZ",
+        read_only=True,
+        help_text="Timestamp when the code fix was created.",
+    )
+    updated_at = serializers.DateTimeField(
+        format="%Y-%m-%dT%H:%M:%SZ",
+        read_only=True,
+        help_text="Timestamp when the code fix was last updated.",
+    )
+
+    class Meta:
+        model = CodeFixV2
+        fields = [
+            "id",
+            "commits",
+            "pulls",
+            "downloads",
+            "patch",
+            "affected_advisory_id",
+            "affected_package_purl",
+            "fixed_package_purl",
+            "notes",
+            "references",
+            "is_reviewed",
+            "created_at",
+            "updated_at",
+        ]
+        read_only_fields = ["created_at", "updated_at"]
+
+
 class CodeFixViewSet(viewsets.ReadOnlyModelViewSet):
     """
     API endpoint that allows viewing CodeFix entries.
@@ -735,6 +787,25 @@ def get_queryset(self):
         return queryset
 
 
+class CodeFixV2ViewSet(viewsets.ReadOnlyModelViewSet):
+    """
+    API endpoint that allows viewing CodeFix entries.
+    """
+
+    queryset = CodeFixV2.objects.all()
+    serializer_class = CodeFixV2Serializer
+
+    def get_queryset(self):
+        """
+        Optionally filter by vulnerability ID.
+        """
+        queryset = super().get_queryset()
+        advisory_id = self.request.query_params.get("advisory_id")
+        if advisory_id:
+            queryset = queryset.filter(advisory__avid=advisory_id)
+        return queryset
+
+
 class CreateListRetrieveUpdateViewSet(
     mixins.CreateModelMixin,
     mixins.ListModelMixin,
@@ -1043,10 +1114,10 @@ def bulk_search(self, request):
             # Collect vulnerabilities associated with these packages
             advisories = set()
             for package in packages:
-                advisories.update(package.affected_by_vulnerabilities.all())
-                advisories.update(package.fixing_vulnerabilities.all())
+                advisories.update(package.affected_by_advisories.all())
+                advisories.update(package.fixing_advisories.all())
 
-            advisory_data = {adv.avid: VulnerabilityV2Serializer(adv).data for adv in advisories}
+            advisory_data = {adv.avid: AdvisoryV2Serializer(adv).data for adv in advisories}
 
             if not purl_only:
                 package_data = AdvisoryPackageV2Serializer(
@@ -1071,8 +1142,8 @@ def bulk_search(self, request):
         # Collect vulnerabilities associated with these packages
         advisories = set()
         for package in packages:
-            advisories.update(package.affected_by_vulnerabilities.all())
-            advisories.update(package.fixing_vulnerabilities.all())
+            advisories.update(package.affected_by_advisories.all())
+            advisories.update(package.fixing_advisories.all())
 
         advisory_data = {adv.advisory_id: AdvisoryV2Serializer(adv).data for adv in advisories}
 

vulnerabilities/models.py

@@ -221,6 +221,14 @@ class VulnerabilityStatusType(models.IntegerChoices):
     INVALID = 3, "Invalid"
 
 
+class AdvisoryStatusType(models.IntegerChoices):
+    """List of vulnerability statuses."""
+
+    PUBLISHED = 1, "Published"
+    DISPUTED = 2, "Disputed"
+    INVALID = 3, "Invalid"
+
+
 # FIXME: Remove when migration from Vulnerability to Advisory is completed
 class Vulnerability(models.Model):
     """
@@ -2713,7 +2721,7 @@ class AdvisoryV2(models.Model):
     )
 
     status = models.IntegerField(
-        choices=VulnerabilityStatusType.choices, default=VulnerabilityStatusType.PUBLISHED
+        choices=AdvisoryStatusType.choices, default=AdvisoryStatusType.PUBLISHED
     )
 
     exploitability = models.DecimalField(
@@ -2764,7 +2772,7 @@ def get_absolute_url(self):
         """
         Return this Vulnerability details absolute URL.
         """
-        return reverse("advisory_details", args=[self.id])
+        return reverse("advisory_details", args=[self.avid])
 
     def to_advisory_data(self) -> "AdvisoryDataV2":
         from vulnerabilities.importer import AdvisoryDataV2
@@ -2950,6 +2958,12 @@ def _vulnerable(self, vulnerable=True):
         """
         return self.with_is_vulnerable().filter(is_vulnerable=vulnerable)
 
+    def vulnerable(self):
+        """
+        Return only packages that are vulnerable.
+        """
+        return self.filter(affected_by_advisories__isnull=False)
+
     def with_is_vulnerable(self):
         """
         Annotate Package with ``is_vulnerable`` boolean attribute.
@@ -2958,6 +2972,12 @@ def with_is_vulnerable(self):
             is_vulnerable=Exists(AdvisoryV2.objects.filter(affecting_packages__pk=OuterRef("pk")))
         )
 
+    def from_purl(self, purl: Union[PackageURL, str]):
+        """
+        Return a new Package given a ``purl`` PackageURL object or PURL string.
+        """
+        return PackageV2.objects.create(**purl_to_dict(purl=purl))
+
 
 class PackageV2(PackageURLMixin):
     """

vulnerabilities/templates/package_details_v2.html

@@ -146,7 +146,7 @@
                                 {% for advisory in affected_by_advisories %}
                                 <tr>
                                     <td>
-                                        <a href="/advisories/{{advisory.id}}">
+                                        <a href="{{advisory.get_absolute_url}}">
                                             {{advisory.avid }}
                                         </a>
                                         <br />
@@ -267,7 +267,7 @@
                                 {% for advisory in fixing_advisories %}
                                 <tr>
                                     <td>
-                                        <a href="/advisories/{{advisory.id}}">
+                                        <a href="{{advisory.get_absolute_url}}">
                                             {{advisory.avid }}
                                         </a>
                                     </td>

vulnerabilities/tests/test_api_v2.py

@@ -19,9 +19,12 @@
 
 from vulnerabilities.api_v2 import PackageV2Serializer
 from vulnerabilities.api_v2 import VulnerabilityListSerializer
+from vulnerabilities.models import AdvisoryV2
 from vulnerabilities.models import Alias
 from vulnerabilities.models import ApiUser
+from vulnerabilities.models import CodeFixV2
 from vulnerabilities.models import Package
+from vulnerabilities.models import PackageV2
 from vulnerabilities.models import PipelineRun
 from vulnerabilities.models import PipelineSchedule
 from vulnerabilities.models import Vulnerability
@@ -786,3 +789,120 @@ def test_schedule_update_with_staff_session_permitted(self, mock_create_new_job)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertNotEqual(response.status_code, status.HTTP_403_FORBIDDEN)
         self.assertEqual(self.schedule1.run_interval, 2)
+
+
+class CodeFixV2APITest(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:pypi/affected_package@1.0.0")
+        self.fixed_package = PackageV2.objects.from_purl(purl="pkg:pypi/fixed_package@1.0.1")
+
+        self.codefix = CodeFixV2.objects.create(
+            advisory=self.advisory,
+            affected_package=self.affected_package,
+            fixed_package=self.fixed_package,
+            notes="Security patch",
+            is_reviewed=True,
+        )
+        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("advisory-codefix-list")
+
+    def test_list_all_codefixes(self):
+        response = self.client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["count"] == 1
+        assert response.data["results"][0]["affected_advisory_id"] == self.advisory.avid
+
+    def test_filter_codefix_by_advisory_id_success(self):
+        response = self.client.get(self.url, {"advisory_id": self.advisory.avid})
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["count"] == 1
+        assert response.data["results"][0]["affected_advisory_id"] == self.advisory.avid
+
+    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 AdvisoriesPackageV2Tests(APITestCase):
+    def setUp(self):
+        self.advisory = AdvisoryV2.objects.create(
+            datasource_id="ghsa",
+            advisory_id="GHSA-1234",
+            avid="ghsa/GHSA-1234",
+            unique_content_id="f" * 64,
+            url="https://example.com/advisory",
+            date_collected="2025-07-01T00:00:00Z",
+        )
+
+        self.package = PackageV2.objects.from_purl(purl="pkg:pypi/sample@1.0.0")
+
+        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.package.affected_by_advisories.add(self.advisory)
+        self.package.save()
+
+    def test_list_with_purl_filter(self):
+        url = reverse("advisories-package-v2-list")
+        with self.assertNumQueries(16):
+            response = self.client.get(url, {"purl": "pkg:pypi/sample@1.0.0"})
+        assert response.status_code == 200
+        assert "packages" in response.data["results"]
+        assert "advisories" in response.data["results"]
+        assert self.advisory.avid in response.data["results"]["advisories"]
+
+    def test_bulk_lookup(self):
+        url = reverse("advisories-package-v2-bulk-lookup")
+        with self.assertNumQueries(11):
+            response = self.client.post(url, {"purls": ["pkg:pypi/sample@1.0.0"]}, format="json")
+        assert response.status_code == 200
+        assert "packages" in response.data
+        assert "advisories" in response.data
+        assert self.advisory.avid in response.data["advisories"]
+
+    def test_bulk_search_plain(self):
+        url = reverse("advisories-package-v2-bulk-search")
+        payload = {"purls": ["pkg:pypi/sample@1.0.0"], "plain_purl": True, "purl_only": False}
+        with self.assertNumQueries(11):
+            response = self.client.post(url, payload, format="json")
+        assert response.status_code == 200
+        assert "packages" in response.data
+        assert "advisories" in response.data
+
+    def test_bulk_search_purl_only(self):
+        url = reverse("advisories-package-v2-bulk-search")
+        payload = {"purls": ["pkg:pypi/sample@1.0.0"], "plain_purl": False, "purl_only": True}
+        with self.assertNumQueries(11):
+            response = self.client.post(url, payload, format="json")
+        assert response.status_code == 200
+        assert "pkg:pypi/sample@1.0.0" in response.data
+
+    def test_lookup_single_package(self):
+        url = reverse("advisories-package-v2-lookup")
+        with self.assertNumQueries(9):
+            response = self.client.post(url, {"purl": "pkg:pypi/sample@1.0.0"}, format="json")
+        assert response.status_code == 200
+        assert any(pkg["purl"] == "pkg:pypi/sample@1.0.0" for pkg in response.data)
+
+    def test_get_all_vulnerable_purls(self):
+        url = reverse("advisories-package-v2-all")
+        with self.assertNumQueries(4):
+            response = self.client.get(url)
+        assert response.status_code == 200
+        assert "pkg:pypi/sample@1.0.0" in response.data

vulnerabilities/views.py

@@ -325,8 +325,8 @@ def get_context_data(self, **kwargs):
 class AdvisoryDetails(DetailView):
     model = models.AdvisoryV2
     template_name = "advisory_detail.html"
-    slug_url_kwarg = "id"
-    slug_field = "id"
+    slug_url_kwarg = "avid"
+    slug_field = "avid"
 
     def get_queryset(self):
         return (

vulnerablecode/urls.py

@@ -21,6 +21,7 @@
 from vulnerabilities.api import PackageViewSet
 from vulnerabilities.api import VulnerabilityViewSet
 from vulnerabilities.api_v2 import AdvisoriesPackageV2ViewSet
+from vulnerabilities.api_v2 import CodeFixV2ViewSet
 from vulnerabilities.api_v2 import CodeFixViewSet
 from vulnerabilities.api_v2 import PackageV2ViewSet
 from vulnerabilities.api_v2 import PipelineScheduleV2ViewSet
@@ -66,6 +67,7 @@ def __init__(self, *args, **kwargs):
 )
 api_v2_router.register("vulnerabilities", VulnerabilityV2ViewSet, basename="vulnerability-v2")
 api_v2_router.register("codefixes", CodeFixViewSet, basename="codefix")
+api_v2_router.register("advisory-codefixes", CodeFixV2ViewSet, basename="advisory-codefix")
 api_v2_router.register("schedule", PipelineScheduleV2ViewSet, basename="schedule")
 
 
@@ -102,7 +104,7 @@ def __init__(self, *args, **kwargs):
         name="home",
     ),
     path(
-        "advisories/<int:id>",
+        "advisories/<path:avid>",
         AdvisoryDetails.as_view(),
         name="advisory_details",
     ),