From ef90a3d808f5119ad7ab7cc0efaf430e40dfbb42 Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 17:24:12 +0000
Subject: [PATCH 01/11] Update EZSP decision and device update comments

---
 bellows/types/named.py | 55 +++++++++++++++++++++++++++++++++---------
 1 file changed, 44 insertions(+), 11 deletions(-)

diff --git a/bellows/types/named.py b/bellows/types/named.py
index 66391557..e2780902 100644
--- a/bellows/types/named.py
+++ b/bellows/types/named.py
@@ -996,19 +996,20 @@ class EzspNetworkScanType(basic.enum8):
 
 
 class EmberJoinDecision(basic.enum8):
-    # Decision made by the trust center when a node attempts to join.
+    # The decision made by the Trust Center when a node attempts to join.
 
-    # Allow the node to join. The joining node should have a pre-configured
-    # key. The security data sent to it will be encrypted with that key.
+    # Allow the node to join. The node has the key.
     USE_PRECONFIGURED_KEY = 0x00
-    # Allow the node to join. Send the necessary key (the Network Key in
-    # Standard Security mode, the Trust Center Master in High Security mode)
-    # in-the-clear to the joining device.
+    # Allow the node to join. Send the key to the node.
     SEND_KEY_IN_THE_CLEAR = 0x01
     # Deny join.
     DENY_JOIN = 0x02
     # Take no action.
     NO_ACTION = 0x03
+    # Allow rejoins only.
+    # Note: only used by the SDK host framework, the NCP never reports it in
+    # `trustCenterJoinHandler`.
+    ALLOW_REJOINS_ONLY = 0x04
 
 
 class EmberInitialSecurityBitmask(basic.bitmap16):
@@ -2449,7 +2450,10 @@ class EzspPolicyId(basic.enum8):
     ZLL_POLICY = 0x08
     # Controls whether Trust Center (insecure) rejoins for devices using the well-known
     # link key are accepted. If rejoining using the well-known key is allowed, it is
-    # disabled again after emAllowTcRejoinsUsingWellKnownKeyTimeoutSec seconds.
+    # disabled again after
+    # sli_zigbee_allow_tc_rejoins_using_well_known_key_timeout_sec seconds.
+    # Note: the timeout is set with `CONFIG_TC_REJOINS_USING_WELL_KNOWN_KEY_TIMEOUT_S`.
+    # A timeout of 0 never disables it.
     TC_REJOINS_USING_WELL_KNOWN_KEY_POLICY = 0x09
     # Controls whether the ZigBee RF4CE stack will use standard profile-dependent
     # behavior during the discovery and pairing process. The profiles supported at the
@@ -2470,14 +2474,25 @@ class EzspPolicyId(basic.enum8):
 class EzspDecisionBitmask(basic.bitmap8):
     """EZSP Decision bitmask."""
 
+    # This is the policy decision bitmask that controls the trust center decision
+    # strategies. The bitmask is modified and extracted from the
+    # sl_zigbee_ezsp_decision_id_t for supporting bitmask operations.
+    # Note: used for `TRUST_CENTER_POLICY` from EZSPv8 onwards, replacing the trust
+    # center decisions of `EzspDecisionId`. The NCP boots with `ALLOW_JOINS |
+    # ALLOW_UNSECURED_REJOINS`.
+
     # XXX: this is intentionally an 8-bit bitmap! The SDK source treats it as a 16-bit
     # value but it is always serialized to an 8-bit integer
 
     # Disallow joins and rejoins.
     DEFAULT_CONFIGURATION = 0x0000
     # Send the network key to all joining devices.
+    # Note: EmberZNet 6.7.x (EZSPv8) also accepts unsecured rejoins with only this bit
+    # set.
     ALLOW_JOINS = 0x0001
     # Send the network key to all rejoining devices.
+    # Note: from EmberZNet 6.9.0 (EZSPv8) onwards, unsecured rejoins are denied without
+    # it.
     ALLOW_UNSECURED_REJOINS = 0x0002
     # Send the network key in the clear.
     SEND_KEY_IN_CLEAR = 0x0004
@@ -2492,6 +2507,9 @@ class EzspDecisionBitmask(basic.bitmap8):
 class EzspDecisionId(basic.enum8):
     # Identifies a policy decision.
 
+    # Note: `TRUST_CENTER_POLICY` decisions, EZSPv4 to EZSPv7 only. EZSPv8 and later
+    # use `EzspDecisionBitmask` instead and no longer define these values.
+
     # Send the network key in the clear to all joining and rejoining devices.
     ALLOW_JOINS = 0x00
     # Send the network key in the clear to all joining devices.  Rejoining
@@ -2520,6 +2538,7 @@ class EzspDecisionId(basic.enum8):
     BDB_JOIN_USES_INSTALL_CODE_KEY = 0x06
     # Delay sending the network key to a new joining device.
     DEFER_JOINS_REJOINS_HAVE_LINK_KEY = 0x07
+
     # BINDING_MODIFICATION_POLICY default decision. Do not allow the local
     # binding table to be changed by remote nodes.
     DISALLOW_BINDING_MODIFICATION = 0x10
@@ -2552,8 +2571,11 @@ class EzspDecisionId(basic.enum8):
     # TC_KEY_REQUEST_POLICY decision. When the Trust Center receives a request for a
     # Trust Center link key, it will reply to it with the corresponding key.
     ALLOW_TC_KEY_REQUESTS_AND_SEND_CURRENT_KEY = 0x51
-    # TC_KEY_REQUEST_POLICY decision. When the Trust Center receives a request
-    # for a Trust Center link key, it will generate a key to send to the joiner.
+    # SL_ZIGBEE_EZSP_TC_KEY_REQUEST_POLICY decision. When the Trust Center
+    # receives a request for a Trust Center link key, it will generate a key to
+    # send to the joiner. After generation, the key will be added to the
+    # transient key table and After verification, this key will be added into the
+    # link key table
     ALLOW_TC_KEY_REQUEST_AND_GENERATE_NEW_KEY = 0x52
     # APP_KEY_REQUEST_POLICY decision. When the Trust Center receives a request
     # for an application link key, it will be ignored.
@@ -2590,11 +2612,22 @@ class EzspDecisionId(basic.enum8):
 
 
 class EmberDeviceUpdate(basic.enum8):
-    # The status of the device update.
-
+    # The Status of the Update Device message sent to the Trust Center.
+    # The device may have joined or rejoined insecurely, rejoined securely, or
+    # left.  MAC Security has been deprecated and therefore there is no secure
+    # join.
+    # These map to the actual values within the APS Command frame so they cannot
+    # be arbitrarily changed.
+    # Note: this is the Status field of the APS Update Device command, Zigbee R23.2
+    # 4.4.11.2.4 and Table 4-14. The spec reserves 0x04 to 0x07.
+
+    # Note: spec name "Standard Device Secured Rejoin".
     STANDARD_SECURITY_SECURED_REJOIN = 0x0
+    # Note: spec name "Standard Device Unsecured Join".
     STANDARD_SECURITY_UNSECURED_JOIN = 0x1
     DEVICE_LEFT = 0x2
+    # Note: spec name "Standard Device Trust Center Rejoin", a NWK rejoin without NWK
+    # security.
     STANDARD_SECURITY_UNSECURED_REJOIN = 0x3
 
     # Removed in EZSPv5

From e6fa916108d6002c95d393662d7c7765cfb1a067 Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 17:42:10 +0000
Subject: [PATCH 02/11] Migrate permit APIs into EZSP handlers and a context
 manager

---
 bellows/ezsp/__init__.py      |  32 +++++++-
 bellows/ezsp/protocol.py      |  11 ++-
 bellows/ezsp/v4/__init__.py   |   7 +-
 bellows/ezsp/v5/__init__.py   |   7 +-
 bellows/ezsp/v8/__init__.py   |  27 ++++---
 bellows/zigbee/application.py |  56 +++++++++-----
 tests/test_application.py     | 134 ++++++++++++++++++++++++++++------
 tests/test_ezsp.py            |  63 +++++++++++++++-
 tests/test_ezsp_v10.py        |  16 +---
 tests/test_ezsp_v13.py        |  28 ++++---
 tests/test_ezsp_v4.py         |  21 +++++-
 tests/test_ezsp_v5.py         |  17 ++++-
 tests/test_ezsp_v8.py         |  40 +++++-----
 tests/test_ezsp_v9.py         |  16 +---
 14 files changed, 346 insertions(+), 129 deletions(-)

diff --git a/bellows/ezsp/__init__.py b/bellows/ezsp/__init__.py
index a2113829..550412f9 100644
--- a/bellows/ezsp/__init__.py
+++ b/bellows/ezsp/__init__.py
@@ -5,7 +5,7 @@
 import asyncio
 from asyncio import timeout as asyncio_timeout
 import collections
-from collections.abc import Callable, Generator
+from collections.abc import AsyncIterator, Callable, Generator
 import contextlib
 import dataclasses
 import functools
@@ -18,6 +18,7 @@
 from bellows.ash import NcpFailure
 import bellows.config as conf
 from bellows.exception import (
+    ControllerError,
     EzspError,
     InvalidCommandError,
     InvalidCommandPayload,
@@ -52,6 +53,9 @@
 NETWORK_OPS_TIMEOUT = 10
 NETWORK_COORDINATOR_STARTUP_RESET_WAIT = 2
 
+WILDCARD_EUI64 = t.EUI64.convert("FF:FF:FF:FF:FF:FF:FF:FF")
+WELL_KNOWN_LINK_KEY = t.KeyData(b"ZigBeeAlliance09")
+
 
 class EZSP:
     _BY_VERSION = {
@@ -628,6 +632,32 @@ async def set_source_routing(self, enabled: bool) -> None:
                 )
             )
 
