From 24e883c698a53e8be1cd18d76c30d63d38c5ba49 Mon Sep 17 00:00:00 2001 From: "Daniel A. Wozniak" Date: Fri, 26 Jun 2026 06:17:25 -0700 Subject: [PATCH 1/2] Fix #68915: Syndic reconnect after Master of Masters restart - Invalidate auth on pub_channel before close in Syndic.reconnect() so stale tokens are not reused on reconnection. - Start ZeroMQSocketMonitor on PublishClient.connect() with a reconnect_callback; on subsequent EVENT_CONNECTED events (after initial connection), the callback is scheduled via asyncio.ensure_future to trigger full re-authentication. - Add unit tests for ZeroMQSocketMonitor.monitor_callback reconnect behaviour and for Syndic.reconnect() auth invalidation. - Add changelog entry. --- changelog/68923.fixed.md | 1 + salt/minion.py | 2 + salt/transport/zeromq.py | 15 +++- tests/pytests/unit/syndic/test_syndic.py | 77 +++++++++++++++++++- tests/pytests/unit/transport/test_zeromq.py | 78 +++++++++++++++++++++ 5 files changed, 171 insertions(+), 2 deletions(-) create mode 100644 changelog/68923.fixed.md diff --git a/changelog/68923.fixed.md b/changelog/68923.fixed.md new file mode 100644 index 000000000000..9ef4c3e73025 --- /dev/null +++ b/changelog/68923.fixed.md @@ -0,0 +1 @@ +Syndic now invalidates auth and triggers re-authentication when the ZeroMQ publish channel reconnects after a Master of Masters restart. diff --git a/salt/minion.py b/salt/minion.py index aa4f27954088..8eef8ad98da4 100644 --- a/salt/minion.py +++ b/salt/minion.py @@ -5586,6 +5586,8 @@ async def _process_cmd_socket(self, payload): async def reconnect(self): if hasattr(self, "pub_channel"): self.pub_channel.on_recv(None) + if hasattr(self.pub_channel, "auth"): + self.pub_channel.auth.invalidate() if hasattr(self.pub_channel, "close"): self.pub_channel.close() del self.pub_channel diff --git a/salt/transport/zeromq.py b/salt/transport/zeromq.py index 84046009b70e..67964c9d8817 100644 --- a/salt/transport/zeromq.py +++ b/salt/transport/zeromq.py @@ -327,6 +327,12 @@ async def connect( self._socket.connect(master_pub_uri) if connect_callback: await connect_callback(True) + # Start ZMQ socket monitor to detect reconnections and trigger re-auth + if HAS_ZMQ_MONITOR and not getattr(self, "_monitor", None): + self._monitor = ZeroMQSocketMonitor( + self._socket, reconnect_callback=connect_callback + ) + self._monitor.start_io_loop(self.io_loop) async def connect_uri(self, uri, connect_callback=None, disconnect_callback=None): self._connect_called = True @@ -1574,7 +1580,7 @@ def _send_recv(self, socket, _TimeoutError=tornado.gen.TimeoutError): class ZeroMQSocketMonitor: __EVENT_MAP = None - def __init__(self, socket): + def __init__(self, socket, reconnect_callback=None): """ Create ZMQ monitor sockets @@ -1585,6 +1591,8 @@ def __init__(self, socket): self._monitor_socket = self._socket.get_monitor_socket() self._monitor_task = None self._running = asyncio.Event() + self._reconnect_callback = reconnect_callback + self._initial_connect_done = False def start_io_loop(self, io_loop): log.trace("Event monitor start!") @@ -1637,6 +1645,11 @@ def monitor_callback(self, msg): log.debug("ZeroMQ event: %s", evt) if evt["event"] == zmq.EVENT_MONITOR_STOPPED: self.stop() + elif evt["event"] == zmq.EVENT_CONNECTED: + if self._initial_connect_done and self._reconnect_callback: + log.debug("ZMQ socket reconnected, triggering re-authentication") + asyncio.ensure_future(self._reconnect_callback(True)) + self._initial_connect_done = True def stop(self): if self._socket is None: diff --git a/tests/pytests/unit/syndic/test_syndic.py b/tests/pytests/unit/syndic/test_syndic.py index 930e61fbd4d7..f89417bcabf0 100644 --- a/tests/pytests/unit/syndic/test_syndic.py +++ b/tests/pytests/unit/syndic/test_syndic.py @@ -4,9 +4,11 @@ import collections +import pytest + import salt.exceptions import salt.minion -from tests.support.mock import MagicMock, call +from tests.support.mock import AsyncMock, MagicMock, call def _syndic_manager(opts=None): @@ -229,3 +231,76 @@ def test_delayed_queue_prevents_memory_leak(): assert final_delayed_size <= initial_delayed_size + ( 3 * len(syndic_manager._syndics) ) + + +@pytest.fixture +def syndic_opts(): + return { + "conf_file": "conf/minion", + "id": "syndic", + "master": "127.0.0.1", + "master_port": 4506, + "master_uri": "tcp://127.0.0.1:4506", + "acceptance_wait_time": 10, + "acceptance_wait_time_max": 20, + "syndic_failover": "ordered", + "syndic_retries": 3, + "pki_dir": "/tmp", + "sock_dir": "/tmp", + "cachedir": "/tmp", + "auth_tries": 1, + "auth_timeout": 5, + "keysize": 2048, + "__role": "syndic", + "zmq_filtering": False, + "zmq_monitor": False, + "ipv6": False, + "recon_default": 1000, + "recon_max": 5000, + "recon_randomize": False, + } + + +async def test_syndic_reconnect_invalidates_auth(syndic_opts): + """ + When Syndic.reconnect() is called and pub_channel has an auth attribute, + the auth should be invalidated before the channel is closed. + """ + syndic = salt.minion.Syndic.__new__(salt.minion.Syndic) + syndic.opts = syndic_opts + syndic.connected = True + syndic.destroy = MagicMock() # prevent incomplete __del__ teardown + + mock_auth = MagicMock() + mock_pub_channel = MagicMock() + mock_pub_channel.auth = mock_auth + + syndic.pub_channel = mock_pub_channel + syndic.eval_master = AsyncMock(return_value=("127.0.0.1", MagicMock())) + + await syndic.reconnect() + + mock_auth.invalidate.assert_called_once() + mock_pub_channel.on_recv.assert_called_with(None) + mock_pub_channel.close.assert_called_once() + + +async def test_syndic_reconnect_without_auth_attribute(syndic_opts): + """ + When Syndic.reconnect() is called and pub_channel has no auth attribute, + reconnect should complete without error. + """ + syndic = salt.minion.Syndic.__new__(salt.minion.Syndic) + syndic.opts = syndic_opts + syndic.connected = True + syndic.destroy = MagicMock() # prevent incomplete __del__ teardown + + mock_pub_channel = MagicMock(spec=["on_recv", "close"]) + + syndic.pub_channel = mock_pub_channel + syndic.eval_master = AsyncMock(return_value=("127.0.0.1", MagicMock())) + + await syndic.reconnect() + + mock_pub_channel.on_recv.assert_called_with(None) + mock_pub_channel.close.assert_called_once() diff --git a/tests/pytests/unit/transport/test_zeromq.py b/tests/pytests/unit/transport/test_zeromq.py index 29e1c0aed21f..92f4760e3cbb 100644 --- a/tests/pytests/unit/transport/test_zeromq.py +++ b/tests/pytests/unit/transport/test_zeromq.py @@ -13,6 +13,7 @@ import tornado.concurrent import tornado.gen import tornado.ioloop +import zmq import zmq.eventloop.future from pytestshellutils.utils import ports @@ -2528,3 +2529,80 @@ def test_backoff_timer(): next_iteration += next_iteration * percent * ourcount assert ourcount == 39 assert backoff() == maximum + + +def _make_zmq_event_msg(event_id): + """Build a minimal ZMQ monitor message matching parse_monitor_message format. + + Frame 1: 6 bytes — 16-bit event id (signed short) + 32-bit value (signed int). + Frame 2: endpoint bytestring. + """ + import struct + + frame1 = struct.pack("=hi", event_id, 0) + frame2 = b"tcp://127.0.0.1:4505" + return [frame1, frame2] + + +async def test_zmq_monitor_callback_reconnect_triggers_callback(): + """ + ZeroMQSocketMonitor.monitor_callback fires reconnect_callback after initial + connection is established (i.e., on subsequent EVENT_CONNECTED events). + """ + mock_socket = MagicMock() + mock_socket.get_monitor_socket.return_value = MagicMock() + callback = AsyncMock() + + monitor = salt.transport.zeromq.ZeroMQSocketMonitor( + mock_socket, reconnect_callback=callback + ) + + # Simulate first EVENT_CONNECTED — should NOT call callback yet + msg = _make_zmq_event_msg(zmq.EVENT_CONNECTED) + monitor.monitor_callback(msg) + callback.assert_not_called() + assert monitor._initial_connect_done is True + + # Simulate second EVENT_CONNECTED (reconnect) — should schedule callback + monitor.monitor_callback(msg) + # Give the event loop a chance to run the scheduled coroutine + await asyncio.sleep(0) + callback.assert_called_once_with(True) + + +def test_zmq_monitor_callback_no_callback_on_reconnect(): + """ + ZeroMQSocketMonitor.monitor_callback does not raise when no reconnect_callback + is provided and an EVENT_CONNECTED is received after initial connection. + """ + mock_socket = MagicMock() + mock_socket.get_monitor_socket.return_value = MagicMock() + + monitor = salt.transport.zeromq.ZeroMQSocketMonitor(mock_socket) + + msg = _make_zmq_event_msg(zmq.EVENT_CONNECTED) + # First connect + monitor.monitor_callback(msg) + # Second connect (reconnect) — no callback, should not raise + monitor.monitor_callback(msg) + + +def test_zmq_monitor_callback_first_connect_no_callback(): + """ + ZeroMQSocketMonitor.monitor_callback does not call reconnect_callback on the + very first EVENT_CONNECTED — only on subsequent reconnects. + """ + mock_socket = MagicMock() + mock_socket.get_monitor_socket.return_value = MagicMock() + callback = AsyncMock() + + monitor = salt.transport.zeromq.ZeroMQSocketMonitor( + mock_socket, reconnect_callback=callback + ) + assert monitor._initial_connect_done is False + + msg = _make_zmq_event_msg(zmq.EVENT_CONNECTED) + monitor.monitor_callback(msg) + + callback.assert_not_called() + assert monitor._initial_connect_done is True From 6f9edc4088f75ef69afc7e6634cb06045309f0b2 Mon Sep 17 00:00:00 2001 From: "Daniel A. Wozniak" Date: Fri, 3 Jul 2026 18:24:02 -0700 Subject: [PATCH 2/2] Add integration test for syndic reconnect after MoM restart Regression test for #68915. Verifies that a syndic re-establishes its ZeroMQ connection to the Master of Masters after the MoM process is restarted, and that job dispatch to downstream minions through the syndic remains functional after the reconnect. --- .../syndic/sync/test_syndic_reconnect.py | 207 ++++++++++++++++++ 1 file changed, 207 insertions(+) create mode 100644 tests/pytests/scenarios/syndic/sync/test_syndic_reconnect.py diff --git a/tests/pytests/scenarios/syndic/sync/test_syndic_reconnect.py b/tests/pytests/scenarios/syndic/sync/test_syndic_reconnect.py new file mode 100644 index 000000000000..4ea3c5242cc9 --- /dev/null +++ b/tests/pytests/scenarios/syndic/sync/test_syndic_reconnect.py @@ -0,0 +1,207 @@ +# Regression test for #68915 +""" +Integration test: syndic reconnects to Master of Masters after MoM restart. + +Before the fix, Syndic.reconnect() did not invalidate the stale auth token +on the pub_channel before closing it. When the ZeroMQ socket detected a +reconnection it would attempt to reuse the old token, causing authentication +to fail and leaving the syndic unable to forward jobs. + +This test exercises the full MoM→syndic→minion topology with a real process +restart of the MoM, then asserts that a ``test.ping`` dispatched through the +MoM still reaches the downstream minion after the restart. + +Why a functional/integration test rather than a unit test? + The reconnect path involves ZeroMQ socket-monitor callbacks + (``ZeroMQSocketMonitor.monitor_callback``) and asyncio scheduling + (``asyncio.ensure_future``), which only fire inside a live event loop that + owns a real ZMQ context. Unit mocks cannot exercise the timing of the + reconnect-triggered re-authentication handshake. +""" +import logging +import time + +import pytest +from saltfactories.utils import random_string + +from tests.conftest import FIPS_TESTRUN + +log = logging.getLogger(__name__) + +pytestmark = [ + pytest.mark.skip_on_fips_enabled_platform, +] + + +# --------------------------------------------------------------------------- +# Function-scoped fixtures — we need to be able to restart the MoM inside +# the test body, so we cannot share package-scope fixtures with the other +# tests in this directory. +# --------------------------------------------------------------------------- + + +@pytest.fixture +def mom(request, salt_factories): + """Master of Masters — the top-level master.""" + config_defaults = { + "transport": request.config.getoption("--transport"), + } + config_overrides = { + "interface": "127.0.0.1", + "auto_accept": True, + "order_masters": True, + "gather_job_timeout": 30, + "timeout": 30, + "fips_mode": FIPS_TESTRUN, + "publish_signing_algorithm": ( + "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1" + ), + } + factory = salt_factories.salt_master_daemon( + random_string("mom-"), + defaults=config_defaults, + overrides=config_overrides, + extra_cli_arguments_after_first_start_failure=["--log-level=info"], + ) + with factory.started(start_timeout=120): + yield factory + + +@pytest.fixture +def syndic(mom): + """Syndic that connects to *mom*.""" + ret_port = mom.config["ret_port"] + port = mom.config["publish_port"] + addr = mom.config["interface"] + + config_defaults = { + "transport": mom.config["transport"], + "interface": "127.0.0.2", + "publish_port": f"{port}", + } + master_overrides = { + "interface": "127.0.0.2", + "auto_accept": True, + "syndic_master": f"{addr}", + "syndic_master_port": f"{ret_port}", + "fips_mode": FIPS_TESTRUN, + "publish_signing_algorithm": ( + "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1" + ), + } + minion_overrides = { + "master": "127.0.0.2", + "publish_port": f"{port}", + "fips_mode": FIPS_TESTRUN, + "encryption_algorithm": "OAEP-SHA224" if FIPS_TESTRUN else "OAEP-SHA1", + "signing_algorithm": "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1", + } + factory = mom.salt_syndic_daemon( + random_string("syndic-"), + defaults=config_defaults, + master_overrides=master_overrides, + minion_overrides=minion_overrides, + extra_cli_arguments_after_first_start_failure=["--log-level=info"], + ) + factory.after_terminate(factory.minion.terminate) + factory.after_terminate(factory.master.terminate) + with factory.started(start_timeout=120): + yield factory + + +@pytest.fixture +def minion(syndic): + """Downstream minion connected to the syndic's internal master.""" + config_defaults = { + "transport": syndic.config["transport"], + } + port = syndic.master.config["ret_port"] + addr = syndic.master.config["interface"] + config_overrides = { + "master": f"{addr}:{port}", + "fips_mode": FIPS_TESTRUN, + "encryption_algorithm": "OAEP-SHA224" if FIPS_TESTRUN else "OAEP-SHA1", + "signing_algorithm": "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1", + } + factory = syndic.master.salt_minion_daemon( + random_string("minion-"), + defaults=config_defaults, + overrides=config_overrides, + extra_cli_arguments_after_first_start_failure=["--log-level=info"], + ) + with factory.started(start_timeout=120): + yield factory + + +def test_syndic_reconnects_after_mom_restart(mom, syndic, minion): + """ + Regression test for #68915. + + Verify that after the Master of Masters (MoM) is restarted: + + 1. The syndic re-establishes its ZeroMQ publish-channel connection and + completes re-authentication (the fix invalidates the stale auth token + before closing the channel so a fresh handshake is performed). + 2. A ``test.ping`` dispatched from the MoM reaches the downstream minion + through the syndic and returns a truthy result. + + Without the fix, step 2 times out or returns an empty result because the + syndic's reconnect attempt reuses a stale auth token and the MoM rejects + the subscription. + """ + salt_cli = mom.salt_cli(timeout=60) + minion_id = minion.id + syndic_id = syndic.id + + # Baseline: verify the topology is working before the restart. + # Targeting "*" returns {syndic_id: True, minion_id: True} for a + # syndic topology; the syndic itself acts as a pseudo-minion on the MoM. + ret = salt_cli.run("test.ping", minion_tgt="*", _timeout=30) + assert ret.returncode == 0, f"Baseline ping failed before MoM restart: {ret}" + assert isinstance(ret.data, dict), f"Unexpected baseline result type: {ret.data!r}" + assert ( + ret.data.get(minion_id) is True + ), f"Downstream minion not responding in baseline: {ret.data}" + + log.info("Baseline ping passed. Restarting MoM (%s)…", mom.id) + + # Stop the MoM, pause briefly to let the syndic detect the disconnect, + # then bring the MoM back up. + with mom.stopped(): + log.info("MoM stopped. Waiting for syndic to detect the disconnect…") + time.sleep(3) + + log.info("MoM restarted. Waiting for syndic to reconnect and re-authenticate…") + + # Give the syndic up to 120 s to reconnect, re-authenticate, and register + # with the MoM before we send the post-restart ping. The ZMQ reconnect + # interval is randomised (recon_default + up to recon_max), so allow + # generous time for the syndic to reconnect, re-authenticate with the + # restarted MoM, and for the minion to relay job returns back through + # the syndic. + last_ret = None + deadline = time.monotonic() + 120 + while time.monotonic() < deadline: + last_ret = salt_cli.run("test.ping", minion_tgt="*", _timeout=20) + if ( + last_ret.returncode == 0 + and isinstance(last_ret.data, dict) + and last_ret.data.get(minion_id) is True + ): + log.info( + "Post-restart ping succeeded after %.1f s", + 120 - (deadline - time.monotonic()), + ) + break + log.debug( + "Post-restart ping attempt: returncode=%r data=%r", + last_ret.returncode, + last_ret.data, + ) + time.sleep(5) + else: + pytest.fail( + f"Syndic did not reconnect to MoM within 120 s after restart. " + f"Last response: returncode={last_ret.returncode!r}, " + f"data={last_ret.data!r}" + )