Skip to content

Fix #1428: Make VpnClient objects immutable after creation #1431

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sameerrandive-dev wants to merge 1 commit into
base: master
Choose a base branch
from
+146 −11
Open

Fix #1428: Make VpnClient objects immutable after creation #1431

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 52 additions & 6 deletions openwisp_controller/config/base/vpn.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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()
Comment on lines 911 to +923

@coderabbitai coderabbitai Bot Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Verify whether repository tests cover saving a newly constructed VpnClient
# with an existing primary key and immutable-field changes.
rg -n -C3 'VpnClient\([^)]*pk=|VpnClient\([^)]*id=' openwisp_controller || true
rg -n -C3 '_state\.adding|force_insert|update_fields' openwisp_controller/config openwisp_controller/subnet_division || true

Repository: openwisp/openwisp-controller

Length of output: 36210

🌐 Web query:

For the Django version supported by this repository, how does Model.save() behave when a new model instance has a manually supplied primary key matching an existing row?

💡 Result:

In Django, when you manually supply a primary key value for a new model instance that matches an existing row in the database, Django assumes you intend to update the existing record rather than create a new one [1][2]. When the save method is called, Django uses an internal algorithm to decide whether to execute an SQL INSERT or UPDATE statement [1][3]. For models where the primary key does not define a default or db_default, Django checks the primary key attribute: - If the primary key is set to anything other than None, Django attempts an UPDATE [1][2]. - If the UPDATE fails to affect any rows (for instance, because the provided primary key does not actually exist in the database), Django then executes an INSERT [1][2]. For models that do define a default or db_default on the primary key, Django executes an UPDATE if the instance is considered "existing" and the primary key matches a value currently in the database; otherwise, it executes an INSERT [1][4]. Because of this behavior, it is strongly recommended to avoid manually specifying primary key values for new objects unless you can guarantee that the value is not already in use [1][2]. If you unintentionally use a primary key that matches an existing record, you will overwrite the data in that row [1]. If you instead intend to create a new record and ensure no overwrite occurs, you can use the force_insert=True argument with the save method [5].

Citations:

Don't rely on _state.adding alone here. A freshly constructed VpnClient(pk=<existing pk>, ...) can still take this branch, so save() may skip _check_immutable_fields() even when Django updates an existing row. Check whether self.pk already exists before treating it as a create.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@openwisp_controller/config/base/vpn.py` around lines 911 - 923, The save()
method in VpnClient currently uses only _state.adding to decide whether this is
a create, which can miss cases where a new instance is constructed with an
existing pk. Update the branching in save() to treat objects with an existing
self.pk as updates, and ensure _check_immutable_fields() is called for those
cases even when _state.adding is still true. Keep the auto-provisioning path in
save() for true creates, but gate it on both _state.adding and absence of an
existing primary key.

super().save(*args, **kwargs)

def _auto_x509(self):
Expand Down
96 changes: 92 additions & 4 deletions openwisp_controller/config/tests/test_vpn.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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")
Expand Down
3 changes: 2 additions & 1 deletion openwisp_controller/subnet_division/rule_types/vpn.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
Loading