# ------------------------------------ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. # ------------------------------------ import asyncio from azure.keyvault.certificates import ApiVersion, CertificatePolicy, CertificateContentType, WellKnownIssuerNames import pytest from _shared.test_case_async import KeyVaultTestCase from _async_test_case import get_decorator, AsyncCertificatesClientPreparer from devtools_testutils.aio import recorded_by_proxy_async all_api_versions = get_decorator(is_async=True) exclude_2016_10_01 = get_decorator(is_async=True, api_versions=[v for v in ApiVersion if v != ApiVersion.V2016_10_01]) def print(*args): assert all(arg is not None for arg in args) @pytest.mark.asyncio @pytest.mark.playback_test_only("Can't run in live pipelines, and there's no reason to.") async def test_create_certificate(): vault_url = "vault_url" # pylint:disable=unused-variable # [START create_certificate_client] from azure.identity.aio import DefaultAzureCredential from azure.keyvault.certificates.aio import CertificateClient # Create a KeyVaultCertificate using default Azure credentials credential = DefaultAzureCredential() certificate_client = CertificateClient(vault_url=vault_url, credential=credential) # the client and credential should be closed when no longer needed # (both are also async context managers) await certificate_client.close() await credential.close() # [END create_certificate_client] class TestExamplesKeyVault(KeyVaultTestCase): @pytest.mark.asyncio @pytest.mark.parametrize("api_version", all_api_versions) @AsyncCertificatesClientPreparer(logging_enable=True) @recorded_by_proxy_async async def test_example_certificate_crud_operations(self, certificate_client, **kwargs): cert_name = self.get_resource_name("cert-name") # [START create_certificate] from azure.keyvault.certificates import CertificatePolicy, CertificateContentType, WellKnownIssuerNames # specify the certificate policy cert_policy = CertificatePolicy( issuer_name=WellKnownIssuerNames.self, subject="CN=*.microsoft.com", san_dns_names=["sdk.azure-int.net"], exportable=True, key_type="RSA", key_size=2048, reuse_key=False, content_type=CertificateContentType.pkcs12, validity_in_months=24, ) certificate = await certificate_client.create_certificate(certificate_name=cert_name, policy=cert_policy) print(certificate.id) print(certificate.name) print(certificate.policy.issuer_name) # [END create_certificate] # [START get_certificate] # get the latest version of a certificate certificate = await certificate_client.get_certificate(cert_name) print(certificate.id) print(certificate.name) print(certificate.policy.issuer_name) # [END get_certificate] version = certificate.properties.version # [START get_certificate_version] certificate = await certificate_client.get_certificate_version(cert_name, version) print(certificate.id) print(certificate.properties.version) # [END get_certificate_version] # [START update_certificate] # update attributes of an existing certificate tags = {"foo": "updated tag"} updated_certificate = await certificate_client.update_certificate_properties( certificate_name=certificate.name, tags=tags ) print(updated_certificate.properties.version) print(updated_certificate.properties.updated_on) print(updated_certificate.properties.tags) # [END update_certificate] # [START delete_certificate] # delete a certificate deleted_certificate = await certificate_client.delete_certificate(cert_name) print(deleted_certificate.name) # if the vault has soft-delete enabled, the certificate's # scheduled purge date, deleted_on, and recovery id are available print(deleted_certificate.deleted_on) print(deleted_certificate.scheduled_purge_date) print(deleted_certificate.recovery_id) # [END delete_certificate] @pytest.mark.asyncio @pytest.mark.parametrize("api_version", all_api_versions) @AsyncCertificatesClientPreparer(logging_enable=True) @recorded_by_proxy_async async def test_example_certificate_list_operations(self, certificate_client, **kwargs): # specify the certificate policy cert_policy = CertificatePolicy( issuer_name=WellKnownIssuerNames.self, subject="CN=*.microsoft.com", san_dns_names=["sdk.azure-int.net"], exportable=True, key_type="RSA", key_size=2048, reuse_key=False, content_type=CertificateContentType.pkcs12, validity_in_months=24, ) certificate_name = self.get_resource_name("cert") await certificate_client.create_certificate(certificate_name, cert_policy) # [START list_properties_of_certificates] # list certificates certificates = certificate_client.list_properties_of_certificates() async for certificate in certificates: print(certificate.id) print(certificate.created_on) print(certificate.name) print(certificate.updated_on) print(certificate.enabled) # [END list_properties_of_certificates] # create a second version of the cert await certificate_client.create_certificate(certificate_name, cert_policy) # [START list_properties_of_certificate_versions] # get an iterator of all versions of a certificate certificate_versions = certificate_client.list_properties_of_certificate_versions(certificate_name) async for certificate in certificate_versions: print(certificate.id) print(certificate.updated_on) print(certificate.version) # [END list_properties_of_certificate_versions] await certificate_client.delete_certificate(certificate_name) # [START list_deleted_certificates] # get an iterator of deleted certificates (requires soft-delete enabled for the vault) deleted_certificates = certificate_client.list_deleted_certificates() async for certificate in deleted_certificates: print(certificate.id) print(certificate.name) print(certificate.scheduled_purge_date) print(certificate.recovery_id) print(certificate.deleted_on) # [END list_deleted_certificates] @pytest.mark.asyncio @pytest.mark.parametrize("api_version", exclude_2016_10_01) @AsyncCertificatesClientPreparer(logging_enable=True) @recorded_by_proxy_async async def test_example_certificate_backup_restore(self, certificate_client, **kwargs): # specify the certificate policy cert_policy = CertificatePolicy( issuer_name=WellKnownIssuerNames.self, subject="CN=*.microsoft.com", san_dns_names=["sdk.azure-int.net"], exportable=True, key_type="RSA", key_size=2048, reuse_key=False, content_type=CertificateContentType.pkcs12, validity_in_months=24, ) cert_name = self.get_resource_name("cert-name") create_certificate_poller = certificate_client.create_certificate( certificate_name=cert_name, policy=cert_policy ) await create_certificate_poller # [START backup_certificate] # backup certificate certificate_backup = await certificate_client.backup_certificate(cert_name) # returns the raw bytes of the backed up certificate print(certificate_backup) # [END backup_certificate] await certificate_client.delete_certificate(certificate_name=cert_name) await certificate_client.purge_deleted_certificate(certificate_name=cert_name) if self.is_live: await asyncio.sleep(60) # [START restore_certificate] # restores a certificate backup restored_certificate = await certificate_client.restore_certificate_backup(certificate_backup) print(restored_certificate.id) print(restored_certificate.name) print(restored_certificate.properties.version) # [END restore_certificate] @pytest.mark.asyncio @pytest.mark.parametrize("api_version", all_api_versions) @AsyncCertificatesClientPreparer(logging_enable=True) @recorded_by_proxy_async async def test_example_certificate_recover(self, certificate_client, **kwargs): # specify the certificate policy cert_policy = CertificatePolicy( issuer_name=WellKnownIssuerNames.self, subject="CN=*.microsoft.com", san_dns_names=["sdk.azure-int.net"], exportable=True, key_type="RSA", key_size=2048, reuse_key=False, content_type=CertificateContentType.pkcs12, validity_in_months=24, ) cert_name = self.get_resource_name("cert-name") create_certificate_poller = certificate_client.create_certificate( certificate_name=cert_name, policy=cert_policy ) await create_certificate_poller await certificate_client.delete_certificate(certificate_name=cert_name) # [START get_deleted_certificate] # get a deleted certificate (requires soft-delete enabled for the vault) deleted_certificate = await certificate_client.get_deleted_certificate(cert_name) print(deleted_certificate.name) # [END get_deleted_certificate] # [START recover_deleted_certificate] # recover deleted certificate to its latest version (requires soft-delete enabled for the vault) recovered_certificate = await certificate_client.recover_deleted_certificate(cert_name) print(recovered_certificate.id) print(recovered_certificate.name) # [END recover_deleted_certificate] @pytest.mark.asyncio @pytest.mark.parametrize("api_version", all_api_versions) @AsyncCertificatesClientPreparer(logging_enable=True) @recorded_by_proxy_async async def test_example_contacts(self, certificate_client, **kwargs): # [START set_contacts] from azure.keyvault.certificates import CertificateContact # Create a list of the contacts that you want to set for this key vault. contact_list = [ CertificateContact(email="admin@contoso.com", name="John Doe", phone="1111111111"), CertificateContact(email="admin2@contoso.com", name="John Doe2", phone="2222222222"), ] contacts = await certificate_client.set_contacts(contact_list) for contact in contacts: print(contact.name) print(contact.email) print(contact.phone) # [END set_contacts] # [START get_contacts] contacts = await certificate_client.get_contacts() # Loop through the certificate contacts for this key vault. for contact in contacts: print(contact.name) print(contact.email) print(contact.phone) # [END get_contacts] # [START delete_contacts] deleted_contacts = await certificate_client.delete_contacts() for deleted_contact in deleted_contacts: print(deleted_contact.name) print(deleted_contact.email) print(deleted_contact.phone) # [END delete_contacts] @pytest.mark.asyncio @pytest.mark.parametrize("api_version", all_api_versions) @AsyncCertificatesClientPreparer(logging_enable=True) @recorded_by_proxy_async async def test_example_issuers(self, certificate_client, **kwargs): # [START create_issuer] from azure.keyvault.certificates import AdministratorContact # First we specify the AdministratorContact for a issuer. admin_contacts = [ AdministratorContact(first_name="John", last_name="Doe", email="admin@microsoft.com", phone="4255555555") ] issuer = await certificate_client.create_issuer( issuer_name="issuer1", provider="Test", account_id="keyvaultuser", admin_contacts=admin_contacts, enabled=True, ) print(issuer.name) print(issuer.provider) print(issuer.account_id) for contact in issuer.admin_contacts: print(contact.first_name) print(contact.last_name) print(contact.email) print(contact.phone) # [END create_issuer] # [START get_issuer] issuer = await certificate_client.get_issuer("issuer1") print(issuer.name) print(issuer.provider) print(issuer.account_id) for contact in issuer.admin_contacts: print(contact.first_name) print(contact.last_name) print(contact.email) print(contact.phone) # [END get_issuer] await certificate_client.create_issuer( issuer_name="issuer2", provider="Test", account_id="keyvaultuser", enabled=True ) # [START list_properties_of_issuers] issuers = certificate_client.list_properties_of_issuers() async for issuer in issuers: print(issuer.name) print(issuer.provider) # [END list_properties_of_issuers] # [START delete_issuer] deleted_issuer = await certificate_client.delete_issuer("issuer1") print(deleted_issuer.name) print(deleted_issuer.provider) print(deleted_issuer.account_id) for contact in deleted_issuer.admin_contacts: print(contact.first_name) print(contact.last_name) print(contact.email) print(contact.phone) # [END delete_issuer]