Merge pull request #288 from SigmaHQ:issue-cs-17

Introduced interpret_special option to ReplaceStringTransformation

Author: thomaspatzke

sigma/processing/transformations.py

@@ -765,14 +765,17 @@ class ReplaceStringTransformation(StringValueTransformation):
     capture groups. Normally, the replacement operates on the plain string representation of the
     SigmaString. This allows also to include special characters and placeholders in the replacement.
     By enabling the skip_special parameter, the replacement is only applied to the plain string
-    parts of a SigmaString and special characters and placeholders are left untouched.
+    parts of a SigmaString and special characters and placeholders are left untouched. The
+    interpret_special option determines for skip_special if special characters and placeholders are
+    interpreted in the replacement result or not.
 
     The replacement is implemented with re.sub() and can use all features available there.
     """
 
     regex: str
     replacement: str
     skip_special: bool = False
+    interpret_special: bool = False
 
     def __post_init__(self):
         super().__post_init__()
@@ -787,7 +790,9 @@ def apply_string_value(self, field: str, val: SigmaString) -> SigmaString:
         if isinstance(val, SigmaString):
             if self.skip_special:
                 return val.map_parts(
-                    lambda s: self.re.sub(self.replacement, s), lambda p: isinstance(p, str)
+                    lambda s: self.re.sub(self.replacement, s),
+                    lambda p: isinstance(p, str),
+                    self.interpret_special,
                 )
             else:
                 sigma_string_plain = str(val)

sigma/types.py

@@ -501,14 +501,21 @@ def map_parts(
         self,
         func: Callable[[Union[str, SpecialChars]], Optional[Union[str, SpecialChars]]],
         filter_func: Callable[[Union[str, SpecialChars]], bool] = lambda x: True,
+        interpret_special: bool = False,
     ) -> "SigmaString":
         s = self.__class__()
-        s.s = tuple(
-            filter(
-                lambda x: x is not None,  # filter out None results
-                (func(item) if filter_func(item) else item for item in self.iter_parts()),
-            )
-        )
+        parts = []
+        for item in self.iter_parts():
+            if filter_func(item):
+                result = func(item)
+                if result is not None:
+                    if interpret_special:
+                        parts.extend(SigmaString(result).s)
+                    else:
+                        parts.append(result)
+            else:
+                parts.append(item)
+        s.s = tuple(parts)
         return s
 
     def convert(

tests/test_processing_transformations.py

@@ -1416,13 +1416,43 @@ def test_replace_string_skip_specials(dummy_pipeline):
             },
         }
     )
-    transformation = ReplaceStringTransformation("^.*\\\\", "/", True)
+    transformation = ReplaceStringTransformation("^.*\\\\", "/?/", True)
     transformation.apply(dummy_pipeline, sigma_rule)
     assert sigma_rule.detection.detections["test"] == SigmaDetection(
         [
             SigmaDetection(
                 [
-                    SigmaDetectionItem("field1", [], [SigmaString("*/value")]),
+                    SigmaDetectionItem("field1", [], [SigmaString("*/\\?/value")]),
+                    SigmaDetectionItem("field2", [], [SigmaNumber(123)]),
+                ]
+            )
+        ]
+    )
+
+
+def test_replace_string_skip_specials_with_interpret_specials(dummy_pipeline):
+    sigma_rule = SigmaRule.from_dict(
+        {
+            "title": "Test",
+            "logsource": {"category": "test"},
+            "detection": {
+                "test": [
+                    {
+                        "field1": "*\\value",
+                        "field2": 123,
+                    }
+                ],
+                "condition": "test",
+            },
+        }
+    )
+    transformation = ReplaceStringTransformation("^.*\\\\", "/?/", True, True)
+    transformation.apply(dummy_pipeline, sigma_rule)
+    assert sigma_rule.detection.detections["test"] == SigmaDetection(
+        [
+            SigmaDetection(
+                [
+                    SigmaDetectionItem("field1", [], [SigmaString("*/?/value")]),
                     SigmaDetectionItem("field2", [], [SigmaNumber(123)]),
                 ]
             )

tests/test_types.py

@@ -491,6 +491,18 @@ def test_string_map_parts(sigma_string):
     )
 
 
+def test_string_map_parts_interpret_special(sigma_string):
+    assert sigma_string.map_parts(lambda x: x.upper(), lambda x: isinstance(x, str), True).s == (
+        SpecialChars.WILDCARD_MULTI,
+        "TEST",
+        SpecialChars.WILDCARD_MULTI,
+        "STR",
+        SpecialChars.WILDCARD_MULTI,
+        "ING",
+        SpecialChars.WILDCARD_MULTI,
+    )
+
+
 def test_cased_string(sigma_string):
     assert SigmaCasedString.from_sigma_string(sigma_string) == SigmaCasedString("*Test*Str\\*ing*")