fix(vrtManager): request values are not written into XPath expressions

Four lookups built XPath expressions from values that come from
requests (CodeQL py/xpath-injection, plus get_capabilities, which
CodeQL did not flag; the pages are for superusers):

- The architecture in the create wizard's URLs went into
  arch[@name='...'] in get_emulator, get_machine_types and
  get_capabilities: "nope' or '1'='1" matched every architecture. It is
  now an XPath variable ($arch), which is never parsed as XPath.
- Network filter rules were found with a predicate built from the
  posted action, direction and priority: a quote raised SyntaxError (an
  error page). The rule is now found by comparing its attributes.
- unset_qos put the posted direction into the path, so "*" removed
  every bandwidth setting. It accepts inbound and outbound, as set_qos
  does; the network page shows an error for anything else.

No other XPath in vrtManager is built from a value; the remaining
expressions are constants. The tests fail on the previous code.

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(

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

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

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)

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 = """<capabilities><guest><arch name='x86_64'><wordsize>64</wordsize><emulator>/usr/bin/qemu-x86</emulator>
+<machine canonical='pc-q35'>q35</machine></arch>
+<arch name='aarch64'><emulator>/usr/bin/qemu-arm</emulator><machine>virt</machine></arch></guest></capabilities>"""
+
+FILTER = """<filter name='f'><rule action='drop' direction='in' priority='1'><tcp/></rule>
+<rule action='drop' direction='in' priority='2'><udp/></rule></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("<rule "), 2)
+        self.assertEqual(nwf.delete_rule("it's", "in", "1").count("<rule "), 2)
+        kept = nwf.delete_rule("drop", "in", "2")
+        self.assertEqual(kept.count("<rule "), 1)
+        self.assertIn("priority=\"1\"", kept)
+        # add_rule appends to the matching rule only
+        added = nwf.add_rule("<rule action='drop' direction='in' priority='2'><icmp/></rule>")
+        self.assertEqual(added.count("<rule "), 2)
+        self.assertIn("<icmp", added.split('priority="2"')[1])
+
+    def test_qos_direction_must_be_known(self):
+        net = wvmNetwork.__new__(wvmNetwork)
+        net.wvm = FakeConn()
+        net._XMLDesc = lambda flags: "<network><bandwidth><inbound average='1'/><outbound average='2'/></bandwidth></network>"
+        with self.assertRaises(ValueError):
+            net.unset_qos("*")
+        net.unset_qos("inbound")
+        self.assertNotIn("inbound", net.wvm.defined)
+        self.assertIn("outbound", net.wvm.defined)