From ae4266a27c713fe72f9c71155346942a7995fa6e Mon Sep 17 00:00:00 2001
From: Jeffrey 'Alex' Clark <alex.clark@mongodb.com>
Date: Fri, 31 Jul 2026 13:23:05 -0400
Subject: [PATCH 1/9] PYTHON-5981 Reject aggregate/pipeline as aggregation
 options

---
 doc/changelog.rst                    |  9 +++++++++
 pymongo/asynchronous/aggregation.py  |  5 +++++
 pymongo/synchronous/aggregation.py   |  5 +++++
 test/asynchronous/test_collection.py | 13 +++++++++++++
 test/test_collection.py              | 13 +++++++++++++
 5 files changed, 45 insertions(+)

diff --git a/doc/changelog.rst b/doc/changelog.rst
index fb7d300b2e..afdf7d5adf 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -39,6 +39,15 @@ PyMongo 4.18 brings a number of changes including:
 - Fixed a bug on Windows, and on macOS when using PyOpenSSL, where
   ``SSL_CERT_FILE``/``SSL_CERT_DIR`` were merged with, rather than replacing,
   the OS/certifi certificate store.
+- Aggregation helpers now raise :exc:`~pymongo.errors.ConfigurationError` when
+  passed an ``aggregate`` or ``pipeline`` keyword argument. Previously these
+  keys silently replaced the target namespace and pipeline of the generated
+  ``aggregate`` command, so an application forwarding untrusted option names to
+  :meth:`~pymongo.asynchronous.collection.AsyncCollection.aggregate`,
+  :meth:`~pymongo.asynchronous.collection.AsyncCollection.aggregate_raw_batches`,
+  :meth:`~pymongo.asynchronous.database.AsyncDatabase.aggregate`, or
+  :meth:`~pymongo.asynchronous.collection.AsyncCollection.list_search_indexes`
+  could be redirected to read or overwrite an unintended collection.
 
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------
diff --git a/pymongo/asynchronous/aggregation.py b/pymongo/asynchronous/aggregation.py
index f1f77acc73..8b2331fc5c 100644
--- a/pymongo/asynchronous/aggregation.py
+++ b/pymongo/asynchronous/aggregation.py
@@ -60,6 +60,11 @@ def __init__(
             raise ConfigurationError(
                 "The explain option is not supported. Use AsyncDatabase.command instead."
             )
+        for name in ("aggregate", "pipeline"):
+            if name in options:
+                raise ConfigurationError(
+                    f"The {name} option cannot be specified as a keyword argument"
+                )
 
         self._target = target
 
diff --git a/pymongo/synchronous/aggregation.py b/pymongo/synchronous/aggregation.py
index d540484fa8..9cdd32952d 100644
--- a/pymongo/synchronous/aggregation.py
+++ b/pymongo/synchronous/aggregation.py
@@ -60,6 +60,11 @@ def __init__(
             raise ConfigurationError(
                 "The explain option is not supported. Use Database.command instead."
             )
+        for name in ("aggregate", "pipeline"):
+            if name in options:
+                raise ConfigurationError(
+                    f"The {name} option cannot be specified as a keyword argument"
+                )
 
         self._target = target
 
diff --git a/test/asynchronous/test_collection.py b/test/asynchronous/test_collection.py
index d787080d02..3f10c70f88 100644
--- a/test/asynchronous/test_collection.py
+++ b/test/asynchronous/test_collection.py
@@ -1568,6 +1568,19 @@ async def test_aggregate(self):
         with self.write_concern_collection() as coll:
             await coll.aggregate([{"$out": "output-collection"}])
 
+    async def test_aggregate_reserved_options(self):
+        db = self.db
+        with self.assertRaises(ConfigurationError):
+            await db.test.aggregate([], aggregate="other")
+        with self.assertRaises(ConfigurationError):
+            await db.test.aggregate_raw_batches([], aggregate="other")
+        with self.assertRaises(ConfigurationError):
+            await db.aggregate([], aggregate="other")
+        with self.assertRaises(ConfigurationError):
+            await db.test.list_search_indexes(aggregate="other")
+        with self.assertRaises(ConfigurationError):
+            await db.test.list_search_indexes(pipeline=[{"$out": "other"}])
+
     async def test_aggregate_raw_bson(self):
         db = self.db
         await db.drop_collection("test")
diff --git a/test/test_collection.py b/test/test_collection.py
index 7a52f94081..78aaa12ab6 100644
--- a/test/test_collection.py
+++ b/test/test_collection.py
@@ -1550,6 +1550,19 @@ def test_aggregate(self):
         with self.write_concern_collection() as coll:
             coll.aggregate([{"$out": "output-collection"}])
 
+    def test_aggregate_reserved_options(self):
+        db = self.db
+        with self.assertRaises(ConfigurationError):
+            db.test.aggregate([], aggregate="other")
+        with self.assertRaises(ConfigurationError):
+            db.test.aggregate_raw_batches([], aggregate="other")
+        with self.assertRaises(ConfigurationError):
+            db.aggregate([], aggregate="other")
+        with self.assertRaises(ConfigurationError):
+            db.test.list_search_indexes(aggregate="other")
+        with self.assertRaises(ConfigurationError):
+            db.test.list_search_indexes(pipeline=[{"$out": "other"}])
+
     def test_aggregate_raw_bson(self):
         db = self.db
         db.drop_collection("test")

From 4fa67f7ba8254950764130816b41b6788fed18ee Mon Sep 17 00:00:00 2001
From: Jeffrey 'Alex' Clark <alex.clark@mongodb.com>
Date: Tue, 18 Aug 2026 16:29:27 -0400
Subject: [PATCH 2/9] PYTHON-5981 Assert the resulting command in aggregation
 option tests

The reserved-option tests only asserted that ConfigurationError is raised,
so one of them passed for the wrong reason:
list_search_indexes(aggregate=...) raises OperationFailure on a non-Atlas
server before the option is applied.

Assert the behavior instead. Command monitoring shows the supplied
namespace never reaches the wire, which holds on any topology, and
separate tests show a supplied pipeline can neither read ($unionWith)
nor overwrite ($out) another collection.
---
 test/asynchronous/test_collection.py | 84 ++++++++++++++++++++++++++++
 test/test_collection.py              | 82 +++++++++++++++++++++++++++
 2 files changed, 166 insertions(+)

diff --git a/test/asynchronous/test_collection.py b/test/asynchronous/test_collection.py
index 3f10c70f88..af6dd4bf3b 100644
--- a/test/asynchronous/test_collection.py
+++ b/test/asynchronous/test_collection.py
@@ -1569,6 +1569,9 @@ async def test_aggregate(self):
             await coll.aggregate([{"$out": "output-collection"}])
 
     async def test_aggregate_reserved_options(self):
+        # The reserved command fields must not be settable as options: an
+        # application that forwards a caller-supplied options mapping would
+        # otherwise let the caller retarget the command at any collection.
         db = self.db
         with self.assertRaises(ConfigurationError):
             await db.test.aggregate([], aggregate="other")
@@ -1581,6 +1584,87 @@ async def test_aggregate_reserved_options(self):
         with self.assertRaises(ConfigurationError):
             await db.test.list_search_indexes(pipeline=[{"$out": "other"}])
 
+    async def test_aggregate_reserved_options_do_not_reach_server(self):
+        # Assert the security property rather than the error type: the injected
+        # namespace must never reach the wire, and the secret must never reach
+        # the caller. Checking the command as sent keeps this independent of
+        # whether the server would have accepted the injected pipeline, so it
+        # fails on a vulnerable driver for the right reason on any topology.
+        listener = OvertCommandListener()
+        client = await self.async_single_client(event_listeners=[listener])
+        db = client[self.db.name]
+        await self.db.drop_collection("secrets")
+        self.addAsyncCleanup(self.db.drop_collection, "secrets")
+        await self.db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
+
+        attempts = {
+            "aggregate": lambda: db.test.aggregate([], aggregate="secrets"),
+            "aggregate_raw_batches": lambda: db.test.aggregate_raw_batches([], aggregate="secrets"),
+            "database aggregate": lambda: db.aggregate([], aggregate="secrets"),
+            "list_search_indexes aggregate": lambda: db.test.list_search_indexes(
+                aggregate="secrets"
+            ),
+        }
+        for name, attempt in attempts.items():
+            with self.subTest(entry_point=name):
+                listener.reset()
+                leaked = []
+                # A vulnerable driver raises nothing; a server that rejects the
+                # injected pipeline raises OperationFailure. Neither outcome is
+                # what this test asserts on, so both are tolerated here.
+                with contextlib.suppress(ConfigurationError, OperationFailure):
+                    leaked = await (await attempt()).to_list()
+                targets = [
+                    event.command.get("aggregate")
+                    for event in listener.started_events
+                    if event.command_name == "aggregate"
+                ]
+                self.assertNotIn(
+                    "secrets", targets, f"{name} sent the injected namespace to the server"
+                )
+                self.assertEqual(leaked, [], f"{name} returned another collection's documents")
+
+    @async_client_context.require_version_min(4, 4, -1)
+    async def test_pipeline_option_cannot_read_another_collection(self):
+        # The pipeline option alone is a distinct primitive: it needs no
+        # aggregate key, so the command still names the intended collection
+        # while $unionWith pulls in another one. Requires $unionWith (4.4+).
+        db = self.db
+        await db.drop_collection("secrets")
+        self.addAsyncCleanup(db.drop_collection, "secrets")
+        await db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
+
+        leaked = []
+        with contextlib.suppress(ConfigurationError, OperationFailure):
+            leaked = await (
+                await db.test.list_search_indexes(pipeline=[{"$unionWith": "secrets"}])
+            ).to_list()
+        self.assertNotIn(
+            "sentinel",
+            [doc.get("api_key") for doc in leaked],
+            "injected $unionWith read another collection",
+        )
+
+    async def test_pipeline_option_cannot_write_another_collection(self):
+        # The injected pipeline also evades the $out/$merge write detection,
+        # which inspects only the legitimate pipeline argument. Assert the
+        # target collection is left untouched.
+        db = self.db
+        await db.drop_collection("secrets")
+        await db.drop_collection("billing")
+        self.addAsyncCleanup(db.drop_collection, "secrets")
+        self.addAsyncCleanup(db.drop_collection, "billing")
+        await db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
+        await db.billing.insert_one({"_id": 1, "balance": 100})
+
+        with contextlib.suppress(ConfigurationError, OperationFailure):
+            await db.test.list_search_indexes(pipeline=[{"$match": {}}, {"$out": "billing"}])
+        self.assertEqual(
+            await db.billing.find().to_list(),
+            [{"_id": 1, "balance": 100}],
+            "injected $out overwrote another collection",
+        )
+
     async def test_aggregate_raw_bson(self):
         db = self.db
         await db.drop_collection("test")
diff --git a/test/test_collection.py b/test/test_collection.py
index 78aaa12ab6..bb154a1f2f 100644
--- a/test/test_collection.py
+++ b/test/test_collection.py
@@ -1551,6 +1551,9 @@ def test_aggregate(self):
             coll.aggregate([{"$out": "output-collection"}])
 
     def test_aggregate_reserved_options(self):
+        # The reserved command fields must not be settable as options: an
+        # application that forwards a caller-supplied options mapping would
+        # otherwise let the caller retarget the command at any collection.
         db = self.db
         with self.assertRaises(ConfigurationError):
             db.test.aggregate([], aggregate="other")
@@ -1563,6 +1566,85 @@ def test_aggregate_reserved_options(self):
         with self.assertRaises(ConfigurationError):
             db.test.list_search_indexes(pipeline=[{"$out": "other"}])
 
+    def test_aggregate_reserved_options_do_not_reach_server(self):
+        # Assert the security property rather than the error type: the injected
+        # namespace must never reach the wire, and the secret must never reach
+        # the caller. Checking the command as sent keeps this independent of
+        # whether the server would have accepted the injected pipeline, so it
+        # fails on a vulnerable driver for the right reason on any topology.
+        listener = OvertCommandListener()
+        client = self.single_client(event_listeners=[listener])
+        db = client[self.db.name]
+        self.db.drop_collection("secrets")
+        self.addCleanup(self.db.drop_collection, "secrets")
+        self.db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
+
+        attempts = {
+            "aggregate": lambda: db.test.aggregate([], aggregate="secrets"),
+            "aggregate_raw_batches": lambda: db.test.aggregate_raw_batches([], aggregate="secrets"),
+            "database aggregate": lambda: db.aggregate([], aggregate="secrets"),
+            "list_search_indexes aggregate": lambda: db.test.list_search_indexes(
+                aggregate="secrets"
+            ),
+        }
+        for name, attempt in attempts.items():
+            with self.subTest(entry_point=name):
+                listener.reset()
+                leaked = []
+                # A vulnerable driver raises nothing; a server that rejects the
+                # injected pipeline raises OperationFailure. Neither outcome is
+                # what this test asserts on, so both are tolerated here.
+                with contextlib.suppress(ConfigurationError, OperationFailure):
+                    leaked = (attempt()).to_list()
+                targets = [
+                    event.command.get("aggregate")
+                    for event in listener.started_events
+                    if event.command_name == "aggregate"
+                ]
+                self.assertNotIn(
+                    "secrets", targets, f"{name} sent the injected namespace to the server"
+                )
+                self.assertEqual(leaked, [], f"{name} returned another collection's documents")
+
+    @client_context.require_version_min(4, 4, -1)
+    def test_pipeline_option_cannot_read_another_collection(self):
+        # The pipeline option alone is a distinct primitive: it needs no
+        # aggregate key, so the command still names the intended collection
+        # while $unionWith pulls in another one. Requires $unionWith (4.4+).
+        db = self.db
+        db.drop_collection("secrets")
+        self.addCleanup(db.drop_collection, "secrets")
+        db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
+
+        leaked = []
+        with contextlib.suppress(ConfigurationError, OperationFailure):
+            leaked = (db.test.list_search_indexes(pipeline=[{"$unionWith": "secrets"}])).to_list()
+        self.assertNotIn(
+            "sentinel",
+            [doc.get("api_key") for doc in leaked],
+            "injected $unionWith read another collection",
+        )
+
+    def test_pipeline_option_cannot_write_another_collection(self):
+        # The injected pipeline also evades the $out/$merge write detection,
+        # which inspects only the legitimate pipeline argument. Assert the
+        # target collection is left untouched.
+        db = self.db
+        db.drop_collection("secrets")
+        db.drop_collection("billing")
+        self.addCleanup(db.drop_collection, "secrets")
+        self.addCleanup(db.drop_collection, "billing")
+        db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
+        db.billing.insert_one({"_id": 1, "balance": 100})
+
+        with contextlib.suppress(ConfigurationError, OperationFailure):
+            db.test.list_search_indexes(pipeline=[{"$match": {}}, {"$out": "billing"}])
+        self.assertEqual(
+            db.billing.find().to_list(),
+            [{"_id": 1, "balance": 100}],
+            "injected $out overwrote another collection",
+        )
+
     def test_aggregate_raw_bson(self):
         db = self.db
         db.drop_collection("test")

From 002e8e9240793dcee01eb5fc4bab94554a8a3dee Mon Sep 17 00:00:00 2001
From: Jeffrey 'Alex' Clark <alex.clark@mongodb.com>
Date: Tue, 18 Aug 2026 16:29:32 -0400
Subject: [PATCH 3/9] PYTHON-5981 Drop redundant reserved option tests

Both tests are blocked by the same ConfigurationError as
test_aggregate_reserved_options, so they cannot fail independently.
---
 test/asynchronous/test_collection.py | 41 ----------------------------
 test/test_collection.py              | 39 --------------------------
 2 files changed, 80 deletions(-)

diff --git a/test/asynchronous/test_collection.py b/test/asynchronous/test_collection.py
index af6dd4bf3b..beab868640 100644
--- a/test/asynchronous/test_collection.py
+++ b/test/asynchronous/test_collection.py
@@ -1624,47 +1624,6 @@ async def test_aggregate_reserved_options_do_not_reach_server(self):
                 )
                 self.assertEqual(leaked, [], f"{name} returned another collection's documents")
 
-    @async_client_context.require_version_min(4, 4, -1)
-    async def test_pipeline_option_cannot_read_another_collection(self):
-        # The pipeline option alone is a distinct primitive: it needs no
-        # aggregate key, so the command still names the intended collection
-        # while $unionWith pulls in another one. Requires $unionWith (4.4+).
-        db = self.db
-        await db.drop_collection("secrets")
-        self.addAsyncCleanup(db.drop_collection, "secrets")
-        await db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
-
-        leaked = []
-        with contextlib.suppress(ConfigurationError, OperationFailure):
-            leaked = await (
-                await db.test.list_search_indexes(pipeline=[{"$unionWith": "secrets"}])
-            ).to_list()
-        self.assertNotIn(
-            "sentinel",
-            [doc.get("api_key") for doc in leaked],
-            "injected $unionWith read another collection",
-        )
-
-    async def test_pipeline_option_cannot_write_another_collection(self):
-        # The injected pipeline also evades the $out/$merge write detection,
-        # which inspects only the legitimate pipeline argument. Assert the
-        # target collection is left untouched.
-        db = self.db
-        await db.drop_collection("secrets")
-        await db.drop_collection("billing")
-        self.addAsyncCleanup(db.drop_collection, "secrets")
-        self.addAsyncCleanup(db.drop_collection, "billing")
-        await db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
-        await db.billing.insert_one({"_id": 1, "balance": 100})
-
-        with contextlib.suppress(ConfigurationError, OperationFailure):
-            await db.test.list_search_indexes(pipeline=[{"$match": {}}, {"$out": "billing"}])
-        self.assertEqual(
-            await db.billing.find().to_list(),
-            [{"_id": 1, "balance": 100}],
-            "injected $out overwrote another collection",
-        )
-
     async def test_aggregate_raw_bson(self):
         db = self.db
         await db.drop_collection("test")
diff --git a/test/test_collection.py b/test/test_collection.py
index bb154a1f2f..5564deccb1 100644
--- a/test/test_collection.py
+++ b/test/test_collection.py
@@ -1606,45 +1606,6 @@ def test_aggregate_reserved_options_do_not_reach_server(self):
                 )
                 self.assertEqual(leaked, [], f"{name} returned another collection's documents")
 
