RPF-WP-0015-T02: select the lease issued for the mounted refresh; leave predecessors

ESO refresh (15m) and role TTL (15m, max 1h) overlap, so the first window
saw two live handles and aborted before revocation.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 680842@bnt-lap001
Assistant-Session: 54eb9e8e-7d6b-4f0c-b7a6-a6db62075883

scripts/audit-core-database-lease-recovery.py

@@ -160,21 +160,7 @@ class Authority:
     def bao(self, args: list[str], *, label: str) -> str:
         return safe_run(["bao", *args], label=label, env=self.env).stdout.strip()
 
-    def single_runtime_lease(self) -> dict[str, Any]:
-        raw = self.bao(
-            ["list", "-format=json", f"sys/leases/lookup/{LEASE_PREFIX}"],
-            label="list exact runtime lease handles",
-        )
-        try:
-            handles = json.loads(raw)
-        except json.JSONDecodeError as exc:
-            raise ProcedureError("runtime lease list is invalid JSON") from exc
-        if not isinstance(handles, list) or len(handles) != 1:
-            count = len(handles) if isinstance(handles, list) else "unknown"
-            raise ProcedureError(
-                f"exact runtime prefix must contain one live handle (observed {count})"
-            )
-        lease_id = f"{LEASE_PREFIX}/{handles[0]}"
+    def _lookup(self, lease_id: str) -> dict[str, Any]:
         lookup_raw = self.bao(
             ["write", "-format=json", "sys/leases/lookup", f"lease_id={lease_id}"],
             label="lookup exact runtime lease metadata",
@@ -194,6 +180,28 @@ class Authority:
             "ttl": ttl,
         }
 
+    def runtime_lease_for(self, secret: dict[str, Any]) -> dict[str, Any]:
+        """The one handle issued for the mounted Secret's refresh.
+
+        ESO refreshes every 15m while the role's TTL is 15m (max 1h), so the
+        predecessor of the mounted credential can still be live. Select only the
+        handle coherent with the Secret refresh; every other handle must be an
+        older predecessor, which is never revoked.
+        """
+        raw = self.bao(
+            ["list", "-format=json", f"sys/leases/lookup/{LEASE_PREFIX}"],
+            label="list exact runtime lease handles",
+        )
+        try:
+            handles = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise ProcedureError("runtime lease list is invalid JSON") from exc
+        if not isinstance(handles, list) or not 1 <= len(handles) <= MAX_RUNTIME_HANDLES:
+            count = len(handles) if isinstance(handles, list) else "unknown"
+            raise ProcedureError(f"exact runtime prefix has an unexpected handle count ({count})")
+        leases = [self._lookup(f"{LEASE_PREFIX}/{handle}") for handle in handles]
+        return select_runtime_lease(leases, secret)
+
     def revoke(self, lease_id: str) -> None:
         self.bao(["lease", "revoke", lease_id], label="revoke exact runtime lease")
 
@@ -307,6 +315,24 @@ def pod_state(remote: Any) -> dict[str, Any]:
     }
 
 
+MAX_RUNTIME_HANDLES = 8
+
+
+def select_runtime_lease(leases: list[dict[str, Any]], secret: dict[str, Any]) -> dict[str, Any]:
+    coherent = [
+        lease for lease in leases
+        if abs((lease["issue_time"] - secret["refresh_time"]).total_seconds()) <= 5
+    ]
+    if len(coherent) != 1:
+        raise ProcedureError(
+            f"exactly one runtime lease must match the mounted Secret refresh (observed {len(coherent)})"
+        )
+    selected = coherent[0]
+    if any(lease["issue_time"] >= selected["issue_time"] for lease in leases if lease is not selected):
+        raise ProcedureError("a runtime lease newer than the mounted credential exists")
+    return {**selected, "predecessor_handles": len(leases) - 1}
+
+
 def assert_lease_matches_refresh(lease: dict[str, Any], secret: dict[str, Any]) -> None:
     delta = abs((lease["issue_time"] - secret["refresh_time"]).total_seconds())
     if (
@@ -380,7 +406,7 @@ def exercise(args: argparse.Namespace) -> dict[str, Any]:
     attest_chain_state(remote)
     before_pod = pod_state(remote)
     before_secret = secret_state(remote)
-    lease = authority.single_runtime_lease()
+    lease = authority.runtime_lease_for(before_secret)
     assert_lease_matches_refresh(lease, before_secret)
     baseline_load = driver.run("baseline")
     if baseline_load.get("status") != "ready":
@@ -395,7 +421,7 @@ def exercise(args: argparse.Namespace) -> dict[str, Any]:
         # Close the race with ESO's ordinary refresh before the destructive step.
         if secret_state(remote) != before_secret or pod_state(remote) != before_pod:
             raise ProcedureError("baseline changed before revocation")
-        current = authority.single_runtime_lease()
+        current = authority.runtime_lease_for(before_secret)
         if current["fingerprint"] != lease["fingerprint"]:
             raise ProcedureError("runtime lease changed before revocation")
         authority.revoke(lease["id"])