Merge pull request from GHSA-8cm5-jfj2-26q7

- Further escape any % signs added ('%%') when setting the sqlalchemy URL to avoid invalid interpolation syntax errors.

src/fides/api/db/database.py

@@ -31,7 +31,8 @@ def get_alembic_config(database_url: str) -> Config:
     directory = path.join(migrations_dir, "../alembic/migrations")
     config = Config(path.join(migrations_dir, "../alembic/alembic.ini"))
     config.set_main_option("script_location", directory.replace("%", "%%"))
-    config.set_main_option("sqlalchemy.url", database_url)
+    # Avoids invalid interpolation syntax errors if % in string
+    config.set_main_option("sqlalchemy.url", database_url.replace("%", "%%"))
     return config
 
 

src/fides/config/database_settings.py

@@ -4,7 +4,7 @@
 
 from copy import deepcopy
 from typing import Dict, Optional, Union, cast
-from urllib.parse import quote, urlencode
+from urllib.parse import quote, quote_plus, urlencode
 
 from pydantic import Field, PostgresDsn, validator
 
@@ -71,7 +71,7 @@ class DatabaseSettings(FidesSettings):
     )
     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.",
+        description="Additional connection parameters used when connecting to the application database.",
     )
 
     # These must be at the end because they require other values to construct
@@ -96,6 +96,13 @@ class DatabaseSettings(FidesSettings):
         exclude=True,
     )
 
+    @validator("password", pre=True)
+    def escape_password(cls, value: Optional[str]) -> Optional[str]:
+        """Escape password"""
+        if value and isinstance(value, str):
+            return quote_plus(value)
+        return value
+
     @validator("sync_database_uri", pre=True)
     @classmethod
     def assemble_sync_database_uri(

tests/ctl/core/config/test_config.py

@@ -5,6 +5,7 @@
 import pytest
 from pydantic import ValidationError
 
+from fides.api.db.database import get_alembic_config
 from fides.config import check_required_webserver_config_values, get_config
 from fides.config.database_settings import DatabaseSettings
 from fides.config.redis_settings import RedisSettings
@@ -191,6 +192,46 @@ def test_database_url_test_mode_disabled() -> None:
     )
 
 
+@pytest.mark.unit
+def test_password_escaped_by_database_settings_validation() -> None:
+    database_settings = DatabaseSettings(
+        user="postgres",
+        password="fidesp@ssword",
+        server="fides-db",
+        port="5432",
+        db="database",
+        test_db="test_database",
+    )
+    assert (
+        database_settings.async_database_uri
+        == "postgresql+asyncpg://postgres:fidesp%40ssword@fides-db:5432/test_database"
+    )
+
+    assert (
+        database_settings.sync_database_uri
+        == "postgresql+psycopg2://postgres:fidesp%40ssword@fides-db:5432/test_database"
+    )
+
+    assert (
+        database_settings.sqlalchemy_database_uri
+        == "postgresql://postgres:fidesp%40ssword@fides-db:5432/database"
+    )
+
+    assert (
+        database_settings.sqlalchemy_test_database_uri
+        == "postgresql://postgres:fidesp%40ssword@fides-db:5432/test_database"
+    )
+
+
+def test_get_alembic_config_with_special_char_in_database_url():
+    database_url = (
+        "postgresql+psycopg2://postgres:fidesp%40ssword@fides-db:5432/test_database"
+    )
+    # this would fail with - ValueError: invalid interpolation syntax
+    # if not handled
+    get_alembic_config(database_url)
+
+
 @patch.dict(
     os.environ,
     {