TC-DA-1.7: Use steps_function, add missing checks

src/python_testing/TC_DA_1_7.py

@@ -18,15 +18,16 @@
 import logging
 from glob import glob
 from pathlib import Path
-from typing import Optional
+from typing import List, Optional
 
 import chip.clusters as Clusters
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
 from cryptography.x509 import AuthorityKeyIdentifier, Certificate, SubjectKeyIdentifier, load_der_x509_certificate
-from matter_testing_support import MatterBaseTest, async_test_body, bytes_from_hex, default_matter_test_main, hex_from_bytes
+from matter_testing_support import (MatterBaseTest, TestStep, async_test_body, bytes_from_hex, default_matter_test_main,
+                                    hex_from_bytes)
 from mobly import asserts
 
 # Those are SDK samples that are known to be non-production.
@@ -107,20 +108,66 @@ class TC_DA_1_7(MatterBaseTest):
             --script-args "--storage-path admin_storage.json --commissioning-method on-network \
                 --discriminator 12 34 --passcode 20202021 20202021 --bool-arg allow_sdk_dac:true"
     '''
-    @async_test_body
-    async def test_TC_DA_1_7(self):
+
+    def expected_number_of_DUTs(self):
         # For real tests, we require more than one DUT
         # On the CI, this doesn't make sense to do since all the examples use the same DAC
         # To specify more than 1 DUT, use a list of discriminators and passcodes
         allow_sdk_dac = self.user_params.get("allow_sdk_dac", False)
-        if allow_sdk_dac:
-            asserts.assert_equal(len(self.matter_test_config.discriminators), 1, "Only one device can be tested with SDK DAC")
-        if not allow_sdk_dac:
-            asserts.assert_equal(len(self.matter_test_config.discriminators), 2, "This test requires 2 DUTs")
+        post_cert_test = self.user_params.get("post_cert_test", False)
+        if allow_sdk_dac or post_cert_test:
+            return 1
+        return 2
+
+    def steps_one_dut(self, DUT: int) -> List[TestStep]:
+        return [TestStep(f'{DUT}', f'Test DUT{DUT} DAC chain as follows:'),
+                TestStep(f'{DUT}.1', f'TH sends CertificateChainRequest Command to DUT{DUT} with the CertificateType set to PAICertificate',
+                         'Verify that the DUT returns a CertificateChainResponse. Save the returned Certificate as `pai_cert`.'),
+                TestStep(f'{DUT}.2', 'TH sends CertificateChainRequest Command to DUT1 with the CertificateType set to DACCertificate',
+                         'Verify that the DUT returns a CertificateChainResponse. Save the returned Certificate as `dac_cert`.'),
+                TestStep(f'{DUT}.3', 'TH extracts the Authority Key Identifier from the PAI certificate',
+                         ('* Verify that the Authority Key Identifier is signed by a PAA in the DCL. (Ensure that it is not SDK’s test PAAs)\n'
+                          '* Verify that PAI authority key ID must not be one of:\n'
+                          '  - 78: 5C: E7: 05: B8: 6B: 8F: 4E: 6F: C7: 93: AA: 60: CB: 43: EA: 69: 68:82: D5\n'
+                          '  - 6A: FD: 22: 77: 1F: 51: 1F: EC: BF: 16: 41: 97: 67: 10: DC: DC: 31: A1: 71: 7E\n'
+                          '* Save the selected PAA certificate as `paa_cert`\n')),
+                TestStep(f'{DUT}.4', 'TH extracts ASN.1 DER bytes for the entire issuer field of `dac_cert` and subject field of `pai_cert`.',
+                         'Verify that the `dac_cert` `issuer field is byte-forbyte equivalent to the `pai_cert`subject field.'),
+                TestStep(f'{DUT}.5', 'TH extracts ASN.1 DER bytes for the entire issuer field of `pai_cert` and subject field of `paa_cert`.',
+                         'Verify that the `pai_cert` issuer field is byte-forbyte equivalent to the `paa_cert` subject field.'),
+                TestStep(f'{DUT}.6', f'TH extracts the public key from the DAC and saves as `pk_{DUT}`.')
+                ]
+
+    def steps_TC_DA_1_7(self):
+        steps = [TestStep(0, "Commission DUT if not already done", is_commissioning=True)]
+        steps += self.steps_one_dut(1)
+        if self.expected_number_of_DUTs() == 2:
+            steps += self.steps_one_dut(2)
+        steps += [TestStep(3, "Verify that `pk_1` is not equal to `pk_2",
+                           "Public keys do not match")]
+        return steps
+
+    @async_test_body
+    async def test_TC_DA_1_7(self):
+        # post_cert_tests (or sdk) can use the qr or manual code
+        # We don't currently support this in cert because the base doesn't support multiple QR/manual
+        num = 0
+        if self.matter_test_config.discriminators:
+            num += len(self.matter_test_config.discriminators)
+        if self.matter_test_config.qr_code_content:
+            num += 1
+        if self.matter_test_config.manual_code:
+            num += 1
+        asserts.assert_equal(num, self.expected_number_of_DUTs(),
+                             "Unexpected number of devices specified - this test expects two DUTs at cert, one device for CI")
         pk = []
+        # Commissioning - already done.
+        self.step(0)
+
         for i in range(len(self.matter_test_config.dut_node_ids)):
-            pk.append(await self.single_DUT(i, self.matter_test_config.dut_node_ids[i]))
+            pk.append(await self.single_DUT(i+1, self.matter_test_config.dut_node_ids[i]))
 
+        self.step(3)
         asserts.assert_equal(len(pk), len(set(pk)), "Found matching public keys in different DUTs")
 
     async def single_DUT(self, dut_index: int, dut_node_id: int) -> bytes:
@@ -132,26 +179,28 @@ async def single_DUT(self, dut_index: int, dut_node_id: int) -> bytes:
         paa_by_skid = load_all_paa(conf.paa_trust_store_path)
         logging.info("Found %d PAAs" % len(paa_by_skid))
 
-        logging.info("DUT {} Step 1: Commissioning, already done".format(dut_index))
+        # Test plan step introducing test for each DUT
+        self.step(f'{dut_index}')
         dev_ctrl = self.default_controller
 
-        logging.info("DUT {} Step 2: Get PAI of DUT1 with certificate chain request".format(dut_index))
+        self.step(f'{dut_index}.1')
         result = await dev_ctrl.SendCommand(dut_node_id, 0,
                                             Clusters.OperationalCredentials.Commands.CertificateChainRequest(2))
         pai = result.certificate
         asserts.assert_less_equal(len(pai), 600, "PAI cert must be at most 600 bytes")
         key = 'pai_{}'.format(dut_index)
         self.record_data({key: hex_from_bytes(pai)})
 
-        logging.info("DUT {} Step 3: Get DAC of DUT1 with certificate chain request".format(dut_index))
+        self.step(f'{dut_index}.2')
         result = await dev_ctrl.SendCommand(dut_node_id, 0,
                                             Clusters.OperationalCredentials.Commands.CertificateChainRequest(1))
         dac = result.certificate
         asserts.assert_less_equal(len(dac), 600, "DAC cert must be at most 600 bytes")
         key = 'dac_{}'.format(dut_index)
         self.record_data({key: hex_from_bytes(dac)})
 
-        logging.info("DUT {} Step 4 check 1: Ensure PAI's AKID matches a PAA and signature is valid".format(dut_index))
+        self.step(f'{dut_index}.3')
+        logging.info("DUT {} Step 3 check 1: Ensure PAI's AKID matches a PAA and signature is valid".format(dut_index))
         pai_cert = load_der_x509_certificate(pai)
         pai_akid = extract_akid(pai_cert)
         if pai_akid not in paa_by_skid:
@@ -168,15 +217,23 @@ async def single_DUT(self, dut_index: int, dut_node_id: int) -> bytes:
             asserts.fail("DUT %d: Failed to verify PAI signature against PAA public key: %s" % (dut_index, str(e)))
         logging.info("Validated PAI signature against PAA")
 
-        logging.info("DUT {} Step 4 check 2: Verify PAI AKID not in denylist of SDK PAIs".format(dut_index))
+        logging.info("DUT {} Step 3 check 2: Verify PAI AKID not in denylist of SDK PAIs".format(dut_index))
         if allow_sdk_dac:
-            logging.warn("===> TEST STEP SKIPPED: Allowing SDK DACs!")
+            logging.warning("===> TEST STEP SKIPPED: Allowing SDK DACs!")
         else:
             for candidate in FORBIDDEN_AKID:
                 asserts.assert_not_equal(hex_from_bytes(pai_akid), hex_from_bytes(candidate), "PAI AKID must not be in denylist")
 
-        logging.info("DUT {} Step 5: Extract subject public key of DAC and save".format(dut_index))
+        self.step(f'{dut_index}.4')
+        # dac issuer == pai subject
         dac_cert = load_der_x509_certificate(dac)
+        asserts.assert_equal(dac_cert.issuer, pai_cert.subject, "DAC issuer does not match PAI subject")
+
+        self.step(f'{dut_index}.5')
+        # pai issues == paa subject
+        asserts.assert_equal(pai_cert.issuer, paa_cert.subject, "PAI issuer does not match PAA subject")
+
+        self.step(f'{dut_index}.6')
         pk = dac_cert.public_key().public_bytes(encoding=Encoding.X962, format=PublicFormat.UncompressedPoint)
         logging.info("Subject public key pk: %s" % hex_from_bytes(pk))
         key = 'pk_{}'.format(dut_index)