If a malformed message gets into Rabbit, Celery worker fails to start about py-amqp

[ask]

It cannot leave the message unacknowledged, as that means it will reduce the prefetch limit.

The only other way it could deal with it is to log the event and discard the message, as requeueing it will cause a loop.

[djsmith42]

@ask That works for me.

[djsmith42]

Correction! This bug is actually in amqp/serialization.py (separate project). The version I have (1.4.9) assumes all headers are UTF-8 encoded with this code:

    def read_longstr(self):
        """Read a string that's up to 2**32 bytes.

        The encoding isn't specified in the AMQP spec, so
        assume it's utf-8

        """
        self.bitcount = self.bits = 0
        slen = unpack('>I', self.input.read(4))[0]
        return self.input.read(slen).decode('utf-8')

But that is not a safe assumption, as header values are totally unconstrained. They could be binary data, for example.

Which is why the UnicodeDecodeError is thrown. I see that the current amqp/serialization.py code is very different in master. Does the current code not have this limitation?

[djsmith42]

Here's a minimal snippet that can repro this crash and put your Celery worker into an unrecoverable loop:

my_task.apply_async(args=['test'], headers={'foo': '\x8d'})

The '\x8d is not UTF-8, and thus when the worker tries to read the header table, it raise UnicodeDecodeError, which will take down your Celery worker and prevent it from coming back up:

[2016-01-23 04:15:51,355: ERROR/MainProcess celery.worker] Unrecoverable error: UnicodeDecodeError('utf8', '\x8d', 0, 1, 'invalid start byte')

I would love to submit a PR to fix this, but I'm not super clear on what version of py-amqp to work on. What do you recommend?

Once I know which version to work on, I'm thinking about modifying the py-amqp serialization module to ignore header values that cannot be decoded with UTF-8.

[djsmith42]

Never mind, I found the 1.4 branch. Here's my PR: #78

[danhenriquesc]

Hi! I'm having the same issue with Celery 4.1.0 and kombu 4.1.0. Any idea?

[auvipy]

can you cleanly apply the patch on pyamqp?

[vazir]

The bug is still present, it does not require a feedback but a simple fix...

[auvipy]

could you tell which versions of celery you are using and facing this issue?