(self, request, *args, **kwargs)
| 258 | |
| 259 | @extend_schema(request=PackageQuerySerializer) |
| 260 | def create(self, request, *args, **kwargs): |
| 261 | serializer = PackageQuerySerializer(data=request.data) |
| 262 | serializer.is_valid(raise_exception=True) |
| 263 | |
| 264 | purls = serializer.validated_data["purls"] |
| 265 | details = serializer.validated_data["details"] |
| 266 | reachability = serializer.validated_data["reachability"] |
| 267 | ignore_qualifiers_subpath = serializer.validated_data["ignore_qualifiers_subpath"] |
| 268 | max_advisories = serializer.validated_data["max_advisories"] |
| 269 | |
| 270 | if not purls: |
| 271 | query = PackageV2.objects.all_vulnerable_purls().order_by("package_url") |
| 272 | page = self.paginate_queryset(query) |
| 273 | return self.get_paginated_response(page) |
| 274 | |
| 275 | plain_purls = None |
| 276 | |
| 277 | if ignore_qualifiers_subpath: |
| 278 | plain_purls = [ |
| 279 | str( |
| 280 | PackageURL( |
| 281 | type=p.type, |
| 282 | namespace=p.namespace, |
| 283 | name=p.name, |
| 284 | version=p.version, |
| 285 | ) |
| 286 | ) |
| 287 | for p in map(PackageURL.from_string, purls) |
| 288 | ] |
| 289 | |
| 290 | if not details: |
| 291 | affecting_exists = ImpactedPackageAffecting.objects.filter( |
| 292 | package_id=OuterRef("pk"), |
| 293 | impacted_package__advisory___all_impacts_unfurled_at__isnull=False, |
| 294 | impacted_package__advisory__is_latest=True, |
| 295 | ) |
| 296 | # Return back vulnerable PURLs only |
| 297 | if ignore_qualifiers_subpath: |
| 298 | query = ( |
| 299 | PackageV2.objects.filter_plain_purls(plain_purls) |
| 300 | .values_list("plain_package_url", flat=True) |
| 301 | .order_by("plain_package_url") |
| 302 | ) |
| 303 | else: |
| 304 | query = ( |
| 305 | PackageV2.objects.filter_purls(purls) |
| 306 | .order_by("package_url") |
| 307 | .values_list("package_url", flat=True) |
| 308 | ) |
| 309 | |
| 310 | query = query.annotate( |
| 311 | is_vulnerable=Exists(affecting_exists), |
| 312 | ).filter(is_vulnerable=True) |
| 313 | |
| 314 | page = self.paginate_queryset(query) |
| 315 | return self.get_paginated_response(page) |
| 316 | |
| 317 | if ignore_qualifiers_subpath: |
nothing calls this directly
no test coverage detected