From d0d827b0074e4d78cdc8284d18c1d535db0bf39f Mon Sep 17 00:00:00 2001
From: liufeng <liufeng@kylinos.cn>
Date: Sat, 1 Aug 2026 17:27:03 +0800
Subject: [PATCH] Validate uncompressed size in OP_COMPRESSED messages

The process_compression_header method previously discarded the
uncompressed_size field from the compression sub-header. A malicious
or compromised server could send a small compressed envelope
(passing the max_message_size check) that decompresses to a very
large payload, causing memory exhaustion.

This change returns the uncompressed_size from the compression header
and validates it against max_message_size before accepting the
compressed payload.
---
 pymongo/network_layer.py                      | 23 +++++++++++++++----
 test/asynchronous/test_async_network_layer.py | 19 +++++++++++++++
 2 files changed, 38 insertions(+), 4 deletions(-)

diff --git a/pymongo/network_layer.py b/pymongo/network_layer.py
index 102f560d65..87bcd18455 100644
--- a/pymongo/network_layer.py
+++ b/pymongo/network_layer.py
@@ -604,7 +604,20 @@ def buffer_updated(self, nbytes: int) -> None:
             self._compression_index += nbytes
             if self._compression_index >= 9:
                 self._expecting_compression = False
-                self._op_code, self._compressor_id = self.process_compression_header()
+                (
+                    self._op_code,
+                    uncompressed_size,
+                    self._compressor_id,
+                ) = self.process_compression_header()
+                if uncompressed_size > self._max_message_size:
+                    self.close(
+                        ProtocolError(
+                            f"Uncompressed message size ({uncompressed_size!r}) "
+                            f"is larger than server max message size "
+                            f"({self._max_message_size!r})"
+                        )
+                    )
+                    return
             return
 
         self._message_index += nbytes
@@ -658,10 +671,12 @@ def process_header(self) -> tuple[int, int, int, bool]:
 
         return length - 16, op_code, response_to, expecting_compression
 
-    def process_compression_header(self) -> tuple[int, int]:
+    def process_compression_header(self) -> tuple[int, int, int]:
         """Unpack a MongoDB Wire Protocol compression header."""
-        op_code, _, compressor_id = _UNPACK_COMPRESSION_HEADER(self._compression_header)
-        return op_code, compressor_id
+        op_code, uncompressed_size, compressor_id = _UNPACK_COMPRESSION_HEADER(
+            self._compression_header
+        )
+        return op_code, uncompressed_size, compressor_id
 
     def _resolve_pending_messages(self, exc: Optional[Exception] = None) -> None:
         pending = list(self._pending_messages)
diff --git a/test/asynchronous/test_async_network_layer.py b/test/asynchronous/test_async_network_layer.py
index 5adb7aaeac..31b7812bad 100644
--- a/test/asynchronous/test_async_network_layer.py
+++ b/test/asynchronous/test_async_network_layer.py
@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import asyncio
+import struct
 import sys
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -88,6 +89,24 @@ def test_length_exceeds_max_raises(self):
         with self.assertRaisesRegex(ProtocolError, "larger than server max"):
             self.protocol.process_header()
 
+    def test_compression_uncompressed_size_exceeds_max_closes(self):
+        self.protocol._max_message_size = 1024
+        self.protocol._header = memoryview(
+            bytearray(
+                pack_msg_header(
+                    length=35, request_id=1, response_to=0, op_code=2012
+                )
+            )
+        )
+        self.protocol.process_header()
+        # Now feed compression sub-header with uncompressed_size > max
+        self.protocol._compression_header[:] = struct.pack(
+            "<iiB", 2013, 9999, 2
+        )
+        self.protocol._compression_index = 9
+        self.protocol.buffer_updated(0)
+        self.protocol.transport.abort.assert_called()
+
 
 class TestClose(AsyncUnitTest):
     async def asyncSetUp(self):