Merge branch 'main' into ui/compliance-assessment-detail

backend/core/migrations/0003_library_dependencies_and_more.py

@@ -0,0 +1,24 @@
+# Generated by Django 5.0.2 on 2024-02-13 15:19
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0002_initial'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='library',
+            name='dependencies',
+            field=models.ManyToManyField(blank=True, to='core.library', verbose_name='Dependencies'),
+        ),
+        migrations.AlterField(
+            model_name='requirementassessment',
+            name='compliance_assessment',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='requirement_assessments', to='core.complianceassessment', verbose_name='Compliance assessment'),
+        ),
+    ]

backend/core/models.py

@@ -40,11 +40,14 @@ class Library(ReferentialObjectMixin, AbstractBaseModel, FolderMixin):
         help_text=_("Packager of the library"),
         verbose_name=_("Packager"),
     )
+    dependencies = models.ManyToManyField(
+        "self", blank=True, verbose_name=_("Dependencies"), symmetrical=False
+    )
 
     @property
     def reference_count(self) -> int:
         """
-        Returns the number of distinct risk and compliance assessments that reference objects from this library
+        Returns the number of distinct dependent libraries and risk and compliance assessments that reference objects from this library
         """
         return (
             RiskAssessment.objects.filter(
@@ -62,13 +65,19 @@ def reference_count(self) -> int:
             )
             .distinct()
             .count()
+            + Library.objects.filter(dependencies=self).distinct().count()
         )
 
     def delete(self, *args, **kwargs):
         if self.reference_count > 0:
             raise ValueError(
                 "This library is still referenced by some risk or compliance assessments"
             )
+        dependent_libraries = Library.objects.filter(dependencies=self)
+        if dependent_libraries:
+            raise ValueError(
+                f"This library is a dependency of {dependent_libraries.count()} other libraries"
+            )
         super(Library, self).delete(*args, **kwargs)
 
 

backend/core/tests/test_models.py

@@ -1115,3 +1115,30 @@ def test_library_reference_count_must_be_zero_for_library_deletion(
         library.delete()
 
         assert Library.objects.count() == 0
+
+    @pytest.mark.usefixtures("domain_project_fixture")
+    def test_library_cannot_be_deleted_if_it_is_a_dependency_of_other_libraries(self):
+        dependency_library = Library.objects.create(
+            name="Dependency Library",
+            description="Dependency Library description",
+            folder=Folder.get_root_folder(),
+            locale="en",
+            version=1,
+        )
+        library = Library.objects.create(
+            name="Library",
+            description="Library description",
+            folder=Folder.get_root_folder(),
+            locale="en",
+            version=1,
+        )
+        library.dependencies.add(dependency_library)
+
+        with pytest.raises(ValueError):
+            dependency_library.delete()
+
+        library.delete()
+        assert Library.objects.count() == 1
+
+        dependency_library.delete()
+        assert Library.objects.count() == 0

backend/library/utils.py

@@ -1,24 +1,19 @@
+import json
+import os
+from typing import List, Union
+
+import yaml
+from ciso_assistant import settings
 from core.models import (
     Framework,
-    RequirementNode,
-    RequirementNode,
-    RequirementLevel,
     Library,
+    RequirementNode,
+    RiskMatrix,
+    SecurityFunction,
+    Threat,
 )
-from core.models import Threat, SecurityFunction, RiskMatrix
-from django.contrib import messages
-from django.contrib.auth.models import Permission
-from iam.models import Folder, RoleAssignment
-from ciso_assistant import settings
-from django.utils.translation import gettext_lazy as _
-
-from .validators import *
-
-import os
-import yaml
-import json
-
-from typing import Union, List
+from django.db import transaction
+from iam.models import Folder
 
 
 def get_available_library_files():
@@ -69,8 +64,8 @@ def get_library_names(libraries):
         names: list of available library names
     """
     names = []
-    for l in libraries:
-        names.append(l.get("name"))
+    for lib in libraries:
+        names.append(lib.get("name"))
     return names
 
 
@@ -85,9 +80,9 @@ def get_library(urn: str) -> dict | None:
         library: library with the given urn
     """
     libraries = get_available_libraries()
-    for l in libraries:
-        if l["urn"] == urn:
-            return l
+    for lib in libraries:
+        if lib["urn"] == urn:
+            return lib
     return None
 
 
@@ -465,33 +460,26 @@ def init(self) -> Union[str, None]:
             ) is not None:
                 return security_function_import_error
 
