mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-10 04:06:47 +00:00
fix(virtsecrets): a secret's usage value goes into its XML as text
create_secret wrote the usage value into the secret XML with an f-string,
unchecked: a quote, "<" or "&" broke the XML or added to it. The page's
pattern ([a-z0-9. ]+) was the only check and refused real values: a
volume secret's path ("/") and an iSCSI IQN ("-", ":"). The XML is now
built with lxml, so the value is text whatever it holds, and the input
keeps only required and the form's 100 characters; libvirt reports a
value it does not take.
A live test creates a volume, an iSCSI and a Ceph secret with real
values through the page.
This commit is contained in:
parent
45b92c8638
commit
66151d42b1
5 changed files with 90 additions and 13 deletions
|
|
@ -46,8 +46,7 @@
|
|||
<div class="row">
|
||||
<label class="col-sm-3 col-form-label">{% trans "Data" %}</label>
|
||||
<div class="col-sm-6">
|
||||
<input type="text" name="data" class="form-control" value="" required
|
||||
pattern="[a-z0-9\. ]+" />
|
||||
<input type="text" name="data" class="form-control" value="" maxlength="100" required />
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
|
|
|||
0
virtsecrets/tests/__init__.py
Normal file
0
virtsecrets/tests/__init__.py
Normal file
28
virtsecrets/tests/test_create_secret.py
Normal file
28
virtsecrets/tests/test_create_secret.py
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
"""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."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from django.test import SimpleTestCase
|
||||
from lxml import etree
|
||||
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 test_the_usage_value_is_text_of_its_element(self):
|
||||
for usage_type, element, data in (
|
||||
("volume", "volume", "/var/lib/libvirt/images/vm-1.qcow2"),
|
||||
("iscsi", "target", "iqn.2026-10.test.wvc:lun"),
|
||||
("ceph", "name", "client.libvirt secret"),
|
||||
("volume", "volume", "/a'b<c>&d\""),
|
||||
):
|
||||
with self.subTest(usage_type=usage_type, data=data):
|
||||
secret = self.defined_xml(usage_type, data)
|
||||
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)
|
||||
52
virtsecrets/tests/test_live_secrets.py
Normal file
52
virtsecrets/tests/test_live_secrets.py
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
"""Live check that the secrets page creates a secret for each usage type
|
||||
with a real usage value: a volume path, an iSCSI IQN and a Ceph name. Runs
|
||||
only with TEST_LIBVIRT_HOST set."""
|
||||
|
||||
import os
|
||||
import unittest
|
||||
|
||||
import libvirt
|
||||
from computes.models import Compute
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase
|
||||
from django.urls import reverse
|
||||
from instances.tests import livetest
|
||||
from vrtManager.connection import connection_manager
|
||||
|
||||
P = livetest.PREFIX
|
||||
USAGES = {
|
||||
libvirt.VIR_SECRET_USAGE_TYPE_VOLUME: ("volume", f"/var/lib/libvirt/{P}secrets/vm-1.qcow2"),
|
||||
libvirt.VIR_SECRET_USAGE_TYPE_ISCSI: ("iscsi", f"iqn.2026-10.test.wvc:{P}secret"),
|
||||
libvirt.VIR_SECRET_USAGE_TYPE_CEPH: ("ceph", f"client.{P}secret secret"),
|
||||
}
|
||||
|
||||
|
||||
@unittest.skipUnless(livetest.enabled(), "Set TEST_LIBVIRT_HOST to run the live libvirt tests")
|
||||
class LiveSecretsTestCase(TestCase):
|
||||
def setUp(self):
|
||||
args = dict(
|
||||
hostname=os.environ["TEST_LIBVIRT_HOST"],
|
||||
login=os.environ.get("TEST_LIBVIRT_LOGIN", ""),
|
||||
password=os.environ.get("TEST_LIBVIRT_PASSWORD", ""),
|
||||
type=int(os.environ.get("TEST_LIBVIRT_TYPE", 4)),
|
||||
)
|
||||
self.conn = connection_manager.get_connection(*args.values())
|
||||
self.remove_secrets()
|
||||
self.addCleanup(self.remove_secrets)
|
||||
self.compute = Compute.objects.create(name="live-secrets", **args)
|
||||
self.client.force_login(get_user_model().objects.create_superuser("live-secrets", "ls@example.com", "x"))
|
||||
|
||||
def remove_secrets(self):
|
||||
for secret in self.conn.listAllSecrets():
|
||||
if (secret.usageType(), secret.usageID()) in {(t, v) for t, (_, v) in USAGES.items()}:
|
||||
secret.undefine()
|
||||
|
||||
def test_each_usage_type_takes_a_real_value(self):
|
||||
for usage_type, (name, value) in USAGES.items():
|
||||
with self.subTest(name):
|
||||
response = self.client.post(
|
||||
reverse("virtsecrets", args=[self.compute.id]),
|
||||
{"create": "", "ephemeral": "no", "private": "no", "usage_type": name, "data": value},
|
||||
)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertEqual(self.conn.secretLookupByUsage(usage_type, value).usageID(), value)
|
||||
|
|
@ -1,21 +1,19 @@
|
|||
import base64
|
||||
|
||||
from lxml import etree
|
||||
from vrtManager.connection import wvmConnect
|
||||
|
||||
# the element of <usage> that holds the value, per usage type
|
||||
USAGE_ELEMENTS = {"ceph": "name", "volume": "volume", "iscsi": "target"}
|
||||
|
||||
|
||||
class wvmSecrets(wvmConnect):
|
||||
def create_secret(self, ephemeral, private, secret_type, data):
|
||||
xml = f"""<secret ephemeral='{ephemeral}' private='{private}'>
|
||||
<usage type='{secret_type}'>"""
|
||||
if secret_type == "ceph":
|
||||
xml += f"""<name>{data}</name>"""
|
||||
if secret_type == "volume":
|
||||
xml += f"""<volume>{data}</volume>"""
|
||||
if secret_type == "iscsi":
|
||||
xml += f"""<target>{data}</target>"""
|
||||
xml += """</usage>
|
||||
</secret>"""
|
||||
self.wvm.secretDefineXML(xml)
|
||||
# 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 get_secret_value(self, uuid):
|
||||
secrt = self.get_secret(uuid)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue