Add a pipeline to collect Sigma rules

vulnerabilities/improvers/__init__.py

@@ -30,6 +30,7 @@
     enhance_with_metasploit as enhance_with_metasploit_v2,
 )
 from vulnerabilities.pipelines.v2_improvers import flag_ghost_packages as flag_ghost_packages_v2
+from vulnerabilities.pipelines.v2_improvers import sigma_rules
 from vulnerabilities.pipelines.v2_improvers import unfurl_version_range as unfurl_version_range_v2
 from vulnerabilities.utils import create_registry
 
@@ -70,5 +71,6 @@
         compute_advisory_todo_v2.ComputeToDo,
         unfurl_version_range_v2.UnfurlVersionRangePipeline,
         compute_advisory_todo.ComputeToDo,
+        sigma_rules.SigmaRulesImproverPipeline,
     ]
 )

vulnerabilities/migrations/0104_detectionrule.py

@@ -0,0 +1,68 @@
+# Generated by Django 4.2.25 on 2025-12-04 00:14
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0103_codecommit_impactedpackage_affecting_commits_and_more"),
+    ]
+
+    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",
+                    ),
+                ),
+            ],
+        ),
+    ]

vulnerabilities/models.py

@@ -3414,3 +3414,45 @@ class CodeCommit(models.Model):
 
     class Meta:
         unique_together = ("commit_hash", "vcs_url")
