Bound rejected backup fixture cleanup to successful replacement recovery

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

scripts/cleanup_nextcloud_recovery_fixture.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Remove only the rejected WP-0029 drill copy after successful real recovery."""
+import argparse
+import base64
+import json
+import os
+from pathlib import Path
+import urllib.error
+import urllib.request
+from migrate_nextcloud_backup_account import HOST, OPERATOR_PATH, NoRedirect, quota, require
+from state_hub_preflight_lane import bao, data
+
+ARTIFACT = 'wp0029-recovery-forgejo-dump-20260904T001507Z.zip.age'
+SIZE = 147171923
+
+
+def run(restore, receipt):
+    proof = json.loads(restore.read_text())
+    require(proof.get('status') == 'restored' and proof.get('cleanup') is True,
+            'successful_replacement_recovery_required')
+    require(proof.get('offsite_artifact') == 'wp0029-recovery-forgejo-dump-20260905-verified.zip.age',
+            'unexpected_replacement_artifact')
+    entry = data(bao(['read', '-format=json', OPERATOR_PATH]))['data']['data']
+    require(entry['BACKUP_USERNAME'] == 'Backup', 'backup_owner_required')
+    auth = (entry['BACKUP_USERNAME'], entry['BACKUP_PASSWORD'])
+    headers = {'Authorization': 'Basic '+base64.b64encode((auth[0]+':'+auth[1]).encode()).decode()}
+    url = HOST+'/remote.php/dav/files/Backup/railiance-backups/forgejo/'+ARTIFACT
+    opener = urllib.request.build_opener(NoRedirect())
+    receipt['quota_before'] = quota(auth)
+    try:
+        with opener.open(urllib.request.Request(url, method='HEAD', headers=headers), timeout=60) as response:
+            require(response.status == 200 and int(response.headers['Content-Length']) == SIZE,
+                    'rejected_fixture_identity_mismatch')
+            etag = response.headers.get('ETag')
+            require(bool(etag) and not etag.startswith('W/'), 'strong_fixture_etag_required')
+    except urllib.error.HTTPError as error:
+        if error.code != 404:
+            raise
+        receipt.update(status='already_absent', artifact=ARTIFACT)
+        return
+    headers['If-Match'] = etag
+    with opener.open(urllib.request.Request(url, method='DELETE', headers=headers), timeout=60) as response:
+        require(response.status == 204, 'fixture_cleanup_failed')
+    receipt.update(status='removed', artifact=ARTIFACT, delete_http_status=204, quota_after=quota(auth))
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument('--restore-receipt', required=True, type=Path)
+    p.add_argument('--receipt', required=True)
+    args = p.parse_args()
+    receipt = {'schema':'platform.rejected-recovery-fixture-cleanup.v1', 'status':'failed'}
+    fd = os.open(args.receipt, os.O_WRONLY|os.O_CREAT|os.O_EXCL, 0o600)
+    try:
+        run(args.restore_receipt, receipt)
+    except Exception:
+        receipt['error'] = 'bounded_fixture_cleanup_failed'
+    finally:
+        with os.fdopen(fd, 'w') as output:
+            json.dump(receipt, output, indent=2)
+    return int(receipt['status'] == 'failed')
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

tests/test_nextcloud_recovery_cleanup.py

@@ -0,0 +1,53 @@
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
+import cleanup_nextcloud_recovery_fixture as cleanup
+
+
+class RecoveryFixtureCleanup(unittest.TestCase):
+    def proof(self, directory, status='restored'):
+        path = Path(directory) / 'proof.json'
+        path.write_text(json.dumps({'status': status, 'cleanup': True,
+            'offsite_artifact': 'wp0029-recovery-forgejo-dump-20260905-verified.zip.age'}))
+        return path
+
+    def test_no_credentials_before_successful_recovery(self):
+        with tempfile.TemporaryDirectory() as directory, patch.object(cleanup, 'bao') as bao:
+            with self.assertRaises(Exception):
+                cleanup.run(self.proof(directory, 'failed'), {})
+            bao.assert_not_called()
+
+    def exercise(self, size):
+        opener = MagicMock()
+        head = MagicMock(status=200, headers={'Content-Length': str(size), 'ETag': '"fixture-version"'})
+        delete = MagicMock(status=204)
+        opener.open.side_effect = [MagicMock(__enter__=lambda _: head, __exit__=lambda *a: False),
+                                    MagicMock(__enter__=lambda _: delete, __exit__=lambda *a: False)]
+        with tempfile.TemporaryDirectory() as directory, \
+                patch.object(cleanup, 'bao'), \
+                patch.object(cleanup, 'data', return_value={'data': {'data': {'BACKUP_USERNAME': 'Backup', 'BACKUP_PASSWORD': 'fixture'}}}), \
+                patch.object(cleanup, 'quota', return_value={}), \
+                patch.object(cleanup.urllib.request, 'build_opener', return_value=opener):
+            receipt = {}
+            if size != cleanup.SIZE:
+                with self.assertRaises(Exception):
+                    cleanup.run(self.proof(directory), receipt)
+                self.assertEqual(opener.open.call_count, 1)
+            else:
+                cleanup.run(self.proof(directory), receipt)
+                request = opener.open.call_args_list[1].args[0]
+                self.assertEqual(request.method, 'DELETE')
+                self.assertTrue(request.full_url.endswith('/forgejo/' + cleanup.ARTIFACT))
+                self.assertEqual(request.get_header('If-match'), '"fixture-version"')
+                self.assertEqual(receipt['status'], 'removed')
+
+    def test_mismatched_object_is_preserved(self):
+        self.exercise(1)
+
+    def test_exact_object_has_conditional_delete(self):
+        self.exercise(cleanup.SIZE)