diff --git a/panos/base.py b/panos/base.py index c1ee2def..7ac361bc 100644 --- a/panos/base.py +++ b/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 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 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 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") diff --git a/panos/firewall.py b/panos/firewall.py index d131ddea..e4703271 100644 --- a/panos/firewall.py +++ b/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") diff --git a/tests/test_base.py b/tests/test_base.py index 2ba1437d..c072ef55 100644 --- a/tests/test_base.py +++ b/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 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"any" + + 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"any" + + 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"