Rename and update DetectionRule model

Update sigma rules improver

Signed-off-by: ziad hany <ziadhany2016@gmail.com>

Author: ziadhany

vulnerabilities/migrations/0105_detectionrule_delete_advisorydetectionrule.py

@@ -0,0 +1,71 @@
+# Generated by Django 4.2.25 on 2025-12-03 02:30
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0104_advisorydetectionrule"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="DetectionRule",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
+                    ),
+                ),
+                (
+                    "rule_type",
+                    models.CharField(
+                        choices=[
+                            ("yara", "Yara"),
+                            ("yara-x", "Yara-X"),
+                            ("sigma", "Sigma"),
+                            ("clamav", "ClamAV"),
+                            ("suricata", "Suricata"),
+                        ],
+                        help_text="The type of the detection rule content (e.g., YARA, Sigma).",
+                        max_length=50,
+                    ),
+                ),
+                (
+                    "source_url",
+                    models.URLField(
+                        help_text="URL to the original source or reference for this rule.",
+                        max_length=1024,
+                    ),
+                ),
+                (
+                    "rule_metadata",
+                    models.JSONField(
+                        blank=True,
+                        help_text="Additional structured data such as tags, or author information.",
+                        null=True,
+                    ),
+                ),
+                (
+                    "rule_text",
+                    models.TextField(help_text="The content of the detection signature."),
+                ),
+                (
+                    "advisory",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="detection_rules",
+                        to="vulnerabilities.advisoryv2",
+                    ),
+                ),
+            ],
+        ),
+        migrations.DeleteModel(
+            name="AdvisoryDetectionRule",
+        ),
+    ]

vulnerabilities/models.py

@@ -3416,29 +3416,43 @@ class Meta:
         unique_together = ("commit_hash", "vcs_url")
 
 
-class AdvisoryDetectionRule(models.Model):
+class DetectionRuleTypes(models.TextChoices):
+    """Defines the supported formats for security detection rules."""
+
+    YARA = "yara", "Yara"
+    YARA_X = "yara-x", "Yara-X"
+    SIGMA = "sigma", "Sigma"
+    CLAMAV = "clamav", "ClamAV"
+    SURICATA = "suricata", "Suricata"
+
+
+class DetectionRule(models.Model):
     """
-    A detection rule (YARA, Sigma, ClamAV) linked to an advisory.
+    A Detection Rule is code used to identify malicious activity or security threats.
     """
 
-    RULE_TYPES = [
-        ("yara", "YARA"),
-        ("sigma", "Sigma Detection Rule"),
-        ("clamav", "ClamAV Signature"),
-    ]
+    rule_type = models.CharField(
+        max_length=50,
+        choices=DetectionRuleTypes.choices,
+        help_text="The type of the detection rule content (e.g., YARA, Sigma).",
+    )
 
