TC-OPCREDS 3.5: Automate

src/controller/python/OpCredsBinding.cpp

@@ -83,6 +83,8 @@ class OperationalCredentialsAdapter : public OperationalCredentialsDelegate
 
     void SetMaximallyLargeCertsUsed(bool enabled) { mExampleOpCredsIssuer.SetMaximallyLargeCertsUsed(enabled); }
 
+    void SetCertificateValidityPeriod(uint32_t validity) { mExampleOpCredsIssuer.SetCertificateValidityPeriod(validity); }
+
 private:
     CHIP_ERROR GenerateNOCChain(const ByteSpan & csrElements, const ByteSpan & csrNonce, const ByteSpan & attestationSignature,
                                 const ByteSpan & attestationChallenge, const ByteSpan & DAC, const ByteSpan & PAI,
@@ -607,6 +609,15 @@ PyChipError pychip_OpCreds_SetMaximallyLargeCertsUsed(OpCredsContext * context,
     return ToPyChipError(CHIP_NO_ERROR);
 }
 
+PyChipError pychip_OpCreds_SetCertificateValidityPeriod(OpCredsContext * context, uint32_t validity)
+{
+    VerifyOrReturnError(context != nullptr && context->mAdapter != nullptr, ToPyChipError(CHIP_ERROR_INCORRECT_STATE));
+
+    context->mAdapter->SetCertificateValidityPeriod(validity);
+
+    return ToPyChipError(CHIP_NO_ERROR);
+}
+
 void pychip_OpCreds_FreeDelegate(OpCredsContext * context)
 {
     Platform::Delete(context);

src/controller/python/chip/CertificateAuthority.py

@@ -21,6 +21,7 @@
 import ctypes
 import logging
 from ctypes import c_void_p
+from datetime import timedelta
 from typing import List, Optional
 
 import chip.exceptions
@@ -30,6 +31,9 @@
 
 LOGGER = logging.getLogger(__name__)
 
+# By default, let's set certificate validity to 10 years.
+CERTIFICATE_VALIDITY_PERIOD_SEC = int(timedelta(days=10*365).total_seconds())
+
 
 class CertificateAuthority:
     '''  This represents an operational Root Certificate Authority (CA) with a root key key pair with associated
@@ -77,11 +81,15 @@ def __init__(self, chipStack: ChipStack.ChipStack, caIndex: int, persistentStora
         self._Handle().pychip_OpCreds_SetMaximallyLargeCertsUsed.restype = PyChipError
         self._Handle().pychip_OpCreds_SetMaximallyLargeCertsUsed.argtypes = [ctypes.c_void_p, ctypes.c_bool]
 
+        self._Handle().pychip_OpCreds_SetCertificateValidityPeriod.restype = PyChipError
+        self._Handle().pychip_OpCreds_SetCertificateValidityPeriod.argtypes = [ctypes.c_void_p, ctypes.c_uint32]
+
         if (persistentStorage is None):
             persistentStorage = self._chipStack.GetStorageManager()
 
         self._persistentStorage = persistentStorage
         self._maximizeCertChains = False
+        self._certificateValidityPeriodSec = CERTIFICATE_VALIDITY_PERIOD_SEC
 
         self._closure = self._chipStack.Call(
             lambda: self._Handle().pychip_OpCreds_InitializeDelegate(
@@ -189,6 +197,10 @@ def adminList(self) -> list[FabricAdmin.FabricAdmin]:
     def maximizeCertChains(self) -> bool:
         return self._maximizeCertChains
 
+    @property
+    def certificateValidityPeriodSec(self) -> int:
+        return self._certificateValidityPeriodSec
+
     @maximizeCertChains.setter
     def maximizeCertChains(self, enabled: bool):
         self._chipStack.Call(
@@ -197,6 +209,17 @@ def maximizeCertChains(self, enabled: bool):
 
         self._maximizeCertChains = enabled
 
+    @certificateValidityPeriodSec.setter
+    def certificateValidityPeriodSec(self, validity: int):
+        if validity < 0:
+            raise ValueError("Validity period must be a non-negative integer")
+
+        self._chipStack.Call(
+            lambda: self._Handle().pychip_OpCreds_SetCertificateValidityPeriod(ctypes.c_void_p(self._closure), ctypes.c_uint32(validity))
+        ).raise_on_error()
+
+        self._certificateValidityPeriodSec = validity
+
     def __del__(self):
         self.Shutdown()
 
@@ -258,7 +281,7 @@ def LoadAuthoritiesFromStorage(self):
             ca = self.NewCertificateAuthority(int(caIndex))
             ca.LoadFabricAdminsFromStorage()
 
-    def NewCertificateAuthority(self, caIndex: Optional[int] = None, maximizeCertChains: bool = False):
+    def NewCertificateAuthority(self, caIndex: Optional[int] = None, maximizeCertChains: bool = False, certificateValidityPeriodSec: Optional[int] = None):
         ''' Creates a new CertificateAuthority instance with the provided CA Index and the PersistentStorage
             instance previously setup in the constructor.
 
@@ -282,8 +305,12 @@ def NewCertificateAuthority(self, caIndex: Optional[int] = None, maximizeCertCha
             caList[str(caIndex)] = []
             self._persistentStorage.SetReplKey(key='caList', value=caList)
 
+        if certificateValidityPeriodSec is None:
+            certificateValidityPeriodSec = CERTIFICATE_VALIDITY_PERIOD_SEC
+
         ca = CertificateAuthority(chipStack=self._chipStack, caIndex=caIndex, persistentStorage=self._persistentStorage)
         ca.maximizeCertChains = maximizeCertChains
+        ca.certificateValidityPeriodSec = certificateValidityPeriodSec
         self._activeCaList.append(ca)
 
         return ca

src/controller/python/chip/utils/CommissioningBuildingBlocks.py

@@ -20,7 +20,7 @@
 import typing
 
 import chip.clusters as Clusters
-from chip.ChipDeviceCtrl import ChipDeviceController as ChipDeviceController
+from chip.ChipDeviceCtrl import ChipDeviceController, NOCChain
 from chip.clusters import GeneralCommissioning as generalCommissioning
 from chip.clusters import OperationalCredentials as opCreds
 from chip.clusters.Types import NullValue
@@ -144,7 +144,7 @@ async def CreateControllersOnFabric(fabricAdmin: FabricAdmin,
     return controllerList
 
 
-async def AddNOCForNewFabricFromExisting(commissionerDevCtrl, newFabricDevCtrl, existingNodeId, newNodeId):
+async def AddNOCForNewFabricFromExisting(commissionerDevCtrl, newFabricDevCtrl, existingNodeId, newNodeId) -> tuple[bool, typing.Optional[Clusters.OperationalCredentials.Commands.NOCResponse], typing.Optional[NOCChain]]:
     ''' Perform sequence to commission new fabric using existing commissioned fabric.
 
     Args:
@@ -156,7 +156,7 @@ async def AddNOCForNewFabricFromExisting(commissionerDevCtrl, newFabricDevCtrl,
         newNodeId (int): Node ID to use for the target node on the new fabric.
 
     Return:
-        tuple:  (bool, Optional[nocResp], Optional[rcacResp]: True if successful, False otherwise, along with nocResp, rcacResp value.
+        tuple:  (bool, Optional[nocResp], Optional[NOCChain]: True if successful, False otherwise, along with nocResp, rcacResp value.
 
     '''
     nocResp = None
@@ -173,7 +173,7 @@ async def AddNOCForNewFabricFromExisting(commissionerDevCtrl, newFabricDevCtrl,
             chainForAddNOC.nocBytes is None or chainForAddNOC.ipkBytes is None):
         # Expiring the failsafe timer in an attempt to clean up.
         await commissionerDevCtrl.SendCommand(existingNodeId, 0, generalCommissioning.Commands.ArmFailSafe(0))
-        return False, nocResp
+        return False, nocResp, None
 
     await commissionerDevCtrl.SendCommand(existingNodeId, 0, opCreds.Commands.AddTrustedRootCertificate(chainForAddNOC.rcacBytes))
     nocResp = await commissionerDevCtrl.SendCommand(existingNodeId,
@@ -184,8 +184,6 @@ async def AddNOCForNewFabricFromExisting(commissionerDevCtrl, newFabricDevCtrl,
                                                                             newFabricDevCtrl.nodeId,
                                                                             newFabricDevCtrl.fabricAdmin.vendorId))
 
-    rcacResp = chainForAddNOC.rcacBytes
-
     if nocResp.statusCode is not opCreds.Enums.NodeOperationalCertStatusEnum.kOk:
         # Expiring the failsafe timer in an attempt to clean up.
         await commissionerDevCtrl.SendCommand(existingNodeId, 0, generalCommissioning.Commands.ArmFailSafe(0))
@@ -196,12 +194,12 @@ async def AddNOCForNewFabricFromExisting(commissionerDevCtrl, newFabricDevCtrl,
     if resp.errorCode is not generalCommissioning.Enums.CommissioningErrorEnum.kOk:
         # Expiring the failsafe timer in an attempt to clean up.
         await commissionerDevCtrl.SendCommand(existingNodeId, 0, generalCommissioning.Commands.ArmFailSafe(0))
-        return False, nocResp
+        return False, nocResp, chainForAddNOC
 
     if not await _IsNodeInFabricList(newFabricDevCtrl, newNodeId):
-        return False, nocResp
+        return False, nocResp, chainForAddNOC
 
-    return True, nocResp, rcacResp
+    return True, nocResp, chainForAddNOC
 
 
 async def UpdateNOC(devCtrl, existingNodeId, newNodeId):

src/python_testing/TC_OPCREDS_3_2.py

@@ -94,10 +94,11 @@ async def test_TC_OPCREDS_3_2(self):
 
         cr2_new_admin_ctrl = cr2_new_fabric_admin.NewController(
             nodeId=cr2_nodeid)
-        success, nocResp, rcacResp = await CommissioningBuildingBlocks.AddNOCForNewFabricFromExisting(
+        success, nocResp, chain = await CommissioningBuildingBlocks.AddNOCForNewFabricFromExisting(
             commissionerDevCtrl=dev_ctrl, newFabricDevCtrl=cr2_new_admin_ctrl,
             existingNodeId=self.dut_node_id, newNodeId=cr2_dut_node_id
         )
+        rcacResp = chain.rcacBytes
 
         fabric_index_CR2 = nocResp.fabricIndex
         tlvReaderRCAC_CR2 = TLVReader(rcacResp).get()["Any"]
@@ -114,10 +115,11 @@ async def test_TC_OPCREDS_3_2(self):
 
         cr3_new_admin_ctrl = cr3_new_fabric_admin.NewController(
             nodeId=cr3_nodeid)
-        success, nocResp, rcacResp = await CommissioningBuildingBlocks.AddNOCForNewFabricFromExisting(
+        success, nocResp, chain = await CommissioningBuildingBlocks.AddNOCForNewFabricFromExisting(
             commissionerDevCtrl=dev_ctrl, newFabricDevCtrl=cr3_new_admin_ctrl,
             existingNodeId=self.dut_node_id, newNodeId=cr3_dut_node_id
         )
+        rcacResp = chain.rcacBytes
 
         fabric_index_CR3 = nocResp.fabricIndex
         tlvReaderRCAC_CR3 = TLVReader(rcacResp).get()["Any"]