PYTHON-5994 Use exact match for file ID in GridFS delete methods (#3046)

Co-authored-by: Iris <58442094+sleepyStick@users.noreply.github.com>

Author: aclark4life

gridfs/asynchronous/grid_file.py

@@ -284,8 +284,8 @@ async def delete(self, file_id: Any, session: Optional[AsyncClientSession] = Non
            ``delete`` no longer ensures indexes.
         """
         _disallow_transactions(session)
-        await self._files.delete_one({"_id": file_id}, session=session)
-        await self._chunks.delete_many({"files_id": file_id}, session=session)
+        await self._files.delete_one({"_id": {"$eq": file_id}}, session=session)
+        await self._chunks.delete_many({"files_id": {"$eq": file_id}}, session=session)
 
     async def list(self, session: Optional[AsyncClientSession] = None) -> list[str]:
         """List the names of all files stored in this instance of
@@ -341,7 +341,7 @@ async def find_one(
            Added ``session`` parameter.
         """
         if filter is not None and not isinstance(filter, abc.Mapping):
-            filter = {"_id": filter}
+            filter = {"_id": {"$eq": filter}}
 
         _disallow_transactions(session)
         async for f in self.find(filter, *args, session=session, **kwargs):
@@ -455,6 +455,8 @@ async def exists(
         if kwargs:
             f = await self._files.find_one(kwargs, ["_id"], session=session)
         else:
+            if document_or_id is not None and not isinstance(document_or_id, abc.Mapping):
+                document_or_id = {"_id": {"$eq": document_or_id}}
             f = await self._files.find_one(document_or_id, ["_id"], session=session)
 
         return f is not None
@@ -831,8 +833,8 @@ async def delete(self, file_id: Any, session: Optional[AsyncClientSession] = Non
            Added ``session`` parameter.
         """
         _disallow_transactions(session)
-        res = await self._files.delete_one({"_id": file_id}, session=session)
-        await self._chunks.delete_many({"files_id": file_id}, session=session)
+        res = await self._files.delete_one({"_id": {"$eq": file_id}}, session=session)
+        await self._chunks.delete_many({"files_id": {"$eq": file_id}}, session=session)
         if not res.deleted_count:
             raise NoFile(f"no file could be deleted because none matched {file_id}")
 
@@ -1044,7 +1046,7 @@ async def rename(
         """
         _disallow_transactions(session)
         result = await self._files.update_one(
-            {"_id": file_id}, {"$set": {"filename": new_filename}}, session=session
+            {"_id": {"$eq": file_id}}, {"$set": {"filename": new_filename}}, session=session
         )
         if not result.matched_count:
             raise NoFile(
@@ -1198,8 +1200,12 @@ async def _ensure_indexes(self) -> None:
 
     async def abort(self) -> None:
         """Remove all chunks/files that may have been uploaded and close."""
-        await self._coll.chunks.delete_many({"files_id": self._file["_id"]}, session=self._session)
-        await self._coll.files.delete_one({"_id": self._file["_id"]}, session=self._session)
+        await self._coll.chunks.delete_many(
+            {"files_id": {"$eq": self._file["_id"]}}, session=self._session
+        )
+        await self._coll.files.delete_one(
+            {"_id": {"$eq": self._file["_id"]}}, session=self._session
+        )
         object.__setattr__(self, "_closed", True)
 
     @property
@@ -1248,7 +1254,9 @@ def __setattr__(self, name: str, value: Any) -> None:
             self._file[name] = value
             if self._closed:
                 if _IS_SYNC:
-                    self._coll.files.update_one({"_id": self._file["_id"]}, {"$set": {name: value}})
+                    self._coll.files.update_one(
+                        {"_id": {"$eq": self._file["_id"]}}, {"$set": {name: value}}
+                    )
                 else:
                     raise AttributeError(
                         "AsyncGridIn does not support __setattr__ after being closed(). Set the attribute before closing the file or use AsyncGridIn.set() instead"
@@ -1257,7 +1265,9 @@ def __setattr__(self, name: str, value: Any) -> None:
     async def set(self, name: str, value: Any) -> None:
         self._file[name] = value
         if self._closed:
-            await self._coll.files.update_one({"_id": self._file["_id"]}, {"$set": {name: value}})
+            await self._coll.files.update_one(
+                {"_id": {"$eq": self._file["_id"]}}, {"$set": {name: value}}
+            )
 
     async def _flush_data(self, data: Any, force: bool = False) -> None:
         """Flush `data` to a chunk."""
@@ -1582,7 +1592,9 @@ async def readlines(self, size: int = -1) -> list[bytes]:
     async def open(self) -> None:
         if not self._file:
             _disallow_transactions(self._session)
-            self._file = await self._files.find_one({"_id": self._file_id}, session=self._session)
+            self._file = await self._files.find_one(
+                {"_id": {"$eq": self._file_id}}, session=self._session
+            )
             if not self._file:
                 raise NoFile(
                     f"no file in gridfs collection {self._files!r} with _id {self._file_id!r}"
@@ -1853,7 +1865,7 @@ def __aiter__(self) -> _AsyncGridOutChunkIterator:
         return self
 
     def _create_cursor(self) -> None:
-        filter = {"files_id": self._id}
+        filter = {"files_id": {"$eq": self._id}}
         if self._next_chunk > 0:
             filter["n"] = {"$gte": self._next_chunk}
         _disallow_transactions(self._session)

gridfs/grid_file_shared.py

@@ -120,7 +120,9 @@ def setter(self: Any, value: Any) -> Any:
         if warn_str:
             warnings.warn(warn_str, stacklevel=2, category=DeprecationWarning)
         if self._closed:
-            self._coll.files.update_one({"_id": self._file["_id"]}, {"$set": {field_name: value}})
+            self._coll.files.update_one(
+                {"_id": {"$eq": self._file["_id"]}}, {"$set": {field_name: value}}
+            )
         self._file[field_name] = value
 
     if read_only:

gridfs/synchronous/grid_file.py

@@ -284,8 +284,8 @@ def delete(self, file_id: Any, session: Optional[ClientSession] = None) -> None:
            ``delete`` no longer ensures indexes.
         """
         _disallow_transactions(session)
-        self._files.delete_one({"_id": file_id}, session=session)
-        self._chunks.delete_many({"files_id": file_id}, session=session)
+        self._files.delete_one({"_id": {"$eq": file_id}}, session=session)
+        self._chunks.delete_many({"files_id": {"$eq": file_id}}, session=session)
 
     def list(self, session: Optional[ClientSession] = None) -> list[str]:
         """List the names of all files stored in this instance of
@@ -339,7 +339,7 @@ def find_one(
            Added ``session`` parameter.
         """
         if filter is not None and not isinstance(filter, abc.Mapping):
-            filter = {"_id": filter}
+            filter = {"_id": {"$eq": filter}}
 
         _disallow_transactions(session)
         for f in self.find(filter, *args, session=session, **kwargs):
@@ -453,6 +453,8 @@ def exists(
         if kwargs:
             f = self._files.find_one(kwargs, ["_id"], session=session)
         else:
+            if document_or_id is not None and not isinstance(document_or_id, abc.Mapping):
+                document_or_id = {"_id": {"$eq": document_or_id}}
             f = self._files.find_one(document_or_id, ["_id"], session=session)
 
         return f is not None
@@ -827,8 +829,8 @@ def delete(self, file_id: Any, session: Optional[ClientSession] = None) -> None:
            Added ``session`` parameter.
         """
         _disallow_transactions(session)
-        res = self._files.delete_one({"_id": file_id}, session=session)
-        self._chunks.delete_many({"files_id": file_id}, session=session)
+        res = self._files.delete_one({"_id": {"$eq": file_id}}, session=session)
+        self._chunks.delete_many({"files_id": {"$eq": file_id}}, session=session)
         if not res.deleted_count:
             raise NoFile(f"no file could be deleted because none matched {file_id}")
 
@@ -1036,7 +1038,7 @@ def rename(
         """
         _disallow_transactions(session)
         result = self._files.update_one(
-            {"_id": file_id}, {"$set": {"filename": new_filename}}, session=session
+            {"_id": {"$eq": file_id}}, {"$set": {"filename": new_filename}}, session=session
         )
         if not result.matched_count:
             raise NoFile(
@@ -1186,8 +1188,10 @@ def _ensure_indexes(self) -> None:
 
     def abort(self) -> None:
         """Remove all chunks/files that may have been uploaded and close."""
-        self._coll.chunks.delete_many({"files_id": self._file["_id"]}, session=self._session)
-        self._coll.files.delete_one({"_id": self._file["_id"]}, session=self._session)
+        self._coll.chunks.delete_many(
+            {"files_id": {"$eq": self._file["_id"]}}, session=self._session
+        )
+        self._coll.files.delete_one({"_id": {"$eq": self._file["_id"]}}, session=self._session)
         object.__setattr__(self, "_closed", True)
 
     @property
@@ -1236,7 +1240,9 @@ def __setattr__(self, name: str, value: Any) -> None:
             self._file[name] = value
             if self._closed:
                 if _IS_SYNC:
-                    self._coll.files.update_one({"_id": self._file["_id"]}, {"$set": {name: value}})
+                    self._coll.files.update_one(
+                        {"_id": {"$eq": self._file["_id"]}}, {"$set": {name: value}}
+                    )
                 else:
                     raise AttributeError(
                         "GridIn does not support __setattr__ after being closed(). Set the attribute before closing the file or use GridIn.set() instead"
@@ -1245,7 +1251,9 @@ def __setattr__(self, name: str, value: Any) -> None:
     def set(self, name: str, value: Any) -> None:
         self._file[name] = value
         if self._closed:
-            self._coll.files.update_one({"_id": self._file["_id"]}, {"$set": {name: value}})
+            self._coll.files.update_one(
+                {"_id": {"$eq": self._file["_id"]}}, {"$set": {name: value}}
+            )
 
     def _flush_data(self, data: Any, force: bool = False) -> None:
         """Flush `data` to a chunk."""
@@ -1570,7 +1578,9 @@ def readlines(self, size: int = -1) -> list[bytes]:
     def open(self) -> None:
         if not self._file:
             _disallow_transactions(self._session)
-            self._file = self._files.find_one({"_id": self._file_id}, session=self._session)
+            self._file = self._files.find_one(
+                {"_id": {"$eq": self._file_id}}, session=self._session
+            )
             if not self._file:
                 raise NoFile(
                     f"no file in gridfs collection {self._files!r} with _id {self._file_id!r}"
@@ -1841,7 +1851,7 @@ def __iter__(self) -> GridOutChunkIterator:
         return self
 
     def _create_cursor(self) -> None:
-        filter = {"files_id": self._id}
+        filter = {"files_id": {"$eq": self._id}}
         if self._next_chunk > 0:
             filter["n"] = {"$gte": self._next_chunk}
         _disallow_transactions(self._session)

test/asynchronous/test_gridfs_bucket.py

@@ -33,6 +33,7 @@
 import gridfs
 from bson.binary import Binary
 from bson.int64 import Int64
+from bson.min_key import MinKey
 from bson.objectid import ObjectId
 from bson.son import SON
 from gridfs.errors import CorruptGridFile, NoFile
@@ -42,6 +43,7 @@
     NotPrimaryError,
     ServerSelectionTimeoutError,
     WriteConcernError,
+    WriteError,
 )
 from pymongo.read_preferences import ReadPreference
 from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest
@@ -475,6 +477,34 @@ async def test_abort(self):
             await gin.write(b"test4")
         self.assertEqual(0, await self.db.fs.chunks.count_documents({"files_id": gin._id}))
 
+    # Server versions older than 5.0 do not support document values with
+    # "$"-prefixed keys.
+    @async_client_context.require_version_min(5, 0)
+    async def test_abort_with_injected_file_id_does_not_delete_other_files(self):
+        bucket = self.fs
+        file1_bytes = b"\x11\x22\x33\x44"
+        await bucket.upload_from_stream("file1", file1_bytes)
+
+        upload_stream = bucket.open_upload_stream_with_id(
+            {"$gt": MinKey()}, "file2", chunk_size_bytes=2
+        )
+        await upload_stream.write(b"\x55\x66\x77\x88")
+        try:
+            await upload_stream.abort()
+        except WriteError:
+            # Newer server versions reject an equality match on "_id" whose
+            # operand itself resembles a query operator, even when the
+            # driver safely wraps it in "$eq". The chunks are still cleaned
+            # up in that case; only the (never created) files document
+            # delete is rejected.
+            pass
+
+        self.assertEqual(
+            file1_bytes, await (await bucket.open_download_stream_by_name("file1")).read()
+        )
+        with self.assertRaises(NoFile):
+            await bucket.open_download_stream_by_name("file2")
+
     async def test_download_to_stream(self):
         file1 = BytesIO(b"hello world")
         # Test with one chunk.