in src/aws_encryption_sdk/streaming_client.py [0:0]
def _has_mpl_attrs_post_init(self):
"""If the MPL is present in the runtime, perform MPL-specific post-init logic
to validate the new object has a valid state.
"""
if not exactly_one_arg_is_not_none(self.materials_manager, self.key_provider, self.keyring):
raise TypeError("Exactly one of keyring, materials_manager, or key_provider must be provided")
if self.materials_manager is None:
if self.key_provider is not None:
# No CMM, provided legacy native `key_provider` => create legacy native DefaultCryptoMaterialsManager
self.materials_manager = DefaultCryptoMaterialsManager(
master_key_provider=self.key_provider
)
elif self.keyring is not None:
try:
mat_prov: AwsCryptographicMaterialProviders = AwsCryptographicMaterialProviders(
config=MaterialProvidersConfig()
)
cmm = mat_prov.create_default_cryptographic_materials_manager(
CreateDefaultCryptographicMaterialsManagerInput(
keyring=self.keyring
)
)
cmm_handler: CryptoMaterialsManager = CryptoMaterialsManagerFromMPL(cmm)
self.materials_manager = cmm_handler
except AwsCryptographicMaterialProvidersException as mpl_exception:
# Wrap MPL error into the ESDK error type
# so customers only have to catch ESDK error types.
raise AWSEncryptionSDKClientError(mpl_exception)
# If the provided materials_manager is directly from the MPL, wrap it in a native interface
# for internal use.
elif (self.materials_manager is not None
and isinstance(self.materials_manager, MPL_ICryptographicMaterialsManager)):
self.materials_manager = CryptoMaterialsManagerFromMPL(self.materials_manager)