-    @client_context.require_version_min(4, 4, -1)
-    def test_pipeline_option_cannot_read_another_collection(self):
-        # The pipeline option alone is a distinct primitive: it needs no
-        # aggregate key, so the command still names the intended collection
-        # while $unionWith pulls in another one. Requires $unionWith (4.4+).
-        db = self.db
-        db.drop_collection("secrets")
-        self.addCleanup(db.drop_collection, "secrets")
-        db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
-
-        leaked = []
-        with contextlib.suppress(ConfigurationError, OperationFailure):
-            leaked = (db.test.list_search_indexes(pipeline=[{"$unionWith": "secrets"}])).to_list()
-        self.assertNotIn(
-            "sentinel",
-            [doc.get("api_key") for doc in leaked],
-            "injected $unionWith read another collection",
-        )
-
-    def test_pipeline_option_cannot_write_another_collection(self):
-        # The injected pipeline also evades the $out/$merge write detection,
-        # which inspects only the legitimate pipeline argument. Assert the
-        # target collection is left untouched.
-        db = self.db
-        db.drop_collection("secrets")
-        db.drop_collection("billing")
-        self.addCleanup(db.drop_collection, "secrets")
-        self.addCleanup(db.drop_collection, "billing")
-        db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
-        db.billing.insert_one({"_id": 1, "balance": 100})
-
-        with contextlib.suppress(ConfigurationError, OperationFailure):
-            db.test.list_search_indexes(pipeline=[{"$match": {}}, {"$out": "billing"}])
-        self.assertEqual(
-            db.billing.find().to_list(),
-            [{"_id": 1, "balance": 100}],
-            "injected $out overwrote another collection",
-        )
-
     def test_aggregate_raw_bson(self):
         db = self.db
         db.drop_collection("test")

