Bind wrapped issuance to native metadata and preserve immutable recipient attempts

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

src/secrets_engine/openbao.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 import hashlib
 import json
 import os
+import re
 import shutil
 import stat
 import subprocess
@@ -400,23 +401,34 @@ class OpenBaoClient:
         self._run_ok(["lease", "revoke", lease_id])
 
     def _parse_wrap_response(self, stdout: str, *, ttl: str, creation_path: str) -> WrappedResponse:
+        # Verify native wrapping metadata. Never reconstruct a purported source
+        # path or TTL from the caller when the backend omitted or changed it.
         try:
             payload = json.loads(stdout)
-        except json.JSONDecodeError as exc:
-            raise BackendError("wrapped response is not JSON") from exc
+        except (json.JSONDecodeError, TypeError):
+            raise BackendError("wrapped response is not JSON") from None
         if not isinstance(payload, dict):
             raise BackendError("wrapped response is invalid")
         wrap = payload.get("wrap_info")
         if not isinstance(wrap, dict) or not wrap.get("token"):
             raise BackendError("response wrapping was not applied")
-        token = str(wrap["token"])
-        accessor = str(wrap.get("accessor") or "")
-        wrap_ttl = wrap.get("ttl")
+        if payload.get("data") or payload.get("auth"):
+            raise BackendError("wrapped response unexpectedly included unwrapped material")
+        match = re.fullmatch(r"([1-9][0-9]*)([smh])", ttl)
+        seconds = int(match[1]) * {"s": 1, "m": 60, "h": 3600}[match[2]] if match else 0
+        if not 0 < seconds <= 900:
+            raise BackendError("wrap TTL is outside the supported bound")
+        token, accessor = wrap.get("token"), wrap.get("accessor")
+        if (not isinstance(token, str) or not token or any(ch.isspace() for ch in token)
+                or not isinstance(accessor, str) or not accessor
+                or type(wrap.get("ttl")) is not int or wrap["ttl"] != seconds
+                or wrap.get("creation_path") != creation_path):
+            raise BackendError("native wrap identity, TTL or creation path unverified")
         return WrappedResponse(
             wrap_token=token,
-            accessor_fingerprint=accessor_fingerprint(accessor) if accessor else "",
-            ttl=str(wrap_ttl if wrap_ttl is not None else ttl),
-            creation_path=creation_path,
+            accessor_fingerprint=accessor_fingerprint(accessor),
+            ttl=str(seconds),
+            creation_path=wrap["creation_path"],
         )
 
     def wrap_kv_get(self, mount: str, path: str, *, ttl: str) -> WrappedResponse:
@@ -425,7 +437,7 @@ class OpenBaoClient:
         proc = self._run(["kv", "get", f"-wrap-ttl={ttl}", "-format=json", target])
         if proc.returncode != 0:
             raise BackendError("wrapped KV read failed")
-        return self._parse_wrap_response(proc.stdout, ttl=ttl, creation_path=target)
+        return self._parse_wrap_response(proc.stdout, ttl=ttl, creation_path=f"{mount}/data/{path}")
 
     def wrap_approle_secret_id(self, role_name: str, *, ttl: str) -> WrappedResponse:
         """Wrap a single-use AppRole secret_id. Fail if returned unwrapped."""