Allow configuring a custom CA with redis tls

CHANGELOG.md

@@ -22,6 +22,7 @@ The types of changes are:
 - Add Google Tag Manager and Privacy Center ENV vars to sample app [#2949](https://github.com/ethyca/fides/pull/2949)
 - Add `notice_key` field to Privacy Notice UI form [#3403](https://github.com/ethyca/fides/pull/3403)
 - 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/files)
 

src/fides/api/util/cache.py

@@ -171,6 +171,7 @@ def get_cache() -> FidesopsRedis:
             username=CONFIG.redis.user,
             password=CONFIG.redis.password,
             ssl=CONFIG.redis.ssl,
+            ssl_ca_certs=CONFIG.redis.ssl_ca_certs,
             ssl_cert_reqs=CONFIG.redis.ssl_cert_reqs,
         )
         logger.debug("New Redis connection created.")

src/fides/core/config/redis_settings.py

@@ -1,5 +1,5 @@
 from typing import Dict, Optional
-from urllib.parse import quote_plus
+from urllib.parse import quote, quote_plus, urlencode
 
 from pydantic import Field, validator
 
@@ -53,7 +53,11 @@ class RedisSettings(FidesSettings):
     )
     ssl_cert_reqs: Optional[str] = Field(
         default="required",
-        description="If using TLS encryption, set this to 'required' if you wish to enforce the Redis cache to provide a certificate. Note that not all cache providers support this e.g. AWS Elasticache.",
+        description="If using TLS encryption, set this to 'required' if you wish to enforce the Redis cache to provide a certificate. Note that not all cache providers support this without setting ssl_ca_certs (e.g. AWS Elasticache).",
+    )
+    ssl_ca_certs: str = Field(
+        default="",
+        description="If using TLS encryption rooted with a custom Certificate Authority, set this to the path of the CA certificate.",
     )
     user: str = Field(
         default="", description="The user with which to login to the Redis cache."
@@ -79,7 +83,7 @@ def assemble_connection_url(
             return v
 
         connection_protocol = "redis"
-        params = ""
+        params_str = ""
         use_tls = values.get("ssl", False)
 
         # These vars are intentionally fetched with `or ""` as the default to account
@@ -92,15 +96,18 @@ def assemble_connection_url(
             # If using TLS update the connection URL format
             connection_protocol = "rediss"
             cert_reqs = values.get("ssl_cert_reqs", "none")
-            params = f"?ssl_cert_reqs={quote_plus(cert_reqs)}"
+            params = {"ssl_cert_reqs": quote_plus(cert_reqs)}
+            if ssl_ca_certs := values.get("ssl_ca_certs", ""):
+                params["ssl_ca_certs"] = quote(ssl_ca_certs, safe="/")
+            params_str = "?" + urlencode(params, quote_via=quote, safe="/")
 
         # Configure a basic auth prefix if either user or password is provided, e.g.
         # redis://<user>:<password>@<host>
         auth_prefix = ""
         if password or user:
             auth_prefix = f"{quote_plus(user)}:{quote_plus(password)}@"
 
-        connection_url = f"{connection_protocol}://{auth_prefix}{values.get('host', '')}:{values.get('port', '')}/{db_index}{params}"
+        connection_url = f"{connection_protocol}://{auth_prefix}{values.get('host', '')}:{values.get('port', '')}/{db_index}{params_str}"
         return connection_url
 
     class Config:

tests/ctl/core/config/test_config.py

@@ -7,6 +7,7 @@
 
 from fides.core.config import check_required_webserver_config_values, get_config
 from fides.core.config.database_settings import DatabaseSettings
+from fides.core.config.redis_settings import RedisSettings
 from fides.core.config.security_settings import SecuritySettings
 
 REQUIRED_ENV_VARS = {
@@ -391,3 +392,29 @@ def test_check_required_webserver_config_values_error(capfd) -> None:
 def test_check_required_webserver_config_values_success_from_path() -> None:
     config = get_config()
     assert check_required_webserver_config_values(config=config) is None
+
+
+class TestBuildingRedisURLs:
+    def test_generic(self) -> None:
+        redis_settings = RedisSettings()
+        assert redis_settings.connection_url == "redis://:testpassword@redis:6379/"
+
+    def test_configured(self) -> None:
+        redis_settings = RedisSettings(
+            db_index=1, host="myredis", port="6380", password="supersecret"
+        )
+        assert redis_settings.connection_url == "redis://:supersecret@myredis:6380/1"
+
+    def test_tls(self) -> None:
+        redis_settings = RedisSettings(ssl=True, ssl_cert_reqs="none")
+        assert (
+            redis_settings.connection_url
+            == "rediss://:testpassword@redis:6379/?ssl_cert_reqs=none"
+        )
+
+    def test_tls_custom_ca(self) -> None:
+        redis_settings = RedisSettings(ssl=True, ssl_ca_certs="/path/to/my/cert.crt")
+        assert (
+            redis_settings.connection_url
+            == "rediss://:testpassword@redis:6379/?ssl_cert_reqs=required&ssl_ca_certs=/path/to/my/cert.crt"
+        )