Feat/auto create suggested controls in audit

backend/app_tests/api/test_api_compliance_assessments.py

@@ -129,6 +129,7 @@ def test_get_compliance_assessments(self, test):
                     "id": str(Framework.objects.all()[0].id),
                     "str": str(Framework.objects.all()[0]),
                     "implementation_groups_definition": None,
+                    "reference_controls": [],
                     "min_score": 1,
                     "max_score": 4,
                     "ref_id": str(Framework.objects.all()[0].ref_id),
@@ -168,6 +169,7 @@ def test_create_compliance_assessments(self, test):
                     "id": str(Framework.objects.all()[0].id),
                     "str": str(Framework.objects.all()[0]),
                     "implementation_groups_definition": None,
+                    "reference_controls": [],
                     "min_score": Framework.objects.all()[0].min_score,
                     "max_score": Framework.objects.all()[0].max_score,
                     "ref_id": str(Framework.objects.all()[0].ref_id),
@@ -219,6 +221,7 @@ def test_update_compliance_assessments(self, test):
                     "id": str(Framework.objects.all()[0].id),
                     "str": str(Framework.objects.all()[0]),
                     "implementation_groups_definition": None,
+                    "reference_controls": [],
                     "min_score": Framework.objects.all()[0].min_score,
                     "max_score": Framework.objects.all()[0].max_score,
                     "ref_id": str(Framework.objects.all()[0].ref_id),

backend/core/models.py

@@ -9,31 +9,25 @@
 from django.apps import apps
 from django.contrib.auth import get_user_model
 from django.core import serializers
-from django.utils.translation import get_language
-from library.helpers import (
-    update_translations_in_object,
-    update_translations_as_string,
-    update_translations,
-    get_referential_translation,
-)
-
-import os
-import json
-import yaml
-import re
-
 from django.core.exceptions import ValidationError
 from django.core.validators import MaxValueValidator
 from django.db import models, transaction
 from django.db.models import Q
 from django.forms.models import model_to_dict
 from django.urls import reverse
 from django.utils.html import format_html
+from django.utils.translation import get_language
 from django.utils.translation import gettext_lazy as _
-from iam.models import Folder, FolderMixin, PublishInRootFolderMixin
-from library.helpers import update_translations, update_translations_in_object
 from structlog import get_logger
 
+from iam.models import Folder, FolderMixin, PublishInRootFolderMixin
+from library.helpers import (
+    get_referential_translation,
+    update_translations,
+    update_translations_as_string,
+    update_translations_in_object,
+)
+
 from .base_models import AbstractBaseModel, ETADueDateMixin, NameDescriptionMixin
 from .utils import camel_case, sha256
 from .validators import validate_file_name, validate_file_size
@@ -925,6 +919,18 @@ def library_entry(self):
 
         return res
 
+    @property
+    def reference_controls(self):
+        _reference_controls = ReferenceControl.objects.filter(
+            requirements__framework=self
+        ).distinct()
+        reference_controls = []
+        for control in _reference_controls:
+            reference_controls.append(
+                {"str": control.display_long, "urn": control.urn, "id": control.id}
+            )
+        return reference_controls
+
     def get_requirement_nodes(self):
         # Prefetch related objects if they exist to reduce database queries.
         # Adjust prefetch_related paths according to your model relationships.
@@ -2080,7 +2086,9 @@ def save(self, *args, **kwargs) -> None:
             self.scores_definition = self.framework.scores_definition
         super().save(*args, **kwargs)
 
-    def create_requirement_assessments(self, baseline: Self | None = None):
+    def create_requirement_assessments(
+        self, baseline: Self | None = None
+    ) -> list["RequirementAssessment"]:
         requirements = RequirementNode.objects.filter(framework=self.framework)
         requirement_assessments = []
         for requirement in requirements:
