diff --git a/CHANGELOG.md b/CHANGELOG.md index b88676933..cbcacc28a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ### Breaking ### Features +* generic_adder: add support for templated http urls ### Improvements diff --git a/logprep/abc/getter.py b/logprep/abc/getter.py index 70095f30c..64d2e7860 100644 --- a/logprep/abc/getter.py +++ b/logprep/abc/getter.py @@ -114,10 +114,12 @@ def get_json(self) -> dict | list: content = self._resolve_content(raw) return self._parse_json(content) - def get_collection(self) -> dict | list: + def get_collection(self, content_field: str | None = None) -> dict | list: """Gets and parses the raw content to yaml or json""" content = self._resolve_content_by_content_type() + content = Getter._apply_content_field(content, content_field) + if isinstance(content, str): content = self._parse_yaml_or_json(content) @@ -143,11 +145,10 @@ def _parse_newline_separated_list(content: str) -> list: """Helper which tries to convert content to list""" return content.splitlines() - def get_list(self, content_field: str | None = None) -> list: - """Gets list and fails otherwise""" - - content = self._resolve_content_by_content_type() - + @staticmethod + def _apply_content_field( + content: dict | list | str, content_field: str | None = None + ) -> dict | list | str: if isinstance(content, dict) and content_field is not None: content = content[content_field] elif content_field is not None: @@ -155,6 +156,15 @@ def get_list(self, content_field: str | None = None) -> list: f"Expected mapping type when content_field is set, got {type(content)}" ) + return content + + def get_list(self, content_field: str | None = None) -> list: + """Gets list and fails otherwise""" + + content = self._resolve_content_by_content_type() + + content = Getter._apply_content_field(content, content_field) + if isinstance(content, str): content = self._parse_newline_separated_list(content) diff --git a/logprep/ng/processor/generic_adder/processor.py b/logprep/ng/processor/generic_adder/processor.py index 12f930595..d722000e9 100644 --- a/logprep/ng/processor/generic_adder/processor.py +++ b/logprep/ng/processor/generic_adder/processor.py @@ -25,10 +25,13 @@ """ import typing +from typing import Sequence from logprep.ng.abc.processor import Processor +from logprep.processor.base.exceptions import ProcessingWarning from logprep.processor.base.rule import Rule from logprep.processor.generic_adder.rule import GenericAdderRule +from logprep.util.getter import RefreshableGetter from logprep.util.helper import add_fields_to @@ -37,8 +40,26 @@ class GenericAdder(Processor): rule_class = GenericAdderRule + @property + def _rules(self) -> Sequence[GenericAdderRule]: + """Returns all rules""" + return typing.cast(Sequence[GenericAdderRule], self.rules) + + async def setup(self): + await super().setup() + for rule in self._rules: + rule.init_generic_adder(self._job_tag_for_cleanup) + def _apply_rules(self, event: dict, rule: Rule) -> None: rule = typing.cast(GenericAdderRule, rule) - items_to_add = rule.add + + try: + items_to_add = rule.add(event) + except Exception as error: + raise ProcessingWarning(str(error), rule, event) from error if items_to_add: add_fields_to(event, items_to_add, rule, rule.merge_with_target, rule.overwrite_target) + + def _shut_down(self) -> None: + RefreshableGetter.remove_callbacks_for_tag(self._job_tag_for_cleanup) + return super()._shut_down() diff --git a/logprep/processor/generic_adder/processor.py b/logprep/processor/generic_adder/processor.py index ae84b7ee1..0d214502d 100644 --- a/logprep/processor/generic_adder/processor.py +++ b/logprep/processor/generic_adder/processor.py @@ -25,10 +25,12 @@ """ import typing +from typing import Sequence from logprep.abc.processor import Processor -from logprep.processor.base.rule import Rule +from logprep.processor.base.exceptions import ProcessingWarning from logprep.processor.generic_adder.rule import GenericAdderRule +from logprep.util.getter import RefreshableGetter from logprep.util.helper import add_fields_to @@ -37,8 +39,24 @@ class GenericAdder(Processor): rule_class = GenericAdderRule - def _apply_rules(self, event: dict, rule: Rule): - rule = typing.cast(GenericAdderRule, rule) - items_to_add = rule.add + @property + def _rules(self) -> Sequence[GenericAdderRule]: + """Returns all rules""" + return typing.cast(Sequence[GenericAdderRule], self.rules) + + def setup(self): + super().setup() + for rule in self._rules: + rule.init_generic_adder(self._job_tag_for_cleanup) + + def _apply_rules(self, event: dict, rule: GenericAdderRule): + try: + items_to_add = rule.add(event) + except Exception as error: + raise ProcessingWarning(str(error), rule, event) from error if items_to_add: add_fields_to(event, items_to_add, rule, rule.merge_with_target, rule.overwrite_target) + + def _shut_down(self) -> None: + RefreshableGetter.remove_callbacks_for_tag(self._job_tag_for_cleanup) + return super()._shut_down() diff --git a/logprep/processor/generic_adder/rule.py b/logprep/processor/generic_adder/rule.py index e82e209b8..ee1292775 100644 --- a/logprep/processor/generic_adder/rule.py +++ b/logprep/processor/generic_adder/rule.py @@ -83,13 +83,59 @@ # pylint: enable=anomalous-backslash-in-string import copy +import logging +import os import typing from attrs import define, field, validators +from logprep.filter.expression.filter_expression import FilterExpression from logprep.processor.base.rule import InvalidRuleDefinitionError from logprep.processor.field_manager.rule import FieldManagerRule -from logprep.util.getter import GetterFactory, RefreshableGetter +from logprep.util.converters import convert_from_dict +from logprep.util.environ import ENV_VARS +from logprep.util.getter import GetterFactory, HttpGetter, RefreshableGetter +from logprep.util.helper import ( + MISSING, + DottedTemplate, + FieldValue, + get_dotted_field_value, + get_dotted_field_value_with_missing, +) + +logger = logging.getLogger("GenericAdder") + + +@define(kw_only=True, frozen=True) +class AddFromUrlConfig: + url: str = field( + validator=[validators.instance_of(str), validators.matches_re(r"^https?://.+")] + ) + + target_field: str | None = field( + default=None, validator=validators.optional(validators.instance_of(str)) + ) + + target_field_mapping: dict[str, str] = field( + validator=validators.deep_mapping( + key_validator=validators.instance_of(str), + value_validator=validators.instance_of(str), + ), + factory=dict, + ) + + def __attrs_post_init__(self) -> None: + if not self.target_field and not self.target_field_mapping: + raise ValueError("adding values from url requires target_field or target_field_mapping") + + +def _convert_add_from_url( + value: AddFromUrlConfig | dict | None, +) -> AddFromUrlConfig | None: + if value is None: + return None + + return convert_from_dict(AddFromUrlConfig, value) class GenericAdderRule(FieldManagerRule): @@ -113,11 +159,11 @@ class Config(FieldManagerRule.Config): """Contains a dictionary of field names and values that should be added. If dot notation is being used, then all fields on the path are being automatically created.""" - add_from_file: list = field( - validator=[ - validators.instance_of(list), - validators.deep_iterable(member_validator=validators.instance_of(str)), - ], + add_from_file: list[str] = field( + validator=validators.deep_iterable( + iterable_validator=validators.instance_of(list), + member_validator=validators.instance_of(str), + ), converter=lambda x: x if isinstance(x, list) else [x], factory=list, eq=False, @@ -141,6 +187,13 @@ class Config(FieldManagerRule.Config): authenticity and integrity of the loaded values. """ + + add_from_url: AddFromUrlConfig | None = field( + default=None, + validator=validators.optional(validators.instance_of(AddFromUrlConfig)), + converter=_convert_add_from_url, + ) + only_first_existing_file: bool = field( validator=validators.instance_of(bool), default=False, eq=False ) @@ -160,20 +213,47 @@ def _refresh_add(self): def __attrs_post_init__(self): self._base_add = copy.deepcopy(self.add) - if self.add_from_file: - for add_file in self.add_from_file: # pylint: disable=not-an-iterable - getter = GetterFactory.from_string(add_file) - if isinstance(getter, RefreshableGetter): - # TODO: This never gets cleaned up, Memory leak on a lot of new generic adders / generic resolvers - getter.add_callback( - f"generic_adder:{self.id}:{add_file}", self._refresh_add - ) - self._add_from_path() + if (self.add_from_file or self.add) and self.add_from_url is not None: + raise ValueError( + "only one of add_from_file + add or add_from_url is allowed per GenericAdder rule" + ) + + if not self.add and not self.add_from_file and self.add_from_url is None: + raise ValueError( + "one of add, add_from_file or add_from_url is neccessary per GenericAdder rule" + ) + + if ( + self.add_from_url is not None + and self.add_from_url.target_field + and self.add_from_url.target_field_mapping + ): + raise ValueError( + "only one of target_field or target_field_mapping is allowed per GenericAdder rule" + ) + + if ( + self.add_from_url is not None + and not self.add_from_url.target_field + and not self.add_from_url.target_field_mapping + ): + raise ValueError( + "one of target_field or target_field_mapping is neccessary per GenericAdder rule" + ) + + # Eagerly loaded from file + for add_file in self.add_from_file: # pylint: disable=not-an-iterable + getter = GetterFactory.from_string(add_file) + if isinstance(getter, RefreshableGetter): + # TODO: This never gets cleaned up, Memory leak on a lot of new generic adders / generic resolvers + getter.add_callback(f"generic_adder:{self.id}:{add_file}", self._refresh_add) + self._add_from_path() def _add_from_path(self): """Reads add fields from file""" missing_files = [] - for add_file in self.add_from_file: # pylint: disable=not-an-iterable + + for add_file in self.add_from_file: try: add_dict = GetterFactory.from_string(add_file).get_yaml() except FileNotFoundError: @@ -195,8 +275,157 @@ def _add_from_path(self): f"The following required files do not exist: '{missing_files}'" ) - @property - def add(self) -> dict: + def __init__(self, filter_rule: FilterExpression, config: Config, processor_name: str): + super().__init__(filter_rule, config, processor_name) + self._dynamic_content: dict[str, FieldValue] = {} + self._callback_tag = "" + self._is_dynamic: bool = False + self._dynamic_template: DottedTemplate + self._dynamic_identifiers: tuple[str, ...] = () + self._static_uri: str | None = None + + def init_generic_adder(self, job_tag: str) -> None: + self._callback_tag = job_tag + + config = typing.cast(GenericAdderRule.Config, self._config) + if config.add_from_file or config.add: + return + + assert config.add_from_url is not None + + base_template = DottedTemplate(config.add_from_url.url) + resolved_template = DottedTemplate(base_template.safe_substitute({**ENV_VARS})) + self._dynamic_template = resolved_template + self._dynamic_identifiers = tuple(resolved_template.get_identifiers()) + + if len(self._dynamic_identifiers) > 0: + self._is_dynamic = True + + if not self._is_dynamic: + static_uri = resolved_template.substitute() + http_getter = GetterFactory.from_string(static_uri) + + assert isinstance(http_getter, HttpGetter) + + self._update_static_content(http_getter, static_uri) + + http_getter.add_callback( + self._callback_tag, + self._update_static_content, + deduplication_key=(self._callback_tag, static_uri, id(self)), + fnc_args=[http_getter, static_uri], + ) + + self._static_uri = static_uri + + def _dynamic_add_from_url(self, event: dict) -> dict[str, FieldValue]: + config = typing.cast(GenericAdderRule.Config, self._config) + + assert config.add_from_url + + key_val = { + identifier: get_dotted_field_value(event, identifier) + for identifier in self._dynamic_identifiers + } + for identifier, val in key_val.items(): + if val is None: + raise ValueError( + f"missing event field {identifier!r} for dynamic generic adder path" + ) + if not isinstance(val, (str, int)): + raise ValueError( + f"value for generic adder field {identifier!r} is not a scalar value" + ) + pass + + dynamic_resolved = self._dynamic_template.substitute(key_val) + content: FieldValue = None + if dynamic_resolved not in self._dynamic_content: + http_getter = GetterFactory.from_string(dynamic_resolved) + assert isinstance(http_getter, HttpGetter) + + http_getter.keep_alive() + + content = http_getter.get_collection() + self._dynamic_content[dynamic_resolved] = content + + tag = self._callback_tag + + http_getter.add_callback( + tag, + self._update_dynamic_content, + deduplication_key=(tag, dynamic_resolved, id(self)), + fnc_args=[http_getter, dynamic_resolved], + ) + + http_getter.add_cleanup_callback( + tag, + self._cleanup, + deduplication_key=(tag, dynamic_resolved, id(self)), + fnc_args=[dynamic_resolved], + ) + else: + RefreshableGetter.keep_alive_for_target(dynamic_resolved) + content = self._dynamic_content[dynamic_resolved] + + return self._content_to_items_to_add(content) + + def _content_to_items_to_add(self, content: FieldValue): + items_to_add: dict[str, FieldValue] = {} + + config = typing.cast(GenericAdderRule.Config, self._config) + assert config.add_from_url + + if config.add_from_url.target_field: + items_to_add[config.add_from_url.target_field] = content + else: + assert config.add_from_url.target_field_mapping is not None + + if not isinstance(content, dict): + raise ValueError("add_from_url.target_field_mapping requires a mapping response") + + for ( + mapping_source_field, + mapping_target_field, + ) in config.add_from_url.target_field_mapping.items(): + item = get_dotted_field_value_with_missing(content, mapping_source_field) + if item is MISSING: + logger.warning( + "could not add source_field: %s for target_field: %s because was missing", + mapping_source_field, + mapping_target_field, + ) + continue + + items_to_add[mapping_target_field] = item + + return items_to_add + + def _update_dynamic_content(self, http_getter: HttpGetter, resolved_uri: str): + content = http_getter.get_collection() + self._dynamic_content[resolved_uri] = content + + def _update_static_content(self, getter: HttpGetter, uri: str) -> None: + try: + self._update_dynamic_content(getter, uri) + except Exception as error: + self.mark_failed(error) + else: + self.clear_failed() + + def _cleanup(self, resolved_uri: str): + self._dynamic_content.pop(resolved_uri, None) + + def add(self, event: dict) -> dict: """Returns the fields to add""" config = typing.cast(GenericAdderRule.Config, self._config) - return config.add + + if config.add_from_file or config.add: + return config.add + + if not self._is_dynamic: + assert self._static_uri + return self._content_to_items_to_add(self._dynamic_content[self._static_uri]) + + assert config.add_from_url is not None + return self._dynamic_add_from_url(event) diff --git a/tests/unit/ng/processor/generic_adder/test_generic_adder.py b/tests/unit/ng/processor/generic_adder/test_generic_adder.py index 8b1738f1b..803acad83 100644 --- a/tests/unit/ng/processor/generic_adder/test_generic_adder.py +++ b/tests/unit/ng/processor/generic_adder/test_generic_adder.py @@ -9,11 +9,13 @@ from copy import deepcopy import pytest +import responses from logprep.factory import Factory from logprep.ng.abc.event import InputMeta, LogEvent from logprep.ng.processor.generic_adder.processor import GenericAdder -from logprep.processor.base.exceptions import InvalidRuleDefinitionError +from logprep.processor.base.exceptions import InvalidRuleDefinitionError, ProcessingWarning +from logprep.util.getter import HttpGetter, RefreshableGetter from tests.unit.ng.processor.base import BaseProcessorTestCase from tests.unit.processor.generic_adder.test_generic_adder import ( failure_test_cases as non_ng_failure_test_cases, @@ -95,10 +97,158 @@ async def test_add_only_copies(self): log_event = LogEvent(event, original=b"", input_meta=InputMeta()) await instance.process(log_event) - rule_add = instance.rules[0].add + rule_add = instance.rules[0].add({}) assert event["some_list_field"] == ["some_value"] assert event["some_list_field"] is not rule_add["some_list_field"], "only copies in events" assert event["some_dict_field"] == {"some_key": "some_value"} assert event["some_dict_field"] is not rule_add["some_dict_field"], "only copies in events" + + @responses.activate + async def test_adds_response_from_event_templated_url(self): + resolved_url = "https://values.example/acme" + response_content = {"user": {"name": "Alice"}, "risk": {"score": 7}} + responses.add(responses.GET, resolved_url, json=response_content) + processor = self._create_test_instance( + { + "rules": [ + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://values.example/${tenant.id}", + "target_field": "enrichment", + } + }, + } + ] + } + ) + await processor.setup() + event = {"tenant": {"id": "acme"}} + + result = await processor.process(LogEvent(event, original=b"", input_meta=InputMeta())) + + assert result.errors == [] + assert event == { + "tenant": {"id": "acme"}, + "enrichment": response_content, + } + assert responses.calls[0].request.url == resolved_url + + @responses.activate + async def test_dynamic_url_failure_is_event_scoped(self): + failed_url = "https://values.example/acme" + successful_url = "https://values.example/beta" + responses.add(responses.GET, failed_url, status=500) + responses.add(responses.GET, successful_url, json={"risk": {"score": 7}}) + RefreshableGetter.reset() + processor = self._create_test_instance( + { + "rules": [ + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://values.example/${tenant}", + "target_field": "enrichment", + } + }, + } + ] + } + ) + await processor.setup() + rule = processor.rules[0] + failed_event = {"tenant": "acme"} + successful_event = {"tenant": "beta"} + + failed_result = await processor.process( + LogEvent(failed_event, original=b"", input_meta=InputMeta()) + ) + successful_result = await processor.process( + LogEvent(successful_event, original=b"", input_meta=InputMeta()) + ) + + assert failed_result.errors == [] + assert len(failed_result.warnings) == 1 + assert isinstance(failed_result.warnings[0], ProcessingWarning) + assert failed_event == { + "tenant": "acme", + "tags": ["_generic_adder_failure"], + } + assert rule.data_error is None + assert len(HttpGetter._target_to_data_caches[failed_url].callbacks) == 0 + assert len(HttpGetter._target_to_data_caches[failed_url].cleanup_callbacks) == 0 + + assert successful_result.errors == [] + assert successful_result.warnings == [] + assert successful_event == { + "tenant": "beta", + "enrichment": {"risk": {"score": 7}}, + } + + async def test_missing_dynamic_url_field_adds_warning_without_clearing_event(self): + processor = self._create_test_instance( + { + "rules": [ + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://values.example/${tenant.id}", + "target_field": "enrichment", + } + }, + } + ] + } + ) + await processor.setup() + event = {"message": "preserved"} + + result = await processor.process(LogEvent(event, original=b"", input_meta=InputMeta())) + + assert result.errors == [] + assert len(result.warnings) == 1 + assert "missing event field 'tenant.id'" in str(result.warnings[0]) + assert event == { + "message": "preserved", + "tags": ["_generic_adder_failure"], + } + + @responses.activate + async def test_mapping_response_type_error_adds_warning_without_clearing_event(self): + url = "https://values.example/acme" + responses.add(responses.GET, url, json=["not", "a", "mapping"]) + RefreshableGetter.reset() + processor = self._create_test_instance( + { + "rules": [ + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://values.example/${tenant}", + "target_field_mapping": { + "risk.score": "enrichment.score", + }, + } + }, + } + ] + } + ) + await processor.setup() + event = {"tenant": "acme"} + + result = await processor.process(LogEvent(event, original=b"", input_meta=InputMeta())) + + assert result.errors == [] + assert len(result.warnings) == 1 + assert "target_field_mapping requires a mapping response" in str(result.warnings[0]) + assert event == { + "tenant": "acme", + "tags": ["_generic_adder_failure"], + } diff --git a/tests/unit/processor/generic_adder/test_generic_adder.py b/tests/unit/processor/generic_adder/test_generic_adder.py index 6ef8fcd99..3fff06c8d 100644 --- a/tests/unit/processor/generic_adder/test_generic_adder.py +++ b/tests/unit/processor/generic_adder/test_generic_adder.py @@ -7,11 +7,14 @@ from copy import deepcopy import pytest +import responses from logprep.factory import Factory -from logprep.processor.base.exceptions import InvalidRuleDefinitionError +from logprep.processor.base.exceptions import InvalidRuleDefinitionError, ProcessingWarning from logprep.processor.generic_adder.processor import GenericAdder +from logprep.util.getter import HttpGetter, RefreshableGetter from tests.unit.processor.base import BaseProcessorTestCase +from tests.conftest import mock_env RULES_DIR_MISSING = "tests/testdata/unit/generic_adder/rules_missing" RULES_DIR_INVALID = "tests/testdata/unit/generic_adder/rules_invalid" @@ -295,7 +298,7 @@ { "add_generic_test": "Test", "event_id": 123, - "\\u\\0\\1\\x\\z": "whatever", # pylint: disable=anomalous-backslash-in-string + "\\u\\0\\1\\x\\z": "whatever", }, { "add_generic_test": "Test", @@ -303,8 +306,8 @@ "comp\\lex.field": "value", "comp\\lex.nested": {"field": 42}, "nested": {"comp\\lex.field": 1337}, - "\\u\\0\\1\\x\y": 1338, # pylint: disable=anomalous-backslash-in-string - "\\u\\0\\1\\x\z": "whatever", # pylint: disable=anomalous-backslash-in-string + "\\u\\0\\1\\x\\y": 1338, + "\\u\\0\\1\\x\\z": "whatever", }, id="Add from rule definition with escaping", ), @@ -454,10 +457,197 @@ def test_add_only_copies(self): event = {} instance.process(event) - rule_add = instance.rules[0].add + rule_add = instance.rules[0].add({}) assert event["some_list_field"] == ["some_value"] assert event["some_list_field"] is not rule_add["some_list_field"], "only copies in events" assert event["some_dict_field"] == {"some_key": "some_value"} assert event["some_dict_field"] is not rule_add["some_dict_field"], "only copies in events" + + @responses.activate + def test_adds_mapped_response_fields_from_event_templated_url(self): + resolved_url = "https://values.example/acme" + responses.add( + responses.GET, + resolved_url, + json={ + "user": {"name": "Alice"}, + "risk": {"score": 7}, + }, + ) + configuration = { + "dynamic_generic_adder": { + "type": "generic_adder", + "rules": [ + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://${GENERIC_ADDER_HOST}/${tenant.id}", + "target_field_mapping": { + "user.name": "enrichment.user", + "risk.score": "enrichment.score", + }, + } + }, + } + ], + } + } + + RefreshableGetter.reset() + with mock_env({"GENERIC_ADDER_HOST": "values.example"}): + processor = typing.cast(GenericAdder, self._create_test_instance(configuration)) + processor.setup() + + first_event = {"tenant": {"id": "acme"}} + second_event = {"tenant": {"id": "acme"}} + first_result = processor.process(first_event) + second_result = processor.process(second_event) + + processor.shut_down() + + assert first_result.errors == [] + assert second_result.errors == [] + assert first_event == { + "tenant": {"id": "acme"}, + "enrichment": {"user": "Alice", "score": 7}, + } + assert second_event == first_event + assert len(responses.calls) == 1 + assert responses.calls[0].request.url == resolved_url + + @responses.activate + def test_dynamic_url_failure_is_event_scoped(self): + failed_url = "https://values.example/acme" + successful_url = "https://values.example/beta" + responses.add(responses.GET, failed_url, status=500) + responses.add(responses.GET, successful_url, json={"risk": {"score": 7}}) + RefreshableGetter.reset() + processor = typing.cast( + GenericAdder, + self._create_test_instance( + { + "dynamic_generic_adder": { + "type": "generic_adder", + "rules": [ + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://values.example/${tenant}", + "target_field": "enrichment", + } + }, + } + ], + } + } + ), + ) + processor.setup() + rule = processor.rules[0] + failed_event = {"tenant": "acme"} + successful_event = {"tenant": "beta"} + + failed_result = processor.process(failed_event) + successful_result = processor.process(successful_event) + + assert failed_result.errors == [] + assert len(failed_result.warnings) == 1 + assert isinstance(failed_result.warnings[0], ProcessingWarning) + assert failed_event == { + "tenant": "acme", + "tags": ["_generic_adder_failure"], + } + assert rule.data_error is None + assert len(HttpGetter._target_to_data_caches[failed_url].callbacks) == 0 + assert len(HttpGetter._target_to_data_caches[failed_url].cleanup_callbacks) == 0 + + assert successful_result.errors == [] + assert successful_result.warnings == [] + assert successful_event == { + "tenant": "beta", + "enrichment": {"risk": {"score": 7}}, + } + + processor.shut_down() + + def test_missing_dynamic_url_field_adds_warning_without_clearing_event(self): + processor = typing.cast( + GenericAdder, + self._create_test_instance( + { + "dynamic_generic_adder": { + "type": "generic_adder", + "rules": [ + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://values.example/${tenant.id}", + "target_field": "enrichment", + } + }, + } + ], + } + } + ), + ) + processor.setup() + event = {"message": "preserved"} + + result = processor.process(event) + + assert result.errors == [] + assert len(result.warnings) == 1 + assert "missing event field 'tenant.id'" in str(result.warnings[0]) + assert event == { + "message": "preserved", + "tags": ["_generic_adder_failure"], + } + + @responses.activate + def test_mapping_response_type_error_adds_warning_without_clearing_event(self): + url = "https://values.example/acme" + responses.add(responses.GET, url, json=["not", "a", "mapping"]) + RefreshableGetter.reset() + processor = typing.cast( + GenericAdder, + self._create_test_instance( + { + "dynamic_generic_adder": { + "type": "generic_adder", + "rules": [ + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://values.example/${tenant}", + "target_field_mapping": { + "risk.score": "enrichment.score", + }, + } + }, + } + ], + } + } + ), + ) + processor.setup() + event = {"tenant": "acme"} + + result = processor.process(event) + + processor.shut_down() + + assert result.errors == [] + assert len(result.warnings) == 1 + assert "target_field_mapping requires a mapping response" in str(result.warnings[0]) + assert event == { + "tenant": "acme", + "tags": ["_generic_adder_failure"], + } diff --git a/tests/unit/processor/generic_adder/test_generic_adder_rule.py b/tests/unit/processor/generic_adder/test_generic_adder_rule.py index 4052f3196..62c8aae4c 100644 --- a/tests/unit/processor/generic_adder/test_generic_adder_rule.py +++ b/tests/unit/processor/generic_adder/test_generic_adder_rule.py @@ -1,14 +1,15 @@ # pylint: disable=missing-docstring # pylint: disable=protected-access import json +import typing from pathlib import Path import pytest import responses -from logprep.processor.generic_adder.rule import GenericAdderRule +from logprep.processor.generic_adder.rule import AddFromUrlConfig, GenericAdderRule from logprep.util.defaults import ENV_NAME_LOGPREP_GETTER_CONFIG -from logprep.util.getter import HttpGetter, RefreshableGetter +from logprep.util.getter import GetterFactory, HttpGetter, RefreshableGetter from tests.conftest import mock_env @@ -28,6 +29,200 @@ def fixture_rule_definition(): class TestGenericAdderRule: + @pytest.mark.parametrize( + ("url_config", "expected_target_field", "expected_target_field_mapping"), + [ + pytest.param( + { + "url": "https://values.example/${tenant.id}", + "target_field": "enrichment", + }, + "enrichment", + {}, + id="whole-response", + ), + pytest.param( + { + "url": "https://values.example/${tenant.id}", + "target_field_mapping": { + "user.name": "enrichment.user", + "risk.score": "enrichment.score", + }, + }, + None, + { + "user.name": "enrichment.user", + "risk.score": "enrichment.score", + }, + id="field-mapping", + ), + ], + ) + def test_converts_add_from_url_configuration( + self, url_config, expected_target_field, expected_target_field_mapping + ): + rule = GenericAdderRule.create_from_dict( + { + "filter": "*", + "generic_adder": {"add_from_url": url_config}, + } + ) + + config = typing.cast(GenericAdderRule.Config, rule._config) + + assert isinstance(config.add_from_url, AddFromUrlConfig) + assert config.add_from_url.target_field == expected_target_field + assert config.add_from_url.target_field_mapping == expected_target_field_mapping + + @pytest.mark.parametrize( + ("url_config", "error_message"), + [ + pytest.param( + {"url": "https://values.example/${tenant}"}, + "requires target_field or target_field_mapping", + id="missing-target", + ), + pytest.param( + { + "url": "https://values.example/${tenant}", + "target_field": "enrichment", + "target_field_mapping": {"risk": "enrichment.risk"}, + }, + "only one of target_field or target_field_mapping", + id="ambiguous-target", + ), + ], + ) + def test_rejects_invalid_add_from_url_target_configuration(self, url_config, error_message): + with pytest.raises(ValueError, match=error_message): + GenericAdderRule.create_from_dict( + { + "filter": "*", + "generic_adder": {"add_from_url": url_config}, + } + ) + + def test_rejects_rule_without_addition_source(self): + with pytest.raises( + ValueError, + match="one of add, add_from_file or add_from_url", + ): + GenericAdderRule.create_from_dict( + { + "filter": "*", + "generic_adder": {}, + } + ) + + @responses.activate + def test_resolves_dotted_event_field_and_adds_complete_response(self): + resolved_url = "https://values.example/acme" + response_content = { + "user": {"name": "Alice"}, + "risk": {"score": 7}, + } + responses.add(responses.GET, resolved_url, json=response_content) + rule = GenericAdderRule.create_from_dict( + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://values.example/${tenant.id}", + "target_field": "enrichment", + } + }, + } + ) + rule.init_generic_adder("generic-adder-test") + + additions = rule.add({"tenant": {"id": "acme"}}) + + assert additions == {"enrichment": response_content} + assert responses.calls[0].request.url == resolved_url + + @responses.activate + def test_static_url_loads_during_setup_and_registers_only_refresh_callback(self): + url = "https://values.example/static" + response_content = {"risk": {"score": 7}} + responses.add(responses.GET, url, json=response_content) + RefreshableGetter.reset() + rule = GenericAdderRule.create_from_dict( + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": url, + "target_field": "enrichment", + } + }, + } + ) + + rule.init_generic_adder("generic-adder-test") + + assert rule.add({}) == {"enrichment": response_content} + assert rule.add({}) == {"enrichment": response_content} + assert len(responses.calls) == 1 + assert len(HttpGetter._target_to_data_caches[url].callbacks) == 1 + assert len(HttpGetter._target_to_data_caches[url].cleanup_callbacks) == 0 + + @responses.activate + def test_static_url_recovers_after_failed_initial_load(self, tmp_path): + url = "https://values.example/static" + response_content = {"risk": {"score": 7}} + responses.add(responses.GET, url, status=500) + RefreshableGetter.reset() + getter_config = tmp_path / "http_getter.json" + getter_config.write_text(json.dumps({url: {"refresh_interval": 1}})) + rule = GenericAdderRule.create_from_dict( + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": url, + "target_field": "enrichment", + } + }, + } + ) + + with mock_env({ENV_NAME_LOGPREP_GETTER_CONFIG: str(getter_config)}): + rule.init_generic_adder("generic-adder-test") + getter = GetterFactory.from_string(url) + assert isinstance(getter, HttpGetter) + assert getter.scheduler is not None + + assert rule.data_error is not None + assert len(getter.shared.callbacks) == 1 + assert len(getter.shared.cleanup_callbacks) == 0 + + responses.replace(responses.GET, url, json=response_content) + getter.scheduler.run_all() + + assert rule.data_error is None + assert rule.add({}) == {"enrichment": response_content} + + def test_target_field_mapping_skips_missing_values_but_preserves_none(self, caplog): + rule = GenericAdderRule.create_from_dict( + { + "filter": "*", + "generic_adder": { + "add_from_url": { + "url": "https://values.example/${tenant}", + "target_field_mapping": { + "present": "enrichment.present", + "missing": "enrichment.missing", + }, + } + }, + } + ) + + additions = rule._content_to_items_to_add({"present": None}) + + assert additions == {"enrichment.present": None} + assert "source_field: missing" in caplog.text + @pytest.mark.parametrize( "testcase, other_rule_definition, is_equal", [ @@ -101,7 +296,7 @@ def test_rule_accepts_bool_type(self): "generic_adder": {"add": {"added_bool_field": True}}, } rule = GenericAdderRule.create_from_dict(rule_definition) - assert isinstance(rule.add.get("added_bool_field"), bool) + assert isinstance(rule.add({}).get("added_bool_field"), bool) @responses.activate def test_rule_callback_updates_additions_and_preserves_original_add(self, tmp_path): @@ -146,11 +341,11 @@ def test_rule_callback_updates_additions_and_preserves_original_add(self, tmp_pa with mock_env({ENV_NAME_LOGPREP_GETTER_CONFIG: str(http_getter_conf)}): scheduler = HttpGetter(protocol="http", target=url).scheduler rule = GenericAdderRule.create_from_dict(rule_definition) - assert rule.add == expected_1 + assert rule.add({}) == expected_1 HttpGetter.refresh() - assert rule.add == expected_1 + assert rule.add({}) == expected_1 scheduler.run_all() - assert rule.add == expected_2 - assert rule.add == expected_2 + assert rule.add({}) == expected_2 + assert rule.add({}) == expected_2 scheduler.run_all() - assert rule.add == expected_3 + assert rule.add({}) == expected_3 diff --git a/tests/unit/util/test_getter.py b/tests/unit/util/test_getter.py index 014524d1a..a274311ee 100644 --- a/tests/unit/util/test_getter.py +++ b/tests/unit/util/test_getter.py @@ -1472,6 +1472,31 @@ def test_get_collection_parses_json_if_yaml_fails(self, mock_parse_yaml): http_getter.get_collection() mock_parse_yaml.assert_called_once() + @responses.activate + def test_get_collection_extracts_content_field(self): + responses.add( + responses.GET, + "http://something", + json={"payload": {"answer": 42}}, + ) + + http_getter = GetterFactory.from_string("http://something") + + assert http_getter.get_collection("payload") == {"answer": 42} + + @responses.activate + def test_get_collection_rejects_content_field_for_non_mapping_content(self): + responses.add( + responses.GET, + "http://something", + json=["one", "two"], + ) + + http_getter = GetterFactory.from_string("http://something") + + with pytest.raises(ValueError, match="Expected mapping type when content_field is set"): + http_getter.get_collection("payload") + @mock.patch("logprep.abc.getter.Getter.get_collection", return_value="not a dict") def test_get_dict_raises_exception_if_result_not_dict(self, _): http_getter = GetterFactory.from_string("http://something")