PYTHON-5968 Reject extra fields in Extended JSON timestamps (#2961)

Co-authored-by: Steven Silvester <steve.silvester@mongodb.com>

Author: winklemad

bson/json_util.py

@@ -788,7 +788,13 @@ def _parse_binary(doc: Any, json_options: JSONOptions) -> Union[Binary, uuid.UUI
 
 
 def _parse_timestamp(doc: Any, dummy0: Any) -> Timestamp:
+    if len(doc) != 1:
+        raise TypeError(f"Bad $timestamp, extra field(s): {doc}")
     tsp = doc["$timestamp"]
+    if not isinstance(tsp, Mapping):
+        raise TypeError(f'$timestamp value must be a document with "t" and "i" components: {doc}')
+    if set(tsp) != {"t", "i"}:
+        raise TypeError(f'$timestamp must include exactly "t" and "i" components: {doc}')
     return Timestamp(tsp["t"], tsp["i"])
 
 

doc/changelog.rst

@@ -25,6 +25,8 @@ PyMongo 4.18 brings a number of changes including:
   the bytes remaining in the array now raises
   :class:`~bson.errors.InvalidBSON` instead of reading past the end of the
   buffer.
+- Fixed :func:`bson.json_util.loads` to reject ``$timestamp`` values containing
+  fields other than ``t`` and ``i``.
 
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------

doc/contributors.rst

@@ -109,3 +109,4 @@ The following is a list of people who have contributed to
 - Noah Stapp (NoahStapp)
 - Cal Jacobson (cj81499)
 - Sophia Yang (sophiayangDB)
+- Madan Kumar (winklemad)

test/test_json_util.py

@@ -405,6 +405,28 @@ def test_timestamp(self):
         self.assertEqual(dct, rtdct)
         self.assertEqual('{"ts": {"$timestamp": {"t": 4, "i": 13}}}', res)
 
+    def test_timestamp_with_invalid_fields(self):
+        invalid_values = [
+            '{"t": 4, "i": 13, "extra": 1}',
+            '{"t": 4, "unexpected": 13}',
+        ]
+        for value in invalid_values:
+            with self.subTest(value=value):
+                with self.assertRaisesRegex(
+                    TypeError, r'\$timestamp must include exactly "t" and "i" components'
+                ):
+                    json_util.loads(f'{{"ts": {{"$timestamp": {value}}}}}')
+
+    def test_timestamp_with_extra_wrapper_fields(self):
+        with self.assertRaisesRegex(TypeError, r"Bad \$timestamp, extra field\(s\)"):
+            json_util.loads('{"ts": {"$timestamp": {"t": 4, "i": 13}, "extra": 1}}')
+
+    def test_timestamp_with_non_document_value(self):
+        for value in ('["t", "i"]', '"ti"', "5"):
+            with self.subTest(value=value):
+                with self.assertRaisesRegex(TypeError, r"\$timestamp value must be a document"):
+                    json_util.loads(f'{{"ts": {{"$timestamp": {value}}}}}')
+
     def test_uuid_default(self):
         # Cannot directly encode native UUIDs with the default
         # uuid_representation.