feat: Sigma rules pt.4

pyproject.toml

@@ -18,6 +18,7 @@ dependencies = [
     "tenacity",
     "PyYAML",
     "typing-extensions",
+    "pySigma==1.3.3",
 ]
 
 classifiers = [

src/cosl/rules.py

@@ -96,6 +96,8 @@
 )
 
 import yaml
+from sigma.exceptions import SigmaError
+from sigma.rule import SigmaRule  # type: ignore[reportMissingTypeStubs]
 
 from . import CosTool, JujuTopology
 from .types import (
@@ -652,11 +654,16 @@ def add(self, rule_dict: Mapping[str, Any]) -> None:
         if not rule_dict:
             return
         rule_copy = copy.deepcopy(dict(rule_dict))
-        if "rules" in rule_copy:
-            for r in rule_copy["rules"]:
-                self.rules.append(self._inject_topology(r))
-        else:
-            self.rules.append(self._inject_topology(rule_copy))
+        rules = rule_copy.get("rules") or [rule_copy]
+        for rule in rules:
+            try:
+                # leverage pysigma's rule validation by casting to SigmaRule and back
+                sigma_rule = SigmaRule.from_dict(cast(Dict[str, Any], rule))
+                self.rules.append(
+                    self._inject_topology(cast(SigmaRuleFormat, sigma_rule.to_dict()))
+                )
+            except (KeyError, AttributeError, SigmaError) as e:
+                logger.error("Invalid sigma_rule: %s", e)
 
     def add_path(self, dir_path: Union[str, Path], *, recursive: bool = False) -> None:
         """Add sigma rules from a directory or file path.

tests/features/sigma_rules.feature

@@ -45,7 +45,7 @@ Feature: Sigma Rules processing
   # --- Loading from filesystem ---
 
   Scenario: Loading a single Sigma rule file
-    When I load the sigma rule file "ssh_failed_login.yaml"
+    When I load the valid sigma rule file "ssh_failed_login.yaml"
     Then the rules collection contains 1 rule
     And the rule titled "Failed SSH Login Attempt" exists
     And the rule has id "5f3a4e20-1b2c-4d5e-9f8a-7b6c3d4e5f6a"
@@ -56,22 +56,22 @@ Feature: Sigma Rules processing
     Then the rules collection contains 5 rules
 
   Scenario: A collection file expands into multiple rules
-    When I load the sigma rule file "collection.yaml"
+    When I load the valid sigma rule file "collection.yaml"
     Then the rules collection contains 2 rules
     And the rule titled "Disk Space Critical" exists
     And the rule titled "Memory Exhaustion Warning" exists
 
   Scenario: Loading a nonexistent path does nothing
-    When I load the sigma rule file "nonexistent.yaml"
+    When I load the valid sigma rule file "nonexistent.yaml"
     Then the rules collection is empty
 
   Scenario: Topology is injected when loading from file
-    When I load the sigma rule file "high_cpu_process.yaml"
+    When I load the valid sigma rule file "high_cpu_process.yaml"
     Then the rule has label "juju_model" set to "testmodel"
     And the rule has label "juju_application" set to "myapp"
 
   Scenario: Existing file labels are preserved on load
-    When I load the sigma rule file "unauthorized_api_access.yaml"
+    When I load the valid sigma rule file "unauthorized_api_access.yaml"
     Then the rule has label "team" set to "security"
     And the rule has label "juju_model" set to "testmodel"
 
@@ -80,3 +80,13 @@ Feature: Sigma Rules processing
   Scenario: Adding a rule does not mutate the caller's input
     When I add a sigma rule and keep a reference to the original dict
     Then the original dict is unchanged
+
+  # --- Rule validation ---
+
+  Scenario: Loading an invalid rule
+    When I load the invalid sigma rule file "invalid_rule.yaml"
+    Then the rules collection is empty
+
+  Scenario: Loading a mix of valid and invalid rules
+    When I load the invalid sigma rule file "valid_and_invalid.yaml"
+    Then the rules collection contains 1 rule

tests/sigma_rules/invalid_rules/invalid_rule.yaml

@@ -0,0 +1,11 @@
+# invalid rule: logsource has been removed
+title: Failed SSH Login Attempt
+id: 5f3a4e20-1b2c-4d5e-9f8a-7b6c3d4e5f6a
+status: experimental
+description: Detects failed SSH login attempts
+detection:
+  selection:
+    event_type: authentication_failure
+    service: sshd
+  condition: selection
+level: medium

tests/sigma_rules/invalid_rules/valid_and_invalid.yaml

@@ -0,0 +1,25 @@
+rules:
+  # invalid (detection has been removed)
+  - title: Disk Space Critical
+    id: c3d4e5f6-a7b8-9012-cdef-345678901234
+    status: stable
+    description: Detects critically low disk space
+    logsource:
+      category: system
+      product: linux
+    level: critical
+
+  - title: Memory Exhaustion Warning
+    id: d4e5f6a7-b8c9-0123-defa-456789012345
+    status: experimental
+    description: Detects when system memory is nearly exhausted
+    logsource:
+      category: system
+      product: linux
+    detection:
+      selection:
+        metric: memory_percent
+      filter:
+        memory_percent|gte: 90
+      condition: selection and filter
+    level: high

tests/test_rules_sigma.py

@@ -10,7 +10,8 @@
 from cosl.juju_topology import JujuTopology
 from cosl.rules import SigmaRules
 
-SIGMA_RULES_DIR = Path(__file__).resolve().parent / "sigma_rules"
+VALID_SIGMA_RULES_DIR = Path(__file__).resolve().parent / "sigma_rules" / "valid_rules"
+INVALID_SIGMA_RULES_DIR = Path(__file__).resolve().parent / "sigma_rules" / "invalid_rules"
 MODEL_UUID = "53316f3c-b681-47b8-b272-9f8a2a858e0e"
 
 # --- Scenarios (auto-collect all from feature file) ---
@@ -98,14 +99,19 @@ def when_add_with_labels(sigma, label_a, label_b):
     )
 
 
-@when(parsers.parse('I load the sigma rule file "{filename}"'))
+@when(parsers.parse('I load the valid sigma rule file "{filename}"'))
 def when_load_file(sigma, filename):
-    sigma.add_path(SIGMA_RULES_DIR / filename)
+    sigma.add_path(VALID_SIGMA_RULES_DIR / filename)
+
+
+@when(parsers.parse('I load the invalid sigma rule file "{filename}"'))
+def when_load_file(sigma, filename):
+    sigma.add_path(INVALID_SIGMA_RULES_DIR / filename)
 
 
 @when("I load the sigma rules directory")
 def when_load_directory(sigma):
-    sigma.add_path(SIGMA_RULES_DIR)
+    sigma.add_path(VALID_SIGMA_RULES_DIR)
 
 
 @when(

tox.ini

@@ -9,7 +9,7 @@ isolated_build=true
 src_path = {toxinidir}/src
 tst_path = {toxinidir}/tests
 all_path = {[vars]src_path} {[vars]tst_path}
-uv_flags = --frozen --isolated --extra=dev
+uv_flags = --frozen --extra=dev
 
 [testenv]
 allowlist_externals = uv