Add support for the aps_encrypted bit

pyproject.toml

@@ -14,7 +14,7 @@ readme = "README.md"
 license = {text = "Apache-2.0"}
 requires-python = ">=3.12"
 dependencies = [
-    "zigpy",
+    "zigpy>=3.0.0",
     "aiohttp",
     "aiospinel>=1.2.0",
 ]

tests/test_api.py

@@ -28,7 +28,6 @@ def _send_aps(*, aps_ack: bool) -> p.SendUnicast:
         cluster_id=0x0006,
         src_ep=1,
         dst_ep=1,
-        aps_seq=55,
         radius=30,
         priority=0,
         asdu=b"\x01\x02",
@@ -262,7 +261,7 @@ async def test_request_confirmed(
 ) -> None:
     """An APS-ack send resolves once the end-to-end APS ack arrives."""
     await api.request_confirmed(_send_aps(aps_ack=True))
-    assert transport.sent(p.SendUnicast)[-1].aps_seq == 55
+    assert transport.sent(p.SendUnicast)[-1].aps_ack
 
 
 async def test_cancel_on_abandon(
@@ -407,7 +406,9 @@ async def test_notifications(
         p.ReceivedAps(
             source=t.NWK(0xAB12),
             destination=t.NWK(0x0000),
-            has_group=t.Bool(False),
+            has_group=t.uint1_t(0),
+            aps_encrypted=t.uint1_t(0),
+            reserved=t.uint6_t(0),
             group=t.uint16_t(0),
             profile_id=t.uint16_t(0x0104),
             cluster_id=t.uint16_t(0x0006),

tests/test_application.py

@@ -668,7 +668,6 @@ async def test_send_packet_unicast(
     assert not send.aps_encryption
     assert not send.sleepy_destination
     assert send.profile_id == 0x0104
-    assert send.aps_seq == 33
     assert send.radius == 30
     assert send.priority == 0
     assert bytes(send.asdu) == b"\x01\x02\x03"
@@ -750,7 +749,6 @@ async def test_send_packet_broadcast(
     send = server.sent(p.SendBroadcast)[-1]
     assert send.destination == t.NWK(0xFFFC)
     assert send.dst_ep == 255
-    assert send.aps_seq == 33
     assert bytes(send.asdu) == b"\x01\x02\x03"
 
 
@@ -766,7 +764,6 @@ async def test_send_packet_groupcast(
 
     send = server.sent(p.SendGroupcast)[-1]
     assert send.group_id == 0x0002
-    assert send.aps_seq == 33
     assert bytes(send.asdu) == b"\x01\x02\x03"
 
 
@@ -1070,7 +1067,9 @@ async def test_on_notification_received_aps(
         p.ReceivedAps(
             source=DEVICE_NWK,
             destination=t.NWK(0x0000),
-            has_group=t.Bool(False),
+            has_group=t.uint1_t(0),
+            aps_encrypted=t.uint1_t(0),
+            reserved=t.uint6_t(0),
             group=t.uint16_t(0),
             profile_id=t.uint16_t(0x0000),
             cluster_id=t.uint16_t(zdo_t.ZDOCmd.Node_Desc_req),
@@ -1086,6 +1085,35 @@ async def test_on_notification_received_aps(
     assert reply.asdu[0] == 0x77
 
 
+@pytest.mark.parametrize("aps_encrypted", [0, 1])
+async def test_on_notification_received_aps_zdo_reply_encryption(
+    app: ControllerApplication, server: SyntheticZiggurat, aps_encrypted: int
+) -> None:
+    """A local ZDO reply uses the same APS security as the request."""
+    add_initialized_device(app)
+
+    await server.send_notification(
+        p.ReceivedAps(
+            source=DEVICE_NWK,
+            destination=t.NWK(0x0000),
+            has_group=t.uint1_t(0),
+            aps_encrypted=t.uint1_t(aps_encrypted),
+            reserved=t.uint6_t(0),
+            group=t.uint16_t(0),
+            profile_id=t.uint16_t(0x0000),
+            cluster_id=t.uint16_t(zdo_t.ZDOCmd.Node_Desc_req),
+            src_ep=t.uint8_t(0),
+            dst_ep=t.uint8_t(0),
+            lqi=t.uint8_t(255),
+            rssi=t.int8s(-40),
+            data=t.LongOctetString(b"\x77" + t.NWK(0x0000).serialize()),
+        )
+    )
+    reply = await server.wait_for(p.SendUnicast)
+    assert reply.cluster_id == zdo_t.ZDOCmd.Node_Desc_rsp
+    assert reply.aps_encryption == aps_encrypted
+
+
 @pytest.mark.parametrize(
     ("destination", "group", "expected_dst"),
     [
@@ -1125,7 +1153,9 @@ async def test_on_notification_received_aps_address_modes(
         p.ReceivedAps(
             source=DEVICE_NWK,
             destination=destination,
-            has_group=t.Bool(group is not None),
+            has_group=t.uint1_t(group is not None),
+            aps_encrypted=t.uint1_t(0),
+            reserved=t.uint6_t(0),
             group=t.uint16_t(group or 0),
             profile_id=t.uint16_t(0x0104),
             cluster_id=t.uint16_t(0x0006),
@@ -1150,6 +1180,48 @@ async def test_on_notification_received_aps_address_modes(
     await app.shutdown(db=False)
 
 
+@pytest.mark.parametrize(
+    ("aps_encrypted", "expected_tx_options"),
+    [
+        (0, t.TransmitOptions.NONE),
+        (1, t.TransmitOptions.APS_Encryption),
+    ],
+)
+async def test_on_notification_received_aps_encrypted(
+    server: SyntheticZiggurat,
+    aps_encrypted: int,
+    expected_tx_options: t.TransmitOptions,
+) -> None:
+    app = RecordingApplication(make_app_config(server.url))
+    await app.connect()
+    await app.start_network()
+    add_initialized_device(app)
+
+    await server.send_notification(
+        p.ReceivedAps(
+            source=DEVICE_NWK,
+            destination=t.NWK(0x0000),
+            has_group=t.uint1_t(0),
+            aps_encrypted=t.uint1_t(aps_encrypted),
+            reserved=t.uint6_t(0),
+            group=t.uint16_t(0),
+            profile_id=t.uint16_t(0x0104),
+            cluster_id=t.uint16_t(0x0006),
+            src_ep=t.uint8_t(1),
+            dst_ep=t.uint8_t(1),
+            lqi=t.uint8_t(200),
+            rssi=t.int8s(-70),
+            data=t.LongOctetString(b"\x00\x01\x00\x00\x00"),
+        )
+    )
+    await flush(app)
+
+    assert len(app.packets) == 1
+    assert app.packets[0].tx_options == expected_tx_options
+
+    await app.shutdown(db=False)
+
+
 async def test_on_notification_frame_counter(
     app: ControllerApplication, server: SyntheticZiggurat
 ) -> None:

zigpy_ziggurat/zigbee/application.py

@@ -614,6 +614,9 @@ def _maybe_handle_local_zdo_request(self, packet: t.ZigbeePacket) -> None:
         coordinator = self._device
         nwk = self.state.node_info.nwk
 
+        # Answer with the same APS security the request arrived with
+        aps_encryption = t.TransmitOptions.APS_Encryption in packet.tx_options
+
         if hdr.command_id == zdo_t.ZDOCmd.Node_Desc_req:
             # Joining devices read our node descriptor to learn the trust center's
             # stack compliance revision before attempting the link key exchange
@@ -635,6 +638,7 @@ def _maybe_handle_local_zdo_request(self, packet: t.ZigbeePacket) -> None:
                     nwk,
                     node_desc,
                     tsn=hdr.tsn,
+                    aps_encryption=aps_encryption,
                 )
             )
         elif hdr.command_id == zdo_t.ZDOCmd.Active_EP_req:
@@ -645,6 +649,7 @@ def _maybe_handle_local_zdo_request(self, packet: t.ZigbeePacket) -> None:
                     nwk,
                     endpoints,
                     tsn=hdr.tsn,
+                    aps_encryption=aps_encryption,
                 )
             )
         elif hdr.command_id == zdo_t.ZDOCmd.Simple_Desc_req:
@@ -667,6 +672,7 @@ def _maybe_handle_local_zdo_request(self, packet: t.ZigbeePacket) -> None:
                     nwk,
                     descriptor,
                     tsn=hdr.tsn,
+                    aps_encryption=aps_encryption,
                 )
             )
 
@@ -794,6 +800,11 @@ def _handle_received_aps_command(self, command: p.ReceivedAps) -> None:
             cluster_id=command.cluster_id,
             lqi=command.lqi,
             rssi=command.rssi,
+            tx_options=(
+                t.TransmitOptions.APS_Encryption
+                if command.aps_encrypted
+                else t.TransmitOptions.NONE
+            ),
             data=t.SerializableBytes(command.data),
         )
         self.packet_received(packet)
@@ -841,7 +852,6 @@ async def send_packet(self, packet: t.ZigbeePacket) -> None:
                     profile_id=packet.profile_id,
                     cluster_id=packet.cluster_id or 0x0000,
                     src_ep=packet.src_ep or 0,
-                    aps_seq=packet.tsn,
                     radius=radius,
                     priority=priority,
                     asdu=asdu,
@@ -854,7 +864,6 @@ async def send_packet(self, packet: t.ZigbeePacket) -> None:
                     cluster_id=packet.cluster_id or 0x0000,
                     src_ep=packet.src_ep or 0,
                     dst_ep=packet.dst_ep or 0,
-                    aps_seq=packet.tsn,
                     radius=radius,
                     priority=priority,
                     asdu=asdu,
@@ -899,7 +908,6 @@ async def send_packet(self, packet: t.ZigbeePacket) -> None:
                     cluster_id=packet.cluster_id or 0x0000,
                     src_ep=packet.src_ep or 0,
                     dst_ep=packet.dst_ep or 0,
-                    aps_seq=packet.tsn,
                     radius=radius,
                     priority=priority,
                     route_control=route_control,

zigpy_ziggurat/zigbee/protocol.py

@@ -204,7 +204,6 @@ def build(
         cluster_id: int,
         src_ep: int,
         dst_ep: int,
-        aps_seq: int,
         radius: int,
         priority: int,
         asdu: bytes,
@@ -225,7 +224,7 @@ def build(
             cluster_id=t.uint16_t(cluster_id),
             src_ep=t.uint8_t(src_ep),
             dst_ep=t.uint8_t(dst_ep),
-            aps_seq=t.uint8_t(aps_seq),
+            _aps_seq=t.uint8_t(0),
             radius=t.uint8_t(radius),
             priority=t.int8s(priority),
             route=route_control,
@@ -251,7 +250,6 @@ def build(
         cluster_id: int,
         src_ep: int,
         dst_ep: int,
-        aps_seq: int,
         radius: int,
         priority: int,
         asdu: bytes,
@@ -263,7 +261,7 @@ def build(
             cluster_id=t.uint16_t(cluster_id),
             src_ep=t.uint8_t(src_ep),
             dst_ep=t.uint8_t(dst_ep),
-            aps_seq=t.uint8_t(aps_seq),
+            _aps_seq=t.uint8_t(0),
             radius=t.uint8_t(radius),
             priority=t.int8s(priority),
             asdu=t.LongOctetString(asdu),
@@ -281,7 +279,6 @@ def build(
         profile_id: int,
         cluster_id: int,
         src_ep: int,
-        aps_seq: int,
         radius: int,
         priority: int,
         asdu: bytes,
@@ -292,7 +289,7 @@ def build(
             profile_id=t.uint16_t(profile_id),
             cluster_id=t.uint16_t(cluster_id),
             src_ep=t.uint8_t(src_ep),
-            aps_seq=t.uint8_t(aps_seq),
+            _aps_seq=t.uint8_t(0),
             radius=t.uint8_t(radius),
             priority=t.int8s(priority),
             asdu=t.LongOctetString(asdu),

zigpy_ziggurat/zigbee/wire.py

@@ -267,7 +267,7 @@ class SendUnicastPayload(t.Struct):
     cluster_id: t.uint16_t
     src_ep: t.uint8_t
     dst_ep: t.uint8_t
-    aps_seq: t.uint8_t
+    _aps_seq: t.uint8_t
     radius: t.uint8_t
     priority: t.int8s
     route: RouteControl
@@ -289,7 +289,7 @@ class SendBroadcastPayload(t.Struct):
     cluster_id: t.uint16_t
     src_ep: t.uint8_t
     dst_ep: t.uint8_t
-    aps_seq: t.uint8_t
+    _aps_seq: t.uint8_t
     radius: t.uint8_t
     priority: t.int8s
     asdu: t.LongOctetString
@@ -301,7 +301,7 @@ class SendGroupcastPayload(t.Struct):
     profile_id: t.uint16_t
     cluster_id: t.uint16_t
     src_ep: t.uint8_t
-    aps_seq: t.uint8_t
+    _aps_seq: t.uint8_t
     radius: t.uint8_t
     priority: t.int8s
     asdu: t.LongOctetString
@@ -389,7 +389,9 @@ class LastResetPayload(t.Struct):
 class ReceivedApsPayload(t.Struct):
     source: t.NWK
     destination: t.NWK
-    has_group: t.Bool
+    has_group: t.uint1_t
+    aps_encrypted: t.uint1_t
+    reserved: t.uint6_t
     group: t.uint16_t
     profile_id: t.uint16_t
     cluster_id: t.uint16_t