+    @contextlib.asynccontextmanager
+    async def allow_joins(
+        self, *, link_keys: dict[t.EUI64, t.KeyData] | None = None
+    ) -> AsyncIterator[None]:
+        """Allow joins with the well-known key, or only with the given link keys."""
+        install_codes_only = link_keys is not None
+
+        if link_keys is None:
+            link_keys = {WILDCARD_EUI64: WELL_KNOWN_LINK_KEY}
+
+        try:
+            for ieee, key in link_keys.items():
+                status = await self._protocol.add_transient_link_key(ieee, key)
+
+                if status != t.sl_Status.OK:
+                    raise ControllerError(f"Failed to set link key for {ieee}")
+
+            await self._protocol.set_join_policy(
+                allow_joins=True, install_codes_only=install_codes_only
+            )
+
+            yield
+        finally:
+            await self._protocol.clear_transient_link_keys()
+            await self._protocol.set_join_policy(allow_joins=False)
+
     def start_ezsp(self):
         """Mark EZSP as running."""
         self._ezsp_event.set()
diff --git a/bellows/ezsp/protocol.py b/bellows/ezsp/protocol.py
index 841c1b3a..b1b2d5b7 100644
--- a/bellows/ezsp/protocol.py
+++ b/bellows/ezsp/protocol.py
@@ -283,8 +283,11 @@ def __getattr__(self, name: str) -> Callable:
 
         return functools.partial(self.command, name)
 
