[Key Vault] Make issuer_name optional to support cert import (Azure#19141)

Author: mccoyp

sdk/keyvault/azure-keyvault-certificates/CHANGELOG.md

@@ -5,6 +5,7 @@ This is the last version to support Python 3.5. The next version will require Py
 ### Changed
 - Key Vault API version 7.2 is now the default
 - Updated minimum `msrest` version to 0.6.21
+- The `issuer_name` parameter for `CertificatePolicy` is now optional
 
 ### Added
 - Added class `KeyVaultCertificateIdentifier` that parses out a full ID returned by Key Vault,

sdk/keyvault/azure-keyvault-certificates/azure/keyvault/certificates/_client.py

@@ -35,6 +35,9 @@
     from azure.core.paging import ItemPaged
 
 
+NO_SAN_OR_SUBJECT = "You need to set either subject or one of the subject alternative names parameters in the policy"
+
+
 class CertificateClient(KeyVaultClientBase):
     """A high-level interface for managing a vault's certificates.
 
@@ -68,17 +71,20 @@ def begin_create_certificate(self, certificate_name, policy, **kwargs):
         an :class:`~azure.core.exceptions.HttpResponseError`
 
         :param str certificate_name: The name of the certificate.
-        :param policy: The management policy for the certificate.
+        :param policy: The management policy for the certificate. Either subject or one of the subject alternative
+            name properties are required.
         :type policy:
-         ~azure.keyvault.certificates.CertificatePolicy
+            ~azure.keyvault.certificates.CertificatePolicy
         :keyword bool enabled: Whether the certificate is enabled for use.
         :keyword tags: Application specific metadata in the form of key-value pairs.
         :paramtype tags: dict[str, str]
         :returns: An LROPoller for the create certificate operation. Waiting on the poller
-         gives you the certificate if creation is successful, the CertificateOperation if not.
+            gives you the certificate if creation is successful, the CertificateOperation if not.
         :rtype: ~azure.core.polling.LROPoller[~azure.keyvault.certificates.KeyVaultCertificate or
-         ~azure.keyvault.certificates.CertificateOperation]
-        :raises: :class:`~azure.core.exceptions.HttpResponseError`
+            ~azure.keyvault.certificates.CertificateOperation]
+        :raises:
+            :class:`ValueError` if the certificate policy is invalid,
+            :class:`~azure.core.exceptions.HttpResponseError` for other errors.
 
         Keyword arguments
             - *enabled (bool)* - Determines whether the object is enabled.
@@ -92,12 +98,14 @@ def begin_create_certificate(self, certificate_name, policy, **kwargs):
                 :caption: Create a certificate
                 :dedent: 8
         """
+        if not (policy.san_emails or policy.san_user_principal_names or policy.san_dns_names or policy.subject):
+            raise ValueError(NO_SAN_OR_SUBJECT)
+
         polling_interval = kwargs.pop("_polling_interval", None)
         if polling_interval is None:
             polling_interval = 5
         enabled = kwargs.pop("enabled", None)
 
-
         if enabled is not None:
             attributes = self._models.CertificateAttributes(enabled=enabled)
         else:
@@ -106,7 +114,7 @@ def begin_create_certificate(self, certificate_name, policy, **kwargs):
         parameters = self._models.CertificateCreateParameters(
             certificate_policy=policy._to_certificate_policy_bundle(),
             certificate_attributes=attributes,
-            tags=kwargs.pop("tags", None)
+            tags=kwargs.pop("tags", None),
         )
 
         cert_bundle = self._client.create_certificate(
@@ -332,7 +340,6 @@ def begin_recover_deleted_certificate(self, certificate_name, **kwargs):
 
         return KeyVaultOperationPoller(polling_method)
 
-
     @distributed_trace
     def import_certificate(self, certificate_name, certificate_bytes, **kwargs):
         # type: (str, bytes, **Any) -> KeyVaultCertificate
@@ -459,8 +466,7 @@ def update_certificate_properties(self, certificate_name, version=None, **kwargs
             attributes = None
 
         parameters = self._models.CertificateUpdateParameters(
-            certificate_attributes=attributes,
-            tags=kwargs.pop("tags", None)
+            certificate_attributes=attributes, tags=kwargs.pop("tags", None)
         )
 
         bundle = self._client.update_certificate(
@@ -528,7 +534,8 @@ def restore_certificate_backup(self, backup, **kwargs):
         bundle = self._client.restore_certificate(
             vault_base_url=self.vault_url,
             parameters=self._models.CertificateRestoreParameters(certificate_bundle_backup=backup),
-            error_map=_error_map, **kwargs
+            error_map=_error_map,
+            **kwargs
         )
         return KeyVaultCertificate._from_certificate_bundle(certificate_bundle=bundle)
 
@@ -795,9 +802,7 @@ def merge_certificate(self, certificate_name, x509_certificates, **kwargs):
             attributes = None
 
         parameters = self._models.CertificateMergeParameters(
-            x509_certificates=x509_certificates,
-            certificate_attributes=attributes,
-            tags=kwargs.pop("tags", None)
+            x509_certificates=x509_certificates, certificate_attributes=attributes, tags=kwargs.pop("tags", None)
         )
 
         bundle = self._client.merge_certificate(
@@ -884,9 +889,7 @@ def create_issuer(self, issuer_name, provider, **kwargs):
         else:
             admin_details = None
         if organization_id or admin_details:
-            organization_details = self._models.OrganizationDetails(
-                id=organization_id, admin_details=admin_details
-            )
+            organization_details = self._models.OrganizationDetails(id=organization_id, admin_details=admin_details)
         else:
             organization_details = None
         if enabled is not None:
@@ -902,11 +905,7 @@ def create_issuer(self, issuer_name, provider, **kwargs):
         )
 
         issuer_bundle = self._client.set_certificate_issuer(
-            vault_base_url=self.vault_url,
-            issuer_name=issuer_name,
-            parameter=parameters,
-            error_map=_error_map,
-            **kwargs
+            vault_base_url=self.vault_url, issuer_name=issuer_name, parameter=parameters, error_map=_error_map, **kwargs
         )
         return CertificateIssuer._from_issuer_bundle(issuer_bundle=issuer_bundle)
 
@@ -951,9 +950,7 @@ def update_issuer(self, issuer_name, **kwargs):
         else:
             admin_details = None
         if organization_id or admin_details:
-            organization_details = self._models.OrganizationDetails(
-                id=organization_id, admin_details=admin_details
-            )
+            organization_details = self._models.OrganizationDetails(id=organization_id, admin_details=admin_details)
         else:
             organization_details = None
         if enabled is not None:
@@ -965,15 +962,11 @@ def update_issuer(self, issuer_name, **kwargs):
             provider=kwargs.pop("provider", None),
             credentials=issuer_credentials,
             organization_details=organization_details,
-            attributes=issuer_attributes
+            attributes=issuer_attributes,
         )
 
         issuer_bundle = self._client.update_certificate_issuer(
-            vault_base_url=self.vault_url,
-            issuer_name=issuer_name,
-            parameter=parameters,
-            error_map=_error_map,
-            **kwargs
+            vault_base_url=self.vault_url, issuer_name=issuer_name, parameter=parameters, error_map=_error_map, **kwargs
         )
         return CertificateIssuer._from_issuer_bundle(issuer_bundle=issuer_bundle)
 

sdk/keyvault/azure-keyvault-certificates/azure/keyvault/certificates/_models.py

@@ -618,17 +618,19 @@ def request_id(self):
 class CertificatePolicy(object):
     """Management policy for a certificate.
 
-    :param str issuer_name: Name of the referenced issuer object or reserved names; for example,
-        'Self' or 'Unknown"
+    :param Optional[str] issuer_name: Optional. Name of the referenced issuer object or reserved names; for example,
+        :attr:`~azure.keyvault.certificates.WellKnownIssuerNames.self` or
+        :attr:`~azure.keyvault.certificates.WellKnownIssuerNames.unknown`
     :keyword str subject: The subject name of the certificate. Should be a valid X509
-        distinguished name. Either subject or one of the subject alternative name parameters
-        are required.
+        distinguished name. Either subject or one of the subject alternative name parameters are required for
+        creating a certificate. This will be ignored when importing a certificate; the subject will be parsed from
+        the imported certificate.
     :keyword Iterable[str] san_emails: Subject alternative emails of the X509 object. Either
-        subject or one of the subject alternative name parameters are required.
+        subject or one of the subject alternative name parameters are required for creating a certificate.
     :keyword Iterable[str] san_dns_names: Subject alternative DNS names of the X509 object. Either
-        subject or one of the subject alternative name parameters are required.
+        subject or one of the subject alternative name parameters are required for creating a certificate.
     :keyword Iterable[str] san_user_principal_names: Subject alternative user principal names of the X509 object.
-        Either subject or one of the subject alternative name parameters are required.
+        Either subject or one of the subject alternative name parameters are required for creating a certificate.
     :keyword bool exportable: Indicates if the private key can be exported. For valid values,
         see KeyType.
     :keyword key_type: The type of key pair to be used for the certificate.
@@ -659,7 +661,7 @@ class CertificatePolicy(object):
     # pylint:disable=too-many-instance-attributes
     def __init__(
         self,
-        issuer_name,  # type: str
+        issuer_name=None,  # type: Optional[str]
         **kwargs  # type: Any
     ):
         # type: (...) -> None
@@ -682,12 +684,6 @@ def __init__(
         self._san_dns_names = kwargs.pop("san_dns_names", None) or None
         self._san_user_principal_names = kwargs.pop("san_user_principal_names", None) or None
 
-        if not (
-            self._san_emails or self._san_user_principal_names or self._san_dns_names or self._subject
-        ):
-            raise ValueError("You need to set either subject or one of the subject alternative names " +
-                            "parameters")
-
     @classmethod
     def get_default(cls):
         return cls(issuer_name=WellKnownIssuerNames.self, subject="CN=DefaultPolicy")
@@ -987,7 +983,7 @@ def lifetime_actions(self):
 
     @property
     def issuer_name(self):
-        # type: () -> str
+        # type: () -> Optional[str]
         """Name of the referenced issuer object or reserved names for the issuer
         of the certificate.
 

sdk/keyvault/azure-keyvault-certificates/azure/keyvault/certificates/aio/_client.py

@@ -23,6 +23,7 @@
     IssuerProperties,
 )
 from ._polling_async import CreateCertificatePollerAsync
+from .._client import NO_SAN_OR_SUBJECT
 from .._shared import AsyncKeyVaultClientBase
 from .._shared._polling_async import AsyncDeleteRecoverPollingMethod
 from .._shared.exceptions import error_map as _error_map
@@ -62,17 +63,20 @@ async def create_certificate(
         an :class:`~azure.core.exceptions.HttpResponseError`
 
         :param str certificate_name: The name of the certificate.
-        :param policy: The management policy for the certificate.
+        :param policy: The management policy for the certificate. Either subject or one of the subject alternative
+            name properties are required.
         :type policy:
-         ~azure.keyvault.certificates.CertificatePolicy
+            ~azure.keyvault.certificates.CertificatePolicy
         :keyword bool enabled: Whether the certificate is enabled for use.
         :keyword tags: Application specific metadata in the form of key-value pairs.
         :paramtype tags: dict[str, str]
         :returns: A coroutine for the creation of the certificate. Awaiting the coroutine
-         returns the created KeyVaultCertificate if creation is successful, the CertificateOperation if not.
+            returns the created KeyVaultCertificate if creation is successful, the CertificateOperation if not.
         :rtype: ~azure.keyvault.certificates.KeyVaultCertificate or
-         ~azure.keyvault.certificates.CertificateOperation
-        :raises: :class:`~azure.core.exceptions.HttpResponseError`
+            ~azure.keyvault.certificates.CertificateOperation
+        :raises:
+            :class:`ValueError` if the certificate policy is invalid,
+            :class:`~azure.core.exceptions.HttpResponseError` for other errors.
 
         Example:
             .. literalinclude:: ../tests/test_examples_certificates_async.py
@@ -82,6 +86,9 @@ async def create_certificate(
                 :caption: Create a certificate
                 :dedent: 8
         """
+        if not (policy.san_emails or policy.san_user_principal_names or policy.san_dns_names or policy.subject):
+            raise ValueError(NO_SAN_OR_SUBJECT)
+
         polling_interval = kwargs.pop("_polling_interval", None)
         if polling_interval is None:
             polling_interval = 5
@@ -95,7 +102,7 @@ async def create_certificate(
         parameters = self._models.CertificateCreateParameters(
             certificate_policy=policy._to_certificate_policy_bundle(),
             certificate_attributes=attributes,
-            tags=kwargs.pop("tags", None)
+            tags=kwargs.pop("tags", None),
         )
 
         cert_bundle = await self._client.create_certificate(
@@ -435,8 +442,7 @@ async def update_certificate_properties(
             attributes = None
 
         parameters = self._models.CertificateUpdateParameters(
-            certificate_attributes=attributes,
-            tags=kwargs.pop("tags", None)
+            certificate_attributes=attributes, tags=kwargs.pop("tags", None)
         )
 
         bundle = await self._client.update_certificate(
@@ -774,9 +780,7 @@ async def merge_certificate(
             attributes = None
 
         parameters = self._models.CertificateMergeParameters(
-            x509_certificates=x509_certificates,
-            certificate_attributes=attributes,
-            tags=kwargs.pop("tags", None)
+            x509_certificates=x509_certificates, certificate_attributes=attributes, tags=kwargs.pop("tags", None)
         )
 
         bundle = await self._client.merge_certificate(
@@ -861,9 +865,7 @@ async def create_issuer(self, issuer_name: str, provider: str, **kwargs: "Any")
         else:
             admin_details = None
         if organization_id or admin_details:
-            organization_details = self._models.OrganizationDetails(
-                id=organization_id, admin_details=admin_details
-            )
+            organization_details = self._models.OrganizationDetails(id=organization_id, admin_details=admin_details)
         else:
             organization_details = None
         if enabled is not None:
@@ -879,11 +881,7 @@ async def create_issuer(self, issuer_name: str, provider: str, **kwargs: "Any")
         )
 
         issuer_bundle = await self._client.set_certificate_issuer(
-            vault_base_url=self.vault_url,
-            issuer_name=issuer_name,
-            parameter=parameters,
-            error_map=_error_map,
-            **kwargs
+            vault_base_url=self.vault_url, issuer_name=issuer_name, parameter=parameters, error_map=_error_map, **kwargs
         )
         return CertificateIssuer._from_issuer_bundle(issuer_bundle=issuer_bundle)
 
@@ -928,9 +926,7 @@ async def update_issuer(self, issuer_name: str, **kwargs: "Any") -> CertificateI
         else:
             admin_details = None
         if organization_id or admin_details:
-            organization_details = self._models.OrganizationDetails(
-                id=organization_id, admin_details=admin_details
-            )
+            organization_details = self._models.OrganizationDetails(id=organization_id, admin_details=admin_details)
         else:
             organization_details = None
         if enabled is not None:
@@ -942,15 +938,11 @@ async def update_issuer(self, issuer_name: str, **kwargs: "Any") -> CertificateI
             provider=kwargs.pop("provider", None),
             credentials=issuer_credentials,
             organization_details=organization_details,
-            attributes=issuer_attributes
+            attributes=issuer_attributes,
         )
 
         issuer_bundle = await self._client.update_certificate_issuer(
-            vault_base_url=self.vault_url,
-            issuer_name=issuer_name,
-            parameter=parameters,
-            error_map=_error_map,
-            **kwargs
+            vault_base_url=self.vault_url, issuer_name=issuer_name, parameter=parameters, error_map=_error_map, **kwargs
         )
         return CertificateIssuer._from_issuer_bundle(issuer_bundle=issuer_bundle)
 

sdk/keyvault/azure-keyvault-certificates/tests/test_certificates_client.py

@@ -24,8 +24,10 @@
     CertificateContentType,
     LifetimeAction,
     CertificateIssuer,
-    IssuerProperties
+    IssuerProperties,
+    WellKnownIssuerNames
 )
+from azure.keyvault.certificates._client import NO_SAN_OR_SUBJECT
 import pytest
 
 from _shared.test_case import KeyVaultTestCase
@@ -680,6 +682,19 @@ def test_list_deleted_certificates(self, client, **kwargs):
         assert "The 'include_pending' parameter to `list_deleted_certificates` is only available for API versions v7.0 and up" in str(excinfo.value)
 
 
+def test_policy_expected_errors_for_create_cert():
+    """Either a subject or subject alternative name property are required for creating a certificate"""
+    client = CertificateClient("...", object())
+
+    with pytest.raises(ValueError, match=NO_SAN_OR_SUBJECT):
+        policy = CertificatePolicy()
+        client.begin_create_certificate("...", policy=policy)
+
+    with pytest.raises(ValueError, match=NO_SAN_OR_SUBJECT):
+        policy = CertificatePolicy(issuer_name=WellKnownIssuerNames.self)
+        client.begin_create_certificate("...", policy=policy)
+
+
 def test_service_headers_allowed_in_logs():
     service_headers = {"x-ms-keyvault-network-info", "x-ms-keyvault-region", "x-ms-keyvault-service-version"}
     client = CertificateClient("...", object())