normalization of timestamp in pre detector (#646)

* add parsing of timestamp
* make timestamp field configurable
* add config options for source_format and timezones
* add processing warning tests
* update changelog

---------

Co-authored-by: ekneg54 <[email protected]>

CHANGELOG.md

@@ -3,6 +3,11 @@
 ## next release
 ### Breaking
 ### Features
+
+* `pre_detector` now normalizes timestamps with configurable parameters timestamp_field, source_format, source_timezone and target_timezone
+* `pre_detector` now writes tags in failure cases
+* `ProcessingWarnings` now can write `tags` to the event
+
 ### Improvements
 ### Bugfix
 

logprep/abc/processor.py

@@ -324,6 +324,9 @@ def _handle_warning_error(self, event, rule, error, failure_tags=None):
         else:
             add_and_overwrite(event, "tags", sorted(list({*tags, *failure_tags})))
         if isinstance(error, ProcessingWarning):
+            if error.tags:
+                tags = tags if tags else []
+                add_and_overwrite(event, "tags", sorted(list({*error.tags, *tags, *failure_tags})))
             self.result.warnings.append(error)
         else:
             self.result.warnings.append(ProcessingWarning(str(error), rule, event))

logprep/processor/base/exceptions.py

@@ -71,7 +71,8 @@ def __init__(self, message: str, rule: "Rule", event: dict):
 class ProcessingWarning(Warning):
     """A warning occurred - log the warning, but continue processing the event."""
 
-    def __init__(self, message: str, rule: "Rule", event: dict):
+    def __init__(self, message: str, rule: "Rule", event: dict, tags: List[str] = None):
+        self.tags = tags if tags else []
         rule.metrics.number_of_warnings += 1
         message = f"{message}, {rule.id=}, {rule.description=}, {event=}"
         super().__init__(f"{self.__class__.__name__}: {message}")

logprep/processor/pre_detector/processor.py

@@ -36,10 +36,11 @@
 from attr import define, field, validators
 
 from logprep.abc.processor import Processor
+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_field_to, get_dotted_field_value
-from logprep.util.time import TimeParser
+from logprep.util.time import TimeParser, TimeParserException
 
 
 class PreDetector(Processor):
@@ -75,7 +76,7 @@ class Config(Processor.Config):
         """
         Path to a YML file or a list of paths to YML files with dictionaries of IPs.
         For string format see :ref:`getters`.
-        It is used by the Predetector to throw alerts if one of the IPs is found
+        It is used by the PreDetector to throw alerts if one of the IPs is found
         in fields that were defined in a rule.
 
         It uses IPs or networks in the CIDR format as keys and can contain expiration
@@ -92,17 +93,37 @@ class Config(Processor.Config):
     def _ip_alerter(self):
         return IPAlerter(self._config.alert_ip_list_path)
 
-    def _apply_rules(self, event, rule):
+    def normalize_timestamp(self, rule: PreDetectorRule, timestamp: str) -> str:
+        """method for normalizing the timestamp"""
+        try:
+            parsed_datetime = TimeParser.parse_datetime(
+                timestamp, rule.source_format, rule.source_timezone
+            )
+            return (
+                parsed_datetime.astimezone(rule.target_timezone).isoformat().replace("+00:00", "Z")
+            )
+        except TimeParserException as error:
+            error_message = "Could not parse timestamp"
+            raise (
+                ProcessingWarning(
+                    error_message,
+                    rule,
+                    self.result.event,
+                    tags=["_pre_detector_timeparsing_failure"],
+                )
+            ) from error
+
+    def _apply_rules(self, event: dict, rule: PreDetectorRule):
         if not (
             self._ip_alerter.has_ip_fields(rule)
             and not self._ip_alerter.is_in_alerts_list(rule, event)
         ):
             self._get_detection_result(event, rule)
         for detection, _ in self.result.data:
             detection["creation_timestamp"] = TimeParser.now().isoformat()
-            timestamp = get_dotted_field_value(event, "@timestamp")
+            timestamp = get_dotted_field_value(event, rule.timestamp_field)
             if timestamp is not None:
-                detection["@timestamp"] = timestamp
+                detection[rule.timestamp_field] = self.normalize_timestamp(rule, timestamp)
 
     def _get_detection_result(self, event: dict, rule: PreDetectorRule):
         pre_detection_id = get_dotted_field_value(event, "pre_detection_id")
@@ -114,7 +135,9 @@ def _get_detection_result(self, event: dict, rule: PreDetectorRule):
         self.result.data.append((detection_result, self._config.outputs))
 
     @staticmethod
-    def _generate_detection_result(pre_detection_id: str, event: dict, rule: PreDetectorRule):
+    def _generate_detection_result(
+        pre_detection_id: str, event: dict, rule: PreDetectorRule
+    ) -> dict:
         detection_result = rule.detection_data
         detection_result["rule_filter"] = rule.filter_str
         detection_result["description"] = rule.description

logprep/processor/pre_detector/rule.py

@@ -88,6 +88,31 @@
     ip_fields:
     - some_ip_field
 
+The pre_detector also has the option to normalize the timestamp. 
+To configure this the following parameters can be set in the rule configuration.
+
+..  code-block:: yaml
+    :linenos:
+    :caption: Example
+
+    filter: 'some_field: "very malicious!"'
+    pre_detector:
+      case_condition: directly
+      id: RULE_ONE_ID
+      mitre:
+      - attack.something1
+      - attack.something2
+      severity: critical
+      title: Rule one
+      timestamp_field: <field which includes the timestamp to be normalized>
+      source_format: <the format of the timestamp in strftime format or ISO8601 or UNIX>
+      sorce_timezone: <the timezone of the timestamp>
+      target_timezone: <the timezone after normalization>
+    description: Some malicious event.
+
+All of these new parameters are configurable and default to 
+standard values if not explicitly set.
+
 .. autoclass:: logprep.processor.pre_detector.rule.PreDetectorRule.Config
    :members:
    :undoc-members:
@@ -97,6 +122,7 @@
 
 from functools import cached_property
 from typing import Optional, Union
+from zoneinfo import ZoneInfo
 
 from attrs import asdict, define, field, validators
 
@@ -106,6 +132,15 @@
 class PreDetectorRule(Rule):
     """Check if documents match a filter."""
 
+    special_field_types = {
+        *Rule.special_field_types,
+        "source_format",
+        "source_timezone",
+        "target_timezone",
+        "timestamp_field",
+        "failure_tags",
+    }
+
     @define(kw_only=True)
     class Config(Rule.Config):  # pylint: disable=too-many-instance-attributes
         """RuleConfig for Predetector"""
@@ -122,7 +157,7 @@ class Config(Rule.Config):  # pylint: disable=too-many-instance-attributes
         """The type of the triggered rule, mostly `directly`."""
         ip_fields: list = field(validator=validators.instance_of(list), factory=list)
         """Specify a list of fields that can be compared to a list of IPs,
-        which can be configured in the pipeline for the predetector.
+        which can be configured in the pipeline for the pre_detector.
         If this field was specified, then the rule will *only* trigger in case one of
         the IPs from the list is also available in the specified fields."""
         sigma_fields: Union[list, bool] = field(
@@ -133,6 +168,25 @@ class Config(Rule.Config):  # pylint: disable=too-many-instance-attributes
             validator=validators.optional(validators.instance_of(str)), default=None
         )
         """A link to the rule if applicable."""
+        source_format: list = field(
+            validator=validators.instance_of(str),
+            default="ISO8601",
+        )
+        """the source format that can be given for normalizing the timestamp defaults to :code:`ISO8601`"""
+        timestamp_field: str = field(validator=validators.instance_of(str), default="@timestamp")
+        """the field which has the given timestamp to be normalized defaults to :code:`@timestamp`"""
+        source_timezone: ZoneInfo = field(
+            validator=[validators.instance_of(ZoneInfo)], converter=ZoneInfo, default="UTC"
+        )
+        """ timezone of source_fields defaults to :code:`UTC`"""
+        target_timezone: ZoneInfo = field(
+            validator=[validators.instance_of(ZoneInfo)], converter=ZoneInfo, default="UTC"
+        )
+        """ timezone for target_field defaults to :code:`UTC`"""
+        failure_tags: list = field(
+            validator=validators.instance_of(list), default=["pre_detector_failure"]
+        )
+        """ tags to be added if processing of the rule fails"""
 
     def __eq__(self, other: "PreDetectorRule") -> bool:
         return all(
@@ -160,4 +214,20 @@ def ip_fields(self) -> list:
     def description(self) -> str:
         return self._config.description
 
+    @property
+    def source_format(self) -> str:
+        return self._config.source_format
+
+    @property
+    def target_timezone(self) -> str:
+        return self._config.target_timezone
+
+    @property
+    def source_timezone(self) -> str:
+        return self._config.source_timezone
+
+    @property
+    def timestamp_field(self) -> str:
+        return self._config.timestamp_field
+
     # pylint: enable=C0111