result_backend setting supports rediss:// protocol URLs (#4696)

* Implements support for rediss protocol URLs in setting result_backend

* Adds documentation for rediss:// urls in result_broker setting

* Adds tests for rediss:// urls in result_backend

* Urllib import compatible with Python 3

* Adds moar tests

* Happify lint.

* Fixes test input for rediss URLs

celery/app/backends.py

@@ -21,6 +21,7 @@ BACKEND_ALIASES = {
     'rpc': 'celery.backends.rpc.RPCBackend',
     'cache': 'celery.backends.cache:CacheBackend',
     'redis': 'celery.backends.redis:RedisBackend',
+    'rediss': 'celery.backends.redis:RedisBackend',
     'sentinel': 'celery.backends.redis:SentinelBackend',
     'mongodb': 'celery.backends.mongodb:MongoBackend',
     'db': 'celery.backends.database:DatabaseBackend',

celery/backends/redis.py

@@ -3,6 +3,7 @@
 from __future__ import absolute_import, unicode_literals
 
 from functools import partial
+from ssl import CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
 
 from kombu.utils.functional import retry_over_time
 from kombu.utils.objects import cached_property
@@ -20,6 +21,12 @@ from celery.utils.time import humanize_seconds
 
 from . import async, base
 
+try:
+    from urllib.parse import unquote
+except ImportError:
+    # Python 2
+    from urlparse import unquote
+
 try:
     import redis
     from kombu.transport.redis import get_redis_error_classes
@@ -44,6 +51,23 @@ You need to install the redis library with support of \
 sentinel in order to use the Redis result store backend.
 """
 
+W_REDIS_SSL_CERT_OPTIONAL = """
+Setting ssl_cert_reqs=CERT_OPTIONAL when connecting to redis means that \
+celery might not valdate the identity of the redis broker when connecting. \
+This leaves you vulnerable to man in the middle attacks.
+"""
+
+W_REDIS_SSL_CERT_NONE = """
+Setting ssl_cert_reqs=CERT_NONE when connecting to redis means that celery \
+will not valdate the identity of the redis broker when connecting. This \
+leaves you vulnerable to man in the middle attacks.
+"""
+
+E_REDIS_SSL_CERT_REQS_MISSING = """
+A rediss:// URL must have parameter ssl_cert_reqs be CERT_REQUIRED, \
+CERT_OPTIONAL, or CERT_NONE
+"""
+
 E_LOST = 'Connection to Redis lost: Retry (%s/%s) %s.'
 
 logger = get_logger(__name__)
@@ -197,6 +221,26 @@ class RedisBackend(base.BaseKeyValueStoreBackend, async.AsyncBackendMixin):
         else:
             connparams['db'] = path
 
+        if scheme == 'rediss':
+            connparams['connection_class'] = redis.SSLConnection
+            # The following parameters, if present in the URL, are encoded. We
+            # must add the decoded values to connparams.
+            for ssl_setting in ['ssl_ca_certs', 'ssl_certfile', 'ssl_keyfile']:
+                ssl_val = query.pop(ssl_setting, None)
+                if ssl_val:
+                    connparams[ssl_setting] = unquote(ssl_val)
+            ssl_cert_reqs = query.pop('ssl_cert_reqs', 'MISSING')
+            if ssl_cert_reqs == 'CERT_REQUIRED':
+                connparams['ssl_cert_reqs'] = CERT_REQUIRED
+            elif ssl_cert_reqs == 'CERT_OPTIONAL':
+                logger.warn(W_REDIS_SSL_CERT_OPTIONAL)
+                connparams['ssl_cert_reqs'] = CERT_OPTIONAL
+            elif ssl_cert_reqs == 'CERT_NONE':
+                logger.warn(W_REDIS_SSL_CERT_NONE)
+                connparams['ssl_cert_reqs'] = CERT_NONE
+            else:
+                raise ValueError(E_REDIS_SSL_CERT_REQS_MISSING)
+
         # db may be string and start with / like in kombu.
         db = connparams.get('db') or 0
         db = db.strip('/') if isinstance(db, string_t) else db

docs/userguide/configuration.rst

@@ -875,10 +875,13 @@ Configuring the backend URL
     requirements.
 
 This backend requires the :setting:`result_backend`
-setting to be set to a Redis URL::
+setting to be set to a Redis or `Redis over TLS`_ URL::
 
     result_backend = 'redis://:password@host:port/db'
 
+.. _`Redis over TLS`:
+    https://www.iana.org/assignments/uri-schemes/prov/rediss
+
 For example::
 
     result_backend = 'redis://localhost/0'
@@ -887,6 +890,10 @@ is the same as::
 
     result_backend = 'redis://'
 
+Use the ``rediss://`` protocol to connect to redis over TLS::
+
+    result_backend = 'rediss://:password@host:port/db?ssl_cert_reqs=CERT_REQUIRED'
+
 The fields of the URL are defined as follows:
 
 #. ``password``
@@ -906,6 +913,17 @@ The fields of the URL are defined as follows:
     Database number to use. Default is 0.
     The db can include an optional leading slash.
 
+When using a TLS connection (protocol is ``rediss://``), you may pass in all values in :setting:`broker_use_ssl` as query parameters. Paths to certificates must be URL encoded, and ``ssl_cert_reqs`` is required. Example:
+
+.. code-block:: python
+
+    result_backend = 'rediss://:password@host:port/db?\
+        ssl_cert_reqs=CERT_REQUIRED\
+        &ssl_ca_certs=%2Fvar%2Fssl%2Fmyca.pem\                  # /var/ssl/myca.pem
+        &ssl_certfile=%2Fvar%2Fssl%2Fredis-server-cert.pem\     # /var/ssl/redis-server-cert.pem
+        &ssl_keyfile=%2Fvar%2Fssl%2Fprivate%2Fworker-key.pem'   # /var/ssl/private/worker-key.pem
+
+
 .. setting:: redis_backend_use_ssl
 
 ``redis_backend_use_ssl``

t/unit/backends/test_redis.py

@@ -270,6 +270,74 @@ class test_RedisBackend:
         from redis.connection import SSLConnection
         assert x.connparams['connection_class'] is SSLConnection
 
+    @skip.unless_module('redis')
+    def test_backend_ssl_url(self):
+        self.app.conf.redis_socket_timeout = 30.0
+        self.app.conf.redis_socket_connect_timeout = 100.0
+        x = self.Backend(
+            'rediss://:bosco@vandelay.com:123//1?ssl_cert_reqs=CERT_REQUIRED',
+            app=self.app,
+        )
+        assert x.connparams
+        assert x.connparams['host'] == 'vandelay.com'
+        assert x.connparams['db'] == 1
+        assert x.connparams['port'] == 123
+        assert x.connparams['password'] == 'bosco'
+        assert x.connparams['socket_timeout'] == 30.0
+        assert x.connparams['socket_connect_timeout'] == 100.0
+        assert x.connparams['ssl_cert_reqs'] == ssl.CERT_REQUIRED
+
+        from redis.connection import SSLConnection
+        assert x.connparams['connection_class'] is SSLConnection
+
+    @skip.unless_module('redis')
+    def test_backend_ssl_url_options(self):
+        x = self.Backend(
+            (
+                'rediss://:bosco@vandelay.com:123//1?ssl_cert_reqs=CERT_NONE'
+                '&ssl_ca_certs=%2Fvar%2Fssl%2Fmyca.pem'
+                '&ssl_certfile=%2Fvar%2Fssl%2Fredis-server-cert.pem'
+                '&ssl_keyfile=%2Fvar%2Fssl%2Fprivate%2Fworker-key.pem'
+            ),
+            app=self.app,
+        )
+        assert x.connparams
+        assert x.connparams['host'] == 'vandelay.com'
+        assert x.connparams['db'] == 1
+        assert x.connparams['port'] == 123
+        assert x.connparams['password'] == 'bosco'
+        assert x.connparams['ssl_cert_reqs'] == ssl.CERT_NONE
+        assert x.connparams['ssl_ca_certs'] == '/var/ssl/myca.pem'
+        assert x.connparams['ssl_certfile'] == '/var/ssl/redis-server-cert.pem'
+        assert x.connparams['ssl_keyfile'] == '/var/ssl/private/worker-key.pem'
+
+    @skip.unless_module('redis')
+    def test_backend_ssl_url_cert_none(self):
+        x = self.Backend(
+            'rediss://:bosco@vandelay.com:123//1?ssl_cert_reqs=CERT_OPTIONAL',
+            app=self.app,
+        )
+        assert x.connparams
+        assert x.connparams['host'] == 'vandelay.com'
+        assert x.connparams['db'] == 1
+        assert x.connparams['port'] == 123
+        assert x.connparams['ssl_cert_reqs'] == ssl.CERT_OPTIONAL
+
+        from redis.connection import SSLConnection
+        assert x.connparams['connection_class'] is SSLConnection
+
+    @skip.unless_module('redis')
+    @pytest.mark.parametrize("uri", [
+        'rediss://:bosco@vandelay.com:123//1?ssl_cert_reqs=CERT_KITTY_CATS',
+        'rediss://:bosco@vandelay.com:123//1'
+    ])
+    def test_backend_ssl_url_invalid(self, uri):
+        with pytest.raises(ValueError):
+            self.Backend(
+                uri,
+                app=self.app,
+            )
+
     def test_compat_propertie(self):
         x = self.Backend(
             'redis://:bosco@vandelay.com:123//1', app=self.app,