[Key Vault] Handle cryptography RSA keys without local key material (#34330)

Author: mccoyp

sdk/keyvault/azure-keyvault-keys/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "python",
   "TagPrefix": "python/keyvault/azure-keyvault-keys",
-  "Tag": "python/keyvault/azure-keyvault-keys_d3b8eaba1a"
+  "Tag": "python/keyvault/azure-keyvault-keys_6a80b2f740"
 }

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

@@ -64,10 +64,10 @@ class KeyProperties(object):
 
     :keyword bool managed: Whether the key's lifetime is managed by Key Vault.
     :keyword tags: Application specific metadata in the form of key-value pairs.
-    :paramtype tags: dict[str, str]
+    :paramtype tags: dict[str, str] or None
     :keyword release_policy: The azure.keyvault.keys.KeyReleasePolicy specifying the rules under which the key
         can be exported.
-    :paramtype release_policy: ~azure.keyvault.keys.KeyReleasePolicy
+    :paramtype release_policy: ~azure.keyvault.keys.KeyReleasePolicy or None
     """
 
     def __init__(self, key_id: str, attributes: "Optional[_models.KeyAttributes]" = None, **kwargs: Any) -> None:
@@ -288,6 +288,8 @@ def __init__(self, encoded_policy: bytes, **kwargs: Any) -> None:
 class ReleaseKeyResult(object):
     """The result of a key release operation.
 
+    :ivar str value: A signed token containing the released key.
+
     :param str value: A signed token containing the released key.
     """
 

sdk/keyvault/azure-keyvault-keys/azure/keyvault/keys/crypto/_models.py

@@ -120,19 +120,14 @@ def get_signature_algorithm(padding: AsymmetricPadding, algorithm: HashAlgorithm
 class KeyVaultRSAPublicKey(RSAPublicKey):
     """An `RSAPublicKey` implementation based on a key managed by Key Vault.
 
-    Only synchronous clients and operations are supported at this time.
+    This class should not be instantiated directly. Instead, use the
+    :func:`~azure.keyvault.keys.crypto.CryptographyClient.create_rsa_public_key` method to create a key based on the
+    client's key. Only synchronous clients and operations are supported at this time.
     """
 
-    def __init__(self, client: "CryptographyClient", key_material: JsonWebKey) -> None:
-        """Creates a `KeyVaultRSAPublicKey` from a `CryptographyClient` and key.
-
-        :param client: The client that will be used to communicate with Key Vault.
-        :type client: ~azure.keyvault.keys.crypto.CryptographyClient
-        :param key_material: They Key Vault key's material, as a `JsonWebKey`.
-        :type key_material: ~azure.keyvault.keys.JsonWebKey
-        """
+    def __init__(self, client: "CryptographyClient", key_material: Optional[JsonWebKey] = None) -> None:
         self._client: "CryptographyClient" = client
-        self._key: JsonWebKey = key_material
+        self._key: Optional[JsonWebKey] = key_material
 
     def encrypt(self, plaintext: bytes, padding: AsymmetricPadding) -> bytes:
         """Encrypts the given plaintext.
@@ -156,7 +151,15 @@ def key_size(self) -> int:
 
         :returns: The key's size.
         :rtype: int
+
+        :raises ValueError: if the client is unable to obtain the key material from Key Vault.
         """
+        if self._key is None:
+            raise ValueError(
+                "Key material could not be obtained from Key Vault. Only remote cryptographic operations "
+                "(encrypt, verify) can be performed."
+            )
+
         public_key = self.public_numbers().public_key()
         return public_key.key_size
 
@@ -165,7 +168,15 @@ def public_numbers(self) -> RSAPublicNumbers:
 
         :returns: The public numbers of the key.
         :rtype: RSAPublicNumbers
+
+        :raises ValueError: if the client is unable to obtain the key material from Key Vault.
         """
+        if self._key is None:
+            raise ValueError(
+                "Key material could not be obtained from Key Vault. Only remote cryptographic operations "
+                "(encrypt, verify) can be performed."
+            )
+
         e = int.from_bytes(self._key.e, "big")  # type: ignore[attr-defined]
         n = int.from_bytes(self._key.n, "big")  # type: ignore[attr-defined]
         return RSAPublicNumbers(e, n)
@@ -184,7 +195,15 @@ def public_bytes(self, encoding: Encoding, format: PublicFormat) -> bytes:
 
         :returns: The serialized key.
         :rtype: bytes