From 1dd3dc2025c4cb9d37ec8f25170d62c7ffac26d5 Mon Sep 17 00:00:00 2001
From: Jeffrey 'Alex' Clark <alex.clark@mongodb.com>
Date: Mon, 17 Aug 2026 18:27:31 -0400
Subject: [PATCH 4/9] PYTHON-5981 Trim redundant test comments

---
 test/asynchronous/test_collection.py | 8 ++------
 test/test_collection.py              | 8 ++------
 2 files changed, 4 insertions(+), 12 deletions(-)

diff --git a/test/asynchronous/test_collection.py b/test/asynchronous/test_collection.py
index beab868640..f965a5adf4 100644
--- a/test/asynchronous/test_collection.py
+++ b/test/asynchronous/test_collection.py
@@ -1569,9 +1569,7 @@ async def test_aggregate(self):
             await coll.aggregate([{"$out": "output-collection"}])
 
     async def test_aggregate_reserved_options(self):
-        # The reserved command fields must not be settable as options: an
-        # application that forwards a caller-supplied options mapping would
-        # otherwise let the caller retarget the command at any collection.
+        # Reserved command fields must not be settable as options.
         db = self.db
         with self.assertRaises(ConfigurationError):
             await db.test.aggregate([], aggregate="other")
@@ -1587,9 +1585,7 @@ async def test_aggregate_reserved_options(self):
     async def test_aggregate_reserved_options_do_not_reach_server(self):
         # Assert the security property rather than the error type: the injected
         # namespace must never reach the wire, and the secret must never reach
