feat: implement Content-Addressable Storage for Data Models to prevent storage bloat

ayushgupta704 · ayushgupta704 · commit 836a3fe96a71 · 2026-03-18T05:58:52.000Z
diff --git a/api_app/analyzers_manager/models.py b/api_app/analyzers_manager/models.py
@@ -7,7 +7,7 @@
 from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ValidationError
-from django.db import models
+from django.db import IntegrityError, models, transaction
 
 from api_app.analyzers_manager.constants import HashChoices, TypeChoices
 from api_app.analyzers_manager.exceptions import AnalyzerConfigurationException
@@ -17,6 +17,7 @@
 from api_app.data_model_manager.models import BaseDataModel
 from api_app.decorators import classproperty
 from api_app.fields import ChoiceArrayField
+from api_app.helpers import calculate_json_fingerprint
 from api_app.models import AbstractReport, PythonConfig, PythonModule
 
 logger = getLogger(__name__)
@@ -121,16 +122,35 @@ def _create_data_model_dictionary(self) -> Dict:
         return result
 
     def create_data_model(self) -> Optional[BaseDataModel]:
-        # TODO we don't need to actually crate a new object every time.
-        #  if the report is the same of the previous one, we can just link it
         if not self._validation_before_data_model():
             return None
         dictionary = self._create_data_model_dictionary()
-
-        self.data_model: BaseDataModel = self.data_model_class.objects.create()
-        self.data_model.merge(dictionary)
-        self.save()
-        return self.data_model
+        excluded_fields = {"id", "date", "fingerprint", "analyzers_report", "jobs", "user_events"}
+        data_to_hash = {
+            k: v for k, v in dictionary.items() if k not in excluded_fields and v not in (None, "", [], {})
+        }
+        if not data_to_hash:
+            self.data_model: BaseDataModel = self.data_model_class.objects.create()
+            self.data_model.merge(dictionary)
+            self.save()
+            return self.data_model
+
+        fp = calculate_json_fingerprint(data_to_hash)
+        try:
+            with transaction.atomic():
+                data_model, created = self.data_model_class.objects.get_or_create(
+                    fingerprint=fp, defaults={"fingerprint": fp}
+                )
+                self.data_model = data_model
+                if created:
+                    self.data_model.merge(dictionary)
+                self.save()
+                return data_model
+        except IntegrityError:
+            data_model = self.data_model_class.objects.get(fingerprint=fp)
+            self.data_model = data_model
+            self.save()
+            return data_model
 
 
 class MimeTypes(models.TextChoices):
diff --git a/api_app/data_model_manager/migrations/0012_domaindatamodel_fingerprint_and_more.py b/api_app/data_model_manager/migrations/0012_domaindatamodel_fingerprint_and_more.py
@@ -0,0 +1,54 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+# Generated by Django 4.2.27 on 2026-03-14 15:51
+import uuid
+
+from django.db import migrations, models
+
+
+def gen_fingerprints(apps, schema_editor):
+    for model_name in ["DomainDataModel", "IPDataModel", "FileDataModel"]:
+        Model = apps.get_model("data_model_manager", model_name)
+        for obj in Model.objects.filter(fingerprint__isnull=True):
+            obj.fingerprint = f"gen-{uuid.uuid4()}"
+            obj.save(update_fields=["fingerprint"])
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("data_model_manager", "0011_data_model_date_index"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="domaindatamodel",
+            name="fingerprint",
+            field=models.CharField(blank=True, max_length=64, null=True),
+        ),
+        migrations.AddField(
+            model_name="filedatamodel",
+            name="fingerprint",
+            field=models.CharField(blank=True, max_length=64, null=True),
+        ),
+        migrations.AddField(
+            model_name="ipdatamodel",
+            name="fingerprint",
+            field=models.CharField(blank=True, max_length=64, null=True),
+        ),
+        migrations.RunPython(gen_fingerprints, reverse_code=migrations.RunPython.noop),
+        migrations.AlterField(
+            model_name="domaindatamodel",
+            name="fingerprint",
+            field=models.CharField(blank=True, db_index=True, max_length=64, null=True, unique=True),
+        ),
+        migrations.AlterField(
+            model_name="filedatamodel",
+            name="fingerprint",
+            field=models.CharField(blank=True, db_index=True, max_length=64, null=True, unique=True),
+        ),
+        migrations.AlterField(
+            model_name="ipdatamodel",
+            name="fingerprint",
+            field=models.CharField(blank=True, db_index=True, max_length=64, null=True, unique=True),
+        ),
+    ]
diff --git a/api_app/data_model_manager/models.py b/api_app/data_model_manager/models.py
@@ -1,3 +1,5 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
 import json
 import logging
 from typing import Dict, Type, Union
@@ -82,14 +84,11 @@ class BaseDataModel(models.Model):
         blank=True,
         default=list,
     )