+
+        :raises ValueError: if the client is unable to obtain the key material from Key Vault.
         """
+        if self._key is None:
+            raise ValueError(
+                "Key material could not be obtained from Key Vault. Only remote cryptographic operations "
+                "(encrypt, verify) can be performed."
+            )
+
         public_key = self.public_numbers().public_key()
         return public_key.public_bytes(encoding=encoding, format=format)
 
@@ -250,12 +269,18 @@ def recover_data_from_signature(
 
         :returns: The signed data.
         :rtype: bytes
-        :raises:
-            NotImplementedError if the local version of `cryptography` doesn't support this method.
-            :class:`~cryptography.exceptions.InvalidSignature` if the signature is invalid.
-            :class:`~cryptography.exceptions.UnsupportedAlgorithm` if the signature data recovery is not supported with
+        :raises NotImplementedError: if the local version of `cryptography` doesn't support this method.
+        :raises ~cryptography.exceptions.InvalidSignature: if the signature is invalid.
+        :raises ~cryptography.exceptions.UnsupportedAlgorithm: if the signature data recovery is not supported with
             the provided `padding` type.
+        :raises ValueError: if the client is unable to obtain the key material from Key Vault.
         """
+        if self._key is None:
+            raise ValueError(
+                "Key material could not be obtained from Key Vault. Only remote cryptographic operations "
+                "(encrypt, verify) can be performed."
+            )
+
         public_key = self.public_numbers().public_key()
         try:
             return public_key.recover_data_from_signature(signature=signature, padding=padding, algorithm=algorithm)
@@ -270,9 +295,13 @@ def __eq__(self, other: object) -> bool:
         :param object other: Another object to compare with this instance. Currently, only comparisons with
             `KeyVaultRSAPrivateKey` or `JsonWebKey` instances are supported.
 
-        :returns: True if the objects are equal; False otherwise.
+        :returns: True if the objects are equal; False if the objects are unequal or if key material can't be obtained
+            from Key Vault for comparison.
         :rtype: bool
         """
+        if self._key is None:
+            return False
+
         if isinstance(other, KeyVaultRSAPublicKey):
             return all(getattr(self._key, field) == getattr(other._key, field) for field in self._key._FIELDS)
         if isinstance(other, JsonWebKey):
@@ -289,19 +318,14 @@ def verifier(  # pylint:disable=docstring-missing-param,docstring-missing-return
 class KeyVaultRSAPrivateKey(RSAPrivateKey):
     """An `RSAPrivateKey` implementation based on a key managed by Key Vault.
 
-    Only synchronous clients and operations are supported at this time.
+    This class should not be instantiated directly. Instead, use the
+    :func:`~azure.keyvault.keys.crypto.CryptographyClient.create_rsa_private_key` method to create a key based on the
+    client's key. Only synchronous clients and operations are supported at this time.
     """
 
-    def __init__(self, client: "CryptographyClient", key_material: JsonWebKey) -> None:
-        """Creates a `KeyVaultRSAPrivateKey` from a `CryptographyClient` and key.
-
-        :param client: The client that will be used to communicate with Key Vault.
-        :type client: ~azure.keyvault.keys.crypto.CryptographyClient
-        :param key_material: They Key Vault key's material, as a `JsonWebKey`.
-        :type key_material: ~azure.keyvault.keys.JsonWebKey
-        """
+    def __init__(self, client: "CryptographyClient", key_material: Optional[JsonWebKey]) -> None:
         self._client: "CryptographyClient" = client
-        self._key: JsonWebKey = key_material
+        self._key: Optional[JsonWebKey] = key_material
 
     def decrypt(self, ciphertext: bytes, padding: AsymmetricPadding) -> bytes:
         """Decrypts the provided ciphertext.
@@ -325,7 +349,15 @@ def key_size(self) -> int:
 
         :returns: The key's size.
         :rtype: int
+
+        :raises ValueError: if the client is unable to obtain the key material from Key Vault.
         """
+        if self._key is None:
+            raise ValueError(
+                "Key material could not be obtained from Key Vault. Only remote cryptographic operations "
+                "(decrypt, sign) can be performed."
+            )
+
         # Key size only requires public modulus, which we can always get
         # Relying on private numbers instead would cause issues for keys stored in KV (which doesn't return private key)
         return self.public_key().key_size
@@ -374,7 +406,15 @@ def private_numbers(self) -> RSAPrivateNumbers:
 
         :returns: The private numbers of the key.
         :rtype: ~cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateNumbers
+
+        :raises ValueError: if the client is unable to obtain the key material from Key Vault.
         """
+        if self._key is None:
+            raise ValueError(
+                "Key material could not be obtained from Key Vault. Only remote cryptographic operations "
+                "(decrypt, sign) can be performed."
+            )
+
         # Fetch public numbers from JWK
         e = int.from_bytes(self._key.e, "big")  # type: ignore[attr-defined]
         n = int.from_bytes(self._key.n, "big")  # type: ignore[attr-defined]
