in metaflow/plugins/azure/azure_secret_manager_secrets_provider.py [0:0]
def get_secret_as_dict(self, secret_id, options={}, role=None):
# https://learn.microsoft.com/en-us/azure/app-service/app-service-key-vault-references?tabs=azure-cli has a lot of details on
# the patterns used in key vault
# Vault names and Managed HSM pool names are selected by the user and are globally unique.
# Vault name and Managed HSM pool name must be a 3-24 character string, containing only 0-9, a-z, A-Z, and not consecutive -.
# object-type The type of the object. As of 05/08/24 only "secrets", are supported
# object-name An object-name is a user provided name for and must be unique within a key vault. The name must be a 1-127 character string, starting with a letter and containing only 0-9, a-z, A-Z, and -.
# object-version An object-version is a system-generated, 32 character string identifier that is optionally used to address a unique version of an object.
# We allow these forms of secret_id:
#
# 1. Full path like https://<key-vault-name><.vault-domain>/secrets/<secret-name>/<secret-version>. This is what you
# see in Azure portal and is easy to copy paste.
#
# 2. Full path but without the version like https://<key-vault-name><.vault-domain>/secrets/<secret-name>
#
# 3. Simple string like mysecret. This corresponds to the SecretName.
#
# 4. Simple string with <secret-name>/<secret-version> suffix like mysecret/123
# The latter two forms require METAFLOW_AZURE_KEY_VAULT_PREFIX to be set.
# if the secret_id is None/empty/does not start with https then return false
if secret_id is None or secret_id == "":
raise MetaflowAzureKeyVaultBadSecret("empty secret id is not supported")
# check if the passed in secret is a short-form ( #3/#4 in the above comment)
if not secret_id.startswith("https://"):
# check if the secret_id is of form `secret_name` OR `secret_name/secret_version`
if not self._is_partial_secret_valid(secret_id=secret_id):
raise MetaflowAzureKeyVaultBadSecret(
"unsupported partial secret %s" % secret_id
)
domain = AZURE_KEY_VAULT_PREFIX.rstrip("/")
full_secret = "%s/secrets/%s" % (domain, secret_id)
# if the secret id is passed as a URL - then check if the url is fully qualified
if secret_id.startswith("https://"):
if not self._is_secret_id_fully_qualified_url(secret_id=secret_id):
raise MetaflowException("unsupported secret %s" % secret_id)
full_secret = secret_id
# at this point I know that the secret URL is good so we can start creating the Secret Client
az_credentials = create_cacheable_azure_credential()
res = urlparse(full_secret)
az_vault_url = "%s://%s" % (
res.scheme,
res.netloc,
) # https://myvault.vault.azure.net
secret_data = res.path.strip("/").split("/")[1:]
secret_name = secret_data[0]
secret_version = None
if len(secret_data) > 1:
secret_version = secret_data[1]
from azure.keyvault.secrets import SecretClient
client = SecretClient(vault_url=az_vault_url, credential=az_credentials)
key_vault_secret_val = client.get_secret(
name=secret_name, version=secret_version
)
result = {}
if options.get("env_var_name") is not None:
env_var_name = options["env_var_name"]
sanitized_key = self._sanitize_key_as_env_var(env_var_name)
else:
sanitized_key = self._sanitize_key_as_env_var(key_vault_secret_val.name)
response_payload = key_vault_secret_val.value
result[sanitized_key] = response_payload
return result