fix(config): decrypt sensitive settings when loading from database
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -13,6 +13,7 @@ from typing import Any, Optional, Union
|
|||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
from app.models import ApplicationSettings
|
from app.models import ApplicationSettings
|
||||||
|
from app.utils.settings_service import get_setting_metadata
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -41,6 +42,17 @@ def load_settings_from_db(settings_obj: object, db_session: Session) -> None:
|
|||||||
key = db_setting.key
|
key = db_setting.key
|
||||||
value = db_setting.value
|
value = db_setting.value
|
||||||
|
|
||||||
|
# Decrypt sensitive settings before applying to the settings object
|
||||||
|
metadata = get_setting_metadata(key)
|
||||||
|
if metadata.get("sensitive", False) and value:
|
||||||
|
try:
|
||||||
|
from app.utils.encryption import decrypt_value
|
||||||
|
|
||||||
|
value = decrypt_value(value)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Failed to decrypt setting {key}: {e}")
|
||||||
|
continue
|
||||||
|
|
||||||
# Check if the setting exists in the Settings class
|
# Check if the setting exists in the Settings class
|
||||||
if hasattr(settings_obj, key):
|
if hasattr(settings_obj, key):
|
||||||
# Get the field info to determine the type
|
# Get the field info to determine the type
|
||||||
|
|||||||
@@ -161,6 +161,77 @@ class TestLoadSettingsFromDb:
|
|||||||
# Should not raise, just log warning
|
# Should not raise, just log warning
|
||||||
load_settings_from_db(mock_settings, mock_db)
|
load_settings_from_db(mock_settings, mock_db)
|
||||||
|
|
||||||
|
@patch("app.utils.config_loader.get_setting_metadata")
|
||||||
|
@patch("app.utils.encryption.decrypt_value")
|
||||||
|
def test_decrypts_sensitive_settings(self, mock_decrypt, mock_metadata):
|
||||||
|
"""Test that sensitive settings are decrypted before being applied."""
|
||||||
|
from app.models import ApplicationSettings
|
||||||
|
|
||||||
|
mock_metadata.return_value = {"sensitive": True}
|
||||||
|
mock_decrypt.return_value = "plaintext_token"
|
||||||
|
|
||||||
|
mock_settings = MagicMock()
|
||||||
|
mock_settings.__fields__ = {
|
||||||
|
"onedrive_refresh_token": MagicMock(annotation=str),
|
||||||
|
}
|
||||||
|
|
||||||
|
mock_db = MagicMock()
|
||||||
|
mock_db.query.return_value.all.return_value = [
|
||||||
|
ApplicationSettings(key="onedrive_refresh_token", value="enc:gAAAAABencryptedvalue"),
|
||||||
|
]
|
||||||
|
|
||||||
|
load_settings_from_db(mock_settings, mock_db)
|
||||||
|
|
||||||
|
mock_decrypt.assert_called_once_with("enc:gAAAAABencryptedvalue")
|
||||||
|
assert mock_settings.onedrive_refresh_token == "plaintext_token"
|
||||||
|
|
||||||
|
@patch("app.utils.config_loader.get_setting_metadata")
|
||||||
|
def test_non_sensitive_settings_not_decrypted(self, mock_metadata):
|
||||||
|
"""Test that non-sensitive settings are applied as-is without decryption."""
|
||||||
|
from app.models import ApplicationSettings
|
||||||
|
|
||||||
|
mock_metadata.return_value = {"sensitive": False}
|
||||||
|
|
||||||
|
mock_settings = MagicMock()
|
||||||
|
mock_settings.__fields__ = {
|
||||||
|
"some_setting": MagicMock(annotation=str),
|
||||||
|
}
|
||||||
|
|
||||||
|
mock_db = MagicMock()
|
||||||
|
mock_db.query.return_value.all.return_value = [
|
||||||
|
ApplicationSettings(key="some_setting", value="plain_value"),
|
||||||
|
]
|
||||||
|
|
||||||
|
with patch("app.utils.encryption.decrypt_value") as mock_decrypt:
|
||||||
|
load_settings_from_db(mock_settings, mock_db)
|
||||||
|
mock_decrypt.assert_not_called()
|
||||||
|
|
||||||
|
assert mock_settings.some_setting == "plain_value"
|
||||||
|
|
||||||
|
@patch("app.utils.config_loader.get_setting_metadata")
|
||||||
|
def test_decryption_failure_skips_setting(self, mock_metadata):
|
||||||
|
"""Test that a setting is skipped if decryption fails."""
|
||||||
|
from app.models import ApplicationSettings
|
||||||
|
|
||||||
|
mock_metadata.return_value = {"sensitive": True}
|
||||||
|
|
||||||
|
mock_settings = MagicMock()
|
||||||
|
mock_settings.__fields__ = {
|
||||||
|
"onedrive_refresh_token": MagicMock(annotation=str),
|
||||||
|
}
|
||||||
|
|
||||||
|
mock_db = MagicMock()
|
||||||
|
mock_db.query.return_value.all.return_value = [
|
||||||
|
ApplicationSettings(key="onedrive_refresh_token", value="enc:corrupted"),
|
||||||
|
]
|
||||||
|
|
||||||
|
with patch("app.utils.encryption.decrypt_value", side_effect=Exception("Decryption error")):
|
||||||
|
load_settings_from_db(mock_settings, mock_db)
|
||||||
|
|
||||||
|
# The attribute should not have been overwritten with a real value,
|
||||||
|
# since decryption failed and the setting was skipped
|
||||||
|
assert isinstance(mock_settings.onedrive_refresh_token, MagicMock)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
class TestReloadSettingsFromDb:
|
class TestReloadSettingsFromDb:
|
||||||
|
|||||||
Reference in New Issue
Block a user