@@ -420,7 +460,15 @@ def private_bytes(
 
         :returns: The serialized key.
         :rtype: bytes
+
+        :raises ValueError: if the client is unable to obtain the key material from Key Vault.
         """
+        if self._key is None:
+            raise ValueError(
+                "Key material could not be obtained from Key Vault. Only remote cryptographic operations "
+                "(decrypt, sign) can be performed."
+            )
+
         try:
             private_numbers = self.private_numbers()
         except ValueError as exc:

sdk/keyvault/azure-keyvault-keys/tests/_async_test_case.py

@@ -9,7 +9,6 @@
 from azure.core.pipeline import AsyncPipeline
 from azure.core.pipeline.transport import AioHttpTransport, HttpRequest
 from azure.keyvault.keys import KeyReleasePolicy
-from azure.keyvault.keys._shared.client_base import DEFAULT_VERSION, ApiVersion
 from devtools_testutils import AzureRecordedTestCase
 from _test_case import HSM_SUPPORTED_VERSIONS
 
@@ -42,7 +41,7 @@ def get_release_policy(attestation_uri, **kwargs):
 def get_test_parameters(only_hsm=False, only_vault=False, api_versions=None):
     """generates a list of parameter pairs for test case parameterization, where [x, y] = [api_version, is_hsm]"""
     combinations = []
-    versions = api_versions or ApiVersion
+    versions = api_versions or pytest.api_version  # pytest.api_version -> [DEFAULT_VERSION] if live, ApiVersion if not
 
     for api_version in versions:
         if not only_vault and api_version in HSM_SUPPORTED_VERSIONS:
@@ -74,7 +73,7 @@ def __init__(self, *args, **kwargs):
     def __call__(self, fn):
         async def _preparer(test_class, api_version, is_hsm, **kwargs):
 
-            self._skip_if_not_configured(api_version, is_hsm)
+            self._skip_if_not_configured(is_hsm)
             if not self.is_logging_enabled:
                 kwargs.update({"logging_enable": False})
             endpoint_url = self.managed_hsm_url if is_hsm else self.vault_url
@@ -98,10 +97,8 @@ def _set_mgmt_settings_real_values(self):
         if self.is_live:
             os.environ["AZURE_TENANT_ID"] = os.environ["KEYVAULT_TENANT_ID"]
             os.environ["AZURE_CLIENT_ID"] = os.environ["KEYVAULT_CLIENT_ID"]
-            os.environ["AZURE_CLIENT_SECRET"] = os.environ["KEYVAULT_CLIENT_SECRET"]
+            os.environ["AZURE_CLIENT_SECRET"] = os.environ.get("KEYVAULT_CLIENT_SECRET", "")  # Empty for user auth
 
-    def _skip_if_not_configured(self, api_version, is_hsm):
-        if self.is_live and api_version != DEFAULT_VERSION:
-            pytest.skip("This test only uses the default API version for live tests")
+    def _skip_if_not_configured(self, is_hsm):
         if self.is_live and is_hsm and self.managed_hsm_url is None:
             pytest.skip("No HSM endpoint for live testing")

sdk/keyvault/azure-keyvault-keys/tests/_test_case.py

@@ -9,7 +9,7 @@
 from azure.core.pipeline import Pipeline
 from azure.core.pipeline.transport import HttpRequest, RequestsTransport
 from azure.keyvault.keys import KeyReleasePolicy
-from azure.keyvault.keys._shared.client_base import DEFAULT_VERSION, ApiVersion
+from azure.keyvault.keys._shared.client_base import ApiVersion
 from devtools_testutils import AzureRecordedTestCase
 
 
@@ -44,7 +44,7 @@ def get_release_policy(attestation_uri, **kwargs):
 def get_test_parameters(only_hsm=False, only_vault=False, api_versions=None):
     """generates a list of parameter pairs for test case parameterization, where [x, y] = [api_version, is_hsm]"""
     combinations = []
-    versions = api_versions or pytest.api_version
+    versions = api_versions or pytest.api_version  # pytest.api_version -> [DEFAULT_VERSION] if live, ApiVersion if not
 
     for api_version in versions:
         if not only_vault and api_version in HSM_SUPPORTED_VERSIONS:
@@ -79,7 +79,7 @@ def __init__(self, *args, **kwargs):
     def __call__(self, fn):
         def _preparer(test_class, api_version, is_hsm, **kwargs):
 
-            self._skip_if_not_configured(api_version, is_hsm)
+            self._skip_if_not_configured(is_hsm)
             if not self.is_logging_enabled:
                 kwargs.update({"logging_enable": False})
             endpoint_url = self.managed_hsm_url if is_hsm else self.vault_url
@@ -102,11 +102,8 @@ def _set_mgmt_settings_real_values(self):
         if self.is_live:
             os.environ["AZURE_TENANT_ID"] = os.environ["KEYVAULT_TENANT_ID"]
             os.environ["AZURE_CLIENT_ID"] = os.environ["KEYVAULT_CLIENT_ID"]
-            os.environ["AZURE_CLIENT_SECRET"] = os.environ["KEYVAULT_CLIENT_SECRET"]
+            os.environ["AZURE_CLIENT_SECRET"] = os.environ.get("KEYVAULT_CLIENT_SECRET", "")  # Empty for user auth
 
-    def _skip_if_not_configured(self, api_version, is_hsm):
-
-        if self.is_live and api_version != DEFAULT_VERSION:
-            pytest.skip("This test only uses the default API version for live tests")
+    def _skip_if_not_configured(self, is_hsm):
         if self.is_live and is_hsm and self.managed_hsm_url is None:
             pytest.skip("No HSM endpoint for live testing")