1# ------------------------------------ 2# Copyright (c) Microsoft Corporation. 3# Licensed under the MIT License. 4# ------------------------------------ 5import asyncio 6import os 7from azure.keyvault.certificates import CertificatePolicy 8from azure.keyvault.certificates.aio import CertificateClient 9from azure.identity.aio import DefaultAzureCredential 10from azure.core.exceptions import HttpResponseError 11 12# ---------------------------------------------------------------------------------------------------------- 13# Prerequisites: 14# 1. An Azure Key Vault (https://docs.microsoft.com/en-us/azure/key-vault/quick-create-cli) 15# 16# 2. azure-keyvault-certificates and azure-identity packages (pip install these) 17# 18# 3. Set Environment variables AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_CLIENT_SECRET, VAULT_URL 19# (See https://github.com/Azure/azure-sdk-for-python/tree/master/sdk/keyvault/azure-keyvault-keys#authenticate-the-client) 20# 21# ---------------------------------------------------------------------------------------------------------- 22# Sample - demonstrates the basic list operations on a vault(certificate) resource for Azure Key Vault. 23# The vault has to be soft-delete enabled to perform one of the following operations: https://docs.microsoft.com/en-us/azure/key-vault/key-vault-ovw-soft-delete 24# 25# 1. Create certificate (create_certificate) 26# 27# 2. List certificates from the Key Vault (list_properties_of_certificates) 28# 29# 3. List certificate versions from the Key Vault (list_properties_of_certificate_versions) 30# 31# 4. List deleted certificates from the Key Vault (list_deleted_certificates). The vault has to be soft-delete enabled to perform this operation. 32# 33# ---------------------------------------------------------------------------------------------------------- 34 35 36async def run_sample(): 37 # Instantiate a certificate client that will be used to call the service. Notice that the client is using default Azure credentials. 38 # To make default credentials work, ensure that environment variables 'AZURE_CLIENT_ID', 39 # 'AZURE_CLIENT_SECRET' and 'AZURE_TENANT_ID' are set with the service principal credentials. 40 VAULT_URL = os.environ["VAULT_URL"] 41 credential = DefaultAzureCredential() 42 client = CertificateClient(vault_url=VAULT_URL, credential=credential) 43 try: 44 # Let's create a certificate for holding storage and bank accounts credentials. If the certificate 45 # already exists in the Key Vault, then a new version of the certificate is created. 46 print("\n.. Create Certificate") 47 bank_cert_name = "BankListCertificate" 48 storage_cert_name = "StorageListCertificate" 49 50 bank_certificate = await client.create_certificate( 51 certificate_name=bank_cert_name, policy=CertificatePolicy.get_default() 52 ) 53 storage_certificate = await client.create_certificate( 54 certificate_name=storage_cert_name, policy=CertificatePolicy.get_default() 55 ) 56 57 print("Certificate with name '{0}' was created.".format(bank_certificate.name)) 58 print("Certificate with name '{0}' was created.".format(storage_certificate.name)) 59 60 # Let's list the certificates. 61 print("\n.. List certificates from the Key Vault") 62 certificates = client.list_properties_of_certificates() 63 async for certificate in certificates: 64 print("Certificate with name '{0}' was found.".format(certificate.name)) 65 66 # You've decided to add tags to the certificate you created. Calling create_certificate on an existing 67 # certificate creates a new version of the certificate in the Key Vault with the new value. 68 69 tags = {"a": "b"} 70 71 bank_certificate = await client.create_certificate( 72 certificate_name=bank_cert_name, policy=CertificatePolicy.get_default(), tags=tags 73 ) 74 print( 75 "Certificate with name '{0}' was created again with tags '{1}'".format( 76 bank_certificate.name, bank_certificate.properties.tags 77 ) 78 ) 79 80 # You need to check all the different tags your bank account certificate had previously. Lets print all the versions of this certificate. 81 print("\n.. List versions of the certificate using its name") 82 certificate_versions = client.list_properties_of_certificate_versions(bank_cert_name) 83 async for certificate_version in certificate_versions: 84 print( 85 "Bank Certificate with name '{0}' with version '{1}' has tags: '{2}'.".format( 86 certificate_version.name, certificate_version.version, certificate_version.tags 87 ) 88 ) 89 90 # The bank account and storage accounts got closed. Let's delete bank and storage accounts certificates. 91 await client.delete_certificate(bank_cert_name) 92 await client.delete_certificate(storage_cert_name) 93 94 # You can list all the deleted and non-purged certificates, assuming Key Vault is soft-delete enabled. 95 print("\n.. List deleted certificates from the Key Vault") 96 deleted_certificates = client.list_deleted_certificates() 97 async for deleted_certificate in deleted_certificates: 98 print( 99 "Certificate with name '{0}' has recovery id '{1}'".format( 100 deleted_certificate.name, deleted_certificate.recovery_id 101 ) 102 ) 103 104 except HttpResponseError as e: 105 if "(NotSupported)" in e.message: 106 print("\n{0} Please enable soft delete on Key Vault to perform this operation.".format(e.message)) 107 else: 108 print("\nrun_sample has caught an error. {0}".format(e.message)) 109 110 finally: 111 print("\nrun_sample done") 112 await credential.close() 113 await client.close() 114 115 116if __name__ == "__main__": 117 try: 118 loop = asyncio.get_event_loop() 119 loop.run_until_complete(run_sample()) 120 loop.close() 121 122 except Exception as e: 123 print("Top level Error: {0}".format(str(e))) 124