diff --git a/dejacode/urls.py b/dejacode/urls.py index 5b381868..eb656555 100644 --- a/dejacode/urls.py +++ b/dejacode/urls.py @@ -55,6 +55,8 @@ from reporting.api import ReportViewSet from vulnerabilities.api import VulnerabilityAnalysisViewSet from vulnerabilities.api import VulnerabilityViewSet +from vulnerabilities.triage.api import AnalysisPresetViewSet +from vulnerabilities.triage.api import TriageRulesetViewSet from workflow.api import RequestTemplateViewSet from workflow.api import RequestViewSet @@ -83,6 +85,8 @@ api_router.register("usage_policies", UsagePolicyViewSet) api_router.register("vulnerabilities", VulnerabilityViewSet) api_router.register("vulnerability_analyses", VulnerabilityAnalysisViewSet) +api_router.register("triage_rulesets", TriageRulesetViewSet) +api_router.register("analysis_presets", AnalysisPresetViewSet) urlpatterns = [ diff --git a/product_portfolio/api.py b/product_portfolio/api.py index 5c74d9c0..571012dc 100644 --- a/product_portfolio/api.py +++ b/product_portfolio/api.py @@ -7,6 +7,8 @@ # from django.core.exceptions import ValidationError +from django.db import transaction +from django.shortcuts import get_object_or_404 import django_filters from rest_framework import permissions @@ -51,6 +53,11 @@ from product_portfolio.models import ProductPolicyViolation from product_portfolio.models import ScanCodeProject from vulnerabilities.api import VulnerabilityAnalysisSerializer +from vulnerabilities.triage.engine import delete_triage_records_for_assignment +from vulnerabilities.triage.engine import reevaluate_product_rulesets +from vulnerabilities.triage.models import ProductTriageRuleset +from vulnerabilities.triage.models import TriageRecord +from vulnerabilities.triage.models import TriageRuleset base_extra_kwargs = { "licenses": { @@ -363,6 +370,37 @@ class Meta: ) +class TriageRecordSerializer(serializers.ModelSerializer): + advisory_id = serializers.ReadOnlyField(source="vulnerability.advisory_id") + ruleset = serializers.ReadOnlyField(source="ruleset.name") + request = serializers.StringRelatedField() + + class Meta: + model = TriageRecord + fields = ( + "advisory_id", + "ruleset", + "recommended_action", + "matched_rules", + "request", + "detected_date", + "last_checked", + ) + + +class TriageRulesetAssignmentSerializer(serializers.Serializer): + uuid = serializers.UUIDField(read_only=True) + name = serializers.CharField(read_only=True) + recommended_action = serializers.CharField(read_only=True) + precedence = serializers.IntegerField(read_only=True) + assigned = serializers.BooleanField(read_only=True) + + +class AssignTriageRulesetSerializer(serializers.Serializer): + ruleset = serializers.UUIDField() + assigned = serializers.BooleanField() + + class ProductViewSet( ObjectPermissionsMixin, SendAboutFilesMixin, @@ -441,6 +479,69 @@ def policy_violations(self, request, uuid): serializer = ProductPolicyViolationSerializer(violations, many=True) return Response(serializer.data) + @action(detail=True, url_path="triage_records") + def triage_records(self, request, uuid): + """List active triage recommendations for this product, one per vulnerability.""" + product = self.get_object() + records = product.triage_records.highest_precedence().select_related( + "vulnerability", "ruleset", "request" + ) + serializer = TriageRecordSerializer(records, many=True) + return Response(serializer.data) + + @action( + detail=True, + methods=["get", "post"], + url_path="manage_triage_rulesets", + serializer_class=AssignTriageRulesetSerializer, + ) + def manage_triage_rulesets(self, request, uuid): + """ + GET: list every enabled triage ruleset in this product's dataspace, each flagged + with whether it is currently assigned to this product. + + POST: assign or unassign a single ruleset for this product. + Body: {"ruleset": "", "assigned": true} + """ + product = self.get_object() + + if request.method == "POST": + serializer = AssignTriageRulesetSerializer(data=request.data) + if not serializer.is_valid(): + return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) + + ruleset = get_object_or_404( + TriageRuleset.objects.scope(product.dataspace).filter(enabled=True), + uuid=serializer.validated_data["ruleset"], + ) + assigned = serializer.validated_data["assigned"] + + with transaction.atomic(): + assignment = ProductTriageRuleset.objects.filter( + product=product, ruleset=ruleset + ).first() + if assigned and not assignment: + ProductTriageRuleset.objects.create( + product=product, ruleset=ruleset, dataspace=product.dataspace + ) + reevaluate_product_rulesets(product) + elif not assigned and assignment: + assignment.delete() + delete_triage_records_for_assignment(ruleset=ruleset, product=product) + reevaluate_product_rulesets(product) + return Response(status=status.HTTP_200_OK) + + assigned_ruleset_ids = set( + product.product_triage_rulesets.values_list("ruleset_id", flat=True) + ) + rulesets = TriageRuleset.objects.filter(dataspace=product.dataspace, enabled=True).order_by( + "-precedence", "name" + ) + for ruleset in rulesets: + ruleset.assigned = ruleset.id in assigned_ruleset_ids + serializer = TriageRulesetAssignmentSerializer(rulesets, many=True) + return Response(serializer.data) + @action(detail=True, methods=["post"], serializer_class=LoadSBOMsFormSerializer) def load_sboms(self, request, *args, **kwargs): """ diff --git a/product_portfolio/tests/test_api.py b/product_portfolio/tests/test_api.py index 23803db8..e68bb5fd 100644 --- a/product_portfolio/tests/test_api.py +++ b/product_portfolio/tests/test_api.py @@ -24,6 +24,7 @@ from component_catalog.models import Component from component_catalog.models import ComponentKeyword from component_catalog.models import Package +from component_catalog.tests import make_package from dje.models import Dataspace from dje.models import History from dje.tests import MaxQueryMixin @@ -47,8 +48,14 @@ from product_portfolio.models import ProductRelationStatus from product_portfolio.models import ProductStatus from product_portfolio.models import ScanCodeProject +from product_portfolio.tests import make_product_package from vulnerabilities.tests import make_vulnerability from vulnerabilities.tests import make_vulnerability_analysis +from vulnerabilities.triage.engine import evaluate_ruleset +from vulnerabilities.triage.models import ProductTriageRuleset +from vulnerabilities.triage.models import TriageAction +from vulnerabilities.triage.tests import make_product_triage_ruleset +from vulnerabilities.triage.tests import make_triage_ruleset class ProductAPITestCase(MaxQueryMixin, TestCase): @@ -758,6 +765,163 @@ def test_api_product_endpoint_policy_violations_action(self): self.assertEqual(status.HTTP_200_OK, response.status_code) self.assertEqual([], response.data) + def test_api_product_endpoint_triage_records_action(self): + url = reverse("api_v2:product-triage-records", args=[self.product1.uuid]) + + self.client.login(username=self.base_user.username, password="secret") + response = self.client.get(url) + self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code) + + add_perm(self.base_user, "add_product") + assign_perm("view_product", self.base_user, self.product1) + + response = self.client.get(url) + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual([], response.data) + + package = make_package(self.dataspace) + make_product_package(self.product1, package=package) + vulnerability = make_vulnerability(self.dataspace, affecting=package, risk_score=9.0) + ruleset = make_triage_ruleset( + self.dataspace, + recommended_action=TriageAction.UPGRADE, + rules_config={"risk_score": {"is_active": True, "min_risk_score": 8.0}}, + ) + make_product_triage_ruleset(self.product1, ruleset=ruleset) + evaluate_ruleset(ruleset, self.product1) + + response = self.client.get(url) + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual(1, len(response.data)) + entry = response.data[0] + self.assertEqual(vulnerability.advisory_id, entry["advisory_id"]) + self.assertEqual(ruleset.name, entry["ruleset"]) + self.assertEqual(TriageAction.UPGRADE, entry["recommended_action"]) + self.assertEqual(["risk_score"], entry["matched_rules"]) + self.assertIsNone(entry["request"]) + self.assertIn("detected_date", entry) + self.assertIn("last_checked", entry) + + def test_api_product_endpoint_manage_triage_rulesets_get(self): + url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid]) + self.client.login(username=self.base_user.username, password="secret") + response = self.client.get(url) + self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code) + + add_perm(self.base_user, "add_product") + assign_perm("view_product", self.base_user, self.product1) + + assigned_ruleset = make_triage_ruleset(self.dataspace, name="Assigned Ruleset") + make_product_triage_ruleset(self.product1, ruleset=assigned_ruleset) + unassigned_ruleset = make_triage_ruleset(self.dataspace, name="Unassigned Ruleset") + make_triage_ruleset(self.dataspace, name="Disabled Ruleset", enabled=False) + + response = self.client.get(url) + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual(2, len(response.data)) + entries_by_name = {entry["name"]: entry for entry in response.data} + self.assertTrue(entries_by_name[assigned_ruleset.name]["assigned"]) + self.assertFalse(entries_by_name[unassigned_ruleset.name]["assigned"]) + self.assertNotIn("Disabled Ruleset", entries_by_name) + + def test_api_product_endpoint_manage_triage_rulesets_post_assigns(self): + url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid]) + self.client.login(username=self.base_user.username, password="secret") + add_perm(self.base_user, "add_product") + assign_perm("view_product", self.base_user, self.product1) + assign_perm("change_product", self.base_user, self.product1) + + package = make_package(self.dataspace) + make_product_package(self.product1, package=package) + vulnerability = make_vulnerability(self.dataspace, affecting=package, risk_score=9.0) + ruleset = make_triage_ruleset( + self.dataspace, + recommended_action=TriageAction.UPGRADE, + rules_config={"risk_score": {"is_active": True, "min_risk_score": 8.0}}, + ) + + data = {"ruleset": str(ruleset.uuid), "assigned": True} + response = self.client.post(url, data=data, content_type="application/json") + + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertTrue( + ProductTriageRuleset.objects.filter(product=self.product1, ruleset=ruleset).exists() + ) + triage_record = self.product1.triage_records.get() + self.assertEqual(vulnerability, triage_record.vulnerability) + + def test_api_product_endpoint_manage_triage_rulesets_post_unassigns(self): + url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid]) + self.client.login(username=self.base_user.username, password="secret") + add_perm(self.base_user, "add_product") + assign_perm("view_product", self.base_user, self.product1) + assign_perm("change_product", self.base_user, self.product1) + + ruleset = make_triage_ruleset(self.dataspace) + make_product_triage_ruleset(self.product1, ruleset=ruleset) + + data = {"ruleset": str(ruleset.uuid), "assigned": False} + response = self.client.post(url, data=data, content_type="application/json") + + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertFalse( + ProductTriageRuleset.objects.filter(product=self.product1, ruleset=ruleset).exists() + ) + + def test_api_product_endpoint_manage_triage_rulesets_post_unassigns_form_encoded(self): + # Regression: a form-encoded "false" string must not be treated as truthy. + url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid]) + self.client.login(username=self.base_user.username, password="secret") + add_perm(self.base_user, "add_product") + assign_perm("view_product", self.base_user, self.product1) + assign_perm("change_product", self.base_user, self.product1) + + ruleset = make_triage_ruleset(self.dataspace) + make_product_triage_ruleset(self.product1, ruleset=ruleset) + + data = {"ruleset": str(ruleset.uuid), "assigned": "false"} + response = self.client.post(url, data=data) + + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertFalse( + ProductTriageRuleset.objects.filter(product=self.product1, ruleset=ruleset).exists() + ) + + def test_api_product_endpoint_manage_triage_rulesets_post_requires_both_fields(self): + url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid]) + self.client.login(username=self.base_user.username, password="secret") + add_perm(self.base_user, "add_product") + assign_perm("view_product", self.base_user, self.product1) + assign_perm("change_product", self.base_user, self.product1) + + response = self.client.post(url, data={}, content_type="application/json") + + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + + def test_api_product_endpoint_manage_triage_rulesets_post_rejects_non_dict_body(self): + url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid]) + self.client.login(username=self.base_user.username, password="secret") + add_perm(self.base_user, "add_product") + assign_perm("view_product", self.base_user, self.product1) + assign_perm("change_product", self.base_user, self.product1) + + response = self.client.post(url, data=[], content_type="application/json") + + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + + def test_api_product_endpoint_manage_triage_rulesets_post_rejects_disabled_ruleset(self): + url = reverse("api_v2:product-manage-triage-rulesets", args=[self.product1.uuid]) + self.client.login(username=self.base_user.username, password="secret") + add_perm(self.base_user, "add_product") + assign_perm("view_product", self.base_user, self.product1) + assign_perm("change_product", self.base_user, self.product1) + + ruleset = make_triage_ruleset(self.dataspace, enabled=False) + data = {"ruleset": str(ruleset.uuid), "assigned": True} + response = self.client.post(url, data=data, content_type="application/json") + + self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code) + class ProductRelatedAPITestCase(TestCase): def setUp(self): diff --git a/vulnerabilities/api.py b/vulnerabilities/api.py index b8be931b..e85335e7 100644 --- a/vulnerabilities/api.py +++ b/vulnerabilities/api.py @@ -9,6 +9,7 @@ from django.db.models import Prefetch +import django_filters from rest_framework import serializers from rest_framework import viewsets @@ -134,6 +135,9 @@ def get_queryset(self): class VulnerabilityAnalysisSerializer(DataspacedSerializer, serializers.ModelSerializer): advisory_uid = serializers.ReadOnlyField(source="vulnerability.advisory_uid") + applied_by_preset = serializers.ReadOnlyField(source="applied_by_preset.name", allow_null=True) + created_by = serializers.StringRelatedField() + last_modified_by = serializers.StringRelatedField() class Meta: model = VulnerabilityAnalysis @@ -148,6 +152,9 @@ class Meta: "responses", "detail", "is_reachable", + "applied_by_preset", + "created_by", + "last_modified_by", "first_issued", "last_updated", ) @@ -170,6 +177,9 @@ class Meta: class VulnerabilityAnalysisFilterSet(DataspacedAPIFilterSet): uuid = MultipleUUIDFilter() last_updated = LastModifiedDateFilter() + applied_by_preset__isnull = django_filters.BooleanFilter( + field_name="applied_by_preset", lookup_expr="isnull" + ) class Meta: model = VulnerabilityAnalysis @@ -203,5 +213,8 @@ def get_queryset(self): .select_related( "vulnerability", "product_package", + "applied_by_preset", + "created_by", + "last_modified_by", ) ) diff --git a/vulnerabilities/tests/test_api.py b/vulnerabilities/tests/test_api.py index ccae963b..5f0f6574 100644 --- a/vulnerabilities/tests/test_api.py +++ b/vulnerabilities/tests/test_api.py @@ -20,6 +20,9 @@ from vulnerabilities.models import VulnerabilityAnalysis from vulnerabilities.tests import make_vulnerability from vulnerabilities.tests import make_vulnerability_analysis +from vulnerabilities.triage.engine import apply_preset_for_vulnerabilities +from vulnerabilities.triage.models import AnalysisPreset +from vulnerabilities.triage.tests import make_analysis_preset class VulnerabilitiesAPITestCase(MaxQueryMixin, TestCase): @@ -154,6 +157,49 @@ def test_api_vulnerability_analysis_detail_endpoint(self): self.assertEqual(str(analysis1.uuid), response.data["uuid"]) self.assertTrue(response.data["is_reachable"]) + def test_api_vulnerability_analysis_detail_endpoint_applied_by_preset_and_authors(self): + self.client.login(username="super_user", password="secret") + + human_analysis = make_vulnerability_analysis( + self.product_package1, + self.vulnerability1, + created_by=self.super_user, + last_modified_by=self.super_user, + ) + detail_url = reverse("api_v2:vulnerabilityanalysis-detail", args=[human_analysis.uuid]) + response = self.client.get(detail_url) + self.assertIsNone(response.data["applied_by_preset"]) + self.assertEqual(self.super_user.username, response.data["created_by"]) + self.assertEqual(self.super_user.username, response.data["last_modified_by"]) + + preset = make_analysis_preset(self.dataspace, state=AnalysisPreset.State.NOT_AFFECTED) + apply_preset_for_vulnerabilities(preset, self.product1, [self.vulnerability2.pk]) + auto_analysis = VulnerabilityAnalysis.objects.get(vulnerability=self.vulnerability2) + detail_url = reverse("api_v2:vulnerabilityanalysis-detail", args=[auto_analysis.uuid]) + + response = self.client.get(detail_url) + self.assertEqual(preset.name, response.data["applied_by_preset"]) + self.assertIsNone(response.data["created_by"]) + self.assertIsNone(response.data["last_modified_by"]) + + def test_api_vulnerability_analysis_list_endpoint_filters_applied_by_preset(self): + self.client.login(username="super_user", password="secret") + make_vulnerability_analysis(self.product_package1, self.vulnerability1) + preset = make_analysis_preset(self.dataspace, state=AnalysisPreset.State.NOT_AFFECTED) + apply_preset_for_vulnerabilities(preset, self.product1, [self.vulnerability2.pk]) + + data = {"applied_by_preset__isnull": "true"} + response = self.client.get(self.analysis_list_url, data) + self.assertEqual(1, response.data["count"]) + self.assertContains(response, self.vulnerability1.advisory_id) + self.assertNotContains(response, self.vulnerability2.advisory_id) + + data = {"applied_by_preset__isnull": "false"} + response = self.client.get(self.analysis_list_url, data) + self.assertEqual(1, response.data["count"]) + self.assertNotContains(response, self.vulnerability1.advisory_id) + self.assertContains(response, self.vulnerability2.advisory_id) + def test_api_vulnerability_analysis_endpoint_create(self): self.client.login(username="super_user", password="secret") response = self.client.post(self.analysis_list_url) diff --git a/vulnerabilities/triage/api.py b/vulnerabilities/triage/api.py new file mode 100644 index 00000000..2e90bbf4 --- /dev/null +++ b/vulnerabilities/triage/api.py @@ -0,0 +1,139 @@ +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# DejaCode is a trademark of nexB Inc. +# SPDX-License-Identifier: AGPL-3.0-only +# See https://github.com/aboutcode-org/dejacode for support or download. +# See https://aboutcode.org for more information about AboutCode FOSS projects. +# + +from rest_framework import mixins +from rest_framework import serializers + +from dje.api import CreateRetrieveUpdateListViewSet +from dje.api import DataspacedHyperlinkedRelatedField +from dje.api import DataspacedSerializer +from dje.api import ExtraPermissionsViewSetMixin +from dje.api_custom import TabPermission +from vulnerabilities.triage.models import AnalysisPreset +from vulnerabilities.triage.models import TriageRuleset + + +class AnalysisPresetSerializer(DataspacedSerializer): + class Meta: + model = AnalysisPreset + fields = ( + "api_url", + "uuid", + "name", + "description", + "state", + "justification", + "responses", + "detail", + "is_reachable", + ) + extra_kwargs = { + "api_url": { + "view_name": "api_v2:analysispreset-detail", + "lookup_field": "uuid", + }, + } + + def validate(self, data): + content_fields = ("state", "justification", "responses", "detail") + + def get_value(field_name): + if field_name in data: + return data[field_name] + if self.instance is not None: + return getattr(self.instance, field_name) + return None + + if not any(get_value(field_name) for field_name in content_fields): + raise serializers.ValidationError( + "At least one of state, justification, responses or detail must be provided." + ) + return data + + +class AnalysisPresetViewSet( + ExtraPermissionsViewSetMixin, + mixins.DestroyModelMixin, + CreateRetrieveUpdateListViewSet, +): + queryset = AnalysisPreset.objects.all() + serializer_class = AnalysisPresetSerializer + lookup_field = "uuid" + extra_permissions = (TabPermission,) + search_fields = ( + "name", + "description", + ) + ordering_fields = ("name",) + allow_reference_access = True + + +class TriageRulesetSerializer(DataspacedSerializer): + analysis_preset = DataspacedHyperlinkedRelatedField( + view_name="api_v2:analysispreset-detail", + lookup_field="uuid", + required=False, + allow_null=True, + ) + request_template = DataspacedHyperlinkedRelatedField( + view_name="api_v2:requesttemplate-detail", + lookup_field="uuid", + required=False, + allow_null=True, + ) + + class Meta: + model = TriageRuleset + fields = ( + "api_url", + "uuid", + "name", + "description", + "recommended_action", + "precedence", + "enabled", + "rules_config", + "analysis_preset", + "request_template", + ) + extra_kwargs = { + "api_url": { + "view_name": "api_v2:triageruleset-detail", + "lookup_field": "uuid", + }, + } + + def validate_request_template(self, value): + if value and not value.created_by_id: + raise serializers.ValidationError( + "This request template has no creator and cannot be used to open requests." + ) + return value + + +class TriageRulesetViewSet( + ExtraPermissionsViewSetMixin, + mixins.DestroyModelMixin, + CreateRetrieveUpdateListViewSet, +): + queryset = TriageRuleset.objects.all() + serializer_class = TriageRulesetSerializer + lookup_field = "uuid" + extra_permissions = (TabPermission,) + search_fields = ( + "name", + "description", + ) + ordering_fields = ( + "name", + "precedence", + ) + allow_reference_access = True + + def get_queryset(self): + return super().get_queryset().select_related("analysis_preset", "request_template") diff --git a/vulnerabilities/triage/tests/test_api.py b/vulnerabilities/triage/tests/test_api.py new file mode 100644 index 00000000..8c89c309 --- /dev/null +++ b/vulnerabilities/triage/tests/test_api.py @@ -0,0 +1,172 @@ +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# DejaCode is a trademark of nexB Inc. +# SPDX-License-Identifier: AGPL-3.0-only +# See https://github.com/aboutcode-org/dejacode for support or download. +# See https://aboutcode.org for more information about AboutCode FOSS projects. +# + +from django.contrib.contenttypes.models import ContentType +from django.test import TestCase +from django.urls import reverse + +from rest_framework import status + +from dje.models import Dataspace +from dje.tests import create_superuser +from product_portfolio.models import Product +from vulnerabilities.triage.models import AnalysisPreset +from vulnerabilities.triage.models import TriageAction +from vulnerabilities.triage.models import TriageRuleset +from vulnerabilities.triage.tests import make_analysis_preset +from vulnerabilities.triage.tests import make_triage_ruleset +from workflow.models import RequestTemplate + + +class AnalysisPresetAPITestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.alternate = Dataspace.objects.create(name="Alternate") + self.super_user = create_superuser("super_user", self.dataspace) + + self.list_url = reverse("api_v2:analysispreset-list") + self.preset = make_analysis_preset( + self.dataspace, name="Preset1", state=AnalysisPreset.State.NOT_AFFECTED + ) + self.detail_url = reverse("api_v2:analysispreset-detail", args=[self.preset.uuid]) + make_analysis_preset(self.alternate, name="OtherPreset") + + def test_api_analysispreset_list_endpoint_user_available_scope(self): + self.client.login(username="super_user", password="secret") + response = self.client.get(self.list_url) + self.assertEqual(1, response.data["count"]) + self.assertContains(response, self.preset.name) + + def test_api_analysispreset_detail_endpoint(self): + self.client.login(username="super_user", password="secret") + response = self.client.get(self.detail_url) + self.assertEqual(self.preset.name, response.data["name"]) + self.assertEqual(AnalysisPreset.State.NOT_AFFECTED, response.data["state"]) + + def test_api_analysispreset_endpoint_create(self): + self.client.login(username="super_user", password="secret") + data = {"name": "New Preset", "detail": "Some detail"} + response = self.client.post(self.list_url, data) + self.assertEqual(status.HTTP_201_CREATED, response.status_code) + preset = AnalysisPreset.objects.get(name="New Preset") + self.assertEqual("Some detail", preset.detail) + + def test_api_analysispreset_endpoint_create_rejects_no_content_field(self): + self.client.login(username="super_user", password="secret") + data = {"name": "No content", "is_reachable": True} + response = self.client.post(self.list_url, data) + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + msg = "At least one of state, justification, responses or detail must be provided." + self.assertIn(msg, response.data["non_field_errors"]) + + def test_api_analysispreset_endpoint_update(self): + self.client.login(username="super_user", password="secret") + data = {"detail": "Updated detail"} + response = self.client.patch(self.detail_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.preset.refresh_from_db() + self.assertEqual("Updated detail", self.preset.detail) + + def test_api_analysispreset_endpoint_delete(self): + self.client.login(username="super_user", password="secret") + response = self.client.delete(self.detail_url) + self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code) + self.assertFalse(AnalysisPreset.objects.filter(pk=self.preset.pk).exists()) + + +class TriageRulesetAPITestCase(TestCase): + def setUp(self): + self.dataspace = Dataspace.objects.create(name="nexB") + self.alternate = Dataspace.objects.create(name="Alternate") + self.super_user = create_superuser("super_user", self.dataspace) + + self.list_url = reverse("api_v2:triageruleset-list") + self.ruleset = make_triage_ruleset( + self.dataspace, + name="Ruleset1", + recommended_action=TriageAction.UPGRADE, + precedence=100, + ) + self.detail_url = reverse("api_v2:triageruleset-detail", args=[self.ruleset.uuid]) + make_triage_ruleset(self.alternate, name="OtherRuleset") + + def test_api_triageruleset_list_endpoint_user_available_scope(self): + self.client.login(username="super_user", password="secret") + response = self.client.get(self.list_url) + self.assertEqual(1, response.data["count"]) + self.assertContains(response, self.ruleset.name) + + def test_api_triageruleset_detail_endpoint(self): + self.client.login(username="super_user", password="secret") + response = self.client.get(self.detail_url) + self.assertEqual(self.ruleset.name, response.data["name"]) + self.assertEqual(TriageAction.UPGRADE, response.data["recommended_action"]) + self.assertEqual(100, response.data["precedence"]) + + def test_api_triageruleset_endpoint_create(self): + self.client.login(username="super_user", password="secret") + data = { + "name": "New Ruleset", + "precedence": 200, + "recommended_action": TriageAction.NOTIFY, + "rules_config": {"risk_score": {"is_active": True, "min_risk_score": 8.0}}, + } + response = self.client.post(self.list_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_201_CREATED, response.status_code) + ruleset = TriageRuleset.objects.get(name="New Ruleset") + self.assertEqual( + {"is_active": True, "min_risk_score": 8.0}, ruleset.rules_config["risk_score"] + ) + + def test_api_triageruleset_endpoint_create_rejects_request_template_with_no_creator(self): + self.client.login(username="super_user", password="secret") + request_template = RequestTemplate.objects.create( + name="Broken Template", + description="Header", + dataspace=self.dataspace, + content_type=ContentType.objects.get_for_model(Product), + ) + request_template_url = reverse( + "api_v2:requesttemplate-detail", args=[request_template.uuid] + ) + data = { + "name": "New Ruleset", + "precedence": 200, + "request_template": request_template_url, + } + response = self.client.post(self.list_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + msg = "This request template has no creator and cannot be used to open requests." + self.assertIn(msg, response.data["request_template"]) + + def test_api_triageruleset_endpoint_create_rejects_cross_dataspace_analysis_preset(self): + self.client.login(username="super_user", password="secret") + other_preset = make_analysis_preset(self.alternate, name="OtherPreset") + other_preset_url = reverse("api_v2:analysispreset-detail", args=[other_preset.uuid]) + data = { + "name": "New Ruleset", + "precedence": 200, + "analysis_preset": other_preset_url, + } + response = self.client.post(self.list_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code) + self.assertIn("analysis_preset", response.data) + + def test_api_triageruleset_endpoint_update(self): + self.client.login(username="super_user", password="secret") + data = {"enabled": False} + response = self.client.patch(self.detail_url, data=data, content_type="application/json") + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.ruleset.refresh_from_db() + self.assertFalse(self.ruleset.enabled) + + def test_api_triageruleset_endpoint_delete(self): + self.client.login(username="super_user", password="secret") + response = self.client.delete(self.detail_url) + self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code) + self.assertFalse(TriageRuleset.objects.filter(pk=self.ruleset.pk).exists())