Add config option to pre-detector for copying log fields to detection events

Author: mhoff

.pre-commit-config.yaml

@@ -29,6 +29,11 @@ repos:
     hooks:
       - id: isort
         name: isort (python)
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.18.2
+    hooks:
+      - id: mypy
+        additional_dependencies: [attrs]
   - repo: https://github.com/google/yamlfmt
     rev: v0.15.0
     hooks:

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Features
 * add support for python 3.14
+* allow pre-detector to copy a configurable list of fields from log to detection event
 
 ### Improvements
 * add workflow to partially run & check the compose example
@@ -15,6 +16,7 @@
 * fix docker-compose and k8s example setups
 * fix handling of non-string values (e.g. int) as replacement argument for `generic_resolver`
 * fix documentation for `generic_resolver` rule `append_to_list -> merge_with_target` option
+* fix grokker using a fixed directory for downloaded patterns, potentially leading to conflicts between processes
 
 ## 17.0.3
 ### Breaking

doc/source/installation.rst

@@ -40,6 +40,7 @@ contribute to them.
     git clone https://github.com/fkie-cad/Logprep.git
     cd Logprep
     pip install .
+    pip install .[dev] # if you intend to contribute
 
 To see if the installation was successful run
 :code:`logprep --version`.

logprep/abc/processor.py

@@ -2,6 +2,7 @@
 
 import logging
 import os
+import typing
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Type
 
@@ -105,27 +106,43 @@ class Config(Component.Config):
     __slots__ = [
         "_event",
         "_rule_tree",
-        "result",
+        "_result",
         "_bypass_rule_tree",
     ]
 
-    rule_class: ClassVar["Type[Rule] | None"] = None
+    rule_class: ClassVar[Type["Rule"] | None] = None
     _event: dict
     _rule_tree: RuleTree
     _strategy = None
     _bypass_rule_tree: bool
-    result: ProcessorResult | None
+    _result: ProcessorResult | None
 
     def __init__(self, name: str, configuration: "Processor.Config"):
         super().__init__(name, configuration)
         self._rule_tree = RuleTree(config=self._config.tree_config)
         self.load_rules(rules_targets=self._config.rules)
-        self.result = None
+        self._result = None
         self._bypass_rule_tree = False
         if os.environ.get("LOGPREP_BYPASS_RULE_TREE"):
             self._bypass_rule_tree = True
             logger.debug("Bypassing rule tree for processor %s", self.name)
 
+    @property
+    def result(self) -> ProcessorResult:
+        """Returns the current result object which is guaranteed to be non-None
+        during processing of an event.
+
+        Returns
+        -------
+        ProcessorResult
+            The current result to be modified in-place
+        """
+        return typing.cast(ProcessorResult, self._result)
+
+    @result.setter
+    def result(self, value: ProcessorResult):
+        self._result = value
+
     @property
     def rules(self):
         """Returns all rules
@@ -161,7 +178,7 @@ def process(self, event: dict) -> ProcessorResult:
             extra data and a list of target outputs.
 
         """
-        self.result = ProcessorResult(processor_name=self.name, event=event)  # type: ignore
+        self._result = ProcessorResult(processor_name=self.name, event=event)  # type: ignore
         logger.debug("%s processing event %s", self.describe(), event)
         if self._bypass_rule_tree:
             self._process_all_rules(event)

logprep/ng/processor/grokker/processor.py

@@ -31,6 +31,7 @@
 
 import logging
 import re
+import tempfile
 from pathlib import Path
 from zipfile import ZipFile
 
@@ -99,24 +100,24 @@ def setup(self) -> None:
         super().setup()
         custom_patterns_dir = self._config.custom_patterns_dir
         if re.search(r"http(s)?:\/\/.*?\.zip", custom_patterns_dir):
-            patterns_tmp_path = Path("/tmp/grok_patterns")
-            self._download_zip_file(source_file=custom_patterns_dir, target_dir=patterns_tmp_path)
-            for rule in self.rules:
-                rule.set_mapping_actions(patterns_tmp_path)
-            return
+            with tempfile.TemporaryDirectory("grok") as patterns_tmp_path:
+                self._download_zip_file(
+                    source_file=custom_patterns_dir, target_dir=Path(patterns_tmp_path)
+                )
+                for rule in self.rules:
+                    rule.set_mapping_actions(patterns_tmp_path)
+                return
         if custom_patterns_dir:
             for rule in self.rules:
                 rule.set_mapping_actions(custom_patterns_dir)
             return
         for rule in self.rules:
             rule.set_mapping_actions()
 
