feat(encryption): Add Fernet Key Store (#103511)

Introduces `FernetKeyStore` which is a util singleton class used for
loading and retrieving Fernet keys.

For now, keys are loaded on first usage, and support for loading keys on
star up will be introduced as a part of a separate PR.

With the `FernetKeyStore` we are also able to simplify the logic in
`EncryptedFiled` base class because now all of the logic for handling
keys lives elsewhere.

Author: vgrozdanic

src/sentry/conf/server.py

@@ -20,6 +20,7 @@
     SENTRY_API_PAGINATION_ALLOWLIST_DO_NOT_MODIFY,
 )
 from sentry.conf.types.bgtask import BgTaskConfig
+from sentry.conf.types.encrypted_field import EncryptedFieldSettings
 from sentry.conf.types.kafka_definition import ConsumerDefinition
 from sentry.conf.types.logging_config import LoggingConfig
 from sentry.conf.types.region_config import RegionConfig
@@ -794,18 +795,14 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     },
 }
 
-# Fernet keys for database encryption.
-# First key in the dict is used as a primary key, and if
-# encryption method options is "fernet", the first key will be
-# used to decrypt the data.
-#
-# Other keys are used only for data decryption. This structure
-# is used to allow easier key rotation when "fernet" is used
-# as an encryption method.
-DATABASE_ENCRYPTION_FERNET_KEYS = {
-    os.getenv("DATABASE_ENCRYPTION_KEY_ID_1"): os.getenv("DATABASE_ENCRYPTION_FERNET_KEY_1"),
+# Settings for encrypted database fields.
+DATABASE_ENCRYPTION_SETTINGS: EncryptedFieldSettings = {
+    "method": "plaintext",
+    "fernet_primary_key_id": os.getenv("DATABASE_ENCRYPTION_FERNET_PRIMARY_KEY_ID"),
+    "fernet_keys_location": os.getenv("DATABASE_ENCRYPTION_FERNET_KEYS_LOCATION"),
 }
 
+
 #######################
 # Taskworker settings #
 #######################

src/sentry/conf/types/encrypted_field.py

@@ -0,0 +1,8 @@
+from typing import Literal, TypedDict
+
+
+class EncryptedFieldSettings(TypedDict):
+    method: Literal["fernet"] | Literal["plaintext"]
+    # fernet config
+    fernet_primary_key_id: str | None
+    fernet_keys_location: str | None

src/sentry/db/models/fields/encryption.py

@@ -6,11 +6,11 @@
 from typing import Any, Literal, TypedDict
 
 import sentry_sdk
-from cryptography.fernet import Fernet, InvalidToken
-from django.conf import settings
+from cryptography.fernet import InvalidToken
 from django.db.models import CharField, Field
 
 from sentry import options
+from sentry.utils.security.encrypted_field_key_store import FernetKeyStore
 
 logger = logging.getLogger(__name__)
 
@@ -102,8 +102,10 @@ def get_prep_value(self, value: Any) -> str | None:
         if value is None:
             return value
 
+        # Get the encryption method from the options
+        # xxx(vgrozdanic): this is a temporary solution during a rollout
+        # so that we can quickly rollback if needed.
         encryption_method = options.get("database.encryption.method")
-
         # Default to plaintext if method is not recognized
         if encryption_method not in self._encryption_handlers:
             logger.error(
@@ -179,15 +181,8 @@ def _encrypt_fernet(self, value: Any) -> str:
         Always returns formatted string: enc:fernet:key_id:base64data
         The key_id is required to support key rotation.
         """
-        key_id, key = self._get_fernet_key_for_encryption()
-        if not key:
-            raise ValueError(
-                "Fernet encryption key is required but not found. "
-                "Please set DATABASE_ENCRYPTION_FERNET_KEYS in your settings."
-            )
-
         try:
-            f = Fernet(key)
+            key_id, f = FernetKeyStore.get_primary_fernet()
             value_bytes = self._get_value_in_bytes(value)
             encrypted_data = f.encrypt(value_bytes)
             return self._format_encrypted_value(encrypted_data, MARKER_FERNET, key_id)
@@ -215,14 +210,8 @@ def _decrypt_fernet(self, value: str) -> bytes:
             logger.warning("Failed to decode base64 data: %s", e)
             raise ValueError("Invalid base64 encoding") from e
 
-        # Get the decryption key
-        key = self._get_fernet_key(key_id)
-        if not key:
-            logger.warning("No decryption key found for key_id '%s'", key_id)
-            raise ValueError(f"Cannot decrypt without key for key_id '{key_id}'")
-
         try:
-            f = Fernet(key)
+            f = FernetKeyStore.get_fernet_for_key_id(key_id)
             decrypted = f.decrypt(encrypted_data)
             return decrypted
         except InvalidToken:  # noqa
@@ -280,70 +269,6 @@ def _decrypt_with_fallback(self, value: str) -> bytes | str:
         logger.warning("No handler found for marker '%s'", marker)
         return value
 
-    def _get_fernet_key_for_encryption(self) -> tuple[str, bytes]:
-        """Get the first Fernet key for encryption along with its key_id."""
-        keys = getattr(settings, "DATABASE_ENCRYPTION_FERNET_KEYS", None)
-        if keys is None:
-            raise ValueError("DATABASE_ENCRYPTION_FERNET_KEYS is not configured")
-
-        if not isinstance(keys, dict):
-            logger.error("DATABASE_ENCRYPTION_FERNET_KEYS must be a dict, got %s", type(keys))
-            raise ValueError("DATABASE_ENCRYPTION_FERNET_KEYS must be a dictionary")
-
-        if not keys:
-            raise ValueError("DATABASE_ENCRYPTION_FERNET_KEYS is empty")
-
-        # Use the first key for encryption
-        key_id = next(iter(keys.keys()))
-        key = keys[key_id]
-
-        if not key_id or not key:
-            raise ValueError(
-                f"DATABASE_ENCRYPTION_FERNET_KEYS has invalid key_id or key ({key_id}, {key})"
-            )
-
-        if isinstance(key, str):
-            key = key.encode("utf-8")
-
-        # Validate key
-        try:
-            Fernet(key)
-            return (key_id, key)
-        except Exception as e:
-            sentry_sdk.capture_exception(e)
-            logger.exception("Invalid Fernet key for key_id '%s'", key_id)
-            raise ValueError(f"Invalid Fernet key for key_id '{key_id}'")
-
-    def _get_fernet_key(self, key_id: str) -> bytes | None:
-        """Get the Fernet key for the specified key_id from Django settings."""
-        keys = getattr(settings, "DATABASE_ENCRYPTION_FERNET_KEYS", None)
-        if keys is None:
-            return None
-
-        if not isinstance(keys, dict):
-            logger.error("DATABASE_ENCRYPTION_FERNET_KEYS must be a dict, got %s", type(keys))
-            return None
-
-        # Return specific key by key_id
-        if key_id not in keys:
-            logger.warning("Fernet key with id '%s' not found, cannot decrypt data", key_id)
-            return None
-        key = keys[key_id]
-
-        if isinstance(key, str):
-            # If key is a string, encode it
-            key = key.encode("utf-8")
-
-        # Validate key length (Fernet requires 32 bytes base64 encoded)
-        try:
-            Fernet(key)
-            return key
-        except Exception as e:
-            sentry_sdk.capture_exception(e)
-            logger.exception("Invalid Fernet key")
-
-        return None
-
 
 class EncryptedCharField(EncryptedField, CharField):
     def from_db_value(self, value: Any, expression: Any, connection: Any) -> Any:

src/sentry/utils/security/encrypted_field_key_store.py

@@ -0,0 +1,102 @@
+from logging import getLogger
+from pathlib import Path
+
+import sentry_sdk
+from cryptography.fernet import Fernet
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
+
+logger = getLogger(__name__)
+
+
+class FernetKeyStore:
+    _keys: dict[str, Fernet] | None = {}
+    _is_loaded = False
+
+    @classmethod
+    def _path_to_keys(cls) -> Path | None:
+        settings_path = settings.DATABASE_ENCRYPTION_SETTINGS.get("fernet_keys_location")
+
+        return Path(settings_path) if settings_path is not None else None
+
+    @classmethod
+    def load_keys(cls) -> None:
+        """
+        Reads all files in the given directory.
+        Filename = Key ID
+        File Content = Fernet Key
+        """
+        path = cls._path_to_keys()
+
+        if path is None:
+            # No keys directory is configured, so we don't need to load any keys.
+            cls._keys = None
+            cls._is_loaded = True
+            return
+
+        if not path.exists() or not path.is_dir():
+            raise ImproperlyConfigured(f"Key directory not found: {path}")
+
+        # Clear the keys dictionary to avoid stale data
+        cls._keys = {}
+
+        for file_path in path.iterdir():
+            if file_path.is_file():
+                # Skip hidden files
+                if file_path.name.startswith("."):
+                    continue
+
+                try:
+                    with open(file_path) as f:
+                        key_content = f.read().strip()
+
+                    if not key_content:
+                        logger.warning("Empty key file found: %s", file_path.name)
+                        continue
+
+                    # Store Fernet object in the dictionary
+                    # Objects are stored instead of keys for performance optimization.
+                    cls._keys[file_path.name] = Fernet(key_content.encode("utf-8"))
+
+                except Exception as e:
+                    logger.exception("Error reading key %s", file_path.name)
+                    sentry_sdk.capture_exception(e)
+
+        cls._is_loaded = True
+        logger.info("Successfully loaded %d Fernet encryption keys.", len(cls._keys))
+
+    @classmethod
+    def get_fernet_for_key_id(cls, key_id: str) -> Fernet:
+        """Retrieves a Fernet object for a specific key ID"""
+        if not cls._is_loaded:
+            # Fallback: if the keys are not already loaded, load them on first access.
+            logger.warning("Loading Fernet encryption keys on first access instead of on startup.")
+            cls.load_keys()
+
+        if cls._keys is None:
+            # This can happen only if the keys settings are misconfigured
+            raise ValueError(
+                "Fernet encryption keys are not loaded. Please configure the keys directory."
+            )
+
+        fernet = cls._keys.get(key_id)
+        if not fernet:
+            raise ValueError(f"Encryption key with ID '{key_id}' not found.")
+
+        return fernet
+
+    @classmethod
+    def get_primary_fernet(cls) -> tuple[str, Fernet]:
+        """
+        Reads the configuration and returns the primary key ID and the Fernet object
+        initialized with the primary Fernet key.
+
+        The primary Fernet is the one that is used to encrypt the data, while
+        decryption can be done with any of the registered Fernet keys.
+        """
+
+        primary_key_id = settings.DATABASE_ENCRYPTION_SETTINGS.get("fernet_primary_key_id")
+        if primary_key_id is None:
+            raise ValueError("Fernet primary key ID is not configured.")
+
+        return primary_key_id, cls.get_fernet_for_key_id(primary_key_id)