-        # the caller. Checking the command as sent keeps this independent of
-        # whether the server would have accepted the injected pipeline, so it
-        # fails on a vulnerable driver for the right reason on any topology.
+        # the caller.
         listener = OvertCommandListener()
         client = await self.async_single_client(event_listeners=[listener])
         db = client[self.db.name]
diff --git a/test/test_collection.py b/test/test_collection.py
index 5564deccb1..d13caa38b4 100644
--- a/test/test_collection.py
+++ b/test/test_collection.py
@@ -1551,9 +1551,7 @@ def test_aggregate(self):
             coll.aggregate([{"$out": "output-collection"}])
 
     def test_aggregate_reserved_options(self):
-        # The reserved command fields must not be settable as options: an
-        # application that forwards a caller-supplied options mapping would
-        # otherwise let the caller retarget the command at any collection.
+        # Reserved command fields must not be settable as options.
         db = self.db
         with self.assertRaises(ConfigurationError):
             db.test.aggregate([], aggregate="other")
@@ -1569,9 +1567,7 @@ def test_aggregate_reserved_options(self):
     def test_aggregate_reserved_options_do_not_reach_server(self):
         # Assert the security property rather than the error type: the injected
         # namespace must never reach the wire, and the secret must never reach
-        # the caller. Checking the command as sent keeps this independent of
-        # whether the server would have accepted the injected pipeline, so it
-        # fails on a vulnerable driver for the right reason on any topology.
+        # the caller.
         listener = OvertCommandListener()
         client = self.single_client(event_listeners=[listener])
         db = client[self.db.name]

