From c8b0cb3a4ad9d4b097b009d6eab80f25a77f317a Mon Sep 17 00:00:00 2001
From: catborise <catborise@gmail.com>
Date: Mon, 19 Oct 2026 12:09:34 +0300
Subject: [PATCH] fix(storages): pool sources take real device and export
 paths, a leading special character is refused

The source check of a new logical or NFS pool accepted only letters,
digits and "/", which refused paths such as /dev/mapper/vg0-data,
/dev/disk/by-id/wwn-..., /dev/disk/by-path/pci-0000:00:1f.2-ata-1 or
/export/vm_images. The target check let any first character but "-"
through, so a quote, "<" or "&" reached the pool XML and libvirt failed
on it.

Target and source now share one expression: letters, digits, "/", "_",
".", ":" and "-" (not first, not right after "/"), still in linear time.
The Vagrant computes' NFS export becomes /srv/wvc-nfs, created through the
page in the live pool test.
---
 README.md                          |  6 +++---
 Vagrantfile                        |  6 +++---
 storages/forms.py                  | 13 ++++++++-----
 storages/tests/test_pool_target.py | 31 +++++++++++++++++++++++++++++-
 4 files changed, 44 insertions(+), 12 deletions(-)

diff --git a/README.md b/README.md
index 21f4a719..5d3f46e5 100644
--- a/README.md
+++ b/README.md
@@ -703,20 +703,20 @@ export TEST_LIBVIRT_LOGIN=root TEST_LIBVIRT_PASSWORD=   # login/password for TCP
 export TEST_LIBVIRT_GUEST_IMAGE=/var/lib/libvirt/images/linux.qcow2   # optional, see below
 export TEST_LIBVIRT_LVM_DEVICE=/dev/vdb   # optional: an empty disk of 4 GiB or more on the host, wiped by the LVM tests
 export TEST_LIBVIRT_ISCSI_TARGET=iqn.2026-10.test.wvc:lun   # optional: a target on the host's 127.0.0.1, open to any initiator, with a LUN
-export TEST_LIBVIRT_NFS_EXPORT=/srv/wvcnfs   # optional: a directory the host exports to its 127.0.0.1 over NFS
+export TEST_LIBVIRT_NFS_EXPORT=/srv/wvc-nfs   # optional: a directory the host exports to its 127.0.0.1 over NFS
 python manage.py test
 ```
 
 `TEST_LIBVIRT_GUEST_IMAGE` enables the tests that need a guest OS (disk and NIC hot-unplug, vCPU hotplug, guest agent, ACPI power off). It must be an installed, BIOS-bootable Linux qcow2 image with qemu-guest-agent, in a storage pool of the host other than `wvc-test`; the tests boot copies of it from overlays and only read the image. Shut down any VM using it first.
 
-Two disposable compute nodes for these tests come from the `Vagrantfile`: AlmaLinux 10 with nested KVM, set up by `dev/libvirt-bootstrap.sh` (with this checkout's gstfsd), a SASL user for TCP (`wvc`, password `wvc-test`) and your `~/.ssh/id_ed25519.pub` for root; migration runs between them, and each has an empty second disk (`/dev/vdb`) for the LVM tests and serves a 1 GiB iSCSI LUN on 127.0.0.1 for the iSCSI test and an NFS export (`/srv/wvcnfs`) to 127.0.0.1 for the NFS pool test. They accept TLS logins (SASL) with certificates from a throwaway CA that `vagrant up` makes in `.vagrant/tls`; the TLS run takes its client certificate from there through `HOME`. It needs Vagrant with the `vagrant-libvirt` plugin, a KVM host with `qemu-kvm` and `cyrus-sasl-md5`, and a user in the `libvirt` group. On a host that also runs Docker, let the VMs' network (the bridge of `virsh net-info vagrant-libvirt`, `virbr1` here) through Docker's forward rules (`sudo iptables -I DOCKER-USER -i virbr1 -j ACCEPT` and `sudo iptables -I DOCKER-USER -o virbr1 -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT`).
+Two disposable compute nodes for these tests come from the `Vagrantfile`: AlmaLinux 10 with nested KVM, set up by `dev/libvirt-bootstrap.sh` (with this checkout's gstfsd), a SASL user for TCP (`wvc`, password `wvc-test`) and your `~/.ssh/id_ed25519.pub` for root; migration runs between them, and each has an empty second disk (`/dev/vdb`) for the LVM tests and serves a 1 GiB iSCSI LUN on 127.0.0.1 for the iSCSI test and an NFS export (`/srv/wvc-nfs`) to 127.0.0.1 for the NFS pool test. They accept TLS logins (SASL) with certificates from a throwaway CA that `vagrant up` makes in `.vagrant/tls`; the TLS run takes its client certificate from there through `HOME`. It needs Vagrant with the `vagrant-libvirt` plugin, a KVM host with `qemu-kvm` and `cyrus-sasl-md5`, and a user in the `libvirt` group. On a host that also runs Docker, let the VMs' network (the bridge of `virsh net-info vagrant-libvirt`, `virbr1` here) through Docker's forward rules (`sudo iptables -I DOCKER-USER -i virbr1 -j ACCEPT` and `sudo iptables -I DOCKER-USER -o virbr1 -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT`).
 
 ```bash
 vagrant up --provider=libvirt
 C1=$(vagrant ssh compute1 -c "hostname -I" | awk '{print $1}') C2=$(vagrant ssh compute2 -c "hostname -I" | awk '{print $1}')
 for h in $C1 $C2; do ssh-keygen -R $h; ssh -o StrictHostKeyChecking=accept-new root@$h true; done   # new VMs, new host keys
 export TEST_LIBVIRT_HOST=$C1 TEST_LIBVIRT_TYPE=1 TEST_LIBVIRT_LOGIN=wvc TEST_LIBVIRT_PASSWORD=wvc-test
