feat(pro): audit logs forwarding capabilities (#4333)

* step1: prerequisite for auditlog data enrichment with folder

* backend requirements

* auditlog sink

* pass #1 on frontend

* fixup

* fix webhooks that cannot be disabled

* wip

* cef_experiment

* combine migrations

* fixup

* reduce scope for now

* fixup

* wip

* fixup

* misc

---------

Co-authored-by: Eric <71850047+eric-intuitem@users.noreply.github.com>

Author: ab-smith

backend/core/base_models.py

@@ -1,7 +1,7 @@
 from django.db import models, transaction
 from django.utils.translation import gettext_lazy as _
 from django.urls.base import reverse_lazy
-from django.core.exceptions import ValidationError
+from django.core.exceptions import ObjectDoesNotExist, ValidationError
 import uuid
 
 
@@ -31,6 +31,24 @@ def scoped_id(self, scope: models.QuerySet) -> int:
     def __str__(self) -> str:
         return self.name if hasattr(self, "name") and self.name else str(self.id)
 
+    def get_additional_data(self) -> dict:
+        # Attached to every django-auditlog LogEntry at creation (create/update/
+        # delete and m2m). Runs with the instance in memory, so the folder is
+        # captured even on delete — used to scope audit events forwarded to SIEMs.
+        folder_id = getattr(self, "folder_id", None)
+        if folder_id is None:
+            from iam.models import Folder
+
+            # Runs in auditlog's synchronous delete receiver: a cascade may have
+            # already removed the FK target get_folder traverses, raising
+            # DoesNotExist. Never let metadata enrichment break the delete.
+            try:
+                folder = Folder.get_folder(self)
+                folder_id = folder.id if folder else None
+            except ObjectDoesNotExist:
+                folder_id = None
+        return {"folder_id": str(folder_id) if folder_id else None}
+
     def is_unique_in_scope(self, scope: models.QuerySet, fields_to_check: list) -> bool:
         """
         Checks if the object is unique in the given scope based on the given fields.

backend/core/custom_middleware.py

@@ -4,17 +4,16 @@
 from auditlog.cid import correlation_id, set_cid
 from auditlog.context import set_extra_data
 from django.utils.functional import SimpleLazyObject
-from auditlog.models import LogEntry
-from django.db.models.signals import post_save
-from django.dispatch import receiver
-
-import structlog
-
-logger = structlog.getLogger(__name__)
 
 
 # Fix the actor resolution with the custom middleware
 # Based on https://stackoverflow.com/questions/40740061/auditlog-with-django-and-drf
+#
+# Folder/actor enrichment of LogEntry is no longer done here via a post_save
+# signal: folder is captured inline at log creation through
+# AbstractBaseModel.get_additional_data(), and actor/actor_email are native
+# LogEntry columns. This fixes delete events (instance still in memory) and
+# covers m2m changes.
 class AuditlogMiddleware(middleware.AuditlogMiddleware):
     @staticmethod
     def _get_actor(request):
@@ -33,42 +32,3 @@ def __call__(self, request):
 
         with set_extra_data(context_data=self.get_extra_data(request)):
             return self.get_response(request)
-
-
-# Add a post-save signal to add the additional info after the log entry is saved
-# Think about the potential perf overhead of this
-@receiver(post_save, sender=LogEntry)
-def add_user_info_to_log_entry(sender, instance, created, **kwargs):
-    if not created or not instance.actor_id:
-        return
-
-    obj = None
-    model_class = instance.content_type.model_class()
-    if model_class is not None:
-        try:
-            obj = model_class.objects.get(pk=instance.object_pk)
-        except model_class.DoesNotExist:
-            logger.debug("audit log enrichment failed: no model_class.")
-            pass
-
-    # Only update if this is a new log entry and it has an actor
-    try:
-        user_uuid = str(instance.actor_id)
-        user_email = instance.actor.email
-        folder = (
-            "/".join([f.name for f in obj.get_folder_full_path()])
-            if obj and hasattr(obj, "get_folder_full_path")
-            else None
-        )
-
-        LogEntry.objects.filter(pk=instance.pk).update(
-            additional_data={
-                "user_uuid": user_uuid,
-                "user_email": user_email,
-                "folder": folder,
-            }
-        )
-    except Exception:
-        # Fail silently if there's any issue
-        logger.debug("audit log enrichment with actor failed.")
-        pass

backend/pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
     "fido2>=2.1.1,<3",
     "humanize>=4.15.0,<5",
     "huey>=2.5.5,<3",
+    "kafka-python>=2.0.6,<3",
     "python-docx>=1.2.0,<2",
     "docxtpl>=0.20.2,<0.21",
     "numpy>=2.4.1,<3",

backend/serdes/views.py

@@ -20,9 +20,7 @@
 from iam.models import User
 from serdes.serializers import LoadBackupSerializer
 
-from auditlog.models import LogEntry
 from django.db.models.signals import post_save
-from core.custom_middleware import add_user_info_to_log_entry
 from django.apps import apps
 from django.conf import settings
 from auditlog.context import disable_auditlog
@@ -77,9 +75,9 @@ class LoadBackupView(APIView):
     serializer_class = LoadBackupSerializer
 
     def load_backup(self, request, decompressed_data, backup_version, current_version):
-        # Temporarily disconnect the problematic signal
-        post_save.disconnect(add_user_info_to_log_entry, sender=LogEntry)
-
+        # The LogEntry enrichment receiver disconnected here in #1707 is gone:
+        # folder is now captured inline via AbstractBaseModel.get_additional_data,
+        # so loaddata of auditlog.logentry fixtures no longer triggers enrichment.
         backup_buffer = io.StringIO()
         try:
             management.call_command(
@@ -178,7 +176,6 @@ def fixture_callback(sender, **kwargs):
             return Response({}, status=status.HTTP_400_BAD_REQUEST)
         finally:
             post_save.disconnect(fixture_callback)
-            post_save.connect(add_user_info_to_log_entry, sender=LogEntry)
 
         # Enforce LICENSE_SEATS after successful restore
         license_seats = getattr(settings, "LICENSE_SEATS", None)

backend/uv.lock

@@ -4,3 +4,6 @@
 class WebhooksConfig(AppConfig):
     default_auto_field = "django.db.models.BigAutoField"
     name = "webhooks"
+
+    def ready(self):
+        from . import signals  # noqa: F401  connects the audit-log forwarding receiver

backend/webhooks/apps.py

@@ -0,0 +1,83 @@
+# Generated by Django 6.0.6 on 2026-06-15 09:04
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("webhooks", "0003_remove_webhookendpoint_new_owner_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="webhookendpoint",
+            name="body_format",
+            field=models.CharField(
+                choices=[
+                    ("ciso_native", "CISO Assistant (HMAC-signed)"),
+                    ("ocsf", "OCSF"),
+                    ("raw", "Raw LogEntry"),
+                ],
+                default="ocsf",
+                help_text="Canonical event schema for audit sinks.",
+                max_length=20,
+            ),
+        ),
+        migrations.AddField(
+            model_name="webhookendpoint",
+            name="headers",
+            field=models.JSONField(
+                blank=True,
+                default=dict,
+                help_text='Static headers added to each request, e.g. {"Authorization": "Splunk <token>"}. Used for audit-sink auth.',
+            ),
+        ),
+        migrations.AddField(
+            model_name="webhookendpoint",
+            name="kafka_config",
+            field=models.JSONField(
+                blank=True,
+                default=dict,
+                help_text="Kafka transport: {bootstrap_servers, topic, config:{...}}.",
+            ),
+        ),
+        migrations.AddField(
+            model_name="webhookendpoint",
+            name="kind",
+            field=models.CharField(
+                choices=[("integration", "Integration"), ("audit_sink", "Audit sink")],
+                default="integration",
+                max_length=20,
+            ),
+        ),
+        migrations.AddField(
+            model_name="webhookendpoint",
+            name="transport",
+            field=models.CharField(
+                choices=[("http", "HTTP"), ("kafka", "Kafka")],
+                default="http",
+                help_text="Delivery transport (audit sinks only).",
+                max_length=10,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="webhookendpoint",
+            name="secret",
+            field=models.CharField(
+                blank=True,
+                default="",
+                help_text="HMAC signing secret (integration webhooks only).",
+                max_length=100,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="webhookendpoint",
+            name="url",
+            field=models.URLField(
+                blank=True,
+                default="",
+                help_text="Consumer URL (HTTP transport).",
+                max_length=512,
+            ),
+        ),
+    ]

backend/webhooks/migrations/0004_webhookendpoint_body_format_webhookendpoint_headers_and_more.py

@@ -33,6 +33,19 @@ class PayloadFormats(models.TextChoices):
         THIN = "thin", "Thin"
         FULL = "full", "Full"
 
+    class Kind(models.TextChoices):
+        INTEGRATION = "integration", "Integration"
+        AUDIT_SINK = "audit_sink", "Audit sink"
+
+    class Transport(models.TextChoices):
+        HTTP = "http", "HTTP"
+        KAFKA = "kafka", "Kafka"
+
+    class BodyFormat(models.TextChoices):
+        CISO_NATIVE = "ciso_native", "CISO Assistant (HMAC-signed)"
+        OCSF = "ocsf", "OCSF"
+        RAW = "raw", "Raw LogEntry"
+
     payload_format = models.CharField(
         verbose_name="Payload Format",
         max_length=10,
@@ -41,6 +54,33 @@ class PayloadFormats(models.TextChoices):
         help_text="The format of the webhook payload sent to this endpoint.",
     )
 
+    # An "audit_sink" forwards the audit log (LogEntry stream) to an external
+    # SIEM; an "integration" is the user-facing model-event webhook. Audit sinks
+    # are admin/org-managed and hidden from the user webhook list.
+    kind = models.CharField(
+        max_length=20,
+        choices=Kind.choices,
+        default=Kind.INTEGRATION,
+    )
+    transport = models.CharField(
+        max_length=10,
+        choices=Transport.choices,
+        default=Transport.HTTP,
+        help_text="Delivery transport (audit sinks only).",
+    )
+    body_format = models.CharField(
+        max_length=20,
+        choices=BodyFormat.choices,
+        default=BodyFormat.OCSF,
+        help_text="Canonical event schema for audit sinks.",
+    )
+    headers = models.JSONField(
+        default=dict,
+        blank=True,
+        help_text="Static headers added to each request, e.g. "
+        '{"Authorization": "Splunk <token>"}. Used for audit-sink auth.',
+    )
+
     owner = models.ForeignKey(
         Actor,
         related_name="webhook_endpoints",
@@ -51,10 +91,24 @@ class PayloadFormats(models.TextChoices):
     )
 
     url = models.URLField(
-        max_length=512, help_text="The consumer URL to send webhook events to."
+        max_length=512,
+        blank=True,
+        default="",
+        help_text="Consumer URL (HTTP transport).",
     )
 
-    secret = models.CharField(max_length=100, help_text="HMAC signing secret.")
+    kafka_config = models.JSONField(
+        default=dict,
+        blank=True,
+        help_text="Kafka transport: {bootstrap_servers, topic, config:{...}}.",
+    )
+
+    secret = models.CharField(
+        max_length=100,
+        blank=True,
+        default="",
+        help_text="HMAC signing secret (integration webhooks only).",
+    )
 
     event_types = models.ManyToManyField(
         WebhookEventType,
@@ -78,6 +132,8 @@ def __str__(self):
 
     def clean(self):
         super().clean()
+        if self.transport != self.Transport.HTTP:
+            return
         if getattr(settings, "WEBHOOK_ALLOW_PRIVATE_IPS", False):
             return
         try:
@@ -91,8 +147,6 @@ def clean(self):
             )
 
     def save(self, *args, **kwargs):
-        """
-        On save, ensure a secret exists if one wasn't provided.
-        """
-        self.full_clean()  # Run validation
+        """Run full model validation (clean + field checks) before persisting."""
+        self.full_clean()
         super().save(*args, **kwargs)