Merge pull request #78 from maykinmedia/chore/encrypted-private-key-s…

…upport

Move support for encrypted private key

digid_eherkenning/admin.py

@@ -23,10 +23,7 @@ class DigidConfigurationAdmin(CustomPrivateMediaMixin, SingletonModelAdmin):
         (
             _("X.509 Certificate"),
             {
-                "fields": (
-                    "certificate",
-                    "key_passphrase",
-                ),
+                "fields": ("certificate",),
             },
         ),
         (
@@ -88,10 +85,7 @@ class EherkenningConfigurationAdmin(CustomPrivateMediaMixin, SingletonModelAdmin
         (
             _("X.509 Certificate"),
             {
-                "fields": (
-                    "certificate",
-                    "key_passphrase",
-                ),
+                "fields": ("certificate",),
             },
         ),
         (

digid_eherkenning/management/commands/generate_eherkenning_dienstcatalogus.py

@@ -30,7 +30,6 @@ def add_arguments(self, parser):
         dests_to_delete = [
             "want_assertions_encrypted",
             "want_assertions_signed",
-            "key_passphrase",
             "technical_contact_person_telephone",
             "technical_contact_person_email",
             "organization_url",

digid_eherkenning/migrations/0009_decrypt_private_keys.py

@@ -0,0 +1,63 @@
+# Generated by Django 4.2.13 on 2024-07-18 11:08
+
+from io import BytesIO
+from pathlib import Path
+
+from django.core.files import File
+from django.db import migrations
+
+from simple_certmanager.utils import (
+    BadPassword,
+    KeyIsNotEncrypted,
+    decrypted_key_to_pem,
+    load_pem_x509_private_key,
+)
+
+
+def decrypt_encrypted_keys(apps, _):
+    DigiDConfiguration = apps.get_model("digid_eherkenning", "DigiDConfiguration")
+    EherkenningConfiguration = apps.get_model(
+        "digid_eherkenning", "EherkenningConfiguration"
+    )
+
+    for model in (DigiDConfiguration, EherkenningConfiguration):
+        config = model.objects.first()
+
+        # check if there is something to decrypt in the first place
+        if (
+            # no configuration record exists
+            config is None
+            # key is not encrypted (or we don't have the passphrase to decrypt it)
+            or not (pw := config.key_passphrase)
+            or not (cert := config.certificate)
+            # no private key file is uploaded
+            or not (privkey := cert.private_key)
+            # or the DB thinks there is a file but it's not actually there (anymore)
+            or not privkey.storage.exists(privkey.name)
+        ):
+            continue
+
+        with privkey.open("rb") as keyfile:
+            try:
+                key = load_pem_x509_private_key(keyfile.read(), password=pw)
+            # key material + passphrase set up are not in a consistent state, this would
+            # also affect runtime behaviour, but we opt to make our migration robust.
+            except (KeyIsNotEncrypted, BadPassword):
+                continue
+
+        decrypted_data = decrypted_key_to_pem(key)
+        # replace existing data
+        existing_name = Path(privkey.name).name  # strip nested directories of the name
+        privkey.save(name=existing_name, content=File(BytesIO(decrypted_data)))
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("digid_eherkenning", "0008_update_loa_fields"),
+    ]
+
+    operations = [
+        # we cannot reverse it since we may not know the encryption key anymore
+        migrations.RunPython(decrypt_encrypted_keys, migrations.RunPython.noop),
+    ]

digid_eherkenning/migrations/0010_remove_digidconfiguration_key_passphrase_and_more.py

@@ -0,0 +1,21 @@
+# Generated by Django 4.2.13 on 2024-07-18 14:46
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("digid_eherkenning", "0009_decrypt_private_keys"),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name="digidconfiguration",
+            name="key_passphrase",
+        ),
+        migrations.RemoveField(
+            model_name="eherkenningconfiguration",
+            name="key_passphrase",
+        ),
+    ]

digid_eherkenning/models/base.py

@@ -82,12 +82,6 @@ class BaseConfiguration(SingletonModel):
             "expect 'text/xml'."
         ),
     )
