[chores] Updated DeviceCertificate model and tests

Added selenium tests and refactored DeviceCertificate model

Author: stktyagi

openwisp_controller/config/base/device_certificate.py

@@ -9,6 +9,9 @@
 from openwisp_controller.config import settings as app_settings
 from openwisp_utils.base import TimeStampedEditableModel
 
+MAC_ADDRESS_OID = "1.3.6.1.4.1.65901.1"
+DEVICE_UUID_OID = "1.3.6.1.4.1.65901.2"
+
 
 class AbstractDeviceCertificate(TimeStampedEditableModel):
     config = models.ForeignKey(
@@ -84,63 +87,69 @@ def _get_common_name(self):
         common_name = f"{common_name}-{unique_slug}"
         return common_name
 
-    def _auto_create_cert(self, name, common_name):
-        """
-        Automatically creates and assigns a client x509 certificate
-        using Blueprint cloning and custom hardware OID injection.
-        """
+    def _build_cert(self, name, common_name):
+        """Build (but do not save) a Cert instance from template + blueprint."""
         ca = self.template.ca
         blueprint = self.template.blueprint_cert
-        device = self.config.device
         cert_model = self.__class__.cert.field.related_model
 
-        # blueprint property cloning with CA fallback
-        key_length = blueprint.key_length if blueprint else ca.key_length
-        digest = blueprint.digest if blueprint else str(ca.digest)
-        country_code = blueprint.country_code if blueprint else ca.country_code
-        state = blueprint.state if blueprint else ca.state
-        city = blueprint.city if blueprint else ca.city
-        organization_name = (
-            blueprint.organization_name if blueprint else ca.organization_name
+        attrs = self._clone_blueprint_attrs(ca, blueprint)
+        extensions = self._build_extensions(blueprint)
+        cert = cert_model(
+            name=name,
+            ca=ca,
+            common_name=common_name,
+            extensions=extensions,
+            **attrs,
+        )
+        return self._auto_create_cert_extra(cert)
+
+    def _clone_blueprint_attrs(self, ca, blueprint):
+        """
+        Extracts base X.509 attributes (such as key length, digest, and
+        location data) from the provided blueprint certificate.
+        """
+        source = blueprint or ca
+        digest = str(source.digest) if not blueprint else source.digest
+        return dict(
+            key_length=source.key_length,
+            digest=digest,
+            country_code=source.country_code,
+            state=source.state,
+            city=source.city,
+            organization_name=source.organization_name,
+            email=source.email,
         )
-        email = blueprint.email if blueprint else ca.email
 
+    def _build_extensions(self, blueprint):
+        """Compiles the list of X.509 extensions for the new certificate."""
         if blueprint and blueprint.extensions:
             extensions = copy.deepcopy(blueprint.extensions)
         else:
             extensions = [{"name": "nsCertType", "value": "client", "critical": False}]
+        extensions.extend(self._get_hardware_oid_extensions())
+        return extensions
 
-        # inject MAC and UUID as custom OIDs, prerequisite: #228 in django-x509
-        mac_oid = "1.3.6.1.4.1.65901.1"
-        uuid_oid = "1.3.6.1.4.1.65901.2"
-        extensions.extend(
-            [
-                {
-                    "oid": mac_oid,
-                    "value": f"ASN1:UTF8:string:{device.mac_address}",
-                    "critical": False,
-                },
-                {
-                    "oid": uuid_oid,
-                    "value": f"ASN1:UTF8:string:{device.id}",
-                    "critical": False,
-                },
-            ]
-        )
-        cert = cert_model(
-            name=name,
-            ca=ca,
-            key_length=key_length,
-            digest=digest,
-            country_code=country_code,
-            state=state,
-            city=city,
-            organization_name=organization_name,
-            email=email,
-            common_name=common_name,
-            extensions=extensions,
-        )
-        cert = self._auto_create_cert_extra(cert)
+    def _get_hardware_oid_extensions(self):
+        device = self.config.device
+        return [
+            {
+                "oid": MAC_ADDRESS_OID,
+                "value": f"ASN1:UTF8:string:{device.mac_address}",
+                "critical": False,
+            },
+            {
+                "oid": DEVICE_UUID_OID,
+                "value": f"ASN1:UTF8:string:{device.id}",
+                "critical": False,
+            },
+        ]
+
+    def _auto_create_cert(self, name, common_name):
+        """
+        Automatically creates and assigns a client x509 certificate
+        """
+        cert = self._build_cert(name=name, common_name=common_name)
         cert.full_clean()
         cert.save()
         self.cert = cert

openwisp_controller/config/handlers.py

@@ -45,6 +45,12 @@ def device_registered_notification(sender, instance, is_new, **kwargs):
     )
 
 
+def _hardware_fields_changed(update_fields):
+    return update_fields is None or (
+        "name" in update_fields or "mac_address" in update_fields
+    )
+
+
 @receiver(pre_save, sender=Device, dispatch_uid="capture_old_hardware_properties")
 def capture_old_hardware_properties(sender, instance, **kwargs):
     """
@@ -53,10 +59,8 @@ def capture_old_hardware_properties(sender, instance, **kwargs):
     """
     if not instance.pk:
         return
-    update_fields = kwargs.get("update_fields")
-    if update_fields is not None:
-        if "name" not in update_fields and "mac_address" not in update_fields:
-            return
+    if not _hardware_fields_changed(kwargs.get("update_fields")):
+        return
     try:
         old_instance = sender.objects.only("name", "mac_address").get(pk=instance.pk)
         instance._old_name = old_instance.name
@@ -72,10 +76,8 @@ def detect_hardware_drift(sender, instance, created, **kwargs):
     """
     if created or not app_settings.REGENERATE_CERTS_ON_HARDWARE_CHANGE:
         return
-    update_fields = kwargs.get("update_fields")
-    if update_fields is not None:
-        if "name" not in update_fields and "mac_address" not in update_fields:
-            return
+    if not _hardware_fields_changed(kwargs.get("update_fields")):
+        return
     name_changed = getattr(instance, "_old_name", instance.name) != instance.name
     mac_changed = (
         getattr(instance, "_old_mac", instance.mac_address) != instance.mac_address

openwisp_controller/config/tasks.py

@@ -1,4 +1,3 @@
-import copy
 import logging
 
 import requests
@@ -233,7 +232,6 @@ def regenerate_device_certificates_task(device_id, expected_cert_ids=None):
         return
     Device = load_model("config", "Device")
     DeviceCertificate = load_model("config", "DeviceCertificate")
-    Cert = load_model("django_x509", "Cert")
     try:
         device = Device.objects.get(id=device_id)
     except Device.DoesNotExist:
@@ -262,60 +260,19 @@ def regenerate_device_certificates_task(device_id, expected_cert_ids=None):
                 continue
             old_cert = dc.cert
             old_cert.revoke()
-            blueprint = dc.template.blueprint_cert
-            if blueprint and blueprint.extensions:
-                extensions = copy.deepcopy(blueprint.extensions)
-            else:
-                extensions = [
-                    {"name": "nsCertType", "value": "client", "critical": False}
-                ]
-            ca = dc.template.ca
-            key_length = blueprint.key_length if blueprint else ca.key_length
-            digest = blueprint.digest if blueprint else str(ca.digest)
-            country_code = blueprint.country_code if blueprint else ca.country_code
-            state = blueprint.state if blueprint else ca.state
-            city = blueprint.city if blueprint else ca.city
-            organization_name = (
-                blueprint.organization_name if blueprint else ca.organization_name
-            )
-            email = blueprint.email if blueprint else ca.email
-            extensions.extend(
-                [
-                    {
-                        "oid": "1.3.6.1.4.1.65901.1",
-                        "value": f"ASN1:UTF8:string:{device.mac_address}",
-                        "critical": False,
-                    },
-                    {
-                        "oid": "1.3.6.1.4.1.65901.2",
-                        "value": f"ASN1:UTF8:string:{device.id}",
-                        "critical": False,
-                    },
-                ]
-            )
-            new_cert = Cert(
-                name=device.name,
-                ca=ca,
-                key_length=key_length,
-                digest=digest,
-                country_code=country_code,
-                state=state,
-                city=city,
-                organization_name=organization_name,
-                email=email,
-                common_name=dc._get_common_name(),
-                extensions=extensions,
+            new_cert = dc._build_cert(
+                name=device.name, common_name=dc._get_common_name()
             )
-            new_cert = dc._auto_create_cert_extra(new_cert)
             new_cert.full_clean()
             new_cert.save()
             dc.cert = new_cert
             dc.save()
             configs_to_update.add(dc.config)
             certs_regenerated += 1
     for config in configs_to_update:
+        config.refresh_from_db()
         config.update_status_if_checksum_changed()
-    if certs_regenerated:
+    if certs_regenerated > 0:
         try:
             message = _(
                 "Hardware drift detected on device {device_name}. "

openwisp_controller/config/tests/test_selenium.py

@@ -22,6 +22,9 @@
 Device = load_model("config", "Device")
 DeviceGroup = load_model("config", "DeviceGroup")
 Cert = load_model("django_x509", "Cert")
+Ca = load_model("django_x509", "Ca")
+DeviceCertificate = load_model("config", "DeviceCertificate")
+Notification = load_model("openwisp_notifications", "Notification")
 
 
 class SeleniumTestMixin(BaseSeleniumTestMixin):
@@ -486,6 +489,70 @@ def test_relevant_templates_duplicates(self):
             self.find_element(by=By.NAME, value="_save").click()
             self.wait_for_presence(By.CSS_SELECTOR, ".messagelist .success", timeout=5)
 
+    def test_e2e_certificate_provisioning(self):
+        """
+        End-to-end flow: create CA, create certificate
+        template, assign to device, verify certificate generation.
+        """
+        org = self._get_org()
+        ca = Ca.objects.create(name="test-ca", organization=org)
+        template = self._create_template(
+            organization=org, type="cert", ca=ca, auto_cert=True
+        )
+        device = self._create_device(organization=org, name="e2e-router")
+        self._create_config(device=device)
+
+        self.login()
+        self.open(
+            reverse(f"admin:{self.config_app_label}_device_change", args=[device.id])
+            + "#config-group"
+        )
+        self.hide_loading_overlay()
+        self.find_element(by=By.XPATH, value=f'//*[@value="{template.id}"]').click()
+        self.web_driver.execute_script(
+            'document.querySelector("#ow-user-tools").style.display="none"'
+        )
+        self.find_element(by=By.NAME, value="_continue").click()
+        self.wait_for_presence(By.CSS_SELECTOR, ".messagelist .success", timeout=5)
+        device.config.refresh_from_db()
+        device_cert = DeviceCertificate.objects.get(
+            config=device.config, template=template
+        )
+        self.assertIsNotNone(
+            device_cert.cert, "Certificate was not generated after UI assignment!"
+        )
+        self.assertFalse(device_cert.cert.revoked)
+        self.assertEqual(device_cert.cert.name, "e2e-router")
+
+    def test_hardware_drift_notification(self):
+        org = self._get_org()
+        ca = Ca.objects.create(name="test-ca", organization=org)
+        template = self._create_template(
+            organization=org, type="cert", ca=ca, auto_cert=True
+        )
+        device = self._create_device(organization=org, name="old-router-name")
+        config = self._create_config(device=device)
+        config.templates.add(template)
+        self.login()
+        self.open(
+            reverse(f"admin:{self.config_app_label}_device_change", args=[device.id])
+        )
+        self.hide_loading_overlay()
+        name_input = self.find_element(by=By.NAME, value="name", wait_for="presence")
+        name_input.clear()
+        name_input.send_keys("renamed-router")
+        # Hide user tools because it covers the save button
+        self.web_driver.execute_script(
+            'document.querySelector("#ow-user-tools").style.display="none"'
+        )
+        self.find_element(by=By.NAME, value="_save").click()
+        self.wait_for_presence(By.CSS_SELECTOR, ".messagelist .success", timeout=5)
+        self.find_element(by=By.ID, value="openwisp_notifications").click()
+        notification = self.wait_for_visibility(
+            By.CLASS_NAME, "ow-notification-elem", timeout=10
+        )
+        self.assertIn("Hardware drift detected", notification.text)
+
 
 @tag("selenium_tests")
 class TestDeviceGroupAdmin(