PYTHON-5786 Add test variant for the Secure Frontend Processor (monguard)

.evergreen/generated_configs/variants.yml

@@ -564,6 +564,17 @@ buildvariants:
       VERSION: latest
     tags: [coverage_tag]
 
+  # Sfp tests
+  - name: sfp-rhel8
+    tasks:
+      - name: .test-no-orchestration
+    display_name: SFP RHEL8
+    run_on:
+      - rhel87-small
+    expansions:
+      TEST_NAME: sfp
+    tags: [pr]
+
   # Stable api tests
   - name: stable-api-require-v1-rhel8-auth
     tasks:

.evergreen/scripts/generate_config.py

@@ -464,6 +464,19 @@ def create_atlas_connect_variants():
     ]
 
 
+def create_sfp_variants():
+    host = DEFAULT_HOST
+    return [
+        create_variant(
+            [".test-no-orchestration"],
+            get_variant_name("SFP", host),
+            tags=["pr"],
+            host=host,
+            expansions=dict(TEST_NAME="sfp"),
+        )
+    ]
+
+
 def create_coverage_report_variants():
     return [create_variant(["coverage-report"], "Coverage Report", host=DEFAULT_HOST)]
 

.evergreen/scripts/setup_tests.py

@@ -436,6 +436,19 @@ def handle_test_env() -> None:
         # We do not want the default client_context to be initialized.
         write_env("DISABLE_CONTEXT")
 
+    if test_name == "sfp":
+        secrets = get_secrets("drivers/sfp")
+
+        # Write file with SFP Atlas X509 client certificate:
+        decoded = base64.b64decode(secrets["SFP_ATLAS_X509_BASE64"]).decode("utf8")
+        cert_file = ROOT / ".evergreen/atlas_x509_sfp_client_certificate.pem"
+        with cert_file.open("w") as file:
+            file.write(decoded)
+        write_env("SFP_ATLAS_X509_CERT", str(cert_file))
+
+        # We do not want the default client_context to be initialized.
+        write_env("DISABLE_CONTEXT")
+
     if test_name == "numpy":
         UV_ARGS.append("--with numpy")
 

.evergreen/scripts/utils.py

@@ -44,6 +44,7 @@ class Distro:
     "mockupdb": "mockupdb",
     "ocsp": "ocsp",
     "perf": "perf",
+    "sfp": "sfp",
     "numpy": "",
 }
 
@@ -59,6 +60,7 @@ class Distro:
     "aws_lambda",
     "mockupdb",
     "ocsp",
+    "sfp",
 ]
 
 # Mapping of env variables to options

.gitignore

@@ -32,6 +32,7 @@ specifications/
 results.json
 .synchro-modified
 .evergreen/atlas_x509_dev_client_certificate.pem
+.evergreen/atlas_x509_sfp_client_certificate.pem
 
 # Lambda temp files
 test/lambda/.aws-sam

pyproject.toml

