Add support for client side certificate authentication and verifying

server certificate using provided CA bundle for message bus (RabbitMQ)
connections.

Option names are consistent with the same option names for MongoDB.

Update affected code so connection and URLs are only retrieved in a
single place.

Author: Kami

st2actions/st2actions/notifier/notifier.py

@@ -14,10 +14,10 @@
 # limitations under the License.
 
 from __future__ import absolute_import
+
 from datetime import datetime
 import json
 
-from kombu import Connection
 from oslo_config import cfg
 
 from st2common import log as logging
@@ -268,6 +268,6 @@ def _get_runner_ref(self, action_ref):
 
 
 def get_notifier():
-    with Connection(transport_utils.get_messaging_urls()) as conn:
+    with transport_utils.get_connection() as conn:
         return Notifier(conn, [NOTIFIER_ACTIONUPDATE_WORK_QUEUE],
                         trigger_dispatcher=TriggerDispatcher(LOG))

st2actions/st2actions/resultstracker/resultstracker.py

@@ -14,11 +14,11 @@
 # limitations under the License.
 
 from __future__ import absolute_import
+
 import eventlet
 import six
 
 from collections import defaultdict
-from kombu import Connection
 
 from st2common.query.base import QueryContext
 from st2common import log as logging
@@ -111,5 +111,5 @@ def get_querier(self, query_module_name):
 
 
 def get_tracker():
-    with Connection(transport_utils.get_messaging_urls()) as conn:
+    with transport_utils.get_connection() as conn:
         return ResultsTracker(conn, [RESULTSTRACKER_ACTIONSTATE_WORK_QUEUE])

st2actions/st2actions/scheduler/entrypoint.py

@@ -14,7 +14,6 @@
 # limitations under the License.
 
 from __future__ import absolute_import
-from kombu import Connection
 
 from st2common import log as logging
 from st2common.util import date
@@ -105,5 +104,5 @@ def _create_execution_queue_item_db_from_liveaction(self, liveaction, delay=None
 
 
 def get_scheduler_entrypoint():
-    with Connection(transport_utils.get_messaging_urls()) as conn:
+    with transport_utils.get_connection() as conn:
         return SchedulerEntrypoint(conn, [ACTIONSCHEDULER_REQUEST_QUEUE])

st2actions/st2actions/worker.py

@@ -17,8 +17,6 @@
 import sys
 import traceback
 
-from kombu import Connection
-
 from st2actions.container.base import RunnerContainer
 from st2common import log as logging
 from st2common.constants import action as action_constants
@@ -250,5 +248,5 @@ def _resume_action(self, liveaction_db):
 
 
 def get_worker():
-    with Connection(transport_utils.get_messaging_urls()) as conn:
+    with transport_utils.get_connection() as conn:
         return ActionExecutionDispatcher(conn, ACTIONRUNNER_QUEUES)

st2actions/tests/integration/test_action_state_consumer.py

@@ -20,8 +20,6 @@
 
 import mock
 
-from kombu import Connection
-
 from st2common.transport.queues import RESULTSTRACKER_ACTIONSTATE_WORK_QUEUE
 from st2actions.resultstracker.resultstracker import ResultsTracker
 from st2common.models.db.executionstate import ActionExecutionStateDB
@@ -63,7 +61,7 @@ def setUpClass(cls):
 
     @mock.patch.object(TestQuerier, 'query', mock.MagicMock(return_value=(False, {})))
     def test_process_message(self):
-        with Connection(transport_utils.get_messaging_urls()) as conn:
+        with transport_utils.get_connection() as conn:
             tracker = ResultsTracker(conn, [RESULTSTRACKER_ACTIONSTATE_WORK_QUEUE])
             tracker._bootstrap()
             state = ActionStateConsumerTests.get_state(

st2common/st2common/config.py

@@ -226,7 +226,28 @@ def register_opts(ignore_errors=False):
             help='How many times should we retry connection before failing.'),
         cfg.IntOpt(
             'connection_retry_wait', default=10000,
-            help='How long should we wait between connection retries.')
+            help='How long should we wait between connection retries.'),
+        cfg.BoolOpt(
+            'ssl', default=False,
+            help='Use SSL / TLS to connection to the messaging server. Same as '
+                 'appending "?ssl=true" at the end of the connection URL string.'),
+        cfg.StrOpt(
+            'ssl_keyfile', default=None,
+            help='Private keyfile used to identify the local connection against RabbitMQ.'),
+        cfg.StrOpt(
+            'ssl_certfile', default=None,
+            help='Certificate file used to identify the local connection (client).'),
+        cfg.StrOpt(
+            'ssl_cert_reqs', default=None, choices='none, optional, required',
+            help='Specifies whether a certificate is required from the other side of the '
+                 'connection, and whether it will be validated if provided.'),
+        cfg.StrOpt(
+            'ssl_ca_certs', default=None,
+            help='ca_certs file contains a set of concatenated CA certificates, which are '
+                 'used to validate certificates passed from RabbitMQ.'),
+        cfg.StrOpt(
+            'login_method', default=None,
+            help='Login method to use (AMQPLAIN, PLAIN, EXTERNAL, etc.).')
     ]
 
     do_register_opts(messaging_opts, 'messaging', ignore_errors)