-    advisory = models.ForeignKey(
-        AdvisoryV2,
-        related_name="detection_rules",
-        on_delete=models.CASCADE,
+    source_url = models.URLField(
+        max_length=1024, help_text="URL to the original source or reference for this rule."
     )
 
-    rule_text = models.TextField(help_text="Full text of the detection rule, script, or signature.")
+    rule_metadata = models.JSONField(
+        null=True,
+        blank=True,
+        help_text="Additional structured data such as tags, or author information.",
+    )
 
-    rule_type = models.CharField(max_length=100, choices=RULE_TYPES, blank=True)
+    rule_text = models.TextField(help_text="The content of the detection signature.")
 
-    source_url = models.URLField(
+    advisory = models.ForeignKey(
+        AdvisoryV2,
+        related_name="detection_rules",
+        on_delete=models.SET_NULL,
         null=True,
         blank=True,
-        help_text="URL or reference to the source of the rule (vendor feed, GitHub repo, etc.).",
     )

vulnerabilities/pipelines/v2_improvers/sigma_rules.py

@@ -9,13 +9,13 @@
 
 from pathlib import Path
 
-import saneyaml
 from aboutcode.pipeline import LoopProgress
 from fetchcode.vcs import fetch_via_vcs
 from yaml import YAMLError
 
 from vulnerabilities.models import AdvisoryAlias
-from vulnerabilities.models import AdvisoryDetectionRule
+from vulnerabilities.models import DetectionRule
+from vulnerabilities.models import DetectionRuleTypes
 from vulnerabilities.pipelines import VulnerableCodePipeline
 from vulnerabilities.utils import find_all_cve
 
@@ -50,42 +50,48 @@ def collect_and_store_rules(self):
         self.log(f"Enhancing the vulnerability with {rules_count:,d} rule records")
         progress = LoopProgress(total_iterations=rules_count, logger=self.log)
         for file_path in progress.iter(yaml_files):
-            cve_ids = find_all_cve(str(file_path))
-            if not cve_ids or len(cve_ids) > 1:
+            if any(part in [".github", "images", "documentation"] for part in file_path.parts):
                 continue
 
-            cve_id = cve_ids[0]
-
             with open(file_path, "r") as f:
                 try:
-                    rule_data = saneyaml.load(f)
+                    rule_data = f.read()
                 except YAMLError as err:
                     self.log(f"Invalid YAML in {file_path}: {err}. Skipping.")
                     continue
 
-            advisories = set()
-            try:
-                if alias := AdvisoryAlias.objects.get(alias=cve_id):
-                    for adv in alias.advisories.all():
-                        advisories.add(adv)
-            except AdvisoryAlias.DoesNotExist:
-                self.log(f"Advisory {file_path.name} not found.")
-                continue
-
-            rule_text = saneyaml.dump(rule_data)
             rule_url = f"https://raw.githubusercontent.com/SigmaHQ/sigma/refs/heads/master/{file_path.relative_to(base_directory)}"
+            cve_ids = find_all_cve(str(file_path))
+            found_advisories = set()
+            for cve_id in cve_ids:
+                try:
+                    alias = AdvisoryAlias.objects.get(alias=cve_id)
+                    for adv in alias.advisories.all():
+                        found_advisories.add(adv)
+                except AdvisoryAlias.DoesNotExist:
+                    self.log(f"Advisory {file_path.name} not found.")
+                    continue
 
-            for advisory in advisories:
-                AdvisoryDetectionRule.objects.update_or_create(
-                    advisory=advisory,
-                    rule_type="sigma",
+            for adv in found_advisories:
+                DetectionRule.objects.update_or_create(
+                    rule_text=rule_data,
+                    advisory=adv,
                     defaults={
-                        "rule_text": rule_text,
+                        "rule_type": DetectionRuleTypes.SIGMA,
                         "source_url": rule_url,
                     },
                 )
 
-        self.log(f"Successfully added {rules_count:,d} rules advisory")
+            if not found_advisories:
+                DetectionRule.objects.update_or_create(
+                    rule_text=rule_data,
+                    advisory=None,
+                    defaults={
+                        "rule_type": DetectionRuleTypes.SIGMA,
+                        "source_url": rule_url,
+                    },
+                )
+            self.log(f"Successfully processed rules.")
 
     def clean_downloads(self):
         if self.vcs_response:

vulnerabilities/tests/pipelines/v2_improvers/test_sigma_rules.py

@@ -15,8 +15,8 @@
 import pytest
 
 from vulnerabilities.models import AdvisoryAlias
-from vulnerabilities.models import AdvisoryDetectionRule
 from vulnerabilities.models import AdvisoryV2
+from vulnerabilities.models import DetectionRule
 from vulnerabilities.pipelines.v2_improvers.sigma_rules import SigmaRulesImproverPipeline
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -67,6 +67,6 @@ def test_sigma_rules_db_improver(mock_fetch_via_vcs):
     improver = SigmaRulesImproverPipeline()
     improver.execute()
 
-    assert len(AdvisoryDetectionRule.objects.all()) == 3
-    sigma_rule = AdvisoryDetectionRule.objects.first()
+    assert len(DetectionRule.objects.all()) == 3
+    sigma_rule = DetectionRule.objects.first()
     assert sigma_rule.rule_type == "sigma"