@@ -2581,6 +2589,41 @@ def infer_result(
                 return (RequirementAssessment.Result.NON_COMPLIANT, None)
         return (None, None)
 
+    def create_applied_controls_from_suggestions(self) -> list[AppliedControl]:
+        applied_controls: list[AppliedControl] = []
+        for reference_control in self.requirement.reference_controls.all():
+            try:
+                _name = reference_control.name or reference_control.ref_id
+                applied_control, created = AppliedControl.objects.get_or_create(
+                    name=_name,
+                    folder=self.folder,
+                    reference_control=reference_control,
+                    category=reference_control.category,
+                    description=reference_control.description,
+                )
+                if created:
+                    logger.info(
+                        "Successfully created applied control from reference_control",
+                        applied_control=applied_control,
+                        reference_control=reference_control,
+                    )
+                else:
+                    logger.info(
+                        "Applied control already exists, skipping creation and using existing one",
+                        applied_control=applied_control,
+                        reference_control=reference_control,
+                    )
+                applied_controls.append(applied_control)
+            except Exception as e:
+                logger.error(
+                    "An error occurred while creating applied control from reference control",
+                    reference_control=reference_control,
+                    exc_info=e,
+                )
+                continue
+        self.applied_controls.set(applied_controls)
+        return applied_controls
+
     class Meta:
         verbose_name = _("Requirement assessment")
         verbose_name_plural = _("Requirement assessments")

backend/core/serializers.py

@@ -541,7 +541,14 @@ class ComplianceAssessmentReadSerializer(AssessmentReadSerializer):
     project = FieldsRelatedField(["id", "folder"])
     folder = FieldsRelatedField()
     framework = FieldsRelatedField(
-        ["id", "min_score", "max_score", "implementation_groups_definition", "ref_id"]
+        [
+            "id",
+            "min_score",
+            "max_score",
+            "implementation_groups_definition",
+            "ref_id",
+            "reference_controls",
+        ]
     )
     selected_implementation_groups = serializers.ReadOnlyField(
         source="get_selected_implementation_groups"
@@ -559,6 +566,9 @@ class ComplianceAssessmentWriteSerializer(BaseModelSerializer):
         required=False,
         allow_null=True,
     )
+    create_applied_controls_from_suggestions = serializers.BooleanField(
+        write_only=True, required=False, default=False
+    )
 
     def create(self, validated_data: Any):
         return super().create(validated_data)

backend/core/tests/fixtures.py

@@ -0,0 +1,39 @@
+import pytest
+
+from core.models import (
+    Project,
+    StoredLibrary,
+)
+from iam.models import Folder
+
+
+@pytest.fixture
+def domain_project_fixture():
+    folder = Folder.objects.create(
+        name="test folder", description="test folder description"
+    )
+    project = Project.objects.create(name="test project", folder=folder)
+    return project
+
+
+@pytest.fixture
+def risk_matrix_fixture():
+    library = StoredLibrary.objects.filter(
+        urn="urn:intuitem:risk:library:critical_risk_matrix_5x5"
+    ).last()
+    assert library is not None
+    library.load()
+
+
+@pytest.fixture
+def iso27001_csf1_1_frameworks_fixture():
+    iso27001_library = StoredLibrary.objects.get(
+        urn="urn:intuitem:risk:library:iso27001-2022", locale="en"
+    )
+    assert iso27001_library is not None
+    iso27001_library.load()
+    csf_1_1_library = StoredLibrary.objects.get(
+        urn="urn:intuitem:risk:library:nist-csf-1.1", locale="en"
+    )
+    assert csf_1_1_library is not None
+    csf_1_1_library.load()

backend/core/tests/test_models.py

@@ -18,7 +18,6 @@
     Evidence,
     RiskAcceptance,
     Asset,
-    StoredLibrary,
     Threat,
     RiskMatrix,
     LoadedLibrary,
@@ -28,41 +27,12 @@
 from django.core.files.uploadedfile import SimpleUploadedFile
 from iam.models import Folder
 
-User = get_user_model()
+from .fixtures import *
 
-SAMPLE_640x480_JPG = BASE_DIR / "app_tests" / "sample_640x480.jpg"
 
+User = get_user_model()
 
-@pytest.fixture
-def domain_project_fixture():
-    folder = Folder.objects.create(
-        name="test folder", description="test folder description"
-    )
-    project = Project.objects.create(name="test project", folder=folder)
-    return project
-
-
-@pytest.fixture
-def risk_matrix_fixture():
-    library = StoredLibrary.objects.filter(
-        urn="urn:intuitem:risk:library:critical_risk_matrix_5x5"
-    ).last()
-    assert library is not None
-    library.load()
-
-
-@pytest.fixture
-def iso27001_csf1_1_frameworks_fixture():
-    iso27001_library = StoredLibrary.objects.get(
-        urn="urn:intuitem:risk:library:iso27001-2022", locale="en"
-    )
-    assert iso27001_library is not None
-    iso27001_library.load()
-    csf_1_1_library = StoredLibrary.objects.get(
-        urn="urn:intuitem:risk:library:nist-csf-1.1", locale="en"
-    )
-    assert csf_1_1_library is not None
-    csf_1_1_library.load()
+SAMPLE_640x480_JPG = BASE_DIR / "app_tests" / "sample_640x480.jpg"
 
 
 @pytest.mark.django_db

backend/core/tests/test_requirement_assessment.py

@@ -0,0 +1,71 @@
+import pytest
+from django.contrib.auth import get_user_model
+from core.models import (
+    StoredLibrary,
+    Framework,
+    ComplianceAssessment,
+    Project,
+    RequirementAssessment,
+)
+from iam.models import Folder
+
+from .fixtures import *
+
+User = get_user_model()
+
+
+@pytest.fixture
+def enisa_5g_scm_framework_fixture():
+    enisa_5g_scm_library = StoredLibrary.objects.get(
+        urn="urn:intuitem:risk:library:enisa-5g-scm-v1.3", locale="en"
+    )
+    assert enisa_5g_scm_library is not None
+    enisa_5g_scm_library.load()
+
+
+@pytest.mark.django_db
+class TestRequirementAssessment:
+    @pytest.mark.usefixtures("domain_project_fixture", "enisa_5g_scm_framework_fixture")
+    def test_create_applied_controls_from_suggestions(self):
+        enisa_5g_scm = Framework.objects.first()
+        compliance_assessment = ComplianceAssessment.objects.create(
+            name="test compliance assessment",
+            framework=enisa_5g_scm,
+            folder=Folder.objects.filter(
+                content_type=Folder.ContentType.DOMAIN
+            ).first(),
+            project=Project.objects.first(),
+        )
+
+        requirement_assessments: list[RequirementAssessment] = (
+            compliance_assessment.create_requirement_assessments()
+        )
+        assert requirement_assessments is not None
+        assert len(requirement_assessments) > 0
+
+        for requirement_assessment in requirement_assessments:
+            requirement_assessment.create_applied_controls_from_suggestions()
+
+            if len(requirement_assessment.requirement.reference_controls.all()) == 0:
+                assert requirement_assessment.applied_controls.all().count() == 0
+                return
+
+            assert requirement_assessment.applied_controls.all() is not None
+            assert len(requirement_assessment.applied_controls.all()) > 0
+            for control in requirement_assessment.applied_controls.all():
+                assert (
+                    control.reference_control
+                    in requirement_assessment.requirement.reference_controls.all()
+                )
+
+            # NOTE: running create_applied_controls_from_suggestions agani MUST not create
+            # any new applied control.
+
+            applied_controls_count: int = (
+                requirement_assessment.applied_controls.all().count()
+            )
+            requirement_assessment.create_applied_controls_from_suggestions()
+            assert (
+                requirement_assessment.applied_controls.all().count()
+                == applied_controls_count
+            )

backend/core/urls.py

@@ -90,6 +90,14 @@
     ),  # NOTE: This has to be placed before the allauth urls, otherwise our ACS implementation will not be used
     path("accounts/", include("allauth.urls")),
     path("_allauth/", include("allauth.headless.urls")),
+    path(
+        "requirement-assessments/<uuid:pk>/suggestions/applied-controls/",
+        RequirementAssessmentViewSet.create_suggested_applied_controls,
+    ),
+    path(
+        "compliance-assessments/<uuid:pk>/suggestions/applied-controls/",
+        ComplianceAssessmentViewSet.create_suggested_applied_controls,
+    ),
 ]
 
 # Additional modules take precedence over the default modules