-export TEST_LIBVIRT_SSH_HOST=$C2 TEST_LIBVIRT_SSH_LOGIN=root TEST_LIBVIRT_MIGRATE_ADDRESS=$C2 TEST_LIBVIRT_LVM_DEVICE=/dev/vdb TEST_LIBVIRT_ISCSI_TARGET=iqn.2026-10.test.wvc:lun TEST_LIBVIRT_NFS_EXPORT=/srv/wvcnfs
+export TEST_LIBVIRT_SSH_HOST=$C2 TEST_LIBVIRT_SSH_LOGIN=root TEST_LIBVIRT_MIGRATE_ADDRESS=$C2 TEST_LIBVIRT_LVM_DEVICE=/dev/vdb TEST_LIBVIRT_ISCSI_TARGET=iqn.2026-10.test.wvc:lun TEST_LIBVIRT_NFS_EXPORT=/srv/wvc-nfs
 python manage.py test
 HOME=$PWD/.vagrant/tls TEST_LIBVIRT_TYPE=3 TEST_LIBVIRT_SSH_HOST= python manage.py test   # compute1 again, over TLS
 vagrant destroy -f
diff --git a/Vagrantfile b/Vagrantfile
index 9aaa033c..37586e16 100644
--- a/Vagrantfile
+++ b/Vagrantfile
@@ -4,7 +4,7 @@
 # key for root (SASL user wvc, password wvc-test). The empty second disk
 # /dev/vdb is for the LVM tests (TEST_LIBVIRT_LVM_DEVICE); each also serves a
 # 1 GiB iSCSI LUN on 127.0.0.1 for the iSCSI tests (TEST_LIBVIRT_ISCSI_TARGET),
-# exports /srv/wvcnfs to 127.0.0.1 for the NFS pool test (TEST_LIBVIRT_NFS_EXPORT),
+# exports /srv/wvc-nfs to 127.0.0.1 for the NFS pool test (TEST_LIBVIRT_NFS_EXPORT),
 # and accepts TLS logins (SASL) with a certificate from a throwaway CA.
 
 # The CA is made here once; each compute signs its own certificate with it.
@@ -62,8 +62,8 @@ Vagrant.configure(2) do |config|
       targetcli saveconfig
     }
     dnf -y install nfs-utils
-    install -d /srv/wvcnfs
-    echo "/srv/wvcnfs 127.0.0.1(rw,no_root_squash)" > /etc/exports.d/wvc.exports
+    install -d /srv/wvc-nfs
+    echo "/srv/wvc-nfs 127.0.0.1(rw,no_root_squash)" > /etc/exports.d/wvc.exports
     systemctl enable --now nfs-server
     exportfs -r
     install -D -m 644 /tmp/wvc/cacert.pem /etc/pki/CA/cacert.pem