+
+
+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 is code used to identify malicious activity or security threats.
+    """
+
+    rule_type = models.CharField(
+        max_length=50,
+        choices=DetectionRuleTypes.choices,
+        help_text="The type of the detection rule content (e.g., YARA, Sigma).",
+    )
+
+    source_url = models.URLField(
+        max_length=1024, help_text="URL to the original source or reference for this rule."
+    )
+
+    rule_metadata = models.JSONField(
+        null=True,
+        blank=True,
+        help_text="Additional structured data such as tags, or author information.",
+    )
+
+    rule_text = models.TextField(help_text="The content of the detection signature.")
+
+    advisory = models.ForeignKey(
+        AdvisoryV2,
+        related_name="detection_rules",
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+    )

vulnerabilities/pipelines/v2_improvers/sigma_rules.py

@@ -0,0 +1,126 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# 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 vulnerabilities.models import AdvisoryAlias
+from vulnerabilities.models import DetectionRule
+from vulnerabilities.models import DetectionRuleTypes
+from vulnerabilities.pipelines import VulnerableCodePipeline
+from vulnerabilities.utils import find_all_cve
+
+
+class SigmaRulesImproverPipeline(VulnerableCodePipeline):
+    pipeline_id = "sigma_rules"
+    repo_url = "git+https://github.com/SigmaHQ/sigma"
+    license_url = "https://github.com/SigmaHQ/Detection-Rule-License"
+
+    @classmethod
+    def steps(cls):
+        return (
+            cls.clone_repo,
+            cls.collect_and_store_rules,
+            cls.clean_downloads,
+        )
+
+    def clone_repo(self):
+        self.log(f"Cloning `{self.repo_url}`")
+        self.vcs_response = fetch_via_vcs(self.repo_url)
+
+    def collect_and_store_rules(self):
+        """
+        Collect Sigma YAML rules from the destination directory and store/update
+        them as DetectionRule objects.
+        """
+
+        base_directory = Path(self.vcs_response.dest_dir)
+        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):
+            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"AdvisoryAlias {cve_id}: {file_path.name} not found.")
+                    continue
+
+            for adv in found_advisories:
+                DetectionRule.objects.update_or_create(
+                    rule_text=raw_text,
+                    rule_type=DetectionRuleTypes.SIGMA,
+                    advisory=adv,
+                    defaults={
+                        "rule_metadata": rule_metadata,
+                        "source_url": rule_url,
+                    },
+                )
+
+            if not found_advisories:
+                DetectionRule.objects.update_or_create(
+                    rule_text=raw_text,
+                    rule_type=DetectionRuleTypes.SIGMA,
+                    advisory=None,
+                    defaults={
+                        "rule_metadata": rule_metadata,
+                        "source_url": rule_url,
+                    },
+                )
+
+    def clean_downloads(self):
+        if self.vcs_response:
+            self.log(f"Removing cloned repository")
+            self.vcs_response.delete()
+
+    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

@@ -0,0 +1,84 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import os
+from datetime import datetime
+from unittest import mock
+from unittest.mock import MagicMock
+
+import pytest
+
+from vulnerabilities.models import AdvisoryAlias
+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__))
+
+TEST_REPO_DIR = os.path.join(BASE_DIR, "../../test_data/sigma")
+
+
+@pytest.mark.django_db
+@mock.patch("vulnerabilities.pipelines.v2_improvers.sigma_rules.fetch_via_vcs")
+def test_sigma_rules_db_improver(mock_fetch_via_vcs):
+    mock_vcs = MagicMock()
+    mock_vcs.dest_dir = TEST_REPO_DIR
+    mock_vcs.delete = MagicMock()
+    mock_fetch_via_vcs.return_value = mock_vcs
+
+    adv1 = AdvisoryV2.objects.create(
+        advisory_id="VCIO-123-0001",
+        datasource_id="ds",
+        avid="ds/VCIO-123-0001",
+        unique_content_id="sgsdg45",
+        url="https://test.com",
+        date_collected=datetime.now(),
+    )
+    adv2 = AdvisoryV2.objects.create(
+        advisory_id="VCIO-123-1002",
+        datasource_id="ds",
+        avid="ds/VCIO-123-1002",
+        unique_content_id="6hd4d6f",
+        url="https://test.com",
+        date_collected=datetime.now(),
+    )
+    adv3 = AdvisoryV2.objects.create(
+        advisory_id="VCIO-123-1003",
+        datasource_id="ds",
+        avid="ds/VCIO-123-1003",
+        unique_content_id="sd6h4sh",
+        url="https://test.com",
+        date_collected=datetime.now(),
+    )
+
+    alias1 = AdvisoryAlias.objects.create(alias="CVE-2025-33053")
+    alias2 = AdvisoryAlias.objects.create(alias="CVE-2025-10035")
+    alias3 = AdvisoryAlias.objects.create(alias="CVE-2010-5278")
+    adv1.aliases.add(alias1)
+    adv2.aliases.add(alias2)
+    adv3.aliases.add(alias3)
+
+    improver = SigmaRulesImproverPipeline()
+    improver.execute()
+
+    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",
+    }
+    assert sigma_rule.advisory == adv1
+    assert (
+        sigma_rule.rule_text
+        == "title: Potential Exploitation of RCE Vulnerability CVE-2025-33053 - Image Load\nid: 04fc4b22-91a6-495a-879d-0144fec5ec03\nrelated:\n    - id: abe06362-a5b9-4371-8724-ebd00cd48a04\n      type: similar\n    - id: 9a2d8b3e-f5a1-4c68-9e21-7d9e1cf8a123\n      type: similar\nstatus: experimental\ndescription: |\n    Detects potential exploitation of remote code execution vulnerability CVE-2025-33053\n    by monitoring suspicious image loads from WebDAV paths. The exploit involves malicious executables from\n    attacker-controlled WebDAV servers loading the Windows system DLLs like gdi32.dll, netapi32.dll, etc.\nreferences:\n    - https://msrc.microsoft.com/update-guide/en-US/vulnerability/CVE-2025-33053\n    - https://research.checkpoint.com/2025/stealth-falcon-zero-day/\nauthor: Swachchhanda Shrawan Poudel (Nextron Systems)\ndate: 2025-06-13\ntags:\n    - attack.command-and-control\n    - attack.execution\n    - attack.defense-evasion\n    - attack.t1218\n    - attack.lateral-movement\n    - attack.t1105\n    - detection.emerging-threats\n    - cve.2025-33053\nlogsource:\n    category: image_load\n    product: windows\ndetection:\n    selection_img_path:\n        Image|startswith: '\\\\\\\\'\n        Image|contains: '\\DavWWWRoot\\'\n    selection_img_bin:\n        Image|endswith:\n            - '\\route.exe'\n            - '\\netsh.exe'\n            - '\\makecab.exe'\n            - '\\dxdiag.exe'\n            - '\\ipconfig.exe'\n            - '\\explorer.exe'\n    condition: all of selection_*\nfalsepositives:\n    - Unknown\nlevel: high"
+    )

vulnerabilities/tests/test_data/sigma/CVE-2010-5278/web_cve_2010_5278_exploitation_attempt.yml

@@ -0,0 +1,26 @@
+title: CVE-2010-5278 Exploitation Attempt
+id: a4a899e8-fd7a-49dd-b5a8-7044def72d61
+status: test
+description: |
+  MODx manager - Local File Inclusion:Directory traversal vulnerability in manager/controllers/default/resource/tvs.php in MODx Revolution 2.0.2-pl, and possibly earlier,
+  when magic_quotes_gpc is disabled, allows remote attackers to read arbitrary files via a .. (dot dot) in the class_key parameter.
+references:
+    - https://github.com/projectdiscovery/nuclei-templates
+author: Subhash Popuri (@pbssubhash)
+date: 2021-08-25
+modified: 2023-01-02
+tags:
+    - attack.initial-access
+    - attack.t1190
+    - cve.2010-5278
+    - detection.emerging-threats
+logsource:
+    category: webserver
+detection:
+    selection:
+        cs-uri-query|contains: /manager/controllers/default/resource/tvs.php?class_key=../../../../../../../../../../windows/win.ini%00
+    condition: selection
+falsepositives:
+    - Scanning from Nuclei
+    - Unknown
+level: critical