"""
Manage the Vault (or OpenBao) PKI secret engine, request X.509 certificates.
.. versionadded:: 1.1.0
.. important::
This module requires the general :ref:`Vault setup <vault-setup>`.
"""
# pylint: disable=too-many-lines
import logging
import typing
from datetime import datetime
from datetime import timedelta
from datetime import timezone
from salt.exceptions import CommandExecutionError
from salt.exceptions import SaltInvocationError
from saltext.vault.utils import vault
from saltext.vault.utils.vault import helpers as hlp
try:
from salt.utils import x509 as x509util
from saltext.vault.utils.vault import pki
HAS_CRYPTOGRAPHY = True
except ImportError: # pragma: no cover
HAS_CRYPTOGRAPHY = False
if typing.TYPE_CHECKING:
from saltext.vault.utils._types import SaltContext
from saltext.vault.utils._types import SaltFunctions
from saltext.vault.utils._types import SaltGrains
from saltext.vault.utils._types import SaltLogger
from saltext.vault.utils._types import SaltOpts
__opts__: SaltOpts
__context__: SaltContext
__salt__: SaltFunctions
__grains__: SaltGrains
log: "SaltLogger" = logging.getLogger(__name__) # type: ignore
__virtualname__ = "vault_pki"
def __virtual__():
return __virtualname__
[docs]
def list_roles(mount="pki"):
"""
List configured PKI roles.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#list-roles>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/roles" {
capabilities = ["list"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.list_roles
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/roles"
try:
return vault.api_list(endpoint, __opts__, __context__)["data"]["keys"]
except vault.VaultNotFoundError:
return []
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def read_role(name, mount="pki"):
"""
Get configuration of specific PKI role.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#read-role>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/roles/<name>" {
capabilities = ["read"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.read_role my_role
name
Name of the role.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/roles/{name}"
try:
res = vault.api_get(endpoint, __opts__, __context__)
return res["data"]
except vault.VaultNotFoundError:
return None
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def write_role(
name,
mount="pki",
issuer_ref=None,
ttl=None,
max_ttl=None,
allow_localhost=None,
allowed_domains=None,
server_flag=None,
client_flag=None,
key_usage=None,
no_store=None,
require_cn=None,
**kwargs,
):
"""
Create or update PKI role.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#create-update-role>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/roles/<name>" {
capabilities = ["create", "update", "patch"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.write_role myrole
name
Name of the role.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
issuer_ref
Name or ID of the issuer which should be used with this role. If not set, the default issuer is used.
ttl
Specifies the Time To Live value to be used for the validity period of the requested certificate,
provided as a string duration with time suffix. Hour is the largest suffix.
The value specified is strictly used for future validity.
If not set, uses the system default value or the value of ``max_ttl``, whichever is shorter.
max_ttl
Specifies the maximum Time To Live provided as a string duration with time suffix.
Hour is the largest suffix. If not set, defaults to the system maximum lease TTL.
allow_localhost
Specifies if clients can request certificates for ``localhost`` as one of the requested common names.
allowed_domains
Specifies the domains this role is allowed to issue certificates for.
This is used with the ``allow_bare_domains``, ``allow_subdomains``, and ``allow_glob_domains`` options to
determine the type of matching between these domains and the values of common name, DNS-typed SAN entries, and Email-typed SAN entries.
When ``allow_any_name`` is used, this attribute has no effect.
server_flag
Specifies if certificates are flagged for server authentication use.
See `RFC 5280 Section 4.2.1.12 <https://datatracker.ietf.org/doc/html/rfc5280#section-4.2.1.12>`__
for information about the Extended Key Usage field.
If not set, defaults to true.
client_flag
Specifies if certificates are flagged for client authentication use.
See `RFC 5280 Section 4.2.1.12 <https://datatracker.ietf.org/doc/html/rfc5280#section-4.2.1.12>`__
for information about the Extended Key Usage field.
If not set, defaults to true.
key_usage
Specifies the allowed key usage constraint on issued certificates.
If unset, defaults to ``["DigitalSignature", "KeyAgreement", "KeyEncipherment"]``
no_store
If set, certificates issued/signed against this role are not stored in the storage backend.
require_cn
If set to false, makes the common_name field optional while generating a certificate. Defaults to true.
kwargs:
Any other params which can be understood by the Vault API.
"""
endpoint = f"{mount}/roles/{name}"
api_write = vault.api_put
if read_role(name, mount=mount) is not None:
api_write = vault.api_patch
payload = {k: v for k, v in kwargs.items() if not k.startswith("_")}
if issuer_ref is not None:
payload["issuer_ref"] = issuer_ref
if ttl is not None:
payload["ttl"] = ttl
if max_ttl is not None:
payload["max_ttl"] = max_ttl
if allow_localhost is not None:
payload["allow_localhost"] = allow_localhost
if allowed_domains is not None:
if not isinstance(allowed_domains, list):
allowed_domains = [allowed_domains]
payload["allowed_domains"] = allowed_domains
if server_flag is not None:
payload["server_flag"] = server_flag
if client_flag is not None:
payload["client_flag"] = client_flag
if key_usage is not None:
if not isinstance(key_usage, list):
key_usage = [key_usage]
payload["key_usage"] = key_usage
if no_store is not None:
payload["no_store"] = no_store
if require_cn is not None:
payload["require_cn"] = require_cn
try:
api_write(endpoint, __opts__, __context__, payload=payload, safe_to_retry=True)
return True
except vault.VaultUnsupportedOperationError as err: # pragma: no cover
raise CommandExecutionError(
f"Vault version too old. Please upgrade to v1.11.0+: {err}"
) from err
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def delete_role(name, mount="pki"):
"""
Delete PKI role from Vault.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#delete-role>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/roles/<name>" {
capabilities = ["delete"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.delete_role myrole
name
Name of the role.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/roles/{name}"
try:
vault.api_delete(endpoint, __opts__, __context__)
return True
except vault.VaultNotFoundError:
return False
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def list_issuers(mount="pki"):
"""
List issuers information.
Returns ``{ "<issuer_id>" : { "is_default": False, "issuer_name": "...", "key_id": "...", "serial_number": "...."}}``
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#list-issuers>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/issuers" {
capabilities = ["list"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.list_issuers
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/issuers"
try:
return vault.api_list(endpoint, __opts__, __context__, is_unauthd=True)["data"]["key_info"]
except vault.VaultNotFoundError:
return {}
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def get_issuer_id(ref="default", mount="pki"):
"""
.. versionadded:: 1.9.0
Get the issuer ID of a reference, which can be an issuer ID or an issuer name.
Ensures the returned issuer ID exists.
Required policy: See :func:`list_issuers`
CLI Example:
.. code-block:: bash
salt '*' vault_pki.get_issuer_id foobar
ref
Reference to an issuer. Either ``issuer_name`` or ``issuer_id``.
Defaults to ``default``.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
if ref == "default":
issuer_id = get_default_issuer(mount=mount)
if issuer_id is None:
raise CommandExecutionError(f"No default issuer has been configured on mount '{mount}'")
return issuer_id
issuers = list_issuers(mount=mount)
if ref in issuers:
return ref
for issuer_id, info in issuers.items():
if info.get("issuer_name") == ref:
return issuer_id
raise CommandExecutionError(
f"No issuer is associated with reference '{ref}' on mount '{mount}'"
)
[docs]
def read_issuer(ref="default", mount="pki"):
"""
Read an issuer's information.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#read-issuer-certificate>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/issuer/<name>" {
capabilities = ["read"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.read_issuer
ref
Reference of the issuer. Can be issuer ID, issuer name or literal ``default``
which means default issuer. Defaults to ``default``.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/issuer/{ref}"
try:
return vault.api_get(endpoint, __opts__, __context__, is_unauthd=True)["data"]
except vault.VaultNotFoundError:
return None
except vault.VaultServerError as err:
if "unable to find PKI issuer for reference" in str(err):
return None
if ref == "default" and "no default issuer currently configured" in str(err):
return None
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def update_issuer(
ref="default",
mount="pki",
manual_chain=None,
usage=None,
aia_urls=None,
crl_endpoints=None,
ocsp_servers=None,
name=None,
aia_url_templating=None,
delta_crl_endpoints=None,
leaf_not_after_behavior=None,
revocation_signature_algorithm=None,
):
"""
Update issuer's information.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#update-issuer>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/issuer/<name>" {
capabilities = ["patch"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.update_issuer ref usage=["crl-signing"]
ref
Reference of the issuer. Can be issuer ID, issuer name or literal ``default``,
referring to the default issuer. Defaults to ``default``.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
manual_chain
Chain of issuer references to build this issuer's computed CAChain field from, when non-empty.
usage
Allowed usages for this issuer. Valid options are:
* ``read-only`` - to allow this issuer to be read; implicit; always allowed;
* ``issuing-certificates`` - to allow this issuer to be used for issuing other certificates;
* ``crl-signing`` - to allow this issuer to be used for signing CRLs.
This is separate from the CRLSign KeyUsage on the x509 certificate, but this usage cannot be set
unless that KeyUsage is allowed on the x509 certificate;
* ``ocsp-signing`` - to allow this issuer to be used for signing OCSP responses.
aia_urls
Specifies the URL values for the Issuing Certificate field as an array.
crl_endpoints
Specifies the URL values for the CRL Distribution Points field as an array.
ocsp_servers
Specifies the URL values for the OCSP Servers field as an array.
name
.. versionadded:: 1.9.0
Custom name for the issuer. Must be unique and not equal to ``default``.
aia_url_templating
.. versionadded:: 1.9.0
Render ``aia_urls``/``crl_endpoints``/``ocsp_servers``/``delta_crl_endpoints`` as templates.
Supported variables: ``{{issuer_id}}``, ``{{cluster_path}}``, ``{{cluster_aia_path}}``
delta_crl_endpoints
.. versionadded:: 1.9.0
(Requires Vault 1.20+ or OpenBao)
Specifies the URL values for the Delta CRL Distribution Points field.
This can be an array or a comma-separated string list.
leaf_not_after_behavior
.. versionadded:: 1.9.0
Behavior of a leaf's ``NotAfter`` field during issuance when it exceeds the issuer's validity.
Valid options:
* ``err``: Error, unless during CA/ACME issuance. (default)
* ``always_enforce_err``: Error, including during CA/ACME issuance. (Vault 1.18.2+ only)
* ``truncate``: Silently truncate the requested NotAfter to that of the issuer.
* ``permit``: Allow signed certificate validities to exceed that of the issuer.
revocation_signature_algorithm
.. versionadded:: 1.9.0
Which signature algorithm to use when building CRLs.
See Go's `x509.SignatureAlgorithm <https://pkg.go.dev/crypto/x509#SignatureAlgorithm>`__ constant for possible values.
Default (empty string) is to autoselect.
"""
endpoint = f"{mount}/issuer/{ref}"
payload = {}
for param, val in (
("manual_chain", manual_chain),
("usage", usage),
("issuing_certificates", aia_urls),
("crl_distribution_points", crl_endpoints),
("ocsp_servers", ocsp_servers),
("issuer_name", name),
("enable_aia_url_templating", aia_url_templating),
("delta_crl_distribution_points", delta_crl_endpoints),
("leaf_not_after_behavior", leaf_not_after_behavior),
("revocation_signature_algorithm", revocation_signature_algorithm),
):
if val is not None:
payload[param] = val
try:
vault.api_patch(
endpoint,
__opts__,
__context__,
payload=payload,
)
return True
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def read_issuer_certificate(name="default", mount="pki", include_chain=False):
"""
Read an issuer's certificate.
Returns certificate(s) in PEM format
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#read-issuer-certificate>`__.
Required policy: See :func:`read_issuer`
CLI Example:
.. code-block:: bash
salt '*' vault_pki.read_issuer_certificate
name
Name of the issuer. Can be issuer ID, issuer name or literal ``default``
which means default issuer. Defaults to ``default``.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
include_chain
If set to true, appends the CA chain to the certificate (in case of intermediate issuer)
"""
cert_data = read_issuer(name, mount)
if not cert_data:
raise CommandExecutionError("Issuer does not exist")
if include_chain:
return "".join(cert_data["ca_chain"])
return cert_data["certificate"]
[docs]
def get_default_issuer(mount="pki"):
"""
Return the issuer ID of the default issuer.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#list-issuers>`__.
Required policy: See :func:`list_issuers`
CLI Example:
.. code-block:: bash
salt '*' vault_pki.get_default_issuer
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
all_issuers = list_issuers(mount)
for k, v in all_issuers.items():
if v["is_default"]:
return k
# In case there is no default issuer
return None
[docs]
def set_default_issuer(name, mount="pki"):
"""
Set the default issuer.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#set-issuers-configuration>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/config/issuers" {
capabilities = ["create", "update"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.set_default_issuer myca
name
Name or ID of the default issuer to set.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/config/issuers"
payload = {"default": name}
try:
vault.api_put(endpoint, __opts__, __context__, payload=payload)
return True
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def get_key_id(ref, mount="pki"):
"""
.. versionadded:: 1.9.0
Get the key ID of a reference, which can be a key ID or a key name. Ensures the returned key ID exists.
Required policy: See :func:`list_keys`
CLI Example:
.. code-block:: bash
salt '*' vault_pki.get_key_id foobar
ref
Reference to a key. Either ``key_name`` or ``key_id``.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
keys = list_keys(mount=mount)
if ref in keys:
return ref
for key_id, info in keys.items():
if info.get("key_name") == ref:
return key_id
raise CommandExecutionError(f"No key is associated with reference '{ref}' on mount '{mount}'")
[docs]
def list_keys(mount="pki"):
"""
.. versionadded:: 1.9.0
Get a mapping of keys provisioned in this mount to some of their properties (currently only ``key_name``).
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#list-keys>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/keys" {
capabilities = ["list"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.list_keys
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
try:
res = vault.api_list(f"{mount}/keys", __opts__, __context__)["data"]
except vault.VaultNotFoundError:
return {}
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
keys = res["key_info"]
for key in res["keys"]:
if key not in keys: # pragma: no cover
keys[key] = {}
return keys
[docs]
def generate_key(
key_type="internal",
key_name=None,
key_algo=None,
key_bits=None,
managed_key_name=None,
managed_key_id=None,
mount="pki",
):
"""
.. versionadded:: 1.9.0
Generate a new private key for use in the PKI mount.
This key can be used with :func:`generate_root` and :func:`generate_intermediate`,
using the ``key_type=existing`` variant by passing the returned ``key_id`` as ``key_ref``.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#generate-key>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/keys/generate/<key_type>" {
capabilities = ["create", "update"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.generate_key key_name=my_rsa_key key_bits=4096
salt '*' vault_pki.generate_key exported key_algo=ed25519
key_type
Key type to generate. Valid values are:
* ``internal``: The private key is not returned and cannot be retrieved later.
* ``exported``: The private key is returned in the response.
* ``kms``: Request a key from a key management system. The private key is not returned and cannot be retrieved later.
Defaults to ``internal``.
key_name
Specify a name for the generated key. Optional.
key_algo
Key algorithm. Either ``rsa``, ``ed25519`` or ``ec``. Defaults to ``rsa``.
key_bits
Number of bits to use for the generated key. Valid values depend on the ``key_type``:
* ``rsa``: 2048 (default), 3072, 4096, 8192.
* ``ec``: 224, 256 (default), 384, 521
* ``ed25519``: ignored
Defaults to ``0`` (universal default).
managed_key_name
When ``key_type`` is ``kms``, the managed key's configured name. Either this or ``managed_key_id`` is required then.
managed_key_id
When ``key_type`` is ``kms``, the managed key's UUID. Either this or ``managed_key_name`` is required then.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
key_type = hlp.in_vals(("exported", "internal", "kms"), key_type=key_type)
if key_name == "default":
raise SaltInvocationError("key_name cannot be `default`. This is a reserved word.")
if key_type == "kms":
hlp.one_of(
_reason="key_type is `kms`",
managed_key_name=managed_key_name,
managed_key_id=managed_key_id,
)
else:
hlp.none_of(
_reason="key_type is not `kms`",
managed_key_name=managed_key_name,
managed_key_id=managed_key_id,
)
endpoint = f"{mount}/keys/generate/{key_type}"
payload = hlp.filter_unset(
{
"key_name": key_name,
"key_type": key_algo,
"key_bits": key_bits,
"managed_key_name": managed_key_name,
"managed_key_id": managed_key_id,
}
)
try:
return vault.api_post(endpoint, __opts__, __context__, payload=payload)["data"]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def generate_root(
common_name,
mount="pki",
key_type="internal",
issuer_name=None,
key_name=None,
ttl=None,
key_algo="rsa",
key_bits=0,
max_path_length=-1,
key_ref=None,
managed_key_name=None,
managed_key_id=None,
encoding=None,
**kwargs,
):
"""
Generate a new root issuer.
Returns ``{ "certificate" : "-----BEGIN CERTIFICATE...", "issuer_id": "...", "key_id": "...", }``.
If key_type is ``exported``, also returns the private key.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#generate-root>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/root/generate/<key_type>" {
capabilities = ["create", "update"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.generate_root my-root
common_name
Subject common name (``CN``) for the certificate. Required.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
key_type
.. versionchanged:: 1.9.0
This parameter used to be called ``type``.
Key type of the root to generate. Valid values are:
* ``existing``: Use an existing key for the generated root, specified in ``key_ref``.
* ``internal``: The private key is not returned and cannot be retrieved later.
* ``exported``: The private key is returned in the response.
* ``kms``: Request a key from a key management system. The private key is not returned and cannot be retrieved later.
Defaults to ``internal``.
issuer_name
Provides a name to the specified issuer.
The name must be unique across all issuers and not be the reserved value ``default``.
key_name
When a new key is created with this request, optionally specifies the name for this.
The global ref ``default`` may not be used as a name.
ttl
Specifies the requested Time To Live (after which the certificate expires).
This cannot be larger than the engine's max (or, if not set, the system max).
key_algo
.. versionchanged:: 1.9.0
This parameter used to be called ``key_type``, which now refers to key generation/exportability instead.
Specifies the desired key algorithm, either ``rsa``, ``ed25519`` or ``ec``. Defaults to ``rsa``.
key_bits
Number of bits to use for the generated key. Valid values depend on the ``key_type``:
* ``rsa``: 2048 (default), 3072, 4096, 8192.
* ``ec``: 224, 256 (default), 384, 521
* ``ed25519``: ignored
Defaults to ``0`` (universal default).
max_path_length
basicConstraints ``pathlen`` parameter, which indicates the maximum number of CAs that can appear below this one in a chain.
If set to ``0``, this CA can only issue leaf certificates, not other CAs.
A negative value means no limit. Defaults to ``-1``.
key_ref
Reference to an existing private key on this ``mount``, either ``key_name`` or ``key_id``.
Required when ``key_type`` is ``existing``, otherwise ignored.
managed_key_name
When ``key_type`` is ``kms``, the managed key's configured name.
Either this or ``managed_key_id`` is required then.
managed_key_id
When ``key_type`` is ``kms``, the managed key's UUID.
Either this or ``managed_key_name`` is required then.
encoding
.. versionchanged:: 1.9.0
This parameter used to be called ``format``. To align it with other endpoints,
it was renamed to ``encoding``. The previous name still works, but is warned about.
Output format. Can be either ``pem`` or ``der``. Defaults to ``pem``.
kwargs
Unknown keyword arguments are passed through to the Vault API.
See the API method docs linked above for details. Here is an incomplete list:
* key_name
* private_key_format
* pkcs12_encoder/pkcs12_password/jks_password/jks_private_key_alias
* alt_names/ip_sans/uri_sans/other_sans (in contrast to other functions in this module, no special handling for alt_names is applied)
* exclude_cn_from_sans
* key_usage
* permitted_dns_domains/excluded_dns_domains (no special handling for permitted_alt_names/excluded_alt_names either)
* permitted_ip_ranges/excluded_ip_ranges
* permitted_email_addresses/excluded_email_addresses
* permitted_uri_domains/excluded_uri_domains
* ou
* organization
* country
* locality
* province
* street_address
* postal_code
* serial_number
* signature_bits
* not_before_duration
* not_after
* use_pss
"""
if key_type in ("rsa", "ec", "ed25519"):
log.warning(
"The `key_type` parameter to this function has changed meaning. To specify a key's algorithm, use ``key_algo`` instead."
)
key_algo = key_type
if "type" in kwargs:
log.warning(
"The `type` parameter to this function is deprecated. Use ``key_type`` instead."
)
key_type = kwargs.pop("type")
key_type = hlp.in_vals(("existing", "exported", "internal", "kms"), key_type=key_type)
if key_type == "kms":
hlp.one_of(
_reason="key_type is `kms`",
managed_key_name=managed_key_name,
managed_key_id=managed_key_id,
)
else:
hlp.none_of(
_reason="key_type is not `kms`",
managed_key_name=managed_key_name,
managed_key_id=managed_key_id,
)
if key_type == "existing":
if key_ref is None:
raise SaltInvocationError("key_type `existing` requires `key_ref` to be set")
else:
hlp.none_of(_reason="key_type is not `existing`", key_ref=key_ref)
if issuer_name == "default":
raise SaltInvocationError("issuer_name cannot be `default`. This is a reserved word.")
if key_name == "default":
raise SaltInvocationError("key_name cannot be `default`. This is a reserved word.")
# Compatibility with releases pre-1.9.0
if "format" in kwargs:
log.warning(
"The `format` parameter to `vault_pki.generate_root` was renamed to `encoding`. Please update your call."
)
encoding = kwargs.pop("format")
if encoding is None:
encoding = "pem"
endpoint = f"{mount}/root/generate/{key_type}"
payload = {k: v for k, v in kwargs.items() if not k.startswith("_")}
payload["common_name"] = common_name
payload["format"] = encoding
if issuer_name is not None:
payload["issuer_name"] = issuer_name
if ttl is not None:
payload["ttl"] = ttl
if max_path_length is not None and max_path_length > -1:
payload["max_path_length"] = max_path_length
if key_type == "existing":
payload["key_ref"] = key_ref
else:
payload["key_type"] = key_algo
if key_name is not None:
payload["key_name"] = key_name
if key_bits > 0:
payload["key_bits"] = key_bits
if managed_key_name is not None:
payload["managed_key_name"] = managed_key_name
if managed_key_id is not None:
payload["managed_key_id"] = managed_key_id
try:
resp = vault.api_post(endpoint, __opts__, __context__, payload=payload)["data"]
ret = {
"certificate": resp["certificate"],
"issuer_id": resp["issuer_id"],
"key_id": resp["key_id"],
}
if key_type == "exported":
ret["private_key"] = resp["private_key"]
return ret
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def delete_key(ref, mount="pki"):
"""
Delete a private key from Vault.
There must be no issuers depending on the key for this to succeed.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#delete-key>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/key/<ref>" {
capabilities = ["delete"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.delete_key ref
ref
Reference to the key, either ``key_name`` or ``key_id``.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/key/{ref}"
try:
vault.api_delete(endpoint, __opts__, __context__)
return True
# Don't need to catch VaultNotFoundError, it's not thrown for missing key
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def delete_issuer(ref, mount="pki", include_key=False):
"""
Delete issuer from Vault.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#delete-issuer>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/issuer/<ref>" {
capabilities = ["delete"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.delete_issuer ref
ref
Reference to the issuer, either ``issuer_name`` or ``issuer_id``.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
include_key
If set to true, also deletes the private key if imported.
Defaults to false, i.e. the private key is preserved.
"""
endpoint = f"{mount}/issuer/{ref}"
key_id = None
if include_key:
issuer_info = read_issuer(ref, mount=mount)
if issuer_info:
key_id = issuer_info["key_id"]
try:
vault.api_delete(endpoint, __opts__, __context__)
if key_id:
delete_key(key_id, mount=mount)
return True
# Don't need to catch VaultNotFoundError, it's not thrown for missing issuer
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def import_issuer(cert, chain=None, private_key=None, private_key_passphrase=None, mount="pki"):
"""
.. versionadded:: 1.9.0
Import a CA certificate and (optionally) corresponding private key.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#import-ca-certificates-and-keys>`__.
Required policy:
.. code-block:: vaultpolicy
# without private_key
path "<mount>/issuer/import/cert" {
capabilities = ["create", "update"]
}
# with private_key
path "<mount>/issuer/import/bundle" {
capabilities = ["create", "update"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.import_issuer /etc/tls/my_intermediate_cert.pem
salt '*' vault_pki.import_issuer /etc/tls/my_intermediate_cert.pem private_key=/etc/tls/my_intermediate.key
cert
Certificate to import. Any input accepted by the :py:mod:`x509_v2 modules <salt.modules.x509_v2>` is accepted.
Included CA chain is respected when ``chain`` is not specified.
chain
CA chain for the certificate. Defaults to the chain in ``cert``, if present.
private_key
Import corresponding private key for ``cert``. Optional.
private_key_passphrase
When ``private_key`` is specified and encrypted, the passphrase to decrypt it.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/issuers/import/"
if not chain:
cert, chain = x509util.load_cert(cert, load_chain=True)
# Ensure this works in the wrapper
cert = x509util.to_pem(cert).decode()
chain = [x509util.to_pem(chain_cert).decode() for chain_cert in chain]
payload = {"pem_bundle": _x509v2("encode_certificate", cert, append_certs=chain)}
if private_key:
endpoint += "bundle"
payload["pem_bundle"] += "\n" + _x509v2(
"encode_private_key", private_key, private_key_passphrase=private_key_passphrase
)
else:
endpoint += "cert"
try:
return vault.api_post(endpoint, __opts__, __context__, payload=payload)["data"]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def read_issuer_crl(ref="default", mount="pki", delta=False):
"""
Get issuer CRL.
.. note::
If CA cannot sign CRLs, returns None.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#read-issuer-crl>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/issuer/<ref>" {
capabilities = ["read"]
}
path "<mount>/issuer/<ref>/crl" {
capabilities = ["read"]
}
path "<mount>/issuer/<ref>/crl/delta" {
capabilities = ["read"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.read_issuer_crl ref
ref
Reference to the issuer, either ``issuer_name`` or ``issuer_id``.
Defaults to ``default``.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
delta
If set to true, returns delta CRL instead of complete one.
"""
# Check if issuer can sign CRLs at all. If not,
# there is no point to check for CRL as this throws error
try:
issuer = vault.api_get(f"{mount}/issuer/{ref}", __opts__, __context__)["data"]
except vault.VaultServerError as err:
if "unable to find PKI issuer" in str(err):
return None
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
if "crl-signing" not in issuer["usage"].split(","):
return None
endpoint = f"{mount}/issuer/{ref}/crl"
if delta:
endpoint = endpoint + "/delta"
try:
return vault.api_get(endpoint, __opts__, __context__, is_unauthd=True)["data"]["crl"]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def list_revoked_certificates(mount="pki"):
"""
List revoked certificates serial numbers
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#list-revoked-certificates>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/certs/revoked" {
capabilities = ["list"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.list_revoked_certificates
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/certs/revoked"
try:
return vault.api_list(endpoint, __opts__, __context__)["data"]["keys"]
except vault.VaultNotFoundError:
return []
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def list_certificates(mount="pki"):
"""
List issued certificates serial numbers
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#list-certificates>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/certs" {
capabilities = ["list"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.list_certificates
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/certs"
try:
return vault.api_list(endpoint, __opts__, __context__)["data"]["keys"]
except vault.VaultNotFoundError:
return []
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def read_certificate(serial, mount="pki"):
"""
Read issued certificate.
Returns certificate in PEM format
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#read-certificate>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/cert/<serial>" {
capabilities = ["read"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.read_certificate 7e:85:c5:d1:85:94:9a:46:08:b5:1b:9c:22:cb:35:e5:ea:f3:56:3f
serial
Specifies the serial of the key to read. Valid values are:
* ``<serial>`` for the certificate with the given serial number, in hyphen-separated or colon-separated hexadecimal.
* ``ca`` for the default issuer's CA certificate
* ``crl`` for the default issuer's CRL
* ``ca_chain`` for the default issuer's CA trust chain.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/cert/{serial}"
try:
return vault.api_get(endpoint, __opts__, __context__, is_unauthd=True)["data"][
"certificate"
]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def read_certificate_full(serial, mount="pki"):
"""
.. versionadded:: 1.7.0
Get full certificate information as a dictionary, including the certificate (`certificate`)
and its CA chain certificates (`ca_chain`, a list of strings) in PEM format.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#read-certificate>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/cert/<serial>" {
capabilities = ["read"]
}
path "<mount>/issuer/<name>" {
capabilities = ["read"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.read_certificate_full 7e:85:c5:d1:85:94:9a:46:08:b5:1b:9c:22:cb:35:e5:ea:f3:56:3f
serial
Specifies the serial of the certificate to read. Valid values are:
* ``<serial>`` for the certificate with the given serial number, in hyphen-separated or colon-separated hexadecimal.
* ``ca`` for the default issuer's CA certificate
* ``crl`` for the default issuer's CRL
* ``ca_chain`` for the default issuer's CA trust chain.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/cert/{serial}"
try:
data = vault.api_get(endpoint, __opts__, __context__, is_unauthd=True)["data"]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
# Ensure trailing newline so callers can concatenate certificate
# and ca_chain entries without corrupting PEM boundaries.
if not data["certificate"].endswith("\n"):
data["certificate"] += "\n"
if serial == "ca_chain":
return data
# Vault may omit issuer_id and the immediate issuer cert from ca_chain.
# Resolve the signing issuer and rebuild a complete chain.
if serial in ("ca", "crl"):
# These special values always reference the default issuer
issuer_ref = "default"
else:
# Prefer explicit issuer_id, which is only set for revoked certificates.
# Otherwise, iterate over all issuers and find the most fitting one.
# Newer Vault versions include authority_key_id in the response, OpenBao does not.
issuer_ref = data.get("issuer_id") or _find_signing_issuer(
data["certificate"], authority_key_id=data.get("authority_key_id"), mount=mount
)
# Do not fall back to default issuer, it has been checked already.
if not issuer_ref:
raise CommandExecutionError("Failed to determine cert issuer")
issuer_data = read_issuer(ref=issuer_ref, mount=mount)
if not issuer_data:
raise CommandExecutionError(f"Failed to lookup issuer `{issuer_ref}`")
chain = issuer_data["ca_chain"]
issuer_cert = issuer_data["certificate"]
if issuer_cert not in chain:
chain.insert(0, issuer_cert)
data["ca_chain"] = chain
return data
def _find_signing_issuer(leaf_pem, authority_key_id=None, mount="pki"):
"""
Find the configured issuer whose certificate SubjectKeyIdentifier matches the
certificate's AuthorityKeyIdentifier. Returns the matching ``issuer_id`` or ``None``.
"""
if not HAS_CRYPTOGRAPHY: # pragma: no cover
return None
try:
leaf = x509util.load_cert(leaf_pem)
except (SaltInvocationError, CommandExecutionError):
return None
if not authority_key_id:
try:
authority_key_id = leaf.extensions.get_extension_for_class(
x509util.cx509.AuthorityKeyIdentifier
).value.key_identifier
except x509util.cx509.ExtensionNotFound:
return None
if not isinstance(authority_key_id, bytes):
authority_key_id = bytes.fromhex(authority_key_id.replace(":", ""))
try:
issuers = list_issuers(mount=mount)
except CommandExecutionError as err:
log.error(str(err), exc_info_on_loglevel=logging.DEBUG)
return None
candidates = []
now = datetime.now(tz=timezone.utc)
for issuer_id in issuers:
try:
issuer_data = read_issuer(ref=issuer_id, mount=mount)
except CommandExecutionError:
continue
if not issuer_data or "certificate" not in issuer_data:
continue
try:
issuer_cert = x509util.load_cert(issuer_data["certificate"])
except (SaltInvocationError, CommandExecutionError):
continue
try:
issuer_ski = issuer_cert.extensions.get_extension_for_class(
x509util.cx509.SubjectKeyIdentifier
).value.key_identifier
except x509util.cx509.ExtensionNotFound:
continue
if issuer_ski != authority_key_id:
continue
try:
leaf.verify_directly_issued_by(issuer_cert) # requires cryptography >=40
except (ValueError, TypeError, x509util.InvalidSignature):
continue
candidates.append((issuer_id, issuer_data, issuer_cert))
for predicate in (
lambda c: not c[1].get("revoked"),
lambda c: c[2].not_valid_after_utc > now >= c[2].not_valid_before_utc,
):
filtered = [c for c in candidates if predicate(c)]
if filtered:
candidates = filtered
if len(candidates) > 1:
default = read_issuer("default", mount=mount)
if default:
for cand in candidates:
if cand[0] == default["issuer_id"]:
return cand[0]
if candidates:
return candidates[0][0]
return None
[docs]
def issue_certificate(
role_name,
common_name=None,
mount="pki",
issuer_ref=None,
alt_names=None,
ttl=None,
encoding=None,
exclude_cn_from_sans=False,
**kwargs,
):
"""
Generate and issue a new leaf certificate and private key.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#generate-certificate-and-key>`__.
Required policy:
.. code-block:: vaultpolicy
# When not specifying issuer_ref
path "<mount>/issue/<role_name>" {
capabilities = ["update"]
}
# When specifying issuer_ref
path "<mount>/issuer/<issuer_ref>/issue/<role_name>" {
capabilities = ["update"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.issue_certificate myrole common_name="www.example.com"
role_name
PKI role to use for issuing the certificate. Required.
common_name
Subject common name (``CN``) for the certificate.
Required, unless the role explicitly sets ``require_cn`` to false.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
issuer_ref
Specify an explicit issuer instead of taking it from the role definition.
Can be issuer_name or issuer_id.
alt_names
Any alternative names to add to the certificate.
Can be specified either as dict (``{ "<type>": "<value>" }``),
a dict of lists (``{ "<type>": ["<value1>", "<value2>", ...] }``)
or list of SAN strings (``["<type1>:<value1>", ...]``).
``<type>`` can be ``dns``, ``email``, ``uri``, ``ip`` or any OID for otherName SANs.
``<value>`` is the corresponding value. Note that otherName SANs need to omit ``UTF8:``.
ttl
Specifies the requested Time To Live (after which the certificate will be expired).
This cannot be larger than the engine's max (or, if not set, the system max).
Can be an integer, which is interpreted as seconds, or a time string such as ``1h``.
encoding
.. versionchanged:: 1.9.0
This parameter used to be called ``format``. To align it with other endpoints,
it was renamed to ``encoding``. The previous name still works, but is warned about.
Output format. Can be either ``pem`` or ``der``. Defaults to ``pem``.
exclude_cn_from_sans
If set to true, the Common Name is not added to the SANs.
Useful if the CN is not a hostname or email address.
kwargs
Any additional parameter accepted by the Vault API.
"""
endpoint = f"{mount}/issue/{role_name}"
if issuer_ref is not None:
endpoint = f"{mount}/issuer/{issuer_ref}/issue/{role_name}"
# Compatibility with releases pre-1.9.0
if "format" in kwargs:
log.warning(
"The `format` parameter to `vault_pki.issue_certificate` was renamed to `encoding`. Please update your call."
)
encoding = kwargs.pop("format")
if encoding is None:
encoding = "pem"
payload = {k: v for k, v in kwargs.items() if not k.startswith("_")}
payload["common_name"] = common_name
payload["format"] = encoding
payload["exclude_cn_from_sans"] = exclude_cn_from_sans
if ttl is not None:
payload["ttl"] = ttl
if alt_names is not None:
if not HAS_CRYPTOGRAPHY: # pragma: no cover
raise CommandExecutionError(
"Missing `cryptography` library, which is required for this operation"
)
dns_sans, ip_sans, uri_sans, other_sans = pki.split_sans(pki.norm_sans(alt_names))
payload["alt_names"] = ",".join(dns_sans)
payload["ip_sans"] = ",".join(ip_sans)
payload["uri_sans"] = ",".join(uri_sans)
payload["other_sans"] = ",".join(other_sans)
try:
return vault.api_post(endpoint, __opts__, __context__, payload=payload)["data"]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def sign_certificate(
role_name=None,
common_name=None,
mount="pki",
csr=None,
private_key=None,
private_key_passphrase=None,
digest="sha256",
issuer_ref=None,
alt_names=None,
ttl=None,
sign_verbatim=False,
encoding="pem",
exclude_cn_from_sans=False,
serial_number=None,
user_ids=None,
not_after=None,
# Vault sign-verbatim only args
key_usage=None,
ext_key_usage=None,
ext_key_usage_oids=None,
**kwargs,
):
"""
Issue a new leaf certificate from an existing private key or CSR.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#sign-certificate>`__.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#sign-verbatim>`__
Required policy:
.. code-block:: vaultpolicy
# When sign_verbatim is false and not specifying issuer_ref
path "<mount>/sign/<role_name>" {
capabilities = ["update"]
}
# When sign_verbatim is false and specifying issuer_ref
path "<mount>/issuer/<issuer_ref>/sign/<role_name>" {
capabilities = ["update"]
}
# When sign_verbatim is true and neither specifying issuer_ref nor role_name
path "<mount>/sign-verbatim" {
capabilities = ["update"]
}
# When sign_verbatim is true and specifying role_name, but not issuer_ref
path "<mount>/sign-verbatim/<role_name>" {
capabilities = ["update"]
}
# When sign_verbatim is true and specifying issuer_ref, but not role_name
path "<mount>/issuer/<issuer_ref>/sign-verbatim" {
capabilities = ["update"]
}
# When sign_verbatim is true and specifying both issuer_ref and role_name
path "<mount>/issuer/<issuer_ref>/sign-verbatim/<role_name>" {
capabilities = ["update"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.sign_certificate myrole common_name="www.example.com" private_key=/private/key/path.key
salt '*' vault_pki.sign_certificate myrole common_name="www.example.com" csr=/csr/path.csr
role_name
PKI role to use for issuing the certificate.
Required, unless ``sign_verbatim`` is true.
common_name
Subject common name (``CN``) for the certificate.
Required, unless the role explicitly sets ``require_cn`` to false or
``sign_verbatim`` is true.
Ignored (i.e. also not required) when a ``csr`` is passed that specifies
it and the role's ``use_csr_common_name`` is true (the default value).
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
csr
Pass the CSR which should be used for issuing the certificate. Can be text or path.
Either ``csr`` or ``private_key`` parameter can be set, not both.
private_key
Private key for which a certificate should be issued. Can be text or path.
Either ``csr`` or ``private_key`` parameter can be set, not both.
.. note::
This parameter requires the :py:mod:`x509_v2 execution module <salt.modules.x509_v2>` to be available.
When this parameter is set, a CSR is generated in place. You can influence the resulting CSR by providing
keyword arguments for :py:func:`x509.create_csr <salt.modules.x509_v2.create_csr>`, which are passed through.
See ``kwargs`` below.
private_key_passphrase
Passphrase for the ``private_key``, if encrypted. Not used in case of ``csr``.
digest
Digest to use for generating the CSR. Not used in case of ``csr``. Defaults to ``sha256``
issuer_ref
Specify an explicit issuer instead of taking it from the role definition.
Can be issuer_name or issuer_id.
alt_names
Any alternative names to add to the certificate.
Can be specified either as dict (``{ "<type>": "<value>" }``),
a dict of lists (``{ "<type>": ["<value1>", "<value2>", ...] }``)
or list of SAN strings (``["<type1>:<value1>", ...]``).
``<type>`` can be ``dns``, ``email``, ``uri``, ``ip`` or any OID for otherName SANs.
``<value>`` is the corresponding value. Note that otherName SANs need to omit ``UTF8:``.
Ignored when a ``csr`` is passed and the role's ``use_csr_sans`` is true (the default value).
ttl
Specifies the requested Time To Live (after which the certificate will be expired).
This cannot be larger than the engine's max (or, if not set, the system max).
Can be an integer, which is interpreted as seconds, or a time string such as ``1h``.
sign_verbatim
If set to true, the resulting certificate follows the CSR more or less exactly, including extensions.
Otherwise, only ``CN`` can be set for the subject, any other subject parameters (like ``O``) are
taken from the role.
.. warning::
This option uses a potentially dangerous endpoint. Be careful when using that option, as roles
are not restricting what can be issued anymore.
encoding
Output format. Can be either ``pem`` or ``der``. Defaults to ``pem``.
exclude_cn_from_sans
If set to true, the Common Name is not added to the SANs.
Useful if the CN is not a hostname or email address.
Has no effect when ``sign_verbatim`` is true.
serial_number
Single value for the **subject** SERIALNUMBER (OID: 2.5.4.5) name attribute (NOT the certificate's serial number!).
user_ids
List of User ID (``UID``) subject attributes.
Each one is added to the generated CSR's subject Name as a distinct RDN.
not_after
Absolute value of the Not After field of the certificate in UTC format ``YYYY-MM-ddTHH:MM:SSZ``.
When set, ``ttl`` is ignored.
key_usage
When ``sign_verbatim`` is true, list of key usages to encode onto the certificate if the
CSR does not specify a ``keyUsage`` extension. For non-verbatim issuance, this parameter
must not be specified because Vault takes it from the role.
Valid values can be found at https://golang.org/pkg/crypto/x509/#KeyUsage - simply drop the
``KeyUsage`` part of the value. Values are case-insensitive. Pass an empty list to specify
no constraints.
ext_key_usage
When ``sign_verbatim`` is true, list of extended key usages to encode onto the certificate if the
CSR does not specify an ``extendedKeyUsage`` extension. For non-verbatim issuance, this parameter
must not be specified because Vault takes it from the role.
Valid values can be found at https://golang.org/pkg/crypto/x509/#ExtKeyUsage - simply drop the
``ExtKeyUsage`` part of the value. Values are case-insensitive. Pass an empty list to specify
no constraints.
ext_key_usage_oids
When ``sign_verbatim`` is true, list of extended key usage oids to encode onto the certificate if the
CSR does not specify an ``extendedKeyUsage`` extension.
Useful for adding EKUs not supported by the Go standard library.
For non-verbatim issuance, this parameter must not be specified because Vault takes it from the role.
kwargs
Any additional parameter accepted by the Vault API or, if ``private_key`` is set, the
:py:func:`x509_v2 module <salt.modules.x509_v2.create_csr>`.
Note that ``CN`` and ``subjectAltName`` are overwritten with the
``common_name``/``alt_names`` parameters to this function, regardless of ``sign_verbatim``.
"""
if not HAS_CRYPTOGRAPHY: # pragma: no cover
raise CommandExecutionError(
"Missing `cryptography` library, which is required for this operation"
)
hlp.one_of(csr=csr, private_key=private_key)
if not sign_verbatim:
if not role_name:
raise SaltInvocationError("`role_name` is required when `sign_verbatim` is false")
hlp.none_of(
key_usage=key_usage,
ext_key_usage=ext_key_usage,
ext_key_usage_oids=ext_key_usage_oids,
_reason="sign_verbatim is false",
)
sign = "sign-verbatim" if sign_verbatim else "sign"
if issuer_ref is not None:
endpoint = f"{mount}/issuer/{issuer_ref}/{sign}"
else:
endpoint = f"{mount}/{sign}"
if role_name:
endpoint += f"/{role_name}"
csr_args, extra_args = pki.split_csr_kwargs(kwargs)
payload = {k: v for k, v in extra_args.items() if not k.startswith("_") and v is not None}
if not sign_verbatim:
payload["common_name"] = common_name
payload["exclude_cn_from_sans"] = exclude_cn_from_sans
if serial_number is not None:
payload["serial_number"] = serial_number
if user_ids is not None:
payload["user_ids"] = user_ids
else:
if key_usage is not None:
payload["key_usage"] = key_usage
if ext_key_usage is not None:
payload["ext_key_usage"] = ext_key_usage
if ext_key_usage_oids is not None:
payload["ext_key_usage_oids"] = ext_key_usage_oids
if ttl is not None:
payload["ttl"] = ttl
if not_after is not None:
payload["not_after"] = not_after
payload["format"] = encoding
norm_sans = None
if alt_names is not None:
norm_sans = pki.norm_sans(alt_names)
if not sign_verbatim:
dns_sans, ip_sans, uri_sans, other_sans = pki.split_sans(norm_sans)
payload["alt_names"] = ",".join(dns_sans)
payload["ip_sans"] = ",".join(ip_sans)
payload["uri_sans"] = ",".join(uri_sans)
payload["other_sans"] = ",".join(other_sans)
# In case private_key is passed, we're going to build a CSR in place.
if private_key is not None:
if norm_sans:
# Ensure we get the specified SANs, regardless of a role's use_csr_sans or sign_verbatim.
csr_args["subjectAltName"] = [
f"{k}:{vv}" if k.upper() in pki.SUPPORTED_SAN_TYPES else f"otherName:{k};UTF8:{vv}"
for k, v in norm_sans.items()
for vv in v
]
else:
# Ensure alt_names is the only parameter to specify SANs
csr_args.pop("subjectAltName", None)
# Ensure we get the specified CN, regardless of a role's use_csr_common_name or sign_verbatim
if common_name is not None:
csr_args["CN"] = common_name
else:
csr_args.pop("CN", None)
# Ensure user_ids and serial_number work the same across regular and verbatim signing
csr_args = pki.sync_verbatim_csr_subject(
csr_args, user_ids=user_ids, serial_number=serial_number
)
try:
csr = _x509v2(
"create_csr",
private_key=private_key,
private_key_passphrase=private_key_passphrase,
digest=digest,
**csr_args,
)
except SaltInvocationError as err:
if not norm_sans or "otherName is currently not implemented" not in str(err):
raise
# use_csr_sans does not matter for otherName SANs
csr_args["subjectAltName"] = [
f"{k}:{vv}"
for k, v in norm_sans.items()
if k.upper() in pki.SUPPORTED_SAN_TYPES
for vv in v
]
_, _, _, other_sans = pki.split_sans(norm_sans)
# sign-verbatim does not document, but still respects this
payload["other_sans"] = ",".join(other_sans)
csr = _x509v2(
"create_csr",
private_key=private_key,
private_key_passphrase=private_key_passphrase,
digest=digest,
**csr_args,
)
else:
# Ensure we load file paths and pass the CSR in PEM encoding
csr = _x509v2("encode_csr", csr)
payload["csr"] = csr
try:
return vault.api_post(endpoint, __opts__, __context__, payload=payload)["data"]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def revoke_certificate(
serial=None, certificate=None, private_key=None, private_key_passphrase=None, mount="pki"
):
"""
Revoke an issued certificate.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#revoke-certificate>`__.
Required policy:
.. code-block:: vaultpolicy
# when `private_key` is unspecified
path "<mount>/revoke" {
capabilities = ["create", "update"]
}
# when `private_key` is passed
path "<mount>/revoke-with-key" {
capabilities = ["create", "update"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.revoke_certificate 7e:85:c5:d1:85:94:9a:46:08:b5:1b:9c:22:cb:35:e5:ea:f3:56:3f
salt '*' vault_pki.revoke_certificate certificate=/etc/tls/my_cert.pem
salt '*' vault_pki.revoke_certificate certificate=/etc/tls/my_cert.pem private_key=/etc/tls/my_key.pem
serial
Specifies the serial of the certificate to revoke. Either ``serial`` or ``certificate`` must be specified.
certificate
Specifies the certificate (PEM or path) to revoke. Either ``serial`` or ``certificate`` must be specified.
.. note::
This parameter requires the :py:mod:`x509_v2 execution module <salt.modules.x509_v2>` to be available.
private_key
.. versionadded:: 1.9.0
Private key corresponding to the certificate issued by Vault that is attempted to be revoked.
Optional. When this is passed, a different, less trusted API endpoint is used.
private_key_passphrase
.. versionadded:: 1.9.0
Passphrase for ``private_key``, if specified and encrypted. Optional.
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/revoke"
payload = {}
hlp.one_of(serial=serial, certificate=certificate)
if private_key:
endpoint += "-with-key"
try:
if certificate is not None:
payload["certificate"] = _x509v2("encode_certificate", certificate)
elif serial is not None:
if isinstance(serial, int):
serial = hlp.dec2hex(serial)
payload["serial_number"] = serial
else: # pragma: no cover
raise RuntimeError("This path should not have been hit")
if private_key:
payload["private_key"] = _x509v2(
"encode_private_key", private_key, private_key_passphrase=private_key_passphrase
)
vault.api_put(endpoint, __opts__, __context__, payload=payload)
return True
except vault.VaultInvocationError:
return False
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def read_urls(mount="pki"):
"""
Fetch the URLs to be encoded in generated certificates.
No URL configuration is returned until the configuration is set.
`API method docs <https://developer.hashicorp.com/vault/api-docs/secret/pki#read-urls>`__.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/config/urls" {
capabilities = ["read"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.read_urls
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/config/urls"
try:
return vault.api_get(endpoint, __opts__, __context__)["data"]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def write_urls(
issuing_certificates=None,
crl_endpoints=None,
delta_crl_endpoints=None,
ocsp_servers=None,
aia_url_templating=None,
mount="pki",
):
"""
.. versionadded:: 1.9.0
Set issuing certificate endpoints, CRL distribution points, and OCSP server
endpoints that will be encoded into issued certificates. This behaves
as PATCH. To unset a value, set it to an empty string.
`API method docs <https://www.vaultproject.io/api-docs/secret/pki#set-urls>`_.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/config/urls" {
capabilities = ["create", "update"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.set_urls ocsp_servers=ocsp.my.ca
issuing_certificates
Specifies the URL values for the Issuing Certificate field as a list.
(see RFC 5280 Section 4.2.2.1 for details)
crl_endpoints
Specifies the URL values for the CRL Distribution Points field as a list.
(see RFC 5280 Section 4.2.1.13 for details)
delta_crl_endpoints
(Requires Vault 1.20+ or OpenBao)
Specifies the URL values for the Delta CRL Distribution Points field.
(see RFC 5280 Section 4.2.1.15 for details)
ocsp_servers
Specifies the URL values for the OCSP Servers field as a list.
(see RFC 5280 Section 4.2.2.1 for details)
aia_url_templating
Render ``issuing_certificates``/``crl_endpoints``/``ocsp_servers``/``delta_crl_endpoints`` as templates.
Supported variables: ``{{issuer_id}}``, ``{{cluster_path}}``, ``{{cluster_aia_path}}``
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/config/urls"
payload = hlp.filter_unset(
{
"issuing_certificates": issuing_certificates,
"crl_distribution_points": crl_endpoints,
"delta_crl_distribution_points": delta_crl_endpoints,
"ocsp_servers": ocsp_servers,
"enable_templating": aia_url_templating,
}
)
if not payload:
raise CommandExecutionError("You need to specify at least one parameter.")
try:
return vault.api_post(endpoint, __opts__, __context__, payload=payload)
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def read_cluster_config(mount="pki"):
"""
.. versionadded:: 1.9.0
Fetch cluster-local configuration, which is used in templated AIA URLs.
``path`` populates ``{{cluster_path}}`` and ``aia_path`` populates ``{{cluster_aia_path}}``.
`API method docs <https://www.vaultproject.io/api-docs/secret/pki#read-cluster-configuration>`_.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/config/cluster" {
capabilities = ["read"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.read_cluster_config
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
endpoint = f"{mount}/config/cluster"
try:
return vault.api_get(endpoint, __opts__, __context__)["data"]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
[docs]
def write_cluster_config(path=None, aia_path=None, mount="pki"):
"""
.. versionadded:: 1.9.0
Set cluster-local configuration, which is used in templated AIA URLs.
`API method docs <https://www.vaultproject.io/api-docs/secret/pki#set-cluster-configuration>`_.
Required policy:
.. code-block:: vaultpolicy
path "<mount>/config/cluster" {
capabilities = ["create", "update"]
}
CLI Example:
.. code-block:: bash
salt '*' vault_pki.write_cluster_config path='https://pr-a.vault.example.com/v1/ns1/pki-root'
path
URL of this performance replication cluster's API mount path,
including any namespaces as path components.
Example: ``https://pr-a.vault.example.com/v1/ns1/pki-root``.
aia_path
URL of this performance replication cluster's AIA distribution point;
may refer to an external, non-Vault responder.
This is for resolving AIA URLs and providing the ``{{cluster_aia_path}}`` template parameter
and will not be used for other purposes.
As such, unlike ``path``, this can safely use an insecure transit mechanism (like HTTP without TLS).
mount
Mount path the PKI backend is mounted to. Defaults to ``pki``.
"""
hlp.x_of(_max=2, path=path, aia_path=aia_path)
endpoint = f"{mount}/config/cluster"
payload = hlp.filter_unset({"path": path, "aia_path": aia_path})
try:
return vault.api_post(endpoint, __opts__, __context__, payload=payload)["data"]
except vault.VaultException as err:
raise CommandExecutionError(f"{type(err).__name__}: {err}") from err
def _x509v2(fun, *args, **kwargs):
try:
func = __salt__[f"x509.{fun}"]
except KeyError as err: # pragma: no cover
raise CommandExecutionError(
f"Missing `x509.{fun}`, provided by the builtin `x509_v2` execution module"
) from err
return func(*args, **kwargs)