70cb5ea702c69f2289ea7fbb9f26b14202c905b6-PR-1245

Build # 70cb5ea702c69f2289ea7fbb9f26b14202c905b6-PR-1245

Build Type

github-actions

Committed by GitHub

Commit Message

<a href="https://github.com/airbnb/streamalert/commit/<a class=hub.com/airbnb/streamalert/commit/<a class="double-link" href="https://git"><a class=hub.com/airbnb/streamalert/commit/<a class="double-link" href="https://git"><a class=hub.com/airbnb/streamalert/commit/<a class="double-link" href="https://git"><a class=hub.com/airbnb/streamalert/commit/70cb5ea702c69f2289ea7fbb9f26b14202c905b6">70cb5ea70">&lt;a href=&quot;https://github.com/airbnb/streamalert/commit/</a><a class="double-link" href="https://github.com/airbnb/streamalert/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/airbnb/streamalert/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/airbnb/streamalert/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/airbnb/streamalert/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/airbnb/streamalert/commit/70cb5ea702c69f2289ea7fbb9f26b14202c905b6">70cb5ea70</a><a href="https://github.com/airbnb/streamalert/commit/70cb5ea702c69f2289ea7fbb9f26b14202c905b6">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/airbnb/streamalert/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/airbnb/streamalert/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/airbnb/streamalert/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/airbnb/streamalert/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/airbnb/streamalert/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/airbnb/streamalert/commit/70cb5ea702c69f2289ea7fbb9f26b14202c905b6&quot;&gt;70cb5ea70&lt;/a&gt;&lt;a href=&quot;https://github.com/airbnb/streamalert/commit/70cb5ea702c69f2289ea7fbb9f26b14202c905b6&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/airbnb/streamalert/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/airbnb/streamalert/comm... (continued)

Pull Request Pull Request #1245: Normalization reboot - Add condition support to normalizer

Run Details

39 of 39 new or added lines in 2 files covered. (100.0%)

