[fix] Do not allow applying more than 1 templates of 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,55 @@ def _get_templates_from_pk_set(cls, pk_set):
             templates = pk_set
         return templates
 
+    @classmethod
+    def validate_duplicate_vpn_templates(
+        cls, action, instance, templates, raw_data=None
+    ):
+        """
+        Validates if there are duplicate templates for the same VPN server.
+        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(
+                f'The templates {template_list} are all linked'
+                f' to the same VPN server: "{vpn_name}".'
+            )
+        if len(error_lines) > 1:
+            raise ValidationError('\n'.join(error_lines))
+
     @classmethod
     def clean_templates(cls, action, instance, pk_set, raw_data=None, **kwargs):
         """
@@ -203,6 +253,9 @@ def clean_templates(cls, action, instance, pk_set, raw_data=None, **kwargs):
         )
         if not templates:
             return
+        cls.validate_duplicate_vpn_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_templates_same_vpn(self):
+        """
+        Ensure that assigning multiple templates of type 'vpn' referencing the same VPN
+        to a device's config does not create duplicate VpnClient objects.
+        """
+        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,55 @@ class TestTransactionConfig(
     TestVpnX509Mixin,
     TransactionTestCase,
 ):
+    def test_multiple_vpn_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 template one by one'):
+            with self.assertRaises(ValidationError) as context_manager:
+                config.templates.add(vpn1_template2)
+            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".',
+            )
+
+        with self.subTest('Add duplicate 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
+                )
+            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".',
+            )
+
     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
             )