-    async def pre_permit(self, time_s: int) -> None:
-        """Schedule task before allowing new joins."""
+    @abc.abstractmethod
+    async def set_join_policy(
+        self, *, allow_joins: bool, install_codes_only: bool = False
+    ) -> None:
+        raise NotImplementedError
 
     @abc.abstractmethod
     async def add_transient_link_key(
@@ -292,6 +295,10 @@ async def add_transient_link_key(
     ) -> t.sl_Status:
         raise NotImplementedError
 
+    @abc.abstractmethod
+    async def clear_transient_link_keys(self) -> None:
+        raise NotImplementedError
+
     @abc.abstractmethod
     async def read_child_data(
         self,
diff --git a/bellows/ezsp/v4/__init__.py b/bellows/ezsp/v4/__init__.py
index abbde484..9f94bae6 100644
--- a/bellows/ezsp/v4/__init__.py
+++ b/bellows/ezsp/v4/__init__.py
@@ -36,7 +36,9 @@ def _ezsp_frame_rx(self, data: bytes) -> tuple[int, int, bytes]:
         """Handler for received data frame."""
         return data[0], data[2], data[3:]
 
-    async def pre_permit(self, time_s: int) -> None:
+    async def set_join_policy(
+        self, *, allow_joins: bool, install_codes_only: bool = False
+    ) -> None:
         pass
 
     async def add_transient_link_key(
@@ -44,6 +46,9 @@ async def add_transient_link_key(
     ) -> t.sl_Status:
         return t.sl_Status.OK
 
+    async def clear_transient_link_keys(self) -> None:
+        pass
+
     async def read_child_data(
         self,
     ) -> AsyncGenerator[tuple[t.NWK, t.EUI64, t.EmberNodeType]]:
diff --git a/bellows/ezsp/v5/__init__.py b/bellows/ezsp/v5/__init__.py
index 0a741a12..aeb2bc10 100644
--- a/bellows/ezsp/v5/__init__.py
+++ b/bellows/ezsp/v5/__init__.py
@@ -41,11 +41,8 @@ async def add_transient_link_key(
         (status,) = await self.addTransientLinkKey(partner=ieee, transientKey=key)
         return t.sl_Status.from_ember_status(status)
 
-    async def pre_permit(self, time_s: int) -> None:
-        """Add pre-shared TC Link key."""
-        wild_card_ieee = t.EUI64.convert("FF:FF:FF:FF:FF:FF:FF:FF")
-        tc_link_key = t.KeyData(b"ZigBeeAlliance09")
-        await self.add_transient_link_key(wild_card_ieee, tc_link_key)
+    async def clear_transient_link_keys(self) -> None:
+        await self.clearTransientLinkKeys()
 
     async def read_address_table(self) -> AsyncGenerator[tuple[t.NWK, t.EUI64]]:
         (status, addr_table_size) = await self.getConfigurationValue(
diff --git a/bellows/ezsp/v8/__init__.py b/bellows/ezsp/v8/__init__.py
index b2b55dde..b5b5cd22 100644
--- a/bellows/ezsp/v8/__init__.py
+++ b/bellows/ezsp/v8/__init__.py
@@ -1,5 +1,4 @@
 """"EZSP Protocol version 8 protocol handler."""
-import asyncio
 import logging
 
 import voluptuous
@@ -36,18 +35,22 @@ def _ezsp_frame_rx(self, data: bytes) -> tuple[int, int, bytes]:
 
         return seq, frame_id, data
 
-    async def pre_permit(self, time_s: int) -> None:
-        """Temporarily change TC policy while allowing new joins."""
-        await super().pre_permit(time_s)
-        await self.setPolicy(
-            policyId=t.EzspPolicyId.TRUST_CENTER_POLICY,
-            decisionId=(
+    async def set_join_policy(
+        self, *, allow_joins: bool, install_codes_only: bool = False
+    ) -> None:
+        if not allow_joins:
+            decision = self.tc_policy
+        elif install_codes_only:
+            decision = (
+                t.EzspDecisionBitmask.ALLOW_JOINS
+                | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY
+            )
+        else:
+            decision = (
                 t.EzspDecisionBitmask.ALLOW_JOINS
                 | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS
-            ),
-        )
-        await asyncio.sleep(time_s + 2)
+            )
+
         await self.setPolicy(
-            policyId=t.EzspPolicyId.TRUST_CENTER_POLICY,
-            decisionId=self.tc_policy,
+            policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, decisionId=decision
         )
diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py
index 9fe94ac5..9ece9389 100644
--- a/bellows/zigbee/application.py
+++ b/bellows/zigbee/application.py
@@ -3,6 +3,7 @@
 import asyncio
 from asyncio import timeout as asyncio_timeout
 from collections.abc import AsyncGenerator
+import contextlib
 from datetime import UTC, datetime
 import importlib.metadata
 import logging
@@ -116,6 +117,7 @@ def __init__(self, config: dict) -> None:
         self._ezsp = None
         self._multicast = None
         self._mfg_id_task: asyncio.Task | None = None
+        self._join_window: tuple[asyncio.Event, asyncio.Task] | None = None
         self._pending_requests = {}
         self._watchdog_failures = 0
         self._watchdog_feed_counter = 0
@@ -1335,9 +1337,43 @@ async def send_packet(self, packet: zigpy.types.ZigbeePacket) -> None:
             finally:
                 del self._pending_requests[pending_tag]
 
+    async def _close_join_window(self) -> None:
+        if self._join_window is None:
+            return
+
+        close_event, task = self._join_window
+        self._join_window = None
+
+        # The window task always runs to completion so cleanup is never interrupted
+        close_event.set()
+        await asyncio.wait([task])
+
+    async def _open_join_window(
+        self, time_s: int, link_keys: dict[t.EUI64, t.KeyData] | None = None
+    ) -> None:
+        await self._close_join_window()
+
+        stack = contextlib.AsyncExitStack()
+        await stack.enter_async_context(self._ezsp.allow_joins(link_keys=link_keys))
+        close_event = asyncio.Event()
+
+        async def join_window() -> None:
+            async with stack:
+                with contextlib.suppress(TimeoutError):
+                    await asyncio.wait_for(close_event.wait(), time_s + 2)
+
+        self._join_window = (
+            close_event,
+            self.create_task(join_window(), "join_window"),
+        )
+
     async def permit(self, time_s: int = 60, node: t.EmberNodeId = None) -> None:
         """Permit joining."""
-        self.create_task(self._ezsp.pre_permit(time_s), "pre_permit")
+        if time_s == 0:
+            await self._close_join_window()
+        else:
+            await self._open_join_window(time_s)
+
         await super().permit(time_s, node)
 
     def permit_ncp(self, time_s=60):
@@ -1348,22 +1384,8 @@ async def permit_with_link_key(
         self, node: t.EUI64, link_key: t.KeyData, time_s: int = 60
     ) -> None:
         """Permits a new device to join with the given IEEE and link key."""
-
-        status = await self._ezsp.add_transient_link_key(node, link_key)
-
-        if t.sl_Status.from_ember_status(status) != t.sl_Status.OK:
-            raise ControllerError("Failed to set link key")
-
-        if self._ezsp.ezsp_version >= 8:
-            await self._ezsp.setPolicy(
-                policyId=t.EzspPolicyId.TRUST_CENTER_POLICY,
-                decisionId=(
-                    t.EzspDecisionBitmask.ALLOW_JOINS
-                    | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY
-                ),
-            )
-
-        return await super().permit(time_s)
+        await self._open_join_window(time_s, link_keys={node: link_key})
+        await super().permit(time_s)
 
     async def _subscribe_to_multicast_group(
         self, group_id: zigpy.types.Group, endpoint_id: int
diff --git a/tests/test_application.py b/tests/test_application.py
index 326ef973..1488c83e 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -732,25 +732,117 @@ async def test_permit_ncp(app):
     assert app._ezsp._protocol.permitJoining.mock_calls == [call(60)]
 
 
-async def test_permit_with_link_key(app, ieee):
+POLICY_OPEN = call(
+    policyId=t.EzspPolicyId.TRUST_CENTER_POLICY,
+    decisionId=(
+        t.EzspDecisionBitmask.ALLOW_JOINS
+        | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS
+    ),
+)
+POLICY_INSTALL_CODES_ONLY = call(
+    policyId=t.EzspPolicyId.TRUST_CENTER_POLICY,
+    decisionId=(
+        t.EzspDecisionBitmask.ALLOW_JOINS
+        | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY
+    ),
+)
+POLICY_CLOSED = call(policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, decisionId=0)
+
+
+async def test_permit(app):
+    proto = app._ezsp._protocol
+
+    def check_trust_center_ready(*args, **kwargs):
+        # The Trust Center must be ready before the network opens
+        assert proto.add_transient_link_key.mock_calls == [
+            call(ezsp.WILDCARD_EUI64, ezsp.WELL_KNOWN_LINK_KEY)
+        ]
+        assert proto.setPolicy.mock_calls == [POLICY_OPEN]
+
+    with patch(
+        "zigpy.application.ControllerApplication.permit",
+        side_effect=check_trust_center_ready,
+    ) as permit_mock:
+        await app.permit(10)
+
+    assert permit_mock.mock_calls == [call(10, None)]
+    assert len(proto.clearTransientLinkKeys.mock_calls) == 0
+
     with patch("zigpy.application.ControllerApplication.permit") as permit_mock:
-        await app.permit_with_link_key(
-            ieee,
-            zigpy_t.KeyData.convert("11:22:33:44:55:66:77:88:11:22:33:44:55:66:77:88"),
-            60,
-        )
+        await app.permit(0)
 
-    assert permit_mock.await_count == 1
-    assert app._ezsp._protocol.add_transient_link_key.mock_calls == [
-        call(
-            ieee,
-            zigpy_t.KeyData.convert("11:22:33:44:55:66:77:88:11:22:33:44:55:66:77:88"),
-        )
+    assert permit_mock.mock_calls == [call(0, None)]
+    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED]
+
+
+async def test_permit_window_expires(app):
+    proto = app._ezsp._protocol
+
+    async def expire_immediately(aw, timeout):
+        aw.close()
+        raise TimeoutError
+
+    with (
+        patch("zigpy.application.ControllerApplication.permit"),
+        patch(
+            "bellows.zigbee.application.asyncio.wait_for",
+            side_effect=expire_immediately,
+        ),
+    ):
+        await app.permit(10)
+        _, task = app._join_window
+        await asyncio.wait([task])
+
+    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED]
+
+
+async def test_permit_overlapping(app):
+    proto = app._ezsp._protocol
+
+    with patch("zigpy.application.ControllerApplication.permit"):
+        await app.permit(60)
+        await app.permit(10)
+
+    # The first window is closed before the second one opens
+    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED, POLICY_OPEN]
+
+    with patch("zigpy.application.ControllerApplication.permit"):
+        await app.permit(0)
+
+    assert proto.clearTransientLinkKeys.mock_calls == [call(), call()]
+    assert proto.setPolicy.mock_calls == [
+        POLICY_OPEN,
+        POLICY_CLOSED,
+        POLICY_OPEN,
+        POLICY_CLOSED,
     ]
 
 
+async def test_permit_with_link_key(app, ieee):
+    proto = app._ezsp._protocol
+    key = zigpy_t.KeyData.convert("11:22:33:44:55:66:77:88:11:22:33:44:55:66:77:88")
+
+    with patch("zigpy.application.ControllerApplication.permit") as permit_mock:
+        await app.permit_with_link_key(ieee, key, 60)
+
+    assert permit_mock.mock_calls == [call(60)]
+    assert proto.add_transient_link_key.mock_calls == [call(ieee, key)]
+    assert proto.setPolicy.mock_calls == [POLICY_INSTALL_CODES_ONLY]
+
+    # The policy is restored once the window closes
+    with patch("zigpy.application.ControllerApplication.permit"):
+        await app.permit(0)
+
+    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.setPolicy.mock_calls == [POLICY_INSTALL_CODES_ONLY, POLICY_CLOSED]
+
+
 async def test_permit_with_link_key_failure(app, ieee):
-    app._ezsp._protocol.add_transient_link_key.return_value = t.EmberStatus.ERR_FATAL
+    proto = app._ezsp._protocol
+    proto.add_transient_link_key.return_value = t.sl_Status.FAIL
 
     with patch("zigpy.application.ControllerApplication.permit") as permit_mock:
         with pytest.raises(ControllerError):
@@ -762,8 +854,12 @@ async def test_permit_with_link_key_failure(app, ieee):
                 60,
             )
 
-    assert permit_mock.await_count == 0
-    assert app._ezsp._protocol.add_transient_link_key.await_count == 1
+    assert len(permit_mock.mock_calls) == 0
+    assert proto.add_transient_link_key.await_count == 1
+
+    # Joins were never allowed
+    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.setPolicy.mock_calls == [POLICY_CLOSED]
 
 
 @pytest.fixture
@@ -1963,14 +2059,6 @@ async def test_cleanup_tc_link_key(app):
     assert ezsp.eraseKeyTableEntry.mock_calls == [call(index=sentinel.index)]
 
 
-# @patch("zigpy.application.ControllerApplication.permit", new=AsyncMock())
-async def test_permit(app):
-    """Test permit method."""
-    await app.permit(10)
-    await asyncio.sleep(0.1)
-    assert app._ezsp._protocol.add_transient_link_key.await_count == 1
-
-
 @patch("bellows.zigbee.application.MFG_ID_RESET_DELAY", new=0.01)
 @pytest.mark.parametrize(
     "ieee, expected_mfg_id",
diff --git a/tests/test_ezsp.py b/tests/test_ezsp.py
index bccf17b3..8c082213 100644
--- a/tests/test_ezsp.py
+++ b/tests/test_ezsp.py
@@ -12,12 +12,13 @@
 from bellows import config, uart
 from bellows.ash import NcpFailure
 from bellows.exception import (
+    ControllerError,
     EzspError,
     InvalidCommandError,
     InvalidCommandPayload,
     InvalidTxPower,
 )
-from bellows.ezsp import EZSP, EZSP_LATEST, xncp
+from bellows.ezsp import EZSP, EZSP_LATEST, WELL_KNOWN_LINK_KEY, WILDCARD_EUI64, xncp
 from bellows.ezsp.v9.commands import GetTokenDataRsp
 import bellows.types as t
 
@@ -463,6 +464,66 @@ async def test_set_disable_source_routing(ezsp_f):
     )
 
 
+@pytest.mark.parametrize(
+    "link_keys, expected_keys, install_codes_only",
+    [
+        (None, {WILDCARD_EUI64: WELL_KNOWN_LINK_KEY}, False),
+        (
+            {
+                t.EUI64.convert("00:11:22:33:44:55:66:77"): t.KeyData(
+                    b"0123456789abcdef"
+                )
+            },
+            {
+                t.EUI64.convert("00:11:22:33:44:55:66:77"): t.KeyData(
+                    b"0123456789abcdef"
+                )
+            },
+            True,
+        ),
+    ],
+)
+async def test_allow_joins(ezsp_f, link_keys, expected_keys, install_codes_only):
+    proto = ezsp_f._protocol
+    proto.add_transient_link_key = AsyncMock(return_value=t.sl_Status.OK)
+    proto.set_join_policy = AsyncMock()
+    proto.clear_transient_link_keys = AsyncMock()
+
+    async with ezsp_f.allow_joins(link_keys=link_keys):
+        assert proto.add_transient_link_key.mock_calls == [
+            call(ieee, key) for ieee, key in expected_keys.items()
+        ]
+        assert proto.set_join_policy.mock_calls == [
+            call(allow_joins=True, install_codes_only=install_codes_only)
+        ]
+        assert len(proto.clear_transient_link_keys.mock_calls) == 0
+
+    assert proto.clear_transient_link_keys.mock_calls == [call()]
+    assert proto.set_join_policy.mock_calls == [
+        call(allow_joins=True, install_codes_only=install_codes_only),
+        call(allow_joins=False),
+    ]
+
+
+async def test_allow_joins_link_key_failure(ezsp_f):
+    proto = ezsp_f._protocol
+    proto.add_transient_link_key = AsyncMock(return_value=t.sl_Status.FAIL)
+    proto.set_join_policy = AsyncMock()
+    proto.clear_transient_link_keys = AsyncMock()
+
+    link_keys = {
+        t.EUI64.convert("00:11:22:33:44:55:66:77"): t.KeyData(b"0123456789abcdef")
+    }
+
+    with pytest.raises(ControllerError):
+        async with ezsp_f.allow_joins(link_keys=link_keys):
+            pytest.fail("Joins must not be allowed")  # pragma: no cover
+
+    # Joins are never allowed but everything is still cleaned up
+    assert proto.clear_transient_link_keys.mock_calls == [call()]
+    assert proto.set_join_policy.mock_calls == [call(allow_joins=False)]
+
+
 async def test_leave_network_error(ezsp_f):
     """Test EZSP leaveNetwork command failure."""
 
diff --git a/tests/test_ezsp_v10.py b/tests/test_ezsp_v10.py
index 3ae6063d..2b950b53 100644
--- a/tests/test_ezsp_v10.py
+++ b/tests/test_ezsp_v10.py
@@ -1,4 +1,4 @@
-from unittest.mock import AsyncMock, MagicMock, call, patch
+from unittest.mock import AsyncMock, MagicMock, call
 
 import pytest
 
@@ -33,20 +33,6 @@ def test_ezsp_frame_rx(ezsp_f):
     assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234]
 
 
-async def test_pre_permit(ezsp_f):
-    """Test pre permit."""
-    p1 = patch.object(ezsp_f, "setPolicy", new=AsyncMock())
-    p2 = patch.object(
-        ezsp_f,
-        "addTransientLinkKey",
-        new=AsyncMock(return_value=[t.EmberStatus.SUCCESS]),
-    )
-    with p1 as pre_permit_mock, p2 as tclk_mock:
-        await ezsp_f.pre_permit(-1.9)
-    assert pre_permit_mock.await_count == 2
-    assert tclk_mock.await_count == 1
-
-
 async def test_write_child_data(ezsp_f) -> None:
     ezsp_f.setChildData.return_value = [t.EmberStatus.SUCCESS]
 
diff --git a/tests/test_ezsp_v13.py b/tests/test_ezsp_v13.py
index 3646e4d9..a53dcbc3 100644
--- a/tests/test_ezsp_v13.py
+++ b/tests/test_ezsp_v13.py
@@ -1,4 +1,4 @@
-from unittest.mock import AsyncMock, MagicMock, call, patch
+from unittest.mock import MagicMock, call
 
 import pytest
 import zigpy.exceptions
@@ -33,18 +33,22 @@ def test_ezsp_frame_rx(ezsp_f):
     assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234]
 
 
-async def test_pre_permit(ezsp_f):
-    """Test pre permit."""
-    p1 = patch.object(ezsp_f, "setPolicy", new=AsyncMock())
-    p2 = patch.object(
-        ezsp_f,
-        "importTransientKey",
-        new=AsyncMock(return_value=[t.sl_Status.OK]),
+async def test_add_transient_link_key(ezsp_f) -> None:
+    ezsp_f.importTransientKey.return_value = [t.sl_Status.OK]
+
+    status = await ezsp_f.add_transient_link_key(
+        ieee=t.EUI64.convert("00:11:22:33:44:55:66:77"),
+        key=t.KeyData(b"ZigBeeAlliance09"),
     )
-    with p1 as pre_permit_mock, p2 as tclk_mock:
-        await ezsp_f.pre_permit(-1.9)
-    assert pre_permit_mock.await_count == 2
-    assert tclk_mock.await_count == 1
+
+    assert status == t.sl_Status.OK
+    assert ezsp_f.importTransientKey.mock_calls == [
+        call(
+            eui64=t.EUI64.convert("00:11:22:33:44:55:66:77"),
+            plaintext_key=t.KeyData(b"ZigBeeAlliance09"),
+            flags=t.SecurityManagerContextFlags.NONE,
+        )
+    ]
 
 
 async def test_read_link_keys(ezsp_f):
diff --git a/tests/test_ezsp_v4.py b/tests/test_ezsp_v4.py
index e8f68821..6b855b0b 100644
--- a/tests/test_ezsp_v4.py
+++ b/tests/test_ezsp_v4.py
@@ -34,9 +34,24 @@ def test_ezsp_frame_rx(ezsp_f):
     assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234]
 
 
-async def test_pre_permit(ezsp_f):
-    """Test pre permit."""
-    await ezsp_f.pre_permit(1.9)
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"allow_joins": True},
+        {"allow_joins": True, "install_codes_only": True},
+        {"allow_joins": False},
+    ],
+)
+async def test_set_join_policy(ezsp_f, kwargs) -> None:
+    # It's a no-op
+    await ezsp_f.set_join_policy(**kwargs)
+    assert len(ezsp_f.setPolicy.mock_calls) == 0
+
+
+async def test_clear_transient_link_keys(ezsp_f) -> None:
+    # It's a no-op
+    await ezsp_f.clear_transient_link_keys()
+    assert len(ezsp_f.clearTransientLinkKeys.mock_calls) == 0
 
 
 async def test_read_child_data(ezsp_f):
