From 5141637c4e0a4f57e6148738d5857ac5cffdf483 Mon Sep 17 00:00:00 2001
From: Nikita Sychev <upml@yandex-team.ru>
Date: Mon, 26 Oct 2026 12:21:00 +0500
Subject: [PATCH] Survive a cluster peer that has not learned our public key
 yet

Fixes #70565
---
 changelog/70565.fixed.md                  |  1 +
 salt/channel/server.py                    | 13 +++++-
 tests/pytests/unit/channel/test_server.py | 51 +++++++++++++++++++++++
 3 files changed, 63 insertions(+), 2 deletions(-)
 create mode 100644 changelog/70565.fixed.md

diff --git a/changelog/70565.fixed.md b/changelog/70565.fixed.md
new file mode 100644
index 000000000000..3f1d4e414275
--- /dev/null
+++ b/changelog/70565.fixed.md
@@ -0,0 +1 @@
+Fixed a clustered master logging `CRITICAL Unhandled error while polling master events` with `KeyError: 'aes'` while the master still awaits for peer's public key.
diff --git a/salt/channel/server.py b/salt/channel/server.py
index ad19306a003e..8720d3eb6f63 100644
--- a/salt/channel/server.py
+++ b/salt/channel/server.py
@@ -4027,8 +4027,17 @@ async def handle_pool_publish(self, payload):
                 if peer == master_id:
                     log.debug("Skip our own cluster peer event %s", tag)
                     return
-                aes = data["peers"][master_id]["aes"]
-                sig = data["peers"][master_id]["sig"]
+                entry = data["peers"].get(master_id) or {}
+                if "aes" not in entry:
+                    # The sender records an empty entry for a peer whose public
+                    # key it does not hold yet. It re-announces once the key
+                    # arrives, so skip this round.
+                    log.info(
+                        "Peer %s has no AES key for us yet, awaiting re-announce", peer
+                    )
+                    return
+                aes = entry["aes"]
+                sig = entry["sig"]
                 key_str = self.master_key.master_key.decrypt(
                     aes, algorithm=self.opts["cluster_encryption_algorithm"]
                 )
diff --git a/tests/pytests/unit/channel/test_server.py b/tests/pytests/unit/channel/test_server.py
index 97adf659f8e2..00c3ee19720e 100644
--- a/tests/pytests/unit/channel/test_server.py
+++ b/tests/pytests/unit/channel/test_server.py
@@ -1,5 +1,6 @@
 import asyncio
 import ctypes
+import logging
 import multiprocessing
 import pathlib
 import time
@@ -1435,3 +1436,53 @@ async def test_join_reply_refreshes_master_keys_cache_70090(tmp_path, key_data):
         "cluster_key (mirrors the ``self.key = self.cluster_key`` line "
         "at the end of MasterKeys._setup_keys)."
     )
+
+
+@pytest.mark.parametrize(
+    "our_entry",
+    [
+        # ``send_aes_key_event`` records an empty entry if there is no public key.
+        {},
+        # The host is not yet in the ``peers`` dict at all.
+        None,
+    ],
+)
+async def test_handle_pool_publish_peer_without_our_key_yet(
+    cluster_master_opts, caplog, our_entry
+):
+    """
+    Regression test for issue #70565.
+
+    A peer that has not learned our public key yet must not raise out of the
+    event-polling loop.
+    """
+    from tests.support.mock import AsyncMock, MagicMock
+
+    bare_id = cluster_master_opts["id"].removesuffix("_master")
+    peers = {} if our_entry is None else {bare_id: our_entry}
+    sibling_data = {"peer_id": "salt-master-2", "peers": peers}
+    tag = salt.utils.event.tagify("salt-master-2", "peer", "cluster")
+    payload = salt.utils.event.SaltEvent.pack(tag, sibling_data)
+
+    channel = server.MasterPubServerChannel.__new__(server.MasterPubServerChannel)
+    channel.opts = cluster_master_opts
+    channel.peer_keys = {}
+    channel.auth_errors = {"salt-master-2": [], "salt-master-3": []}
+    channel.transport = MagicMock()
+    channel.transport.publish_payload = AsyncMock()
+    # Never reached once the empty entry is recognised; a bare mock is enough
+    # to prove the handler returns before any crypto is attempted.
+    channel.master_key = MagicMock()
+
+    with caplog.at_level(logging.INFO):
+        await channel.handle_pool_publish(payload)
+
+    assert (
+        "salt-master-2" not in channel.peer_keys
+    ), "nothing should be cached from an announce that carries no key for us"
+    assert not [
+        record for record in caplog.records if record.levelno >= logging.ERROR
+    ], "a peer that has not learned our key yet is a transient, not an error"
+    assert any(
+        "no AES key for us yet" in record.getMessage() for record in caplog.records
+    ), "the skipped round must say why it was skipped"