Introduce start and end anchors into label filter regexes (#165)

grafana · Jun 25, 2024 · 0309734 · 0309734
1 parent bfa75d9
commit 0309734
Show file tree

Hide file tree

Showing 9 changed files with 203 additions and 406 deletions.
diff --git a/sigma/backends/loki/loki.py b/sigma/backends/loki/loki.py
@@ -45,11 +45,11 @@
     SigmaNull,
     SigmaNumber,
     SigmaFieldReference,
+    SpecialChars,
 )
 from warnings import warn
 from yaml import dump
 
-
 Conditions = Union[
     ConditionItem,
     ConditionNOT,
@@ -255,6 +255,9 @@ class LogQLBackend(TextQueryBackend):
     filter_chars: ClassVar[str] = ""
 
     field_replace_pattern: ClassVar[Pattern] = re.compile("[^a-zA-Z0-9_:]+")
+    anchor_replace_pattern: ClassVar[Pattern] = re.compile(
+        "^(?P<ext>\\(\\?[^)]\\))?(?P<start>\\^)?(?P<body>.*?)(?P<end>\\$)?$"
+    )
 
     negated_line_filter_operator: ClassVar[Dict[str, str]] = {
         "|=": "!=",
@@ -374,14 +377,27 @@ def __init__(
     # As str equality is case-insensitive in Sigma, but LogQL regexes are case-sensitive,
     # we need to do the same for *ALL* string values and prepend with (?i)
     def convert_str_to_re(
-        self, value: SigmaString, case_insensitive: bool = True
+        self,
+        value: SigmaString,
+        case_insensitive: bool = True,
+        field_filter: bool = False,
     ) -> SigmaRegularExpression:
         """Convert a SigmaString into a regular expression, replacing any
         wildcards with equivalent regular expression operators, and enforcing
         case-insensitive matching"""
         return SigmaRegularExpression(
             ("(?i)" if case_insensitive else "")
+            + (
+                "^"
+                if field_filter and not value.startswith(SpecialChars.WILDCARD_MULTI)
+                else ""
+            )
             + re.escape(str(value)).replace("\\?", ".").replace("\\*", ".*")
+            + (
+                "$"
+                if field_filter and not value.endswith(SpecialChars.WILDCARD_MULTI)
+                else ""
+            )
         )
 
     def select_log_parser(self, rule: SigmaRule) -> Union[str, LogQLLogParser]:
@@ -534,8 +550,16 @@ def convert_field_expression_to_line_filter(
             )
         elif isinstance(expr.value, SigmaRegularExpression):
             # Could include field name if entries are logfmt and doesn't start with wildcard
+            regexp = expr.value.regexp
+            anchors = LogQLBackend.anchor_replace_pattern.match(expr.value.regexp)
+            if anchors and anchors.group("body") and (
+                anchors.group("start") or anchors.group("end")
+            ):
+                regexp = (
+                    anchors.group("ext") if anchors.group("ext") else ""
+                ) + anchors.group("body")
             return LogQLLineFilterInfo(
-                value=expr.value.regexp,
+                value=regexp,
                 negated=getattr(expr, "negated", False),
                 deftype=LogQLDeferredType.REGEXP,
             )
@@ -671,7 +695,12 @@ def update_parsed_conditions(
                 and (not self.case_sensitive or condition.value.contains_special())
                 and not isinstance(condition.value, SigmaCasedString)
             ):
-                condition.value = self.convert_str_to_re(condition.value)
+                condition.value = self.convert_str_to_re(
+                    condition.value,
+                    field_filter=isinstance(
+                        condition, ConditionFieldEqualsValueExpression
+                    ),
+                )
         if isinstance(condition, ConditionItem):
             if isinstance(condition, ConditionNOT):
                 negated = not negated
@@ -979,7 +1008,7 @@ def convert_condition_field_eq_val_str(
             and isinstance(cond.value, SigmaString)
             and len(cond.value) > 0
         ):
-            cond.value = self.convert_str_to_re(cond.value)
+            cond.value = self.convert_str_to_re(cond.value, True, True)
             return super().convert_condition_field_eq_val_re(cond, state)
         return super().convert_condition_field_eq_val_str(cond, state)
 
@@ -990,7 +1019,7 @@ def convert_condition_field_eq_val_str_case_sensitive(
         modifiers, Sigma introduces wildcards that are then not handled correctly
         by Loki. So, in those cases, we convert the string to a regular expression."""
         if isinstance(cond.value, SigmaString) and cond.value.contains_special():
-            cond.value = self.convert_str_to_re(cond.value, False)
+            cond.value = self.convert_str_to_re(cond.value, False, True)
             return super().convert_condition_field_eq_val_re(cond, state)
         return super().convert_condition_field_eq_val_str_case_sensitive(cond, state)