[fix] Prevent multiple VPN client templates to the same VPN server #832

Fixes #832

Author: pandafy

openwisp_controller/config/base/config.py

@@ -1,6 +1,7 @@
 import collections
 import logging
 import re
+from collections import defaultdict
 
 from cache_memoize import cache_memoize
 from django.core.exceptions import ObjectDoesNotExist, PermissionDenied, ValidationError
@@ -185,6 +186,60 @@ def _get_templates_from_pk_set(cls, pk_set):
             templates = pk_set
         return templates
 
+    @classmethod
+    def clean_duplicate_vpn_client_templates(
+        cls, action, instance, templates, raw_data=None
+    ):
+        """
+        Multiple VPN client templates related to the same VPN server are not allowed:
+        it hardly makes sense, preventing it keeps things simple and avoids headaches.
+        Raises a ValidationError if duplicates are found.
+        """
+        if action != "pre_add":
+            return
+
+        def format_template_list(names):
+            quoted = [f'"{name}"' for name in names]
+            if len(quoted) == 2:
+                return " and ".join(quoted)
+            return ", ".join(quoted[:-1]) + " and " + quoted[-1]
+
+        def add_vpn_templates(templates_queryset):
+            for template in templates_queryset.filter(type="vpn"):
+                if template.name not in vpn_templates[template.vpn.name]:
+                    vpn_templates[template.vpn.name].append(template.name)
+
+        raw_data = raw_data or {}
+        vpn_templates = defaultdict(list)
+        if not raw_data:
+            # When raw_data is present, validation is triggered by a
+            # ConfigForm submission.
+            # In this case, the "templates" queryset already contains only the templates
+            # that are intended to be assigned. Templates that would be removed
+            # (e.g., in a pre_clear action) have already been excluded from the
+            # queryset.
+            add_vpn_templates(instance.templates)
+        add_vpn_templates(templates)
+
+        error_lines = [
+            _(
+                "You cannot select multiple VPN client templates related to"
+                " the same VPN server."
+            )
+        ]
+        for vpn_name, template_names in vpn_templates.items():
+            if len(template_names) < 2:
+                continue
+            template_list = format_template_list(sorted(template_names))
+            error_lines.append(
+                _(
+                    "The templates {template_list} are all linked"
+                    ' to the same VPN server: "{vpn_name}".'
+                ).format(template_list=template_list, vpn_name=vpn_name)
+            )
+        if len(error_lines) > 1:
+            raise ValidationError("\n".join(str(line) for line in error_lines))
+
     @classmethod
     def clean_templates(cls, action, instance, pk_set, raw_data=None, **kwargs):
         """
@@ -203,6 +258,9 @@ def clean_templates(cls, action, instance, pk_set, raw_data=None, **kwargs):
         )
         if not templates:
             return
+        cls.clean_duplicate_vpn_client_templates(
+            action, instance, templates, raw_data=raw_data
+        )
         backend = instance.get_backend_instance(template_instances=templates)
         try:
             cls.clean_netjsonconfig_backend(backend)

openwisp_controller/config/tests/test_api.py

@@ -1551,6 +1551,76 @@ def test_device_detail_api_change_config(self):
             self.assertEqual(response.status_code, 200)
             self.assertEqual(device.config.templates.count(), 0)
 
+    def test_multiple_vpn_client_templates_same_vpn(self):
+        """
+        Assigning multiple templates of type 'vpn' referencing the same VPN
+        to a device's config raises error.
+        """
+        org = self._get_org()
+        vpn = self._create_vpn(organization=org)
+        # Create two templates of type 'vpn' referencing the same VPN
+        vpn_template1 = self._create_template(
+            type="vpn", vpn=vpn, organization=org, name="VPN Client 1"
+        )
+        vpn_template2 = self._create_template(
+            type="vpn", vpn=vpn, organization=org, name="VPN Client 2"
+        )
+        device = self._create_device(organization=org)
+        config = self._create_config(device=device)
+        path = reverse("config_api:device_detail", args=[device.pk])
+        data = {
+            "name": device.name,
+            "organization": str(org.id),
+            "mac_address": device.mac_address,
+            "config": {
+                "backend": "netjsonconfig.OpenWrt",
+                "context": {"lan_ip": "192.168.1.1"},
+                "config": {"interfaces": [{"name": "wlan0", "type": "wireless"}]},
+            },
+        }
+        expected_error_message = (
+            "You cannot select multiple VPN client templates related"
+            " to the same VPN server.\n"
+            'The templates "VPN Client 1" and "VPN Client 2" are all '
+            'linked to the same VPN server: "test".'
+        )
+
+        with self.subTest("Add both templates at once"):
+            data["config"]["templates"] = [str(vpn_template1.pk), str(vpn_template2.pk)]
+            response = self.client.put(path, data, content_type="application/json")
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(
+                str(response.data["config"][0]),
+                expected_error_message,
+            )
+            self.assertEqual(config.templates.count(), 0)
+            self.assertEqual(config.vpnclient_set.count(), 0)
+
+        with self.subTest("Add one template at a time"):
+            data["config"]["templates"] = [str(vpn_template1.pk)]
+            response = self.client.put(path, data, content_type="application/json")
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(config.templates.count(), 1)
+            self.assertEqual(config.vpnclient_set.count(), 1)
+
+            # Now add the second template
+            data["config"]["templates"] = [str(vpn_template1.pk), str(vpn_template2.pk)]
+            response = self.client.put(path, data, content_type="application/json")
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(
+                str(response.data["config"][0]),
+                expected_error_message,
+            )
+            self.assertEqual(config.templates.filter(id=vpn_template1.id).count(), 1)
+            self.assertEqual(config.vpnclient_set.count(), 1)
+
+        with self.subTest("Change existing template with another"):
+            data["config"]["templates"] = [str(vpn_template2.pk)]
+            response = self.client.put(path, data, content_type="application/json")
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(config.templates.filter(id=vpn_template2.id).count(), 1)
+            self.assertEqual(config.vpnclient_set.count(), 1)
+
     def test_device_patch_with_templates_of_same_org(self):
         org1 = self._create_org(name="testorg")
         d1 = self._create_device(name="org1-config", organization=org1)

openwisp_controller/config/tests/test_config.py

@@ -892,6 +892,61 @@ class TestTransactionConfig(
     TestVpnX509Mixin,
     TransactionTestCase,
 ):
+    def test_multiple_vpn_client_templates_same_vpn(self):
+        vpn1 = self._create_vpn(name="vpn1")
+        vpn2 = self._create_vpn(name="vpn2")
+        vpn1_template1 = self._create_template(
+            name="vpn1-template1", type="vpn", vpn=vpn1
+        )
+        vpn1_template2 = self._create_template(
+            name="vpn1-template2", type="vpn", vpn=vpn1
+        )
+        vpn2_template1 = self._create_template(
+            name="vpn2-template1", type="vpn", vpn=vpn2
+        )
+        vpn2_template2 = self._create_template(
+            name="vpn2-template2", type="vpn", vpn=vpn2
+        )
+        vpn2_template3 = self._create_template(
+            name="vpn2-template3", type="vpn", vpn=vpn2
+        )
+        config = self._create_config(device=self._create_device())
+        config.templates.add(vpn1_template1)
+        with self.subTest("Adding duplicate vpn-client template one at time"):
+            with self.assertRaises(ValidationError) as context_manager:
+                config.templates.add(vpn1_template2)
+            try:
+                self.assertEqual(
+                    context_manager.exception.message,
+                    "You cannot select multiple VPN client templates related to the"
+                    " same VPN server.\n"
+                    'The templates "vpn1-template1" and "vpn1-template2" are all'
+                    ' linked to the same VPN server: "vpn1".',
+                )
+            except AssertionError:
+                self.fail("ValidationError not raised")
+
+        with self.subTest("Add multiple vpn client templates for multiple VPN"):
+            config.refresh_from_db()
+            self.assertEqual(config.templates.count(), 1)
+            self.assertEqual(config.vpnclient_set.count(), 1)
+            with self.assertRaises(ValidationError) as context_manager:
+                config.templates.add(
+                    vpn1_template2, vpn2_template1, vpn2_template2, vpn2_template3
+                )
+            try:
+                self.assertEqual(
+                    context_manager.exception.message,
+                    "You cannot select multiple VPN client templates related to the"
+                    " same VPN server.\n"
+                    'The templates "vpn1-template1" and "vpn1-template2" are all'
+                    ' linked to the same VPN server: "vpn1".\n'
+                    'The templates "vpn2-template1", "vpn2-template2" and'
+                    ' "vpn2-template3" are all linked to the same VPN server: "vpn2".',
+                )
+            except AssertionError:
+                self.fail("ValidationError not raised")
+
     def test_certificate_renew_invalidates_checksum_cache(self):
         config = self._create_config(organization=self._get_org())
         vpn_template = self._create_template(

openwisp_controller/config/tests/test_vpn.py

@@ -249,7 +249,7 @@ def test_vpn_client_deletion(self):
         def _assert_vpn_client_cert(cert, vpn_client, cert_ct, vpn_client_ct):
             self.assertEqual(Cert.objects.filter(pk=cert.pk).count(), 1)
             self.assertEqual(VpnClient.objects.filter(pk=vpn_client.pk).count(), 1)
-            vpnclient.delete()
+            c.templates.remove(t)
             self.assertEqual(
                 Cert.objects.filter(pk=cert.pk, revoked=False).count(), cert_ct
             )