Merge branch 'develop' into fix-2169

Author: waldbauer-certat

CHANGELOG.md

@@ -29,6 +29,8 @@ CHANGELOG
   The `LogLevel` and `ReturnType` Enums were added to `intelmq.lib.datatypes`.
 - `intelmq.lib.bot`:
   - Enhance behaviour if an unconfigured bot is started (PR#2054 by Sebastian Wagner).
+  - Fix line recovery and message dumping of the `ParserBot` (PR#2192 by Sebastian Wagner).
+    - Previously the dumped message was always the last message of a report if the report contained muliple lines leading to data-loss.
 
 ### Development
 
@@ -73,6 +75,7 @@ CHANGELOG
   - Added support for `Accessible AMQP`, `Device Identification Report` (IPv4 and IPv6) (PR#2134 by Mateo Durante).
   - Added file name mapping for `SSL-POODLE-Vulnerable-Servers IPv6` (file name `scan6_ssl_poodle`) (PR#2134 by Mateo Durante).
 - `intelmq.bots.parsers.cymru.parser_cap_program`: The parser mapped the hostname into `source.fqdn` which is not allowed by the IntelMQ Data Format. Added a check (PR#2215 by Sebastian Waldbauer, fixes #2169)
+- `intelmq.bots.parsers.generic.parser_csv`: Use RewindableFileHandle to use the original current line for line recovery (PR#2192 by Sebastian Wagner).
 
 #### Experts
 - `intelmq.bots.experts.domain_valid`: New bot for checking domain's validity (PR#1966 by Marius Karotkis).

intelmq/bots/parsers/abusech/parser_ip.py

@@ -69,7 +69,7 @@ def parse(self, report: dict):
         for line in data_lines:
             yield line.strip()
 
-    def parse_line(self, line, report):
+    def parse_line(self, line: str, report):
         event = self.new_event(report)
         self.__process_defaults(event, line, report['feed.url'])
         self.__process_fields(event, line, report['feed.url'])
@@ -114,7 +114,7 @@ def __process_additional(event, line, feed_url):
     def __sanitize_csv_lines(s: str):
         return s.replace('"', '')
 
-    def recover_line(self, line):
+    def recover_line(self, line: str):
         return '\n'.join(self.comments + [self.header_line, line])
 
 

intelmq/bots/parsers/cymru/parser_full_bogons.py

@@ -26,7 +26,7 @@ def parse(self, report):
         for row in raw_report.splitlines():
             yield row.strip()
 
-    def parse_line(self, val, report):
+    def parse_line(self, val: str, report):
         if not len(val) or val.startswith('#') or val.startswith('//'):
             return
 

intelmq/bots/parsers/generic/parser_csv.py

@@ -28,6 +28,7 @@
 from intelmq.lib.bot import ParserBot
 from intelmq.lib.exceptions import InvalidArgument, InvalidValue
 from intelmq.lib.harmonization import DateTime
+from intelmq.lib.utils import RewindableFileHandle
 
 TIME_CONVERSIONS = {'timestamp': DateTime.from_timestamp,
                     'windows_nt': DateTime.from_windows_nt,
@@ -101,8 +102,10 @@ def parse(self, report):
         if self.skip_header:
             self.tempdata.append(raw_report[:raw_report.find('\n')])
             raw_report = raw_report[raw_report.find('\n') + 1:]
-        for row in csv.reader(io.StringIO(raw_report),
+        self._handle = RewindableFileHandle(io.StringIO(raw_report))
+        for row in csv.reader(self._handle,
                               delimiter=str(self.delimiter)):
+            self._current_line = self._handle.current_line
 
             if self.filter_text and self.filter_type:
                 text_in_row = self.filter_text in self.delimiter.join(row)
@@ -115,7 +118,7 @@ def parse(self, report):
             else:
                 yield row
 
-    def parse_line(self, row, report):
+    def parse_line(self, row: list, report):
         event = self.new_event(report)
 
         for keygroup, value, required in zip(self.columns, row, self.columns_required):

intelmq/bots/parsers/microsoft/parser_ctip.py

@@ -210,19 +210,23 @@ class MicrosoftCTIPParserBot(ParserBot):
     def parse(self, report):
         raw_report = utils.base64_decode(report.get("raw"))
         if raw_report.startswith('['):
+            # Interflow
             self.recover_line = self.recover_line_json
             yield from self.parse_json(report)
         elif raw_report.startswith('{'):
+            # Azure
             self.recover_line = self.recover_line_json_stream
             yield from self.parse_json_stream(report)
+        else:
+            raise ValueError("Can't parse the received message. It is neither a JSON list nor a JSON dictionary. Please report this bug.")
 
     def parse_line(self, line, report):
         if line.get('version', None) == 1.5:
             yield from self.parse_interflow(line, report)
         else:
             yield from self.parse_azure(line, report)
 
-    def parse_interflow(self, line, report):
+    def parse_interflow(self, line: dict, report):
         raw = self.recover_line(line)
         if line['indicatorthreattype'] != 'Botnet':
             raise ValueError('Unknown indicatorthreattype %r, only Botnet is supported.' % line['indicatorthreattype'])
@@ -257,7 +261,7 @@ def parse_interflow(self, line, report):
         yield event
 
     def parse_azure(self, line, report):
-        raw = self.recover_line(line)
+        raw = self.recover_line()
 
         event = self.new_event(report)
 

intelmq/bots/parsers/shadowserver/parser.py

@@ -197,7 +197,7 @@ def parse_line(self, row, report):
         # Now add additional constant fields.
         event.update(conf.get('constant_fields', {}))
 
-        event.add('raw', self.recover_line(row))
+        event.add('raw', self.recover_line())
 
         # Add everything which could not be resolved to extra.
         for f in fields:

intelmq/lib/bot.py

@@ -30,8 +30,6 @@
 from datetime import datetime, timedelta
 from typing import Any, List, Optional, Union
 
-import psutil
-
 import intelmq.lib.message as libmessage
 from intelmq import (DEFAULT_LOGGING_PATH,
                      HARMONIZATION_CONF_FILE,
@@ -942,7 +940,7 @@ class ParserBot(Bot):
     _csv_params = {}
     _ignore_lines_starting = []
     _handle = None
-    _current_line = None
+    _current_line: Optional[str] = None
 
     def __init__(self, bot_id: str, start: bool = False, sighup_event=None,
                  disable_multithreading: bool = None):
@@ -956,6 +954,7 @@ def __init__(self, bot_id: str, start: bool = False, sighup_event=None,
     def parse_csv(self, report: libmessage.Report):
         """
         A basic CSV parser.
+        The resulting lines are lists.
         """
         raw_report: str = utils.base64_decode(report.get("raw")).strip()
         raw_report = raw_report.translate({0: None})
@@ -971,6 +970,7 @@ def parse_csv(self, report: libmessage.Report):
     def parse_csv_dict(self, report: libmessage.Report):
         """
         A basic CSV Dictionary parser.
+        The resulting lines are dictionaries with the column names as keys.
         """
         raw_report: str = utils.base64_decode(report.get("raw")).strip()
         raw_report: str = raw_report.translate({0: None})
@@ -1024,6 +1024,7 @@ def parse(self, report: libmessage.Report):
         for line in utils.base64_decode(report.get("raw")).splitlines():
             line = line.strip()
             if not any([line.startswith(prefix) for prefix in self._ignore_lines_starting]):
+                self._current_line = line
                 yield line
 
     def parse_line(self, line: Any, report: libmessage.Report):
@@ -1063,14 +1064,14 @@ def process(self):
                     events: list[libmessage.Event] = [value]
             except Exception:
                 self.logger.exception('Failed to parse line.')
-                self.__failed.append((traceback.format_exc(), line))
+                self.__failed.append((traceback.format_exc(), self._current_line))
             else:
                 events_count += len(events)
                 self.send_message(*events)
 
-        for exc, line in self.__failed:
+        for exc, original_line in self.__failed:
             report_dump: libmessage.Message = report.copy()
-            report_dump.change('raw', self.recover_line(line))
+            report_dump.change('raw', self.recover_line(original_line))
             if self.error_dump_message:
                 self._dump_message(exc, report_dump)
             if self.destination_queues and '_on_error' in self.destination_queues:
@@ -1115,21 +1116,34 @@ def recover_line(self, line: Optional[str] = None) -> str:
         line = line if line else self._current_line
         return '\n'.join(tempdata + [line])
 
-    def recover_line_csv(self, line: str) -> str:
-        out = io.StringIO()
-        writer = csv.writer(out, **self._csv_params)
-        writer.writerow(line)
+    def recover_line_csv(self, line: Optional[list]) -> str:
+        """
+        Parameter:
+            line: Optional line as list. If absent, the current line is used as string.
+        """
+        if line:
+            out = io.StringIO()
+            writer = csv.writer(out, **self._csv_params)
+            writer.writerow(line)
+            result = out.getvalue()
+        else:
+            result = self._current_line
         tempdata = '\r\n'.join(self.tempdata) + '\r\n' if self.tempdata else ''
-        return tempdata + out.getvalue()
+        return tempdata + result
 
-    def recover_line_csv_dict(self, line: str) -> str:
+    def recover_line_csv_dict(self, line: Union[dict, str, None] = None) -> str:
         """
         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()
-        out.write(self._current_line)
+        if isinstance(line, dict):
+            writer.writerow(line)
+        elif isinstance(line, str):
+            out.write(line)
+        else:
+            out.write(self._current_line)
 
         return out.getvalue().strip()
 
@@ -1138,20 +1152,29 @@ def recover_line_json(self, line: dict) -> str:
         Reverse of parse for JSON pulses.
 
         Recovers a fully functional report with only the problematic pulse.
+        Using a string as input here is not possible, as the input may span over multiple lines.
+        Output is not identical to the input, but has the same content.
+
+        Parameters:
+            The line as dict.
+
+        Returns:
+            str: The JSON-encoded line as string.
         """
         return json.dumps([line])
 
-    def recover_line_json_stream(self, line=None) -> str:
+    def recover_line_json_stream(self, line: Optional[str] = None) -> str:
         """
-        recover_line for json streams, just returns the current line, unparsed.
+        recover_line for JSON streams (one JSON element per line, no outer structure),
+        just returns the current line, unparsed.
 
         Parameters:
-            line: None, not required, only for compatibility with other recover_line methods
+            line: The line itself as dict, if available, falls back to original current line
 
         Returns:
             str: unparsed JSON line.
         """
-        return self._current_line
+        return line if line else self._current_line
 
 
 class CollectorBot(Bot):

intelmq/lib/mixins/sql.py

@@ -1,10 +1,11 @@
 """ SQLMixin for IntelMQ
 
-SPDX-FileCopyrightText: 2021 Birger Schacht
+SPDX-FileCopyrightText: 2021 Birger Schacht, 2022 Intevation GmbH
 SPDX-License-Identifier: AGPL-3.0-or-later
 
 Based on the former SQLBot base class
 """
+from intelmq.lib import exceptions
 
 
 class SQLMixin:
@@ -18,36 +19,39 @@ class SQLMixin:
 
     POSTGRESQL = "postgresql"
     SQLITE = "sqlite"
-    default_engine = "postgresql"
+    _default_engine = "postgresql"
     engine = None
     # overwrite the default value from the OutputBot
     message_jsondict_as_string = True
 
     def __init__(self, *args, **kwargs):
+        self._init_sql()
+
+        super().__init__(*args, **kwargs)
+
+    def _init_sql(self):
         self.logger.debug("Running SQL Mixin initialization.")
-        self.engine_name = getattr(self, 'engine', self.default_engine).lower()
+        self._engine_name = getattr(self, 'engine', self._default_engine).lower()
         engines = {SQLMixin.POSTGRESQL: (self._init_postgresql, "%s"),
                    SQLMixin.SQLITE: (self._init_sqlite, "?")}
         for key, val in engines.items():
-            if self.engine_name == key:
+            if self._engine_name == key:
                 val[0]()
                 self.format_char = val[1]
                 break
         else:
-            raise ValueError(f"Wrong parameter 'engine' {self.engine_name!r}, possible values are {engines}")
-
-        super().__init__()
+            raise ValueError(f"Wrong parameter 'engine' {self._engine_name!r}, possible values are {engines}")
 
     def _connect(self, engine, connect_args: dict, autocommitable: bool = False):
-        self.engine = engine  # imported external library that connects to the DB
+        self._engine = engine  # imported external library that connects to the DB
         self.logger.debug(f"Connecting to database with connect_args: {connect_args}.")
 
         try:
-            self.con = self.engine.connect(**connect_args)
+            self.con = self._engine.connect(**connect_args)
             if autocommitable:  # psycopg2 has it, sqlite3 has not
                 self.con.autocommit = getattr(self, 'autocommit', True)  # True prevents deadlocks
             self.cur = self.con.cursor()
-        except (self.engine.Error, Exception):
+        except (self._engine.Error, Exception):
             self.logger.exception('Failed to connect to database.')
             self.stop()
         self.logger.info("Connected to database.")
@@ -88,24 +92,24 @@ def execute(self, query: str, values: tuple, rollback=False):
             # note: this assumes, the DB was created with UTF-8 support!
             self.cur.execute(query, values)
             self.logger.debug('Done.')
-        except (self.engine.InterfaceError, self.engine.InternalError,
-                self.engine.OperationalError, AttributeError):
+        except (self._engine.InterfaceError, self._engine.InternalError,
+                self._engine.OperationalError, AttributeError):
             if rollback:
                 try:
                     self.con.rollback()
                     self.logger.exception('Executed rollback command '
                                           'after failed query execution.')
-                except self.engine.OperationalError:
+                except self._engine.OperationalError:
                     self.logger.exception('Executed rollback command '
                                           'after failed query execution.')
-                    self.init()
+                    self._init_sql()
                 except Exception:
                     self.logger.exception('Cursor has been closed, connecting '
                                           'again.')
-                    self.init()
+                    self._init_sql()
             else:
                 self.logger.exception('Database connection problem, connecting again.')
-                self.init()
+                self._init_sql()
         else:
             return True
         return False

intelmq/lib/pipeline.py

@@ -29,7 +29,7 @@
 class PipelineFactory:
 
     @staticmethod
-    def create(logger, broker=None, direction=None, queues=None, pipeline_args=None, load_balance=False, is_multithreaded=False):
+    def create(logger, broker=None, direction=None, queues=None, pipeline_args: Optional[dict] = None, load_balance=False, is_multithreaded=False):
         """
         direction: "source" or "destination", optional, needed for queues
         queues: needs direction to be set, calls set_queues