diff --git a/networks/views.py b/networks/views.py
index e1bf8b67..26104c75 100644
--- a/networks/views.py
+++ b/networks/views.py
@@ -258,7 +258,11 @@ def network(request, compute_id, pool):
return HttpResponseRedirect(request.get_full_path())
if "unset_qos" in request.POST:
qos_dir = request.POST.get("qos_direction", "")
- conn.unset_qos(qos_dir)
+ try:
+ conn.unset_qos(qos_dir)
+ except ValueError as err:
+ messages.error(request, err)
+ return HttpResponseRedirect(request.get_full_path())
if conn.is_active():
messages.success(
diff --git a/vrtManager/connection.py b/vrtManager/connection.py
index b777d18c..44a8e253 100644
--- a/vrtManager/connection.py
+++ b/vrtManager/connection.py
@@ -410,7 +410,7 @@ class wvmConnect(object):
def guests(ctx):
result = dict()
- for arch_el in ctx.xpath("/capabilities/guest/arch[@name='{}']".format(arch)):
+ for arch_el in ctx.xpath("/capabilities/guest/arch[@name=$arch]", arch=arch):
result["wordsize"] = arch_el.find("wordsize").text
result["emulator"] = arch_el.find("emulator").text
result["domain"] = [v for v in arch_el.xpath("domain/@type")]
@@ -625,7 +625,12 @@ class wvmConnect(object):
"""
:return: emulator list
"""
- return util.get_xml_path(self.get_cap_xml(), "/capabilities/guest/arch[@name='{}']/emulator".format(arch))
+ # arch comes from the URL: an XPath variable, not part of the expression
+ def emulator(ctx):
+ found = ctx.xpath("/capabilities/guest/arch[@name=$arch]/emulator", arch=arch)
+ return found[0].text if found else None
+
+ return util.get_xml_path(self.get_cap_xml(), func=emulator)
def get_machine_types(self, arch):
"""
@@ -634,9 +639,9 @@ class wvmConnect(object):
def machines(ctx):
result = list()
- canonical_name = ctx.xpath("/capabilities/guest/arch[@name='{}']/machine[@canonical]".format(arch))
+ canonical_name = ctx.xpath("/capabilities/guest/arch[@name=$arch]/machine[@canonical]", arch=arch)
if not canonical_name:
- canonical_name = ctx.xpath("/capabilities/guest/arch[@name='{}']/machine".format(arch))
+ canonical_name = ctx.xpath("/capabilities/guest/arch[@name=$arch]/machine", arch=arch)
for archi in canonical_name:
result.append(archi.text)
return result
diff --git a/vrtManager/network.py b/vrtManager/network.py
index 2202f56f..51a7d6b6 100644
--- a/vrtManager/network.py
+++ b/vrtManager/network.py
@@ -379,6 +379,9 @@ class wvmNetwork(wvmConnect):
self.wvm.networkDefineXML(new_xml)
def unset_qos(self, direction):
+ # direction goes into the XPath below
+ if direction not in ("inbound", "outbound"):
+ raise ValueError("Direction must be inbound or outbound")
tree = etree.fromstring(self._XMLDesc(0))
for direct in tree.xpath(f"/network/bandwidth/{direction}"):
parent = direct.getparent()
diff --git a/vrtManager/nwfilters.py b/vrtManager/nwfilters.py
index 7e8d5350..4ad768e8 100644
--- a/vrtManager/nwfilters.py
+++ b/vrtManager/nwfilters.py
@@ -84,13 +84,19 @@ class wvmNWFilter(wvmConnect):
break
return ElementTree.tostring(tree).decode()
+ @staticmethod
+ def _rules(tree, action, direction, priority):
+ # compared, not put into an XPath predicate: the values come from forms
+ return [
+ rule
+ for rule in tree.findall("./rule")
+ if (rule.get("action"), rule.get("direction"), rule.get("priority")) == (action, direction, priority)
+ ]
+
def delete_rule(self, action, direction, priority):
tree = ElementTree.fromstring(self._XMLDesc(0))
- rule_tree = tree.findall(
- "./rule[@action='%s'][@direction='%s'][@priority='%s']"
- % (action, direction, priority)
- )
+ rule_tree = self._rules(tree, action, direction, priority)
if rule_tree:
tree.remove(rule_tree[0])
@@ -111,10 +117,7 @@ class wvmNWFilter(wvmConnect):
rule_direction = rule.get("direction")
rule_priority = rule.get("priority")
rule_directives = rule.find("./")
- rule_tree = tree.findall(
- "./rule[@action='%s'][@direction='%s'][@priority='%s']"
- % (rule_action, rule_direction, rule_priority)
- )
+ rule_tree = self._rules(tree, rule_action, rule_direction, rule_priority)
if rule_tree:
rule_tree[0].append(rule_directives)
diff --git a/vrtManager/test_xpath_values.py b/vrtManager/test_xpath_values.py
new file mode 100644
index 00000000..04757c09
--- /dev/null
+++ b/vrtManager/test_xpath_values.py
@@ -0,0 +1,65 @@
+"""Values from requests do not change the XPath queries they are used in."""
+
+import unittest
+
+from django.conf import settings
+
+if not settings.configured:
+ settings.configure(MAC_OUI="52:54:10")
+
+from vrtManager.connection import wvmConnect
+from vrtManager.network import wvmNetwork
+from vrtManager.nwfilters import wvmNWFilter
+
+CAPS = """64/usr/bin/qemu-x86
+q35
+/usr/bin/qemu-armvirt"""
+
+FILTER = """
+"""
+
+
+class FakeConn:
+ def __init__(self):
+ self.defined = None
+
+ def networkDefineXML(self, xml):
+ self.defined = xml
+
+
+class XPathValueTests(unittest.TestCase):
+ def setUp(self):
+ self.conn = wvmConnect.__new__(wvmConnect)
+ self.conn.get_cap_xml = lambda: CAPS
+
+ def test_arch_lookups(self):
+ self.assertEqual(self.conn.get_emulator("x86_64"), "/usr/bin/qemu-x86")
+ self.assertEqual(self.conn.get_machine_types("aarch64"), ["virt"])
+ injected = "nope' or '1'='1"
+ self.assertIsNone(self.conn.get_emulator(injected))
+ self.assertEqual(self.conn.get_machine_types(injected), [])
+ self.assertEqual(self.conn.get_capabilities(injected), {})
+
+ def test_nwfilter_rule_is_matched_by_its_attributes(self):
+ nwf = wvmNWFilter.__new__(wvmNWFilter)
+ nwf._XMLDesc = lambda flags: FILTER
+ # values with quotes or brackets are compared, not parsed as XPath
+ self.assertEqual(nwf.delete_rule("drop'][@priority='2", "in", "1").count("")
+ self.assertEqual(added.count(""
+ with self.assertRaises(ValueError):
+ net.unset_qos("*")
+ net.unset_qos("inbound")
+ self.assertNotIn("inbound", net.wvm.defined)
+ self.assertIn("outbound", net.wvm.defined)