|
30 | 30 | from vulnerabilities.models import AdvisorySeverity |
31 | 31 | from vulnerabilities.models import AdvisoryV2 |
32 | 32 | from vulnerabilities.models import AdvisoryWeakness |
| 33 | +from vulnerabilities.models import DetectionRule |
33 | 34 | from vulnerabilities.models import Group |
34 | 35 | from vulnerabilities.models import GroupedAdvisory |
35 | 36 | from vulnerabilities.models import ImpactedPackageAffecting |
@@ -704,3 +705,47 @@ def get_fixing_advisories_bulk(packages): |
704 | 705 | result[package.id] = grouped |
705 | 706 |
|
706 | 707 | return result |
| 708 | + |
| 709 | + |
| 710 | +class DetectionRuleFilter(filters.FilterSet): |
| 711 | + advisory_avid = filters.CharFilter(field_name="related_advisories__avid", lookup_expr="exact") |
| 712 | + |
| 713 | + rule_text_contains = filters.CharFilter(field_name="rule_text", lookup_expr="icontains") |
| 714 | + |
| 715 | + class Meta: |
| 716 | + model = DetectionRule |
| 717 | + fields = ["rule_type"] |
| 718 | + |
| 719 | + |
| 720 | +class DetectionRuleSerializer(serializers.ModelSerializer): |
| 721 | + advisory_avid = serializers.SerializerMethodField() |
| 722 | + |
| 723 | + class Meta: |
| 724 | + model = DetectionRule |
| 725 | + fields = ["rule_type", "source_url", "rule_metadata", "rule_text", "advisory_avid"] |
| 726 | + |
| 727 | + def get_advisory_avid(self, obj): |
| 728 | + avids = {advisory.avid for advisory in obj.related_advisories.all()} |
| 729 | + return sorted(avids) |
| 730 | + |
| 731 | + |
| 732 | +class DetectionRuleViewSet(viewsets.ReadOnlyModelViewSet): |
| 733 | + advisories_prefetch = Prefetch( |
| 734 | + "related_advisories", queryset=AdvisoryV2.objects.only("id", "avid").distinct() |
| 735 | + ) |
| 736 | + queryset = DetectionRule.objects.prefetch_related(advisories_prefetch) |
| 737 | + serializer_class = DetectionRuleSerializer |
| 738 | + throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle] |
| 739 | + filter_backends = [filters.DjangoFilterBackend] |
| 740 | + filterset_class = DetectionRuleFilter |
| 741 | + |
| 742 | + def get_queryset(self): |
| 743 | + queryset = super().get_queryset() |
| 744 | + query_params = ["advisory_avid", "rule_text_contains", "rule_type"] |
| 745 | + has_query_params = any( |
| 746 | + query_param in self.request.query_params for query_param in query_params |
| 747 | + ) |
| 748 | + if not has_query_params: |
| 749 | + return queryset.none() |
| 750 | + |
| 751 | + return queryset |
0 commit comments