From 2923cf024c0508b3187b5f3e6a2f682ba74395a8 Mon Sep 17 00:00:00 2001 From: svader0 Date: Mon, 10 Aug 2026 12:01:43 -0500 Subject: [PATCH] Encrypt Tool Configuration credentials on every write path Encryption lived in the edit view, so the create views, the REST serializer, the admin and loaddata all stored the credential as it was typed, and api_key was never encrypted by anything. Moving it into the model field covers every writer and every reader, and drops the manual calls at the two sites that had them. Because reads no longer go through prepare_for_view, a value stored before this change is returned as it is rather than as an empty string, so opening and saving an affected configuration stops discarding its password and SSH key. The columns become text: AES-GCM output is a little over twice the input, which did not fit the old widths. The migration encrypts the rows already stored. --- .../0292_encrypt_tool_config_credentials.py | 94 ++++++++++++++ dojo/db_utils.py | 33 +++++ dojo/tool_config/admin.py | 2 +- dojo/tool_config/models.py | 10 +- dojo/tool_config/ui/forms.py | 4 + dojo/tool_config/ui/views.py | 6 +- dojo/tools/api_sonarqube/api_client.py | 7 +- .../test_tool_config_credential_encryption.py | 121 ++++++++++++++++++ 8 files changed, 261 insertions(+), 16 deletions(-) create mode 100644 dojo/db_migrations/0292_encrypt_tool_config_credentials.py create mode 100644 unittests/test_tool_config_credential_encryption.py diff --git a/dojo/db_migrations/0292_encrypt_tool_config_credentials.py b/dojo/db_migrations/0292_encrypt_tool_config_credentials.py new file mode 100644 index 00000000000..e4a6863de9f --- /dev/null +++ b/dojo/db_migrations/0292_encrypt_tool_config_credentials.py @@ -0,0 +1,94 @@ +import logging + +from django.db import migrations + +import dojo.db_utils + +logger = logging.getLogger(__name__) + +CREDENTIAL_COLUMNS = ("password", "ssh", "api_key") + +# Read in bounded chunks so a large Tool_Configuration table never loads every +# row at once. +BATCH_SIZE = 500 + + +def encrypt_plaintext_credentials(apps, schema_editor): + """ + Encrypt the Tool_Configuration credentials that earlier releases stored in the clear. + + Only one write path ever encrypted, so most rows hold plaintext. Migration + 0272 could not pick them up because it only upgraded values already carrying + the legacy "AES.1:" prefix. + + This reads the columns through a cursor rather than the model so it sees the + stored bytes instead of the value EncryptedTextField decodes. Anything + already prefixed is left alone, which also protects a value encrypted under + a key this deployment no longer has from being overwritten. + """ + from dojo.utils import dojo_crypto_encrypt + + connection = schema_editor.connection + encrypted = 0 + last_id = 0 + while True: + with connection.cursor() as cursor: + cursor.execute( + "SELECT id, password, ssh, api_key FROM dojo_tool_configuration " + "WHERE id > %s ORDER BY id LIMIT %s", + [last_id, BATCH_SIZE], + ) + page = cursor.fetchall() + if not page: + break + last_id = page[-1][0] + + for row_id, *values in page: + updates = { + column: dojo_crypto_encrypt(value) + for column, value in zip(CREDENTIAL_COLUMNS, values, strict=True) + if value and not value.startswith(dojo.db_utils.ENCRYPTED_VALUE_PREFIXES) + } + if not updates: + continue + assignments = ", ".join(f"{column} = %s" for column in updates) + with connection.cursor() as cursor: + cursor.execute( + f"UPDATE dojo_tool_configuration SET {assignments} WHERE id = %s", + [*updates.values(), row_id], + ) + encrypted += 1 + + if encrypted: + logger.info("Encrypted credentials for %d Tool_Configuration rows", encrypted) + + +def noop_reverse(apps, schema_editor): + # Decrypting on the way back would put the credentials in the clear again, + # which is the state this migration exists to leave. + pass + + +class Migration(migrations.Migration): + dependencies = [ + ("dojo", "0291_dojometa_location_product"), + ] + + operations = [ + migrations.AlterField( + model_name="tool_configuration", + name="password", + field=dojo.db_utils.EncryptedTextField(blank=True, null=True), + ), + migrations.AlterField( + model_name="tool_configuration", + name="ssh", + field=dojo.db_utils.EncryptedTextField(blank=True, null=True), + ), + migrations.AlterField( + model_name="tool_configuration", + name="api_key", + field=dojo.db_utils.EncryptedTextField(blank=True, null=True, verbose_name="API Key"), + ), + migrations.RunPython(encrypt_plaintext_credentials, noop_reverse), + ] diff --git a/dojo/db_utils.py b/dojo/db_utils.py index 11d7432ec37..866cfbcac12 100644 --- a/dojo/db_utils.py +++ b/dojo/db_utils.py @@ -1,5 +1,11 @@ """Database-level helpers that are not tied to any particular model or app.""" +from django.db import models + +# Stored-format prefixes written by dojo_crypto_encrypt(). "AES.1" is the legacy +# OFB format, "AES.2" the current GCM one; prepare_for_view() reads both. +ENCRYPTED_VALUE_PREFIXES = ("AES.1:", "AES.2:") + # SQLSTATEs meaning "this transaction lost a concurrency race": 40P01 # deadlock_detected and 40001 serialization_failure. Postgres aborts one participant # and lets the other commit, so the aborted work is not invalid -- it just has to run @@ -20,3 +26,30 @@ def is_transient_db_conflict(exc): getattr(err, "sqlstate", None) in TRANSIENT_DB_CONFLICT_SQLSTATES for err in (exc, exc.__cause__) ) + + +class EncryptedTextField(models.TextField): + + """ + TextField whose value is encrypted at rest and decrypted on read. + + Encrypting in the field rather than at each write site means every writer is + covered: forms, serializers, the admin, ``loaddata`` and ``QuerySet.update()``. + + Values stored before a column moved to this field are plaintext. They are + returned as they are and encrypted the next time the row is written, so an + unconverted row stays readable instead of decoding to an empty string. + """ + + def get_prep_value(self, value): + value = super().get_prep_value(value) + if not value or value.startswith(ENCRYPTED_VALUE_PREFIXES): + return value + from dojo.utils import dojo_crypto_encrypt # noqa: PLC0415 circular import + return dojo_crypto_encrypt(value) + + def from_db_value(self, value, expression, connection): + if not value or not value.startswith(ENCRYPTED_VALUE_PREFIXES): + return value + from dojo.utils import prepare_for_view # noqa: PLC0415 circular import + return prepare_for_view(value) diff --git a/dojo/tool_config/admin.py b/dojo/tool_config/admin.py index cb07719c546..983bd518000 100644 --- a/dojo/tool_config/admin.py +++ b/dojo/tool_config/admin.py @@ -21,7 +21,7 @@ def __init__(self, *args, **kwargs): # keep password from db to use if the user entered no password self.password_from_db = self.instance.password self.ssh_from_db = self.instance.ssh - self.api_key = self.instance.api_key + self.api_key_from_db = self.instance.api_key def clean(self): cleaned_data = super().clean() diff --git a/dojo/tool_config/models.py b/dojo/tool_config/models.py index f8281c098db..4ffbbf51533 100644 --- a/dojo/tool_config/models.py +++ b/dojo/tool_config/models.py @@ -1,6 +1,8 @@ from django.db import models from django.utils.translation import gettext_lazy as _ +from dojo.db_utils import EncryptedTextField + class Tool_Configuration(models.Model): name = models.CharField(max_length=200, null=False) @@ -17,12 +19,12 @@ class Tool_Configuration(models.Model): extras = models.CharField(max_length=255, null=True, blank=True, help_text=_("Additional definitions that will be " "consumed by scanner")) username = models.CharField(max_length=200, null=True, blank=True) - password = models.CharField(max_length=900, null=True, blank=True) + password = EncryptedTextField(null=True, blank=True) auth_title = models.CharField(max_length=200, null=True, blank=True, verbose_name=_("Title for SSH/API Key")) - ssh = models.CharField(max_length=9000, null=True, blank=True) - api_key = models.CharField(max_length=900, null=True, blank=True, - verbose_name=_("API Key")) + ssh = EncryptedTextField(null=True, blank=True) + api_key = EncryptedTextField(null=True, blank=True, + verbose_name=_("API Key")) class Meta: ordering = ["name"] diff --git a/dojo/tool_config/ui/forms.py b/dojo/tool_config/ui/forms.py index cc769725d78..74552d1db1e 100644 --- a/dojo/tool_config/ui/forms.py +++ b/dojo/tool_config/ui/forms.py @@ -8,6 +8,10 @@ class ToolConfigForm(forms.ModelForm): tool_type = forms.ModelChoiceField(queryset=Tool_Type.objects.all(), label="Tool Type") ssh = forms.CharField(widget=forms.Textarea(attrs={}), required=False, label="SSH Key") + # The credential columns are text so ciphertext of any length fits; keep the + # single-line inputs a ModelForm would otherwise turn into textareas. + password = forms.CharField(widget=forms.TextInput(attrs={}), required=False) + api_key = forms.CharField(widget=forms.TextInput(attrs={}), required=False, label="API Key") class Meta: model = Tool_Configuration diff --git a/dojo/tool_config/ui/views.py b/dojo/tool_config/ui/views.py index 163341a1500..7009675b2b3 100644 --- a/dojo/tool_config/ui/views.py +++ b/dojo/tool_config/ui/views.py @@ -10,7 +10,7 @@ from dojo.tool_config.factory import create_API from dojo.tool_config.models import Tool_Configuration from dojo.tool_config.ui.forms import ToolConfigForm -from dojo.utils import add_breadcrumb, dojo_crypto_encrypt, prepare_for_view +from dojo.utils import add_breadcrumb logger = logging.getLogger(__name__) @@ -57,8 +57,6 @@ def edit_tool_config(request, ttid): tform = ToolConfigForm(request.POST, instance=tool_config) if tform.is_valid(): form_copy = tform.save(commit=False) - form_copy.password = dojo_crypto_encrypt(tform.cleaned_data["password"]) - form_copy.ssh = dojo_crypto_encrypt(tform.cleaned_data["ssh"]) try: api = create_API(form_copy) if api and hasattr(api, "test_connection"): @@ -80,8 +78,6 @@ def edit_tool_config(request, ttid): str(e), extra_tags="alert-danger") else: - tool_config.password = prepare_for_view(tool_config.password) - tool_config.ssh = prepare_for_view(tool_config.ssh) tform = ToolConfigForm(instance=tool_config) add_breadcrumb(title="Edit Tool Configuration", top_level=False, request=request) diff --git a/dojo/tools/api_sonarqube/api_client.py b/dojo/tools/api_sonarqube/api_client.py index e7d78fae1da..237b1e1501b 100644 --- a/dojo/tools/api_sonarqube/api_client.py +++ b/dojo/tools/api_sonarqube/api_client.py @@ -2,8 +2,6 @@ from django.conf import settings from requests.exceptions import JSONDecodeError as RequestsJSONDecodeError -from dojo.utils import prepare_for_view - class SonarQubeAPI: def __init__(self, tool_config): @@ -46,10 +44,7 @@ def __init__(self, tool_config): self.default_headers = {"User-Agent": "DefectDojo"} self.sonar_api_url = tool_config.url if tool_config.authentication_type == "Password": - self.session.auth = ( - tool_config.username, - prepare_for_view(tool_config.password), - ) + self.session.auth = (tool_config.username, tool_config.password) elif tool_config.authentication_type == "API": self.session.auth = (tool_config.api_key, "") else: diff --git a/unittests/test_tool_config_credential_encryption.py b/unittests/test_tool_config_credential_encryption.py new file mode 100644 index 00000000000..8ec1bca80cf --- /dev/null +++ b/unittests/test_tool_config_credential_encryption.py @@ -0,0 +1,121 @@ +""" +Regression test: Tool_Configuration credentials must be encrypted at rest no +matter which path wrote them. + +Encryption used to live in the edit view, so the create views, the REST +serializer, the admin and ``loaddata`` all stored the credential in the clear, +and ``api_key`` was never encrypted by any path at all. The checks below read +the columns through a cursor, because the model field decrypts on the way out +and would hide the very thing under test. +""" +from django.db import connection + +from dojo.models import Tool_Configuration, Tool_Type +from dojo.tool_config.api.serializer import ToolConfigurationSerializer +from dojo.tool_config.ui.forms import ToolConfigForm +from dojo.utils import dojo_crypto_encrypt + +from .dojo_test_case import DojoTestCase + +CREDENTIALS = { + "password": "pw-plaintext-canary", + "ssh": "ssh-plaintext-canary", + "api_key": "apikey-plaintext-canary", +} + + +def stored_values(pk): + """The credential columns exactly as Postgres holds them.""" + with connection.cursor() as cursor: + cursor.execute( + "SELECT password, ssh, api_key FROM dojo_tool_configuration WHERE id = %s", [pk], + ) + return dict(zip(CREDENTIALS, cursor.fetchone(), strict=True)) + + +class ToolConfigCredentialEncryptionTest(DojoTestCase): + def setUp(self): + self.tool_type, _ = Tool_Type.objects.get_or_create(name="SonarQube") + + def assert_encrypted_at_rest(self, tool_config): + stored = stored_values(tool_config.pk) + for field, plaintext in CREDENTIALS.items(): + with self.subTest(field=field): + self.assertTrue(stored[field].startswith("AES.2:"), stored[field]) + self.assertNotIn(plaintext, stored[field]) + # and the application still sees the original value + fresh = Tool_Configuration.objects.get(pk=tool_config.pk) + for field, plaintext in CREDENTIALS.items(): + self.assertEqual(plaintext, getattr(fresh, field)) + + def test_model_create_encrypts(self): + # Covers every writer that goes through the ORM, including the admin and loaddata. + self.assert_encrypted_at_rest(Tool_Configuration.objects.create( + name="via-model", tool_type=self.tool_type, authentication_type="API", **CREDENTIALS, + )) + + def test_serializer_create_encrypts(self): + serializer = ToolConfigurationSerializer(data={ + "name": "via-api", "tool_type": self.tool_type.pk, "authentication_type": "API", + **CREDENTIALS, + }) + self.assertTrue(serializer.is_valid(), serializer.errors) + self.assert_encrypted_at_rest(serializer.save()) + + def test_serializer_update_encrypts(self): + tool_config = Tool_Configuration.objects.create( + name="via-api-update", tool_type=self.tool_type, authentication_type="API", + ) + serializer = ToolConfigurationSerializer(tool_config, data=CREDENTIALS, partial=True) + self.assertTrue(serializer.is_valid(), serializer.errors) + self.assert_encrypted_at_rest(serializer.save()) + + def test_form_create_encrypts(self): + form = ToolConfigForm(data={ + "name": "via-form", "tool_type": self.tool_type.pk, "authentication_type": "API", + "url": "https://example.invalid", **CREDENTIALS, + }) + self.assertTrue(form.is_valid(), form.errors) + self.assert_encrypted_at_rest(form.save()) + + def test_queryset_update_encrypts(self): + tool_config = Tool_Configuration.objects.create( + name="via-queryset", tool_type=self.tool_type, authentication_type="API", + ) + Tool_Configuration.objects.filter(pk=tool_config.pk).update(**CREDENTIALS) + self.assert_encrypted_at_rest(tool_config) + + def test_existing_plaintext_row_is_readable_and_encrypted_on_save(self): + # Rows written before this change hold plaintext. They must keep working, + # not decode to "" the way the old edit view made them. + tool_config = Tool_Configuration.objects.create( + name="legacy-plaintext", tool_type=self.tool_type, authentication_type="API", + ) + with connection.cursor() as cursor: + cursor.execute( + "UPDATE dojo_tool_configuration SET password = %s, ssh = %s, api_key = %s " + "WHERE id = %s", + [*CREDENTIALS.values(), tool_config.pk], + ) + + legacy = Tool_Configuration.objects.get(pk=tool_config.pk) + for field, plaintext in CREDENTIALS.items(): + self.assertEqual(plaintext, getattr(legacy, field)) + + legacy.save() + self.assert_encrypted_at_rest(legacy) + + def test_an_already_encrypted_value_is_stored_as_is(self): + # A caller that hands over ciphertext (a fixture, or code that encrypted + # for itself) must not have it encrypted a second time and become + # undecryptable. + ciphertext = dojo_crypto_encrypt(CREDENTIALS["password"]) + tool_config = Tool_Configuration.objects.create( + name="pre-encrypted", tool_type=self.tool_type, authentication_type="API", + password=ciphertext, + ) + self.assertEqual(ciphertext, stored_values(tool_config.pk)["password"]) + self.assertEqual( + CREDENTIALS["password"], + Tool_Configuration.objects.get(pk=tool_config.pk).password, + )