Completely deserialize zigpy's ZDO request when proxying

Author: puddly

tests/test_application.py

@@ -10,7 +10,7 @@
 import zigpy_znp.config as conf
 
 import zigpy.device
-from zigpy.zdo.types import ZDOCmd
+from zigpy.zdo.types import ZDOCmd, SizePrefixedSimpleDescriptor
 
 from zigpy_znp.uart import ZnpMtProtocol
 
@@ -511,32 +511,41 @@ async def test_zdo_request_interception(application, mocker):
     app, znp_server = application
     await app.startup(auto_form=False)
 
-    device = app.add_device(ieee=t.EUI64(range(8)), nwk=0x0011)
+    device = app.add_device(ieee=t.EUI64(range(8)), nwk=0xFA9E)
 
     # Send back a request response
     active_ep_req = znp_server.reply_once_to(
-        request=c.ZDOCommands.ActiveEpReq.Req(
-            DstAddr=device.nwk, NWKAddrOfInterest=device.nwk
+        request=c.ZDOCommands.SimpleDescReq.Req(
+            DstAddr=device.nwk, NWKAddrOfInterest=device.nwk, Endpoint=1
         ),
         responses=[
-            c.ZDOCommands.ActiveEpReq.Rsp(Status=t.Status.Success),
-            c.ZDOCommands.ActiveEpRsp.Callback(
+            c.ZDOCommands.SimpleDescReq.Rsp(Status=t.Status.Success),
+            c.ZDOCommands.SimpleDescRsp.Callback(
                 Src=device.nwk,
                 Status=t.ZDOStatus.SUCCESS,
-                ActiveEndpoints=[0, 1, 2],
                 NWK=device.nwk,
+                SimpleDescriptor=SizePrefixedSimpleDescriptor(
+                    *dict(
+                        endpoint=1,
+                        profile=49246,
+                        device_type=256,
+                        device_version=2,
+                        input_clusters=[0, 3, 4, 5, 6, 8, 2821, 4096],
+                        output_clusters=[5, 25, 32, 4096],
+                    ).values()
+                ),
             ),
         ],
     )
 
     status, message = await app.request(
         device=device,
         profile=260,
-        cluster=ZDOCmd.Active_EP_req,
+        cluster=ZDOCmd.Simple_Desc_req,
         src_ep=0,
         dst_ep=0,
-        sequence=0,
-        data=b"test",
+        sequence=1,
+        data=b"\x01\x9e\xfa\x01",
         use_ieee=False,
     )
 

zigpy_znp/zigbee/application.py

@@ -12,9 +12,9 @@
 import zigpy.profiles
 import zigpy.zcl.foundation
 
-from zigpy.zdo.types import ZDOCmd
+from zigpy.zdo.types import ZDOCmd, ZDOHeader, CLUSTERS as ZDO_CLUSTERS
 
-from zigpy.types import ExtendedPanId
+from zigpy.types import ExtendedPanId, deserialize as list_deserialize
 from zigpy.zcl.clusters.security import IasZone
 
 import zigpy_znp.config as conf
@@ -35,8 +35,8 @@
     ZDOCmd.Node_Desc_req: (
         ZDOCmd.Node_Desc_rsp,
         (
-            lambda addr, ep: c.ZDOCommands.NodeDescReq.Req(
-                DstAddr=addr, NWKAddrOfInterest=addr
+            lambda addr, NWKAddrOfInterest: c.ZDOCommands.NodeDescReq.Req(
+                DstAddr=addr, NWKAddrOfInterest=NWKAddrOfInterest
             )
         ),
         (
@@ -49,8 +49,8 @@
     ZDOCmd.Active_EP_req: (
         ZDOCmd.Active_EP_rsp,
         (
-            lambda addr, ep: c.ZDOCommands.ActiveEpReq.Req(
-                DstAddr=addr, NWKAddrOfInterest=addr
+            lambda addr, NWKAddrOfInterest: c.ZDOCommands.ActiveEpReq.Req(
+                DstAddr=addr, NWKAddrOfInterest=NWKAddrOfInterest
             )
         ),
         (
@@ -63,8 +63,8 @@
     ZDOCmd.Simple_Desc_req: (
         ZDOCmd.Simple_Desc_rsp,
         (
-            lambda addr, ep: c.ZDOCommands.SimpleDescReq.Req(
-                DstAddr=addr, NWKAddrOfInterest=addr, Endpoint=ep
+            lambda addr, NWKAddrOfInterest, EndPoint: c.ZDOCommands.SimpleDescReq.Req(
+                DstAddr=addr, NWKAddrOfInterest=NWKAddrOfInterest, Endpoint=EndPoint
             )
         ),
         (
@@ -474,8 +474,15 @@ async def _send_zdo_request(
 
         assert dst_ep == ZDO_ENDPOINT
 
+        # Deserialize the ZDO request
+        zdo_hdr, data = ZDOHeader.deserialize(cluster, data)
+        field_names, field_types = ZDO_CLUSTERS[cluster]
+        zdo_args, _ = list_deserialize(data, field_types)
+        zdo_kwargs = dict(zip(field_names, zdo_args))
+
+        # Call the converter with the ZDO request's kwargs
         rsp_cluster, req_factory, callback_factory, converter = ZDO_CONVERTERS[cluster]
-        request = req_factory(dst_addr.address, ep=src_ep)
+        request = req_factory(dst_addr.address, **zdo_kwargs)
         callback = callback_factory(dst_addr.address)
 
         LOGGER.debug(