diff --git a/tests/test_ezsp_v5.py b/tests/test_ezsp_v5.py
index 5f078365..aad1922f 100644
--- a/tests/test_ezsp_v5.py
+++ b/tests/test_ezsp_v5.py
@@ -32,19 +32,28 @@ def test_ezsp_frame_rx(ezsp_f):
     assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234]
 
 
-async def test_pre_permit(ezsp_f):
-    """Test pre permit."""
+async def test_add_transient_link_key(ezsp_f) -> None:
     ezsp_f.addTransientLinkKey.return_value = [t.EmberStatus.SUCCESS]
-    await ezsp_f.pre_permit(1)
 
+    status = await ezsp_f.add_transient_link_key(
+        ieee=t.EUI64.convert("00:11:22:33:44:55:66:77"),
+        key=t.KeyData(b"ZigBeeAlliance09"),
+    )
+
+    assert status == t.sl_Status.OK
     assert ezsp_f.addTransientLinkKey.mock_calls == [
         call(
-            partner=t.EUI64.convert("ff:ff:ff:ff:ff:ff:ff:ff"),
+            partner=t.EUI64.convert("00:11:22:33:44:55:66:77"),
             transientKey=t.KeyData(b"ZigBeeAlliance09"),
         )
     ]
 
 
+async def test_clear_transient_link_keys(ezsp_f) -> None:
+    await ezsp_f.clear_transient_link_keys()
+    assert ezsp_f.clearTransientLinkKeys.mock_calls == [call()]
+
+
 async def test_read_address_table(ezsp_f):
     def get_addr_table_node_id(addressTableIndex):
         return (
diff --git a/tests/test_ezsp_v8.py b/tests/test_ezsp_v8.py
index e3d93de8..5806f7bf 100644
--- a/tests/test_ezsp_v8.py
+++ b/tests/test_ezsp_v8.py
@@ -32,26 +32,30 @@ def test_ezsp_frame_rx(ezsp_f):
     assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234]
 
 
-async def test_pre_permit(ezsp_f):
-    """Test pre permit."""
-    ezsp_f.addTransientLinkKey.return_value = (t.EmberStatus.SUCCESS,)
+@pytest.mark.parametrize(
+    "kwargs, decision",
+    [
+        (
+            {"allow_joins": True},
+            t.EzspDecisionBitmask.ALLOW_JOINS
+            | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS,
+        ),
+        (
+            {"allow_joins": True, "install_codes_only": True},
+            t.EzspDecisionBitmask.ALLOW_JOINS
+            | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY,
+        ),
+        # Closing restores the configured policy
+        ({"allow_joins": False}, t.EzspDecisionBitmask.ALLOW_JOINS),
+    ],
+)
+async def test_set_join_policy(ezsp_f, kwargs, decision) -> None:
+    ezsp_f.tc_policy = t.EzspDecisionBitmask.ALLOW_JOINS
+
+    await ezsp_f.set_join_policy(**kwargs)
 
-    await ezsp_f.pre_permit(-1.9)
     assert ezsp_f.setPolicy.mock_calls == [
-        call(
-            policyId=t.EzspPolicyId.TRUST_CENTER_POLICY,
-            decisionId=(
-                t.EzspDecisionBitmask.ALLOW_JOINS
-                | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS
-            ),
-        ),
-        call(policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, decisionId=0),
-    ]
-    assert ezsp_f.addTransientLinkKey.mock_calls == [
-        call(
-            partner=t.EUI64.convert("ff:ff:ff:ff:ff:ff:ff:ff"),
-            transientKey=t.KeyData(b"ZigBeeAlliance09"),
-        )
+        call(policyId=t.EzspPolicyId.TRUST_CENTER_POLICY, decisionId=decision)
     ]
 
 
diff --git a/tests/test_ezsp_v9.py b/tests/test_ezsp_v9.py
index 9219282a..a50895c7 100644
--- a/tests/test_ezsp_v9.py
+++ b/tests/test_ezsp_v9.py
@@ -1,4 +1,4 @@
-from unittest.mock import AsyncMock, MagicMock, call, patch
+from unittest.mock import MagicMock, call
 
 import pytest
 
@@ -31,20 +31,6 @@ def test_ezsp_frame_rx(ezsp_f):
     assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234]
 
 
-async def test_pre_permit(ezsp_f):
-    """Test pre permit."""
-    p1 = patch.object(ezsp_f, "setPolicy", new=AsyncMock())
-    p2 = patch.object(
-        ezsp_f,
-        "addTransientLinkKey",
-        new=AsyncMock(return_value=[t.EmberStatus.SUCCESS]),
-    )
-    with p1 as pre_permit_mock, p2 as tclk_mock:
-        await ezsp_f.pre_permit(-1.9)
-    assert pre_permit_mock.await_count == 2
-    assert tclk_mock.await_count == 1
-
-
 async def test_write_child_data(ezsp_f) -> None:
     ezsp_f.setChildData.return_value = [t.EmberStatus.SUCCESS]
 

From bec468aae89630638c93af6169a935c5b4a682f5 Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 17:46:15 +0000
Subject: [PATCH 03/11] Enable `ALLOW_UNSECURED_REJOINS` for EZSPv8+

