add match_ocsf

sigma/processing/conditions/rule.py

@@ -28,19 +28,31 @@ class LogsourceCondition(RuleProcessingCondition):
     the condition returns true if any of the associated rules have the required log source fields.
     """
 
+    class_uid: Optional[str] = field(default=None)
     category: Optional[str] = field(default=None)
     product: Optional[str] = field(default=None)
     service: Optional[str] = field(default=None)
 
     def __post_init__(self) -> None:
-        self.logsource = SigmaLogSource(self.category, self.product, self.service)
+        self.logsource = SigmaLogSource(
+            self.category,
+            self.product,
+            self.service,
+            custom_attributes={"class_uid": self.class_uid},
+        )
 
     def match(
         self,
         rule: Union[SigmaRule, SigmaCorrelationRule],
     ) -> bool:
         if isinstance(rule, SigmaRule):
-            return rule.logsource in self.logsource
+            res = (
+                str(rule.logsource.category) == str(self.logsource.category)
+                and str(rule.logsource.product) == str(self.logsource.product)
+                and str(rule.logsource.service) == str(self.logsource.service)
+            )
+            res = res and self.match_ocsf(rule=rule)
+            return res
         elif isinstance(rule, SigmaCorrelationRule):
             # Will only return true if the rules have been resolved in advance
             for ref in rule.rules:
@@ -49,6 +61,14 @@ def match(
                         return True
             return False
 
+    def match_ocsf(self, rule: SigmaRule) -> bool:
+        rule_ocsf = rule.custom_attributes.get("ocsf")
+        if rule_ocsf:
+            res = str(rule_ocsf["class_uid"]) == str(self.logsource.custom_attributes["class_uid"])
+        else:
+            res = True
+        return res
+
 
 @dataclass
 class RuleContainsFieldCondition(RuleDetectionItemCondition):

tests/test_processing_conditions.py

@@ -90,7 +90,7 @@ def test_processing_condition_multiple_pipelines_set(dummy_processing_pipeline):
 
 
 def test_logsource_match(sigma_rule):
-    assert LogsourceCondition(category="test_category").match(
+    assert not LogsourceCondition(category="test_category").match(
         sigma_rule,
     )