certtools / intelmq / 564

pending completion

Build # 564

Build Type

push

travis-ci

Committed by

aaronkaplan

Commit Message

Revert "distinguish which type message is. NOTE: this is an ugly patch! It's just there to note where to fix something or where the symptom can be found. Feel freet to propose a better fix. It works for me now. Fixes #590"

This reverts commit 468f4aac3.

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

5091 of 7175 relevant lines covered (70.95%)

2.13 hits per line

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

75.65

/intelmq/lib/bot.py

# -*- coding: utf-8 -*-
"""

"""

import csv
import datetime
import io
import json
import re
import os
import signal
import sys
import time
import traceback

from intelmq import (DEFAULT_LOGGING_LEVEL, DEFAULT_LOGGING_PATH,
                     DEFAULTS_CONF_FILE, HARMONIZATION_CONF_FILE,
                     PIPELINE_CONF_FILE, RUNTIME_CONF_FILE, SYSTEM_CONF_FILE)
from intelmq.lib import exceptions, utils
from intelmq.lib.message import MessageFactory
from intelmq.lib.pipeline import PipelineFactory

__all__ = ['Bot', 'ParserBot']


class Bot(object):

    def __init__(self, bot_id):
        self.__log_buffer = []
        self.parameters = Parameters()

        self.__current_message = None
        self.__message_counter = 0
        self.__error_retries_counter = 0
        self.__source_pipeline = None
        self.__destination_pipeline = None
        self.logger = None

        try:
            version_info = sys.version.splitlines()[0].strip()
            self.__log_buffer.append(('info',
                                      '{} initialized with id {} and version '
                                      '{} as process {}.'
                                      ''.format(self.__class__.__name__,
                                                bot_id, version_info,
                                                os.getpid())))

            self.__load_defaults_configuration()
            self.__load_system_configuration()

            self.__check_bot_id(bot_id)
            self.__bot_id = bot_id

            if self.parameters.logging_handler == 'syslog':
                syslog = self.parameters.logging_syslog
            else:
                syslog = False
            self.logger = utils.log(self.__bot_id, syslog=syslog,
                                    log_level=self.parameters.logging_level)
        except:
            self.__log_buffer.append(('critical', traceback.format_exc()))
            self.stop()
        else:
            for line in self.__log_buffer:
                getattr(self.logger, line[0])(line[1])

        try:
            self.logger.info('Bot is starting.')
            self.__load_runtime_configuration()
            self.__load_pipeline_configuration()
            self.__load_harmonization_configuration()

            self.init()

            signal.signal(signal.SIGHUP, self.__sighup)
        except:
            self.logger.exception('Bot initialization failed.')
            raise

    def __sighup(self, signum, stack):
        self.logger.info('Received SIGHUP, initializing again.')
        self.__disconnect_pipelines()
        self.logger.handlers = []  # remove all existing handlers
        self.__init__(self.__bot_id)
        self.__connect_pipelines()

    def init(self):
        pass

    def shutdown(self):
        pass

    def start(self, starting=True, error_on_pipeline=True,
              error_on_message=False, source_pipeline=None,
              destination_pipeline=None):
        self.__source_pipeline = source_pipeline
        self.__destination_pipeline = destination_pipeline
        self.logger.info('Bot starts processings.')

        while True:
            try:
                if not starting and (error_on_pipeline or error_on_message):
                    self.logger.info('Bot will restart in %s seconds.' %
                                     self.parameters.error_retry_delay)
                    time.sleep(self.parameters.error_retry_delay)
                    self.logger.info('Bot woke up')
                    self.logger.info('Trying to start processing again.')

                if error_on_message:
                    error_on_message = False

                if error_on_pipeline:
                    self.__connect_pipelines()
                    error_on_pipeline = False

                if starting:
                    self.logger.info("Start processing.")
                    starting = False

                self.process()
                self.__error_retries_counter = 0  # reset counter

                self.logger.info("Idling for {!s}s now.".format(self.parameters.rate_limit))
                time.sleep(self.parameters.rate_limit)

            except exceptions.PipelineError:
                error_on_pipeline = True
                error_traceback = traceback.format_exc()

                if self.parameters.error_log_exception:
                    self.logger.exception('Pipeline failed.')
                else:
                    self.logger.error('Pipeline failed.')
                self.__disconnect_pipelines()

            except Exception:
                error_on_message = True
                error_traceback = traceback.format_exc()

                if self.parameters.error_log_exception:
                    self.logger.exception("Bot has found a problem.")
                else:
                    self.logger.error("Bot has found a problem.")

                if self.parameters.error_log_message:
                    # Dump full message if explicitly requested by config
                    self.logger.info("Current Message(event): {!r}."
                                     "".format(self.__current_message))

            except KeyboardInterrupt:
                self.logger.error("Received KeyboardInterrupt.")
                self.stop()
                break

            finally:
                if self.parameters.testing:
                    self.stop()
                    break

                if error_on_message or error_on_pipeline:
                    self.__error_retries_counter += 1

                    # reached the maximum number of retries
                    if (self.__error_retries_counter >
                            self.parameters.error_max_retries):

                        if error_on_message:

                            if self.parameters.error_dump_message:
                                self._dump_message(error_traceback,
                                                   message=self.__current_message)
                                self.__current_message = None

                            # remove message from pipeline
                            self.acknowledge_message()

                            # when bot acknowledge the message,
                            # dont need to wait again
                            error_on_message = False

                        # error_procedure: stop
                        if self.parameters.error_procedure == "stop":
                            self.stop()

                        # error_procedure: pass
                        else:
                            self.__error_retries_counter = 0  # reset counter

    def __del__(self):
        self.stop()

    def stop(self):
        self.shutdown()

        self.__disconnect_pipelines()

        if self.logger:
            self.logger.info("Bot stopped.")
        else:
            self.__log_buffer.append(('info', 'Bot stopped.'))
            self.__print_log_buffer()

        if not self.parameters.testing:
            self.__terminate()

    def __terminate(self):
        try:
            self.logger.error("Exiting.")
        except:
            print("Exiting")
        exit(-1)

    def __print_log_buffer(self):
        for level, message in self.__log_buffer:
            if self.logger:
                getattr(self.logger, level)(message)
            print(level.upper(), '-', message)
        self.__log_buffer = []

    def __check_bot_id(self, str):
        res = re.search('[^0-9a-zA-Z\-]+', str)
        if res:
            self.__log_buffer.append(('error',
                                      "Invalid bot id, must match '"
                                      "[^0-9a-zA-Z\-]+'."))
            self.stop()

    def __connect_pipelines(self):
        self.logger.debug("Loading source pipeline.")
        self.__source_pipeline = PipelineFactory.create(self.parameters)
        self.logger.debug("Loading source queue.")
        self.__source_pipeline.set_queues(self.__source_queues, "source")
        self.logger.debug("Source queue loaded {}."
                         "".format(self.__source_queues))
        self.__source_pipeline.connect()
        self.logger.debug("Connected to source queue.")

        self.logger.debug("Loading destination pipeline.")
        self.__destination_pipeline = PipelineFactory.create(self.parameters)
        self.logger.debug("Loading destination queues.")
        self.__destination_pipeline.set_queues(self.__destination_queues,
                                               "destination")
        self.logger.debug("Destination queues loaded {}."
                         "".format(self.__destination_queues))
        self.__destination_pipeline.connect()
        self.logger.debug("Connected to destination queues.")

        self.logger.info("Pipeline ready.")

    def __disconnect_pipelines(self):
        """ Disconnecting pipelines. """
        if self.__source_pipeline:
            self.__source_pipeline.disconnect()
            self.__source_pipeline = None
            self.logger.debug("Disconnecting from source pipeline.")
        if self.__destination_pipeline:
            self.__destination_pipeline.disconnect()
            self.__destination_pipeline = None
            self.logger.debug("Disconnecting from destination pipeline.")

    def send_message(self, *messages):
        for message in messages:
            if not message:
                self.logger.warning("Ignoring empty message at sending.")
                continue

            self.logger.debug("Sending message.")
            self.__message_counter += 1
            if self.__message_counter % 500 == 0:
                self.logger.info("Processed %s messages." % self.__message_counter)

            raw_message = MessageFactory.serialize(message)
            self.__destination_pipeline.send(raw_message)

    def receive_message(self):
        self.logger.debug('Waiting for incoming message.')
        message = None
        while not message:
            message = self.__source_pipeline.receive()
            if not message:
                self.logger.warning('Empty message received.')
                continue
        self.__current_message = MessageFactory.unserialize(message)

        if 'raw' in self.__current_message and len(self.__current_message['raw']) > 400:
            tmp_msg = self.__current_message.to_dict(hierarchical=False)
            tmp_msg['raw'] = tmp_msg['raw'][:397] + '...'
        else:
            tmp_msg = self.__current_message
        self.logger.debug('Received message {!r}.'.format(tmp_msg))

        return self.__current_message

    def acknowledge_message(self):
        self.__source_pipeline.acknowledge()

    def _dump_message(self, error_traceback, message):
        if message is None:
            return

        self.logger.info('Dumping message from pipeline to dump file.')
        timestamp = datetime.datetime.utcnow()
        timestamp = timestamp.isoformat()

        dump_file = "%s%s.dump" % (self.parameters.logging_path, self.__bot_id)

        new_dump_data = dict()
        new_dump_data[timestamp] = dict()
        new_dump_data[timestamp]["bot_id"] = self.__bot_id
        new_dump_data[timestamp]["source_queue"] = self.__source_queues
        new_dump_data[timestamp]["traceback"] = error_traceback

        new_dump_data[timestamp]["message"] = message.serialize()

        try:
            with open(dump_file, 'r') as fp:
                dump_data = json.load(fp)
                dump_data.update(new_dump_data)
        except:
            dump_data = new_dump_data

        with open(dump_file, 'w') as fp:
            json.dump(dump_data, fp, indent=4, sort_keys=True)

        self.logger.warn('Message dumped.')

    def __load_defaults_configuration(self):
        self.__log_buffer.append(('debug', "Loading defaults configuration."))
        config = utils.load_configuration(DEFAULTS_CONF_FILE)

        setattr(self.parameters, 'testing', False)
        setattr(self.parameters, 'logging_path', DEFAULT_LOGGING_PATH)
        setattr(self.parameters, 'logging_level', DEFAULT_LOGGING_LEVEL)

        for option, value in config.items():
            setattr(self.parameters, option, value)
            self.__log_buffer.append(('debug',
                                      "Defaults configuration: parameter {!r} "
                                      "loaded  with value {!r}.".format(option,
                                                                        value)))

    def __load_system_configuration(self):
        self.__log_buffer.append(('debug', "Loading system configuration"))
        config = utils.load_configuration(SYSTEM_CONF_FILE)

        for option, value in config.items():
            setattr(self.parameters, option, value)
            self.__log_buffer.append(('debug',
                                      "System configuration: parameter {!r} "
                                      "loaded  with value {!r}.".format(option,
                                                                        value)))

    def __load_runtime_configuration(self):
        self.logger.debug("Loading runtime configuration")
        config = utils.load_configuration(RUNTIME_CONF_FILE)

        if self.__bot_id in list(config.keys()):
            for option, value in config[self.__bot_id].items():
                setattr(self.parameters, option, value)
                self.logger.debug("Runtime configuration: parameter {!r} "
                                  "loaded with value {!r}.".format(option, value))

    def __load_pipeline_configuration(self):
        self.logger.debug("Loading pipeline configuration")
        config = utils.load_configuration(PIPELINE_CONF_FILE)

        self.__source_queues = None
        self.__destination_queues = None

        if self.__bot_id in list(config.keys()):

            if 'source-queue' in config[self.__bot_id].keys():
                self.__source_queues = config[self.__bot_id]['source-queue']
                self.logger.debug("Pipeline configuration: parameter "
                                  "'source-queue' loaded with the value {!r}."
                                  "".format(self.__source_queues))

            if 'destination-queues' in config[self.__bot_id].keys():

                self.__destination_queues = config[
                    self.__bot_id]['destination-queues']
                self.logger.debug("Pipeline configuration: parameter"
                                  "'destination-queues' loaded with the value "
                                  "{!r}.".format(", ".join(self.__destination_queues)))

        else:
            self.logger.error("Pipeline configuration: no key "
                              "{!r}.".format(self.__bot_id))
            self.stop()

    def __load_harmonization_configuration(self):
        self.logger.debug("Loading Harmonization configuration.")
        config = utils.load_configuration(HARMONIZATION_CONF_FILE)

        for message_types in config.keys():
            for key in config[message_types].keys():
                for _key in config.keys():
                    if _key.startswith("%s." % key):
                        raise exceptions.ConfigurationError(
                            'harmonization',
                            "Key %s is not valid." % _key)


