Fix buffer overflow in frame_writer after frame_max is increased

`frame_writer` allocates a `bytearray` on intialization with a length based on the
`connection.frame_max` value. If `connection.frame_max` is changed to a larger value, this causes an
error like `pack_into requires a buffer of at least 408736 bytes`.

From what I can tell, `connection._on_tune` can arbitrarily change this value, although I'm not
totally sure in what cases it actually occurs.

For context, we're upgrading from Celery 3.1.25 -> Celery 4.1, librabbitmq 1.6.1 -> amqp 2.6.0,
kombo 3.0.37 -> 4.2.2.post1. This has worked fine in a small production environment, but fails on
our main deployment. The main culprit we see are lots of errors with `pack_into requires a buffer
of at least <x> bytes`, where x is always larger than our default `frame_max` of 131072. Despite
this, `bigbody` is False, which ends up with us attempting to write a frame that is too large into
the existing buffer. The only way I can see this happening is if `frame_max` increases.

2 files changed, 31 insertions, 2 deletions

diff --git a/amqp/method_framing.py b/amqp/method_framing.py
index a749bd7..380cec0 100644
--- a/amqp/method_framing.py
+++ b/amqp/method_framing.py
@@ -85,6 +85,20 @@ def frame_handler(connection, callback,
     return on_frame
 
 
+class Buffer(object):
+    def __init__(self, buf):
+        self.buf = buf
+
+    @property
+    def buf(self):
+        return self._buf
+
+    @buf.setter
+    def buf(self, buf):
+        self._buf = buf
+        self.view = memoryview(buf)
+
+
 def frame_writer(connection, transport,
                  pack=pack, pack_into=pack_into, range=range, len=len,
                  bytes=bytes, str_to_bytes=str_to_bytes, text_t=text_t):
@@ -94,11 +108,17 @@ def frame_writer(connection, transport,
     # memoryview first supported in Python 2.7
     # Initial support was very shaky, so could be we have to
     # check for a bugfix release.
-    buf = bytearray(connection.frame_max - 8)
-    view = memoryview(buf)
+    buffer_store = Buffer(bytearray(connection.frame_max - 8))
 
     def write_frame(type_, channel, method_sig, args, content):
         chunk_size = connection.frame_max - 8
+        # frame_max can be updated via connection._on_tune. If
+        # it became larger, then we need to resize the buffer
+        # to prevent overflow.
+        if chunk_size > len(buffer_store.buf):
+            buffer_store.buf = bytearray(chunk_size)
+        buf = buffer_store.buf
+        view = buffer_store.view
         offset = 0
         properties = None
         args = str_to_bytes(args)
diff --git a/t/unit/test_method_framing.py b/t/unit/test_method_framing.py
index 232ed8d..6a1ccac 100644
--- a/t/unit/test_method_framing.py
+++ b/t/unit/test_method_framing.py
@@ -138,3 +138,12 @@ class test_frame_writer:
         assert isinstance(memory, memoryview)
         assert 'body'.encode('utf-16') in memory.tobytes()
         assert msg.properties['content_encoding'] == 'utf-16'
+
+    def test_frame_max_update(self):
+        msg = Message(body='t' * (self.connection.frame_max + 10))
+        frame = 2, 1, spec.Basic.Publish, b'x' * 10, msg
+        self.connection.frame_max += 100
+        self.g(*frame)
+        self.write.assert_called()
+        memory = self.write.call_args[0][0]
+        assert isinstance(memory, memoryview)