blob: bce921cfb785bf236e11ceaca2baf71123de6dab [file] [log] [blame]
#
# Copyright (c) 2021 Project CHIP Authors
# All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Needed to use types in type hints before they are fully defined.
from __future__ import annotations
import ctypes
import logging
from ctypes import c_void_p
from typing import List
import chip.exceptions
from chip import ChipStack, FabricAdmin
from chip.native import PyChipError
from chip.storage import PersistentStorage
class CertificateAuthority:
''' This represents an operational Root Certificate Authority (CA) with a root key key pair with associated
public key (i.e "Root PK") . This manages a set of FabricAdmin objects, each administering a fabric identified
by a unique FabricId scoped to it.
Each CertificateAuthority instance is tied to a 'CA index' that is used to look-up the list of fabrics already setup
previously in the provided PersistentStorage object.
>> C++ Binding Details
Each CertificateAuthority instance is associated with a single instance of the OperationalCredentialsAdapter.
This adapter instance implements the OperationalCredentialsDelegate and is meant to provide a Python adapter to the
functions in that delegate. It relies on the in-built ExampleOperationalCredentialsIssuer to then generate certificate
material for the CA. This instance also uses the 'CA index' to store/look-up the associated credential material from
the provided PersistentStorage object.
'''
@classmethod
def _Handle(cls):
return chip.native.GetLibraryHandle()
@classmethod
def logger(cls):
return logging.getLogger('CertificateAuthority')
def __init__(self, chipStack: ChipStack.ChipStack, caIndex: int, persistentStorage: PersistentStorage = None):
''' Initializes the CertificateAuthority. This will set-up the associated C++ OperationalCredentialsAdapter
as well.
Arguments:
chipStack: A reference to a chip.ChipStack object.
caIndex: An index used to look-up details about stored credential material and fabrics
from persistent storage.
persistentStorage: An optional reference to a PersistentStorage object. If one is provided, it will pick that over
the default PersistentStorage object retrieved from the chipStack.
'''
self.logger().warning(f"New CertificateAuthority at index {caIndex}")
self._chipStack = chipStack
self._caIndex = caIndex
self._Handle().pychip_OpCreds_InitializeDelegate.restype = c_void_p
self._Handle().pychip_OpCreds_InitializeDelegate.argtypes = [ctypes.py_object, ctypes.c_uint32, ctypes.c_void_p]
self._Handle().pychip_OpCreds_SetMaximallyLargeCertsUsed.restype = PyChipError
self._Handle().pychip_OpCreds_SetMaximallyLargeCertsUsed.argtypes = [ctypes.c_void_p, ctypes.c_bool]
if (persistentStorage is None):
persistentStorage = self._chipStack.GetStorageManager()
self._persistentStorage = persistentStorage
self._maximizeCertChains = False
self._closure = self._chipStack.Call(
lambda: self._Handle().pychip_OpCreds_InitializeDelegate(
ctypes.py_object(self), ctypes.c_uint32(self._caIndex), self._persistentStorage.GetSdkStorageObject())
)
if (self._closure is None):
raise ValueError("Encountered error initializing OpCreds adapter")
self._isActive = True
self._activeAdmins = []
def LoadFabricAdminsFromStorage(self):
''' If FabricAdmins had been setup previously, this re-creates them using information from persistent storage.
Otherwise, it initializes the REPL keys in persistent storage to sane defaults. This includes a top-level
key identifying the CA (using the associated CA Index) initialized to an empty list.
This expects a 'caList' key to be present in the REPL config.
Each FabricAdmin that is added there-after will insert a dictionary item into that list containing
'fabricId' and 'vendorId' keys.
'''
if (not (self._isActive)):
raise RuntimeError("Object isn't active")
self.logger().warning("Loading fabric admins from storage...")
caList = self._persistentStorage.GetReplKey(key='caList')
if (str(self._caIndex) not in caList):
caList[str(self._caIndex)] = []
self._persistentStorage.SetReplKey(key='caList', value=caList)
fabricAdminMetadataList = self._persistentStorage.GetReplKey(key='caList')[str(self._caIndex)]
for adminMetadata in fabricAdminMetadataList:
self.NewFabricAdmin(vendorId=int(adminMetadata['vendorId']), fabricId=int(adminMetadata['fabricId']))
def NewFabricAdmin(self, vendorId: int, fabricId: int):
''' Creates a new FabricAdmin object initialized with the provided vendorId and fabricId values.
This will update the REPL keys in persistent storage IF a 'caList' key is present. If it isn't,
will avoid making any updates.
'''
if (not (self._isActive)):
raise RuntimeError(
"CertificateAuthority object was previously shutdown and is no longer valid!")
if (vendorId is None or fabricId is None):
raise ValueError("Invalid values for fabricId and vendorId")
for existingAdmin in self._activeAdmins:
if (existingAdmin.fabricId == fabricId):
raise ValueError(f"Provided fabricId of {fabricId} collides with an existing FabricAdmin instance!")
fabricAdmin = FabricAdmin.FabricAdmin(self, vendorId=vendorId, fabricId=fabricId)
caList = self._persistentStorage.GetReplKey('caList')
if (caList is not None):
replFabricEntry = {'fabricId': fabricId, 'vendorId': vendorId}
if (replFabricEntry not in caList[str(self._caIndex)]):
caList[str(self._caIndex)].append(replFabricEntry)
self._persistentStorage.SetReplKey(key='caList', value=caList)
self._activeAdmins.append(fabricAdmin)
return fabricAdmin
def Shutdown(self):
''' Shuts down all active FabricAdmin objects managed by this CertificateAuthority before
shutting itself down.
You cannot interact with this object there-after.
'''
if (self._isActive):
for admin in self._activeAdmins:
admin.Shutdown()
self._activeAdmins = []
self._Handle().pychip_OpCreds_FreeDelegate.argtypes = [ctypes.c_void_p]
self._chipStack.Call(
lambda: self._Handle().pychip_OpCreds_FreeDelegate(
ctypes.c_void_p(self._closure))
)
self._isActive = False
def GetOpCredsContext(self):
''' Returns a pointer to the underlying C++ OperationalCredentialsAdapter.
'''
if (not (self._isActive)):
raise RuntimeError("Object isn't active")
return self._closure
@property
def caIndex(self) -> int:
return self._caIndex
@property
def adminList(self) -> list[FabricAdmin.FabricAdmin]:
return self._activeAdmins
@property
def maximizeCertChains(self) -> bool:
return self._maximizeCertChains
@maximizeCertChains.setter
def maximizeCertChains(self, enabled: bool):
self._chipStack.Call(
lambda: self._Handle().pychip_OpCreds_SetMaximallyLargeCertsUsed(ctypes.c_void_p(self._closure), ctypes.c_bool(enabled))
).raise_on_error()
self._maximizeCertChains = enabled
def __del__(self):
self.Shutdown()
class CertificateAuthorityManager:
''' Manages a set of CertificateAuthority instances.
'''
@classmethod
def _Handle(cls):
return chip.native.GetLibraryHandle()
@classmethod
def logger(cls):
return logging.getLogger('CertificateAuthorityManager')
def __init__(self, chipStack: ChipStack.ChipStack, persistentStorage: PersistentStorage = None):
''' Initializes the manager.
chipStack: Reference to a chip.ChipStack object that is used to initialize
CertificateAuthority instances.
persistentStorage: If provided, over-rides the default instance in the provided chipStack
when initializing CertificateAuthority instances.
'''
self._activeCaIndexList = []
self._chipStack = chipStack
if (persistentStorage is None):
persistentStorage = self._chipStack.GetStorageManager()
self._persistentStorage = persistentStorage
self._activeCaList = []
self._isActive = True
def _AllocateNextCaIndex(self):
''' Allocate the next un-used CA index.
'''
nextCaIndex = 1
for ca in self._activeCaList:
nextCaIndex = ca.caIndex + 1
return nextCaIndex
def LoadAuthoritiesFromStorage(self):
''' Loads any existing CertificateAuthority instances present in persistent storage.
If the 'caList' key is not present in the REPL config, it will create one.
'''
if (not (self._isActive)):
raise RuntimeError("Object is not active")
self.logger().warning("Loading certificate authorities from storage...")
#
# Persist details to storage (read modify write).
#
caList = self._persistentStorage.GetReplKey('caList')
if (caList is None):
caList = {}
for caIndex in caList:
ca = self.NewCertificateAuthority(int(caIndex))
ca.LoadFabricAdminsFromStorage()
def NewCertificateAuthority(self, caIndex: int = None, maximizeCertChains: bool = False):
''' Creates a new CertificateAuthority instance with the provided CA Index and the PersistentStorage
instance previously setup in the constructor.
This will write to the REPL keys in persistent storage to setup an empty list for the 'CA Index'
item.
'''
if (not (self._isActive)):
raise RuntimeError("Object is not active")
if (caIndex is None):
caIndex = self._AllocateNextCaIndex()
#
# Persist details to storage (read modify write).
#
caList = self._persistentStorage.GetReplKey('caList')
if (caList is None):
caList = {}
if (str(caIndex) not in caList):
caList[str(caIndex)] = []
self._persistentStorage.SetReplKey(key='caList', value=caList)
ca = CertificateAuthority(chipStack=self._chipStack, caIndex=caIndex, persistentStorage=self._persistentStorage)
ca.maximizeCertChains = maximizeCertChains
self._activeCaList.append(ca)
return ca
def Shutdown(self):
''' Shuts down all active CertificateAuthority instances tracked by this manager, before shutting itself down.
You cannot interact with this object there-after.
'''
for ca in self._activeCaList:
ca.Shutdown()
self._activeCaList = []
self._isActive = False
@property
def activeCaList(self) -> List[CertificateAuthority]:
return self._activeCaList
def __del__(self):
self.Shutdown()