@@ -138,6 +138,7 @@ markers = [
     "auth: tests that rely on authentication",
     "ocsp: tests that rely on ocsp",
     "atlas_connect: tests that rely on an atlas connection",
+    "sfp: tests that rely on an Atlas Secure Frontend Processor connection",
     "perf: benchmark tests",
     "search_index: search index helper tests",
     "kms: client-side field-level encryption tests using kms",

test/asynchronous/test_sfp.py

@@ -0,0 +1,111 @@
+# Copyright 2026-present MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test connectivity and authentication through an Atlas Secure Frontend Processor (SFP/Monguard)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from typing import Any
+
+import pytest
+
+sys.path[0:0] = [""]
+
+from bson import ObjectId
+from pymongo.asynchronous.mongo_client import AsyncMongoClient
+from pymongo.server_api import ServerApi
+from test.asynchronous import AsyncPyMongoTestCase
+
+_IS_SYNC = False
+
+pytestmark = pytest.mark.sfp
+
+# Each authenticated test must run under each of these variations:
+# no additional configuration, a compressor enabled, and Server API v1.
+VARIATIONS: dict[str, dict[str, Any]] = {
+    "baseline": {},
+    "compressor": {"compressors": "zlib"},
+    "server_api": {"server_api": ServerApi("1")},
+}
+
+
+def _require_env(name: str) -> str:
+    value = os.environ.get(name)
+    if not value:
+        raise Exception(f"Must set {name} env variable to test.")
+    return value
+
+
+class TestAtlasSFP(AsyncPyMongoTestCase):
+    async def assert_ping(self, client: AsyncMongoClient) -> None:
+        response = await client.admin.command("ping")
+        self.assertEqual(response["ok"], 1)
+
+    async def assert_connection_status(self, client: AsyncMongoClient, authenticated: bool) -> None:
+        response = await client.admin.command("connectionStatus")
+        self.assertEqual(response["ok"], 1)
+        users = response["authInfo"]["authenticatedUsers"]
+        if authenticated:
+            self.assertGreaterEqual(len(users), 1)
+        else:
+            self.assertEqual(users, [])
+
+    async def assert_crud(self, client: AsyncMongoClient) -> None:
+        # Use a unique collection name for each test run and drop it
+        # afterward, regardless of test success or failure.
+        collection = client.db[f"sfp_test_{ObjectId()}"]
+        self.addAsyncCleanup(collection.drop)
+        result = await collection.insert_one({"_id": 0})
+        self.assertEqual(result.inserted_id, 0)
+        document = await collection.find_one({"_id": 0})
+        self.assertEqual(document, {"_id": 0})
+
+    async def test_unauthenticated(self):
+        client = self.simple_client(_require_env("SFP_ATLAS_URI"))
+        await self.assert_ping(client)
+        await self.assert_connection_status(client, authenticated=False)
+
+    async def test_scram_sha_256(self):
+        uri = _require_env("SFP_ATLAS_URI")
+        username = _require_env("SFP_ATLAS_USER")
+        password = _require_env("SFP_ATLAS_PASSWORD")
+        for variation, kwargs in VARIATIONS.items():
+            with self.subTest(variation=variation):
+                client = self.simple_client(
+                    uri,
+                    username=username,
+                    password=password,
+                    authMechanism="SCRAM-SHA-256",
+                    **kwargs,
+                )
+                await self.assert_ping(client)
+                await self.assert_connection_status(client, authenticated=True)
+                await self.assert_crud(client)
+
+    async def test_x509(self):
+        uri = _require_env("SFP_ATLAS_X509_URI")
+        cert = _require_env("SFP_ATLAS_X509_CERT")
+        for variation, kwargs in VARIATIONS.items():
+            with self.subTest(variation=variation):
+                client = self.simple_client(uri, tlsCertificateKeyFile=cert, **kwargs)
+                await self.assert_ping(client)
+                await self.assert_connection_status(client, authenticated=True)
+                await self.assert_crud(client)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_sfp.py

@@ -0,0 +1,111 @@
+# Copyright 2026-present MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test connectivity and authentication through an Atlas Secure Frontend Processor (SFP/Monguard)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from typing import Any
+
+import pytest
+
+sys.path[0:0] = [""]
+
+from bson import ObjectId
+from pymongo.server_api import ServerApi
+from pymongo.synchronous.mongo_client import MongoClient
+from test import PyMongoTestCase
+
+_IS_SYNC = True
+
+pytestmark = pytest.mark.sfp
+
+# Each authenticated test must run under each of these variations:
+# no additional configuration, a compressor enabled, and Server API v1.
+VARIATIONS: dict[str, dict[str, Any]] = {
+    "baseline": {},
+    "compressor": {"compressors": "zlib"},
+    "server_api": {"server_api": ServerApi("1")},
+}
+
+
+def _require_env(name: str) -> str:
+    value = os.environ.get(name)
+    if not value:
+        raise Exception(f"Must set {name} env variable to test.")
+    return value
+
+
+class TestAtlasSFP(PyMongoTestCase):
+    def assert_ping(self, client: MongoClient) -> None:
+        response = client.admin.command("ping")
+        self.assertEqual(response["ok"], 1)
+
+    def assert_connection_status(self, client: MongoClient, authenticated: bool) -> None:
+        response = client.admin.command("connectionStatus")
+        self.assertEqual(response["ok"], 1)
+        users = response["authInfo"]["authenticatedUsers"]
+        if authenticated:
+            self.assertGreaterEqual(len(users), 1)
+        else:
+            self.assertEqual(users, [])
+
+    def assert_crud(self, client: MongoClient) -> None:
+        # Use a unique collection name for each test run and drop it
+        # afterward, regardless of test success or failure.
+        collection = client.db[f"sfp_test_{ObjectId()}"]
+        self.addCleanup(collection.drop)
+        result = collection.insert_one({"_id": 0})
+        self.assertEqual(result.inserted_id, 0)
+        document = collection.find_one({"_id": 0})
+        self.assertEqual(document, {"_id": 0})
+
+    def test_unauthenticated(self):
+        client = self.simple_client(_require_env("SFP_ATLAS_URI"))
+        self.assert_ping(client)
+        self.assert_connection_status(client, authenticated=False)
+
+    def test_scram_sha_256(self):
+        uri = _require_env("SFP_ATLAS_URI")
+        username = _require_env("SFP_ATLAS_USER")
+        password = _require_env("SFP_ATLAS_PASSWORD")
+        for variation, kwargs in VARIATIONS.items():
+            with self.subTest(variation=variation):
+                client = self.simple_client(
+                    uri,
+                    username=username,
+                    password=password,
+                    authMechanism="SCRAM-SHA-256",
+                    **kwargs,
+                )
+                self.assert_ping(client)
+                self.assert_connection_status(client, authenticated=True)
+                self.assert_crud(client)
+
+    def test_x509(self):
+        uri = _require_env("SFP_ATLAS_X509_URI")
+        cert = _require_env("SFP_ATLAS_X509_CERT")
+        for variation, kwargs in VARIATIONS.items():
+            with self.subTest(variation=variation):
+                client = self.simple_client(uri, tlsCertificateKeyFile=cert, **kwargs)
+                self.assert_ping(client)
+                self.assert_connection_status(client, authenticated=True)
+                self.assert_crud(client)
+
+
+if __name__ == "__main__":
+    unittest.main()