st2common/st2common/persistence/execution.py

@@ -19,7 +19,6 @@
 from st2common.models.db.execution import ActionExecutionDB
 from st2common.models.db.execution import ActionExecutionOutputDB
 from st2common.persistence.base import Access
-from st2common.transport import utils as transport_utils
 
 __all__ = [
     'ActionExecution',
@@ -38,8 +37,7 @@ def _get_impl(cls):
     @classmethod
     def _get_publisher(cls):
         if not cls.publisher:
-            cls.publisher = transport.execution.ActionExecutionPublisher(
-                urls=transport_utils.get_messaging_urls())
+            cls.publisher = transport.execution.ActionExecutionPublisher()
         return cls.publisher
 
     @classmethod
@@ -57,8 +55,7 @@ def _get_impl(cls):
     @classmethod
     def _get_publisher(cls):
         if not cls.publisher:
-            cls.publisher = transport.execution.ActionExecutionOutputPublisher(
-                urls=transport_utils.get_messaging_urls())
+            cls.publisher = transport.execution.ActionExecutionOutputPublisher()
         return cls.publisher
 
     @classmethod

st2common/st2common/persistence/executionstate.py

@@ -14,10 +14,14 @@
 # limitations under the License.
 
 from __future__ import absolute_import
+
 from st2common import transport
 from st2common.models.db.executionstate import actionexecstate_access
 from st2common.persistence import base as persistence
-from st2common.transport import utils as transport_utils
+
+__all__ = [
+    'ActionExecutionState'
+]
 
 
 class ActionExecutionState(persistence.Access):
@@ -31,6 +35,5 @@ def _get_impl(cls):
     @classmethod
     def _get_publisher(cls):
         if not cls.publisher:
-            cls.publisher = transport.actionexecutionstate.ActionExecutionStatePublisher(
-                urls=transport_utils.get_messaging_urls())
+            cls.publisher = transport.actionexecutionstate.ActionExecutionStatePublisher()
         return cls.publisher

st2common/st2common/persistence/liveaction.py

@@ -14,10 +14,14 @@
 # limitations under the License.
 
 from __future__ import absolute_import
+
 from st2common import transport
 from st2common.models.db.liveaction import liveaction_access
 from st2common.persistence import base as persistence
-from st2common.transport import utils as transport_utils
+
+__all__ = [
+    'LiveAction'
+]
 
 
 class LiveAction(persistence.StatusBasedResource):
@@ -31,8 +35,7 @@ def _get_impl(cls):
     @classmethod
     def _get_publisher(cls):
         if not cls.publisher:
-            cls.publisher = transport.liveaction.LiveActionPublisher(
-                urls=transport_utils.get_messaging_urls())
+            cls.publisher = transport.liveaction.LiveActionPublisher()
         return cls.publisher
 
     @classmethod

st2common/st2common/persistence/sensor.py

@@ -14,10 +14,14 @@
 # limitations under the License.
 
 from __future__ import absolute_import
+
 from st2common import transport
 from st2common.models.db.sensor import sensor_type_access
 from st2common.persistence.base import ContentPackResource
-from st2common.transport import utils as transport_utils
+
+__all__ = [
+    'SensorType'
+]
 
 
 class SensorType(ContentPackResource):
@@ -31,6 +35,5 @@ def _get_impl(cls):
     @classmethod
     def _get_publisher(cls):
         if not cls.publisher:
-            cls.publisher = transport.reactor.SensorCUDPublisher(
-                urls=transport_utils.get_messaging_urls())
+            cls.publisher = transport.reactor.SensorCUDPublisher()
         return cls.publisher