class ParserBot(Bot):

    def parse_csv(self, report):
        """
        A basic CSV parser.
        """
        raw_report = utils.base64_decode(report.get("raw"))
        for line in csv.reader(io.StringIO(raw_report)):
            yield line

    def parse(self, report):
        """
        A generator yielding the single elements of the data.

        Comments, headers etc. can be processed here. Data needed by
        `self.parse_line` can be saved in `self.tempdata` (list).

        Default parser yields stripped lines.
        Override for your use or use an exisiting parser, e.g.:
            parse = ParserBot.parse_csv
        """
        for line in utils.base64_decode(report.get("raw")).splitlines():
            yield line.strip()

    def parse_line(self, line, report):
        """
        A generator which can yield one or more messages contained in line.

        Report has the full message, thus you can access some metadata.
        Override for your use.
        """
        raise NotImplementedError

    def process(self):
        self.tempdata = []  # temporary data for parse, parse_line and recover_line
        self.__failed = []
        report = self.receive_message()

        for line in self.parse(report):
            if not line:
                continue
            try:
                # filter out None
                events = list(filter(bool, self.parse_line(line, report)))
            except Exception as exc:
                self.logger.exception('Failed to parse line.')
                self.__failed.append((exc, line))
            else:
                self.send_message(*events)

        for exc, line in self.__failed:
            self._dump_message(exc, self.recover_line(line))

        self.acknowledge_message()

    def recover_line(self, line):
        """
        Reverse of parse for single lines.

        Recovers a fully functional report with only the problematic line.
        """
        return '\n'.join(self.tempdata + [line])

    def recover_line_csv(self, line):
        out = io.StringIO()
        writer = csv.writer(out)
        writer.writerow(line)
        return out.getvalue()

    csv_params = {}

    def recover_line_csv_dict(self, line):
        """
        Converts dictionaries to csv. self.csv_fieldnames must be list of fields.
        """
        out = io.StringIO()
        writer = csv.DictWriter(out, self.csv_fieldnames, **self.csv_params)
        writer.writeheader()
        writer.writerow(line)
        return out.getvalue()