-    def _download_zip_file(self, source_file: str, target_dir: Path) -> None:
-        if not target_dir.exists():
-            logger.debug("start grok pattern download...")
-            archive = Path(f"{target_dir}.zip")
-            archive.touch()
-            archive.write_bytes(GetterFactory.from_string(source_file).get_raw())
+    def _download_zip_file(self, source_file: str, target_dir: Path):
+        logger.debug("start grok pattern download...")
+        with tempfile.TemporaryFile("wb+") as archive:
+            archive.write(GetterFactory.from_string(source_file).get_raw())
             logger.debug("finished grok pattern download.")
-            with ZipFile(str(archive), mode="r") as zip_file:
+            with ZipFile(archive, mode="r") as zip_file:
                 zip_file.extractall(target_dir)

logprep/ng/processor/pre_detector/processor.py

@@ -28,6 +28,7 @@
 .. automodule:: logprep.processor.pre_detector.rule
 """
 
+import typing
 from functools import cached_property
 from uuid import uuid4
 
@@ -38,7 +39,12 @@
 from logprep.processor.base.exceptions import ProcessingWarning
 from logprep.processor.pre_detector.ip_alerter import IPAlerter
 from logprep.processor.pre_detector.rule import PreDetectorRule
-from logprep.util.helper import add_fields_to, get_dotted_field_value
+from logprep.util.helper import (
+    FieldValue,
+    add_fields_to,
+    copy_fields_to_event,
+    get_dotted_field_value,
+)
 from logprep.util.time import TimeParser, TimeParserException
 
 
@@ -92,16 +98,16 @@ class Config(Processor.Config):
     def _ip_alerter(self) -> IPAlerter:
         return IPAlerter(self._config.alert_ip_list_path)
 
-    def normalize_timestamp(self, rule: PreDetectorRule, timestamp: str) -> str:
+    def normalize_timestamp(self, rule: PreDetectorRule, timestamp: FieldValue) -> str:
         """method for normalizing the timestamp"""
         try:
             parsed_datetime = TimeParser.parse_datetime(
-                timestamp, rule.source_format, rule.source_timezone
+                typing.cast(str, timestamp), rule.source_format, rule.source_timezone
             )
             return (
                 parsed_datetime.astimezone(rule.target_timezone).isoformat().replace("+00:00", "Z")
             )
-        except TimeParserException as error:
+        except (TimeParserException, TypeError) as error:
             raise ProcessingWarning(
                 "Could not parse timestamp",
                 rule,
@@ -132,15 +138,19 @@ def _get_detection_result(self, event: dict, rule: PreDetectorRule) -> None:
 
     @staticmethod
     def _generate_detection_result(
-        pre_detection_id: str, event: dict, rule: PreDetectorRule
+        pre_detection_id: FieldValue, event: dict, rule: PreDetectorRule
     ) -> dict:
         detection_result = {
             **rule.detection_data,
             "rule_filter": rule.filter_str,
             "description": rule.description,
             "pre_detection_id": pre_detection_id,
         }
-
-        if host_name := get_dotted_field_value(event, "host.name"):
-            detection_result.update({"host": {"name": host_name}})
+        copy_fields_to_event(
+            target_event=detection_result,
+            source_event=event,
+            dotted_field_names=rule.copy_fields_to_detection_event,
+            rule=rule,
+            skip_missing=True,
+        )
         return detection_result

logprep/processor/grokker/processor.py

@@ -31,6 +31,7 @@
 
 import logging
 import re
+import tempfile
 from pathlib import Path
 from zipfile import ZipFile
 
@@ -106,16 +107,18 @@ def _apply_rules(self, event: dict, rule: GrokkerRule):
         if not matches:
             raise ProcessingWarning("no grok pattern matched", rule, event)
 
-    def setup(self):
+    def setup(self) -> None:
         """Loads the action mapping. Has to be called before processing"""
         super().setup()
         custom_patterns_dir = self._config.custom_patterns_dir
         if re.search(r"http(s)?:\/\/.*?\.zip", custom_patterns_dir):
-            patterns_tmp_path = Path("/tmp/grok_patterns")
-            self._download_zip_file(source_file=custom_patterns_dir, target_dir=patterns_tmp_path)
-            for rule in self.rules:
-                rule.set_mapping_actions(patterns_tmp_path)
-            return
+            with tempfile.TemporaryDirectory("grok") as patterns_tmp_path:
+                self._download_zip_file(
+                    source_file=custom_patterns_dir, target_dir=Path(patterns_tmp_path)
+                )
+                for rule in self.rules:
+                    rule.set_mapping_actions(patterns_tmp_path)
+                return
         if custom_patterns_dir:
             for rule in self.rules:
                 rule.set_mapping_actions(custom_patterns_dir)
@@ -124,11 +127,9 @@ def setup(self):
             rule.set_mapping_actions()
 
     def _download_zip_file(self, source_file: str, target_dir: Path):
-        if not target_dir.exists():
-            logger.debug("start grok pattern download...")
-            archive = Path(f"{target_dir}.zip")
-            archive.touch()
-            archive.write_bytes(GetterFactory.from_string(source_file).get_raw())
+        logger.debug("start grok pattern download...")
+        with tempfile.TemporaryFile("wb+") as archive:
+            archive.write(GetterFactory.from_string(source_file).get_raw())
             logger.debug("finished grok pattern download.")
-            with ZipFile(str(archive), mode="r") as zip_file:
+            with ZipFile(archive, mode="r") as zip_file:
                 zip_file.extractall(target_dir)

logprep/processor/pre_detector/processor.py

@@ -28,6 +28,7 @@
 .. automodule:: logprep.processor.pre_detector.rule
 """
 