---
 bellows/ezsp/v10/config.py |  2 +-
 bellows/ezsp/v11/config.py |  2 +-
 bellows/ezsp/v12/config.py |  2 +-
 bellows/ezsp/v13/config.py |  2 +-
 bellows/ezsp/v14/config.py |  2 +-
 bellows/ezsp/v16/config.py |  2 +-
 bellows/ezsp/v17/config.py |  2 +-
 bellows/ezsp/v18/config.py |  2 +-
 bellows/ezsp/v19/config.py |  2 +-
 bellows/ezsp/v8/config.py  | 23 +++++++++++++++++++++--
 bellows/ezsp/v9/config.py  |  2 +-
 bellows/types/named.py     |  3 +++
 tests/test_ezsp.py         | 24 ++++++++++++++++++++++++
 13 files changed, 58 insertions(+), 12 deletions(-)

diff --git a/bellows/ezsp/v10/config.py b/bellows/ezsp/v10/config.py
index d77116c0..de47bb8c 100644
--- a/bellows/ezsp/v10/config.py
+++ b/bellows/ezsp/v10/config.py
@@ -3,7 +3,7 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v8.config import EZSP_POLICIES_SHARED
 from ..v9 import config as v9_config
 
 EZSP_SCHEMA = {
diff --git a/bellows/ezsp/v11/config.py b/bellows/ezsp/v11/config.py
index 99526935..3f969d74 100644
--- a/bellows/ezsp/v11/config.py
+++ b/bellows/ezsp/v11/config.py
@@ -3,7 +3,7 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v8.config import EZSP_POLICIES_SHARED
 from ..v10 import config as v10_config
 
 EZSP_SCHEMA = {
diff --git a/bellows/ezsp/v12/config.py b/bellows/ezsp/v12/config.py
index 36688809..992da249 100644
--- a/bellows/ezsp/v12/config.py
+++ b/bellows/ezsp/v12/config.py
@@ -3,7 +3,7 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v8.config import EZSP_POLICIES_SHARED
 from ..v11 import config as v11_config
 
 EZSP_SCHEMA = {
diff --git a/bellows/ezsp/v13/config.py b/bellows/ezsp/v13/config.py
index f71825db..39654556 100644
--- a/bellows/ezsp/v13/config.py
+++ b/bellows/ezsp/v13/config.py
@@ -3,7 +3,7 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v8.config import EZSP_POLICIES_SHARED
 from ..v12 import config as v12_config
 
 EZSP_SCHEMA = {
diff --git a/bellows/ezsp/v14/config.py b/bellows/ezsp/v14/config.py
index c56d354c..8d16a3c3 100644
--- a/bellows/ezsp/v14/config.py
+++ b/bellows/ezsp/v14/config.py
@@ -3,7 +3,7 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v8.config import EZSP_POLICIES_SHARED
 from ..v13 import config as v13_config
 
 EZSP_SCHEMA = {
diff --git a/bellows/ezsp/v16/config.py b/bellows/ezsp/v16/config.py
index d502f1ec..77c9c35a 100644
--- a/bellows/ezsp/v16/config.py
+++ b/bellows/ezsp/v16/config.py
@@ -3,7 +3,7 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v8.config import EZSP_POLICIES_SHARED
 from ..v14 import config as v14_config
 
 EZSP_SCHEMA = {
diff --git a/bellows/ezsp/v17/config.py b/bellows/ezsp/v17/config.py
index 45bf59e5..64891d97 100644
--- a/bellows/ezsp/v17/config.py
+++ b/bellows/ezsp/v17/config.py
@@ -3,7 +3,7 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v8.config import EZSP_POLICIES_SHARED
 from ..v16 import config as v16_config
 
 EZSP_SCHEMA = {
diff --git a/bellows/ezsp/v18/config.py b/bellows/ezsp/v18/config.py
index 167fa57a..c4873e00 100644
--- a/bellows/ezsp/v18/config.py
+++ b/bellows/ezsp/v18/config.py
@@ -3,7 +3,7 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v8.config import EZSP_POLICIES_SHARED
 from ..v17 import config as v17_config
 
 EZSP_SCHEMA = {
diff --git a/bellows/ezsp/v19/config.py b/bellows/ezsp/v19/config.py
index 87252f27..f16469ab 100644
--- a/bellows/ezsp/v19/config.py
+++ b/bellows/ezsp/v19/config.py
@@ -3,7 +3,7 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v8.config import EZSP_POLICIES_SHARED
 from ..v18 import config as v18_config
 
 EZSP_SCHEMA = {
diff --git a/bellows/ezsp/v8/config.py b/bellows/ezsp/v8/config.py
index 7ee0adf0..da38778b 100644
--- a/bellows/ezsp/v8/config.py
+++ b/bellows/ezsp/v8/config.py
@@ -1,11 +1,30 @@
 import voluptuous as vol
 
 from bellows.config import cv_optional_int, cv_uint16, extend_vol_schema
-from bellows.types import EzspConfigId, EzspPolicyId
+from bellows.types import EzspConfigId, EzspDecisionBitmask, EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
+from ..v4 import config as v4_config
 from ..v7 import config as v7_config
 
+# EZSPv8 replaces the trust center decisions with a decision bitmask
+EZSP_POLICIES_SHARED = {
+    **{
+        key: value
+        for key, value in v4_config.EZSP_POLICIES_SHARED.items()
+        if key != EzspPolicyId.TRUST_CENTER_POLICY.name
+    },
+    # Same as the NCP's boot default and the SDK host framework. Trust Center rejoins
+    # are always accepted. New joins still need permit joining and the well-known
+    # transient key, which only exists while joins are allowed.
+    vol.Optional(
+        EzspPolicyId.TRUST_CENTER_POLICY.name,
+        default=(
+            EzspDecisionBitmask.ALLOW_JOINS
+            | EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS
+        ),
+    ): cv_uint16,
+}
+
 EZSP_SCHEMA = extend_vol_schema(
     v7_config.EZSP_SCHEMA,
     {
diff --git a/bellows/ezsp/v9/config.py b/bellows/ezsp/v9/config.py
index 9c678ff5..95b5a001 100644
--- a/bellows/ezsp/v9/config.py
+++ b/bellows/ezsp/v9/config.py
@@ -3,8 +3,8 @@
 from bellows.config import cv_uint16
 from bellows.types import EzspPolicyId
 
-from ..v4.config import EZSP_POLICIES_SHARED
 from ..v8 import config as v8_config
+from ..v8.config import EZSP_POLICIES_SHARED
 
 EZSP_SCHEMA = {
     **v8_config.EZSP_SCHEMA,
diff --git a/bellows/types/named.py b/bellows/types/named.py
index e2780902..909361a1 100644
--- a/bellows/types/named.py
+++ b/bellows/types/named.py
@@ -2493,6 +2493,9 @@ class EzspDecisionBitmask(basic.bitmap8):
     # Send the network key to all rejoining devices.
     # Note: from EmberZNet 6.9.0 (EZSPv8) onwards, unsecured rejoins are denied without
     # it.
+    # Note: these are Trust Center rejoins (spec name). The network key is APS
+    # encrypted with the device's link key, devices using the well-known key are
+    # rejected unless `TC_REJOINS_USING_WELL_KNOWN_KEY_POLICY` is enabled.
     ALLOW_UNSECURED_REJOINS = 0x0002
     # Send the network key in the clear.
     SEND_KEY_IN_CLEAR = 0x0004
diff --git a/tests/test_ezsp.py b/tests/test_ezsp.py
index 8c082213..a3ae636d 100644
--- a/tests/test_ezsp.py
+++ b/tests/test_ezsp.py
@@ -524,6 +524,30 @@ async def test_allow_joins_link_key_failure(ezsp_f):
     assert proto.set_join_policy.mock_calls == [call(allow_joins=False)]
 
 
+@pytest.mark.parametrize(
+    "version, default",
+    [(v, t.EzspDecisionId.ALLOW_PRECONFIGURED_KEY_JOINS) for v in (4, 5, 6, 7)]
+    + [
+        (
+            v,
+            t.EzspDecisionBitmask.ALLOW_JOINS
+            | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS,
+        )
+        for v in EZSP._BY_VERSION
+        if v >= 8
+    ],
+)
+def test_trust_center_policy_default(version, default):
+    schema = EZSP._BY_VERSION[version].SCHEMAS[config.CONF_EZSP_POLICIES]
+
+    policies = schema({})
+    assert policies[t.EzspPolicyId.TRUST_CENTER_POLICY.name] == default
+
+    # The user's configuration takes precedence
+    policies = schema({t.EzspPolicyId.TRUST_CENTER_POLICY.name: 0x02})
+    assert policies[t.EzspPolicyId.TRUST_CENTER_POLICY.name] == 0x02
+
+
 async def test_leave_network_error(ezsp_f):
     """Test EZSP leaveNetwork command failure."""
 

From eda1160af5612168d0a7183f97338a571e109b53 Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 18:20:00 +0000
Subject: [PATCH 04/11] Only call `handle_join` for secured rejoins, leave
 `Device_annce` to zigpy

---
 bellows/zigbee/application.py |  7 +++-
 tests/test_application.py     | 75 ++++++++++++++++++++---------------
 2 files changed, 48 insertions(+), 34 deletions(-)

diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py
index 9ece9389..fc9346dc 100644
--- a/bellows/zigbee/application.py
+++ b/bellows/zigbee/application.py
@@ -1007,7 +1007,12 @@ def _handle_tc_join_handler(
             if self._mfg_id_task and not self._mfg_id_task.done():
                 self._mfg_id_task.cancel()
             self._mfg_id_task = asyncio.create_task(self._reset_mfg_id(mfg_id))
-        self.handle_join(nwk, ieee, parent_nwk)
+
+        # Unsecured joins and rejoins are unauthenticated: anyone can send one for any
+        # EUI64 and the device may never receive the network key. The device's own
+        # network-secured Device_annce reports the join once it has the key.
+        if device_update_status == t.EmberDeviceUpdate.STANDARD_SECURITY_SECURED_REJOIN:
+            self.handle_join(nwk, ieee, parent_nwk)
 
     async def _reset_mfg_id(self, mfg_id: int) -> None:
         """Resets manufacturer id if was temporary overridden by a joining device."""
diff --git a/tests/test_application.py b/tests/test_application.py
index 1488c83e..8ea514b1 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -661,46 +661,55 @@ def test_unexpected_send_success(app, aps, ieee):
     )
 
 
-async def test_join_handler(app, ieee):
+@pytest.mark.parametrize(
+    "status, decision, joins, cleans_up_tclk",
+    [
+        # Only a secured rejoin proves the device holds the network key
+        (
+            t.EmberDeviceUpdate.STANDARD_SECURITY_SECURED_REJOIN,
+            t.EmberJoinDecision.NO_ACTION,
+            True,
+            False,
+        ),
+        # Unsecured joins and rejoins are reported by the device's Device_annce
+        (
+            t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN,
+            t.EmberJoinDecision.USE_PRECONFIGURED_KEY,
+            False,
+            True,
+        ),
+        (
+            t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_REJOIN,
+            t.EmberJoinDecision.USE_PRECONFIGURED_KEY,
+            False,
+            False,
+        ),
+        (
+            t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN,
+            t.EmberJoinDecision.DENY_JOIN,
+            False,
+            True,
+        ),
+        (
+            t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_REJOIN,
+            t.EmberJoinDecision.DENY_JOIN,
+            False,
+            False,
+        ),
+    ],
+)
+async def test_join_handler(app, ieee, status, decision, joins, cleans_up_tclk):
     # Calls device.initialize, leaks a task
     app.handle_join = MagicMock()
     app.cleanup_tc_link_key = AsyncMock()
     app.ezsp_callback_handler(
-        "trustCenterJoinHandler",
-        [
-            1,
-            ieee,
-            t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN,
-            t.EmberJoinDecision.NO_ACTION,
-            sentinel.parent,
-        ],
+        "trustCenterJoinHandler", [1, ieee, status, decision, sentinel.parent]
     )
     await asyncio.sleep(0)
+
     assert ieee not in app.devices
-    assert app.handle_join.call_count == 1
-    assert app.handle_join.call_args[0][0] == 1
-    assert app.handle_join.call_args[0][1] == ieee
-    assert app.handle_join.call_args[0][2] is sentinel.parent
-    assert app.cleanup_tc_link_key.await_count == 1
-    assert app.cleanup_tc_link_key.call_args[0][0] is ieee
-
-    # cleanup TCLK, but no join handling
-    app.handle_join.reset_mock()
-    app.cleanup_tc_link_key.reset_mock()
-    app.ezsp_callback_handler(
-        "trustCenterJoinHandler",
-        [
-            1,
-            ieee,
-            t.EmberDeviceUpdate.STANDARD_SECURITY_UNSECURED_JOIN,
-            t.EmberJoinDecision.DENY_JOIN,
-            sentinel.parent,
-        ],
-    )
-    await asyncio.sleep(0)
-    assert app.cleanup_tc_link_key.await_count == 1
-    assert app.cleanup_tc_link_key.call_args[0][0] is ieee
-    assert app.handle_join.call_count == 0
+    assert app.handle_join.mock_calls == [call(1, ieee, sentinel.parent)] * joins
+    assert app.cleanup_tc_link_key.mock_calls == [call(ieee)] * cleans_up_tclk
 
 
 def test_leave_handler(app, ieee):

From 2e85a8f9c7182143dd324a9b19dfe9358047de98 Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 21:19:16 +0000
Subject: [PATCH 05/11] `ALLOW_UNSECURED_REJOINS` during the IC join period as
 well

---
 bellows/ezsp/v8/__init__.py | 1 +
 tests/test_application.py   | 1 +
 2 files changed, 2 insertions(+)

diff --git a/bellows/ezsp/v8/__init__.py b/bellows/ezsp/v8/__init__.py
index b5b5cd22..73059743 100644
--- a/bellows/ezsp/v8/__init__.py
+++ b/bellows/ezsp/v8/__init__.py
@@ -43,6 +43,7 @@ async def set_join_policy(
         elif install_codes_only:
             decision = (
                 t.EzspDecisionBitmask.ALLOW_JOINS
+                | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS
                 | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY
             )
         else:
diff --git a/tests/test_application.py b/tests/test_application.py
index 8ea514b1..a0bdd2d7 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -752,6 +752,7 @@ async def test_permit_ncp(app):
     policyId=t.EzspPolicyId.TRUST_CENTER_POLICY,
     decisionId=(
         t.EzspDecisionBitmask.ALLOW_JOINS
+        | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS
         | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY
     ),
 )

From 3ac9a062771f7ce1b16d86107faa9461cd37676b Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 21:20:31 +0000
Subject: [PATCH 06/11] Build policies on user config

---
 bellows/ezsp/v8/__init__.py |  6 ++++--
 tests/test_ezsp_v8.py       | 17 +++++++++++++----
 2 files changed, 17 insertions(+), 6 deletions(-)

diff --git a/bellows/ezsp/v8/__init__.py b/bellows/ezsp/v8/__init__.py
index 73059743..ea7412e8 100644
--- a/bellows/ezsp/v8/__init__.py
+++ b/bellows/ezsp/v8/__init__.py
@@ -42,13 +42,15 @@ async def set_join_policy(
             decision = self.tc_policy
         elif install_codes_only:
             decision = (
-                t.EzspDecisionBitmask.ALLOW_JOINS
+                self.tc_policy
+                | t.EzspDecisionBitmask.ALLOW_JOINS
                 | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS
                 | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY
             )
         else:
             decision = (
-                t.EzspDecisionBitmask.ALLOW_JOINS
+                self.tc_policy
+                | t.EzspDecisionBitmask.ALLOW_JOINS
                 | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS
             )
 
diff --git a/tests/test_ezsp_v8.py b/tests/test_ezsp_v8.py
index 5806f7bf..d815ef83 100644
--- a/tests/test_ezsp_v8.py
+++ b/tests/test_ezsp_v8.py
@@ -37,20 +37,29 @@ def test_ezsp_frame_rx(ezsp_f):
     [
         (
             {"allow_joins": True},
-            t.EzspDecisionBitmask.ALLOW_JOINS
+            t.EzspDecisionBitmask.DEFER_JOINS
+            | t.EzspDecisionBitmask.ALLOW_JOINS
             | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS,
         ),
         (
             {"allow_joins": True, "install_codes_only": True},
-            t.EzspDecisionBitmask.ALLOW_JOINS
+            t.EzspDecisionBitmask.DEFER_JOINS
+            | t.EzspDecisionBitmask.ALLOW_JOINS
+            | t.EzspDecisionBitmask.ALLOW_UNSECURED_REJOINS
             | t.EzspDecisionBitmask.JOINS_USE_INSTALL_CODE_KEY,
         ),
         # Closing restores the configured policy
-        ({"allow_joins": False}, t.EzspDecisionBitmask.ALLOW_JOINS),
+        ({"allow_joins": False}, t.EzspDecisionBitmask.DEFER_JOINS),
     ],
 )
 async def test_set_join_policy(ezsp_f, kwargs, decision) -> None:
-    ezsp_f.tc_policy = t.EzspDecisionBitmask.ALLOW_JOINS
+    ezsp_f.setPolicy.return_value = [t.EmberStatus.SUCCESS]
+
+    # Join windows build on top of the user's policy
+    await ezsp_f.update_policies(
+        {t.EzspPolicyId.TRUST_CENTER_POLICY.name: t.EzspDecisionBitmask.DEFER_JOINS}
+    )
+    ezsp_f.setPolicy.reset_mock()
 
     await ezsp_f.set_join_policy(**kwargs)
 

From 0a5be5cf38414ffa0807083b81d42fb4a8d29415 Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 21:21:01 +0000
Subject: [PATCH 07/11] Clear transient keys before adding new ones

---
 bellows/ezsp/__init__.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/bellows/ezsp/__init__.py b/bellows/ezsp/__init__.py
index 550412f9..fd756e72 100644
--- a/bellows/ezsp/__init__.py
+++ b/bellows/ezsp/__init__.py
@@ -643,6 +643,9 @@ async def allow_joins(
             link_keys = {WILDCARD_EUI64: WELL_KNOWN_LINK_KEY}
 
         try:
+            # Keys from a previous window may remain if its cleanup failed
+            await self._protocol.clear_transient_link_keys()
+
             for ieee, key in link_keys.items():
                 status = await self._protocol.add_transient_link_key(ieee, key)
 

From 2be4b5513cc1516f265bc7a9ea7dd2b183a45bf5 Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 21:21:20 +0000
Subject: [PATCH 08/11] Defensively run `set_join_policy(allow_joins=False)`

---
 bellows/ezsp/__init__.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/bellows/ezsp/__init__.py b/bellows/ezsp/__init__.py
index fd756e72..6992ea5b 100644
--- a/bellows/ezsp/__init__.py
+++ b/bellows/ezsp/__init__.py
@@ -658,8 +658,10 @@ async def allow_joins(
 
             yield
         finally:
-            await self._protocol.clear_transient_link_keys()
-            await self._protocol.set_join_policy(allow_joins=False)
+            try:
+                await self._protocol.clear_transient_link_keys()
+            finally:
+                await self._protocol.set_join_policy(allow_joins=False)
 
     def start_ezsp(self):
         """Mark EZSP as running."""

From eef6d5c2cace1a04a6bd4ce5e1539861bd37943e Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 21:21:32 +0000
Subject: [PATCH 09/11] Fix tests

---
 tests/test_application.py | 14 +++++++-------
 tests/test_ezsp.py        | 25 ++++++++++++++++++++++---
 2 files changed, 29 insertions(+), 10 deletions(-)

diff --git a/tests/test_application.py b/tests/test_application.py
index a0bdd2d7..09f6f657 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -776,13 +776,13 @@ def check_trust_center_ready(*args, **kwargs):
         await app.permit(10)
 
     assert permit_mock.mock_calls == [call(10, None)]
-    assert len(proto.clearTransientLinkKeys.mock_calls) == 0
+    assert proto.clearTransientLinkKeys.mock_calls == [call()]
 
     with patch("zigpy.application.ControllerApplication.permit") as permit_mock:
         await app.permit(0)
 
     assert permit_mock.mock_calls == [call(0, None)]
-    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.clearTransientLinkKeys.mock_calls == [call(), call()]
     assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED]
 
 
@@ -804,7 +804,7 @@ async def expire_immediately(aw, timeout):
         _, task = app._join_window
         await asyncio.wait([task])
 
-    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.clearTransientLinkKeys.mock_calls == [call(), call()]
     assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED]
 
 
@@ -816,13 +816,13 @@ async def test_permit_overlapping(app):
         await app.permit(10)
 
     # The first window is closed before the second one opens
-    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.clearTransientLinkKeys.mock_calls == [call()] * 3
     assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED, POLICY_OPEN]
 
     with patch("zigpy.application.ControllerApplication.permit"):
         await app.permit(0)
 
-    assert proto.clearTransientLinkKeys.mock_calls == [call(), call()]
+    assert proto.clearTransientLinkKeys.mock_calls == [call()] * 4
     assert proto.setPolicy.mock_calls == [
         POLICY_OPEN,
         POLICY_CLOSED,
@@ -846,7 +846,7 @@ async def test_permit_with_link_key(app, ieee):
     with patch("zigpy.application.ControllerApplication.permit"):
         await app.permit(0)
 
-    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.clearTransientLinkKeys.mock_calls == [call(), call()]
     assert proto.setPolicy.mock_calls == [POLICY_INSTALL_CODES_ONLY, POLICY_CLOSED]
 
 
@@ -868,7 +868,7 @@ async def test_permit_with_link_key_failure(app, ieee):
     assert proto.add_transient_link_key.await_count == 1
 
     # Joins were never allowed
-    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.clearTransientLinkKeys.mock_calls == [call(), call()]
     assert proto.setPolicy.mock_calls == [POLICY_CLOSED]
 
 
diff --git a/tests/test_ezsp.py b/tests/test_ezsp.py
index a3ae636d..77a017af 100644
--- a/tests/test_ezsp.py
+++ b/tests/test_ezsp.py
@@ -496,15 +496,34 @@ async def test_allow_joins(ezsp_f, link_keys, expected_keys, install_codes_only)
         assert proto.set_join_policy.mock_calls == [
             call(allow_joins=True, install_codes_only=install_codes_only)
         ]
-        assert len(proto.clear_transient_link_keys.mock_calls) == 0
 
-    assert proto.clear_transient_link_keys.mock_calls == [call()]
+        # Leftover keys from a previous window are cleared before the new ones
+        assert proto.clear_transient_link_keys.mock_calls == [call()]
+
+    assert proto.clear_transient_link_keys.mock_calls == [call(), call()]
     assert proto.set_join_policy.mock_calls == [
         call(allow_joins=True, install_codes_only=install_codes_only),
         call(allow_joins=False),
     ]
 
 
+async def test_allow_joins_clear_keys_failure(ezsp_f):
+    proto = ezsp_f._protocol
+    proto.add_transient_link_key = AsyncMock(return_value=t.sl_Status.OK)
+    proto.set_join_policy = AsyncMock()
+    proto.clear_transient_link_keys = AsyncMock(side_effect=[None, TimeoutError()])
+
+    with pytest.raises(TimeoutError):
+        async with ezsp_f.allow_joins():
+            pass
+
+    # The policy is restored even though the keys could not be cleared
+    assert proto.set_join_policy.mock_calls == [
+        call(allow_joins=True, install_codes_only=False),
+        call(allow_joins=False),
+    ]
+
+
 async def test_allow_joins_link_key_failure(ezsp_f):
     proto = ezsp_f._protocol
     proto.add_transient_link_key = AsyncMock(return_value=t.sl_Status.FAIL)
@@ -520,7 +539,7 @@ async def test_allow_joins_link_key_failure(ezsp_f):
             pytest.fail("Joins must not be allowed")  # pragma: no cover
 
     # Joins are never allowed but everything is still cleaned up
-    assert proto.clear_transient_link_keys.mock_calls == [call()]
+    assert proto.clear_transient_link_keys.mock_calls == [call(), call()]
     assert proto.set_join_policy.mock_calls == [call(allow_joins=False)]
 
 

From 8d971bda934c3ccca6d47cb92f795d308e501263 Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 21:28:53 +0000
Subject: [PATCH 10/11] Extend the join window instead of closing+opening when
 overlapping

---
 bellows/zigbee/application.py | 64 +++++++++++++++++++-------
 tests/test_application.py     | 85 ++++++++++++++++++++++++++++-------
 2 files changed, 116 insertions(+), 33 deletions(-)

diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py
index fc9346dc..b58ff92d 100644
--- a/bellows/zigbee/application.py
+++ b/bellows/zigbee/application.py
@@ -4,6 +4,7 @@
 from asyncio import timeout as asyncio_timeout
 from collections.abc import AsyncGenerator
 import contextlib
+import dataclasses
 from datetime import UTC, datetime
 import importlib.metadata
 import logging
@@ -72,6 +73,8 @@
 RESET_ATTEMPT_BACKOFF_TIME = 5
 NETWORK_UP_TIMEOUT_S = 10
 MAX_WATCHDOG_FAILURES = 4
+# The Trust Center accepts joins slightly longer than the network is open
+JOIN_WINDOW_GRACE_S = 2
 IEEE_PREFIX_MFG_ID = {
     "04:CF:8C": 0x115F,  # Xiaomi
     "54:EF:44": 0x115F,  # Lumi
@@ -100,6 +103,14 @@
 }
 
 
+@dataclasses.dataclass
+class JoinWindow:
+    link_keys: dict[t.EUI64, t.KeyData] | None
+    close_event: asyncio.Event
+    timer: asyncio.TimerHandle
+    task: asyncio.Task
+
+
 class ControllerApplication(zigpy.application.ControllerApplication):
     SCHEMA = CONFIG_SCHEMA
 
@@ -117,7 +128,8 @@ def __init__(self, config: dict) -> None:
         self._ezsp = None
         self._multicast = None
         self._mfg_id_task: asyncio.Task | None = None
-        self._join_window: tuple[asyncio.Event, asyncio.Task] | None = None
+        self._join_window: JoinWindow | None = None
+        self._join_window_lock = asyncio.Lock()
         self._pending_requests = {}
         self._watchdog_failures = 0
         self._watchdog_feed_counter = 0
@@ -1346,16 +1358,33 @@ async def _close_join_window(self) -> None:
         if self._join_window is None:
             return
 
-        close_event, task = self._join_window
+        window = self._join_window
         self._join_window = None
 
+        window.timer.cancel()
+        window.close_event.set()
+
         # The window task always runs to completion so cleanup is never interrupted
-        close_event.set()
-        await asyncio.wait([task])
+        await asyncio.wait([window.task])
 
     async def _open_join_window(
         self, time_s: int, link_keys: dict[t.EUI64, t.KeyData] | None = None
     ) -> None:
+        loop = asyncio.get_running_loop()
+        window = self._join_window
+
+        # Reopening the window would briefly reject joins, extend it instead
+        if (
+            window is not None
+            and not window.close_event.is_set()
+            and window.link_keys == link_keys
+        ):
+            window.timer.cancel()
+            window.timer = loop.call_later(
+                time_s + JOIN_WINDOW_GRACE_S, window.close_event.set
+            )
+            return
+
         await self._close_join_window()
 
         stack = contextlib.AsyncExitStack()
@@ -1364,22 +1393,24 @@ async def _open_join_window(
 
         async def join_window() -> None:
             async with stack:
-                with contextlib.suppress(TimeoutError):
-                    await asyncio.wait_for(close_event.wait(), time_s + 2)
+                await close_event.wait()
 
-        self._join_window = (
-            close_event,
-            self.create_task(join_window(), "join_window"),
+        self._join_window = JoinWindow(
+            link_keys=link_keys,
+            close_event=close_event,
+            timer=loop.call_later(time_s + JOIN_WINDOW_GRACE_S, close_event.set),
+            task=self.create_task(join_window(), "join_window"),
         )
 
     async def permit(self, time_s: int = 60, node: t.EmberNodeId = None) -> None:
         """Permit joining."""
-        if time_s == 0:
-            await self._close_join_window()
-        else:
-            await self._open_join_window(time_s)
+        async with self._join_window_lock:
+            if time_s == 0:
+                await self._close_join_window()
+            else:
+                await self._open_join_window(time_s)
 
-        await super().permit(time_s, node)
+            await super().permit(time_s, node)
 
     def permit_ncp(self, time_s=60):
         assert 0 <= time_s <= 254
@@ -1389,8 +1420,9 @@ async def permit_with_link_key(
         self, node: t.EUI64, link_key: t.KeyData, time_s: int = 60
     ) -> None:
         """Permits a new device to join with the given IEEE and link key."""
-        await self._open_join_window(time_s, link_keys={node: link_key})
-        await super().permit(time_s)
+        async with self._join_window_lock:
+            await self._open_join_window(time_s, link_keys={node: link_key})
+            await super().permit(time_s)
 
     async def _subscribe_to_multicast_group(
         self, group_id: zigpy.types.Group, endpoint_id: int
diff --git a/tests/test_application.py b/tests/test_application.py
index 09f6f657..17976fe6 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -789,48 +789,99 @@ def check_trust_center_ready(*args, **kwargs):
 async def test_permit_window_expires(app):
     proto = app._ezsp._protocol
 
-    async def expire_immediately(aw, timeout):
-        aw.close()
-        raise TimeoutError
-
     with (
         patch("zigpy.application.ControllerApplication.permit"),
-        patch(
-            "bellows.zigbee.application.asyncio.wait_for",
-            side_effect=expire_immediately,
-        ),
+        # The window expires 10ms after opening
+        patch("bellows.zigbee.application.JOIN_WINDOW_GRACE_S", -9.99),
     ):
         await app.permit(10)
-        _, task = app._join_window
-        await asyncio.wait([task])
+        await asyncio.wait([app._join_window.task])
 
     assert proto.clearTransientLinkKeys.mock_calls == [call(), call()]
     assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED]
 
+    # An expired window is reopened, not extended
+    with patch("zigpy.application.ControllerApplication.permit"):
+        await app.permit(10)
+
+    assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED, POLICY_OPEN]
+
 
-async def test_permit_overlapping(app):
+async def test_permit_extend(app):
     proto = app._ezsp._protocol
 
-    with patch("zigpy.application.ControllerApplication.permit"):
+    with (
+        patch("zigpy.application.ControllerApplication.permit") as permit_mock,
+        patch("bellows.zigbee.application.JOIN_WINDOW_GRACE_S", -9.99),
+    ):
         await app.permit(60)
+        window = app._join_window
         await app.permit(10)
 
-    # The first window is closed before the second one opens
-    assert proto.clearTransientLinkKeys.mock_calls == [call()] * 3
-    assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED, POLICY_OPEN]
+    # The open window is extended, joins are never rejected in between
+    assert app._join_window is window
+    assert permit_mock.mock_calls == [call(60, None), call(10, None)]
+    assert proto.clearTransientLinkKeys.mock_calls == [call()]
+    assert proto.setPolicy.mock_calls == [POLICY_OPEN]
+
+    # The window now closes on the timer of the last call
+    await asyncio.wait([window.task])
+
+    assert proto.clearTransientLinkKeys.mock_calls == [call(), call()]
+    assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED]
+
+
+async def test_permit_switch_link_keys(app, ieee):
+    proto = app._ezsp._protocol
+    key = zigpy_t.KeyData.convert("11:22:33:44:55:66:77:88:11:22:33:44:55:66:77:88")
 
     with patch("zigpy.application.ControllerApplication.permit"):
-        await app.permit(0)
+        await app.permit(60)
+        await app.permit_with_link_key(ieee, key, 60)
+        await app.permit_with_link_key(ieee, key, 60)
 
-    assert proto.clearTransientLinkKeys.mock_calls == [call()] * 4
+    # A window with other keys is replaced, one with the same keys is extended
     assert proto.setPolicy.mock_calls == [
         POLICY_OPEN,
         POLICY_CLOSED,
+        POLICY_INSTALL_CODES_ONLY,
+    ]
+
+    with patch("zigpy.application.ControllerApplication.permit"):
+        await app.permit(60)
+
+    assert proto.setPolicy.mock_calls == [
         POLICY_OPEN,
         POLICY_CLOSED,
+        POLICY_INSTALL_CODES_ONLY,
+        POLICY_CLOSED,
+        POLICY_OPEN,
     ]
 
 
+async def test_permit_concurrent(app):
+    proto = app._ezsp._protocol
+
+    async def add_transient_link_key(ieee, key):
+        # Let the other permit call run while this window is opening
+        await asyncio.sleep(0)
+        return t.sl_Status.OK
+
+    proto.add_transient_link_key.side_effect = add_transient_link_key
+
+    with patch("zigpy.application.ControllerApplication.permit") as permit_mock:
+        await asyncio.gather(app.permit(60), app.permit(10))
+
+    # The second call waits for the first window to open and then extends it
+    assert permit_mock.mock_calls == [call(60, None), call(10, None)]
+    assert proto.setPolicy.mock_calls == [POLICY_OPEN]
+
+    with patch("zigpy.application.ControllerApplication.permit"):
+        await app.permit(0)
+
+    assert proto.setPolicy.mock_calls == [POLICY_OPEN, POLICY_CLOSED]
+
+
 async def test_permit_with_link_key(app, ieee):
     proto = app._ezsp._protocol
     key = zigpy_t.KeyData.convert("11:22:33:44:55:66:77:88:11:22:33:44:55:66:77:88")

From 86bc8d81841cf2d915dba5257e4cdb15351aa466 Mon Sep 17 00:00:00 2001
From: puddly <32534428+puddly@users.noreply.github.com>
Date: Mon, 2 Nov 2026 21:29:03 +0000
Subject: [PATCH 11/11] Remove `JOIN_WINDOW_GRACE_S`

---
 bellows/zigbee/application.py |  8 ++------
 tests/test_application.py     | 17 +++++------------
 2 files changed, 7 insertions(+), 18 deletions(-)

diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py
index b58ff92d..f6996efe 100644
--- a/bellows/zigbee/application.py
+++ b/bellows/zigbee/application.py
@@ -73,8 +73,6 @@
 RESET_ATTEMPT_BACKOFF_TIME = 5
 NETWORK_UP_TIMEOUT_S = 10
 MAX_WATCHDOG_FAILURES = 4
-# The Trust Center accepts joins slightly longer than the network is open
-JOIN_WINDOW_GRACE_S = 2
 IEEE_PREFIX_MFG_ID = {
     "04:CF:8C": 0x115F,  # Xiaomi
     "54:EF:44": 0x115F,  # Lumi
@@ -1380,9 +1378,7 @@ async def _open_join_window(
             and window.link_keys == link_keys
         ):
             window.timer.cancel()
-            window.timer = loop.call_later(
-                time_s + JOIN_WINDOW_GRACE_S, window.close_event.set
-            )
+            window.timer = loop.call_later(time_s, window.close_event.set)
             return
 
         await self._close_join_window()
@@ -1398,7 +1394,7 @@ async def join_window() -> None:
         self._join_window = JoinWindow(
             link_keys=link_keys,
             close_event=close_event,
-            timer=loop.call_later(time_s + JOIN_WINDOW_GRACE_S, close_event.set),
+            timer=loop.call_later(time_s, close_event.set),
             task=self.create_task(join_window(), "join_window"),
         )
 
diff --git a/tests/test_application.py b/tests/test_application.py
index 17976fe6..1a721fce 100644
--- a/tests/test_application.py
+++ b/tests/test_application.py
@@ -789,12 +789,8 @@ def check_trust_center_ready(*args, **kwargs):
 async def test_permit_window_expires(app):
     proto = app._ezsp._protocol
 
-    with (
-        patch("zigpy.application.ControllerApplication.permit"),
-        # The window expires 10ms after opening
-        patch("bellows.zigbee.application.JOIN_WINDOW_GRACE_S", -9.99),
-    ):
-        await app.permit(10)
+    with patch("zigpy.application.ControllerApplication.permit"):
+        await app.permit(0.01)
         await asyncio.wait([app._join_window.task])
 
     assert proto.clearTransientLinkKeys.mock_calls == [call(), call()]
@@ -810,17 +806,14 @@ async def test_permit_window_expires(app):
 async def test_permit_extend(app):
     proto = app._ezsp._protocol
 
-    with (
-        patch("zigpy.application.ControllerApplication.permit") as permit_mock,
-        patch("bellows.zigbee.application.JOIN_WINDOW_GRACE_S", -9.99),
-    ):
+    with patch("zigpy.application.ControllerApplication.permit") as permit_mock:
         await app.permit(60)
         window = app._join_window
-        await app.permit(10)
+        await app.permit(0.01)
 
     # The open window is extended, joins are never rejected in between
     assert app._join_window is window
-    assert permit_mock.mock_calls == [call(60, None), call(10, None)]
+    assert permit_mock.mock_calls == [call(60, None), call(0.01, None)]
     assert proto.clearTransientLinkKeys.mock_calls == [call()]
     assert proto.setPolicy.mock_calls == [POLICY_OPEN]