From c5ce281201738f2e20288233a23ab18d1dd8e4e7 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Mon, 19 Oct 2026 19:51:33 +0000 Subject: [PATCH 1/3] Forward ZGP indications --- tests/test_api.py | 85 ++++++++++++++++++++++++++++++ tests/test_send_receive.py | 27 ++++++++++ zigpy_deconz/api.py | 29 +++++++++- zigpy_deconz/zigbee/application.py | 13 ++++- 4 files changed, 151 insertions(+), 3 deletions(-) diff --git a/tests/test_api.py b/tests/test_api.py index 55e4179..78b6777 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -9,6 +9,7 @@ import pytest import zigpy.config import zigpy.types as zigpy_t +import zigpy.zgp if sys.version_info[:2] < (3, 11): from async_timeout import timeout as asyncio_timeout @@ -335,6 +336,90 @@ async def test_command_not_connected(api): await api._command(cmd=deconz_api.CommandId.version, reserved=0) +def green_power_indication(api, gpdf: bytes) -> bytes: + _, rx_schema = deconz_api.COMMAND_SCHEMAS[deconz_api.CommandId.zigbee_green_power] + + return deconz_api.Command( + command_id=deconz_api.CommandId.zigbee_green_power, + seq=api._seq, + payload=t.serialize_dict( + { + "status": deconz_api.Status.SUCCESS, + "frame_length": t.uint16_t(7 + len(gpdf)), + "payload_length": t.uint16_t(len(gpdf)), + "gpdf": t.Bytes(gpdf), + }, + rx_schema, + ), + ).serialize() + + +async def test_green_power_indication(api, mock_command_rsp): + """A GPDF received directly is passed to zigpy unprocessed, with its MIC.""" + await api.connect() + + # The spec's test vector for SecurityLevel 0b10 (A.1.5.4.2.1), repeated by the GPD + gpdf = bytes.fromhex("8C 10 21 43 65 87 02 00 00 00 20 CF 78 7E 72") + api.data_received(green_power_indication(api, gpdf)) + api.data_received(green_power_indication(api, gpdf)) + + await asyncio.sleep(0.01) + + # Protected GPDFs are filtered by their frame counter in zigpy + assert ( + api._app.gp_packet_received.mock_calls + == [ + call( + zigpy_t.ZigbeeGpPacket( + application_id=zigpy.zgp.ApplicationID.SrcID, + src_id=zigpy.zgp.SrcID(0x87654321), + command_id=zigpy.zgp.GPDCommandID.Off, + payload=zigpy_t.SerializableBytes(b""), + frame_counter=zigpy_t.uint32_t(2), + security_level=zigpy.zgp.SecurityLevel.FullFrameCounterAndMIC, + security_key_type=None, + security_status=zigpy.zgp.SecurityStatus.Unprocessed, + mic=zigpy_t.uint32_t(0x727E78CF), + auto_commissioning=False, + rx_after_tx=False, + individual_key=False, + ) + ) + ] + * 2 + ) + + +async def test_green_power_indication_unprotected_repeated(api, mock_command_rsp): + """Without a MAC sequence number, repetitions of an unprotected GPDF are dropped.""" + await api.connect() + + toggle = bytes.fromhex("0C 44 33 22 11 22") + on = bytes.fromhex("0C 44 33 22 11 21") + + for gpdf in (toggle, toggle, on, toggle): + api.data_received(green_power_indication(api, gpdf)) + + await asyncio.sleep(0.01) + + assert [c.args[0].command_id for c in api._app.gp_packet_received.mock_calls] == [ + zigpy.zgp.GPDCommandID.Toggle, + zigpy.zgp.GPDCommandID.On, + ] + + +async def test_green_power_indication_invalid(api, mock_command_rsp, caplog): + await api.connect() + + # A Channel Configuration sent by another sink (Direction 0b1) + with caplog.at_level(logging.DEBUG): + api.data_received(green_power_indication(api, bytes.fromhex("8D 80 F3 1F"))) + await asyncio.sleep(0.01) + + assert "Ignoring GPDF" in caplog.text + assert len(api._app.gp_packet_received.mock_calls) == 0 + + async def test_data_received(api, mock_command_rsp): await api.connect() diff --git a/tests/test_send_receive.py b/tests/test_send_receive.py index 2416c06..2ac3298 100644 --- a/tests/test_send_receive.py +++ b/tests/test_send_receive.py @@ -7,6 +7,7 @@ import pytest import zigpy.exceptions import zigpy.types as zigpy_t +import zigpy.zgp from zigpy_deconz.api import Status, TXStatus import zigpy_deconz.exception @@ -207,6 +208,32 @@ async def test_send_packet_broadcast(app, tx_packet): # noqa: F811 assert req["radius"] == 12 +async def test_send_packet_green_power_endpoint(app, tx_packet): # noqa: F811 + """The Green Power endpoint is the one source endpoint that is kept as is.""" + tx_packet.src_ep = zigpy.zgp.GP_ENDPOINT + tx_packet.dst_ep = zigpy.zgp.GP_ENDPOINT + + with patch_data_request(app) as mock_req: + await app.send_packet(tx_packet) + + assert mock_req.mock_calls[0].kwargs["src_ep"] == zigpy.zgp.GP_ENDPOINT + + +async def test_send_packet_alias(app, tx_packet): # noqa: F811 + """The firmware cannot send on behalf of a Green Power alias.""" + tx_packet.dst = zigpy_t.AddrModeAddress( + addr_mode=zigpy_t.AddrMode.Broadcast, + address=zigpy_t.BroadcastAddress.RX_ON_WHEN_IDLE, + ) + tx_packet.alias_src = zigpy_t.NWK(0x1234) + tx_packet.alias_seq = zigpy_t.uint8_t(0) + + with patch_data_request(app) as mock_req, pytest.raises(NotImplementedError): + await app.send_packet(tx_packet) + + assert len(mock_req.mock_calls) == 0 + + async def test_send_packet_enqueue_failure(app, tx_packet): # noqa: F811 with patch_data_request(app, fail_enqueue=True): # noqa: F811 with pytest.raises(zigpy.exceptions.DeliveryError) as e: diff --git a/zigpy_deconz/api.py b/zigpy_deconz/api.py index 3ec9e3d..e5ea448 100644 --- a/zigpy_deconz/api.py +++ b/zigpy_deconz/api.py @@ -14,7 +14,8 @@ else: from asyncio import timeout as asyncio_timeout # pragma: no cover -from zigpy.datastructures import PriorityLock +from zigpy.datastructures import Debouncer, PriorityLock +from zigpy.device import GP_DUPLICATE_TIMEOUT from zigpy.types import ( APSStatus, Bool, @@ -25,6 +26,7 @@ ZigbeePacket, ) from zigpy.zdo.types import SimpleDescriptor +from zigpy.zgp.util import parse_gpdf from zigpy_deconz.exception import CommandError, MismatchedResponseError, ParsingError import zigpy_deconz.types as t @@ -404,7 +406,9 @@ class Command(Struct): "status": Status, "frame_length": t.uint16_t, "payload_length": t.uint16_t, - "reserved": t.LongOctetString, + # The NWK frame of a received GPDF, raw: the firmware performs no security + # processing and strips the MAC header + "gpdf": t.Bytes, }, ), } @@ -433,6 +437,11 @@ def __init__(self, app: Callable, device_config: dict[str, Any]): self._free_slots_available_event = asyncio.Event() self._free_slots_available_event.set() + # The firmware delivers every repetition of a GPDF and strips the MAC header, + # so the MAC sequence number of an unprotected GPDF is not available to filter + # them (spec A.3.6.1.2): its content is the only key + self._gp_debouncer = Debouncer() + self._data_poller_event = asyncio.Event() self._data_poller_event.set() self._data_poller_task: asyncio.Task | None = None @@ -768,6 +777,22 @@ async def _data_poller(self): status=rsp["status"], device_state=rsp["device_state"] ) + def _handle_zigbee_green_power(self, status: Status, gpdf: t.Bytes) -> None: + try: + packet = parse_gpdf(gpdf) + except ValueError: + LOGGER.debug("Ignoring GPDF %r", gpdf, exc_info=True) + return + + if packet.frame_counter is None and self._gp_debouncer.filter( + (packet.src_id, packet.command_id, packet.payload.serialize()), + GP_DUPLICATE_TIMEOUT.total_seconds(), + ): + LOGGER.debug("Dropping repeated GPDF: %r", packet) + return + + self._app.gp_packet_received(packet) + def _handle_device_state_changed( self, status: t.Status, diff --git a/zigpy_deconz/zigbee/application.py b/zigpy_deconz/zigbee/application.py index 8fa2fdb..cf77cdb 100644 --- a/zigpy_deconz/zigbee/application.py +++ b/zigpy_deconz/zigbee/application.py @@ -23,6 +23,7 @@ import zigpy.types import zigpy.util import zigpy.zdo.types as zdo_t +import zigpy.zgp import zigpy_deconz from zigpy_deconz import types as t @@ -545,6 +546,10 @@ async def add_endpoint(self, descriptor: zdo_t.SimpleDescriptor) -> None: async def send_packet(self, packet): LOGGER.debug("Sending packet: %r", packet) + if packet.alias_src is not None: + # The APS data request has no alias parameters + raise NotImplementedError("The firmware cannot send with an alias") + if zigpy.types.TransmitOptions.APS_Encryption in packet.tx_options: # APS encryption is performed with the link key, not the network key, so # the "use NWK key" option must not be set @@ -578,7 +583,13 @@ async def send_packet(self, packet): ), profile=packet.profile_id, cluster=packet.cluster_id, - src_ep=min(1, packet.src_ep), + # The Green Power endpoint is registered on its own, every + # other source endpoint collapses onto the first one + src_ep=( + packet.src_ep + if packet.src_ep == zigpy.zgp.GP_ENDPOINT + else min(1, packet.src_ep) + ), aps_payload=packet.data.serialize(), tx_options=tx_options, relays=packet.source_route, From 064bcc91bd39351c80d80a2a1937f5d83b5ae005 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Fri, 23 Oct 2026 04:02:47 +0000 Subject: [PATCH 2/3] Bump zigpy to 2.4.0 --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 763c124..e4e4751 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,7 +15,7 @@ license = {text = "GPL-3.0"} requires-python = ">=3.8" dependencies = [ "voluptuous", - "zigpy>=0.80.0", + "zigpy>=2.4.0", 'async-timeout; python_version<"3.11"', ] From f931178874d78fe1656b4bc50b784887af3015ef Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Fri, 23 Oct 2026 04:06:34 +0000 Subject: [PATCH 3/3] Include the optional GP packet `endpoint` when debouncing --- zigpy_deconz/api.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/zigpy_deconz/api.py b/zigpy_deconz/api.py index e5ea448..931c02e 100644 --- a/zigpy_deconz/api.py +++ b/zigpy_deconz/api.py @@ -785,7 +785,12 @@ def _handle_zigbee_green_power(self, status: Status, gpdf: t.Bytes) -> None: return if packet.frame_counter is None and self._gp_debouncer.filter( - (packet.src_id, packet.command_id, packet.payload.serialize()), + ( + packet.src_id, + packet.command_id, + packet.endpoint, + packet.payload.serialize(), + ), GP_DUPLICATE_TIMEOUT.total_seconds(), ): LOGGER.debug("Dropping repeated GPDF: %r", packet)