diff --git a/api_app/classes.py b/api_app/classes.py index f347bcfd6d..f24b428bf2 100644 --- a/api_app/classes.py +++ b/api_app/classes.py @@ -140,16 +140,16 @@ def __str__(self): return f"{self.__class__.__name__}" def config(self, runtime_configuration: typing.Dict): - """ - Configure the plugin with runtime parameters. - - Args: - runtime_configuration (dict): Runtime configuration parameters. - """ self.__parameters = self._config.read_configured_params(self._user, runtime_configuration) for parameter in self.__parameters: attribute_name = f"_{parameter.name}" if parameter.is_secret else parameter.name - setattr(self, attribute_name, parameter.value) + value = parameter.value + # decrypt secrets that were stored encrypted + if parameter.is_secret and isinstance(value, str) and value.startswith("gAAAAA"): + from api_app.models import PluginConfig + + value = PluginConfig._decrypt_value(value) + setattr(self, attribute_name, value) logger.debug( f"Adding to {self.__class__.__name__} param {attribute_name} with value {parameter.value} " ) @@ -349,6 +349,14 @@ def _get_health_check_url(self, user: User = None) -> typing.Optional[str]: if not param.configured or not param.value: continue url = param.value + # Decrypt if the value is Fernet-encrypted (secret parameter) + if isinstance(url, str) and url.startswith("gAAAAA"): + try: + from api_app.models import PluginConfig + + url = PluginConfig._decrypt_value(url) + except Exception: + pass logger.info(f"Url retrieved to verify is {param.name} for {self}") return url if hasattr(self, "url") and self.url: diff --git a/api_app/migrations/0073_encrypt_plugin_config_secrets.py b/api_app/migrations/0073_encrypt_plugin_config_secrets.py new file mode 100644 index 0000000000..74da405aae --- /dev/null +++ b/api_app/migrations/0073_encrypt_plugin_config_secrets.py @@ -0,0 +1,65 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +import base64 +import hashlib +import json + +from django.conf import settings +from django.db import migrations + + +def _get_fernet(): + from cryptography.fernet import Fernet + + key = getattr(settings, "PLUGIN_CONFIG_FERNET_KEY", None) + if key is None: + key = base64.urlsafe_b64encode( + hashlib.sha256(settings.SECRET_KEY.encode()).digest() + ) + return Fernet(key) + + +def encrypt_existing_secrets(apps, schema_editor): + PluginConfig = apps.get_model("api_app", "PluginConfig") + fernet = _get_fernet() + + for pc in PluginConfig.objects.filter( + parameter__is_secret=True, + value__isnull=False, + ): + if isinstance(pc.value, str) and pc.value.startswith("gAAAAA"): + continue + encrypted = fernet.encrypt(json.dumps(pc.value).encode()).decode() + PluginConfig.objects.filter(pk=pc.pk).update(value=encrypted) + + +def decrypt_existing_secrets(apps, schema_editor): + PluginConfig = apps.get_model("api_app", "PluginConfig") + fernet = _get_fernet() + + for pc in PluginConfig.objects.filter( + parameter__is_secret=True, + value__isnull=False, + ): + if not (isinstance(pc.value, str) and pc.value.startswith("gAAAAA")): + continue + try: + decrypted = json.loads(fernet.decrypt(pc.value.encode()).decode()) + PluginConfig.objects.filter(pk=pc.pk).update(value=decrypted) + except Exception: + pass + + +class Migration(migrations.Migration): + + dependencies = [ + ("api_app", "0072_update_check_system"), + ] + + operations = [ + migrations.RunPython( + encrypt_existing_secrets, + reverse_code=decrypt_existing_secrets, + ), + ] diff --git a/api_app/models.py b/api_app/models.py index bd3dca557f..3161a86d8f 100644 --- a/api_app/models.py +++ b/api_app/models.py @@ -1019,6 +1019,31 @@ def is_secret(self): """Returns whether the parameter is marked as secret.""" return self.parameter.is_secret + @staticmethod + def _encrypt_value(value): + """Fernet-encrypt a value (serialized as JSON).""" + from cryptography.fernet import Fernet + + f = Fernet(settings.PLUGIN_CONFIG_FERNET_KEY) + return f.encrypt(json.dumps(value).encode()).decode() + + @staticmethod + def _decrypt_value(encrypted_value): + """Fernet-decrypt a value back to its original Python object.""" + from cryptography.fernet import Fernet + + f = Fernet(settings.PLUGIN_CONFIG_FERNET_KEY) + return json.loads(f.decrypt(encrypted_value.encode()).decode()) + + def save(self, *args, **kwargs): + if ( + self.is_secret() + and self.value is not None + and not (isinstance(self.value, str) and self.value.startswith("gAAAAA")) + ): + self.value = self._encrypt_value(self.value) + super().save(*args, **kwargs) + @property def plugin_name(self): """Returns the name of the plugin associated with this configuration.""" diff --git a/intel_owl/settings/security.py b/intel_owl/settings/security.py index 8be1ea5172..c0fbe9972a 100644 --- a/intel_owl/settings/security.py +++ b/intel_owl/settings/security.py @@ -2,6 +2,9 @@ # See the file 'LICENSE' for copying permission. # Security Stuff +import base64 +import hashlib + from django.core.management.utils import get_random_secret_key from ._util import get_secret @@ -26,6 +29,11 @@ CSRF_TRUSTED_ORIGINS = [f"{WEB_CLIENT_URL}:80/"] ALLOWED_HOSTS = ["*"] +# Fernet key for encrypting plugin secrets at rest. +# Falls back to SECRET_KEY if PLUGIN_CONFIG_SECRET_KEY is not set. +_raw_secret = get_secret("PLUGIN_CONFIG_SECRET_KEY", SECRET_KEY) +PLUGIN_CONFIG_FERNET_KEY = base64.urlsafe_b64encode(hashlib.sha256(_raw_secret.encode()).digest()) + # https://docs.djangoproject.com/en/4.2/ref/settings/#data-upload-max-memory-size DATA_UPLOAD_MAX_MEMORY_SIZE = 100 * (10**6) FILE_UPLOAD_MAX_MEMORY_SIZE = 100 * (10**6) diff --git a/requirements/project-requirements.txt b/requirements/project-requirements.txt index a2eeaaada9..0da6544aaa 100644 --- a/requirements/project-requirements.txt +++ b/requirements/project-requirements.txt @@ -118,3 +118,4 @@ DeepDiff==8.6.1 lxml==6.0.2 Faker==36.1.0 beautifulsoup4==4.14.2 +cryptography==46.0.5 diff --git a/tests/api_app/test_plugin_config_encryption.py b/tests/api_app/test_plugin_config_encryption.py new file mode 100644 index 0000000000..c4b9ea03f7 --- /dev/null +++ b/tests/api_app/test_plugin_config_encryption.py @@ -0,0 +1,102 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +from api_app.choices import PythonModuleBasePaths +from api_app.models import Parameter, PluginConfig, PythonModule +from api_app.visualizers_manager.models import VisualizerConfig +from tests import CustomTestCase + + +class PluginConfigEncryptionTestCase(CustomTestCase): + def setUp(self): + super().setUp() + self.vc, _ = VisualizerConfig.objects.get_or_create( + name="test_encryption", + description="test encryption", + python_module=PythonModule.objects.get( + base_path=PythonModuleBasePaths.Visualizer.value, + module="yara.Yara", + ), + disabled=False, + ) + self.secret_param = Parameter.objects.create( + python_module=self.vc.python_module, + name="test_api_key", + type="str", + is_secret=True, + required=False, + ) + self.non_secret_param = Parameter.objects.create( + python_module=self.vc.python_module, + name="test_max_retries", + type="int", + is_secret=False, + required=False, + ) + + def tearDown(self): + self.secret_param.delete() + self.non_secret_param.delete() + self.vc.delete() + super().tearDown() + + def test_secret_value_encrypted_on_save(self): + pc = PluginConfig.objects.create( + owner=self.user, + for_organization=False, + parameter=self.secret_param, + value="my_super_secret_api_key_12345", + visualizer_config=self.vc, + ) + pc.refresh_from_db() + self.assertIsInstance(pc.value, str) + self.assertTrue(pc.value.startswith("gAAAAA")) + pc.delete() + + def test_encrypt_decrypt_roundtrip(self): + original = "my_super_secret_api_key_12345" + encrypted = PluginConfig._encrypt_value(original) + self.assertTrue(encrypted.startswith("gAAAAA")) + self.assertEqual(PluginConfig._decrypt_value(encrypted), original) + + def test_non_secret_value_unchanged(self): + pc = PluginConfig.objects.create( + owner=self.user, + for_organization=False, + parameter=self.non_secret_param, + value=10, + visualizer_config=self.vc, + ) + pc.refresh_from_db() + self.assertEqual(pc.value, 10) + pc.delete() + + def test_no_double_encryption(self): + pc = PluginConfig.objects.create( + owner=self.user, + for_organization=False, + parameter=self.secret_param, + value="test_secret_value", + visualizer_config=self.vc, + ) + pc.refresh_from_db() + first_encrypted = pc.value + + # saving again should not re-encrypt + pc.save() + pc.refresh_from_db() + self.assertEqual(pc.value, first_encrypted) + self.assertEqual(PluginConfig._decrypt_value(pc.value), "test_secret_value") + pc.delete() + + def test_encrypt_decrypt_dict(self): + original = {"key": "value", "nested": {"a": 1}} + encrypted = PluginConfig._encrypt_value(original) + self.assertTrue(encrypted.startswith("gAAAAA")) + self.assertEqual(PluginConfig._decrypt_value(encrypted), original) + + def test_encrypt_decrypt_list(self): + original = ["secret1", "secret2", "secret3"] + encrypted = PluginConfig._encrypt_value(original) + self.assertTrue(encrypted.startswith("gAAAAA")) + self.assertEqual(PluginConfig._decrypt_value(encrypted), original) diff --git a/tests/api_app/test_views.py b/tests/api_app/test_views.py index e78b69a32a..6774581596 100644 --- a/tests/api_app/test_views.py +++ b/tests/api_app/test_views.py @@ -678,7 +678,7 @@ def test_plugin_config(self): for config in [*org_config, *user_config]: if config["attribute"] == "mynewparameter": - self.assertEqual(config["value"], "supersecret") + self.assertEqual(PluginConfig._decrypt_value(config["value"]), "supersecret") # if the user is admin of an org, he should get the org secret self.client.force_authenticate(user=self.admin) @@ -690,7 +690,7 @@ def test_plugin_config(self): for config in [*org_config, *user_config]: if config["attribute"] == "mynewparameter": - self.assertEqual(config["value"], "supersecret") + self.assertEqual(PluginConfig._decrypt_value(config["value"]), "supersecret") # second personal item secret_owner = PluginConfig( @@ -712,10 +712,10 @@ def test_plugin_config(self): for config in org_config: if config["attribute"] == "mynewparameter": - self.assertEqual(config["value"], "supersecret") + self.assertEqual(PluginConfig._decrypt_value(config["value"]), "supersecret") for config in user_config: if config["attribute"] == "mynewparameter": - self.assertEqual(config["value"], "supersecret_user_only") + self.assertEqual(PluginConfig._decrypt_value(config["value"]), "supersecret_user_only") # other users cannot see user's personal items self.client.force_authenticate(user=self.admin) @@ -727,11 +727,11 @@ def test_plugin_config(self): for config in org_config: if config["attribute"] == "mynewparameter": - self.assertEqual(config["value"], "supersecret") + self.assertEqual(PluginConfig._decrypt_value(config["value"]), "supersecret") for config in user_config: if config["attribute"] == "mynewparameter": - self.assertNotEqual(config["value"], "supersecret_user_only") - self.assertEqual(config["value"], "supersecret") + self.assertNotEqual(PluginConfig._decrypt_value(config["value"]), "supersecret_user_only") + self.assertEqual(PluginConfig._decrypt_value(config["value"]), "supersecret") # if a standard user who does not belong to any org tries to get a secret, # they should not find anything @@ -768,7 +768,7 @@ def test_plugin_config(self): self.assertEqual(config["value"], "redacted") secret_owner.refresh_from_db() - self.assertEqual(secret_owner.value, "supersecret_user_only") + self.assertEqual(PluginConfig._decrypt_value(secret_owner.value), "supersecret_user_only") # third superuser secret secret_owner = PluginConfig( @@ -790,10 +790,10 @@ def test_plugin_config(self): self.assertEqual(config["value"], "redacted") for config in user_config: if config["attribute"] == "mynewparameter": - self.assertEqual(config["value"], "supersecret_low_privilege") + self.assertEqual(PluginConfig._decrypt_value(config["value"]), "supersecret_low_privilege") param.delete() - PluginConfig.objects.filter(value__startswith="supersecret").delete() + PluginConfig.objects.filter(value__startswith="gAAAAA").delete() org.delete() def test_plugin_config_list(self): @@ -856,7 +856,7 @@ def test_plugin_config_list(self): self.assertIn("organization", needle) self.assertEqual(needle["organization"], "testorg0") self.assertIn("value", needle) - self.assertEqual(needle["value"], "value") + self.assertEqual(PluginConfig._decrypt_value(needle["value"]), "value") self.assertIn("attribute", needle) self.assertEqual(needle["attribute"], "test") self.assertIn("required", needle) @@ -883,7 +883,7 @@ def test_plugin_config_list(self): self.assertIn("organization", needle) self.assertEqual(needle["organization"], "testorg0") self.assertIn("value", needle) - self.assertEqual(needle["value"], "value") + self.assertEqual(PluginConfig._decrypt_value(needle["value"]), "value") self.assertIn("attribute", needle) self.assertEqual(needle["attribute"], "test") self.assertIn("required", needle) @@ -998,7 +998,7 @@ def test_update(self): response = self.client.patch(uri, payload, format="json") self.assertEqual(response.status_code, 200) pc1 = PluginConfig.objects.get(id=pc.pk) - self.assertEqual(pc1.value, "new_org_supersecret") + self.assertEqual(PluginConfig._decrypt_value(pc1.value), "new_org_supersecret") # admin can update org secret self.client.force_authenticate(user=self.admin) @@ -1012,7 +1012,7 @@ def test_update(self): response = self.client.patch(uri, payload, format="json") self.assertEqual(response.status_code, 200) pc1 = PluginConfig.objects.get(id=pc.pk) - self.assertEqual(pc1.value, "new_org_supersecret_admin") + self.assertEqual(PluginConfig._decrypt_value(pc1.value), "new_org_supersecret_admin") # user can not update org secret self.client.force_authenticate(user=self.user) @@ -1049,7 +1049,7 @@ def test_update(self): response = self.client.patch(uri, payload, format="json") self.assertEqual(response.status_code, 200) pc_user = PluginConfig.objects.get(id=secret_owner.pk) - self.assertEqual(pc_user.value, "new_supersecret_user_only") + self.assertEqual(PluginConfig._decrypt_value(pc_user.value), "new_supersecret_user_only") # other users cannot update user's personal items self.client.force_authenticate(user=self.guest) @@ -1063,14 +1063,14 @@ def test_update(self): response = self.client.patch(uri, payload, format="json") self.assertEqual(response.status_code, 403) pc_user = PluginConfig.objects.get(id=secret_owner.pk) - self.assertEqual(pc_user.value, "new_supersecret_user_only") - self.assertNotEqual(pc_user.value, "new_supersecret") + self.assertEqual(PluginConfig._decrypt_value(pc_user.value), "new_supersecret_user_only") + self.assertNotEqual(PluginConfig._decrypt_value(pc_user.value), "new_supersecret") secret_owner.delete() pc.delete() param.delete() - PluginConfig.objects.filter(value__startswith="supersecret").delete() + PluginConfig.objects.filter(value__startswith="gAAAAA").delete() org.delete() def test_create(self): @@ -1121,7 +1121,7 @@ def test_create(self): response = self.client.post(uri, payload, format="json") self.assertEqual(response.status_code, 201) content = response.json() - self.assertEqual(content[0]["value"], "new_org_supersecret") + self.assertEqual(PluginConfig._decrypt_value(content[0]["value"]), "new_org_supersecret") self.assertEqual(content[0]["owner"], self.superuser.username) pc = PluginConfig.objects.get(id=content[0]["id"]) pc2 = PluginConfig.objects.get(id=content[1]["id"]) @@ -1144,7 +1144,7 @@ def test_create(self): response = self.client.post(uri, payload, format="json") self.assertEqual(response.status_code, 201) content = response.json() - self.assertEqual(content[0]["value"], "new_org_supersecret_admin") + self.assertEqual(PluginConfig._decrypt_value(content[0]["value"]), "new_org_supersecret_admin") self.assertEqual(content[0]["owner"], self.admin.username) pc = PluginConfig.objects.get(id=content[0]["id"]) self.assertTrue(pc.for_organization) @@ -1162,7 +1162,7 @@ def test_create(self): response = self.client.post(uri, payload, format="json") self.assertEqual(response.status_code, 201) content = response.json() - self.assertEqual(content[0]["value"], "new_supersecret_admin") + self.assertEqual(PluginConfig._decrypt_value(content[0]["value"]), "new_supersecret_admin") self.assertEqual(content[0]["owner"], self.admin.username) pc1 = PluginConfig.objects.get(id=content[0]["id"]) self.assertFalse(pc1.for_organization) @@ -1197,7 +1197,7 @@ def test_create(self): response = self.client.post(uri, payload, format="json") self.assertEqual(response.status_code, 201) content = response.json() - self.assertEqual(content[0]["value"], "new_supersecret_user_only") + self.assertEqual(PluginConfig._decrypt_value(content[0]["value"]), "new_supersecret_user_only") self.assertEqual(content[0]["owner"], self.user.username) pc = PluginConfig.objects.get(id=content[0]["id"]) self.assertFalse(pc.for_organization) @@ -1244,7 +1244,7 @@ def test_create(self): response = self.client.post(uri, payload, format="json") self.assertEqual(response.status_code, 201) content = response.json() - self.assertEqual(content[0]["value"], "new_user_secret") + self.assertEqual(PluginConfig._decrypt_value(content[0]["value"]), "new_user_secret") self.assertEqual(content[0]["owner"], self.user.username) pc1 = PluginConfig.objects.get(id=content[0]["id"]) self.assertFalse(pc1.for_organization) @@ -1252,7 +1252,7 @@ def test_create(self): pc.delete() param.delete() - PluginConfig.objects.filter(value__startswith="supersecret").delete() + PluginConfig.objects.filter(value__startswith="gAAAAA").delete() org.delete() def test_delete(self):