Fix RQScheduler when run with SSL connection (#1383)

* Quick and dirty set up of SSL

* copy connection kwargs in scheduler

* fix

* chmod the cert

* Skip SSL tests in CI

Dockerfile

@@ -1,11 +1,19 @@
 FROM ubuntu:latest
 
-RUN apt-get update
-RUN apt-get install -y redis-server python3-pip
+RUN apt-get update \
+    && apt-get install -y \
+    redis-server \
+    python3-pip \
+    stunnel
+
+COPY tests/ssl_config/private.pem tests/ssl_config/stunnel.conf /etc/stunnel/
 
 COPY . /tmp/rq
 WORKDIR /tmp/rq
-RUN pip3 install -r /tmp/rq/requirements.txt -r /tmp/rq/dev-requirements.txt
-RUN python3 /tmp/rq/setup.py build && python3 /tmp/rq/setup.py install
+RUN pip3 install -r /tmp/rq/requirements.txt -r /tmp/rq/dev-requirements.txt \
+    && python3 /tmp/rq/setup.py build \
+    && python3 /tmp/rq/setup.py install
 
-CMD redis-server& RUN_SLOW_TESTS_TOO=1 pytest /tmp/rq/tests/ --durations=5 -v --log-cli-level 10
+CMD stunnel \
+    & redis-server \
+    & RUN_SLOW_TESTS_TOO=1 RUN_SSL_TESTS=1 pytest /tmp/rq/tests/ --durations=5 -v --log-cli-level 10

rq/scheduler.py

@@ -3,10 +3,11 @@ import os
 import signal
 import time
 import traceback
-
 from datetime import datetime
 from multiprocessing import Process
 
+from redis import Redis, SSLConnection
+
 from .defaults import DEFAULT_LOGGING_DATE_FORMAT, DEFAULT_LOGGING_FORMAT
 from .job import Job
 from .logutils import setup_loghandlers
@@ -14,8 +15,6 @@ from .queue import Queue
 from .registry import ScheduledJobRegistry
 from .utils import current_timestamp, enum
 
-from redis import Redis, SSLConnection
-
 SCHEDULER_KEY_TEMPLATE = 'rq:scheduler:%s'
 SCHEDULER_LOCKING_KEY_TEMPLATE = 'rq:scheduler-lock:%s'
 
@@ -39,7 +38,9 @@ class RQScheduler(object):
         self._acquired_locks = set()
         self._scheduled_job_registries = []
         self.lock_acquisition_time = None
-        self._connection_kwargs = connection.connection_pool.connection_kwargs
+        # Copy the connection kwargs before mutating them in order to not change the arguments
+        # used by the current connection pool to create new connections
+        self._connection_kwargs = connection.connection_pool.connection_kwargs.copy()
         # Redis does not accept parser_class argument which is sometimes present
         # on connection_pool kwargs, for example when hiredis is used
         self._connection_kwargs.pop('parser_class', None)
@@ -47,6 +48,7 @@ class RQScheduler(object):
         connection_class = connection.connection_pool.connection_class
         if issubclass(connection_class, SSLConnection):
             self._connection_kwargs['ssl'] = True
+
         self._connection = None
         self.interval = interval
         self._stop_requested = False

run_tests_in_docker.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
 
-docker build . -t rqtest && docker run --rm rqtest
+docker build . -t rqtest && docker run -it --rm rqtest

tests/__init__.py

@@ -3,9 +3,11 @@ from __future__ import (absolute_import, division, print_function,
                         unicode_literals)
 
 import logging
+import os
 
 from redis import Redis
 from rq import pop_connection, push_connection
+from rq.job import cancel_job
 
 try:
     import unittest
@@ -13,12 +15,17 @@ except ImportError:
     import unittest2 as unittest  # noqa
 
 
-def find_empty_redis_database():
+def find_empty_redis_database(ssl=False):
     """Tries to connect to a random Redis database (starting from 4), and
     will use/connect it when no keys are in there.
     """
     for dbnum in range(4, 17):
-        testconn = Redis(db=dbnum)
+        connection_kwargs = { 'db': dbnum }
+        if ssl:
+            connection_kwargs['port'] = 9736
+            connection_kwargs['ssl'] = True
+            connection_kwargs['ssl_cert_reqs'] = None # disable certificate validation
+        testconn = Redis(**connection_kwargs)
         empty = testconn.dbsize() == 0
         if empty:
             return testconn
@@ -26,16 +33,10 @@ def find_empty_redis_database():
 
 
 def slow(f):
-    import os
-    from functools import wraps
-
-    @wraps(f)
-    def _inner(*args, **kwargs):
-        if os.environ.get('RUN_SLOW_TESTS_TOO'):
-            f(*args, **kwargs)
-
-    return _inner
+    return unittest.skipUnless(os.environ.get('RUN_SLOW_TESTS_TOO'), "Slow tests disabled")(f)
 
+def ssl_test(f):
+    return unittest.skipUnless(os.environ.get('RUN_SSL_TESTS'), "SSL tests disabled")(f)
 
 class RQTestCase(unittest.TestCase):
     """Base class to inherit test cases from for RQ.

tests/ssl_config/private.pem

@@ -0,0 +1,85 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIJKwIBAAKCAgEAwN/TmlUJWSo8rWLAf94FUqWlFieMnitFbeOkpZsVI5ROdUVl
+NvvCF1h/o6+PTff6kRuRDWMdxQed22Pk40K79mGz8rjgNCRBJehPIUgi27BZZac3
+diae4aTgHsp6I0sw4+vT/4xbwfQoF+S2WdRfeoOV3odbFOKrxz2FKNb/p0I8/IbK
+Dgp/IpcX6z/LmYA0yD77eGxL9TzTW06hoLZByifKp0Q/MmQe6n4h4S1bG2dhAg5G
+2twa+B4+lh5j45/WA+OvWzCMkRjI8NuDidxFKdx+ddqqmJdXR6Aivi15oCDzJsvA
+eRHtFddgHa7+jj2+rx6+D8E9bkwiTQHS23rLWVnB0Fydm2a+G7PyXUGk+Ss+ekyT
++83HZfoPDN58k4ZPPG7xhOLYC5bDCNmRo0P4L4CkNj91KQYMdhpuX2LjOtYRR2B7
+fmOXAlWIkeo8rJ+i+hCepkXTRTPG0FOzRVnYQfN2IbCFwSizqqRDSO7wlOBs7Q1U
+bDzgQi2JmpxuUf+/7A6WSAJirxXgTVEhj9YaxKZuGXzx/1+AQ2Dzp1u4Dh0dygxD
+BghornbBr5KdXRyAC71jszRnFNdHZriijwvgmKV70Jz5WGNxennHcE45HEUiFbI6
+AZCJ+zqqlJfZGt5lWO1EPCALrBn5dKm8BzcYniIx1+AGC+mG7oy4NVePc9sCAwEA
+AQKCAgEAm6SDx6kTsCaLbIeiPA1YUkdlnykvKnxUvMbVGOa6+kk1vyDO+r3S9K/v
+4JFNnWedhfeu6BSx80ugMWi9Tj+OGtbhNd/G3YzcHdEH+h2SM6JtocB82xVzZTd9
+vJs8ULreqy6llzUW3r8+k3l3RapBmkYRbM/hykrYwCF/EWPeToT/XfEPoKEL00gG
+f0qt7CMvdOCOYbFS4oXBMY+UknJBSPcvbCeAsBNnd2dtw56sRML534TR3M992/fc
+HZxMk2VqeR0FZxsYdAaCMQuTbG6aSZurWUOqIxUN07kAEGP2ICg2z3ngylKS9esl
+nw6WUQa2l+7BBUm1XwqFK4trMr421W5hwdsUCt4iwgYjBdc/uJtOPsnF8wVol7I9
+YWooHfnSvztaIYq4qVNU8iCn6KYZ6s+2CMafto/gugQlTNGksUhP0cu70oh3t8bC
+oeNf8O9ZRfwZzhsSTScVWpNpJxTB19Ofm2o/yU0JUiiH4fGVSSlTzVmP6/9g2tqU
+iuTjcuM55sOtFmTIWDY3aeKvnGz2peQEgtfdxQa5nkRwt719SplsP3iyjJdArgE/
+x2xC162CwDVGCrq1H3JD9/fpZedC3CaYrXDMqI1vAsBcoKBbF3lNAxDnT+8tP2g5
+1pGuvaR3+UOUG6sd/8bHycPZU5ba9XcpqXTNG7JRAlji/bdunaECggEBAOzhi6a+
+Pmf6Ou6cAveEeGwki5d7qY+4Sw9q7eqOGE/4n3gF7ZQjbkdjSvE4D9Tk4soOcXv2
+1o4Hh+Cmgwp4U6BLGgBW94mTUVqXtVkD0HFLpORjSd4HLSu9NCJfCjWH1Gtc/IyM
+vq6zeSwLIFDm7TZe8hvrfN5sxI6FMsi5T87sXQS1GjlBTVSiIAm2m/q27Hmkrs7u
+wI22yYmVgnWy7LbReSfhweYzdBQSMItYL+aXQvRsLhHWm+rLzdu8nslZ1gBgiqrs
+8lly9SasM1d1E4vFvbtt1w4ZLTdetyq5FgWackgrj1dpHis116onxBa9lTRnAumw
+O4Dqr1JroTD6anMCggEBANBxAsl/LkhTIUW5biJ8DI6zavI38iZhcrYbDl8G+9i/
+JUj4tuSgq8YX3wdoMqkaDX8s6L7YFxYY7Dom4wBhDYrxqiih7RLyxu9UxLjx5TeO
+f9m9SBwCxa+i05M8gAEyK9jmd/k76yuAqDDeZGuy/rH/itP+BJpsC3QX+8chKIjh
+/lN3le1OM3TmE9OdGwFG7CxPelKeghd9ES1yvq7yyL7RpCLcwNkKer8X+PQISrUe
+Q77vmc94p+Zgdacmt2Eu3hgCOk+swtouTmp4W1k0oJTcOIeT+2OF2U2/mZA5B1us
+smhFvpxObh3RHaxG3R1ciK5xWHWyx78qooc/n1Id7vkCggEBAI+XfV8bbZr7/aNM
+oSPHgnQThybRiIyda6qx5/zKHATGMmzAMy8cdyoBD5m/oSEtiihvru01SQQZno1Y
+gpDjNdYyEFXqYe1chvFCi2SlQkKbVx427b0QXppn++Vl9TtT1jkqydCtNJ2UH7zK
+FdHU2jCeR2cTTcNK7a9zIMC6TJ2jfBNxcK8KXcUS7hbVQiItppVqdajs435EMlEb
+d1S/nGyJ+EZrvG09/Xx5NkIRuB+wy558wUSA8kzXNDeiVCK8OVRLMWPBdHsyi1bh
+BdJbHvkYahXm1HkwW893s9LLFYVaBTKobSDQkMAiyFPV/TDHxV1ZoFNmR/uyx4pP
+wgt9kO8CggEBAMN2NjbdnHkV+0125WBREzV96fvZmqmDGB7MoF1cHy7RkBUtpdQf
+FvVbzTkU7OzGEYIAiwDrgjqmhF7DuHrSh/CTTg1sSvRJ1WL5CsCjlV7TsfBtHwGl
+V9urxNt9EEwO0C9Fb5u4JH9W1mF9Ko4T++LOz1CcE5T7XIIxO1kwLuKtieCbc2xk
+uLwWROFbocdAypeCsCJpoXSFQ2ZrA4TrBnRqApDukaj1usUXpcyxOd091CloZcO4
+UTonmix0keIAISRCcovkZZRTeBU/Z+nu/+aX3CrHCiX5jhzqXwZvdAbzmxlMzcGl
+in1La5fxm8e8zi9G+rzkOYt6X46UisJmb4ECggEBAM2NtCiX85y0YswAx8GpZXz7
+8yM9qmR1RJwDA8mnsJYRpyohIbHiPvGGd67W/MyOe2j8EPlMraK9PG/Q9PfkChc0
+su5kjH/o2etgSYjykV0e3xKIuGb57gkQjgN6ZXTMBRxo+PqOp8BG/PkiTEbJErod
+K72zYfnvF1/YfrTHF+uGhF7rUl8Z66nNh1uZLURVE/O1+YRbJrFVi9hxdT+3FGv6
+ilq32bGCMopgFOee0CRS4IYJtYJufq+EgmXBt5l6yjr6A1OLUcNQ0tsT88VDgTQe
+rvaAxK/9DXs3J7gjgsu4Qc/I6oLg+KSCEOSEbZsaYuICas143lC1cLfThlxAYoM=
+-----END RSA PRIVATE KEY-----
+-----BEGIN CERTIFICATE-----
+MIIF0TCCA7mgAwIBAgIUH0n4JVFqZVeehn7EeRAkjWh0wrowDQYJKoZIhvcNAQEL
+BQAweDEfMB0GCSqGSIb3DQEJARYQdGVzdEBnZXRyZXNxLmNvbTEPMA0GA1UEAwwG
+cnEuY29tMQswCQYDVQQKDAJSUTEMMAoGA1UECwwDRW5nMQswCQYDVQQGEwJDQTEN
+MAsGA1UECAwEVGVzdDENMAsGA1UEBwwEVGVzdDAeFw0yMDExMjUxOTAzMzJaFw0y
+NTExMjUxOTAzMzJaMHgxHzAdBgkqhkiG9w0BCQEWEHRlc3RAZ2V0cmVzcS5jb20x
+DzANBgNVBAMMBnJxLmNvbTELMAkGA1UECgwCUlExDDAKBgNVBAsMA0VuZzELMAkG
+A1UEBhMCQ0ExDTALBgNVBAgMBFRlc3QxDTALBgNVBAcMBFRlc3QwggIiMA0GCSqG
+SIb3DQEBAQUAA4ICDwAwggIKAoICAQDA39OaVQlZKjytYsB/3gVSpaUWJ4yeK0Vt
+46SlmxUjlE51RWU2+8IXWH+jr49N9/qRG5ENYx3FB53bY+TjQrv2YbPyuOA0JEEl
+6E8hSCLbsFllpzd2Jp7hpOAeynojSzDj69P/jFvB9CgX5LZZ1F96g5Xeh1sU4qvH
+PYUo1v+nQjz8hsoOCn8ilxfrP8uZgDTIPvt4bEv1PNNbTqGgtkHKJ8qnRD8yZB7q
+fiHhLVsbZ2ECDkba3Br4Hj6WHmPjn9YD469bMIyRGMjw24OJ3EUp3H512qqYl1dH
+oCK+LXmgIPMmy8B5Ee0V12Adrv6OPb6vHr4PwT1uTCJNAdLbestZWcHQXJ2bZr4b
+s/JdQaT5Kz56TJP7zcdl+g8M3nyThk88bvGE4tgLlsMI2ZGjQ/gvgKQ2P3UpBgx2
+Gm5fYuM61hFHYHt+Y5cCVYiR6jysn6L6EJ6mRdNFM8bQU7NFWdhB83YhsIXBKLOq
+pENI7vCU4GztDVRsPOBCLYmanG5R/7/sDpZIAmKvFeBNUSGP1hrEpm4ZfPH/X4BD
+YPOnW7gOHR3KDEMGCGiudsGvkp1dHIALvWOzNGcU10dmuKKPC+CYpXvQnPlYY3F6
+ecdwTjkcRSIVsjoBkIn7OqqUl9ka3mVY7UQ8IAusGfl0qbwHNxieIjHX4AYL6Ybu
+jLg1V49z2wIDAQABo1MwUTAdBgNVHQ4EFgQUFBBOTl94RoNjXrxR9+idaPA6WMEw
+HwYDVR0jBBgwFoAUFBBOTl94RoNjXrxR9+idaPA6WMEwDwYDVR0TAQH/BAUwAwEB
+/zANBgkqhkiG9w0BAQsFAAOCAgEAltcc8+Vz+sLnoVrappVJ3iRa20T8J9XwrRt8
+zs7WiMORHIh3PIKJVSjd328HwdFBHUJEMc5Vgrwg8rVQYoxRoz2kFj9fMF0fYync
+ipjL+p4bLGdyWDEHIziJSLULkjgypsW3rRi4MdB8kV8r8zHWVz4enFrztnw8e2Qz
+i/7FIIxc5i07kttCY4+u8VVZWrzaNt3KUrDQ3yJiBODp1pIMcmCUgx6AG7vhi9Js
+v1y27GKRW88pIGSHPWDcko2X9JuJuNHdBPYBU2rJXkhA6bh36LUuSJ0ZY2tvHPUw
+NZWi2DoYb3xaevdUDHS25+LUhFullQRvuS/1r9l8sCRp17xZBUh0rtDJa+keoq3O
+EADybpmoRKOfNoZLMeJabo/VbQX9qNYVN3rgzCZ/yOdotEKOrr90tw/JSS4CTtMw
+athKFIHWQwqcL1/xTM3EQ/HpxA6d1qayozMPVj5NnfpYjaBK+PncBTN01u/O45Pw
++GGvvILPCsRYLIXp1lM5O3kbL9qffNLYHngQ/yW+R85AzMqbBIB9aaY3M0b4zdVo
+eIr8vDfTUh1bnzyKLiVWugOPVwfeU0ePg06Kr2yVPwtia4dW7YXm0dXHxn+7sMjg
+stJ4aqjlOiudLyb3wsRgnFDSzM5YZwtz3hCnbKhgDf5Qayywj/9VJWGpVbuQkmoq
+QQRVNAs=
+-----END CERTIFICATE-----

tests/ssl_config/stunnel.conf

@@ -0,0 +1,13 @@
+cert=/etc/stunnel/private.pem
+fips=no
+foreground=yes
+sslVersion=all
+socket=l:TCP_NODELAY=1
+socket=r:TCP_NODELAY=1
+pid=/var/run/stunnel.pid
+debug=0
+output=/etc/stunnel/stunnel.log
+
+[redis]
+accept = 0.0.0.0:9736
+connect = 127.0.0.1:6379

tests/test_scheduler.py

@@ -1,9 +1,8 @@
 import os
-import time
-
 from datetime import datetime, timedelta, timezone
 from multiprocessing import Process
 
+import mock
 from rq import Queue
 from rq.compat import PY2
 from rq.exceptions import NoSuchJobError
@@ -13,10 +12,9 @@ from rq.scheduler import RQScheduler
 from rq.utils import current_timestamp
 from rq.worker import Worker
 
-from .fixtures import kill_worker, say_hello
-from tests import RQTestCase
+from tests import RQTestCase, find_empty_redis_database, ssl_test
 
-import mock
+from .fixtures import kill_worker, say_hello
 
 
 class TestScheduledJobRegistry(RQTestCase):
@@ -75,20 +73,22 @@ class TestScheduledJobRegistry(RQTestCase):
         registry = ScheduledJobRegistry(queue=queue)
 
         from datetime import timezone
+
         # If we pass in a datetime with no timezone, `schedule()`
         # assumes local timezone so depending on your local timezone,
         # the timestamp maybe different
-
+        #
         # we need to account for the difference between a timezone
         # with DST active and without DST active.  The time.timezone
         # property isn't accurate when time.daylight is non-zero,
         # we'll test both.
-
+        #
         # first, time.daylight == 0 (not in DST).
         # mock the sitatuoin for American/New_York not in DST (UTC - 5)
         # time.timezone = 18000
         # time.daylight = 0
         # time.altzone = 14400
+
         mock_day = mock.patch('time.daylight', 0)
         mock_tz = mock.patch('time.timezone', 18000)
         mock_atz = mock.patch('time.altzone', 14400)
@@ -294,6 +294,21 @@ class TestWorker(RQTestCase):
         registry = FinishedJobRegistry(queue=queue)
         self.assertEqual(len(registry), 1)
 
+    @ssl_test
+    def test_work_with_ssl(self):
+        connection = find_empty_redis_database(ssl=True)
+        queue = Queue(connection=connection)
+        worker = Worker(queues=[queue], connection=connection)
+        p = Process(target=kill_worker, args=(os.getpid(), False, 5))
+
+        p.start()
+        queue.enqueue_at(datetime(2019, 1, 1, tzinfo=timezone.utc), say_hello)
+        worker.work(burst=False, with_scheduler=True)
+        p.join(1)
+        self.assertIsNotNone(worker.scheduler)
+        registry = FinishedJobRegistry(queue=queue)
+        self.assertEqual(len(registry), 1)
+
 
 class TestQueue(RQTestCase):