6508 of 6825 relevant lines covered (95.36%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.49

/streamalert/shared/normalize.py

"""
Copyright 2017-present Airbnb, Inc.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

   http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from collections import defaultdict
import logging
import itertools

from streamalert.shared.config import TopLevelConfigKeys
from streamalert.shared.exceptions import ConfigError
from streamalert.shared.logger import get_logger


LOGGER = get_logger(__name__)
LOGGER_DEBUG_ENABLED = LOGGER.isEnabledFor(logging.DEBUG)

CONST_FUNCTION = 'function'
CONST_PATH = 'path'
CONST_CONDITION = 'condition'
CONST_VALUES = 'values'

class NormalizedType:
    """The class encapsulates normalization information for each normalized type"""

    VALID_KEYS = {CONST_PATH, CONST_FUNCTION, CONST_CONDITION}
    CONST_STR = 'str'
    CONST_DICT = 'dict'

    def __init__(self, log_type, normalized_type, params):
        """Init NormalizatedType
        Args:
            log_type (str): log type name, e.g. osquery:differential
            normalized_type (str): Normalized type name defined in conf/, e.g. 'sourceAddress',
                'destination_ip' may be normalized to 'ip_address'.
            params (list): a list of str or dict contains normalization configuration read from
                conf/schemas/*.json. The params can be a list of str or a list of dict to specify
                the path to the keys which will be normalized.
                e.g.
                    ['path', 'to', 'the', 'key']
                or
                    [
                        {
                            'path': ['detail', 'sourceIPAddress'],
                            'function': 'source ip address'
                        },
                        {
                            'path': ['path', 'to', 'the', 'key'],
                            'function': 'destination ip address'
                        }
                    ]
        """
        self._log_type = log_type
        self._log_source = log_type.split(':')[0]
        self._normalized_type = normalized_type
        self._parsed_params = self._parse_params(params)

    def __eq__(self, other):
        """Compare two NormalizedType instances and it is very helpful in unit test when use
        assert_equal
        """
        if not (self._log_type == other.log_type
                and self._log_source == other.log_source
                and self._normalized_type == other.normalized_type):
            return False

        if len(self._parsed_params) != len(other.parsed_params):
            return False

        for idx in range(len(self._parsed_params)):
            if self._parsed_params[idx][CONST_PATH] == other.parsed_params[idx][CONST_PATH]:
                continue

            return False

        return True

    @property
    def log_type(self):
        """Return the log type name, e.g. 'osquery:differential'"""
        return self._log_type

    @property
    def log_source(self):
        """Return the log source name, e.g. 'osquery'"""
        return self._log_source

    @property
    def normalized_type(self):
        """Return the normalized type, e.g. 'ip_address'"""
        return self._normalized_type

    @property
    def parsed_params(self):
        """Return the normalization configuration which is a list of dict, e.g.
            [
                {
                    'path': ['path', 'to', 'the', 'key'],
                    'function': None
                }
            ]

            or
            [
                {
                    'path': ['detail', 'sourceIPAddress'],
                    'function': 'source ip address'
                },
                {
                    'path': ['path', 'to', 'the', 'destination', 'ip'],
                    'function': 'destination ip address'
                }
            ]
        """
        return self._parsed_params

    def _parse_params(self, params):
        """Extract path and function information from params argument

        Args:
            params (list): a list of str or dict contains normalization configuration.
        """
        param_type = self._parse_param_type(params)

        if param_type == self.CONST_STR:
            # Format params to include 'function' field which is set to None.
            return [
                {
                    CONST_PATH: params,
                    CONST_FUNCTION: None
                }
            ]

        return params

    def _parse_param_type(self, params):
        """Parse all param type in params

        Args:
            params (list): a list of str or dict contains normalization configuration.
        """
        if not isinstance(params, list):
            raise ConfigError(
                'Unsupported params {} for normalization. Convert params to a list'.format(params)
            )

        if all(isinstance(param, str) for param in params):
            return self.CONST_STR

        if all(isinstance(param, dict) and set(param.keys()).issubset(self.VALID_KEYS)
               for param in params
              ):
            return self.CONST_DICT

        # FIXME: should we raise exception here? Or may just return False and log a warming message
        raise ConfigError(
            ('Unsupported type(s) used in {} or missing keys. Valid types are str or dict and '
             'valid keys are {}').format(params, self.VALID_KEYS)
        )


class Normalizer:
    """Normalizer class to handle log key normalization in payloads"""

    NORMALIZATION_KEY = 'streamalert_normalization'

    # Store the normalized types mapping to original keys from the records
    _types_config = dict()

    @classmethod
    def match_types(cls, record, normalized_types):
        """Check for normalized types within record

        Args:
            record (dict): Parsed payload of log
            normalized_types (dict): Normalized types mapping

        Returns:
            dict: A dict of normalized keys with a list of values

        Example:
            return
            {
                'region': [
                    {
                        'values': ['us-east-1']
                        'function': 'AWS region'
                    },
                    {
                        'values': ['us-west-2']
                        'function': 'AWS region'
                    }
                ]
            }
        """
        results = {}
        for type_name, type_info in normalized_types.items():
            result = list(cls._extract_values(record, type_info))

            if result:
                results[type_name] = result

        return results

    @classmethod
    def _find_value(cls, record, path):
        """Retrieve value from a record based on a json path"""
        found_value = False
        value = record
        for key in path:
            value = value.get(key)
            if not value:
                found_value = False
                break
            found_value = True

        if not found_value:
            return False, None

        return True, value

    @classmethod
    def _extract_values(cls, record, paths_to_normalize):
        """Recursively extract lists of path parts from a dictionary

        Args:
            record (dict): Parsed payload of log
            paths_to_normalize (set): Normalized keys for which to extract paths
            path (list=None): Parts of current path for which keys are being extracted

        Yields:
            dict: A dict contians the values of normalized types. For example,
                {
                    'values': ['1.1.1.2']
                    'function': 'Source ip address'
                }
        """
        for param in paths_to_normalize.parsed_params:
            if param.get(CONST_CONDITION) and not cls._match_condition(record, param['condition']):
                # If optional 'condition' block is configured, it will only extract values if
                # condition is matched.
                continue

            found_value, value = cls._find_value(record, param.get(CONST_PATH))

            if found_value:
                yield {
                    CONST_FUNCTION: param.get(CONST_FUNCTION) or None,
                    # if value not a list, it will be cast to a str even it is a dict or other
                    # types
                    CONST_VALUES: value if isinstance(value, list) else [str(value)]
                }

    @classmethod
    def _match_condition(cls, record, condition):
        """Apply condition to a record before normalization kicked in.

        Returns:
            bool: Return True if the value of the condition path matches to the condition, otherwise
                return False. It is False if the path doesn't exist.
        """
        if not condition.get('path'):
            return False

        found_value, value = cls._find_value(record, condition['path'])
        if not found_value:
            return False

        # cast value to a str in all lowercases
        value = str(value).lower()

        # Only support extract one condition. The result is not quaranteed if multiple conditions
        # configured.
        # FIXME: log a warning if more than one condition configured.
        if condition.get('is'):
            return value == condition['is']

        if condition.get('is_not'):
            return value != condition['is_not']

        if condition.get('in'):
            return value in condition['in']

        if condition.get('not_in'):
            return value not in condition['not_in']

        if condition.get('contains'):
            return condition['contains'] in value

        if condition.get('not_contains'):
            return condition['not_contains'] not in value

        return False

    @classmethod
    def normalize(cls, record, log_type):
        """Apply data normalization to a record

        Args:
            record (dict): The parsed log without data normalization
            log_type (str): Type of log for which to apply normalizaiton
        """
        log_normalized_types = cls._types_config.get(log_type)
        if not log_normalized_types:
            LOGGER.debug('No normalized types defined for log type: %s', log_type)
            return

        # Add normalized keys to the record
        record.update({cls.NORMALIZATION_KEY: cls.match_types(record, log_normalized_types)})

    @classmethod
    def get_values_for_normalized_type(cls, record, datatype):
        """Fetch values by normalized_type.

        Args:
            record (dict): parsed payload of any log
            datatype (str): normalized type being found

        Returns:
            set: The values for the normalized type specified
        """
        normalization_results = record.get(cls.NORMALIZATION_KEY, {}).get(datatype)
        if not normalization_results:
            return
        return set(itertools.chain(*[result.get(CONST_VALUES) for result in normalization_results]))

    @classmethod
    def load_from_config(cls, config):
        """Extract and store the data types from the config in the proper format

        Args:
            config (dict): Config read from 'conf/' directory

        Returns:
            Normalizer: Class to be used for normalization logic
        """
        if cls._types_config:
            return cls  # config is already populated

        cls._types_config = cls._parse_normalization(config)

        return cls  # there are no instance methods, so just return the class

    @classmethod
    def _parse_normalization(cls, config):
        """Load and parse normalization config from conf/schemas/*.json. Normalization will be
        configured along with log schema and a path will be provided to find the original key.

        For example: conf/schemas/cloudwatch.json looks like
            'cloudwatch:events': {
                'schema': {
                    'account': 'string',
                    'source': 'string',
                    'other_key': 'string'
                },
                'configuration': {
                    'normalization': {
                        'region': ['path', 'to', 'original', 'key'],
                        'ip_address': [
                            {
                                'path': ['detail', 'sourceIPAddress'],
                                'function': 'source ip address'
                            },
                            {
                                'path': ['path', 'to', 'original', 'key'],
                                'function': 'destination ip address'
                            }
                        ]
                    }
                }
            }

        Args:
            config (dict): Config read from 'conf/' directory

        Returns:
            dict: return a dict contains normalization information per log type basis.
                {
                    'cloudwatch:events': {
                        'region': NormalizedType(),
                        'ip_address': NormalizedType()
                    }
                }
        """
        normalized_config = defaultdict(dict)
        for log_type, val in config.get(TopLevelConfigKeys.LOGS, {}).items():
            result = defaultdict(dict)

            log_type_normalization = val.get('configuration', {}).get('normalization', {})

            for normalized_type, params in log_type_normalization.items():
                # add normalization info if it is defined in log type configuration field
                result[normalized_type] = NormalizedType(log_type, normalized_type, params)

            if result:
                normalized_config[log_type] = result

        # return None is normalized_config is an empty defaultdict.
        return normalized_config or None

1	"""
2	Copyright 2017-present Airbnb, Inc.
3
4	Licensed under the Apache License, Version 2.0 (the "License");
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	http://www.apache.org/licenses/LICENSE-2.0
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	"""
16	from collections import defaultdict	1✔
17	import logging	1✔
18	import itertools	1✔
19
20	from streamalert.shared.config import TopLevelConfigKeys	1✔
21	from streamalert.shared.exceptions import ConfigError	1✔
22	from streamalert.shared.logger import get_logger	1✔
23
24
25	LOGGER = get_logger(__name__)	1✔
26	LOGGER_DEBUG_ENABLED = LOGGER.isEnabledFor(logging.DEBUG)	1✔
27
28	CONST_FUNCTION = 'function'	1✔
29	CONST_PATH = 'path'	1✔
30	CONST_CONDITION = 'condition'	1✔
31	CONST_VALUES = 'values'	1✔
32
33	class NormalizedType:	1✔
34	"""The class encapsulates normalization information for each normalized type"""
35
36	VALID_KEYS = {CONST_PATH, CONST_FUNCTION, CONST_CONDITION}	1✔
37	CONST_STR = 'str'	1✔
38	CONST_DICT = 'dict'	1✔
39
40	def __init__(self, log_type, normalized_type, params):	1✔
41	"""Init NormalizatedType
42	Args:
43	log_type (str): log type name, e.g. osquery:differential
44	normalized_type (str): Normalized type name defined in conf/, e.g. 'sourceAddress',
45	'destination_ip' may be normalized to 'ip_address'.
46	params (list): a list of str or dict contains normalization configuration read from
47	conf/schemas/*.json. The params can be a list of str or a list of dict to specify
48	the path to the keys which will be normalized.
49	e.g.
50	['path', 'to', 'the', 'key']
51	or
52	[
53	{
54	'path': ['detail', 'sourceIPAddress'],
55	'function': 'source ip address'
56	},
57	{
58	'path': ['path', 'to', 'the', 'key'],
59	'function': 'destination ip address'
60	}
61	]
62	"""
63	self._log_type = log_type	1✔
64	self._log_source = log_type.split(':')[0]	1✔
65	self._normalized_type = normalized_type	1✔
66	self._parsed_params = self._parse_params(params)	1✔
67
68	def __eq__(self, other):	1✔
69	"""Compare two NormalizedType instances and it is very helpful in unit test when use
70	assert_equal
71	"""
72	if not (self._log_type == other.log_type	1✔
73	and self._log_source == other.log_source
74	and self._normalized_type == other.normalized_type):
75	return False	×
76
77	if len(self._parsed_params) != len(other.parsed_params):	1✔
78	return False	×
79
80	for idx in range(len(self._parsed_params)):	1✔
81	if self._parsed_params[idx][CONST_PATH] == other.parsed_params[idx][CONST_PATH]:	1✔
82	continue	1✔
83
84	return False	×
85
86	return True	1✔
87
88	@property	1✔
89	def log_type(self):
90	"""Return the log type name, e.g. 'osquery:differential'"""
91	return self._log_type	1✔
92
93	@property	1✔
94	def log_source(self):
95	"""Return the log source name, e.g. 'osquery'"""
96	return self._log_source	1✔
97
98	@property	1✔
99	def normalized_type(self):
100	"""Return the normalized type, e.g. 'ip_address'"""
101	return self._normalized_type	1✔
102
103	@property	1✔
104	def parsed_params(self):
105	"""Return the normalization configuration which is a list of dict, e.g.
106	[
107	{
108	'path': ['path', 'to', 'the', 'key'],
109	'function': None
110	}
111	]
112
113	or
114	[
115	{
116	'path': ['detail', 'sourceIPAddress'],
117	'function': 'source ip address'
118	},
119	{
120	'path': ['path', 'to', 'the', 'destination', 'ip'],
121	'function': 'destination ip address'
122	}
123	]
124	"""
125	return self._parsed_params	1✔
126
127	def _parse_params(self, params):	1✔
128	"""Extract path and function information from params argument
129
130	Args:
131	params (list): a list of str or dict contains normalization configuration.
132	"""
133	param_type = self._parse_param_type(params)	1✔
134
135	if param_type == self.CONST_STR:	1✔
136	# Format params to include 'function' field which is set to None.
137	return [	1✔
138	{
139	CONST_PATH: params,
140	CONST_FUNCTION: None
141	}
142	]
143
144	return params	1✔
145
146	def _parse_param_type(self, params):	1✔
147	"""Parse all param type in params
148
149	Args:
150	params (list): a list of str or dict contains normalization configuration.
151	"""
152	if not isinstance(params, list):	1✔
153	raise ConfigError(	1✔
154	'Unsupported params {} for normalization. Convert params to a list'.format(params)
155	)
156
157	if all(isinstance(param, str) for param in params):	1✔
158	return self.CONST_STR	1✔
159
160	if all(isinstance(param, dict) and set(param.keys()).issubset(self.VALID_KEYS)	1✔
161	for param in params
162	):
163	return self.CONST_DICT	1✔
164
165	# FIXME: should we raise exception here? Or may just return False and log a warming message
166	raise ConfigError(	×
167	('Unsupported type(s) used in {} or missing keys. Valid types are str or dict and '
168	'valid keys are {}').format(params, self.VALID_KEYS)
169	)
170
171
172	class Normalizer:	1✔
173	"""Normalizer class to handle log key normalization in payloads"""
174
175	NORMALIZATION_KEY = 'streamalert_normalization'	1✔
176
177	# Store the normalized types mapping to original keys from the records
178	_types_config = dict()	1✔
179
180	@classmethod	1✔
181	def match_types(cls, record, normalized_types):
182	"""Check for normalized types within record
183
184	Args:
185	record (dict): Parsed payload of log
186	normalized_types (dict): Normalized types mapping
187
188	Returns:
189	dict: A dict of normalized keys with a list of values
190
191	Example:
192	return
193	{
194	'region': [
195	{
196	'values': ['us-east-1']
197	'function': 'AWS region'
198	},
199	{
200	'values': ['us-west-2']
201	'function': 'AWS region'
202	}
203	]
204	}
205	"""
206	results = {}	1✔
207	for type_name, type_info in normalized_types.items():	1✔
208	result = list(cls._extract_values(record, type_info))	1✔
209
210	if result:	1✔
211	results[type_name] = result	1✔
212
213	return results	1✔
214
215	@classmethod	1✔
216	def _find_value(cls, record, path):
217	"""Retrieve value from a record based on a json path"""
218	found_value = False	1✔
219	value = record	1✔
220	for key in path:	1✔
221	value = value.get(key)	1✔
222	if not value:	1✔
223	found_value = False	1✔
224	break	1✔
225	found_value = True	1✔
226
227	if not found_value:	1✔
228	return False, None	1✔
229
230	return True, value	1✔
231
232	@classmethod	1✔
233	def _extract_values(cls, record, paths_to_normalize):
234	"""Recursively extract lists of path parts from a dictionary
235
236	Args:
237	record (dict): Parsed payload of log
238	paths_to_normalize (set): Normalized keys for which to extract paths
239	path (list=None): Parts of current path for which keys are being extracted
240
241	Yields:
242	dict: A dict contians the values of normalized types. For example,
243	{
244	'values': ['1.1.1.2']
245	'function': 'Source ip address'
246	}
247	"""
248	for param in paths_to_normalize.parsed_params:	1✔
249	if param.get(CONST_CONDITION) and not cls._match_condition(record, param['condition']):	1✔
250	# If optional 'condition' block is configured, it will only extract values if
251	# condition is matched.
252	continue	1✔
253
254	found_value, value = cls._find_value(record, param.get(CONST_PATH))	1✔
255
256	if found_value:	1✔
257	yield {	1✔
258	CONST_FUNCTION: param.get(CONST_FUNCTION) or None,
259	# if value not a list, it will be cast to a str even it is a dict or other
260	# types
261	CONST_VALUES: value if isinstance(value, list) else [str(value)]
262	}
263
264	@classmethod	1✔
265	def _match_condition(cls, record, condition):
266	"""Apply condition to a record before normalization kicked in.
267
268	Returns:
269	bool: Return True if the value of the condition path matches to the condition, otherwise
270	return False. It is False if the path doesn't exist.
271	"""
272	if not condition.get('path'):	1✔
273	return False	×
274
275	found_value, value = cls._find_value(record, condition['path'])	1✔
276	if not found_value:	1✔
277	return False	×
278
279	# cast value to a str in all lowercases
280	value = str(value).lower()	1✔
281
282	# Only support extract one condition. The result is not quaranteed if multiple conditions
283	# configured.
284	# FIXME: log a warning if more than one condition configured.
285	if condition.get('is'):	1✔
286	return value == condition['is']	1✔
287
288	if condition.get('is_not'):	1✔
289	return value != condition['is_not']	1✔
290
291	if condition.get('in'):	1✔
292	return value in condition['in']	1✔
293
294	if condition.get('not_in'):	1✔
295	return value not in condition['not_in']	1✔
296
297	if condition.get('contains'):	1✔
298	return condition['contains'] in value	1✔
299
300	if condition.get('not_contains'):	1✔
301	return condition['not_contains'] not in value	1✔
302
303	return False	×
304
305	@classmethod	1✔
306	def normalize(cls, record, log_type):
307	"""Apply data normalization to a record
308
309	Args:
310	record (dict): The parsed log without data normalization
311	log_type (str): Type of log for which to apply normalizaiton
312	"""
313	log_normalized_types = cls._types_config.get(log_type)	1✔
314	if not log_normalized_types:	1✔
315	LOGGER.debug('No normalized types defined for log type: %s', log_type)	1✔
316	return	1✔
317
318	# Add normalized keys to the record
319	record.update({cls.NORMALIZATION_KEY: cls.match_types(record, log_normalized_types)})	1✔
320
321	@classmethod	1✔
322	def get_values_for_normalized_type(cls, record, datatype):
323	"""Fetch values by normalized_type.
324
325	Args:
326	record (dict): parsed payload of any log
327	datatype (str): normalized type being found
328
329	Returns:
330	set: The values for the normalized type specified
331	"""
332	normalization_results = record.get(cls.NORMALIZATION_KEY, {}).get(datatype)	1✔
333	if not normalization_results:	1✔
334	return	1✔
335	return set(itertools.chain(*[result.get(CONST_VALUES) for result in normalization_results]))	1✔
336
337	@classmethod	1✔
338	def load_from_config(cls, config):
339	"""Extract and store the data types from the config in the proper format
340
341	Args:
342	config (dict): Config read from 'conf/' directory
343
344	Returns:
345	Normalizer: Class to be used for normalization logic
346	"""
347	if cls._types_config:	1✔
348	return cls # config is already populated	1✔
349
350	cls._types_config = cls._parse_normalization(config)	1✔
351
352	return cls # there are no instance methods, so just return the class	1✔
353
354	@classmethod	1✔
355	def _parse_normalization(cls, config):
356	"""Load and parse normalization config from conf/schemas/*.json. Normalization will be
357	configured along with log schema and a path will be provided to find the original key.
358
359	For example: conf/schemas/cloudwatch.json looks like
360	'cloudwatch:events': {
361	'schema': {
362	'account': 'string',
363	'source': 'string',
364	'other_key': 'string'
365	},
366	'configuration': {
367	'normalization': {
368	'region': ['path', 'to', 'original', 'key'],
369	'ip_address': [
370	{
371	'path': ['detail', 'sourceIPAddress'],
372	'function': 'source ip address'
373	},
374	{
375	'path': ['path', 'to', 'original', 'key'],
376	'function': 'destination ip address'
377	}
378	]
379	}
380	}
381	}
382
383	Args:
384	config (dict): Config read from 'conf/' directory
385
386	Returns:
387	dict: return a dict contains normalization information per log type basis.
388	{
389	'cloudwatch:events': {
390	'region': NormalizedType(),
391	'ip_address': NormalizedType()
392	}
393	}
394	"""
395	normalized_config = defaultdict(dict)	1✔
396	for log_type, val in config.get(TopLevelConfigKeys.LOGS, {}).items():	1✔
397	result = defaultdict(dict)	1✔
398
399	log_type_normalization = val.get('configuration', {}).get('normalization', {})	1✔
400
401	for normalized_type, params in log_type_normalization.items():	1✔
402	# add normalization info if it is defined in log type configuration field
403	result[normalized_type] = NormalizedType(log_type, normalized_type, params)	1✔
404
405	if result:	1✔
406	normalized_config[log_type] = result	1✔
407
408	# return None is normalized_config is an empty defaultdict.
409	return normalized_config or None	1✔

airbnb / streamalert / 70cb5ea702c69f2289ea7fbb9f26b14202c905b6-PR-1245

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous