Django Signals for Automated Webhook Notifications

.dockerignore

@@ -1,9 +1,15 @@
 *.pyc
-*.DS_Store
+**/.DS_Store
 *~$*
 .git*
 .pytest*
 .idea*
-venv/
-env/
-**/node_modules/
+.venv
+.dockerignore
+Dockerfile
+db
+**/__pycache__/
+**/.pytest_cache/
+**/.ruff_cache/
+static/
+pytest-report.html

backend/.gitignore

@@ -1,22 +1,21 @@
-!__init__.py
-*.pyc
-__pycache__
 *.DS_Store
 *~$*
-staticfiles/*
-static/
 .env
+.venv
+venv
+**/node_modules/
 .vscode
 *.sqlite3
-staticfiles/
-venv/
+django_secret_key
 temp/
-db/attachments/
-db/django_secret_key
-db/pg_password.txt
-./db/
-.coverage
-pytest-report.html
-enterprise/
-*.log
-*.bak
+db/
+.dccache
+/backend/profiles
+./backend/ciso_assistant/.meta
+caddy_data/
+**/dist/
+**/.meta
+charts/custom-values.yaml
+**/charts/*/charts
+charts/ciso-assistant-next/custom.yaml
+docker-compose.yml

backend/Dockerfile

@@ -39,14 +39,11 @@ RUN apt-get update && apt-get install -y \
     && locale-gen sv_SE.UTF-8 \
     && locale-gen id_ID.UTF-8 \
     && pip install --no-cache-dir --upgrade pip poetry==2.0.1
-
+    
 COPY pyproject.toml poetry.lock ./
+RUN poetry install --no-root && rm -rf $POETRY_CACHE_DIR
 
-RUN poetry install --no-root \
-    && rm -rf $POETRY_CACHE_DIR
-
-#watch out for local files during dev and maintenance of .dockerignore
 COPY . .
 
 EXPOSE 8000
-ENTRYPOINT ["poetry", "run", "bash", "startup.sh"]
+ENTRYPOINT ["poetry", "run", "bash", "startup.sh"]

backend/ciso_assistant/settings.py

@@ -495,3 +495,5 @@ def set_ciso_assistant_url(_, __, event_dict):
 
 AUDITLOG_RETENTION_DAYS = int(os.environ.get("AUDITLOG_RETENTION_DAYS", 90))
 AUDITLOG_MAX_RECORDS = int(os.environ.get("AUDITLOG_MAX_RECORDS", 50000))
+
+NOTIFICATION_WEBHOOK_URL = os.environ.get("NOTIFICATION_WEBHOOK_URL")

backend/core/apps.py

@@ -11,6 +11,8 @@ class CoreConfig(AppConfig):
     verbose_name = "Core"
 
     def ready(self):
-        # avoid post_migrate handler if we are in the main, as it interferes with restore
+        # Avoid post_migrate handler if we are in the main, as it interferes with restore
         if not os.environ.get("RUN_MAIN"):
             post_migrate.connect(startup, sender=self)
+        # Import signals to register Django signal handlers
+        import core.signals

backend/core/signals.py

@@ -0,0 +1,139 @@
+import logging
+from django.db.models.signals import pre_save, post_save
+from django.dispatch import receiver
+from django.conf import settings
+from core.models import ComplianceAssessment, AppliedControl
+import requests
+
+logger = logging.getLogger(__name__)
+
+WEBHOOK_URL = getattr(settings, 'NOTIFICATION_WEBHOOK_URL', None)
+
+
+def send_webhook_notification(payload):
+    if not WEBHOOK_URL:
+        logger.warning('Notification webhook URL is not set. Skipping notification.')
+        return
+    try:
+        logger.info('Sending notification to webhook.')
+        response = requests.post(WEBHOOK_URL, json=payload, timeout=5)
+        response.raise_for_status()
+        logger.info(f'Notification sent successfully. Response: {response.status_code}')
+    except requests.RequestException as e:
+        logger.error(f'Error sending notification: {e}')
+
+
+def get_applied_control_payload(instance, old_status):
+    return {
+        'type': 'applied_control_status_changed',
+        'id': str(instance.pk),
+        'name': instance.name,
+        'old_status': old_status,
+        'new_status': instance.status,
+        'ref_id': instance.ref_id,
+        'category': instance.category,
+        'csf_function': instance.csf_function,
+        'priority': instance.priority,
+        'effort': instance.effort,
+        'control_impact': instance.control_impact,
+        'cost': instance.cost,
+        'progress_field': instance.progress_field,
+        'start_date': str(instance.start_date) if instance.start_date else None,
+        'eta': str(instance.eta) if instance.eta else None,
+        'expiry_date': str(instance.expiry_date) if instance.expiry_date else None,
+        'link': instance.link,
+        'reference_control': str(instance.reference_control) if instance.reference_control else None,
+        'owners': [{'id': u.id, 'name': str(u)} for u in instance.owner.all()],
+        'assets': [{'id': a.id, 'name': str(a)} for a in instance.assets.all()],
+        'evidences': [{'id': e.id, 'name': str(e)} for e in instance.evidences.all()],
+        'security_exceptions': [{'id': se.id, 'name': str(se)} for se in instance.security_exceptions.all()],
+        # Add other fields as needed for more context
+    }
+
+def get_compliance_assessment_payload(instance, old_status):
+    # Calcul des pourcentages de conformité et de progression
+    compliance_percentage = None
+    progress_percentage = None
+    try:
+        if hasattr(instance, 'get_global_score'):
+            compliance_percentage = instance.get_global_score()
+    except Exception as e:
+        logger.warning(f'Could not compute compliance_percentage: {e}')
+    try:
+        if hasattr(instance, 'get_progress'):
+            progress_percentage = instance.get_progress()
+    except Exception as e:
+        logger.warning(f'Could not compute progress_percentage: {e}')
+    return {
+        'type': 'compliance_assessment_status_changed',
+        'id': str(instance.pk),
+        'name': instance.name,
+        'old_status': old_status,
+        'new_status': instance.status,
+        'ref_id': instance.ref_id,
+        'framework': str(instance.framework) if instance.framework else None,
+        'perimeter': str(instance.perimeter) if hasattr(instance, 'perimeter') and instance.perimeter else None,
+        'min_score': instance.min_score,
+        'max_score': instance.max_score,
+        'show_documentation_score': instance.show_documentation_score,
+        'assets': [str(a) for a in instance.assets.all()],
+        'campaign': str(instance.campaign) if instance.campaign else None,
+        'evidences': [str(e) for e in instance.evidences.all()],
+        'authors': [str(u) for u in instance.authors.all()] if hasattr(instance, 'authors') else [],
+        'compliance_percentage': compliance_percentage,
+        'progress_percentage': progress_percentage,
+        # Add other fields as needed
+    }
+
+
+@receiver(pre_save, sender=ComplianceAssessment)
+def compliance_assessment_status_change(sender, instance, **kwargs):
+    if not instance.pk:
+        logger.info(f'[pre_save] ComplianceAssessment: new instance, will notify creation after save')
+        # On ne peut pas envoyer la notification ici car l'ID n'est pas encore assigné
+        # On la gère dans post_save ci-dessous
+        return
+    try:
+        old = sender.objects.get(pk=instance.pk)
+    except sender.DoesNotExist:
+        logger.info(f'[pre_save] ComplianceAssessment: instance with pk={instance.pk} does not exist in DB')
+        return
+    old_status = old.status
+    new_status = instance.status
+    logger.info(f'[pre_save] ComplianceAssessment: old_status={old_status}, new_status={new_status}')
+    if old_status != new_status:
+        logger.info(f'[pre_save] Status changed for ComplianceAssessment id={instance.pk}: {old_status} -> {new_status}')
+        payload = get_compliance_assessment_payload(instance, old_status)
+        send_webhook_notification(payload)
+    else:
+        logger.info(f'[pre_save] No status change for ComplianceAssessment id={instance.pk}: status remains {new_status}')
+
+# Ajout : notification à la création d'un audit
+@receiver(post_save, sender=ComplianceAssessment)
+def compliance_assessment_created(sender, instance, created, **kwargs):
+    if created:
+        logger.info(f'[post_save] ComplianceAssessment created: id={instance.pk}, name={instance.name}')
+        payload = get_compliance_assessment_payload(instance, old_status=None)
+        payload['type'] = 'compliance_assessment_created'
+        send_webhook_notification(payload)
+
+
+@receiver(pre_save, sender=AppliedControl)
+def applied_control_status_change(sender, instance, **kwargs):
+    if not instance.pk:
+        logger.info(f'[pre_save] AppliedControl: new instance, no status to compare (id will be assigned after save)')
+        return
+    try:
+        old = sender.objects.get(pk=instance.pk)
+    except sender.DoesNotExist:
+        logger.info(f'[pre_save] AppliedControl: instance with pk={instance.pk} does not exist in DB')
+        return
+    old_status = old.status
+    new_status = instance.status
+    logger.info(f'[pre_save] AppliedControl: old_status={old_status}, new_status={new_status}')
+    if old_status != new_status:
+        logger.info(f'[pre_save] Status changed for AppliedControl id={instance.pk}: {old_status} -> {new_status}')
+        payload = get_applied_control_payload(instance, old_status)
+        send_webhook_notification(payload)
+    else:
+        logger.info(f'[pre_save] No status change for AppliedControl id={instance.pk}: status remains {new_status}') 

backend/manage.py

@@ -3,7 +3,7 @@
 
 import os
 import sys
-
+# import signals
 
 def main():
     """Run administrative tasks."""

docker-compose.yml

@@ -1,14 +1,15 @@
 services:
   backend:
     container_name: backend
-    image: ghcr.io/intuitem/ciso-assistant-community/backend:latest
-    pull_policy: always
+    image: backend
     restart: always
     environment:
       - ALLOWED_HOSTS=backend,localhost
       - CISO_ASSISTANT_URL=https://localhost:8443
       - DJANGO_DEBUG=True
       - AUTH_TOKEN_TTL=7200
+      # Set this to your real webhook endpoint before deployment
+      - NOTIFICATION_WEBHOOK_URL=https://your-webhook-endpoint.example.com/notifications
 
     volumes:
       - ./db:/code/db