Update sigma rule importer to store metadata

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

Author: ziadhany

vulnerabilities/pipelines/v2_improvers/sigma_rules.py

@@ -6,12 +6,12 @@
 # See https://github.com/aboutcode-org/vulnerablecode for support or download.
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
-
+import datetime
 from pathlib import Path
 
+import yaml
 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 DetectionRule
@@ -40,58 +40,58 @@ def clone_repo(self):
     def collect_and_store_rules(self):
         """
         Collect Sigma YAML rules from the destination directory and store/update
-        them as AdvisoryDetectionRule objects.
+        them as DetectionRule objects.
         """
 
         base_directory = Path(self.vcs_response.dest_dir)
-        yaml_files = list(base_directory.rglob("**/*.yml"))
-        rules_count = len(yaml_files)
+        yaml_files = [
+            p
+            for p in base_directory.rglob("**/*.yml")
+            if not any(part in [".github", "images", "documentation"] for part in p.parts)
+        ]
 
+        rules_count = len(yaml_files)
         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):
-            if any(part in [".github", "images", "documentation"] for part in file_path.parts):
-                continue
-
-            with open(file_path, "r") as f:
-                try:
-                    rule_data = f.read()
-                except YAMLError as err:
-                    self.log(f"Invalid YAML in {file_path}: {err}. Skipping.")
-                    continue
+            raw_text = file_path.read_text(encoding="utf-8")
+            rule_documents = list(yaml.load_all(raw_text, yaml.FullLoader))
 
+            rule_metadata = extract_sigma_metadata(rule_documents)
             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.")
+                    self.log(f"AdvisoryAlias {cve_id}: {file_path.name} not found.")
                     continue
 
             for adv in found_advisories:
                 DetectionRule.objects.update_or_create(
-                    rule_text=rule_data,
+                    rule_text=raw_text,
+                    rule_type=DetectionRuleTypes.SIGMA,
                     advisory=adv,
                     defaults={
-                        "rule_type": DetectionRuleTypes.SIGMA,
+                        "rule_metadata": rule_metadata,
                         "source_url": rule_url,
                     },
                 )
 
             if not found_advisories:
                 DetectionRule.objects.update_or_create(
-                    rule_text=rule_data,
+                    rule_text=raw_text,
+                    rule_type=DetectionRuleTypes.SIGMA,
                     advisory=None,
                     defaults={
-                        "rule_type": DetectionRuleTypes.SIGMA,
+                        "rule_metadata": rule_metadata,
                         "source_url": rule_url,
                     },
                 )
-            self.log(f"Successfully processed rules.")
 
     def clean_downloads(self):
         if self.vcs_response:
@@ -100,3 +100,27 @@ def clean_downloads(self):
 
     def on_failure(self):
         self.clean_downloads()
+
+
+def extract_sigma_metadata(rule_documents):
+    """
+    Extract Sigma metadata from Sigma YAML rules
+    """
+    if not rule_documents:
+        return None
+
+    first_document = rule_documents[0]
+    metadata = {
+        "status": first_document.get("status"),
+        "author": first_document.get("author"),
+        "date": first_document.get("date"),
+        "title": first_document.get("title"),
+        "id": first_document.get("id"),
+    }
+
+    rule_date = metadata.get("date")
+
+    if isinstance(rule_date, (datetime.date, datetime.datetime)):
+        metadata["date"] = rule_date.isoformat()
+
+    return metadata

vulnerabilities/tests/pipelines/v2_improvers/test_sigma_rules.py

@@ -70,3 +70,10 @@ def test_sigma_rules_db_improver(mock_fetch_via_vcs):
     assert len(DetectionRule.objects.all()) == 3
     sigma_rule = DetectionRule.objects.first()
     assert sigma_rule.rule_type == "sigma"
+    assert sigma_rule.rule_metadata == {
+        "author": "Swachchhanda Shrawan Poudel (Nextron Systems)",
+        "date": "2025-06-13",
+        "id": "04fc4b22-91a6-495a-879d-0144fec5ec03",
+        "status": "experimental",
+        "title": "Potential Exploitation of RCE Vulnerability CVE-2025-33053 - Image " "Load",
+    }