Merge b29c888 into f96b50b

CHANGELOG.rst

@@ -12,6 +12,8 @@ This project adheres to `Semantic Versioning`_ starting with version 1.0.
 
 Added
 -----
+- port of 1.2.10 (support for RabbitMQ TLS authentication and ``port`` key in
+  event broker endpoint config)
 
 Fixed
 -----
@@ -22,7 +24,6 @@ Changed
 Removed
 -------
 
-
 [1.3.8] - 2019-10-08
 ^^^^^^^^^^^^^^^^^^^^
 
@@ -218,6 +219,18 @@ Removed
 -------
 - Removed ``--report`` argument from ``rasa test nlu``. All output files are stored in the ``--out`` directory.
 
+[1.2.10] - 2019-09-17
+^^^^^^^^^^^^^^^^^^^^
+
+- Added support for RabbitMQ TLS authentication. The following environment variables
+  need to be set:
+  ``RABBITMQ_SSL_CLIENT_CERTIFICATE`` - path to the SSL client certificate (required)
+  ``RABBITMQ_SSL_CLIENT_KEY`` - path to the SSL client key (required)
+  ``RABBITMQ_SSL_CA_FILE`` - path to the SSL CA file (optional, for certificate
+  verification)
+  ``RABBITMQ_SSL_KEY_PASSWORD`` - SSL private key password (optional)
+- Added ability to define the RabbitMQ port using the ``port`` key in the
+  ``event_broker`` endpoint config.
 
 [1.2.9] - 2019-09-17
 ^^^^^^^^^^^^^^^^^^^^

rasa/core/brokers/pika.py

@@ -5,6 +5,7 @@
 
 import time
 
+import rasa.core.brokers.utils as rasa_broker_utils
 from rasa.core.brokers.event_channel import EventChannel
 from rasa.utils.endpoints import EndpointConfig
 
@@ -18,6 +19,7 @@ def initialise_pika_connection(
     host: Text,
     username: Text,
     password: Text,
+    port: Union[Text, int] = 5672,
     connection_attempts: int = 20,
     retry_delay_in_seconds: Union[int, float] = 5,
 ) -> "BlockingConnection":
