Fix security policy partial commit

panos/base.py

@@ -556,6 +556,19 @@ def element_str(self, pretty_print=False):
             return parsed.toprettyxml(indent="\t", encoding="utf-8")
         return ET.tostring(self.element(), encoding="utf-8")
 
+    def element_str_inner(self):
+        """XML of this object's child elements only, without the root entry/member wrapper.
+
+        Used by create() when issuing set() against the entry's own xpath. PAN-OS
+        expects inner content only at that xpath — passing the full <entry> wrapper
+        causes a schema error ("may need to override template object first").
+
+        Returns:
+            bytes: concatenated XML of child elements, empty bytes if none.
+        """
+        root = self.element()
+        return b"".join(ET.tostring(child) for child in root)
+
     def _root_element(self):
         if self.SUFFIX == ENTRY:
             return ET.Element("entry", {"name": self.uid})
@@ -669,13 +682,33 @@ def create(self):
             device.id + ': create called on %s object "%s"' % (type(self), self.uid)
         )
         device.set_config_changed()
-        element = self.element_str()
-        if self.HA_SYNC:
-            device.active().xapi.set(
-                self.xpath_short(), element, retry_on_peer=self.HA_SYNC
-            )
+        # For entry/member objects, use edit() against the entry's own xpath rather than
+        # For entry/member objects, use set() against the entry's own xpath rather than
+        # the parent container xpath. PAN-OS records set() on an entry xpath as a CREATE
+        # For entry/member objects, use set() against the entry's own xpath with inner
+        # content only (no <entry> wrapper). PAN-OS records this as a CREATE owned by
+        # the calling admin. Using the parent container xpath is recorded as an EDIT on
+        # the container and changes admin lock ownership, breaking partial commits for
+        # other admins. Passing the full <entry> wrapper at the entry xpath causes a
+        # PAN-OS schema error ("may need to override template object first").
+        if self.SUFFIX in (ENTRY, MEMBER):
+            if self.HA_SYNC:
+                device.active().xapi.set(
+                    self.xpath(), self.element_str_inner(), retry_on_peer=self.HA_SYNC
+                )
+            else:
+                device.xapi.set(
+                    self.xpath(), self.element_str_inner(), retry_on_peer=self.HA_SYNC
+                )
         else:
-            device.xapi.set(self.xpath_short(), element, retry_on_peer=self.HA_SYNC)
+            if self.HA_SYNC:
+                device.active().xapi.set(
+                    self.xpath_short(), self.element_str(), retry_on_peer=self.HA_SYNC
+                )
+            else:
+                device.xapi.set(
+                    self.xpath_short(), self.element_str(), retry_on_peer=self.HA_SYNC
+                )
         for child in self.children:
             child._check_child_methods("create")
 

panos/firewall.py

@@ -622,7 +622,7 @@ def element(self):
                 ET.SubElement(partial, "shared-object").text = "excluded"
             if self.exclude_policy_and_objects:
                 ET.SubElement(partial, "policy-and-objects").text = "excluded"
-            fe.append(partial)
+            root.append(partial)
 
         if self.force:
             fe = ET.SubElement(root, "force")

tests/test_base.py

@@ -583,6 +583,65 @@ def test_create_without_ha_sync(self, m_uid):
         for c in self.obj.children:
             c._check_child_methods.assert_called_once_with("create")
 
+    @mock.patch("panos.base.PanObject.uid", new_callable=mock.PropertyMock)
+    def test_create_entry_suffix_uses_set_on_entry_xpath_with_inner_element(self, m_uid):
+        # ENTRY-suffix objects must use xapi.set() against the entry's own xpath with
+        # inner content only (no <entry> wrapper). Passing the full element_str() at
+        # the entry xpath causes PAN-OS schema error; using parent xpath causes an EDIT
+        # on the container changing admin lock ownership and breaking partial commits.
+        PanDeviceId = "42"
+        PanDeviceXpath = "path/to/entry"
+        PanDeviceInnerElement = b"<from>any</from>"
+
+        self.obj.SUFFIX = Base.ENTRY
+
+        spec = {"id": PanDeviceId}
+        m_panos = mock.Mock(**spec)
+        self.obj.nearest_pandevice = mock.Mock(return_value=m_panos)
+        self.obj.xpath = mock.Mock(return_value=PanDeviceXpath)
+        self.obj.element_str_inner = mock.Mock(return_value=PanDeviceInnerElement)
+        m_uid.return_value = "uid"
+
+        ret_val = self.obj.create()
+
+        self.assertIsNone(ret_val)
+        m_panos.set_config_changed.assert_called_once_with()
+        m_panos.active().xapi.set.assert_called_once_with(
+            PanDeviceXpath,
+            PanDeviceInnerElement,
+            retry_on_peer=self.obj.HA_SYNC,
+        )
+        self.obj.xpath.assert_called_once_with()
+        self.obj.element_str_inner.assert_called_once_with()
+
+    @mock.patch("panos.base.PanObject.uid", new_callable=mock.PropertyMock)
+    def test_create_entry_suffix_without_ha_sync(self, m_uid):
+        PanDeviceId = "42"
+        PanDeviceXpath = "path/to/entry"
+        PanDeviceInnerElement = b"<from>any</from>"
+
+        self.obj.SUFFIX = Base.ENTRY
+        self.obj.HA_SYNC = False
+
+        spec = {"id": PanDeviceId}
+        m_panos = mock.Mock(**spec)
+        self.obj.nearest_pandevice = mock.Mock(return_value=m_panos)
+        self.obj.xpath = mock.Mock(return_value=PanDeviceXpath)
+        self.obj.element_str_inner = mock.Mock(return_value=PanDeviceInnerElement)
+        m_uid.return_value = "uid"
+
+        ret_val = self.obj.create()
+
+        self.assertIsNone(ret_val)
+        m_panos.set_config_changed.assert_called_once_with()
+        m_panos.xapi.set.assert_called_once_with(
+            PanDeviceXpath,
+            PanDeviceInnerElement,
+            retry_on_peer=self.obj.HA_SYNC,
+        )
+        self.obj.xpath.assert_called_once_with()
+        self.obj.element_str_inner.assert_called_once_with()
+
     @mock.patch("panos.base.PanObject.uid", new_callable=mock.PropertyMock)
     def test_delete_with_ha_sync_no_parent(self, m_uid):
         PanDeviceId = "42"