Allow for reading deleted secret versions (kv2) without an exception (#…

…907)

* add test to verify the desired behavior

* blackify

* update test

* nominal fix (no flag)

* revert kv_v2

* add VaultError.from_status class method

* add optional text and json fields to VaultError instances

* update raise_for_error to incorportate new VaultError fields and class method

* update RawAdapter to provide text and json in exceptions

* update kv2 read_secret_version to handle latest deleted secret

* we already got the text

* missed some conditionals

* cleanerrrr

* add raise_on_deleted_version with deprecated default

* add unit tests

* update integration tests

* fix unit syntax

* refactor and test

* text prop mock

* blackify

* who's mocking whom

* update docs

hvac/adapters.py

@@ -237,6 +237,34 @@ class RawAdapter(Adapter):
     but always returns Response objects for requests.
     """
 
+    def _raise_for_error(self, method: str, url: str, response: requests.Response):
+        msg = json = text = errors = None
+        try:
+            text = response.text
+        except Exception:
+            pass
+
+        if response.headers.get("Content-Type") == "application/json":
+            try:
+                json = response.json()
+            except Exception:
+                pass
+            else:
+                errors = json.get("errors")
+
+        if errors is None:
+            msg = text
+
+        utils.raise_for_error(
+            method,
+            url,
+            response.status_code,
+            msg,
+            errors=errors,
+            text=text,
+            json=json,
+        )
+
     def get_login_token(self, response):
         """Extracts the client token from a login response.
 
@@ -309,17 +337,7 @@ def request(self, method, url, headers=None, raise_exception=True, **kwargs):
         )
 
         if not response.ok and (raise_exception and not self.ignore_exceptions):
-            text = errors = None
-            if response.headers.get("Content-Type") == "application/json":
-                try:
-                    errors = response.json().get("errors")
-                except Exception:
-                    pass
-            if errors is None:
-                text = response.text
-            utils.raise_for_error(
-                method, url, response.status_code, text, errors=errors
-            )
+            self._raise_for_error(method, url, response)
 
         return response
 

hvac/api/secrets_engines/kv_v2.py

@@ -1,5 +1,8 @@
 #!/usr/bin/env python
 """KvV2 methods module."""
+
+import warnings
+
 from hvac import exceptions, utils
 from hvac.api.vault_api_base import VaultApiBase
 
@@ -69,12 +72,44 @@ def read_configuration(self, mount_point=DEFAULT_MOUNT_POINT):
         )
         return self._adapter.get(url=api_path)
 
-    def read_secret(self, path, mount_point=DEFAULT_MOUNT_POINT):
-        return self.read_secret_version(path, mount_point=mount_point)
-
-    def read_secret_version(self, path, version=None, mount_point=DEFAULT_MOUNT_POINT):
+    def read_secret(
+        self, path, mount_point=DEFAULT_MOUNT_POINT, raise_on_deleted_version=None
+    ):
         """Retrieve the secret at the specified location.
 
+        Equivalent to calling read_secret_version with version=None.
+
+        Supported methods:
+            GET: /{mount_point}/data/{path}. Produces: 200 application/json
+
+
+        :param path: Specifies the path of the secret to read. This is specified as part of the URL.
+        :type path: str | unicode
+        :param mount_point: The "path" the secret engine was mounted on.
+        :type mount_point: str | unicode
+        :param raise_on_deleted_version: Changes the behavior when the requested version is deleted.
+            If True an exception will be raised.
+            If False, some metadata about the deleted secret is returned.
+            If None (pre-v3), a default of True will be used and a warning will be issued.
+        :type raise_on_deleted_version: bool
+        :return: The JSON response of the request.
+        :rtype: dict
+        """
+        return self.read_secret_version(
+            path,
+            mount_point=mount_point,
+            raise_on_deleted_version=raise_on_deleted_version,
+        )
+
+    def read_secret_version(
+        self,
+        path,
+        version=None,
+        mount_point=DEFAULT_MOUNT_POINT,
+        raise_on_deleted_version=None,
+    ):
+        """Retrieve the secret at the specified location, with the specified version.
+
         Supported methods:
             GET: /{mount_point}/data/{path}. Produces: 200 application/json
 
@@ -85,19 +120,52 @@ def read_secret_version(self, path, version=None, mount_point=DEFAULT_MOUNT_POIN
         :type version: int
         :param mount_point: The "path" the secret engine was mounted on.
         :type mount_point: str | unicode
+        :param raise_on_deleted_version: Changes the behavior when the requested version is deleted.
+            If True an exception will be raised.
+            If False, some metadata about the deleted secret is returned.
+            If None (pre-v3), a default of True will be used and a warning will be issued.
+        :type raise_on_deleted_version: bool
         :return: The JSON response of the request.
         :rtype: dict
         """
