From 89839460210949c1ab59c57a3fbfcbc1b03b13d3 Mon Sep 17 00:00:00 2001 From: Sameer Randive Date: Thu, 9 Jul 2026 11:39:58 +0530 Subject: [PATCH] [fix] Prevent modifications to persisted VpnClient objects #1428 Make VpnClient objects immutable after creation while allowing managed IP updates. Fixes #1428 --- openwisp_controller/config/base/vpn.py | 58 +++++++++-- openwisp_controller/config/tests/test_vpn.py | 96 ++++++++++++++++++- .../subnet_division/rule_types/vpn.py | 3 +- 3 files changed, 146 insertions(+), 11 deletions(-) diff --git a/openwisp_controller/config/base/vpn.py b/openwisp_controller/config/base/vpn.py index f3270c852..411fdf0b3 100644 --- a/openwisp_controller/config/base/vpn.py +++ b/openwisp_controller/config/base/vpn.py @@ -831,6 +831,11 @@ class AbstractVpnClient(models.Model): db_index=True, ) _auto_ip_stopper_funcs = [] + # Fields which the internal provisioning machinery may legitimately + # update after creation (subnet_division reassigns the provisioned IP + # asynchronously); every other field is immutable once persisted: + # to change a VPN client, remove the VPN template and add it again. + _updatable_fields = ("ip",) class Meta: abstract = True @@ -867,14 +872,55 @@ def _get_unique_checks(self, exclude=None, include_meta_constraints=False): unique_checks.append((self.__class__, ("vpn", "vni"))) return unique_checks, date_checks + def clean(self): + super().clean() + self._check_immutable_fields() + + def _check_immutable_fields(self): + """Raises ValidationError if fields of a persisted instance changed. + + VPN clients are immutable: they are created when a VPN template + is applied to a configuration and deleted when it is removed; + allowing updates would desynchronize the provisioned artifacts + (certificate, IP, keys) from the VPN server. + """ + if self._state.adding: + return + current = self._meta.model.objects.filter(pk=self.pk).first() + if current is None: + return + changed_fields = [ + field.name + for field in self._meta.concrete_fields + if not field.primary_key + and field.name not in self._updatable_fields + and getattr(self, field.attname) != getattr(current, field.attname) + ] + if changed_fields: + raise ValidationError( + { + field: _( + "VPN clients cannot be modified after creation. " + "To change this VPN client, remove the VPN template " + "and add it again." + ) + for field in changed_fields + } + ) + def save(self, *args, **kwargs): """Performs automatic provisioning if ``auto_cert`` is True.""" - if self.auto_cert: - self._auto_x509() - self._auto_ip() - self._auto_wireguard() - self._auto_vxlan() - self._auto_secret() + if self._state.adding: + if self.auto_cert: + self._auto_x509() + self._auto_ip() + self._auto_wireguard() + self._auto_vxlan() + self._auto_secret() + else: + # not called automatically by save(): enforce immutability + # also for direct ORM saves which skip full_clean() + self._check_immutable_fields() super().save(*args, **kwargs) def _auto_x509(self): diff --git a/openwisp_controller/config/tests/test_vpn.py b/openwisp_controller/config/tests/test_vpn.py index 211ad4a06..05f5c35e8 100644 --- a/openwisp_controller/config/tests/test_vpn.py +++ b/openwisp_controller/config/tests/test_vpn.py @@ -272,19 +272,34 @@ def _assert_vpn_client_cert(cert, vpn_client, cert_ct, vpn_client_ct): _assert_vpn_client_cert(cert, vpnclient, 0, 0) with self.subTest( - 'Test VpnClient post_delete handler when "auto_cert" field is set to "False"' # noqa + 'Changing "auto_cert" after creation is rejected (immutability)' ): t = self._create_template( name="vpn-test-2", type="vpn", vpn=vpn, auto_cert=True ) c.templates.add(t) vpnclient = c.vpnclient_set.first() - cert = vpnclient.cert - # Set auto_cert field to false vpnclient.auto_cert = False + with self.assertRaises(ValidationError) as context_manager: + vpnclient.full_clean() + self.assertIn("auto_cert", context_manager.exception.message_dict) + c.templates.remove(t) + + with self.subTest( + 'Test VpnClient post_delete handler when "auto_cert" field is set to "False"' # noqa + ): + t = self._create_template( + name="vpn-test-3", type="vpn", vpn=vpn, auto_cert=False + ) + cert = self._create_cert(name="manual-cert", ca=vpn.ca, organization=org) + vpnclient = VpnClient( + config=c, vpn=vpn, template=t, cert=cert, auto_cert=False + ) vpnclient.full_clean() vpnclient.save() - _assert_vpn_client_cert(cert, vpnclient, 1, 0) + vpnclient.delete() + # the certificate must not be revoked because auto_cert is False + self.assertEqual(Cert.objects.filter(pk=cert.pk, revoked=False).count(), 1) def test_vpn_client_get_common_name(self): vpn = self._create_vpn() @@ -494,6 +509,79 @@ def test_cert_validation(self): self.assertIn("ca", message_dict) self.assertIn("CA is required with this VPN backend", message_dict["ca"]) + def _create_vpn_client_via_template(self): + vpn = self._create_vpn() + template = self._create_template( + name="vpn-immutability", type="vpn", vpn=vpn, auto_cert=True + ) + config = self._create_config(device=self._create_device()) + config.templates.add(template) + return config.vpnclient_set.select_related("vpn", "cert", "config").first() + + def test_vpn_client_creation_allowed(self): + client = self._create_vpn_client_via_template() + self.assertIsNotNone(client) + self.assertIsNotNone(client.cert) + self.assertTrue(client.auto_cert) + + def test_vpn_client_noop_save_allowed(self): + client = self._create_vpn_client_via_template() + cert_count = Cert.objects.count() + client.full_clean() + client.save() + self.assertEqual(VpnClient.objects.count(), 1) + self.assertEqual(Cert.objects.count(), cert_count) + + def test_vpn_client_immutable_after_creation(self): + client = self._create_vpn_client_via_template() + vpn2 = self._create_vpn(name="immutability-vpn2", ca=client.vpn.ca) + for field, value in ( + ("auto_cert", False), + ("vpn", vpn2), + ("public_key", "changed-key"), + ("secret", "changed-secret"), + ("vni", 42), + ): + with self.subTest(field=field): + client.refresh_from_db() + setattr(client, field, value) + with self.assertRaises(ValidationError) as context_manager: + client.full_clean() + self.assertIn(field, context_manager.exception.message_dict) + + def test_vpn_client_direct_save_rejected(self): + # save() without full_clean() must also enforce immutability + client = self._create_vpn_client_via_template() + client.auto_cert = False + with self.assertRaises(ValidationError): + client.save() + client.refresh_from_db() + self.assertTrue(client.auto_cert) + + def test_vpn_client_ip_update_allowed(self): + # internal machinery (subnet_division) must be able to + # reassign the provisioned IP after creation + client = self._create_vpn_client_via_template() + subnet = Subnet(name="immutability-subnet", subnet="10.99.0.0/24") + subnet.full_clean() + subnet.save() + ip = subnet.request_ip() + client.ip = ip + client.full_clean() + client.save(update_fields=["ip"]) + client.refresh_from_db() + self.assertEqual(client.ip, ip) + + def test_vpn_client_delete_recreate_flow(self): + client = self._create_vpn_client_via_template() + config, template = client.config, client.template + config.templates.remove(template) + self.assertEqual(config.vpnclient_set.count(), 0) + config.templates.add(template) + new_client = config.vpnclient_set.first() + self.assertIsNotNone(new_client) + self.assertNotEqual(new_client.pk, client.pk) + class TestVpnTransaction(BaseTestVpn, TestWireguardVpnMixin, TransactionTestCase): @mock.patch.object(create_vpn_dh, "delay") diff --git a/openwisp_controller/subnet_division/rule_types/vpn.py b/openwisp_controller/subnet_division/rule_types/vpn.py index 2680f89fb..a50010041 100644 --- a/openwisp_controller/subnet_division/rule_types/vpn.py +++ b/openwisp_controller/subnet_division/rule_types/vpn.py @@ -52,7 +52,8 @@ def post_provision_handler(cls, instance, provisioned, **kwargs): instance.ip.delete() instance.ip = provisioned["ip_addresses"][0] instance.full_clean() - instance.save() + # "ip" is the only field VpnClient allows updating after creation + instance.save(update_fields=["ip"]) @classmethod def destroy_provisioned_subnets_ips(cls, instance, **kwargs):