-    key_passphrase = models.CharField(
-        _("key passphrase"),
-        blank=True,
-        help_text=_("Passphrase for the private key used by the SOAP client."),
-        max_length=100,
-    )
     signature_algorithm = models.CharField(
         _("signature algorithm"),
         blank=True,

digid_eherkenning/models/digid.py

@@ -78,7 +78,6 @@ def as_dict(self) -> dict:
             # optional in runtime code
             "want_assertions_encrypted": self.want_assertions_encrypted,
             "want_assertions_signed": self.want_assertions_signed,
-            "key_passphrase": self.key_passphrase or None,
             "signature_algorithm": self.signature_algorithm,
             "digest_algorithm": self.digest_algorithm or None,
             "technical_contact_person_telephone": self.technical_contact_person_telephone

digid_eherkenning/models/eherkenning.py

@@ -287,7 +287,6 @@ def as_dict(self) -> EHerkenningConfig:
             # optional in runtime code
             "want_assertions_encrypted": self.want_assertions_encrypted,
             "want_assertions_signed": self.want_assertions_signed,
-            "key_passphrase": self.key_passphrase or None,
             "signature_algorithm": self.signature_algorithm,
             "digest_algorithm": self.digest_algorithm or None,
             "technical_contact_person_telephone": self.technical_contact_person_telephone

digid_eherkenning/saml2/base.py

@@ -246,7 +246,6 @@ def create_config_dict(self, conf):
                 "wantAssertionsSigned": conf.get("want_assertions_signed", False),
                 "soapClientKey": conf["key_file"].path,
                 "soapClientCert": conf["cert_file"].path,
-                "soapClientPassphrase": conf.get("key_passphrase", None),
                 # algorithm for requests with HTTP-redirect binding.
                 # AuthnRequest with HTTP-POST uses RSA_SHA256, which is hardcoded in OneLogin_Saml2_Auth.login_post
                 "signatureAlgorithm": conf.get(
@@ -283,7 +282,6 @@ def create_config_dict(self, conf):
                 "NameIDFormat": "urn:oasis:names:tc:SAML:1.1:nameid-format:unspecified",
                 "x509cert": certificate,
                 "privateKey": privkey,
-                "privateKeyPassphrase": conf.get("key_passphrase", None),
             },
         }
 
@@ -412,7 +410,6 @@ def handle_logout_request(
                 logout_response,
                 self.saml2_settings.get_sp_key(),
                 self.saml2_settings.get_sp_cert(),
-                key_passphrase=self.saml2_settings.get_sp_key_passphrase(),
                 sign_algorithm=OneLogin_Saml2_Constants.RSA_SHA256,
                 digest_algorithm=OneLogin_Saml2_Constants.SHA256,
             )

digid_eherkenning/saml2/eherkenning.py

@@ -491,7 +491,6 @@ def create_config_dict(self, conf: EHerkenningConfig) -> EHerkenningSAMLConfig:
                     "NameIDFormat": "urn:oasis:names:tc:SAML:1.1:nameid-format:unspecified",
                     "x509cert": certificate,
                     "privateKey": privkey,
-                    "privateKeyPassphrase": conf.get("key_passphrase", None),
                 },
             }
         )

digid_eherkenning/types.py

@@ -32,7 +32,6 @@ class EHerkenningConfig(TypedDict):
     services: list[ServiceConfig]
     want_assertions_encrypted: str
     want_assertions_signed: str
-    key_passphrase: str
     signature_algorithm: str
     digest_algorithm: str
     technical_contact_person_telephone: Optional[str]

pyproject.toml