diff --git a/storages/forms.py b/storages/forms.py
index 4025d0e1..d316acd3 100644
--- a/storages/forms.py
+++ b/storages/forms.py
@@ -3,6 +3,12 @@ import re
 from django import forms
 from django.utils.translation import gettext_lazy as _
 
+# pool target and source paths, both put into the pool XML: letters, digits,
+# "/", "_", ".", ":" (/dev/disk/by-path names) and "-" not first and not right
+# after "/". Written without nested repetition, which backtracked
+# exponentially on crafted input.
+POOL_PATH = re.compile(r"(?!-)(?:[/0-9a-zA-Z_.:]|(?<!/)-)+")
+
 
 class AddStgPool(forms.Form):
     name = forms.CharField(
@@ -38,10 +44,7 @@ class AddStgPool(forms.Form):
     def clean_target(self):
         storage_type = self.cleaned_data["stg_type"]
         target = self.cleaned_data["target"]
-        # letters, digits, "_", ".", "/" and "-" not right after "/"; not
-        # starting with "-". Written without nested repetition, which
-        # backtracked exponentially on crafted input.
-        have_symbol = re.match(r"^[^-](?:[/0-9a-zA-Z_.]|(?<!/)-)*$", target)
+        have_symbol = POOL_PATH.fullmatch(target)
         if storage_type in ["dir", "netfs"] and not have_symbol:
             raise forms.ValidationError(
                 _("The target must not contain any special characters")
@@ -53,7 +56,7 @@ class AddStgPool(forms.Form):
     def clean_source(self):
         storage_type = self.cleaned_data["stg_type"]
         source = self.cleaned_data["source"]
-        have_symbol = re.match(r"^[a-zA-Z0-9\/]+$", source)
+        have_symbol = POOL_PATH.fullmatch(source)
         if storage_type in ["logical", "netfs"]:
             if not source:
                 raise forms.ValidationError(_("No device or path has been entered"))
diff --git a/storages/tests/test_pool_target.py b/storages/tests/test_pool_target.py
index 207dcb06..47fd2246 100644
--- a/storages/tests/test_pool_target.py
+++ b/storages/tests/test_pool_target.py
@@ -1,4 +1,6 @@
-"""The storage pool target check runs in linear time and accepts what it did."""
+"""The storage pool target and source checks run in linear time and take
+paths of letters, digits and "/", "_", ".", ":" and "-" (not first, not right
+after "/"): both values go into the pool XML."""
 
 import time
 
@@ -12,6 +14,12 @@ def target_error(target):
     return "target" in form.errors
 
 
+def source_error(source, stg_type="logical"):
+    form = AddStgPool({"name": "pool1", "stg_type": stg_type, "source": source, "target": "/mnt/pool1"})
+    form.is_valid()
+    return "source" in form.errors
+
+
 class PoolTargetTests(SimpleTestCase):
     def test_accepted_and_refused_targets(self):
         for target, refused in (
@@ -22,10 +30,31 @@ class PoolTargetTests(SimpleTestCase):
             ("/a/-b", True),  # a dash right after a slash
             ("-x", True),
             ("/a b", True),
+            ("'/a", True),  # any first character but "-" passed
+            ("</a", True),
         ):
             with self.subTest(target=target):
                 self.assertEqual(target_error(target), refused)
 
+    def test_accepted_and_refused_sources(self):
+        for source, refused in (
+            ("/dev/vdb", False),
+            ("/dev/mapper/vg0-data", False),
+            ("/dev/disk/by-id/wwn-0x5000c500a1b2c3d4", False),
+            ("/dev/disk/by-path/pci-0000:00:1f.2-ata-1", False),
+            ("/export/vm_images.v2", False),
+            ("", True),
+            ("-x", True),  # would read as an option
+            ("/a/-b", True),
+            ("/a b", True),
+            ("/a'b", True),  # the value goes into the pool XML
+            ("/a<b", True),
+            ("'/a", True),
+        ):
+            for stg_type in ("logical", "netfs"):
+                with self.subTest(source=source, stg_type=stg_type):
+                    self.assertEqual(source_error(source, stg_type), refused)
+
     def test_a_crafted_target_is_checked_at_once(self):
         # the previous expression backtracked exponentially: 22 dashes took over a second
         start = time.monotonic()