class Parameters(object):
    pass

1	# -- coding: utf-8 --
2	"""	3✔
3
4	"""
5
6	import csv	3✔
7	import datetime	3✔
8	import io	3✔
9	import json	3✔
10	import re	3✔
11	import os	3✔
12	import signal	3✔
13	import sys	3✔
14	import time	3✔
15	import traceback	3✔
16
17	from intelmq import (DEFAULT_LOGGING_LEVEL, DEFAULT_LOGGING_PATH,	3✔
18	DEFAULTS_CONF_FILE, HARMONIZATION_CONF_FILE,
19	PIPELINE_CONF_FILE, RUNTIME_CONF_FILE, SYSTEM_CONF_FILE)
20	from intelmq.lib import exceptions, utils	3✔
21	from intelmq.lib.message import MessageFactory	3✔
22	from intelmq.lib.pipeline import PipelineFactory	3✔
23
24	__all__ = ['Bot', 'ParserBot']	3✔
25
26
27	class Bot(object):	3✔
28
29	def __init__(self, bot_id):	3✔
30	self.__log_buffer = []	3✔
31	self.parameters = Parameters()	3✔
32
33	self.__current_message = None	3✔
34	self.__message_counter = 0	3✔
35	self.__error_retries_counter = 0	3✔
36	self.__source_pipeline = None	3✔
37	self.__destination_pipeline = None	3✔
38	self.logger = None	3✔
39
40	try:	3✔
41	version_info = sys.version.splitlines()[0].strip()	3✔
42	self.__log_buffer.append(('info',	3✔
43	'{} initialized with id {} and version '
44	'{} as process {}.'
45	''.format(self.__class__.__name__,
46	bot_id, version_info,
47	os.getpid())))
48
49	self.__load_defaults_configuration()	3✔
50	self.__load_system_configuration()	3✔
51
52	self.__check_bot_id(bot_id)	3✔
53	self.__bot_id = bot_id	3✔
54
55	if self.parameters.logging_handler == 'syslog':	3✔
56	syslog = self.parameters.logging_syslog	×
57	else:
58	syslog = False	3✔
59	self.logger = utils.log(self.__bot_id, syslog=syslog,	3✔
60	log_level=self.parameters.logging_level)
61	except:	×
62	self.__log_buffer.append(('critical', traceback.format_exc()))	×
63	self.stop()	×
64	else:
65	for line in self.__log_buffer:	3✔
66	getattr(self.logger, line[0])(line[1])	3✔
67
68	try:	3✔
69	self.logger.info('Bot is starting.')	3✔
70	self.__load_runtime_configuration()	3✔
71	self.__load_pipeline_configuration()	3✔
72	self.__load_harmonization_configuration()	3✔
73
74	self.init()	3✔
75
76	signal.signal(signal.SIGHUP, self.__sighup)	3✔
77	except:	×
78	self.logger.exception('Bot initialization failed.')	×
79	raise	×
80
81	def __sighup(self, signum, stack):	3✔
82	self.logger.info('Received SIGHUP, initializing again.')	×
83	self.__disconnect_pipelines()	×
84	self.logger.handlers = [] # remove all existing handlers	×
85	self.__init__(self.__bot_id)	×
86	self.__connect_pipelines()	×
87
88	def init(self):	3✔
89	pass	3✔
90
91	def shutdown(self):	3✔
92	pass	3✔
93
94	def start(self, starting=True, error_on_pipeline=True,	3✔
95	error_on_message=False, source_pipeline=None,
96	destination_pipeline=None):
97	self.__source_pipeline = source_pipeline	3✔
98	self.__destination_pipeline = destination_pipeline	3✔
99	self.logger.info('Bot starts processings.')	3✔
100
101	while True:	3✔
102	try:	3✔
103	if not starting and (error_on_pipeline or error_on_message):	3✔
104	self.logger.info('Bot will restart in %s seconds.' %	×
105	self.parameters.error_retry_delay)
106	time.sleep(self.parameters.error_retry_delay)	×
107	self.logger.info('Bot woke up')	×
108	self.logger.info('Trying to start processing again.')	×
109
110	if error_on_message:	3✔
111	error_on_message = False	×
112
113	if error_on_pipeline:	3✔
114	self.__connect_pipelines()	3✔
115	error_on_pipeline = False	3✔
116
117	if starting:	3✔
118	self.logger.info("Start processing.")	3✔
119	starting = False	3✔
120
121	self.process()	3✔
122	self.__error_retries_counter = 0 # reset counter	3✔
123
124	self.logger.info("Idling for {!s}s now.".format(self.parameters.rate_limit))	3✔
125	time.sleep(self.parameters.rate_limit)	3✔
126
127	except exceptions.PipelineError:	3✔
128	error_on_pipeline = True	3✔
129	error_traceback = traceback.format_exc()	3✔
130
131	if self.parameters.error_log_exception:	3✔
132	self.logger.exception('Pipeline failed.')	3✔
133	else:
134	self.logger.error('Pipeline failed.')	×
135	self.__disconnect_pipelines()	3✔
136
137	except Exception:	3✔
138	error_on_message = True	3✔
139	error_traceback = traceback.format_exc()	3✔
140
141	if self.parameters.error_log_exception:	3✔
142	self.logger.exception("Bot has found a problem.")	3✔
143	else:
144	self.logger.error("Bot has found a problem.")	×
145
146	if self.parameters.error_log_message:	3✔
147	# Dump full message if explicitly requested by config
148	self.logger.info("Current Message(event): {!r}."	3✔
149	"".format(self.__current_message))
150
151	except KeyboardInterrupt:	×
152	self.logger.error("Received KeyboardInterrupt.")	×
153	self.stop()	×
154	break	×
155
156	finally:
157	if self.parameters.testing:	3✔
158	self.stop()	3✔
159	break	3✔
160
161	if error_on_message or error_on_pipeline:	×
162	self.__error_retries_counter += 1	×
163
164	# reached the maximum number of retries
165	if (self.__error_retries_counter >	×
166	self.parameters.error_max_retries):
167
168	if error_on_message:	×
169
170	if self.parameters.error_dump_message:	×
171	self._dump_message(error_traceback,	×
172	message=self.__current_message)
173	self.__current_message = None	×
174
175	# remove message from pipeline
176	self.acknowledge_message()	×
177
178	# when bot acknowledge the message,
179	# dont need to wait again
180	error_on_message = False	×
181
182	# error_procedure: stop
183	if self.parameters.error_procedure == "stop":	×
184	self.stop()	×
185
186	# error_procedure: pass
187	else:
188	self.__error_retries_counter = 0 # reset counter	×
189
190	def __del__(self):	3✔
191	self.stop()	×
192
193	def stop(self):	3✔
194	self.shutdown()	3✔
195
196	self.__disconnect_pipelines()	3✔
197
198	if self.logger:	3✔
199	self.logger.info("Bot stopped.")	3✔
200	else:
201	self.__log_buffer.append(('info', 'Bot stopped.'))	×
202	self.__print_log_buffer()	×
203
204	if not self.parameters.testing:	3✔
205	self.__terminate()	×
206
207	def __terminate(self):	3✔
208	try:	×
209	self.logger.error("Exiting.")	×
210	except:	×
211	print("Exiting")	×
212	exit(-1)	×
213
214	def __print_log_buffer(self):	3✔
215	for level, message in self.__log_buffer:	×
216	if self.logger:	×
217	getattr(self.logger, level)(message)	×
218	print(level.upper(), '-', message)	×
219	self.__log_buffer = []	×
220
221	def __check_bot_id(self, str):	3✔
222	res = re.search('[^0-9a-zA-Z\-]+', str)	3✔
223	if res:	3✔
224	self.__log_buffer.append(('error',	×
225	"Invalid bot id, must match '"
226	"[^0-9a-zA-Z\-]+'."))
227	self.stop()	×
228
229	def __connect_pipelines(self):	3✔
230	self.logger.debug("Loading source pipeline.")	3✔
231	self.__source_pipeline = PipelineFactory.create(self.parameters)	3✔
232	self.logger.debug("Loading source queue.")	3✔
233	self.__source_pipeline.set_queues(self.__source_queues, "source")	3✔
234	self.logger.debug("Source queue loaded {}."	3✔
235	"".format(self.__source_queues))
236	self.__source_pipeline.connect()	3✔
237	self.logger.debug("Connected to source queue.")	3✔
238
239	self.logger.debug("Loading destination pipeline.")	3✔
240	self.__destination_pipeline = PipelineFactory.create(self.parameters)	3✔
241	self.logger.debug("Loading destination queues.")	3✔
242	self.__destination_pipeline.set_queues(self.__destination_queues,	3✔
243	"destination")
244	self.logger.debug("Destination queues loaded {}."	3✔
245	"".format(self.__destination_queues))
246	self.__destination_pipeline.connect()	3✔
247	self.logger.debug("Connected to destination queues.")	3✔
248
249	self.logger.info("Pipeline ready.")	3✔
250
251	def __disconnect_pipelines(self):	3✔
252	""" Disconnecting pipelines. """
253	if self.__source_pipeline:	3✔
254	self.__source_pipeline.disconnect()	3✔
255	self.__source_pipeline = None	3✔
256	self.logger.debug("Disconnecting from source pipeline.")	3✔
257	if self.__destination_pipeline:	3✔
258	self.__destination_pipeline.disconnect()	3✔
259	self.__destination_pipeline = None	3✔
260	self.logger.debug("Disconnecting from destination pipeline.")	3✔
261
262	def send_message(self, *messages):	3✔
263	for message in messages:	3✔
264	if not message:	3✔
265	self.logger.warning("Ignoring empty message at sending.")	3✔
266	continue	3✔
267
268	self.logger.debug("Sending message.")	3✔
269	self.__message_counter += 1	3✔
270	if self.__message_counter % 500 == 0:	3✔
271	self.logger.info("Processed %s messages." % self.__message_counter)	×
272
273	raw_message = MessageFactory.serialize(message)	3✔
274	self.__destination_pipeline.send(raw_message)	3✔
275
276	def receive_message(self):	3✔
277	self.logger.debug('Waiting for incoming message.')	3✔
278	message = None	3✔
279	while not message:	3✔
280	message = self.__source_pipeline.receive()	3✔
281	if not message:	3✔
282	self.logger.warning('Empty message received.')	×
283	continue	×
284	self.__current_message = MessageFactory.unserialize(message)	3✔
285
286	if 'raw' in self.__current_message and len(self.__current_message['raw']) > 400:	3✔
287	tmp_msg = self.__current_message.to_dict(hierarchical=False)	3✔
288	tmp_msg['raw'] = tmp_msg['raw'][:397] + '...'	3✔
289	else:
290	tmp_msg = self.__current_message	3✔
291	self.logger.debug('Received message {!r}.'.format(tmp_msg))	3✔
292
293	return self.__current_message	3✔
294
295	def acknowledge_message(self):	3✔
296	self.__source_pipeline.acknowledge()	3✔
297
298	def _dump_message(self, error_traceback, message):	3✔
299	if message is None:	3✔
300	return	×
301
302	self.logger.info('Dumping message from pipeline to dump file.')	3✔
303	timestamp = datetime.datetime.utcnow()	3✔
304	timestamp = timestamp.isoformat()	3✔
305
306	dump_file = "%s%s.dump" % (self.parameters.logging_path, self.__bot_id)	3✔
307
308	new_dump_data = dict()	3✔
309	new_dump_data[timestamp] = dict()	3✔
310	new_dump_data[timestamp]["bot_id"] = self.__bot_id	3✔
311	new_dump_data[timestamp]["source_queue"] = self.__source_queues	3✔
312	new_dump_data[timestamp]["traceback"] = error_traceback	3✔
313
314	new_dump_data[timestamp]["message"] = message.serialize()	3✔
315
316	try:	×
317	with open(dump_file, 'r') as fp:	×
318	dump_data = json.load(fp)	×
319	dump_data.update(new_dump_data)	×
320	except:	×
321	dump_data = new_dump_data	×
322
323	with open(dump_file, 'w') as fp:	×
324	json.dump(dump_data, fp, indent=4, sort_keys=True)	×
325
326	self.logger.warn('Message dumped.')	×
327
328	def __load_defaults_configuration(self):	3✔
329	self.__log_buffer.append(('debug', "Loading defaults configuration."))	3✔
330	config = utils.load_configuration(DEFAULTS_CONF_FILE)	3✔
331
332	setattr(self.parameters, 'testing', False)	3✔
333	setattr(self.parameters, 'logging_path', DEFAULT_LOGGING_PATH)	3✔
334	setattr(self.parameters, 'logging_level', DEFAULT_LOGGING_LEVEL)	3✔
335
336	for option, value in config.items():	3✔
337	setattr(self.parameters, option, value)	3✔
338	self.__log_buffer.append(('debug',	3✔
339	"Defaults configuration: parameter {!r} "
340	"loaded with value {!r}.".format(option,
341	value)))
342
343	def __load_system_configuration(self):	3✔
344	self.__log_buffer.append(('debug', "Loading system configuration"))	3✔
345	config = utils.load_configuration(SYSTEM_CONF_FILE)	3✔
346
347	for option, value in config.items():	3✔
348	setattr(self.parameters, option, value)	3✔
349	self.__log_buffer.append(('debug',	3✔
350	"System configuration: parameter {!r} "
351	"loaded with value {!r}.".format(option,
352	value)))
353
354	def __load_runtime_configuration(self):	3✔
355	self.logger.debug("Loading runtime configuration")	3✔
356	config = utils.load_configuration(RUNTIME_CONF_FILE)	3✔
357
358	if self.__bot_id in list(config.keys()):	3✔
359	for option, value in config[self.__bot_id].items():	3✔
360	setattr(self.parameters, option, value)	×
361	self.logger.debug("Runtime configuration: parameter {!r} "	×
362	"loaded with value {!r}.".format(option, value))
363
364	def __load_pipeline_configuration(self):	3✔
365	self.logger.debug("Loading pipeline configuration")	3✔
366	config = utils.load_configuration(PIPELINE_CONF_FILE)	3✔
367
368	self.__source_queues = None	3✔
369	self.__destination_queues = None	3✔
370
371	if self.__bot_id in list(config.keys()):	3✔
372
373	if 'source-queue' in config[self.__bot_id].keys():	3✔
374	self.__source_queues = config[self.__bot_id]['source-queue']	3✔
375	self.logger.debug("Pipeline configuration: parameter "	3✔
376	"'source-queue' loaded with the value {!r}."
377	"".format(self.__source_queues))
378
379	if 'destination-queues' in config[self.__bot_id].keys():	3✔
380
381	self.__destination_queues = config[	3✔
382	self.__bot_id]['destination-queues']
383	self.logger.debug("Pipeline configuration: parameter"	3✔
384	"'destination-queues' loaded with the value "
385	"{!r}.".format(", ".join(self.__destination_queues)))
386
387	else:
388	self.logger.error("Pipeline configuration: no key "	×
389	"{!r}.".format(self.__bot_id))
390	self.stop()	×
391
392	def __load_harmonization_configuration(self):	3✔
393	self.logger.debug("Loading Harmonization configuration.")	3✔
394	config = utils.load_configuration(HARMONIZATION_CONF_FILE)	3✔
395
396	for message_types in config.keys():	3✔
397	for key in config[message_types].keys():	3✔
398	for _key in config.keys():	3✔
399	if _key.startswith("%s." % key):	3✔
400	raise exceptions.ConfigurationError(	×
401	'harmonization',
402	"Key %s is not valid." % _key)
403
404
405	class ParserBot(Bot):	3✔
406
407	def parse_csv(self, report):	3✔
408	"""
409	A basic CSV parser.
410	"""
411	raw_report = utils.base64_decode(report.get("raw"))	3✔
412	for line in csv.reader(io.StringIO(raw_report)):	3✔
413	yield line	3✔
414
415	def parse(self, report):	3✔
416	"""
417	A generator yielding the single elements of the data.
418
419	Comments, headers etc. can be processed here. Data needed by
420	`self.parse_line` can be saved in `self.tempdata` (list).
421
422	Default parser yields stripped lines.
423	Override for your use or use an exisiting parser, e.g.:
424	parse = ParserBot.parse_csv
425	"""
426	for line in utils.base64_decode(report.get("raw")).splitlines():	3✔
427	yield line.strip()	3✔
428
429	def parse_line(self, line, report):	3✔
430	"""
431	A generator which can yield one or more messages contained in line.
432
433	Report has the full message, thus you can access some metadata.
434	Override for your use.
435	"""
436	raise NotImplementedError	×
437
438	def process(self):	3✔
439	self.tempdata = [] # temporary data for parse, parse_line and recover_line	3✔
440	self.__failed = []	3✔
441	report = self.receive_message()	3✔
442
443	for line in self.parse(report):	3✔
444	if not line:	3✔
445	continue	3✔
446	try:	3✔
447	# filter out None
448	events = list(filter(bool, self.parse_line(line, report)))	3✔
449	except Exception as exc:	3✔
450	self.logger.exception('Failed to parse line.')	3✔
451	self.__failed.append((exc, line))	3✔
452	else:
453	self.send_message(*events)	3✔
454
455	for exc, line in self.__failed:	3✔
456	self._dump_message(exc, self.recover_line(line))	3✔
457
458	self.acknowledge_message()	3✔
459
460	def recover_line(self, line):	3✔
461	"""
462	Reverse of parse for single lines.
463
464	Recovers a fully functional report with only the problematic line.
465	"""
466	return '\n'.join(self.tempdata + [line])	3✔
467
468	def recover_line_csv(self, line):	3✔
469	out = io.StringIO()	3✔
470	writer = csv.writer(out)	3✔
471	writer.writerow(line)	3✔
472	return out.getvalue()	3✔
473
474	csv_params = {}	3✔
475
476	def recover_line_csv_dict(self, line):	3✔
477	"""
478	Converts dictionaries to csv. self.csv_fieldnames must be list of fields.
479	"""
480	out = io.StringIO()	×
481	writer = csv.DictWriter(out, self.csv_fieldnames, **self.csv_params)	×
482	writer.writeheader()	×
483	writer.writerow(line)	×
484	return out.getvalue()	×
485
486
487	class Parameters(object):	3✔
488	pass	3✔

certtools / intelmq / 564

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