fix: Escape user-controlled values in constructed XML commands

UserId.get_user_tags, UserId.get_groups, UserId.get_group_members, and
VsysOperations.create_import built XML payloads via str.format / string
concatenation. Special characters in caller-supplied names (& < > ')
either crashed ET.fromstring on the receiving end or altered the
structure of the command sent to the device. Wrap the interpolated
values with xml.sax.saxutils.escape (element text) and quoteattr
(attribute values), and add regression tests.

Author: btorresgil

panos/base.py

@@ -29,6 +29,7 @@
 import time
 import xml.dom.minidom as minidom
 import xml.etree.ElementTree as ET
+from xml.sax.saxutils import escape as xml_escape
 
 import pan.commit
 import pan.xapi
@@ -3386,7 +3387,7 @@ def create_import(self, vsys=None):
 
         if vsys != "shared" and vsys is not None and self.XPATH_IMPORT is not None:
             xpath = self.xpath_import_base(vsys)
-            element = "<member>{0}</member>".format(self.uid)
+            element = "<member>{0}</member>".format(xml_escape(self.uid))
             device = self.nearest_pandevice()
             device.active().xapi.set(xpath, element, retry_on_peer=True)
 

panos/userid.py

@@ -19,6 +19,7 @@
 
 import xml.etree.ElementTree as ET
 from copy import deepcopy
+from xml.sax.saxutils import escape, quoteattr
 
 from pan.xapi import PanXapiError
 
@@ -550,7 +551,7 @@ def get_groups(self, style=None):
             "<show><user><group><list>",
         ]
         if style is not None:
-            msg.append("<entry name='{0}'/>".format(style))
+            msg.append("<entry name={0}/>".format(quoteattr(style)))
         msg.append("</list></group></user></show>")
         cmd = "".join(msg)
         vsys = self.device.vsys or "vsys1"
@@ -594,7 +595,11 @@ def get_group_members(self, group):
             list
 
         """
-        cmd = "<show><user><group><name>" + group + "</name></group></user></show>"
+        cmd = (
+            "<show><user><group><name>"
+            + escape(group)
+            + "</name></group></user></show>"
+        )
         vsys = self.device.vsys or "vsys1"
 
         resp = self.device.op(cmd, vsys=vsys, cmd_xml=False)
@@ -644,12 +649,12 @@ def get_user_tags(self, user=None, prefix=None):
         if user is None:
             msg.append(
                 "<all>"
-                + "<limit>{0}</limit>".format(limit)
-                + "<start-point>{0}</start-point>".format(start)
+                + "<limit>{0}</limit>".format(escape(str(limit)))
+                + "<start-point>{0}</start-point>".format(escape(str(start)))
                 + "</all>"
             )
         else:
-            msg.append("<user>{0}</user>".format(user))
+            msg.append("<user>{0}</user>".format(escape(user)))
         msg.append("</registered-user></object></show>")
 
         cmd = ET.fromstring("".join(msg))

tests/test_base.py

@@ -23,6 +23,8 @@
 import pan.xapi
 import panos.base as Base
 import panos.errors as Err
+import panos.firewall
+import panos.network
 
 
 OBJECT_NAME = "MyObjectName"
@@ -1701,5 +1703,30 @@ def test_times_out(self, mocksleep):
         assert mocksleep.call_count == 0
 
 
+class TestVsysOperationsCreateImport(unittest.TestCase):
+    def test_create_import_escapes_uid(self):
+        evil_name = "zone&<bad>"
+        fw = panos.firewall.Firewall(
+            "fw1", "user", "passwd", "authkey", serial="S", vsys="vsys1"
+        )
+        vlan = panos.network.Vlan(evil_name)
+        fw.add(vlan)
+
+        with mock.patch.object(
+            panos.network.Vlan,
+            "XPATH_IMPORT",
+            new_callable=mock.PropertyMock,
+            return_value="/network/vlan",
+        ):
+            fw.xapi.set = mock.Mock()
+            vlan.create_import("vsys1")
+
+        args, _ = fw.xapi.set.call_args
+        element = args[1]
+        self.assertEqual(element, "<member>zone&amp;&lt;bad&gt;</member>")
+        parsed = ET.fromstring(element)
+        self.assertEqual(parsed.text, evil_name)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_userid.py

@@ -17,6 +17,7 @@
     import mock
 import sys
 import unittest
+import xml.etree.ElementTree as ET
 
 import panos.firewall
 import panos.panorama
@@ -100,5 +101,63 @@ def test_batch_untag_user(self):
         )
 
 
+    def test_get_user_tags_escapes_user(self):
+        evil = "admin</user><all><limit>999999</limit></all><user>x"
+        fw = panos.firewall.Firewall(
+            "fw1", "user", "passwd", "authkey", serial="Serial", vsys="vsys1"
+        )
+        empty_response = ET.fromstring(
+            b"<response status='success'><result/></response>"
+        )
+        fw.op = mock.Mock(return_value=empty_response)
+
+        fw.userid.get_user_tags(user=evil)
+
+        sent = fw.op.call_args[1]["cmd"]
+        parsed = ET.fromstring(sent)
+        users = parsed.findall("./object/registered-user/user")
+        self.assertEqual(len(users), 1)
+        self.assertEqual(users[0].text, evil)
+        self.assertIsNone(parsed.find("./object/registered-user/all"))
+
+    def test_get_groups_escapes_style(self):
+        evil = "x'/><evil/><entry name='y"
+        fw = panos.firewall.Firewall(
+            "fw1", "user", "passwd", "authkey", serial="Serial", vsys="vsys1"
+        )
+        empty_response = ET.fromstring(
+            b"<response status='success'><result>\nTotal: 0\n</result></response>"
+        )
+        fw.op = mock.Mock(return_value=empty_response)
+
+        fw.userid.get_groups(style=evil)
+
+        sent = fw.op.call_args[0][0]
+        parsed = ET.fromstring(sent)
+        entries = parsed.findall("./user/group/list/entry")
+        self.assertEqual(len(entries), 1)
+        self.assertEqual(entries[0].attrib["name"], evil)
+        self.assertIsNone(parsed.find(".//evil"))
+
+    def test_get_group_members_escapes_group(self):
+        evil = "g</name><evil/><name>x"
+        fw = panos.firewall.Firewall(
+            "fw1", "user", "passwd", "authkey", serial="Serial", vsys="vsys1"
+        )
+        empty_response = ET.fromstring(
+            b"<response status='success'><result>\nTotal: 0\n</result></response>"
+        )
+        fw.op = mock.Mock(return_value=empty_response)
+
+        fw.userid.get_group_members(evil)
+
+        sent = fw.op.call_args[0][0]
+        parsed = ET.fromstring(sent)
+        names = parsed.findall("./user/group/name")
+        self.assertEqual(len(names), 1)
+        self.assertEqual(names[0].text, evil)
+        self.assertIsNone(parsed.find(".//evil"))
+
+
 if __name__ == "__main__":
     unittest.main()