Skip to content

Commit 24423fa

Browse files
shivamshrma09shivamshrma09
committed
Add AdvisoryV2 API endpoint to api/v2/ #2224
Signed-off-by: shivamshrma09 <shivamsharma27107@gmail.com>
1 parent 6df203d commit 24423fa

File tree

3 files changed

+208
-0
lines changed

3 files changed

+208
-0
lines changed

‎vulnerabilities/api_v2.py‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1025,6 +1025,83 @@ def get_view_name(self):
10251025
return "Pipeline Jobs"
10261026

10271027

1028+
class AdvisoryV2FilterSet(filters.FilterSet):
1029+
alias = CharInFilter(
1030+
field_name="aliases__alias",
1031+
lookup_expr="in",
1032+
label="Alias",
1033+
help_text="Filter by one or more aliases (e.g. CVE-2021-1234). Multi-value supported (comma-separated).",
1034+
)
1035+
advisory_id = CharInFilter(
1036+
field_name="avid",
1037+
lookup_expr="in",
1038+
label="Advisory ID",
1039+
help_text="Filter by one or more advisory IDs (avid). Multi-value supported (comma-separated).",
1040+
)
1041+
datasource_id = filters.CharFilter(
1042+
field_name="datasource_id",
1043+
label="Datasource ID",
1044+
help_text="Filter by datasource ID (e.g. nginx_importer_v2).",
1045+
)
1046+
1047+
class Meta:
1048+
model = AdvisoryV2
1049+
fields = ["alias", "advisory_id", "datasource_id"]
1050+
1051+
1052+
@extend_schema_view(
1053+
list=extend_schema(
1054+
parameters=[
1055+
OpenApiParameter(
1056+
name="alias",
1057+
description="Filter by one or more aliases (e.g. CVE-2021-1234). Comma-separated.",
1058+
required=False,
1059+
type={"type": "array", "items": {"type": "string"}},
1060+
location=OpenApiParameter.QUERY,
1061+
),
1062+
OpenApiParameter(
1063+
name="advisory_id",
1064+
description="Filter by one or more advisory IDs (avid). Comma-separated.",
1065+
required=False,
1066+
type={"type": "array", "items": {"type": "string"}},
1067+
location=OpenApiParameter.QUERY,
1068+
),
1069+
OpenApiParameter(
1070+
name="datasource_id",
1071+
description="Filter by datasource ID.",
1072+
required=False,
1073+
type=str,
1074+
location=OpenApiParameter.QUERY,
1075+
),
1076+
]
1077+
)
1078+
)
1079+
class AdvisoryV2ViewSet(viewsets.ReadOnlyModelViewSet):
1080+
"""
1081+
Lookup for advisories by advisory ID, alias, or datasource.
1082+
"""
1083+
1084+
queryset = (
1085+
AdvisoryV2.objects.prefetch_related(
1086+
"aliases",
1087+
"references",
1088+
"severities",
1089+
"weaknesses",
1090+
"related_ssvcs",
1091+
"source_ssvcs",
1092+
)
1093+
.order_by("datasource_id", "advisory_id")
1094+
.distinct()
1095+
)
1096+
serializer_class = AdvisoryV2Serializer
1097+
lookup_field = "avid"
1098+
# avid contains slashes (e.g. nginx_importer_v2/CVE-2021-1234)
1099+
lookup_value_regex = r"[^/]+/[^/]+"
1100+
filter_backends = [filters.DjangoFilterBackend]
1101+
filterset_class = AdvisoryV2FilterSet
1102+
throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
1103+
1104+
10281105
class PackageV3ViewSet(viewsets.ReadOnlyModelViewSet):
10291106
queryset = PackageV2.objects.all()
10301107
serializer_class = PackageV3Serializer

‎vulnerabilities/tests/test_api_v2.py‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,10 @@
1717
from rest_framework.test import APIClient
1818
from rest_framework.test import APITestCase
1919

