From fb7ffda401270ed913d357cf4fd951317c47595f Mon Sep 17 00:00:00 2001 From: nmingam <192914892+nmingam@users.noreply.github.com> Date: Thu, 23 Apr 2026 12:13:33 +0200 Subject: [PATCH 1/8] Fix gpepIncomingMessageHandler parsing on EZSP v13/v14/v16 Real Green Power frames emitted by a Friends of Hue class switch never reached the host: bellows dropped them with 'Data is too short' at deserialization time, as reported in zigpy/zigpy#1814 with a capture from a Busch-Jaeger 6716 U switch on a Silabs ZBT-1 stick. Three things were wrong: - The v4 schema for this callback, inherited up to v16, treated the GP address as five scattered fields (addrType, addr:uint32, applicationId, address:EUI64, endpoint). The NCP actually sends a single 10-byte EmberGpAddress struct: applicationId, an 8-byte id union, and endpoint. The existing EmberGpAddress type also assumed the wrong layout (14 bytes with separate gpdIeeeAddress and sourceId fields). - The v17 override uses sl_GpStatus, a strict enum that only accepts 0x00..0x07. Current ZBT-1 firmware returns higher status bytes for frames that were not matched in the proxy table (0x7C observed on the captured frame), which dropped the whole callback before the address was even read. - EZSP v16 appends an SlRxPacketInfo struct after the LVBytes payload. zigbee-herdsman's ember adapter gates the read on 'version >= 0x10', so v13 and v14 do not carry this trailer but v16 does. The previous override that flowed through from v13 was one field short on v16. Fix EmberGpAddress to the correct 10-byte layout, add a v13 override of gpepIncomingMessageHandler that uses a plain uint8_t for status so unexpected values reach the host, and add a dedicated v16 override that appends the SlRxPacketInfo trailer. v14 picks the v13 override up through the _REPLACEMENTS loop. v17+ keep their own schema which now also benefits from the struct fix. Add tests driven by the exact bytes captured by the community tester, a smoke test on v14 to verify the schema flows through inheritance, and a v16 test that appends a synthetic SlRxPacketInfo to the same capture to exercise the trailer parsing. --- bellows/ezsp/v13/commands.py | 36 +++++++++++++ bellows/ezsp/v16/commands.py | 28 ++++++++++ bellows/types/struct.py | 39 +++++++++++--- tests/test_ezsp_v13.py | 100 +++++++++++++++++++++++++++++++++++ tests/test_ezsp_v14.py | 26 +++++++++ tests/test_ezsp_v16.py | 38 +++++++++++++ 6 files changed, 261 insertions(+), 6 deletions(-) diff --git a/bellows/ezsp/v13/commands.py b/bellows/ezsp/v13/commands.py index f87664ae..eda2a280 100644 --- a/bellows/ezsp/v13/commands.py +++ b/bellows/ezsp/v13/commands.py @@ -90,6 +90,42 @@ "status": t.sl_Status, }, ), + # Redefine the Green Power incoming callback so real GPDFs from a + # Friends of Hue class switch actually parse. The v4 schema used + # scattered address fields (``addrType`` + ``addr:uint32`` + + # ``applicationId`` + ``address:EUI64`` + ``endpoint``) that do not + # match the 10-byte ``EmberGpAddress`` the NCP actually sends. + # + # ``status`` is kept as a plain ``uint8_t`` on purpose. The strict + # ``sl_GpStatus`` enum only accepts 0x00..0x07, but firmware versions + # shipped with current ZBT-1 / SkyConnect sticks return higher status + # bytes (e.g. 0x7C observed from a Busch-Jaeger 6716 U switch) when + # the frame was not matched against a proxy/sink table entry. Using a + # plain ``uint8_t`` lets the frame reach the host so zigpy can decide + # what to do with it. + # + # v14 and v16 inherit this override through the ``_REPLACEMENTS`` + # mechanism in ``bellows/ezsp/v14/commands.py``; v17 and later keep + # their own schema which already uses ``EmberGpAddress``. + "gpepIncomingMessageHandler": ( + 0x00C5, + {}, + { + "status": t.uint8_t, + "gpdLink": t.uint8_t, + "sequenceNumber": t.uint8_t, + "addr": t.EmberGpAddress, + "gpdfSecurityLevel": t.EmberGpSecurityLevel, + "gpdfSecurityKeyType": t.EmberGpKeyType, + "autoCommissioning": t.Bool, + "bidirectionalInfo": t.uint8_t, + "gpdSecurityFrameCounter": t.uint32_t, + "gpdCommandId": t.uint8_t, + "mic": t.uint32_t, + "proxyTableIndex": t.uint8_t, + "gpdCommandPayload": t.LVBytes, + }, + ), } del COMMANDS["becomeTrustCenter"] # this one was likely removed earlier diff --git a/bellows/ezsp/v16/commands.py b/bellows/ezsp/v16/commands.py index 51dc4248..92ef7a4d 100644 --- a/bellows/ezsp/v16/commands.py +++ b/bellows/ezsp/v16/commands.py @@ -1,5 +1,33 @@ +import bellows.types as t + from ..v14.commands import COMMANDS as COMMANDS_v14 COMMANDS = { **COMMANDS_v14, + # EZSP v16 appends an ``EmberRxPacketInfo`` struct after the LVBytes + # payload in ``gpepIncomingMessageHandler``. zigbee-herdsman gates the + # read on ``version >= 0x10`` (see + # ``src/adapter/ember/ezsp/ezsp.ts`` in the ``ember`` adapter), so v13 + # and v14 do not carry this trailer but v16 does. v17 and v18 override + # the whole command again, so this entry only matters for v16. + "gpepIncomingMessageHandler": ( + 0x00C5, + {}, + { + "status": t.uint8_t, + "gpdLink": t.uint8_t, + "sequenceNumber": t.uint8_t, + "addr": t.EmberGpAddress, + "gpdfSecurityLevel": t.EmberGpSecurityLevel, + "gpdfSecurityKeyType": t.EmberGpKeyType, + "autoCommissioning": t.Bool, + "bidirectionalInfo": t.uint8_t, + "gpdSecurityFrameCounter": t.uint32_t, + "gpdCommandId": t.uint8_t, + "mic": t.uint32_t, + "proxyTableIndex": t.uint8_t, + "gpdCommandPayload": t.LVBytes, + "packetInfo": t.SlRxPacketInfo, + }, + ), } diff --git a/bellows/types/struct.py b/bellows/types/struct.py index 91a17c01..68c61e20 100644 --- a/bellows/types/struct.py +++ b/bellows/types/struct.py @@ -357,16 +357,43 @@ class EmberTokTypeStackZllSecurity(EzspStruct): class EmberGpAddress(EzspStruct): - # A GP address structure. - # The GPD's EUI64. - gpdIeeeAddress: named.EUI64 - # The GPD's source ID. - sourceId: basic.uint32_t - # The GPD Application ID. + """A GP address structure. + + On the wire this is a 10-byte layout: + + - ``applicationId`` (1 byte) — 0 means ``SrcID`` mode, 2 means ``IEEE`` + mode. + - ``id`` (8 bytes) — a raw union. When ``applicationId == 0`` the source + ID lives in the first 4 bytes (little-endian) and the remaining 4 + bytes are padding. When ``applicationId == 2`` the full 8 bytes are + the GPD EUI64. + - ``endpoint`` (1 byte). + + The historical declaration here treated the union as two separate + fields (``gpdIeeeAddress`` + ``sourceId``) for a total of 14 bytes, + which does not match what the NCP sends in + ``gpepIncomingMessageHandler``. See the ``source_id`` and + ``gpd_ieee_address`` helpers below to access the right view. + """ + + # The GPD Application ID: 0 = source ID mode, 2 = IEEE mode. applicationId: basic.uint8_t + # Raw 8-byte union. Use :attr:`source_id` or :attr:`gpd_ieee_address` + # to interpret it based on :attr:`applicationId`. + id: basic.FixedList[basic.uint8_t, 8] # The GPD endpoint. endpoint: basic.uint8_t + @property + def source_id(self) -> int: + """Return the 32-bit source ID (only valid when applicationId == 0).""" + return int.from_bytes(bytes(self.id[:4]), "little") + + @property + def gpd_ieee_address(self) -> named.EUI64: + """Return the EUI64 (only valid when applicationId == 2).""" + return named.EUI64(bytes(self.id)) + class NV3StackTrustCenterToken(EzspStruct): """NV3 stack trust center token value.""" diff --git a/tests/test_ezsp_v13.py b/tests/test_ezsp_v13.py index 69c5909f..03495956 100644 --- a/tests/test_ezsp_v13.py +++ b/tests/test_ezsp_v13.py @@ -227,3 +227,103 @@ async def test_factory_reset(ezsp_f) -> None: assert ezsp_f.tokenFactoryReset.mock_calls == [ call(excludeOutgoingFC=False, excludeBootCounter=False) ] + + +# --------------------------------------------------------------------------- +# gpepIncomingMessageHandler (0x00C5) +# +# Real payload captured on 2026-04-22 from a Busch-Jaeger 6716 U +# "Friends of Hue" switch paired through a Silabs ZBT-1 stick running +# EZSP v13/v14. The raw bytes come straight from the bellows warning log +# reported in zigpy/zigpy#1814 before this fix landed. +# --------------------------------------------------------------------------- + + +# Payload without the EZSP frame envelope, as passed to ``deserialize_dict``. +BJ6716U_GPEP_PAYLOAD: bytes = bytes.fromhex( + "7ccdec" # status, gpdLink, sequenceNumber + "00" # EmberGpAddress.applicationId = 0 (SrcID) + "86f8710186f87101" # EmberGpAddress.id (8-byte union, + # sourceId = 0x0171F886 in first 4 LE) + "00" # EmberGpAddress.endpoint + "00" # gpdfSecurityLevel + "00" # gpdfSecurityKeyType + "00" # autoCommissioning + "00" # bidirectionalInfo + "ffffffff" # gpdSecurityFrameCounter (commissioning) + "e0" # gpdCommandId = 0xE0 (Commissioning) + "ffffffff" # mic + "ff" # proxyTableIndex + "2e" # LVBytes length = 46 + "02c5f2" # commissioning: deviceId, options, extOptions + "1ce9ae2f9e4f85f15de37c1ccbd94387" # encrypted GPD key (16 bytes) + "0013911a" # KeyMIC + "ec1d0000" # OutgoingCounter = 0x00001DEC + "04" # AppInfo + "11" # NumGPDCommands = 17 + "1011121314151617" # RecallScene 0-7 + "22" # Toggle + "6062636465666768" # Press/Release variants +) + + +def test_gpep_incoming_real_frame_bj6716u(): + """Parse the commissioning frame captured by a community tester. + + Before this schema override the v4 layout inherited up to v16 treated + the address as five scattered fields and ran off the end of the + buffer. The symptom was ``ValueError: Data is too short`` exactly as + reported in zigpy/zigpy#1814. + """ + _, _, rx_schema = bellows.ezsp.v13.commands.COMMANDS[ + "gpepIncomingMessageHandler" + ] + result, rest = t.deserialize_dict(BJ6716U_GPEP_PAYLOAD, rx_schema) + + assert rest == b"" + # ``status`` comes through as a plain byte — the strict ``sl_GpStatus`` + # enum only covers 0x00..0x07 and would have dropped the whole frame. + assert result["status"] == 0x7C + assert result["gpdLink"] == 0xCD + assert result["sequenceNumber"] == 0xEC + + addr = result["addr"] + assert addr.applicationId == 0 + assert addr.source_id == 0x0171F886 + assert addr.endpoint == 0 + + assert result["gpdfSecurityLevel"] == 0 + assert result["gpdfSecurityKeyType"] == 0 + assert result["gpdSecurityFrameCounter"] == 0xFFFFFFFF + assert result["gpdCommandId"] == 0xE0 # GPD Commissioning + # Commissioning payload: deviceId=0x02, then options, then the 17 + # advertised GPD commands at the tail. + payload = result["gpdCommandPayload"] + assert len(payload) == 46 + assert payload[:3] == b"\x02\xc5\xf2" + assert payload[-8:] == bytes.fromhex("6062636465666768") + + +def test_gpep_incoming_via_frame_rx(ezsp_f): + """The same frame wrapped in the EZSP v13 envelope reaches the callback. + + Builds a synthetic EZSP frame (``seq`` + padding + ``frame_id`` LE + + payload) and feeds it to the protocol handler. The callback must be + dispatched with the parsed result — no ``Failed to parse frame`` + warning. + """ + envelope = ( + bytes([0x42, 0x00, 0x01]) # seq + control bytes + + t.uint16_t(0x00C5).serialize() # frame_id LE + + BJ6716U_GPEP_PAYLOAD + ) + + ezsp_f(envelope) + + assert ezsp_f._handle_callback.call_count == 1 + assert ezsp_f._handle_callback.call_args[0][0] == "gpepIncomingMessageHandler" + parsed = ezsp_f._handle_callback.call_args[0][1] + # ``parsed`` is the list of values in schema order. + assert parsed[0] == 0x7C # status + assert parsed[3].source_id == 0x0171F886 # addr.source_id + assert parsed[9] == 0xE0 # gpdCommandId diff --git a/tests/test_ezsp_v14.py b/tests/test_ezsp_v14.py index 49bf152b..21c26f11 100644 --- a/tests/test_ezsp_v14.py +++ b/tests/test_ezsp_v14.py @@ -226,3 +226,29 @@ async def test_send_broadcast(ezsp_f) -> None: message=b"hello", ) ] + + +def test_gpep_incoming_inherits_v13_schema(ezsp_f): + """v14 must pick up the v13 override of ``gpepIncomingMessageHandler``. + + The fix is declared in v13; v14 re-exports commands through the + ``_REPLACEMENTS`` loop, and the plain ``uint8_t`` status passes + through unchanged because the loop only rewrites ``EmberStatus`` and + ``EzspStatus``. Feeding the real Busch-Jaeger 6716 U capture must + dispatch the callback without warnings. + """ + from tests.test_ezsp_v13 import BJ6716U_GPEP_PAYLOAD + + envelope = ( + bytes([0x42, 0x00, 0x01]) + + t.uint16_t(0x00C5).serialize() + + BJ6716U_GPEP_PAYLOAD + ) + + ezsp_f(envelope) + + assert ezsp_f._handle_callback.call_count == 1 + assert ezsp_f._handle_callback.call_args[0][0] == "gpepIncomingMessageHandler" + parsed = ezsp_f._handle_callback.call_args[0][1] + assert parsed[3].source_id == 0x0171F886 + assert parsed[9] == 0xE0 # gpdCommandId diff --git a/tests/test_ezsp_v16.py b/tests/test_ezsp_v16.py index ee75c22b..b12b4049 100644 --- a/tests/test_ezsp_v16.py +++ b/tests/test_ezsp_v16.py @@ -31,3 +31,41 @@ def test_ezsp_frame_rx(ezsp_f): assert ezsp_f._handle_callback.call_count == 1 assert ezsp_f._handle_callback.call_args[0][0] == "version" assert ezsp_f._handle_callback.call_args[0][1] == [0x01, 0x02, 0x1234] + + +def test_gpep_incoming_v16_expects_trailing_packet_info(ezsp_f): + """v16 appends an ``SlRxPacketInfo`` struct after the LVBytes payload. + + zigbee-herdsman gates the read on ``version >= 0x10``. We simulate + that wire format by taking the v13 payload captured from a real + Busch-Jaeger 6716 U and appending a synthetic packet info trailer. + """ + from tests.test_ezsp_v13 import BJ6716U_GPEP_PAYLOAD + + packet_info = t.SlRxPacketInfo( + sender_short_id=t.NWK(0x1234), + sender_long_id=t.EUI64.convert("00:11:22:33:44:55:66:77"), + binding_index=0xFF, + address_index=0xFF, + last_hop_lqi=200, + last_hop_rssi=-45, + last_hop_timestamp=0xDEADBEEF, + ).serialize() + + envelope = ( + bytes([0x42, 0x00, 0x01]) + + t.uint16_t(0x00C5).serialize() + + BJ6716U_GPEP_PAYLOAD + + packet_info + ) + + ezsp_f(envelope) + + assert ezsp_f._handle_callback.call_count == 1 + assert ezsp_f._handle_callback.call_args[0][0] == "gpepIncomingMessageHandler" + parsed = ezsp_f._handle_callback.call_args[0][1] + assert parsed[3].source_id == 0x0171F886 + assert parsed[9] == 0xE0 # gpdCommandId + assert parsed[-1].sender_short_id == 0x1234 + assert parsed[-1].last_hop_lqi == 200 + assert parsed[-1].last_hop_rssi == -45 From 7eca82a52a1defbf179a5916e5b5b566227bf465 Mon Sep 17 00:00:00 2001 From: nmingam <192914892+nmingam@users.noreply.github.com> Date: Thu, 23 Apr 2026 14:05:05 +0200 Subject: [PATCH 2/8] Forward gpepIncomingMessageHandler to zigpy's GreenPower manager Decoding the callback is only half the job: zigpy's GreenPower manager listens for GP frames through ControllerApplication. packet_received() on endpoint 242, cluster 0x0021, and expects a ZCL GP Notification (server command 0x00). Until now bellows dispatched nothing at all for this callback, so real GPDFs from a Friends of Hue switch like the Busch-Jaeger 6716 U never reached the host stack. Add a dedicated branch in ezsp_callback_handler that reencapsulates the flattened GPDF as a ZCL GP Notification and calls packet_received, mirroring what zigbee-herdsman's ember adapter does. Key behaviour: - SrcID-mode GPDs (applicationId == 0) are forwarded. IEEE-mode GPDs are dropped with a debug log because the current zigpy GP stack explicitly does not support them. - On EZSP v16+ the trailing SlRxPacketInfo supplies the proxy NWK, LQI and RSSI. On v13/v14 we fall back to the coordinator short address and the gpdLink field, since no proxy info is on the wire. - Malformed callbacks (short arg tuple) are dropped silently so a firmware mismatch cannot crash the dispatcher. - A new green_power_rx counter tracks how many GPDFs the NCP has handed us, mirroring the existing rx counters. Add tests that cover the v13/v14 shape, the v16 shape with SlRxPacketInfo, an operational data command (Toggle), the IEEE drop path, and the short-args drop path. Each test asserts the ZCL envelope, the NotificationSchema round-trip, and the packet routing fields zigpy needs. --- bellows/zigbee/application.py | 116 ++++++++++++++++++++++++++ tests/test_application.py | 150 ++++++++++++++++++++++++++++++++++ 2 files changed, 266 insertions(+) diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py index 1bd0271c..59e79643 100644 --- a/bellows/zigbee/application.py +++ b/bellows/zigbee/application.py @@ -21,7 +21,9 @@ import zigpy.state import zigpy.types import zigpy.util +from zigpy.zcl.clusters.greenpower import NotificationOptions, NotificationSchema import zigpy.zdo.types as zdo_t +import zigpy.zgp.types as zgp_t import bellows from bellows.config import ( @@ -56,6 +58,7 @@ COUNTER_RX_BCAST = "broadcast_rx" COUNTER_RX_MCAST = "multicast_rx" COUNTER_RX_UNICAST = "unicast_rx" +COUNTER_RX_GP = "green_power_rx" COUNTER_UNKNOWN_DEVICE = "unknown_device_rx" COUNTER_WATCHDOG = "watchdog_reset_requests" COUNTERS_EZSP = "ezsp_counters" @@ -696,6 +699,119 @@ def ezsp_callback_handler(self, frame_name, args): self.handle_route_error(status, nwk) elif frame_name == "idConflictHandler": self._handle_id_conflict(*args) + elif frame_name == "gpepIncomingMessageHandler": + self._handle_gp_frame(args) + + def _handle_gp_frame(self, args: tuple) -> None: + """Forward a decoded gpepIncomingMessageHandler callback to zigpy. + + The NCP has already done the GP stub processing (proxy-table lookup, + duplicate suppression, etc.) and hands us a flattened GPDF. zigpy's + GP manager on the other side expects a ZigbeePacket on endpoint 242, + cluster 0x0021, carrying a ZCL GP Notification (server command 0x00) + so it can route commissioning, data, and decommissioning frames + through the same entry point. + + We therefore rebuild that envelope here, mirroring what zigbee- + herdsman's ember adapter does. Any GPDF that cannot be converted + (for example an IEEE-addressed frame, which the current zigpy GP + stack explicitly does not support) is logged and dropped. + """ + if len(args) < 13: + LOGGER.debug("gpepIncomingMessageHandler: short args %r, dropping", args) + return + + ( + _status, + gpd_link, + sequence_number, + addr, + gpdf_security_level, + gpdf_security_key_type, + _auto_commissioning, + _bidirectional_info, + gpd_security_frame_counter, + gpd_command_id, + _mic, + _proxy_table_index, + gpd_command_payload, + *rest, + ) = args + + # v16+ appends an SlRxPacketInfo after the payload. Pull the + # proxy NWK out of it if it is there, otherwise fall back to the + # coordinator short address (the coordinator acts as the proxy). + packet_info = rest[0] if rest else None + + if addr.applicationId == zgp_t.ApplicationID.SrcID: + source_id = addr.source_id + else: + LOGGER.debug( + "GP frame with unsupported applicationId %s (only SrcID is " + "handled), dropping", + addr.applicationId, + ) + return + + options = NotificationOptions( + application_id=zgp_t.ApplicationID(addr.applicationId), + also_unicast=0, + also_derived_group=0, + also_commissioned_group=0, + security_level=zgp_t.SecurityLevel(gpdf_security_level), + security_key_type=zgp_t.SecurityKeyType(gpdf_security_key_type), + appoint_temp_master=0, + tx_queue_full=0, + _reserved=0, + ) + + notification = NotificationSchema( + options=options, + gpd_id=zgp_t.DeviceID(source_id), + frame_counter=zigpy.types.uint32_t(gpd_security_frame_counter), + command_id=zigpy.types.uint8_t(gpd_command_id), + payload=zigpy.types.LVBytes(bytes(gpd_command_payload)), + ) + + # ZCL header: frame_control=0x01 (cluster-specific, client→server, + # not manufacturer-specific, no default response disabled), TSN is + # the GPD MAC sequence number so duplicates can be tracked, and the + # command is GP Notification (0x00). + zcl_bytes = ( + bytes([0x01, int(sequence_number) & 0xFF, 0x00]) + notification.serialize() + ) + + if packet_info is not None: + proxy_nwk = int(packet_info.sender_short_id) + lqi = int(packet_info.last_hop_lqi) + rssi = int(packet_info.last_hop_rssi) + else: + proxy_nwk = int(self.state.node_info.nwk) + lqi = int(gpd_link) + rssi = 0 + + self.state.counters[COUNTERS_CTRL][COUNTER_RX_GP].increment() + + self.packet_received( + zigpy.types.ZigbeePacket( + src=zigpy.types.AddrModeAddress( + addr_mode=zigpy.types.AddrMode.NWK, + address=zigpy.types.NWK(proxy_nwk), + ), + src_ep=zigpy.types.uint8_t(zgp_t.GP_ENDPOINT), + dst=zigpy.types.AddrModeAddress( + addr_mode=zigpy.types.AddrMode.NWK, + address=self.state.node_info.nwk, + ), + dst_ep=zigpy.types.uint8_t(zgp_t.GP_ENDPOINT), + tsn=zigpy.types.uint8_t(int(sequence_number) & 0xFF), + profile_id=zigpy.types.uint16_t(0xA1E0), + cluster_id=zigpy.types.uint16_t(zgp_t.GP_CLUSTER_ID), + data=zigpy.types.SerializableBytes(zcl_bytes), + lqi=zigpy.types.uint8_t(lqi), + rssi=zigpy.types.int8s(rssi), + ) + ) def _handle_frame( self, diff --git a/tests/test_application.py b/tests/test_application.py index 018dd632..4830e930 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -541,6 +541,156 @@ def test_frame_handler_ignored(app, aps_frame): assert app.packet_received.call_count == 0 +# --------------------------------------------------------------------------- +# gpepIncomingMessageHandler forwarding +# --------------------------------------------------------------------------- +# +# Expected behaviour: the NCP delivers a flattened GPDF via +# ``gpepIncomingMessageHandler``. bellows must reencapsulate it as a ZCL +# GP Notification on endpoint 242, cluster 0x0021, so zigpy's GreenPower +# manager (listening through ``packet_received``) can route it. These +# tests cover both the v13/v14 wire format (no trailing SlRxPacketInfo) +# and the v16+ wire format (with trailer). +from zigpy.zcl.clusters.greenpower import NotificationSchema # noqa: E402 +import zigpy.zgp.types as zgp_t # noqa: E402 + + +def _gp_addr_srcid(source_id: int, endpoint: int = 0x00) -> t.EmberGpAddress: + """Build a 10-byte EmberGpAddress for a SrcID-mode GPD.""" + id_bytes = source_id.to_bytes(4, "little") + bytes(4) + return t.EmberGpAddress( + applicationId=t.uint8_t(0), + id=t.FixedList[t.uint8_t, 8](id_bytes), + endpoint=t.uint8_t(endpoint), + ) + + +def _gp_args_v13( + *, + source_id: int = 0x0171F886, + status: int = 0x7C, + gpd_link: int = 0xCD, + sequence_number: int = 0xEC, + security_level: int = 0, + security_key_type: int = 0, + frame_counter: int = 0xFFFFFFFF, + command_id: int = 0xE0, + payload: bytes = b"\x02\xc5\xf2" + bytes(43), +) -> list: + """Build the args list as delivered on EZSP v13/v14 (no trailer).""" + return [ + t.uint8_t(status), + t.uint8_t(gpd_link), + t.uint8_t(sequence_number), + _gp_addr_srcid(source_id), + t.EmberGpSecurityLevel(security_level), + t.EmberGpKeyType(security_key_type), + t.Bool(False), # autoCommissioning + t.uint8_t(0), # bidirectionalInfo + t.uint32_t(frame_counter), + t.uint8_t(command_id), + t.uint32_t(0xFFFFFFFF), # mic + t.uint8_t(0xFF), # proxyTableIndex + t.LVBytes(payload), + ] + + +def test_gp_frame_handler_forwards_notification(app): + """A GPDF is forwarded to zigpy as a ZCL GP Notification.""" + app.ezsp_callback_handler("gpepIncomingMessageHandler", _gp_args_v13()) + + assert app.packet_received.call_count == 1 + + packet = app.packet_received.mock_calls[0].args[0] + assert packet.dst_ep == zgp_t.GP_ENDPOINT + assert packet.src_ep == zgp_t.GP_ENDPOINT + assert packet.cluster_id == zgp_t.GP_CLUSTER_ID + assert packet.profile_id == 0xA1E0 + assert packet.dst.address == app.state.node_info.nwk + # On v13/v14 there is no packetInfo, so bellows uses the coordinator + # short address as a stand-in for the proxy NWK. + assert packet.src.address == app.state.node_info.nwk + + # ZCL header: 0x01 (cluster-specific, client→server) + TSN + cmd 0x00 + raw = packet.data.serialize() + assert raw[0] == 0x01 + assert raw[2] == 0x00 # GP Notification + notification = NotificationSchema.deserialize(raw[3:])[0] + assert int(notification.gpd_id) == 0x0171F886 + assert int(notification.frame_counter) == 0xFFFFFFFF + assert int(notification.command_id) == 0xE0 + # Commissioning payload is carried verbatim so zigpy can parse it + assert bytes(notification.payload) == b"\x02\xc5\xf2" + bytes(43) + + assert ( + app.state.counters[bellows.zigbee.application.COUNTERS_CTRL][ + bellows.zigbee.application.COUNTER_RX_GP + ] + == 1 + ) + + +def test_gp_frame_handler_v16_uses_packet_info(app): + """On v16+, the trailing SlRxPacketInfo provides the proxy NWK + LQI/RSSI.""" + packet_info = t.SlRxPacketInfo( + sender_short_id=t.NWK(0x1234), + sender_long_id=t.EUI64.convert("00:11:22:33:44:55:66:77"), + binding_index=t.uint8_t(0xFF), + address_index=t.uint8_t(0xFF), + last_hop_lqi=t.uint8_t(210), + last_hop_rssi=t.int8s(-40), + last_hop_timestamp=t.uint32_t(0xDEADBEEF), + ) + args = _gp_args_v13() + [packet_info] + + app.ezsp_callback_handler("gpepIncomingMessageHandler", args) + + assert app.packet_received.call_count == 1 + packet = app.packet_received.mock_calls[0].args[0] + assert packet.src.address == 0x1234 + assert packet.lqi == 210 + assert packet.rssi == -40 + + +def test_gp_frame_handler_data_command_toggle(app): + """An operational data command (e.g. Toggle) is forwarded the same way.""" + app.ezsp_callback_handler( + "gpepIncomingMessageHandler", + _gp_args_v13(command_id=0x22, payload=b"", frame_counter=42), + ) + + assert app.packet_received.call_count == 1 + packet = app.packet_received.mock_calls[0].args[0] + notification = NotificationSchema.deserialize(packet.data.serialize()[3:])[0] + assert int(notification.command_id) == 0x22 + assert int(notification.frame_counter) == 42 + assert bytes(notification.payload) == b"" + + +def test_gp_frame_handler_ieee_application_id_dropped(app): + """IEEE-addressed GPDs are not supported by the zigpy GP stack yet. + + Rather than fabricate a best-effort SrcID, drop the frame with a + debug log so interoperability issues surface cleanly. + """ + args = _gp_args_v13() + args[3] = t.EmberGpAddress( + applicationId=t.uint8_t(2), + id=t.FixedList[t.uint8_t, 8](bytes(8)), + endpoint=t.uint8_t(1), + ) + + app.ezsp_callback_handler("gpepIncomingMessageHandler", args) + + assert app.packet_received.call_count == 0 + + +def test_gp_frame_handler_short_args_dropped(app): + """Malformed callbacks never crash the dispatcher.""" + app.ezsp_callback_handler("gpepIncomingMessageHandler", [0x00, 0x00]) + assert app.packet_received.call_count == 0 + + @pytest.mark.parametrize( "msg_type", ( From 91c7c016810c8649f2d8876cba238d5fae916289 Mon Sep 17 00:00:00 2001 From: nmingam <192914892+nmingam@users.noreply.github.com> Date: Fri, 24 Apr 2026 08:37:06 +0200 Subject: [PATCH 3/8] Inline GP constants and reformat to fix CI Two CI failures on this PR: 1. pytest on Python 3.14 (and all other versions) fails with `AttributeError: module 'zigpy.zgp.types' has no attribute 'GP_ENDPOINT'`. These constants (``GP_ENDPOINT``, ``GP_CLUSTER_ID``) are not exported by the current zigpy release -- they would be added by zigpy/zigpy#1814, which is still unreleased. Depending on them here introduces a hard coupling to an unreleased upstream. 2. pre-commit (black) rewrites the aligned hex-dump comments in the BJ6716U capture and the split `bellows.ezsp.v13.commands.COMMANDS` lookup. Fix both by: - Defining ``GP_ENDPOINT`` (242), ``GP_CLUSTER_ID`` (0x0021) and ``GP_PROFILE_ID`` (0xA1E0) as module-level constants in ``bellows.zigbee.application``, citing the ZGP 1.1b profile. These values are stable and do not need to travel across repos. - Dropping the now-unused ``zigpy.zgp.types`` import from ``tests/test_application.py`` and replacing the assertions with the same literals. - Running black to compact the inline comments so pre-commit is idempotent. No behaviour change. All 22 EZSP v13/v14/v16 tests pass against a venv with PyPI zigpy (which does not ship GP_ENDPOINT), confirming the fix removes the upstream dependency. --- bellows/zigbee/application.py | 15 +++++++--- tests/test_application.py | 7 ++--- tests/test_ezsp_v13.py | 56 +++++++++++++++++------------------ 3 files changed, 41 insertions(+), 37 deletions(-) diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py index 59e79643..b9c17c1a 100644 --- a/bellows/zigbee/application.py +++ b/bellows/zigbee/application.py @@ -78,6 +78,13 @@ DEFAULT_TX_POWER = 8 # dBm +# Zigbee Green Power spec constants — hardcoded rather than imported from +# ``zigpy.zgp`` so bellows does not depend on the upstream zigpy release +# shipping these symbols. These values are fixed by the ZGP 1.1b profile. +GP_ENDPOINT = 242 +GP_CLUSTER_ID = 0x0021 +GP_PROFILE_ID = 0xA1E0 + LIB_VERSION = importlib.metadata.version("bellows") LOGGER = logging.getLogger(__name__) @@ -798,15 +805,15 @@ def _handle_gp_frame(self, args: tuple) -> None: addr_mode=zigpy.types.AddrMode.NWK, address=zigpy.types.NWK(proxy_nwk), ), - src_ep=zigpy.types.uint8_t(zgp_t.GP_ENDPOINT), + src_ep=zigpy.types.uint8_t(GP_ENDPOINT), dst=zigpy.types.AddrModeAddress( addr_mode=zigpy.types.AddrMode.NWK, address=self.state.node_info.nwk, ), - dst_ep=zigpy.types.uint8_t(zgp_t.GP_ENDPOINT), + dst_ep=zigpy.types.uint8_t(GP_ENDPOINT), tsn=zigpy.types.uint8_t(int(sequence_number) & 0xFF), - profile_id=zigpy.types.uint16_t(0xA1E0), - cluster_id=zigpy.types.uint16_t(zgp_t.GP_CLUSTER_ID), + profile_id=zigpy.types.uint16_t(GP_PROFILE_ID), + cluster_id=zigpy.types.uint16_t(GP_CLUSTER_ID), data=zigpy.types.SerializableBytes(zcl_bytes), lqi=zigpy.types.uint8_t(lqi), rssi=zigpy.types.int8s(rssi), diff --git a/tests/test_application.py b/tests/test_application.py index 4830e930..c0347fd7 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -552,7 +552,6 @@ def test_frame_handler_ignored(app, aps_frame): # tests cover both the v13/v14 wire format (no trailing SlRxPacketInfo) # and the v16+ wire format (with trailer). from zigpy.zcl.clusters.greenpower import NotificationSchema # noqa: E402 -import zigpy.zgp.types as zgp_t # noqa: E402 def _gp_addr_srcid(source_id: int, endpoint: int = 0x00) -> t.EmberGpAddress: @@ -602,9 +601,9 @@ def test_gp_frame_handler_forwards_notification(app): assert app.packet_received.call_count == 1 packet = app.packet_received.mock_calls[0].args[0] - assert packet.dst_ep == zgp_t.GP_ENDPOINT - assert packet.src_ep == zgp_t.GP_ENDPOINT - assert packet.cluster_id == zgp_t.GP_CLUSTER_ID + assert packet.dst_ep == 242 # ZGP endpoint + assert packet.src_ep == 242 + assert packet.cluster_id == 0x0021 # Green Power cluster assert packet.profile_id == 0xA1E0 assert packet.dst.address == app.state.node_info.nwk # On v13/v14 there is no packetInfo, so bellows uses the coordinator diff --git a/tests/test_ezsp_v13.py b/tests/test_ezsp_v13.py index 03495956..a69917f9 100644 --- a/tests/test_ezsp_v13.py +++ b/tests/test_ezsp_v13.py @@ -241,29 +241,29 @@ async def test_factory_reset(ezsp_f) -> None: # Payload without the EZSP frame envelope, as passed to ``deserialize_dict``. BJ6716U_GPEP_PAYLOAD: bytes = bytes.fromhex( - "7ccdec" # status, gpdLink, sequenceNumber - "00" # EmberGpAddress.applicationId = 0 (SrcID) - "86f8710186f87101" # EmberGpAddress.id (8-byte union, - # sourceId = 0x0171F886 in first 4 LE) - "00" # EmberGpAddress.endpoint - "00" # gpdfSecurityLevel - "00" # gpdfSecurityKeyType - "00" # autoCommissioning - "00" # bidirectionalInfo - "ffffffff" # gpdSecurityFrameCounter (commissioning) - "e0" # gpdCommandId = 0xE0 (Commissioning) - "ffffffff" # mic - "ff" # proxyTableIndex - "2e" # LVBytes length = 46 - "02c5f2" # commissioning: deviceId, options, extOptions - "1ce9ae2f9e4f85f15de37c1ccbd94387" # encrypted GPD key (16 bytes) - "0013911a" # KeyMIC - "ec1d0000" # OutgoingCounter = 0x00001DEC - "04" # AppInfo - "11" # NumGPDCommands = 17 - "1011121314151617" # RecallScene 0-7 - "22" # Toggle - "6062636465666768" # Press/Release variants + "7ccdec" # status, gpdLink, sequenceNumber + "00" # EmberGpAddress.applicationId = 0 (SrcID) + "86f8710186f87101" # EmberGpAddress.id (8-byte union, + # sourceId = 0x0171F886 in first 4 LE) + "00" # EmberGpAddress.endpoint + "00" # gpdfSecurityLevel + "00" # gpdfSecurityKeyType + "00" # autoCommissioning + "00" # bidirectionalInfo + "ffffffff" # gpdSecurityFrameCounter (commissioning) + "e0" # gpdCommandId = 0xE0 (Commissioning) + "ffffffff" # mic + "ff" # proxyTableIndex + "2e" # LVBytes length = 46 + "02c5f2" # commissioning: deviceId, options, extOptions + "1ce9ae2f9e4f85f15de37c1ccbd94387" # encrypted GPD key (16 bytes) + "0013911a" # KeyMIC + "ec1d0000" # OutgoingCounter = 0x00001DEC + "04" # AppInfo + "11" # NumGPDCommands = 17 + "1011121314151617" # RecallScene 0-7 + "22" # Toggle + "6062636465666768" # Press/Release variants ) @@ -275,9 +275,7 @@ def test_gpep_incoming_real_frame_bj6716u(): buffer. The symptom was ``ValueError: Data is too short`` exactly as reported in zigpy/zigpy#1814. """ - _, _, rx_schema = bellows.ezsp.v13.commands.COMMANDS[ - "gpepIncomingMessageHandler" - ] + _, _, rx_schema = bellows.ezsp.v13.commands.COMMANDS["gpepIncomingMessageHandler"] result, rest = t.deserialize_dict(BJ6716U_GPEP_PAYLOAD, rx_schema) assert rest == b"" @@ -313,7 +311,7 @@ def test_gpep_incoming_via_frame_rx(ezsp_f): warning. """ envelope = ( - bytes([0x42, 0x00, 0x01]) # seq + control bytes + bytes([0x42, 0x00, 0x01]) # seq + control bytes + t.uint16_t(0x00C5).serialize() # frame_id LE + BJ6716U_GPEP_PAYLOAD ) @@ -324,6 +322,6 @@ def test_gpep_incoming_via_frame_rx(ezsp_f): assert ezsp_f._handle_callback.call_args[0][0] == "gpepIncomingMessageHandler" parsed = ezsp_f._handle_callback.call_args[0][1] # ``parsed`` is the list of values in schema order. - assert parsed[0] == 0x7C # status + assert parsed[0] == 0x7C # status assert parsed[3].source_id == 0x0171F886 # addr.source_id - assert parsed[9] == 0xE0 # gpdCommandId + assert parsed[9] == 0xE0 # gpdCommandId From 4b22bf3ebac38555fdb8fecd6814d0b76a59ea05 Mon Sep 17 00:00:00 2001 From: nmingam <192914892+nmingam@users.noreply.github.com> Date: Fri, 24 Apr 2026 22:47:05 +0200 Subject: [PATCH 4/8] Cover EmberGpAddress.gpd_ieee_address property codecov/patch flagged one uncovered line in bellows/types/struct.py (line 395, the gpd_ieee_address accessor). IEEE-addressed GPDFs are dropped upstream of this property in `_handle_gp_frame`, so no integration path reaches it. Add a direct unit test on the struct, plus a symmetric test for source_id so both accessor paths are exercised explicitly. --- tests/test_types.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/tests/test_types.py b/tests/test_types.py index a724d96a..57184210 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -28,3 +28,33 @@ def test_ember_node_type_to_zdo_logical_type(node_type, logical_type): node_type = t.EmberNodeType(node_type) assert node_type.zdo_logical_type == zdo_t.LogicalType(logical_type) + + +def test_ember_gp_address_source_id(): + """In SrcID mode (applicationId == 0) the first 4 bytes are the 32-bit ID. + + The remaining 4 bytes of the 8-byte union are padding per the GP spec. + """ + addr = t.EmberGpAddress( + applicationId=t.uint8_t(0), + id=t.FixedList[t.uint8_t, 8](b"\x86\xf8\x71\x01" + b"\x00" * 4), + endpoint=t.uint8_t(0), + ) + assert addr.source_id == 0x0171F886 + + +def test_ember_gp_address_gpd_ieee_address(): + """In IEEE mode (applicationId == 2) the full 8 bytes form the EUI64. + + Covers the ``gpd_ieee_address`` property used when the GP stack gets an + IEEE-addressed frame. bellows' dispatcher currently drops these frames + without touching the property, so this is the only place the accessor + is exercised. + """ + raw = b"\x11\x22\x33\x44\x55\x66\x77\x88" + addr = t.EmberGpAddress( + applicationId=t.uint8_t(2), + id=t.FixedList[t.uint8_t, 8](raw), + endpoint=t.uint8_t(3), + ) + assert addr.gpd_ieee_address == t.EUI64(raw) From f2578e88bc711591eb710dcfb02be8c8c0ba604b Mon Sep 17 00:00:00 2001 From: nmingam <192914892+nmingam@users.noreply.github.com> Date: Sun, 26 Apr 2026 19:23:47 +0200 Subject: [PATCH 5/8] bellows: drop EmberGpAddress helper accessors Read the 8-byte union directly at the call sites instead of going through the source_id / gpd_ieee_address properties. The class docstring also points out that the EZSP wire layout serializes applicationId first, even though the C SDK union puts id first. --- bellows/types/struct.py | 40 ++++++++--------------------------- bellows/zigbee/application.py | 9 ++++---- tests/test_ezsp_v13.py | 4 ++-- tests/test_ezsp_v14.py | 2 +- tests/test_ezsp_v16.py | 2 +- tests/test_types.py | 37 ++++++++------------------------ 6 files changed, 26 insertions(+), 68 deletions(-) diff --git a/bellows/types/struct.py b/bellows/types/struct.py index 68c61e20..a8b59ccd 100644 --- a/bellows/types/struct.py +++ b/bellows/types/struct.py @@ -357,43 +357,21 @@ class EmberTokTypeStackZllSecurity(EzspStruct): class EmberGpAddress(EzspStruct): - """A GP address structure. - - On the wire this is a 10-byte layout: - - - ``applicationId`` (1 byte) — 0 means ``SrcID`` mode, 2 means ``IEEE`` - mode. - - ``id`` (8 bytes) — a raw union. When ``applicationId == 0`` the source - ID lives in the first 4 bytes (little-endian) and the remaining 4 - bytes are padding. When ``applicationId == 2`` the full 8 bytes are - the GPD EUI64. - - ``endpoint`` (1 byte). - - The historical declaration here treated the union as two separate - fields (``gpdIeeeAddress`` + ``sourceId``) for a total of 14 bytes, - which does not match what the NCP sends in - ``gpepIncomingMessageHandler``. See the ``source_id`` and - ``gpd_ieee_address`` helpers below to access the right view. + """sl_zigbee_gp_address_t: GPD address used by the GP callbacks. + + The C SDK declares the union ``id`` field first, but the EZSP wire + layout for ``gpepIncomingMessageHandler`` serializes ``applicationId`` + first. The order below matches what the NCP actually sends; see the + real-frame test in ``tests/test_ezsp_v13.py``. """ - # The GPD Application ID: 0 = source ID mode, 2 = IEEE mode. applicationId: basic.uint8_t - # Raw 8-byte union. Use :attr:`source_id` or :attr:`gpd_ieee_address` - # to interpret it based on :attr:`applicationId`. + # 8-byte union. When applicationId == 0 (SrcID), the first 4 bytes + # hold the 32-bit source ID (little-endian) and the rest is padding; + # when applicationId == 2 (IEEE), the 8 bytes are the GPD EUI64. id: basic.FixedList[basic.uint8_t, 8] - # The GPD endpoint. endpoint: basic.uint8_t - @property - def source_id(self) -> int: - """Return the 32-bit source ID (only valid when applicationId == 0).""" - return int.from_bytes(bytes(self.id[:4]), "little") - - @property - def gpd_ieee_address(self) -> named.EUI64: - """Return the EUI64 (only valid when applicationId == 2).""" - return named.EUI64(bytes(self.id)) - class NV3StackTrustCenterToken(EzspStruct): """NV3 stack trust center token value.""" diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py index b9c17c1a..ae4576f2 100644 --- a/bellows/zigbee/application.py +++ b/bellows/zigbee/application.py @@ -750,16 +750,15 @@ def _handle_gp_frame(self, args: tuple) -> None: # coordinator short address (the coordinator acts as the proxy). packet_info = rest[0] if rest else None - if addr.applicationId == zgp_t.ApplicationID.SrcID: - source_id = addr.source_id - else: + if addr.applicationId != zgp_t.ApplicationID.SrcID: LOGGER.debug( - "GP frame with unsupported applicationId %s (only SrcID is " - "handled), dropping", + "GP frame with unsupported applicationId %s, dropping", addr.applicationId, ) return + source_id = int.from_bytes(bytes(addr.id[:4]), "little") + options = NotificationOptions( application_id=zgp_t.ApplicationID(addr.applicationId), also_unicast=0, diff --git a/tests/test_ezsp_v13.py b/tests/test_ezsp_v13.py index a69917f9..016fb688 100644 --- a/tests/test_ezsp_v13.py +++ b/tests/test_ezsp_v13.py @@ -287,7 +287,7 @@ def test_gpep_incoming_real_frame_bj6716u(): addr = result["addr"] assert addr.applicationId == 0 - assert addr.source_id == 0x0171F886 + assert int.from_bytes(bytes(addr.id[:4]), "little") == 0x0171F886 assert addr.endpoint == 0 assert result["gpdfSecurityLevel"] == 0 @@ -323,5 +323,5 @@ def test_gpep_incoming_via_frame_rx(ezsp_f): parsed = ezsp_f._handle_callback.call_args[0][1] # ``parsed`` is the list of values in schema order. assert parsed[0] == 0x7C # status - assert parsed[3].source_id == 0x0171F886 # addr.source_id + assert int.from_bytes(bytes(parsed[3].id[:4]), "little") == 0x0171F886 assert parsed[9] == 0xE0 # gpdCommandId diff --git a/tests/test_ezsp_v14.py b/tests/test_ezsp_v14.py index 21c26f11..02b39182 100644 --- a/tests/test_ezsp_v14.py +++ b/tests/test_ezsp_v14.py @@ -250,5 +250,5 @@ def test_gpep_incoming_inherits_v13_schema(ezsp_f): assert ezsp_f._handle_callback.call_count == 1 assert ezsp_f._handle_callback.call_args[0][0] == "gpepIncomingMessageHandler" parsed = ezsp_f._handle_callback.call_args[0][1] - assert parsed[3].source_id == 0x0171F886 + assert int.from_bytes(bytes(parsed[3].id[:4]), "little") == 0x0171F886 assert parsed[9] == 0xE0 # gpdCommandId diff --git a/tests/test_ezsp_v16.py b/tests/test_ezsp_v16.py index b12b4049..e31f079c 100644 --- a/tests/test_ezsp_v16.py +++ b/tests/test_ezsp_v16.py @@ -64,7 +64,7 @@ def test_gpep_incoming_v16_expects_trailing_packet_info(ezsp_f): assert ezsp_f._handle_callback.call_count == 1 assert ezsp_f._handle_callback.call_args[0][0] == "gpepIncomingMessageHandler" parsed = ezsp_f._handle_callback.call_args[0][1] - assert parsed[3].source_id == 0x0171F886 + assert int.from_bytes(bytes(parsed[3].id[:4]), "little") == 0x0171F886 assert parsed[9] == 0xE0 # gpdCommandId assert parsed[-1].sender_short_id == 0x1234 assert parsed[-1].last_hop_lqi == 200 diff --git a/tests/test_types.py b/tests/test_types.py index 57184210..8783865b 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -30,31 +30,12 @@ def test_ember_node_type_to_zdo_logical_type(node_type, logical_type): assert node_type.zdo_logical_type == zdo_t.LogicalType(logical_type) -def test_ember_gp_address_source_id(): - """In SrcID mode (applicationId == 0) the first 4 bytes are the 32-bit ID. - - The remaining 4 bytes of the 8-byte union are padding per the GP spec. - """ - addr = t.EmberGpAddress( - applicationId=t.uint8_t(0), - id=t.FixedList[t.uint8_t, 8](b"\x86\xf8\x71\x01" + b"\x00" * 4), - endpoint=t.uint8_t(0), - ) - assert addr.source_id == 0x0171F886 - - -def test_ember_gp_address_gpd_ieee_address(): - """In IEEE mode (applicationId == 2) the full 8 bytes form the EUI64. - - Covers the ``gpd_ieee_address`` property used when the GP stack gets an - IEEE-addressed frame. bellows' dispatcher currently drops these frames - without touching the property, so this is the only place the accessor - is exercised. - """ - raw = b"\x11\x22\x33\x44\x55\x66\x77\x88" - addr = t.EmberGpAddress( - applicationId=t.uint8_t(2), - id=t.FixedList[t.uint8_t, 8](raw), - endpoint=t.uint8_t(3), - ) - assert addr.gpd_ieee_address == t.EUI64(raw) +def test_ember_gp_address_roundtrip(): + """Wire layout: applicationId(1) + id(8) + endpoint(1).""" + raw = b"\x00" + b"\x86\xf8\x71\x01" + bytes(4) + b"\x00" + addr, rest = t.EmberGpAddress.deserialize(raw) + assert rest == b"" + assert addr.applicationId == 0 + assert addr.endpoint == 0 + assert bytes(addr.id) == b"\x86\xf8\x71\x01" + bytes(4) + assert addr.serialize() == raw From 4bc2d5072f47edc359dc9f0e0eb2fbcbed6af657 Mon Sep 17 00:00:00 2001 From: nmingam <192914892+nmingam@users.noreply.github.com> Date: Sun, 26 Apr 2026 19:33:18 +0200 Subject: [PATCH 6/8] bellows: trim comments on gpepIncomingMessageHandler overrides --- bellows/ezsp/v13/commands.py | 20 +++----------------- bellows/ezsp/v16/commands.py | 7 +------ 2 files changed, 4 insertions(+), 23 deletions(-) diff --git a/bellows/ezsp/v13/commands.py b/bellows/ezsp/v13/commands.py index eda2a280..a67e129c 100644 --- a/bellows/ezsp/v13/commands.py +++ b/bellows/ezsp/v13/commands.py @@ -90,23 +90,9 @@ "status": t.sl_Status, }, ), - # Redefine the Green Power incoming callback so real GPDFs from a - # Friends of Hue class switch actually parse. The v4 schema used - # scattered address fields (``addrType`` + ``addr:uint32`` + - # ``applicationId`` + ``address:EUI64`` + ``endpoint``) that do not - # match the 10-byte ``EmberGpAddress`` the NCP actually sends. - # - # ``status`` is kept as a plain ``uint8_t`` on purpose. The strict - # ``sl_GpStatus`` enum only accepts 0x00..0x07, but firmware versions - # shipped with current ZBT-1 / SkyConnect sticks return higher status - # bytes (e.g. 0x7C observed from a Busch-Jaeger 6716 U switch) when - # the frame was not matched against a proxy/sink table entry. Using a - # plain ``uint8_t`` lets the frame reach the host so zigpy can decide - # what to do with it. - # - # v14 and v16 inherit this override through the ``_REPLACEMENTS`` - # mechanism in ``bellows/ezsp/v14/commands.py``; v17 and later keep - # their own schema which already uses ``EmberGpAddress``. + # status is kept as uint8_t (not sl_GpStatus) to accept values + # outside the strict 0x00..0x07 range that NCPs return for frames + # with no matching proxy/sink entry. "gpepIncomingMessageHandler": ( 0x00C5, {}, diff --git a/bellows/ezsp/v16/commands.py b/bellows/ezsp/v16/commands.py index 92ef7a4d..4466fb0d 100644 --- a/bellows/ezsp/v16/commands.py +++ b/bellows/ezsp/v16/commands.py @@ -4,12 +4,7 @@ COMMANDS = { **COMMANDS_v14, - # EZSP v16 appends an ``EmberRxPacketInfo`` struct after the LVBytes - # payload in ``gpepIncomingMessageHandler``. zigbee-herdsman gates the - # read on ``version >= 0x10`` (see - # ``src/adapter/ember/ezsp/ezsp.ts`` in the ``ember`` adapter), so v13 - # and v14 do not carry this trailer but v16 does. v17 and v18 override - # the whole command again, so this entry only matters for v16. + # v16 appends an SlRxPacketInfo trailer to the v13/v14 layout. "gpepIncomingMessageHandler": ( 0x00C5, {}, From 2df0ea51232a262bb34f3cd4df35167423ae6578 Mon Sep 17 00:00:00 2001 From: nmingam <192914892+nmingam@users.noreply.github.com> Date: Sun, 26 Apr 2026 19:58:54 +0200 Subject: [PATCH 7/8] bellows: simplify GP frame forwarding Build the ZCL header through foundation.ZCLHeader.cluster instead of a magic byte literal, drop the verbose docstring, and bail out on EZSP < v13 callbacks where the 4th argument is a legacy uint8_t addrType instead of an EmberGpAddress. --- bellows/zigbee/application.py | 53 +++++++++++++++-------------------- tests/test_application.py | 10 +++++++ 2 files changed, 32 insertions(+), 31 deletions(-) diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py index ae4576f2..5a1559ff 100644 --- a/bellows/zigbee/application.py +++ b/bellows/zigbee/application.py @@ -21,6 +21,7 @@ import zigpy.state import zigpy.types import zigpy.util +from zigpy.zcl import foundation from zigpy.zcl.clusters.greenpower import NotificationOptions, NotificationSchema import zigpy.zdo.types as zdo_t import zigpy.zgp.types as zgp_t @@ -78,9 +79,8 @@ DEFAULT_TX_POWER = 8 # dBm -# Zigbee Green Power spec constants — hardcoded rather than imported from -# ``zigpy.zgp`` so bellows does not depend on the upstream zigpy release -# shipping these symbols. These values are fixed by the ZGP 1.1b profile. +# TODO: replace with zigpy.zgp.GP_ENDPOINT / GP_CLUSTER_ID / GP_PROFILE_ID +# once those are released upstream. GP_ENDPOINT = 242 GP_CLUSTER_ID = 0x0021 GP_PROFILE_ID = 0xA1E0 @@ -710,20 +710,7 @@ def ezsp_callback_handler(self, frame_name, args): self._handle_gp_frame(args) def _handle_gp_frame(self, args: tuple) -> None: - """Forward a decoded gpepIncomingMessageHandler callback to zigpy. - - The NCP has already done the GP stub processing (proxy-table lookup, - duplicate suppression, etc.) and hands us a flattened GPDF. zigpy's - GP manager on the other side expects a ZigbeePacket on endpoint 242, - cluster 0x0021, carrying a ZCL GP Notification (server command 0x00) - so it can route commissioning, data, and decommissioning frames - through the same entry point. - - We therefore rebuild that envelope here, mirroring what zigbee- - herdsman's ember adapter does. Any GPDF that cannot be converted - (for example an IEEE-addressed frame, which the current zigpy GP - stack explicitly does not support) is logged and dropped. - """ + """Forward gpepIncomingMessageHandler as a ZCL GP Notification.""" if len(args) < 13: LOGGER.debug("gpepIncomingMessageHandler: short args %r, dropping", args) return @@ -745,10 +732,16 @@ def _handle_gp_frame(self, args: tuple) -> None: *rest, ) = args - # v16+ appends an SlRxPacketInfo after the payload. Pull the - # proxy NWK out of it if it is there, otherwise fall back to the - # coordinator short address (the coordinator acts as the proxy). - packet_info = rest[0] if rest else None + # On EZSP < v13 the 4th argument is a uint8_t addrType, not an + # EmberGpAddress. Drop those frames rather than crashing on + # attribute access; bellows does not currently parse the legacy + # layout. + if not isinstance(addr, t.EmberGpAddress): + LOGGER.debug( + "gpepIncomingMessageHandler: unsupported legacy address layout, " + "dropping" + ) + return if addr.applicationId != zgp_t.ApplicationID.SrcID: LOGGER.debug( @@ -758,6 +751,8 @@ def _handle_gp_frame(self, args: tuple) -> None: return source_id = int.from_bytes(bytes(addr.id[:4]), "little") + # v16+ appends an SlRxPacketInfo after the payload. + packet_info = rest[0] if rest else None options = NotificationOptions( application_id=zgp_t.ApplicationID(addr.applicationId), @@ -770,7 +765,6 @@ def _handle_gp_frame(self, args: tuple) -> None: tx_queue_full=0, _reserved=0, ) - notification = NotificationSchema( options=options, gpd_id=zgp_t.DeviceID(source_id), @@ -778,20 +772,17 @@ def _handle_gp_frame(self, args: tuple) -> None: command_id=zigpy.types.uint8_t(gpd_command_id), payload=zigpy.types.LVBytes(bytes(gpd_command_payload)), ) - - # ZCL header: frame_control=0x01 (cluster-specific, client→server, - # not manufacturer-specific, no default response disabled), TSN is - # the GPD MAC sequence number so duplicates can be tracked, and the - # command is GP Notification (0x00). - zcl_bytes = ( - bytes([0x01, int(sequence_number) & 0xFF, 0x00]) + notification.serialize() - ) + tsn = zigpy.types.uint8_t(int(sequence_number) & 0xFF) + zcl_header = foundation.ZCLHeader.cluster(tsn=tsn, command_id=0x00) + zcl_bytes = zcl_header.serialize() + notification.serialize() if packet_info is not None: proxy_nwk = int(packet_info.sender_short_id) lqi = int(packet_info.last_hop_lqi) rssi = int(packet_info.last_hop_rssi) else: + # No SlRxPacketInfo on v13/v14: the coordinator stands in as + # the proxy and gpdLink is the only signal-quality byte. proxy_nwk = int(self.state.node_info.nwk) lqi = int(gpd_link) rssi = 0 @@ -810,7 +801,7 @@ def _handle_gp_frame(self, args: tuple) -> None: address=self.state.node_info.nwk, ), dst_ep=zigpy.types.uint8_t(GP_ENDPOINT), - tsn=zigpy.types.uint8_t(int(sequence_number) & 0xFF), + tsn=tsn, profile_id=zigpy.types.uint16_t(GP_PROFILE_ID), cluster_id=zigpy.types.uint16_t(GP_CLUSTER_ID), data=zigpy.types.SerializableBytes(zcl_bytes), diff --git a/tests/test_application.py b/tests/test_application.py index c0347fd7..d910d4f7 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -690,6 +690,16 @@ def test_gp_frame_handler_short_args_dropped(app): assert app.packet_received.call_count == 0 +def test_gp_frame_handler_legacy_address_layout_dropped(app): + """EZSP < v13 delivers a uint8_t addrType, not an EmberGpAddress.""" + args = _gp_args_v13() + args[3] = t.uint8_t(0) # legacy addrType in place of EmberGpAddress + + app.ezsp_callback_handler("gpepIncomingMessageHandler", args) + + assert app.packet_received.call_count == 0 + + @pytest.mark.parametrize( "msg_type", ( From af28f20d9e2c44e697bb258623449bff8eb1eefb Mon Sep 17 00:00:00 2001 From: nmingam <192914892+nmingam@users.noreply.github.com> Date: Sun, 26 Apr 2026 20:13:32 +0200 Subject: [PATCH 8/8] bellows: assert GP forwarding via packet equality Build the expected ZigbeePacket up front and compare with == instead of checking each field one by one. --- tests/test_application.py | 154 +++++++++++++++++++++++++------------- 1 file changed, 101 insertions(+), 53 deletions(-) diff --git a/tests/test_application.py b/tests/test_application.py index d910d4f7..21f7bddd 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -541,21 +541,15 @@ def test_frame_handler_ignored(app, aps_frame): assert app.packet_received.call_count == 0 -# --------------------------------------------------------------------------- -# gpepIncomingMessageHandler forwarding -# --------------------------------------------------------------------------- -# -# Expected behaviour: the NCP delivers a flattened GPDF via -# ``gpepIncomingMessageHandler``. bellows must reencapsulate it as a ZCL -# GP Notification on endpoint 242, cluster 0x0021, so zigpy's GreenPower -# manager (listening through ``packet_received``) can route it. These -# tests cover both the v13/v14 wire format (no trailing SlRxPacketInfo) -# and the v16+ wire format (with trailer). -from zigpy.zcl.clusters.greenpower import NotificationSchema # noqa: E402 +from zigpy.zcl import foundation # noqa: E402 +from zigpy.zcl.clusters.greenpower import ( # noqa: E402 + NotificationOptions, + NotificationSchema, +) +import zigpy.zgp.types as zgp_t # noqa: E402 def _gp_addr_srcid(source_id: int, endpoint: int = 0x00) -> t.EmberGpAddress: - """Build a 10-byte EmberGpAddress for a SrcID-mode GPD.""" id_bytes = source_id.to_bytes(4, "little") + bytes(4) return t.EmberGpAddress( applicationId=t.uint8_t(0), @@ -594,33 +588,78 @@ def _gp_args_v13( ] -def test_gp_frame_handler_forwards_notification(app): - """A GPDF is forwarded to zigpy as a ZCL GP Notification.""" - app.ezsp_callback_handler("gpepIncomingMessageHandler", _gp_args_v13()) +def _expected_gp_packet( + *, + src_nwk, + dst_nwk, + source_id: int, + sequence_number: int, + command_id: int, + payload: bytes, + frame_counter: int, + lqi: int, + rssi: int, + security_level: int = 0, + security_key_type: int = 0, +) -> zigpy.types.ZigbeePacket: + options = NotificationOptions( + application_id=zgp_t.ApplicationID.SrcID, + also_unicast=0, + also_derived_group=0, + also_commissioned_group=0, + security_level=zgp_t.SecurityLevel(security_level), + security_key_type=zgp_t.SecurityKeyType(security_key_type), + appoint_temp_master=0, + tx_queue_full=0, + _reserved=0, + ) + notification = NotificationSchema( + options=options, + gpd_id=zgp_t.DeviceID(source_id), + frame_counter=zigpy.types.uint32_t(frame_counter), + command_id=zigpy.types.uint8_t(command_id), + payload=zigpy.types.LVBytes(payload), + ) + tsn = zigpy.types.uint8_t(sequence_number & 0xFF) + zcl_bytes = ( + foundation.ZCLHeader.cluster(tsn=tsn, command_id=0x00).serialize() + + notification.serialize() + ) + return zigpy.types.ZigbeePacket( + src=zigpy.types.AddrModeAddress( + addr_mode=zigpy.types.AddrMode.NWK, + address=zigpy.types.NWK(src_nwk), + ), + src_ep=zigpy.types.uint8_t(242), + dst=zigpy.types.AddrModeAddress( + addr_mode=zigpy.types.AddrMode.NWK, address=dst_nwk + ), + dst_ep=zigpy.types.uint8_t(242), + tsn=tsn, + profile_id=zigpy.types.uint16_t(0xA1E0), + cluster_id=zigpy.types.uint16_t(0x0021), + data=zigpy.types.SerializableBytes(zcl_bytes), + lqi=zigpy.types.uint8_t(lqi), + rssi=zigpy.types.int8s(rssi), + ) - assert app.packet_received.call_count == 1 - packet = app.packet_received.mock_calls[0].args[0] - assert packet.dst_ep == 242 # ZGP endpoint - assert packet.src_ep == 242 - assert packet.cluster_id == 0x0021 # Green Power cluster - assert packet.profile_id == 0xA1E0 - assert packet.dst.address == app.state.node_info.nwk - # On v13/v14 there is no packetInfo, so bellows uses the coordinator - # short address as a stand-in for the proxy NWK. - assert packet.src.address == app.state.node_info.nwk - - # ZCL header: 0x01 (cluster-specific, client→server) + TSN + cmd 0x00 - raw = packet.data.serialize() - assert raw[0] == 0x01 - assert raw[2] == 0x00 # GP Notification - notification = NotificationSchema.deserialize(raw[3:])[0] - assert int(notification.gpd_id) == 0x0171F886 - assert int(notification.frame_counter) == 0xFFFFFFFF - assert int(notification.command_id) == 0xE0 - # Commissioning payload is carried verbatim so zigpy can parse it - assert bytes(notification.payload) == b"\x02\xc5\xf2" + bytes(43) +def test_gp_frame_handler_forwards_notification(app): + """A GPDF is forwarded as the exact ZCL GP Notification packet.""" + app.ezsp_callback_handler("gpepIncomingMessageHandler", _gp_args_v13()) + expected = _expected_gp_packet( + src_nwk=int(app.state.node_info.nwk), + dst_nwk=app.state.node_info.nwk, + source_id=0x0171F886, + sequence_number=0xEC, + command_id=0xE0, + payload=b"\x02\xc5\xf2" + bytes(43), + frame_counter=0xFFFFFFFF, + lqi=0xCD, + rssi=0, + ) + assert app.packet_received.mock_calls == [call(expected)] assert ( app.state.counters[bellows.zigbee.application.COUNTERS_CTRL][ bellows.zigbee.application.COUNTER_RX_GP @@ -644,34 +683,43 @@ def test_gp_frame_handler_v16_uses_packet_info(app): app.ezsp_callback_handler("gpepIncomingMessageHandler", args) - assert app.packet_received.call_count == 1 - packet = app.packet_received.mock_calls[0].args[0] - assert packet.src.address == 0x1234 - assert packet.lqi == 210 - assert packet.rssi == -40 + expected = _expected_gp_packet( + src_nwk=0x1234, + dst_nwk=app.state.node_info.nwk, + source_id=0x0171F886, + sequence_number=0xEC, + command_id=0xE0, + payload=b"\x02\xc5\xf2" + bytes(43), + frame_counter=0xFFFFFFFF, + lqi=210, + rssi=-40, + ) + assert app.packet_received.mock_calls == [call(expected)] def test_gp_frame_handler_data_command_toggle(app): - """An operational data command (e.g. Toggle) is forwarded the same way.""" + """A non-commissioning command (Toggle) is forwarded the same way.""" app.ezsp_callback_handler( "gpepIncomingMessageHandler", _gp_args_v13(command_id=0x22, payload=b"", frame_counter=42), ) - assert app.packet_received.call_count == 1 - packet = app.packet_received.mock_calls[0].args[0] - notification = NotificationSchema.deserialize(packet.data.serialize()[3:])[0] - assert int(notification.command_id) == 0x22 - assert int(notification.frame_counter) == 42 - assert bytes(notification.payload) == b"" + expected = _expected_gp_packet( + src_nwk=int(app.state.node_info.nwk), + dst_nwk=app.state.node_info.nwk, + source_id=0x0171F886, + sequence_number=0xEC, + command_id=0x22, + payload=b"", + frame_counter=42, + lqi=0xCD, + rssi=0, + ) + assert app.packet_received.mock_calls == [call(expected)] def test_gp_frame_handler_ieee_application_id_dropped(app): - """IEEE-addressed GPDs are not supported by the zigpy GP stack yet. - - Rather than fabricate a best-effort SrcID, drop the frame with a - debug log so interoperability issues surface cleanly. - """ + """IEEE-addressed GPDs are dropped: the zigpy GP stack only handles SrcID.""" args = _gp_args_v13() args[3] = t.EmberGpAddress( applicationId=t.uint8_t(2),