@@ -30,7 +30,7 @@ dependencies = [
     "cryptography>=40.0.0",
     "django>=4.2.0",
     "django-sessionprofile",
-    "django-simple-certmanager",
+    "django-simple-certmanager>=2.2.0",
     "django-solo",
     "lxml>=4.7.1",
     "furl",
@@ -121,7 +121,6 @@ exclude_also = [
     "@(abc\\.)?abstractmethod",
     "raise NotImplementedError",
     "\\.\\.\\.",
-    "pass",
 ]
 omit = [
     "*/migrations/*",

tests/test_migrations.py

@@ -1,6 +1,15 @@
+from io import BytesIO
+from typing import Callable, Literal
+
+from django.core.files import File
 from django.db import IntegrityError
 
 import pytest
+from cryptography.hazmat.primitives.asymmetric import rsa
+from simple_certmanager.constants import CertificateTypes
+from simple_certmanager.models import Certificate
+from simple_certmanager.test.certificate_generation import key_to_pem
+from simple_certmanager.utils import load_pem_x509_private_key
 
 from digid_eherkenning.choices import AssuranceLevels
 
@@ -30,3 +39,166 @@ def test_fixing_misconfigured_eherkenning(migrator):
     config.loa = ""
     with pytest.raises(IntegrityError):
         config.save()
+
+
+@pytest.mark.django_db
+def test_decrypt_private_keys_with_passphrase(
+    migrator, encrypted_keypair: tuple[bytes, bytes]
+):
+    old_state = migrator.apply_initial_migration(
+        ("digid_eherkenning", "0008_update_loa_fields")
+    )
+    OldDigiDConfiguration = old_state.apps.get_model(
+        "digid_eherkenning", "DigiDConfiguration"
+    )
+    Certificate = old_state.apps.get_model("simple_certmanager", "Certificate")
+    key, cert = encrypted_keypair
+    certificate = Certificate.objects.create(
+        label="Test certificate",
+        type=CertificateTypes.key_pair,
+        public_certificate=File(BytesIO(cert), name="client_cert.pem"),
+        private_key=File(BytesIO(key), name="client_key.pem"),
+    )
+    OldDigiDConfiguration.objects.create(
+        certificate=certificate, key_passphrase="SUPERSECRET🔐"
+    )
+
+    new_state = migrator.apply_tested_migration(
+        ("digid_eherkenning", "0009_decrypt_private_keys")
+    )
+
+    DigiDConfiguration = new_state.apps.get_model(
+        "digid_eherkenning", "DigiDConfiguration"
+    )
+    config = DigiDConfiguration.objects.get()
+    with config.certificate.private_key.open("rb") as privkey:
+        try:
+            load_pem_x509_private_key(privkey.read(), password=None)
+        except Exception:
+            pytest.fail("Expected private key to be decrypted.")
+
+
+def _decryption_skip_cases_idfn(case):
+    model_name, has_config, passphrase, encrypt_key, *_ = case
+    return f"test_decryption_skip_cases({model_name=} {has_config=} {passphrase=} {encrypt_key=})"
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        # EH has to work too - working setup with the correct key
+        (
+            "EHerkenningConfiguration",
+            True,
+            "SUPERSECRET🔐",
+            True,
+            lambda key, cert: {
+                "public_certificate": File(BytesIO(cert), name="client_cert.pem"),
+                "private_key": File(BytesIO(key), name="client_key.pem"),
+            },
+        ),
+        # No config actually exists
+        (
+            "DigiDConfiguration",
+            False,
+            "",
+            False,
+            lambda *args: None,
+        ),
+        # Config exists, but no cert is configured (required for django-solo to
+        # properly work)
+        (
+            "DigiDConfiguration",
+            True,
+            "foo",
+            False,
+            lambda *args: None,
+        ),
+        # Config exists, certificate only has certificate (no privkey)
+        (
+            "DigiDConfiguration",
+            True,
+            "foo",
+            False,
+            lambda key, cert: {
+                "public_certificate": File(BytesIO(cert), name="client_cert.pem"),
+                "private_key": "",
+            },
+        ),
+        # Config exists, passphrase set, but key is not encrypted
+        (
+            "DigiDConfiguration",
+            True,
+            "foo",
+            False,
+            lambda key, cert: {
+                "public_certificate": File(BytesIO(cert), name="client_cert.pem"),
+                "private_key": File(BytesIO(key), name="client_key.pem"),
+            },
+        ),
+        # Config exists, passphrase is wrong
+        (
+            "DigiDConfiguration",
+            True,
+            "foo",
+            True,
+            lambda key, cert: {
+                "public_certificate": File(BytesIO(cert), name="client_cert.pem"),
+                "private_key": File(BytesIO(key), name="client_key.pem"),
+            },
+        ),
+        # Config exists, passphrase is missing
+        (
+            "DigiDConfiguration",
+            True,
+            "",
+            True,
+            lambda key, cert: {
+                "public_certificate": File(BytesIO(cert), name="client_cert.pem"),
+                "private_key": File(BytesIO(key), name="client_key.pem"),
+            },
+        ),
+    ],
+    ids=_decryption_skip_cases_idfn,
+)
+@pytest.mark.django_db
+def test_decryption_migration_robustness(
+    migrator,
+    leaf_keypair: tuple[rsa.RSAPrivateKey, bytes],
+    encrypted_keypair: tuple[bytes, bytes],
+    case: tuple[
+        Literal["DigiDConfiguration", "EHerkenningConfiguration"],
+        bool,
+        str,
+        bool,
+        Callable[[bytes, bytes], dict | None],
+    ],
+):
+    model_name, has_config, passphrase, encrypt_key, certificate_kwargs_callback = case
+    old_state = migrator.apply_initial_migration(
+        ("digid_eherkenning", "0008_update_loa_fields")
+    )
+    OldConfig = old_state.apps.get_model("digid_eherkenning", model_name)
+    Certificate = old_state.apps.get_model("simple_certmanager", "Certificate")
+    encrypted_key, cert = encrypted_keypair
+    key = encrypted_key if encrypt_key else key_to_pem(leaf_keypair[0], passphrase="")
+
+    certificate_kwargs = certificate_kwargs_callback(key, cert)
+    certificate = (
+        None
+        if certificate_kwargs is None
+        else Certificate.objects.create(
+            label="Test certificate",
+            type=CertificateTypes.key_pair,
+            **certificate_kwargs,
+        )
+    )
+    if has_config:
+        OldConfig.objects.create(certificate=certificate, key_passphrase=passphrase)
+
+    try:
+        migrator.apply_tested_migration(
+            ("digid_eherkenning", "0009_decrypt_private_keys")
+        )
+    except Exception:
+        pytest.fail("Expected migration not to crash")