From 42ccc9d7828219ab81651843679179302e76ec08 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 8 Oct 2026 12:13:58 -0500
Subject: [PATCH] PYTHON-6110 [v4.18] Bound Regex element scans to the
 remaining document bytes

Decoding a malformed Regex element could scan past the end of the
caller's buffer. The decoder now searches for each string's NUL within
the document instead of scanning the buffer, raising InvalidBSON on
malformed elements on every accepted buffer type.

(cherry picked from commit 8fb5685985c2bae714d0ce8fdc7ee2226af7a36d)
---
 bson/_cbsonmodule.c | 23 ++++++++++++++++-------
 test/test_bson.py   | 45 +++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 61 insertions(+), 7 deletions(-)

diff --git a/bson/_cbsonmodule.c b/bson/_cbsonmodule.c
index d383cee308..0caf84a54d 100644
--- a/bson/_cbsonmodule.c
+++ b/bson/_cbsonmodule.c
@@ -2466,24 +2466,33 @@ static PyObject* get_value(PyObject* self, PyObject* name, const char* buffer,
             PyObject* pattern;
             int flags;
             size_t flags_length, i;
-            size_t pattern_length = strlen(buffer + *position);
-            if (pattern_length > BSON_MAX_SIZE || max < pattern_length) {
+            /* PYTHON-6110: Search for each NUL within the bytes remaining
+             * in the document rather than scanning the buffer. */
+            const char* start = buffer + *position;
+            const char* pattern_nul = memchr(start, 0, max);
+            if (!pattern_nul) {
+                goto invalid;
+            }
+            size_t pattern_length = (size_t)(pattern_nul - start);
+            if (pattern_length > BSON_MAX_SIZE) {
                 goto invalid;
             }
             pattern = PyUnicode_DecodeUTF8(
-                buffer + *position, pattern_length,
+                start, pattern_length,
                 options->unicode_decode_error_handler);
             if (!pattern) {
                 goto invalid;
             }
             *position += (unsigned)pattern_length + 1;
-            flags_length = strlen(buffer + *position);
-            if (flags_length > BSON_MAX_SIZE ||
-                    (BSON_MAX_SIZE - pattern_length) < flags_length) {
+            start += pattern_length + 1;
+            const char* flags_nul = memchr(start, 0, max - pattern_length);
+            if (!flags_nul) {
                 Py_DECREF(pattern);
                 goto invalid;
             }
-            if (max < pattern_length + flags_length) {
+            flags_length = (size_t)(flags_nul - start);
+            if (flags_length > BSON_MAX_SIZE ||
+                    (BSON_MAX_SIZE - pattern_length) < flags_length) {
                 Py_DECREF(pattern);
                 goto invalid;
             }
diff --git a/test/test_bson.py b/test/test_bson.py
index 0800806b84..8367fdb95f 100644
--- a/test/test_bson.py
+++ b/test/test_bson.py
@@ -416,6 +416,51 @@ def test_invalid_field_name(self):
         # Assert that the InvalidBSON error message is not empty.
         self.assertTrue(str(ctx.exception))
 
+    def test_malformed_regex(self):
+        # PYTHON-6110: decoding a malformed Regex element must raise
+        # InvalidBSON on every accepted buffer type.
+        malformed = []
+        for pattern in ("", "x", "xx"):
+            doc = bytearray(encode({"a": Regex(pattern, "")}))
+            # Drop the trailing flags NUL and document EOO bytes, leaving
+            # the flags string unterminated after the pattern's NUL.
+            del doc[-2:]
+            # Keep the length prefix consistent with the shorter document.
+            doc[:4] = struct.pack("<i", len(doc))
+            malformed.append(bytes(doc))
+        for data in malformed:
+            with self.assertRaises(InvalidBSON):
+                decode(data)
+            with self.assertRaises(InvalidBSON):
+                decode(bytearray(data))
+            with self.assertRaises(InvalidBSON):
+                decode(memoryview(data))
+            with self.assertRaises(InvalidBSON):
+                decode(array.array("B", data))
+            with self.assertRaises(InvalidBSON):
+                list(decode_iter(data))
+            with self.assertRaises(InvalidBSON):
+                decode_all(data)
+            with mmap.mmap(-1, len(data)) as mm:
+                mm.write(data)
+                mm.seek(0)
+                with self.assertRaises(InvalidBSON):
+                    decode(mm)
+                with self.assertRaises(InvalidBSON):
+                    decode_all(mm)
+
+    def test_regex_empty_flags(self):
+        # PYTHON-6110: a valid regex with an empty flags string must
+        # continue to decode.
+        valid = encode({"a": Regex("b", "")})
+        expected = {"a": Regex("b", 0)}
+        self.assertEqual(decode(valid), expected)
+        with mmap.mmap(-1, len(valid)) as mm:
+            mm.write(valid)
+            mm.seek(0)
+            self.assertEqual(decode(mm), expected)
+            self.assertEqual(decode_all(mm), [expected])
+
     def test_data_timestamp(self):
         self.assertEqual(
             {"test": Timestamp(4, 20)},