|
| 1 | +# ------------------------------------ |
| 2 | +# Copyright (c) Microsoft Corporation. |
| 3 | +# Licensed under the MIT License. |
| 4 | +# ------------------------------------ |
| 5 | +import asyncio |
| 6 | +import os |
| 7 | +from azure.identity.aio import DefaultAzureCredential |
| 8 | +from azure.keyvault.certificates import CertificateContentType, CertificatePolicy, WellKnownIssuerNames |
| 9 | +from azure.keyvault.certificates.aio import CertificateClient |
| 10 | + |
| 11 | +# ---------------------------------------------------------------------------------------------------------- |
| 12 | +# Prerequisites: |
| 13 | +# 1. An Azure Key Vault (https://docs.microsoft.com/en-us/azure/key-vault/quick-create-cli) |
| 14 | +# |
| 15 | +# 2. azure-keyvault-certificates and azure-identity packages (pip install these) |
| 16 | +# |
| 17 | +# 3. Set up your environment to use azure-identity's DefaultAzureCredential. To authenticate a service principal with |
| 18 | +# environment variables, set AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, and AZURE_TENANT_ID |
| 19 | +# (See https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/keyvault/azure-keyvault-administration#authenticate-the-client) |
| 20 | +# |
| 21 | +# 4. A PFX certificate on your machine. Set an environment variable, PFX_CERT_PATH, with the path to this certificate. |
| 22 | +# |
| 23 | +# 5. A PEM-formatted certificate on your machine. Set an environment variable, PEM_CERT_PATH, with the path to this |
| 24 | +# certificate. |
| 25 | +# |
| 26 | +# ---------------------------------------------------------------------------------------------------------- |
| 27 | +# Sample - demonstrates importing a PFX and PEM-formatted certificate into Azure Key Vault |
| 28 | +# |
| 29 | +# 1. Import an existing PFX certificate (import_certificate) |
| 30 | +# |
| 31 | +# 2. Import an existing PEM-formatted certificate (import_certificate) |
| 32 | +# |
| 33 | +# ---------------------------------------------------------------------------------------------------------- |
| 34 | + |
| 35 | +async def run_sample(): |
| 36 | + # Instantiate a certificate client that will be used to call the service. |
| 37 | + # Here we use the DefaultAzureCredential, but any azure-identity credential can be used. |
| 38 | + VAULT_URL = os.environ["VAULT_URL"] |
| 39 | + credential = DefaultAzureCredential() |
| 40 | + client = CertificateClient(vault_url=VAULT_URL, credential=credential) |
| 41 | + |
| 42 | + # Let's import a PFX certificate first. |
| 43 | + # Assuming you already have a PFX containing your key pair, you can import it into Key Vault. |
| 44 | + # You can do this without setting a policy, but the policy is needed if you want the private key to be exportable |
| 45 | + # or to configure actions when a certificate is close to expiration. |
| 46 | + pfx_cert_name = "pfxCert" |
| 47 | + with open(os.environ["PFX_CERT_PATH"], "rb") as f: |
| 48 | + pfx_cert_bytes = f.read() |
| 49 | + imported_pfx_cert = await client.import_certificate( |
| 50 | + certificate_name=pfx_cert_name, certificate_bytes=pfx_cert_bytes |
| 51 | + ) |
| 52 | + print("PFX certificate '{}' imported successfully.".format(imported_pfx_cert.name)) |
| 53 | + |
| 54 | + # Now let's import a PEM-formatted certificate. |
| 55 | + # To import a PEM-formatted certificate, you must provide a CertificatePolicy that sets the content_type to |
| 56 | + # CertificateContentType.pem or the certificate will fail to import (the default content type is PFX). |
| 57 | + pem_cert_name = "pemCert" |
| 58 | + with open(os.environ["PEM_CERT_PATH"], "rb") as f: |
| 59 | + pem_cert_bytes = f.read() |
| 60 | + pem_cert_policy = CertificatePolicy(issuer_name=WellKnownIssuerNames.self, content_type=CertificateContentType.pem) |
| 61 | + imported_pem_cert = await client.import_certificate( |
| 62 | + certificate_name=pem_cert_name, certificate_bytes=pem_cert_bytes, policy=pem_cert_policy |
| 63 | + ) |
| 64 | + print("PEM-formatted certificate '{}' imported successfully.".format(imported_pem_cert.name)) |
| 65 | + |
| 66 | + await credential.close() |
| 67 | + await client.close() |
| 68 | + |
| 69 | +if __name__ == "__main__": |
| 70 | + loop = asyncio.get_event_loop() |
| 71 | + loop.run_until_complete(run_sample()) |
| 72 | + loop.close() |
0 commit comments