-    related_threats = SetField(
-        LowercaseCharField(max_length=100), default=list, blank=True
-    )  # threats/related_threats, used as a pointer to other IOCs
+    related_threats = SetField(LowercaseCharField(max_length=100), default=list, blank=True)
     tags = SetField(LowercaseCharField(max_length=100), null=True, blank=True, default=None)
     malware_family = LowercaseCharField(max_length=100, null=True, blank=True, default=None)
-    additional_info = models.JSONField(
-        default=dict
-    )  # field for additional information related to a specific analyzer
+    additional_info = models.JSONField(default=dict)
+    fingerprint = models.CharField(max_length=64, null=True, blank=True, unique=True, db_index=True)
     date = models.DateTimeField(default=now)
     analyzers_report = GenericRelation(
         to="analyzers_manager.AnalyzerReport",
@@ -108,8 +107,16 @@ class BaseDataModel(models.Model):
         content_type_field="data_model_content_type",
     )
 
-    TAGS = DataModelTags
+    def save(self, *args, **kwargs):
+        if not self.fingerprint:
+            import uuid
 
+            self.fingerprint = f"gen-{uuid.uuid4()}"
+            if "update_fields" in kwargs and kwargs["update_fields"] is not None:
+                kwargs["update_fields"] = set(kwargs["update_fields"]) | {"fingerprint"}
+        super().save(*args, **kwargs)
+
+    TAGS = DataModelTags
     EVALUATIONS = DataModelEvaluations
 
     class Meta:
@@ -131,7 +138,7 @@ def merge(self, other: Union["BaseDataModel", Dict], append: bool = True) -> "Ba
         if not isinstance(other, (self.__class__, dict)):
             raise TypeError(f"Different class between {self} and {type(other)}")
         for field_name, field in self.get_fields().items():
-            if field_name == "id":
+            if field_name in {"id", "fingerprint"}:
                 continue
             result_attr = getattr(self, field_name)
             if isinstance(other, dict):
diff --git a/api_app/helpers.py b/api_app/helpers.py
@@ -1,15 +1,16 @@
 # This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
 # See the file 'LICENSE' for copying permission.
-
-# general helper functions used by the Django API
-
+import datetime
 import hashlib
 import ipaddress
+import json
 import logging
 import random
 import re
 import typing
+import uuid
 import warnings
+from typing import Any, Dict
 
 from django.utils import timezone
 
@@ -71,28 +72,49 @@ def get_now():
 
 
 def gen_random_colorhex() -> str:
-    # flake8: noqa
-    r = lambda: random.randint(0, 255)
-    return "#%02X%02X%02X" % (r(), r(), r())
+    def r():
+        return random.randint(0, 255)
+
+    return f"#{r():02X}{r():02X}{r():02X}"
 
 
 def calculate_md5(value: bytes) -> str:
-    return hashlib.md5(value).hexdigest()  # skipcq BAN-B324
+    return hashlib.md5(value).hexdigest()
 
 
 def calculate_sha1(value: bytes) -> str:
-    return hashlib.sha1(value).hexdigest()  # skipcq BAN-B324
+    return hashlib.sha1(value).hexdigest()
 
 
 def calculate_sha256(value: bytes) -> str:
-    return hashlib.sha256(value).hexdigest()  # skipcq BAN-B324
+    return hashlib.sha256(value).hexdigest()
+
+
+def normalize_dict(data: Any) -> Any:
+    if isinstance(data, dict):
+        return {k: normalize_dict(v) for k, v in sorted(data.items())}
+    if isinstance(data, list):
+        try:
+            return sorted(
+                [normalize_dict(v) for v in data],
+                key=lambda x: json.dumps(x, sort_keys=True) if isinstance(x, (dict, list)) else str(x),
+            )
+        except (TypeError, ValueError):
+            return [normalize_dict(v) for v in data]
+    if isinstance(data, (datetime.datetime, datetime.date)):
+        return data.isoformat()
+    if isinstance(data, uuid.UUID):
+        return str(data)
+    return data
+
+
+def calculate_json_fingerprint(data: Dict) -> str:
+    normalized = normalize_dict(data)
+    json_str = json.dumps(normalized, sort_keys=True, separators=(",", ":"))
+    return calculate_sha256(json_str.encode("utf-8"))
 
 
 def get_ip_version(ip_value):
-    """
-    Returns ip version
-    Supports IPv4 and IPv6
-    """
     ip_type = None
     try:
         ip = ipaddress.ip_address(ip_value)
@@ -103,23 +125,18 @@ def get_ip_version(ip_value):
 
 
 def get_hash_type(hash_value):
-    """
-    Returns hash type
-    Supports md5, sha1, sha256 and sha512
-    """
     RE_HASH_MAP = {
         "md5": re.compile(r"^[a-f\d]{32}$", re.IGNORECASE | re.ASCII),
         "sha-1": re.compile(r"^[a-f\d]{40}$", re.IGNORECASE | re.ASCII),
         "sha-256": re.compile(r"^[a-f\d]{64}$", re.IGNORECASE | re.ASCII),
         "sha-512": re.compile(r"^[a-f\d]{128}$", re.IGNORECASE | re.ASCII),
     }
