WIP

Author: thomaspatzke

poetry.lock

@@ -27,6 +27,7 @@ pyparsing = "^3.1"
 pyyaml = "^6.0"
 requests = "^2.31"
 jinja2 = "^3.1"
+types-pyyaml = "^6.0.12.20240917"
 
 [tool.poetry.group.dev.dependencies]
 black = "^24.4.2"

pyproject.toml

@@ -1,18 +1,19 @@
 from collections import defaultdict
 import re
-from typing import ClassVar, Dict, Optional, Pattern, Tuple
+from typing import Any, ClassVar, Dict, List, Optional, Pattern, Tuple, cast
 
 from sigma.conversion.base import TextQueryBackend
 from sigma.conversion.state import ConversionState
 from sigma.pipelines.test import dummy_test_pipeline
 from sigma.processing.pipeline import ProcessingItem, ProcessingPipeline
 from sigma.processing.transformations import FieldMappingTransformation
+from sigma.rule.rule import SigmaRule
 from sigma.types import CompareOperators, SigmaCompareExpression
 
 
 class TextQueryTestBackend(TextQueryBackend):
-    name: str = "Test backend"
-    formats: Dict[str, str] = {
+    name: ClassVar[str] = "Test backend"
+    formats: ClassVar[Dict[str, str]] = {
         "default": "Default format",
         "test": "Dummy test format",
         "state": "Test format that obtains information from state",
@@ -29,15 +30,15 @@ class TextQueryTestBackend(TextQueryBackend):
     eq_token: ClassVar[str] = "="
 
     field_quote: ClassVar[str] = "'"
-    field_quote_pattern: ClassVar[Pattern] = re.compile("^\\w+$")
+    field_quote_pattern: ClassVar[Pattern[str]] = re.compile("^\\w+$")
 
     str_quote: ClassVar[str] = '"'
     escape_char: ClassVar[str] = "\\"
     wildcard_multi: ClassVar[str] = "*"
     wildcard_single: ClassVar[str] = "?"
     add_escaped: ClassVar[str] = ":"
     filter_chars: ClassVar[str] = "&"
-    bool_values: ClassVar[Dict[bool, str]] = {
+    bool_values: ClassVar[Dict[bool, Optional[str]]] = {
         True: "1",
         False: "0",
     }
@@ -52,7 +53,7 @@ class TextQueryTestBackend(TextQueryBackend):
 
     re_expression: ClassVar[str] = "{field}=/{regex}/"
     re_escape_char: ClassVar[str] = "\\"
-    re_escape: ClassVar[Tuple[str]] = ("/", "bar")
+    re_escape: ClassVar[List[str]] = ["/", "bar"]
 
     case_sensitive_match_expression = "{field} casematch {value}"
     case_sensitive_startswith_expression: ClassVar[str] = "{field} startswith_cased {value}"
@@ -115,8 +116,12 @@ class TextQueryTestBackend(TextQueryBackend):
         "test": "Test correlation method",
     }
     default_correlation_method: ClassVar[str] = "test"
-    default_correlation_query: ClassVar[str] = {"test": "{search}\n{aggregate}\n{condition}"}
-    temporal_correlation_query: ClassVar[str] = {"test": "{search}\n\n{aggregate}\n\n{condition}"}
+    default_correlation_query: ClassVar[Dict[str, str]] = {
+        "test": "{search}\n{aggregate}\n{condition}"
+    }
+    temporal_correlation_query: ClassVar[Dict[str, str]] = {
+        "test": "{search}\n\n{aggregate}\n\n{condition}"
+    }
 
     correlation_search_single_rule_expression: ClassVar[str] = "{query}"
     correlation_search_multi_rule_expression: ClassVar[str] = "{queries}"
@@ -169,33 +174,39 @@ def __init__(
         processing_pipeline: Optional[ProcessingPipeline] = None,
         collect_errors: bool = False,
         testparam: Optional[str] = None,
-        **kwargs,
+        **kwargs: Dict[str, Any],
     ):
         super().__init__(processing_pipeline, collect_errors, **kwargs)
         self.testparam = testparam
 
-    def finalize_query_test(self, rule, query, index, state):
-        return "[ " + self.finalize_query_default(rule, query, index, state) + " ]"
+    def finalize_query_test(
+        self, rule: SigmaRule, query: str, index: int, state: ConversionState
+    ) -> str:
+        return "[ " + cast(str, self.finalize_query_default(rule, query, index, state)) + " ]"
 
-    def finalize_output_test(self, queries):
-        return self.finalize_output_default(queries)
+    def finalize_output_test(self, queries: List[str]) -> str:
+        return cast(str, self.finalize_output_default(queries))
 
-    def finalize_query_state(self, rule, query, index, state: ConversionState):
+    def finalize_query_state(
+        self, rule: SigmaRule, query: str, index: int, state: ConversionState
+    ) -> str:
         return (
             "index="
-            + state.processing_state.get("index", "default")
+            + cast(str, state.processing_state.get("index", "default"))
             + " ("
-            + self.finalize_query_default(rule, query, index, state)
+            + cast(str, self.finalize_query_default(rule, query, index, state))
             + ")"
         )
 
-    def finalize_output_state(self, queries):
-        return self.finalize_output_default(queries)
+    def finalize_output_state(self, queries: List[str]) -> str:
+        return cast(str, self.finalize_output_default(queries))
 
-    def finalize_query_list_of_dict(self, rule, query, index, state):
-        return self.finalize_query_default(rule, query, index, state)
+    def finalize_query_list_of_dict(
+        self, rule: SigmaRule, query: str, index: int, state: ConversionState
+    ) -> str:
+        return cast(str, self.finalize_query_default(rule, query, index, state))
 
-    def finalize_output_list_of_dict(self, queries):
+    def finalize_output_list_of_dict(self, queries: List[str]) -> List[Dict[str, Optional[str]]]:
         return [
             (
                 {"query": query, "test": self.testparam}
@@ -205,18 +216,22 @@ def finalize_output_list_of_dict(self, queries):
             for query in self.finalize_output_default(queries)
         ]
 
-    def finalize_query_bytes(self, rule, query, index, state):
-        return self.finalize_query_default(rule, query, index, state)
+    def finalize_query_bytes(
+        self, rule: SigmaRule, query: str, index: int, state: ConversionState
+    ) -> str:
+        return cast(str, self.finalize_query_default(rule, query, index, state))
 
-    def finalize_output_bytes(self, queries):
+    def finalize_output_bytes(self, queries: List[str]) -> bytes:
         return bytes("\x00".join(self.finalize_output_default(queries)), "utf-8")
 
-    def finalize_query_str(self, rule, query, index, state):
-        return self.finalize_query_default(rule, query, index, state)
+    def finalize_query_str(
+        self, rule: SigmaRule, query: str, index: int, state: ConversionState
+    ) -> str:
+        return cast(str, self.finalize_query_default(rule, query, index, state))
 
-    def finalize_output_str(self, queries):
+    def finalize_output_str(self, queries: List[str]) -> str:
         return "\n".join(self.finalize_output_default(queries))
 
 
 class MandatoryPipelineTestBackend(TextQueryTestBackend):
-    requires_pipeline: bool = True
+    requires_pipeline: ClassVar[bool] = True

sigma/backends/test/backend.py

@@ -1,7 +1,7 @@
 from dataclasses import dataclass, field
 from functools import reduce
 from pathlib import Path
-from typing import Callable, Dict, Iterable, List, Optional, Union, IO
+from typing import Callable, Dict, Iterable, List, Optional, Union, IO, cast
 from uuid import UUID
 
 import yaml
@@ -15,13 +15,16 @@
 )
 from sigma.rule import SigmaRule, SigmaRuleBase
 from sigma.filters import SigmaFilter
+from typing import TypeVar, Union
+
+NestedDict = Dict[str, Union[str, int, float, bool, None, "NestedDict"]]
 
 
 @dataclass
 class SigmaCollection:
     """Collection of Sigma rules"""
 
-    rules: List[SigmaRuleBase]
+    rules: List[Union[SigmaRule, SigmaCorrelationRule]]
     errors: List[SigmaError] = field(default_factory=list)
     ids_to_rules: Dict[UUID, SigmaRuleBase] = field(
         init=False, repr=False, hash=False, compare=False
@@ -30,7 +33,7 @@ class SigmaCollection:
         init=False, repr=False, hash=False, compare=False
     )
 
-    def __post_init__(self):
+    def __post_init__(self) -> None:
         """
         Map rule identifiers to rules and resolve rule references in correlation rules.
         """
@@ -42,7 +45,7 @@ def __post_init__(self):
             if rule.name is not None:
                 self.names_to_rules[rule.name] = rule
 
-    def resolve_rule_references(self):
+    def resolve_rule_references(self) -> None:
         """
         Resolve rule references in correlation rules to the actual rule objects and sort the rules
         by reference order (rules that are referenced by other rules come first).
@@ -55,12 +58,21 @@ def resolve_rule_references(self):
                 rule.resolve_rule_references(self)
 
         # Extract all filters from the rules
-        filters: List[SigmaFilter] = [rule for rule in self.rules if isinstance(rule, SigmaFilter)]
+        filters: List[SigmaFilter] = [
+            cast(SigmaFilter, rule) for rule in self.rules if isinstance(rule, SigmaFilter)
+        ]
         self.rules = [rule for rule in self.rules if not isinstance(rule, SigmaFilter)]
 
         # Apply filters on each rule and replace the rule with the filtered rule
         self.rules = (
-            [reduce(lambda r, f: f.apply_on_rule(r), filters, rule) for rule in self.rules]
+            [
+                reduce(
+                    lambda r, f: f.apply_on_rule(r) if isinstance(r, SigmaRule) else r,
+                    filters,
+                    rule,
+                )
+                for rule in self.rules
+            ]
             if filters
             else self.rules
         )
@@ -71,7 +83,7 @@ def resolve_rule_references(self):
     @classmethod
     def from_dicts(
         cls,
-        rules: List[dict],
+        rules: List[NestedDict],
         collect_errors: bool = False,
         source: Optional[SigmaRuleLocation] = None,
     ) -> "SigmaCollection":
@@ -84,7 +96,7 @@ def from_dicts(
         errors = []
         parsed_rules = list()
         prev_rule = None
-        global_rule = dict()
+        global_rule: NestedDict = dict()
 
         for i, rule in zip(range(1, len(rules) + 1), rules):
             if isinstance(

sigma/collection.py

@@ -148,7 +148,7 @@ def __init__(
         self,
         processing_pipeline: Optional[ProcessingPipeline] = None,
         collect_errors: bool = False,
-        **backend_options: Dict,
+        **backend_options: Dict[str, Any],
     ):
         self.processing_pipeline = processing_pipeline
         self.errors = list()
@@ -846,7 +846,7 @@ class variables. If this is not sufficient, the respective methods can be implem
     re_escape_char: ClassVar[Optional[str]] = (
         None  # Character used for escaping in regular expressions
     )
-    re_escape: ClassVar[Tuple[str]] = ()  # List of strings that are escaped
+    re_escape: ClassVar[List[str]] = []  # List of strings that are escaped
     re_escape_escape_char: bool = True  # If True, the escape character is also escaped
     re_flag_prefix: bool = (
         True  # If True, the flags are prepended as (?x) group at the beginning of the regular expression, e.g. (?i). If this is not supported by the target, it should be set to False.