Merge branch 'main' into ebios-rm-tiles

backend/ciso_assistant/settings.py

@@ -153,6 +153,7 @@ def set_ciso_assistant_url(_, __, event_dict):
     "allauth.socialaccount",
     "allauth.socialaccount.providers.saml",
     "allauth.mfa",
+    "huey.contrib.djhuey",
 ]
 
 MIDDLEWARE = [
@@ -224,6 +225,7 @@ def set_ciso_assistant_url(_, __, event_dict):
     "MIN_REFRESH_INTERVAL": 60,
 }
 
+
 # Empty outside of debug mode so that allauth middleware does not raise an error
 STATIC_URL = ""
 
@@ -372,6 +374,16 @@ def set_ciso_assistant_url(_, __, event_dict):
     # OTHER SETTINGS
 }
 
+HUEY = {
+    "huey_class": "huey.SqliteHuey",  # Huey implementation to use.
+    "name": "huey-ciso-assistant",  # Use db name for huey.
+    "results": True,  # Store return values of tasks.
+    "store_none": False,  # If a task returns None, do not save to results.
+    "immediate": DEBUG,  # If DEBUG=True, run synchronously.
+    "utc": True,  # Use UTC for all times internally.
+    "filename": "db/huey.sqlite3",
+}
+
 # SSO with allauth
 
 ACCOUNT_USER_MODEL_USERNAME_FIELD = None

backend/core/helpers.py

@@ -15,6 +15,9 @@
 from iam.models import Folder, Permission, RoleAssignment, User
 from library.helpers import get_referential_translation
 
+from statistics import mean
+import math
+
 from .models import *
 from .utils import camel_case
 
@@ -889,6 +892,10 @@ def viewable_items(model):
 
     viewable_controls = viewable_items(AppliedControl)
     controls_count = viewable_controls.count()
+    progress_avg = math.ceil(
+        mean([x.progress() for x in viewable_items(ComplianceAssessment)] or [0])
+    )
+
     data = {
         "controls": {
             "total": controls_count,
@@ -908,17 +915,19 @@ def viewable_items(model):
             "acceptances": viewable_items(RiskAcceptance).count(),
         },
         "compliance": {
+            "used_frameworks": viewable_items(ComplianceAssessment)
+            .values("framework_id")
+            .distinct()
+            .count(),
             "audits": viewable_items(ComplianceAssessment).count(),
             "active_audits": viewable_items(ComplianceAssessment)
             .filter(status__in=["in_progress", "in_review", "done"])
             .count(),
             "evidences": viewable_items(Evidence).count(),
-            "compliant_items": viewable_items(RequirementAssessment)
-            .filter(result="compliant")
-            .count(),
             "non_compliant_items": viewable_items(RequirementAssessment)
             .filter(result="non_compliant")
             .count(),
+            "progress_avg": progress_avg,
         },
         "audits_stats": build_audits_stats(user),
         "csf_functions": csf_functions(user),

backend/core/migrations/0043_historicalmetric.py

@@ -0,0 +1,49 @@
+# Generated by Django 5.1.1 on 2024-11-29 09:46
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0042_asset_filtering_labels"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="HistoricalMetric",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("date", models.DateField(db_index=True, verbose_name="Date")),
+                ("data", models.JSONField(verbose_name="Historical Data")),
+                ("model", models.TextField(db_index=True, verbose_name="Model")),
+                (
+                    "object_id",
+                    models.UUIDField(db_index=True, verbose_name="Object ID"),
+                ),
+                (
+                    "updated_at",
+                    models.DateTimeField(auto_now=True, verbose_name="Updated at"),
+                ),
+            ],
+            options={
+                "indexes": [
+                    models.Index(
+                        fields=["model", "object_id", "date"],
+                        name="core_histor_model_e05191_idx",
+                    ),
+                    models.Index(
+                        fields=["date", "model"], name="core_histor_date_ddb7df_idx"
+                    ),
+                ],
+                "unique_together": {("model", "object_id", "date")},
+            },
+        ),
+    ]

backend/core/models.py

@@ -1747,6 +1747,34 @@ class Status(models.TextChoices):
     fields_to_check = ["name"]
 
 
+## historical data
+class HistoricalMetric(models.Model):
+    date = models.DateField(verbose_name=_("Date"), db_index=True)
+    data = models.JSONField(verbose_name=_("Historical Data"))
+    model = models.TextField(verbose_name=_("Model"), db_index=True)
+    object_id = models.UUIDField(verbose_name=_("Object ID"), db_index=True)
+    updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated at"))
+
+    class Meta:
+        unique_together = ("model", "object_id", "date")
+        indexes = [
+            models.Index(fields=["model", "object_id", "date"]),
+            models.Index(fields=["date", "model"]),
+        ]
+
+    @classmethod
+    def update_daily_metric(cls, model, object_id, data):
+        """
+        Upsert method to update or create a daily metric. Should be generic enough for other metrics.
+        """
+        return cls.objects.update_or_create(
+            model=model,
+            object_id=object_id,
+            date=now().date(),
+            defaults={"data": data},
+        )
+
+
 ########################### Secondary objects #########################
 
 
@@ -1817,9 +1845,38 @@ class Meta:
         verbose_name = _("Risk assessment")
         verbose_name_plural = _("Risk assessments")
 
