Python 使用 Azure Key Vault 证书并发送发布请求

Question

我已在 Azure 密钥保管库中上传了 .pfx 证书，并尝试使用该证书向 API 发送请求

我正在使用以下代码读取证书和秘密

credential = DefaultAzureCredential()

certificate_client = CertificateClient(
    vault_url="https://something.vault.azure.net/", credential=credential
)

certificate = certificate_client.get_certificate("aemocert")

secret_client = SecretClient(
    vault_url="https://something.vault.azure.net/", credential=credential
)
secret = secret_client.get_secret(certificate.name)

现在我必须将收到的数据转换为证书，以便我可以在以下请求中使用它

url= "https://myurl"
headers = {"X-id": "dsd"}
data = {"test":"test"}

requests.request(
    "POST",
    "url = url,
    headers=headers,
    data=data,
    cert=**cert**,
    timeout=10,
)

我尝试过以下代码

base64_bytes = secret.value.encode("ascii")
message_bytes = base64.b64decode(base64_bytes)
cert = x509.load_der_x509_certificate(message_bytes)

但出现此错误

异常：ValueError：解析asn1值时出错：ParseError { kind： UnexpectedTag { 实际：标签 { 值：2，构造：false，类： 通用 } }，位置：[“证书::tbs_cert”] }

如果我使用以下代码

cert = crypto.load_pkcs12(base64.b64decode(secret.value))

我收到错误

异常：类型错误：“PKCS12”对象不可订阅

Answer 1

如果我们使用 openssl 生成 .pfx 文件并将其上传到 Azure KeyVault 证书，则以下代码可以正常工作。由于证书中上传的任何内容都可以作为秘密隐式使用，因此该代码可以正常工作。 先决条件：-

1. 在 Azure 中设置 KeyVault。
2. 上传.pfx文件（使用.pem和.key(private_key)文件生成）。
3. 创建一个应用程序，并向 Azure 中新创建的应用程序授予对 Key Vault 的访问权限。请注意 Azure 中创建的新应用程序的详细信息，因为它将用于连接到 KeyVault 以获取证书和密钥文件数据。

,,

import base64
import orjson, ssl, json
from urllib.parse import urlencode
from azure.identity import EnvironmentCredential, ClientSecretCredential
from azure.keyvault.secrets import SecretClient
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.hazmat.primitives.serialization import Encoding
from cryptography.hazmat.primitives.serialization import PrivateFormat
from cryptography.hazmat.primitives.serialization import NoEncryption
from OpenSSL.SSL import SSLv23_METHOD
from OpenSSL import SSL, crypto
from urllib3.contrib.pyopenssl import inject_into_urllib3
import certifi
import urllib3
import urllib3.contrib.pyopenssl

# generate these values by first registering an Application in azure and giving that respective rights on Azure KeyVault to be referred.
client_id = 'azure-client-id'
tenant_id = 'azure-tenant-id'
client_secret = 'azure-client-secret'
vault_url = 'azure-vault-url'


credentials = ClientSecretCredential(
    client_id = client_id,
    client_secret= client_secret,
    tenant_id= tenant_id
)

# generate the secret client.
secret = SecretClient(vault_url, credentials)
retrieved_secret = secret.get_secret("name-same-as-that-of-certificate-uploaded")

cert_bytes = base64.b64decode(retrieved_secret.value)

private_key, public_certificate, additional_certificates = pkcs12.load_key_and_certificates(
    data=cert_bytes,
    password=None
)
# generating certificate for requests.
cert = public_certificate.public_bytes(encoding=Encoding.PEM)
x509 = crypto.load_certificate(crypto.FILETYPE_PEM, cert)

# generating private key for requests.
key_plain = private_key.private_bytes(encoding=Encoding.PEM,
                                      format=PrivateFormat.PKCS8,
                                      encryption_algorithm=NoEncryption())


ctx = SSL.Context(SSLv23_METHOD)
ctx.verify_mode = True

pkey = crypto.load_privatekey(crypto.FILETYPE_PEM, key_plain)

payload = {"param1": "value1","param2": "value2"}


# don't forget to keep '?' at end of URL if parameters needs to be passed as a part of request.
url = "https://api-url?"

# set your headers accordingly as per the API.
headers = {
        'Content-Type': "application/x-www-form-urlencoded",
        }


# This only works with PyOpenSSL. If you want it to work with the stdlib, you
# have to get the appropriate functionality in the stdlib, which doesn't allow
# loading certs/keys from memory.
urllib3.contrib.pyopenssl.inject_into_urllib3()

ctx = urllib3.util.ssl_.create_urllib3_context()
ctx.load_verify_locations(cafile=certifi.where())

# At this point, ctx._ctx is a PyOpenSSL Context object. If you have a PyOpenSSL
# X509 object and PKey object, you can then do this:
ctx._ctx.use_certificate(x509)
ctx._ctx.use_privatekey(pkey)

encoded_args = urlencode(payload)
# You can now load a PoolManager and tell it to always use this SSLContext.
http = urllib3.PoolManager(ssl_context=ctx)

# example to make post request by passing parameters in URL.
# refer this link for other options to pass data in request -> https://github.com/urllib3/urllib3/blob/main/docs/user-guide.rst
output = http.request(method="POST", url=(url+ encoded_args), headers=headers)
print(output.json())


# example to make post request with payload.
# token = 'your-api-token'
# fetching the actual balance using token generated above.
balance_url = "https://your-api.com"
payload = {
               "key1": f"value1",
        }

headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json'
    }

encoded_data = orjson.dumps(payload)
balance_output = http.request(method="POST", url=balance_url, headers=headers, body=encoded_data)
print(balance_output.json())