Unable to get local issuer when using AzureKeyVault certificate to authenticate web request

Question

In an Azure ML notebook, I am trying to pull a certificate (.pfx) that is stored in Azure Key Vault then use that certificate to authenticate to a CyberArk Web Service to retrieve a username and password.

I am successful in pulling the certificate from Azure Key Vault and breaking it down to the cert and key but I am having an issue when I make the web request to CyberArk. Specifically, it returns the error Unable to get local issuer.

My code is below:

from requests import Session
import certifi
import base64
import tempfile
import os
from azure.identity import DefaultAzureCredential,AzureCliCredential
from azure.keyvault.secrets import SecretClient
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.hazmat.primitives import serialization

#credential = DefaultAzureCredential()
credential = AzureCliCredential()

secret_client = SecretClient(vault_url="https://xxxxxxxxx.vault.azure.net/", credential=credential)

certificate = secret_client.get_secret("xxxxxxxx")

URL = 'https://xxxxxxxxx/AIMWebService/api/Accounts?AppID=xxxxxx&Query=Safe=xxxxxx;Object=xxxxxxx'

headers = {'Content-Type': 'application/json'}

cert_bytes = base64.b64decode(certificate.value)

private_key, public_certificate, additional_certificates = serialization.pkcs12.load_key_and_certificates(
    data=cert_bytes,
    password=None
)

key = tempfile.NamedTemporaryFile(dir="xxxxxxx",delete=False)
cert = tempfile.NamedTemporaryFile(dir="xxxxxxxx",delete=False)

key.write(
    private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )
)
key.flush()
cert.write(
    public_certificate.public_bytes(serialization.Encoding.PEM),
)
cert.flush()
session = Session()
session.cert = (cert.name, key.name)
response = session.get(URL)
print(response)

`

I am not super sure what the issue is but I think it has to do with the verifying certificates normally found in the cacerts or windows cert store but I am not sure how that all translates to using certificates in the Azure.

response = session.get(URL,verify=certify.where())

I tried verifying with the certify but I still get the same error when I send the request. I am able to pull a list of additional certificates from the returned .pfx but I wasn't able to find a way to convert the lists of certificates to into a verifying .pem file I can send with requests.

private_key, public_certificate, additional_certificates = serialization.pkcs12.load_key_and_certificates(
    data=cert_bytes,
    password=None
)

Answer 1

I had to change several things for AzureKeyVault authentication in an Azure ML notebook but with the session.verify correction and additional certs verification I was finally able to authenticate.

from requests import Session
import base64
import tempfile
import os
from azure.identity import DefaultAzureCredential,AzureCliCredential,ClientSecretCredential
from azure.keyvault.secrets import SecretClient
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.hazmat.primitives import serialization

credential = DefaultAzureCredential()

secret_client = SecretClient(vault_url="https://xxxxxxx.vault.azure.net/", credential=credential)

certificate = secret_client.get_secret("xxxxxxx")

URL = 'xxxxxxx'

headers = {'Content-Type': 'application/json'}

cert_bytes = base64.b64decode(certificate.value)

private_key, public_certificate, additional_certificates = serialization.pkcs12.load_key_and_certificates(
    data=cert_bytes,
    password=None
)

#print(private_key)
#print(public_certificate)
#for item in additional_certificates:
#    print(item)

key = tempfile.NamedTemporaryFile(delete=False)
cert = tempfile.NamedTemporaryFile(delete=False)
issuer = tempfile.NamedTemporaryFile(delete=False)

key.write(
    private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )
)
key.flush()
cert.write(
    public_certificate.public_bytes(serialization.Encoding.PEM),
)
cert.flush()
issuer.write(
    additional_certificates[0].public_bytes(serialization.Encoding.PEM),
)
issuer.flush()
issuer.write(
    additional_certificates[1].public_bytes(serialization.Encoding.PEM),
)
issuer.flush()
session = Session()
session.cert = (cert.name, key.name)
session.verify=issuer.name
response = session.get(URL)
print(response.json())
session.close()
cert.close()
key.close()
issuer.close()
os.remove(cert.name)
os.remove(key.name)
os.remove(issuer.name)