From ed8bc558ff2a06b855c6492320d64c675e91834e Mon Sep 17 00:00:00 2001 From: krystian-elisity Date: Mon, 22 Jun 2026 11:58:29 +0200 Subject: [PATCH] fix(ha): clear stale ha_failed when refreshing HA active state refresh_ha_active() updated each device's _ha_active flag from the live HA state but left ha_failed untouched. A device marked failed by a prior transient connection error therefore stayed "failed" even after it was reached again, so XapiWrapper kept routing API calls to the HA peer (the passive device) indefinitely. In a Panorama HA pair this sends User-ID / config calls to the passive member, which never propagates them. The else-branch is only reached after successfully reading each device's live HA state, proving it is reachable, so ha_failed can be cleared there. Non-authoritative states (disabled/initial) return early and are untouched. Co-Authored-By: Claude Opus 4.8 (1M context) --- panos/base.py | 2 ++ tests/test_firewall.py | 42 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/panos/base.py b/panos/base.py index c1ee2def..25fa4dd3 100644 --- a/panos/base.py +++ b/panos/base.py @@ -4765,6 +4765,8 @@ def refresh_ha_active(self): else: for fw, state in ((self, self_state), (self.ha_peer, peer_state)): fw._ha_active = state == "active" + # Reached this device for its state, so it is no longer failed. + fw.ha_failed = False return self_state def synchronize_config(self): diff --git a/tests/test_firewall.py b/tests/test_firewall.py index e883dcb0..1392f1d6 100644 --- a/tests/test_firewall.py +++ b/tests/test_firewall.py @@ -14,6 +14,12 @@ import unittest +# Python 2 has no unittest.mock; fall back to the standalone mock package. +try: + from unittest import mock +except ImportError: + import mock + import panos import panos.firewall @@ -51,5 +57,41 @@ def test_id_returns_no_id(self): self.assertEqual(expected, ret_val) +class TestFirewallHa(unittest.TestCase): + def _ha_pair(self): + fw = panos.firewall.Firewall("10.0.0.1", "user", "pass") + peer = panos.firewall.Firewall("10.0.0.2", "user", "pass") + fw.set_ha_peers(peer) + return fw, peer + + def test_refresh_ha_active_clears_stale_ha_failed(self): + """A live state refresh clears ha_failed on the reachable device.""" + fw, peer = self._ha_pair() + # A transient error failed this device and promoted the peer to active. + fw.set_failed() + self.assertTrue(fw.ha_failed) + # Live HA now reports this device is active again. + fw.show_highavailability_state = mock.Mock(return_value=("active", None)) + peer.show_highavailability_state = mock.Mock(return_value=("passive", None)) + + fw.refresh_ha_active() + + self.assertFalse(fw.ha_failed) + + def test_refresh_ha_active_keeps_ha_failed_when_state_not_authoritative(self): + """A non-authoritative state (initial/disabled) leaves ha_failed untouched.""" + fw, peer = self._ha_pair() + # A transient error failed this device. + fw.set_failed() + self.assertTrue(fw.ha_failed) + # HA is still initializing, so the reported state is not authoritative. + fw.show_highavailability_state = mock.Mock(return_value=("initial", None)) + peer.show_highavailability_state = mock.Mock(return_value=("active", None)) + + fw.refresh_ha_active() + + self.assertTrue(fw.ha_failed) + + if __name__ == "__main__": unittest.main()