From a95ec40f69748a464920a999928317fe599e5c32 Mon Sep 17 00:00:00 2001 From: catborise Date: Thu, 29 Oct 2026 09:29:20 +0300 Subject: [PATCH] fix(virtsecrets): a secret in use is not deleted, and a password is stored as typed - A secret is in use when a VM names it in a disk, a host device or its vTPM, or a pool in its source, by UUID or by its usage within the usage type, as they run and as saved. The vTPM, host devices and pools' saved definitions were not checked before. - A Ceph key is taken in base64, as `ceph auth get-key` prints it; any other value is a password or passphrase stored as typed, spaces included. - A new secret takes its value at once; a value refused leaves no secret. - TLS and vTPM secrets can be created; the usage has no length limit. - Every POST redirects; the list is read once and only with its users; a secret, VM or pool gone meanwhile is left out. - domain_trees() is shared by the in-use checks of secrets, network filters, networks and pools; the storage page's Ceph secret picker reads the same listing. --- nwfilters/tests/test_filters.py | 4 +- storages/tests/test_audit_rows.py | 2 +- storages/tests/test_new_pool.py | 13 +- storages/tests/test_pool_page.py | 2 +- storages/views.py | 8 +- virtsecrets/forms.py | 12 +- .../templates/create_secret_block.html | 36 ++-- virtsecrets/templates/secrets.html | 6 +- virtsecrets/tests/test_audit_rows.py | 11 +- virtsecrets/tests/test_create_secret.py | 93 +++++++--- virtsecrets/tests/test_live_secrets.py | 2 +- virtsecrets/tests/test_secret_in_use.py | 165 ++++++++++++++---- virtsecrets/tests/test_secrets_page.py | 19 +- virtsecrets/views.py | 99 ++++------- vrtManager/connection.py | 55 ++++-- vrtManager/nwfilters.py | 22 +-- vrtManager/virtsecrets.py | 129 +++++++++++--- 17 files changed, 429 insertions(+), 249 deletions(-) diff --git a/nwfilters/tests/test_filters.py b/nwfilters/tests/test_filters.py index 476d06a9..84bea436 100644 --- a/nwfilters/tests/test_filters.py +++ b/nwfilters/tests/test_filters.py @@ -27,7 +27,7 @@ def nwfilter(name, xml, uuid=UUID): def vm(name, xml, saved=None, active=False, persistent=True): def describe(flags=0): - return saved if flags else xml + return (saved or xml) if flags else xml return MagicMock(**{"name.return_value": name, "XMLDesc.side_effect": describe, "isActive.return_value": active, "isPersistent.return_value": persistent}) @@ -70,7 +70,7 @@ class UsersTests(SimpleTestCase): ]) self.assertEqual(conn.get_users(), {"a": ["stopped"], "b": ["running"], "c": ["running"], "d": ["transient"]}) stopped, running, transient = conn.wvm.listAllDomains.return_value - self.assertEqual((stopped.XMLDesc.call_count, running.XMLDesc.call_count, transient.XMLDesc.call_count), (1, 2, 1)) + self.assertEqual((stopped.XMLDesc.call_count, running.XMLDesc.call_count, transient.XMLDesc.call_count), (2, 2, 1)) def test_a_vm_gone_meanwhile_uses_nothing_other_errors_are_raised(self): gone = vm("gone", "") diff --git a/storages/tests/test_audit_rows.py b/storages/tests/test_audit_rows.py index 1163af15..a30914b7 100644 --- a/storages/tests/test_audit_rows.py +++ b/storages/tests/test_audit_rows.py @@ -127,7 +127,7 @@ class PoolAndVolumeRowTests(TransactionTestCase): conn.get_size.return_value = (1, 1) conn.get_storages_info.return_value = [] conn.get_storages.return_value = ["default"] - conn.get_secrets.return_value = [] + conn.get_secrets_info.return_value = [] conn.update_volumes.return_value = [] conn.create_volume.return_value = "disk1.qcow2" setattr(self, name, conn) diff --git a/storages/tests/test_new_pool.py b/storages/tests/test_new_pool.py index 3634f319..b28262fb 100644 --- a/storages/tests/test_new_pool.py +++ b/storages/tests/test_new_pool.py @@ -2,13 +2,13 @@ path another one has, an RBD pool only a Ceph secret of the host; a refused pool opens the dialog again with what was entered and why.""" -from unittest.mock import MagicMock, patch +from unittest.mock import patch from computes.models import Compute from django.contrib.auth import get_user_model from django.test import SimpleTestCase, TestCase from django.urls import reverse -from libvirt import VIR_SECRET_USAGE_TYPE_CEPH, VIR_SECRET_USAGE_TYPE_VOLUME, libvirtError +from libvirt import libvirtError from lxml import html from storages.forms import AddStgPool @@ -63,11 +63,10 @@ class DialogTests(TestCase): self.addCleanup(patcher.stop) pools = [{**pool, "status": 1, "autostart": 1, "volumes": 0, "size": 1, "free": 1, "used_percent": 0} for pool in POOLS] self.conn.get_storages_info.return_value = pools - secrets = {SECRET: (VIR_SECRET_USAGE_TYPE_CEPH, "client.libvirt"), - "00000000-0000-0000-0000-000000000003": (VIR_SECRET_USAGE_TYPE_VOLUME, "/var/lib/libvirt/images/a.img")} - self.conn.get_secrets.return_value = list(secrets) - self.conn.get_secret.side_effect = lambda uuid: MagicMock(**{"usageType.return_value": secrets[uuid][0], - "usageID.return_value": secrets[uuid][1]}) + self.conn.get_secrets_info.return_value = [ + {"uuid": SECRET, "usage_type": "ceph", "usage": "client.libvirt"}, + {"uuid": "00000000-0000-0000-0000-000000000003", "usage_type": "volume", "usage": "/var/lib/libvirt/images/a.img"}, + ] self.url = reverse("storages", args=[self.compute.pk]) def page(self, response): diff --git a/storages/tests/test_pool_page.py b/storages/tests/test_pool_page.py index fb623785..23e61556 100644 --- a/storages/tests/test_pool_page.py +++ b/storages/tests/test_pool_page.py @@ -139,7 +139,7 @@ class PoolListTests(TestCase): ] with patch("storages.views.wvmStorages") as storages: storages.return_value.get_storages_info.return_value = pools - storages.return_value.get_secrets.return_value = [] + storages.return_value.get_secrets_info.return_value = [] page = html.fromstring(self.client.get(reverse("storages", args=[compute.pk])).content) cards = [" ".join(" ".join(card.itertext()).split()) for card in page.xpath("//div[@class='card h-100']")] self.assertEqual(cards, [ diff --git a/storages/views.py b/storages/views.py index 0eb657a9..eef84bbf 100644 --- a/storages/views.py +++ b/storages/views.py @@ -10,7 +10,7 @@ from django.http import HttpResponseRedirect, JsonResponse from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.utils.translation import gettext_lazy as _ -from libvirt import VIR_SECRET_USAGE_TYPE_CEPH, libvirtError +from libvirt import libvirtError from operations import claims, jobs from operations.utils import busy_message from logs.audit import audit, failed @@ -79,11 +79,7 @@ def storages(request, compute_id): storages = conn.get_storages_info() iso_pool = next((pool for pool in storages if pool["name"] == "iso"), None) # the secrets an RBD pool can use, by the name they were given - ceph_secrets = [] - for uuid in conn.get_secrets(): - secret = conn.get_secret(uuid) - if secret.usageType() == VIR_SECRET_USAGE_TYPE_CEPH: - ceph_secrets.append({"uuid": uuid, "usage": secret.usageID()}) + ceph_secrets = [secret for secret in conn.get_secrets_info() if secret["usage_type"] == "ceph"] if request.method == "POST" and "create" in request.POST: form = AddStgPool(request.POST, pools=storages, ceph_secrets=[s["uuid"] for s in ceph_secrets]) diff --git a/virtsecrets/forms.py b/virtsecrets/forms.py index 232973ad..be266ef5 100644 --- a/virtsecrets/forms.py +++ b/virtsecrets/forms.py @@ -1,11 +1,13 @@ from django import forms +from django.utils.translation import gettext_lazy as _ class AddSecret(forms.Form): - ephemeral = forms.ChoiceField(required=True, choices=(("no", "no"), ("yes", "yes"))) - private = forms.ChoiceField(required=True, choices=(("no", "no"), ("yes", "yes"))) usage_type = forms.ChoiceField( - required=True, - choices=(("ceph", "ceph"), ("volume", "volume"), ("iscsi", "iscsi")), + choices=(("ceph", "Ceph"), ("iscsi", "iSCSI"), ("volume", _("Volume")), ("tls", "TLS"), ("vtpm", "vTPM")), ) - data = forms.CharField(max_length=100, required=True) + usage = forms.CharField() + # a password keeps its spaces; none defines the secret without a value + value = forms.CharField(required=False, strip=False) + ephemeral = forms.BooleanField(required=False) + private = forms.BooleanField(required=False) diff --git a/virtsecrets/templates/create_secret_block.html b/virtsecrets/templates/create_secret_block.html index 9ab34929..26e33296 100644 --- a/virtsecrets/templates/create_secret_block.html +++ b/virtsecrets/templates/create_secret_block.html @@ -15,24 +15,6 @@
{% csrf_token %}
-
- -
- -
-
-
- -
- -
-
@@ -44,11 +26,25 @@
- +
- +
+
+ +
+ +
+
+
+ + +
+
+ + +
diff --git a/virtsecrets/templates/secrets.html b/virtsecrets/templates/secrets.html index 1f3e4189..9763b13c 100644 --- a/virtsecrets/templates/secrets.html +++ b/virtsecrets/templates/secrets.html @@ -41,8 +41,8 @@ {% for secret in secrets_all %} {{ secret.uuid }} - {{ secret.usageType }} - {{ secret.usage|default_if_none:"" }} + {{ secret.usage_type }} + {{ secret.usage }} {% csrf_token %} @@ -69,7 +69,7 @@
- +
diff --git a/virtsecrets/tests/test_audit_rows.py b/virtsecrets/tests/test_audit_rows.py index 4727dab1..3d37ff8f 100644 --- a/virtsecrets/tests/test_audit_rows.py +++ b/virtsecrets/tests/test_audit_rows.py @@ -27,11 +27,11 @@ class SecretRowTests(TestCase): patcher = patch("virtsecrets.views.wvmSecrets") self.conn = patcher.start().return_value self.addCleanup(patcher.stop) - self.conn.get_secrets.return_value = [] + self.conn.get_secrets_info.return_value = [] def test_secret_actions(self): - self.client.post(self.url, {"create": "", "ephemeral": "no", "private": "yes", "usage_type": "ceph", - "data": "client.admin"}) + self.client.post(self.url, {"create": "", "private": "on", "usage_type": "ceph", "usage": "client.admin", + "value": VALUE}) self.client.post(self.url, {"set_value": "", "uuid": UUID, "value": VALUE}) self.client.post(self.url, {"delete": "", "uuid": UUID}) self.assertEqual(rows(), [ @@ -42,9 +42,12 @@ class SecretRowTests(TestCase): self.assertEqual(set(Logs.objects.values_list("host", flat=True)), {"c1"}) def test_a_secret_value_is_never_logged(self): + # an unexpected error naming the value is raised, its row names only the error's class self.conn.set_secret_value.side_effect = ValueError(f"bad value {VALUE}") - self.client.post(self.url, {"set_value": "", "uuid": UUID, "value": VALUE}) + with self.assertRaises(ValueError): + self.client.post(self.url, {"set_value": "", "uuid": UUID, "value": VALUE}) self.assertEqual(rows(), [(f"Secret {UUID}: set value: ValueError", False)]) + self.conn.set_secret_value.side_effect = OperationError("A Ceph key is base64 text") with patch.object(Logs.objects, "create", side_effect=RuntimeError("db down")), \ self.assertLogs("logs.audit", "WARNING") as logged: self.client.post(self.url, {"set_value": "", "uuid": UUID, "value": VALUE}) diff --git a/virtsecrets/tests/test_create_secret.py b/virtsecrets/tests/test_create_secret.py index 14139034..330cfe62 100644 --- a/virtsecrets/tests/test_create_secret.py +++ b/virtsecrets/tests/test_create_secret.py @@ -1,55 +1,91 @@ -"""A secret's usage value goes into its XML as text, whatever it holds: a -volume path, an iSCSI IQN or a Ceph name, and characters XML gives a meaning.""" +"""A secret's usage goes into its XML as text, whatever it holds. Its value +is a Ceph key in base64, as `ceph auth get-key` prints it, or a password or +passphrase as typed; a new secret takes its value at once, or none, and a +value refused leaves no secret behind.""" from unittest.mock import patch +import libvirt from computes.models import Compute from django.contrib.auth import get_user_model from django.test import SimpleTestCase, TestCase from django.urls import reverse from lxml import etree, html from virtsecrets.forms import AddSecret +from vrtManager.util import OperationError from vrtManager.virtsecrets import wvmSecrets -class CreateSecretTests(SimpleTestCase): - def defined_xml(self, usage_type, data): - with patch("vrtManager.connection.connection_manager.get_connection") as get: - wvmSecrets("host", "login", "pw", 1).create_secret("no", "yes", usage_type, data) - return etree.fromstring(get.return_value.secretDefineXML.call_args.args[0]) +def connection(): + with patch("vrtManager.connection.connection_manager.get_connection") as get: + conn = wvmSecrets("host", "login", "pw", 1) + return conn, get.return_value - def test_the_usage_value_is_text_of_its_element(self): - for usage_type, element, data in ( + +class CreateSecretTests(SimpleTestCase): + def test_the_usage_is_text_of_its_element(self): + for usage_type, element, usage in ( ("volume", "volume", "/var/lib/libvirt/images/vm-1.qcow2"), ("iscsi", "target", "iqn.2026-10.test.wvc:lun"), ("ceph", "name", "client.libvirt secret"), + ("tls", "name", "migration"), + ("vtpm", "name", "VM web-1's vTPM"), ("volume", "volume", "/a'b&d\""), ): - with self.subTest(usage_type=usage_type, data=data): - secret = self.defined_xml(usage_type, data) + with self.subTest(usage_type=usage_type, usage=usage): + conn, wvm = connection() + conn.create_secret(usage_type, usage, private=True) + secret = etree.fromstring(wvm.secretDefineXML.call_args.args[0]) self.assertEqual((secret.get("ephemeral"), secret.get("private")), ("no", "yes")) self.assertEqual(secret.find("usage").get("type"), usage_type) - self.assertEqual(secret.findtext(f"usage/{element}"), data) + self.assertEqual(secret.findtext(f"usage/{element}"), usage) + wvm.secretDefineXML.return_value.setValue.assert_not_called() + + def test_a_new_secret_takes_its_value_at_once(self): + for usage_type, value, stored in (("ceph", "cHcA/w==", b"pw\x00\xff"), ("iscsi", " pass word ", b" pass word ")): + with self.subTest(usage_type=usage_type): + conn, wvm = connection() + conn.create_secret(usage_type, "lun", value) + wvm.secretDefineXML.return_value.setValue.assert_called_once_with(stored) + + def test_a_value_refused_leaves_no_secret(self): + conn, wvm = connection() + with self.assertRaisesRegex(OperationError, "base64"): + conn.create_secret("ceph", "client.libvirt", "not base64!") + wvm.secretDefineXML.assert_not_called() + wvm.secretDefineXML.return_value.setValue.side_effect = libvirt.libvirtError("refused") + with self.assertRaises(libvirt.libvirtError): + conn.create_secret("iscsi", "lun", "pw") + wvm.secretDefineXML.return_value.undefine.assert_called_once_with() class SecretValueTests(SimpleTestCase): - def secrets(self): - with patch("vrtManager.connection.connection_manager.get_connection") as get: - conn = wvmSecrets("host", "login", "pw", 1) - secret = get.return_value.secretLookupByUUIDString.return_value - return conn, secret + def set_value(self, usage_type, value): + conn, wvm = connection() + secret = wvm.secretLookupByUUIDString.return_value + secret.XMLDesc.return_value = f"x" + conn.set_secret_value("uuid", value) + return secret.setValue - def test_a_base64_value_is_stored_as_its_bytes(self): - conn, secret = self.secrets() - conn.set_secret_value("uuid", "cHcA/w==") - secret.setValue.assert_called_once_with(b"pw\x00\xff") + def test_a_ceph_key_is_stored_as_its_bytes_a_password_as_typed(self): + self.set_value("ceph", "cHcA/w==").assert_called_once_with(b"pw\x00\xff") + # a password that happens to be base64 is not decoded + self.set_value("iscsi", "cHcA/w==").assert_called_once_with(b"cHcA/w==") + self.set_value("vtpm", " pässword ").assert_called_once_with(" pässword ".encode()) - def test_a_value_that_is_not_base64_is_refused(self): - conn, secret = self.secrets() - for value in ("b'cHc='", "not base64!"): - with self.subTest(value=value), self.assertRaises(ValueError): - conn.set_secret_value("uuid", value) - secret.setValue.assert_not_called() + def test_a_ceph_key_that_is_not_base64_is_refused_without_the_value(self): + for value in ("b'cHc='", "not base64!", "kéy"): + with self.subTest(value=value), self.assertRaises(OperationError) as refused: + self.set_value("ceph", value) + self.assertNotIn(value, str(refused.exception)) + + +class CreateSecretFormTests(SimpleTestCase): + def test_a_password_keeps_its_spaces_and_the_flags_are_checkboxes(self): + form = AddSecret({"usage_type": "iscsi", "usage": "iqn.2026-10.test.wvc:lun", "value": " pw ", "private": "on"}) + self.assertTrue(form.is_valid(), form.errors) + self.assertEqual((form.cleaned_data["value"], form.cleaned_data["ephemeral"], form.cleaned_data["private"]), + (" pw ", False, True)) class CreateSecretDialogTests(TestCase): @@ -57,7 +93,8 @@ class CreateSecretDialogTests(TestCase): compute = Compute.objects.create(name="secret-usage", hostname="localhost", login="", password="", type=4) self.client.force_login(get_user_model().objects.create_superuser("usage", "usage@example.com", "pw")) with patch("virtsecrets.views.wvmSecrets") as conn: - conn.return_value.get_secrets.return_value = [] + conn.return_value.get_secrets_info.return_value = [] + conn.return_value.get_users.return_value = {} response = self.client.get(reverse("virtsecrets", args=[compute.id])) offered = html.fromstring(response.content).xpath("//select[@id='secret_usage_type']/option/@value") self.assertEqual(offered, [value for value, _ in AddSecret.base_fields["usage_type"].choices]) diff --git a/virtsecrets/tests/test_live_secrets.py b/virtsecrets/tests/test_live_secrets.py index f55bc7f4..98ea5620 100644 --- a/virtsecrets/tests/test_live_secrets.py +++ b/virtsecrets/tests/test_live_secrets.py @@ -46,7 +46,7 @@ class LiveSecretsTestCase(TestCase): with self.subTest(name): response = self.client.post( reverse("virtsecrets", args=[self.compute.id]), - {"create": "", "ephemeral": "no", "private": "no", "usage_type": name, "data": value}, + {"create": "", "usage_type": name, "usage": value}, ) self.assertEqual(response.status_code, 302) self.assertEqual(self.conn.secretLookupByUsage(usage_type, value).usageID(), value) diff --git a/virtsecrets/tests/test_secret_in_use.py b/virtsecrets/tests/test_secret_in_use.py index 2786b63b..c04d8213 100644 --- a/virtsecrets/tests/test_secret_in_use.py +++ b/virtsecrets/tests/test_secret_in_use.py @@ -1,9 +1,12 @@ -"""A secret a VM disk or a pool needs is not deleted: without it that VM or -pool no longer starts. A disk or pool names its secret by UUID or by usage. -libvirt's test driver has no secrets, so the connection is a mock.""" +"""A secret a VM or a pool needs is not deleted: without it that VM or pool +no longer starts. A VM names its secret in a disk, a host device or its vTPM, +a pool in its source, by UUID or by usage within the usage type; as they run +and as saved. libvirt's test driver has no secrets, so the connection is a +mock.""" from unittest.mock import MagicMock, patch +import libvirt from computes.models import Compute from django.contrib.auth import get_user_model from django.contrib.messages import get_messages @@ -12,53 +15,139 @@ from django.urls import reverse from vrtManager.util import OperationError from vrtManager.virtsecrets import wvmSecrets -UUID = "8f2c0000-0000-4000-8000-000000000001" +CEPH = "8f2c0000-0000-4000-8000-000000000001" +ISCSI = "8f2c0000-0000-4000-8000-000000000002" +VTPM = "8f2c0000-0000-4000-8000-000000000003" -def libvirt_object(name, xml): - obj = MagicMock() - obj.name.return_value = name - obj.XMLDesc.return_value = xml - return obj +def secret(uuid, usage_type, usage): + return MagicMock(**{"UUIDString.return_value": uuid, + "XMLDesc.return_value": f"{usage}"}) -def disk_vm(name, secret): - return libvirt_object(name, f"" - f"{secret}") +def owner(name, xml, saved=None, active=False, persistent=True): + """A domain or a pool: xml as it runs, saved as defined for its next start.""" + return MagicMock(**{"name.return_value": name, "XMLDesc.side_effect": lambda flags=0: (saved or xml) if flags else xml, + "isActive.return_value": active, "isPersistent.return_value": persistent}) -class SecretInUseTests(SimpleTestCase): - def delete(self, domains=(), pools=()): - with patch("vrtManager.connection.connection_manager.get_connection") as get: - wvm = get.return_value - wvm.listAllDomains.return_value = list(domains) - wvm.listAllStoragePools.return_value = list(pools) - secret = wvm.secretLookupByUUIDString.return_value - secret.usageID.return_value = "client.libvirt secret" - wvmSecrets("host", "login", "pw", 1).delete_secret(UUID) - return secret +def vm(name, devices, **kwargs): + return owner(name, f"{devices}", **kwargs) - def test_a_secret_a_vm_disk_or_a_pool_names_is_not_deleted(self): - pool = libvirt_object("ceph-rbd", "" - "") - with self.assertRaises(OperationError) as refused: - self.delete([disk_vm("db-2", f"")], [pool]) - self.assertIn("VMs db-2", str(refused.exception)) - self.assertIn("pools ceph-rbd", str(refused.exception)) + +def pool(name, auth, **kwargs): + return owner(name, f"{auth}", **kwargs) + + +def gone(code): + err = libvirt.libvirtError("gone") + err.get_error_code = lambda: code + return MagicMock(**{"name.return_value": "gone", "XMLDesc.side_effect": err}) + + +# the same usage, of two types: a reference by usage names the one of its type +SECRETS = [secret(CEPH, "ceph", "client.libvirt"), secret(ISCSI, "iscsi", "client.libvirt"), secret(VTPM, "vtpm", "tpm")] + + +def connection(domains=(), pools=(), secrets=SECRETS): + with patch("vrtManager.connection.connection_manager.get_connection") as get: + conn = wvmSecrets("host", "login", "pw", 1) + get.return_value.configure_mock(**{"listAllDomains.return_value": list(domains), + "listAllStoragePools.return_value": list(pools), + "listAllSecrets.return_value": list(secrets)}) + return conn + + +def users(**kwargs): + conn = connection(**kwargs) + return conn.get_users(conn.get_secrets_info()) + + +class UsersTests(SimpleTestCase): + def test_vms_name_secrets_in_disks_host_devices_and_vtpm(self): + found = users(domains=[ + vm("rbd-1", f""), + vm("lun-1", "" + ""), + vm("tpm-1", f""), + ]) + self.assertEqual(found, {CEPH: {"vms": ["rbd-1"], "pools": []}, ISCSI: {"vms": ["lun-1"], "pools": []}, + VTPM: {"vms": ["tpm-1"], "pools": []}}) + + def test_a_pool_names_its_secret_by_the_auth_type_and_as_saved(self): + found = users(pools=[ + pool("ceph-rbd", ""), + # chap is iSCSI; named only for its next start + pool("lun", "", saved="" + "", active=False), + ]) + self.assertEqual(found, {CEPH: {"vms": [], "pools": ["ceph-rbd"]}, ISCSI: {"vms": [], "pools": ["lun"]}}) + + def test_a_vm_uses_a_secret_as_it_runs_and_as_saved(self): + disk = "" + # saved only: read whether it runs or not, as it may stop between the reads + for kwargs in ({"active": True, "saved": f"{disk}"}, + {"active": False, "saved": f"{disk}"}, + {"active": False}, {"active": True, "persistent": False}): + devices = "" if "saved" in kwargs else disk + with self.subTest(**kwargs): + self.assertEqual(users(domains=[vm("db-1", devices, **kwargs)]), {CEPH: {"vms": ["db-1"], "pools": []}}) + + def test_what_is_gone_meanwhile_is_left_out_other_errors_are_raised(self): + found = users(domains=[gone(libvirt.VIR_ERR_NO_DOMAIN)], pools=[gone(libvirt.VIR_ERR_NO_STORAGE_POOL)], + secrets=[gone(libvirt.VIR_ERR_NO_SECRET)]) + self.assertEqual(found, {}) + with self.assertRaises(libvirt.libvirtError): + users(domains=[gone(libvirt.VIR_ERR_OPERATION_FAILED)]) + + +class NoValueReadTests(SimpleTestCase): + def test_the_list_and_its_users_read_no_value(self): + for listed in SECRETS: + listed.value.side_effect = AssertionError("a secret's value was read") + conn = connection(domains=[vm("db-1", f"")]) + conn.get_users(conn.get_secrets_info()) + conn.wvm.secretLookupByUUIDString.assert_not_called() + + +class DeleteTests(SimpleTestCase): + def test_a_secret_a_vm_or_a_pool_names_is_not_deleted(self): + conn = connection(domains=[vm("db-2", f"")], + pools=[pool("ceph-rbd", "")]) + with self.assertRaisesRegex(OperationError, f"Secret {CEPH} is used by VMs db-2; pools ceph-rbd"): + conn.delete_secret(CEPH) + conn.wvm.secretLookupByUUIDString.return_value.undefine.assert_not_called() def test_an_unused_secret_is_deleted(self): - other = disk_vm("web-1", "") - self.delete([other]).undefine.assert_called_once_with() + conn = connection(domains=[vm("web-1", f"")]) + conn.delete_secret(ISCSI) + conn.wvm.secretLookupByUUIDString.assert_called_once_with(ISCSI) + conn.wvm.secretLookupByUUIDString.return_value.undefine.assert_called_once_with() class SecretInUsePageTests(TestCase): - def test_the_page_says_who_needs_it(self): + def setUp(self): compute = Compute.objects.create(name="secret-use", hostname="localhost", login="", password="", type=4) self.client.force_login(get_user_model().objects.create_superuser("secret-use", "su@example.com", "pw")) - url = reverse("virtsecrets", args=[compute.id]) + self.url = reverse("virtsecrets", args=[compute.id]) + + def test_the_page_says_who_needs_it(self): with patch("virtsecrets.views.wvmSecrets") as conn: - conn.return_value.get_secrets.return_value = [] - conn.return_value.delete_secret.side_effect = OperationError(f"Secret {UUID} is used by VMs db-2") - response = self.client.post(url, {"delete": "", "uuid": UUID}) - self.assertRedirects(response, url, fetch_redirect_response=False) - self.assertEqual([str(m) for m in get_messages(response.wsgi_request)], [f"Secret {UUID} is used by VMs db-2"]) + conn.return_value.delete_secret.side_effect = OperationError(f"Secret {CEPH} is used by VMs db-2") + response = self.client.post(self.url, {"delete": "", "uuid": CEPH}) + self.assertRedirects(response, self.url, fetch_redirect_response=False) + self.assertEqual([str(m) for m in get_messages(response.wsgi_request)], [f"Secret {CEPH} is used by VMs db-2"]) + + def test_a_post_to_a_host_not_reached_is_redirected(self): + with patch("virtsecrets.views.wvmSecrets", side_effect=libvirt.libvirtError("unable to connect")): + response = self.client.post(self.url, {"delete": "", "uuid": CEPH}) + self.assertRedirects(response, self.url, fetch_redirect_response=False) + + def test_the_list_is_not_shown_without_its_users(self): + # it would offer to delete the secrets in use + with patch("virtsecrets.views.wvmSecrets") as conn: + conn.return_value.get_secrets_info.return_value = [{"uuid": CEPH, "usage_type": "ceph", "usage": "client.libvirt", + "ephemeral": False, "private": False}] + conn.return_value.get_users.side_effect = libvirt.libvirtError("connection reset") + response = self.client.get(self.url) + self.assertEqual(response.context["secrets_all"], []) diff --git a/virtsecrets/tests/test_secrets_page.py b/virtsecrets/tests/test_secrets_page.py index 1a94f6a0..f552f530 100644 --- a/virtsecrets/tests/test_secrets_page.py +++ b/virtsecrets/tests/test_secrets_page.py @@ -7,7 +7,6 @@ from computes.models import Compute from django.contrib.auth import get_user_model from django.test import TestCase from django.urls import reverse -from libvirt import VIR_SECRET_USAGE_TYPE_CEPH, VIR_SECRET_USAGE_TYPE_VTPM from lxml import html @@ -16,23 +15,21 @@ class SecretsPageTests(TestCase): compute = Compute.objects.create(name="secrets-page", hostname="localhost", login="", password="", type=4) self.client.force_login(get_user_model().objects.create_superuser("page", "page@example.com", "pw")) with patch("virtsecrets.views.wvmSecrets") as conn: - conn.return_value.get_secrets.return_value = ["0e5b1f0a-8f3e-4c1a-9d2b-5c6d7e8f9a0b"] - secret = conn.return_value.get_secret.return_value - secret.usageID.return_value = "client.libvirt" - secret.UUIDString.return_value = "0e5b1f0a-8f3e-4c1a-9d2b-5c6d7e8f9a0b" - secret.usageType.return_value = usage_type - secret.value.return_value = b"cephx key" - conn.return_value.get_secret_value.return_value = "Y2VwaHgga2V5" + conn.return_value.get_secrets_info.return_value = [ + {"uuid": "0e5b1f0a-8f3e-4c1a-9d2b-5c6d7e8f9a0b", "usage_type": usage_type, "usage": "client.libvirt", + "ephemeral": False, "private": False}] + conn.return_value.get_users.return_value = {} + conn.return_value.wvm.secretLookupByUUIDString.return_value.value.return_value = b"cephx key" response = self.client.get(reverse("virtsecrets", args=[compute.id])) self.assertEqual(response.status_code, 200) return response def test_a_vtpm_secret_is_listed(self): - self.assertContains(self.page(VIR_SECRET_USAGE_TYPE_VTPM), "vtpm") + self.assertContains(self.page("vtpm"), "vtpm") def test_the_page_holds_no_secret_value(self): - response = self.page(VIR_SECRET_USAGE_TYPE_CEPH) - self.assertFalse(b"Y2VwaHgga2V5" in response.content, "the secret's value is on the page") + response = self.page("ceph") + self.assertFalse(b"cephx key" in response.content, "the secret's value is on the page") field = html.fromstring(response.content).get_element_by_id("secret_value_0e5b1f0a-8f3e-4c1a-9d2b-5c6d7e8f9a0b") self.assertIsNone(field.get("value")) self.assertEqual(field.get("type"), "password") diff --git a/virtsecrets/views.py b/virtsecrets/views.py index 0e2448a4..6fee6464 100644 --- a/virtsecrets/views.py +++ b/virtsecrets/views.py @@ -3,17 +3,8 @@ from computes.models import Compute from django.contrib import messages from django.http import HttpResponseRedirect from django.shortcuts import get_object_or_404, render -from libvirt import ( - VIR_SECRET_USAGE_TYPE_CEPH, - VIR_SECRET_USAGE_TYPE_ISCSI, - VIR_SECRET_USAGE_TYPE_NONE, - VIR_SECRET_USAGE_TYPE_TLS, - VIR_SECRET_USAGE_TYPE_VOLUME, - VIR_SECRET_USAGE_TYPE_VTPM, - libvirtError, -) +from libvirt import libvirtError from logs.audit import audit -from vrtManager.util import OperationError from vrtManager.virtsecrets import wvmSecrets from virtsecrets.forms import AddSecret @@ -21,74 +12,46 @@ from virtsecrets.forms import AddSecret @superuser_only def secrets(request, compute_id): - """ - :param request: - :param compute_id: - :return: - """ - - secrets_all = [] + """The host's secrets with the VMs and pools that use them; one created, + given a value or deleted, after which the list is read again. A value is + never shown or logged.""" compute = get_object_or_404(Compute, pk=compute_id) - secret_usage_types = { - VIR_SECRET_USAGE_TYPE_NONE: "none", - VIR_SECRET_USAGE_TYPE_VOLUME: "volume", - VIR_SECRET_USAGE_TYPE_CEPH: "ceph", - VIR_SECRET_USAGE_TYPE_ISCSI: "iscsi", - VIR_SECRET_USAGE_TYPE_TLS: "tls", - VIR_SECRET_USAGE_TYPE_VTPM: "vtpm", - } - # the create dialog offers the usage types the form takes - usage_choices = AddSecret.base_fields["usage_type"].choices + secrets_all = [] try: - conn = wvmSecrets( - compute.hostname, compute.login, compute.password, compute.type - ) - secrets = conn.get_secrets() + conn = wvmSecrets(compute.hostname, compute.login, compute.password, compute.type) - for uuid in secrets: - secrt = conn.get_secret(uuid) - secrets_all.append( - { - "usage": secrt.usageID(), - "uuid": secrt.UUIDString(), - "usageType": secret_usage_types[secrt.usageType()], - } - ) if request.method == "POST": + uuid = request.POST.get("uuid", "") if "create" in request.POST: form = AddSecret(request.POST) if form.is_valid(): data = form.cleaned_data - with audit(request, compute, f"Secret {data['usage_type']} {data['data']}", "add"): - conn.create_secret( - data["ephemeral"], - data["private"], - data["usage_type"], - data["data"], - ) - return HttpResponseRedirect(request.get_full_path()) + with audit(request, compute, f"Secret {data['usage_type']} {data['usage']}", "add"): + conn.create_secret(data["usage_type"], data["usage"], data["value"], + data["ephemeral"], data["private"]) else: for msg_err in form.errors.values(): messages.error(request, msg_err.as_text()) - if "delete" in request.POST: - uuid = request.POST.get("uuid", "") - try: - with audit(request, compute, f"Secret {uuid}", "delete"): - conn.delete_secret(uuid) - except OperationError as err: # a VM disk or a pool needs it - messages.error(request, str(err)) - return HttpResponseRedirect(request.get_full_path()) - if "set_value" in request.POST: - uuid = request.POST.get("uuid", "") - value = request.POST.get("value", "") - try: - with audit(request, compute, f"Secret {uuid}", "set value"): - conn.set_secret_value(uuid, value) - except Exception as err: - messages.error(request, err) - return HttpResponseRedirect(request.get_full_path()) - except libvirtError as err: - messages.error(request, err) + elif "set_value" in request.POST: + with audit(request, compute, f"Secret {uuid}", "set value"): + conn.set_secret_value(uuid, request.POST.get("value", "")) + elif "delete" in request.POST: + with audit(request, compute, f"Secret {uuid}", "delete"): + conn.delete_secret(uuid) + return HttpResponseRedirect(request.get_full_path()) - return render(request, "secrets.html", locals()) + # shown only with their users: a list without them would offer to delete secrets in use + secrets = conn.get_secrets_info() + users = conn.get_users(secrets) + for secret in secrets: + secret.update(users.get(secret["uuid"], {"vms": [], "pools": []})) + secrets_all = secrets + except libvirtError as err: # OperationError too: a secret in use, a Ceph key not base64 + messages.error(request, err) + if request.method == "POST": # a host not reached or a secret gone as well + return HttpResponseRedirect(request.get_full_path()) + + usage_choices = AddSecret.base_fields["usage_type"].choices + return render(request, "secrets.html", {"secrets_all": secrets_all, "compute": compute, + "usage_choices": usage_choices}) diff --git a/vrtManager/connection.py b/vrtManager/connection.py index 5dd5eb8b..5d62439e 100644 --- a/vrtManager/connection.py +++ b/vrtManager/connection.py @@ -843,11 +843,27 @@ class wvmConnect(object): def get_iface(self, name): return self.wvm.interfaceLookupByName(name) - def get_secrets(self): - return self.wvm.listSecrets() - - def get_secret(self, uuid): - return self.wvm.secretLookupByUUIDString(uuid) + def get_secrets_info(self): + """Every secret of the host: UUID, usage type and usage (a volume + path, an iSCSI target, a Ceph or TLS or vTPM name), whether it is + ephemeral and private. One listing, one XML read per secret; a secret + gone meanwhile is left out.""" + secrets = [] + for secret in self.wvm.listAllSecrets(0): + try: + tree = etree.fromstring(secret.XMLDesc(0)) + except libvirtError as err: + if err.get_error_code() != libvirt.VIR_ERR_NO_SECRET: + raise + continue + secrets.append({ + "uuid": secret.UUIDString(), + "usage_type": tree.xpath("string(usage/@type)") or "none", + "usage": tree.xpath("string(usage/*)"), + "ephemeral": tree.get("ephemeral") == "yes", + "private": tree.get("private") == "yes", + }) + return sorted(secrets, key=lambda secret: (secret["usage_type"], secret["usage"])) def get_storage(self, name): return self.wvm.storagePoolLookupByName(name) @@ -968,16 +984,27 @@ class wvmConnect(object): instances.append(name) return instances - def domains_using(self, match, running_only=False): - """Names of the domains with a definition, parsed, that match(tree) - holds for: the live and the persistent one of each domain, or with - running_only the live one of each running domain.""" + def domain_trees(self, running_only=False): + """Each domain's name with its definitions, parsed: the live one, and + for a persistent domain the one saved for its next boot too, read + whether it runs or not (it may stop meanwhile); with running_only the + live one of each running domain. A domain gone meanwhile is left out.""" listing = libvirt.VIR_CONNECT_LIST_DOMAINS_ACTIVE if running_only else 0 - flags = (0,) if running_only else (0, libvirt.VIR_DOMAIN_XML_INACTIVE) - return sorted( - dom.name() for dom in self.wvm.listAllDomains(listing) - if any(match(etree.fromstring(dom.XMLDesc(f))) for f in flags) - ) + for dom in self.wvm.listAllDomains(listing): + try: + xmls = [dom.XMLDesc(0)] + if not running_only and dom.isPersistent(): + xmls.append(dom.XMLDesc(libvirt.VIR_DOMAIN_XML_INACTIVE)) + except libvirtError as err: + if err.get_error_code() != libvirt.VIR_ERR_NO_DOMAIN: + raise + continue + yield dom.name(), [etree.fromstring(xml) for xml in xmls] + + def domains_using(self, match, running_only=False): + """Names of the domains with a definition (domain_trees) that + match(tree) holds for.""" + return sorted(name for name, trees in self.domain_trees(running_only) if any(map(match, trees))) def paths_in_use(self, skip_uuid=None, own=()): """Paths (names for network disks) a domain uses as a disk or a diff --git a/vrtManager/nwfilters.py b/vrtManager/nwfilters.py index 27ab44d0..c963f3e4 100644 --- a/vrtManager/nwfilters.py +++ b/vrtManager/nwfilters.py @@ -2,7 +2,6 @@ import copy import re from functools import cached_property -from libvirt import VIR_DOMAIN_XML_INACTIVE, VIR_ERR_NO_DOMAIN, libvirtError from lxml import etree from vrtManager import util @@ -43,24 +42,15 @@ class wvmNWFilters(wvmConnect): return sorted(filters, key=lambda nwfilter: nwfilter["name"]) def get_users(self): - """The VMs whose interfaces reference each filter, by filter name: as - they run and, for a persistent VM that runs, as saved for its next - boot. One listing, one or two XML reads per VM.""" + """The VMs whose interfaces reference each filter, by filter name, in + any of their definitions (domain_trees).""" users = {} - for dom in self.wvm.listAllDomains(0): - try: - xmls = [dom.XMLDesc(0)] - if dom.isActive() and dom.isPersistent(): - xmls.append(dom.XMLDesc(VIR_DOMAIN_XML_INACTIVE)) - except libvirtError as err: # a VM undefined meanwhile uses none - if err.get_error_code() != VIR_ERR_NO_DOMAIN: - raise - continue + for vm, trees in self.domain_trees(): names = set() - for xml in xmls: - names.update(etree.fromstring(xml).xpath("devices/interface/filterref/@filter")) + for tree in trees: + names.update(tree.xpath("devices/interface/filterref/@filter")) for name in names: - users.setdefault(name, []).append(dom.name()) + users.setdefault(name, []).append(vm) return users def create_nwfilter(self, xml): diff --git a/vrtManager/virtsecrets.py b/vrtManager/virtsecrets.py index 346da177..8dba8070 100644 --- a/vrtManager/virtsecrets.py +++ b/vrtManager/virtsecrets.py @@ -1,37 +1,118 @@ import base64 +from libvirt import VIR_ERR_NO_STORAGE_POOL, VIR_STORAGE_XML_INACTIVE, libvirtError from lxml import etree from vrtManager import util from vrtManager.connection import wvmConnect -# the element of that holds the value, per usage type -USAGE_ELEMENTS = {"ceph": "name", "volume": "volume", "iscsi": "target"} +# the element of that holds the usage, per usage type +USAGE_ELEMENTS = {"ceph": "name", "volume": "volume", "iscsi": "target", "tls": "name", "vtpm": "name"} + +# the usage type a reference by usage means: the type the gives, or +# for a pool's the auth type; a disk's encryption passphrase is a volume's +REF_USAGE_TYPES = {"ceph": "ceph", "iscsi": "iscsi", "chap": "iscsi", "passphrase": "volume"} + + +def value_bytes(usage_type, value): + """The bytes a value typed on the page stands for: a Ceph key as + `ceph auth get-key` prints it, in base64; any other secret's password or + passphrase as it is typed.""" + if usage_type != "ceph": + return value.encode() + try: + return base64.b64decode(value, validate=True) + except ValueError: # not base64, or not ASCII; the message never holds the value + raise util.OperationError("A Ceph key is base64 text, as `ceph auth get-key` prints it") from None + + +def secrets_named(tree, path, by_usage): + """UUIDs of the secrets the elements at path name, by UUID or by + usage; by_usage maps (usage type, usage) to a UUID.""" + uuids = set() + for ref in tree.xpath(path): + if ref.get("uuid"): + uuids.add(ref.get("uuid")) + else: + ref_type = ref.get("type") or ref.getparent().get("type") + uuids.add(by_usage.get((REF_USAGE_TYPES.get(ref_type), ref.get("usage")))) + return uuids class wvmSecrets(wvmConnect): - def create_secret(self, ephemeral, private, secret_type, data): - # built with lxml: data is text, whatever characters it holds - secret = etree.Element("secret", ephemeral=ephemeral, private=private) - usage = etree.SubElement(secret, "usage", type=secret_type) - etree.SubElement(usage, USAGE_ELEMENTS[secret_type]).text = data - self.wvm.secretDefineXML(etree.tostring(secret, encoding="unicode")) + def pool_trees(self): + """Each pool's name with its definitions, parsed: the live one, and + for a persistent pool the one saved for its next start too, read + whether it is active or not (it may stop meanwhile). A pool gone + meanwhile is left out.""" + for pool in self.wvm.listAllStoragePools(0): + try: + xmls = [pool.XMLDesc(0)] + if pool.isPersistent(): + xmls.append(pool.XMLDesc(VIR_STORAGE_XML_INACTIVE)) + except libvirtError as err: + if err.get_error_code() != VIR_ERR_NO_STORAGE_POOL: + raise + continue + yield pool.name(), [etree.fromstring(xml) for xml in xmls] + + def get_users(self, secrets): + """The VMs and pools that name each of secrets (get_secrets_info), by + UUID: a VM's disks, host devices and vTPM, a pool's source, in any of + their definitions. Not seen: a LUKS volume's own encryption, the TLS + settings of qemu.conf, snapshots.""" + by_usage = {(secret["usage_type"], secret["usage"]): secret["uuid"] for secret in secrets} + users = {} + + def add(kind, name, uuids): + for uuid in uuids - {None}: + users.setdefault(uuid, {"vms": [], "pools": []})[kind].append(name) + + for name, trees in self.domain_trees(): + uuids = set() + for tree in trees: + uuids.update(secrets_named(tree, "devices//secret", by_usage)) + uuids.update(tree.xpath("devices/tpm/backend/encryption/@secret")) + add("vms", name, uuids) + for name, trees in self.pool_trees(): + uuids = set() + for tree in trees: + uuids.update(secrets_named(tree, "source//secret", by_usage)) + add("pools", name, uuids) + return users + + def create_secret(self, usage_type, usage, value="", ephemeral=False, private=False): + """A secret for usage, with value if one is given: a Ceph key that is + not base64 defines nothing, and a value libvirt refuses undefines the + new secret again.""" + value = value_bytes(usage_type, value) if value else None + # built with lxml: the usage is text, whatever characters it holds + secret = etree.Element("secret", ephemeral="yes" if ephemeral else "no", private="yes" if private else "no") + usage_element = etree.SubElement(secret, "usage", type=usage_type) + etree.SubElement(usage_element, USAGE_ELEMENTS[usage_type]).text = usage + defined = self.wvm.secretDefineXML(etree.tostring(secret, encoding="unicode")) + if value is not None: + try: + defined.setValue(value) + except libvirtError: + defined.undefine() + raise def set_secret_value(self, uuid, value): - """value: the new value as base64 text.""" - secrt = self.get_secret(uuid) - # validate: a value that is not base64 is refused, not decoded to other bytes - secrt.setValue(base64.b64decode(value, validate=True)) + """value as typed on the page; its bytes follow the secret's usage + type (value_bytes).""" + secret = self.wvm.secretLookupByUUIDString(uuid) + usage_type = etree.fromstring(secret.XMLDesc(0)).xpath("string(usage/@type)") + secret.setValue(value_bytes(usage_type, value)) def delete_secret(self, uuid): - """Refused while a VM disk or a pool names it, by UUID or by usage: - without it that VM or pool would no longer start.""" - secrt = self.get_secret(uuid) - refs = "//secret[@uuid=$uuid or @usage=$usage]" - usage = secrt.usageID() or "" - vms = self.domains_using(lambda domain: domain.xpath(f"devices/disk{refs}", uuid=uuid, usage=usage)) - pools = sorted(pool.name() for pool in self.wvm.listAllStoragePools() - if etree.fromstring(pool.XMLDesc(0)).xpath(f"source{refs}", uuid=uuid, usage=usage)) - if vms or pools: - users = "; ".join(f"{kind} {', '.join(names)}" for kind, names in (("VMs", vms), ("pools", pools)) if names) - raise util.OperationError(f"Secret {uuid} is used by {users}; remove it from them first") - secrt.undefine() + """Refused while a VM or a pool names it (get_users): without it that + VM or pool would no longer start.""" + secret = self.wvm.secretLookupByUUIDString(uuid) + if users := self.get_users(self.get_secrets_info()).get(uuid): + named = [] + if users["vms"]: + named.append(f"VMs {', '.join(users['vms'])}") + if users["pools"]: + named.append(f"pools {', '.join(users['pools'])}") + raise util.OperationError(f"Secret {uuid} is used by {'; '.join(named)}; remove it from them first") + secret.undefine()