Add a pipeline to collect Sigma rules

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

Author: ziadhany

vulnerabilities/improvers/__init__.py

@@ -31,6 +31,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
 
@@ -72,5 +73,6 @@
         unfurl_version_range_v2.UnfurlVersionRangePipeline,
         compute_advisory_todo.ComputeToDo,
         collect_ssvc_trees.CollectSSVCPipeline,
+        sigma_rules.SigmaRulesImproverPipeline,
     ]
 )

vulnerabilities/migrations/0104_advisorydetectionrule.py

@@ -0,0 +1,59 @@
+# Generated by Django 4.2.25 on 2025-11-24 19:30
+
+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="AdvisoryDetectionRule",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
+                    ),
+                ),
+                (
+                    "rule_text",
+                    models.TextField(
+                        help_text="Full text of the detection rule, script, or signature."
+                    ),
+                ),
+                (
+                    "rule_type",
+                    models.CharField(
+                        blank=True,
+                        choices=[
+                            ("yara", "YARA"),
+                            ("sigma", "Sigma Detection Rule"),
+                            ("clamav", "ClamAV Signature"),
+                        ],
+                        max_length=100,
+                    ),
+                ),
+                (
+                    "source_url",
+                    models.URLField(
+                        blank=True,
+                        help_text="URL or reference to the source of the rule (vendor feed, GitHub repo, etc.).",
+                        null=True,
+                    ),
+                ),
+                (
+                    "advisory",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="detection_rules",
+                        to="vulnerabilities.advisoryv2",
+                    ),
+                ),
+            ],
+        ),
+    ]

vulnerabilities/models.py

@@ -3489,3 +3489,31 @@ def __str__(self):
 
     class Meta:
         unique_together = ("vector", "source_advisory")
+
+
+class AdvisoryDetectionRule(models.Model):
+    """
+    A detection rule (YARA, Sigma, ClamAV) linked to an advisory.
+    """
+
+    RULE_TYPES = [
+        ("yara", "YARA"),
+        ("sigma", "Sigma Detection Rule"),
+        ("clamav", "ClamAV Signature"),
+    ]
+
+    advisory = models.ForeignKey(
+        AdvisoryV2,
+        related_name="detection_rules",
+        on_delete=models.CASCADE,
+    )
+
+    rule_text = models.TextField(help_text="Full text of the detection rule, script, or signature.")
+
+    rule_type = models.CharField(max_length=100, choices=RULE_TYPES, blank=True)
+
+    source_url = models.URLField(
+        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

@@ -0,0 +1,96 @@
+#
+# 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.
+#
+
+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.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 AdvisoryDetectionRule objects.
+        """
+
+        base_directory = Path(self.vcs_response.dest_dir)
+        yaml_files = list(base_directory.rglob("**/*.yml"))
+        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):
+            cve_ids = find_all_cve(str(file_path))
+            if not cve_ids or len(cve_ids) > 1:
+                continue
+
+            cve_id = cve_ids[0]
+
+            with open(file_path, "r") as f:
+                try:
+                    rule_data = saneyaml.load(f)
+                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)}"
+
+            for advisory in advisories:
+                AdvisoryDetectionRule.objects.update_or_create(
+                    advisory=advisory,
+                    rule_type="sigma",
+                    defaults={
+                        "rule_text": rule_text,
+                        "source_url": rule_url,
+                    },
+                )
+
+        self.log(f"Successfully added {rules_count:,d} rules advisory")
+
+    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()

vulnerabilities/tests/pipelines/v2_improvers/test_sigma_rules.py

@@ -0,0 +1,72 @@
+#
+# 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 AdvisoryDetectionRule
+from vulnerabilities.models import AdvisoryV2
+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(AdvisoryDetectionRule.objects.all()) == 3
+    sigma_rule = AdvisoryDetectionRule.objects.first()
+    assert sigma_rule.rule_type == "sigma"

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

vulnerabilities/tests/test_data/sigma/CVE-2025-10035/proc_creation_win_exploit_cve_2025_10035.yml

@@ -0,0 +1,78 @@
+title: Potential Exploitation of GoAnywhere MFT Vulnerability
+id: 6c76b3d0-afe4-4870-9443-ffe6773c5fef
+status: experimental
+description: |
+    Detects suspicious command execution by child processes of the GoAnywhere Managed File Transfer (MFT) application, which may indicate exploitation such as CVE-2025-10035.
+    This behavior is indicative of post-exploitation activity related to CVE-2025-10035, as observed in campaigns by the threat actor Storm-1175.
+references:
+    - https://www.microsoft.com/en-us/security/blog/2025/10/06/investigating-active-exploitation-of-cve-2025-10035-goanywhere-managed-file-transfer-vulnerability/
+author: MSFT (idea), Swachchhanda Shrawan Poudel (Nextron Systems)
+date: 2025-10-07
+tags:
+    - attack.initial-access
+    - attack.t1190
+    - attack.execution
+    - attack.t1059.001
+    - attack.persistence
+    - attack.t1133
+    - detection.emerging-threats
+    - cve.2025-10035
+logsource:
+    category: process_creation
+    product: windows
+detection:
+    # Detects the GoAnywhere Tomcat parent process based on path and command line arguments
+    selection_parent:
+        ParentImage|contains: '\GoAnywhere\tomcat\'
+    selection_powershell_img:
+        Image|endswith:
+            - '\powershell.exe'
+            - '\powershell_ise.exe'
+            - '\pwsh.exe'
+    selection_powershell_cmd:
+        - CommandLine|contains|all:
+              - 'IEX'
+              - 'enc'
+              - 'Hidden'
+              - 'bypass'
+        - CommandLine|re:
+              - 'net\s+user'
+              - 'net\s+group'
+              - 'query\s+session'
+        - CommandLine|contains:
+              - 'whoami'
+              - 'systeminfo'
+              - 'dsquery'
+              - 'localgroup administrators'
+              - 'nltest'
+              - 'samaccountname='
+              - 'adscredentials'
+              - 'o365accountconfiguration'
+              - '.DownloadString('
+              - '.DownloadFile('
+              - 'FromBase64String('
+              - 'System.IO.Compression'
+              - 'System.IO.MemoryStream'
+              - 'curl'
+    selection_child_cmd:
+        Image|endswith: '\cmd.exe'
+        CommandLine|contains:
+            - 'powershell'
+            - 'whoami'
+            - 'net.exe'
+            - 'net1.exe'
+            - 'rundll32'
+            - 'quser'
+            - 'nltest'
+            - 'curl'
+    selection_child_others:
+        CommandLine|contains:
+            - 'bitsadmin'
+            - 'certutil'
+            - 'mshta'
+            - 'cscript'
+            - 'wscript'
+    condition: selection_parent and (all of selection_powershell_* or 1 of selection_child_*)
+falsepositives:
+    - Legitimate administrative scripts or built-in GoAnywhere functions could potentially trigger this rule. Tuning may be required based on normal activity in your environment.
+level: high