[feature] Introduced standalone certificate templates and device bindings

openwisp_controller/config/admin.py

@@ -117,7 +117,12 @@ class Media:
         css = {"all": (f"{prefix}css/admin.css",)}
         js = list(CopyableFieldsAdmin.Media.js) + [
             f"{prefix}js/{file_}"
-            for file_ in ("preview.js", "unsaved_changes.js", "switcher.js")
+            for file_ in (
+                "preview.js",
+                "unsaved_changes.js",
+                "switcher.js",
+                "template_ui.js",
+            )
         ]
 
     def get_extra_context(self, pk=None):
@@ -1059,6 +1064,8 @@ class TemplateAdmin(MultitenantAdminMixin, BaseConfigAdmin, SystemDefinedVariabl
         "organization",
         "type",
         "backend",
+        "ca",
+        "blueprint_cert",
         "default",
         "required",
         "created",
@@ -1073,13 +1080,15 @@ class TemplateAdmin(MultitenantAdminMixin, BaseConfigAdmin, SystemDefinedVariabl
         "created",
     ]
     search_fields = ["name"]
-    multitenant_shared_relations = ("vpn",)
+    multitenant_shared_relations = ("vpn", "ca", "blueprint_cert")
     fields = [
         "name",
         "organization",
         "type",
         "backend",
         "vpn",
+        "ca",
+        "blueprint_cert",
         "auto_cert",
         "tags",
         "default",
@@ -1091,7 +1100,7 @@ class TemplateAdmin(MultitenantAdminMixin, BaseConfigAdmin, SystemDefinedVariabl
         "modified",
     ]
     readonly_fields = ["system_context"]
-    autocomplete_fields = ["vpn"]
+    autocomplete_fields = ["vpn", "ca", "blueprint_cert"]
 
     @admin.action(permissions=["add"])
     def clone_selected_templates(self, request, queryset):

openwisp_controller/config/api/serializers.py

@@ -17,6 +17,7 @@
 DeviceGroup = load_model("config", "DeviceGroup")
 Config = load_model("config", "Config")
 Organization = load_model("openwisp_users", "Organization")
+DeviceCertificate = load_model("config", "DeviceCertificate")
 
 
 class BaseMeta:
@@ -38,6 +39,8 @@ class Meta(BaseMeta):
             "type",
             "backend",
             "vpn",
+            "ca",
+            "blueprint_cert",
             "tags",
             "default",
             "required",
@@ -46,6 +49,16 @@ class Meta(BaseMeta):
             "created",
             "modified",
         ]
+        extra_kwargs = {
+            "blueprint_cert": {
+                "error_messages": {
+                    "does_not_exist": _(
+                        "This certificate does not exist or is already "
+                        "assigned to an active device."
+                    )
+                }
+            }
+        }
 
     def validate_vpn(self, value):
         """
@@ -62,12 +75,73 @@ def validate_config(self, value):
         """
         Display appropriate field name.
         """
-        if self.initial_data.get("type") == "generic" and value == {}:
+        template_type = self.initial_data.get("type")
+        if template_type == "generic" and value == {}:
             raise serializers.ValidationError(
                 _("The configuration field cannot be empty.")
             )
         return value
 
+    def validate(self, data):
+        """
+        Explicitly validate certificate template fields and locks for the API.
+        """
+        template_type = data.get("type", getattr(self.instance, "type", "generic"))
+        ca = data.get("ca", getattr(self.instance, "ca", None))
+        blueprint_cert = data.get(
+            "blueprint_cert", getattr(self.instance, "blueprint_cert", None)
+        )
+        if template_type == "cert" and not ca:
+            raise serializers.ValidationError(
+                {
+                    "ca": _(
+                        "A Certificate Authority is required when "
+                        "the template type is certificate."
+                    )
+                }
+            )
+        elif template_type != "cert":
+            data["ca"] = None
+            data["blueprint_cert"] = None
+        if blueprint_cert and ca:
+            if blueprint_cert.ca_id != ca.id:
+                raise serializers.ValidationError(
+                    {
+                        "blueprint_cert": _(
+                            "The selected certificate must match "
+                            "the selected Certificate Authority."
+                        )
+                    }
+                )
+        if self.instance and self.instance.pk:
+            if Config.objects.filter(templates=self.instance).exists():
+
+                if "ca" in data and data["ca"] != getattr(
+                    self.instance, "ca_id", self.instance.ca
+                ):
+                    raise serializers.ValidationError(
+                        {
+                            "ca": _(
+                                "This template is already assigned to active devices. "
+                                "You cannot change the CA or Blueprint Certificate "
+                                "on an active template."
+                            )
+                        }
+                    )
+                if "blueprint_cert" in data and data["blueprint_cert"] != getattr(
+                    self.instance, "blueprint_cert_id", self.instance.blueprint_cert
+                ):
+                    raise serializers.ValidationError(
+                        {
+                            "blueprint_cert": _(
+                                "This template is already assigned to active devices. "
+                                "You cannot change the CA or Blueprint Certificate "
+                                "on an active template."
+                            )
+                        }
+                    )
+        return super().validate(data)
+
 
 class VpnSerializer(BaseSerializer):
     config = serializers.JSONField(initial={})
@@ -203,6 +277,9 @@ def _update_config(self, device, config_data):
                     vpn_list = config.templates.filter(type="vpn").values_list("vpn")
                     if vpn_list:
                         config.vpnclient_set.exclude(vpn__in=vpn_list).delete()
+                    DeviceCertificate.objects.filter(config=config).exclude(
+                        template_id__in=config_templates
+                    ).delete()
                     config.templates.set(config_templates, clear=True)
             config.save()
         except ValidationError as error:

openwisp_controller/config/base/config.py

@@ -3,6 +3,7 @@
 import re
 from collections import defaultdict
 
+import swapper
 from cache_memoize import cache_memoize
 from django.core.cache import cache
 from django.core.exceptions import ObjectDoesNotExist, PermissionDenied, ValidationError
@@ -63,6 +64,13 @@ class AbstractConfig(ChecksumCacheMixin, BaseConfig):
         related_name="vpn_relations",
         blank=True,
     )
+    device_certificates = models.ManyToManyField(
+        swapper.get_model_name("config", "Template"),
+        through=swapper.get_model_name("config", "DeviceCertificate"),
+        related_name="config_device_certificates",
+        blank=True,
+        verbose_name=_("device certificates"),
+    )
 
     STATUS = Choices("modified", "applied", "error", "deactivating", "deactivated")
     status = StatusField(

openwisp_controller/config/base/device_certificate.py

@@ -0,0 +1,41 @@
+from django.core.exceptions import ValidationError
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+from swapper import get_model_name, load_model
+
+
+class AbstractDeviceCertificate(models.Model):
+    config = models.ForeignKey(
+        get_model_name("config", "Config"), on_delete=models.CASCADE
+    )
+    template = models.ForeignKey(
+        get_model_name("config", "Template"), on_delete=models.CASCADE
+    )
+    cert = models.OneToOneField(
+        get_model_name("django_x509", "Cert"), on_delete=models.CASCADE
+    )
+    auto_cert = models.BooleanField(default=False)
+
+    class Meta:
+        abstract = True
+        unique_together = ("config", "template")
+        verbose_name = _("Device certificate")
+        verbose_name_plural = _("Device certificates")
+
+    def __str__(self):
+        return f"{self.config.device.name} - {self.cert.name}"
+
+    def clean(self):
+        Template = load_model("config", "Template")
+        if (
+            self.cert_id
+            and Template.objects.filter(blueprint_cert_id=self.cert_id).exists()
+        ):
+            raise ValidationError(
+                {
+                    "cert": _(
+                        "This certificate is currently used as a blueprint "
+                        "by a template and cannot be directly assigned to a device."
+                    )
+                }
+            )

openwisp_controller/config/base/template.py

@@ -22,7 +22,11 @@
 
 logger = logging.getLogger(__name__)
 
-TYPE_CHOICES = (("generic", _("Generic")), ("vpn", _("VPN-client")))
+TYPE_CHOICES = (
+    ("generic", _("Generic")),
+    ("vpn", _("VPN-client")),
+    ("cert", _("Certificate")),
+)
 
 
 def default_auto_cert():
@@ -55,6 +59,34 @@ class AbstractTemplate(ShareableOrgMixinUniqueName, BaseConfig):
         null=True,
         on_delete=models.CASCADE,
     )
+    ca = models.ForeignKey(
+        get_model_name("django_x509", "Ca"),
+        on_delete=models.CASCADE,
+        verbose_name=_("Certificate Authority"),
+        blank=True,
+        null=True,
+        help_text=_(
+            "The Certificate Authority that will sign certificates generated "
+            "by this template."
+        ),
+    )
+
+    def get_unassigned_certs():
+        Cert = load_model("django_x509", "Cert")
+        return {"pk__in": Cert.objects.exclude(devicecertificate__isnull=False)}
+
+    blueprint_cert = models.ForeignKey(
+        get_model_name("django_x509", "Cert"),
+        on_delete=models.SET_NULL,
+        verbose_name=_("Blueprint Certificate"),
+        blank=True,
+        null=True,
+        limit_choices_to=get_unassigned_certs,
+        help_text=_(
+            "Optional: Select an unassigned certificate to copy extensions and "
+            "properties from."
+        ),
+    )
     type = models.CharField(
         _("type"),
         max_length=16,
@@ -90,7 +122,8 @@ class AbstractTemplate(ShareableOrgMixinUniqueName, BaseConfig):
         help_text=_(
             "whether tunnel specific configuration (cryptographic keys, ip addresses, "
             "etc) should be automatically generated and managed behind the scenes "
-            "for each configuration using this template, valid only for the VPN type"
+            "for each configuration using this template, valid only for the VPN and "
+            "certificate template types"
         ),
     )
     default_values = JSONField(
@@ -221,6 +254,26 @@ def clean(self, *args, **kwargs):
         * automatically determines configuration if necessary
         * if flagged as required forces it also to be default
         """
+        if not self._state.adding:
+            try:
+                current = self.__class__.objects.get(pk=self.pk)
+                if (
+                    current.ca_id != self.ca_id
+                    or current.blueprint_cert_id != self.blueprint_cert_id
+                ):
+                    Config = load_model("config", "Config")
+                    if Config.objects.filter(templates=self).exists():
+                        raise ValidationError(
+                            {
+                                "ca": _(
+                                    "This template is already assigned "
+                                    "to active devices. You cannot change the CA "
+                                    "or Blueprint Certificate on an active template."
+                                )
+                            }
+                        )
+            except self.__class__.DoesNotExist:
+                pass
         self._validate_org_relation("vpn")
         if not self.default_values:
             self.default_values = {}
@@ -234,15 +287,56 @@ def clean(self, *args, **kwargs):
             )
         elif self.type != "vpn":
             self.vpn = None
