STOMP bots: several updates, fixes and improvements + some refactoring

The updates, fixes and improvements regard the *STOMP collector* and
*STOMP output* bots. Important changes are described below...

From now on, newer versions of the `stomp.py` package are supported --
including the latest (8.1.0).

Now both STOMP bots coerce the `port` configuration parameter to int --
so that a string representing an integer number is also acceptable (even
if not recommended) as a value of that parameter.

In the *STOMP output* bot, a bug has been fixed: `AttributeError` caused
by attempts to get unset attributes (`ssl_ca_cert` and companions...).

The *STOMP collector*'s reconnection mechanism has been fixed: from now
on, no reconnection attempts are made after `shutdown()`. Apart from
that, reconnection is not attempted at all for versions of `stomp.py`
older than 4.1.21 (as it did not work properly anyway).

Also regarding the *STOMP collector* bot, the following (undocumented
and unused) attributes of `StompCollectorBot` instances are no longer
set in `init()`: `ssl_ca_cert`, `ssl_cl_cert`, `ssl_cl_cert_key`.

Various checks have been improved/enhanced. Now, for example, both STOMP
bot classes implement the `check()` static/class method -- whose role is
to check ("statically", without the need to run the bot) configuration
parameters; in particular, it checks whether necessary certificate files
are accessible. When it comes to runtime (on-initialization) checks, one
notable improvement is that now also the *STOMP output* bot will raise a
`MissingDependencyError` if the `stomp.py` version is older than 4.1.8
(an analogous check has already been implemented by *STOMP collector*).

The code of those bot classes have also been significantly refactored
-- in particular, several common operations have been factored out and
placed in a new mix-in class: `intelmq.lib.mixins.StompMixin`; its
definition resides in a new module: `intelmq.lib.mixins.stomp`.

Author: zuo

intelmq/bots/collectors/stomp/collector.py

@@ -3,13 +3,13 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
 # -*- coding: utf-8 -*-
-import os.path
 
 from intelmq.lib.bot import CollectorBot
-from intelmq.lib.exceptions import MissingDependencyError
+from intelmq.lib.mixins import StompMixin
 
 try:
     import stomp
+    import stomp.exception
 except ImportError:
     stomp = None
 else:
@@ -18,9 +18,10 @@ class StompListener(stomp.PrintingListener):
         the stomp listener gets called asynchronously for
         every STOMP message
         """
-        def __init__(self, n6stompcollector, conn, destination):
+        def __init__(self, n6stompcollector, conn, destination, connect_kwargs=None):
             self.stompbot = n6stompcollector
             self.conn = conn
+            self.connect_kwargs = connect_kwargs
             self.destination = destination
             super().__init__()
             if stomp.__version__ >= (5, 0, 0):
@@ -29,15 +30,23 @@ def __init__(self, n6stompcollector, conn, destination):
 
         def on_heartbeat_timeout(self):
             self.stompbot.logger.info("Heartbeat timeout. Attempting to re-connect.")
-            connect_and_subscribe(self.conn, self.stompbot.logger, self.destination)
-
-        def on_error(self, headers, message):
-            self.stompbot.logger.error('Received an error: %r.', message)
-
-        def on_message(self, headers, message):
-            self.stompbot.logger.debug('Receive message %r...', message[:500])
+            if self.stompbot._auto_reconnect:
+                connect_and_subscribe(self.conn, self.stompbot.logger, self.destination,
+                                      connect_kwargs=self.connect_kwargs)
+
+        def on_error(self, frame, body=None):
+            if body is None:
+                # `stomp.py >= 6.1.0`
+                body = frame.body
+            self.stompbot.logger.error('Received an error: %r.', body)
+
+        def on_message(self, frame, body=None):
+            if body is None:
+                # `stomp.py >= 6.1.0`
+                body = frame.body
+            self.stompbot.logger.debug('Receive message %r...', body[:500])
             report = self.stompbot.new_report()
-            report.add("raw", message.rstrip())
+            report.add("raw", body.rstrip())
             report.add("feed.url", "stomp://" +
                        self.stompbot.server +
                        ":" + str(self.stompbot.port) +
@@ -46,19 +55,23 @@ def on_message(self, headers, message):
 
         def on_disconnected(self):
             self.stompbot.logger.debug('Detected disconnect')
-            connect_and_subscribe(self.conn, self.stompbot.logger, self.destination)
+            if self.stompbot._auto_reconnect:
+                connect_and_subscribe(self.conn, self.stompbot.logger, self.destination,
+                                      connect_kwargs=self.connect_kwargs)
 
 
-def connect_and_subscribe(conn, logger, destination, start=False):
+def connect_and_subscribe(conn, logger, destination, start=False, connect_kwargs=None):
     if start:
         conn.start()
-    conn.connect(wait=True)
+    if connect_kwargs is None:
+        connect_kwargs = dict(wait=True)
+    conn.connect(**connect_kwargs)
     conn.subscribe(destination=destination,
                    id=1, ack='auto')
     logger.info('Successfully connected and subscribed.')
 
 
-class StompCollectorBot(CollectorBot):
+class StompCollectorBot(CollectorBot, StompMixin):
     """Collect data from a STOMP Interface"""
     """ main class for the STOMP protocol collector """
     exchange: str = ''
@@ -73,36 +86,22 @@ class StompCollectorBot(CollectorBot):
     __conn = False  # define here so shutdown method can check for it
 
     def init(self):
-        if stomp is None:
-            raise MissingDependencyError("stomp")
-        elif stomp.__version__ < (4, 1, 8):
-            raise MissingDependencyError("stomp", version="4.1.8",
-                                         installed=stomp.__version__)
-
-        self.ssl_ca_cert = self.ssl_ca_certificate
-        self.ssl_cl_cert = self.ssl_client_certificate
-        self.ssl_cl_cert_key = self.ssl_client_certificate_key
-
-        # check if certificates exist
-        for f in [self.ssl_ca_cert, self.ssl_cl_cert, self.ssl_cl_cert_key]:
-            if not os.path.isfile(f):
-                raise ValueError("Could not open file %r." % f)
-
-        _host = [(self.server, self.port)]
-        self.__conn = stomp.Connection(host_and_ports=_host, use_ssl=True,
-                                       ssl_key_file=self.ssl_cl_cert_key,
-                                       ssl_cert_file=self.ssl_cl_cert,
-                                       ssl_ca_certs=self.ssl_ca_cert,
-                                       heartbeats=(self.heartbeat,
-                                                   self.heartbeat))
-
-        self.__conn.set_listener('', StompListener(self, self.__conn, self.exchange))
+        self.stomp_bot_runtime_initial_check()
+
+        # (note: older versions of `stomp.py` do not play well with reconnects)
+        self._auto_reconnect = (stomp.__version__ >= (4, 1, 21))
+
+        self.__conn, connect_kwargs = self.prepare_stomp_connection()
+        self.__conn.set_listener('', StompListener(self, self.__conn, self.exchange,
+                                                   connect_kwargs=connect_kwargs))
         connect_and_subscribe(self.__conn, self.logger, self.exchange,
-                              start=stomp.__version__ < (4, 1, 20))
+                              start=stomp.__version__ < (4, 1, 20),
+                              connect_kwargs=connect_kwargs)
 
     def shutdown(self):
         if not stomp or not self.__conn:
             return
+        self._auto_reconnect = False
         try:
             self.__conn.disconnect()
         except stomp.exception.NotConnectedException:
@@ -111,5 +110,9 @@ def shutdown(self):
     def process(self):
         pass
 
+    @classmethod
+    def check(cls, parameters):
+        return cls.stomp_bot_parameters_check(parameters) or None
+
 
 BOT = StompCollectorBot

intelmq/bots/outputs/stomp/output.py

@@ -3,19 +3,17 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
 # -*- coding: utf-8 -*-
-import os.path
 
 from intelmq.lib.bot import OutputBot
-from intelmq.lib.exceptions import MissingDependencyError
-
+from intelmq.lib.mixins import StompMixin
 
 try:
     import stomp
 except ImportError:
     stomp = None
 
 
-class StompOutputBot(OutputBot):
+class StompOutputBot(OutputBot, StompMixin):
     """Send events to a STMOP server"""
     """ main class for the STOMP protocol output bot """
     exchange: str = "/exchange/_push"
@@ -35,29 +33,18 @@ class StompOutputBot(OutputBot):
     _conn = None
 
     def init(self):
-        if stomp is None:
-            raise MissingDependencyError("stomp")
-
-        # check if certificates exist
-        for f in [self.ssl_ca_cert, self.ssl_cl_cert, self.ssl_cl_cert_key]:
-            if not os.path.isfile(f):
-                raise ValueError("Could not open SSL (certificate) file '%s'." % f)
-
-        _host = [(self.server, self.port)]
-        self._conn = stomp.Connection(host_and_ports=_host, use_ssl=True,
-                                      ssl_key_file=self.ssl_cl_cert_key,
-                                      ssl_cert_file=self.ssl_cl_cert,
-                                      ssl_ca_certs=self.ssl_ca_cert,
-                                      heartbeats=(self.heartbeat,
-                                                  self.heartbeat))
+        self.stomp_bot_runtime_initial_check()
+        (self._conn,
+         self._connect_kwargs) = self.prepare_stomp_connection()
         self.connect()
 
     def connect(self):
         self.logger.debug('Connecting.')
         # based on the documentation at:
         # https://github.com/jasonrbriggs/stomp.py/wiki/Simple-Example
-        self._conn.start()
-        self._conn.connect(wait=True)
+        if stomp.__version__ < (4, 1, 20):
+            self._conn.start()
+        self._conn.connect(**self._connect_kwargs)
         self.logger.debug('Connected.')
 
     def shutdown(self):
@@ -73,5 +60,9 @@ def process(self):
                         destination=self.exchange)
         self.acknowledge_message()
 
+    @classmethod
+    def check(cls, parameters):
+        return cls.stomp_bot_parameters_check(parameters) or None
+
 
 BOT = StompOutputBot

intelmq/lib/mixins/__init__.py

@@ -5,5 +5,6 @@
 from intelmq.lib.mixins.http import HttpMixin
 from intelmq.lib.mixins.cache import CacheMixin
 from intelmq.lib.mixins.sql import SQLMixin
+from intelmq.lib.mixins.stomp import StompMixin
 
-__all__ = ['HttpMixin', 'CacheMixin', 'SQLMixin']
+__all__ = ['HttpMixin', 'CacheMixin', 'SQLMixin', 'StompMixin']

intelmq/lib/mixins/stomp.py

@@ -0,0 +1,145 @@
+""" StompMixin for IntelMQ
+
+SPDX-FileCopyrightText: 2017 Sebastian Wagner, 2023 NASK
+SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+
+from typing import (
+    Any,
+    Callable,
+    List,
+    NoReturn,
+    Tuple,
+)
+
+try:
+    import stomp
+except ImportError:
+    stomp = None
+
+from intelmq.lib.exceptions import MissingDependencyError
+
+
+class StompMixin:
+
+    """A mixin that provides certain common methods for STOMP bots."""
+
+    #
+    # STOMP bot attributes relevant to this mixin
+
+    server: str
+    port: int
+    heartbeat: int
+
+    ssl_ca_certificate: str  # TODO: could be pathlib.Path
+    ssl_client_certificate: str  # TODO: could be pathlib.Path
+    ssl_client_certificate_key: str  # TODO: could be patlib.Path
+
+    #
+    # Helper methods intended to be used in subclasses
+
+    @classmethod
+    def stomp_bot_parameters_check(cls, parameters: dict) -> List[List[str]]:
+        """Intended to be used in bots' `check()` static/class method."""
+        logs = []
+        cls.__verify_parameters(
+            get_param=parameters.get,
+            on_error=lambda msg: logs.append(['error', msg]),
+        )
+        return logs
+
+    def stomp_bot_runtime_initial_check(self) -> None:
+        """Intended to be used in bots' `init()` instance method."""
+        self.__verify_dependency()
+        self.__verify_parameters(
+            get_param=self.__get_own_attribute,
+            on_error=self.__raise_value_error,
+        )
+
+    def prepare_stomp_connection(self) -> Tuple['stomp.Connection', dict]:
+        """
+        Get a `(<STOMP connection>, <STOMP connect arguments>)` pair.
+
+        * `<STOMP connection>` is a new instance of `stomp.Connection`,
+           with the SSL stuff *already configured*, but *without* any
+           invocations of `connect()` made yet;
+
+        * `<STOMP connect arguments>` is a dict of arguments -- ready
+          to be passed to the `connect()` method of the aforementioned
+          `<STOMP connection>` object.
+        """
+        ssl_kwargs, connect_kwargs = self.__get_ssl_and_connect_kwargs()
+        host_and_ports = [(self.server, int(self.port))]
+        stomp_connection = stomp.Connection(host_and_ports=host_and_ports,
+                                            heartbeats=(self.heartbeat,
+                                                        self.heartbeat))
+        stomp_connection.set_ssl(host_and_ports, **ssl_kwargs)
+        return stomp_connection, connect_kwargs
+
+    #
+    # Implementation details
+
+    _DEPENDENCY_NAME_REMARK = (
+        "Note that the actual name of the pip-installable "
+        "distribution package is 'stomp.py', not 'stomp'.")
+
+    @classmethod
+    def __verify_dependency(cls) -> None:
+        # Note: the pip-installable package's name is 'stomp.py', but
+        # e.g. the apt-installable package's name is 'python3-stomp' (or
+        # similar) -- that's why we pass to the `MissingDependencyError`
+        # constructor the name 'stomp', but also pass the value of the
+        # `_DEPENDENCY_NAME_REMARK` constant as `additional_text`...
+        if stomp is None:
+            raise MissingDependencyError('stomp',
+                                         additional_text=cls._DEPENDENCY_NAME_REMARK)
+        if stomp.__version__ < (4, 1, 8):
+            raise MissingDependencyError('stomp', version="4.1.8",
+                                         installed=stomp.__version__,
+                                         additional_text=cls._DEPENDENCY_NAME_REMARK)
+
+    @classmethod
+    def __verify_parameters(cls,
+                            get_param: Callable[[str], Any],
+                            on_error: Callable[[str], None]) -> None:
+        for param_name in [
+            'ssl_ca_certificate',
+            'ssl_client_certificate',
+            'ssl_client_certificate_key',
+        ]:
+            cls.__verify_file_param(param_name, get_param, on_error)
+
+    @classmethod
+    def __verify_file_param(cls,
+                            param_name: str,
+                            get_param: Callable[[str], Any],
+                            on_error: Callable[[str], None]) -> None:
+        path = get_param(param_name)
+        if path is None:
+            on_error(f"Parameter {param_name!r} is not given "
+                     f"(or is set to None).")
+            return
+        try:
+            open(path, 'rb').close()
+        except OSError as exc:
+            # (note: the filename is mentioned in the included exc message)
+            on_error(f"Cannot open file specified as parameter "
+                     f"{param_name!r} ({str(exc)!r}).")
+
+    def __get_own_attribute(self, param_name: str) -> Any:
+        return getattr(self, param_name, None)
+
+    def __raise_value_error(self, msg: str) -> NoReturn:
+        raise ValueError(msg)
+
+    def __get_ssl_and_connect_kwargs(self) -> Tuple[dict, dict]:
+        # Note: the `ca_certs` argument to `set_ssl()` must always be
+        # provided, otherwise the `stomp.py`'s machinery would *not*
+        # perform any certificate verification!
+        ssl_kwargs = dict(
+            ca_certs=self.ssl_ca_certificate,
+            cert_file=self.ssl_client_certificate,
+            key_file=self.ssl_client_certificate_key,
+        )
+        connect_kwargs = dict(wait=True)
+        return ssl_kwargs, connect_kwargs