From 7cf834f0921a08f59416711f64f6a4a751de893a Mon Sep 17 00:00:00 2001
From: Jeffrey 'Alex' Clark <alex.clark@mongodb.com>
Date: Tue, 18 Aug 2026 09:27:03 -0400
Subject: [PATCH 5/9] PYTHON-5981 Address review feedback

---
 doc/changelog.rst                    | 13 ++++++----
 test/asynchronous/test_collection.py | 38 ----------------------------
 test/test_collection.py              | 38 ----------------------------
 3 files changed, 8 insertions(+), 81 deletions(-)

diff --git a/doc/changelog.rst b/doc/changelog.rst
index afdf7d5adf..e752f9bcb2 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -42,12 +42,15 @@ PyMongo 4.18 brings a number of changes including:
 - Aggregation helpers now raise :exc:`~pymongo.errors.ConfigurationError` when
   passed an ``aggregate`` or ``pipeline`` keyword argument. Previously these
   keys silently replaced the target namespace and pipeline of the generated
-  ``aggregate`` command, so an application forwarding untrusted option names to
-  :meth:`~pymongo.asynchronous.collection.AsyncCollection.aggregate`,
-  :meth:`~pymongo.asynchronous.collection.AsyncCollection.aggregate_raw_batches`,
-  :meth:`~pymongo.asynchronous.database.AsyncDatabase.aggregate`, or
+  ``aggregate`` command. This affects
+  :meth:`~pymongo.asynchronous.collection.AsyncCollection.aggregate` and
+  :meth:`~pymongo.synchronous.collection.Collection.aggregate`,
+  :meth:`~pymongo.asynchronous.collection.AsyncCollection.aggregate_raw_batches`
+  and :meth:`~pymongo.synchronous.collection.Collection.aggregate_raw_batches`,
+  :meth:`~pymongo.asynchronous.database.AsyncDatabase.aggregate` and
+  :meth:`~pymongo.synchronous.database.Database.aggregate`, and
   :meth:`~pymongo.asynchronous.collection.AsyncCollection.list_search_indexes`
