| # | |
| # Copyright (c) 2025 Project CHIP Authors | |
| # All rights reserved. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| # | |
| import ipaddress | |
| import logging | |
| from dataclasses import dataclass | |
| from cryptography.hazmat.primitives import hashes | |
| from cryptography.hazmat.primitives.kdf.hkdf import HKDF | |
| from mobly import asserts | |
| import matter.clusters as Clusters | |
| from matter.ChipDeviceCtrl import ChipDeviceController | |
| from matter.clusters.Types import NullValue | |
| from matter.testing.matter_testing import AttributeMatcher | |
| from matter.testing.spec_parsing import build_xml_clusters, dm_from_spec_version | |
| logger = logging.getLogger(__name__) | |
| GROUP_EPOCH_KEY_LENGTH_BYTES = 16 | |
| _GROUP_KEY_V1_INFO = b"GroupKey v1.0" | |
| _GROUP_KEY_HASH_INFO = b"GroupKeyHash" | |
| def derive_group_operational_key(epoch_key: bytes, compressed_fabric_id: bytes) -> bytes: | |
| """Derive the operational group encryption key from an epoch key and compressed fabric ID.""" | |
| asserts.assert_equal(len(epoch_key), GROUP_EPOCH_KEY_LENGTH_BYTES, | |
| "Epoch key must be 16 bytes") | |
| asserts.assert_equal(len(compressed_fabric_id), 8, | |
| "Compressed fabric ID must be 8 bytes") | |
| return HKDF( | |
| algorithm=hashes.SHA256(), | |
| length=GROUP_EPOCH_KEY_LENGTH_BYTES, | |
| salt=compressed_fabric_id, | |
| info=_GROUP_KEY_V1_INFO, | |
| ).derive(epoch_key) | |
| def derive_group_session_id(epoch_key: bytes, compressed_fabric_id: bytes) -> int: | |
| """Derive the 16-bit Group Session ID for an epoch key on a fabric.""" | |
| operational_key = derive_group_operational_key(epoch_key, compressed_fabric_id) | |
| session_id_bytes = HKDF( | |
| algorithm=hashes.SHA256(), | |
| length=2, | |
| salt=b"", | |
| info=_GROUP_KEY_HASH_INFO, | |
| ).derive(operational_key) | |
| return int.from_bytes(session_id_bytes, byteorder="big") | |
| def find_colliding_epoch_key(reference_epoch_key: bytes, compressed_fabric_id: bytes, | |
| max_attempts: int = 65536 * 10) -> bytes: | |
| """Find an epoch key which lead to a Group Session ID collision with reference_epoch_key on the same fabric. | |
| Group Session IDs are 16 bits wide, so a collision is expected after around 2^16 (65536) attempts. | |
| Limit the number of attempts to avoid infinite loop or long execution time. | |
| 10 times the expected convergence attempts should provide less than 0.005% chance of not finding a collision. | |
| """ | |
| target_session_id = derive_group_session_id(reference_epoch_key, compressed_fabric_id) | |
| for attempt in range(1, max_attempts + 1): | |
| candidate = attempt.to_bytes(GROUP_EPOCH_KEY_LENGTH_BYTES, byteorder="big") | |
| if candidate == reference_epoch_key: | |
| continue | |
| if derive_group_session_id(candidate, compressed_fabric_id) == target_session_id: | |
| # found an epoch key whose Group Session ID collides with the one of reference_epoch_key | |
| return candidate | |
| raise ValueError("Could not find a colliding key") | |
| def group_id_from_node_id(node_id: int) -> int: | |
| """Extracts the 16-bit Group ID from a Group-scoped Node ID.""" | |
| return node_id & 0xFFFF | |
| def get_auxiliary_acl_equivalence_set(aux_acl, parts_list) -> set[tuple[int, int, int]]: | |
| """Expands AuxiliaryACL entries into a set of (fabric_index, group_id, endpoint_id) tuples. | |
| This implements the equivalence class logic for verifying auxiliary entries, accounting | |
| for various encodings and wildcards (empty target lists). It also strictly validates | |
| that Groupcast auxiliary entries have the correct privilege and auth mode. | |
| Args: | |
| aux_acl: The list of AuxiliaryACL entries read from the DUT. | |
| parts_list: The list of endpoints from the Root Node's Descriptor PartsList attribute. | |
| Returns: | |
| A set of (fabric_index, group_id, endpoint_id) tuples representing the granted access. | |
| """ | |
| equivalence_set = set() | |
| for entry in aux_acl: | |
| # We only process Groupcast auxiliary entries. | |
| if entry.auxiliaryType != Clusters.AccessControl.Enums.AccessControlAuxiliaryTypeEnum.kGroupcast: | |
| continue | |
| # Strictly validate metadata for Groupcast auxiliary entries. | |
| asserts.assert_equal(entry.privilege, Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kOperate, | |
| f"Groupcast auxiliary entry MUST have Operate privilege, but has {entry.privilege}") | |
| asserts.assert_equal(entry.authMode, Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kGroup, | |
| f"Groupcast auxiliary entry MUST have Group auth mode, but has {entry.authMode}") | |
| subjects = entry.subjects if (entry.subjects is not None and entry.subjects is not NullValue) else [] | |
| targets = entry.targets if (entry.targets is not None and entry.targets is not NullValue) else [] | |
| for subject in subjects: | |
| group_id = group_id_from_node_id(subject) | |
| if not targets: | |
| # Wildcard: empty target list represents all endpoints in the parts list (excluding root). | |
| for endpoint_id in parts_list: | |
| if endpoint_id != 0: | |
| equivalence_set.add((entry.fabricIndex, group_id, endpoint_id)) | |
| else: | |
| for target in targets: | |
| endpoint_id = target.endpoint | |
| if endpoint_id is None or endpoint_id is NullValue: | |
| # Wildcard target: applies to all endpoints in the parts list (excluding root). | |
| for ep in parts_list: | |
| if ep != 0: | |
| equivalence_set.add((entry.fabricIndex, group_id, ep)) | |
| else: | |
| equivalence_set.add((entry.fabricIndex, group_id, endpoint_id)) | |
| return equivalence_set | |
| async def get_feature_map(test) -> tuple: | |
| """Get supported features.""" | |
| feature_map = await test.read_single_attribute_check_success( | |
| cluster=Clusters.Groupcast, | |
| attribute=Clusters.Groupcast.Attributes.FeatureMap, | |
| endpoint=0 | |
| ) | |
| ln_enabled = bool(feature_map & Clusters.Groupcast.Bitmaps.Feature.kListener) | |
| sd_enabled = bool(feature_map & Clusters.Groupcast.Bitmaps.Feature.kSender) | |
| pga_enabled = bool(feature_map & Clusters.Groupcast.Bitmaps.Feature.kPerGroup) | |
| asserts.assert_true(sd_enabled or ln_enabled, | |
| "At least one of the following features must be enabled: Listener or Sender.") | |
| logger.info( | |
| "FeatureMap: %s : LN supported: %s | SD supported: %s | PGA supported: %s", feature_map, ln_enabled, sd_enabled, pga_enabled) | |
| return ln_enabled, sd_enabled, pga_enabled | |
| async def valid_endpoints_list(test, ln_enabled: bool) -> list: | |
| """Get the JoinGroup cmd endpoints list based on enabled features such as Listener/Sender. | |
| For Senders: return empty list. | |
| For Listeners: return the list of non-root endpoints that have at least one cluster exposing | |
| one command requiring Operate privilege. | |
| """ | |
| endpoints_list: list = [] | |
| if ln_enabled: | |
| operate_only_commands_dict = await get_operate_only_commands( | |
| test.default_controller, test.dut_node_id, exclude_ep0=True) | |
| endpoints_list = sorted(operate_only_commands_dict.keys()) | |
| logger.info( | |
| "Endpoints with at least one Operate-privilege command: %s", endpoints_list) | |
| asserts.assert_greater(len(endpoints_list), 0, | |
| "Listener feature is enabled. Endpoint list should not be empty. There should be a valid endpoint for the GroupCast JoinGroup Command.") | |
| return endpoints_list | |
| async def is_groupcast_on_root_node(test) -> bool: | |
| """Check if Groupcast cluster is present on the RootNode endpoint (EP0).""" | |
| server_list = await test.read_single_attribute_check_success( | |
| cluster=Clusters.Descriptor, | |
| attribute=Clusters.Descriptor.Attributes.ServerList, | |
| endpoint=0) | |
| return Clusters.Groupcast.id in server_list | |
| async def is_groupcast_adopted(test, groups_endpoint: int) -> bool: | |
| """Return whether Groupcast is adopted for Groups/GroupKeyManagement behavior. | |
| Adoption requires the GroupKeyManagement Groupcast feature bit and Groups | |
| ClusterRevision >= 5 on the given endpoint. | |
| """ | |
| grpkey_feature_map = await test.read_single_attribute_check_success( | |
| cluster=Clusters.GroupKeyManagement, | |
| attribute=Clusters.GroupKeyManagement.Attributes.FeatureMap, | |
| endpoint=0) | |
| grpkey_groupcast_enabled = (grpkey_feature_map & Clusters.GroupKeyManagement.Bitmaps.Feature.kGroupcast) != 0 | |
| groups_cluster_revision = await test.read_single_attribute_check_success( | |
| endpoint=groups_endpoint, | |
| cluster=Clusters.Groups, | |
| attribute=Clusters.Groups.Attributes.ClusterRevision) | |
| return grpkey_groupcast_enabled and groups_cluster_revision >= 5 | |
| def generate_membership_entry_matcher( | |
| group_id: int, | |
| key_set_id: int | None = None, | |
| has_auxiliary_acl: bool | None = None, | |
| endpoints: list | None = None, | |
| mcastAddrPolicy: Clusters.Groupcast.Enums.MulticastAddrPolicyEnum | None = None, | |
| test_for_exists: bool = True, | |
| ) -> AttributeMatcher: | |
| """Create a matcher that checks if Membership attribute contains (or does not contain) an entry matching the specified criteria. | |
| Args: | |
| group_id: The groupID to match (required) | |
| key_set_id: The keySetID to match (optional) | |
| has_auxiliary_acl: The HasAuxiliaryACL value to match (optional) | |
| endpoints: The endpoints list to match (optional) | |
| mcastAddrPolicy: The multicast address policy to match (optional) | |
| test_for_exists: If True, membership entry exists. (default: True) | |
| Returns: | |
| An AttributeMatcher that returns True when: | |
| - test_for_exists=True: A Membership entry matches all specified criteria | |
| - test_for_exists=False: No Membership entry matches the specified criteria | |
| """ | |
| def predicate(report) -> bool: | |
| if report.attribute != Clusters.Groupcast.Attributes.Membership: | |
| return False | |
| found_match = False | |
| for entry in report.value: | |
| if entry.groupID != group_id: | |
| continue | |
| if key_set_id is not None and entry.keySetID != key_set_id: | |
| continue | |
| if has_auxiliary_acl is not None: | |
| if entry.hasAuxiliaryACL is None or entry.hasAuxiliaryACL != has_auxiliary_acl: | |
| continue | |
| if endpoints is not None: | |
| if entry.endpoints is None or entry.endpoints != endpoints: | |
| continue | |
| if mcastAddrPolicy is not None: | |
| if entry.mcastAddrPolicy is None or entry.mcastAddrPolicy != mcastAddrPolicy: | |
| continue | |
| found_match = True | |
| break | |
| return found_match if test_for_exists else not found_match | |
| desc_parts = [f"groupID={group_id}"] | |
| if key_set_id is not None: | |
| desc_parts.append(f"keySetID={key_set_id}") | |
| if has_auxiliary_acl is not None: | |
| desc_parts.append(f"hasAuxiliaryACL={has_auxiliary_acl}") | |
| if endpoints is not None: | |
| desc_parts.append(f"endpoints={endpoints}") | |
| if mcastAddrPolicy is not None: | |
| desc_parts.append(f"mcastAddrPolicy={mcastAddrPolicy}") | |
| if test_for_exists: | |
| description = f"Membership has entry with {', '.join(desc_parts)}" | |
| else: | |
| description = f"Membership does NOT have entry with {', '.join(desc_parts)}" | |
| return AttributeMatcher.from_callable(description=description, matcher=predicate) | |
| def generate_membership_empty_matcher() -> AttributeMatcher: | |
| """Create a matcher that checks if Membership attribute is empty (no groups present). | |
| Returns: | |
| An AttributeMatcher that returns True when the Membership list is empty. | |
| """ | |
| def predicate(report) -> bool: | |
| if report.attribute != Clusters.Groupcast.Attributes.Membership: | |
| return False | |
| return len(report.value) == 0 | |
| description = "Membership list is empty (no groups present)" | |
| return AttributeMatcher.from_callable(description=description, matcher=predicate) | |
| def generate_fabric_under_test_matcher(expected_fabric_index: int) -> AttributeMatcher: | |
| """Create a matcher that checks if FabricUnderTest attribute has the expected value. | |
| Args: | |
| expected_fabric_index: The expected fabric index value. | |
| Returns: | |
| An AttributeMatcher that returns True when FabricUnderTest equals the expected value. | |
| """ | |
| def predicate(report) -> bool: | |
| if report.attribute != Clusters.Groupcast.Attributes.FabricUnderTest: | |
| return False | |
| return report.value == expected_fabric_index | |
| description = f"FabricUnderTest == {expected_fabric_index}" | |
| return AttributeMatcher.from_callable(description=description, matcher=predicate) | |
| def generate_usedMcastAddrCount_entry_matcher(expected_count: int) -> AttributeMatcher: | |
| """Create a matcher that checks if UsedMcastAddrCount attribute has the expected value. | |
| Args: | |
| expected_count: The expected UsedMcastAddrCount value. | |
| Returns: | |
| An AttributeMatcher that returns True when UsedMcastAddrCount equals the expected value. | |
| """ | |
| def predicate(report) -> bool: | |
| if report.attribute != Clusters.Groupcast.Attributes.UsedMcastAddrCount: | |
| return False | |
| return report.value == expected_count | |
| description = f"UsedMcastAddrCount == {expected_count}" | |
| return AttributeMatcher.from_callable(description=description, matcher=predicate) | |
| @dataclass | |
| class OperateOnlyCommand: | |
| cluster_object: Clusters.ClusterObjects.Cluster | |
| command_object: Clusters.ClusterObjects.ClusterCommand | |
| async def get_operate_only_commands(dev_ctrl: ChipDeviceController, node_id: int, exclude_ep0: bool = True, endpoint_id_to_search: int | None = None) -> dict[int, list[OperateOnlyCommand]]: | |
| """ | |
| Reads all AcceptedCommandList attributes and the SpecificationVersion to determine all | |
| commands that only require Operate privilege. | |
| Args: | |
| dev_ctrl: The ChipDeviceController instance. | |
| node_id: The node ID of the device to query. | |
| exclude_ep0: Boolean to determine if endpoint 0 should be excluded in the search for valid cluster commands | |
| endpoint_id_to_search: Optional argument. When specified, search for commands will only be on clusters on the specified endpoint. Search all endpoints if not specified | |
| Returns: | |
| A list of OperateOnlyCommand dataclass objects for each command that only requires | |
| Operate privilege. | |
| """ | |
| # Helper function to perform wildcard read and get spec info | |
| async def get_device_composition_and_spec(dev_ctrl, node_id) -> tuple[dict, int]: | |
| wildcard_read = await dev_ctrl.Read(node_id, [()]) | |
| attributes = wildcard_read.attributes | |
| spec_version = attributes[0][Clusters.BasicInformation][Clusters.BasicInformation.Attributes.SpecificationVersion] | |
| return attributes, spec_version | |
| # Helper function to parse spec | |
| def get_xml_clusters(spec_version: int): | |
| dm = dm_from_spec_version(spec_version) | |
| xml_clusters, _ = build_xml_clusters(dm) | |
| return xml_clusters | |
| def find_commands_on_endpoint_and_cluster(endpoint_id, endpoint_data, operate_only_commands_dict): | |
| for cluster, cluster_data in endpoint_data.items(): | |
| if cluster.Attributes.AcceptedCommandList in cluster_data: | |
| command_list = cluster_data[cluster.Attributes.AcceptedCommandList] | |
| for cmd_id in command_list: | |
| try: | |
| xml_command = xml_clusters[cluster.id].accepted_commands[cmd_id] | |
| if xml_command.privilege == Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kOperate: | |
| cluster_object = Clusters.ClusterObjects.ALL_CLUSTERS[cluster.id] | |
| command_object = Clusters.ClusterObjects.ALL_ACCEPTED_COMMANDS[cluster.id][cmd_id] | |
| # Only client-to-server commands (no response commands) | |
| if not command_object.is_client: | |
| continue | |
| if endpoint_id not in operate_only_commands_dict: | |
| operate_only_commands_dict[endpoint_id] = [] | |
| # In this codebase, all generated ClusterCommand subclasses have defaults for all fields. | |
| operate_only_commands_dict[endpoint_id].append(OperateOnlyCommand( | |
| cluster_object=cluster_object, command_object=command_object)) | |
| except KeyError: | |
| logger.warning( | |
| "Command ID %s on cluster %s not found in spec XMLs. This may be a manufacturer-specific command.", | |
| cmd_id, cluster.id) | |
| # Main logic | |
| attributes, spec_version = await get_device_composition_and_spec(dev_ctrl, node_id) | |
| xml_clusters = get_xml_clusters(spec_version) | |
| operate_only_commands_dict = {} | |
| if endpoint_id_to_search is not None: | |
| asserts.assert_false((exclude_ep0 and endpoint_id_to_search == 0), | |
| "Endpoint 0 was both specified to be searched in and to be ignored.") | |
| endpoint_data = attributes.get(endpoint_id_to_search) | |
| if endpoint_data is None: | |
| asserts.fail(f"Endpoint {endpoint_id_to_search} not found on the device.") | |
| find_commands_on_endpoint_and_cluster(endpoint_id_to_search, endpoint_data, operate_only_commands_dict) | |
| else: | |
| for endpoint_id, endpoint_data in attributes.items(): | |
| if exclude_ep0 and endpoint_id == 0: | |
| continue | |
| find_commands_on_endpoint_and_cluster(endpoint_id, endpoint_data, operate_only_commands_dict) | |
| return operate_only_commands_dict | |
| def get_iana_multicast_address() -> bytes: | |
| """Returns the 16-byte IANA-assigned multicast address for Groupcast (ff05::fa).""" | |
| return bytes.fromhex("ff0500000000000000000000000000fa") | |
| def get_per_group_multicast_address(fabric_id: int, group_id: int) -> bytes: | |
| """Returns the 16-byte per-group multicast address (ff35:0040:fd<Fabric ID>00:<Group ID>).""" | |
| # The first 32 bits will always be a fixed value. 0xFF3 defined by RFC 3306, | |
| # 0x05 represents scope, 0x00 is a reserved byte, and 0x40 represents length | |
| # of network prefix (64 bits) | |
| prefix_scope_plen = 0xFF350040 | |
| # Create 64 bit network prefix. Consists of FD (locally assigned ULA prefix) and then | |
| # the upper 56 bits of fabric ID (in big endian format) | |
| network_prefix = 0xfd00000000000000 | ((fabric_id >> 8) & 0x00ffffffffffffff) | |
| # Create 32 bit group identifier portion. Constists of the lower 8 bits of fabric id, | |
| # a reserved 0x00 byte, then followed by 16 bit group id | |
| group_id_field = ((fabric_id << 24) & 0xff000000) | (group_id & 0xffff) | |
| # Combine all portions into 128-bit address | |
| addr_int = (prefix_scope_plen << 96) | (network_prefix << 32) | group_id_field | |
| return ipaddress.IPv6Address(addr_int).packed |