+    def upsert_daily_metrics(self):
+        per_treatment = self.get_per_treatment()
+
+        total = RiskScenario.objects.filter(risk_assessment=self).count()
+        data = {
+            "scenarios": {
+                "total": total,
+                "per_treatment": per_treatment,
+            },
+        }
+
+        HistoricalMetric.update_daily_metric(
+            model=self.__class__.__name__, object_id=self.id, data=data
+        )
+
     def __str__(self) -> str:
         return f"{self.name} - {self.version}"
 
+    def get_per_treatment(self) -> dict:
+        output = dict()
+        for treatment in RiskScenario.TREATMENT_OPTIONS:
+            output[treatment[0]] = (
+                RiskScenario.objects.filter(risk_assessment=self)
+                .filter(treatment=treatment[0])
+                .count()
+            )
+        return output
+
+    def save(self, *args, **kwargs) -> None:
+        super().save(*args, **kwargs)
+        self.upsert_daily_metrics()
+
     @property
     def path_display(self) -> str:
         return f"{self.project.folder}/{self.project}/{self.name} - {self.version}"
@@ -2114,14 +2171,17 @@ class RiskScenario(NameDescriptionMixin):
     ]
 
     QUALIFICATIONS = [
-        ("Financial", _("Financial")),
-        ("Legal", _("Legal")),
-        ("Reputation", _("Reputation")),
-        ("Operational", _("Operational")),
         ("Confidentiality", _("Confidentiality")),
         ("Integrity", _("Integrity")),
         ("Availability", _("Availability")),
+        ("Proof", _("Proof")),
         ("Authenticity", _("Authenticity")),
+        ("Privacy", _("Privacy")),
+        ("Safety", _("Safety")),
+        ("Reputation", _("Reputation")),
+        ("Operational", _("Operational")),
+        ("Legal", _("Legal")),
+        ("Financial", _("Financial")),
     ]
 
     DEFAULT_SOK_OPTIONS = {
@@ -2399,6 +2459,7 @@ def save(self, *args, **kwargs):
         else:
             self.residual_level = -1
         super(RiskScenario, self).save(*args, **kwargs)
+        self.risk_assessment.upsert_daily_metrics()
 
 
 class ComplianceAssessment(Assessment):
@@ -2422,12 +2483,36 @@ class Meta:
         verbose_name = _("Compliance assessment")
         verbose_name_plural = _("Compliance assessments")
 
+    def upsert_daily_metrics(self):
+        per_status = dict()
+        per_result = dict()
+        for item in self.get_requirements_status_count():
+            per_status[item[1]] = item[0]
+
+        for item in self.get_requirements_result_count():
+            per_result[item[1]] = item[0]
+        total = RequirementAssessment.objects.filter(compliance_assessment=self).count()
+        data = {
+            "reqs": {
+                "total": total,
+                "per_status": per_status,
+                "per_result": per_result,
+                "progress_perc": self.progress(),
+                "score": self.get_global_score(),
+            },
+        }
+
+        HistoricalMetric.update_daily_metric(
+            model=self.__class__.__name__, object_id=self.id, data=data
+        )
+
     def save(self, *args, **kwargs) -> None:
         if self.min_score is None:
             self.min_score = self.framework.min_score
             self.max_score = self.framework.max_score
             self.scores_definition = self.framework.scores_definition
         super().save(*args, **kwargs)
+        self.upsert_daily_metrics()
 
     def create_requirement_assessments(
         self, baseline: Self | None = None
@@ -3037,6 +3122,10 @@ class Meta:
         verbose_name = _("Requirement assessment")
         verbose_name_plural = _("Requirement assessments")
 
+    def save(self, *args, **kwargs) -> None:
+        super().save(*args, **kwargs)
+        self.compliance_assessment.upsert_daily_metrics()
+
 
 ########################### RiskAcesptance is a domain object relying on secondary objects #########################
 

backend/core/views.py

@@ -17,6 +17,8 @@
 from pathlib import Path
 import humanize
 
+# from icecream import ic
+
 from django.utils.decorators import method_decorator
 from django.views.decorators.cache import cache_page
 from django.views.decorators.vary import vary_on_cookie
@@ -2233,6 +2235,31 @@ def create_suggested_applied_controls(request, pk):
             requirement_assessment.create_applied_controls_from_suggestions()
         return Response(status=status.HTTP_200_OK)
 
+    @action(detail=True, methods=["get"], url_path="progress_ts")
+    def progress_ts(self, request, pk):
+        try:
+            raw = (
+                HistoricalMetric.objects.filter(
+                    model="ComplianceAssessment", object_id=pk
+                )
+                .annotate(progress=F("data__reqs__progress_perc"))
+                .values("date", "progress")
+                .order_by("date")
+            )
+
+            # Transform the data into the required format
+            formatted_data = [
+                [entry["date"].isoformat(), entry["progress"]] for entry in raw
+            ]
+
+            return Response({"data": formatted_data})
+
+        except HistoricalMetric.DoesNotExist:
+            return Response(
+                {"error": "No metrics found for this assessment"},
+                status=status.HTTP_404_NOT_FOUND,
+            )
+
 
 class RequirementAssessmentViewSet(BaseModelViewSet):
     """

backend/iam/models.py

@@ -537,7 +537,11 @@ def is_editor(self) -> bool:
 
     @classmethod
     def get_editors(cls) -> List[Self]:
-        return [user for user in cls.objects.all() if user.is_editor]
+        return [
+            user
+            for user in cls.objects.all()
+            if user.is_editor and not user.is_third_party
+        ]
 
 
 class Role(NameDescriptionMixin, FolderMixin):