-            self.auto_cert = False
+            if self.type != "cert":
+                self.auto_cert = False
         if self.type == "vpn" and not self.config:
             self.config = self.vpn.auto_client(
                 auto_cert=self.auto_cert, template_backend_class=self.backend_class
             )
+        if self.type == "cert":
+            self._validate_org_relation("ca")
+            self._validate_org_relation("blueprint_cert")
+            if not self.ca:
+                raise ValidationError(
+                    {
+                        "ca": _(
+                            "A Certificate Authority is required when the template "
+                            "type is certificate."
+                        )
+                    }
+                )
+            if self.blueprint_cert and self.blueprint_cert.ca_id != self.ca_id:
+                raise ValidationError(
+                    {
+                        "blueprint_cert": _(
+                            "The selected certificate must match the selected "
+                            "Certificate Authority."
+                        )
+                    }
+                )
+            if self.blueprint_cert_id:
+                DeviceCertificate = load_model("config", "DeviceCertificate")
+                if DeviceCertificate.objects.filter(
+                    cert_id=self.blueprint_cert_id
+                ).exists():
+                    raise ValidationError(
+                        {
+                            "blueprint_cert": _(
+                                "This certificate is already assigned to a device. "
+                                "Please select an unassigned certificate to "
+                                "use as a blueprint."
+                            )
+                        }
+                    )
+            if self.config is None:
+                self.config = {}
+        else:
+            self.ca = None
+            self.blueprint_cert = None
         if self.required and not self.default:
             self.default = True
         super().clean(*args, **kwargs)
-        if not self.config:
+        if not self.config and self.type != "cert":
             raise ValidationError(_("The configuration field cannot be empty."))
 
     def get_context(self, system=False):