PYTHON-6110 Bound Regex element scans to the remaining document bytes

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;
             }

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)},