+import typing
 from functools import cached_property
 from uuid import uuid4
 
@@ -37,7 +38,12 @@
 from logprep.processor.base.exceptions import ProcessingWarning
 from logprep.processor.pre_detector.ip_alerter import IPAlerter
 from logprep.processor.pre_detector.rule import PreDetectorRule
-from logprep.util.helper import add_fields_to, get_dotted_field_value
+from logprep.util.helper import (
+    FieldValue,
+    add_fields_to,
+    copy_fields_to_event,
+    get_dotted_field_value,
+)
 from logprep.util.time import TimeParser, TimeParserException
 
 
@@ -101,19 +107,19 @@ class Config(Processor.Config):
     rule_class = PreDetectorRule
 
     @cached_property
-    def _ip_alerter(self):
+    def _ip_alerter(self) -> IPAlerter:
         return IPAlerter(self._config.alert_ip_list_path)
 
-    def normalize_timestamp(self, rule: PreDetectorRule, timestamp: str) -> str:
+    def normalize_timestamp(self, rule: PreDetectorRule, timestamp: FieldValue) -> str:
         """method for normalizing the timestamp"""
         try:
             parsed_datetime = TimeParser.parse_datetime(
-                timestamp, rule.source_format, rule.source_timezone
+                typing.cast(str, timestamp), rule.source_format, rule.source_timezone
             )
             return (
                 parsed_datetime.astimezone(rule.target_timezone).isoformat().replace("+00:00", "Z")
             )
-        except TimeParserException as error:
+        except (TimeParserException, TypeError) as error:
             raise ProcessingWarning(
                 "Could not parse timestamp",
                 rule,
@@ -143,16 +149,19 @@ def _get_detection_result(self, event: dict, rule: PreDetectorRule):
 
     @staticmethod
     def _generate_detection_result(
-        pre_detection_id: str, event: dict, rule: PreDetectorRule
+        pre_detection_id: FieldValue, event: dict, rule: PreDetectorRule
     ) -> dict:
-        detection_result = rule.detection_data
-        detection_result.update(
-            {
-                "rule_filter": rule.filter_str,
-                "description": rule.description,
-                "pre_detection_id": pre_detection_id,
-            }
+        detection_result = {
+            **rule.detection_data,
+            "rule_filter": rule.filter_str,
+            "description": rule.description,
+            "pre_detection_id": pre_detection_id,
+        }
+        copy_fields_to_event(
+            target_event=detection_result,
+            source_event=event,
+            dotted_field_names=rule.copy_fields_to_detection_event,
+            rule=rule,
+            skip_missing=True,
         )
-        if host_name := get_dotted_field_value(event, "host.name"):
-            detection_result.update({"host": {"name": host_name}})
         return detection_result