@@ -27,11 +29,13 @@ def initialise_pika_connection(
         host: Pika host
         username: username for authentication with Pika host
         password: password for authentication with Pika host
+        port: port of the Pika host
         connection_attempts: number of channel attempts before giving up
         retry_delay_in_seconds: delay in seconds between channel attempts
 
     Returns:
         Pika `BlockingConnection` with provided parameters
+
     """
 
     import pika
@@ -47,12 +51,14 @@ def initialise_pika_connection(
         # host seems to be just the host, so we use our parameters
         parameters = pika.ConnectionParameters(
             host,
+            port=port,
             credentials=pika.PlainCredentials(username, password),
             connection_attempts=connection_attempts,
             # Wait between retries since
             # it can take some time until
             # RabbitMQ comes up.
             retry_delay=retry_delay_in_seconds,
+            ssl_options=rasa_broker_utils.create_rabbitmq_ssl_options(host),
         )
     return pika.BlockingConnection(parameters)
 
@@ -62,6 +68,7 @@ def initialise_pika_channel(
     queue: Text,
     username: Text,
     password: Text,
+    port: Union[Text, int] = 5672,
     connection_attempts: int = 20,
     retry_delay_in_seconds: Union[int, float] = 5,
 ) -> "BlockingChannel":
@@ -72,15 +79,17 @@ def initialise_pika_channel(
         queue: Pika queue to declare
         username: username for authentication with Pika host
         password: password for authentication with Pika host
+        port: port of the Pika host
         connection_attempts: number of channel attempts before giving up
         retry_delay_in_seconds: delay in seconds between channel attempts
 
     Returns:
         Pika `BlockingChannel` with declared queue
+
     """
 
     connection = initialise_pika_connection(
-        host, username, password, connection_attempts, retry_delay_in_seconds
+        host, username, password, port, connection_attempts, retry_delay_in_seconds
     )
 
     return _declare_pika_channel_with_queue(connection, queue)
@@ -127,6 +136,7 @@ def __init__(
         host: Text,
         username: Text,
         password: Text,
+        port: Union[int, Text] = 5672,
         queue: Text = "rasa_core_events",
         loglevel: Union[Text, int] = logging.WARNING,
     ):
@@ -136,6 +146,7 @@ def __init__(
         self.host = host
         self.username = username
         self.password = password
+        self.port = port
         self.channel = None  # delay opening channel until first event
 
     def __del__(self) -> None:
@@ -153,6 +164,7 @@ def _open_channel(
             self.queue,
             self.username,
             self.password,
+            self.port,
             connection_attempts,
             retry_delay_in_seconds,
         )

rasa/core/brokers/utils.py

@@ -1,12 +1,14 @@
 import logging
+import os
 import typing
-from typing import Optional
+from typing import Optional, Text
 
 import rasa.utils.common as rasa_utils
 from rasa.utils.endpoints import EndpointConfig
 
 if typing.TYPE_CHECKING:
     from rasa.core.brokers.event_channel import EventChannel
+    import pika
 
 logger = logging.getLogger(__name__)
 
@@ -52,3 +54,50 @@ def load_event_channel_from_module_string(
             "Not using any event channel. Error: {}".format(broker_config.type, e)
         )
         return None
+
+
+def create_rabbitmq_ssl_options(
+    rabbitmq_host: Optional[Text] = None
+) -> Optional["pika.SSLOptions"]:
+    """Create RabbitMQ SSL options.
+
+    Requires the following environment variables to be set:
+
+        RABBITMQ_SSL_CLIENT_CERTIFICATE - path to the SSL client certificate (required)
+        RABBITMQ_SSL_CLIENT_KEY - path to the SSL client key (required)
+        RABBITMQ_SSL_CA_FILE - path to the SSL CA file for verification (optional)
+        RABBITMQ_SSL_KEY_PASSWORD - SSL private key password (optional)
+
+    Details on how to enable RabbitMQ TLS support can be found here:
+    https://www.rabbitmq.com/ssl.html#enabling-tls
+
+    Args:
+        rabbitmq_host: RabbitMQ hostname
+
+    Returns:
+        Pika SSL context of type `pika.SSLOptions` if
+        the RABBITMQ_SSL_CLIENT_CERTIFICATE and RABBITMQ_SSL_CLIENT_KEY
+        environment variables are valid paths, else `None`.
+
+    """
+
+    client_certificate_path = os.environ.get("RABBITMQ_SSL_CLIENT_CERTIFICATE")
+    client_key_path = os.environ.get("RABBITMQ_SSL_CLIENT_KEY")
+
+    if client_certificate_path and client_key_path:
+        import pika
+        import rasa.server
+
+        logger.debug(
+            "Configuring SSL context for RabbitMQ host '{}'.".format(rabbitmq_host)
+        )
+
+        ca_file_path = os.environ.get("RABBITMQ_SSL_CA_FILE")
+        key_password = os.environ.get("RABBITMQ_SSL_KEY_PASSWORD")
+
+        ssl_context = rasa.server.create_ssl_context(
+            client_certificate_path, client_key_path, ca_file_path, key_password
+        )
+        return pika.SSLOptions(ssl_context, rabbitmq_host)
+    else:
+        return None

rasa/server.py

@@ -1,12 +1,15 @@
 import logging
+import multiprocessing
 import os
 import tempfile
 import traceback
-import multiprocessing
+import typing
 from functools import wraps, reduce
 from inspect import isawaitable
 from typing import Any, Callable, List, Optional, Text, Union
-from ssl import SSLContext
+
+if typing.TYPE_CHECKING:
+    from ssl import SSLContext
 
 from sanic import Sanic, response
 from sanic.request import Request
@@ -226,14 +229,28 @@ async def authenticate(request: Request):
 def create_ssl_context(
     ssl_certificate: Optional[Text],
     ssl_keyfile: Optional[Text],
-    ssl_password: Optional[Text],
-) -> Optional[SSLContext]:
-    """Create a SSL context (for the sanic server) if a proper certificate is passed."""
+    ssl_ca_file: Optional[Text] = None,
+    ssl_password: Optional[Text] = None,
+) -> Optional["SSLContext"]:
+    """Create an SSL context if a proper certificate is passed.
+
+    Args:
+        ssl_certificate: path to the SSL client certificate
+        ssl_keyfile: path to the SSL key file
+        ssl_ca_file: path to the SSL CA file for verification (optional)
+        ssl_password: SSL private key password (optional)
+
+    Returns:
+        SSL context if a valid certificate chain can be loaded, `None` otherwise.
+
+    """
 
     if ssl_certificate:
         import ssl
 
-        ssl_context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
+        ssl_context = ssl.create_default_context(
+            purpose=ssl.Purpose.CLIENT_AUTH, cafile=ssl_ca_file
+        )
         ssl_context.load_cert_chain(
             ssl_certificate, keyfile=ssl_keyfile, password=ssl_password
         )