Database parameters (#3450)

Co-authored-by: Thomas <ThomasLaPiana@users.noreply.github.com>
Co-authored-by: Thomas <thomas.lapiana+github@pm.me>

CHANGELOG.md

@@ -24,6 +24,8 @@ The types of changes are:
 - Add `identity` query param to the consent reporting API view [#3418](https://github.com/ethyca/fides/pull/3418)
 - Added the ability to use custom CAs with Redis via TLS [#3451](https://github.com/ethyca/fides/pull/3451)
 - Add default experience configs on startup [#3449](https://github.com/ethyca/fides/pull/3449)
+- Load default privacy notices on startup [#3401](https://github.com/ethyca/fides/pull/3401)
+- Add ability for users to pass in additional parameters for application database connection [#3450](https://github.com/ethyca/fides/pull/3450)
 - Load default privacy notices on startup [#3401](https://github.com/ethyca/fides/pull/3401/files)
 - Add privacy centre button text customisations [#3432](https://github.com/ethyca/fides/pull/3432)
 - Add privacy centre favicon customisation [#3432](https://github.com/ethyca/fides/pull/3432)

src/fides/api/ctl/database/session.py

@@ -1,3 +1,4 @@
+import ssl
 from typing import AsyncGenerator
 
 from sqlalchemy import create_engine
@@ -7,14 +8,23 @@
 from fides.api.db.session import ExtendedSession
 from fides.core.config import CONFIG
 
+# Associated with a workaround in fides.core.config.database_settings
+# ref: https://github.com/sqlalchemy/sqlalchemy/discussions/5975
+connect_args = {}
+if CONFIG.database.params.get("sslrootcert"):
+    ssl_ctx = ssl.create_default_context(cafile=CONFIG.database.params["sslrootcert"])
+    ssl_ctx.verify_mode = ssl.CERT_REQUIRED
+    connect_args["ssl"] = ssl_ctx
+
 # Parameters are hidden for security
-engine = create_async_engine(
+async_engine = create_async_engine(
     CONFIG.database.async_database_uri,
+    connect_args=connect_args,
     echo=False,
     hide_parameters=not CONFIG.dev_mode,
     logging_name="AsyncEngine",
 )
-async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+async_session = sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
 
 sync_engine = create_engine(
     CONFIG.database.sync_database_uri,

src/fides/core/config/database_settings.py

@@ -2,7 +2,9 @@
 
 # pylint: disable=C0115,C0116, E0213
 
-from typing import Dict, Optional
+from copy import deepcopy
+from typing import Dict, Optional, Union, cast
+from urllib.parse import quote, urlencode
 
 from pydantic import Field, PostgresDsn, validator
 
@@ -67,6 +69,10 @@ class DatabaseSettings(FidesSettings):
         default="defaultuser",
         description="The database user with which to login to the application database.",
     )
+    params: Dict = Field(
+        default={},  # Can't use the default_factory since it breaks docs generation
+        description="Additional connection parameters used when connecting to the applicaiton database.",
+    )
 
     # These must be at the end because they require other values to construct
     sqlalchemy_database_uri: str = Field(
@@ -93,7 +99,7 @@ class DatabaseSettings(FidesSettings):
     @validator("sync_database_uri", pre=True)
     @classmethod
     def assemble_sync_database_uri(
-        cls, value: Optional[str], values: Dict[str, str]
+        cls, value: Optional[str], values: Dict[str, Union[str, Dict]]
     ) -> str:
         """Join DB connection credentials into a connection string"""
         if isinstance(value, str) and value:
@@ -108,19 +114,33 @@ def assemble_sync_database_uri(
                 host=values["server"],
                 port=values.get("port"),
                 path=f"/{db_name or ''}",
+                query=urlencode(
+                    cast(Dict, values["params"]), quote_via=quote, safe="/"
+                ),
             )
         )
 
     @validator("async_database_uri", pre=True)
     @classmethod
     def assemble_async_database_uri(
-        cls, value: Optional[str], values: Dict[str, str]
+        cls, value: Optional[str], values: Dict[str, Union[str, Dict]]
     ) -> str:
         """Join DB connection credentials into an async connection string."""
         if isinstance(value, str) and value:
             return value
 
         db_name = values["test_db"] if get_test_mode() else values["db"]
+
+        # Workaround https://github.com/MagicStack/asyncpg/issues/737
+        # Required due to the unique way in which Asyncpg handles SSL
+        params = cast(Dict, deepcopy(values["params"]))
+        if "sslmode" in params:
+            params["ssl"] = params.pop("sslmode")
+        # This must be constructed in fides.api.ctl.database.session as part of the ssl context
+        # ref: https://github.com/sqlalchemy/sqlalchemy/discussions/5975
+        params.pop("sslrootcert", None)
+        # End workaround
+
         return str(
             PostgresDsn.build(
                 scheme="postgresql+asyncpg",
@@ -129,12 +149,15 @@ def assemble_async_database_uri(
                 host=values["server"],
                 port=values.get("port"),
                 path=f"/{db_name or ''}",
+                query=urlencode(params, quote_via=quote, safe="/"),
             )
         )
 
     @validator("sqlalchemy_database_uri", pre=True)
     @classmethod
-    def assemble_db_connection(cls, v: Optional[str], values: Dict[str, str]) -> str:
+    def assemble_db_connection(
+        cls, v: Optional[str], values: Dict[str, Union[str, Dict]]
+    ) -> str:
         """Join DB connection credentials into a synchronous connection string."""
         if isinstance(v, str) and v:
             return v
@@ -146,13 +169,16 @@ def assemble_db_connection(cls, v: Optional[str], values: Dict[str, str]) -> str
                 host=values["server"],
                 port=values.get("port"),
                 path=f"/{values.get('db') or ''}",
+                query=urlencode(
+                    cast(Dict, values["params"]), quote_via=quote, safe="/"
+                ),
             )
         )
 
     @validator("sqlalchemy_test_database_uri", pre=True)
     @classmethod
     def assemble_test_db_connection(
-        cls, v: Optional[str], values: Dict[str, str]
+        cls, v: Optional[str], values: Dict[str, Union[str, Dict]]
     ) -> str:
         """Join DB connection credentials into a connection string"""
         if isinstance(v, str) and v:
@@ -165,6 +191,9 @@ def assemble_test_db_connection(
                 host=values["server"],
                 port=values["port"],
                 path=f"/{values.get('test_db') or ''}",
+                query=urlencode(
+                    cast(Dict, values["params"]), quote_via=quote, safe="/"
+                ),
             )
         )
 

tests/ctl/core/config/test_config.py

@@ -210,6 +210,7 @@ def test_config_from_path() -> None:
     os.environ,
     {
         "FIDES__DATABASE__SERVER": "envserver",
+        "FIDES__DATABASE__PARAMS": '{"sslmode": "verify-full", "sslrootcert": "/etc/ssl/private/myca.crt"}',
         "FIDES__REDIS__HOST": "envhost",
         **REQUIRED_ENV_VARS,
     },
@@ -221,6 +222,10 @@ def test_overriding_config_from_env_vars() -> None:
     assert config.database.server == "envserver"
     assert config.redis.host == "envhost"
     assert config.security.app_encryption_key == "OLMkv91j8DHiDAULnK5Lxx3kSCov30b3"
+    assert config.database.params == {
+        "sslmode": "verify-full",
+        "sslrootcert": "/etc/ssl/private/myca.crt",
+    }
 
 
 def test_config_app_encryption_key_validation() -> None:
@@ -354,6 +359,40 @@ def test_validating_included_async_database_uri(self) -> None:
         assert incorrect_value not in database_settings.async_database_uri
         assert correct_value in database_settings.async_database_uri
 
+    def test_builds_with_params(self) -> None:
+        """
+        Test that when params are passed, they are correctly
+        encoded as query parameters on the resulting database uris
+        """
+        os.environ["FIDES__TEST_MODE"] = "False"
+        database_settings = DatabaseSettings(
+            user="postgres",
+            password="fides",
+            server="fides-db",
+            port="5432",
+            db="database",
+            params={
+                "sslmode": "verify-full",
+                "sslrootcert": "/etc/ssl/private/myca.crt",
+            },
+        )
+        assert (
+            database_settings.async_database_uri
+            == "postgresql+asyncpg://postgres:fides@fides-db:5432/database?ssl=verify-full"
+            # Q: But why! Where did the sslrootcert parameter go?
+            # A: asyncpg cannot accept it, and an ssl context must be
+            #    passed to the create_async_engine function.
+            # Q: But wait! `ssl` is a different name than what we
+            #    passed in the parameters!
+            # A: That was more of a statement, but Jeopardy rules
+            #    aside, asyncpg has a different set of names
+            #    for these extremely standardized parameter names...
+        )
+        assert (
+            database_settings.sync_database_uri
+            == "postgresql+psycopg2://postgres:fides@fides-db:5432/database?sslmode=verify-full&sslrootcert=/etc/ssl/private/myca.crt"
+        )
+
 
 @pytest.mark.unit
 def test_check_required_webserver_config_values_success(test_config_path: str) -> None: