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()