Handle Tool Configuration credential storage in the model layer

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),
+    ]

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)

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()

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"]

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

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)
 

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:

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,
+        )