Harden WP-0024 recovery execution gates

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa

tests/test_audit_core_database_lease_recovery.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import tempfile
+import unittest
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SPEC = importlib.util.spec_from_file_location(
+    "audit_core_database_lease_recovery",
+    ROOT / "scripts" / "audit-core-database-lease-recovery.py",
+)
+assert SPEC and SPEC.loader
+module = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(module)
+
+
+def approved_receipt(now: datetime) -> dict:
+    return {
+        "procedure": module.PROCEDURE,
+        "task_id": module.TASK_ID,
+        "status": "approved",
+        "approval_id": "operator-window-1",
+        "window": {
+            "start": (now - timedelta(minutes=1)).isoformat(),
+            "end": (now + timedelta(minutes=9)).isoformat(),
+        },
+        "abort_operator": "operator-a",
+        "owners": {
+            owner: {"acknowledged": True, "message_id": f"ack-{owner}"}
+            for owner in ("audit-core", "rapp-postgres", "railiance-platform")
+        },
+        "synthetic_load": {"contract_id": "load-1", "driver_revision": "abc123"},
+    }
+
+
+class DatabaseLeaseRecoveryTests(unittest.TestCase):
+    def test_approval_requires_exact_owner_and_window_contract(self) -> None:
+        now = datetime(2026, 8, 22, 12, 0, tzinfo=UTC)
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "approval.json"
+            path.write_text(json.dumps(approved_receipt(now)), encoding="utf-8")
+            result = module.load_approval(path, now, require_open_window=True)
+        self.assertEqual("operator-window-1", result["approval_id"])
+
+    def test_pending_or_overlong_approval_is_rejected(self) -> None:
+        now = datetime(2026, 8, 22, 12, 0, tzinfo=UTC)
+        receipt = approved_receipt(now)
+        receipt["status"] = "pending"
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "approval.json"
+            path.write_text(json.dumps(receipt), encoding="utf-8")
+            with self.assertRaises(module.ProcedureError):
+                module.load_approval(path, now, require_open_window=True)
+
+    def test_lease_must_align_to_refresh_and_have_headroom(self) -> None:
+        issue = datetime(2026, 8, 22, 12, 0, tzinfo=UTC)
+        lease = {
+            "issue_time": issue,
+            "expire_time": issue + timedelta(minutes=15),
+            "ttl": 800,
+        }
+        secret = {"refresh_time": issue + timedelta(seconds=1)}
+        module.assert_lease_matches_refresh(lease, secret)
+        lease["ttl"] = 20
+        with self.assertRaises(module.ProcedureError):
+            module.assert_lease_matches_refresh(lease, secret)
+
+    def test_load_driver_rejects_extra_evidence_fields(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            driver = Path(directory) / "driver"
+            driver.write_text(
+                "#!/bin/sh\nprintf '%s\\n' '{\"contract_id\":\"load-1\",\"fixture_id\":\"f-1\",\"status\":\"ready\",\"secret_values_observed\":false,\"token\":\"forbidden\"}'\n",
+                encoding="utf-8",
+            )
+            os.chmod(driver, 0o700)
+            revision = "sha256:" + module.hashlib.sha256(driver.read_bytes()).hexdigest()
+            with self.assertRaises(module.ProcedureError):
+                module.LoadDriver(driver, "load-1", revision).run("baseline")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_audit_core_recovery_preflight.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import importlib.util
+import json
+import tempfile
 import unittest
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 
@@ -96,6 +98,78 @@ host:      50 bytes
             {"username", "password", "host"}, module.secret_key_names(description)
         )
 
+    def test_snapshot_receipt_matches_live_cluster_and_freshness(self) -> None:
+        now = datetime(2026, 8, 22, 12, 0, tzinfo=UTC)
+        receipt = {
+            "receipt_version": 1,
+            "receipt_id": "snapshot-20260822",
+            "created_at": (now - timedelta(hours=1)).isoformat(),
+            "operator": "operator-a",
+            "source_cluster": "railiance01",
+            "source_namespace": "openbao",
+            "source_pod": "openbao-0",
+            "cluster_id": "cluster-1",
+            "raft_applied_index": 900,
+            "snapshot_created": True,
+            "source_initialized": True,
+            "source_unsealed": True,
+            "snapshot_sha256": "sha256:" + "ab" * 32,
+            "snapshot_encrypted": True,
+            "encrypted_snapshot_sha256": "sha256:" + "cd" * 32,
+            "encrypted_location_ref": "custody:openbao:snapshot-20260822",
+            "encrypted_copy_off_host": True,
+            "encryption_verified": True,
+            "hash_verified": True,
+            "no_secret_material_recorded": True,
+        }
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "receipt.json"
+            path.write_text(json.dumps(receipt), encoding="utf-8")
+            result = module.validate_snapshot_receipt(
+                path,
+                live_openbao={"cluster_id": "cluster-1", "raft_applied_index": 1000},
+                now=now,
+                max_age_hours=24,
+            )
+        self.assertTrue(result["verified"])
+        self.assertEqual(1.0, result["age_hours"])
+        self.assertFalse(result["secret_values_observed"])
+
+    def test_snapshot_receipt_rejects_stale_or_secret_material(self) -> None:
+        now = datetime(2026, 8, 22, 12, 0, tzinfo=UTC)
+        receipt = {
+            "receipt_version": 1,
+            "receipt_id": "snapshot-old",
+            "created_at": (now - timedelta(hours=25)).isoformat(),
+            "operator": "operator-a",
+            "source_cluster": "railiance01",
+            "source_namespace": "openbao",
+            "source_pod": "openbao-0",
+            "cluster_id": "cluster-1",
+            "raft_applied_index": 900,
+            "snapshot_created": True,
+            "source_initialized": True,
+            "source_unsealed": True,
+            "snapshot_sha256": "sha256:" + "ab" * 32,
+            "snapshot_encrypted": True,
+            "encrypted_snapshot_sha256": "sha256:" + "cd" * 32,
+            "encrypted_location_ref": "custody:old",
+            "encrypted_copy_off_host": True,
+            "encryption_verified": True,
+            "hash_verified": True,
+            "no_secret_material_recorded": True,
+        }
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "receipt.json"
+            path.write_text(json.dumps(receipt), encoding="utf-8")
+            with self.assertRaises(module.PreflightError):
+                module.validate_snapshot_receipt(
+                    path,
+                    live_openbao={"cluster_id": "cluster-1", "raft_applied_index": 1000},
+                    now=now,
+                    max_age_hours=24,
+                )
+
 
 if __name__ == "__main__":
     unittest.main()