-    def import_library(self) -> Union[str, None]:
-        if (error_message := self.init()) is not None:
-            return error_message
-
-        if self._library_data.get("dependencies"):
-            for dependency in self._library_data["dependencies"]:
-                if not Library.objects.filter(urn=dependency).exists():
-                    import_library_view(get_library(dependency))
+    def check_and_import_dependencies(self):
+        """Check and import library dependencies."""
+        dependencies = self._library_data.get("dependencies", [])
+        for dependency in dependencies:
+            if not Library.objects.filter(urn=dependency).exists():
+                import_library_view(get_library(dependency))
 
+    def create_or_update_library(self):
+        """Create or update the library object."""
         _urn = self._library_data["urn"]
+        _locale = self._library_data["locale"]
         _default_locale = not Library.objects.filter(urn=_urn).exists()
 
-        # todo: import only if new or newer version
-        # if Library.objects.filter(
-        #     urn=self._library_data['urn'],
-        #     locale=self._library_data["locale"]
-        # ).exists():
-        #     return "A library with the same URN and the same locale value has already been loaded !"
-        _urn = self._library_data["urn"]
-        _locale = self._library_data["locale"]
         library_object, _created = Library.objects.update_or_create(
             defaults={
                 "ref_id": self._library_data["ref_id"],
                 "name": self._library_data.get("name"),
                 "description": self._library_data.get("description", None),
                 "urn": _urn,
-                "locale": self._library_data["locale"],
+                "locale": _locale,
                 "default_locale": _default_locale,
                 "version": self._library_data.get("version", None),
                 "provider": self._library_data.get("provider", None),
@@ -501,25 +489,42 @@ def import_library(self) -> Union[str, None]:
             urn=_urn,
             locale=_locale,
         )
+        return library_object
 
-        import_error_msg = None
-        try:
-            if self._framework_importer is not None:
-                self._framework_importer.import_framework(library_object)
+    def import_objects(self, library_object):
+        """Import library objects."""
+        if self._framework_importer is not None:
+            self._framework_importer.import_framework(library_object)
+
+        for threat in self._threats:
+            threat.import_threat(library_object)
 
-            for threat in self._threats:
-                threat.import_threat(library_object)
+        for security_function in self._security_functions:
+            security_function.import_security_function(library_object)
 
-            for security_function in self._security_functions:
-                security_function.import_security_function(library_object)
+        for risk_matrix in self._risk_matrices:
+            risk_matrix.import_risk_matrix(library_object)
 
-            for risk_matrix in self._risk_matrices:
-                risk_matrix.import_risk_matrix(library_object)
+    def import_library(self):
+        """Main method to import a library."""
+        if (error_message := self.init()) is not None:
+            return error_message
 
+        self.check_and_import_dependencies()
+
+        try:
+            with transaction.atomic():
+                library_object = self.create_or_update_library()
+                self.import_objects(library_object)
+                library_object.dependencies.set(
+                    Library.objects.filter(
+                        urn__in=self._library_data.get("dependencies", [])
+                    )
+                )
         except Exception as e:
-            print("lib exception", e)
-            library_object.delete()
-            raise e
+            # TODO: Switch to proper logging
+            print(f"Library import exception: {e}")
+            raise
 
 
 def import_library_view(library: dict) -> Union[str, None]:
@@ -536,5 +541,6 @@ def import_library_view(library: dict) -> Union[str, None]:
     optional_error : Union[str,None]
         A string describing the error if the function fails and returns None on success.
     """
+    # NOTE: We should just use LibraryImporter.import_library at this point
     library_importer = LibraryImporter(library)
     return library_importer.import_library()