5.0.3 Regression with SSL handshake

[tomwojcik]

If I'm not mistaken, kombu is using py-ampq if librabbitmq is not available.
So celery is using kombu which is using py-amqp by default.
No matter what celery version is being used, if py-amqp has no version specified in requirements.txt, the latest one is used.

When I connect to rabbitmq using pika, everything seems to be working fine.

parameters = pika.URLParameters('***')
connection = pika.BlockingConnection(parameters)
connection.is_open
connection.close()

But after installing Celery 5.0.1 or latest commit hash from master, when I do the same with kombu

from kombu import Connection

conn = Connection('***')
conn.connect()

I get

---------------------------------------------------------------------------
SSLCertVerificationError                  Traceback (most recent call last)
<ipython-input-4-3e58ad3f6019> in <module>
----> 1 conn.connect()
/usr/local/lib/python3.8/site-packages/kombu/connection.py in connect(self)
    274     def connect(self):
    275         """Establish connection to server immediately."""
--> 276         return self._ensure_connection(
    277             max_retries=1, reraise_as_library_errors=False
    278         )
/usr/local/lib/python3.8/site-packages/kombu/connection.py in _ensure_connection(self, errback, max_retries, interval_start, interval_step, interval_max, callback, reraise_as_library_errors, timeout)
    433             ctx = self._dummy_context
    434         with ctx():
--> 435             return retry_over_time(
    436                 self._connection_factory, self.recoverable_connection_errors,
    437                 (), {}, on_error, max_retries,
/usr/local/lib/python3.8/site-packages/kombu/utils/functional.py in retry_over_time(fun, catch, args, kwargs, errback, max_retries, interval_start, interval_step, interval_max, callback, timeout)
    323     for retries in count():
    324         try:
--> 325             return fun(*args, **kwargs)
    326         except catch as exc:
    327             if max_retries is not None and retries >= max_retries:
/usr/local/lib/python3.8/site-packages/kombu/connection.py in _connection_factory(self)
    864         self.declared_entities.clear()
    865         self._default_channel = None
--> 866         self._connection = self._establish_connection()
    867         self._closed = False
    868         return self._connection
/usr/local/lib/python3.8/site-packages/kombu/connection.py in _establish_connection(self)
    799     def _establish_connection(self):
    800         self._debug('establishing connection...')
--> 801         conn = self.transport.establish_connection()
    802         self._debug('connection established: %r', self)
    803         return conn
/usr/local/lib/python3.8/site-packages/kombu/transport/pyamqp.py in establish_connection(self)
    126         conn = self.Connection(**opts)
    127         conn.client = self.client
--> 128         conn.connect()
    129         return conn
    130 
/usr/local/lib/python3.8/site-packages/amqp/connection.py in connect(self, callback)
    321                 socket_settings=self.socket_settings,
    322             )
--> 323             self.transport.connect()
    324             self.on_inbound_frame = self.frame_handler_cls(
    325                 self, self.on_inbound_method)
/usr/local/lib/python3.8/site-packages/amqp/transport.py in connect(self)
    112                 return
    113             self._connect(self.host, self.port, self.connect_timeout)
--> 114             self._init_socket(
    115                 self.socket_settings, self.read_timeout, self.write_timeout,
    116             )
/usr/local/lib/python3.8/site-packages/amqp/transport.py in _init_socket(self, socket_settings, read_timeout, write_timeout)
    223                     pack('ll', sec, usec),
    224                 )
--> 225         self._setup_transport()
    226 
    227         self._write(AMQP_PROTOCOL_HEADER)
/usr/local/lib/python3.8/site-packages/amqp/transport.py in _setup_transport(self)
    403         """Wrap the socket in an SSL object."""
    404         self.sock = self._wrap_socket(self.sock, **self.sslopts)
--> 405         self.sock.do_handshake()
    406         self._quick_recv = self.sock.read
    407 
/usr/local/lib/python3.8/ssl.py in do_handshake(self, block)
   1307             if timeout == 0.0 and block:
   1308                 self.settimeout(None)
-> 1309             self._sslobj.do_handshake()
   1310         finally:
   1311             self.settimeout(timeout)
SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1125)

After adding amqp==5.0.2 as a dependency everything seems to be working fine.

Therefore I conclude that version 5.0.3 comes with some serious regressions.

[auvipy]

@matusvalo would you mind checking this when you have time?

[tomwojcik]

I host rabbitmq on heroku with a free cloudamqp plugin.
There's one person on SO who runs into the same problem with Amazon MQ.

[auvipy]

this is the PR you should be checking #347

[matusvalo]

I will check it...

[djmitche]

I will address this in a different patch.

Originally posted by @moisesguimaraes in #347 (comment)

I think that's the patch that needed to be made before release..

[moisesguimaraes]

Oh, sorry about that. I have the changes stashed here. I'm running the tests locally and will put a PR shortly.

[auvipy]

I host rabbitmq on heroku with a free cloudamqp plugin.
There's one person on SO who runs into the same problem with Amazon MQ.

you could check this fix #350

[tomwojcik]

I host rabbitmq on heroku with a free cloudamqp plugin.
There's one person on SO who runs into the same problem with Amazon MQ.

you could check this fix #350

I confirm it works :)