| # See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments | |
| # for details about the block below. | |
| # | |
| # These are separated into different runs because the logs for these tests are HUGE. The attribute one individually | |
| # reads every attribute on every cluster 4 times. If there's a failure, having these in separate runs makes it significantly | |
| # easier to navigate the logs | |
| # | |
| # === BEGIN CI TEST ARGUMENTS === | |
| # test-runner-runs: | |
| # run1: | |
| # app: ${ALL_CLUSTERS_APP} | |
| # factory-reset: true | |
| # quiet: true | |
| # app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json | |
| # script-args: > | |
| # --storage-path admin_storage.json | |
| # --commissioning-method on-network | |
| # --discriminator 1234 | |
| # --passcode 20202021 | |
| # --trace-to json:${TRACE_TEST_JSON}.json | |
| # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto | |
| # --tests test_TC_ACE_2_1 | |
| # run2: | |
| # app: ${ALL_CLUSTERS_APP} | |
| # factory-reset: true | |
| # quiet: true | |
| # app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json | |
| # script-args: > | |
| # --storage-path admin_storage.json | |
| # --commissioning-method on-network | |
| # --discriminator 1234 | |
| # --passcode 20202021 | |
| # --trace-to json:${TRACE_TEST_JSON}.json | |
| # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto | |
| # --tests test_TC_ACE_2_2 | |
| # run3: | |
| # app: ${ALL_CLUSTERS_APP} | |
| # factory-reset: true | |
| # quiet: true | |
| # app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json | |
| # script-args: > | |
| # --storage-path admin_storage.json | |
| # --commissioning-method on-network | |
| # --discriminator 1234 | |
| # --passcode 20202021 | |
| # --trace-to json:${TRACE_TEST_JSON}.json | |
| # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto | |
| # --bool-arg ci_only_linux_skip_ota_cluster_disallowed_for_certification:True | |
| # --tests test_TC_ACE_2_3 | |
| # run4: | |
| # app: ${ALL_CLUSTERS_APP} | |
| # factory-reset: true | |
| # quiet: true | |
| # app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json | |
| # script-args: > | |
| # --storage-path admin_storage.json | |
| # --commissioning-method on-network | |
| # --discriminator 1234 | |
| # --passcode 20202021 | |
| # --trace-to json:${TRACE_TEST_JSON}.json | |
| # --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto | |
| # --tests test_TC_ACE_2_4 | |
| # === END CI TEST ARGUMENTS === | |
| import logging | |
| from copy import deepcopy | |
| from enum import Enum, auto | |
| import matter.clusters as Clusters | |
| from matter.clusters.Attribute import ValueDecodeFailure | |
| from matter.exceptions import ChipStackError | |
| from matter.interaction_model import InteractionModelError, Status | |
| from matter.testing.basic_composition import BasicCompositionTests | |
| from matter.testing.decorators import async_test_body | |
| from matter.testing.global_attribute_ids import (GlobalAttributeIds, is_standard_attribute_id, is_standard_cluster_id, | |
| is_standard_command_id) | |
| from matter.testing.problem_notices import AttributePathLocation, ClusterPathLocation, CommandPathLocation | |
| from matter.testing.runner import TestStep, default_matter_test_main | |
| from matter.testing.spec_parsing import XmlCluster | |
| from matter.tlv import uint | |
| log = logging.getLogger(__name__) | |
| class AccessTestType(Enum): | |
| READ = auto() | |
| WRITE = auto() | |
| INVOKE = auto() | |
| SUBSCRIBE = auto() | |
| def step_number_with_privilege(step: int, substep: str, privilege: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum) -> str: | |
| return f'{step}{substep}_{privilege.name}' | |
| def operation_allowed(spec_requires: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum, | |
| acl_set_to: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum | None) -> bool: | |
| ''' Determines if the action is allowed on the device based on the spec_requirements and the current ACL privilege granted. | |
| The spec parsing uses kUnknownEnumValue to indicate that NO access is allowed for this attribute | |
| or command (ex. command with no write access). | |
| ACL uses None to indicate that no access has been granted to this controller. | |
| In both of these cases, the action is disallowed. In all other cases, access is allowed if the ACL | |
| grants a privilege at or above the privilege required in the spec. | |
| ''' | |
| if spec_requires == Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue or acl_set_to is None: | |
| return False | |
| return spec_requires <= acl_set_to | |
| def checkable_attributes(cluster_id, cluster, xml_cluster) -> list[uint]: | |
| all_attrs = cluster[GlobalAttributeIds.ATTRIBUTE_LIST_ID] | |
| def is_known_cluster_attribute(attribute_id) -> bool: | |
| ''' Returns true if this is a non-manufacturer specific attribute that has information in the XML and has python codegen data''' | |
| return is_standard_attribute_id(attribute_id) and attribute_id in xml_cluster.attributes and attribute_id in Clusters.ClusterObjects.ALL_ATTRIBUTES.get(cluster_id, {}) | |
| return [attr_id for attr_id in all_attrs if is_known_cluster_attribute(attr_id)] | |
| def checkable_commands(cluster_id, cluster, xml_cluster) -> list[uint]: | |
| all_cmds = cluster[GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID] | |
| def is_known_cluster_cmd(command_id) -> bool: | |
| ''' Returns true if this is a non-manufacturer specific command that has information in the XML and has python codegen data''' | |
| return is_standard_command_id(command_id) and command_id in xml_cluster.accepted_commands and command_id in Clusters.ClusterObjects.ALL_ACCEPTED_COMMANDS.get(cluster_id, {}) | |
| return [cmd_id for cmd_id in all_cmds if is_known_cluster_cmd(cmd_id)] | |
| class AccessChecker(BasicCompositionTests): | |
| @async_test_body | |
| async def setup_class(self): | |
| # TODO: Make this into a proper default in the class so we're not overriding the command lines | |
| self.user_params["use_pase_only"] = False | |
| super().setup_class() | |
| await self.setup_class_helper() | |
| self.build_spec_xmls() | |
| acl_attr = Clusters.AccessControl.Attributes.Acl | |
| self.default_acl = await self.read_single_attribute_check_success(endpoint=0, cluster=Clusters.AccessControl, attribute=acl_attr) | |
| self._record_errors() | |
| # We need to run this test from two controllers so we can test access to the ACL cluster while retaining access to the ACL cluster | |
| fabric_admin = self.certificate_authority_manager.activeCaList[0].adminList[0] | |
| self.TH2_nodeid = self.matter_test_config.controller_node_id + 1 | |
| self.TH2 = fabric_admin.NewController(nodeId=self.TH2_nodeid) | |
| # Both the tests in this suite are potentially long-running if there are a large number of attributes on the DUT | |
| # and the network is slow. TC_ACE_2_4 (subscription test) is especially long as it tests ALL attributes. | |
| # Set the default to 10 minutes to account for comprehensive subscription testing. | |
| @property | |
| def default_timeout(self) -> int: | |
| return 1500 # 25 minutes for comprehensive subscription testing | |
| @async_test_body | |
| async def setup_test(self): | |
| super().setup_test() | |
| self.success = True | |
| @async_test_body | |
| async def teardown_test(self): | |
| await self.default_controller.WriteAttribute(self.dut_node_id, attributes=[ | |
| (0, Clusters.AccessControl.Attributes.Acl(self.default_acl))]) | |
| async def _setup_acl(self, privilege: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum | None): | |
| if privilege is None: | |
| return | |
| new_acl = deepcopy(self.default_acl) | |
| new_entry = Clusters.AccessControl.Structs.AccessControlEntryStruct( | |
| privilege=privilege, authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase, subjects=[self.TH2_nodeid]) | |
| new_acl.append(new_entry) | |
| await self.default_controller.WriteAttribute(self.dut_node_id, attributes=[(0, Clusters.AccessControl.Attributes.Acl(new_acl))]) | |
| def _record_errors(self): | |
| ''' Checks through all the endpoints and records all the spec warnings in one go so we don't get repeats''' | |
| all_clusters = set() | |
| attrs: dict[uint, set] = {} | |
| cmds: dict[uint, set] = {} | |
| for endpoint_id, endpoint in self.endpoints_tlv.items(): | |
| all_clusters |= set(endpoint.keys()) | |
| for cluster_id, device_cluster_data in endpoint.items(): | |
| # Find all the attributes for this cluster across all endpoint | |
| if cluster_id not in attrs: | |
| attrs[cluster_id] = set() | |
| if cluster_id not in cmds: | |
| cmds[cluster_id] = set() | |
| if GlobalAttributeIds.ATTRIBUTE_LIST_ID not in device_cluster_data: | |
| location = ClusterPathLocation(endpoint_id=endpoint_id, cluster_id=cluster_id) | |
| self.record_error(test_name="Access Checker", location=location, | |
| problem="Cluster does not have the AttributeList attribute") | |
| else: | |
| # discard MEI attributes as we do not have access information for them. | |
| attrs[cluster_id].update( | |
| set(filter(is_standard_attribute_id, device_cluster_data[GlobalAttributeIds.ATTRIBUTE_LIST_ID]))) | |
| if GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID not in device_cluster_data: | |
| location = ClusterPathLocation(endpoint_id=endpoint_id, cluster_id=cluster_id) | |
| self.record_error(test_name="Access Checker", location=location, | |
| problem="Cluster does not have the AcceptedCommandList attribute") | |
| else: | |
| cmds[cluster_id].update( | |
| set(filter(is_standard_command_id, device_cluster_data[GlobalAttributeIds.ACCEPTED_COMMAND_LIST_ID]))) | |
| # Remove MEI clusters - we don't have information available to check these. | |
| all_clusters = list(filter(is_standard_cluster_id, all_clusters)) | |
| for cluster_id in all_clusters: | |
| location = ClusterPathLocation(endpoint_id=0, cluster_id=cluster_id) | |
| if cluster_id not in self.xml_clusters: | |
| self.record_error(test_name="Access Checker", location=location, problem="Cluster not present in spec data") | |
| self.success = False | |
| continue | |
| if cluster_id not in Clusters.ClusterObjects.ALL_ATTRIBUTES: | |
| self.record_error(test_name="Access Checker", location=location, problem="Unknown cluster") | |
| self.success = False | |
| continue | |
| # check that we have information for all the required attributes | |
| xml_cluster = self.xml_clusters[cluster_id] | |
| for attribute_id in attrs[cluster_id]: | |
| location = AttributePathLocation(endpoint_id=endpoint_id, cluster_id=cluster_id, attribute_id=attribute_id) | |
| if attribute_id not in xml_cluster.attributes: | |
| self.record_error(test_name="Access Checker", location=location, | |
| problem="Cluster attribute not found in spec XML") | |
| self.success = False | |
| continue | |
| if attribute_id not in Clusters.ClusterObjects.ALL_ATTRIBUTES[cluster_id]: | |
| self.record_error(test_name="Access Checker", location=location, | |
| problem="Unknown attribute") | |
| self.success = False | |
| continue | |
| # Check that we have information for all the required commands | |
| for command_id in cmds[cluster_id]: | |
| location = CommandPathLocation(endpoint_id=endpoint_id, cluster_id=cluster_id, command_id=command_id) | |
| if command_id not in xml_cluster.accepted_commands: | |
| self.record_error(test_name="Access Checker", location=location, | |
| problem="Cluster command not found in spec XML") | |
| self.success = False | |
| continue | |
| if command_id not in Clusters.ClusterObjects.ALL_ACCEPTED_COMMANDS.get(cluster_id, {}): | |
| self.record_error(test_name="Access Checker", location=location, | |
| problem="Unknown command") | |
| self.success = False | |
| continue | |
| async def _maybe_run_command_access_test_for_cluster_privilege(self, endpoint_id, cluster_id, device_cluster_data, xml_cluster: XmlCluster, privilege: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum): | |
| """ Runs a command only if the required cluster privilege is HIGHER than the specified privilege. In this way, | |
| no commands are actually run on the device, which means there are no side effects. However, we can differentiate | |
| ACL rejections from commands being unsupported. | |
| """ | |
| ota_exception = self.user_params.get('ci_only_linux_skip_ota_cluster_disallowed_for_certification', False) | |
| if cluster_id == Clusters.OtaSoftwareUpdateRequestor.id and ota_exception: | |
| log.warning('WARNING: Skipping OTA cluster check for CI. THIS IS DISALLOWED FOR CERTIFICATION') | |
| return | |
| log.info('Testing commands on %s at privilege %s', xml_cluster.name, privilege) | |
| for command_id in checkable_commands(cluster_id, device_cluster_data, xml_cluster): | |
| # TODO(spec): ChangeToModeByCoreTag in ModeBase.adoc / data_model/1.7/clusters/ModeBase.xml | |
| # is missing `<access invokePrivilege="operate"/>`. As a result, spec_parsing defaults | |
| # its privilege to UNKNOWN, causing TC_AccessChecker to expect UNSUPPORTED_ACCESS at Operate | |
| # privilege even though the device implementation correctly allows Operate access. | |
| if xml_cluster.accepted_commands[command_id].name == "ChangeToModeByCoreTag": | |
| log.warning("Skipping ChangeToModeByCoreTag access check due to missing invokePrivilege in spec XML") | |
| continue | |
| spec_requires = xml_cluster.accepted_commands[command_id].privilege | |
| command = Clusters.ClusterObjects.ALL_ACCEPTED_COMMANDS.get(cluster_id, {}).get(command_id) | |
| if command is None: | |
| continue | |
| location = CommandPathLocation(endpoint_id=endpoint_id, cluster_id=cluster_id, command_id=command_id) | |
| name = f"Command test - privilege {privilege}" | |
| if operation_allowed(spec_requires, privilege): | |
| # In this test, we're only checking that the disallowed commands are rejected so that there are | |
| # no side effects. Commands are checked with admin privilege in their cluster tests. The error that | |
| # may be let through here is if the spec requires operate and the implementation requires admin. | |
| continue | |
| log.info(' Testing command %s from cluster %s - at privilege %s, requires %s', | |
| xml_cluster.accepted_commands[command_id].name, xml_cluster.name, privilege, spec_requires) | |
| try: | |
| timed = None | |
| if command.must_use_timed_invoke: | |
| # This command requires a timedRequest. Setting the timed value to largest value (unsigned int). | |
| # We're sending the command right away, so this value doesn't matter, but we do need to set a value here to trigger the timed request message. | |
| timed = 65535 | |
| await self.send_single_cmd(cmd=command(), dev_ctrl=self.TH2, endpoint=endpoint_id, timedRequestTimeoutMs=timed) | |
| # If this was successful, that's an error | |
| self.record_error(test_name=name, location=location, | |
| problem=f"Unexpected success sending command {command} with privilege {privilege}") | |
| self.success = False | |
| log.info(' Received unexpected SUCCESS') | |
| except InteractionModelError as e: | |
| if e.status != Status.UnsupportedAccess: | |
| self.record_error(test_name=name, location=location, | |
| problem=f'Unexpected error sending command {command} with privilege {privilege} - expected UNSUPPORTED_ACCESS, got {e.status}') | |
| self.success = False | |
| log.info(' Received unexpected error %s', e) | |
| else: | |
| log.info(' Received expected error') | |
| async def _run_read_access_test_for_cluster_privilege(self, endpoint_id, cluster_id, device_cluster_data, xml_cluster: XmlCluster, privilege: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum): | |
| # TODO: This assumes all attributes are readable. Which they are currently. But we don't have a general way to mark otherwise. | |
| for attribute_id in checkable_attributes(cluster_id, device_cluster_data, xml_cluster): | |
| spec_requires = xml_cluster.attributes[attribute_id].read_access | |
| attribute = Clusters.ClusterObjects.ALL_ATTRIBUTES.get(cluster_id, {}).get(attribute_id) | |
| cluster_class = Clusters.ClusterObjects.ALL_CLUSTERS.get(cluster_id) | |
| if attribute is None or cluster_class is None: | |
| continue | |
| if operation_allowed(spec_requires, privilege): | |
| ret = await self.read_single_attribute_check_success(dev_ctrl=self.TH2, endpoint=endpoint_id, cluster=cluster_class, attribute=attribute, assert_on_error=False, test_name=f"Read access Checker - {privilege}") | |
| if ret is None: | |
| self.success = False | |
| else: | |
| ret = await self.read_single_attribute_expect_error(dev_ctrl=self.TH2, endpoint=endpoint_id, cluster=cluster_class, attribute=attribute, error=Status.UnsupportedAccess, assert_on_error=False, test_name=f"Read access Checker - {privilege}") | |
| if ret is None: | |
| self.success = False | |
| async def _subscribe_single_attribute_check_success(self, endpoint_id, cluster_id, attribute_id, attribute, cluster_class, privilege, test_name): | |
| """Helper to subscribe to an attribute expecting success.""" | |
| try: | |
| subscription = await self.TH2.ReadAttribute( | |
| nodeId=self.dut_node_id, | |
| attributes=[(endpoint_id, attribute)], | |
| reportInterval=(0, 30), | |
| keepSubscriptions=False, | |
| fabricFiltered=False | |
| ) | |
| if subscription.subscriptionId is None: | |
| self.record_error(test_name=test_name, | |
| location=AttributePathLocation(endpoint_id=endpoint_id, | |
| cluster_id=cluster_id, attribute_id=attribute_id), | |
| problem=f"Subscription activation failed - expected success with privilege {privilege}") | |
| self.success = False | |
| return False | |
| log.info("Successfully established subscription (ID: %s) with privilege %s", subscription.subscriptionId, privilege) | |
| subscription.Shutdown() | |
| return True | |
| except ChipStackError as e: # chipstack-ok | |
| # Unexpected ChipStackError | |
| log.error("Unexpected ChipStackError subscribing to attribute %s: %s", attribute, e) | |
| self.record_error(test_name=test_name, | |
| location=AttributePathLocation(endpoint_id=endpoint_id, | |
| cluster_id=cluster_id, attribute_id=attribute_id), | |
| problem=f"Unexpected ChipStackError during subscription: {e} (error code: 0x{e.err:08X})") | |
| self.success = False | |
| return False | |
| async def _subscribe_single_attribute_expect_error(self, endpoint_id, cluster_id, attribute_id, attribute, cluster_class, privilege, test_name): | |
| """Helper to subscribe to an attribute expecting UnsupportedAccess error.""" | |
| try: | |
| subscription = await self.TH2.ReadAttribute( | |
| nodeId=self.dut_node_id, | |
| attributes=[(endpoint_id, attribute)], | |
| reportInterval=(0, 30), | |
| keepSubscriptions=False, | |
| fabricFiltered=False | |
| ) | |
| # Subscription succeeded but should have failed | |
| subscription.Shutdown() | |
| self.record_error(test_name=test_name, | |
| location=AttributePathLocation(endpoint_id=endpoint_id, | |
| cluster_id=cluster_id, attribute_id=attribute_id), | |
| problem=f"Subscription succeeded but should have failed with privilege {privilege}") | |
| self.success = False | |
| return False | |
| except ChipStackError as e: # chipstack-ok | |
| if e.err == 0x00000580: # INVALID_ACTION - some attributes don't support subscription | |
| # This occurs for attributes that cannot be subscribed to when testing with insufficient privileges. | |
| # The attributes are write-only or have special subscription restrictions. | |
| # These are the current clusters where this is noticed: | |
| # - AccessControl | |
| # - NetworkCommissioning | |
| # - CameraAvStreamManagement | |
| # - OperationalCredentials | |
| log.warning("INVALID_ACTION: %s.%s (Cluster=0x%04X, Attribute=0x%04X, Endpoint=%s, Privilege=%s)", | |
| cluster_class.__name__, attribute.__name__, cluster_id, attribute_id, endpoint_id, privilege.name) | |
| return None # Indicates skip | |
| # Unexpected ChipStackError | |
| log.error("Unexpected ChipStackError subscribing to attribute %s: %s", attribute, e) | |
| self.record_error(test_name=test_name, | |
| location=AttributePathLocation(endpoint_id=endpoint_id, | |
| cluster_id=cluster_id, attribute_id=attribute_id), | |
| problem=f"Unexpected ChipStackError during subscription: {e} (error code: 0x{e.err:08X})") | |
| self.success = False | |
| return False | |
| async def _run_subscribe_access_test_for_cluster_privilege(self, endpoint_id, cluster_id, device_cluster_data, xml_cluster: XmlCluster, privilege: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum): | |
| """Tests subscription access control by verifying subscriptions can be established. | |
| This follows the same pattern as _run_read_access_test_for_cluster_privilege, | |
| but uses subscription parameters in the ReadAttribute call. | |
| """ | |
| for attribute_id in checkable_attributes(cluster_id, device_cluster_data, xml_cluster): | |
| spec_requires = xml_cluster.attributes[attribute_id].read_access | |
| attribute = Clusters.ClusterObjects.ALL_ATTRIBUTES.get(cluster_id, {}).get(attribute_id) | |
| cluster_class = Clusters.ClusterObjects.ALL_CLUSTERS.get(cluster_id) | |
| if attribute is None or cluster_class is None: | |
| continue | |
| test_name = f'Subscribe access checker - {privilege}' | |
| log.info("Subscribing to attribute %s cluster %s privilege %s", attribute, xml_cluster.name, privilege) | |
| if operation_allowed(spec_requires, privilege): | |
| result = await self._subscribe_single_attribute_check_success( | |
| endpoint_id, cluster_id, attribute_id, attribute, cluster_class, privilege, test_name) | |
| else: | |
| result = await self._subscribe_single_attribute_expect_error( | |
| endpoint_id, cluster_id, attribute_id, attribute, cluster_class, privilege, test_name) | |
| if result is None: # Skip this attribute (INVALID_ACTION) | |
| continue | |
| async def _run_write_access_test_for_cluster_privilege(self, endpoint_id, cluster_id, cluster, xml_cluster: XmlCluster, privilege: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum, wildcard_read): | |
| for attribute_id in checkable_attributes(cluster_id, cluster, xml_cluster): | |
| spec_requires = xml_cluster.attributes[attribute_id].write_access | |
| is_optional_write = xml_cluster.attributes[attribute_id].write_optional | |
| attribute = Clusters.ClusterObjects.ALL_ATTRIBUTES.get(cluster_id, {}).get(attribute_id) | |
| cluster_class = Clusters.ClusterObjects.ALL_CLUSTERS.get(cluster_id) | |
| if attribute is None or cluster_class is None: | |
| continue | |
| location = AttributePathLocation(endpoint_id=endpoint_id, cluster_id=cluster_id, attribute_id=attribute_id) | |
| test_name = f'Write access checker - {privilege}' | |
| log.info("Testing attribute %s on endpoint %s", attribute, endpoint_id) | |
| if attribute == Clusters.AccessControl.Attributes.Acl and privilege == Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister: | |
| log.info("Skipping ACL attribute check for admin privilege as this is known to be writeable and is being used for this test") | |
| continue | |
| # Because we read everything with admin, we should have this in the wildcard read | |
| # This will only not work if we end up with write-only attributes. We do not currently have any of these. | |
| val = wildcard_read.attributes[endpoint_id][cluster_class][attribute] | |
| if isinstance(val, ValueDecodeFailure): | |
| self.record_error(test_name=test_name, location=location, | |
| problem=f"Attribute {attribute} returned a read error {val} - unable to write current value") | |
| self.success = False | |
| continue | |
| if isinstance(val, list): | |
| # Use an empty list for writes in case the list is large and does not fit | |
| val = [] | |
| resp = await self.TH2.WriteAttribute(nodeId=self.dut_node_id, attributes=[(endpoint_id, attribute(val))]) | |
| if spec_requires == Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue: | |
| # not writeable - expect either an unsupported write response or an unsupported access response (since ACL privilege check happens before writable check) | |
| if resp[0].Status not in (Status.UnsupportedWrite, Status.UnsupportedAccess): | |
| self.record_error(test_name=test_name, location=location, | |
| problem=f"Unexpected error writing non-writeable attribute - expected either Unsupported Access or UnsupportedWrite, got {resp[0].Status.name}") | |
| self.success = False | |
| elif is_optional_write and resp[0].Status == Status.UnsupportedWrite: | |
| # unsupported optional writeable attribute - this is fine, no error | |
| continue | |
| elif operation_allowed(spec_requires, privilege): | |
| # Write the default attribute. We don't care if this fails, as long as it fails with a DIFFERENT error than the access | |
| # This is OK because access is required to be checked BEFORE any other thing to avoid leaking device information. | |
| # For example, because we don't have any range information, we might be writing an out of range value, but this will | |
| # get rejected by the ACL check before the range check. | |
| # See section: 8.4.3.2 | |
| if resp[0].Status == Status.UnsupportedAccess: | |
| self.record_error(test_name=test_name, location=location, | |
| problem="Unexpected UnsupportedAccess writing attribute") | |
| self.success = False | |
| else: | |
| if resp[0].Status != Status.UnsupportedAccess: | |
| self.record_error(test_name=test_name, location=location, | |
| problem=f"Unexpected error writing attribute - expected Unsupported Access, got {resp[0].Status.name}") | |
| self.success = False | |
| if resp[0].Status == Status.Success and isinstance(val, list): | |
| # Reset the value to the original if we managed to write an empty list | |
| val = wildcard_read.attributes[endpoint_id][cluster_class][attribute] | |
| await self.TH2.WriteAttribute(nodeId=self.dut_node_id, attributes=[(endpoint_id, attribute(val))]) | |
| async def run_access_test(self, test_type: AccessTestType): | |
| # Step precondition, 1 and 2 are handled in the class setup, but need to be marked for every test | |
| self.step("precondition") | |
| self.step(1) | |
| self.step(2) | |
| # Read all the attributes on TH2 using admin access | |
| check_step = 3 | |
| if test_type == AccessTestType.WRITE: | |
| self.step(3) | |
| await self._setup_acl(privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister) | |
| self.step(4) | |
| wildcard_read = await self.TH2.Read(self.dut_node_id, [()]) | |
| check_step = 5 | |
| self.step(check_step) | |
| enum = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum | |
| privilege_enum = [p for p in enum if p != enum.kUnknownEnumValue and p != enum.kProxyView] | |
| for privilege in privilege_enum: | |
| log.info("Testing for %s", privilege) | |
| self.step(step_number_with_privilege(check_step, 'a', privilege)) | |
| await self._setup_acl(privilege=privilege) | |
| self.step(step_number_with_privilege(check_step, 'b', privilege)) | |
| for endpoint_id, endpoint in self.endpoints_tlv.items(): | |
| for cluster_id, device_cluster_data in endpoint.items(): | |
| if not is_standard_cluster_id(cluster_id) or cluster_id not in self.xml_clusters or cluster_id not in Clusters.ClusterObjects.ALL_ATTRIBUTES: | |
| # These cases have already been recorded by the _record_errors function | |
| continue | |
| xml_cluster = self.xml_clusters[cluster_id] | |
| if test_type == AccessTestType.READ: | |
| await self._run_read_access_test_for_cluster_privilege(endpoint_id, cluster_id, device_cluster_data, xml_cluster, privilege) | |
| elif test_type == AccessTestType.WRITE: | |
| await self._run_write_access_test_for_cluster_privilege(endpoint_id, cluster_id, device_cluster_data, xml_cluster, privilege, wildcard_read) | |
| elif test_type == AccessTestType.INVOKE: | |
| await self._maybe_run_command_access_test_for_cluster_privilege(endpoint_id, cluster_id, device_cluster_data, xml_cluster, privilege) | |
| elif test_type == AccessTestType.SUBSCRIBE: | |
| await self._run_subscribe_access_test_for_cluster_privilege(endpoint_id, cluster_id, device_cluster_data, xml_cluster, privilege) | |
| else: | |
| self.fail_current_test("Unsupported test type") | |
| if not self.success: | |
| self.fail_current_test("One or more access violations was found") | |
| def steps_TC_ACE_2_1(self): | |
| steps = [TestStep("precondition", "DUT is commissioned", is_commissioning=True), | |
| TestStep(1, "TH_commissioner performs a wildcard read (done during test setup)"), | |
| TestStep(2, "TH_commissioner reads the ACL attribute (done during test setup)"), | |
| TestStep(3, "Repeat steps 3a and 3b for each permission level")] | |
| enum = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum | |
| privilege_enum = [p for p in enum if p != enum.kUnknownEnumValue and p != enum.kProxyView] | |
| for p in privilege_enum: | |
| steps.append(TestStep(step_number_with_privilege(3, 'a', p), | |
| "TH_commissioner gives TH_second_commissioner the specified privilege")) | |
| steps.append(TestStep(step_number_with_privilege(3, 'b', p), | |
| "TH_second_controller reads all the attributes and checks for appropriate permission errors")) | |
| return steps | |
| def desc_TC_ACE_2_1(self): | |
| return "[TC-ACE-2.1] Attribute read privilege enforcement - [DUT as Server]" | |
| @async_test_body | |
| async def test_TC_ACE_2_1(self): | |
| await self.run_access_test(AccessTestType.READ) | |
| def steps_TC_ACE_2_2(self): | |
| steps = [TestStep("precondition", "DUT is commissioned", is_commissioning=True), | |
| TestStep(1, "TH_commissioner performs a wildcard read (done during test setup)"), | |
| TestStep(2, "TH_commissioner reads the ACL attribute (done during test setup)"), | |
| TestStep(3, "TH_commissioner grants TH_second_controller admin permission"), | |
| TestStep(4, "TH_second_controller performs a wildcard read"), | |
| TestStep(5, "Repeat steps 5a and 5b for each permission level")] | |
| enum = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum | |
| privilege_enum = [p for p in enum if p != enum.kUnknownEnumValue and p != enum.kProxyView] | |
| for p in privilege_enum: | |
| steps.append(TestStep(step_number_with_privilege(5, 'a', p), | |
| "TH_commissioner gives TH_second_commissioner the specified privilege")) | |
| steps.append(TestStep(step_number_with_privilege(5, 'b', p), | |
| "TH_second_commissioner writes all the attributes and checks for appropriate permission errors")) | |
| return steps | |
| def desc_TC_ACE_2_2(self): | |
| return "[TC-ACE-2.2] Attribute write privilege enforcement - [DUT as Server]" | |
| @async_test_body | |
| async def test_TC_ACE_2_2(self): | |
| await self.run_access_test(AccessTestType.WRITE) | |
| def steps_TC_ACE_2_3(self): | |
| steps = [TestStep("precondition", "DUT is commissioned", is_commissioning=True), | |
| TestStep(1, "TH_commissioner performs a wildcard read (done during test setup)"), | |
| TestStep(2, "TH_commissioner reads the ACL attribute (done during test setup)"), | |
| TestStep(3, "Repeat steps 3a and 3b for each permission level")] | |
| enum = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum | |
| privilege_enum = [p for p in enum if p != enum.kUnknownEnumValue and p != enum.kProxyView] | |
| for p in privilege_enum: | |
| steps.append(TestStep(step_number_with_privilege(3, 'a', p), | |
| "TH_commissioner gives TH_second_controller the specified privilege")) | |
| steps.append(TestStep(step_number_with_privilege(3, 'b', p), | |
| """For each standard command on each standard cluster on each endpoint, | |
| TH_second_controller checks the permission requirements for that command. | |
| If the permission required for the command is HIGHER than the permission level being tested, | |
| TH_second_controller sends the command to the DUT using default values. | |
| Regardless of the command contents, the DUT should return an access error since access must be checked | |
| before the command is processed. Receipt of an UNSUPPORTED_COMMAND error is a conformance failure.""", | |
| "DUT returns UNSUPPORTED_ACCESS error")) | |
| return steps | |
| def desc_TC_ACE_2_3(self): | |
| return "[TC-ACE-2.3] Command Privilege Enforcement - [DUT as Server]" | |
| @async_test_body | |
| async def test_TC_ACE_2_3(self): | |
| await self.run_access_test(AccessTestType.INVOKE) | |
| def steps_TC_ACE_2_4(self): | |
| steps = [TestStep("precondition", "DUT is commissioned", is_commissioning=True), | |
| TestStep(1, "TH_commissioner performs a wildcard read (done during test setup)"), | |
| TestStep(2, "TH_commissioner reads the ACL attribute (done during test setup)"), | |
| TestStep(3, "Repeat steps 3a and 3b for each permission level")] | |
| enum = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum | |
| privilege_enum = [p for p in enum if p != enum.kUnknownEnumValue and p != enum.kProxyView] | |
| for p in privilege_enum: | |
| steps.append(TestStep(step_number_with_privilege(3, 'a', p), | |
| "TH_commissioner gives TH_second_commissioner the specified privilege")) | |
| steps.append(TestStep(step_number_with_privilege(3, 'b', p), | |
| "TH_second_controller subscribes to all the attributes and verifies the subscription is established and priming report is received with appropriate permission errors")) | |
| return steps | |
| def desc_TC_ACE_2_4(self): | |
| return "[TC-ACE-2.4] Attribute read subscription report - [DUT as Server]" | |
| @async_test_body | |
| async def test_TC_ACE_2_4(self): | |
| await self.run_access_test(AccessTestType.SUBSCRIBE) | |
| if __name__ == "__main__": | |
| default_matter_test_main() |