-  could be redirected to read or overwrite an unintended collection.
+  and :meth:`~pymongo.synchronous.collection.Collection.list_search_indexes`.
 
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------
diff --git a/test/asynchronous/test_collection.py b/test/asynchronous/test_collection.py
index f965a5adf4..5fd4542d36 100644
--- a/test/asynchronous/test_collection.py
+++ b/test/asynchronous/test_collection.py
@@ -1582,44 +1582,6 @@ async def test_aggregate_reserved_options(self):
         with self.assertRaises(ConfigurationError):
             await db.test.list_search_indexes(pipeline=[{"$out": "other"}])
 
-    async def test_aggregate_reserved_options_do_not_reach_server(self):
-        # Assert the security property rather than the error type: the injected
-        # namespace must never reach the wire, and the secret must never reach
-        # the caller.
-        listener = OvertCommandListener()
-        client = await self.async_single_client(event_listeners=[listener])
-        db = client[self.db.name]
-        await self.db.drop_collection("secrets")
-        self.addAsyncCleanup(self.db.drop_collection, "secrets")
-        await self.db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
-
-        attempts = {
-            "aggregate": lambda: db.test.aggregate([], aggregate="secrets"),
-            "aggregate_raw_batches": lambda: db.test.aggregate_raw_batches([], aggregate="secrets"),
-            "database aggregate": lambda: db.aggregate([], aggregate="secrets"),
-            "list_search_indexes aggregate": lambda: db.test.list_search_indexes(
-                aggregate="secrets"
-            ),
-        }
-        for name, attempt in attempts.items():
-            with self.subTest(entry_point=name):
-                listener.reset()
-                leaked = []
-                # A vulnerable driver raises nothing; a server that rejects the
-                # injected pipeline raises OperationFailure. Neither outcome is
-                # what this test asserts on, so both are tolerated here.
-                with contextlib.suppress(ConfigurationError, OperationFailure):
-                    leaked = await (await attempt()).to_list()
-                targets = [
-                    event.command.get("aggregate")
-                    for event in listener.started_events
-                    if event.command_name == "aggregate"
-                ]
-                self.assertNotIn(
-                    "secrets", targets, f"{name} sent the injected namespace to the server"
-                )
-                self.assertEqual(leaked, [], f"{name} returned another collection's documents")
-
     async def test_aggregate_raw_bson(self):
         db = self.db
         await db.drop_collection("test")
diff --git a/test/test_collection.py b/test/test_collection.py
index d13caa38b4..2bf32b8bf4 100644
--- a/test/test_collection.py
+++ b/test/test_collection.py
@@ -1564,44 +1564,6 @@ def test_aggregate_reserved_options(self):
         with self.assertRaises(ConfigurationError):
             db.test.list_search_indexes(pipeline=[{"$out": "other"}])
 