20+
from vulnerabilities.api_v2 import AdvisoryV2Serializer
2021
from vulnerabilities.api_v2 import PackageV2Serializer
2122
from vulnerabilities.api_v2 import VulnerabilityListSerializer
23+
from vulnerabilities.models import AdvisoryAlias
2224
from vulnerabilities.models import AdvisoryV2
2325
from vulnerabilities.models import Alias
2426
from vulnerabilities.models import ApiUser
@@ -903,3 +905,130 @@ def test_get_all_vulnerable_purls(self):
903905
response = self.client.get(url)
904906
assert response.status_code == 200
905907
assert "pkg:pypi/sample@1.0.0" in response.data
908+
909+
910+
class AdvisoryV2ViewSetTest(APITestCase):
911+
def setUp(self):
912+
self.advisory1 = AdvisoryV2.objects.create(
913+
datasource_id="nginx_importer_v2",
914+
advisory_id="CVE-2021-1234",
915+
avid="nginx_importer_v2/CVE-2021-1234",
916+
unique_content_id="a" * 64,
917+
url="https://example.com/advisory1",
918+
date_collected="2024-01-01T00:00:00Z",
919+
summary="Test advisory 1",
920+
)
921+
self.advisory2 = AdvisoryV2.objects.create(
922+
datasource_id="pypa_importer_v2",
923+
advisory_id="PYSEC-2022-5678",
924+
avid="pypa_importer_v2/PYSEC-2022-5678",
925+
unique_content_id="b" * 64,
926+
url="https://example.com/advisory2",
927+
date_collected="2024-01-01T00:00:00Z",
928+
summary="Test advisory 2",
929+
)
930+
931+
self.alias1 = AdvisoryAlias.objects.create(alias="CVE-2021-1234")
932+
self.advisory1.aliases.add(self.alias1)
933+
934+
self.alias2 = AdvisoryAlias.objects.create(alias="GHSA-xxxx-yyyy-zzzz")
935+
self.advisory2.aliases.add(self.alias2)
936+
937+
cache.clear()
938+
self.client = APIClient(enforce_csrf_checks=True)
939+
940+
def test_list_advisories(self):
941+
"""
942+
Test listing all advisories without filters.
943+
"""
944+
url = reverse("advisory-v2-list")
945+
response = self.client.get(url, format="json")
946+
self.assertEqual(response.status_code, status.HTTP_200_OK)
947+
self.assertIn("results", response.data)
948+
self.assertEqual(response.data["count"], 2)
949+
950+
def test_retrieve_advisory_by_avid(self):
951+
"""
952+
Test retrieving a specific advisory by its avid.
953+
The avid contains a slash, handled by lookup_value_regex.
954+
"""
955+
url = reverse("advisory-v2-detail", kwargs={"avid": self.advisory1.avid})
956+
response = self.client.get(url, format="json")
957+
self.assertEqual(response.status_code, status.HTTP_200_OK)
958+
self.assertEqual(response.data["advisory_id"], self.advisory1.avid)
959+
self.assertEqual(response.data["url"], self.advisory1.url)
960+
self.assertIn("CVE-2021-1234", response.data["aliases"])
961+
962+
def test_retrieve_nonexistent_advisory_returns_404(self):
963+
"""
964+
Test that a non-existent advisory returns 404.
965+
"""
966+
url = reverse("advisory-v2-detail", kwargs={"avid": "fake_source/FAKE-0000"})
967+
response = self.client.get(url, format="json")
968+
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
969+
970+
def test_filter_by_alias(self):
971+
"""
972+
Test filtering advisories by alias returns only matching advisory.
973+
"""
974+
url = reverse("advisory-v2-list")
975+
response = self.client.get(url, {"alias": "CVE-2021-1234"}, format="json")
976+
self.assertEqual(response.status_code, status.HTTP_200_OK)
977+
self.assertEqual(response.data["count"], 1)
978+
result = response.data["results"][0]
979+
self.assertIn("CVE-2021-1234", result["aliases"])
980+
981+
def test_filter_by_advisory_id(self):
982+
"""
983+
Test filtering advisories by advisory_id (avid).
984+
"""
985+
url = reverse("advisory-v2-list")
986+
response = self.client.get(
987+
url, {"advisory_id": "nginx_importer_v2/CVE-2021-1234"}, format="json"
988+
)
989+
self.assertEqual(response.status_code, status.HTTP_200_OK)
990+
self.assertEqual(response.data["count"], 1)
991+
self.assertEqual(response.data["results"][0]["advisory_id"], self.advisory1.avid)
992+
993+
def test_filter_by_datasource_id(self):
994+
"""
995+
Test filtering advisories by datasource_id returns only that source's advisories.
996+
"""
997+
url = reverse("advisory-v2-list")
998+
response = self.client.get(url, {"datasource_id": "nginx_importer_v2"}, format="json")
999+
self.assertEqual(response.status_code, status.HTTP_200_OK)
1000+
self.assertEqual(response.data["count"], 1)
1001+
self.assertEqual(response.data["results"][0]["advisory_id"], self.advisory1.avid)
1002+
1003+
def test_filter_by_nonexistent_alias_returns_empty(self):
1004+
"""
1005+
Test that filtering by a non-existent alias returns an empty list.
1006+
"""
1007+
url = reverse("advisory-v2-list")
1008+
response = self.client.get(url, {"alias": "CVE-9999-9999"}, format="json")
1009+
self.assertEqual(response.status_code, status.HTTP_200_OK)
1010+
self.assertEqual(response.data["count"], 0)
1011+
1012+
def test_advisory_serializer_fields(self):
1013+
"""
1014+
Test that AdvisoryV2Serializer returns all required fields.
1015+
"""
1016+
serializer = AdvisoryV2Serializer(self.advisory1)
1017+
data = serializer.data
1018+
expected_fields = [
1019+
"advisory_id",
1020+
"url",
1021+
"aliases",
1022+
"summary",
1023+
"severities",
1024+
"weaknesses",
1025+
"references",
1026+
"exploitability",
1027+
"weighted_severity",
1028+
"risk_score",
1029+
"related_ssvc_trees",
1030+
]
1031+
for field in expected_fields:
1032+
self.assertIn(field, data)
1033+
self.assertEqual(data["advisory_id"], self.advisory1.avid)
1034+
self.assertIn("CVE-2021-1234", data["aliases"])

‎vulnerablecode/urls.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
from vulnerabilities.api import CPEViewSet
2121
from vulnerabilities.api import PackageViewSet
2222
from vulnerabilities.api import VulnerabilityViewSet
23+
from vulnerabilities.api_v2 import AdvisoryV2ViewSet
2324
from vulnerabilities.api_v2 import CodeFixV2ViewSet
2425
from vulnerabilities.api_v2 import CodeFixViewSet
2526
from vulnerabilities.api_v2 import PackageV2ViewSet
@@ -66,6 +67,7 @@ def __init__(self, *args, **kwargs):
6667
api_v2_router.register("codefixes", CodeFixViewSet, basename="codefix")
6768
api_v2_router.register("pipelines", PipelineScheduleV2ViewSet, basename="pipelines")
6869
api_v2_router.register("advisory-codefixes", CodeFixV2ViewSet, basename="advisory-codefix")
70+
api_v2_router.register("advisories", AdvisoryV2ViewSet, basename="advisory-v2")
6971

7072
api_v3_router = OptionalSlashRouter()
7173

0 commit comments

Comments
 (0)