33703505adb99480a1e32d5da20d782e28f51560-PR-1238

Build # 33703505adb99480a1e32d5da20d782e28f51560-PR-1238

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/33703505adb99480a1e32d5da20d782e28f51560">33703505a">&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/33703505adb99480a1e32d5da20d782e28f51560">33703505a</a><a href="https://github.com/airbnb/streamalert/commit/33703505adb99480a1e32d5da20d782e28f51560">&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/33703505adb99480a1e32d5da20d782e28f51560&quot;&gt;33703505a&lt;/a&gt;&lt;a href=&quot;https://github.com/airbnb/streamalert/commit/33703505adb99480a1e32d5da20d782e28f51560&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 #1238: Normalization reboot - Refactor normalization code

Run Details

85 of 85 new or added lines in 3 files covered. (100.0%)

6471 of 6785 relevant lines covered (95.37%)

0.95 hits per line

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

95.92

/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_VALUES = 'values'

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

    VALID_KEYS = {CONST_PATH, CONST_FUNCTION}
    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()) == 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 _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:
            found_value = False
            value = record
            for key in param.get(CONST_PATH):
                value = value.get(key)
                if not value:
                    break
                found_value = True

            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 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_VALUES = 'values'	1✔
31
32	class NormalizedType:	1✔
33	"""The class encapsulates normalization information for each normalized type"""
34
35	VALID_KEYS = {CONST_PATH, CONST_FUNCTION}	1✔
36	CONST_STR = 'str'	1✔
37	CONST_DICT = 'dict'	1✔
38
39	def __init__(self, log_type, normalized_type, params):	1✔
40	"""Init NormalizatedType
41	Args:
42	log_type (str): log type name, e.g. osquery:differential
43	normalized_type (str): Normalized type name defined in conf/, e.g. 'sourceAddress',
44	'destination_ip' may be normalized to 'ip_address'.
45	params (list): a list of str or dict contains normalization configuration read from
46	conf/schemas/*.json. The params can be a list of str or a list of dict to specify
47	the path to the keys which will be normalized.
48	e.g.
49	['path', 'to', 'the', 'key']
50	or
51	[
52	{
53	'path': ['detail', 'sourceIPAddress'],
54	'function': 'source ip address'
55	},
56	{
57	'path': ['path', 'to', 'the', 'key'],
58	'function': 'destination ip address'
59	}
60	]
61	"""
62	self._log_type = log_type	1✔
63	self._log_source = log_type.split(':')[0]	1✔
64	self._normalized_type = normalized_type	1✔
65	self._parsed_params = self._parse_params(params)	1✔
66
67	def __eq__(self, other):	1✔
68	"""Compare two NormalizedType instances and it is very helpful in unit test when use
69	assert_equal
70	"""
71	if not (self._log_type == other.log_type	1✔
72	and self._log_source == other.log_source
73	and self._normalized_type == other.normalized_type):
74	return False	×
75
76	if len(self._parsed_params) != len(other.parsed_params):	1✔
77	return False	×
78
79	for idx in range(len(self._parsed_params)):	1✔
80	if self._parsed_params[idx][CONST_PATH] == other.parsed_params[idx][CONST_PATH]:	1✔
81	continue	1✔
82
83	return False	×
84
85	return True	1✔
86
87	@property	1✔
88	def log_type(self):
89	"""Return the log type name, e.g. 'osquery:differential'"""
90	return self._log_type	1✔
91
92	@property	1✔
93	def log_source(self):
94	"""Return the log source name, e.g. 'osquery'"""
95	return self._log_source	1✔
96
97	@property	1✔
98	def normalized_type(self):
99	"""Return the normalized type, e.g. 'ip_address'"""
100	return self._normalized_type	1✔
101
102	@property	1✔
103	def parsed_params(self):
104	"""Return the normalization configuration which is a list of dict, e.g.
105	[
106	{
107	'path': ['path', 'to', 'the', 'key'],
108	'function': None
109	}
110	]
111
112	or
113	[
114	{
115	'path': ['detail', 'sourceIPAddress'],
116	'function': 'source ip address'
117	},
118	{
119	'path': ['path', 'to', 'the', 'destination', 'ip'],
120	'function': 'destination ip address'
121	}
122	]
123	"""
124	return self._parsed_params	1✔
125
126	def _parse_params(self, params):	1✔
127	"""Extract path and function information from params argument
128
129	Args:
130	params (list): a list of str or dict contains normalization configuration.
131	"""
132	param_type = self._parse_param_type(params)	1✔
133
134	if param_type == self.CONST_STR:	1✔
135	# Format params to include 'function' field which is set to None.
136	return [	1✔
137	{
138	CONST_PATH: params,
139	CONST_FUNCTION: None
140	}
141	]
142
143	return params	1✔
144
145	def _parse_param_type(self, params):	1✔
146	"""Parse all param type in params
147
148	Args:
149	params (list): a list of str or dict contains normalization configuration.
150	"""
151	if not isinstance(params, list):	1✔
152	raise ConfigError(	1✔
153	'Unsupported params {} for normalization. Convert params to a list'.format(params)
154	)
155
156	if all(isinstance(param, str) for param in params):	1✔
157	return self.CONST_STR	1✔
158
159	if all(isinstance(param, dict) and set(param.keys()) == self.VALID_KEYS	1✔
160	for param in params
161	):
162	return self.CONST_DICT	1✔
163
164	# FIXME: should we raise exception here? Or may just return False and log a warming message
165	raise ConfigError(	×
166	('Unsupported type(s) used in {} or missing keys. Valid types are str or dict and '
167	'valid keys are {}').format(params, self.VALID_KEYS)
168	)
169
170
171	class Normalizer:	1✔
172	"""Normalizer class to handle log key normalization in payloads"""
173
174	NORMALIZATION_KEY = 'streamalert:normalization'	1✔
175
176	# Store the normalized types mapping to original keys from the records
177	_types_config = dict()	1✔
178
179	@classmethod	1✔
180	def match_types(cls, record, normalized_types):
181	"""Check for normalized types within record
182
183	Args:
184	record (dict): Parsed payload of log
185	normalized_types (dict): Normalized types mapping
186
187	Returns:
188	dict: A dict of normalized keys with a list of values
189
190	Example:
191	return
192	{
193	'region': [
194	{
195	'values': ['us-east-1']
196	'function': 'AWS region'
197	},
198	{
199	'values': ['us-west-2']
200	'function': 'AWS region'
201	}
202	]
203	}
204	"""
205	results = {}	1✔
206	for type_name, type_info in normalized_types.items():	1✔
207	result = list(cls._extract_values(record, type_info))	1✔
208
209	if result:	1✔
210	results[type_name] = result	1✔
211
212	return results	1✔
213
214
215	@classmethod	1✔
216	def _extract_values(cls, record, paths_to_normalize):
217	"""Recursively extract lists of path parts from a dictionary
218
219	Args:
220	record (dict): Parsed payload of log
221	paths_to_normalize (set): Normalized keys for which to extract paths
222	path (list=None): Parts of current path for which keys are being extracted
223
224	Yields:
225	dict: A dict contians the values of normalized types. For example,
226	{
227	'values': ['1.1.1.2']
228	'function': 'Source ip address'
229	}
230	"""
231	for param in paths_to_normalize.parsed_params:	1✔
232	found_value = False	1✔
233	value = record	1✔
234	for key in param.get(CONST_PATH):	1✔
235	value = value.get(key)	1✔
236	if not value:	1✔
237	break	1✔
238	found_value = True	1✔
239
240	if found_value:	1✔
241	yield {	1✔
242	CONST_FUNCTION: param.get(CONST_FUNCTION) or None,
243	# if value not a list, it will be cast to a str even it is a dict or other
244	# types
245	CONST_VALUES: value if isinstance(value, list) else [str(value)]
246	}
247
248	@classmethod	1✔
249	def normalize(cls, record, log_type):
250	"""Apply data normalization to a record
251
252	Args:
253	record (dict): The parsed log without data normalization
254	log_type (str): Type of log for which to apply normalizaiton
255	"""
256	log_normalized_types = cls._types_config.get(log_type)	1✔
257	if not log_normalized_types:	1✔
258	LOGGER.debug('No normalized types defined for log type: %s', log_type)	1✔
259	return	1✔
260
261	# Add normalized keys to the record
262	record.update({cls.NORMALIZATION_KEY: cls.match_types(record, log_normalized_types)})	1✔
263
264	@classmethod	1✔
265	def get_values_for_normalized_type(cls, record, datatype):
266	"""Fetch values by normalized_type.
267
268	Args:
269	record (dict): parsed payload of any log
270	datatype (str): normalized type being found
271
272	Returns:
273	set: The values for the normalized type specified
274	"""
275	normalization_results = record.get(cls.NORMALIZATION_KEY, {}).get(datatype)	1✔
276	if not normalization_results:	1✔
277	return	1✔
278	return set(itertools.chain(*[result.get(CONST_VALUES) for result in normalization_results]))	1✔
279
280	@classmethod	1✔
281	def load_from_config(cls, config):
282	"""Extract and store the data types from the config in the proper format
283
284	Args:
285	config (dict): Config read from 'conf/' directory
286
287	Returns:
288	Normalizer: Class to be used for normalization logic
289	"""
290	if cls._types_config:	1✔
291	return cls # config is already populated	1✔
292
293	cls._types_config = cls._parse_normalization(config)	1✔
294
295	return cls # there are no instance methods, so just return the class	1✔
296
297	@classmethod	1✔
298	def _parse_normalization(cls, config):
299	"""Load and parse normalization config from conf/schemas/*.json. Normalization will be
300	configured along with log schema and a path will be provided to find the original key.
301
302	For example: conf/schemas/cloudwatch.json looks like
303	'cloudwatch:events': {
304	'schema': {
305	'account': 'string',
306	'source': 'string',
307	'other_key': 'string'
308	},
309	'configuration': {
310	'normalization': {
311	'region': ['path', 'to', 'original', 'key'],
312	'ip_address': [
313	{
314	'path': ['detail', 'sourceIPAddress'],
315	'function': 'source ip address'
316	},
317	{
318	'path': ['path', 'to', 'original', 'key'],
319	'function': 'destination ip address'
320	}
321	]
322	}
323	}
324	}
325
326	Args:
327	config (dict): Config read from 'conf/' directory
328
329	Returns:
330	dict: return a dict contains normalization information per log type basis.
331	{
332	'cloudwatch:events': {
333	'region': NormalizedType(),
334	'ip_address': NormalizedType()
335	}
336	}
337	"""
338	normalized_config = defaultdict(dict)	1✔
339	for log_type, val in config.get(TopLevelConfigKeys.LOGS, {}).items():	1✔
340	result = defaultdict(dict)	1✔
341
342	log_type_normalization = val.get('configuration', {}).get('normalization', {})	1✔
343
344	for normalized_type, params in log_type_normalization.items():	1✔
345	# add normalization info if it is defined in log type configuration field
346	result[normalized_type] = NormalizedType(log_type, normalized_type, params)	1✔
347
348	if result:	1✔
349	normalized_config[log_type] = result	1✔
350
351	# return None is normalized_config is an empty defaultdict.
352	return normalized_config or None	1✔

airbnb / streamalert / 33703505adb99480a1e32d5da20d782e28f51560-PR-1238

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