-    def test_aggregate_reserved_options_do_not_reach_server(self):
-        # Assert the security property rather than the error type: the injected
-        # namespace must never reach the wire, and the secret must never reach
-        # the caller.
-        listener = OvertCommandListener()
-        client = self.single_client(event_listeners=[listener])
-        db = client[self.db.name]
-        self.db.drop_collection("secrets")
-        self.addCleanup(self.db.drop_collection, "secrets")
-        self.db.secrets.insert_one({"_id": 1, "api_key": "sentinel"})
-
-        attempts = {
-            "aggregate": lambda: db.test.aggregate([], aggregate="secrets"),
-            "aggregate_raw_batches": lambda: db.test.aggregate_raw_batches([], aggregate="secrets"),
-            "database aggregate": lambda: db.aggregate([], aggregate="secrets"),
-            "list_search_indexes aggregate": lambda: db.test.list_search_indexes(
-                aggregate="secrets"
-            ),
-        }
-        for name, attempt in attempts.items():
-            with self.subTest(entry_point=name):
-                listener.reset()
-                leaked = []
-                # A vulnerable driver raises nothing; a server that rejects the
-                # injected pipeline raises OperationFailure. Neither outcome is
-                # what this test asserts on, so both are tolerated here.
-                with contextlib.suppress(ConfigurationError, OperationFailure):
-                    leaked = (attempt()).to_list()
-                targets = [
-                    event.command.get("aggregate")
-                    for event in listener.started_events
-                    if event.command_name == "aggregate"
-                ]
-                self.assertNotIn(
-                    "secrets", targets, f"{name} sent the injected namespace to the server"
-                )
-                self.assertEqual(leaked, [], f"{name} returned another collection's documents")
-
     def test_aggregate_raw_bson(self):
         db = self.db
         db.drop_collection("test")

From 5a8f894a1f732645cc0066ad93fcd61236147642 Mon Sep 17 00:00:00 2001
From: Jeffrey 'Alex' Clark <aclark@aclark.net>
Date: Tue, 18 Aug 2026 11:32:48 -0400
Subject: [PATCH 6/9] Update pymongo/asynchronous/aggregation.py

Co-authored-by: Noah Stapp <noah.stapp@mongodb.com>
---
 pymongo/asynchronous/aggregation.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pymongo/asynchronous/aggregation.py b/pymongo/asynchronous/aggregation.py
