check

src/python_testing/TC_OPCREDS_3_1.py

@@ -15,6 +15,7 @@
 #    limitations under the License.
 #
 
+import copy
 import logging
 import random
 
@@ -25,6 +26,7 @@
 import chip.discovery as Discovery
 from chip.exceptions import ChipStackError
 from chip.interaction_model import InteractionModelError, Status
+from chip.tlv import TLVReader, TLVWriter
 from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, type_matches
 from mobly import asserts
 
@@ -101,16 +103,15 @@ async def test_TC_OPCREDS_3_1(self):
             # Expiring the failsafe timer in an attempt to clean up.
             await TH1.SendCommand(newNodeId, 0, Clusters.GeneralCommissioning.Commands.ArmFailSafe(0))
             asserts.assert_fail("Unable to generate NOC chain for DUT - this is a script failure, please report this as a bug")
-        # TODO: This is actually easier because we just need to parse the TLV
-        root_cert_temp = x509.load_der_x509_certificate(TH1_certs_real.rcacBytes)
-        root_public_key_th1 = hex_from_bytes(root_cert_temp.public_key().public_bytes(
-            encoding=Encoding.X962, format=PublicFormat.UncompressedPoint))
+        th1_rcac_decoded = TLVReader(TH1_certs_real.rcacBytes).get()["Any"]
+        # public key is field 9
+        root_public_key_th1 = th1_rcac_decoded[9]
 
         self.print_step(
             11, "TH1 obtains or generates Root Certificate with a different Root CA ID and the corresponding ICAC, NOC and IPK using csrResponse")
         TH1_CA_fake = self.certificate_authority_manager.NewCertificateAuthority()
         TH1_fabric_admin_fake = TH1_CA_fake.NewFabricAdmin(vendorId=0xFFF1, fabricId=2)
-        TH1_fake = TH1_fabric_admin_fake.NewController(nodeId=dev_ctrl.nodeId)
+        TH1_fake = TH1_fabric_admin_fake.NewController(nodeId=self.default_controller.nodeId)
         TH1_certs_fake = TH1_fake.IssueNOCChain(csrResponse, newNodeId)
         if (TH1_certs_real.rcacBytes is None or
                 TH1_certs_real.icacBytes is None or
@@ -121,12 +122,30 @@ async def test_TC_OPCREDS_3_1(self):
 
         self.print_step(
             12, "TH1 generates an INVALID Root Certificate where the signature does not match the public key and saves it as `Root_CA_Malformed`")
-        TH1_root_CA_malformed = copy.deepcopy(root_cert_temp)
-        TH1_root_CA_malformed.signature[0] = TH1_root_CA_malformed.signature[0] + 1
-        # TH1_root_CA_malformed_der = TH1_root_CA_malformed.public_bytes(encoding)
+        TH1_root_CA_malformed_decoded = copy.deepcopy(th1_rcac_decoded)
+        # signature is field 11
+        print(TH1_root_CA_malformed_decoded[11])
+        malformed_sig_int = int.from_bytes(TH1_root_CA_malformed_decoded[11], 'big') + 1
+        malformed_sig = malformed_sig_int.to_bytes(len(TH1_root_CA_malformed_decoded[11]), 'big')
+        print(malformed_sig)
+        TH1_root_CA_malformed_decoded[11] = malformed_sig
+
+        writer = TLVWriter(bytearray())
+        writer.startStructure(None)
+        for tag, val in TH1_root_CA_malformed_decoded.items():
+            TH1_root_CA_malformed = writer.put(tag, val)
+        writer.endContainer()
+        TH1_root_CA_malformed = writer.encoding
 
         self.print_step(
             13, "TH1 sends AddTrustedRootCertificate command to DUT to install `Root_CA_Malformed` and verifies INVALID_COMMAND is returned")
+        cmd = opcreds.Commands.AddTrustedRootCertificate(TH1_root_CA_malformed)
+        try:
+            await self.send_single_cmd(dev_ctrl=TH1, node_id=newNodeId, cmd=cmd)
+            asserts.assert_fail("Unexpected success adding trusted root cert with malformed signature")
+        except InteractionModelError as e:
+            asserts.assert_equal(e.status, Status.InvalidCommand,
+                                 "Unexpected error adding trusted root cert with malformed signature")
 
         self.print_step(
             14, "TH1 sends AddTrustedRootCertificate command to DUT with RootCACertificate set to `Root_CA_Certificate_TH1`, verify SUCCESS")