+
+        if raise_on_deleted_version is None:
+            msg = (
+                "The raise_on_deleted parameter will change its default value to False in hvac v3.0.0. "
+                "The current default of True will presere previous behavior. "
+                "To use the old behavior with no warning, explicitly set this value to True. "
+                "See https://github.com/hvac/hvac/pull/907"
+            )
+            warnings.warn(
+                message=msg,
+                category=DeprecationWarning,
+                stacklevel=2,
+            )
+            raise_on_deleted_version = True
+
         params = {}
         if version is not None:
             params["version"] = version
         api_path = utils.format_url(
             "/v1/{mount_point}/data/{path}", mount_point=mount_point, path=path
         )
-        return self._adapter.get(
-            url=api_path,
-            params=params,
-        )
+        try:
+            return self._adapter.get(
+                url=api_path,
+                params=params,
+            )
+        except exceptions.InvalidPath as e:
+            if not raise_on_deleted_version:
+                try:
+                    if (
+                        e.json is not None
+                        and e.json["data"]["metadata"]["deletion_time"] != ""
+                    ):
+                        return e.json
+                except KeyError:
+                    pass
+
+            raise
 
     def create_or_update_secret(
         self, path, secret, cas=None, mount_point=DEFAULT_MOUNT_POINT

hvac/exceptions.py

@@ -1,17 +1,37 @@
 class VaultError(Exception):
-    def __init__(self, message=None, errors=None, method=None, url=None):
+    def __init__(
+        self, message=None, errors=None, method=None, url=None, text=None, json=None
+    ):
         if errors:
             message = ", ".join(errors)
 
         self.errors = errors
         self.method = method
         self.url = url
+        self.text = text
+        self.json = json
 
         super().__init__(message)
 
     def __str__(self):
         return f"{self.args[0]}, on {self.method} {self.url}"
 
+    @classmethod
+    def from_status(cls, status_code: int, *args, **kwargs):
+        _STATUS_EXCEPTION_MAP = {
+            400: InvalidRequest,
+            401: Unauthorized,
+            403: Forbidden,
+            404: InvalidPath,
+            429: RateLimitExceeded,
+            500: InternalServerError,
+            501: VaultNotInitialized,
+            502: BadGateway,
+            503: VaultDown,
+        }
+
+        return _STATUS_EXCEPTION_MAP.get(status_code, UnexpectedError)(*args, **kwargs)
+
 
 class InvalidRequest(VaultError):
     pass

hvac/utils.py

@@ -12,7 +12,9 @@
 from hvac import exceptions
 
 
-def raise_for_error(method, url, status_code, message=None, errors=None):
+def raise_for_error(
+    method, url, status_code, message=None, errors=None, text=None, json=None
+):
     """Helper method to raise exceptions based on the status code of a response received back from Vault.
 
     :param method: HTTP method of a request to Vault.
@@ -25,39 +27,26 @@ def raise_for_error(method, url, status_code, message=None, errors=None):
     :type message: str
     :param errors: Optional errors to include in a resulting exception.
     :type errors: list | str
+    :param text: Optional text of the response.
+    :type text: str
+    :param json: Optional deserialized version of a JSON response (object)
+    :type json: object
 
     :raises: hvac.exceptions.InvalidRequest | hvac.exceptions.Unauthorized | hvac.exceptions.Forbidden |
         hvac.exceptions.InvalidPath | hvac.exceptions.RateLimitExceeded | hvac.exceptions.InternalServerError |
         hvac.exceptions.VaultNotInitialized | hvac.exceptions.BadGateway | hvac.exceptions.VaultDown |
         hvac.exceptions.UnexpectedError
 
     """
-    if status_code == 400:
-        raise exceptions.InvalidRequest(message, errors=errors, method=method, url=url)
-    elif status_code == 401:
-        raise exceptions.Unauthorized(message, errors=errors, method=method, url=url)
-    elif status_code == 403:
-        raise exceptions.Forbidden(message, errors=errors, method=method, url=url)
-    elif status_code == 404:
-        raise exceptions.InvalidPath(message, errors=errors, method=method, url=url)
-    elif status_code == 429:
-        raise exceptions.RateLimitExceeded(
-            message, errors=errors, method=method, url=url
-        )
-    elif status_code == 500:
-        raise exceptions.InternalServerError(
-            message, errors=errors, method=method, url=url
-        )
-    elif status_code == 501:
-        raise exceptions.VaultNotInitialized(
-            message, errors=errors, method=method, url=url
-        )
-    elif status_code == 502:
-        raise exceptions.BadGateway(message, errors=errors, method=method, url=url)
-    elif status_code == 503:
-        raise exceptions.VaultDown(message, errors=errors, method=method, url=url)
-    else:
-        raise exceptions.UnexpectedError(message or errors, method=method, url=url)
+    raise exceptions.VaultError.from_status(
+        status_code,
+        message,
+        errors=errors,
+        method=method,
+        url=url,
+        text=text,
+        json=json,
+    )
 
 
 def generate_method_deprecation_message(

tests/integration_tests/api/secrets_engines/test_kv_v2.py

@@ -275,19 +275,31 @@ def test_patch(
                 self.assertEqual(first=v, second=read_secret_result["data"]["data"][k])
 
     @parameterized.expand(
-        [
-            ("successful delete one version written", "hvac"),
+        (
+            combo[0],
+            combo[1],
+            combo[2],
+            dict(enumerate(combo)).get(3, None),
+            raise_on_del,
+            recoverable,
+        )
+        for combo in [
+            ("successful delete one version written", "hvac", 1),
             ("successful delete two versions written", "hvac", 2),
             ("successful delete three versions written", "hvac", 3),
             ("nonexistent path", "no-secret-here", 0, exceptions.InvalidPath),
         ]
+        for raise_on_del in [True, False]
+        for recoverable in [True, False]
     )
     def test_delete_latest_version_of_secret(
         self,
         test_label,
         path,
-        write_secret_before_test=1,
+        write_secret_before_test,
         raises=None,
+        raise_on_del=False,
+        recoverable=None,
         exception_message="",
     ):
         if write_secret_before_test:
@@ -346,6 +358,39 @@ def test_delete_latest_version_of_secret(
                     second="",
                 )
 
+            should_raise = raise_on_del or not recoverable
+
+            try:
+                read_secret_version_result = (
+                    self.client.secrets.kv.v2.read_secret_version(
+                        path=path,
+                        mount_point=self.DEFAULT_MOUNT_POINT,
+                        raise_on_deleted_version=raise_on_del,
+                    )
+                )
+            except exceptions.InvalidPath:
+                if not should_raise:
+                    raise
+            else:
+                logging.debug(
+                    "read_secret_version_result: %s" % read_secret_version_result
+                )
+                self.assertEqual(
+                    first=read_secret_version_result["data"]["data"],
+                    second=None,
+                )
+                self.assertNotEqual(
+                    first=read_secret_version_result["data"]["metadata"][
+                        "deletion_time"
+                    ],
+                    second="",
+                )
+                self.assertEqual(
+                    first=read_secret_version_result["data"]["metadata"]["version"],
+                    second=write_secret_before_test,
+                    msg=repr(read_secret_version_result),
+                )
+
     @parameterized.expand(
         [
             ("successful delete one version written", "hvac", 1, [1]),