Implement ssl_require parameter mapping for various database backends

Author: aminechm

dj_database_url/__init__.py

@@ -170,7 +170,6 @@ def parse(
         conn_max_age,
         conn_health_checks,
         disable_server_side_cursors,
-        ssl_require,
         test_options,
     )
 
@@ -211,6 +210,9 @@ def parse(
     parsed_config["OPTIONS"].update(settings.pop("OPTIONS", {}))
     parsed_config.update(settings)
 
+    if ssl_require:
+        _configure_ssl(parsed_config)
+
     if not parsed_config["OPTIONS"]:
         parsed_config.pop("OPTIONS")
     return parsed_config
@@ -229,12 +231,43 @@ def _parse_value(value: str) -> OptionType:
     return value
 
 
+def _configure_ssl(parsed_config: DBConfig) -> None:
+    assert "OPTIONS" in parsed_config
+    options = parsed_config["OPTIONS"]
+    assert "ENGINE" in parsed_config
+    backend = parsed_config["ENGINE"]
+
+    if backend in (
+        "django.db.backends.mysql",
+        "django.contrib.gis.db.backends.mysql",
+    ):
+        options["ssl_mode"] = "REQUIRED"
+    elif backend in (
+        "django.db.backends.postgresql",
+        "django.contrib.gis.db.backends.postgis",
+        "django_redshift_backend",
+        "django_cockroachdb",
+        "timescale.db.backends.postgresql",
+        "timescale.db.backends.postgis",
+    ):
+        options["sslmode"] = "require"
+    elif backend in (
+        "mssql",
+        "sql_server.pyodbc",
+    ):
+        current_extra = options.get("extra_params", "")
+        if "Encrypt=yes" not in current_extra:
+            if current_extra:
+                options["extra_params"] = f"{current_extra};Encrypt=yes"
+            else:
+                options["extra_params"] = "Encrypt=yes"
+
+
 def _convert_to_settings(
     engine: Optional[str],
     conn_max_age: Optional[int],
     conn_health_checks: bool,
     disable_server_side_cursors: bool,
-    ssl_require: bool,
     test_options: Optional[dict[str, Any]],
 ) -> DBConfig:
     settings: DBConfig = {
@@ -244,9 +277,6 @@ def _convert_to_settings(
     }
     if engine:
         settings["ENGINE"] = engine
-    if ssl_require:
-        settings["OPTIONS"] = {}
-        settings["OPTIONS"]["sslmode"] = "require"
     if test_options:
         settings["TEST"] = test_options
     return settings

tests/test_dj_database_url.py

@@ -673,10 +673,55 @@ def test_register_multiple_times__no_duplicates_in_uses_netloc(self) -> None:
         os.environ,
         {"DATABASE_URL": "postgres://user:password@instance.amazonaws.com:5431/d8r8?"},
     )
-    def test_ssl_require(self) -> None:
+    def test_ssl_require_postgres(self) -> None:
         url = dj_database_url.config(ssl_require=True)
         assert url["OPTIONS"] == {'sslmode': 'require'}
 
+    @mock.patch.dict(
+        os.environ,
+        {"DATABASE_URL": "mysql://user:password@instance.amazonaws.com:3306/dbname"},
+    )
+    def test_ssl_require_mysql(self) -> None:
+        url = dj_database_url.config(ssl_require=True)
+        assert url["OPTIONS"] == {"ssl_mode": "REQUIRED"}
+
+    @mock.patch.dict(
+        os.environ,
+        {"DATABASE_URL": "mssqlms://user:password@instance.amazonaws.com:1234/dbname"},
+    )
+    def test_ssl_require_mssql(self) -> None:
+        url = dj_database_url.config(ssl_require=True)
+        assert url["OPTIONS"] == {"extra_params": "Encrypt=yes"}
+
+    @mock.patch.dict(
+        os.environ,
+        {
+            "DATABASE_URL": (
+                "mssql://user:password@instance.amazonaws.com:1234/dbname"
+                "?extra_params=TrustServerCertificate=yes"
+            )
+        },
+    )
+    def test_ssl_require_mssql_existing_extra_params(self) -> None:
+        url = dj_database_url.config(ssl_require=True)
+        assert url["OPTIONS"] == {
+            "extra_params": "TrustServerCertificate=yes;Encrypt=yes"
+        }
+
+    @mock.patch.dict(
+        os.environ,
+        {
+            "DATABASE_URL": (
+                "mssql://user:password@instance.amazonaws.com:1234/dbname"
+                "?extra_params=Encrypt=yes"
+            )
+        },
+    )
+    def test_ssl_require_mssql_already_encrypted(self) -> None:
+        url = dj_database_url.config(ssl_require=True)
+        # Should NOT append Encrypt=yes again
+        assert url["OPTIONS"] == {"extra_params": "Encrypt=yes"}
+
     def test_options_int_values(self) -> None:
         """Ensure that options with integer values are parsed correctly."""
         url = dj_database_url.parse(