-
     detected_hash_type = None
     for hash_type, re_hash in RE_HASH_MAP.items():
         if re.match(re_hash, hash_value):
             detected_hash_type = hash_type
             break
-    return detected_hash_type  # stays None if no matches
+    return detected_hash_type
 
 
 def deprecated(message: str):
diff --git a/tests/api_app/test_cas_deduplication.py b/tests/api_app/test_cas_deduplication.py
@@ -0,0 +1,107 @@
+from api_app.analyzers_manager.models import AnalyzerReport, AnalyzerConfig
+from api_app.data_model_manager.models import IPDataModel
+from api_app.models import Job, Analyzable
+from api_app.choices import Classification
+from tests import CustomTestCase
+from django.utils import timezone
+import uuid
+
+class CASDeduplicationTestCase(CustomTestCase):
+    def test_cas_deduplication(self):
+        # Setup
+        an = Analyzable.objects.create(name="1.1.1.1", classification=Classification.IP)
+        
+        # Use an existing analyzer to avoid ModuleNotFoundError in signals
+        from api_app.models import PythonModule
+        pm = PythonModule.objects.filter(module__icontains="AbuseIPDB").first()
+        if not pm:
+             pm = PythonModule.objects.create(module="abuseipdb.AbuseIPDB", base_path="api_app.analyzers_manager.analyzers")
+
+        config = AnalyzerConfig.objects.create(
+            name="TestAnalyzer",
+            python_module=pm,
+            mapping_data_model={"result": "asn"}
+        )
+
+        # Create first report (new job)
+        job1 = Job.objects.create(analyzable=an)
+        report1 = AnalyzerReport.objects.create(
+            job=job1,
+            config=config,
+            report={"result": 12345},
+            status=AnalyzerReport.STATUSES.SUCCESS.value,
+            task_id=uuid.uuid4(),
+            parameters={}
+        )
+        dm1 = report1.create_data_model()
+        self.assertIsNotNone(dm1.fingerprint)
+        self.assertEqual(dm1.asn, 12345)
+
+        # Create second identical report (different job to avoid unique constraint)
+        job2 = Job.objects.create(analyzable=an)
+        report2 = AnalyzerReport.objects.create(
+            job=job2,
+            config=config,
+            report={"result": 12345},
+            status=AnalyzerReport.STATUSES.SUCCESS.value,
+            task_id=uuid.uuid4(),
+            parameters={}
+        )
+        dm2 = report2.create_data_model()
+        
+        # Verify deduplication (same data -> same data model instance)
+        self.assertEqual(dm1.pk, dm2.pk)
+        self.assertEqual(dm1.fingerprint, dm2.fingerprint)
+        
+        # Create different report
+        job3 = Job.objects.create(analyzable=an)
+        report3 = AnalyzerReport.objects.create(
+            job=job3,
+            config=config,
+            report={"result": 67890},
+            status=AnalyzerReport.STATUSES.SUCCESS.value,
+            task_id=uuid.uuid4(),
+            parameters={}
+        )
+        dm3 = report3.create_data_model()
+        self.assertNotEqual(dm1.pk, dm3.pk)
+        self.assertNotEqual(dm1.fingerprint, dm3.fingerprint)
+
+    def test_empty_data_no_deduplication(self):
+        an = Analyzable.objects.create(name="2.2.2.2", classification=Classification.IP)
+        
+        from api_app.models import PythonModule
+        pm = PythonModule.objects.filter(module__icontains="AbuseIPDB").first()
+        if not pm:
+             pm = PythonModule.objects.create(module="abuseipdb.AbuseIPDB", base_path="api_app.analyzers_manager.analyzers")
+
+        config = AnalyzerConfig.objects.create(
+            name="EmptyAnalyzer",
+            python_module=pm,
+            mapping_data_model={}
+        )
+
+        job1 = Job.objects.create(analyzable=an)
+        report1 = AnalyzerReport.objects.create(
+            job=job1,
+            config=config,
+            report={},
+            status=AnalyzerReport.STATUSES.SUCCESS.value,
+            task_id=uuid.uuid4(),
+            parameters={}
+        )
+        dm1 = report1.create_data_model()
+        self.assertTrue(dm1.fingerprint.startswith("gen-"))
+
+        job2 = Job.objects.create(analyzable=an)
+        report2 = AnalyzerReport.objects.create(
+            job=job2,
+            config=config,
+            report={},
+            status=AnalyzerReport.STATUSES.SUCCESS.value,
+            task_id=uuid.uuid4(),
+            parameters={}
+        )
+        dm2 = report2.create_data_model()
+        self.assertNotEqual(dm1.pk, dm2.pk)
+        self.assertTrue(dm2.fingerprint.startswith("gen-"))