index 8b2331fc5c..5f70e84fef 100644
--- a/pymongo/asynchronous/aggregation.py
+++ b/pymongo/asynchronous/aggregation.py
@@ -63,7 +63,7 @@ def __init__(
         for name in ("aggregate", "pipeline"):
             if name in options:
                 raise ConfigurationError(
-                    f"The {name} option cannot be specified as a keyword argument"
+                    f"The {name} option cannot be specified as a keyword argument."
                 )
 
         self._target = target

From 09fd11e61f6902532ced48c01f1ec0f1d092410a Mon Sep 17 00:00:00 2001
From: Jeffrey 'Alex' Clark <alex.clark@mongodb.com>
Date: Tue, 18 Aug 2026 11:48:40 -0400
Subject: [PATCH 7/9] PYTHON-5981 Clarify reserved options test comment

---
 test/asynchronous/test_collection.py | 4 +++-
 test/test_collection.py              | 4 +++-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/test/asynchronous/test_collection.py b/test/asynchronous/test_collection.py
index 5fd4542d36..cb072025dd 100644
--- a/test/asynchronous/test_collection.py
+++ b/test/asynchronous/test_collection.py
@@ -1569,7 +1569,9 @@ async def test_aggregate(self):
             await coll.aggregate([{"$out": "output-collection"}])
 
     async def test_aggregate_reserved_options(self):
-        # Reserved command fields must not be settable as options.
+        # "aggregate" and "pipeline" are fields of the aggregate command itself,
+        # so they must not be settable as keyword options: doing so would
+        # replace the command's target namespace or its pipeline.
         db = self.db
         with self.assertRaises(ConfigurationError):
             await db.test.aggregate([], aggregate="other")
diff --git a/test/test_collection.py b/test/test_collection.py
index 2bf32b8bf4..0230caeb65 100644
--- a/test/test_collection.py
+++ b/test/test_collection.py
@@ -1551,7 +1551,9 @@ def test_aggregate(self):
             coll.aggregate([{"$out": "output-collection"}])
 
     def test_aggregate_reserved_options(self):
-        # Reserved command fields must not be settable as options.
+        # "aggregate" and "pipeline" are fields of the aggregate command itself,
+        # so they must not be settable as keyword options: doing so would
+        # replace the command's target namespace or its pipeline.
         db = self.db
         with self.assertRaises(ConfigurationError):
             db.test.aggregate([], aggregate="other")

From f7f118021eb02c02a4861496769ee264dded4fba Mon Sep 17 00:00:00 2001
From: Jeffrey 'Alex' Clark <alex.clark@mongodb.com>
Date: Tue, 18 Aug 2026 11:49:31 -0400
Subject: [PATCH 8/9] PYTHON-5981 Sync aggregation error message

---
 pymongo/synchronous/aggregation.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pymongo/synchronous/aggregation.py b/pymongo/synchronous/aggregation.py
index 9cdd32952d..7e13afa8ec 100644
--- a/pymongo/synchronous/aggregation.py
+++ b/pymongo/synchronous/aggregation.py
@@ -63,7 +63,7 @@ def __init__(
         for name in ("aggregate", "pipeline"):
             if name in options:
                 raise ConfigurationError(
-                    f"The {name} option cannot be specified as a keyword argument"
+                    f"The {name} option cannot be specified as a keyword argument."
                 )
 
         self._target = target

From 33cb231e20000d5fa7325c468e608b37aca8683c Mon Sep 17 00:00:00 2001
From: Jeffrey 'Alex' Clark <alex.clark@mongodb.com>
Date: Wed, 19 Aug 2026 20:30:36 -0400
Subject: [PATCH 9/9] PYTHON-5981 Parametrize reserved options test

---
 test/asynchronous/test_collection.py | 28 ++++++++++++++++++----------
 test/test_collection.py              | 28 ++++++++++++++++++----------
 2 files changed, 36 insertions(+), 20 deletions(-)

diff --git a/test/asynchronous/test_collection.py b/test/asynchronous/test_collection.py
index cb072025dd..813c46d778 100644
--- a/test/asynchronous/test_collection.py
+++ b/test/asynchronous/test_collection.py
@@ -1573,16 +1573,24 @@ async def test_aggregate_reserved_options(self):
         # so they must not be settable as keyword options: doing so would
         # replace the command's target namespace or its pipeline.
         db = self.db
-        with self.assertRaises(ConfigurationError):
-            await db.test.aggregate([], aggregate="other")
-        with self.assertRaises(ConfigurationError):
-            await db.test.aggregate_raw_batches([], aggregate="other")
-        with self.assertRaises(ConfigurationError):
-            await db.aggregate([], aggregate="other")
-        with self.assertRaises(ConfigurationError):
-            await db.test.list_search_indexes(aggregate="other")
-        with self.assertRaises(ConfigurationError):
-            await db.test.list_search_indexes(pipeline=[{"$out": "other"}])
+        reserved_options: list[dict[str, Any]] = [
+            {"aggregate": "other"},
+            {"pipeline": [{"$out": "other"}]},
+            {"aggregate": "other", "pipeline": [{"$out": "other"}]},
+        ]
+        for options in reserved_options:
+            with self.subTest(options=options):
+                # These helpers take the pipeline positionally, so only pass
+                # the options that do not collide with it.
+                if "pipeline" not in options:
+                    with self.assertRaises(ConfigurationError):
+                        await db.test.aggregate([], **options)
+                    with self.assertRaises(ConfigurationError):
+                        await db.test.aggregate_raw_batches([], **options)
+                    with self.assertRaises(ConfigurationError):
+                        await db.aggregate([], **options)
+                with self.assertRaises(ConfigurationError):
+                    await db.test.list_search_indexes(**options)
 
     async def test_aggregate_raw_bson(self):
         db = self.db
diff --git a/test/test_collection.py b/test/test_collection.py
index 0230caeb65..d2dd02cf41 100644
--- a/test/test_collection.py
+++ b/test/test_collection.py
@@ -1555,16 +1555,24 @@ def test_aggregate_reserved_options(self):
         # so they must not be settable as keyword options: doing so would
         # replace the command's target namespace or its pipeline.
         db = self.db
-        with self.assertRaises(ConfigurationError):
-            db.test.aggregate([], aggregate="other")
-        with self.assertRaises(ConfigurationError):
-            db.test.aggregate_raw_batches([], aggregate="other")
-        with self.assertRaises(ConfigurationError):
-            db.aggregate([], aggregate="other")
-        with self.assertRaises(ConfigurationError):
-            db.test.list_search_indexes(aggregate="other")
-        with self.assertRaises(ConfigurationError):
-            db.test.list_search_indexes(pipeline=[{"$out": "other"}])
+        reserved_options: list[dict[str, Any]] = [
+            {"aggregate": "other"},
+            {"pipeline": [{"$out": "other"}]},
+            {"aggregate": "other", "pipeline": [{"$out": "other"}]},
+        ]
+        for options in reserved_options:
+            with self.subTest(options=options):
+                # These helpers take the pipeline positionally, so only pass
+                # the options that do not collide with it.
+                if "pipeline" not in options:
+                    with self.assertRaises(ConfigurationError):
+                        db.test.aggregate([], **options)
+                    with self.assertRaises(ConfigurationError):
+                        db.test.aggregate_raw_batches([], **options)
+                    with self.assertRaises(ConfigurationError):
+                        db.aggregate([], **options)
+                with self.assertRaises(ConfigurationError):
+                    db.test.list_search_indexes(**options)
 
     def test_aggregate_raw_bson(self):
         db = self.db