Explicitly disable joins on startup to mitigate CC2531 + Z-Stack 1.2 bug

Fixes home-assistant/core#43245

tests/application/test_joining.py

@@ -18,6 +18,18 @@
 async def test_permit_join(device, make_application):
     app, znp_server = make_application(server_cls=device)
 
+    # Handle the startup permit join clear
+    znp_server.reply_once_to(
+        request=c.ZDO.MgmtPermitJoinReq.Req(
+            AddrMode=t.AddrMode.NWK, Dst=0x0000, Duration=0, partial=True
+        ),
+        responses=[
+            c.ZDO.MgmtPermitJoinReq.Rsp(Status=t.Status.SUCCESS),
+            c.ZDO.MgmtPermitJoinRsp.Callback(Src=0x0000, Status=t.ZDOStatus.SUCCESS),
+        ],
+        override=True,
+    )
+
     # Handle the permit join request sent by us
     permit_join_unicast = znp_server.reply_once_to(
         request=c.ZDO.MgmtPermitJoinReq.Req(
@@ -53,6 +65,18 @@ async def test_permit_join(device, make_application):
 async def test_permit_join_failure(device, make_application):
     app, znp_server = make_application(server_cls=device)
 
+    # Handle the startup permit join clear
+    znp_server.reply_once_to(
+        request=c.ZDO.MgmtPermitJoinReq.Req(
+            AddrMode=t.AddrMode.NWK, Dst=0x0000, Duration=0, partial=True
+        ),
+        responses=[
+            c.ZDO.MgmtPermitJoinReq.Rsp(Status=t.Status.SUCCESS),
+            c.ZDO.MgmtPermitJoinRsp.Callback(Src=0x0000, Status=t.ZDOStatus.SUCCESS),
+        ],
+        override=True,
+    )
+
     # Handle the permit join request sent by us
     permit_join_unicast = znp_server.reply_once_to(
         request=c.ZDO.MgmtPermitJoinReq.Req(
@@ -111,6 +135,18 @@ async def test_permit_join_with_key(device, status, make_application):
         ],
     )
 
+    # Handle the startup permit join clear
+    znp_server.reply_once_to(
+        request=c.ZDO.MgmtPermitJoinReq.Req(
+            AddrMode=t.AddrMode.NWK, Dst=0x0000, Duration=0, partial=True
+        ),
+        responses=[
+            c.ZDO.MgmtPermitJoinReq.Rsp(Status=t.Status.SUCCESS),
+            c.ZDO.MgmtPermitJoinRsp.Callback(Src=0x0000, Status=t.ZDOStatus.SUCCESS),
+        ],
+        override=True,
+    )
+
     # Handle the permit join request sent by us (or fail)
     permit_join_unicast = znp_server.reply_once_to(
         request=c.ZDO.MgmtPermitJoinReq.Req(
@@ -182,6 +218,18 @@ async def test_new_device_join_and_bind_complex(device, make_application, mocker
     nwk = 0x6A7C
     ieee = t.EUI64.convert("00:17:88:01:08:64:6C:81")
 
+    # Handle the startup permit join clear
+    znp_server.reply_once_to(
+        request=c.ZDO.MgmtPermitJoinReq.Req(
+            AddrMode=t.AddrMode.NWK, Dst=0x0000, Duration=0, partial=True
+        ),
+        responses=[
+            c.ZDO.MgmtPermitJoinReq.Rsp(Status=t.Status.SUCCESS),
+            c.ZDO.MgmtPermitJoinRsp.Callback(Src=0x0000, Status=t.ZDOStatus.SUCCESS),
+        ],
+        override=True,
+    )
+
     # Handle the permit join request sent by us (or fail)
     znp_server.reply_once_to(
         request=c.ZDO.MgmtPermitJoinReq.Req(

tests/conftest.py

@@ -216,7 +216,10 @@ def _flatten_responses(self, request, responses):
             for response in responses:
                 yield from self._flatten_responses(request, response)
 
-    def reply_once_to(self, request, responses):
+    def reply_once_to(self, request, responses, *, override=False):
+        if override:
+            self._listeners[request.header].clear()
+
         future = self.wait_for_response(request)
         called_future = asyncio.get_running_loop().create_future()
 
@@ -612,6 +615,23 @@ def node_desc_responder(self, req):
             ),
         ]
 
+    @reply_to(
+        c.ZDO.MgmtPermitJoinReq.Req(
+            AddrMode=t.AddrMode.NWK, Dst=0x0000, TCSignificance=1, partial=True
+        )
+    )
+    def permit_join(self, request):
+        if request.Duration != 0:
+            rsp = [c.ZDO.PermitJoinInd.Callback(Duration=request.Duration)]
+        else:
+            rsp = []
+
+        return rsp + [
+            c.ZDO.MgmtPermitJoinReq.Rsp(Status=t.Status.SUCCESS),
+            c.ZDO.MgmtPermitJoinRsp.Callback(Src=0x0000, Status=t.ZDOStatus.SUCCESS),
+            c.ZDO.PermitJoinInd.Callback(Duration=0),
+        ]
+
 
 class BaseZStack3Device(BaseZStackDevice):
     def __init__(self, *args, **kwargs):

zigpy_znp/zigbee/application.py

@@ -311,6 +311,11 @@ async def _startup(self, auto_form=False):
         async with async_timeout.timeout(STARTUP_TIMEOUT):
             await started_as_coordinator
 
+        # XXX: The CC2531 running Z-Stack Home 1.2 permanently permits joins on startup
+        # unless they are explicitly disabled. We can't fix this but we can disable them
+        # as early as possible to shrink the window of opportunity for unwanted joins.
+        await self.permit_ncp(time_s=0)
+
         # The CC2531 running Z-Stack Home 1.2 overrides the LED setting if it is changed
         # before the coordinator has started.
         if self.znp_config[conf.CONF_LED_MODE] is not None: