Add a test for Suricata rules

Simplify the rules structure
Fix a duplication avid in UI/API

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

Author: ziadhany

vulnerabilities/api_v2.py

@@ -1412,17 +1412,22 @@ class Meta:
 
 
 class DetectionRuleSerializer(serializers.ModelSerializer):
-    advisory_avid = serializers.SlugRelatedField(
-        many=True, read_only=True, slug_field="avid", source="related_advisories"
-    )
+    advisory_avid = serializers.SerializerMethodField()
 
     class Meta:
         model = DetectionRule
         fields = ["rule_type", "source_url", "rule_metadata", "rule_text", "advisory_avid"]
 
+    def get_advisory_avid(self, obj):
+        avids = set(advisory.avid for advisory in obj.related_advisories.all())
+        return sorted(list(avids))
+
 
 class DetectionRuleViewSet(viewsets.ReadOnlyModelViewSet):
-    queryset = DetectionRule.objects.prefetch_related("related_advisories")
+    advisories_prefetch = Prefetch(
+        "related_advisories", queryset=AdvisoryV2.objects.only("id", "avid").distinct()
+    )
+    queryset = DetectionRule.objects.prefetch_related(advisories_prefetch)
     serializer_class = DetectionRuleSerializer
     throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
     filter_backends = [filters.DjangoFilterBackend]

vulnerabilities/improvers/__init__.py

@@ -34,7 +34,6 @@
 from vulnerabilities.pipelines.v2_improvers import flag_ghost_packages as flag_ghost_packages_v2
 from vulnerabilities.pipelines.v2_improvers import relate_severities
 from vulnerabilities.pipelines.v2_improvers import sigma_rules
-from vulnerabilities.pipelines.v2_improvers import sigma_rules
 from vulnerabilities.pipelines.v2_improvers import suricata_rules
 from vulnerabilities.pipelines.v2_improvers import unfurl_version_range as unfurl_version_range_v2
 from vulnerabilities.pipelines.v2_improvers import yara_rules
@@ -78,13 +77,11 @@
         unfurl_version_range_v2.UnfurlVersionRangePipeline,
         compute_advisory_todo.ComputeToDo,
         collect_ssvc_trees.CollectSSVCPipeline,
-
         relate_severities.RelateSeveritiesPipeline,
         sigma_rules.SigmaHQImproverPipeline,
         sigma_rules.SigmaSamuraiMDRImproverPipeline,
         sigma_rules.SigmaMbabinskiImproverPipeline,
         sigma_rules.P4T12ICKSigmaImproverPipeline,
-
         yara_rules.ProtectionsArtifactsYara,
         yara_rules.YaraRulesYara,
         yara_rules.XumeiquerForensicsYara,
@@ -116,6 +113,7 @@
         yara_rules.Dr4k0niaYaraRules,
         yara_rules.Umair9747YaraRules,
         clamav_rules.ClamVRulesImproverPipeline,
-        suricata_rules.SuricataRulesImproverPipeline,
+        suricata_rules.SudohyakSuricataImproverPipeline,
+        suricata_rules.OISFSuricataImproverPipeline,
     ]
 )

vulnerabilities/migrations/0104_advisorydetectionrule.py

@@ -3689,13 +3689,3 @@ class DetectionRule(models.Model):
         related_name="detection_rules",
         help_text="Advisories associated with this DetectionRule.",
     )
-
-
-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"

vulnerabilities/migrations/0105_alter_advisorydetectionrule_advisory.py

@@ -19,10 +19,11 @@
 import requests
 
 from vulnerabilities.models import AdvisoryAlias
+from vulnerabilities.models import AdvisoryV2
 from vulnerabilities.models import DetectionRule
 from vulnerabilities.models import DetectionRuleTypes
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
-from vulnerabilities.utils import find_all_cve
+from vulnerabilities.utils import find_all_cve_rule
 
 
 def extract_cvd(cvd_path, output_dir):
@@ -93,13 +94,6 @@ def parse_hdb_file(hdb_path: Path) -> List[dict]:
     return signatures
 
 
-def extract_cve_id(name: str):
-    """Normalize underscores and extract the first CVE ID from a string, or None."""
-    normalized = name.replace("_", "-")
-    cves = [cve.upper() for cve in find_all_cve(normalized)]
-    return cves[0] if cves else None
-
-
 class ClamVRulesImproverPipeline(VulnerableCodeBaseImporterPipelineV2):
     """
     Pipeline that downloads ClamAV database (main.cvd), extracts signatures,
@@ -159,36 +153,27 @@ def collect_and_store_advisories(self):
             self.extract_cvd_dir / "main.ndb"
         ):
             name = rule_entry.get("name", "")
-            cve_id = extract_cve_id(name)
-            found_advisories = set()
-
-            if cve_id:
-                try:
-                    if alias := AdvisoryAlias.objects.get(alias=cve_id):
-                        for adv in alias.advisories.all():
-                            found_advisories.add(adv)
-                except AdvisoryAlias.DoesNotExist:
-                    self.log(f"Advisory {cve_id} not found.")
-
-            for adv in found_advisories:
-                DetectionRule.objects.update_or_create(
-                    rule_text=str(rule_entry),
-                    rule_type=DetectionRuleTypes.CLAMAV,
-                    advisory=adv,
-                    defaults={
-                        "source_url": self.MAIN_DATABASE_URL,
-                    },
-                )
-
-            if not found_advisories:
-                DetectionRule.objects.update_or_create(
-                    rule_text=str(rule_entry),
-                    rule_type=DetectionRuleTypes.CLAMAV,
-                    advisory=None,
-                    defaults={
-                        "source_url": self.MAIN_DATABASE_URL,
-                    },
-                )
+            cve_ids = find_all_cve_rule(name)
+
+            advisories = set()
+            for cve_id in cve_ids:
+                alias = AdvisoryAlias.objects.filter(alias=cve_id).first()
+                if alias:
+                    for adv in alias.advisories.all():
+                        advisories.add(adv)
+                else:
+                    advs = AdvisoryV2.objects.filter(advisory_id=cve_id)
+                    for adv in advs:
+                        advisories.add(adv)
+
+            detection_rule, _ = DetectionRule.objects.get_or_create(
+                rule_text=str(rule_entry),
+                rule_type=DetectionRuleTypes.CLAMAV,
+                source_url=self.MAIN_DATABASE_URL,
+            )
+
+            for adv in advisories:
+                detection_rule.related_advisories.add(adv)
 
     def clean_downloads(self):
         """Clean up downloaded files."""