check

Author: cecille

credentials/fetch_paa_certs_from_dcl.py

@@ -37,6 +37,10 @@
 PRODUCTION_NODE_URL_REST = "https://on.dcl.csa-iot.org"
 TEST_NODE_URL_REST = "https://on.test-net.dcl.csa-iot.org"
 
+# TODO: really? We can't just get this by name from the DCL?
+MATTER_CERT_CA_SUBJECT = "MFIxDDAKBgNVBAoMA0NTQTEsMCoGA1UEAwwjTWF0dGVyIENlcnRpZmljYXRpb24gYW5kIFRlc3RpbmcgQ0ExFDASBgorBgEEAYKifAIBDARDNUEw"
+MATTER_CERT_CA_SUBJECT_KEY_ID = "97:E4:69:D0:C5:04:14:C2:6F:C7:01:F7:7E:94:77:39:09:8D:F6:A5"
+
 
 def parse_paa_root_certs(cmdpipe, paa_list):
     """
@@ -80,7 +84,7 @@ def parse_paa_root_certs(cmdpipe, paa_list):
                     paa_list.append(copy.deepcopy(result))
 
 
-def write_paa_root_cert(certificate, subject):
+def write_cert(certificate, subject):
     filename = 'dcld_mirror_' + \
         re.sub('[^a-zA-Z0-9_-]', '', re.sub('[=, ]', '_', subject))
     with open(filename + '.pem', 'w+') as outfile:
@@ -135,10 +139,37 @@ def use_dcld(dcld, production, cmdlist):
 @optgroup.option('--paa-trust-store-path', default='paa-root-certs', type=str, metavar='PATH', help="PAA trust store path (default: paa-root-certs)")
 def main(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_http, paa_trust_store_path):
     """DCL PAA mirroring tools"""
-    fetch_certs(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_http, paa_trust_store_path)
+    fetch_paa_certs(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_http, paa_trust_store_path)
+
+
+def get_cert_from_rest(rest_node_url, subject, subject_key_id):
+    response = requests.get(
+        f"{rest_node_url}/dcl/pki/certificates/{subject}/{subject_key_id}").json()["approvedCertificates"]["certs"][0]
+    certificate = response["pemCert"].rstrip("\n")
+    subject = response["subjectAsText"]
+    return certificate, subject
+
+
+def fetch_cd_signing_certs(store_path):
+    ''' Only supports using main net http currently.'''
+    rest_node_url = PRODUCTION_NODE_URL_REST
+    os.makedirs(store_path, exist_ok=True)
+    original_dir = os.getcwd()
+    os.chdir(store_path)
+
+    cd_signer_ids = requests.get(f"{rest_node_url}/dcl/pki/child-certificates/{MATTER_CERT_CA_SUBJECT}/{MATTER_CERT_CA_SUBJECT_KEY_ID}").json()['childCertificates']['certIds']
+    for signer in cd_signer_ids:
+        subject = signer['subject']
+        subject_key_id = signer['subjectKeyId']
+        certificate, subject = get_cert_from_rest(rest_node_url, subject, subject_key_id)
+
+        print(f"Downloaded CD signing cert with subject: {subject}")
+        write_cert(certificate, subject)
+
+    os.chdir(original_dir)
 
 
-def fetch_certs(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_http, paa_trust_store_path):
+def fetch_paa_certs(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_test_net_http, paa_trust_store_path):
     production = False
     dcld = use_test_net_dcld
 
@@ -171,10 +202,7 @@ def fetch_certs(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_tes
 
     for paa in paa_list:
         if use_rest:
-            response = requests.get(
-                f"{rest_node_url}/dcl/pki/certificates/{paa['subject']}/{paa['subjectKeyId']}").json()["approvedCertificates"]["certs"][0]
-            certificate = response["pemCert"]
-            subject = response["subjectAsText"]
+            certificate, subject = get_cert_from_rest(rest_node_url, paa['subject'], paa['subjectKeyId'])
         else:
             cmdlist = ['query', 'pki', 'x509-cert', '-u',
                        paa['subject'], '-k', paa['subjectKeyId']]
@@ -186,8 +214,8 @@ def fetch_certs(use_main_net_dcld, use_test_net_dcld, use_main_net_http, use_tes
 
         certificate = certificate.rstrip('\n')
 
-        print(f"Downloaded certificate with subject: {subject}")
-        write_paa_root_cert(certificate, subject)
+        print(f"Downloaded PAA certificate with subject: {subject}")
+        write_cert(certificate, subject)
 
     os.chdir(original_dir)
 

src/python_testing/IDT_test_wrapper/post-cert-checks.py

@@ -18,6 +18,7 @@
 import cv2
 import importlib
 import os
+import requests
 import shutil
 import sys
 import time
@@ -153,21 +154,24 @@ def get_setup_code() -> (str, bool):
 class TestConfig(object):
     def __init__(self, code: str, code_type: SetupCodeType):
         tmp_uuid = str(uuid.uuid4())
-        tmpdir_name = f'paas_{tmp_uuid}'
-        path = os.path.join('.', tmpdir_name)
-        os.mkdir(path)
-        fetch_paa_certs_from_dcl.fetch_certs(use_main_net_dcld='', use_test_net_dcld='', use_main_net_http=True, use_test_net_http=False, paa_trust_store_path=tmpdir_name)
-        self.dirname = tmpdir_name
-        self.path = path
+        tmpdir_paa = f'paas_{tmp_uuid}'
+        tmpdir_cd = f'cd_{tmp_uuid}'
+        self.paa_path = os.path.join('.', tmpdir_paa)
+        self.cd_path = os.path.join('.', tmpdir_cd)
+        os.mkdir(self.paa_path)
+        os.mkdir(self.cd_path)
+        fetch_paa_certs_from_dcl.fetch_paa_certs(use_main_net_dcld='', use_test_net_dcld='', use_main_net_http=True, use_test_net_http=False, paa_trust_store_path=tmpdir_paa)
+        fetch_paa_certs_from_dcl.fetch_cd_signing_certs(tmpdir_cd)
         self.admin_storage = f'admin_storage_{tmp_uuid}.json'
         global_test_params = {'use_pase_only': True, 'post_cert_test': True}
-        # TODO: Set the cd_cert_dir to a directory with ONLY the production key signer certs - get from DCL.
         self.config = MatterTestConfig(endpoint=0, dut_node_ids=[1], global_test_params=global_test_params, storage_path=self.admin_storage)
         if code_type == SetupCodeType.QR:
             self.config.qr_code_content = code
         else:
             self.config.manual_code = code
-        self.config.paa_trust_store_path = Path(self.path)
+        self.config.paa_trust_store_path = Path(self.paa_path)
+        # Set for DA-1.2, which uses the CD signing certs for verification. This test is now set to use the production CD signing certs from the DCL.
+        self.config.global_test_params['cd_cert_dir'] = tmpdir_cd
         self.stack = MatterStackState(self.config)
         self.default_controller = self.stack.certificate_authorities[0].adminList[0].NewController(
             nodeId=112233,
@@ -191,7 +195,8 @@ def __exit__(self, *args):
         self.default_controller.Shutdown()
         self.stack.Shutdown()
         os.remove(self.admin_storage)
-        shutil.rmtree(self.path)
+        shutil.rmtree(self.paa_path)
+        shutil.rmtree(self.cd_path)
 
 
 def run_test(test_class: MatterBaseTest, test_name: str, test_config: TestConfig) -> list[str]: