From a9c1da4bd7334777644e32b3640a47f1e3ac0b20 Mon Sep 17 00:00:00 2001 From: codex Date: Sun, 6 Sep 2026 01:05:27 +0200 Subject: [PATCH] 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 | 65 +++++++++++++++++++ tests/test_nextcloud_recovery_cleanup.py | 53 +++++++++++++++ 2 files changed, 118 insertions(+) create mode 100644 scripts/cleanup_nextcloud_recovery_fixture.py create mode 100644 tests/test_nextcloud_recovery_cleanup.py diff --git a/scripts/cleanup_nextcloud_recovery_fixture.py b/scripts/cleanup_nextcloud_recovery_fixture.py new file mode 100644 index 0000000..012b4c0 --- /dev/null +++ b/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()) diff --git a/tests/test_nextcloud_recovery_cleanup.py b/tests/test_nextcloud_recovery_cleanup.py new file mode 100644 index 0000000..6059f41 --- /dev/null +++ b/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)