raise config error if no ae options + use other db

Author: aclark4life

django_mongodb_backend/encryption.py

@@ -44,14 +44,12 @@
 class EncryptedRouter:
     def allow_migrate(self, db, app_label, model_name=None, model=None, **hints):
         if model:
-            return db == (
-                "my_encrypted_database" if getattr(model, "encrypted", False) else "default"
-            )
+            return db == ("other" if getattr(model, "encrypted", False) else "default")
         return db == "default"
 
     def db_for_read(self, model, **hints):
         if getattr(model, "encrypted", False):
-            return "my_encrypted_database"
+            return "other"
         return "default"
 
     db_for_write = db_for_read

django_mongodb_backend/schema.py

@@ -1,3 +1,4 @@
+from django.core.exceptions import ImproperlyConfigured
 from django.db import router
 from django.db.backends.base.schema import BaseDatabaseSchemaEditor
 from django.db.models import Index, UniqueConstraint
@@ -454,6 +455,12 @@ def _create_collection(self, model):
                         provider,
                         credentials,
                     )
+            else:
+                raise ImproperlyConfigured(
+                    "The model has `encrypted=True`, but the connection does not have "
+                    "auto encryption options set. Please set `auto_encryption_opts` "
+                    "in the connection settings."
+                )
         else:
             db.create_collection(db_table)
 

tests/encryption_/routers.py

@@ -10,7 +10,7 @@ def allow_migrate(self, db, app_label, model_name=None, model=None, **hints):
 
     def db_for_read(self, model, **hints):
         if getattr(model, "encrypted", False):
-            return "my_encrypted_database"
+            return "other"
         return None
 
     db_for_write = db_for_read

tests/encryption_/tests.py

@@ -146,7 +146,7 @@ def reload_module(module):
 )
 @override_settings(DATABASE_ROUTERS=[TestEncryptedRouter()])
 class EncryptedFieldTests(TransactionTestCase):
-    databases = {"default", "my_encrypted_database"}
+    databases = {"default", "other"}
     available_apps = ["django_mongodb_backend", "encryption_"]
 
     def setUp(self):
@@ -216,7 +216,7 @@ def tearDownClass(cls):
 
     def test_get_encrypted_fields_map_method(self):
         self.maxDiff = None
-        with connections["my_encrypted_database"].schema_editor() as editor:
+        with connections["other"].schema_editor() as editor:
             db_table = self.patient._meta.db_table
             self.assertCountEqual(
                 {"fields": editor._get_encrypted_fields_map(self.patient)},
@@ -241,7 +241,7 @@ def test_get_encrypted_fields_map_command(self):
         call_command(
             "get_encrypted_fields_map",
             "--database",
-            "my_encrypted_database",
+            "other",
             verbosity=0,
             stdout=out,
         )
@@ -320,6 +320,18 @@ def test_patientrecord(self):
         self.assertFalse(PatientRecord.objects.filter(patient_age__gte=200).exists())
         self.assertTrue(PatientRecord.objects.filter(weight__gte=175.0).exists())
 
+        # Test encrypted patient record in unencrypted database.
+        conn_params = connections["other"].get_connection_params()
+        if conn_params.pop("auto_encryption_opts", False):
+            # Call MongoClient instead of get_new_connection because
+            # get_new_connection will return the encrypted connection
+            # from the connection pool.
+            connection = pymongo.MongoClient(**conn_params)
+            patientrecords = connection["test_other"].patientrecord.find()
+            ssn = patientrecords[0]["ssn"]
+            self.assertTrue(isinstance(ssn, Binary))
+            connection.close()
+
     def test_patient(self):
         self.assertEqual(
             Patient.objects.get(patient_notes="patient notes " * 25).patient_notes,
@@ -337,22 +349,24 @@ def test_patient(self):
         )
 
         # Test decrypted patient record in encrypted database.
-        patients = connections["my_encrypted_database"].database.patient.find()
+        patients = connections["other"].database.patient.find()
         self.assertEqual(len(list(patients)), 1)
-        records = connections["my_encrypted_database"].database.patientrecord.find()
+        records = connections["other"].database.patientrecord.find()
         self.assertTrue("__safeContent__" in records[0])
 
-        # Test encrypted patient record in unencrypted database.
-        conn_params = connections["my_encrypted_database"].get_connection_params()
-        if conn_params.pop("auto_encryption_opts", False):
-            # Call MongoClient instead of get_new_connection because
-            # get_new_connection will return the encrypted connection
-            # from the connection pool.
-            connection = pymongo.MongoClient(**conn_params)
-            patientrecords = connection["test_my_encrypted_database"].patientrecord.find()
-            ssn = patientrecords[0]["ssn"]
-            self.assertTrue(isinstance(ssn, Binary))
-            connection.close()
+    def test_no_auto_encryption_opts(self):
+        """
+        Ensure that the encrypted fields map is not set in the client
+        when auto_encryption_opts is not set.
+        """
+        Patient.objects.using("default").create(
+            patient_id=2,
+            patient_name="John Doe 2",
+            patient_notes="patient notes " * 20,
+            registration_date=datetime(2024, 10, 1, 12, 0, 0),
+            is_active=True,
+            email="[email protected]",
+        ).save()
 
 
 class KMSCredentialsTests(TestCase):